Repository: kroxylicious/kroxylicious (java)

Issue:
Request Metrics are zero
**Describe the bug**
The request metrics (`kroxylicious_inbound_downstream_messages`, `kroxylicious_inbound_downstream_decoded_messages`, `kroxylicious_request_size_bytes`) are scraped as zero despite kroxy handling requests.

**To Reproduce**
Steps to reproduce the behavior:
1. Start up zookeeper and kafka with the default configuration from the apache distribution
2. Start kroxylicious `java -jar kroxylicious/target/kroxylicious-0.2.0-SNAPSHOT.jar -c kroxylicious/example-proxy-config.yml`
3. Produce some messages via kroxylicious `./bin/kafka-console-producer.sh --topic abc --bootstrap-server localhost:9192`
4. `curl localhost:9193/metrics`
5. observe all metrics are 0

**Expected behavior**
With the example config we should have metrics recorded for message count etc.

**Additional context**
The reason is we are pre-registering the meters with a different set of tags from the ones used later, which causes the later meter's data to silently not be scraped, see [comment here](https://github.com/kroxylicious/kroxylicious/pull/122#issuecomment-1413093573).   A limitation of the underlying prometheus java client.

Pull request description:
Fix metrics that were always scraped as zero
### Type of change

- Bugfix

### Description

The message count and size metrics were always zero. This was because when we pre-registered the metrics we used a different set of tags. Data recorded later with the real tags was effectively invisible. This is a limitation of micrometer combined
with the prometheus java client and is quite silent.

To fix I've pulled more work into `io.kroxylicious.proxy.internal.util.Metrics` so that the pre-register and real usages both use the same code with the same tags. We also configure a meter filter that will explode early if we attempt to register a Meter with a different set of tags to existing filters, this is to make the failure visible.

I also removed pre-registration of the summary, this is because the summary is broken down by apiKey and apiVersion, we would have to pre-register the ~250 combinations to pre-register all possibilities.

### Additional Context

Closes #266

Patch:
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
index c31ef4bf..cded1d0b 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java
@@ -27,7 +27,6 @@ import javax.net.ssl.KeyManagerFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.micrometer.core.instrument.Metrics;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -58,12 +57,9 @@ import io.kroxylicious.proxy.internal.KafkaProxyInitializer;
 import io.kroxylicious.proxy.internal.MeterRegistries;
 import io.kroxylicious.proxy.internal.admin.AdminHttpInitializer;
 import io.kroxylicious.proxy.internal.filter.UpstreamBrokerAddressCachingNetFilter;
+import io.kroxylicious.proxy.internal.util.Metrics;
 import io.kroxylicious.proxy.service.HostPort;
 
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
-
 public final class KafkaProxy implements AutoCloseable {
 
     private record EventGroupConfig(EventLoopGroup bossGroup, EventLoopGroup workerGroup, Class<? extends ServerChannel> clazz) {
@@ -192,9 +188,8 @@ public final class KafkaProxy implements AutoCloseable {
 
         // Pre-register counters/summaries to avoid creating them on first request and thus skewing the request latency
         // TODO add a virtual host tag to metrics
-        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES);
-        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES);
-        Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES);
+        Metrics.inboundDownstreamMessagesCounter();
+        Metrics.inboundDownstreamDecodedMessagesCounter();
         return this;
     }
 
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
index cb95c8ee..b6f29f04 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/MeterRegistries.java
@@ -7,9 +7,17 @@ package io.kroxylicious.proxy.internal;
 
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.micrometer.core.instrument.Meter;
 import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.micrometer.core.instrument.config.MeterFilter;
+import io.micrometer.core.instrument.config.MeterFilterReply;
 import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 
@@ -19,6 +27,8 @@ import io.kroxylicious.proxy.micrometer.MicrometerConfigurationHookContributorMa
 public class MeterRegistries {
     private final PrometheusMeterRegistry prometheusMeterRegistry;
 
+    private static final Logger logger = LoggerFactory.getLogger(MeterRegistries.class);
+
     public MeterRegistries(List<MicrometerDefinition> micrometerConfig) {
         configureMicrometer(micrometerConfig);
         this.prometheusMeterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
@@ -27,6 +37,7 @@ public class MeterRegistries {
 
     private void configureMicrometer(List<MicrometerDefinition> micrometerConfig) {
         CompositeMeterRegistry globalRegistry = Metrics.globalRegistry;
+        preventDifferentTagNameRegistration(globalRegistry);
         MicrometerConfigurationHookContributorManager manager = MicrometerConfigurationHookContributorManager.getInstance();
         micrometerConfig
                 .stream()
@@ -34,6 +45,32 @@ public class MeterRegistries {
                 .forEach(micrometerConfigurationHook -> micrometerConfigurationHook.configure(globalRegistry));
     }
 
+    /**
+     * By default, when we register multiple meters with the same name but different tag names,
+     * only the data for one of those meters can be scraped at the prometheus endpoint.
+     * This is a limitation of the prometheus client. So we add a filter that explodes when we
+     * attempt to register a metric with a different set of tags.
+     *
+     * @param registry registry
+     */
+    /* test */ static void preventDifferentTagNameRegistration(CompositeMeterRegistry registry) {
+        registry.config().meterFilter(new MeterFilter() {
+            @Override
+            public MeterFilterReply accept(Meter.Id id) {
+                boolean allTagsSame = registry.find(id.getName()).meters().stream().allMatch(meter -> tagNames(meter.getId()).equals(tagNames(id)));
+                if (!allTagsSame) {
+                    logger.error("Attempted to register a meter with id {} which is already registered but with a different set of tag names", id);
+                    throw new IllegalArgumentException("tags for id " + id + " differ from existing meters registered");
+                }
+                return MeterFilterReply.ACCEPT;
+            }
+        });
+    }
+
+    private static Object tagNames(Meter.Id id1) {
+        return id1.getTags().stream().map(Tag::getKey).collect(Collectors.toSet());
+    }
+
     /**
      * Offers up a prometheus registry if available. Currently, we always have a prometheus registry but in
      * future we may wish to use a different micrometer backend. Clients should use the global
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
index d677186f..281d0c14 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaRequestDecoder.java
@@ -5,8 +5,6 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
-import java.util.List;
-
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -14,8 +12,6 @@ import org.apache.kafka.common.protocol.Readable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.Tag;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
@@ -23,11 +19,7 @@ import io.kroxylicious.proxy.frame.DecodedRequestFrame;
 import io.kroxylicious.proxy.frame.Frame;
 import io.kroxylicious.proxy.frame.OpaqueRequestFrame;
 import io.kroxylicious.proxy.frame.RequestFrame;
-
-import static io.kroxylicious.proxy.internal.util.Metrics.FLOWING_UPSTREAM;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
+import io.kroxylicious.proxy.internal.util.Metrics;
 
 public class KafkaRequestDecoder extends KafkaMessageDecoder {
 
@@ -64,19 +56,15 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
 
         RequestHeaderData header = null;
         final ByteBufAccessorImpl accessor;
-        Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, List.of(Tag.of("flowing", "downstream"))).increment();
+        Metrics.inboundDownstreamMessagesCounter().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) {
-            Metrics.counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, List.of(Tag.of("flowing", "downstream"))).increment();
-            final List<Tag> LIST_OF_TAGS = List.of(
-                    Tag.of("ApiKey", apiKey.name()),
-                    Tag.of("ApiVersion", String.valueOf(apiVersion)),
-                    FLOWING_UPSTREAM);
-            Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES, LIST_OF_TAGS).record(length);
+            Metrics.inboundDownstreamDecodedMessagesCounter().increment();
+            Metrics.payloadSizeBytesUpstreamSummary(apiKey, apiVersion).record(length);
             if (log().isTraceEnabled()) { // avoid boxing
                 log().trace("{}: headerVersion {}", ctx, headerVersion);
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
index 3ae075da..c13972ff 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/codec/KafkaResponseDecoder.java
@@ -5,8 +5,6 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
-import java.util.List;
-
 import org.apache.kafka.common.message.ResponseHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -14,8 +12,6 @@ import org.apache.kafka.common.protocol.Readable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.Tag;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
@@ -25,9 +21,7 @@ import io.kroxylicious.proxy.frame.Frame;
 import io.kroxylicious.proxy.frame.OpaqueFrame;
 import io.kroxylicious.proxy.frame.OpaqueResponseFrame;
 import io.kroxylicious.proxy.internal.InternalResponseFrame;
-
-import static io.kroxylicious.proxy.internal.util.Metrics.FLOWING_DOWNSTREAM;
-import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_REQUEST_SIZE_BYTES;
+import io.kroxylicious.proxy.internal.util.Metrics;
 
 public class KafkaResponseDecoder extends KafkaMessageDecoder {
 
@@ -76,11 +70,7 @@ public class KafkaResponseDecoder extends KafkaMessageDecoder {
             ApiMessage body = BodyDecoder.decodeResponse(apiKey, apiVersion, accessor);
             log().trace("{}: Body: {}", ctx, body);
             KrpcFilter recipient = correlation.recipient();
-            final List<Tag> LIST_OF_TAGS = List.of(
-                    Tag.of("ApiKey", apiKey.name()),
-                    Tag.of("ApiVersion", String.valueOf(apiVersion)),
-                    FLOWING_DOWNSTREAM);
-            Metrics.summary(KROXYLICIOUS_REQUEST_SIZE_BYTES, LIST_OF_TAGS).record(length);
+            Metrics.payloadSizeBytesDownstreamSummary(apiKey, apiVersion).record(length);
             if (recipient == null) {
                 frame = new DecodedResponseFrame<>(apiVersion, correlationId, header, body);
             }
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java
index 1c8e40e5..87676a32 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/util/Metrics.java
@@ -6,22 +6,55 @@
 
 package io.kroxylicious.proxy.internal.util;
 
+import java.util.List;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.Tag;
 
+import static io.micrometer.core.instrument.Metrics.counter;
+import static io.micrometer.core.instrument.Metrics.summary;
+
 public class Metrics {
 
     // creating a constant for all Metrics in the one place so we can easily see what metrics there are
 
-    public static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES = "kroxylicious_inbound_downstream_messages";
+    private static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES = "kroxylicious_inbound_downstream_messages";
+
+    private static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES = "kroxylicious_inbound_downstream_decoded_messages";
+
+    private static final String KROXYLICIOUS_PAYLOAD_SIZE_BYTES = "kroxylicious_payload_size_bytes";
+
+    private static final String FLOWING_TAG = "flowing";
+
+    private static final Tag FLOWING_UPSTREAM = Tag.of(FLOWING_TAG, "upstream");
+
+    private static final Tag FLOWING_DOWNSTREAM = Tag.of(FLOWING_TAG, "downstream");
 
-    public static final String KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES = "kroxylicious_inbound_downstream_decoded_messages";
+    public static Counter inboundDownstreamMessagesCounter() {
+        return counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, List.of(FLOWING_DOWNSTREAM));
+    }
 
-    public static final String KROXYLICIOUS_REQUEST_SIZE_BYTES = "kroxylicious_request_size_bytes";
+    public static Counter inboundDownstreamDecodedMessagesCounter() {
+        return counter(KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, List.of(FLOWING_DOWNSTREAM));
+    }
 
-    public static final String FLOWING_TAG = "flowing";
+    public static DistributionSummary payloadSizeBytesUpstreamSummary(ApiKeys apiKey, short apiVersion) {
+        return payloadSizeBytesSummary(apiKey, apiVersion, FLOWING_UPSTREAM);
+    }
 
-    public static final Tag FLOWING_UPSTREAM = Tag.of(FLOWING_TAG, "upstream");
+    public static DistributionSummary payloadSizeBytesDownstreamSummary(ApiKeys apiKey, short apiVersion) {
+        return payloadSizeBytesSummary(apiKey, apiVersion, FLOWING_DOWNSTREAM);
+    }
 
-    public static final Tag FLOWING_DOWNSTREAM = Tag.of(FLOWING_TAG, "downstream");
+    private static DistributionSummary payloadSizeBytesSummary(ApiKeys apiKey, short apiVersion, Tag flowing) {
+        List<Tag> tags = List.of(
+                Tag.of("ApiKey", apiKey.name()),
+                Tag.of("ApiVersion", String.valueOf(apiVersion)),
+                flowing);
+        return summary(KROXYLICIOUS_PAYLOAD_SIZE_BYTES, tags);
+    }
 
 }


Test patch:
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/MeterRegistriesTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/MeterRegistriesTest.java
new file mode 100644
index 00000000..6b9467c9
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/MeterRegistriesTest.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.internal;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+
+class MeterRegistriesTest {
+
+    @Test
+    public void testPreventingRegistrationOfMetersWithSameNameButDifferentTags() {
+        CompositeMeterRegistry registry = new CompositeMeterRegistry();
+        MeterRegistries.preventDifferentTagNameRegistration(registry);
+        registry.counter("abc", List.of(Tag.of("a", "b")));
+        registry.counter("abc", List.of(Tag.of("a", "c")));
+        Assertions.assertThrows(IllegalArgumentException.class, () -> {
+            registry.counter("abc", List.of(Tag.of("c", "d")));
+        });
+    }
+
+}
\ No newline at end of file