Instrument cassandra executeReactive method

instrumentation/cassandra/cassandra-4.0/javaagent/build.gradle.kts

@@ -12,6 +12,8 @@ muzzle {
 }
 
 dependencies {
+  implementation(project(":instrumentation:cassandra:cassandra-4.0:library"))
+
   library("com.datastax.oss:java-driver-core:4.0.0")
 
   compileOnly("com.google.auto.value:auto-value-annotations")

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

@@ -8,37 +8,22 @@
 import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
-import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.db.DbClientSpanNameExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.db.SqlClientAttributesExtractor;
-import io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor;
+import io.opentelemetry.instrumentation.cassandra.v4_0.CassandraRequest;
+import io.opentelemetry.instrumentation.cassandra.v4_0.CassandraTelemetry;
 import io.opentelemetry.javaagent.bootstrap.internal.CommonConfig;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 
 public final class CassandraSingletons {
-  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.cassandra-4.0";
-
   // using ExecutionInfo because we can get that from ResultSet, AsyncResultSet and DriverException
   private static final Instrumenter<CassandraRequest, ExecutionInfo> INSTRUMENTER;
 
   static {
-    CassandraSqlAttributesGetter attributesGetter = new CassandraSqlAttributesGetter();
+    CassandraTelemetry telemetry =
+        CassandraTelemetry.builder(GlobalOpenTelemetry.get())
+            .setInstrumentationName("io.opentelemetry.cassandra-4.0")
+            .setStatementSanitizationEnabled(CommonConfig.get().isStatementSanitizationEnabled())
+            .build();
 
-    INSTRUMENTER =
-        Instrumenter.<CassandraRequest, ExecutionInfo>builder(
-                GlobalOpenTelemetry.get(),
-                INSTRUMENTATION_NAME,
-                DbClientSpanNameExtractor.create(attributesGetter))
-            .addAttributesExtractor(
-                SqlClientAttributesExtractor.builder(attributesGetter)
-                    .setTableAttribute(SemanticAttributes.DB_CASSANDRA_TABLE)
-                    .setStatementSanitizationEnabled(
-                        CommonConfig.get().isStatementSanitizationEnabled())
-                    .build())
-            .addAttributesExtractor(
-                NetClientAttributesExtractor.create(new CassandraNetAttributesGetter()))
-            .addAttributesExtractor(new CassandraAttributesExtractor())
-            .buildInstrumenter(SpanKindExtractor.alwaysClient());
+    INSTRUMENTER = telemetry.getInstrumenter();
   }
 
   public static Instrumenter<CassandraRequest, ExecutionInfo> instrumenter() {

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

@@ -5,7 +5,10 @@
 
 package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
 
+import static io.opentelemetry.javaagent.instrumentation.cassandra.v4_0.CassandraSingletons.instrumenter;
+
 import com.datastax.oss.driver.api.core.CqlSession;
+import io.opentelemetry.instrumentation.cassandra.v4_0.TracingCqlSession;
 import java.util.function.Function;
 
 public class CompletionStageFunction implements Function<Object, Object> {
@@ -19,6 +22,6 @@ public Object apply(Object session) {
     if (session.getClass().getName().endsWith("cassandra4.TracingCqlSession")) {
       return session;
     }
-    return new TracingCqlSession((CqlSession) session);
+    return new TracingCqlSession((CqlSession) session, instrumenter());
   }
 }

instrumentation/cassandra/cassandra-4.0/library/build.gradle.kts

@@ -0,0 +1,16 @@
+plugins {
+  id("otel.library-instrumentation")
+}
+
+dependencies {
+  library("com.datastax.oss:java-driver-core:4.0.0")
+
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+}
+
+tasks {
+  test {
+    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].getService())
+  }
+}

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

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
+package io.opentelemetry.instrumentation.cassandra.v4_0;
 
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
@@ -16,7 +16,7 @@
 import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 import javax.annotation.Nullable;
 
-final class CassandraAttributesExtractor
+public final class CassandraAttributesExtractor
     implements AttributesExtractor<CassandraRequest, ExecutionInfo> {
 
   @Override

instrumentation/cassandra/cassandra-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_0/CassandraNetAttributesGetter.java → instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/CassandraNetAttributesGetter.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
+package io.opentelemetry.instrumentation.cassandra.v4_0;
 
 import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
 import com.datastax.oss.driver.api.core.metadata.Node;
@@ -12,7 +12,7 @@
 import java.net.SocketAddress;
 import javax.annotation.Nullable;
 
-final class CassandraNetAttributesGetter
+public final class CassandraNetAttributesGetter
     extends InetSocketAddressNetClientAttributesGetter<CassandraRequest, ExecutionInfo> {
 
   @Override

instrumentation/cassandra/cassandra-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_0/CassandraRequest.java → instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/CassandraRequest.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
+package io.opentelemetry.instrumentation.cassandra.v4_0;
 
 import com.datastax.oss.driver.api.core.session.Session;
 import com.google.auto.value.AutoValue;

instrumentation/cassandra/cassandra-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_0/CassandraSqlAttributesGetter.java → instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/CassandraSqlAttributesGetter.java

@@ -3,14 +3,15 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
+package io.opentelemetry.instrumentation.cassandra.v4_0;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import io.opentelemetry.instrumentation.api.instrumenter.db.SqlClientAttributesGetter;
 import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 import javax.annotation.Nullable;
 
-final class CassandraSqlAttributesGetter implements SqlClientAttributesGetter<CassandraRequest> {
+public final class CassandraSqlAttributesGetter
+    implements SqlClientAttributesGetter<CassandraRequest> {
 
   @Override
   public String system(CassandraRequest request) {

instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/CassandraTelemetry.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.cassandra.v4_0;
+
+import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.db.DbClientSpanNameExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.db.SqlClientAttributesExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+
+public final class CassandraTelemetry {
+
+  public static CassandraTelemetryBuilder builder(OpenTelemetry openTelemetry) {
+    return new CassandraTelemetryBuilder(openTelemetry);
+  }
+
+  private final Instrumenter<CassandraRequest, ExecutionInfo> instrumenter;
+
+  CassandraTelemetry(
+      OpenTelemetry openTelemetry,
+      String instrumentationName,
+      boolean statementSanitizationEnabled) {
+    this.instrumenter =
+        createInstrumenter(openTelemetry, instrumentationName, statementSanitizationEnabled);
+  }
+
+  private static Instrumenter<CassandraRequest, ExecutionInfo> createInstrumenter(
+      OpenTelemetry openTelemetry, String instrumenterName, boolean statementSanitizationEnabled) {
+    CassandraSqlAttributesGetter attributesGetter = new CassandraSqlAttributesGetter();
+
+    return Instrumenter.<CassandraRequest, ExecutionInfo>builder(
+            openTelemetry, instrumenterName, DbClientSpanNameExtractor.create(attributesGetter))
+        .addAttributesExtractor(
+            SqlClientAttributesExtractor.builder(attributesGetter)
+                .setTableAttribute(SemanticAttributes.DB_CASSANDRA_TABLE)
+                .setStatementSanitizationEnabled(statementSanitizationEnabled)
+                .build())
+        .addAttributesExtractor(
+            NetClientAttributesExtractor.create(new CassandraNetAttributesGetter()))
+        .addAttributesExtractor(new CassandraAttributesExtractor())
+        .buildInstrumenter(SpanKindExtractor.alwaysClient());
+  }
+
+  public Instrumenter<CassandraRequest, ExecutionInfo> getInstrumenter() {
+    return instrumenter;
+  }
+}

instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/CassandraTelemetryBuilder.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.cassandra.v4_0;
+
+import io.opentelemetry.api.OpenTelemetry;
+
+public class CassandraTelemetryBuilder {
+
+  static final String DEFAULT_INSTRUMENTATION_NAME = "io.opentelemetry.cassandra";
+
+  private final OpenTelemetry openTelemetry;
+
+  private String instrumentationName = DEFAULT_INSTRUMENTATION_NAME;
+  private boolean statementSanitizationEnabled;
+
+  public CassandraTelemetryBuilder(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  public CassandraTelemetryBuilder setInstrumentationName(String instrumentationName) {
+    this.instrumentationName = instrumentationName;
+    return this;
+  }
+
+  public CassandraTelemetryBuilder setStatementSanitizationEnabled(boolean enabled) {
+    this.statementSanitizationEnabled = enabled;
+    return this;
+  }
+
+  public CassandraTelemetry build() {
+    return new CassandraTelemetry(openTelemetry, instrumentationName, statementSanitizationEnabled);
+  }
+}

instrumentation/cassandra/cassandra-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_0/TracingCqlSession.java → instrumentation/cassandra/cassandra-4.0/library/src/main/java/io/opentelemetry/instrumentation/cassandra/v4_0/TracingCqlSession.java

@@ -3,9 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.javaagent.instrumentation.cassandra.v4_0;
-
-import static io.opentelemetry.javaagent.instrumentation.cassandra.v4_0.CassandraSingletons.instrumenter;
+package io.opentelemetry.instrumentation.cassandra.v4_0;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
@@ -25,6 +23,7 @@
 import com.datastax.oss.driver.api.core.type.reflect.GenericType;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -33,9 +32,12 @@
 
 public class TracingCqlSession implements CqlSession {
   private final CqlSession session;
+  private final Instrumenter<CassandraRequest, ExecutionInfo> instrumenter;
 
-  public TracingCqlSession(CqlSession session) {
+  public TracingCqlSession(
+      CqlSession session, Instrumenter<CassandraRequest, ExecutionInfo> instrumenter) {
     this.session = session;
+    this.instrumenter = instrumenter;
   }
 
   @Override
@@ -158,31 +160,31 @@ public <REQUEST extends Request, RESULT> RESULT execute(
   @Override
   public ResultSet execute(String query) {
     CassandraRequest request = CassandraRequest.create(session, query);
-    Context context = instrumenter().start(Context.current(), request);
+    Context context = instrumenter.start(Context.current(), request);
     ResultSet resultSet;
     try (Scope ignored = context.makeCurrent()) {
       resultSet = session.execute(query);
     } catch (RuntimeException e) {
-      instrumenter().end(context, request, getExecutionInfo(e), e);
+      instrumenter.end(context, request, getExecutionInfo(e), e);
       throw e;
     }
-    instrumenter().end(context, request, resultSet.getExecutionInfo(), null);
+    instrumenter.end(context, request, resultSet.getExecutionInfo(), null);
     return resultSet;
   }
 
   @Override
   public ResultSet execute(Statement<?> statement) {
     String query = getQuery(statement);
     CassandraRequest request = CassandraRequest.create(session, query);
-    Context context = instrumenter().start(Context.current(), request);
+    Context context = instrumenter.start(Context.current(), request);
     ResultSet resultSet;
     try (Scope ignored = context.makeCurrent()) {
       resultSet = session.execute(statement);
     } catch (RuntimeException e) {
-      instrumenter().end(context, request, getExecutionInfo(e), e);
+      instrumenter.end(context, request, getExecutionInfo(e), e);
       throw e;
     }
-    instrumenter().end(context, request, resultSet.getExecutionInfo(), null);
+    instrumenter.end(context, request, resultSet.getExecutionInfo(), null);
     return resultSet;
   }
 
@@ -199,21 +201,17 @@ public CompletionStage<AsyncResultSet> executeAsync(String query) {
     return executeAsync(request, () -> session.executeAsync(query));
   }
 
-  private static CompletionStage<AsyncResultSet> executeAsync(
+  private CompletionStage<AsyncResultSet> executeAsync(
       CassandraRequest request, Supplier<CompletionStage<AsyncResultSet>> query) {
     Context parentContext = Context.current();
-    Context context = instrumenter().start(parentContext, request);
+    Context context = instrumenter.start(parentContext, request);
     try (Scope ignored = context.makeCurrent()) {
       CompletionStage<AsyncResultSet> stage = query.get();
       return wrap(
           stage.whenComplete(
               (asyncResultSet, throwable) ->
-                  instrumenter()
-                      .end(
-                          context,
-                          request,
-                          getExecutionInfo(asyncResultSet, throwable),
-                          throwable)),
+                  instrumenter.end(
+                      context, request, getExecutionInfo(asyncResultSet, throwable), throwable)),
           parentContext);
     }
   }

instrumentation/cassandra/cassandra-4.14/javaagent/build.gradle.kts

@@ -0,0 +1,29 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("com.datastax.oss")
+    module.set("java-driver-core")
+    versions.set("[4.14,)")
+    assertInverse.set(true)
+  }
+}
+
+dependencies {
+  implementation(project(":instrumentation:cassandra:cassandra-4.0:library"))
+
+  library("com.datastax.oss:java-driver-core:4.14.1")
+
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+
+  testImplementation("io.projectreactor:reactor-core:3.4.21")
+}
+
+tasks {
+  test {
+    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].getService())
+  }
+}

instrumentation/cassandra/cassandra-4.14/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/cassandra/v4_14/CassandraClientInstrumentationModule.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.cassandra.v4_14;
+
+import static java.util.Collections.singletonList;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import java.util.List;
+
+@AutoService(InstrumentationModule.class)
+public class CassandraClientInstrumentationModule extends InstrumentationModule {
+  public CassandraClientInstrumentationModule() {
+    super("cassandra", "cassandra-4.14");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return singletonList(new SessionBuilderInstrumentation());
+  }
+}