Skip to content

Commit b684fa2

Browse files
niluka-instaniluka-insta
authored
INS-39698 (#4)
* INS-39698 * remove magic compression detection * A logger class to easily detect plugin messages in Kafka * Some scalable fixes * missing log file * Include a timeout for requests to OTel * update plugin version number
1 parent 96d1da3 commit b684fa2

8 files changed

Lines changed: 259 additions & 53 deletions

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121

2222
<groupId>com.kafka.instaclustr</groupId>
2323
<artifactId>apache-kafka-client-metrics-reporter-plugin</artifactId>
24-
<version>1.0.0</version>
24+
<version>1.0.2</version>
2525
<packaging>jar</packaging>
2626

2727
<properties>

‎src/main/java/com/instaclustr/kafka/KafkaClientMetricsReporter.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,16 @@
2020
import org.apache.kafka.common.metrics.MetricsReporter;
2121
import org.apache.kafka.server.telemetry.ClientTelemetry;
2222
import org.apache.kafka.server.telemetry.ClientTelemetryReceiver;
23-
import org.slf4j.Logger;
24-
import org.slf4j.LoggerFactory;
23+
import com.instaclustr.kafka.logging.KafkaClientMetricsLogger;
2524

2625
public class KafkaClientMetricsReporter implements MetricsReporter, ClientTelemetry {
2726

28-
private static final Logger logger = LoggerFactory.getLogger(KafkaClientMetricsReporter.class);
27+
private static final KafkaClientMetricsLogger logger = KafkaClientMetricsLogger.getLogger(KafkaClientMetricsReporter.class);
2928

3029

3130
@Override
3231
public void init(List<KafkaMetric> metrics) {
33-
logger.info("Initializing the client metric reporter: {}", metrics);
32+
logger.info("Initializing the client metric reporter");
3433
}
3534

3635
@Override

‎src/main/java/com/instaclustr/kafka/KafkaClientMetricsReporterConfig.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,7 @@
1414

1515
package com.instaclustr.kafka;
1616

17-
import org.slf4j.Logger;
18-
import org.slf4j.LoggerFactory;
17+
import com.instaclustr.kafka.logging.KafkaClientMetricsLogger;
1918
import org.yaml.snakeyaml.Yaml;
2019

2120
import java.io.FileInputStream;
@@ -24,7 +23,7 @@
2423
public class KafkaClientMetricsReporterConfig {
2524

2625
public Map<String, Object> configurations;
27-
private static final Logger logger = LoggerFactory.getLogger(KafkaClientMetricsReporterConfig.class);
26+
private static final KafkaClientMetricsLogger logger = KafkaClientMetricsLogger.getLogger(KafkaClientMetricsReporterConfig.class);
2827

2928
public KafkaClientMetricsReporterConfig() {
3029
this(System.getenv("KAFKA_CLIENT_METRICS_CONFIG_PATH"));

‎src/main/java/com/instaclustr/kafka/KafkaClientMetricsReporterReceiver.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,16 +16,15 @@
1616

1717
import com.instaclustr.kafka.exporters.MetricsExporter;
1818
import com.instaclustr.kafka.exporters.MetricsExporterFactory;
19+
import com.instaclustr.kafka.logging.KafkaClientMetricsLogger;
1920
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
2021
import org.apache.kafka.server.telemetry.ClientTelemetryPayload;
2122
import org.apache.kafka.server.telemetry.ClientTelemetryReceiver;
22-
import org.slf4j.Logger;
23-
import org.slf4j.LoggerFactory;
2423

2524
public class KafkaClientMetricsReporterReceiver implements ClientTelemetryReceiver {
2625

2726
private final MetricsExporter metricsExporter;
28-
private static final Logger logger = LoggerFactory.getLogger(KafkaClientMetricsReporterReceiver.class);
27+
private static final KafkaClientMetricsLogger logger = KafkaClientMetricsLogger.getLogger(KafkaClientMetricsReporterReceiver.class);
2928

3029
public KafkaClientMetricsReporterReceiver() {
3130
logger.info("Initializing the Kafka Client Metrics Reporter Receiver");

‎src/main/java/com/instaclustr/kafka/exporters/HttpMetricsExporter.java‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,26 +15,30 @@
1515
package com.instaclustr.kafka.exporters;
1616

1717
import com.instaclustr.kafka.helpers.MetricsMetaDataProcessor;
18+
import com.instaclustr.kafka.logging.KafkaClientMetricsLogger;
1819
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
1920
import org.apache.kafka.server.telemetry.ClientTelemetryPayload;
20-
import org.slf4j.Logger;
21-
import org.slf4j.LoggerFactory;
2221
import java.net.URI;
2322
import java.net.http.HttpClient;
2423
import java.net.http.HttpRequest;
2524
import java.net.http.HttpResponse;
2625
import java.time.Duration;
26+
import java.util.Collections;
2727
import java.util.Map;
2828

2929
public class HttpMetricsExporter implements MetricsExporter {
3030
private final String endpoint;
3131
private final HttpClient httpClient;
3232
final Map<String, Object> metadata;
33-
private final Logger logger = LoggerFactory.getLogger(HttpMetricsExporter.class);
33+
private final MetricsMetaDataProcessor metricsMetaDataProcessor;
34+
private final KafkaClientMetricsLogger logger = KafkaClientMetricsLogger.getLogger(HttpMetricsExporter.class);
35+
private final Duration timeoutMillisDuration;
3436

3537
public HttpMetricsExporter(final String endpoint, final int timeoutMillis, final Map<String, Object> metadata) {
3638
this.endpoint = endpoint;
37-
this.metadata = metadata;
39+
this.metadata = metadata != null ? metadata : Collections.emptyMap();
40+
this.metricsMetaDataProcessor = new MetricsMetaDataProcessor(this.metadata);
41+
this.timeoutMillisDuration = Duration.ofMillis(timeoutMillis);
3842
this.httpClient = HttpClient.newBuilder()
3943
.version(HttpClient.Version.HTTP_2)
4044
.connectTimeout(Duration.ofMillis(timeoutMillis))
@@ -45,8 +49,7 @@ public HttpMetricsExporter(final String endpoint, final int timeoutMillis, final
4549
@Override
4650
public void export(final AuthorizableRequestContext requestContext, final ClientTelemetryPayload payload) {
4751
try {
48-
final MetricsMetaDataProcessor processor = new MetricsMetaDataProcessor(metadata);
49-
final byte[] finalBytes = processor.processMetricsData(requestContext, payload.data());
52+
final byte[] finalBytes = metricsMetaDataProcessor.processMetricsData(requestContext, payload.data());
5053

5154
HttpRequest request = buildRequest(finalBytes);
5255
sendAsync(request);
@@ -55,20 +58,19 @@ public void export(final AuthorizableRequestContext requestContext, final Client
5558
}
5659
}
5760

58-
5961
private HttpRequest buildRequest(final byte[] payload) {
6062
return HttpRequest.newBuilder()
6163
.uri(URI.create(endpoint))
6264
.header("Content-Type", "application/x-protobuf")
6365
.POST(HttpRequest.BodyPublishers.ofByteArray(payload))
66+
.timeout(this.timeoutMillisDuration)
6467
.build();
6568
}
6669

6770
private void sendAsync(final HttpRequest request) {
6871
httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
6972
.thenAccept(response -> {
7073
logger.debug("OTLP metrics endpoint status: {}", response.statusCode());
71-
logger.debug("Response body: {}", response.body());
7274
})
7375
.exceptionally(ex -> {
7476
logger.error("Error invoking the OTLP metrics endpoint", ex);

‎src/main/java/com/instaclustr/kafka/exporters/MetricsExporterFactory.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
package com.instaclustr.kafka.exporters;
1616

17+
import java.util.Collections;
1718
import java.util.Map;
1819

1920
public class MetricsExporterFactory {
@@ -33,7 +34,7 @@ public static MetricsExporter create(final Map<String, Object> configurations) {
3334
if (metadataObj != null && !(metadataObj instanceof Map)) {
3435
throw new IllegalArgumentException("Metadata configuration is not a valid map");
3536
}
36-
final Map<String, Object> metadata = (Map<String, Object>) metadataObj;
37+
final Map<String, Object> metadata = metadataObj != null ? (Map<String, Object>) metadataObj : Collections.emptyMap();
3738

3839
switch (((String) exporterMap.get("mode")).toUpperCase()) {
3940
case "HTTP":

‎src/main/java/com/instaclustr/kafka/helpers/MetricsMetaDataProcessor.java‎

Lines changed: 121 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -14,52 +14,123 @@
1414

1515
package com.instaclustr.kafka.helpers;
1616

17+
import com.instaclustr.kafka.logging.KafkaClientMetricsLogger;
18+
import org.apache.kafka.common.compress.Compression;
19+
import org.apache.kafka.common.record.CompressionType;
1720
import org.apache.kafka.common.requests.RequestContext;
21+
import org.apache.kafka.common.utils.BufferSupplier;
1822
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
23+
import org.apache.kafka.shaded.com.google.protobuf.InvalidProtocolBufferException;
1924
import org.apache.kafka.shaded.io.opentelemetry.proto.common.v1.AnyValue;
2025
import org.apache.kafka.shaded.io.opentelemetry.proto.common.v1.KeyValue;
2126
import org.apache.kafka.shaded.io.opentelemetry.proto.metrics.v1.MetricsData;
2227
import org.apache.kafka.shaded.io.opentelemetry.proto.resource.v1.Resource;
23-
import org.slf4j.Logger;
24-
import org.slf4j.LoggerFactory;
2528

29+
import java.io.ByteArrayOutputStream;
30+
import java.io.InputStream;
2631
import java.nio.ByteBuffer;
27-
import java.util.HashMap;
32+
import java.util.ArrayList;
33+
import java.util.Arrays;
34+
import java.util.Collections;
35+
import java.util.List;
2836
import java.util.Map;
2937

3038
public class MetricsMetaDataProcessor {
31-
private static final Logger logger = LoggerFactory.getLogger(MetricsMetaDataProcessor.class);
32-
private final Map<String, Object> metadata;
39+
private static final KafkaClientMetricsLogger logger = KafkaClientMetricsLogger.getLogger(MetricsMetaDataProcessor.class);
40+
private final List<KeyValue> staticMetadataAttributes;
41+
3342

3443
public MetricsMetaDataProcessor(final Map<String, Object> metadata) {
35-
this.metadata = metadata;
44+
this.staticMetadataAttributes = toStaticAttributes(metadata);
3645
}
3746

38-
public byte[] processMetricsData(final AuthorizableRequestContext requestContext, final ByteBuffer buffer) {
39-
final byte[] rawBytes = bufferToBytes(buffer);
47+
public static List<KeyValue> toStaticAttributes(final Map<String, Object> metadata) {
48+
if (metadata == null || metadata.isEmpty()) {
49+
return Collections.emptyList();
50+
}
51+
52+
final List<KeyValue> attrs = new ArrayList<>(metadata.size());
53+
metadata.forEach((key, value) -> attrs.add(toKeyValue(key, value)));
54+
return attrs;
55+
}
4056

57+
public byte[] processMetricsData(final AuthorizableRequestContext requestContext, final ByteBuffer buffer) {
58+
final int bufferRemaining = buffer.remaining();
59+
byte[] rawBytes = null;
4160
try {
42-
MetricsData metricsData = MetricsData.parseFrom(rawBytes);
61+
rawBytes = bufferToBytes(buffer);
62+
final MetricsData metricsData = decodeMetricsData(rawBytes);
63+
if (metricsData == null) {
64+
return rawBytes;
65+
}
4366
return enrichMetricsData(requestContext, metricsData);
4467
} catch (Exception e) {
45-
logger.error("Error processing metrics data: {}", e.getMessage(), e);
68+
logger.error("Error processing metrics data (bufferRemaining={}, rawBytesLength={}): {}",
69+
bufferRemaining, rawBytes != null ? rawBytes.length : "null", e.getMessage(), e);
4670
}
4771

48-
return rawBytes;
72+
return rawBytes != null ? rawBytes : bufferToBytes(buffer);
73+
}
74+
75+
private MetricsData decodeMetricsData(final byte[] rawBytes) {
76+
if (rawBytes == null || rawBytes.length == 0) {
77+
return null;
78+
}
79+
80+
// Fast path: payload is uncompressed OTLP protobuf.
81+
try {
82+
return MetricsData.parseFrom(rawBytes);
83+
} catch (final InvalidProtocolBufferException ignored) {
84+
}
85+
86+
// Slow path: try to decompress with common Kafka compression types.
87+
for (final CompressionType compressionType : Arrays.asList(
88+
CompressionType.ZSTD,
89+
CompressionType.GZIP,
90+
CompressionType.LZ4,
91+
CompressionType.SNAPPY
92+
)) {
93+
try {
94+
final byte[] decompressed = decompress(rawBytes, compressionType);
95+
final MetricsData parsed = MetricsData.parseFrom(decompressed);
96+
logger.debug("Decoded client telemetry payload using compression={}", compressionType.name);
97+
return parsed;
98+
} catch (final Exception ignored) {
99+
}
100+
}
101+
102+
logger.error("Unable to decode client telemetry payload as raw protobuf or with common compression codecs");
103+
return null;
104+
}
105+
106+
private byte[] decompress(final byte[] data, final CompressionType compressionType) throws Exception {
107+
if (compressionType == CompressionType.NONE) {
108+
return data;
109+
}
110+
111+
final ByteBuffer buffer = ByteBuffer.wrap(data);
112+
final Compression compression = Compression.of(compressionType).build();
113+
114+
try (InputStream inputStream = compression.wrapForInput(buffer, (byte) 0, BufferSupplier.NO_CACHING);
115+
ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
116+
117+
inputStream.transferTo(outputStream);
118+
return outputStream.toByteArray();
119+
}
49120
}
50121

51122
private byte[] bufferToBytes(ByteBuffer buffer) {
52-
if (buffer.hasArray()) {
123+
if (buffer.hasArray() && buffer.arrayOffset() == 0 && buffer.remaining() == buffer.array().length) {
53124
return buffer.array();
54125
} else {
55126
byte[] bytes = new byte[buffer.remaining()];
56-
buffer.get(bytes);
127+
buffer.duplicate().get(bytes);
57128
return bytes;
58129
}
59130
}
60131

61132
private boolean shouldEnrichStaticMetaData(final MetricsData metricsData) {
62-
return !this.metadata.isEmpty() && metricsData.getResourceMetricsCount() > 0;
133+
return !this.staticMetadataAttributes.isEmpty() && metricsData.getResourceMetricsCount() > 0;
63134
}
64135

65136
private byte[] enrichMetricsData(final AuthorizableRequestContext context, MetricsData metricsData) {
@@ -78,44 +149,61 @@ private void enrichStaticMetadata(MetricsData.Builder dataBuilder) {
78149
Resource.Builder resourceBuilder =
79150
dataBuilder.getResourceMetricsBuilder(i).getResourceBuilder();
80151

81-
this.metadata.forEach((key, value) ->
82-
resourceBuilder.addAttributes(toKeyValue(key, value))
83-
);
152+
resourceBuilder.addAllAttributes(staticMetadataAttributes);
84153
}
85154
}
86155

87156
private void enrichDynamicMetadata(final AuthorizableRequestContext context, MetricsData.Builder dataBuilder) {
88157

89158
final RequestContext requestContext = (RequestContext) context;
90-
91-
Map<String, String> dynamicMetadata = new HashMap<>();
92-
if (requestContext.clientId() != null) {
93-
dynamicMetadata.put("clientId", requestContext.clientId());
94-
}
95-
if (requestContext.clientInformation != null) {
96-
dynamicMetadata.put("clientSoftwareName", requestContext.clientInformation.softwareName());
97-
dynamicMetadata.put("clientSoftwareVersion", requestContext.clientInformation.softwareVersion());
159+
final List<KeyValue> dynamicAttributes = buildDynamicAttributes(requestContext);
160+
if (dynamicAttributes.isEmpty()) {
161+
return;
98162
}
99163

100164
for (int i = 0; i < dataBuilder.getResourceMetricsCount(); i++) {
101165
Resource.Builder resourceBuilder = dataBuilder.getResourceMetricsBuilder(i).getResourceBuilder();
102-
dynamicMetadata.forEach((key, value) -> resourceBuilder.addAttributes(
103-
KeyValue.newBuilder()
104-
.setKey(key)
105-
.setValue(AnyValue.newBuilder().setStringValue(value))
106-
.build()
107-
));
166+
resourceBuilder.addAllAttributes(dynamicAttributes);
167+
}
168+
}
169+
170+
private static List<KeyValue> buildDynamicAttributes(final RequestContext requestContext) {
171+
final String clientId = requestContext.clientId();
172+
final String softwareName = requestContext.clientInformation != null ? requestContext.clientInformation.softwareName() : null;
173+
final String softwareVersion = requestContext.clientInformation != null ? requestContext.clientInformation.softwareVersion() : null;
174+
175+
if (clientId == null && softwareName == null && softwareVersion == null) {
176+
return Collections.emptyList();
108177
}
178+
179+
final List<KeyValue> attrs = new ArrayList<>(3);
180+
if (clientId != null) {
181+
attrs.add(toStringKeyValue("clientId", clientId));
182+
}
183+
if (softwareName != null) {
184+
attrs.add(toStringKeyValue("clientSoftwareName", softwareName));
185+
}
186+
if (softwareVersion != null) {
187+
attrs.add(toStringKeyValue("clientSoftwareVersion", softwareVersion));
188+
}
189+
return attrs;
190+
}
191+
192+
private static KeyValue toStringKeyValue(final String key, final String value) {
193+
return KeyValue.newBuilder()
194+
.setKey(key)
195+
.setValue(AnyValue.newBuilder().setStringValue(value).build())
196+
.build();
109197
}
110198

111-
private KeyValue toKeyValue(final String key, final Object value) {
199+
private static KeyValue toKeyValue(final String key, final Object value) {
112200
return KeyValue.newBuilder()
113201
.setKey(key)
114202
.setValue(toAnyValue(value))
115203
.build();
116204
}
117205

118-
private AnyValue toAnyValue(final Object value) {
206+
private static AnyValue toAnyValue(final Object value) {
119207
AnyValue.Builder b = AnyValue.newBuilder();
120208
if (value instanceof String) {
121209
b.setStringValue((String) value);

0 commit comments

Comments
 (0)