add spring starter r2dbc support (#11221)

Co-authored-by: Jean Bisutti <jean.bisutti@gmail.com>
Co-authored-by: Lauri Tulmin <ltulmin@splunk.com>

instrumentation/r2dbc-1.0/library-instrumentation-shaded/build.gradle.kts

@@ -12,7 +12,9 @@ dependencies {
 
 tasks {
   shadowJar {
-    exclude("META-INF/**/*")
+    exclude {
+      it.path.startsWith("META-INF") && !it.path.startsWith("META-INF/io/opentelemetry/instrumentation/")
+    }
 
     dependencies {
       // including only :r2dbc-1.0:library excludes its transitive dependencies
@@ -21,13 +23,20 @@ tasks {
     }
     relocate(
       "io.r2dbc.proxy",
-      "io.opentelemetry.javaagent.instrumentation.r2dbc.v1_0.shaded.io.r2dbc.proxy"
+      "io.opentelemetry.instrumentation.r2dbc.v1_0.shaded.io.r2dbc.proxy"
     )
   }
 
   val extractShadowJar by registering(Copy::class) {
     dependsOn(shadowJar)
     from(zipTree(shadowJar.get().archiveFile))
+    exclude("META-INF/**")
     into("build/extracted/shadow")
   }
+
+  val extractShadowJarSpring by registering(Copy::class) {
+    dependsOn(shadowJar)
+    from(zipTree(shadowJar.get().archiveFile))
+    into("build/extracted/shadow-spring")
+  }
 }

instrumentation/r2dbc-1.0/library/src/main/java/io/opentelemetry/instrumentation/r2dbc/v1_0/internal/R2dbcInstrumenterBuilder.java

@@ -41,7 +41,6 @@ public final class R2dbcInstrumenterBuilder {
   }
 
   public Instrumenter<DbExecution, Void> build(boolean statementSanitizationEnabled) {
-
     return Instrumenter.<DbExecution, Void>builder(
             openTelemetry,
             INSTRUMENTATION_NAME,

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

@@ -9,6 +9,18 @@ group = "io.opentelemetry.instrumentation"
 val versions: Map<String, String> by project
 val springBootVersion = versions["org.springframework.boot"]
 
+// r2dbc-proxy is shadowed to prevent org.springframework.boot.autoconfigure.r2dbc.R2dbcAutoConfiguration
+// from being loaded by Spring Boot (by the presence of META-INF/services/io.r2dbc.spi.ConnectionFactoryProvider) - even if the user doesn't want to use R2DBC.
+sourceSets {
+  main {
+    val shadedDep = project(":instrumentation:r2dbc-1.0:library-instrumentation-shaded")
+    output.dir(
+      shadedDep.file("build/extracted/shadow-spring"),
+      "builtBy" to ":instrumentation:r2dbc-1.0:library-instrumentation-shaded:extractShadowJarSpring",
+    )
+  }
+}
+
 dependencies {
   implementation("org.springframework.boot:spring-boot-autoconfigure:$springBootVersion")
   annotationProcessor("org.springframework.boot:spring-boot-autoconfigure-processor:$springBootVersion")
@@ -17,6 +29,7 @@ dependencies {
 
   implementation(project(":instrumentation-annotations-support"))
   implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-2.6:library"))
+  compileOnly(project(path = ":instrumentation:r2dbc-1.0:library-instrumentation-shaded", configuration = "shadow"))
   implementation(project(":instrumentation:spring:spring-kafka-2.7:library"))
   implementation(project(":instrumentation:spring:spring-web:spring-web-3.1:library"))
   implementation(project(":instrumentation:spring:spring-webmvc:spring-webmvc-5.3:library"))
@@ -36,6 +49,7 @@ dependencies {
   library("org.springframework.boot:spring-boot-starter-aop:$springBootVersion")
   library("org.springframework.boot:spring-boot-starter-web:$springBootVersion")
   library("org.springframework.boot:spring-boot-starter-webflux:$springBootVersion")
+  library("org.springframework.boot:spring-boot-starter-data-r2dbc:$springBootVersion")
 
   implementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
   implementation(project(":sdk-autoconfigure-support"))

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/r2dbc/R2dbcAutoConfiguration.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.r2dbc;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.spring.autoconfigure.internal.SdkEnabled;
+import io.r2dbc.spi.ConnectionFactory;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Value;
+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;
+
+@ConditionalOnBean(OpenTelemetry.class)
+@ConditionalOnClass(ConnectionFactory.class)
+@ConditionalOnProperty(name = "otel.instrumentation.r2dbc.enabled", matchIfMissing = true)
+@Conditional(SdkEnabled.class)
+@Configuration(proxyBeanMethods = false)
+public class R2dbcAutoConfiguration {
+
+  public R2dbcAutoConfiguration() {}
+
+  @Bean
+  // static to avoid "is not eligible for getting processed by all BeanPostProcessors" warning
+  static R2dbcInstrumentingPostProcessor r2dbcInstrumentingPostProcessor(
+      ObjectProvider<OpenTelemetry> openTelemetryProvider,
+      @Value("${otel.instrumentation.common.db-statement-sanitizer.enabled:true}")
+          boolean statementSanitizationEnabled) {
+    return new R2dbcInstrumentingPostProcessor(openTelemetryProvider, statementSanitizationEnabled);
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/r2dbc/R2dbcInstrumentingPostProcessor.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.r2dbc;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.r2dbc.v1_0.R2dbcTelemetry;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.ConnectionFactoryOptions;
+import org.springframework.aop.scope.ScopedProxyUtils;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.r2dbc.OptionsCapableConnectionFactory;
+
+class R2dbcInstrumentingPostProcessor implements BeanPostProcessor {
+
+  private final ObjectProvider<OpenTelemetry> openTelemetryProvider;
+  private final boolean statementSanitizationEnabled;
+
+  R2dbcInstrumentingPostProcessor(
+      ObjectProvider<OpenTelemetry> openTelemetryProvider, boolean statementSanitizationEnabled) {
+    this.openTelemetryProvider = openTelemetryProvider;
+    this.statementSanitizationEnabled = statementSanitizationEnabled;
+  }
+
+  @Override
+  public Object postProcessAfterInitialization(Object bean, String beanName) {
+    if (bean instanceof ConnectionFactory && !ScopedProxyUtils.isScopedTarget(beanName)) {
+      ConnectionFactory connectionFactory = (ConnectionFactory) bean;
+      return R2dbcTelemetry.builder(openTelemetryProvider.getObject())
+          .setStatementSanitizationEnabled(statementSanitizationEnabled)
+          .build()
+          .wrapConnectionFactory(connectionFactory, getConnectionFactoryOptions(connectionFactory));
+    }
+    return bean;
+  }
+
+  private static ConnectionFactoryOptions getConnectionFactoryOptions(
+      ConnectionFactory connectionFactory) {
+    OptionsCapableConnectionFactory optionsCapableConnectionFactory =
+        OptionsCapableConnectionFactory.unwrapFrom(connectionFactory);
+    if (optionsCapableConnectionFactory != null) {
+      return optionsCapableConnectionFactory.getOptions();
+    } else {
+      // in practice should never happen
+      // fall back to empty options; or reconstruct them from the R2dbcProperties
+      return ConnectionFactoryOptions.builder().build();
+    }
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -273,6 +273,12 @@
       "description": "Enable the <code>@WithSpan</code> annotation.",
       "defaultValue": true
     },
+    {
+      "name": "otel.instrumentation.common.db-statement-sanitizer.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Enables the DB statement sanitization for R2DBC.",
+      "defaultValue": true
+    },
     {
       "name": "otel.instrumentation.kafka.enabled",
       "type": "java.lang.Boolean",
@@ -332,6 +338,12 @@
       "description": "Enable the Micrometer instrumentation.",
       "defaultValue": false
     },
+    {
+      "name": "otel.instrumentation.r2dbc.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Enable the R2DBC (reactive JDBC) instrumentation. Also see <code>otel.instrumentation.common.db-statement-sanitizer.enabled</code>.",
+      "defaultValue": true
+    },
     {
       "name": "otel.instrumentation.spring-web.enabled",
       "type": "java.lang.Boolean",

instrumentation/spring/spring-boot-autoconfigure/src/main/resources/META-INF/spring.factories

@@ -5,6 +5,7 @@ io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.kafka.Kafk
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.logging.OpenTelemetryAppenderAutoConfiguration,\
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.jdbc.JdbcInstrumentationAutoConfiguration,\
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.micrometer.MicrometerBridgeAutoConfiguration,\
+io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.r2dbc.R2dbcAutoConfiguration,\
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.web.SpringWebInstrumentationAutoConfiguration,\
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webflux.SpringWebfluxInstrumentationAutoConfiguration,\
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webmvc.SpringWebMvc5InstrumentationAutoConfiguration

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.JdbcInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.micrometer.MicrometerBridgeAutoConfiguration
+io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.r2dbc.R2dbcAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.web.SpringWebInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webflux.SpringWebfluxInstrumentationAutoConfiguration
 io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.webmvc.SpringWebMvc6InstrumentationAutoConfiguration

smoke-tests-otel-starter/spring-boot-reactive-2/build.gradle.kts

@@ -11,6 +11,10 @@ dependencies {
 
   implementation(project(":smoke-tests-otel-starter:spring-boot-reactive-common"))
   implementation("org.springframework.boot:spring-boot-starter-webflux")
+  implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
+
+  runtimeOnly("com.h2database:h2")
+  runtimeOnly("io.r2dbc:r2dbc-h2")
 
   testImplementation("org.springframework.boot:spring-boot-starter-test")
   testImplementation("io.projectreactor:reactor-test")

smoke-tests-otel-starter/spring-boot-reactive-3/build.gradle.kts

@@ -16,6 +16,10 @@ dependencies {
 
   implementation(project(":smoke-tests-otel-starter:spring-boot-reactive-common"))
   implementation("org.springframework.boot:spring-boot-starter-webflux")
+  implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")
+
+  runtimeOnly("com.h2database:h2")
+  runtimeOnly("io.r2dbc:r2dbc-h2")
 
   testImplementation("org.springframework.boot:spring-boot-starter-test")
   testImplementation("io.projectreactor:reactor-test")

smoke-tests-otel-starter/spring-boot-reactive-common/build.gradle.kts

@@ -13,6 +13,7 @@ dependencies {
   compileOnly("org.springframework.boot:spring-boot-starter-web")
   compileOnly("org.springframework.boot:spring-boot-starter-webflux")
   compileOnly("org.springframework.boot:spring-boot-starter-test")
+  compileOnly("org.springframework.boot:spring-boot-starter-data-r2dbc")
   api(project(":smoke-tests-otel-starter:spring-smoke-testing"))
 }
 

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/java/io/opentelemetry/spring/smoketest/AbstractOtelReactiveSpringStarterSmokeTest.java

@@ -10,6 +10,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.semconv.HttpAttributes;
 import io.opentelemetry.semconv.UrlAttributes;
+import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -45,6 +46,8 @@ public class AbstractOtelReactiveSpringStarterSmokeTest extends AbstractSpringSt
         .blockLast();
 
     testing.waitAndAssertTraces(
+        trace ->
+            trace.hasSpansSatisfyingExactly(span -> span.hasName("CREATE TABLE testdb.player")),
         trace ->
             trace.hasSpansSatisfyingExactly(
                 span ->
@@ -57,6 +60,24 @@ public class AbstractOtelReactiveSpringStarterSmokeTest extends AbstractSpringSt
                         .hasName("GET /webflux")
                         .hasAttribute(HttpAttributes.HTTP_REQUEST_METHOD, "GET")
                         .hasAttribute(HttpAttributes.HTTP_RESPONSE_STATUS_CODE, 200L)
-                        .hasAttribute(HttpAttributes.HTTP_ROUTE, "/webflux")));
+                        .hasAttribute(HttpAttributes.HTTP_ROUTE, "/webflux"),
+                span ->
+                    span.hasKind(SpanKind.CLIENT)
+                        .satisfies(
+                            s ->
+                                assertThat(s.getName())
+                                    .isEqualToIgnoringCase("SELECT testdb.PLAYER"))
+                        .hasAttribute(DbIncubatingAttributes.DB_NAME, "testdb")
+                        .hasAttributesSatisfying(
+                            a ->
+                                assertThat(a.get(DbIncubatingAttributes.DB_SQL_TABLE))
+                                    .isEqualToIgnoringCase("PLAYER"))
+                        .hasAttribute(DbIncubatingAttributes.DB_OPERATION, "SELECT")
+                        .hasAttributesSatisfying(
+                            a ->
+                                assertThat(a.get(DbIncubatingAttributes.DB_STATEMENT))
+                                    .isEqualToIgnoringCase(
+                                        "SELECT PLAYER.* FROM PLAYER WHERE PLAYER.ID = $? LIMIT ?"))
+                        .hasAttribute(DbIncubatingAttributes.DB_SYSTEM, "h2")));
   }
 }

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/java/io/opentelemetry/spring/smoketest/OtelReactiveSpringStarterSmokeTestController.java

@@ -13,9 +13,16 @@ import reactor.core.publisher.Mono;
 public class OtelReactiveSpringStarterSmokeTestController {
 
   public static final String WEBFLUX = "/webflux";
+  private final PlayerRepository playerRepository;
+
+  public OtelReactiveSpringStarterSmokeTestController(PlayerRepository playerRepository) {
+    this.playerRepository = playerRepository;
+  }
 
   @GetMapping(WEBFLUX)
   public Mono<String> webflux() {
-    return Mono.just("webflux");
+    return playerRepository
+        .findById(1)
+        .map(player -> "Player: " + player.getName() + " Age: " + player.getAge());
   }
 }

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/java/io/opentelemetry/spring/smoketest/Player.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import org.springframework.data.annotation.Id;
+
+public class Player {
+  @Id Integer id;
+  String name;
+  Integer age;
+
+  public Player() {}
+
+  public Player(Integer id, String name, Integer age) {
+    this.id = id;
+    this.name = name;
+    this.age = age;
+  }
+
+  public Integer getId() {
+    return id;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public Integer getAge() {
+    return age;
+  }
+}

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/java/io/opentelemetry/spring/smoketest/PlayerRepository.java

@@ -0,0 +1,10 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+public interface PlayerRepository extends ReactiveCrudRepository<Player, Integer> {}

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/resources/application.yaml

@@ -0,0 +1,6 @@
+spring:
+  r2dbc:
+    url: r2dbc:h2:mem:///testdb
+  jpa:
+    hibernate:
+      ddl-auto: create

smoke-tests-otel-starter/spring-boot-reactive-common/src/main/resources/schema.sql

@@ -0,0 +1 @@
+CREATE TABLE IF NOT EXISTS player(id INT NOT NULL AUTO_INCREMENT, name VARCHAR(255), age INT, PRIMARY KEY (id));

smoke-tests-otel-starter/spring-smoke-testing/src/main/java/io/opentelemetry/spring/smoketest/AbstractSpringStarterSmokeTest.java

@@ -31,10 +31,17 @@ public abstract class AbstractSpringStarterSmokeTest {
   void checkSpringLogs(CapturedOutput output) {
     // warnings are emitted if the auto-configuration have non-fatal problems
     assertThat(output)
+        // not a warning in Spring Boot 2
+        .doesNotContain("is not eligible for getting processed by all BeanPostProcessors")
         // only look for WARN and ERROR log level, e.g. [Test worker] WARN
         .doesNotContain("] WARN")
-        .doesNotContain("] ERROR")
-        // not a warning in Spring Boot 2
-        .doesNotContain("is not eligible for getting processed by all BeanPostProcessors");
+        .satisfies(
+            s -> {
+              if (!s.toString()
+                  .contains(
+                      "Unable to load io.netty.resolver.dns.macos.MacOSDnsServerAddressStreamProvider")) {
+                assertThat(s).doesNotContain("] ERROR");
+              }
+            });
   }
 }