Added dynamodb instrumenter for aws v1_11 sdk (#12756)

Co-authored-by: Lauri Tulmin <ltulmin@splunk.com>
open-telemetry · Nov 30, 2024 · 6fa9553 · 6fa9553
1 parent 7c82dc4
commit 6fa9553
Show file tree

Hide file tree

Showing 11 changed files with 118 additions and 32 deletions.
diff --git a/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AwsSdkInstrumenterFactory.java b/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AwsSdkInstrumenterFactory.java
@@ -186,6 +186,18 @@ Instrumenter<Request<?>, Response<?>> producerInstrumenter() {
         true);
   }
 
+  Instrumenter<Request<?>, Response<?>> dynamoDbInstrumenter() {
+    DynamoDbAttributesExtractor dynamoDbAttributesExtractor = new DynamoDbAttributesExtractor();
+
+    return createInstrumenter(
+        openTelemetry,
+        spanName,
+        SpanKindExtractor.alwaysClient(),
+        attributesExtractors(),
+        singletonList(dynamoDbAttributesExtractor),
+        true);
+  }
+
   private static <REQUEST, RESPONSE> Instrumenter<REQUEST, RESPONSE> createInstrumenter(
       OpenTelemetry openTelemetry,
       SpanNameExtractor<REQUEST> spanNameExtractor,

diff --git a/.../library/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AwsSdkTelemetry.java b/.../library/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AwsSdkTelemetry.java
@@ -49,6 +49,7 @@ public static AwsSdkTelemetryBuilder builder(OpenTelemetry openTelemetry) {
   private final Instrumenter<SqsReceiveRequest, Response<?>> consumerReceiveInstrumenter;
   private final Instrumenter<SqsProcessRequest, Response<?>> consumerProcessInstrumenter;
   private final Instrumenter<Request<?>, Response<?>> producerInstrumenter;
+  private final Instrumenter<Request<?>, Response<?>> dynamoDbInstrumenter;
 
   AwsSdkTelemetry(
       OpenTelemetry openTelemetry,
@@ -65,6 +66,7 @@ public static AwsSdkTelemetryBuilder builder(OpenTelemetry openTelemetry) {
     consumerReceiveInstrumenter = instrumenterFactory.consumerReceiveInstrumenter();
     consumerProcessInstrumenter = instrumenterFactory.consumerProcessInstrumenter();
     producerInstrumenter = instrumenterFactory.producerInstrumenter();
+    dynamoDbInstrumenter = instrumenterFactory.dynamoDbInstrumenter();
   }
 
   /**
@@ -76,6 +78,7 @@ public RequestHandler2 newRequestHandler() {
         requestInstrumenter,
         consumerReceiveInstrumenter,
         consumerProcessInstrumenter,
-        producerInstrumenter);
+        producerInstrumenter,
+        dynamoDbInstrumenter);
   }
 }
diff --git a/.../main/java/io/opentelemetry/instrumentation/awssdk/v1_11/DynamoDbAttributesExtractor.java b/.../main/java/io/opentelemetry/instrumentation/awssdk/v1_11/DynamoDbAttributesExtractor.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.awssdk.v1_11;
+
+import com.amazonaws.Request;
+import com.amazonaws.Response;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import io.opentelemetry.instrumentation.api.internal.AttributesExtractorUtil;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+
+public class DynamoDbAttributesExtractor implements AttributesExtractor<Request<?>, Response<?>> {
+
+  // copied from DbIncubatingAttributes
+  private static final AttributeKey<String> DB_SYSTEM = AttributeKey.stringKey("db.system");
+  // copied from AwsIncubatingAttributes
+  private static final AttributeKey<List<String>> AWS_DYNAMODB_TABLE_NAMES =
+      AttributeKey.stringArrayKey("aws.dynamodb.table_names");
+
+  // copied from DbIncubatingAttributes.DbSystemIncubatingValues
+  private static final String DYNAMODB = "dynamodb";
+
+  @Override
+  public void onStart(AttributesBuilder attributes, Context parentContext, Request<?> request) {
+    AttributesExtractorUtil.internalSet(attributes, DB_SYSTEM, DYNAMODB);
+    String tableName = RequestAccess.getTableName(request.getOriginalRequest());
+    AttributesExtractorUtil.internalSet(
+        attributes, AWS_DYNAMODB_TABLE_NAMES, Collections.singletonList(tableName));
+  }
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      Request<?> request,
+      @Nullable Response<?> response,
+      @Nullable Throwable error) {}
+}
diff --git a/...ry/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/TracingRequestHandler.java b/...ry/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/TracingRequestHandler.java
@@ -31,21 +31,27 @@ final class TracingRequestHandler extends RequestHandler2 {
       ContextKey.named(TracingRequestHandler.class.getName() + ".Timer");
   private static final ContextKey<Boolean> REQUEST_SPAN_SUPPRESSED_KEY =
       ContextKey.named(TracingRequestHandler.class.getName() + ".RequestSpanSuppressed");
+  private static final String SEND_MESSAGE_REQUEST_CLASS =
+      "com.amazonaws.services.sqs.model.SendMessageRequest";
+  private static final String DYNAMODBV2_CLASS_PREFIX = "com.amazonaws.services.dynamodbv2.model.";
 
   private final Instrumenter<Request<?>, Response<?>> requestInstrumenter;
   private final Instrumenter<SqsReceiveRequest, Response<?>> consumerReceiveInstrumenter;
   private final Instrumenter<SqsProcessRequest, Response<?>> consumerProcessInstrumenter;
   private final Instrumenter<Request<?>, Response<?>> producerInstrumenter;
+  private final Instrumenter<Request<?>, Response<?>> dynamoDbInstrumenter;
 
   TracingRequestHandler(
       Instrumenter<Request<?>, Response<?>> requestInstrumenter,
       Instrumenter<SqsReceiveRequest, Response<?>> consumerReceiveInstrumenter,
       Instrumenter<SqsProcessRequest, Response<?>> consumerProcessInstrumenter,
-      Instrumenter<Request<?>, Response<?>> producerInstrumenter) {
+      Instrumenter<Request<?>, Response<?>> producerInstrumenter,
+      Instrumenter<Request<?>, Response<?>> dynamoDbInstrumenter) {
     this.requestInstrumenter = requestInstrumenter;
     this.consumerReceiveInstrumenter = consumerReceiveInstrumenter;
     this.consumerProcessInstrumenter = consumerProcessInstrumenter;
     this.producerInstrumenter = producerInstrumenter;
+    this.dynamoDbInstrumenter = dynamoDbInstrumenter;
   }
 
   @Override
@@ -151,14 +157,17 @@ private void finish(Request<?> request, Response<?> response, @Nullable Throwabl
       }
       return;
     }
-
     instrumenter.end(context, request, response, error);
   }
 
   private Instrumenter<Request<?>, Response<?>> getInstrumenter(Request<?> request) {
-    boolean isSqsProducer =
-        "com.amazonaws.services.sqs.model.SendMessageRequest"
-            .equals(request.getOriginalRequest().getClass().getName());
-    return isSqsProducer ? producerInstrumenter : requestInstrumenter;
+    String className = request.getOriginalRequest().getClass().getName();
+    if (className.startsWith(DYNAMODBV2_CLASS_PREFIX)) {
+      return dynamoDbInstrumenter;
+    }
+    if (className.equals(SEND_MESSAGE_REQUEST_CLASS)) {
+      return producerInstrumenter;
+    }
+    return requestInstrumenter;
   }
 }
diff --git a/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractBaseAwsClientTest.java b/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractBaseAwsClientTest.java
@@ -34,7 +34,6 @@
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -75,7 +74,7 @@ public void assertRequestWithMockedResponse(
       String service,
       String operation,
       String method,
-      Map<String, String> additionalAttributes)
+      List<AttributeAssertion> additionalAttributes)
       throws Exception {
 
     assertThat(response).isNotNull();
@@ -113,8 +112,7 @@ public void assertRequestWithMockedResponse(
                                 stringKey("aws.request_id"), v -> v.isInstanceOf(String.class)));
                       }
 
-                      additionalAttributes.forEach(
-                          (k, v) -> attributes.add(equalTo(stringKey(k), v)));
+                      attributes.addAll(additionalAttributes);
 
                       span.hasName(service + "." + operation)
                           .hasKind(operation.equals("SendMessage") ? PRODUCER : CLIENT)

diff --git a/...c/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractDynamoDbClientTest.java b/...c/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractDynamoDbClientTest.java
@@ -5,13 +5,22 @@
 
 package io.opentelemetry.instrumentation.awssdk.v1_11;
 
+import static io.opentelemetry.api.common.AttributeKey.stringKey;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.semconv.incubating.AwsIncubatingAttributes.AWS_DYNAMODB_TABLE_NAMES;
+import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_SYSTEM;
+import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DbSystemIncubatingValues.DYNAMODB;
+import static java.util.Collections.singletonList;
+
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
 import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
-import com.google.common.collect.ImmutableMap;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.testing.internal.armeria.common.HttpResponse;
 import io.opentelemetry.testing.internal.armeria.common.HttpStatus;
 import io.opentelemetry.testing.internal.armeria.common.MediaType;
+import java.util.Arrays;
+import java.util.List;
 import org.junit.jupiter.api.Test;
 
 public abstract class AbstractDynamoDbClientTest extends AbstractBaseAwsClientTest {
@@ -34,13 +43,14 @@ public void sendRequestWithMockedResponse() throws Exception {
 
     server.enqueue(HttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8, ""));
 
+    List<AttributeAssertion> additionalAttributes =
+        Arrays.asList(
+            equalTo(stringKey("aws.table.name"), "sometable"),
+            equalTo(DB_SYSTEM, DYNAMODB),
+            equalTo(AWS_DYNAMODB_TABLE_NAMES, singletonList("sometable")));
+
     Object response = client.createTable(new CreateTableRequest("sometable", null));
     assertRequestWithMockedResponse(
-        response,
-        client,
-        "DynamoDBv2",
-        "CreateTable",
-        "POST",
-        ImmutableMap.of("aws.table.name", "sometable"));
+        response, client, "DynamoDBv2", "CreateTable", "POST", additionalAttributes);
   }
 }
diff --git a/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractEc2ClientTest.java b/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractEc2ClientTest.java
@@ -41,6 +41,6 @@ public void sendRequestWithMockedResponse() throws Exception {
 
     Object response = client.allocateAddress();
     assertRequestWithMockedResponse(
-        response, client, "EC2", "AllocateAddress", "POST", Collections.emptyMap());
+        response, client, "EC2", "AllocateAddress", "POST", Collections.emptyList());
   }
 }
diff --git a/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractKinesisClientTest.java b/...rc/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractKinesisClientTest.java
@@ -5,14 +5,18 @@
 
 package io.opentelemetry.instrumentation.awssdk.v1_11;
 
+import static io.opentelemetry.api.common.AttributeKey.stringKey;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static java.util.Collections.singletonList;
+
 import com.amazonaws.services.kinesis.AmazonKinesis;
 import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
 import com.amazonaws.services.kinesis.model.DeleteStreamRequest;
-import com.google.common.collect.ImmutableMap;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.testing.internal.armeria.common.HttpResponse;
 import io.opentelemetry.testing.internal.armeria.common.HttpStatus;
 import io.opentelemetry.testing.internal.armeria.common.MediaType;
-import java.util.Map;
+import java.util.List;
 import java.util.function.Function;
 import java.util.stream.Stream;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -42,7 +46,9 @@ public void testSendRequestWithMockedResponse(
 
     server.enqueue(HttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8, ""));
 
-    Map<String, String> additionalAttributes = ImmutableMap.of("aws.stream.name", "somestream");
+    List<AttributeAssertion> additionalAttributes =
+        singletonList(equalTo(stringKey("aws.stream.name"), "somestream"));
+
     Object response = call.apply(client);
     assertRequestWithMockedResponse(
         response, client, "Kinesis", operation, "POST", additionalAttributes);

diff --git a/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractRdsClientTest.java b/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractRdsClientTest.java
@@ -42,6 +42,6 @@ public void sendRequestWithMockedResponse() throws Exception {
 
     Object response = client.deleteOptionGroup(new DeleteOptionGroupRequest());
     assertRequestWithMockedResponse(
-        response, client, "RDS", "DeleteOptionGroup", "POST", Collections.emptyMap());
+        response, client, "RDS", "DeleteOptionGroup", "POST", Collections.emptyList());
   }
 }
diff --git a/...ing/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractS3ClientTest.java b/...ing/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractS3ClientTest.java
@@ -17,6 +17,7 @@
 import static io.opentelemetry.semconv.incubating.RpcIncubatingAttributes.RPC_METHOD;
 import static io.opentelemetry.semconv.incubating.RpcIncubatingAttributes.RPC_SERVICE;
 import static io.opentelemetry.semconv.incubating.RpcIncubatingAttributes.RPC_SYSTEM;
+import static java.util.Collections.singletonList;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.catchThrowable;
 
@@ -27,14 +28,14 @@
 import com.amazonaws.retry.PredefinedRetryPolicies;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
-import com.google.common.collect.ImmutableMap;
 import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.sdk.trace.data.StatusData;
 import io.opentelemetry.testing.internal.armeria.common.HttpResponse;
 import io.opentelemetry.testing.internal.armeria.common.HttpStatus;
 import io.opentelemetry.testing.internal.armeria.common.MediaType;
 import java.time.Duration;
-import java.util.Map;
+import java.util.List;
 import java.util.function.Function;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Test;
@@ -60,7 +61,7 @@ public void testSendRequestWithMockedResponse(
       String operation,
       String method,
       Function<AmazonS3, Object> call,
-      Map<String, String> additionalAttributes)
+      List<AttributeAssertion> additionalAttributes)
       throws Exception {
 
     AmazonS3 client =
@@ -82,12 +83,12 @@ private static Stream<Arguments> provideArguments() {
             "CreateBucket",
             "PUT",
             (Function<AmazonS3, Object>) c -> c.createBucket("testbucket"),
-            ImmutableMap.of("aws.bucket.name", "testbucket")),
+            singletonList(equalTo(stringKey("aws.bucket.name"), "testbucket"))),
         Arguments.of(
             "GetObject",
             "GET",
             (Function<AmazonS3, Object>) c -> c.getObject("someBucket", "someKey"),
-            ImmutableMap.of("aws.bucket.name", "someBucket")));
+            singletonList(equalTo(stringKey("aws.bucket.name"), "someBucket"))));
   }
 
   @Test

diff --git a/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractSnsClientTest.java b/...ng/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractSnsClientTest.java
@@ -5,16 +5,18 @@
 
 package io.opentelemetry.instrumentation.awssdk.v1_11;
 
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
+import static java.util.Collections.singletonList;
 
 import com.amazonaws.services.sns.AmazonSNS;
 import com.amazonaws.services.sns.AmazonSNSClientBuilder;
 import com.amazonaws.services.sns.model.PublishRequest;
-import com.google.common.collect.ImmutableMap;
+import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
 import io.opentelemetry.testing.internal.armeria.common.HttpResponse;
 import io.opentelemetry.testing.internal.armeria.common.HttpStatus;
 import io.opentelemetry.testing.internal.armeria.common.MediaType;
-import java.util.Map;
+import java.util.List;
 import java.util.function.Function;
 import java.util.stream.Stream;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -52,8 +54,8 @@ public void testSendRequestWithMockedResponse(Function<AmazonSNS, Object> call)
 
     server.enqueue(HttpResponse.of(HttpStatus.OK, MediaType.PLAIN_TEXT_UTF_8, body));
 
-    Map<String, String> additionalAttributes =
-        ImmutableMap.of(MESSAGING_DESTINATION_NAME.toString(), "somearn");
+    List<AttributeAssertion> additionalAttributes =
+        singletonList(equalTo(MESSAGING_DESTINATION_NAME, "somearn"));
 
     Object response = call.apply(client);
     assertRequestWithMockedResponse(