OpenSearch Transport v3.0 Implementation (#14823)

Co-authored-by: Lauri Tulmin <[email protected]>

Author: mznet

.fossa.yml

@@ -757,6 +757,9 @@ targets:
     - type: gradle
       path: ./
       target: ':instrumentation:openai:openai-java-1.1:library'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:opensearch:opensearch-java-3.0:javaagent'
     - type: gradle
       path: ./
       target: ':instrumentation:opensearch:opensearch-rest-1.0:javaagent'

instrumentation/opensearch/opensearch-java-3.0/javaagent/build.gradle.kts

@@ -0,0 +1,50 @@
+plugins {
+  id("otel.javaagent-instrumentation")
+}
+
+muzzle {
+  pass {
+    group.set("org.opensearch.client")
+    module.set("opensearch-java")
+    versions.set("[3.0,)")
+  }
+}
+
+otelJava {
+  minJavaVersionSupported.set(JavaVersion.VERSION_11)
+}
+
+dependencies {
+  library("org.opensearch.client:opensearch-java:3.0.0")
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
+
+  testImplementation("org.opensearch.client:opensearch-rest-client:3.0.0")
+  testImplementation(project(":instrumentation:opensearch:opensearch-rest-common:testing"))
+  testInstrumentation(project(":instrumentation:apache-httpclient:apache-httpclient-5.0:javaagent"))
+
+  // For testing AwsSdk2Transport
+  testInstrumentation(project(":instrumentation:apache-httpclient:apache-httpclient-4.0:javaagent"))
+  testInstrumentation(project(":instrumentation:netty:netty-4.1:javaagent"))
+  testImplementation("software.amazon.awssdk:auth:2.22.0")
+  testImplementation("software.amazon.awssdk:identity-spi:2.22.0")
+  testImplementation("software.amazon.awssdk:apache-client:2.22.0")
+  testImplementation("software.amazon.awssdk:netty-nio-client:2.22.0")
+  testImplementation("software.amazon.awssdk:regions:2.22.0")
+}
+
+tasks {
+  test {
+    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
+  }
+
+  val testStableSemconv by registering(Test::class) {
+    testClassesDirs = sourceSets.test.get().output.classesDirs
+    classpath = sourceSets.test.get().runtimeClasspath
+    jvmArgs("-Dotel.semconv-stability.opt-in=database")
+  }
+
+  check {
+    dependsOn(testStableSemconv)
+  }
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchAttributesGetter.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+import io.opentelemetry.instrumentation.api.incubator.semconv.db.DbClientAttributesGetter;
+import io.opentelemetry.semconv.incubating.DbIncubatingAttributes;
+import javax.annotation.Nullable;
+
+final class OpenSearchAttributesGetter
+    implements DbClientAttributesGetter<OpenSearchRequest, Void> {
+
+  @Override
+  public String getDbSystem(OpenSearchRequest request) {
+    return DbIncubatingAttributes.DbSystemNameIncubatingValues.OPENSEARCH;
+  }
+
+  @Override
+  @Nullable
+  public String getDbNamespace(OpenSearchRequest request) {
+    return null;
+  }
+
+  @Override
+  @Nullable
+  public String getDbQueryText(OpenSearchRequest request) {
+    return request.getMethod() + " " + request.getOperation();
+  }
+
+  @Override
+  @Nullable
+  public String getDbOperationName(OpenSearchRequest request) {
+    return request.getMethod();
+  }
+
+  @Nullable
+  @Override
+  public String getResponseStatus(@Nullable Void response, @Nullable Throwable error) {
+    return null; // Response status is handled by HTTP instrumentation
+  }
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchInstrumentationModule.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+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 OpenSearchInstrumentationModule extends InstrumentationModule {
+  public OpenSearchInstrumentationModule() {
+    super("opensearch-java", "opensearch-java-3.0", "opensearch");
+  }
+
+  @Override
+  public List<TypeInstrumentation> typeInstrumentations() {
+    return singletonList(new OpenSearchTransportInstrumentation());
+  }
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchRequest.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+import com.google.auto.value.AutoValue;
+
+@AutoValue
+public abstract class OpenSearchRequest {
+
+  public static OpenSearchRequest create(String method, String endpoint) {
+    return new AutoValue_OpenSearchRequest(method, endpoint);
+  }
+
+  public abstract String getMethod();
+
+  public abstract String getOperation();
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchResponseHandler.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+import static io.opentelemetry.javaagent.instrumentation.opensearch.v3_0.OpenSearchSingletons.instrumenter;
+
+import io.opentelemetry.context.Context;
+import java.util.function.BiConsumer;
+
+public final class OpenSearchResponseHandler implements BiConsumer<Object, Throwable> {
+  private final Context context;
+  private final OpenSearchRequest otelRequest;
+
+  public OpenSearchResponseHandler(Context context, OpenSearchRequest otelRequest) {
+    this.context = context;
+    this.otelRequest = otelRequest;
+  }
+
+  @Override
+  public void accept(Object response, Throwable error) {
+    // OpenSearch responses don't provide response information, so the span is closed with null.
+    instrumenter().end(context, otelRequest, null, error);
+  }
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchSingletons.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.instrumentation.api.incubator.semconv.db.DbClientAttributesExtractor;
+import io.opentelemetry.instrumentation.api.incubator.semconv.db.DbClientMetrics;
+import io.opentelemetry.instrumentation.api.incubator.semconv.db.DbClientSpanNameExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
+
+public final class OpenSearchSingletons {
+  private static final Instrumenter<OpenSearchRequest, Void> INSTRUMENTER = createInstrumenter();
+
+  public static Instrumenter<OpenSearchRequest, Void> instrumenter() {
+    return INSTRUMENTER;
+  }
+
+  private static Instrumenter<OpenSearchRequest, Void> createInstrumenter() {
+    OpenSearchAttributesGetter dbClientAttributesGetter = new OpenSearchAttributesGetter();
+
+    return Instrumenter.<OpenSearchRequest, Void>builder(
+            GlobalOpenTelemetry.get(),
+            "io.opentelemetry.opensearch-java-3.0",
+            DbClientSpanNameExtractor.create(dbClientAttributesGetter))
+        .addAttributesExtractor(DbClientAttributesExtractor.create(dbClientAttributesGetter))
+        .addOperationMetrics(DbClientMetrics.get())
+        .buildInstrumenter(SpanKindExtractor.alwaysClient());
+  }
+
+  private OpenSearchSingletons() {}
+}

instrumentation/opensearch/opensearch-java-3.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/opensearch/v3_0/OpenSearchTransportInstrumentation.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.opensearch.v3_0;
+
+import static io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge.currentContext;
+import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.implementsInterface;
+import static io.opentelemetry.javaagent.instrumentation.opensearch.v3_0.OpenSearchSingletons.instrumenter;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
+import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import java.util.concurrent.CompletableFuture;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.opensearch.client.transport.Endpoint;
+
+public class OpenSearchTransportInstrumentation implements TypeInstrumentation {
+  @Override
+  public ElementMatcher<TypeDescription> typeMatcher() {
+    return implementsInterface(named("org.opensearch.client.transport.OpenSearchTransport"));
+  }
+
+  @Override
+  public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isMethod()
+            .and(isPublic())
+            .and(named("performRequest"))
+            .and(takesArgument(0, Object.class))
+            .and(takesArgument(1, named("org.opensearch.client.transport.Endpoint"))),
+        this.getClass().getName() + "$PerformRequestAdvice");
+
+    transformer.applyAdviceToMethod(
+        isMethod()
+            .and(isPublic())
+            .and(named("performRequestAsync"))
+            .and(takesArgument(0, Object.class))
+            .and(takesArgument(1, named("org.opensearch.client.transport.Endpoint"))),
+        this.getClass().getName() + "$PerformRequestAsyncAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class PerformRequestAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(
+        @Advice.Argument(0) Object request,
+        @Advice.Argument(1) Endpoint<Object, Object, Object> endpoint,
+        @Advice.Local("otelRequest") OpenSearchRequest otelRequest,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      Context parentContext = currentContext();
+      otelRequest =
+          OpenSearchRequest.create(endpoint.method(request), endpoint.requestUrl(request));
+      if (!instrumenter().shouldStart(parentContext, otelRequest)) {
+        return;
+      }
+
+      context = instrumenter().start(parentContext, otelRequest);
+      scope = context.makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void stopSpan(
+        @Advice.Thrown Throwable throwable,
+        @Advice.Local("otelRequest") OpenSearchRequest otelRequest,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+
+      instrumenter().end(context, otelRequest, null, throwable);
+    }
+  }
+
+  @SuppressWarnings("unused")
+  public static class PerformRequestAsyncAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(
+        @Advice.Argument(0) Object request,
+        @Advice.Argument(value = 1, readOnly = false) Endpoint<Object, Object, Object> endpoint,
+        @Advice.Local("otelRequest") OpenSearchRequest otelRequest,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      Context parentContext = currentContext();
+      otelRequest =
+          OpenSearchRequest.create(endpoint.method(request), endpoint.requestUrl(request));
+      if (!instrumenter().shouldStart(parentContext, otelRequest)) {
+        return;
+      }
+
+      context = instrumenter().start(parentContext, otelRequest);
+      scope = context.makeCurrent();
+    }
+
+    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    public static void stopSpan(
+        @Advice.Thrown Throwable throwable,
+        @Advice.Return(readOnly = false) CompletableFuture<Object> future,
+        @Advice.Local("otelRequest") OpenSearchRequest otelRequest,
+        @Advice.Local("otelContext") Context context,
+        @Advice.Local("otelScope") Scope scope) {
+
+      if (scope == null) {
+        return;
+      }
+      scope.close();
+
+      if (throwable != null) {
+        instrumenter().end(context, otelRequest, null, throwable);
+      }
+
+      future.whenComplete(new OpenSearchResponseHandler(context, otelRequest));
+    }
+  }
+}