KAFKA-9106 make metrics exposed via jmx configurable #7674
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,7 +18,10 @@ | |
|
|
||
| import org.apache.kafka.common.KafkaException; | ||
| import org.apache.kafka.common.MetricName; | ||
| import org.apache.kafka.common.Reconfigurable; | ||
| import org.apache.kafka.common.config.ConfigException; | ||
| import org.apache.kafka.common.utils.Sanitizer; | ||
| import org.apache.kafka.common.utils.Utils; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
|
|
@@ -36,16 +39,32 @@ | |
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.function.Predicate; | ||
| import java.util.regex.Pattern; | ||
| import java.util.regex.PatternSyntaxException; | ||
|
|
||
| /** | ||
| * Register metrics in JMX as dynamic mbeans based on the metric names | ||
| */ | ||
| public class JmxReporter implements MetricsReporter, Reconfigurable { | ||
|
Contributor
There was a problem hiding this comment. I think we should make the
Member
Author
There was a problem hiding this comment. We might be able to add default implementation for Reconfigurable methods to the MetricsReporter interface, but that would theoretically break binary compatibility for existing plugins.
Contributor
There was a problem hiding this comment. Adding a new method with default implementation to a Java interface doesn't break binary compatibility. See https://docs.oracle.com/javase/tutorial/java/IandI/defaultmethods.html
Contributor
There was a problem hiding this comment. I guess there is a separate question as to whether having a superclass implement a new interface is a binary-compatible change. Section 13.4.4 of the Java spec seems to imply that it is. https://docs.oracle.com/javase/specs/jls/se8/html/jls-13.html#jls-13.4.4
Member
Author
There was a problem hiding this comment. According to https://wiki.eclipse.org/Evolving_Java-based_APIs_2#Evolving_API_Interfaces adding a default method is binary incompatible. The reasoning is that if the implementing class already implements a different interface with a default method matching the signature of the added new default methods it would break. In practice this will almost never be the case, but it is theoretically possible. I don't know how strict we are about those things in Kafka.
Contributor
There was a problem hiding this comment. We do consider adding new default methods to interfaces to be a compatible change in Kafka. I agree that it is theoretically possible for an existing implementation to contain a different method with the same name, and have an issue that way. But if we treat every method addition as a compatibility break, it defeats the whole point of default methods in Java. This would make it very hard to evolve interfaces and would lead to some pretty ugly hacks. In practice people choose descriptive methods names for public interfaces which makes this is a non-issue. For example, I think the chances that an existing MetricsReporter implements reconfigurableConfigs, validateReconfiguration, or reconfigure are basically zero. We also know what most of the widely-deployed MetricsReporters are, so we can check them.
Contributor
There was a problem hiding this comment. MetricsReporters have been optionally reconfigurable since 1.1.0, before Java 7 support was dropped. Any metrics reporter implementation that implements Reconfigurable is reconfigured by the broker when any of its configs changes. Since then, we have adopted the same approach for other interfaces like Authorizer as well.
Member
Author
There was a problem hiding this comment. @rajinisivaram do I understand correctly that you suggest we leave the Reconfigurable interface optional then?
Contributor
There was a problem hiding this comment. @xvrl: The only reason the interface was optional is that we needed to retain support for Java 7, which didn't have default methods. Since we don't have to worry about Java 7 any more, let's make the interface part of the type.
Member
Author
There was a problem hiding this comment. sorry, I misundertood; I updated the interface to be Reconfigurable and added default methods for backwards compatibility. |
||
|
|
||
| public static final String METRICS_CONFIG_PREFIX = "metrics.jmx."; | ||
|
|
||
| public static final String BLACKLIST_CONFIG = METRICS_CONFIG_PREFIX + "blacklist"; | ||
| public static final String WHITELIST_CONFIG = METRICS_CONFIG_PREFIX + "whitelist"; | ||
|
|
||
| public static final Set<String> RECONFIGURABLE_CONFIGS = Utils.mkSet(WHITELIST_CONFIG, | ||
| BLACKLIST_CONFIG); | ||
|
|
||
| public static final String DEFAULT_WHITELIST = ".*"; | ||
| public static final String DEFAULT_BLACKLIST = ""; | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(JmxReporter.class); | ||
| private static final Object LOCK = new Object(); | ||
| private String prefix; | ||
| private final Map<String, KafkaMbean> mbeans = new HashMap<>(); | ||
| private Predicate<String> mbeanPredicate = s -> true; | ||
|
|
||
| public JmxReporter() { | ||
| this(""); | ||
|
|
@@ -59,15 +78,46 @@ public JmxReporter(String prefix) { | |
| } | ||
|
|
||
| @Override | ||
| public void configure(Map<String, ?> configs) { | ||
| reconfigure(configs); | ||
| } | ||
|
|
||
| @Override | ||
| public Set<String> reconfigurableConfigs() { | ||
| return RECONFIGURABLE_CONFIGS; | ||
| } | ||
|
|
||
| @Override | ||
| public void validateReconfiguration(Map<String, ?> configs) throws ConfigException { | ||
| compilePredicate(configs); | ||
| } | ||
|
|
||
| @Override | ||
| public void reconfigure(Map<String, ?> configs) { | ||
| synchronized (LOCK) { | ||
| this.mbeanPredicate = JmxReporter.compilePredicate(configs); | ||
|
|
||
| mbeans.forEach((name, mbean) -> { | ||
| if (mbeanPredicate.test(name)) { | ||
| reregister(mbean); | ||
| } else { | ||
| unregister(mbean); | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void init(List<KafkaMetric> metrics) { | ||
| synchronized (LOCK) { | ||
| for (KafkaMetric metric : metrics) | ||
| addAttribute(metric); | ||
|
|
||
| mbeans.forEach((name, mbean) -> { | ||
| if (mbeanPredicate.test(name)) { | ||
| reregister(mbean); | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -78,8 +128,10 @@ public boolean containsMbean(String mbeanName) { | |
| @Override | ||
| public void metricChange(KafkaMetric metric) { | ||
| synchronized (LOCK) { | ||
| String mbeanName = addAttribute(metric); | ||
| if (mbeanName != null && mbeanPredicate.test(mbeanName)) { | ||
| reregister(mbeans.get(mbeanName)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -93,7 +145,7 @@ public void metricRemoval(KafkaMetric metric) { | |
| if (mbean.metrics.isEmpty()) { | ||
| unregister(mbean); | ||
| mbeans.remove(mBeanName); | ||
| } else if (mbeanPredicate.test(mBeanName)) | ||
| reregister(mbean); | ||
| } | ||
| } | ||
|
|
@@ -107,15 +159,15 @@ private KafkaMbean removeAttribute(KafkaMetric metric, String mBeanName) { | |
| return mbean; | ||
| } | ||
|
|
||
| private String addAttribute(KafkaMetric metric) { | ||
| try { | ||
| MetricName metricName = metric.metricName(); | ||
| String mBeanName = getMBeanName(prefix, metricName); | ||
| if (!this.mbeans.containsKey(mBeanName)) | ||
| mbeans.put(mBeanName, new KafkaMbean(mBeanName)); | ||
| KafkaMbean mbean = this.mbeans.get(mBeanName); | ||
| mbean.setAttribute(metricName.name(), metric); | ||
| return mBeanName; | ||
| } catch (JMException e) { | ||
| throw new KafkaException("Error creating mbean attribute for metricName :" + metric.metricName(), e); | ||
| } | ||
|
|
@@ -244,4 +296,27 @@ public AttributeList setAttributes(AttributeList list) { | |
|
|
||
| } | ||
|
|
||
| public static Predicate<String> compilePredicate(Map<String, ?> configs) { | ||
| String whitelist = (String) configs.get(WHITELIST_CONFIG); | ||
| String blacklist = (String) configs.get(BLACKLIST_CONFIG); | ||
|
|
||
| if (whitelist == null) { | ||
| whitelist = DEFAULT_WHITELIST; | ||
| } | ||
|
|
||
| if (blacklist == null) { | ||
| blacklist = DEFAULT_BLACKLIST; | ||
| } | ||
|
|
||
| try { | ||
| Pattern whitelistPattern = Pattern.compile(whitelist); | ||
| Pattern blacklistPattern = Pattern.compile(blacklist); | ||
|
|
||
| return s -> whitelistPattern.matcher(s).matches() | ||
| && !blacklistPattern.matcher(s).matches(); | ||
| } catch (PatternSyntaxException e) { | ||
| throw new ConfigException("JMX filter for configuration" + METRICS_CONFIG_PREFIX | ||
| + ".(whitelist/blacklist) is not a valid regular expression"); | ||
| } | ||
| } | ||
| } | ||
Uh oh!
There was an error while loading. Please reload this page.