Fix Cassandra target

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.cassandra.v3_0;
+
+import com.datastax.driver.core.ExecutionInfo;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import io.opentelemetry.semconv.SemanticAttributes;
+import javax.annotation.Nullable;
+
+public class CassandraAttributesExtractor
+    implements AttributesExtractor<CassandraRequest, ExecutionInfo> {
+  @Override
+  public void onStart(AttributesBuilder attributes, Context context, CassandraRequest request) {}
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      CassandraRequest request,
+      @Nullable ExecutionInfo executionInfo,
+      @Nullable Throwable error) {
+    if (executionInfo == null) {
+      return;
+    }
+    attributes.put(
+        SemanticAttributes.SERVER_ADDRESS,
+        executionInfo.getQueriedHost().getSocketAddress().getHostName());
+    attributes.put(
+        SemanticAttributes.SERVER_PORT,
+        executionInfo.getQueriedHost().getSocketAddress().getPort());
+  }
+}

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

@@ -38,6 +38,7 @@ public final class CassandraSingletons {
                     .build())
             .addAttributesExtractor(
                 NetworkAttributesExtractor.create(new CassandraNetworkAttributesGetter()))
+            .addAttributesExtractor(new CassandraAttributesExtractor())
             .buildInstrumenter(SpanKindExtractor.alwaysClient());
   }
 

instrumentation/cassandra/cassandra-3.0/javaagent/src/test/java/CassandraClientTest.java

@@ -86,6 +86,8 @@ void syncTest(Parameter parameter) {
                           .hasNoParent()
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),
@@ -99,6 +101,8 @@ void syncTest(Parameter parameter) {
                           .hasNoParent()
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),
@@ -116,6 +120,8 @@ void syncTest(Parameter parameter) {
                           .hasNoParent()
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),
@@ -153,6 +159,8 @@ void asyncTest(Parameter parameter) {
                           .hasNoParent()
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),
@@ -167,6 +175,8 @@ void asyncTest(Parameter parameter) {
                           .hasParent(trace.getSpan(0))
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),
@@ -189,6 +199,8 @@ void asyncTest(Parameter parameter) {
                           .hasParent(trace.getSpan(0))
                           .hasAttributesSatisfyingExactly(
                               equalTo(SemanticAttributes.NETWORK_TYPE, "ipv4"),
+                              equalTo(SemanticAttributes.SERVER_ADDRESS, "localhost"),
+                              equalTo(SemanticAttributes.SERVER_PORT, cassandraPort),
                               equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
                               equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                               equalTo(SemanticAttributes.DB_SYSTEM, "cassandra"),

instrumentation/cassandra/cassandra-4-common/testing/src/main/java/io/opentelemetry/cassandra/v4/common/AbstractCassandraTest.java

@@ -19,6 +19,9 @@
 import static io.opentelemetry.semconv.SemanticAttributes.DB_STATEMENT;
 import static io.opentelemetry.semconv.SemanticAttributes.DB_SYSTEM;
 import static io.opentelemetry.semconv.SemanticAttributes.NETWORK_TYPE;
+import static io.opentelemetry.semconv.SemanticAttributes.SERVER_ADDRESS;
+import static io.opentelemetry.semconv.SemanticAttributes.SERVER_PORT;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Named.named;
 
 import com.datastax.oss.driver.api.core.CqlSession;
@@ -90,8 +93,20 @@ void syncTest(Parameter parameter) {
                             .hasKind(SpanKind.CLIENT)
                             .hasNoParent()
                             .hasAttributesSatisfyingExactly(
-                                equalTo(NETWORK_TYPE, "ipv4"),
-                                equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
+                                satisfies(
+                                    NETWORK_TYPE,
+                                    val ->
+                                        val.satisfiesAnyOf(
+                                            v -> assertThat(v).isEqualTo("ipv4"),
+                                            v -> assertThat(v).isEqualTo("ipv6"))),
+                                equalTo(SERVER_ADDRESS, "localhost"),
+                                equalTo(SERVER_PORT, cassandraPort),
+                                satisfies(
+                                    NetworkAttributes.NETWORK_PEER_ADDRESS,
+                                    val ->
+                                        val.satisfiesAnyOf(
+                                            v -> assertThat(v).isEqualTo("127.0.0.1"),
+                                            v -> assertThat(v).isEqualTo("0:0:0:0:0:0:0:1"))),
                                 equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                                 equalTo(DB_SYSTEM, "cassandra"),
                                 equalTo(DB_NAME, parameter.keyspace),
@@ -137,8 +152,20 @@ void asyncTest(Parameter parameter) throws Exception {
                             .hasKind(SpanKind.CLIENT)
                             .hasParent(trace.getSpan(0))
                             .hasAttributesSatisfyingExactly(
-                                equalTo(NETWORK_TYPE, "ipv4"),
-                                equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
+                                satisfies(
+                                    NETWORK_TYPE,
+                                    val ->
+                                        val.satisfiesAnyOf(
+                                            v -> assertThat(v).isEqualTo("ipv4"),
+                                            v -> assertThat(v).isEqualTo("ipv6"))),
+                                equalTo(SERVER_ADDRESS, "localhost"),
+                                equalTo(SERVER_PORT, cassandraPort),
+                                satisfies(
+                                    NetworkAttributes.NETWORK_PEER_ADDRESS,
+                                    val ->
+                                        val.satisfiesAnyOf(
+                                            v -> assertThat(v).isEqualTo("127.0.0.1"),
+                                            v -> assertThat(v).isEqualTo("0:0:0:0:0:0:0:1"))),
                                 equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                                 equalTo(DB_SYSTEM, "cassandra"),
                                 equalTo(DB_NAME, parameter.keyspace),

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

@@ -14,6 +14,8 @@
 import io.opentelemetry.context.Context;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import io.opentelemetry.semconv.SemanticAttributes;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import javax.annotation.Nullable;
 
 final class CassandraAttributesExtractor
@@ -36,6 +38,12 @@ public void onEnd(
 
     Node coordinator = executionInfo.getCoordinator();
     if (coordinator != null) {
+      SocketAddress address = coordinator.getEndPoint().resolve();
+      if (address instanceof InetSocketAddress) {
+        attributes.put(
+            SemanticAttributes.SERVER_ADDRESS, ((InetSocketAddress) address).getHostName());
+        attributes.put(SemanticAttributes.SERVER_PORT, ((InetSocketAddress) address).getPort());
+      }
       if (coordinator.getDatacenter() != null) {
         attributes.put(SemanticAttributes.DB_CASSANDRA_COORDINATOR_DC, coordinator.getDatacenter());
       }

instrumentation/cassandra/cassandra-4.4/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_4/CassandraAttributesExtractor.java

@@ -9,16 +9,26 @@
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
 import com.datastax.oss.driver.api.core.cql.Statement;
+import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.api.core.metadata.Node;
+import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.metadata.SniEndPoint;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 import io.opentelemetry.semconv.SemanticAttributes;
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
 import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 final class CassandraAttributesExtractor
     implements AttributesExtractor<CassandraRequest, ExecutionInfo> {
 
+  private static final Logger logger = LoggerFactory.getLogger(CassandraAttributesExtractor.class);
+  private static final Field proxyAddressField = getProxyAddressField();
+
   @Override
   public void onStart(
       AttributesBuilder attributes, Context parentContext, CassandraRequest request) {}
@@ -36,6 +46,8 @@ public void onEnd(
 
     Node coordinator = executionInfo.getCoordinator();
     if (coordinator != null) {
+      updateServerAddressAndPort(attributes, coordinator);
+
       if (coordinator.getDatacenter() != null) {
         attributes.put(SemanticAttributes.DB_CASSANDRA_COORDINATOR_DC, coordinator.getDatacenter());
       }
@@ -74,4 +86,39 @@ public void onEnd(
     }
     attributes.put(SemanticAttributes.DB_CASSANDRA_IDEMPOTENCE, idempotent);
   }
+
+  private static void updateServerAddressAndPort(AttributesBuilder attributes, Node coordinator) {
+    EndPoint endPoint = coordinator.getEndPoint();
+    if (endPoint instanceof DefaultEndPoint) {
+      InetSocketAddress address = ((DefaultEndPoint) endPoint).resolve();
+      attributes.put(SemanticAttributes.SERVER_ADDRESS, address.getHostName());
+      attributes.put(SemanticAttributes.SERVER_PORT, address.getPort());
+    } else if (endPoint instanceof SniEndPoint && proxyAddressField != null) {
+      SniEndPoint sniEndPoint = (SniEndPoint) endPoint;
+      Object object = null;
+      try {
+        object = proxyAddressField.get(sniEndPoint);
+      } catch (IllegalAccessException e) {
+        logger.warn(
+            "Error when retrieving the private field proxyAddress of SniEndPoint using reflection.",
+            e);
+      }
+      if (object instanceof InetSocketAddress) {
+        InetSocketAddress address = (InetSocketAddress) object;
+        attributes.put(SemanticAttributes.SERVER_ADDRESS, address.getHostName());
+        attributes.put(SemanticAttributes.SERVER_PORT, address.getPort());
+      }
+    }
+  }
+
+  @Nullable
+  private static Field getProxyAddressField() {
+    try {
+      Field field = SniEndPoint.class.getDeclaredField("proxyAddress");
+      field.setAccessible(true);
+      return field;
+    } catch (Exception e) {
+      return null;
+    }
+  }
 }

instrumentation/cassandra/cassandra-4.4/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_4/CassandraNetworkAttributesGetter.java

@@ -6,10 +6,12 @@
 package io.opentelemetry.instrumentation.cassandra.v4_4;
 
 import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
+import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.api.core.metadata.Node;
+import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.metadata.SniEndPoint;
 import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
 import java.net.InetSocketAddress;
-import java.net.SocketAddress;
 import javax.annotation.Nullable;
 
 final class CassandraNetworkAttributesGetter
@@ -27,8 +29,12 @@ public InetSocketAddress getNetworkPeerInetSocketAddress(
       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;
+    EndPoint endPoint = coordinator.getEndPoint();
+    if (endPoint instanceof DefaultEndPoint) {
+      return (InetSocketAddress) coordinator.getEndPoint().resolve();
+    } else if (endPoint instanceof SniEndPoint) {
+      return ((SniEndPoint) endPoint).resolve();
+    }
+    return null;
   }
 }

instrumentation/cassandra/cassandra-4.4/testing/src/main/java/io/opentelemetry/testing/cassandra/v4_4/AbstractCassandra44Test.java

@@ -19,12 +19,19 @@
 import static io.opentelemetry.semconv.SemanticAttributes.DB_STATEMENT;
 import static io.opentelemetry.semconv.SemanticAttributes.DB_SYSTEM;
 import static io.opentelemetry.semconv.SemanticAttributes.NETWORK_TYPE;
+import static io.opentelemetry.semconv.SemanticAttributes.SERVER_ADDRESS;
+import static io.opentelemetry.semconv.SemanticAttributes.SERVER_PORT;
 import static org.junit.jupiter.api.Named.named;
 
 import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
+import com.datastax.oss.driver.internal.core.metadata.SniEndPoint;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.cassandra.v4.common.AbstractCassandraTest;
 import io.opentelemetry.instrumentation.api.semconv.network.internal.NetworkAttributes;
+import java.net.InetSocketAddress;
+import java.time.Duration;
 import java.util.stream.Stream;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -57,7 +64,9 @@ void reactiveTest(Parameter parameter) {
                             .hasParent(trace.getSpan(0))
                             .hasAttributesSatisfyingExactly(
                                 equalTo(NETWORK_TYPE, "ipv4"),
-                                equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "127.0.0.1"),
+                                equalTo(SERVER_ADDRESS, "localhost"),
+                                equalTo(SERVER_PORT, cassandraPort),
+                                equalTo(NetworkAttributes.NETWORK_PEER_ADDRESS, "0:0:0:0:0:0:0:1"),
                                 equalTo(NetworkAttributes.NETWORK_PEER_PORT, cassandraPort),
                                 equalTo(DB_SYSTEM, "cassandra"),
                                 equalTo(DB_NAME, parameter.keyspace),
@@ -135,4 +144,22 @@ private static Stream<Arguments> provideReactiveParameters() {
                     "SELECT",
                     "users"))));
   }
+
+  @Override
+  protected CqlSession getSession(String keyspace) {
+    DriverConfigLoader configLoader =
+        DriverConfigLoader.programmaticBuilder()
+            .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(0))
+            .withDuration(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, Duration.ofSeconds(10))
+            .build();
+
+    InetSocketAddress address = new InetSocketAddress("localhost", cassandraPort);
+    return wrap(
+        CqlSession.builder()
+            .addContactEndPoint(new SniEndPoint(address, "localhost"))
+            .withConfigLoader(configLoader)
+            .withLocalDatacenter("datacenter1")
+            .withKeyspace(keyspace)
+            .build());
+  }
 }