diff --git a/.azure/templates/jobs/build_java.yaml b/.azure/templates/jobs/build_java.yaml index c418337..acb05e1 100644 --- a/.azure/templates/jobs/build_java.yaml +++ b/.azure/templates/jobs/build_java.yaml @@ -28,18 +28,14 @@ jobs: - template: '../steps/prerequisites/install_java.yaml' parameters: JDK_VERSION: $(jdk_version) - - bash: "mvn ${MVN_ARGS} verify" - displayName: "Build & Test Java" + - bash: "mvn ${MVN_ARGS} verify spotbugs:check" + displayName: "Build & Test Java + Spotbugs" env: BUILD_REASON: $(Build.Reason) BRANCH: $(Build.SourceBranch) TESTCONTAINERS_RYUK_DISABLED: "TRUE" TESTCONTAINERS_CHECKS_DISABLE: "TRUE" MVN_ARGS: "-e -V -B" - - bash: "mvn ${MVN_ARGS} spotbugs:check" - displayName: "Spotbugs" - env: - MVN_ARGS: "-e -V -B" # We have to TAR the target directory to maintain the permissions of # the files which would otherwise change when downloading the artifact - bash: tar -cvpf target.tar ./target diff --git a/README.md b/README.md index 0836744..c9832ed 100644 --- a/README.md +++ b/README.md @@ -34,50 +34,54 @@ The metrics reporter has the following configurations: ## Running -### Kafka Brokers +### Kafka Brokers and Controllers + +To use the reporter with Kafka brokers and controllers, add the following to your broker configuration: -To use the reporter with Kafka brokers, add the following to your broker configuration: ```properties -metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter -kafka.metrics.reporters=io.strimzi.kafka.metrics.YammerPrometheusMetricsReporter +metric.reporters=io.strimzi.kafka.metrics.prometheus.ServerKafkaMetricsReporter +kafka.metrics.reporters=io.strimzi.kafka.metrics.prometheus.ServerYammerMetricsReporter auto.include.jmx.reporter=false ``` ### Kafka Clients To use the reporter with Kafka producers, consumers or admin clients, add the following to your client configuration: + ```properties -metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter auto.include.jmx.reporter=false ``` ### Kafka Connect and Kafka Streams To use the reporter with Kafka Connect and Kafka Streams, add the following to your Connect runtime or Streams application configuration: + ```properties -metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter auto.include.jmx.reporter=false -admin.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +admin.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter admin.auto.include.jmx.reporter=false -producer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +producer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter producer.auto.include.jmx.reporter=false -consumer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +consumer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter consumer.auto.include.jmx.reporter=false ``` When setting configurations for the Prometheus metrics reporter, they also need to be set with the `admin.`, `producer.` and `consumer.`. For example, to set the `listener` to `http://:8081`: + ```properties -metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter prometheus.metrics.reporter.listener=http://:8081 auto.include.jmx.reporter=false -admin.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +admin.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter admin.prometheus.metrics.reporter.listener=http://:8081 admin.auto.include.jmx.reporter=false -producer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +producer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter producer.prometheus.metrics.reporter.listener=http://:8081 producer.auto.include.jmx.reporter=false -consumer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter +consumer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter consumer.prometheus.metrics.reporter.listener=http://:8081 consumer.auto.include.jmx.reporter=false ``` diff --git a/client-metrics-reporter/pom.xml b/client-metrics-reporter/pom.xml new file mode 100644 index 0000000..8314ea4 --- /dev/null +++ b/client-metrics-reporter/pom.xml @@ -0,0 +1,75 @@ + + + 4.0.0 + + io.strimzi + metrics-reporter + 1.0.0-SNAPSHOT + + + client-metrics-reporter + + + + org.apache.kafka + kafka-clients + + + org.slf4j + slf4j-api + + + com.github.spotbugs + spotbugs-annotations + + + + io.prometheus + prometheus-metrics-model + + + io.prometheus + prometheus-metrics-instrumentation-jvm + + + io.prometheus + prometheus-metrics-exporter-httpserver + + + + org.junit.jupiter + junit-jupiter-api + + + org.slf4j + slf4j-simple + + + io.strimzi + strimzi-test-container + + + org.testcontainers + testcontainers + + + + + + + org.apache.maven.plugins + maven-jar-plugin + 3.4.2 + + + + test-jar + + + + + + + diff --git a/src/main/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporter.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporter.java similarity index 53% rename from src/main/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporter.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporter.java index 7c39995..08af69e 100644 --- a/src/main/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporter.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporter.java @@ -2,14 +2,17 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.prometheus.metrics.model.registry.PrometheusRegistry; import io.prometheus.metrics.model.snapshots.PrometheusNaming; -import io.strimzi.kafka.metrics.http.HttpServers; -import io.strimzi.kafka.metrics.kafka.KafkaCollector; -import io.strimzi.kafka.metrics.kafka.KafkaMetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; +import io.strimzi.kafka.metrics.prometheus.http.HttpServers; +import io.strimzi.kafka.metrics.prometheus.kafka.KafkaCollector; +import io.strimzi.kafka.metrics.prometheus.kafka.KafkaMetricWrapper; import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.MetricsContext; @@ -17,48 +20,55 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.regex.Pattern; /** - * MetricsReporter implementation that expose Kafka metrics in the Prometheus format. - * This can be used by Kafka brokers and clients. + * {@link MetricsReporter} implementation that exposes Kafka client metrics in the Prometheus format. */ -public class KafkaPrometheusMetricsReporter implements MetricsReporter { - - private static final Logger LOG = LoggerFactory.getLogger(KafkaPrometheusMetricsReporter.class); - - private final PrometheusRegistry registry; - private final KafkaCollector kafkaCollector; - @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the configure method - private PrometheusMetricsReporterConfig config; - @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the configure method - private Optional httpServer; +public class ClientMetricsReporter extends AbstractReporter implements MetricsReporter { + + private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsReporter.class); + static final Set PREFIXES = Set.of( + "kafka.admin.client", + "kafka.consumer", + "kafka.producer", + "kafka.connect", + "kafka.streams" + ); + + final PrometheusRegistry registry; + final KafkaCollector kafkaCollector; + + private ClientMetricsReporterConfig config; + Optional httpServer = Optional.empty(); @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the contextChange method - private String prefix; + String prefix; /** * Constructor */ - public KafkaPrometheusMetricsReporter() { + public ClientMetricsReporter() { registry = PrometheusRegistry.defaultRegistry; kafkaCollector = KafkaCollector.getCollector(PrometheusCollector.register(registry)); + kafkaCollector.addReporter(this); } // for testing - KafkaPrometheusMetricsReporter(PrometheusRegistry registry, KafkaCollector kafkaCollector) { + ClientMetricsReporter(PrometheusRegistry registry, KafkaCollector kafkaCollector) { this.registry = registry; this.kafkaCollector = kafkaCollector; + kafkaCollector.addReporter(this); } @Override public void configure(Map map) { - config = new PrometheusMetricsReporterConfig(map, registry); + config = new ClientMetricsReporterConfig(map, registry); httpServer = config.startHttpServer(); - LOG.debug("KafkaPrometheusMetricsReporter configured with {}", config); + LOG.debug("ClientMetricsReporter configured with {}", config); } @Override @@ -68,23 +78,21 @@ public void init(List metrics) { } } + @Override public void metricChange(KafkaMetric metric) { String prometheusName = KafkaMetricWrapper.prometheusName(prefix, metric.metricName()); - if (!config.isAllowed(prometheusName)) { - LOG.trace("Ignoring metric {} as it does not match the allowlist", prometheusName); - } else { - MetricWrapper metricWrapper = new KafkaMetricWrapper(prometheusName, metric, metric.metricName().name()); - kafkaCollector.addMetric(metric.metricName(), metricWrapper); - } + MetricWrapper metricWrapper = new KafkaMetricWrapper(prometheusName, metric, metric.metricName().name()); + addMetric(metric, metricWrapper); } @Override public void metricRemoval(KafkaMetric metric) { - kafkaCollector.removeMetric(metric.metricName()); + removeMetric(metric); } @Override public void close() { + kafkaCollector.removeReporter(this); httpServer.ifPresent(HttpServers::release); } @@ -98,12 +106,15 @@ public void validateReconfiguration(Map configs) throws ConfigExcepti @Override public Set reconfigurableConfigs() { - return Collections.emptySet(); + return Set.of(); } @Override public void contextChange(MetricsContext metricsContext) { String prefix = metricsContext.contextLabels().get(MetricsContext.NAMESPACE); + if (!PREFIXES.contains(prefix)) { + throw new IllegalStateException("ClientMetricsReporter should only be used in Kafka clients"); + } this.prefix = PrometheusNaming.prometheusName(prefix); } @@ -111,4 +122,10 @@ public void contextChange(MetricsContext metricsContext) { Optional getPort() { return Optional.ofNullable(httpServer.isPresent() ? httpServer.get().port() : null); } + + @Override + protected Pattern allowlist() { + return config.allowlist(); + } + } diff --git a/src/main/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfig.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfig.java similarity index 84% rename from src/main/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfig.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfig.java index ac2c2a0..4aeb629 100644 --- a/src/main/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfig.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfig.java @@ -2,12 +2,12 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.strimzi.kafka.metrics.http.HttpServers; -import io.strimzi.kafka.metrics.http.Listener; +import io.strimzi.kafka.metrics.prometheus.http.HttpServers; +import io.strimzi.kafka.metrics.prometheus.http.Listener; import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigException; @@ -22,11 +22,11 @@ import java.util.stream.Collectors; /** -* Configuration for the PrometheusMetricsReporter implementation. +* Configuration for {@link ClientMetricsReporter}. */ -public class PrometheusMetricsReporterConfig extends AbstractConfig { +public class ClientMetricsReporterConfig extends AbstractConfig { - private static final Logger LOG = LoggerFactory.getLogger(PrometheusMetricsReporterConfig.class); + private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsReporterConfig.class); private static final String CONFIG_PREFIX = "prometheus.metrics.reporter."; /** @@ -62,15 +62,15 @@ public class PrometheusMetricsReporterConfig extends AbstractConfig { public static final String ALLOWLIST_CONFIG_DEFAULT = ".*"; private static final String ALLOWLIST_CONFIG_DOC = "A comma separated list of regex patterns to specify the metrics to collect."; - private static final ConfigDef CONFIG_DEF = new ConfigDef() + static final ConfigDef CONFIG_DEF = new ConfigDef() .define(LISTENER_CONFIG, ConfigDef.Type.STRING, LISTENER_CONFIG_DEFAULT, new Listener.ListenerValidator(), ConfigDef.Importance.HIGH, LISTENER_CONFIG_DOC) .define(ALLOWLIST_CONFIG, ConfigDef.Type.LIST, ALLOWLIST_CONFIG_DEFAULT, ConfigDef.Importance.HIGH, ALLOWLIST_CONFIG_DOC) .define(LISTENER_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN, LISTENER_ENABLE_CONFIG_DEFAULT, ConfigDef.Importance.HIGH, LISTENER_ENABLE_CONFIG_DOC); - private final Listener listener; - private final boolean listenerEnabled; - private final Pattern allowlist; - private final PrometheusRegistry registry; + final Listener listener; + final boolean listenerEnabled; + final PrometheusRegistry registry; + final Pattern allowlist; /** * Constructor. @@ -79,7 +79,7 @@ public class PrometheusMetricsReporterConfig extends AbstractConfig { * @param registry the metrics registry */ @SuppressFBWarnings("CT_CONSTRUCTOR_THROW") - public PrometheusMetricsReporterConfig(Map props, PrometheusRegistry registry) { + public ClientMetricsReporterConfig(Map props, PrometheusRegistry registry) { super(CONFIG_DEF, props); this.listener = Listener.parseListener(getString(LISTENER_CONFIG)); this.allowlist = compileAllowlist(getList(ALLOWLIST_CONFIG)); @@ -97,7 +97,15 @@ public boolean isAllowed(String name) { return allowlist.matcher(name).matches(); } - private Pattern compileAllowlist(List allowlist) { + /** + * The configured allowlist. + * @return The Pattern for the allowlist + */ + public Pattern allowlist() { + return allowlist; + } + + Pattern compileAllowlist(List allowlist) { for (String entry : allowlist) { try { Pattern.compile(entry); @@ -129,7 +137,7 @@ public boolean isListenerEnabled() { @Override public String toString() { - return "PrometheusMetricsReporterConfig{" + + return "ClientMetricsReporterConfig{" + ", listener=" + listener + ", listenerEnabled=" + listenerEnabled + ", allowlist=" + allowlist + diff --git a/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporter.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporter.java new file mode 100644 index 0000000..b2fb9bb --- /dev/null +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporter.java @@ -0,0 +1,105 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.common; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.regex.Pattern; + +/** + * Common reporter logic to track metrics that match an allowlist pattern. This filters the metrics as they are added + * and removed so when Prometheus scrapes the /metrics endpoint, we just have to convert them to the Prometheus format. + */ +public abstract class AbstractReporter { + + private static final Logger LOG = LoggerFactory.getLogger(AbstractReporter.class); + + // Metrics that match the allowlist + private final Map allowedMetrics = new ConcurrentHashMap<>(); + // Metrics that don't match the allowlist. This is only used by reporters that are reconfigurable so if the + // allowlist is updated we can update the matching metrics. + private final Map disallowedMetrics = new ConcurrentHashMap<>(); + + /** + * The current allowlist + * @return A {@link Pattern} representing the allowlist + */ + protected abstract Pattern allowlist(); + + /** + * Whether the reporter is reconfigurable. + * @return true for server side reporters, otherwise false + */ + protected boolean isReconfigurable() { + return false; + } + + private boolean matches(String name) { + return allowlist().matcher(name).matches(); + } + + /** + * Add a metric to be collected. + * @param name The name of the metric to add. + * @param metric The metric to add. + */ + public void addMetric(Object name, MetricWrapper metric) { + if (matches(metric.prometheusName())) { + allowedMetrics.put(name, metric); + } else { + LOG.trace("Ignoring metric {} as it does not match the allowlist", metric.prometheusName()); + if (isReconfigurable()) { + disallowedMetrics.put(name, metric); + } + } + } + + /** + * Remove a metric from collection. + * @param name The name of metric to remove. + */ + public void removeMetric(Object name) { + allowedMetrics.remove(name); + if (isReconfigurable()) { + disallowedMetrics.remove(name); + } + } + + /** + * Retrieve the allowed metrics. + * @return A collection of MetricWrapper + */ + public Collection allowedMetrics() { + return allowedMetrics.values(); + } + + /** + * Update the allowed metrics based on the current allowlist pattern. + */ + public void updateAllowedMetrics() { + if (!isReconfigurable()) return; + Map newAllowedMetrics = new HashMap<>(); + for (Map.Entry entry : disallowedMetrics.entrySet()) { + String name = entry.getValue().prometheusName(); + if (matches(name)) { + newAllowedMetrics.put(entry.getKey(), entry.getValue()); + disallowedMetrics.remove(entry.getKey()); + } + } + for (Map.Entry entry : allowedMetrics.entrySet()) { + String name = entry.getValue().prometheusName(); + if (!matches(name)) { + disallowedMetrics.put(entry.getKey(), entry.getValue()); + allowedMetrics.remove(entry.getKey()); + } + } + allowedMetrics.putAll(newAllowedMetrics); + } +} diff --git a/src/main/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilder.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilder.java similarity index 98% rename from src/main/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilder.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilder.java index 6dc8ef9..7c138a8 100644 --- a/src/main/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilder.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilder.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.model.snapshots.CounterSnapshot; import io.prometheus.metrics.model.snapshots.GaugeSnapshot; diff --git a/src/main/java/io/strimzi/kafka/metrics/MetricWrapper.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricWrapper.java similarity index 96% rename from src/main/java/io/strimzi/kafka/metrics/MetricWrapper.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricWrapper.java index 5046535..d62e115 100644 --- a/src/main/java/io/strimzi/kafka/metrics/MetricWrapper.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricWrapper.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.model.snapshots.Labels; diff --git a/src/main/java/io/strimzi/kafka/metrics/MetricsCollector.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricsCollector.java similarity index 90% rename from src/main/java/io/strimzi/kafka/metrics/MetricsCollector.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricsCollector.java index c42121d..0d6dce8 100644 --- a/src/main/java/io/strimzi/kafka/metrics/MetricsCollector.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/MetricsCollector.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.model.snapshots.MetricSnapshot; diff --git a/src/main/java/io/strimzi/kafka/metrics/PrometheusCollector.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollector.java similarity index 88% rename from src/main/java/io/strimzi/kafka/metrics/PrometheusCollector.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollector.java index fb7a3d0..a6ce9e1 100644 --- a/src/main/java/io/strimzi/kafka/metrics/PrometheusCollector.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollector.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; import io.prometheus.metrics.model.registry.MultiCollector; @@ -15,8 +15,7 @@ import java.util.concurrent.atomic.AtomicBoolean; /** - * Prometheus Collector to store and export metrics retrieved by {@link KafkaPrometheusMetricsReporter} - * and {@link YammerPrometheusMetricsReporter}. + * Prometheus Collector to store and export metrics from {@link MetricsCollector} implementations. */ public class PrometheusCollector implements MultiCollector { @@ -26,7 +25,10 @@ public class PrometheusCollector implements MultiCollector { // At runtime this should contain at most one instance of KafkaCollector and one instance of YammerCollector private final List collectors = new ArrayList<>(); - /* for testing */ PrometheusCollector() { } + /** + * Constructor used for testing + */ + /* for testing */ public PrometheusCollector() { } /** * Register this Collector with the provided Prometheus registry diff --git a/src/main/java/io/strimzi/kafka/metrics/http/HttpServers.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/HttpServers.java similarity index 98% rename from src/main/java/io/strimzi/kafka/metrics/http/HttpServers.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/HttpServers.java index 3450e0b..1fedf0e 100644 --- a/src/main/java/io/strimzi/kafka/metrics/http/HttpServers.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/HttpServers.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.http; +package io.strimzi.kafka.metrics.prometheus.http; import io.prometheus.metrics.exporter.httpserver.HTTPServer; import io.prometheus.metrics.model.registry.PrometheusRegistry; diff --git a/src/main/java/io/strimzi/kafka/metrics/http/Listener.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/Listener.java similarity index 90% rename from src/main/java/io/strimzi/kafka/metrics/http/Listener.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/Listener.java index 6dbe2c7..dc70f32 100644 --- a/src/main/java/io/strimzi/kafka/metrics/http/Listener.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/http/Listener.java @@ -2,9 +2,9 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.http; +package io.strimzi.kafka.metrics.prometheus.http; -import io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig; +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigException; @@ -12,10 +12,10 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; -import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.LISTENER_CONFIG; +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.LISTENER_CONFIG; /** - * Class parsing and handling the listener specified via {@link PrometheusMetricsReporterConfig#LISTENER_CONFIG} for + * Class parsing and handling the listener specified via {@link ClientMetricsReporterConfig#LISTENER_CONFIG} for * the HTTP server used to expose the metrics. */ public class Listener { diff --git a/src/main/java/io/strimzi/kafka/metrics/kafka/KafkaCollector.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollector.java similarity index 51% rename from src/main/java/io/strimzi/kafka/metrics/kafka/KafkaCollector.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollector.java index 065031c..5a6a010 100644 --- a/src/main/java/io/strimzi/kafka/metrics/kafka/KafkaCollector.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollector.java @@ -2,17 +2,17 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.kafka; +package io.strimzi.kafka.metrics.prometheus.kafka; import io.prometheus.metrics.model.snapshots.GaugeSnapshot; import io.prometheus.metrics.model.snapshots.InfoSnapshot; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.MetricSnapshot; -import io.strimzi.kafka.metrics.DataPointSnapshotBuilder; -import io.strimzi.kafka.metrics.MetricWrapper; -import io.strimzi.kafka.metrics.MetricsCollector; -import io.strimzi.kafka.metrics.PrometheusCollector; -import org.apache.kafka.common.MetricName; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.DataPointSnapshotBuilder; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.MetricsCollector; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; import org.apache.kafka.common.metrics.KafkaMetric; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; @@ -33,7 +34,7 @@ public class KafkaCollector implements MetricsCollector { private static final KafkaCollector INSTANCE = new KafkaCollector(); private static final AtomicBoolean REGISTERED = new AtomicBoolean(false); - private final Map kafkaMetrics = new ConcurrentHashMap<>(); + private final Set reporters = ConcurrentHashMap.newKeySet(); /* for testing */ KafkaCollector() { } @@ -59,22 +60,19 @@ public static KafkaCollector getCollector(PrometheusCollector prometheusCollecto } /** - * Add a Kafka metric to be collected. - * - * @param name The name of the Kafka metric to add. - * @param metric The Kafka metric to add. + * Add an {@link AbstractReporter} instance to this collector + * @param reporter The reporter instance to add */ - public void addMetric(MetricName name, MetricWrapper metric) { - kafkaMetrics.put(name, metric); + public void addReporter(AbstractReporter reporter) { + reporters.add(reporter); } /** - * Remove a Kafka metric from collection. - * - * @param name The name of Kafka metric to remove. + * Remove an {@link AbstractReporter} instance from this collector + * @param reporter The reporter instance to remove */ - public void removeMetric(MetricName name) { - kafkaMetrics.remove(name); + public void removeReporter(AbstractReporter reporter) { + reporters.remove(reporter); } /** @@ -85,19 +83,21 @@ public void removeMetric(MetricName name) { @Override public List collect() { Map> builders = new HashMap<>(); - for (MetricWrapper metricWrapper : kafkaMetrics.values()) { - String prometheusMetricName = metricWrapper.prometheusName(); - Object metricValue = ((KafkaMetric) metricWrapper.metric()).metricValue(); - Labels labels = metricWrapper.labels(); - LOG.debug("Collecting Kafka metric {} with the following labels: {}", prometheusMetricName, labels); + for (AbstractReporter reporter : reporters) { + for (MetricWrapper metricWrapper : reporter.allowedMetrics()) { + String prometheusMetricName = metricWrapper.prometheusName(); + Object metricValue = ((KafkaMetric) metricWrapper.metric()).metricValue(); + Labels labels = metricWrapper.labels(); + LOG.debug("Collecting Kafka metric {} with the following labels: {}", prometheusMetricName, labels); - if (metricValue instanceof Number) { - double value = ((Number) metricValue).doubleValue(); - GaugeSnapshot.Builder builder = (GaugeSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> GaugeSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.gaugeDataPoint(labels, value)); - } else { - InfoSnapshot.Builder builder = (InfoSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> InfoSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.infoDataPoint(labels, metricValue, metricWrapper.attribute())); + if (metricValue instanceof Number) { + double value = ((Number) metricValue).doubleValue(); + GaugeSnapshot.Builder builder = (GaugeSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> GaugeSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.gaugeDataPoint(labels, value)); + } else { + InfoSnapshot.Builder builder = (InfoSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> InfoSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.infoDataPoint(labels, metricValue, metricWrapper.attribute())); + } } } List snapshots = new ArrayList<>(); diff --git a/src/main/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapper.java b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapper.java similarity index 95% rename from src/main/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapper.java rename to client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapper.java index 2a2c422..6fda5ad 100644 --- a/src/main/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapper.java +++ b/client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapper.java @@ -2,11 +2,11 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.kafka; +package io.strimzi.kafka.metrics.prometheus.kafka; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.PrometheusNaming; -import io.strimzi.kafka.metrics.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.metrics.KafkaMetric; import org.slf4j.Logger; diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfigTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfigTest.java new file mode 100644 index 0000000..f848b2a --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfigTest.java @@ -0,0 +1,108 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import io.strimzi.kafka.metrics.prometheus.http.HttpServers; +import org.apache.kafka.common.config.ConfigException; +import org.junit.jupiter.api.Test; + +import java.net.BindException; +import java.util.Map; +import java.util.Optional; + +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.ALLOWLIST_CONFIG; +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.LISTENER_CONFIG; +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.LISTENER_ENABLE_CONFIG; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +public class ClientMetricsReporterConfigTest { + + @Test + public void testDefaults() { + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(Map.of(), new PrometheusRegistry()); + assertEquals(ClientMetricsReporterConfig.LISTENER_CONFIG_DEFAULT, config.listener()); + assertTrue(config.isAllowed("random_name")); + } + + @Test + public void testOverrides() { + Map props = Map.of( + LISTENER_CONFIG, "http://:0", + ALLOWLIST_CONFIG, "kafka_server.*"); + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + + assertEquals("http://:0", config.listener()); + assertFalse(config.isAllowed("random_name")); + assertTrue(config.isAllowed("kafka_server_metric")); + } + + @Test + public void testAllowList() { + Map props = Map.of(ALLOWLIST_CONFIG, "kafka_server.*,kafka_network.*"); + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + + assertTrue(config.allowlist().pattern().contains("kafka_server.*")); + assertTrue(config.allowlist().pattern().contains("kafka_network.*")); + + assertFalse(config.isAllowed("random_name")); + assertTrue(config.isAllowed("kafka_server_metric")); + assertTrue(config.isAllowed("kafka_network_metric")); + + assertThrows(ConfigException.class, + () -> new ClientMetricsReporterConfig(Map.of(ALLOWLIST_CONFIG, "hell[o,s]world"), null)); + assertThrows(ConfigException.class, + () -> new ClientMetricsReporterConfig(Map.of(ALLOWLIST_CONFIG, "hello\\,world"), null)); + } + + @Test + public void testIsListenerEnabled() { + Map props = Map.of( + LISTENER_ENABLE_CONFIG, "true", + LISTENER_CONFIG, "http://:0"); + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + Optional httpServerOptional = config.startHttpServer(); + + assertTrue(config.isListenerEnabled()); + assertTrue(httpServerOptional.isPresent()); + HttpServers.release(httpServerOptional.get()); + } + + @Test + public void testIsListenerDisabled() { + Map props = Map.of(LISTENER_ENABLE_CONFIG, false); + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + Optional httpServerOptional = config.startHttpServer(); + + assertTrue(httpServerOptional.isEmpty()); + assertFalse(config.isListenerEnabled()); + } + + @Test + public void testStartHttpServer() { + Map props = Map.of(LISTENER_CONFIG, "http://:0"); + ClientMetricsReporterConfig config = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + Optional httpServerOptional = config.startHttpServer(); + assertTrue(httpServerOptional.isPresent()); + + ClientMetricsReporterConfig config2 = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + Optional httpServerOptional2 = config2.startHttpServer(); + assertTrue(httpServerOptional2.isPresent()); + + props = Map.of(LISTENER_CONFIG, "http://:" + httpServerOptional.get().port()); + ClientMetricsReporterConfig config3 = new ClientMetricsReporterConfig(props, new PrometheusRegistry()); + Exception exc = assertThrows(RuntimeException.class, config3::startHttpServer); + assertInstanceOf(BindException.class, exc.getCause()); + + HttpServers.release(httpServerOptional.get()); + HttpServers.release(httpServerOptional2.get()); + } +} + diff --git a/src/test/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporterTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterTest.java similarity index 54% rename from src/test/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporterTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterTest.java index 724fe7b..0885aa9 100644 --- a/src/test/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporterTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterTest.java @@ -2,37 +2,39 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus; import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.strimzi.kafka.metrics.kafka.KafkaCollector; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; +import io.strimzi.kafka.metrics.prometheus.kafka.KafkaCollector; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.KafkaMetricsContext; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; -import static io.strimzi.kafka.metrics.TestUtils.getMetrics; -import static io.strimzi.kafka.metrics.TestUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter.PREFIXES; +import static io.strimzi.kafka.metrics.prometheus.KafkaMetricsUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.getMetrics; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -public class KafkaPrometheusMetricsReporterTest { +public class ClientMetricsReporterTest { - private final Map labels = Collections.singletonMap("key", "value"); - private Map configs; - private PrometheusRegistry registry; - private KafkaCollector kafkaCollector; + static final Map LABELS = Map.of("key", "value"); + Map configs; + PrometheusRegistry registry; + KafkaCollector kafkaCollector; @BeforeEach public void setup() { configs = new HashMap<>(); - configs.put(PrometheusMetricsReporterConfig.LISTENER_CONFIG, "http://:0"); + configs.put(ClientMetricsReporterConfig.LISTENER_CONFIG, "http://:0"); registry = new PrometheusRegistry(); PrometheusCollector prometheusCollector = new PrometheusCollector(); kafkaCollector = new KafkaCollector(prometheusCollector); @@ -41,28 +43,28 @@ public void setup() { @Test public void testLifeCycle() throws Exception { - KafkaPrometheusMetricsReporter reporter = new KafkaPrometheusMetricsReporter(registry, kafkaCollector); - configs.put(PrometheusMetricsReporterConfig.ALLOWLIST_CONFIG, "kafka_server_group_name.*"); + ClientMetricsReporter reporter = new ClientMetricsReporter(registry, kafkaCollector); + configs.put(ClientMetricsReporterConfig.ALLOWLIST_CONFIG, "kafka_producer_group_name.*"); reporter.configure(configs); - reporter.contextChange(new KafkaMetricsContext("kafka.server")); + reporter.contextChange(new KafkaMetricsContext("kafka.producer")); int port = reporter.getPort().orElseThrow(); assertEquals(0, getMetrics(port).size()); // Adding a metric not matching the allowlist does nothing - KafkaMetric metric1 = newKafkaMetric("other", "group", (config, now) -> 0, labels); - reporter.init(Collections.singletonList(metric1)); + KafkaMetric metric1 = newKafkaMetric("other", "group", (config, now) -> 0, LABELS); + reporter.init(List.of(metric1)); List metrics = getMetrics(port); assertEquals(0, metrics.size()); // Adding a metric that matches the allowlist - KafkaMetric metric2 = newKafkaMetric("name", "group", (config, now) -> 0, labels); + KafkaMetric metric2 = newKafkaMetric("name", "group", (config, now) -> 0, LABELS); reporter.metricChange(metric2); metrics = getMetrics(port); assertEquals(1, metrics.size()); // Adding a non-numeric metric - KafkaMetric metric3 = newKafkaMetric("name1", "group", (config, now) -> "hello", labels); + KafkaMetric metric3 = newKafkaMetric("name1", "group", (config, now) -> "hello", LABELS); reporter.metricChange(metric3); metrics = getMetrics(port); assertEquals(2, metrics.size()); @@ -77,26 +79,26 @@ public void testLifeCycle() throws Exception { @Test public void testMultipleReporters() throws Exception { - KafkaPrometheusMetricsReporter reporter1 = new KafkaPrometheusMetricsReporter(registry, kafkaCollector); + ClientMetricsReporter reporter1 = new ClientMetricsReporter(registry, kafkaCollector); reporter1.configure(configs); - reporter1.contextChange(new KafkaMetricsContext("kafka.server")); + reporter1.contextChange(new KafkaMetricsContext("kafka.producer")); Optional port1 = reporter1.getPort(); assertTrue(port1.isPresent()); assertEquals(0, getMetrics(port1.get()).size()); - KafkaPrometheusMetricsReporter reporter2 = new KafkaPrometheusMetricsReporter(registry, kafkaCollector); + ClientMetricsReporter reporter2 = new ClientMetricsReporter(registry, kafkaCollector); reporter2.configure(configs); - reporter2.contextChange(new KafkaMetricsContext("kafka.server")); + reporter2.contextChange(new KafkaMetricsContext("kafka.producer")); Optional port2 = reporter1.getPort(); assertTrue(port2.isPresent()); assertEquals(0, getMetrics(port2.get()).size()); assertEquals(port1, port2); - KafkaMetric metric1 = newKafkaMetric("name", "group", (config, now) -> 0, Collections.singletonMap("name", "metric1")); - reporter1.init(Collections.singletonList(metric1)); - KafkaMetric metric2 = newKafkaMetric("name", "group", (config, now) -> 0, Collections.singletonMap("name", "metric2")); - reporter2.init(Collections.singletonList(metric2)); + KafkaMetric metric1 = newKafkaMetric("name", "group", (config, now) -> 0, Map.of("name", "metric1")); + reporter1.init(List.of(metric1)); + KafkaMetric metric2 = newKafkaMetric("name", "group", (config, now) -> 0, Map.of("name", "metric2")); + reporter2.init(List.of(metric2)); assertEquals(2, getMetrics(port1.get()).size()); reporter1.metricRemoval(metric1); @@ -107,4 +109,21 @@ public void testMultipleReporters() throws Exception { reporter2.close(); } + @Test + public void testReconfigurableConfigs() { + try (ClientMetricsReporter reporter = new ClientMetricsReporter(registry, kafkaCollector)) { + assertTrue(reporter.reconfigurableConfigs().isEmpty()); + } + } + + @Test + public void testContextChange() { + try (ClientMetricsReporter reporter = new ClientMetricsReporter(registry, kafkaCollector)) { + for (String prefix : PREFIXES) { + reporter.contextChange(new KafkaMetricsContext(prefix)); + } + assertThrows(IllegalStateException.class, () -> reporter.contextChange(new KafkaMetricsContext("other"))); + } + } + } diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/KafkaMetricsUtils.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/KafkaMetricsUtils.java new file mode 100644 index 0000000..f6586a8 --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/KafkaMetricsUtils.java @@ -0,0 +1,35 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.metrics.Gauge; +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.metrics.MetricConfig; +import org.apache.kafka.common.utils.Time; + +import java.util.Map; + + +public class KafkaMetricsUtils { + + /** + * Create a new Kafka metric + * @param name The name of the metric + * @param group The group of the metric + * @param gauge The gauge providing the value of the metric + * @param labels The labels of the metric + * @return The Kafka metric + */ + public static KafkaMetric newKafkaMetric(String name, String group, Gauge gauge, Map labels) { + return new KafkaMetric( + new Object(), + new MetricName(name, group, "", labels), + gauge, + new MetricConfig(), + Time.SYSTEM); + } + +} diff --git a/src/test/java/io/strimzi/kafka/metrics/TestUtils.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java similarity index 77% rename from src/test/java/io/strimzi/kafka/metrics/TestUtils.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java index 0a05d99..3b6855e 100644 --- a/src/test/java/io/strimzi/kafka/metrics/TestUtils.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus; import io.prometheus.metrics.model.snapshots.CounterSnapshot; import io.prometheus.metrics.model.snapshots.GaugeSnapshot; @@ -11,12 +11,6 @@ import io.prometheus.metrics.model.snapshots.MetricSnapshot; import io.prometheus.metrics.model.snapshots.Quantiles; import io.prometheus.metrics.model.snapshots.SummarySnapshot; -import io.strimzi.kafka.metrics.http.Listener; -import org.apache.kafka.common.MetricName; -import org.apache.kafka.common.metrics.Gauge; -import org.apache.kafka.common.metrics.KafkaMetric; -import org.apache.kafka.common.metrics.MetricConfig; -import org.apache.kafka.common.utils.Time; import org.junit.jupiter.api.function.ThrowingConsumer; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.Network; @@ -32,7 +26,6 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -41,18 +34,14 @@ /** * Utility class to create and retrieve metrics */ -@SuppressWarnings("ClassFanOutComplexity") -public class TestUtils { +public class MetricsUtils { - private static final String VERSION = "1.0.0-SNAPSHOT"; - private static final String KAFKA_VERSION = "3.9.0"; - private static final String CLIENTS_IMAGE = "quay.io/strimzi-test-clients/test-clients:latest-kafka-" + KAFKA_VERSION; + public static final String VERSION = "1.0.0-SNAPSHOT"; + private static final String CLIENTS_IMAGE = "quay.io/strimzi-test-clients/test-clients:latest-kafka-3.9.0"; private static final Duration TIMEOUT = Duration.ofSeconds(10L); - public static final String REPORTER_JARS = "target/metrics-reporter-" + VERSION + "/metrics-reporter-" + VERSION + "/libs/"; + public static final String REPORTER_JARS = "target/client-metrics-reporter-" + VERSION + "/client-metrics-reporter-" + VERSION + "/libs/"; public static final String MOUNT_PATH = "/opt/strimzi/metrics-reporter/"; - public static final int PORT = Listener.parseListener(PrometheusMetricsReporterConfig.LISTENER_CONFIG_DEFAULT).port; - public static final String KAFKA_NETWORK_ALIAS = "kafka"; /** * Query the HTTP endpoint and returns the output @@ -88,37 +77,6 @@ public static List getMetrics(String host, int port) throws Exception { return metrics; } - /** - * Create a new Kafka metric - * @param name The name of the metric - * @param group The group of the metric - * @param gauge The gauge providing the value of the metric - * @param labels The labels of the metric - * @return The Kafka metric - */ - public static KafkaMetric newKafkaMetric(String name, String group, Gauge gauge, Map labels) { - return new KafkaMetric( - new Object(), - new MetricName(name, group, "", labels), - gauge, - new MetricConfig(), - Time.SYSTEM); - } - - /** - * Create a new Yammer metric - * @param valueSupplier The supplier providing the value of the metric - * @return The Yammer metric - */ - public static com.yammer.metrics.core.Gauge newYammerMetric(Supplier valueSupplier) { - return new com.yammer.metrics.core.Gauge<>() { - @Override - public T value() { - return valueSupplier.get(); - } - }; - } - /** * Check a Gauge snapshot * @param snapshot the gauge snapshot @@ -135,6 +93,7 @@ public static void assertGaugeSnapshot(MetricSnapshot snapshot, double expectedV } /** +<<<<<<< HEAD:src/test/java/io/strimzi/kafka/metrics/TestUtils.java * Check a Counter snapshot * @param snapshot the counter snapshot * @param expectedValue the expected value @@ -150,6 +109,8 @@ public static void assertCounterSnapshot(MetricSnapshot snapshot, double expecte } /** +======= +>>>>>>> a479b7a (Split the reporter into client and server modules):client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java * Check an Info snapshot * @param snapshot the info snapshot * @param labels the existing labels @@ -165,6 +126,7 @@ public static void assertInfoSnapshot(MetricSnapshot snapshot, Labels labels, St } /** +<<<<<<< HEAD:src/test/java/io/strimzi/kafka/metrics/TestUtils.java * Check a Summary snapshot * @param snapshot the summary snapshot * @param expectedCount the expected count @@ -184,6 +146,8 @@ public static void assertSummarySnapshot(MetricSnapshot snapshot, int expectedCo } /** +======= +>>>>>>> a479b7a (Split the reporter into client and server modules):client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/MetricsUtils.java * Filter metrics that start with a specified prefix * @param allMetrics all the metric names * @param prefix the prefix @@ -199,11 +163,11 @@ public static List filterMetrics(List allMetrics, String prefix) return metrics; } - public static void verify(GenericContainer container, String prefix, ThrowingConsumer> condition) { + public static void verify(GenericContainer container, String prefix, int port, ThrowingConsumer> condition) { assertTimeoutPreemptively(TIMEOUT, () -> { while (true) { try { - List filteredMetrics = filterMetrics(getMetrics(container.getHost(), container.getMappedPort(PORT)), prefix); + List filteredMetrics = filterMetrics(getMetrics(container.getHost(), container.getMappedPort(port)), prefix); condition.accept(filteredMetrics); return; } catch (Throwable t) { @@ -214,10 +178,10 @@ public static void verify(GenericContainer container, String prefix, Throwing }); } - public static GenericContainer clientContainer(Map env) { + public static GenericContainer clientContainer(Map env, int port) { return new GenericContainer<>(CLIENTS_IMAGE) .withNetwork(Network.SHARED) - .withExposedPorts(PORT) + .withExposedPorts(port) .withCopyFileToContainer(MountableFile.forHostPath(REPORTER_JARS), MOUNT_PATH) .withEnv(env) .waitingFor(Wait.forHttp("/metrics").forStatusCode(200)); diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporterTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporterTest.java new file mode 100644 index 0000000..f8258ad --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/AbstractReporterTest.java @@ -0,0 +1,94 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.common; + +import org.junit.jupiter.api.Test; + +import java.util.regex.Pattern; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class AbstractReporterTest { + + @Test + public void testAllowedMetricsNotReconfigurable() { + Pattern pattern = Pattern.compile("pattern_.*"); + TestReporter reporter = new TestReporter(pattern, false); + reporter.addMetric("pattern_metric", new MetricWrapper(null, null, null, null) { + @Override + public String prometheusName() { + return "pattern_metric"; + } + }); + assertEquals(1, reporter.allowedMetrics().size()); + + reporter.addMetric("pattern2_metric", new MetricWrapper(null, null, null, null) { + @Override + public String prometheusName() { + return "pattern2_metric"; + } + }); + assertEquals(1, reporter.allowedMetrics().size()); + + reporter.allowlist = Pattern.compile("(pattern_.*)|(pattern2_.*)"); + reporter.updateAllowedMetrics(); + assertEquals(1, reporter.allowedMetrics().size()); + + reporter.removeMetric("pattern_metric"); + assertTrue(reporter.allowedMetrics().isEmpty()); + } + + @Test + public void testAllowedMetricsReconfigurable() { + Pattern pattern = Pattern.compile("pattern_.*"); + TestReporter reporter = new TestReporter(pattern, true); + reporter.addMetric("pattern_metric", new MetricWrapper(null, null, null, null) { + @Override + public String prometheusName() { + return "pattern_metric"; + } + }); + assertEquals(1, reporter.allowedMetrics().size()); + + reporter.addMetric("pattern2_metric", new MetricWrapper(null, null, null, null) { + @Override + public String prometheusName() { + return "pattern2_metric"; + } + }); + assertEquals(1, reporter.allowedMetrics().size()); + + reporter.allowlist = Pattern.compile("(pattern_.*)|(pattern2_.*)"); + reporter.updateAllowedMetrics(); + assertEquals(2, reporter.allowedMetrics().size()); + + reporter.removeMetric("pattern_metric"); + assertEquals(1, reporter.allowedMetrics().size()); + reporter.removeMetric("pattern2_metric"); + assertTrue(reporter.allowedMetrics().isEmpty()); + } + + static final class TestReporter extends AbstractReporter { + + private final boolean isReconfigurable; + private Pattern allowlist; + + TestReporter(Pattern allowlist, boolean isReconfigurable) { + this.allowlist = allowlist; + this.isReconfigurable = isReconfigurable; + } + + @Override + protected Pattern allowlist() { + return allowlist; + } + + @Override + protected boolean isReconfigurable() { + return isReconfigurable; + } + } +} diff --git a/src/test/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilderTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilderTest.java similarity index 93% rename from src/test/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilderTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilderTest.java index 85164cc..c494e1b 100644 --- a/src/test/java/io/strimzi/kafka/metrics/DataPointSnapshotBuilderTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/DataPointSnapshotBuilderTest.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.model.snapshots.InfoSnapshot; import io.prometheus.metrics.model.snapshots.Labels; diff --git a/src/test/java/io/strimzi/kafka/metrics/PrometheusCollectorTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollectorTest.java similarity index 67% rename from src/test/java/io/strimzi/kafka/metrics/PrometheusCollectorTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollectorTest.java index ed47928..bab6e54 100644 --- a/src/test/java/io/strimzi/kafka/metrics/PrometheusCollectorTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/common/PrometheusCollectorTest.java @@ -2,27 +2,21 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus.common; import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.prometheus.metrics.model.snapshots.CounterSnapshot; import io.prometheus.metrics.model.snapshots.GaugeSnapshot; import io.prometheus.metrics.model.snapshots.InfoSnapshot; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.MetricSnapshot; import io.prometheus.metrics.model.snapshots.MetricSnapshots; -import io.prometheus.metrics.model.snapshots.Quantile; -import io.prometheus.metrics.model.snapshots.Quantiles; -import io.prometheus.metrics.model.snapshots.SummarySnapshot; import org.junit.jupiter.api.Test; import java.util.ArrayList; import java.util.List; -import static io.strimzi.kafka.metrics.TestUtils.assertCounterSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.assertGaugeSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.assertInfoSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.assertSummarySnapshot; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertGaugeSnapshot; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertInfoSnapshot; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -50,14 +44,9 @@ public void testCollect() { .name("gauge") .dataPoint(DataPointSnapshotBuilder.gaugeDataPoint(labels, value)) .build()); - snapshots.add(CounterSnapshot.builder() - .name("counter") - .dataPoint(DataPointSnapshotBuilder.counterDataPoint(labels, value)) - .build()); return snapshots; }); - int count = 1; - Quantiles quantiles = Quantiles.of(new Quantile(0.9, value)); + String metricName = "name"; prometheusCollector.addCollector(() -> { List snapshots = new ArrayList<>(); @@ -65,18 +54,12 @@ public void testCollect() { .name("info") .dataPoint(DataPointSnapshotBuilder.infoDataPoint(labels, value, metricName)) .build()); - snapshots.add(SummarySnapshot.builder() - .name("summary") - .dataPoint(DataPointSnapshotBuilder.summaryDataPoint(labels, count, value, quantiles)) - .build()); return snapshots; }); MetricSnapshots snapshots = prometheusCollector.collect(); - assertEquals(4, snapshots.size()); + assertEquals(2, snapshots.size()); assertGaugeSnapshot(findSnapshot(snapshots, GaugeSnapshot.class), value, labels); - assertCounterSnapshot(findSnapshot(snapshots, CounterSnapshot.class), value, labels); assertInfoSnapshot(findSnapshot(snapshots, InfoSnapshot.class), labels, metricName, String.valueOf(value)); - assertSummarySnapshot(findSnapshot(snapshots, SummarySnapshot.class), count, value, labels, quantiles); } private MetricSnapshot findSnapshot(MetricSnapshots snapshots, Class clazz) { diff --git a/src/test/java/io/strimzi/kafka/metrics/http/HttpServersTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/HttpServersTest.java similarity index 98% rename from src/test/java/io/strimzi/kafka/metrics/http/HttpServersTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/HttpServersTest.java index 85acb1c..46cf721 100644 --- a/src/test/java/io/strimzi/kafka/metrics/http/HttpServersTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/HttpServersTest.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.http; +package io.strimzi.kafka.metrics.prometheus.http; import io.prometheus.metrics.model.registry.PrometheusRegistry; import org.junit.jupiter.api.Test; diff --git a/src/test/java/io/strimzi/kafka/metrics/http/ListenerTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/ListenerTest.java similarity index 73% rename from src/test/java/io/strimzi/kafka/metrics/http/ListenerTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/ListenerTest.java index be62d10..ab7013f 100644 --- a/src/test/java/io/strimzi/kafka/metrics/http/ListenerTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/http/ListenerTest.java @@ -2,12 +2,11 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.http; +package io.strimzi.kafka.metrics.prometheus.http; import org.apache.kafka.common.config.ConfigException; import org.junit.jupiter.api.Test; -import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.LISTENER_CONFIG; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -35,20 +34,20 @@ public void testListenerParseListener() { @Test public void testValidator() { Listener.ListenerValidator validator = new Listener.ListenerValidator(); - validator.ensureValid(LISTENER_CONFIG, "http://:0"); - validator.ensureValid(LISTENER_CONFIG, "http://123:8080"); - validator.ensureValid(LISTENER_CONFIG, "http://::1:8080"); - validator.ensureValid(LISTENER_CONFIG, "http://[::1]:8080"); - validator.ensureValid(LISTENER_CONFIG, "http://random:8080"); + validator.ensureValid("name", "http://:0"); + validator.ensureValid("name", "http://123:8080"); + validator.ensureValid("name", "http://::1:8080"); + validator.ensureValid("name", "http://[::1]:8080"); + validator.ensureValid("name", "http://random:8080"); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://random")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://random:")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://:-8080")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://random:-8080")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "http://:8080random")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "randomhttp://:8080random")); - assertThrows(ConfigException.class, () -> validator.ensureValid(LISTENER_CONFIG, "randomhttp://:8080")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://random")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://random:")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://:-8080")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://random:-8080")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "http://:8080random")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "randomhttp://:8080random")); + assertThrows(ConfigException.class, () -> validator.ensureValid("name", "randomhttp://:8080")); } } diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestConsumerMetricsIT.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestConsumerMetricsIT.java new file mode 100644 index 0000000..21f9da6 --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestConsumerMetricsIT.java @@ -0,0 +1,98 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.integration; + +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter; +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig; +import io.strimzi.kafka.metrics.prometheus.MetricsUtils; +import io.strimzi.kafka.metrics.prometheus.http.Listener; +import io.strimzi.test.container.StrimziKafkaContainer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestConsumerMetricsIT { + + private static final int PORT = Listener.parseListener(ClientMetricsReporterConfig.LISTENER_CONFIG_DEFAULT).port; + private StrimziKafkaContainer broker; + private Map env; + + @BeforeEach + public void setUp() { + broker = new StrimziKafkaContainer() + .withKraft() + .withNetworkAliases("kafka"); + broker.start(); + + env = new HashMap<>(); + env.put("CLIENT_TYPE", "KafkaConsumer"); + env.put("BOOTSTRAP_SERVERS", "kafka:9091"); + env.put("TOPIC", "my-topic"); + env.put("GROUP_ID", "my-group"); + env.put("ADDITIONAL_CONFIG", "metric.reporters=" + ClientMetricsReporter.class.getName()); + env.put("CLASSPATH", MetricsUtils.MOUNT_PATH + "*"); + env.put("MESSAGE_COUNT", "1000"); + } + + @AfterEach + public void tearDown() { + broker.stop(); + } + + @Test + public void testConsumerMetrics() { + try (GenericContainer consumer = MetricsUtils.clientContainer(env, PORT)) { + consumer.start(); + + List prefixes = List.of( + "jvm_", + "process_", + "kafka_consumer_app_info_", + "kafka_consumer_kafka_metrics_", + "kafka_consumer_consumer_metrics_", + "kafka_consumer_consumer_node_metrics_", + "kafka_consumer_consumer_coordinator_metrics_", + "kafka_consumer_consumer_fetch_manager_metrics_"); + for (String prefix : prefixes) { + MetricsUtils.verify(consumer, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + } + } + + @Test + public void testConsumerMetricsWithAllowlist() { + env.put("ADDITIONAL_CONFIG", + "metric.reporters=" + ClientMetricsReporter.class.getName() + "\n" + + "prometheus.metrics.reporter.allowlist=kafka_consumer_kafka_metrics_.*,kafka_consumer_consumer_coordinator_metrics_.*"); + try (GenericContainer consumer = MetricsUtils.clientContainer(env, PORT)) { + consumer.start(); + + List allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_consumer_kafka_metrics_", + "kafka_consumer_consumer_coordinator_metrics_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(consumer, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + List disallowedPrefixes = List.of( + "kafka_consumer_app_info_", + "kafka_consumer_consumer_metrics_", + "kafka_consumer_consumer_node_metrics_", + "kafka_consumer_consumer_fetch_manager_metrics_"); + for (String prefix : disallowedPrefixes) { + MetricsUtils.verify(consumer, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + } + } +} diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestProducerMetricsIT.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestProducerMetricsIT.java new file mode 100644 index 0000000..e7fe614 --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestProducerMetricsIT.java @@ -0,0 +1,96 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.integration; + +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter; +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig; +import io.strimzi.kafka.metrics.prometheus.MetricsUtils; +import io.strimzi.kafka.metrics.prometheus.http.Listener; +import io.strimzi.test.container.StrimziKafkaContainer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestProducerMetricsIT { + + private static final int PORT = Listener.parseListener(ClientMetricsReporterConfig.LISTENER_CONFIG_DEFAULT).port; + private StrimziKafkaContainer broker; + private Map env; + + @BeforeEach + public void setUp() { + broker = new StrimziKafkaContainer() + .withKraft() + .withNetworkAliases("kafka"); + broker.start(); + + env = new HashMap<>(); + env.put("CLIENT_TYPE", "KafkaProducer"); + env.put("BOOTSTRAP_SERVERS", "kafka:9091"); + env.put("TOPIC", "my-topic"); + env.put("ADDITIONAL_CONFIG", "metric.reporters=" + ClientMetricsReporter.class.getName()); + env.put("CLASSPATH", MetricsUtils.MOUNT_PATH + "*"); + env.put("MESSAGE_COUNT", "1000"); + env.put("DELAY_MS", "100"); + } + + @AfterEach + public void tearDown() { + broker.stop(); + } + + @Test + public void testProducerMetrics() { + try (GenericContainer producer = MetricsUtils.clientContainer(env, PORT)) { + producer.start(); + + List prefixes = List.of( + "jvm_", + "process_", + "kafka_producer_app_info_", + "kafka_producer_kafka_metrics_", + "kafka_producer_producer_metrics_", + "kafka_producer_producer_node_metrics_", + "kafka_producer_producer_topic_metrics_"); + for (String prefix : prefixes) { + MetricsUtils.verify(producer, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + } + } + + @Test + public void testProducerMetricsWithAllowlist() { + env.put("ADDITIONAL_CONFIG", + "metric.reporters=" + ClientMetricsReporter.class.getName() + "\n" + + "prometheus.metrics.reporter.allowlist=kafka_producer_kafka_metrics_.*,kafka_producer_producer_topic_metrics_.*"); + try (GenericContainer producer = MetricsUtils.clientContainer(env, PORT)) { + producer.start(); + + List allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_producer_kafka_metrics_", + "kafka_producer_producer_topic_metrics_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(producer, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + List disallowedPrefixes = List.of( + "kafka_producer_app_info_", + "kafka_producer_producer_metrics_", + "kafka_producer_producer_node_metrics_"); + for (String prefix : disallowedPrefixes) { + MetricsUtils.verify(producer, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + } + } +} diff --git a/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestStreamsMetricsIT.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestStreamsMetricsIT.java new file mode 100644 index 0000000..6c513da --- /dev/null +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestStreamsMetricsIT.java @@ -0,0 +1,153 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.integration; + +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter; +import io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig; +import io.strimzi.kafka.metrics.prometheus.MetricsUtils; +import io.strimzi.kafka.metrics.prometheus.http.Listener; +import io.strimzi.test.container.StrimziKafkaContainer; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestStreamsMetricsIT { + + private static final int PORT = Listener.parseListener(ClientMetricsReporterConfig.LISTENER_CONFIG_DEFAULT).port; + private StrimziKafkaContainer broker; + private Map env; + + @BeforeEach + public void setUp() throws Exception { + broker = new StrimziKafkaContainer() + .withKraft() + .withNetworkAliases("kafka"); + broker.start(); + + try (Admin admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBootstrapServers()))) { + admin.createTopics(List.of( + new NewTopic("source-topic", 1, (short) -1), + new NewTopic("target-topic", 1, (short) -1)) + ).all().get(); + } + + Map producerConfigs = Map.of( + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBootstrapServers(), + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(), + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + try (KafkaProducer producer = new KafkaProducer<>(producerConfigs)) { + for (int i = 0; i < 10; i++) { + producer.send(new ProducerRecord<>("source-topic", "record" + i)); + } + producer.flush(); + } + + env = new HashMap<>(); + env.put("CLIENT_TYPE", "KafkaStreams"); + env.put("BOOTSTRAP_SERVERS", "kafka:9091"); + env.put("APPLICATION_ID", "my-app-id"); + env.put("SOURCE_TOPIC", "source-topic"); + env.put("TARGET_TOPIC", "target-topic"); + env.put("ADDITIONAL_CONFIG", "metric.reporters=" + ClientMetricsReporter.class.getName()); + env.put("CLASSPATH", MetricsUtils.MOUNT_PATH + "*"); + } + + @AfterEach + public void tearDown() { + broker.stop(); + } + + @Test + public void testStreamsMetrics() { + try (GenericContainer streams = MetricsUtils.clientContainer(env, PORT)) { + streams.start(); + + List prefixes = List.of( + "jvm_", + "process_", + "kafka_admin_client_app_info_", + "kafka_admin_client_kafka_metrics_", + "kafka_admin_client_admin_client_metrics_", + "kafka_admin_client_admin_client_node_metrics_", + "kafka_consumer_app_info_", + "kafka_consumer_kafka_metrics_", + "kafka_consumer_consumer_metrics_", + "kafka_consumer_consumer_node_metrics_", + "kafka_consumer_consumer_coordinator_metrics_", + "kafka_consumer_consumer_fetch_manager_metrics_", + "kafka_producer_app_info_", + "kafka_producer_kafka_metrics_", + "kafka_producer_producer_metrics_", + "kafka_producer_producer_node_metrics_", + "kafka_producer_producer_topic_metrics_", + "kafka_streams_stream_metrics_", + "kafka_streams_stream_processor_node_metrics_", + "kafka_streams_stream_state_updater_metrics_", + "kafka_streams_stream_task_metrics_", + "kafka_streams_stream_thread_metrics_", + "kafka_streams_stream_topic_metrics_"); + for (String prefix : prefixes) { + MetricsUtils.verify(streams, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + } + } + + @Test + public void testStreamsMetricsWithAllowlist() { + env.put("ADDITIONAL_CONFIG", + "metric.reporters=" + ClientMetricsReporter.class.getName() + "\n" + + "prometheus.metrics.reporter.allowlist=kafka_consumer_.*,kafka_streams_stream_metrics_.*"); + try (GenericContainer streams = MetricsUtils.clientContainer(env, PORT)) { + streams.start(); + + List allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_consumer_app_info_", + "kafka_consumer_kafka_metrics_", + "kafka_consumer_consumer_metrics_", + "kafka_consumer_consumer_node_metrics_", + "kafka_consumer_consumer_coordinator_metrics_", + "kafka_consumer_consumer_fetch_manager_metrics_", + "kafka_streams_stream_metrics_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(streams, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + List disallowedPrefixes = List.of( + "kafka_admin_client_app_info_", + "kafka_admin_client_kafka_metrics_", + "kafka_admin_client_admin_client_metrics_", + "kafka_admin_client_admin_client_node_metrics_", + "kafka_producer_app_info_", + "kafka_producer_kafka_metrics_", + "kafka_producer_producer_metrics_", + "kafka_producer_producer_node_metrics_", + "kafka_producer_producer_topic_metrics_", + "kafka_streams_stream_processor_node_metrics_", + "kafka_streams_stream_state_updater_metrics_", + "kafka_streams_stream_task_metrics_", + "kafka_streams_stream_thread_metrics_", + "kafka_streams_stream_topic_metrics_"); + for (String prefix : disallowedPrefixes) { + MetricsUtils.verify(streams, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + } + } +} diff --git a/src/test/java/io/strimzi/kafka/metrics/kafka/KafkaCollectorTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollectorTest.java similarity index 71% rename from src/test/java/io/strimzi/kafka/metrics/kafka/KafkaCollectorTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollectorTest.java index 7b1294d..9604483 100644 --- a/src/test/java/io/strimzi/kafka/metrics/kafka/KafkaCollectorTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaCollectorTest.java @@ -2,11 +2,12 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.kafka; +package io.strimzi.kafka.metrics.prometheus.kafka; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.MetricSnapshot; -import io.strimzi.kafka.metrics.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.metrics.Gauge; import org.apache.kafka.common.metrics.KafkaMetric; @@ -17,10 +18,11 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Pattern; -import static io.strimzi.kafka.metrics.TestUtils.assertGaugeSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.assertInfoSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.KafkaMetricsUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertGaugeSnapshot; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertInfoSnapshot; import static org.junit.jupiter.api.Assertions.assertEquals; public class KafkaCollectorTest { @@ -44,6 +46,13 @@ public void setup() { @Test public void testCollectKafkaMetrics() { KafkaCollector collector = new KafkaCollector(); + AbstractReporter reporter = new AbstractReporter() { + @Override + protected Pattern allowlist() { + return Pattern.compile(".*"); + } + }; + collector.addReporter(reporter); List metrics = collector.collect(); assertEquals(0, metrics.size()); @@ -52,7 +61,7 @@ public void testCollectKafkaMetrics() { AtomicInteger value = new AtomicInteger(1); MetricName metricName = new MetricName("name", "group", "description", tagsMap); MetricWrapper metricWrapper = newKafkaMetricWrapper(metricName, (config, now) -> value.get()); - collector.addMetric(metricName, metricWrapper); + reporter.addMetric(metricName, metricWrapper); metrics = collector.collect(); assertEquals(1, metrics.size()); @@ -65,7 +74,11 @@ public void testCollectKafkaMetrics() { assertGaugeSnapshot(metrics.get(0), 3, labels); // Removing a metric - collector.removeMetric(metricName); + reporter.removeMetric(metricName); + metrics = collector.collect(); + assertEquals(0, metrics.size()); + + collector.removeReporter(reporter); metrics = collector.collect(); assertEquals(0, metrics.size()); } @@ -73,6 +86,13 @@ public void testCollectKafkaMetrics() { @Test public void testCollectNonNumericKafkaMetric() { KafkaCollector collector = new KafkaCollector(); + AbstractReporter reporter = new AbstractReporter() { + @Override + protected Pattern allowlist() { + return Pattern.compile(".*"); + } + }; + collector.addReporter(reporter); List metrics = collector.collect(); assertEquals(0, metrics.size()); @@ -81,13 +101,17 @@ public void testCollectNonNumericKafkaMetric() { String nonNumericValue = "myValue"; MetricName metricName = new MetricName("name", "group", "description", tagsMap); MetricWrapper metricWrapper = newKafkaMetricWrapper(metricName, (config, now) -> nonNumericValue); - collector.addMetric(metricName, metricWrapper); - metrics = collector.collect(); + reporter.addMetric(metricName, metricWrapper); + metrics = collector.collect(); assertEquals(1, metrics.size()); MetricSnapshot snapshot = metrics.get(0); assertEquals(metricWrapper.prometheusName(), snapshot.getMetadata().getName()); assertInfoSnapshot(snapshot, labels, "name", nonNumericValue); + + collector.removeReporter(reporter); + metrics = collector.collect(); + assertEquals(0, metrics.size()); } private MetricWrapper newKafkaMetricWrapper(MetricName metricName, Gauge gauge) { diff --git a/src/test/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapperTest.java b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapperTest.java similarity index 87% rename from src/test/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapperTest.java rename to client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapperTest.java index 3e36882..39d83be 100644 --- a/src/test/java/io/strimzi/kafka/metrics/kafka/KafkaMetricWrapperTest.java +++ b/client-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/kafka/KafkaMetricWrapperTest.java @@ -2,7 +2,7 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.kafka; +package io.strimzi.kafka.metrics.prometheus.kafka; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.PrometheusNaming; @@ -10,26 +10,25 @@ import org.apache.kafka.common.metrics.KafkaMetric; import org.junit.jupiter.api.Test; -import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; -import static io.strimzi.kafka.metrics.TestUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.KafkaMetricsUtils.newKafkaMetric; import static org.junit.jupiter.api.Assertions.assertEquals; public class KafkaMetricWrapperTest { @Test public void testKafkaMetricName() { - String metricName = KafkaMetricWrapper.prometheusName("kafka_server", new MetricName("NaMe", "KafKa.neTwork", "", Collections.emptyMap())); + String metricName = KafkaMetricWrapper.prometheusName("kafka_server", new MetricName("NaMe", "KafKa.neTwork", "", Map.of())); assertEquals("kafka_server_kafka_network_name", metricName); } @Test public void testKafkaMetric() { AtomicInteger value = new AtomicInteger(0); - KafkaMetric metric = newKafkaMetric("name", "group", (config, now) -> value.get(), Collections.emptyMap()); + KafkaMetric metric = newKafkaMetric("name", "group", (config, now) -> value.get(), Map.of()); KafkaMetricWrapper wrapper = new KafkaMetricWrapper(KafkaMetricWrapper.prometheusName("kafka_server", metric.metricName()), metric, "name"); assertEquals(value.get(), ((KafkaMetric) wrapper.metric()).metricValue()); value.incrementAndGet(); diff --git a/src/main/assembly/dist.xml b/dist.xml similarity index 98% rename from src/main/assembly/dist.xml rename to dist.xml index cf41569..3cd22c1 100644 --- a/src/main/assembly/dist.xml +++ b/dist.xml @@ -27,4 +27,4 @@ ${artifact.groupId}.${artifact.artifactId}-${artifact.version}${dashClassifier?}.${artifact.extension} - + \ No newline at end of file diff --git a/pom.xml b/pom.xml index 59f9c27..b8f38e9 100644 --- a/pom.xml +++ b/pom.xml @@ -1,5 +1,4 @@ - 4.0.0 @@ -7,10 +6,15 @@ io.strimzi metrics-reporter 1.0.0-SNAPSHOT + pom metrics-reporter Prometheus Metrics Reporter for Apache Kafka server and client components. https://strimzi.io/ + + client-metrics-reporter + server-metrics-reporter + scm:git:git://github.com/strimzi/metrics-reporter.git @@ -117,7 +121,7 @@ 3.1.0 1.7.0 - 3.8.1 + 3.9.0 1.3.6 2.2.0 2.0.13 @@ -137,85 +141,87 @@ - - - org.apache.kafka - kafka-clients - ${kafka.version} - provided - - - org.apache.kafka - kafka-server-common - ${kafka.version} - provided - - - org.apache.kafka - kafka_2.13 - ${kafka.version} - provided - - - com.yammer.metrics - metrics-core - ${yammer.version} - provided - - - org.slf4j - slf4j-api - ${slf4j.version} - provided - - - com.github.spotbugs - spotbugs-annotations - ${spotbugs.version} - provided - + + + + org.apache.kafka + kafka-clients + ${kafka.version} + provided + + + org.apache.kafka + kafka-server-common + ${kafka.version} + provided + + + org.apache.kafka + kafka_2.13 + ${kafka.version} + provided + + + com.yammer.metrics + metrics-core + ${yammer.version} + provided + + + org.slf4j + slf4j-api + ${slf4j.version} + provided + + + com.github.spotbugs + spotbugs-annotations + ${spotbugs.version} + provided + - - io.prometheus - prometheus-metrics-model - ${prometheus.version} - - - io.prometheus - prometheus-metrics-instrumentation-jvm - ${prometheus.version} - - - io.prometheus - prometheus-metrics-exporter-httpserver - ${prometheus.version} - + + io.prometheus + prometheus-metrics-model + ${prometheus.version} + + + io.prometheus + prometheus-metrics-instrumentation-jvm + ${prometheus.version} + + + io.prometheus + prometheus-metrics-exporter-httpserver + ${prometheus.version} + - - org.junit.jupiter - junit-jupiter-api - ${junit.version} - test - - - org.slf4j - slf4j-simple - ${slf4j.version} - test - - - io.strimzi - strimzi-test-container - ${strimzi-test-container.version} - test - - - org.testcontainers - testcontainers - ${testcontainers.version} - test - - + + org.junit.jupiter + junit-jupiter-api + ${junit.version} + test + + + org.slf4j + slf4j-simple + ${slf4j.version} + test + + + io.strimzi + strimzi-test-container + ${strimzi-test-container.version} + test + + + org.testcontainers + testcontainers + ${testcontainers.version} + test + + + @@ -278,7 +284,7 @@ false posix - src/main/assembly/dist.xml + dist.xml @@ -297,7 +303,7 @@ ${project.build.directory}/spotbugs - ${project.basedir}/.spotbugs/spotbugs-exclude.xml + ${project.basedir}/../.spotbugs/spotbugs-exclude.xml diff --git a/server-metrics-reporter/pom.xml b/server-metrics-reporter/pom.xml new file mode 100644 index 0000000..7374be3 --- /dev/null +++ b/server-metrics-reporter/pom.xml @@ -0,0 +1,77 @@ + + + 4.0.0 + + io.strimzi + metrics-reporter + 1.0.0-SNAPSHOT + + + server-metrics-reporter + + + + + io.strimzi + client-metrics-reporter + ${project.parent.version} + + + io.prometheus + prometheus-metrics-model + + + + org.apache.kafka + kafka-clients + + + org.slf4j + slf4j-api + + + com.github.spotbugs + spotbugs-annotations + + + org.apache.kafka + kafka-server-common + + + org.apache.kafka + kafka_2.13 + + + com.yammer.metrics + metrics-core + + + + io.strimzi + client-metrics-reporter + ${project.parent.version} + tests + test-jar + test + + + org.junit.jupiter + junit-jupiter-api + + + org.slf4j + slf4j-simple + + + io.strimzi + strimzi-test-container + + + org.testcontainers + testcontainers + + + + \ No newline at end of file diff --git a/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporter.java b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporter.java new file mode 100644 index 0000000..01d510e --- /dev/null +++ b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporter.java @@ -0,0 +1,84 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import io.strimzi.kafka.metrics.prometheus.kafka.KafkaCollector; +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.common.metrics.MetricsContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +/** + * MetricsReporter implementation that expose Kafka server metrics in the Prometheus format. + */ +public class ServerKafkaMetricsReporter extends ClientMetricsReporter { + + private static final Logger LOG = LoggerFactory.getLogger(ServerKafkaMetricsReporter.class); + + @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the configure method + private ServerMetricsReporterConfig config; + + /** + * Constructor + */ + public ServerKafkaMetricsReporter() { + super(); + } + + // for testing + ServerKafkaMetricsReporter(PrometheusRegistry registry, KafkaCollector kafkaCollector) { + super(registry, kafkaCollector); + } + + @Override + public void configure(Map map) { + config = new ServerMetricsReporterConfig(map, registry); + httpServer = config.startHttpServer(); + LOG.debug("ServerKafkaMetricsReporter configured with {}", config); + } + + @Override + public void reconfigure(Map configs) { + config.reconfigure(configs); + ServerYammerMetricsReporter yammerReporter = ServerYammerMetricsReporter.getInstance(); + if (yammerReporter != null) { + yammerReporter.reconfigure(configs); + } + updateAllowedMetrics(); + } + + @Override + public void validateReconfiguration(Map configs) throws ConfigException { + new ServerMetricsReporterConfig(configs, null); + } + + @Override + public Set reconfigurableConfigs() { + return ServerMetricsReporterConfig.RECONFIGURABLES; + } + + @Override + public void contextChange(MetricsContext metricsContext) { + String prefix = metricsContext.contextLabels().get(MetricsContext.NAMESPACE); + this.prefix = PrometheusNaming.prometheusName(prefix); + } + + @Override + protected boolean isReconfigurable() { + return true; + } + + @Override + protected Pattern allowlist() { + return config.allowlist(); + } +} diff --git a/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfig.java b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfig.java new file mode 100644 index 0000000..9790d27 --- /dev/null +++ b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfig.java @@ -0,0 +1,64 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import org.apache.kafka.common.config.AbstractConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +/** +* Configuration for the PrometheusMetricsReporter implementation. +*/ +public class ServerMetricsReporterConfig extends ClientMetricsReporterConfig { + + private static final Logger LOG = LoggerFactory.getLogger(ServerMetricsReporterConfig.class); + + /** + * The configurations that are reconfigurable + */ + public static final Set RECONFIGURABLES = Set.of(ALLOWLIST_CONFIG); + + private Pattern allowlist; + + /** + * Constructor. + * + * @param props the configuration properties. + * @param registry the metrics registry + */ + public ServerMetricsReporterConfig(Map props, PrometheusRegistry registry) { + super(props, registry); + this.allowlist = compileAllowlist(getList(ALLOWLIST_CONFIG)); + } + + /** + * Update the reconfigurable configurations + * @param props The new configuration + */ + public void reconfigure(Map props) { + AbstractConfig abstractConfig = new AbstractConfig(CONFIG_DEF, props, false); + allowlist = compileAllowlist(abstractConfig.getList(ALLOWLIST_CONFIG)); + LOG.info("Updated allowlist to {}", allowlist); + } + + @Override + public Pattern allowlist() { + return allowlist; + } + + @Override + public String toString() { + return "ServerMetricsReporterConfig{" + + ", listener=" + listener + + ", listenerEnabled=" + listenerEnabled + + ", allowlist=" + allowlist + + '}'; + } +} diff --git a/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporter.java b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporter.java new file mode 100644 index 0000000..c6c7fbc --- /dev/null +++ b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporter.java @@ -0,0 +1,111 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Metric; +import com.yammer.metrics.core.MetricName; +import com.yammer.metrics.core.MetricsRegistry; +import com.yammer.metrics.core.MetricsRegistryListener; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; +import io.strimzi.kafka.metrics.prometheus.yammer.YammerCollector; +import io.strimzi.kafka.metrics.prometheus.yammer.YammerMetricWrapper; +import kafka.metrics.KafkaMetricsReporter; +import kafka.utils.VerifiableProperties; +import org.apache.kafka.server.metrics.KafkaYammerMetrics; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * ServerYammerMetricsReporter to export Yammer broker metrics in the Prometheus format. + */ +public class ServerYammerMetricsReporter extends AbstractReporter implements KafkaMetricsReporter, MetricsRegistryListener { + + private static final Logger LOG = LoggerFactory.getLogger(ServerYammerMetricsReporter.class); + + private static ServerYammerMetricsReporter instance; + + private final PrometheusRegistry registry; + private final YammerCollector yammerCollector; + @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the init method + /* test */ ServerMetricsReporterConfig config; + + /** + * Retrieve the ServerYammerMetricsReporter instance + * @return The ServerYammerMetricsReporter singleton + */ + public static ServerYammerMetricsReporter getInstance() { + return instance; + } + + /** + * Constructor + */ + @SuppressFBWarnings("CT_CONSTRUCTOR_THROW") + public ServerYammerMetricsReporter() { + if (instance != null) { + throw new IllegalStateException("Cannot create multiple instances of ServerYammerMetricsReporter"); + } + instance = this; + registry = PrometheusRegistry.defaultRegistry; + yammerCollector = YammerCollector.getCollector(PrometheusCollector.register(registry)); + yammerCollector.addReporter(this); + } + + // for testing + ServerYammerMetricsReporter(PrometheusRegistry registry, YammerCollector yammerCollector) { + this.registry = registry; + this.yammerCollector = yammerCollector; + yammerCollector.addReporter(this); + } + + @Override + public void init(VerifiableProperties props) { + config = new ServerMetricsReporterConfig(props.props(), registry); + for (MetricsRegistry yammerRegistry : List.of(KafkaYammerMetrics.defaultRegistry(), Metrics.defaultRegistry())) { + yammerRegistry.addListener(this); + } + LOG.debug("ServerYammerMetricsReporter configured with {}", config); + } + + @Override + public void onMetricAdded(MetricName name, Metric metric) { + String prometheusName = YammerMetricWrapper.prometheusName(name); + MetricWrapper metricWrapper = new YammerMetricWrapper(prometheusName, name.getScope(), metric, name.getName()); + addMetric(name, metricWrapper); + } + + @Override + public void onMetricRemoved(MetricName name) { + removeMetric(name); + } + + /** + * Update the reconfigurable configurations + * @param props The new configuration + */ + public void reconfigure(Map props) { + config.reconfigure(props); + updateAllowedMetrics(); + } + + @Override + protected Pattern allowlist() { + return config.allowlist(); + } + + @Override + protected boolean isReconfigurable() { + return true; + } +} diff --git a/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollector.java b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollector.java new file mode 100644 index 0000000..18378c9 --- /dev/null +++ b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollector.java @@ -0,0 +1,142 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.yammer; + +import com.yammer.metrics.core.Counter; +import com.yammer.metrics.core.Gauge; +import com.yammer.metrics.core.Histogram; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.Sampling; +import com.yammer.metrics.core.Timer; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot; +import io.prometheus.metrics.model.snapshots.InfoSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import io.prometheus.metrics.model.snapshots.MetricSnapshot; +import io.prometheus.metrics.model.snapshots.Quantile; +import io.prometheus.metrics.model.snapshots.Quantiles; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.DataPointSnapshotBuilder; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.MetricsCollector; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Collector for Yammer metrics + */ +@SuppressWarnings("ClassFanOutComplexity") +public class YammerCollector implements MetricsCollector { + + private static final Logger LOG = LoggerFactory.getLogger(YammerCollector.class); + private static final AtomicBoolean REGISTERED = new AtomicBoolean(false); + private static final YammerCollector INSTANCE = new YammerCollector(); + private static final List QUANTILES = List.of(0.50, 0.75, 0.95, 0.98, 0.99, 0.999); + + private final Set reporters = ConcurrentHashMap.newKeySet(); + + private YammerCollector() {} + + /** + * Constructor only used for testing + * @param prometheusCollector the PrometheusCollector that will collect metrics + */ + public YammerCollector(PrometheusCollector prometheusCollector) { + prometheusCollector.addCollector(this); + } + + /** + * Retrieve the YammerCollector instance + * + * @param prometheusCollector the PrometheusCollector that will collect metrics + * @return the YammerCollector singleton + */ + public static YammerCollector getCollector(PrometheusCollector prometheusCollector) { + if (REGISTERED.compareAndSet(false, true)) { + prometheusCollector.addCollector(INSTANCE); + } + return INSTANCE; + } + + /** + * Add an {@link AbstractReporter} instance to this collector + * @param reporter The reporter instance to add + */ + public void addReporter(AbstractReporter reporter) { + reporters.add(reporter); + } + + /** + * Collect all the metrics added to this Collector + * + * @return the list of metrics of this collector + */ + @SuppressWarnings({"CyclomaticComplexity", "JavaNCSS"}) + @Override + public List collect() { + Map> builders = new HashMap<>(); + for (AbstractReporter reporter : reporters) { + for (MetricWrapper metricWrapper : reporter.allowedMetrics()) { + String prometheusMetricName = metricWrapper.prometheusName(); + Object metric = metricWrapper.metric(); + Labels labels = metricWrapper.labels(); + LOG.debug("Collecting Yammer metric {} with the following labels: {}", prometheusMetricName, labels); + + if (metric instanceof Counter) { + Counter counter = (Counter) metric; + CounterSnapshot.Builder builder = (CounterSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> CounterSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.counterDataPoint(labels, counter.count())); + } else if (metric instanceof Gauge) { + Object valueObj = ((Gauge) metric).value(); + if (valueObj instanceof Number) { + double value = ((Number) valueObj).doubleValue(); + GaugeSnapshot.Builder builder = (GaugeSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> GaugeSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.gaugeDataPoint(labels, value)); + } else { + InfoSnapshot.Builder builder = (InfoSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> InfoSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.infoDataPoint(labels, valueObj, metricWrapper.attribute())); + } + } else if (metric instanceof Timer) { + Timer timer = (Timer) metric; + SummarySnapshot.Builder builder = (SummarySnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> SummarySnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.summaryDataPoint(labels, timer.count(), timer.sum(), quantiles(timer))); + } else if (metric instanceof Histogram) { + Histogram histogram = (Histogram) metric; + SummarySnapshot.Builder builder = (SummarySnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> SummarySnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.summaryDataPoint(labels, histogram.count(), histogram.sum(), quantiles(histogram))); + } else if (metric instanceof Meter) { + Meter meter = (Meter) metric; + CounterSnapshot.Builder builder = (CounterSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> CounterSnapshot.builder().name(prometheusMetricName)); + builder.dataPoint(DataPointSnapshotBuilder.counterDataPoint(labels, meter.count())); + } else { + LOG.error("The metric {} has an unexpected type: {}", prometheusMetricName, metric.getClass().getName()); + } + } + } + List snapshots = new ArrayList<>(); + for (MetricSnapshot.Builder builder : builders.values()) { + snapshots.add(builder.build()); + } + return snapshots; + } + + private static Quantiles quantiles(Sampling sampling) { + Quantiles.Builder quantilesBuilder = Quantiles.builder(); + for (double quantile : QUANTILES) { + quantilesBuilder.quantile(new Quantile(quantile, sampling.getSnapshot().getValue(quantile))); + } + return quantilesBuilder.build(); + } +} diff --git a/src/main/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapper.java b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapper.java similarity index 95% rename from src/main/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapper.java rename to server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapper.java index 8a65729..965c67d 100644 --- a/src/main/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapper.java +++ b/server-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapper.java @@ -2,13 +2,13 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.yammer; +package io.strimzi.kafka.metrics.prometheus.yammer; import com.yammer.metrics.core.Metric; import com.yammer.metrics.core.MetricName; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.PrometheusNaming; -import io.strimzi.kafka.metrics.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporterTest.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporterTest.java new file mode 100644 index 0000000..e7bef15 --- /dev/null +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerKafkaMetricsReporterTest.java @@ -0,0 +1,76 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.metrics.KafkaMetricsContext; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static io.strimzi.kafka.metrics.prometheus.KafkaMetricsUtils.newKafkaMetric; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.getMetrics; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ServerKafkaMetricsReporterTest extends ClientMetricsReporterTest { + + @Test + public void testReconfigurableConfigs() { + try (ServerKafkaMetricsReporter reporter = new ServerKafkaMetricsReporter(registry, kafkaCollector)) { + assertFalse(reporter.reconfigurableConfigs().isEmpty()); + } + } + + @Test + public void testReconfigure() throws Exception { + try (ServerKafkaMetricsReporter reporter = new ServerKafkaMetricsReporter(registry, kafkaCollector)) { + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "kafka_server_group_name.*"); + reporter.configure(configs); + reporter.contextChange(new KafkaMetricsContext("kafka.server")); + + int port = reporter.getPort().orElseThrow(); + assertEquals(0, getMetrics(port).size()); + + // Adding a metric not matching the allowlist does nothing + KafkaMetric metric1 = newKafkaMetric("other", "group", (config, now) -> 0, LABELS); + reporter.init(List.of(metric1)); + List metrics = getMetrics(port); + assertEquals(0, metrics.size()); + + // Adding a metric matching the allowlist + KafkaMetric metric2 = newKafkaMetric("name", "group", (config, now) -> 0, LABELS); + reporter.metricChange(metric2); + metrics = getMetrics(port); + assertEquals(1, metrics.size()); + assertTrue(metrics.get(0).contains("name")); + + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "kafka_server_group_other.*"); + reporter.reconfigure(configs); + + metrics = getMetrics(port); + assertEquals(1, metrics.size()); + assertTrue(metrics.get(0).contains("other")); + } + } + + @Test + public void testValidateReconfiguration() { + try (ServerKafkaMetricsReporter reporter = new ServerKafkaMetricsReporter(registry, kafkaCollector)) { + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "kafka_server_group_name.*"); + reporter.configure(configs); + reporter.contextChange(new KafkaMetricsContext("kafka.server")); + + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, ".*"); + reporter.validateReconfiguration(configs); + + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "[\\]"); + assertThrows(ConfigException.class, () -> reporter.validateReconfiguration(configs)); + } + } +} diff --git a/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfigTest.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfigTest.java new file mode 100644 index 0000000..edf36ec --- /dev/null +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerMetricsReporterConfigTest.java @@ -0,0 +1,31 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import org.junit.jupiter.api.Test; + +import java.util.Map; + +import static io.strimzi.kafka.metrics.prometheus.ClientMetricsReporterConfig.ALLOWLIST_CONFIG; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +public class ServerMetricsReporterConfigTest { + + @Test + public void testReconfigure() { + Map props = Map.of(ALLOWLIST_CONFIG, "pattern1"); + ServerMetricsReporterConfig config = new ServerMetricsReporterConfig(props, new PrometheusRegistry()); + assertTrue(config.allowlist().pattern().contains("pattern1")); + + props = Map.of(ALLOWLIST_CONFIG, "pattern2"); + config.reconfigure(props); + assertFalse(config.allowlist().pattern().contains("pattern1")); + assertTrue(config.allowlist().pattern().contains("pattern2")); + } +} + diff --git a/src/test/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporterTest.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporterTest.java similarity index 56% rename from src/test/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporterTest.java rename to server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporterTest.java index afa0d46..062e6e5 100644 --- a/src/test/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporterTest.java +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/ServerYammerMetricsReporterTest.java @@ -2,14 +2,16 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics; +package io.strimzi.kafka.metrics.prometheus; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Counter; import com.yammer.metrics.core.Metric; import com.yammer.metrics.core.MetricName; import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.strimzi.kafka.metrics.http.HttpServers; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; +import io.strimzi.kafka.metrics.prometheus.http.HttpServers; +import io.strimzi.kafka.metrics.prometheus.yammer.YammerCollector; import kafka.utils.VerifiableProperties; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -18,22 +20,23 @@ import java.util.Map; import java.util.Properties; -import static io.strimzi.kafka.metrics.TestUtils.getMetrics; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.getMetrics; import static org.junit.jupiter.api.Assertions.assertEquals; -public class YammerPrometheusMetricsReporterTest { +public class ServerYammerMetricsReporterTest { private Properties configs; private PrometheusRegistry registry; - private PrometheusCollector collector; + private YammerCollector yammerCollector; @BeforeEach public void setup() { registry = new PrometheusRegistry(); - collector = new PrometheusCollector(); - registry.register(collector); + PrometheusCollector prometheusCollector = new PrometheusCollector(); + yammerCollector = new YammerCollector(prometheusCollector); + registry.register(prometheusCollector); configs = new Properties(); - configs.put(PrometheusMetricsReporterConfig.LISTENER_CONFIG, "http://:0"); + configs.put(ServerMetricsReporterConfig.LISTENER_CONFIG, "http://:0"); for (Map.Entry entry : Metrics.defaultRegistry().allMetrics().entrySet()) { Metrics.defaultRegistry().removeMetric(entry.getKey()); } @@ -41,8 +44,8 @@ public void setup() { @Test public void testLifeCycle() throws Exception { - YammerPrometheusMetricsReporter reporter = new YammerPrometheusMetricsReporter(registry, collector); - configs.put(PrometheusMetricsReporterConfig.ALLOWLIST_CONFIG, "group_type.*"); + ServerYammerMetricsReporter reporter = new ServerYammerMetricsReporter(registry, yammerCollector); + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "group_type.*"); reporter.init(new VerifiableProperties(configs)); HttpServers.ServerCounter httpServer = null; @@ -71,6 +74,31 @@ public void testLifeCycle() throws Exception { } } + @Test + public void testReconfigure() throws Exception { + ServerYammerMetricsReporter reporter = new ServerYammerMetricsReporter(registry, yammerCollector); + configs.put(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "group_type.*"); + reporter.init(new VerifiableProperties(configs)); + + HttpServers.ServerCounter httpServer = null; + try { + httpServer = reporter.config.startHttpServer().orElseThrow(); + int port = httpServer.port(); + assertEquals(0, getMetrics(port).size()); + + // Adding a metric not matching the allowlist does nothing + newCounter("other", "type", "name"); + List metrics = getMetrics(port); + assertEquals(0, metrics.size()); + + reporter.reconfigure(Map.of(ServerMetricsReporterConfig.ALLOWLIST_CONFIG, "other_type.*")); + metrics = getMetrics(port); + assertEquals(1, metrics.size()); + } finally { + if (httpServer != null) HttpServers.release(httpServer); + } + } + private Counter newCounter(String group, String type, String name) { MetricName metricName = new MetricName(group, type, name, ""); return Metrics.defaultRegistry().newCounter(metricName); diff --git a/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/YammerTestUtils.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/YammerTestUtils.java new file mode 100644 index 0000000..89047c0 --- /dev/null +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/YammerTestUtils.java @@ -0,0 +1,24 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus; + +import java.util.function.Supplier; + +public class YammerTestUtils { + + /** + * Create a new Yammer metric + * @param valueSupplier The supplier providing the value of the metric + * @return The Yammer metric + */ + public static com.yammer.metrics.core.Gauge newYammerMetric(Supplier valueSupplier) { + return new com.yammer.metrics.core.Gauge<>() { + @Override + public T value() { + return valueSupplier.get(); + } + }; + } +} diff --git a/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestServerMetricsIT.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestServerMetricsIT.java new file mode 100644 index 0000000..be27098 --- /dev/null +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/integration/TestServerMetricsIT.java @@ -0,0 +1,148 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.metrics.prometheus.integration; + +import io.strimzi.kafka.metrics.prometheus.MetricsUtils; +import io.strimzi.kafka.metrics.prometheus.ServerKafkaMetricsReporter; +import io.strimzi.kafka.metrics.prometheus.ServerMetricsReporterConfig; +import io.strimzi.kafka.metrics.prometheus.ServerYammerMetricsReporter; +import io.strimzi.kafka.metrics.prometheus.http.Listener; +import io.strimzi.test.container.StrimziKafkaContainer; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.admin.AlterConfigOp; +import org.apache.kafka.clients.admin.ConfigEntry; +import org.apache.kafka.common.config.ConfigResource; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.utility.MountableFile; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.VERSION; +import static io.strimzi.kafka.metrics.prometheus.ServerMetricsReporterConfig.ALLOWLIST_CONFIG; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class TestServerMetricsIT { + + private static final String REPORTER_JARS = "target/server-metrics-reporter-" + VERSION + "/server-metrics-reporter-" + VERSION + "/libs/"; + private static final int PORT = Listener.parseListener(ServerMetricsReporterConfig.LISTENER_CONFIG_DEFAULT).port; + private static final int NODE_ID = 0; + + private Map configs; + private StrimziKafkaContainer broker; + + @BeforeEach + public void setUp() { + configs = new HashMap<>(); + configs.put("metric.reporters", ServerKafkaMetricsReporter.class.getName()); + configs.put("kafka.metrics.reporters", ServerYammerMetricsReporter.class.getName()); + + broker = new StrimziKafkaContainer() + .withNodeId(NODE_ID) + .withKraft() + .withCopyFileToContainer(MountableFile.forHostPath(REPORTER_JARS), MetricsUtils.MOUNT_PATH) + .withExposedPorts(9092, PORT) + .withKafkaConfigurationMap(configs) + .withEnv(Map.of("CLASSPATH", MetricsUtils.MOUNT_PATH + "*")); + } + + @AfterEach + public void tearDown() { + broker.stop(); + } + + @Test + public void testBrokerMetrics() { + broker.start(); + + List prefixes = List.of( + "jvm_", + "process_", + "kafka_controller_", + "kafka_coordinator_", + "kafka_log_", + "kafka_network_", + "kafka_server_"); + for (String prefix : prefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + } + + @Test + public void testBrokerMetricsWithAllowlist() { + configs.put(ALLOWLIST_CONFIG, "kafka_controller.*,kafka_server.*"); + broker.withKafkaConfigurationMap(configs); + broker.start(); + + List allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_controller_", + "kafka_server_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + List disallowPrefixes = List.of( + "kafka_coordinator_", + "kafka_log_", + "kafka_network_"); + for (String prefix : disallowPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + } + + @Test + public void testReconfigureAllowlist() throws Exception { + configs.put(ALLOWLIST_CONFIG, "kafka_controller.*,kafka_server.*"); + broker.withKafkaConfigurationMap(configs); + broker.start(); + + List allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_controller_", + "kafka_server_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + List disallowPrefixes = List.of( + "kafka_coordinator_", + "kafka_log_", + "kafka_network_"); + for (String prefix : disallowPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + + try (Admin admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBootstrapServers()))) { + admin.incrementalAlterConfigs(Map.of( + new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(NODE_ID)), + List.of(new AlterConfigOp( + new ConfigEntry(ALLOWLIST_CONFIG, "kafka_coordinator.*,kafka_log.*,kafka_network.*"), + AlterConfigOp.OpType.SET)) + )).all().get(); + } + + allowedPrefixes = List.of( + "jvm_", + "process_", + "kafka_coordinator_", + "kafka_log_", + "kafka_network_"); + for (String prefix : allowedPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertFalse(metrics.isEmpty())); + } + disallowPrefixes = List.of( + "kafka_controller_", + "kafka_server_"); + for (String prefix : disallowPrefixes) { + MetricsUtils.verify(broker, prefix, PORT, metrics -> assertTrue(metrics.isEmpty())); + } + } +} diff --git a/src/test/java/io/strimzi/kafka/metrics/yammer/YammerCollectorTest.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollectorTest.java similarity index 69% rename from src/test/java/io/strimzi/kafka/metrics/yammer/YammerCollectorTest.java rename to server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollectorTest.java index a8f0e43..668ce54 100644 --- a/src/test/java/io/strimzi/kafka/metrics/yammer/YammerCollectorTest.java +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerCollectorTest.java @@ -2,29 +2,33 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.yammer; +package io.strimzi.kafka.metrics.prometheus.yammer; import com.yammer.metrics.core.Gauge; import com.yammer.metrics.core.MetricName; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.MetricSnapshot; -import io.strimzi.kafka.metrics.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter; +import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper; +import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; +import java.util.regex.Pattern; -import static io.strimzi.kafka.metrics.TestUtils.assertGaugeSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.assertInfoSnapshot; -import static io.strimzi.kafka.metrics.TestUtils.newYammerMetric; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertGaugeSnapshot; +import static io.strimzi.kafka.metrics.prometheus.MetricsUtils.assertInfoSnapshot; +import static io.strimzi.kafka.metrics.prometheus.YammerTestUtils.newYammerMetric; import static org.junit.jupiter.api.Assertions.assertEquals; public class YammerCollectorTest { private Labels labels; private String scope; + private YammerCollector collector; @BeforeEach public void setup() { @@ -35,11 +39,18 @@ public void setup() { scope += "k" + i + ".v" + i + "."; } labels = labelsBuilder.build(); + collector = new YammerCollector(new PrometheusCollector()); } @Test public void testCollectYammerMetrics() { - YammerCollector collector = new YammerCollector(); + AbstractReporter reporter = new AbstractReporter() { + @Override + protected Pattern allowlist() { + return Pattern.compile(".*"); + } + }; + collector.addReporter(reporter); List metrics = collector.collect(); assertEquals(0, metrics.size()); @@ -48,7 +59,7 @@ public void testCollectYammerMetrics() { AtomicInteger value = new AtomicInteger(1); MetricName metricName = new MetricName("group", "type", "name", scope); MetricWrapper metricWrapper = newYammerMetricWrapper(metricName, value::get); - collector.addMetric(metricName, metricWrapper); + reporter.addMetric(metricName, metricWrapper); metrics = collector.collect(); assertEquals(1, metrics.size()); @@ -61,14 +72,20 @@ public void testCollectYammerMetrics() { assertGaugeSnapshot(metrics.get(0), 3, labels); // Removing the metric - collector.removeMetric(metricName); + reporter.removeMetric(metricName); metrics = collector.collect(); assertEquals(0, metrics.size()); } @Test public void testCollectNonNumericYammerMetrics() { - YammerCollector collector = new YammerCollector(); + AbstractReporter reporter = new AbstractReporter() { + @Override + protected Pattern allowlist() { + return Pattern.compile(".*"); + } + }; + collector.addReporter(reporter); List metrics = collector.collect(); assertEquals(0, metrics.size()); @@ -76,7 +93,7 @@ public void testCollectNonNumericYammerMetrics() { String nonNumericValue = "value"; MetricName metricName = new MetricName("group", "type", "name", scope); MetricWrapper metricWrapper = newYammerMetricWrapper(metricName, () -> nonNumericValue); - collector.addMetric(metricName, metricWrapper); + reporter.addMetric(metricName, metricWrapper); metrics = collector.collect(); assertEquals(1, metrics.size()); diff --git a/src/test/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapperTest.java b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapperTest.java similarity index 91% rename from src/test/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapperTest.java rename to server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapperTest.java index 4acc400..6ae12bb 100644 --- a/src/test/java/io/strimzi/kafka/metrics/yammer/YammerMetricWrapperTest.java +++ b/server-metrics-reporter/src/test/java/io/strimzi/kafka/metrics/prometheus/yammer/YammerMetricWrapperTest.java @@ -2,17 +2,17 @@ * Copyright Strimzi authors. * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). */ -package io.strimzi.kafka.metrics.yammer; +package io.strimzi.kafka.metrics.prometheus.yammer; import com.yammer.metrics.core.Gauge; import com.yammer.metrics.core.MetricName; import io.prometheus.metrics.model.snapshots.Labels; import io.prometheus.metrics.model.snapshots.PrometheusNaming; -import io.strimzi.kafka.metrics.TestUtils; import org.junit.jupiter.api.Test; import java.util.concurrent.atomic.AtomicInteger; +import static io.strimzi.kafka.metrics.prometheus.YammerTestUtils.newYammerMetric; import static org.junit.jupiter.api.Assertions.assertEquals; public class YammerMetricWrapperTest { @@ -42,7 +42,7 @@ public void testYammerMetricName() { public void testYammerMetric() { AtomicInteger value = new AtomicInteger(0); MetricName name = new MetricName("group", "type", "name"); - Gauge metric = TestUtils.newYammerMetric(value::get); + Gauge metric = newYammerMetric(value::get); YammerMetricWrapper wrapper = new YammerMetricWrapper(YammerMetricWrapper.prometheusName(name), "", metric, "name"); assertEquals(value.get(), ((Gauge) wrapper.metric()).value()); value.incrementAndGet(); diff --git a/src/main/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporter.java b/src/main/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporter.java deleted file mode 100644 index 2f30d83..0000000 --- a/src/main/java/io/strimzi/kafka/metrics/YammerPrometheusMetricsReporter.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics; - -import com.yammer.metrics.Metrics; -import com.yammer.metrics.core.Metric; -import com.yammer.metrics.core.MetricName; -import com.yammer.metrics.core.MetricsRegistry; -import com.yammer.metrics.core.MetricsRegistryListener; -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; -import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.strimzi.kafka.metrics.yammer.YammerCollector; -import io.strimzi.kafka.metrics.yammer.YammerMetricWrapper; -import kafka.metrics.KafkaMetricsReporter; -import kafka.utils.VerifiableProperties; -import org.apache.kafka.server.metrics.KafkaYammerMetrics; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Arrays; - -/** - * KafkaMetricsReporter to export Kafka broker metrics in the Prometheus format. - */ -public class YammerPrometheusMetricsReporter implements KafkaMetricsReporter, MetricsRegistryListener { - - private static final Logger LOG = LoggerFactory.getLogger(YammerPrometheusMetricsReporter.class); - - private final PrometheusRegistry registry; - private final YammerCollector yammerCollector; - @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the init method - /* test */ PrometheusMetricsReporterConfig config; - - /** - * Constructor - */ - public YammerPrometheusMetricsReporter() { - registry = PrometheusRegistry.defaultRegistry; - yammerCollector = YammerCollector.getCollector(PrometheusCollector.register(registry)); - } - - // for testing - YammerPrometheusMetricsReporter(PrometheusRegistry registry, PrometheusCollector prometheusCollector) { - this.registry = registry; - yammerCollector = YammerCollector.getCollector(prometheusCollector); - } - - @Override - public void init(VerifiableProperties props) { - config = new PrometheusMetricsReporterConfig(props.props(), registry); - for (MetricsRegistry yammerRegistry : Arrays.asList(KafkaYammerMetrics.defaultRegistry(), Metrics.defaultRegistry())) { - yammerRegistry.addListener(this); - } - LOG.debug("YammerPrometheusMetricsReporter configured with {}", config); - } - - @Override - public void onMetricAdded(MetricName name, Metric metric) { - String prometheusName = YammerMetricWrapper.prometheusName(name); - if (!config.isAllowed(prometheusName)) { - LOG.trace("Ignoring metric {} as it does not match the allowlist", prometheusName); - } else { - MetricWrapper metricWrapper = new YammerMetricWrapper(prometheusName, name.getScope(), metric, name.getName()); - yammerCollector.addMetric(name, metricWrapper); - } - } - - @Override - public void onMetricRemoved(MetricName name) { - yammerCollector.removeMetric(name); - } -} diff --git a/src/main/java/io/strimzi/kafka/metrics/yammer/YammerCollector.java b/src/main/java/io/strimzi/kafka/metrics/yammer/YammerCollector.java deleted file mode 100644 index 9c8deeb..0000000 --- a/src/main/java/io/strimzi/kafka/metrics/yammer/YammerCollector.java +++ /dev/null @@ -1,143 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics.yammer; - -import com.yammer.metrics.core.Counter; -import com.yammer.metrics.core.Gauge; -import com.yammer.metrics.core.Histogram; -import com.yammer.metrics.core.Meter; -import com.yammer.metrics.core.MetricName; -import com.yammer.metrics.core.Sampling; -import com.yammer.metrics.core.Timer; -import io.prometheus.metrics.model.snapshots.CounterSnapshot; -import io.prometheus.metrics.model.snapshots.GaugeSnapshot; -import io.prometheus.metrics.model.snapshots.InfoSnapshot; -import io.prometheus.metrics.model.snapshots.Labels; -import io.prometheus.metrics.model.snapshots.MetricSnapshot; -import io.prometheus.metrics.model.snapshots.Quantile; -import io.prometheus.metrics.model.snapshots.Quantiles; -import io.prometheus.metrics.model.snapshots.SummarySnapshot; -import io.strimzi.kafka.metrics.DataPointSnapshotBuilder; -import io.strimzi.kafka.metrics.MetricWrapper; -import io.strimzi.kafka.metrics.MetricsCollector; -import io.strimzi.kafka.metrics.PrometheusCollector; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicBoolean; - -/** - * Collector for Yammer metrics - */ -@SuppressWarnings("ClassFanOutComplexity") -public class YammerCollector implements MetricsCollector { - - private static final Logger LOG = LoggerFactory.getLogger(YammerCollector.class); - private static final YammerCollector INSTANCE = new YammerCollector(); - private static final AtomicBoolean REGISTERED = new AtomicBoolean(false); - private static final List QUANTILES = Arrays.asList(0.50, 0.75, 0.95, 0.98, 0.99, 0.999); - - private final Map yammerMetrics = new ConcurrentHashMap<>(); - - /* for testing */ YammerCollector() {} - - /** - * Retrieve the YammerCollector instance - * - * @param prometheusCollector the PrometheusCollector that will collect metrics - * @return the YammerCollector singleton - */ - public static YammerCollector getCollector(PrometheusCollector prometheusCollector) { - if (REGISTERED.compareAndSet(false, true)) { - prometheusCollector.addCollector(INSTANCE); - } - return INSTANCE; - } - - /** - * Add a Yammer metric to be collected. - * - * @param name The name of the Yammer metric to add. - * @param metric The Yammer metric to add. - */ - public void addMetric(MetricName name, MetricWrapper metric) { - yammerMetrics.put(name, metric); - } - - /** - * Remove a Yammer metric from collection. - * - * @param name The name of the Yammer metric to remove. - */ - public void removeMetric(MetricName name) { - yammerMetrics.remove(name); - } - - /** - * Collect all the metrics added to this Collector - * - * @return the list of metrics of this collector - */ - @SuppressWarnings({"CyclomaticComplexity", "JavaNCSS"}) - @Override - public List collect() { - Map> builders = new HashMap<>(); - for (MetricWrapper metricWrapper : yammerMetrics.values()) { - String prometheusMetricName = metricWrapper.prometheusName(); - Object metric = metricWrapper.metric(); - Labels labels = metricWrapper.labels(); - LOG.debug("Collecting Yammer metric {} with the following labels: {}", prometheusMetricName, labels); - - if (metric instanceof Counter) { - Counter counter = (Counter) metric; - CounterSnapshot.Builder builder = (CounterSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> CounterSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.counterDataPoint(labels, counter.count())); - } else if (metric instanceof Gauge) { - Object valueObj = ((Gauge) metric).value(); - if (valueObj instanceof Number) { - double value = ((Number) valueObj).doubleValue(); - GaugeSnapshot.Builder builder = (GaugeSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> GaugeSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.gaugeDataPoint(labels, value)); - } else { - InfoSnapshot.Builder builder = (InfoSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> InfoSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.infoDataPoint(labels, valueObj, metricWrapper.attribute())); - } - } else if (metric instanceof Timer) { - Timer timer = (Timer) metric; - SummarySnapshot.Builder builder = (SummarySnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> SummarySnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.summaryDataPoint(labels, timer.count(), timer.sum(), quantiles(timer))); - } else if (metric instanceof Histogram) { - Histogram histogram = (Histogram) metric; - SummarySnapshot.Builder builder = (SummarySnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> SummarySnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.summaryDataPoint(labels, histogram.count(), histogram.sum(), quantiles(histogram))); - } else if (metric instanceof Meter) { - Meter meter = (Meter) metric; - CounterSnapshot.Builder builder = (CounterSnapshot.Builder) builders.computeIfAbsent(prometheusMetricName, k -> CounterSnapshot.builder().name(prometheusMetricName)); - builder.dataPoint(DataPointSnapshotBuilder.counterDataPoint(labels, meter.count())); - } else { - LOG.error("The metric {} has an unexpected type: {}", prometheusMetricName, metric.getClass().getName()); - } - } - List snapshots = new ArrayList<>(); - for (MetricSnapshot.Builder builder : builders.values()) { - snapshots.add(builder.build()); - } - return snapshots; - } - - private static Quantiles quantiles(Sampling sampling) { - Quantiles.Builder quantilesBuilder = Quantiles.builder(); - for (double quantile : QUANTILES) { - quantilesBuilder.quantile(new Quantile(quantile, sampling.getSnapshot().getValue(quantile))); - } - return quantilesBuilder.build(); - } -} diff --git a/src/test/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfigTest.java b/src/test/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfigTest.java deleted file mode 100644 index ae7a2b0..0000000 --- a/src/test/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfigTest.java +++ /dev/null @@ -1,109 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics; - -import io.prometheus.metrics.model.registry.PrometheusRegistry; -import io.strimzi.kafka.metrics.http.HttpServers; -import org.apache.kafka.common.config.ConfigException; -import org.junit.jupiter.api.Test; - -import java.net.BindException; -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; - -import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.ALLOWLIST_CONFIG; -import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.LISTENER_CONFIG; -import static io.strimzi.kafka.metrics.PrometheusMetricsReporterConfig.LISTENER_ENABLE_CONFIG; -import static java.util.Collections.emptyMap; -import static java.util.Collections.singletonMap; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertInstanceOf; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; - - -public class PrometheusMetricsReporterConfigTest { - @Test - public void testDefaults() { - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(emptyMap(), new PrometheusRegistry()); - assertEquals(PrometheusMetricsReporterConfig.LISTENER_CONFIG_DEFAULT, config.listener()); - assertTrue(config.isAllowed("random_name")); - } - - @Test - public void testOverrides() { - Map props = new HashMap<>(); - props.put(LISTENER_CONFIG, "http://:0"); - props.put(ALLOWLIST_CONFIG, "kafka_server.*"); - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - - assertEquals("http://:0", config.listener()); - assertFalse(config.isAllowed("random_name")); - assertTrue(config.isAllowed("kafka_server_metric")); - } - - @Test - public void testAllowList() { - Map props = singletonMap(ALLOWLIST_CONFIG, "kafka_server.*,kafka_network.*"); - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - - assertFalse(config.isAllowed("random_name")); - assertTrue(config.isAllowed("kafka_server_metric")); - assertTrue(config.isAllowed("kafka_network_metric")); - - assertThrows(ConfigException.class, - () -> new PrometheusMetricsReporterConfig(singletonMap(ALLOWLIST_CONFIG, "hell[o,s]world"), null)); - assertThrows(ConfigException.class, - () -> new PrometheusMetricsReporterConfig(singletonMap(ALLOWLIST_CONFIG, "hello\\,world"), null)); - } - - @Test - public void testIsListenerEnabled() { - Map props = new HashMap<>(); - props.put(LISTENER_ENABLE_CONFIG, "true"); - props.put(LISTENER_CONFIG, "http://:0"); - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - Optional httpServerOptional = config.startHttpServer(); - - assertTrue(config.isListenerEnabled()); - assertTrue(httpServerOptional.isPresent()); - HttpServers.release(httpServerOptional.get()); - } - - @Test - public void testIsListenerDisabled() { - Map props = singletonMap(LISTENER_ENABLE_CONFIG, false); - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - Optional httpServerOptional = config.startHttpServer(); - - assertTrue(httpServerOptional.isEmpty()); - assertFalse(config.isListenerEnabled()); - } - - @Test - public void testStartHttpServer() { - Map props = new HashMap<>(); - props.put(LISTENER_CONFIG, "http://:0"); - PrometheusMetricsReporterConfig config = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - Optional httpServerOptional = config.startHttpServer(); - assertTrue(httpServerOptional.isPresent()); - - PrometheusMetricsReporterConfig config2 = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - Optional httpServerOptional2 = config2.startHttpServer(); - assertTrue(httpServerOptional2.isPresent()); - - props = new HashMap<>(); - props.put(LISTENER_CONFIG, "http://:" + httpServerOptional.get().port()); - PrometheusMetricsReporterConfig config3 = new PrometheusMetricsReporterConfig(props, new PrometheusRegistry()); - Exception exc = assertThrows(RuntimeException.class, config3::startHttpServer); - assertInstanceOf(BindException.class, exc.getCause()); - - HttpServers.release(httpServerOptional.get()); - HttpServers.release(httpServerOptional2.get()); - } -} - diff --git a/src/test/java/io/strimzi/kafka/metrics/integration/TestBrokerMetricsIT.java b/src/test/java/io/strimzi/kafka/metrics/integration/TestBrokerMetricsIT.java deleted file mode 100644 index d15d5fd..0000000 --- a/src/test/java/io/strimzi/kafka/metrics/integration/TestBrokerMetricsIT.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics.integration; - -import io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter; -import io.strimzi.kafka.metrics.TestUtils; -import io.strimzi.kafka.metrics.YammerPrometheusMetricsReporter; -import io.strimzi.test.container.StrimziKafkaContainer; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.testcontainers.utility.MountableFile; - -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public class TestBrokerMetricsIT { - - private Map configs; - private StrimziKafkaContainer broker; - - @BeforeEach - public void setUp() { - configs = new HashMap<>(); - configs.put("metric.reporters", KafkaPrometheusMetricsReporter.class.getName()); - configs.put("kafka.metrics.reporters", YammerPrometheusMetricsReporter.class.getName()); - - broker = new StrimziKafkaContainer() - .withNodeId(0) - .withKraft() - .withCopyFileToContainer(MountableFile.forHostPath(TestUtils.REPORTER_JARS), TestUtils.MOUNT_PATH) - .withExposedPorts(9092, TestUtils.PORT) - .withKafkaConfigurationMap(configs) - .withEnv(Collections.singletonMap("CLASSPATH", TestUtils.MOUNT_PATH + "*")); - } - - @AfterEach - public void tearDown() { - broker.stop(); - } - - @Test - public void testBrokerMetrics() throws Exception { - broker.start(); - - List prefixes = List.of( - "jvm_", - "process_", - "kafka_controller_", - "kafka_coordinator_", - "kafka_log_", - "kafka_network_", - "kafka_server_"); - List metrics = TestUtils.getMetrics(broker.getHost(), broker.getMappedPort(TestUtils.PORT)); - for (String prefix : prefixes) { - assertFalse(TestUtils.filterMetrics(metrics, prefix).isEmpty()); - } - } - - @Test - public void testBrokerMetricsWithAllowlist() throws Exception { - configs.put("prometheus.metrics.reporter.allowlist", "kafka_controller.*,kafka_server.*"); - broker.withKafkaConfigurationMap(configs); - broker.start(); - - List metrics = TestUtils.getMetrics(broker.getHost(), broker.getMappedPort(TestUtils.PORT)); - List allowedPrefixes = List.of( - "jvm_", - "process_", - "kafka_controller_", - "kafka_server_"); - for (String prefix : allowedPrefixes) { - assertFalse(TestUtils.filterMetrics(metrics, prefix).isEmpty()); - } - List disallowPrefixes = List.of( - "kafka_coordinator_", - "kafka_log_", - "kafka_network_"); - for (String prefix : disallowPrefixes) { - assertTrue(TestUtils.filterMetrics(metrics, prefix).isEmpty()); - } - } -} diff --git a/src/test/java/io/strimzi/kafka/metrics/integration/TestConsumerMetricsIT.java b/src/test/java/io/strimzi/kafka/metrics/integration/TestConsumerMetricsIT.java deleted file mode 100644 index 7395225..0000000 --- a/src/test/java/io/strimzi/kafka/metrics/integration/TestConsumerMetricsIT.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics.integration; - -import io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter; -import io.strimzi.kafka.metrics.TestUtils; -import io.strimzi.test.container.StrimziKafkaContainer; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.GenericContainer; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public class TestConsumerMetricsIT { - - private StrimziKafkaContainer broker; - private Map env; - - @BeforeEach - public void setUp() { - broker = new StrimziKafkaContainer() - .withKraft() - .withNetworkAliases(TestUtils.KAFKA_NETWORK_ALIAS); - broker.start(); - - env = new HashMap<>(); - env.put("CLIENT_TYPE", "KafkaConsumer"); - env.put("BOOTSTRAP_SERVERS", TestUtils.KAFKA_NETWORK_ALIAS + ":9091"); - env.put("TOPIC", "my-topic"); - env.put("GROUP_ID", "my-group"); - env.put("ADDITIONAL_CONFIG", "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName()); - env.put("CLASSPATH", TestUtils.MOUNT_PATH + "*"); - env.put("MESSAGE_COUNT", "1000"); - } - - @AfterEach - public void tearDown() { - broker.stop(); - } - - @Test - public void testConsumerMetrics() { - try (GenericContainer consumer = TestUtils.clientContainer(env)) { - consumer.start(); - - List prefixes = List.of( - "jvm_", - "process_", - "kafka_consumer_app_info_", - "kafka_consumer_kafka_metrics_", - "kafka_consumer_consumer_metrics_", - "kafka_consumer_consumer_node_metrics_", - "kafka_consumer_consumer_coordinator_metrics_", - "kafka_consumer_consumer_fetch_manager_metrics_"); - for (String prefix : prefixes) { - TestUtils.verify(consumer, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - } - } - - @Test - public void testConsumerMetricsWithAllowlist() { - env.put("ADDITIONAL_CONFIG", - "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName() + "\n" + - "prometheus.metrics.reporter.allowlist=kafka_consumer_kafka_metrics_.*,kafka_consumer_consumer_coordinator_metrics_.*"); - try (GenericContainer consumer = TestUtils.clientContainer(env)) { - consumer.start(); - - List allowedPrefixes = List.of( - "jvm_", - "process_", - "kafka_consumer_kafka_metrics_", - "kafka_consumer_consumer_coordinator_metrics_"); - for (String prefix : allowedPrefixes) { - TestUtils.verify(consumer, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - List disallowedPrefixes = List.of( - "kafka_consumer_app_info_", - "kafka_consumer_consumer_metrics_", - "kafka_consumer_consumer_node_metrics_", - "kafka_consumer_consumer_fetch_manager_metrics_"); - for (String prefix : disallowedPrefixes) { - TestUtils.verify(consumer, prefix, metrics -> assertTrue(metrics.isEmpty())); - } - } - } -} diff --git a/src/test/java/io/strimzi/kafka/metrics/integration/TestProducerMetricsIT.java b/src/test/java/io/strimzi/kafka/metrics/integration/TestProducerMetricsIT.java deleted file mode 100644 index 26687df..0000000 --- a/src/test/java/io/strimzi/kafka/metrics/integration/TestProducerMetricsIT.java +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics.integration; - -import io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter; -import io.strimzi.kafka.metrics.TestUtils; -import io.strimzi.test.container.StrimziKafkaContainer; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.GenericContainer; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public class TestProducerMetricsIT { - - private StrimziKafkaContainer broker; - private Map env; - - @BeforeEach - public void setUp() { - broker = new StrimziKafkaContainer() - .withKraft() - .withNetworkAliases(TestUtils.KAFKA_NETWORK_ALIAS); - broker.start(); - - env = new HashMap<>(); - env.put("CLIENT_TYPE", "KafkaProducer"); - env.put("BOOTSTRAP_SERVERS", TestUtils.KAFKA_NETWORK_ALIAS + ":9091"); - env.put("TOPIC", "my-topic"); - env.put("ADDITIONAL_CONFIG", "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName()); - env.put("CLASSPATH", TestUtils.MOUNT_PATH + "*"); - env.put("MESSAGE_COUNT", "1000"); - env.put("DELAY_MS", "100"); - } - - @AfterEach - public void tearDown() { - broker.stop(); - } - - @Test - public void testProducerMetrics() { - try (GenericContainer producer = TestUtils.clientContainer(env)) { - producer.start(); - - List prefixes = List.of( - "jvm_", - "process_", - "kafka_producer_app_info_", - "kafka_producer_kafka_metrics_", - "kafka_producer_producer_metrics_", - "kafka_producer_producer_node_metrics_", - "kafka_producer_producer_topic_metrics_"); - for (String prefix : prefixes) { - TestUtils.verify(producer, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - } - } - - @Test - public void testProducerMetricsWithAllowlist() { - env.put("ADDITIONAL_CONFIG", - "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName() + "\n" + - "prometheus.metrics.reporter.allowlist=kafka_producer_kafka_metrics_.*,kafka_producer_producer_topic_metrics_.*"); - try (GenericContainer producer = TestUtils.clientContainer(env)) { - producer.start(); - - List allowedPrefixes = List.of( - "jvm_", - "process_", - "kafka_producer_kafka_metrics_", - "kafka_producer_producer_topic_metrics_"); - for (String prefix : allowedPrefixes) { - TestUtils.verify(producer, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - List disallowedPrefixes = List.of( - "kafka_producer_app_info_", - "kafka_producer_producer_metrics_", - "kafka_producer_producer_node_metrics_"); - for (String prefix : disallowedPrefixes) { - TestUtils.verify(producer, prefix, metrics -> assertTrue(metrics.isEmpty())); - } - } - } -} diff --git a/src/test/java/io/strimzi/kafka/metrics/integration/TestStreamsMetricsIT.java b/src/test/java/io/strimzi/kafka/metrics/integration/TestStreamsMetricsIT.java deleted file mode 100644 index 8337903..0000000 --- a/src/test/java/io/strimzi/kafka/metrics/integration/TestStreamsMetricsIT.java +++ /dev/null @@ -1,151 +0,0 @@ -/* - * Copyright Strimzi authors. - * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). - */ -package io.strimzi.kafka.metrics.integration; - -import io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter; -import io.strimzi.kafka.metrics.TestUtils; -import io.strimzi.test.container.StrimziKafkaContainer; -import org.apache.kafka.clients.admin.Admin; -import org.apache.kafka.clients.admin.AdminClientConfig; -import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.GenericContainer; - -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public class TestStreamsMetricsIT { - - private StrimziKafkaContainer broker; - private Map env; - - @BeforeEach - public void setUp() throws Exception { - broker = new StrimziKafkaContainer() - .withKraft() - .withNetworkAliases(TestUtils.KAFKA_NETWORK_ALIAS); - broker.start(); - - try (Admin admin = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBootstrapServers()))) { - admin.createTopics(Arrays.asList( - new NewTopic("source-topic", 1, (short) -1), - new NewTopic("target-topic", 1, (short) -1)) - ).all().get(); - } - - Map producerConfigs = Map.of( - ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBootstrapServers(), - ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(), - ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); - try (KafkaProducer producer = new KafkaProducer<>(producerConfigs)) { - for (int i = 0; i < 10; i++) { - producer.send(new ProducerRecord<>("source-topic", "record" + i)); - } - producer.flush(); - } - - env = new HashMap<>(); - env.put("CLIENT_TYPE", "KafkaStreams"); - env.put("BOOTSTRAP_SERVERS", TestUtils.KAFKA_NETWORK_ALIAS + ":9091"); - env.put("APPLICATION_ID", "my-app-id"); - env.put("SOURCE_TOPIC", "source-topic"); - env.put("TARGET_TOPIC", "target-topic"); - env.put("ADDITIONAL_CONFIG", "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName()); - env.put("CLASSPATH", TestUtils.MOUNT_PATH + "*"); - } - - @AfterEach - public void tearDown() { - broker.stop(); - } - - @Test - public void testStreamsMetrics() { - try (GenericContainer streams = TestUtils.clientContainer(env)) { - streams.start(); - - List prefixes = List.of( - "jvm_", - "process_", - "kafka_admin_client_app_info_", - "kafka_admin_client_kafka_metrics_", - "kafka_admin_client_admin_client_metrics_", - "kafka_admin_client_admin_client_node_metrics_", - "kafka_consumer_app_info_", - "kafka_consumer_kafka_metrics_", - "kafka_consumer_consumer_metrics_", - "kafka_consumer_consumer_node_metrics_", - "kafka_consumer_consumer_coordinator_metrics_", - "kafka_consumer_consumer_fetch_manager_metrics_", - "kafka_producer_app_info_", - "kafka_producer_kafka_metrics_", - "kafka_producer_producer_metrics_", - "kafka_producer_producer_node_metrics_", - "kafka_producer_producer_topic_metrics_", - "kafka_streams_stream_metrics_", - "kafka_streams_stream_processor_node_metrics_", - "kafka_streams_stream_state_updater_metrics_", - "kafka_streams_stream_task_metrics_", - "kafka_streams_stream_thread_metrics_", - "kafka_streams_stream_topic_metrics_"); - for (String prefix : prefixes) { - TestUtils.verify(streams, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - } - } - - @Test - public void testStreamsMetricsWithAllowlist() { - env.put("ADDITIONAL_CONFIG", - "metric.reporters=" + KafkaPrometheusMetricsReporter.class.getName() + "\n" + - "prometheus.metrics.reporter.allowlist=kafka_consumer_.*,kafka_streams_stream_metrics_.*"); - try (GenericContainer streams = TestUtils.clientContainer(env)) { - streams.start(); - - List allowedPrefixes = List.of( - "jvm_", - "process_", - "kafka_consumer_app_info_", - "kafka_consumer_kafka_metrics_", - "kafka_consumer_consumer_metrics_", - "kafka_consumer_consumer_node_metrics_", - "kafka_consumer_consumer_coordinator_metrics_", - "kafka_consumer_consumer_fetch_manager_metrics_", - "kafka_streams_stream_metrics_"); - for (String prefix : allowedPrefixes) { - TestUtils.verify(streams, prefix, metrics -> assertFalse(metrics.isEmpty())); - } - List disallowedPrefixes = List.of( - "kafka_admin_client_app_info_", - "kafka_admin_client_kafka_metrics_", - "kafka_admin_client_admin_client_metrics_", - "kafka_admin_client_admin_client_node_metrics_", - "kafka_producer_app_info_", - "kafka_producer_kafka_metrics_", - "kafka_producer_producer_metrics_", - "kafka_producer_producer_node_metrics_", - "kafka_producer_producer_topic_metrics_", - "kafka_streams_stream_processor_node_metrics_", - "kafka_streams_stream_state_updater_metrics_", - "kafka_streams_stream_task_metrics_", - "kafka_streams_stream_thread_metrics_", - "kafka_streams_stream_topic_metrics_"); - for (String prefix : disallowedPrefixes) { - TestUtils.verify(streams, prefix, metrics -> assertTrue(metrics.isEmpty())); - } - } - } -}