Refactor couchbase and hystrix to Instrumenter API (#3802)

instrumentation/couchbase/couchbase-2.0/javaagent/build.gradle.kts

@@ -28,6 +28,9 @@ muzzle {
 }
 
 dependencies {
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+
   implementation(project(":instrumentation:rxjava:rxjava-1.0:library"))
 
   library("com.couchbase.client:java-client:2.0.0")

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseAttributesExtractor.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
+
+import io.opentelemetry.instrumentation.api.instrumenter.db.DbAttributesExtractor;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class CouchbaseAttributesExtractor extends DbAttributesExtractor<CouchbaseRequest, Void> {
+  @Override
+  protected String system(CouchbaseRequest couchbaseRequest) {
+    return SemanticAttributes.DbSystemValues.COUCHBASE;
+  }
+
+  @Override
+  protected @Nullable String user(CouchbaseRequest couchbaseRequest) {
+    return null;
+  }
+
+  @Override
+  protected @Nullable String name(CouchbaseRequest couchbaseRequest) {
+    return couchbaseRequest.bucket();
+  }
+
+  @Override
+  protected @Nullable String connectionString(CouchbaseRequest couchbaseRequest) {
+    return null;
+  }
+
+  @Override
+  protected @Nullable String statement(CouchbaseRequest couchbaseRequest) {
+    return couchbaseRequest.statement();
+  }
+
+  @Override
+  protected @Nullable String operation(CouchbaseRequest couchbaseRequest) {
+    return couchbaseRequest.operation();
+  }
+}

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseBucketInstrumentation.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
 
+import static io.opentelemetry.javaagent.instrumentation.couchbase.v2_0.CouchbaseSingletons.instrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -13,6 +14,7 @@ import static net.bytebuddy.matcher.ElementMatchers.not;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
 
 import com.couchbase.client.java.CouchbaseCluster;
+import io.opentelemetry.instrumentation.rxjava.TracedOnSubscribe;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import io.opentelemetry.javaagent.instrumentation.api.CallDepth;
@@ -59,7 +61,8 @@ public class CouchbaseBucketInstrumentation implements TypeInstrumentation {
       if (callDepth.decrementAndGet() > 0) {
         return;
       }
-      result = Observable.create(CouchbaseOnSubscribe.create(result, bucket, method));
+      CouchbaseRequest request = CouchbaseRequest.create(bucket, method);
+      result = Observable.create(new TracedOnSubscribe<>(result, instrumenter(), request));
     }
   }
 
@@ -83,15 +86,11 @@ public class CouchbaseBucketInstrumentation implements TypeInstrumentation {
         return;
       }
 
-      if (query != null) {
-        // A query can be of many different types. We could track the creation of them and try to
-        // rewind back to when they were created from a string, but for now we rely on toString()
-        // returning something useful. That seems to be the case. If we're starting to see strange
-        // query texts, this is the place to look!
-        result = Observable.create(CouchbaseOnSubscribe.create(result, bucket, query));
-      } else {
-        result = Observable.create(CouchbaseOnSubscribe.create(result, bucket, method));
-      }
+      CouchbaseRequest request =
+          query == null
+              ? CouchbaseRequest.create(bucket, method)
+              : CouchbaseRequest.create(bucket, query);
+      result = Observable.create(new TracedOnSubscribe<>(result, instrumenter(), request));
     }
   }
 }

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseClientTracer.java

@@ -1,52 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
-
-import io.opentelemetry.instrumentation.api.tracer.DatabaseClientTracer;
-import io.opentelemetry.instrumentation.api.tracer.net.NetPeerAttributes;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DbSystemValues;
-import java.lang.reflect.Method;
-import java.net.InetSocketAddress;
-
-public class CouchbaseClientTracer extends DatabaseClientTracer<Void, Method, Void> {
-  private static final CouchbaseClientTracer TRACER = new CouchbaseClientTracer();
-
-  private CouchbaseClientTracer() {
-    super(NetPeerAttributes.INSTANCE);
-  }
-
-  public static CouchbaseClientTracer tracer() {
-    return TRACER;
-  }
-
-  @Override
-  protected String spanName(Void connection, Method method, Void sanitizedStatement) {
-    Class<?> declaringClass = method.getDeclaringClass();
-    String className =
-        declaringClass.getSimpleName().replace("CouchbaseAsync", "").replace("DefaultAsync", "");
-    return className + "." + method.getName();
-  }
-
-  @Override
-  protected Void sanitizeStatement(Method method) {
-    return null;
-  }
-
-  @Override
-  protected String dbSystem(Void connection) {
-    return DbSystemValues.COUCHBASE;
-  }
-
-  @Override
-  protected InetSocketAddress peerAddress(Void connection) {
-    return null;
-  }
-
-  @Override
-  protected String getInstrumentationName() {
-    return "io.opentelemetry.couchbase-2.0";
-  }
-}

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseClusterInstrumentation.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
 
+import static io.opentelemetry.javaagent.instrumentation.couchbase.v2_0.CouchbaseSingletons.instrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -13,6 +14,7 @@ import static net.bytebuddy.matcher.ElementMatchers.not;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
 
 import com.couchbase.client.java.CouchbaseCluster;
+import io.opentelemetry.instrumentation.rxjava.TracedOnSubscribe;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import io.opentelemetry.javaagent.instrumentation.api.CallDepth;
@@ -56,7 +58,8 @@ public class CouchbaseClusterInstrumentation implements TypeInstrumentation {
         return;
       }
 
-      result = Observable.create(CouchbaseOnSubscribe.create(result, null, method));
+      CouchbaseRequest request = CouchbaseRequest.create(null, method);
+      result = Observable.create(new TracedOnSubscribe<>(result, instrumenter(), request));
     }
   }
 }

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseOnSubscribe.java

@@ -1,57 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
-
-import static io.opentelemetry.api.trace.SpanKind.CLIENT;
-import static io.opentelemetry.instrumentation.api.tracer.DatabaseClientTracer.conventionSpanName;
-import static io.opentelemetry.javaagent.instrumentation.couchbase.v2_0.CouchbaseClientTracer.tracer;
-
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.instrumentation.api.db.SqlStatementInfo;
-import io.opentelemetry.instrumentation.rxjava.TracedOnSubscribe;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DbSystemValues;
-import java.lang.reflect.Method;
-import rx.Observable;
-
-public class CouchbaseOnSubscribe<T> extends TracedOnSubscribe<T> {
-  private final String bucket;
-  private final String query;
-
-  public static <T> CouchbaseOnSubscribe<T> create(
-      Observable<T> originalObservable, String bucket, Method method) {
-    Class<?> declaringClass = method.getDeclaringClass();
-    String className =
-        declaringClass.getSimpleName().replace("CouchbaseAsync", "").replace("DefaultAsync", "");
-    String operation = className + "." + method.getName();
-    return new CouchbaseOnSubscribe<>(originalObservable, operation, bucket, operation);
-  }
-
-  public static <T> CouchbaseOnSubscribe<T> create(
-      Observable<T> originalObservable, String bucket, Object query) {
-    SqlStatementInfo statement = CouchbaseQuerySanitizer.sanitize(query);
-    String spanName =
-        conventionSpanName(
-            bucket, statement.getOperation(), statement.getTable(), statement.getFullStatement());
-    return new CouchbaseOnSubscribe<>(
-        originalObservable, spanName, bucket, statement.getFullStatement());
-  }
-
-  private CouchbaseOnSubscribe(
-      Observable<T> originalObservable, String spanName, String bucket, String query) {
-    super(originalObservable, spanName, tracer(), CLIENT);
-
-    this.bucket = bucket;
-    this.query = query;
-  }
-
-  @Override
-  protected void decorateSpan(Span span) {
-    span.setAttribute(SemanticAttributes.DB_SYSTEM, DbSystemValues.COUCHBASE);
-    span.setAttribute(SemanticAttributes.DB_NAME, bucket);
-    span.setAttribute(SemanticAttributes.DB_STATEMENT, query);
-  }
-}

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseRequest.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
+
+import com.google.auto.value.AutoValue;
+import io.opentelemetry.instrumentation.api.db.SqlStatementInfo;
+import java.lang.reflect.Method;
+import javax.annotation.Nullable;
+
+@AutoValue
+public abstract class CouchbaseRequest {
+
+  public static CouchbaseRequest create(@Nullable String bucket, Method method) {
+    Class<?> declaringClass = method.getDeclaringClass();
+    String className =
+        declaringClass.getSimpleName().replace("CouchbaseAsync", "").replace("DefaultAsync", "");
+    String operation = className + "." + method.getName();
+
+    return new AutoValue_CouchbaseRequest(bucket, null, operation, true);
+  }
+
+  public static CouchbaseRequest create(@Nullable String bucket, Object query) {
+    SqlStatementInfo statement = CouchbaseQuerySanitizer.sanitize(query);
+
+    return new AutoValue_CouchbaseRequest(
+        bucket, statement.getFullStatement(), statement.getOperation(), false);
+  }
+
+  @Nullable
+  public abstract String bucket();
+
+  @Nullable
+  public abstract String statement();
+
+  @Nullable
+  public abstract String operation();
+
+  public abstract boolean isMethodCall();
+}

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseSingletons.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
+
+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.SpanNameExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.db.DbSpanNameExtractor;
+
+public final class CouchbaseSingletons {
+
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.couchbase-2.0";
+
+  private static final Instrumenter<CouchbaseRequest, Void> INSTRUMENTER;
+
+  static {
+    CouchbaseAttributesExtractor couchbaseAttributesExtractor = new CouchbaseAttributesExtractor();
+    SpanNameExtractor<CouchbaseRequest> spanNameExtractor =
+        new CouchbaseSpanNameExtractor(DbSpanNameExtractor.create(couchbaseAttributesExtractor));
+
+    INSTRUMENTER =
+        Instrumenter.<CouchbaseRequest, Void>newBuilder(
+                GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME, spanNameExtractor)
+            .addAttributesExtractor(couchbaseAttributesExtractor)
+            .newInstrumenter(SpanKindExtractor.alwaysClient());
+  }
+
+  public static Instrumenter<CouchbaseRequest, Void> instrumenter() {
+    return INSTRUMENTER;
+  }
+
+  private CouchbaseSingletons() {}
+}

instrumentation/couchbase/couchbase-2.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/couchbase/v2_0/CouchbaseSpanNameExtractor.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.couchbase.v2_0;
+
+import io.opentelemetry.instrumentation.api.instrumenter.SpanNameExtractor;
+
+public class CouchbaseSpanNameExtractor implements SpanNameExtractor<CouchbaseRequest> {
+  private final SpanNameExtractor<CouchbaseRequest> dbSpanNameExtractor;
+
+  public CouchbaseSpanNameExtractor(SpanNameExtractor<CouchbaseRequest> dbSpanNameExtractor) {
+    this.dbSpanNameExtractor = dbSpanNameExtractor;
+  }
+
+  @Override
+  public String extract(CouchbaseRequest couchbaseRequest) {
+    if (couchbaseRequest.isMethodCall()) {
+      return couchbaseRequest.operation();
+    }
+    return dbSpanNameExtractor.extract(couchbaseRequest);
+  }
+}

instrumentation/couchbase/couchbase-2.6/javaagent/src/test/groovy/CouchbaseAsyncClient26Test.groovy

@@ -4,11 +4,18 @@
  */
 
 import io.opentelemetry.instrumentation.test.asserts.TraceAssert
+import io.opentelemetry.sdk.trace.data.SpanData
 
 class CouchbaseAsyncClient26Test extends AbstractCouchbaseAsyncClientTest {
 
   @Override
-  void assertCouchbaseCall(TraceAssert trace, int index, Object name, String bucketName = null, Object parentSpan = null, Object statement = null) {
-    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, bucketName, parentSpan, statement)
+  void assertCouchbaseCall(TraceAssert trace,
+                           int index,
+                           Object name,
+                           SpanData parentSpan = null,
+                           String bucketName = null,
+                           Object statement = null,
+                           Object operation = null) {
+    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, parentSpan, bucketName, statement, operation)
   }
 }

instrumentation/couchbase/couchbase-2.6/javaagent/src/test/groovy/CouchbaseClient26Test.groovy

@@ -4,10 +4,17 @@
  */
 
 import io.opentelemetry.instrumentation.test.asserts.TraceAssert
+import io.opentelemetry.sdk.trace.data.SpanData
 
 class CouchbaseClient26Test extends AbstractCouchbaseClientTest {
   @Override
-  void assertCouchbaseCall(TraceAssert trace, int index, Object name, String bucketName = null, Object parentSpan = null, Object statement = null) {
-    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, bucketName, parentSpan, statement)
+  void assertCouchbaseCall(TraceAssert trace,
+                           int index,
+                           Object name,
+                           SpanData parentSpan = null,
+                           String bucketName = null,
+                           Object statement = null,
+                           Object operation = null) {
+    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, parentSpan, bucketName, statement, operation)
   }
 }

instrumentation/couchbase/couchbase-2.6/javaagent/src/test/groovy/CouchbaseSpanUtil.groovy

@@ -12,7 +12,13 @@ import io.opentelemetry.semconv.trace.attributes.SemanticAttributes
 class CouchbaseSpanUtil {
   // Reusable span assertion method.  Cannot directly override AbstractCouchbaseTest.assertCouchbaseSpan because
   // Of the class hierarchy of these tests
-  static void assertCouchbaseCall(TraceAssert trace, int index, Object spanName, String bucketName = null, Object parentSpan = null, Object statement = null) {
+  static void assertCouchbaseCall(TraceAssert trace,
+                                  int index,
+                                  Object spanName,
+                                  SpanData parentSpan = null,
+                                  String bucketName = null,
+                                  Object statement = null,
+                                  Object operation = null) {
     trace.span(index) {
       name spanName
       kind CLIENT
@@ -22,16 +28,15 @@ class CouchbaseSpanUtil {
         childOf((SpanData) parentSpan)
       }
       attributes {
+        "${SemanticAttributes.DB_SYSTEM.key}" "couchbase"
+        "${SemanticAttributes.DB_NAME.key}" bucketName
+        "${SemanticAttributes.DB_STATEMENT.key}" statement
+        "${SemanticAttributes.DB_OPERATION.key}" (operation ?: spanName)
 
         // Because of caching, not all requests hit the server so these attributes may be absent
         "${SemanticAttributes.NET_PEER_NAME.key}" { it == "localhost" || it == "127.0.0.1" || it == null }
         "${SemanticAttributes.NET_PEER_PORT.key}" { it == null || Number }
 
-        "${SemanticAttributes.DB_SYSTEM.key}" "couchbase"
-        if (bucketName != null) {
-          "${SemanticAttributes.DB_NAME.key}" bucketName
-        }
-
         // Because of caching, not all requests hit the server so this tag may be absent
         "couchbase.local.address" { it == null || String }
 
@@ -39,8 +44,6 @@ class CouchbaseSpanUtil {
         // We assign a spanName of 'Bucket.query' and this is shared with n1ql queries
         // that do have operation ids
         "couchbase.operation_id" { it == null || String }
-
-        "${SemanticAttributes.DB_STATEMENT.key}"(statement ?: spanName)
       }
     }
   }

instrumentation/couchbase/couchbase-2.6/javaagent/src/test/groovy/springdata/CouchbaseSpringRepository26Test.groovy

@@ -6,11 +6,18 @@
 package springdata
 
 import io.opentelemetry.instrumentation.test.asserts.TraceAssert
+import io.opentelemetry.sdk.trace.data.SpanData
 
 class CouchbaseSpringRepository26Test extends AbstractCouchbaseSpringRepositoryTest {
 
   @Override
-  void assertCouchbaseCall(TraceAssert trace, int index, Object name, String bucketName = null, Object parentSpan = null, Object statement = null) {
-    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, bucketName, parentSpan, statement)
+  void assertCouchbaseCall(TraceAssert trace,
+                           int index,
+                           Object name,
+                           SpanData parentSpan = null,
+                           String bucketName = null,
+                           Object statement = null,
+                           Object operation = null) {
+    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, parentSpan, bucketName, statement, operation)
   }
 }

instrumentation/couchbase/couchbase-2.6/javaagent/src/test/groovy/springdata/CouchbaseSpringTemplate26Test.groovy

@@ -6,10 +6,17 @@
 package springdata
 
 import io.opentelemetry.instrumentation.test.asserts.TraceAssert
+import io.opentelemetry.sdk.trace.data.SpanData
 
 class CouchbaseSpringTemplate26Test extends AbstractCouchbaseSpringTemplateTest {
   @Override
-  void assertCouchbaseCall(TraceAssert trace, int index, Object name, String bucketName = null, Object parentSpan = null, Object statement = null) {
-    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, bucketName, parentSpan, statement)
+  void assertCouchbaseCall(TraceAssert trace,
+                           int index,
+                           Object name,
+                           SpanData parentSpan = null,
+                           String bucketName = null,
+                           Object statement = null,
+                           Object operation = null) {
+    CouchbaseSpanUtil.assertCouchbaseCall(trace, index, name, parentSpan, bucketName, statement, operation)
   }
 }

instrumentation/couchbase/couchbase-testing/src/main/groovy/AbstractCouchbaseAsyncClientTest.groovy

@@ -32,8 +32,8 @@ abstract class AbstractCouchbaseAsyncClientTest extends AbstractCouchbaseTest {
     assert hasBucket.get()
     assertTraces(1) {
       trace(0, 2) {
-        assertCouchbaseCall(it, 0, "Cluster.openBucket", null)
-        assertCouchbaseCall(it, 1, "ClusterManager.hasBucket", null, span(0))
+        assertCouchbaseCall(it, 0, "Cluster.openBucket")
+        assertCouchbaseCall(it, 1, "ClusterManager.hasBucket", span(0))
       }
     }
 
@@ -74,8 +74,8 @@ abstract class AbstractCouchbaseAsyncClientTest extends AbstractCouchbaseTest {
           hasNoParent()
         }
 
-        assertCouchbaseCall(it, 1, "Cluster.openBucket", null, span(0))
-        assertCouchbaseCall(it, 2, "Bucket.upsert", bucketSettings.name(), span(1))
+        assertCouchbaseCall(it, 1, "Cluster.openBucket", span(0))
+        assertCouchbaseCall(it, 2, "Bucket.upsert", span(1), bucketSettings.name())
       }
     }
 
@@ -123,9 +123,9 @@ abstract class AbstractCouchbaseAsyncClientTest extends AbstractCouchbaseTest {
           hasNoParent()
         }
 
-        assertCouchbaseCall(it, 1, "Cluster.openBucket", null, span(0))
-        assertCouchbaseCall(it, 2, "Bucket.upsert", bucketSettings.name(), span(1))
-        assertCouchbaseCall(it, 3, "Bucket.get", bucketSettings.name(), span(2))
+        assertCouchbaseCall(it, 1, "Cluster.openBucket", span(0))
+        assertCouchbaseCall(it, 2, "Bucket.upsert", span(1), bucketSettings.name())
+        assertCouchbaseCall(it, 3, "Bucket.get", span(2), bucketSettings.name())
       }
     }
 
@@ -172,10 +172,10 @@ abstract class AbstractCouchbaseAsyncClientTest extends AbstractCouchbaseTest {
           hasNoParent()
         }
 
-        assertCouchbaseCall(it, 1, "Cluster.openBucket", null, span(0))
+        assertCouchbaseCall(it, 1, "Cluster.openBucket", span(0))
 
         def dbName = bucketCouchbase.name()
-        assertCouchbaseCall(it, 2, "SELECT $dbName", dbName, span(1), 'SELECT mockrow')
+        assertCouchbaseCall(it, 2, "SELECT $dbName", span(1), dbName, 'SELECT mockrow', 'SELECT')
       }
     }
 

instrumentation/couchbase/couchbase-testing/src/main/groovy/AbstractCouchbaseClientTest.groovy

@@ -71,8 +71,8 @@ abstract class AbstractCouchbaseClientTest extends AbstractCouchbaseTest {
           kind SpanKind.INTERNAL
           hasNoParent()
         }
-        assertCouchbaseCall(it, 1, "Bucket.upsert", bucketSettings.name(), span(0))
-        assertCouchbaseCall(it, 2, "Bucket.get", bucketSettings.name(), span(0))
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), bucketSettings.name())
+        assertCouchbaseCall(it, 2, "Bucket.get", span(0), bucketSettings.name())
       }
     }
 
@@ -112,7 +112,7 @@ abstract class AbstractCouchbaseClientTest extends AbstractCouchbaseTest {
       }
       trace(1, 1) {
         def dbName = bucketCouchbase.name()
-        assertCouchbaseCall(it, 0, "SELECT $dbName", dbName, null, 'SELECT mockrow')
+        assertCouchbaseCall(it, 0, "SELECT $dbName", null, dbName, 'SELECT mockrow', 'SELECT')
       }
     }
 

instrumentation/couchbase/couchbase-testing/src/main/groovy/springdata/AbstractCouchbaseSpringRepositoryTest.groovy

@@ -83,7 +83,7 @@ abstract class AbstractCouchbaseSpringRepositoryTest extends AbstractCouchbaseTe
     assertTraces(1) {
       trace(0, 1) {
         def dbName = bucketCouchbase.name()
-        assertCouchbaseCall(it, 0, dbName, dbName, null, ~/^ViewQuery\(doc\/all\).*/)
+        assertCouchbaseCall(it, 0, dbName, null, dbName, ~/^ViewQuery\(doc\/all\).*/, { it == null })
       }
     }
   }
@@ -99,7 +99,7 @@ abstract class AbstractCouchbaseSpringRepositoryTest extends AbstractCouchbaseTe
     result == doc
     assertTraces(1) {
       trace(0, 1) {
-        assertCouchbaseCall(it, 0, "Bucket.upsert", bucketCouchbase.name())
+        assertCouchbaseCall(it, 0, "Bucket.upsert", null, bucketCouchbase.name())
       }
     }
 
@@ -129,8 +129,8 @@ abstract class AbstractCouchbaseSpringRepositoryTest extends AbstractCouchbaseTe
           kind SpanKind.INTERNAL
           hasNoParent()
         }
-        assertCouchbaseCall(it, 1, "Bucket.upsert", bucketCouchbase.name(), span(0))
-        assertCouchbaseCall(it, 2, "Bucket.get", bucketCouchbase.name(), span(0))
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), bucketCouchbase.name())
+        assertCouchbaseCall(it, 2, "Bucket.get", span(0), bucketCouchbase.name())
       }
     }
 
@@ -160,8 +160,8 @@ abstract class AbstractCouchbaseSpringRepositoryTest extends AbstractCouchbaseTe
           kind SpanKind.INTERNAL
           hasNoParent()
         }
-        assertCouchbaseCall(it, 1, "Bucket.upsert", bucketCouchbase.name(), span(0))
-        assertCouchbaseCall(it, 2, "Bucket.upsert", bucketCouchbase.name(), span(0))
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), bucketCouchbase.name())
+        assertCouchbaseCall(it, 2, "Bucket.upsert", span(0), bucketCouchbase.name())
       }
     }
 
@@ -194,9 +194,9 @@ abstract class AbstractCouchbaseSpringRepositoryTest extends AbstractCouchbaseTe
         }
 
         def dbName = bucketCouchbase.name()
-        assertCouchbaseCall(it, 1, "Bucket.upsert", dbName, span(0))
-        assertCouchbaseCall(it, 2, "Bucket.remove", dbName, span(0))
-        assertCouchbaseCall(it, 3, dbName, dbName, span(0), ~/^ViewQuery\(doc\/all\).*/)
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), dbName)
+        assertCouchbaseCall(it, 2, "Bucket.remove", span(0), dbName)
+        assertCouchbaseCall(it, 3, dbName, span(0), dbName, ~/^ViewQuery\(doc\/all\).*/, { it == null })
       }
     }
   }

instrumentation/couchbase/couchbase-testing/src/main/groovy/springdata/AbstractCouchbaseSpringTemplateTest.groovy

@@ -84,8 +84,8 @@ class AbstractCouchbaseSpringTemplateTest extends AbstractCouchbaseTest {
           kind SpanKind.INTERNAL
           hasNoParent()
         }
-        assertCouchbaseCall(it, 1, "Bucket.upsert", testName, span(0))
-        assertCouchbaseCall(it, 2, "Bucket.get", testName, span(0))
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), testName)
+        assertCouchbaseCall(it, 2, "Bucket.get", span(0), testName)
       }
     }
 
@@ -113,8 +113,8 @@ class AbstractCouchbaseSpringTemplateTest extends AbstractCouchbaseTest {
           kind SpanKind.INTERNAL
           hasNoParent()
         }
-        assertCouchbaseCall(it, 1, "Bucket.upsert", testName, span(0))
-        assertCouchbaseCall(it, 2, "Bucket.remove", testName, span(0))
+        assertCouchbaseCall(it, 1, "Bucket.upsert", span(0), testName)
+        assertCouchbaseCall(it, 2, "Bucket.remove", span(0), testName)
       }
     }
     clearExportedData()
@@ -126,7 +126,7 @@ class AbstractCouchbaseSpringTemplateTest extends AbstractCouchbaseTest {
     result == null
     assertTraces(1) {
       trace(0, 1) {
-        assertCouchbaseCall(it, 0, "Bucket.get", testName)
+        assertCouchbaseCall(it, 0, "Bucket.get", null, testName)
       }
     }
 

instrumentation/couchbase/couchbase-testing/src/main/groovy/util/AbstractCouchbaseTest.groovy

@@ -103,7 +103,13 @@ abstract class AbstractCouchbaseTest extends AgentInstrumentationSpecification {
       .socketConnectTimeout(timeout.intValue())
   }
 
-  void assertCouchbaseCall(TraceAssert trace, int index, Object spanName, String bucketName = null, Object parentSpan = null, Object statement = null) {
+  void assertCouchbaseCall(TraceAssert trace,
+                           int index,
+                           Object spanName,
+                           SpanData parentSpan = null,
+                           String bucketName = null,
+                           Object statement = null,
+                           Object operation = null) {
     trace.span(index) {
       name spanName
       kind CLIENT
@@ -114,10 +120,9 @@ abstract class AbstractCouchbaseTest extends AgentInstrumentationSpecification {
       }
       attributes {
         "${SemanticAttributes.DB_SYSTEM.key}" "couchbase"
-        if (bucketName != null) {
-          "${SemanticAttributes.DB_NAME.key}" bucketName
-        }
-        "${SemanticAttributes.DB_STATEMENT.key}"(statement ?: spanName)
+        "${SemanticAttributes.DB_NAME.key}" bucketName
+        "${SemanticAttributes.DB_STATEMENT.key}" statement
+        "${SemanticAttributes.DB_OPERATION.key}" (operation ?: spanName)
       }
     }
   }

instrumentation/hystrix-1.4/javaagent/build.gradle.kts

@@ -11,6 +11,9 @@ muzzle {
 }
 
 dependencies {
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+
   implementation(project(":instrumentation:rxjava:rxjava-1.0:library"))
 
   library("com.netflix.hystrix:hystrix-core:1.4.0")

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/ExperimentalAttributesExtractor.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.hystrix;
+
+import static io.opentelemetry.api.common.AttributeKey.booleanKey;
+import static io.opentelemetry.api.common.AttributeKey.stringKey;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class ExperimentalAttributesExtractor extends AttributesExtractor<HystrixRequest, Void> {
+  private static final AttributeKey<String> HYSTRIX_COMMAND = stringKey("hystrix.command");
+  private static final AttributeKey<String> HYSTRIX_GROUP = stringKey("hystrix.group");
+  private static final AttributeKey<Boolean> HYSTRIX_CIRCUIT_OPEN =
+      booleanKey("hystrix.circuit_open");
+
+  @Override
+  protected void onStart(AttributesBuilder attributes, HystrixRequest hystrixRequest) {
+    String commandName = hystrixRequest.command().getCommandKey().name();
+    String groupName = hystrixRequest.command().getCommandGroup().name();
+    boolean circuitOpen = hystrixRequest.command().isCircuitBreakerOpen();
+
+    set(attributes, HYSTRIX_COMMAND, commandName);
+    set(attributes, HYSTRIX_GROUP, groupName);
+    set(attributes, HYSTRIX_CIRCUIT_OPEN, circuitOpen);
+  }
+
+  @Override
+  protected void onEnd(
+      AttributesBuilder attributes, HystrixRequest hystrixRequest, @Nullable Void unused) {}
+}

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/HystrixCommandInstrumentation.java

@@ -7,11 +7,13 @@ package io.opentelemetry.javaagent.instrumentation.hystrix;
 
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.extendsClass;
 import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed;
+import static io.opentelemetry.javaagent.instrumentation.hystrix.HystrixSingletons.instrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.namedOneOf;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
 
 import com.netflix.hystrix.HystrixInvokableInfo;
+import io.opentelemetry.instrumentation.rxjava.TracedOnSubscribe;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import net.bytebuddy.asm.Advice;
@@ -52,7 +54,8 @@ public class HystrixCommandInstrumentation implements TypeInstrumentation {
         @Advice.Return(readOnly = false) Observable<?> result,
         @Advice.Thrown Throwable throwable) {
 
-      result = Observable.create(new HystrixOnSubscribe<>(result, command, "execute"));
+      HystrixRequest request = HystrixRequest.create(command, "execute");
+      result = Observable.create(new TracedOnSubscribe<>(result, instrumenter(), request));
     }
   }
 
@@ -65,7 +68,8 @@ public class HystrixCommandInstrumentation implements TypeInstrumentation {
         @Advice.Return(readOnly = false) Observable<?> result,
         @Advice.Thrown Throwable throwable) {
 
-      result = Observable.create(new HystrixOnSubscribe<>(result, command, "fallback"));
+      HystrixRequest request = HystrixRequest.create(command, "fallback");
+      result = Observable.create(new TracedOnSubscribe<>(result, instrumenter(), request));
     }
   }
 }

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/HystrixOnSubscribe.java

@@ -1,34 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.hystrix;
-
-import static io.opentelemetry.api.trace.SpanKind.INTERNAL;
-import static io.opentelemetry.javaagent.instrumentation.hystrix.HystrixTracer.tracer;
-
-import com.netflix.hystrix.HystrixInvokableInfo;
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.instrumentation.rxjava.TracedOnSubscribe;
-import rx.Observable;
-
-public class HystrixOnSubscribe<T> extends TracedOnSubscribe<T> {
-  private static final String OPERATION_NAME = "hystrix.cmd";
-
-  private final HystrixInvokableInfo<?> command;
-  private final String methodName;
-
-  public HystrixOnSubscribe(
-      Observable<T> originalObservable, HystrixInvokableInfo<?> command, String methodName) {
-    super(originalObservable, OPERATION_NAME, tracer(), INTERNAL);
-
-    this.command = command;
-    this.methodName = methodName;
-  }
-
-  @Override
-  protected void decorateSpan(Span span) {
-    tracer().onCommand(span, command, methodName);
-  }
-}

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/HystrixRequest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.hystrix;
+
+import com.google.auto.value.AutoValue;
+import com.netflix.hystrix.HystrixInvokableInfo;
+
+@AutoValue
+public abstract class HystrixRequest {
+
+  public static HystrixRequest create(HystrixInvokableInfo<?> command, String methodName) {
+    return new AutoValue_HystrixRequest(command, methodName);
+  }
+
+  public abstract HystrixInvokableInfo<?> command();
+
+  public abstract String methodName();
+
+  String spanName() {
+    String commandName = command().getCommandKey().name();
+    String groupName = command().getCommandGroup().name();
+    return groupName + "." + commandName + "." + methodName();
+  }
+}

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/HystrixSingletons.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.hystrix;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.instrumentation.api.config.Config;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
+
+public final class HystrixSingletons {
+
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.hystrix-1.4";
+
+  private static final Instrumenter<HystrixRequest, Void> INSTRUMENTER;
+
+  static {
+    InstrumenterBuilder<HystrixRequest, Void> builder =
+        Instrumenter.newBuilder(
+            GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME, HystrixRequest::spanName);
+
+    if (Config.get()
+        .getBooleanProperty("otel.instrumentation.hystrix.experimental-span-attributes", false)) {
+      builder.addAttributesExtractor(new ExperimentalAttributesExtractor());
+    }
+
+    INSTRUMENTER = builder.newInstrumenter();
+  }
+
+  public static Instrumenter<HystrixRequest, Void> instrumenter() {
+    return INSTRUMENTER;
+  }
+
+  private HystrixSingletons() {}
+}

instrumentation/hystrix-1.4/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/hystrix/HystrixTracer.java

@@ -1,45 +0,0 @@
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.hystrix;
-
-import com.netflix.hystrix.HystrixInvokableInfo;
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.instrumentation.api.config.Config;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
-
-public class HystrixTracer extends BaseTracer {
-  private static final HystrixTracer TRACER = new HystrixTracer();
-
-  public static HystrixTracer tracer() {
-    return TRACER;
-  }
-
-  private static final boolean CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES =
-      Config.get()
-          .getBooleanProperty("otel.instrumentation.hystrix.experimental-span-attributes", false);
-
-  @Override
-  protected String getInstrumentationName() {
-    return "io.opentelemetry.hystrix-1.4";
-  }
-
-  public void onCommand(Span span, HystrixInvokableInfo<?> command, String methodName) {
-    if (command != null) {
-      String commandName = command.getCommandKey().name();
-      String groupName = command.getCommandGroup().name();
-      boolean circuitOpen = command.isCircuitBreakerOpen();
-
-      String spanName = groupName + "." + commandName + "." + methodName;
-
-      span.updateName(spanName);
-      if (CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES) {
-        span.setAttribute("hystrix.command", commandName);
-        span.setAttribute("hystrix.group", groupName);
-        span.setAttribute("hystrix.circuit-open", circuitOpen);
-      }
-    }
-  }
-}

instrumentation/hystrix-1.4/javaagent/src/test/groovy/HystrixObservableChainTest.groovy

@@ -71,7 +71,7 @@ class HystrixObservableChainTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableChainTest\$1"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -86,7 +86,7 @@ class HystrixObservableChainTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableChainTest\$2"
             "hystrix.group" "OtherGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(4) {

instrumentation/hystrix-1.4/javaagent/src/test/groovy/HystrixObservableTest.groovy

@@ -63,7 +63,7 @@ class HystrixObservableTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableTest\$1"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -156,7 +156,7 @@ class HystrixObservableTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableTest\$2"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -165,7 +165,7 @@ class HystrixObservableTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableTest\$2"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
       }
@@ -251,7 +251,7 @@ class HystrixObservableTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableTest\$3"
             "hystrix.group" "FailingGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -262,7 +262,7 @@ class HystrixObservableTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixObservableTest\$3"
             "hystrix.group" "FailingGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
       }

instrumentation/hystrix-1.4/javaagent/src/test/groovy/HystrixTest.groovy

@@ -47,7 +47,7 @@ class HystrixTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixTest\$1"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -108,7 +108,7 @@ class HystrixTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixTest\$2"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
         span(2) {
@@ -117,7 +117,7 @@ class HystrixTest extends AgentInstrumentationSpecification {
           attributes {
             "hystrix.command" "HystrixTest\$2"
             "hystrix.group" "ExampleGroup"
-            "hystrix.circuit-open" false
+            "hystrix.circuit_open" false
           }
         }
       }

instrumentation/rxjava/rxjava-1.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/SpanFinishingSubscription.java

@@ -6,24 +6,27 @@
 package io.opentelemetry.instrumentation.rxjava;
 
 import io.opentelemetry.context.Context;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import java.util.concurrent.atomic.AtomicReference;
 import rx.Subscription;
 
-public class SpanFinishingSubscription implements Subscription {
-  private final BaseTracer tracer;
+final class SpanFinishingSubscription<REQUEST> implements Subscription {
+  private final Instrumenter<REQUEST, ?> instrumenter;
   private final AtomicReference<Context> contextRef;
+  private final REQUEST request;
 
-  public SpanFinishingSubscription(BaseTracer tracer, AtomicReference<Context> contextRef) {
-    this.tracer = tracer;
+  SpanFinishingSubscription(
+      Instrumenter<REQUEST, ?> instrumenter, AtomicReference<Context> contextRef, REQUEST request) {
+    this.instrumenter = instrumenter;
     this.contextRef = contextRef;
+    this.request = request;
   }
 
   @Override
   public void unsubscribe() {
     Context context = contextRef.getAndSet(null);
     if (context != null) {
-      tracer.end(context);
+      instrumenter.end(context, request, null, null);
     }
   }
 

instrumentation/rxjava/rxjava-1.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/TracedOnSubscribe.java

@@ -5,42 +5,44 @@
 
 package io.opentelemetry.instrumentation.rxjava;
 
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import rx.Observable;
 import rx.Subscriber;
 import rx.__OpenTelemetryTracingUtil;
 
-public class TracedOnSubscribe<T> implements Observable.OnSubscribe<T> {
+public final class TracedOnSubscribe<T, REQUEST> implements Observable.OnSubscribe<T> {
   private final Observable.OnSubscribe<T> delegate;
-  private final String spanName;
+  private final Instrumenter<REQUEST, ?> instrumenter;
+  private final REQUEST request;
   private final Context parentContext;
-  private final BaseTracer tracer;
-  private final SpanKind spanKind;
 
   public TracedOnSubscribe(
-      Observable<T> originalObservable, String spanName, BaseTracer tracer, SpanKind spanKind) {
+      Observable<T> originalObservable, Instrumenter<REQUEST, ?> instrumenter, REQUEST request) {
     delegate = __OpenTelemetryTracingUtil.extractOnSubscribe(originalObservable);
-    this.spanName = spanName;
-    this.tracer = tracer;
-    this.spanKind = spanKind;
+    this.instrumenter = instrumenter;
+    this.request = request;
 
     parentContext = Context.current();
   }
 
   @Override
   public void call(Subscriber<? super T> subscriber) {
-    Context context = tracer.startSpan(parentContext, spanName, spanKind);
-    decorateSpan(Span.fromContext(context));
-    try (Scope ignored = context.makeCurrent()) {
-      delegate.call(new TracedSubscriber<>(context, subscriber, tracer));
+    /*
+    TODO: can't really call shouldStart() - couchbase async instrumentation nests CLIENT calls
+    which normally should happen in a sequence
+    InstrumentationTypes to the rescue?
+
+    if (!instrumenter.shouldStart(parentContext, request)) {
+      delegate.call(subscriber);
+      return;
     }
-  }
+     */
 
-  protected void decorateSpan(Span span) {
-    // Subclasses can use it to provide addition attributes to the span
+    Context context = instrumenter.start(parentContext, request);
+    try (Scope ignored = context.makeCurrent()) {
+      delegate.call(new TracedSubscriber<>(subscriber, instrumenter, context, request));
+    }
   }
 }

instrumentation/rxjava/rxjava-1.0/library/src/main/java/io/opentelemetry/instrumentation/rxjava/TracedSubscriber.java

@@ -7,22 +7,28 @@ package io.opentelemetry.instrumentation.rxjava;
 
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import java.util.concurrent.atomic.AtomicReference;
 import rx.Subscriber;
 
-public class TracedSubscriber<T> extends Subscriber<T> {
+final class TracedSubscriber<T, REQUEST> extends Subscriber<T> {
 
-  private final AtomicReference<Context> contextRef;
   private final Subscriber<T> delegate;
-  private final BaseTracer tracer;
+  private final Instrumenter<REQUEST, ?> instrumenter;
+  private final AtomicReference<Context> contextRef;
+  private final REQUEST request;
 
-  public TracedSubscriber(Context context, Subscriber<T> delegate, BaseTracer tracer) {
-    contextRef = new AtomicReference<>(context);
+  TracedSubscriber(
+      Subscriber<T> delegate,
+      Instrumenter<REQUEST, ?> instrumenter,
+      Context context,
+      REQUEST request) {
     this.delegate = delegate;
-    this.tracer = tracer;
-    SpanFinishingSubscription subscription = new SpanFinishingSubscription(tracer, contextRef);
-    delegate.add(subscription);
+    this.instrumenter = instrumenter;
+    this.contextRef = new AtomicReference<>(context);
+    this.request = request;
+
+    delegate.add(new SpanFinishingSubscription<>(instrumenter, contextRef, request));
   }
 
   @Override
@@ -60,11 +66,7 @@ public class TracedSubscriber<T> extends Subscriber<T> {
         error = t;
         throw t;
       } finally {
-        if (error != null) {
-          tracer.endExceptionally(context, error);
-        } else {
-          tracer.end(context);
-        }
+        instrumenter.end(context, request, null, error);
       }
     } else {
       delegate.onCompleted();
@@ -75,7 +77,7 @@ public class TracedSubscriber<T> extends Subscriber<T> {
   public void onError(Throwable e) {
     Context context = contextRef.getAndSet(null);
     if (context != null) {
-      tracer.endExceptionally(context, e);
+      instrumenter.end(context, request, null, e);
     }
     // TODO (trask) should this be wrapped in parent of context(?)
     delegate.onError(e);