Implement genai events for bedrock (non-streaming)

Author: anuraaga

instrumentation/aws-sdk/aws-sdk-2.2/javaagent/build.gradle.kts

@@ -147,6 +147,15 @@ testing {
           implementation("software.amazon.awssdk:bedrockruntime:2.25.63")
         }
       }
+
+      targets {
+        all {
+          testTask.configure {
+            // TODO run tests both with and without genai message capture
+            systemProperty("otel.instrumentation.genai.capture-message-content", "true")
+          }
+        }
+      }
     }
   }
 }

instrumentation/aws-sdk/aws-sdk-2.2/library-autoconfigure/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/autoconfigure/AwsSdkSingletons.java

@@ -24,6 +24,7 @@ public final class AwsSdkSingletons {
           .setMessagingReceiveInstrumentationEnabled(messagingReceiveInstrumentationEnabled())
           .setUseConfiguredPropagatorForMessaging(useMessagingPropagator())
           .setRecordIndividualHttpError(recordIndividualHttpError())
+          .setGenaiCaptureMessageContent(genaiCaptureMessageContent())
           .build();
 
   private static boolean hasAgentConfiguration() {
@@ -67,6 +68,11 @@ private static boolean recordIndividualHttpError() {
         "otel.instrumentation.aws-sdk.experimental-record-individual-http-error", false);
   }
 
+  private static boolean genaiCaptureMessageContent() {
+    return getBoolean(
+        "otel.instrumentation.genai.capture-message-content", false);
+  }
+
   private static boolean getBoolean(String name, boolean defaultValue) {
     if (HAS_INSTRUMENTATION_CONFIG) {
       return AgentInstrumentationConfig.get().getBoolean(name, defaultValue);

instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkTelemetry.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.instrumentation.awssdk.v2_2;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.logs.Logger;
 import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.awssdk.v2_2.internal.AwsSdkInstrumenterFactory;
@@ -56,10 +57,12 @@ public static AwsSdkTelemetryBuilder builder(OpenTelemetry openTelemetry) {
   private final Instrumenter<ExecutionAttributes, Response> producerInstrumenter;
   private final Instrumenter<ExecutionAttributes, Response> dynamoDbInstrumenter;
   private final Instrumenter<ExecutionAttributes, Response> bedrockRuntimeInstrumenter;
+  private final Logger eventLogger;
   private final boolean captureExperimentalSpanAttributes;
   @Nullable private final TextMapPropagator messagingPropagator;
   private final boolean useXrayPropagator;
   private final boolean recordIndividualHttpError;
+  private final boolean genAiCaptureMessageContent;
 
   AwsSdkTelemetry(
       OpenTelemetry openTelemetry,
@@ -68,7 +71,8 @@ public static AwsSdkTelemetryBuilder builder(OpenTelemetry openTelemetry) {
       boolean useMessagingPropagator,
       boolean useXrayPropagator,
       boolean recordIndividualHttpError,
-      boolean messagingReceiveInstrumentationEnabled) {
+      boolean messagingReceiveInstrumentationEnabled,
+      boolean genAiCaptureMessageContent) {
     this.useXrayPropagator = useXrayPropagator;
     this.messagingPropagator =
         useMessagingPropagator ? openTelemetry.getPropagators().getTextMapPropagator() : null;
@@ -88,8 +92,10 @@ public static AwsSdkTelemetryBuilder builder(OpenTelemetry openTelemetry) {
     this.producerInstrumenter = instrumenterFactory.producerInstrumenter();
     this.dynamoDbInstrumenter = instrumenterFactory.dynamoDbInstrumenter();
     this.bedrockRuntimeInstrumenter = instrumenterFactory.bedrockRuntimeInstrumenter();
+    this.eventLogger = instrumenterFactory.eventLogger();
     this.captureExperimentalSpanAttributes = captureExperimentalSpanAttributes;
     this.recordIndividualHttpError = recordIndividualHttpError;
+    this.genAiCaptureMessageContent = genAiCaptureMessageContent;
   }
 
   /**
@@ -104,10 +110,12 @@ public ExecutionInterceptor newExecutionInterceptor() {
         producerInstrumenter,
         dynamoDbInstrumenter,
         bedrockRuntimeInstrumenter,
+        eventLogger,
         captureExperimentalSpanAttributes,
         messagingPropagator,
         useXrayPropagator,
-        recordIndividualHttpError);
+        recordIndividualHttpError,
+        genAiCaptureMessageContent);
   }
 
   /**

instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/AwsSdkTelemetryBuilder.java

@@ -24,6 +24,7 @@ public final class AwsSdkTelemetryBuilder {
   private boolean recordIndividualHttpError;
   private boolean useXrayPropagator = true;
   private boolean messagingReceiveInstrumentationEnabled;
+  private boolean genaiCaptureMessageContent;
 
   AwsSdkTelemetryBuilder(OpenTelemetry openTelemetry) {
     this.openTelemetry = openTelemetry;
@@ -115,6 +116,18 @@ public AwsSdkTelemetryBuilder setMessagingReceiveInstrumentationEnabled(
     return this;
   }
 
+  /**
+   * Set whether Generative AI events include full content of user and assistant messages.
+   *
+   * <p>Note that full content can have data privacy and size concerns and care should be taken when
+   * enabling this.
+   */
+  @CanIgnoreReturnValue
+  public AwsSdkTelemetryBuilder setGenaiCaptureMessageContent(boolean genaiCaptureMessageContent) {
+    this.genaiCaptureMessageContent = genaiCaptureMessageContent;
+    return this;
+  }
+
   /**
    * Returns a new {@link AwsSdkTelemetry} with the settings of this {@link AwsSdkTelemetryBuilder}.
    */
@@ -126,6 +139,7 @@ public AwsSdkTelemetry build() {
         useMessagingPropagator,
         useXrayPropagator,
         recordIndividualHttpError,
-        messagingReceiveInstrumentationEnabled);
+        messagingReceiveInstrumentationEnabled,
+        genaiCaptureMessageContent);
   }
 }

instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/internal/AwsSdkInstrumenterFactory.java

@@ -10,6 +10,7 @@
 
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.logs.Logger;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.propagation.TextMapPropagator;
@@ -236,6 +237,10 @@ public Instrumenter<ExecutionAttributes, Response> bedrockRuntimeInstrumenter()
         true);
   }
 
+  public Logger eventLogger() {
+    return openTelemetry.getLogsBridge().get(INSTRUMENTATION_NAME);
+  }
+
   private static <REQUEST, RESPONSE> Instrumenter<REQUEST, RESPONSE> createInstrumenter(
       OpenTelemetry openTelemetry,
       SpanNameExtractor<REQUEST> spanNameExtractor,

instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/internal/BedrockRuntimeAccess.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.instrumentation.awssdk.v2_2.internal;
 
+import io.opentelemetry.api.logs.Logger;
+import io.opentelemetry.context.Context;
 import io.opentelemetry.javaagent.tooling.muzzle.NoMuzzle;
 import java.util.List;
 import javax.annotation.Nullable;
@@ -37,6 +39,11 @@ static boolean isBedrockRuntimeRequest(SdkRequest request) {
     return enabled && BedrockRuntimeImpl.isBedrockRuntimeRequest(request);
   }
 
+  @NoMuzzle
+  static boolean isBedrockRuntimeResponse(SdkResponse response) {
+    return enabled && BedrockRuntimeImpl.isBedrockRuntimeResponse(response);
+  }
+
   @Nullable
   @NoMuzzle
   static String getModelId(SdkRequest request) {
@@ -84,4 +91,25 @@ static Long getUsageInputTokens(SdkResponse response) {
   static Long getUsageOutputTokens(SdkResponse response) {
     return enabled ? BedrockRuntimeImpl.getUsageOutputTokens(response) : null;
   }
+
+  @NoMuzzle
+  static void recordRequestEvents(
+      Context otelContext, Logger eventLogger, SdkRequest request, boolean captureMessageContent) {
+    if (enabled) {
+      BedrockRuntimeImpl.recordRequestEvents(
+          otelContext, eventLogger, request, captureMessageContent);
+    }
+  }
+
+  @NoMuzzle
+  static void recordResponseEvents(
+      Context otelContext,
+      Logger eventLogger,
+      SdkResponse response,
+      boolean captureMessageContent) {
+    if (enabled) {
+      BedrockRuntimeImpl.recordResponseEvents(
+          otelContext, eventLogger, response, captureMessageContent);
+    }
+  }
 }

instrumentation/aws-sdk/aws-sdk-2.2/library/src/main/java/io/opentelemetry/instrumentation/awssdk/v2_2/internal/BedrockRuntimeAttributesGetter.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.instrumentation.awssdk.v2_2.internal;
 
 import static io.opentelemetry.instrumentation.awssdk.v2_2.internal.TracingExecutionInterceptor.SDK_REQUEST_ATTRIBUTE;
+import static java.util.Collections.emptyList;
 
 import io.opentelemetry.instrumentation.api.incubator.semconv.genai.GenAiAttributesGetter;
 import java.util.Arrays;
@@ -25,7 +26,7 @@ private static final class GenAiOperationNameIncubatingValues {
     private GenAiOperationNameIncubatingValues() {}
   }
 
-  private static final class GenAiSystemIncubatingValues {
+  static final class GenAiSystemIncubatingValues {
     static final String AWS_BEDROCK = "aws.bedrock";
 
     private GenAiSystemIncubatingValues() {}
@@ -116,10 +117,13 @@ public Double getRequestTopP(ExecutionAttributes executionAttributes) {
   @Nullable
   @Override
   public List<String> getResponseFinishReasons(
-      ExecutionAttributes executionAttributes, Response response) {
+      ExecutionAttributes executionAttributes, @Nullable Response response) {
+    if (response == null) {
+      return emptyList();
+    }
     String stopReason = BedrockRuntimeAccess.getStopReason(response.getSdkResponse());
     if (stopReason == null) {
-      return null;
+      return emptyList();
     }
     return Arrays.asList(stopReason);
   }
@@ -138,13 +142,21 @@ public String getResponseModel(ExecutionAttributes executionAttributes, Response
 
   @Nullable
   @Override
-  public Long getUsageInputTokens(ExecutionAttributes executionAttributes, Response response) {
+  public Long getUsageInputTokens(
+      ExecutionAttributes executionAttributes, @Nullable Response response) {
+    if (response == null) {
+      return null;
+    }
     return BedrockRuntimeAccess.getUsageInputTokens(response.getSdkResponse());
   }
 
   @Nullable
   @Override
-  public Long getUsageOutputTokens(ExecutionAttributes executionAttributes, Response response) {
+  public Long getUsageOutputTokens(
+      ExecutionAttributes executionAttributes, @Nullable Response response) {
+    if (response == null) {
+      return null;
+    }
     return BedrockRuntimeAccess.getUsageOutputTokens(response.getSdkResponse());
   }
 }