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,6 +30,9 @@
 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.function.Supplier;
@@ -81,7 +84,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 +93,17 @@ 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;
+        }
+        webMetricsHelper.onResponse(response);
     }
 }

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

@@ -16,6 +16,8 @@ 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 +29,8 @@ import io.micronaut.websocket.annotation.*
 import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import spock.lang.PendingFeature
+
+import java.time.Duration
 import spock.lang.Specification
 
 import jakarta.validation.constraints.NotBlank
@@ -152,6 +156,32 @@ 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 client / server metrics ignored uris for client errors"() {
         when:
         EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer, [
@@ -294,6 +324,9 @@ class HttpMetricsSpec extends Specification {
         @Get("/test-http-metrics/exception-handling")
         HttpResponse exceptionHandling()
 
+        @Get(value = "/test-http-metrics/streaming", processes = MediaType.TEXT_PLAIN)
+        String streaming()
+
         @Get("/test-http-metrics-not-found")
         HttpResponse notFound()
     }
@@ -325,6 +358,11 @@ 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')
+        }
+
         @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,9 @@
 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 static io.micronaut.core.util.StringUtils.FALSE;
 import static io.micronaut.core.util.StringUtils.TRUE;
@@ -79,6 +82,20 @@ 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;
+            }
             observation.stop();
         });
     }

micrometer-observation-http/src/test/groovy/io/micronaut/micrometer/observation/ObservationHttpSpec.groovy

@@ -39,6 +39,7 @@ import spock.lang.AutoCleanup
 import spock.lang.Specification
 import spock.util.concurrent.PollingConditions
 
+import java.time.Duration
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.CompletionStage
 
@@ -317,6 +318,31 @@ class ObservationHttpSpec extends Specification {
         testObservationRegistry.clear()
     }
 
+
+    void 'test server observation stays open until streaming response completes'() {
+        when:
+        String response = reactorHttpClient.toBlocking().retrieve('/streaming/body', String)
+
+        then:
+        response == 'stream-1\nstream-2\nstream-3\n'
+
+        and:
+        conditions.eventually {
+            TestObservationRegistryAssert.assertThat(testObservationRegistry)
+                .hasObservationWithNameEqualTo('http.server.requests')
+                .that()
+                .hasContextualNameEqualTo('http get /streaming/body')
+                .hasLowCardinalityKeyValue('uri', '/streaming/body')
+                .hasLowCardinalityKeyValue('streaming', 'done')
+                .hasBeenStarted()
+                .hasBeenStopped()
+        }
+        testObservationRegistry.getCurrentObservation() == null
+
+        cleanup:
+        testObservationRegistry.clear()
+    }
+
     void 'test continue nested HTTP observation - reactive'() {
 
         when:
@@ -497,6 +523,24 @@ class ObservationHttpSpec extends Specification {
         Publisher<String> continuedRx(String name)
     }
 
+
+    @Controller('/streaming')
+    static class StreamingController {
+
+        @Inject
+        ObservationRegistry observationRegistry
+
+        @Get(value = '/body', produces = 'text/plain')
+        Publisher<String> body() {
+            return Flux.range(1, 3)
+                .delayElements(Duration.ofMillis(50))
+                .map { i ->
+                    observationRegistry.currentObservation.lowCardinalityKeyValue('streaming', i == 3 ? 'done' : 'progress')
+                    return "stream-${i}\n"
+                }
+        }
+    }
+
     @Controller('/error')
     static class ErrorController {
 

test-suite/micronaut-serialization/build.gradle

@@ -24,6 +24,10 @@ dependencies {
 
 graalvmNative.toolchainDetection = false
 
+tasks.withType(JavaCompile).configureEach {
+    options.release = 17
+}
+
 micronaut {
     version.set(libs.versions.micronaut.platform)
     runtime("netty")

test-suite/test-observation-datasource/build.gradle

@@ -28,6 +28,10 @@ dependencies {
 
 graalvmNative.toolchainDetection = false
 
+tasks.withType(JavaCompile).configureEach {
+    options.release = 17
+}
+
 micronaut {
     version.set(libs.versions.micronaut.platform)
     runtime("netty")