Inject trace context into AWS Step Functions input

dd-java-agent/instrumentation/aws-java-sfn-2.0/build.gradle

@@ -0,0 +1,29 @@
+muzzle {
+  pass {
+    group = "software.amazon.awssdk"
+    module = "sfn"
+    versions = "[2.15.35,)"
+    assertInverse = true
+  }
+}
+
+apply from: "$rootDir/gradle/java.gradle"
+
+addTestSuiteForDir('latestDepTest', 'test')
+addTestSuiteExtendingForDir('latestDepForkedTest', 'latestDepTest', 'test')
+
+dependencies {
+  compileOnly group: 'software.amazon.awssdk', name: 'sfn', version: '2.15.35'
+
+  // Include httpclient instrumentation for testing because it is a dependency for aws-sdk.
+  testImplementation project(':dd-java-agent:instrumentation:apache-httpclient-4')
+  testImplementation project(':dd-java-agent:instrumentation:aws-java-sdk-2.2')
+  testImplementation 'software.amazon.awssdk:sfn:2.27.2'
+  testImplementation 'org.testcontainers:localstack:1.19.7'
+
+  latestDepTestImplementation group: 'software.amazon.awssdk', name: 'sfn', version: '+'
+}
+
+tasks.withType(Test).configureEach {
+  usesService(testcontainersLimit)
+}

dd-java-agent/instrumentation/aws-java-sfn-2.0/src/main/java/datadog/trace/instrumentation/aws/v2/sfn/InputAttributeInjector.java

@@ -0,0 +1,44 @@
+package datadog.trace.instrumentation.aws.v2.sfn;
+
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+
+public class InputAttributeInjector {
+  public static String buildTraceContext(AgentSpan span) {
+    // Extract span tags
+    StringBuilder spanTagsJSON = new StringBuilder();
+    spanTagsJSON.append('{');
+    span.getTags()
+        .forEach(
+            (tagKey, tagValue) ->
+                spanTagsJSON
+                    .append("\"")
+                    .append(tagKey)
+                    .append("\":\"")
+                    .append(tagValue)
+                    .append("\","));
+    spanTagsJSON.setLength(spanTagsJSON.length() - 1); // remove trailing comma
+    spanTagsJSON.append('}');
+
+    // Build DD trace context object
+    String ddTraceContextJSON =
+        String.format(
+            "\"_datadog\": { \"x-datadog-trace-id\": \"%s\",\"x-datadog-parent-id\":\"%s\", \"x-datadog-tags\": %s }",
+            span.getTraceId().toString(), span.getSpanId(), spanTagsJSON);
+
+    return ddTraceContextJSON;
+  }
+
+  public static StringBuilder getModifiedInput(String request, String ddTraceContextJSON) {
+    StringBuilder modifiedInput = new StringBuilder(request);
+    int startPos = modifiedInput.indexOf("{");
+    int endPos = modifiedInput.lastIndexOf("}");
+    String inputContent = modifiedInput.substring(startPos + 1, endPos);
+    if (inputContent.isEmpty()) {
+      modifiedInput.insert(endPos, ddTraceContextJSON);
+    } else {
+      modifiedInput.insert(
+          endPos, String.format(", %s", ddTraceContextJSON)); // prepend comma to existing input
+    }
+    return modifiedInput;
+  }
+}

dd-java-agent/instrumentation/aws-java-sfn-2.0/src/main/java/datadog/trace/instrumentation/aws/v2/sfn/SfnClientInstrumentation.java

@@ -0,0 +1,50 @@
+package datadog.trace.instrumentation.aws.v2.sfn;
+
+import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.Instrumenter;
+import datadog.trace.agent.tooling.InstrumenterModule;
+import java.util.List;
+import net.bytebuddy.asm.Advice;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+
+/** AWS SDK v2 Step Function instrumentation */
+@AutoService(InstrumenterModule.class)
+public final class SfnClientInstrumentation extends InstrumenterModule.Tracing
+    implements Instrumenter.ForSingleType {
+
+  public SfnClientInstrumentation() {
+    super("sfn", "aws-sdk");
+  }
+
+  @Override
+  public String instrumentedType() {
+    return "software.amazon.awssdk.core.client.builder.SdkDefaultClientBuilder";
+  }
+
+  @Override
+  public void methodAdvice(MethodTransformer transformer) {
+    transformer.applyAdvice(
+        isMethod().and(named("resolveExecutionInterceptors")),
+        SfnClientInstrumentation.class.getName() + "$AwsSfnBuilderAdvice");
+  }
+
+  @Override
+  public String[] helperClassNames() {
+    return new String[] {packageName + ".SfnInterceptor", packageName + ".InputAttributeInjector"};
+  }
+
+  public static class AwsSfnBuilderAdvice {
+    @Advice.OnMethodExit(suppress = Throwable.class)
+    public static void addHandler(@Advice.Return final List<ExecutionInterceptor> interceptors) {
+      for (ExecutionInterceptor interceptor : interceptors) {
+        if (interceptor instanceof SfnInterceptor) {
+          return;
+        }
+      }
+      interceptors.add(new SfnInterceptor());
+    }
+  }
+}

dd-java-agent/instrumentation/aws-java-sfn-2.0/src/main/java/datadog/trace/instrumentation/aws/v2/sfn/SfnInterceptor.java

@@ -0,0 +1,55 @@
+package datadog.trace.instrumentation.aws.v2.sfn;
+
+import datadog.trace.bootstrap.InstanceStore;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.Context;
+import software.amazon.awssdk.core.interceptor.ExecutionAttribute;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.services.sfn.model.StartExecutionRequest;
+import software.amazon.awssdk.services.sfn.model.StartSyncExecutionRequest;
+
+public class SfnInterceptor implements ExecutionInterceptor {
+
+  public static final ExecutionAttribute<AgentSpan> SPAN_ATTRIBUTE =
+      InstanceStore.of(ExecutionAttribute.class)
+          .putIfAbsent("DatadogSpan", () -> new ExecutionAttribute<>("DatadogSpan"));
+
+  public SfnInterceptor() {}
+
+  @Override
+  public SdkRequest modifyRequest(
+      Context.ModifyRequest context, ExecutionAttributes executionAttributes) {
+    final AgentSpan span = executionAttributes.getAttribute(SPAN_ATTRIBUTE);
+    // StartExecutionRequest
+    if (context.request() instanceof StartExecutionRequest) {
+      StartExecutionRequest request = (StartExecutionRequest) context.request();
+      if (request.input() == null) {
+        return request;
+      }
+      String ddTraceContextJSON = InputAttributeInjector.buildTraceContext(span);
+      // Inject the trace context into the Step Function input
+      StringBuilder modifiedInput =
+          InputAttributeInjector.getModifiedInput(request.input(), ddTraceContextJSON);
+
+      return request.toBuilder().input(modifiedInput.toString()).build();
+    }
+
+    // StartSyncExecutionRequest
+    if (context.request() instanceof StartSyncExecutionRequest) {
+      StartSyncExecutionRequest request = (StartSyncExecutionRequest) context.request();
+      if (request.input() == null) {
+        return request;
+      }
+      String ddTraceContextJSON = InputAttributeInjector.buildTraceContext(span);
+      // Inject the trace context into the Step Function input
+      StringBuilder modifiedInput =
+          InputAttributeInjector.getModifiedInput(request.input(), ddTraceContextJSON);
+
+      return request.toBuilder().input(modifiedInput.toString()).build();
+    }
+
+    return context.request();
+  }
+}

dd-java-agent/instrumentation/aws-java-sfn-2.0/src/test/groovy/SfnClientTest.groovy

@@ -0,0 +1,137 @@
+import datadog.trace.agent.test.naming.VersionedNamingTestBase
+import datadog.trace.agent.test.utils.TraceUtils
+import datadog.trace.api.DDSpanTypes
+import datadog.trace.bootstrap.instrumentation.api.Tags
+import groovy.json.JsonSlurper
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.utility.DockerImageName
+import software.amazon.awssdk.services.sfn.SfnClient
+import software.amazon.awssdk.services.sfn.model.StartExecutionResponse
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials
+import spock.lang.Shared
+
+import java.time.Duration
+
+import static datadog.trace.agent.test.utils.TraceUtils.basicSpan
+
+
+abstract class SfnClientTest extends VersionedNamingTestBase {
+  static final LOCALSTACK = new GenericContainer(DockerImageName.parse("localstack/localstack"))
+  .withExposedPorts(4566)
+  .withEnv("SERVICES", "stepfunctions")
+  .withReuse(true)
+  .withStartupTimeout(Duration.ofSeconds(120))
+
+  @Shared SfnClient sfnClient
+
+  @Shared String testStateMachineARN
+
+  def setupSpec() {
+    LOCALSTACK.start()
+    def endPoint = "http://" + LOCALSTACK.getHost() + ":" + LOCALSTACK.getMappedPort(4566)
+    sfnClient = SfnClient.builder()
+      .endpointOverride(URI.create(endPoint))
+      .region(Region.US_EAST_1)
+      .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("test", "test")))
+      .build()
+
+    def response = sfnClient.createStateMachine { builder ->
+      builder.name("testStateMachine")
+        .definition("{\"StartAt\": \"HelloWorld\", \"States\": {\"HelloWorld\": {\"Type\": \"Pass\", \"End\": true}}}")
+        .build()
+    }
+    testStateMachineARN = response.stateMachineArn()
+  }
+
+  def cleanupSpec() {
+    LOCALSTACK.stop()
+  }
+
+  def "Step Functions span is created"() {
+    when:
+    StartExecutionResponse response
+    TraceUtils.runUnderTrace('parent', {
+      response = sfnClient.startExecution { builder ->
+        builder.stateMachineArn(testStateMachineARN)
+          .input("{\"key\": \"value\"}")
+          .build()
+      }
+    })
+
+    def endPoint = "http://" + LOCALSTACK.getHost() + ":" + LOCALSTACK.getMappedPort(4566)
+
+
+    then:
+    assertTraces(1) {
+      trace(2) {
+        basicSpan(it, "parent")
+        span {
+          serviceName "java-aws-sdk"
+          operationName "aws.http"
+          resourceName "Sfn.StartExecution"
+          spanType DDSpanTypes.HTTP_CLIENT
+          errored false
+          measured true
+          childOf(span(0))
+          tags {
+            "$Tags.COMPONENT" "java-aws-sdk"
+            "$Tags.SPAN_KIND" Tags.SPAN_KIND_CLIENT
+            "$Tags.HTTP_URL" endPoint+'/'
+            "$Tags.HTTP_METHOD" "POST"
+            "$Tags.HTTP_STATUS" 200
+            "$Tags.PEER_PORT" LOCALSTACK.getMappedPort(4566)
+            "$Tags.PEER_HOSTNAME" LOCALSTACK.getHost()
+            "aws.service" "Sfn"
+            "aws.operation" "StartExecution"
+            "aws.agent" "java-aws-sdk"
+            "aws.requestId" response.responseMetadata().requestId()
+            "aws_service" "Sfn"
+            defaultTags()
+          }
+        }
+      }
+    }
+  }
+
+  def "Trace context is injected to Step Functions input"() {
+    when:
+    StartExecutionResponse response
+    TraceUtils.runUnderTrace('parent', {
+      response = sfnClient.startExecution { builder ->
+        builder.stateMachineArn(testStateMachineARN)
+          .input("{\"key\": \"value\"}")
+          .build()
+      }
+    })
+
+    then:
+    def execution = sfnClient.describeExecution { builder ->
+      builder.executionArn(response.executionArn())
+        .build()
+    }
+    def input = new JsonSlurper().parseText(execution.input())
+    input["key"] == "value"
+    input["_datadog"]["x-datadog-trace-id"] != null
+    input["_datadog"]["x-datadog-parent-id"] != null
+    input["_datadog"]["x-datadog-tags"] != null
+  }
+}
+
+class SfnClientV0Test extends SfnClientTest {
+  @Override
+  int version() {
+    0
+  }
+
+  @Override
+  String service() {
+    return null
+  }
+
+  @Override
+  String operation() {
+    return null
+  }
+}

settings.gradle

@@ -185,6 +185,7 @@ include ':dd-java-agent:instrumentation:aws-java-sns-1.0'
 include ':dd-java-agent:instrumentation:aws-java-sns-2.0'
 include ':dd-java-agent:instrumentation:aws-java-sqs-1.0'
 include ':dd-java-agent:instrumentation:aws-java-sqs-2.0'
+include ':dd-java-agent:instrumentation:aws-java-sfn-2.0'
 include ':dd-java-agent:instrumentation:aws-lambda-handler'
 include ':dd-java-agent:instrumentation:axis-2'
 include ':dd-java-agent:instrumentation:axway-api'