fixes #1122

gradle/libs.versions.toml

@@ -1,6 +1,6 @@
 [versions]
-micronaut-platform = "4.9.4"
-micronaut = "4.10.9"
+micronaut-platform = "4.10.12"
+micronaut = "4.10.21"
 micronaut-docs = "3.0.0"
 micronaut-gradle-plugin = "4.5.5"
 micronaut-test = "4.9.0"
@@ -21,7 +21,7 @@ hdr-histogram = '2.2.2'
 micronaut-aws = "4.11.0"
 micronaut-cache = "5.3.0"
 micronaut-grpc = "4.12.0"
-micronaut-logging = "1.7.1"
+micronaut-logging = "1.7.4"
 micronaut-r2dbc = "6.1.0"
 micronaut-rxjava2 = "2.8.0"
 micronaut-reactor = "3.9.0"

micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/web/ServerMetricsFilter.java

@@ -30,8 +30,12 @@
 import io.micronaut.http.annotation.ServerFilter;
 import io.micronaut.web.router.UriRouteMatch;
 import jakarta.inject.Provider;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.SignalType;
 
 import java.util.Optional;
+import java.util.concurrent.CompletionStage;
 import java.util.function.Supplier;
 
 import static io.micronaut.core.util.StringUtils.FALSE;
@@ -81,7 +85,7 @@ void onRequest(HttpRequest<?> request) {
 
     @ResponseFilter
     void onResponse(HttpRequest<?> request, HttpResponse<?> response) {
-        WebMetricsHelper httpResponseWebMetricsPublisher = new WebMetricsHelper(
+        WebMetricsHelper webMetricsHelper = new WebMetricsHelper(
             meterRegistryProvider.get(),
             resolvePath(request),
             request.getAttribute(START_ATTRIBUTE, Long.class).orElseGet(System::nanoTime),
@@ -90,6 +94,27 @@ void onResponse(HttpRequest<?> request, HttpResponse<?> response) {
             null,
             reportClientErrorURIs
         );
-        httpResponseWebMetricsPublisher.onResponse(response);
+        Object body = response.body();
+        if (response instanceof io.micronaut.http.MutableHttpResponse<?> mutableHttpResponse && body instanceof Publisher<?> publisher) {
+            mutableHttpResponse.body(Flux.from(publisher)
+                .doOnError(throwable -> webMetricsHelper.error(response, throwable))
+                .doFinally(signalType -> {
+                    if (signalType != SignalType.ON_ERROR) {
+                        webMetricsHelper.onResponse(response);
+                    }
+                }));
+            return;
+        }
+        if (response instanceof io.micronaut.http.MutableHttpResponse<?> mutableHttpResponse && body instanceof CompletionStage<?> completionStage) {
+            mutableHttpResponse.body(completionStage.whenComplete((result, throwable) -> {
+                if (throwable == null) {
+                    webMetricsHelper.onResponse(response);
+                } else {
+                    webMetricsHelper.error(response, throwable);
+                }
+            }));
+            return;
+        }
+        webMetricsHelper.onResponse(response);
     }
 }

micrometer-core/src/test/groovy/io/micronaut/configuration/metrics/binder/web/HttpMetricsSpec.groovy

@@ -7,15 +7,21 @@ import io.micrometer.core.instrument.Timer
 import io.micrometer.core.instrument.distribution.DistributionStatisticConfig
 import io.micrometer.core.instrument.distribution.HistogramSnapshot
 import io.micrometer.core.instrument.search.MeterNotFoundException
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import io.micronaut.configuration.metrics.binder.web.config.HttpClientMeterConfig
+import io.micronaut.configuration.metrics.binder.web.config.HttpMetricsConfig
 import io.micronaut.configuration.metrics.binder.web.config.HttpServerMeterConfig
 import io.micronaut.context.ApplicationContext
 import io.micronaut.context.annotation.Requires
 import io.micronaut.core.util.CollectionUtils
+import io.micronaut.http.HttpAttributes
+import io.micronaut.http.HttpRequest
 import io.micronaut.http.HttpResponse
 import io.micronaut.http.annotation.Controller
 import io.micronaut.http.annotation.Error
 import io.micronaut.http.annotation.Get
+import io.micronaut.http.MediaType
+import io.micronaut.http.client.HttpClient
 import io.micronaut.http.client.annotation.Client
 import io.micronaut.http.client.exceptions.HttpClientResponseException
 import io.micronaut.http.uri.UriBuilder
@@ -27,6 +33,12 @@ import io.micronaut.websocket.annotation.*
 import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import spock.lang.PendingFeature
+
+import java.time.Duration
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.TimeUnit
 import spock.lang.Specification
 
 import jakarta.validation.constraints.NotBlank
@@ -152,6 +164,107 @@ class HttpMetricsSpec extends Specification {
         MICRONAUT_METRICS_BINDERS + ".web.client.slos"        | "0.1,0.2,0.5" | 0                      | 0                      | null            | 3               | null            | false           | 1000000d  | 3.0E10    | 1000000d  | 3.0E10
     }
 
+
+    void "test server metrics track streaming response duration until completion"() {
+        when:
+        EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, ['spec.name': getClass().getSimpleName()])
+        def context = embeddedServer.applicationContext
+        MeterRegistry registry = context.getBean(MeterRegistry)
+        HttpClient rawClient = context.createBean(HttpClient, embeddedServer.URL)
+        long start = System.nanoTime()
+
+        then:
+        rawClient.toBlocking().retrieve('/test-http-metrics/streaming') == 'chunk-1\nchunk-2\nchunk-3\n'
+
+        when:
+        long elapsedNanos = System.nanoTime() - start
+        Timer serverTimer = registry.get(HttpServerMeterConfig.REQUESTS_METRIC).tags('uri', '/test-http-metrics/streaming').timer()
+
+        then:
+        serverTimer.count() == 1
+        serverTimer.totalTime(java.util.concurrent.TimeUnit.NANOSECONDS) >= Duration.ofMillis(250).toNanos()
+        serverTimer.totalTime(java.util.concurrent.TimeUnit.NANOSECONDS) <= elapsedNanos
+
+        cleanup:
+        rawClient.close()
+        embeddedServer.close()
+    }
+
+    void "test server metrics track completion stage response duration until completion"() {
+        when:
+        EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, ['spec.name': getClass().getSimpleName()])
+        def context = embeddedServer.applicationContext
+        MeterRegistry registry = context.getBean(MeterRegistry)
+        HttpClient rawClient = context.createBean(HttpClient, embeddedServer.URL)
+        long start = System.nanoTime()
+
+        then:
+        rawClient.toBlocking().retrieve('/test-http-metrics/completion-stage') == 'completion-stage'
+
+        when:
+        long elapsedNanos = System.nanoTime() - start
+        Timer serverTimer = registry.get(HttpServerMeterConfig.REQUESTS_METRIC).tags('uri', '/test-http-metrics/completion-stage').timer()
+
+        then:
+        serverTimer.count() == 1
+        serverTimer.totalTime(java.util.concurrent.TimeUnit.NANOSECONDS) >= Duration.ofMillis(250).toNanos()
+        serverTimer.totalTime(java.util.concurrent.TimeUnit.NANOSECONDS) <= elapsedNanos
+
+        cleanup:
+        rawClient.close()
+        embeddedServer.close()
+    }
+
+    void "test server metrics completion stage callback records successful response"() {
+        given:
+        SimpleMeterRegistry registry = new SimpleMeterRegistry()
+        ServerMetricsFilter filter = new ServerMetricsFilter({ registry }, new HttpMetricsConfig.ClientErrorsUrisConfig(true))
+        HttpRequest<?> request = HttpRequest.GET('/completion-stage-success')
+        request.setAttribute(HttpAttributes.URI_TEMPLATE, '/completion-stage-success')
+        CompletableFuture<String> completionStage = new CompletableFuture<>()
+        HttpResponse<?> response = HttpResponse.ok(completionStage)
+
+        when:
+        filter.onRequest(request)
+        filter.onResponse(request, response)
+
+        then:
+        registry.find(HttpServerMeterConfig.REQUESTS_METRIC).timer() == null
+
+        when:
+        completionStage.complete('ok')
+        response.body().get(5, TimeUnit.SECONDS)
+        Timer timer = registry.get(HttpServerMeterConfig.REQUESTS_METRIC)
+            .tags('uri', '/completion-stage-success', 'status', '200', 'exception', 'none')
+            .timer()
+
+        then:
+        timer.count() == 1
+    }
+
+    void "test server metrics completion stage callback records failed response"() {
+        given:
+        SimpleMeterRegistry registry = new SimpleMeterRegistry()
+        ServerMetricsFilter filter = new ServerMetricsFilter({ registry }, new HttpMetricsConfig.ClientErrorsUrisConfig(true))
+        HttpRequest<?> request = HttpRequest.GET('/completion-stage-failure')
+        request.setAttribute(HttpAttributes.URI_TEMPLATE, '/completion-stage-failure')
+        CompletableFuture<String> completionStage = new CompletableFuture<>()
+        HttpResponse<?> response = HttpResponse.ok(completionStage)
+
+        when:
+        filter.onRequest(request)
+        filter.onResponse(request, response)
+        completionStage.completeExceptionally(new IllegalStateException('boom'))
+        response.body().get(5, TimeUnit.SECONDS)
+
+        then:
+        thrown(ExecutionException)
+        registry.get(HttpServerMeterConfig.REQUESTS_METRIC)
+            .tags('uri', '/completion-stage-failure', 'status', '200', 'exception', 'IllegalStateException')
+            .timer()
+            .count() == 1
+    }
+
     void "test client / server metrics ignored uris for client errors"() {
         when:
         EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, [
@@ -294,6 +407,12 @@ class HttpMetricsSpec extends Specification {
         @Get("/test-http-metrics/exception-handling")
         HttpResponse exceptionHandling()
 
+        @Get(value = "/test-http-metrics/streaming", produces = MediaType.TEXT_PLAIN)
+        String streaming()
+
+        @Get("/test-http-metrics/completion-stage")
+        String completionStage()
+
         @Get("/test-http-metrics-not-found")
         HttpResponse notFound()
     }
@@ -325,6 +444,19 @@ class HttpMetricsSpec extends Specification {
             throw new MyException("my custom exception")
         }
 
+        @Get(value = "/test-http-metrics/streaming", produces = MediaType.TEXT_PLAIN)
+        Publisher<String> streaming() {
+            return Flux.range(1, 3).delayElements(Duration.ofMillis(100)).map(i -> 'chunk-' + i + '\n')
+        }
+
+        @Get("/test-http-metrics/completion-stage")
+        CompletionStage<String> completionStage() {
+            return CompletableFuture.supplyAsync(() -> {
+                Thread.sleep(300)
+                return 'completion-stage'
+            })
+        }
+
         @Error(exception = MyException)
         HttpResponse<?> myExceptionHandler() {
             return HttpResponse.badRequest()

micrometer-observation-http/src/main/java/io/micronaut/micrometer/observation/http/server/ObservationServerFilter.java

@@ -33,6 +33,11 @@
 import io.micronaut.micrometer.observation.http.server.instrumentation.ServerHttpObservationDocumentation;
 import io.micronaut.micrometer.observation.http.server.instrumentation.ServerRequestObservationContext;
 import io.micronaut.micrometer.observation.http.server.instrumentation.ServerRequestObservationConvention;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.SignalType;
+
+import java.util.concurrent.CompletionStage;
 
 import static io.micronaut.core.util.StringUtils.FALSE;
 import static io.micronaut.core.util.StringUtils.TRUE;
@@ -79,6 +84,29 @@ public void response(HttpRequest<?> request, MutableHttpResponse<?> response) {
         request.getAttribute(MICROMETER_OBSERVATION_ATTRIBUTE_KEY, Observation.class).ifPresent(observation -> {
             response.getAttribute(EXCEPTION, Throwable.class).ifPresent(observation::error);
             ((ServerRequestObservationContext) observation.getContext()).setResponse(response);
+            Object body = response.body();
+            if (body instanceof Publisher<?> publisher) {
+                response.body(Flux.from(publisher)
+                    .doOnError(throwable -> {
+                        observation.error(throwable);
+                        observation.stop();
+                    })
+                    .doFinally(signalType -> {
+                        if (signalType != SignalType.ON_ERROR) {
+                            observation.stop();
+                        }
+                    }));
+                return;
+            }
+            if (body instanceof CompletionStage<?> completionStage) {
+                response.body(completionStage.whenComplete((result, throwable) -> {
+                    if (throwable != null) {
+                        observation.error(throwable);
+                    }
+                    observation.stop();
+                }));
+                return;
+            }
             observation.stop();
         });
     }