refactor messaging wrappers & add the kafka-clients implementation

Author: Cirilla-zmh

messaging-wrappers/README.md

@@ -22,5 +22,6 @@ this tool aims to streamline the tracing and monitoring process.
 ## Component owners
 
 - [Minghui Zhang](https://github.com/Cirilla-zmh), Alibaba Cloud
+- [Steve Rao](https://github.com/steverao), Alibaba Cloud
 
 Learn more about component owners in [component_owners.yml](../.github/component_owners.yml).

messaging-wrappers/api/build.gradle.kts

@@ -8,7 +8,7 @@ description = "OpenTelemetry Messaging Wrappers"
 otelJava.moduleName.set("io.opentelemetry.contrib.messaging.wrappers")
 
 dependencies {
-  api("io.opentelemetry:opentelemetry-api")
+  api("io.opentelemetry.instrumentation:opentelemetry-instrumentation-api-incubator")
 
   compileOnly("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi")
   compileOnly("io.opentelemetry:opentelemetry-api-incubator")

messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/DefaultMessagingProcessWrapperBuilder.java

@@ -3,55 +3,60 @@
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.context.propagation.TextMapGetter;
-import io.opentelemetry.contrib.messaging.wrappers.semconv.DefaultMessagingProcessSpanCustomizer;
+import io.opentelemetry.contrib.messaging.wrappers.semconv.DefaultMessagingAttributesGetter;
 import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest;
-import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessResponse;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.incubator.semconv.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
-public class DefaultMessagingProcessWrapperBuilder<REQUEST extends MessagingProcessRequest, RESPONSE extends MessagingProcessResponse<?>> {
+public class DefaultMessagingProcessWrapperBuilder<REQUEST extends MessagingProcessRequest> {
 
   private OpenTelemetry openTelemetry;
 
-  private TextMapGetter<REQUEST> textMapGetter;
+  protected TextMapGetter<REQUEST> textMapGetter;
 
-  private List<MessagingSpanCustomizer<REQUEST, RESPONSE>> spanCustomizers;
+  protected List<AttributesExtractor<REQUEST, Void>> attributesExtractors;
 
-  public static <REQUEST extends MessagingProcessRequest, RESPONSE extends MessagingProcessResponse<?>> DefaultMessagingProcessWrapperBuilder<REQUEST, RESPONSE> create() {
-    return new DefaultMessagingProcessWrapperBuilder<>();
-  }
-
-  public DefaultMessagingProcessWrapperBuilder<REQUEST, RESPONSE> openTelemetry(OpenTelemetry openTelemetry) {
+  public DefaultMessagingProcessWrapperBuilder<REQUEST> openTelemetry(OpenTelemetry openTelemetry) {
     this.openTelemetry = openTelemetry;
     return this;
   }
 
-  public DefaultMessagingProcessWrapperBuilder<REQUEST, RESPONSE> textMapGetter(TextMapGetter<REQUEST> textMapGetter) {
+  public DefaultMessagingProcessWrapperBuilder<REQUEST> textMapGetter(TextMapGetter<REQUEST> textMapGetter) {
     this.textMapGetter = textMapGetter;
     return this;
   }
 
-  public DefaultMessagingProcessWrapperBuilder<REQUEST, RESPONSE> spanCustomizers(
-          List<MessagingSpanCustomizer<REQUEST, RESPONSE>> spanCustomizers) {
-    this.spanCustomizers = spanCustomizers;
+  /**
+   * This method overrides the original items.
+   * <p>See {@link DefaultMessagingProcessWrapperBuilder#addAttributesExtractor} if you just want to append one.</p>
+   * */
+  public DefaultMessagingProcessWrapperBuilder<REQUEST> attributesExtractors(
+      Collection<AttributesExtractor<REQUEST, Void>> attributesExtractors) {
+    this.attributesExtractors = new ArrayList<>();
+    this.attributesExtractors.addAll(attributesExtractors);
     return this;
   }
 
-  public DefaultMessagingProcessWrapperBuilder<REQUEST, RESPONSE> addSpanCustomizers(
-          MessagingSpanCustomizer<REQUEST, RESPONSE> spanCustomizer) {
-    this.spanCustomizers.add(spanCustomizer);
+  public DefaultMessagingProcessWrapperBuilder<REQUEST> addAttributesExtractor(
+      AttributesExtractor<REQUEST, Void> attributesExtractor) {
+    this.attributesExtractors.add(attributesExtractor);
     return this;
   }
 
-  public MessagingProcessWrapper<REQUEST, RESPONSE> build() {
-    return new MessagingProcessWrapper<>(this.openTelemetry, this.textMapGetter, this.spanCustomizers);
+  public MessagingProcessWrapper<REQUEST> build() {
+    return new MessagingProcessWrapper<>(this.openTelemetry == null ? GlobalOpenTelemetry.get() : this.openTelemetry,
+            this.textMapGetter, this.attributesExtractors);
   }
 
-  private DefaultMessagingProcessWrapperBuilder() {
-    // init by default
-    this.openTelemetry = GlobalOpenTelemetry.get();
-    this.spanCustomizers = new ArrayList<>();
-    this.spanCustomizers.add(DefaultMessagingProcessSpanCustomizer.create());
+  protected DefaultMessagingProcessWrapperBuilder() {
+    // init attributes extractors by default
+    this.attributesExtractors = new ArrayList<>();
+    this.attributesExtractors.add(MessagingAttributesExtractor.create(
+            DefaultMessagingAttributesGetter.create(), MessageOperation.PROCESS));
   }
 }

messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingProcessWrapper.java

@@ -1,6 +1,8 @@
 package io.opentelemetry.contrib.messaging.wrappers;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanBuilder;
 import io.opentelemetry.api.trace.Tracer;
@@ -9,16 +11,12 @@
 import io.opentelemetry.context.propagation.TextMapGetter;
 import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessRequest;
-import io.opentelemetry.contrib.messaging.wrappers.semconv.MessagingProcessResponse;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
 
+import javax.annotation.Nullable;
 import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
-public class MessagingProcessWrapper<REQUEST extends MessagingProcessRequest, RESPONSE extends MessagingProcessResponse<?>> {
-
-  private static final Logger LOG = Logger.getLogger(MessagingProcessWrapper.class.getName());
+public class MessagingProcessWrapper<REQUEST extends MessagingProcessRequest> {
 
   private static final String INSTRUMENTATION_SCOPE = "messaging-process-wrapper";
 
@@ -32,94 +30,57 @@ public class MessagingProcessWrapper<REQUEST extends MessagingProcessRequest, RE
 
   private final TextMapGetter<REQUEST> textMapGetter;
 
-  private final List<MessagingSpanCustomizer<REQUEST, RESPONSE>> spanCustomizers;
+  // no attributes need to be extracted from responses in process operations
+  private final List<AttributesExtractor<REQUEST, Void>> attributesExtractors;
 
-  public Runnable wrap(REQUEST request, Runnable runnable) {
-    return () -> {
-      Span span = handleStart(request);
-      Scope scope = span.makeCurrent();
+  public static <REQUEST extends MessagingProcessRequest> DefaultMessagingProcessWrapperBuilder<REQUEST> defaultBuilder() {
+    return new DefaultMessagingProcessWrapperBuilder<>();
+  }
 
-      try {
-        runnable.run();
-      } catch (Throwable t) {
-        handleEnd(span, request, null, t);
-        scope.close();
-        throw t;
-      }
+  public <E extends Throwable> void doProcess(REQUEST request, ThrowingRunnable<E> runnable) throws E {
+    Span span = handleStart(request);
 
-      handleEnd(span, request, null, null);
-      scope.close();
-    };
-  }
+    try (Scope scope = span.makeCurrent()) {
+      runnable.run();
+    } catch (Throwable t) {
+      handleEnd(span, request, t);
+      throw t;
+    }
 
-  public <R> Callable<R> wrap(REQUEST request, Callable<R> callable) {
-    return () -> {
-      Span span = handleStart(request);
-      Scope scope = span.makeCurrent();
-      RESPONSE response = null;
-
-      R result = null;
-      try {
-        result = callable.call();
-        if (result instanceof MessagingProcessResponse) {
-          response = (RESPONSE) result;
-        }
-      } catch (Throwable t) {
-        handleEnd(span, request, response, t);
-        scope.close();
-        throw t;
-      }
-
-      handleEnd(span, request, response, null);
-      scope.close();
-      return result;
-    };
+    handleEnd(span, request, null);
   }
 
-  public <R> R doProcess(REQUEST request, Callable<R> process) throws Exception {
+  public <R, E extends Throwable> R doProcess(REQUEST request, ThrowingSupplier<R, E> supplier) throws E {
     Span span = handleStart(request);
-    Scope scope = span.makeCurrent();
-    RESPONSE response = null;
 
     R result = null;
-    try {
-      result = process.call();
-      if (result instanceof MessagingProcessResponse) {
-        response = (RESPONSE) result;
-      }
+    try (Scope scope = span.makeCurrent()) {
+      result = supplier.get();
     } catch (Throwable t) {
-      handleEnd(span, request, response, t);
-      scope.close();
+      handleEnd(span, request, t);
       throw t;
     }
 
-    // noop response by default
-    handleEnd(span, request, response, null);
-    scope.close();
+    handleEnd(span, request, null);
     return result;
   }
 
   protected Span handleStart(REQUEST request) {
     Context context = this.textMapPropagator.extract(Context.current(), request, this.textMapGetter);
     SpanBuilder spanBuilder = this.tracer.spanBuilder(getDefaultSpanName(request.getDestination()));
     spanBuilder.setParent(context);
-    for (MessagingSpanCustomizer<REQUEST, RESPONSE> customizer : spanCustomizers) {
-      try {
-        context = customizer.onStart(spanBuilder, context, request);
-      } catch (Exception e) {
-        LOG.log(Level.WARNING, "Exception occurred while customizing span on start.", e);
-      }
+
+    AttributesBuilder builder = Attributes.builder();
+    for (AttributesExtractor<REQUEST, Void> extractor : this.attributesExtractors) {
+      extractor.onStart(builder, context, request);
     }
-    return spanBuilder.startSpan();
+    return spanBuilder.setAllAttributes(builder.build()).startSpan();
   }
 
-  protected void handleEnd(Span span, REQUEST request, RESPONSE response, Throwable t) {
-    for (MessagingSpanCustomizer<REQUEST, RESPONSE> customizer : spanCustomizers) {
-      try {
-        customizer.onEnd(span, Context.current(), request, response, t);
-      } catch (Exception e) {
-        LOG.log(Level.WARNING, "Exception occurred while customizing span on end.", e);
-      }
+  protected void handleEnd(Span span, REQUEST request, Throwable t) {
+    AttributesBuilder builder = Attributes.builder();
+    for (AttributesExtractor<REQUEST, Void> extractor : this.attributesExtractors) {
+      extractor.onEnd(builder, Context.current(), request, null, t);
     }
     span.end();
   }
@@ -132,11 +93,11 @@ protected String getDefaultSpanName(String destination) {
   }
 
   protected MessagingProcessWrapper(OpenTelemetry openTelemetry,
-                          TextMapGetter<REQUEST> textMapGetter,
-                          List<MessagingSpanCustomizer<REQUEST, RESPONSE>> spanCustomizers) {
+                          @Nullable TextMapGetter<REQUEST> textMapGetter,
+                          List<AttributesExtractor<REQUEST, Void>> attributesExtractors) {
     this.textMapPropagator = openTelemetry.getPropagators().getTextMapPropagator();
     this.tracer = openTelemetry.getTracer(INSTRUMENTATION_SCOPE + "-" + INSTRUMENTATION_VERSION);
     this.textMapGetter = textMapGetter;
-    this.spanCustomizers = spanCustomizers;
+    this.attributesExtractors = attributesExtractors;
   }
 }

messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/MessagingSpanCustomizer.java

@@ -0,0 +1,13 @@
+package io.opentelemetry.contrib.messaging.wrappers;
+
+/**
+ * A utility interface representing a {@link Runnable} that may throw.
+ *
+ * <p>Inspired from <a href=https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/util/ThrowingRunnable.java>ThrowingRunnable</a>.
+ *
+ * @param <E> Thrown exception type.
+ */
+@FunctionalInterface
+public interface ThrowingRunnable<E extends Throwable> {
+  void run() throws E;
+}

messaging-wrappers/api/src/main/java/io/opentelemetry/contrib/messaging/wrappers/ThrowingRunnable.java

@@ -0,0 +1,15 @@
+package io.opentelemetry.contrib.messaging.wrappers;
+
+import java.util.function.Supplier;
+
+/**
+ * A utility interface representing a {@link Supplier} that may throw.
+ *
+ * <p>Inspired from <a href=https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/util/ThrowingSupplier.java>ThrowingSupplier</a>.
+ *
+ * @param <E> Thrown exception type.
+ */
+@FunctionalInterface
+public interface ThrowingSupplier<T, E extends Throwable> {
+  T get() throws E;
+}