Skip to content

Commit 5897639

Browse files
fvalerifvaleri
committed
Address comments
Signed-off-by: Federico Valeri <[email protected]>
1 parent 462ae58 commit 5897639

File tree

13 files changed

+90
-116
lines changed

13 files changed

+90
-116
lines changed

‎documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
ae05f2b8839ace3a66d7d1b0eabe1bb6c3ac8b19e39fbd80549883394d1d6ec9
1+
4502368c51039e785930e38d1b134385ac0f20259b840e32f3f96ffee9bbe26c

‎documentation/book/api/index.adoc

+1-2
Original file line numberDiff line numberDiff line change
@@ -1386,7 +1386,6 @@ Retrieves the bridge metrics in Prometheus format.
13861386
===== Content Type
13871387

13881388
* text/plain
1389-
* application/json
13901389

13911390
===== Responses
13921391

@@ -1403,7 +1402,7 @@ Retrieves the bridge metrics in Prometheus format.
14031402

14041403
| 404
14051404
| The metrics endpoint is not enabled.
1406-
| <<Error>>
1405+
| <<>>
14071406

14081407
|===
14091408

‎src/main/java/io/strimzi/kafka/bridge/Application.java

+10-69
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,6 @@
77
import io.strimzi.kafka.bridge.config.BridgeConfig;
88
import io.strimzi.kafka.bridge.config.ConfigRetriever;
99
import io.strimzi.kafka.bridge.http.HttpBridge;
10-
import io.strimzi.kafka.bridge.metrics.JmxMetricsCollector;
11-
import io.strimzi.kafka.bridge.metrics.MetricsCollector;
12-
import io.strimzi.kafka.bridge.metrics.MetricsType;
13-
import io.strimzi.kafka.bridge.metrics.StrimziMetricsCollector;
1410
import io.strimzi.kafka.bridge.tracing.TracingUtil;
1511
import io.vertx.core.Future;
1612
import io.vertx.core.Promise;
@@ -29,17 +25,11 @@
2925
import org.apache.logging.log4j.Logger;
3026

3127
import javax.management.MalformedObjectNameException;
32-
import java.io.BufferedReader;
3328
import java.io.File;
3429
import java.io.IOException;
35-
import java.io.InputStream;
36-
import java.io.InputStreamReader;
37-
import java.nio.charset.StandardCharsets;
38-
import java.nio.file.Files;
3930
import java.util.EnumSet;
4031
import java.util.Map;
4132
import java.util.Set;
42-
import java.util.stream.Collectors;
4333

4434
/**
4535
* Apache Kafka bridge main application class
@@ -83,22 +73,18 @@ private static Future<HttpBridge> deployHttpBridge(BridgeConfig bridgeConfig)
8373
Promise<HttpBridge> httpPromise = Promise.promise();
8474

8575
Vertx vertx = createVertxInstance(bridgeConfig);
86-
MetricsCollector metricsCollector = getMetricsCollector(bridgeConfig);
87-
HttpBridge httpBridge = new HttpBridge(bridgeConfig, metricsCollector);
88-
76+
HttpBridge httpBridge = new HttpBridge(bridgeConfig);
77+
8978
vertx.deployVerticle(httpBridge)
90-
.onComplete(done -> {
91-
if (done.succeeded()) {
92-
LOGGER.info("HTTP verticle instance deployed [{}]", done.result());
93-
if (metricsCollector != null) {
94-
LOGGER.info("Metrics of type '{}' enabled and exposed on /metrics endpoint", bridgeConfig.getMetricsType());
79+
.onComplete(done -> {
80+
if (done.succeeded()) {
81+
LOGGER.info("HTTP verticle instance deployed [{}]", done.result());
82+
httpPromise.complete(httpBridge);
83+
} else {
84+
LOGGER.error("Failed to deploy HTTP verticle instance", done.cause());
85+
httpPromise.fail(done.cause());
9586
}
96-
httpPromise.complete(httpBridge);
97-
} else {
98-
LOGGER.error("Failed to deploy HTTP verticle instance", done.cause());
99-
httpPromise.fail(done.cause());
100-
}
101-
});
87+
});
10288

10389
return httpPromise.future();
10490
}
@@ -129,51 +115,6 @@ private static MicrometerMetricsOptions metricsOptions() {
129115
.setEnabled(true);
130116
}
131117

132-
private static MetricsCollector getMetricsCollector(BridgeConfig bridgeConfig)
133-
throws MalformedObjectNameException, IOException {
134-
if (bridgeConfig.getMetricsType() != null) {
135-
if (bridgeConfig.getMetricsType() == MetricsType.JMX_EXPORTER) {
136-
return getJmxMetricsCollector(bridgeConfig);
137-
} else if (bridgeConfig.getMetricsType() == MetricsType.STRIMZI_REPORTER) {
138-
return new StrimziMetricsCollector();
139-
}
140-
}
141-
return null;
142-
}
143-
144-
/**
145-
* Return a JmxMetricsCollector instance with the YAML configuration filters.
146-
* This is loaded from a custom config file if present or from the default configuration file.
147-
*
148-
* @return JmxCollectorRegistry instance
149-
* @throws MalformedObjectNameException
150-
* @throws IOException
151-
*/
152-
private static JmxMetricsCollector getJmxMetricsCollector(BridgeConfig bridgeConfig) throws MalformedObjectNameException, IOException {
153-
if (bridgeConfig.getJmxExporterConfigPath() != null && Files.exists(bridgeConfig.getJmxExporterConfigPath())) {
154-
// load custom configuration file
155-
LOGGER.info("Loading custom JMX Exporter configuration from {}", bridgeConfig.getJmxExporterConfigPath());
156-
String yaml = Files.readString(bridgeConfig.getJmxExporterConfigPath(), StandardCharsets.UTF_8);
157-
return new JmxMetricsCollector(yaml);
158-
} else {
159-
// load default configuration
160-
if (bridgeConfig.getJmxExporterConfigPath() != null && !Files.exists(bridgeConfig.getJmxExporterConfigPath())) {
161-
LOGGER.warn("Custom JMX Exporter configuration file not found: {}", bridgeConfig.getJmxExporterConfigPath());
162-
}
163-
LOGGER.info("Loading default JMX Exporter configuration");
164-
InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
165-
if (is == null) {
166-
return null;
167-
}
168-
try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
169-
String yaml = reader
170-
.lines()
171-
.collect(Collectors.joining("\n"));
172-
return new JmxMetricsCollector(yaml);
173-
}
174-
}
175-
}
176-
177118
/**
178119
* Generate the command line options
179120
*

‎src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java

+9-4
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,14 @@ public class BridgeConfig extends AbstractConfig {
3131
public static final String METRICS_TYPE = BRIDGE_CONFIG_PREFIX + "metrics";
3232

3333
/** JMX Exporter configuration file path */
34-
public static final String JMX_EXPORTER_CONFIG_PATH = METRICS_TYPE + ".exporter.config.path";
34+
private static final String JMX_EXPORTER_CONFIG_PATH = METRICS_TYPE + ".exporter.config.path";
3535

3636
/** Tracing system to be used in the bridge */
37-
public static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
37+
private static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
38+
39+
/** Default Strimzi Metrics Reporter allow list. */
40+
/* test */ static final String DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST = "kafka_consumer_consumer_metrics.*, " +
41+
"kafka_producer_kafka_metrics_count_count, kafka_producer_producer_metrics.*";
3842

3943
private final KafkaConfig kafkaConfig;
4044
private final HttpConfig httpConfig;
@@ -87,11 +91,12 @@ private static void validateAndApplyDefaults(Map<String, Object> map) {
8791
if (map.get(METRICS_TYPE) != null) {
8892
// get and validate metrics type
8993
MetricsType metricsType = MetricsType.fromString((String) map.get(METRICS_TYPE));
90-
// apply default Strimzi Metrics Reporter configurations if not present
94+
// apply default Strimzi Metrics Reporter configuration if not present
9195
if (metricsType == MetricsType.STRIMZI_REPORTER) {
96+
LOGGER.info("Using default metrics configuration");
9297
map.putIfAbsent("kafka.metric.reporters", "io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter");
9398
map.putIfAbsent("kafka.prometheus.metrics.reporter.listener.enable", "false");
94-
map.putIfAbsent("kafka.prometheus.metrics.reporter.allowlist", ".*");
99+
map.putIfAbsent("kafka.prometheus.metrics.reporter.allowlist", DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST);
95100
}
96101
}
97102
}

‎src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java

+59-10
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
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-
65
package io.strimzi.kafka.bridge.http;
76

87
import com.fasterxml.jackson.databind.JsonNode;
@@ -18,7 +17,10 @@
1817
import io.strimzi.kafka.bridge.config.BridgeConfig;
1918
import io.strimzi.kafka.bridge.http.converter.JsonUtils;
2019
import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
20+
import io.strimzi.kafka.bridge.metrics.JmxMetricsCollector;
2121
import io.strimzi.kafka.bridge.metrics.MetricsCollector;
22+
import io.strimzi.kafka.bridge.metrics.MetricsType;
23+
import io.strimzi.kafka.bridge.metrics.StrimziMetricsCollector;
2224
import io.vertx.core.AbstractVerticle;
2325
import io.vertx.core.Promise;
2426
import io.vertx.core.file.FileSystem;
@@ -43,11 +45,19 @@
4345
import org.apache.logging.log4j.LogManager;
4446
import org.apache.logging.log4j.Logger;
4547

48+
import javax.management.MalformedObjectNameException;
49+
import java.io.BufferedReader;
50+
import java.io.IOException;
51+
import java.io.InputStream;
52+
import java.io.InputStreamReader;
53+
import java.nio.charset.StandardCharsets;
54+
import java.nio.file.Files;
4655
import java.util.HashMap;
4756
import java.util.HashSet;
4857
import java.util.Iterator;
4958
import java.util.Map;
5059
import java.util.Set;
60+
import java.util.stream.Collectors;
5161

5262
import static io.netty.handler.codec.http.HttpHeaderNames.ACCEPT;
5363
import static io.netty.handler.codec.http.HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS;
@@ -59,7 +69,7 @@
5969
/**
6070
* Main bridge class listening for connections and handling HTTP requests.
6171
*/
62-
@SuppressWarnings({"checkstyle:MemberName"})
72+
@SuppressWarnings({"checkstyle:MemberName", "checkstyle:ClassDataAbstractionCoupling", "checkstyle:ClassFanOutComplexity"})
6373
public class HttpBridge extends AbstractVerticle {
6474
private static final Logger LOGGER = LogManager.getLogger(HttpBridge.class);
6575

@@ -76,17 +86,59 @@ public class HttpBridge extends AbstractVerticle {
7686

7787
private final Map<ConsumerInstanceId, Long> timestampMap = new HashMap<>();
7888

79-
private final MetricsCollector metricsCollector;
89+
private MetricsCollector metricsCollector = null;
8090

8191
/**
8292
* Constructor
8393
*
8494
* @param bridgeConfig bridge configuration
85-
* @param metricsCollector metricsCollector instance for scraping metrics from different registries
8695
*/
87-
public HttpBridge(BridgeConfig bridgeConfig, MetricsCollector metricsCollector) {
96+
public HttpBridge(BridgeConfig bridgeConfig) {
8897
this.bridgeConfig = bridgeConfig;
89-
this.metricsCollector = metricsCollector;
98+
if (bridgeConfig.getMetricsType() != null) {
99+
if (bridgeConfig.getMetricsType() == MetricsType.JMX_EXPORTER) {
100+
this.metricsCollector = createJmxMetricsCollector(bridgeConfig);
101+
} else if (bridgeConfig.getMetricsType() == MetricsType.STRIMZI_REPORTER) {
102+
this.metricsCollector = new StrimziMetricsCollector();
103+
}
104+
}
105+
if (metricsCollector != null) {
106+
LOGGER.info("Metrics of type '{}' enabled and exposed on /metrics endpoint", bridgeConfig.getMetricsType());
107+
}
108+
}
109+
110+
/**
111+
* Create a JmxMetricsCollector instance with the YAML configuration filters.
112+
* This is loaded from a custom config file if present or from the default configuration file.
113+
*
114+
* @return JmxCollectorRegistry instance
115+
*/
116+
private static JmxMetricsCollector createJmxMetricsCollector(BridgeConfig bridgeConfig) {
117+
try {
118+
if (bridgeConfig.getJmxExporterConfigPath() == null) {
119+
// load default configuration
120+
LOGGER.info("Using default JMX metrics configuration");
121+
InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
122+
if (is == null) {
123+
return null;
124+
}
125+
try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
126+
String yaml = reader
127+
.lines()
128+
.collect(Collectors.joining("\n"));
129+
return new JmxMetricsCollector(yaml);
130+
}
131+
} else if (Files.exists(bridgeConfig.getJmxExporterConfigPath())) {
132+
// load custom configuration file
133+
LOGGER.info("Loading custom JMX metrics configuration file from {}", bridgeConfig.getJmxExporterConfigPath());
134+
String yaml = Files.readString(bridgeConfig.getJmxExporterConfigPath(), StandardCharsets.UTF_8);
135+
return new JmxMetricsCollector(yaml);
136+
} else {
137+
throw new RuntimeException("Custom JMX metrics configuration file not found");
138+
}
139+
} catch (IOException | MalformedObjectNameException e) {
140+
throw new RuntimeException("Failed to initialize JMX metrics collector", e);
141+
}
90142
}
91143

92144
private void bindHttpServer(Promise<Void> startPromise) {
@@ -235,7 +287,6 @@ private CorsHandler getCorsHandler() {
235287

236288
@Override
237289
public void stop(Promise<Void> stopPromise) {
238-
239290
LOGGER.info("Stopping HTTP-Kafka bridge verticle ...");
240291

241292
this.isReady = false;
@@ -563,9 +614,7 @@ private void metrics(RoutingContext routingContext) {
563614
.setStatusCode(HttpResponseStatus.OK.code())
564615
.end(metricsCollector.scrape());
565616
} else {
566-
HttpBridgeError error = new HttpBridgeError(HttpResponseStatus.NOT_FOUND.code(), "The metrics endpoint is not enabled.");
567-
HttpUtils.sendResponse(routingContext, HttpResponseStatus.NOT_FOUND.code(),
568-
BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
617+
HttpUtils.sendResponse(routingContext, HttpResponseStatus.NOT_FOUND.code(), null, null);
569618
}
570619
}
571620

‎src/main/java/io/strimzi/kafka/bridge/metrics/MetricsCollector.java

+1
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ public abstract class MetricsCollector {
1919
MetricsCollector() {
2020
this.vertxRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();
2121
if (vertxRegistry != null) {
22+
// replace the default Prometheus naming convention
2223
this.vertxRegistry.config().namingConvention(new MetricsNamingConvention());
2324
}
2425
}

‎src/main/resources/openapi.json

+1-16
Original file line numberDiff line numberDiff line change
@@ -1480,22 +1480,7 @@
14801480
"description": "Metrics in Prometheus format retrieved successfully."
14811481
},
14821482
"404": {
1483-
"description": "The metrics endpoint is not enabled.",
1484-
"content": {
1485-
"application/json": {
1486-
"schema": {
1487-
"$ref": "#/components/schemas/Error"
1488-
},
1489-
"examples": {
1490-
"response": {
1491-
"value": {
1492-
"error_code": 404,
1493-
"message": "The metrics endpoint is not enabled."
1494-
}
1495-
}
1496-
}
1497-
}
1498-
}
1483+
"description": "The metrics endpoint is not enabled."
14991484
}
15001485
},
15011486
"operationId": "metrics",

‎src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ public void testStrimziReporterMetricsType() {
117117
assertThat(bridgeConfig.getMetricsType(), is(MetricsType.STRIMZI_REPORTER));
118118
assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter"));
119119
assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.listener.enable=false"));
120-
assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.allowlist=.*"));
120+
assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.allowlist=" + BridgeConfig.DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST));
121121
}
122122

123123
@Test

‎src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java

+1-3
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@
1010
import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
1111
import io.strimzi.kafka.bridge.config.KafkaProducerConfig;
1212
import io.strimzi.kafka.bridge.http.services.ConsumerService;
13-
import io.strimzi.kafka.bridge.metrics.JmxMetricsCollector;
1413
import io.strimzi.kafka.bridge.utils.Urls;
1514
import io.strimzi.test.container.StrimziKafkaContainer;
1615
import io.vertx.core.Vertx;
@@ -77,7 +76,6 @@ public class ConsumerGeneratedNameIT {
7776
private static HttpBridge httpBridge;
7877
private static WebClient client;
7978
private static BridgeConfig bridgeConfig;
80-
private static JmxMetricsCollector jmxMetricsCollector = null;
8179

8280
ConsumerService consumerService() {
8381
return ConsumerService.getInstance(client);
@@ -95,7 +93,7 @@ static void beforeAll(VertxTestContext context) {
9593
if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
9694

9795
bridgeConfig = BridgeConfig.fromMap(config);
98-
httpBridge = new HttpBridge(bridgeConfig, jmxMetricsCollector);
96+
httpBridge = new HttpBridge(bridgeConfig);
9997

10098
LOGGER.info("Deploying in-memory bridge");
10199
vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> context.completeNow()));

‎src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414
import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
1515
import io.strimzi.kafka.bridge.http.services.ConsumerService;
1616
import io.strimzi.kafka.bridge.http.services.ProducerService;
17-
import io.strimzi.kafka.bridge.metrics.StrimziMetricsCollector;
1817
import io.strimzi.kafka.bridge.utils.Urls;
1918
import io.vertx.core.Vertx;
2019
import io.vertx.core.json.JsonArray;
@@ -166,7 +165,7 @@ private void startBridge(VertxTestContext context, Map<String, Object> config) t
166165
CompletableFuture<Boolean> startBridge = new CompletableFuture<>();
167166
if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
168167
BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
169-
HttpBridge httpBridge = new HttpBridge(bridgeConfig, new StrimziMetricsCollector());
168+
HttpBridge httpBridge = new HttpBridge(bridgeConfig);
170169

171170
LOGGER.info("Deploying in-memory bridge");
172171
vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> startBridge.complete(true)));

0 commit comments

Comments
 (0)