Convert Elasticsearch Transport to Instrumenter API

instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/instrumenter/net/NetAttributesExtractor.java

@@ -22,8 +22,15 @@ public abstract class NetAttributesExtractor<REQUEST, RESPONSE>
   @Override
   protected final void onStart(AttributesBuilder attributes, REQUEST request) {
     set(attributes, SemanticAttributes.NET_TRANSPORT, transport(request));
-    set(attributes, SemanticAttributes.NET_PEER_IP, peerIp(request, null));
-    set(attributes, SemanticAttributes.NET_PEER_NAME, peerName(request, null));
+
+    String peerIp = peerIp(request, null);
+    String peerName = peerName(request, null);
+
+    if (peerName != null && !peerName.equals(peerIp)) {
+      set(attributes, SemanticAttributes.NET_PEER_NAME, peerName);
+    }
+    set(attributes, SemanticAttributes.NET_PEER_IP, peerIp);
+
     Integer peerPort = peerPort(request, null);
     if (peerPort != null) {
       set(attributes, SemanticAttributes.NET_PEER_PORT, (long) peerPort);
@@ -36,8 +43,15 @@ protected final void onEnd(
       REQUEST request,
       @Nullable RESPONSE response,
       @Nullable Throwable error) {
-    set(attributes, SemanticAttributes.NET_PEER_IP, peerIp(request, response));
-    set(attributes, SemanticAttributes.NET_PEER_NAME, peerName(request, response));
+
+    String peerIp = peerIp(request, response);
+    String peerName = peerName(request, response);
+
+    if (peerName != null && !peerName.equals(peerIp)) {
+      set(attributes, SemanticAttributes.NET_PEER_NAME, peerName);
+    }
+    set(attributes, SemanticAttributes.NET_PEER_IP, peerIp);
+
     Integer peerPort = peerPort(request, response);
     if (peerPort != null) {
       set(attributes, SemanticAttributes.NET_PEER_PORT, (long) peerPort);

instrumentation-api/src/test/java/io/opentelemetry/instrumentation/api/instrumenter/net/NetAttributesExtractorTest.java

@@ -87,4 +87,40 @@ void normal() {
             entry(SemanticAttributes.NET_PEER_PORT, 42L),
             entry(SemanticAttributes.NET_PEER_IP, "4.3.2.1"));
   }
+
+  @Test
+  public void doesNotSetDuplicateAttributes() {
+    // given
+    Map<String, String> request = new HashMap<>();
+    request.put("transport", "TCP");
+    request.put("peerName", "1.2.3.4");
+    request.put("peerIp", "1.2.3.4");
+    request.put("peerPort", "123");
+
+    Map<String, String> response = new HashMap<>();
+    response.put("peerName", "4.3.2.1");
+    response.put("peerPort", "42");
+    response.put("peerIp", "4.3.2.1");
+
+    TestNetAttributesExtractor extractor = new TestNetAttributesExtractor();
+
+    // when
+    AttributesBuilder startAttributes = Attributes.builder();
+    extractor.onStart(startAttributes, request);
+
+    AttributesBuilder endAttributes = Attributes.builder();
+    extractor.onEnd(endAttributes, request, response, null);
+
+    // then
+    assertThat(startAttributes.build())
+        .containsOnly(
+            entry(SemanticAttributes.NET_TRANSPORT, "TCP"),
+            entry(SemanticAttributes.NET_PEER_PORT, 123L),
+            entry(SemanticAttributes.NET_PEER_IP, "1.2.3.4"));
+
+    assertThat(endAttributes.build())
+        .containsOnly(
+            entry(SemanticAttributes.NET_PEER_PORT, 42L),
+            entry(SemanticAttributes.NET_PEER_IP, "4.3.2.1"));
+  }
 }

instrumentation/elasticsearch/elasticsearch-transport-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/elasticsearch/transport/v5_0/AbstractClientInstrumentation.java

@@ -6,7 +6,7 @@
 package io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.v5_0;
 
 import static io.opentelemetry.javaagent.instrumentation.api.Java8BytecodeBridge.currentContext;
-import static io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticsearchTransportClientTracer.tracer;
+import static io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.v5_0.Elasticsearch5TransportSingletons.instrumenter;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
@@ -15,6 +15,8 @@
 import io.opentelemetry.context.Scope;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticTransportRequest;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.TransportActionListener;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;
@@ -51,27 +53,38 @@ public static void onEnter(
         @Advice.Argument(1) ActionRequest actionRequest,
         @Advice.Local("otelContext") Context context,
         @Advice.Local("otelScope") Scope scope,
+        @Advice.Local("otelRequest") ElasticTransportRequest transportRequest,
         @Advice.Argument(value = 2, readOnly = false)
             ActionListener<ActionResponse> actionListener) {
 
+      transportRequest = ElasticTransportRequest.create(action, actionRequest);
       Context parentContext = currentContext();
-      context = tracer().startSpan(parentContext, null, action);
+      if (!instrumenter().shouldStart(parentContext, transportRequest)) {
+        return;
+      }
+
+      context = instrumenter().start(parentContext, transportRequest);
       scope = context.makeCurrent();
 
-      tracer().onRequest(context, action.getClass(), actionRequest.getClass());
       actionListener =
-          new TransportActionListener<>(actionRequest, actionListener, context, parentContext);
+          new TransportActionListener<>(
+              instrumenter(), transportRequest, actionListener, context, parentContext);
     }
 
     @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
     public static void stopSpan(
         @Advice.Thrown Throwable throwable,
         @Advice.Local("otelContext") Context context,
-        @Advice.Local("otelScope") Scope scope) {
+        @Advice.Local("otelScope") Scope scope,
+        @Advice.Local("otelRequest") ElasticTransportRequest transportRequest) {
+      if (scope == null) {
+        return;
+      }
+
       scope.close();
 
       if (throwable != null) {
-        tracer().endExceptionally(context, throwable);
+        instrumenter().end(context, transportRequest, null, throwable);
       }
     }
   }

instrumentation/elasticsearch/elasticsearch-transport-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/elasticsearch/transport/v5_0/Elasticsearch5TransportExperimentalAttributesExtractor.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.v5_0;
+
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticTransportRequest;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticsearchTransportExperimentalAttributesExtractor;
+import org.elasticsearch.action.DocumentRequest;
+
+public class Elasticsearch5TransportExperimentalAttributesExtractor
+    extends ElasticsearchTransportExperimentalAttributesExtractor {
+
+  @Override
+  protected void onStart(AttributesBuilder attributes, ElasticTransportRequest transportRequest) {
+    super.onStart(attributes, transportRequest);
+
+    Object request = transportRequest.getRequest();
+    if (request instanceof DocumentRequest) {
+      DocumentRequest<?> req = (DocumentRequest<?>) request;
+      attributes.put("elasticsearch.request.write.type", req.type());
+      attributes.put("elasticsearch.request.write.routing", req.routing());
+    }
+  }
+}

instrumentation/elasticsearch/elasticsearch-transport-5.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/elasticsearch/transport/v5_0/Elasticsearch5TransportSingletons.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.v5_0;
+
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticTransportNetAttributesExtractor;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticTransportRequest;
+import io.opentelemetry.javaagent.instrumentation.elasticsearch.transport.ElasticsearchTransportInstrumenterFactory;
+import org.elasticsearch.action.ActionResponse;
+
+public final class Elasticsearch5TransportSingletons {
+
+  private static final Instrumenter<ElasticTransportRequest, ActionResponse> INSTRUMENTER =
+      ElasticsearchTransportInstrumenterFactory.create(
+          "io.opentelemetry.elasticsearch-transport-5.0",
+          new Elasticsearch5TransportExperimentalAttributesExtractor(),
+          new ElasticTransportNetAttributesExtractor());
+
+  public static Instrumenter<ElasticTransportRequest, ActionResponse> instrumenter() {
+    return INSTRUMENTER;
+  }
+
+  private Elasticsearch5TransportSingletons() {}
+}

instrumentation/elasticsearch/elasticsearch-transport-5.0/javaagent/src/test/groovy/Elasticsearch5NodeClientTest.groovy

@@ -243,7 +243,7 @@ class Elasticsearch5NodeClientTest extends AbstractElasticsearchNodeClientTest {
           }
         }
       }
-      trace(3, 2) {
+      trace(3, 1) {
         span(0) {
           name "IndexAction"
           kind CLIENT
@@ -260,17 +260,6 @@ class Elasticsearch5NodeClientTest extends AbstractElasticsearchNodeClientTest {
             "elasticsearch.shard.replication.failed" 0
           }
         }
-        span(1) {
-          name "PutMappingAction"
-          kind CLIENT
-          childOf span(0)
-          attributes {
-            "${SemanticAttributes.DB_SYSTEM.key}" "elasticsearch"
-            "${SemanticAttributes.DB_OPERATION.key}" "PutMappingAction"
-            "elasticsearch.action" "PutMappingAction"
-            "elasticsearch.request" "PutMappingRequest"
-          }
-        }
       }
       trace(4, 1) {
         span(0) {