resolve #1738 | Add jetty http client request processing time metrics

hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/ConsumerSenderMetrics.java

@@ -1,6 +1,7 @@
 package pl.allegro.tech.hermes.common.metric;
 
 import io.micrometer.core.instrument.MeterRegistry;
+import pl.allegro.tech.hermes.metrics.HermesTimer;
 
 import java.util.function.ToDoubleFunction;
 
@@ -66,4 +67,46 @@ public <T> void registerHttp2SerialClientConnectionsGauge(T obj, ToDoubleFunctio
     public <T> void registerHttp2SerialClientPendingConnectionsGauge(T obj, ToDoubleFunction<T> f) {
         gaugeRegistrar.registerGauge(CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_PENDING_CONNECTIONS, obj, f);
     }
+
+    public HermesTimer http1SerialClientRequestQueueWaitingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME)
+        );
+    }
+
+    public HermesTimer http1BatchClientRequestQueueWaitingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_QUEUE_WAITING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_QUEUE_WAITING_TIME)
+        );
+    }
+
+    public HermesTimer http2SerialClientRequestQueueWaitingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME)
+        );
+    }
+
+    public HermesTimer http1SerialClientRequestProcessingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_PROCESSING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_PROCESSING_TIME)
+        );
+    }
+
+    public HermesTimer http1BatchClientRequestProcessingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_PROCESSING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_PROCESSING_TIME)
+        );
+    }
+
+    public HermesTimer http2SerialClientRequestProcessingTimer() {
+        return HermesTimer.from(
+                meterRegistry.timer(Timers.CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_PROCESSING_TIME),
+                hermesMetrics.timer(Timers.CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_PROCESSING_TIME)
+        );
+    }
 }

hermes-common/src/main/java/pl/allegro/tech/hermes/common/metric/Timers.java

@@ -29,4 +29,17 @@ public class Timers {
     public static final String CONSUMER_IDLE_TIME = "idle-time." + GROUP + "." + TOPIC + "." + SUBSCRIPTION;
 
     public static final String OAUTH_PROVIDER_TOKEN_REQUEST_LATENCY = "oauth.provider." + OAUTH_PROVIDER_NAME + ".token-request-latency";
+
+    public static final String CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME =
+            "http-clients.serial.http1.request-queue-waiting-time";
+    public static final String CONSUMER_SENDER_HTTP_1_SERIAL_CLIENT_REQUEST_PROCESSING_TIME =
+            "http-clients.serial.http1.request-processing-time";
+    public static final String CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_QUEUE_WAITING_TIME =
+            "http-clients.serial.http2.request-queue-waiting-time";
+    public static final String CONSUMER_SENDER_HTTP_2_SERIAL_CLIENT_REQUEST_PROCESSING_TIME =
+            "http-clients.serial.http2.request-processing-time";
+    public static final String CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_QUEUE_WAITING_TIME =
+            "http-clients.batch.http1.request-queue-waiting-time";
+    public static final String CONSUMER_SENDER_HTTP_1_BATCH_CLIENT_REQUEST_PROCESSING_TIME =
+            "http-clients.batch.http1.request-processing-time";
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/config/ConsumerSenderConfiguration.java

@@ -42,6 +42,7 @@
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.HttpHeadersProvidersFactory;
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.HttpMessageBatchSenderFactory;
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.HttpRequestFactoryProvider;
+import pl.allegro.tech.hermes.consumers.consumer.sender.http.JettyHttpClientMetrics;
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.JettyHttpMessageSenderProvider;
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.SendingResultHandlers;
 import pl.allegro.tech.hermes.consumers.consumer.sender.http.SslContextFactoryProvider;
@@ -52,6 +53,7 @@
 import pl.allegro.tech.hermes.consumers.consumer.sender.resolver.InterpolatingEndpointAddressResolver;
 import pl.allegro.tech.hermes.consumers.consumer.sender.timeout.FutureAsyncTimeout;
 import pl.allegro.tech.hermes.consumers.consumer.trace.MetadataAppender;
+import pl.allegro.tech.hermes.metrics.HermesTimer;
 
 import java.io.IOException;
 import java.util.List;
@@ -61,10 +63,10 @@
 
 @Configuration
 @EnableConfigurationProperties({
-        SslContextProperties.class,
-        HttpClientsMonitoringProperties.class,
-        SenderAsyncTimeoutProperties.class,
-        BatchProperties.class
+    SslContextProperties.class,
+    HttpClientsMonitoringProperties.class,
+    SenderAsyncTimeoutProperties.class,
+    BatchProperties.class
 })
 public class ConsumerSenderConfiguration {
 
@@ -76,9 +78,19 @@ public Http1ClientProperties http1SerialClientProperties() {
 
     @Bean(name = "http1-serial-client")
     public HttpClient http1SerialClient(
-            HttpClientsFactory httpClientsFactory,
-            @Named("http1-serial-client-parameters") Http1ClientParameters http1ClientParameters) {
-        return httpClientsFactory.createClientForHttp1("jetty-http1-serial-client", http1ClientParameters);
+        HttpClientsFactory httpClientsFactory,
+        @Named("http1-serial-client-parameters") Http1ClientParameters http1ClientParameters,
+        MetricsFacade metricsFacade
+    ) {
+        HttpClient client = httpClientsFactory.createClientForHttp1("jetty-http1-serial-client", http1ClientParameters);
+        if (http1ClientParameters.isRequestProcessingMonitoringEnabled()) {
+            var metrics = metricsFacade.consumerSender();
+            enrichWithMetrics(
+                client, metrics.http1SerialClientRequestQueueWaitingTimer(),
+                metrics.http1SerialClientRequestProcessingTimer()
+            );
+        }
+        return client;
     }
 
     @Bean(name = "http2-serial-client-parameters")
@@ -89,12 +101,22 @@ public Http2ClientProperties http2SerialClientProperties() {
 
     @Bean
     public Http2ClientHolder http2ClientHolder(
-            HttpClientsFactory httpClientsFactory,
-            @Named("http2-serial-client-parameters") Http2ClientProperties http2ClientProperties) {
+        HttpClientsFactory httpClientsFactory,
+        @Named("http2-serial-client-parameters") Http2ClientProperties http2ClientProperties,
+        MetricsFacade metricsFacade
+    ) {
         if (!http2ClientProperties.isEnabled()) {
             return new Http2ClientHolder(null);
         } else {
-            return new Http2ClientHolder(httpClientsFactory.createClientForHttp2("jetty-http2-serial-client", http2ClientProperties));
+            HttpClient client = httpClientsFactory.createClientForHttp2("jetty-http2-serial-client", http2ClientProperties);
+            if (http2ClientProperties.isRequestProcessingMonitoringEnabled()) {
+                var metrics = metricsFacade.consumerSender();
+                enrichWithMetrics(
+                    client, metrics.http2SerialClientRequestQueueWaitingTimer(),
+                    metrics.http2SerialClientRequestProcessingTimer()
+                );
+            }
+            return new Http2ClientHolder(client);
         }
     }
 
@@ -106,9 +128,18 @@ public Http1ClientProperties http1BatchClientProperties() {
 
     @Bean(name = "http1-batch-client")
     public HttpClient http1BatchClient(
-            HttpClientsFactory httpClientsFactory,
-            @Named("http1-batch-client-parameters") Http1ClientParameters http1ClientParameters) {
-        return httpClientsFactory.createClientForHttp1("jetty-http1-batch-client", http1ClientParameters);
+        HttpClientsFactory httpClientsFactory,
+        @Named("http1-batch-client-parameters") Http1ClientParameters http1ClientParameters,
+        MetricsFacade metricsFacade) {
+        HttpClient client = httpClientsFactory.createClientForHttp1("jetty-http1-batch-client", http1ClientParameters);
+        if (http1ClientParameters.isRequestProcessingMonitoringEnabled()) {
+            var metrics = metricsFacade.consumerSender();
+            enrichWithMetrics(
+                client, metrics.http1BatchClientRequestQueueWaitingTimer(),
+                metrics.http1BatchClientRequestProcessingTimer()
+            );
+        }
+        return client;
     }
 
     @Bean(name = "oauth-http-client")
@@ -128,8 +159,8 @@ public MessageBatchSenderFactory httpMessageBatchSenderFactory(SendingResultHand
                                                                    BatchHttpRequestFactory batchHttpRequestFactory
     ) {
         return new HttpMessageBatchSenderFactory(
-                resultHandlers,
-                batchHttpRequestFactory);
+            resultHandlers,
+            batchHttpRequestFactory);
     }
 
     @Bean(initMethod = "start")
@@ -139,12 +170,12 @@ public HttpClientsWorkloadReporter httpClientsWorkloadReporter(MetricsFacade met
                                                                    Http2ClientHolder http2ClientHolder,
                                                                    HttpClientsMonitoringProperties monitoringProperties) {
         return new HttpClientsWorkloadReporter(
-                metrics,
-                http1SerialClient,
-                http1BatchClient,
-                http2ClientHolder,
-                monitoringProperties.isRequestQueueMonitoringEnabled(),
-                monitoringProperties.isConnectionPoolMonitoringEnabled());
+            metrics,
+            http1SerialClient,
+            http1BatchClient,
+            http2ClientHolder,
+            monitoringProperties.isRequestQueueMonitoringEnabled(),
+            monitoringProperties.isConnectionPoolMonitoringEnabled());
     }
 
     @Bean(destroyMethod = "closeProviders")
@@ -162,15 +193,15 @@ public ProtocolMessageSenderProvider jettyHttpMessageSenderProvider(@Named("http
                                                                         SendingResultHandlers sendingResultHandlers,
                                                                         HttpRequestFactoryProvider requestFactoryProvider) {
         return new JettyHttpMessageSenderProvider(
-                httpClient,
-                http2ClientHolder,
-                endpointAddressResolver,
-                metadataAppender,
-                authorizationProviderFactory,
-                httpHeadersProviderFactory,
-                sendingResultHandlers,
-                requestFactoryProvider,
-                ImmutableSet.of("http", "https")
+            httpClient,
+            http2ClientHolder,
+            endpointAddressResolver,
+            metadataAppender,
+            authorizationProviderFactory,
+            httpHeadersProviderFactory,
+            sendingResultHandlers,
+            requestFactoryProvider,
+            ImmutableSet.of("http", "https")
         );
     }
 
@@ -224,21 +255,21 @@ public MetadataAppender<Message> jmsMetadataAppender() {
 
     @Bean(name = "defaultPubSubMessageSenderProvider")
     public ProtocolMessageSenderProvider pubSubMessageSenderProvider(
-            GooglePubSubSenderTargetResolver targetResolver,
-            CredentialsProvider credentialsProvider,
-            ExecutorProvider executorProvider,
-            RetrySettings retrySettings,
-            BatchingSettings batchingSettings,
-            GooglePubSubMessageTransformerCreator googlePubSubMessageTransformerCreator,
-            TransportChannelProvider transportChannelProvider) {
+        GooglePubSubSenderTargetResolver targetResolver,
+        CredentialsProvider credentialsProvider,
+        ExecutorProvider executorProvider,
+        RetrySettings retrySettings,
+        BatchingSettings batchingSettings,
+        GooglePubSubMessageTransformerCreator googlePubSubMessageTransformerCreator,
+        TransportChannelProvider transportChannelProvider) {
         return new GooglePubSubMessageSenderProvider(
-                targetResolver,
-                credentialsProvider,
-                executorProvider,
-                retrySettings,
-                batchingSettings,
-                transportChannelProvider,
-                googlePubSubMessageTransformerCreator
+            targetResolver,
+            credentialsProvider,
+            executorProvider,
+            retrySettings,
+            batchingSettings,
+            transportChannelProvider,
+            googlePubSubMessageTransformerCreator
         );
     }
 
@@ -263,10 +294,18 @@ public EndpointAddressResolver interpolatingEndpointAddressResolver(UriInterpola
     public FutureAsyncTimeout futureAsyncTimeoutFactory(InstrumentedExecutorServiceFactory executorFactory,
                                                         SenderAsyncTimeoutProperties senderAsyncTimeoutProperties) {
         ScheduledExecutorService timeoutExecutorService = executorFactory.getScheduledExecutorService(
-                "async-timeout",
-                senderAsyncTimeoutProperties.getThreadPoolSize(),
-                senderAsyncTimeoutProperties.isThreadPoolMonitoringEnabled()
+            "async-timeout",
+            senderAsyncTimeoutProperties.getThreadPoolSize(),
+            senderAsyncTimeoutProperties.isThreadPoolMonitoringEnabled()
         );
         return new FutureAsyncTimeout(timeoutExecutorService);
     }
+
+    private static void enrichWithMetrics(
+        HttpClient client, HermesTimer requestQueueWaitingTimer, HermesTimer requestProcessingTimer
+    ) {
+        client.getRequestListeners().addListener(
+            new JettyHttpClientMetrics(requestQueueWaitingTimer, requestProcessingTimer)
+        );
+    }
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/config/Http1ClientProperties.java

@@ -20,6 +20,7 @@ public class Http1ClientProperties implements Http1ClientParameters {
 
     private Duration connectionTimeout = Duration.ofSeconds(15);
 
+    private boolean requestProcessingMonitoringEnabled = false;
 
     @Override
     public int getThreadPoolSize() {
@@ -83,4 +84,13 @@ public Duration getConnectionTimeout() {
     public void setConnectionTimeout(Duration connectionTimeout) {
         this.connectionTimeout = connectionTimeout;
     }
+
+    @Override
+    public boolean isRequestProcessingMonitoringEnabled() {
+        return requestProcessingMonitoringEnabled;
+    }
+
+    public void setRequestProcessingMonitoringEnabled(boolean requestProcessingMonitoringEnabled) {
+        this.requestProcessingMonitoringEnabled = requestProcessingMonitoringEnabled;
+    }
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/config/Http2ClientProperties.java

@@ -20,6 +20,8 @@ public class Http2ClientProperties implements Http2ClientParameters {
 
     private Duration connectionTimeout = Duration.ofSeconds(15);
 
+    private boolean requestProcessingMonitoringEnabled = false;
+
     public boolean isEnabled() {
         return enabled;
     }
@@ -81,4 +83,13 @@ public Duration getConnectionTimeout() {
     public void setConnectionTimeout(Duration connectionTimeout) {
         this.connectionTimeout = connectionTimeout;
     }
+
+    @Override
+    public boolean isRequestProcessingMonitoringEnabled() {
+        return this.requestProcessingMonitoringEnabled;
+    }
+
+    public void setRequestProcessingMonitoringEnabled(boolean requestProcessingMonitoringEnabled) {
+        this.requestProcessingMonitoringEnabled = requestProcessingMonitoringEnabled;
+    }
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpClientParameters.java

@@ -15,4 +15,6 @@ public interface HttpClientParameters {
         int getMaxRequestsQueuedPerDestination();
 
         Duration getConnectionTimeout();
+
+        boolean isRequestProcessingMonitoringEnabled();
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/http/HttpClientsFactory.java

@@ -16,23 +16,24 @@ public class HttpClientsFactory {
     private final SslContextFactoryProvider sslContextFactoryProvider;
 
     public HttpClientsFactory(
-            InstrumentedExecutorServiceFactory executorFactory,
-            SslContextFactoryProvider sslContextFactoryProvider) {
+        InstrumentedExecutorServiceFactory executorFactory,
+        SslContextFactoryProvider sslContextFactoryProvider
+    ) {
         this.executorFactory = executorFactory;
         this.sslContextFactoryProvider = sslContextFactoryProvider;
     }
 
     public HttpClient createClientForHttp1(String name, Http1ClientParameters http1ClientParameters) {
         ClientConnector clientConnector = new ClientConnector();
         sslContextFactoryProvider.provideSslContextFactory()
-                .ifPresent(clientConnector::setSslContextFactory);
+            .ifPresent(clientConnector::setSslContextFactory);
         HttpClientTransportOverHTTP transport = new HttpClientTransportOverHTTP(clientConnector);
         HttpClient client = new HttpClient(transport);
 
         ExecutorService executor = executorFactory.getExecutorService(
-                name,
-                http1ClientParameters.getThreadPoolSize(),
-                http1ClientParameters.isThreadPoolMonitoringEnabled());
+            name,
+            http1ClientParameters.getThreadPoolSize(),
+            http1ClientParameters.isThreadPoolMonitoringEnabled());
         client.setExecutor(executor);
         client.setMaxConnectionsPerDestination(http1ClientParameters.getMaxConnectionsPerDestination());
         client.setMaxRequestsQueuedPerDestination(http1ClientParameters.getMaxRequestsQueuedPerDestination());
@@ -46,16 +47,16 @@ public HttpClient createClientForHttp1(String name, Http1ClientParameters http1C
     public HttpClient createClientForHttp2(String name, Http2ClientParameters http2ClientParameters) {
         ClientConnector clientConnector = new ClientConnector();
         sslContextFactoryProvider.provideSslContextFactory()
-                .ifPresentOrElse(clientConnector::setSslContextFactory,
-                        () -> {
-                            throw new  IllegalStateException("Cannot create http/2 client due to lack of ssl context factory");
-                        });
+            .ifPresentOrElse(clientConnector::setSslContextFactory,
+                () -> {
+                    throw new IllegalStateException("Cannot create http/2 client due to lack of ssl context factory");
+                });
         HTTP2Client http2Client = new HTTP2Client(clientConnector);
 
         ExecutorService executor = executorFactory.getExecutorService(
-                name,
-                http2ClientParameters.getThreadPoolSize(),
-                http2ClientParameters.isThreadPoolMonitoringEnabled());
+            name,
+            http2ClientParameters.getThreadPoolSize(),
+            http2ClientParameters.isThreadPoolMonitoringEnabled());
         http2Client.setExecutor(executor);
 
         HttpClientTransportOverHTTP2 transport = new HttpClientTransportOverHTTP2(http2Client);