Reactor netty micrometer

instrumentation/spring/spring-boot-autoconfigure/build.gradle.kts

@@ -8,6 +8,8 @@ group = "io.opentelemetry.instrumentation"
 
 val versions: Map<String, String> by project
 val springBootVersion = versions["org.springframework.boot"]
+val micrometerVersion = "1.10.0"
+val micrometerTraceVersion = "1.0.0"
 
 dependencies {
   implementation("org.springframework.boot:spring-boot-autoconfigure:$springBootVersion")
@@ -32,6 +34,10 @@ dependencies {
   implementation(project(":instrumentation:jdbc:library"))
 
   library("org.springframework.kafka:spring-kafka:2.9.0")
+  library("io.projectreactor.netty:reactor-netty:1.1.0")
+  implementation("io.micrometer:micrometer-tracing-bridge-otel:$micrometerTraceVersion")
+  implementation("io.micrometer:micrometer-tracing:$micrometerTraceVersion")
+  implementation("io.micrometer:micrometer-core:$micrometerVersion")
   library("org.springframework.boot:spring-boot-starter-actuator:$springBootVersion")
   library("org.springframework.boot:spring-boot-starter-aop:$springBootVersion")
   library("org.springframework.boot:spring-boot-starter-web:$springBootVersion")
@@ -68,6 +74,9 @@ dependencies {
   testImplementation("io.opentelemetry:opentelemetry-exporter-otlp")
   testImplementation("io.opentelemetry:opentelemetry-exporter-zipkin")
   testImplementation(project(":instrumentation-annotations"))
+  testImplementation("io.micrometer:micrometer-tracing-bridge-otel:1.0.0")
+  testImplementation("io.micrometer:micrometer-tracing:1.0.0")
+  testImplementation("io.micrometer:micrometer-core:1.10.0")
 }
 
 val latestDepTest = findProperty("testLatestDeps") as Boolean

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/reactornetty/HttpClientBeanPostProcessor.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty;
+
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import reactor.netty.http.client.HttpClient;
+
+final class HttpClientBeanPostProcessor implements BeanPostProcessor {
+
+  HttpClientBeanPostProcessor() {
+  }
+
+  @Override
+  public Object postProcessBeforeInitialization(Object bean, String beanName) {
+    if (bean instanceof HttpClient) {
+      HttpClient httpClient = (HttpClient) bean;
+      return httpClient.metrics(true, s -> s);
+    }
+    return bean;
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/reactornetty/HttpServerBeanPostProcessor.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty;
+
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import reactor.netty.http.server.HttpServer;
+
+final class HttpServerBeanPostProcessor implements BeanPostProcessor {
+
+  HttpServerBeanPostProcessor() {
+  }
+
+  @Override
+  public Object postProcessBeforeInitialization(Object bean, String beanName) {
+    if (bean instanceof HttpServer) {
+      HttpServer httpServer = (HttpServer) bean;
+      return httpServer.metrics(true, s -> s);
+    }
+    return bean;
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/reactornetty/ReactorNettyHttpInitializingBean.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty;
+
+import static reactor.netty.Metrics.OBSERVATION_REGISTRY;
+
+import io.micrometer.tracing.otel.bridge.OtelBaggageManager;
+import io.micrometer.tracing.otel.bridge.OtelCurrentTraceContext;
+import io.micrometer.tracing.otel.bridge.OtelPropagator;
+import io.micrometer.tracing.otel.bridge.OtelTracer;
+import io.micrometer.tracing.propagation.Propagator;
+import io.opentelemetry.api.OpenTelemetry;
+import java.util.Collections;
+import org.springframework.beans.factory.InitializingBean;
+import io.opentelemetry.api.trace.Tracer;
+import reactor.netty.http.observability.ReactorNettyPropagatingReceiverTracingObservationHandler;
+import reactor.netty.http.observability.ReactorNettyPropagatingSenderTracingObservationHandler;
+import reactor.netty.observability.ReactorNettyTracingObservationHandler;
+
+final class ReactorNettyHttpInitializingBean implements InitializingBean {
+
+  private final OpenTelemetry openTelemetry;
+
+  ReactorNettyHttpInitializingBean(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  @Override
+  public void afterPropertiesSet() {
+    if (openTelemetry == null) {
+      return;
+    }
+    OtelCurrentTraceContext otelCurrentTraceContext = new OtelCurrentTraceContext();
+    Tracer tracer = openTelemetry.getTracer("springboot-reactor-netty");
+    OtelTracer otelTracer =
+        new OtelTracer(
+            tracer,
+            otelCurrentTraceContext,
+            event -> {},
+            new OtelBaggageManager(
+                otelCurrentTraceContext, Collections.emptyList(), Collections.emptyList()));
+    Propagator propagator = new OtelPropagator(openTelemetry.getPropagators(), tracer);
+    OBSERVATION_REGISTRY
+        .observationConfig()
+        .observationHandler(new ReactorNettyPropagatingReceiverTracingObservationHandler(otelTracer, propagator))
+        .observationHandler(new ReactorNettyPropagatingSenderTracingObservationHandler(otelTracer, propagator))
+        .observationHandler(new ReactorNettyTracingObservationHandler(otelTracer));
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/reactornetty/ReactorNettyInstrumentationAutoConfiguration.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty;
+
+import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.spring.autoconfigure.internal.SdkEnabled;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.http.server.HttpServer;
+
+/**
+ * Configures {@link HttpClient} and {@link reactor.netty.http.server.HttpServer} for tracing.
+ *
+ */
+@ConditionalOnBean(OpenTelemetry.class)
+@ConditionalOnProperty(name = "otel.instrumentation.reactor-netty.enabled", matchIfMissing = true)
+@Conditional(SdkEnabled.class)
+@Configuration
+public class ReactorNettyInstrumentationAutoConfiguration {
+
+  public ReactorNettyInstrumentationAutoConfiguration() {}
+
+  @ConditionalOnClass({HttpClient.class, ObservationRegistry.class})
+  @Bean
+  static ReactorNettyHttpInitializingBean reactorNettyHttpClientInitializingBean(
+      ObjectProvider<OpenTelemetry> openTelemetryProvider) {
+    return new ReactorNettyHttpInitializingBean(openTelemetryProvider.getIfAvailable());
+  }
+
+  @Bean
+  @ConditionalOnClass(HttpClient.class)
+  static HttpClientBeanPostProcessor httpClientBeanPostProcessor() {
+    return new HttpClientBeanPostProcessor();
+  }
+
+  @Bean
+  @ConditionalOnClass(HttpServer.class)
+  static HttpServerBeanPostProcessor httpServerBeanPostProcessor() {
+    return new HttpServerBeanPostProcessor();
+  }
+
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -4,6 +4,7 @@ io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.kafka.Kafk
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.logging.OpenTelemetryAppenderAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.jdbc.OpenTelemetryJdbcDriverAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.micrometer.MicrometerBridgeAutoConfiguration
+io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty.ReactorNettyInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.web.SpringWebInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webflux.SpringWebfluxInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webmvc.SpringWebMvc6InstrumentationAutoConfiguration

instrumentation/spring/spring-boot-autoconfigure/src/test/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/reactornetty/ReactorNettyInstrumentationAutoConfigurationTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.reactornetty;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.opentelemetry.api.OpenTelemetry;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+class ReactorNettyInstrumentationAutoConfigurationTest {
+
+  private final ApplicationContextRunner contextRunner =
+      new ApplicationContextRunner()
+          .withBean(OpenTelemetry.class, OpenTelemetry::noop)
+          .withConfiguration(
+              AutoConfigurations.of(ReactorNettyInstrumentationAutoConfiguration.class));
+
+  @Test
+  void instrumentationEnabled() {
+    contextRunner
+        .withPropertyValues("otel.instrumentation.reactor-netty.enabled=true")
+        .run(
+            context ->
+                assertThat(
+                        context.getBean(
+                            "reactorNettyHttpClientInitializingBean",
+                            ReactorNettyHttpInitializingBean.class))
+                    .isNotNull());
+  }
+
+  @Test
+  void instrumentationDisabled() {
+    contextRunner
+        .withPropertyValues("otel.instrumentation.reactor-netty.enabled=false")
+        .run(
+            context ->
+                assertThat(context.containsBean("reactorNettyHttpClientInitializingBean"))
+                    .isFalse());
+  }
+
+  @Test
+  void defaultConfiguration() {
+    contextRunner.run(
+        context ->
+            assertThat(
+                    context.getBean(
+                        "reactorNettyHttpClientInitializingBean",
+                        ReactorNettyHttpInitializingBean.class))
+                .isNotNull());
+  }
+}