Skip to content

Commit 69a96a8

Browse files
mimaisonmimaison
committed
Split the reporter into client and server modules
Signed-off-by: Mickael Maison <[email protected]>
1 parent e27b598 commit 69a96a8

File tree

49 files changed

+1945
-1128
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

49 files changed

+1945
-1128
lines changed

‎README.md

+15-11
Original file line numberDiff line numberDiff line change
@@ -37,47 +37,51 @@ The metrics reporter has the following configurations:
3737
### Kafka Brokers
3838

3939
To use the reporter with Kafka brokers, add the following to your broker configuration:
40+
4041
```properties
41-
metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
42-
kafka.metrics.reporters=io.strimzi.kafka.metrics.YammerPrometheusMetricsReporter
42+
metric.reporters=io.strimzi.kafka.metrics.prometheus.ServerKafkaMetricsReporter
43+
kafka.metrics.reporters=io.strimzi.kafka.metrics.prometheus.ServerYammerMetricsReporter
4344
auto.include.jmx.reporter=false
4445
```
4546

4647
### Kafka Clients
4748

4849
To use the reporter with Kafka producers, consumers or admin clients, add the following to your client configuration:
50+
4951
```properties
50-
metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
52+
metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
5153
auto.include.jmx.reporter=false
5254
```
5355

5456
### Kafka Connect and Kafka Streams
5557

5658
To use the reporter with Kafka Connect and Kafka Streams, add the following to your Connect runtime or Streams application configuration:
59+
5760
```properties
58-
metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
61+
metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
5962
auto.include.jmx.reporter=false
60-
admin.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
63+
admin.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
6164
admin.auto.include.jmx.reporter=false
62-
producer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
65+
producer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
6366
producer.auto.include.jmx.reporter=false
64-
consumer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
67+
consumer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
6568
consumer.auto.include.jmx.reporter=false
6669
```
6770

6871
When setting configurations for the Prometheus metrics reporter, they also need to be set with the `admin.`, `producer.` and `consumer.`.
6972
For example, to set the `listener` to `http://:8081`:
73+
7074
```properties
71-
metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
75+
metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
7276
prometheus.metrics.reporter.listener=http://:8081
7377
auto.include.jmx.reporter=false
74-
admin.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
78+
admin.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
7579
admin.prometheus.metrics.reporter.listener=http://:8081
7680
admin.auto.include.jmx.reporter=false
77-
producer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
81+
producer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
7882
producer.prometheus.metrics.reporter.listener=http://:8081
7983
producer.auto.include.jmx.reporter=false
80-
consumer.metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
84+
consumer.metric.reporters=io.strimzi.kafka.metrics.prometheus.ClientMetricsReporter
8185
consumer.prometheus.metrics.reporter.listener=http://:8081
8286
consumer.auto.include.jmx.reporter=false
8387
```

‎client-metrics-reporter/pom.xml

+92
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>io.strimzi</groupId>
8+
<artifactId>metrics-reporter</artifactId>
9+
<version>1.0.0-SNAPSHOT</version>
10+
</parent>
11+
12+
<artifactId>client-metrics-reporter</artifactId>
13+
14+
<dependencies>
15+
<dependency>
16+
<groupId>org.apache.kafka</groupId>
17+
<artifactId>kafka-clients</artifactId>
18+
<version>${kafka.version}</version>
19+
<scope>provided</scope>
20+
</dependency>
21+
<dependency>
22+
<groupId>org.slf4j</groupId>
23+
<artifactId>slf4j-api</artifactId>
24+
<version>${slf4j.version}</version>
25+
<scope>provided</scope>
26+
</dependency>
27+
<dependency>
28+
<groupId>com.github.spotbugs</groupId>
29+
<artifactId>spotbugs-annotations</artifactId>
30+
<version>${spotbugs.version}</version>
31+
<scope>provided</scope>
32+
</dependency>
33+
34+
<dependency>
35+
<groupId>io.prometheus</groupId>
36+
<artifactId>prometheus-metrics-model</artifactId>
37+
<version>${prometheus.version}</version>
38+
</dependency>
39+
<dependency>
40+
<groupId>io.prometheus</groupId>
41+
<artifactId>prometheus-metrics-instrumentation-jvm</artifactId>
42+
<version>${prometheus.version}</version>
43+
</dependency>
44+
<dependency>
45+
<groupId>io.prometheus</groupId>
46+
<artifactId>prometheus-metrics-exporter-httpserver</artifactId>
47+
<version>${prometheus.version}</version>
48+
</dependency>
49+
50+
<dependency>
51+
<groupId>org.junit.jupiter</groupId>
52+
<artifactId>junit-jupiter-api</artifactId>
53+
<version>${junit.version}</version>
54+
<scope>test</scope>
55+
</dependency>
56+
<dependency>
57+
<groupId>org.slf4j</groupId>
58+
<artifactId>slf4j-simple</artifactId>
59+
<version>${slf4j.version}</version>
60+
<scope>test</scope>
61+
</dependency>
62+
<dependency>
63+
<groupId>io.strimzi</groupId>
64+
<artifactId>strimzi-test-container</artifactId>
65+
<version>${strimzi-test-container.version}</version>
66+
<scope>test</scope>
67+
</dependency>
68+
<dependency>
69+
<groupId>org.testcontainers</groupId>
70+
<artifactId>testcontainers</artifactId>
71+
<version>${testcontainers.version}</version>
72+
<scope>test</scope>
73+
</dependency>
74+
</dependencies>
75+
76+
<build>
77+
<plugins>
78+
<plugin>
79+
<groupId>org.apache.maven.plugins</groupId>
80+
<artifactId>maven-jar-plugin</artifactId>
81+
<version>3.4.2</version>
82+
<executions>
83+
<execution>
84+
<goals>
85+
<goal>test-jar</goal>
86+
</goals>
87+
</execution>
88+
</executions>
89+
</plugin>
90+
</plugins>
91+
</build>
92+
</project>

‎src/main/java/io/strimzi/kafka/metrics/KafkaPrometheusMetricsReporter.java ‎client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporter.java

+46-30
Original file line numberDiff line numberDiff line change
@@ -2,63 +2,73 @@
22
* Copyright Strimzi authors.
33
* License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
44
*/
5-
package io.strimzi.kafka.metrics;
5+
package io.strimzi.kafka.metrics.prometheus;
66

77
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
88
import io.prometheus.metrics.model.registry.PrometheusRegistry;
99
import io.prometheus.metrics.model.snapshots.PrometheusNaming;
10-
import io.strimzi.kafka.metrics.http.HttpServers;
11-
import io.strimzi.kafka.metrics.kafka.KafkaCollector;
12-
import io.strimzi.kafka.metrics.kafka.KafkaMetricWrapper;
10+
import io.strimzi.kafka.metrics.prometheus.common.AbstractReporter;
11+
import io.strimzi.kafka.metrics.prometheus.common.MetricWrapper;
12+
import io.strimzi.kafka.metrics.prometheus.common.PrometheusCollector;
13+
import io.strimzi.kafka.metrics.prometheus.http.HttpServers;
14+
import io.strimzi.kafka.metrics.prometheus.kafka.KafkaCollector;
15+
import io.strimzi.kafka.metrics.prometheus.kafka.KafkaMetricWrapper;
1316
import org.apache.kafka.common.config.ConfigException;
1417
import org.apache.kafka.common.metrics.KafkaMetric;
1518
import org.apache.kafka.common.metrics.MetricsContext;
1619
import org.apache.kafka.common.metrics.MetricsReporter;
1720
import org.slf4j.Logger;
1821
import org.slf4j.LoggerFactory;
1922

20-
import java.util.Collections;
2123
import java.util.List;
2224
import java.util.Map;
2325
import java.util.Optional;
2426
import java.util.Set;
27+
import java.util.regex.Pattern;
2528

2629
/**
27-
* MetricsReporter implementation that expose Kafka metrics in the Prometheus format.
28-
* This can be used by Kafka brokers and clients.
30+
* {@link MetricsReporter} implementation that exposes Kafka client metrics in the Prometheus format.
2931
*/
30-
public class KafkaPrometheusMetricsReporter implements MetricsReporter {
31-
32-
private static final Logger LOG = LoggerFactory.getLogger(KafkaPrometheusMetricsReporter.class);
33-
34-
private final PrometheusRegistry registry;
35-
private final KafkaCollector kafkaCollector;
36-
@SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the configure method
37-
private PrometheusMetricsReporterConfig config;
38-
@SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the configure method
39-
private Optional<HttpServers.ServerCounter> httpServer;
32+
public class ClientMetricsReporter extends AbstractReporter implements MetricsReporter {
33+
34+
private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsReporter.class);
35+
static final Set<String> PREFIXES = Set.of(
36+
"kafka.admin.client",
37+
"kafka.consumer",
38+
"kafka.producer",
39+
"kafka.connect",
40+
"kafka.streams"
41+
);
42+
43+
final PrometheusRegistry registry;
44+
final KafkaCollector kafkaCollector;
45+
46+
ClientMetricsReporterConfig config;
47+
Optional<HttpServers.ServerCounter> httpServer = Optional.empty();
4048
@SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the contextChange method
41-
private String prefix;
49+
String prefix;
4250

4351
/**
4452
* Constructor
4553
*/
46-
public KafkaPrometheusMetricsReporter() {
54+
public ClientMetricsReporter() {
4755
registry = PrometheusRegistry.defaultRegistry;
4856
kafkaCollector = KafkaCollector.getCollector(PrometheusCollector.register(registry));
57+
kafkaCollector.addReporter(this);
4958
}
5059

5160
// for testing
52-
KafkaPrometheusMetricsReporter(PrometheusRegistry registry, KafkaCollector kafkaCollector) {
61+
ClientMetricsReporter(PrometheusRegistry registry, KafkaCollector kafkaCollector) {
5362
this.registry = registry;
5463
this.kafkaCollector = kafkaCollector;
64+
kafkaCollector.addReporter(this);
5565
}
5666

5767
@Override
5868
public void configure(Map<String, ?> map) {
59-
config = new PrometheusMetricsReporterConfig(map, registry);
69+
config = new ClientMetricsReporterConfig(map, registry);
6070
httpServer = config.startHttpServer();
61-
LOG.debug("KafkaPrometheusMetricsReporter configured with {}", config);
71+
LOG.debug("ClientMetricsReporter configured with {}", config);
6272
}
6373

6474
@Override
@@ -70,21 +80,18 @@ public void init(List<KafkaMetric> metrics) {
7080

7181
public void metricChange(KafkaMetric metric) {
7282
String prometheusName = KafkaMetricWrapper.prometheusName(prefix, metric.metricName());
73-
if (!config.isAllowed(prometheusName)) {
74-
LOG.trace("Ignoring metric {} as it does not match the allowlist", prometheusName);
75-
} else {
76-
MetricWrapper metricWrapper = new KafkaMetricWrapper(prometheusName, metric, metric.metricName().name());
77-
kafkaCollector.addMetric(metric.metricName(), metricWrapper);
78-
}
83+
MetricWrapper metricWrapper = new KafkaMetricWrapper(prometheusName, metric, metric.metricName().name());
84+
addMetric(metric, metricWrapper);
7985
}
8086

8187
@Override
8288
public void metricRemoval(KafkaMetric metric) {
83-
kafkaCollector.removeMetric(metric.metricName());
89+
removeMetric(metric);
8490
}
8591

8692
@Override
8793
public void close() {
94+
kafkaCollector.removeReporter(this);
8895
httpServer.ifPresent(HttpServers::release);
8996
}
9097

@@ -98,17 +105,26 @@ public void validateReconfiguration(Map<String, ?> configs) throws ConfigExcepti
98105

99106
@Override
100107
public Set<String> reconfigurableConfigs() {
101-
return Collections.emptySet();
108+
return Set.of();
102109
}
103110

104111
@Override
105112
public void contextChange(MetricsContext metricsContext) {
106113
String prefix = metricsContext.contextLabels().get(MetricsContext.NAMESPACE);
114+
if (!PREFIXES.contains(prefix)) {
115+
throw new IllegalStateException("ClientMetricsReporter should only be used in Kafka servers");
116+
}
107117
this.prefix = PrometheusNaming.prometheusName(prefix);
108118
}
109119

110120
// for testing
111121
Optional<Integer> getPort() {
112122
return Optional.ofNullable(httpServer.isPresent() ? httpServer.get().port() : null);
113123
}
124+
125+
@Override
126+
protected Pattern allowlist() {
127+
return config.allowlist();
128+
}
129+
114130
}

‎src/main/java/io/strimzi/kafka/metrics/PrometheusMetricsReporterConfig.java ‎client-metrics-reporter/src/main/java/io/strimzi/kafka/metrics/prometheus/ClientMetricsReporterConfig.java

+22-14
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@
22
* Copyright Strimzi authors.
33
* License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
44
*/
5-
package io.strimzi.kafka.metrics;
5+
package io.strimzi.kafka.metrics.prometheus;
66

77
import io.prometheus.metrics.model.registry.PrometheusRegistry;
8-
import io.strimzi.kafka.metrics.http.HttpServers;
9-
import io.strimzi.kafka.metrics.http.Listener;
8+
import io.strimzi.kafka.metrics.prometheus.http.HttpServers;
9+
import io.strimzi.kafka.metrics.prometheus.http.Listener;
1010
import org.apache.kafka.common.config.AbstractConfig;
1111
import org.apache.kafka.common.config.ConfigDef;
1212
import org.apache.kafka.common.config.ConfigException;
@@ -22,11 +22,11 @@
2222
import java.util.stream.Collectors;
2323

2424
/**
25-
* Configuration for the PrometheusMetricsReporter implementation.
25+
* Configuration for {@link ClientMetricsReporter}.
2626
*/
27-
public class PrometheusMetricsReporterConfig extends AbstractConfig {
27+
public class ClientMetricsReporterConfig extends AbstractConfig {
2828

29-
private static final Logger LOG = LoggerFactory.getLogger(PrometheusMetricsReporterConfig.class);
29+
private static final Logger LOG = LoggerFactory.getLogger(ClientMetricsReporterConfig.class);
3030
private static final String CONFIG_PREFIX = "prometheus.metrics.reporter.";
3131

3232
/**
@@ -62,23 +62,23 @@ public class PrometheusMetricsReporterConfig extends AbstractConfig {
6262
public static final String ALLOWLIST_CONFIG_DEFAULT = ".*";
6363
private static final String ALLOWLIST_CONFIG_DOC = "A comma separated list of regex patterns to specify the metrics to collect.";
6464

65-
private static final ConfigDef CONFIG_DEF = new ConfigDef()
65+
static final ConfigDef CONFIG_DEF = new ConfigDef()
6666
.define(LISTENER_CONFIG, ConfigDef.Type.STRING, LISTENER_CONFIG_DEFAULT, new Listener.ListenerValidator(), ConfigDef.Importance.HIGH, LISTENER_CONFIG_DOC)
6767
.define(ALLOWLIST_CONFIG, ConfigDef.Type.LIST, ALLOWLIST_CONFIG_DEFAULT, ConfigDef.Importance.HIGH, ALLOWLIST_CONFIG_DOC)
6868
.define(LISTENER_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN, LISTENER_ENABLE_CONFIG_DEFAULT, ConfigDef.Importance.HIGH, LISTENER_ENABLE_CONFIG_DOC);
6969

70-
private final Listener listener;
71-
private final boolean listenerEnabled;
72-
private final Pattern allowlist;
73-
private final PrometheusRegistry registry;
70+
final Listener listener;
71+
final boolean listenerEnabled;
72+
final PrometheusRegistry registry;
73+
final Pattern allowlist;
7474

7575
/**
7676
* Constructor.
7777
*
7878
* @param props the configuration properties.
7979
* @param registry the metrics registry
8080
*/
81-
public PrometheusMetricsReporterConfig(Map<?, ?> props, PrometheusRegistry registry) {
81+
public ClientMetricsReporterConfig(Map<?, ?> props, PrometheusRegistry registry) {
8282
super(CONFIG_DEF, props);
8383
this.listener = Listener.parseListener(getString(LISTENER_CONFIG));
8484
this.allowlist = compileAllowlist(getList(ALLOWLIST_CONFIG));
@@ -96,7 +96,15 @@ public boolean isAllowed(String name) {
9696
return allowlist.matcher(name).matches();
9797
}
9898

99-
private Pattern compileAllowlist(List<String> allowlist) {
99+
/**
100+
* The configured allowlist.
101+
* @return The Pattern for the allowlist
102+
*/
103+
public Pattern allowlist() {
104+
return allowlist;
105+
}
106+
107+
Pattern compileAllowlist(List<String> allowlist) {
100108
for (String entry : allowlist) {
101109
try {
102110
Pattern.compile(entry);
@@ -128,7 +136,7 @@ public boolean isListenerEnabled() {
128136

129137
@Override
130138
public String toString() {
131-
return "PrometheusMetricsReporterConfig{" +
139+
return "ClientMetricsReporterConfig{" +
132140
", listener=" + listener +
133141
", listenerEnabled=" + listenerEnabled +
134142
", allowlist=" + allowlist +

0 commit comments

Comments
 (0)