Merge pull request kroxylicious#2017 from SamBarker/network-metrics

Connection count metrics:

This PR adds
`kroxylicious_downstream_connections` - a count of the successful connections from clients
`kroxylicious_downstream_errors` - a count of the exceptions triggered on the downstream/client side of the netty pipeline.

`kroxylicious_upstream_connections` - a count of successful connections established by the proxy to an upstream broker.
`kroxylicious_upstream_attempts` - a count of the number of times the proxy attempts to connect to an upstream broker. 
`kroxylicious_upstream_failures` - a count of the number of times the proxy fails in an attempt to connect to an upstream broker. 
`kroxylicious_upstream_errors` - a count of the exceptions triggered on the upstream/server side of the netty pipeline.

Author: SamBarker

kroxylicious-runtime/src/main/java/io/kroxylicious/proxy/KafkaProxy.java

@@ -19,6 +19,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.micrometer.core.instrument.Tag;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelOption;
@@ -153,9 +154,12 @@ public KafkaProxy startup() {
                     .join();
 
             // 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.inboundDownstreamMessagesCounter();
-            Metrics.inboundDownstreamDecodedMessagesCounter();
+            virtualClusterModels.forEach(virtualClusterModel -> {
+                List<Tag> tags = Metrics.tags(Metrics.FLOWING_TAG, Metrics.DOWNSTREAM, Metrics.VIRTUAL_CLUSTER_TAG, virtualClusterModel.getClusterName());
+                Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, tags);
+                Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, tags);
+            });
+
             return this;
         }
         catch (RuntimeException e) {
@@ -306,4 +310,4 @@ public void close() throws Exception {
         }
     }
 
-}
+}

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

@@ -105,8 +105,9 @@ public class KafkaProxyFrontendHandler
     KafkaProxyFrontendHandler(
                               @NonNull NetFilter netFilter,
                               @NonNull SaslDecodePredicate dp,
-                              EndpointGateway endpointGateway) {
-        this(netFilter, dp, endpointGateway, new ProxyChannelStateMachine());
+                              EndpointGateway endpointGateway,
+                              @NonNull String clusterName) {
+        this(netFilter, dp, endpointGateway, new ProxyChannelStateMachine(clusterName));
     }
 
     @VisibleForTesting
@@ -660,4 +661,4 @@ private ResponseFrame errorResponse(
         }
         return errorResponse;
     }
-}
+}

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

@@ -192,7 +192,7 @@ void addHandlers(SocketChannel ch, EndpointBinding binding) {
         var dp = new SaslDecodePredicate(!authnHandlers.isEmpty());
         // The decoder, this only cares about the filters
         // because it needs to know whether to decode requests
-        KafkaRequestDecoder decoder = new KafkaRequestDecoder(dp, virtualCluster.socketFrameMaxSizeBytes(), apiVersionsService);
+        KafkaRequestDecoder decoder = new KafkaRequestDecoder(dp, virtualCluster.socketFrameMaxSizeBytes(), apiVersionsService, virtualCluster.getClusterName());
         pipeline.addLast("requestDecoder", decoder);
         pipeline.addLast("responseEncoder", new KafkaResponseEncoder());
         pipeline.addLast("responseOrderer", new ResponseOrderer());
@@ -214,7 +214,7 @@ void addHandlers(SocketChannel ch, EndpointBinding binding) {
                 endpointReconciler,
                 new ApiVersionsIntersectFilter(apiVersionsService),
                 new ApiVersionsDowngradeFilter(apiVersionsService));
-        var frontendHandler = new KafkaProxyFrontendHandler(netFilter, dp, binding.endpointGateway());
+        var frontendHandler = new KafkaProxyFrontendHandler(netFilter, dp, binding.endpointGateway(), virtualCluster.getClusterName());
 
         pipeline.addLast("netHandler", frontendHandler);
         addLoggingErrorHandler(pipeline);
@@ -298,4 +298,4 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                             cause.getClass().getSimpleName(), cause.getMessage());
         }
     }
-}
+}

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

@@ -27,6 +27,8 @@
 import io.kroxylicious.proxy.micrometer.MicrometerConfigurationHookService;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
+
 public class MeterRegistries implements AutoCloseable {
     private final PrometheusMeterRegistry prometheusMeterRegistry;
 
@@ -44,6 +46,7 @@ public MeterRegistries(PluginFactoryRegistry pfr, List<MicrometerDefinition> mic
     private List<MicrometerConfigurationHook> registerHooks(List<MicrometerDefinition> micrometerConfig) {
         CompositeMeterRegistry globalRegistry = Metrics.globalRegistry;
         preventDifferentTagNameRegistration(globalRegistry);
+        @SuppressWarnings("unchecked")
         var configurationHooks = micrometerConfig.stream()
                 .map(f -> pfr.pluginFactory(MicrometerConfigurationHookService.class).pluginInstance(f.type()).build(f.config()))
                 .toList();
@@ -62,8 +65,9 @@ private List<MicrometerConfigurationHook> registerHooks(List<MicrometerDefinitio
     @VisibleForTesting
     static void preventDifferentTagNameRegistration(CompositeMeterRegistry registry) {
         registry.config().meterFilter(new MeterFilter() {
+            @NonNull
             @Override
-            public MeterFilterReply accept(Meter.Id id) {
+            public MeterFilterReply accept(@NonNull 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);
@@ -96,4 +100,4 @@ public void close() {
         copy.forEach(Metrics.globalRegistry::remove);
         Metrics.removeRegistry(prometheusMeterRegistry);
     }
-}
+}

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

@@ -16,6 +16,8 @@
 import org.apache.kafka.common.protocol.Errors;
 import org.slf4j.Logger;
 
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Tag;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.DecoderException;
 import io.netty.handler.codec.haproxy.HAProxyMessage;
@@ -27,6 +29,7 @@
 import io.kroxylicious.proxy.internal.ProxyChannelState.Closed;
 import io.kroxylicious.proxy.internal.ProxyChannelState.Forwarding;
 import io.kroxylicious.proxy.internal.codec.FrameOversizedException;
+import io.kroxylicious.proxy.internal.util.Metrics;
 import io.kroxylicious.proxy.model.VirtualClusterModel;
 import io.kroxylicious.proxy.service.HostPort;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
@@ -35,6 +38,13 @@
 import edu.umd.cs.findbugs.annotations.Nullable;
 
 import static io.kroxylicious.proxy.internal.ProxyChannelState.Startup.STARTING_STATE;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_DOWNSTREAM_CONNECTIONS;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_DOWNSTREAM_ERRORS;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_UPSTREAM_CONNECTIONS;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_UPSTREAM_CONNECTION_ATTEMPTS;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_UPSTREAM_CONNECTION_FAILURES;
+import static io.kroxylicious.proxy.internal.util.Metrics.KROXYLICIOUS_UPSTREAM_ERRORS;
+import static io.kroxylicious.proxy.internal.util.Metrics.taggedCounter;
 import static org.slf4j.LoggerFactory.getLogger;
 
 /**
@@ -95,6 +105,22 @@
 public class ProxyChannelStateMachine {
     private static final String DUPLICATE_INITIATE_CONNECT_ERROR = "NetFilter called NetFilterContext.initiateConnect() more than once";
     private static final Logger LOGGER = getLogger(ProxyChannelStateMachine.class);
+    private final Counter downstreamConnectionsCounter;
+    private final Counter upstreamConnectionsCounter;
+    private final Counter downstreamErrorCounter;
+    private final Counter upstreamErrorCounter;
+    private final Counter connectionAttemptsCounter;
+    private final Counter upstreamConnectionFailureCounter;
+
+    public ProxyChannelStateMachine(String clusterName) {
+        List<Tag> tags = Metrics.tags(Metrics.VIRTUAL_CLUSTER_TAG, clusterName);
+        downstreamConnectionsCounter = taggedCounter(KROXYLICIOUS_DOWNSTREAM_CONNECTIONS, tags);
+        downstreamErrorCounter = taggedCounter(KROXYLICIOUS_DOWNSTREAM_ERRORS, tags);
+        upstreamConnectionsCounter = taggedCounter(KROXYLICIOUS_UPSTREAM_CONNECTIONS, tags);
+        connectionAttemptsCounter = taggedCounter(KROXYLICIOUS_UPSTREAM_CONNECTION_ATTEMPTS, tags);
+        upstreamErrorCounter = taggedCounter(KROXYLICIOUS_UPSTREAM_ERRORS, tags);
+        upstreamConnectionFailureCounter = taggedCounter(KROXYLICIOUS_UPSTREAM_CONNECTION_FAILURES, tags);
+    }
 
     /**
      * The current state. This can be changed via a call to one of the {@code on*()} methods.
@@ -370,6 +396,10 @@ void onServerException(Throwable cause) {
                 .setCause(LOGGER.isDebugEnabled() ? cause : null)
                 .addArgument(cause != null ? cause.getMessage() : "")
                 .log("Exception from the server channel: {}. Increase log level to DEBUG for stacktrace");
+        if (state instanceof ProxyChannelState.Connecting) {
+            upstreamConnectionFailureCounter.increment();
+        }
+        upstreamErrorCounter.increment();
         toClosed(cause);
     }
 
@@ -395,6 +425,7 @@ void onClientException(Throwable cause, boolean tlsEnabled) {
                     .log("Exception from the client channel: {}. Increase log level to DEBUG for stacktrace");
             errorCodeEx = Errors.UNKNOWN_SERVER_ERROR.exception();
         }
+        downstreamErrorCounter.increment();
         toClosed(errorCodeEx);
     }
 
@@ -403,6 +434,7 @@ private void toClientActive(
                                 @NonNull KafkaProxyFrontendHandler frontendHandler) {
         setState(clientActive);
         frontendHandler.inClientActive();
+        downstreamConnectionsCounter.increment();
     }
 
     private void toConnecting(
@@ -412,11 +444,13 @@ private void toConnecting(
         setState(connecting);
         backendHandler = new KafkaProxyBackendHandler(this, virtualClusterModel);
         frontendHandler.inConnecting(connecting.remote(), filters, backendHandler);
+        connectionAttemptsCounter.increment();
     }
 
     private void toForwarding(Forwarding forwarding) {
         setState(forwarding);
         Objects.requireNonNull(frontendHandler).inForwarding();
+        upstreamConnectionsCounter.increment();
     }
 
     /**
@@ -510,6 +544,7 @@ private void toSelectingServer(ProxyChannelState.SelectingServer selectingServer
         Objects.requireNonNull(frontendHandler).inSelectingServer();
     }
 
+    @SuppressWarnings("ConstantValue")
     private void toClosed(@Nullable Throwable errorCodeEx) {
         if (state instanceof Closed) {
             return;
@@ -521,7 +556,9 @@ private void toClosed(@Nullable Throwable errorCodeEx) {
         }
 
         // Close the client connection with any error code
-        Objects.requireNonNull(frontendHandler).inClosed(errorCodeEx);
+        if (frontendHandler != null) { // Can be null if the error happens before clientActive (unlikely but possible)
+            frontendHandler.inClosed(errorCodeEx);
+        }
     }
 
     private void setState(@NonNull ProxyChannelState state) {
@@ -533,4 +570,4 @@ private static boolean isMessageApiVersionsRequest(Object msg) {
         return msg instanceof DecodedRequestFrame
                 && ((DecodedRequestFrame<?>) msg).apiKey() == ApiKeys.API_VERSIONS;
     }
-}
+}

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

@@ -5,6 +5,8 @@
  */
 package io.kroxylicious.proxy.internal.codec;
 
+import java.util.List;
+
 import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.message.RequestHeaderData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -13,6 +15,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Tag;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 
@@ -33,11 +37,16 @@ public class KafkaRequestDecoder extends KafkaMessageDecoder {
     private final DecodePredicate decodePredicate;
 
     private final ApiVersionsServiceImpl apiVersionsService;
+    private final Counter inbodundMessageCounter;
+    private final Counter decodedMessagesCounter;
 
-    public KafkaRequestDecoder(DecodePredicate decodePredicate, int socketFrameMaxSize, ApiVersionsServiceImpl apiVersionsService) {
+    public KafkaRequestDecoder(DecodePredicate decodePredicate, int socketFrameMaxSize, ApiVersionsServiceImpl apiVersionsService, String clusterName) {
         super(socketFrameMaxSize);
         this.decodePredicate = decodePredicate;
         this.apiVersionsService = apiVersionsService;
+        List<Tag> tags = Metrics.tags(Metrics.FLOWING_TAG, Metrics.DOWNSTREAM, Metrics.VIRTUAL_CLUSTER_TAG, clusterName);
+        inbodundMessageCounter = Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_MESSAGES, tags);
+        decodedMessagesCounter = Metrics.taggedCounter(Metrics.KROXYLICIOUS_INBOUND_DOWNSTREAM_DECODED_MESSAGES, tags);
     }
 
     @Override
@@ -64,14 +73,14 @@ protected Frame decodeHeaderAndBody(ChannelHandlerContext ctx, ByteBuf in, final
         LOGGER.debug("{}: {} downstream correlation id: {}", ctx, apiKey, correlationId);
         RequestHeaderData header = null;
         final ByteBufAccessorImpl accessor;
-        Metrics.inboundDownstreamMessagesCounter().increment();
+        inbodundMessageCounter.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.inboundDownstreamDecodedMessagesCounter().increment();
+            decodedMessagesCounter.increment();
             Metrics.payloadSizeBytesUpstreamSummary(apiKey, apiVersion).record(length);
             if (log().isTraceEnabled()) { // avoid boxing
                 log().trace("{}: headerVersion {}", ctx, headerVersion);
@@ -198,4 +207,4 @@ private RequestHeaderData readHeader(short headerVersion, Readable accessor) {
         return new RequestHeaderData(accessor, headerVersion);
     }
 
-}
+}