Consolidate network attribute extraction into DbClientAttributesGetter (#15149)

Author: trask

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

@@ -21,6 +21,8 @@
 import io.opentelemetry.instrumentation.api.internal.SemconvStability;
 import io.opentelemetry.instrumentation.api.internal.SpanKey;
 import io.opentelemetry.instrumentation.api.internal.SpanKeyProvider;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesExtractor;
+import io.opentelemetry.instrumentation.api.semconv.network.internal.InternalNetworkAttributesExtractor;
 import javax.annotation.Nullable;
 
 /**
@@ -44,6 +46,8 @@ public final class DbClientAttributesExtractor<REQUEST, RESPONSE>
   private static final AttributeKey<String> DB_OPERATION = AttributeKey.stringKey("db.operation");
 
   private final DbClientAttributesGetter<REQUEST, RESPONSE> getter;
+  private final InternalNetworkAttributesExtractor<REQUEST, RESPONSE> internalNetworkExtractor;
+  private final ServerAttributesExtractor<REQUEST, RESPONSE> serverAttributesExtractor;
 
   /** Creates the database client attributes extractor with default configuration. */
   public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
@@ -53,12 +57,15 @@ public static <REQUEST, RESPONSE> AttributesExtractor<REQUEST, RESPONSE> create(
 
   DbClientAttributesExtractor(DbClientAttributesGetter<REQUEST, RESPONSE> getter) {
     this.getter = getter;
+    internalNetworkExtractor = new InternalNetworkAttributesExtractor<>(getter, true, false);
+    serverAttributesExtractor = ServerAttributesExtractor.create(getter);
   }
 
   @SuppressWarnings("deprecation") // until old db semconv are dropped
   @Override
   public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST request) {
     onStartCommon(attributes, getter, request);
+    serverAttributesExtractor.onStart(attributes, parentContext, request);
   }
 
   @SuppressWarnings("deprecation") // until old db semconv are dropped
@@ -93,6 +100,7 @@ public void onEnd(
       REQUEST request,
       @Nullable RESPONSE response,
       @Nullable Throwable error) {
+    internalNetworkExtractor.onEnd(attributes, request, response);
     onEndCommon(attributes, getter, response, error);
   }
 

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

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.instrumentation.api.incubator.semconv.db;
 
+import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesGetter;
 import javax.annotation.Nullable;
 
 /**
@@ -18,9 +20,11 @@
  * from the attribute methods, but implement as many as possible for best compliance with the
  * OpenTelemetry specification.
  */
-@SuppressWarnings("deprecation") // extending deprecated interface for backward compatibility
+@SuppressWarnings("deprecation") // until DbClientCommonAttributesGetter is removed
 public interface DbClientAttributesGetter<REQUEST, RESPONSE>
-    extends DbClientCommonAttributesGetter<REQUEST, RESPONSE> {
+    extends DbClientCommonAttributesGetter<REQUEST, RESPONSE>,
+        NetworkAttributesGetter<REQUEST, RESPONSE>,
+        ServerAttributesGetter<REQUEST> {
 
   /**
    * @deprecated Use {@link #getDbQueryText(REQUEST)} instead.

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

@@ -18,6 +18,8 @@
 import io.opentelemetry.instrumentation.api.internal.SemconvStability;
 import io.opentelemetry.instrumentation.api.internal.SpanKey;
 import io.opentelemetry.instrumentation.api.internal.SpanKeyProvider;
+import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesExtractor;
+import io.opentelemetry.instrumentation.api.semconv.network.internal.InternalNetworkAttributesExtractor;
 import io.opentelemetry.semconv.AttributeKeyTemplate;
 import java.util.Collection;
 import java.util.Map;
@@ -60,6 +62,8 @@ public static <REQUEST, RESPONSE> SqlClientAttributesExtractorBuilder<REQUEST, R
   private static final String SQL_CALL = "CALL";
 
   private final SqlClientAttributesGetter<REQUEST, RESPONSE> getter;
+  private final InternalNetworkAttributesExtractor<REQUEST, RESPONSE> internalNetworkExtractor;
+  private final ServerAttributesExtractor<REQUEST, RESPONSE> serverAttributesExtractor;
   private final AttributeKey<String> oldSemconvTableAttribute;
   private final boolean statementSanitizationEnabled;
   private final boolean captureQueryParameters;
@@ -74,6 +78,8 @@ public static <REQUEST, RESPONSE> SqlClientAttributesExtractorBuilder<REQUEST, R
     // capturing query parameters disables statement sanitization
     this.statementSanitizationEnabled = !captureQueryParameters && statementSanitizationEnabled;
     this.captureQueryParameters = captureQueryParameters;
+    internalNetworkExtractor = new InternalNetworkAttributesExtractor<>(getter, true, false);
+    serverAttributesExtractor = ServerAttributesExtractor.create(getter);
   }
 
   @SuppressWarnings("deprecation") // until old db semconv are dropped
@@ -139,6 +145,7 @@ public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST
     // getDbQueryText(), and getDbQuerySummary() implementations can override
     // the parsed values from above
     DbClientAttributesExtractor.onStartCommon(attributes, getter, request);
+    serverAttributesExtractor.onStart(attributes, parentContext, request);
   }
 
   private void setQueryParameters(
@@ -171,6 +178,7 @@ public void onEnd(
       REQUEST request,
       @Nullable RESPONSE response,
       @Nullable Throwable error) {
+    internalNetworkExtractor.onEnd(attributes, request, response);
     DbClientAttributesExtractor.onEndCommon(attributes, getter, response, error);
   }
 

instrumentation/cassandra/cassandra-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v3_0/CassandraNetworkAttributesGetter.java

@@ -12,7 +12,6 @@
 import io.opentelemetry.instrumentation.api.incubator.semconv.db.SqlClientAttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
-import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesExtractor;
 import io.opentelemetry.javaagent.bootstrap.internal.AgentCommonConfig;
 import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
 
@@ -38,8 +37,6 @@ public final class CassandraSingletons {
                     .setStatementSanitizationEnabled(
                         AgentCommonConfig.get().isStatementSanitizationEnabled())
                     .build())
-            .addAttributesExtractor(
-                NetworkAttributesExtractor.create(new CassandraNetworkAttributesGetter()))
             .addAttributesExtractor(new CassandraAttributesExtractor())
             .addOperationMetrics(DbClientMetrics.get())
             .buildInstrumenter(SpanKindExtractor.alwaysClient());

instrumentation/cassandra/cassandra-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v3_0/CassandraSingletons.java

@@ -10,6 +10,7 @@
 import com.datastax.driver.core.ExecutionInfo;
 import io.opentelemetry.instrumentation.api.incubator.semconv.db.SqlClientAttributesGetter;
 import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
+import java.net.InetSocketAddress;
 import java.util.Collection;
 import javax.annotation.Nullable;
 
@@ -22,28 +23,21 @@ public String getDbSystem(CassandraRequest request) {
     return DbIncubatingAttributes.DbSystemIncubatingValues.CASSANDRA;
   }
 
-  @Deprecated
-  @Override
-  @Nullable
-  public String getUser(CassandraRequest request) {
-    return null;
-  }
-
   @Override
   @Nullable
   public String getDbNamespace(CassandraRequest request) {
     return request.getSession().getLoggedKeyspace();
   }
 
-  @Deprecated
   @Override
-  @Nullable
-  public String getConnectionString(CassandraRequest request) {
-    return null;
+  public Collection<String> getRawQueryTexts(CassandraRequest request) {
+    return singleton(request.getQueryText());
   }
 
+  @Nullable
   @Override
-  public Collection<String> getRawQueryTexts(CassandraRequest request) {
-    return singleton(request.getQueryText());
+  public InetSocketAddress getNetworkPeerInetSocketAddress(
+      CassandraRequest request, @Nullable ExecutionInfo executionInfo) {
+    return executionInfo == null ? null : executionInfo.getQueriedHost().getSocketAddress();
   }
 }

instrumentation/cassandra/cassandra-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v3_0/CassandraSqlAttributesGetter.java

@@ -12,7 +12,6 @@
 import io.opentelemetry.instrumentation.api.incubator.semconv.db.SqlClientAttributesExtractor;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
-import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesExtractor;
 import io.opentelemetry.javaagent.bootstrap.internal.AgentCommonConfig;
 import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
 
@@ -37,8 +36,6 @@ public final class CassandraSingletons {
                     .setStatementSanitizationEnabled(
                         AgentCommonConfig.get().isStatementSanitizationEnabled())
                     .build())
-            .addAttributesExtractor(
-                NetworkAttributesExtractor.create(new CassandraNetworkAttributesGetter()))
             .addAttributesExtractor(new CassandraAttributesExtractor())
             .addOperationMetrics(DbClientMetrics.get())
             .buildInstrumenter(SpanKindExtractor.alwaysClient());

instrumentation/cassandra/cassandra-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_0/CassandraNetworkAttributesGetter.java

@@ -9,8 +9,11 @@
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import io.opentelemetry.instrumentation.api.incubator.semconv.db.SqlClientAttributesGetter;
 import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.util.Collection;
 import javax.annotation.Nullable;
 
@@ -23,28 +26,31 @@ public String getDbSystem(CassandraRequest request) {
     return DbIncubatingAttributes.DbSystemIncubatingValues.CASSANDRA;
   }
 
-  @Deprecated
-  @Override
-  @Nullable
-  public String getUser(CassandraRequest request) {
-    return null;
-  }
-
   @Override
   @Nullable
   public String getDbNamespace(CassandraRequest request) {
     return request.getSession().getKeyspace().map(CqlIdentifier::toString).orElse(null);
   }
 
-  @Deprecated
   @Override
-  @Nullable
-  public String getConnectionString(CassandraRequest request) {
-    return null;
+  public Collection<String> getRawQueryTexts(CassandraRequest request) {
+    return singleton(request.getQueryText());
   }
 
+  @Nullable
   @Override
-  public Collection<String> getRawQueryTexts(CassandraRequest request) {
-    return singleton(request.getQueryText());
+  public InetSocketAddress getNetworkPeerInetSocketAddress(
+      CassandraRequest request, @Nullable ExecutionInfo executionInfo) {
+    if (executionInfo == null) {
+      return null;
+    }
+    Node coordinator = executionInfo.getCoordinator();
+    if (coordinator == null) {
+      return null;
+    }
+    // resolve() returns an existing InetSocketAddress, it does not do a dns resolve,
+    // at least in the only current EndPoint implementation (DefaultEndPoint)
+    SocketAddress address = coordinator.getEndPoint().resolve();
+    return address instanceof InetSocketAddress ? (InetSocketAddress) address : null;
   }
 }