Merge pull request kroxylicious#2094 from callaertanthony/vc-tag-metrics

add virtualcluster tag to kroxylicious_payload_size_bytes metrics

Author: SamBarker

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/KafkaProxyFrontendHandler.java

@@ -478,7 +478,8 @@ void inConnecting(
             pipeline.addFirst("frameLogger", new LoggingHandler("io.kroxylicious.proxy.internal.UpstreamFrameLogger"));
         }
         addFiltersToPipeline(filters, pipeline, inboundChannel);
-        pipeline.addFirst("responseDecoder", new KafkaResponseDecoder(correlationManager, virtualClusterModel.socketFrameMaxSizeBytes()));
+        pipeline.addFirst("responseDecoder", new KafkaResponseDecoder(correlationManager, virtualClusterModel.socketFrameMaxSizeBytes(),
+                virtualClusterModel.getClusterName()));
         pipeline.addFirst("requestEncoder", new KafkaRequestEncoder(correlationManager));
         if (logNetwork) {
             pipeline.addFirst("networkLogger", new LoggingHandler("io.kroxylicious.proxy.internal.UpstreamNetworkLogger"));

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java

@@ -37,15 +37,17 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
     private final DecodePredicate decodePredicate;
 
     private final ApiVersionsServiceImpl apiVersionsService;
-    private final Counter inbodundMessageCounter;
+    private final Counter inboundMessageCounter;
     private final Counter decodedMessagesCounter;
+    private final String clusterName;
 
     public KafkaRequestDecoder(DecodePredicate decodePredicate, int socketFrameMaxSize, ApiVersionsServiceImpl apiVersionsService, String clusterName) {
         super(socketFrameMaxSize);
         this.decodePredicate = decodePredicate;
         this.apiVersionsService = apiVersionsService;
+        this.clusterName = clusterName;
         List<Tag> tags = Metrics.tags(Metrics.FLOWING_TAG, Metrics.DOWNSTREAM, Metrics.VIRTUAL_CLUSTER_TAG, clusterName);
-        inbodundMessageCounter = Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, tags);
+        inboundMessageCounter = Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, tags);
         decodedMessagesCounter = Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, tags);
     }
 
@@ -73,15 +75,15 @@ protected Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, final
         LOGGER.debug("{}: {} downstream correlation id: {}", ctx, apiKey, correlationId);
         RequestHeaderData header = null;
         final ByteBufAccessorImpl accessor;
-        inbodundMessageCounter.increment();
+        inboundMessageCounter.increment();
         var decodeRequest = decodePredicate.shouldDecodeRequest(apiKey, apiVersion);
         LOGGER.debug("Decode {}/v{} request? {}, Predicate {} ", apiKey, apiVersion, decodeRequest, decodePredicate);
         boolean decodeResponse = decodePredicate.shouldDecodeResponse(apiKey, apiVersion);
         LOGGER.debug("Decode {}/v{} response? {}, Predicate {}", apiKey, apiVersion, decodeResponse, decodePredicate);
         short headerVersion = apiKey.requestHeaderVersion(apiVersion);
         if (decodeRequest) {
             decodedMessagesCounter.increment();
-            Metrics.payloadSizeBytesUpstreamSummary(apiKey, apiVersion).record(length);
+            Metrics.payloadSizeBytesUpstreamSummary(apiKey, apiVersion, clusterName).record(length);
             if (log().isTraceEnabled()) { // avoid boxing
                 log().trace("{}: headerVersion {}", ctx, headerVersion);
             }

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java

@@ -28,10 +28,12 @@ public class KafkaResponseDecoder extends KafkaMessageDecoder {
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaResponseDecoder.class);
 
     private final CorrelationManager correlationManager;
+    private final String clusterName;
 
-    public KafkaResponseDecoder(CorrelationManager correlationManager, int socketRequestMaxSizeBytes) {
+    public KafkaResponseDecoder(CorrelationManager correlationManager, int socketRequestMaxSizeBytes, String clusterName) {
         super(socketRequestMaxSizeBytes);
         this.correlationManager = correlationManager;
+        this.clusterName = clusterName;
     }
 
     @Override
@@ -70,7 +72,7 @@ else if (LOGGER.isDebugEnabled()) {
             ApiMessage body = BodyDecoder.decodeResponse(apiKey, apiVersion, accessor);
             log().trace("{}: Body: {}", ctx, body);
             Filter recipient = correlation.recipient();
-            Metrics.payloadSizeBytesDownstreamSummary(apiKey, apiVersion).record(length);
+            Metrics.payloadSizeBytesDownstreamSummary(apiKey, apiVersion, clusterName).record(length);
             if (recipient == null) {
                 frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
             }
@@ -93,4 +95,4 @@ private ResponseHeaderData readHeader(short headerVersion, Readable accessor) {
         return new ResponseHeaderData(accessor, headerVersion);
     }
 
-}
+}

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java

@@ -52,19 +52,20 @@ public static Counter taggedCounter(String counterName, List<Tag> tags) {
         return counter(counterName, tags);
     }
 
-    public static DistributionSummary payloadSizeBytesUpstreamSummary(ApiKeys apiKey, short apiVersion) {
-        return payloadSizeBytesSummary(apiKey, apiVersion, UPSTREAM);
+    public static DistributionSummary payloadSizeBytesUpstreamSummary(ApiKeys apiKey, short apiVersion, String virtualCluster) {
+        return payloadSizeBytesSummary(apiKey, apiVersion, UPSTREAM, virtualCluster);
     }
 
-    public static DistributionSummary payloadSizeBytesDownstreamSummary(ApiKeys apiKey, short apiVersion) {
-        return payloadSizeBytesSummary(apiKey, apiVersion, DOWNSTREAM);
+    public static DistributionSummary payloadSizeBytesDownstreamSummary(ApiKeys apiKey, short apiVersion, String virtualCluster) {
+        return payloadSizeBytesSummary(apiKey, apiVersion, DOWNSTREAM, virtualCluster);
     }
 
-    private static DistributionSummary payloadSizeBytesSummary(ApiKeys apiKey, short apiVersion, String flowing) {
+    private static DistributionSummary payloadSizeBytesSummary(ApiKeys apiKey, short apiVersion, String flowing, String virtualCluster) {
         List<Tag> tags = tags(
                 "ApiKey", apiKey.name(),
                 "ApiVersion", String.valueOf(apiVersion),
-                FLOWING_TAG, flowing);
+                FLOWING_TAG, flowing,
+                VIRTUAL_CLUSTER_TAG, virtualCluster);
         return summary(KROXYLICIOUS_PAYLOAD_SIZE_BYTES, tags);
     }
 

kroxylicious-runtime/src/test/java/io/kroxylicious/proxy/internal/codec/ResponseDecoderTest.java

@@ -74,7 +74,7 @@ void testApiVersionsExactlyOneFrame_opaque(short apiVersion) throws Exception {
 
     @NonNull
     private static KafkaResponseDecoder createResponseDecoder(CorrelationManager mgr, int socketFrameMaxSizeBytes) {
-        return new KafkaResponseDecoder(mgr, socketFrameMaxSizeBytes);
+        return new KafkaResponseDecoder(mgr, socketFrameMaxSizeBytes, "vc");
     }
 
     @Test
@@ -151,4 +151,4 @@ void supportsFallbackToApiResponseV0() {
                 .extracting("body")
                 .isEqualTo(new ApiVersionsResponseData().setErrorCode(Errors.UNSUPPORTED_VERSION.code()));
     }
-}
+}