Add db.system.name support to r2dbc

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/db/DbClientAttributesExtractor.java

@@ -106,7 +106,7 @@ static <REQUEST, RESPONSE> void onStartCommon(
       }
     }
     if (emitOldDatabaseSemconv()) {
-      attributes.put(DB_SYSTEM, getter.getDbSystemName(request));
+      attributes.put(DB_SYSTEM, getter.getDbSystem(request));
       attributes.put(DB_USER, getter.getUser(request));
       attributes.put(DB_NAME, getter.getDbNamespace(request));
       attributes.put(DB_CONNECTION_STRING, getter.getConnectionString(request));

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/db/DbClientAttributesGetter.java

@@ -41,6 +41,21 @@ default String getDbQuerySummary(REQUEST request) {
   // TODO: make this required to implement
   String getDbSystemName(REQUEST request);
 
+  /**
+   * Returns the old semconv {@code db.system} value. Defaults to {@link #getDbSystemName} for
+   * instrumentations where the old and new values are the same.
+   *
+   * <p>Implementors whose old {@code db.system} value differs from the new {@code db.system.name}
+   * value <b>must override</b> this method; otherwise old-semconv spans will silently emit the new
+   * value for {@code db.system}.
+   *
+   * @deprecated Use {@link #getDbSystemName} instead.
+   */
+  @Deprecated // to be removed in 3.0
+  default String getDbSystem(REQUEST request) {
+    return getDbSystemName(request);
+  }
+
   @Nullable
   String getDbNamespace(REQUEST request);
 

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

@@ -18,21 +18,51 @@
 import io.r2dbc.proxy.core.QueryInfo;
 import io.r2dbc.spi.Connection;
 import io.r2dbc.spi.ConnectionFactoryOptions;
+import java.util.HashMap;
 import java.util.Locale;
+import java.util.Map;
+import javax.annotation.Nullable;
 
 /**
  * This class is internal and is hence not for public use. Its APIs are unstable and can change at
  * any time.
  */
 public final class DbExecution {
+  // copied from DbAttributes.DbSystemNameValues
+  private static final String POSTGRESQL = "postgresql";
+  // copied from DbAttributes.DbSystemNameValues
+  private static final String MYSQL = "mysql";
+  // copied from DbAttributes.DbSystemNameValues
+  private static final String MARIADB = "mariadb";
+  // copied from DbAttributes.DbSystemNameValues
+  private static final String MICROSOFT_SQL_SERVER = "microsoft.sql_server";
+  // copied from DbIncubatingAttributes.DbSystemNameIncubatingValues
+  private static final String ORACLE_DB = "oracle.db";
+  // copied from DbIncubatingAttributes.DbSystemNameIncubatingValues
+  private static final String H2DATABASE = "h2database";
   // copied from DbIncubatingAttributes.DbSystemNameIncubatingValues
   private static final String OTHER_SQL = "other_sql";
 
+  // R2DBC driver identifier → stable semconv db.system.name value
+  private static final Map<String, String> DRIVER_TO_SYSTEM_NAME = buildDriverToSystemName();
+
+  private static Map<String, String> buildDriverToSystemName() {
+    Map<String, String> map = new HashMap<>();
+    map.put("postgresql", POSTGRESQL);
+    map.put("mysql", MYSQL);
+    map.put("mariadb", MARIADB);
+    map.put("mssql", MICROSOFT_SQL_SERVER);
+    map.put("oracle", ORACLE_DB);
+    map.put("h2", H2DATABASE);
+    return map;
+  }
+
+  private final String systemName;
   private final String system;
   private final String user;
-  private final String name;
-  private final String host;
-  private final Integer port;
+  private final String namespace;
+  private final String serverAddress;
+  private final Integer serverPort;
   private final String connectionString;
   private final String rawQueryText;
   private final boolean parameterizedQuery;
@@ -50,23 +80,26 @@ public DbExecution(QueryExecutionInfo queryInfo, ConnectionFactoryOptions factor
                 .split(" ")[0]
             : OTHER_SQL;
     this.user = factoryOptions.hasOption(USER) ? (String) factoryOptions.getValue(USER) : null;
-    this.name =
+    this.namespace =
         factoryOptions.hasOption(DATABASE)
             ? ((String) factoryOptions.getValue(DATABASE)).toLowerCase(Locale.ROOT)
             : null;
     String driver =
         factoryOptions.hasOption(DRIVER) ? (String) factoryOptions.getValue(DRIVER) : null;
     String protocol =
         factoryOptions.hasOption(PROTOCOL) ? (String) factoryOptions.getValue(PROTOCOL) : null;
-    this.host = factoryOptions.hasOption(HOST) ? (String) factoryOptions.getValue(HOST) : null;
-    this.port = factoryOptions.hasOption(PORT) ? (Integer) factoryOptions.getValue(PORT) : null;
+    this.systemName = resolveDbSystemName(driver, protocol);
+    this.serverAddress =
+        factoryOptions.hasOption(HOST) ? (String) factoryOptions.getValue(HOST) : null;
+    this.serverPort =
+        factoryOptions.hasOption(PORT) ? (Integer) factoryOptions.getValue(PORT) : null;
     this.connectionString =
         String.format(
             "%s%s:%s%s",
             driver != null ? driver : "",
             protocol != null ? ":" + protocol : "",
-            host != null ? "//" + host : "",
-            port != null ? ":" + port : "");
+            serverAddress != null ? "//" + serverAddress : "",
+            serverPort != null ? ":" + serverPort : "");
     this.rawQueryText =
         queryInfo.getQueries().stream()
             .map(QueryInfo::getQuery)
@@ -80,14 +113,21 @@ public DbExecution(QueryExecutionInfo queryInfo, ConnectionFactoryOptions factor
     R2dbcSqlCommenterUtil.clearQueries(queryInfo.getConnectionInfo());
   }
 
-  public Integer getPort() {
-    return port;
+  @Nullable
+  public String getServerAddress() {
+    return serverAddress;
+  }
+
+  @Nullable
+  public Integer getServerPort() {
+    return serverPort;
   }
 
-  public String getHost() {
-    return host;
+  public String getSystemName() {
+    return systemName;
   }
 
+  @Deprecated // to be removed in 3.0
   public String getSystem() {
     return system;
   }
@@ -96,8 +136,9 @@ public String getUser() {
     return user;
   }
 
-  public String getName() {
-    return name;
+  @Nullable
+  public String getNamespace() {
+    return namespace;
   }
 
   public String getConnectionString() {
@@ -120,31 +161,10 @@ public void setContext(Context context) {
     this.context = context;
   }
 
-  @Override
-  public String toString() {
-    return "DbExecution{"
-        + "system='"
-        + system
-        + '\''
-        + ", user='"
-        + user
-        + '\''
-        + ", name='"
-        + name
-        + '\''
-        + ", host='"
-        + host
-        + '\''
-        + ", port="
-        + port
-        + ", connectionString='"
-        + connectionString
-        + '\''
-        + ", rawQueryText='"
-        + rawQueryText
-        + '\''
-        + ", context="
-        + context
-        + '}';
+  private static String resolveDbSystemName(@Nullable String driver, @Nullable String protocol) {
+    // Use PROTOCOL when DRIVER is "pool" (r2dbc-pool wraps the real driver in PROTOCOL),
+    // otherwise use DRIVER directly.
+    String rawDriver = "pool".equals(driver) && protocol != null ? protocol : driver;
+    return rawDriver != null ? DRIVER_TO_SYSTEM_NAME.getOrDefault(rawDriver, OTHER_SQL) : OTHER_SQL;
   }
 }

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

@@ -23,6 +23,12 @@ public enum R2dbcSqlAttributesGetter implements SqlClientAttributesGetter<DbExec
 
   @Override
   public String getDbSystemName(DbExecution request) {
+    return request.getSystemName();
+  }
+
+  @Deprecated // to be removed in 3.0
+  @Override
+  public String getDbSystem(DbExecution request) {
     return request.getSystem();
   }
 
@@ -47,7 +53,7 @@ public String getUser(DbExecution request) {
   @Override
   @Nullable
   public String getDbNamespace(DbExecution request) {
-    return request.getName();
+    return request.getNamespace();
   }
 
   @Deprecated // to be removed in 3.0
@@ -75,13 +81,13 @@ public String getErrorType(
   @Nullable
   @Override
   public String getServerAddress(DbExecution request) {
-    return request.getHost();
+    return request.getServerAddress();
   }
 
   @Nullable
   @Override
   public Integer getServerPort(DbExecution request) {
-    return request.getPort();
+    return request.getServerPort();
   }
 
   @Override

instrumentation/r2dbc-1.0/library/src/test/java/io/opentelemetry/instrumentation/r2dbc/v1_0/DbExecutionTest.java

@@ -18,6 +18,8 @@
 import io.r2dbc.spi.ConnectionMetadata;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -27,6 +29,7 @@ class DbExecutionTest {
   @Mock Connection connection;
   @Mock ConnectionMetadata metadata;
 
+  @SuppressWarnings("deprecation") // testing deprecated semconv
   @Test
   void dbExecution() {
     when(connection.getMetadata()).thenReturn(metadata);
@@ -39,13 +42,55 @@ void dbExecution() {
     ConnectionFactoryOptions factoryOptions =
         ConnectionFactoryOptions.parse("r2dbc:mariadb://root:root@localhost:3306/db");
     DbExecution dbExecution = new DbExecution(queryExecutionInfo, factoryOptions);
+    assertThat(dbExecution.getSystemName()).isEqualTo("mariadb");
     assertThat(dbExecution.getSystem()).isEqualTo("testdb");
     assertThat(dbExecution.getUser()).isEqualTo("root");
-    assertThat(dbExecution.getName()).isEqualTo("db");
-    assertThat(dbExecution.getHost()).isEqualTo("localhost");
-    assertThat(dbExecution.getPort()).isEqualTo(3306);
+    assertThat(dbExecution.getNamespace()).isEqualTo("db");
+    assertThat(dbExecution.getServerAddress()).isEqualTo("localhost");
+    assertThat(dbExecution.getServerPort()).isEqualTo(3306);
     assertThat(dbExecution.getConnectionString()).isEqualTo("mariadb://localhost:3306");
     assertThat(dbExecution.getRawQueryText())
         .isEqualTo("SELECT * from person where last_name = 'tom'");
   }
+
+  @SuppressWarnings("deprecation") // testing deprecated semconv
+  @Test
+  void dbExecutionWithPool() {
+    QueryExecutionInfo queryExecutionInfo =
+        MockQueryExecutionInfo.builder()
+            .queryInfo(new QueryInfo("SELECT 1"))
+            .connectionInfo(MockConnectionInfo.builder().build())
+            .build();
+    ConnectionFactoryOptions factoryOptions =
+        ConnectionFactoryOptions.parse("r2dbc:pool:postgresql://user:pass@dbhost:5432/mydb");
+    DbExecution dbExecution = new DbExecution(queryExecutionInfo, factoryOptions);
+    assertThat(dbExecution.getSystemName()).isEqualTo("postgresql");
+    assertThat(dbExecution.getSystem()).isEqualTo("other_sql");
+    assertThat(dbExecution.getUser()).isEqualTo("user");
+    assertThat(dbExecution.getNamespace()).isEqualTo("mydb");
+    assertThat(dbExecution.getServerAddress()).isEqualTo("dbhost");
+    assertThat(dbExecution.getServerPort()).isEqualTo(5432);
+    assertThat(dbExecution.getConnectionString()).isEqualTo("pool:postgresql://dbhost:5432");
+  }
+
+  @ParameterizedTest
+  @CsvSource({
+    "r2dbc:postgresql://localhost/db, postgresql",
+    "r2dbc:mysql://localhost/db, mysql",
+    "r2dbc:mariadb://localhost/db, mariadb",
+    "r2dbc:mssql://localhost/db, microsoft.sql_server",
+    "r2dbc:oracle://localhost/db, oracle.db",
+    "r2dbc:h2:mem:///testdb, h2database",
+    "r2dbc:unknown://localhost/db, other_sql",
+  })
+  void dbSystemName(String url, String expectedSystemName) {
+    QueryExecutionInfo queryExecutionInfo =
+        MockQueryExecutionInfo.builder()
+            .queryInfo(new QueryInfo("SELECT 1"))
+            .connectionInfo(MockConnectionInfo.builder().build())
+            .build();
+    ConnectionFactoryOptions factoryOptions = ConnectionFactoryOptions.parse(url);
+    DbExecution dbExecution = new DbExecution(queryExecutionInfo, factoryOptions);
+    assertThat(dbExecution.getSystemName()).isEqualTo(expectedSystemName);
+  }
 }