Merge Step Functions and Lambda Traces

문서 > 서버리스 > Serverless Monitoring for AWS Step Functions > Merge Step Functions and Lambda Traces

이 페이지는 아직 영어로 제공되지 않습니다. 번역 작업 중입니다.
현재 번역 프로젝트에 대한 질문이나 피드백이 있으신 경우 언제든지 연락주시기 바랍니다.

This page describes how to merge your AWS Step Functions traces with related AWS Lambda traces or nested Step Functions traces. These instructions assume that you have already instrumented these AWS Step Functions and Lambda functions to send traces to Datadog.

Datadog recommends using JSONata to define your Step Function definitions for the most complete end-to-end tracing experience. This approach ensures that any context upstream to the Step Function is preserved and passed down.

Merge Step Functions traces with downstream Lambda traces

With JSONata

Requirements

Runtime	Requirement
Node.js	Datadog Lambda Library for Node.js layer v116+
Python	Datadog Lambda Library for Python layer v103+
Other	Datadog Extension v75+

Your State Machine definition must use JSONata as the query language. To enable this, set your definition’s top-level QueryLanguage field to JSONata.

Setup

On the Lambda Task, set the Payload in the Arguments field as follows:

"Lambda Invoke": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke",
  "Output": "{% $states.result.Payload %}",
  "Arguments": {
    "FunctionName": "MyFunctionName",
    "Payload": "{% ($execInput := $states.context.Execution.Input; $hasDatadogTraceId := $exists($execInput._datadog.`x-datadog-trace-id`); $hasDatadogRootExecutionId := $exists($execInput._datadog.RootExecutionId); $ddTraceContext := $hasDatadogTraceId ? {'x-datadog-trace-id': $execInput._datadog.`x-datadog-trace-id`, 'x-datadog-tags': $execInput._datadog.`x-datadog-tags`} : {'RootExecutionId': $hasDatadogRootExecutionId ? $execInput._datadog.RootExecutionId : $states.context.Execution.Id}; $sfnContext := $merge([$states.context, {'Execution': $sift($states.context.Execution, function($v, $k) { $k != 'Input' })}]); $merge([$states.input, {'_datadog': $merge([$sfnContext, $ddTraceContext, {'serverless-version': 'v1'}])}])) %}"
  }
}

The JSONata expression merges the upstream service’s context with the current Step Functions context object and the Lambda state’s input payload.

Alternatively, if you have business logic defined in the payload, you can replace $states.input at the end of the JSONata expression with your intended value for the Payload key.

With JSONPath

Requirements

Runtime	Requirement
Node.js	Datadog Lambda Library for Node.js layer v112+
Python	Datadog Lambda Library for Python layer v95+

Your State Machine definition is using JSONPath. If your definition’s top-level QueryLanguage field is omitted, it defaults to JSONPath.

Setup

If you have not already, install the Datadog Serverless Framework Plugin v5.40.0+:
```
serverless plugin install --name serverless-plugin-datadog
```
Ensure you have deployed the Datadog Lambda Forwarder, a Lambda function that ships logs from AWS to Datadog, and that you are using v3.130.0+. You might need to update your Forwarder.
Take note of your Forwarder’s ARN.
Add the following to your serverless.yml:
```
custom:
  datadog:
    site: <DATADOG_SITE>
    apiKeySecretArn: <DATADOG_API_KEY_SECRET_ARN>
    forwarderArn: <FORWARDER_ARN>
    enableStepFunctionsTracing: true
    propagateUpstreamTrace: true
    mergeStepFunctionAndLambdaTraces: true
```
- Replace <DATADOG_SITE> with (ensure the correct SITE is selected on the right).
- Replace <DATADOG_API_KEY_SECRET_ARN> with the ARN of the AWS secret where your Datadog API key is securely stored. The key needs to be stored as a plaintext string (not a JSON blob). The secretsmanager:GetSecretValue permission is required. For testing, you can instead use apiKey and set the Datadog API key in plaintext.
- Replace <FORWARDER_ARN> with the ARN of your Datadog Lambda Forwarder. This step configures the log stream subscription for the Forwarder. Ensure that the Step Function log group name begins with /aws/vendedlogs/states/. If it does not, you must set it up manually.
For additional settings, see Datadog Serverless Framework Plugin - Configuration parameters.

If you have not already, install the [Datadog CLI][9] v2.18.0+.
```
npm install -g @datadog/datadog-ci
```
Ensure you have deployed the Datadog Lambda Forwarder, a Lambda function that ships logs from AWS to Datadog, and that you are using v3.130.0+. You may need to update your Forwarder.
Take note of your Forwarder’s ARN.
Instrument your Step Function.
```
datadog-ci stepfunctions instrument \
 --step-function <STEP_FUNCTION_ARN> \
 --forwarder <FORWARDER_ARN> \
 --env <ENVIRONMENT> \
 --propagate-upstream-trace \
 --merge-step-function-and-lambda-traces
```
- Replace <STEP_FUNCTION_ARN> with the ARN of your Step Function. Repeat the --step-function flag for each Step Function you wish to instrument.
- Replace <FORWARDER_ARN> with the ARN of your Datadog Lambda Forwarder, as noted previously. This step configures the log stream subscription for the Forwarder. Ensure that the Step Function log group name begins with “/aws/vendedlogs/states/”. If it does not, you will need to set it up manually.
- Replace <ENVIRONMENT> with the environment tag you would like to apply to your Step Functions.
For more information about the datadog-ci stepfunctions command, see the Datadog CLI documentation.

Modify your Lambda task payload or Step Function task input.

Example:

from aws_cdk import (
    aws_lambda,
    aws_stepfunctions as sfn,
    aws_stepfunctions_tasks as tasks,
)
from datadog_cdk_constructs_v2 import DatadogStepFunctions, DatadogLambda

lambda_function = aws_lambda.Function(...)
lambda_task = tasks.LambdaInvoke(
    self,
    "MyLambdaTask",
    lambda_function=lambda_function,
    payload=sfn.TaskInput.from_object(
        DatadogStepFunctions.build_lambda_payload_to_merge_traces(
            {"custom-key": "custom-value"}
        )
    ),
)

child_state_machine = sfn.StateMachine(...)
invoke_child_state_machine_task = tasks.StepFunctionsStartExecution(
    self,
    "InvokeChildStateMachineTask",
    state_machine=child_state_machine,
    input=sfn.TaskInput.from_object(
        DatadogStepFunctions.build_step_function_task_input_to_merge_traces(
            {"custom-key": "custom-value"}
        )
    ),
)

state_machine = sfn.StateMachine(
    self,
    "CdkPythonTestStateMachine",
    definition_body=sfn.DefinitionBody.from_chainable(
        lambda_task.next(invoke_child_state_machine_task)
    ),
)

datadog_lambda = DatadogLambda(...)
datadog_lambda.add_lambda_functions([lambda_function])

datadog_sfn = DatadogStepFunctions(...)
datadog_sfn.add_state_machines([child_state_machine, state_machine])

For additional code examples in TypeScript and Go, see [CDK Examples for Instrumenting AWS Step Functions][13].

On the Lambda Task, set the Parameters key as follows:

"Parameters": {
  "Payload.$": "States.JsonMerge($$, $, false)",
  ...
}

The JsonMerge [intrinsic function][14] merges the Step Functions context object ($$) with the original Lambda’s input payload ($). Fields of the original payload overwrite the Step Functions context object if their keys are the same.

Example:

"Lambda Read From DynamoDB": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke",
  "Parameters": {
    "Payload.$": "States.JsonMerge($$, $, false)",
    "FunctionName": "${lambdaArn}"
  },
  "End": true
}

Alternatively, if you have business logic defined in the payload, you can also use the following format:

"Lambda Read From DynamoDB": {
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke",
  "Parameters": {
    "Payload": {
      ...
      "Execution.$": "$$.Execution",
      "State.$": "$$.State",
      "StateMachine.$": "$$.StateMachine"
    },
    "FunctionName": "${lambdaArn}"
  },
  "End": true
}

Through managed services

Follow these instructions if your Step Function indirectly invokes a Lambda through EventBridge, SQS, or SNS. To trace through another managed AWS service, contact Datadog Support to open a feature request.

Requirements

Runtime	Requirement
Python	Datadog Lambda Library for Python layer v107+

Your State Machine definition must use JSONata as the query language. To enable this, set your definition’s top-level QueryLanguage field to JSONata.

Merging Step Functions with Lambda traces through managed services is only supported for Python runtimes.

EventBridge

If an EventBridge rule has a Lambda function as a target, edit your EventBridge PutEvents Task to set _datadog in the Detail field as follows:

"EventBridge PutEvents": {
  "Type": "Task",
  "Resource": "arn:aws:states:::events:putEvents",
  "Arguments": {
    "Entries": [
      {
        "Detail": {
          "Message": "Hello from Step Functions!",
          "TaskToken": "{% $states.context.Task.Token %}",
          "_datadog": "{% ($execInput := $states.context.Execution.Input; $hasDatadogTraceId := $exists($execInput._datadog.`x-datadog-trace-id`); $hasDatadogRootExecutionId := $exists($execInput._datadog.RootExecutionId); $ddTraceContext := $hasDatadogTraceId ? {'x-datadog-trace-id': $execInput._datadog.`x-datadog-trace-id`, 'x-datadog-tags': $execInput._datadog.`x-datadog-tags`} : {'RootExecutionId': $hasDatadogRootExecutionId ?  $execInput._datadog.RootExecutionId : $states.context.Execution.Id}; $sfnContext := $merge([$states.context, {'Execution': $sift($states.context.Execution, function($v, $k) { $k != 'Input' })}]); $merge([$sfnContext, $ddTraceContext, {'serverless-version': 'v1'}])) %}"
        },
        "DetailType": "MyDetailType",
        "EventBusName": "MyEventBusName",
        "Source": "MySource"
      }
    ]
  }
}

SQS

If an SQS queue has a Lambda trigger, edit your SQS SendMessage Task to set _datadog in the MessageAttributes field as follows:

"SQS SendMessage": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sqs:sendMessage",
  "Arguments": {
    "MessageBody": "{% $states.input %}",
    "QueueUrl": "https://sqs.<REGION>.amazonaws.com/<ACCOUNT_ID>/<QUEUE_NAME>",
    "MessageAttributes": {
      "_datadog": {
        "DataType": "String",
        "StringValue": "{% ($execInput := $states.context.Execution.Input; $hasDatadogTraceId := $exists($execInput._datadog.`x-datadog-trace-id`); $hasDatadogRootExecutionId := $exists($execInput._datadog.RootExecutionId); $ddTraceContext := $hasDatadogTraceId ? {'x-datadog-trace-id': $execInput._datadog.`x-datadog-trace-id`, 'x-datadog-tags': $execInput._datadog.`x-datadog-tags`} : {'RootExecutionId': $hasDatadogRootExecutionId ?  $execInput._datadog.RootExecutionId : $states.context.Execution.Id}; $sfnContext := $merge([$states.context, {'Execution': $sift($states.context.Execution, function($v, $k) { $k != 'Input' })}]); $merge([$sfnContext, $ddTraceContext, {'serverless-version': 'v1'}])) %}"
      }
    }
  }
}

SNS

If there is a Lambda subscription on the topic, edit the SNS Publish Task to set _datadog in the MessageAttributes field as follows:

"SNS Publish": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sns:publish",
  "Arguments": {
    "TopicArn": "arn:aws:sns:<REGION>:<ACCOUNT_ID>:<TOPIC_NAME>",
    "Message": "{% $states.input %}",
    "MessageAttributes": {
      "_datadog": {
        "DataType": "String",
        "StringValue": "{% ($execInput := $states.context.Execution.Input; $hasDatadogTraceId := $exists($execInput._datadog.`x-datadog-trace-id`); $hasDatadogRootExecutionId := $exists($execInput._datadog.RootExecutionId); $ddTraceContext := $hasDatadogTraceId ? {'x-datadog-trace-id': $execInput._datadog.`x-datadog-trace-id`, 'x-datadog-tags': $execInput._datadog.`x-datadog-tags`} : {'RootExecutionId': $hasDatadogRootExecutionId ?  $execInput._datadog.RootExecutionId : $states.context.Execution.Id}; $sfnContext := $merge([$states.context, {'Execution': $sift($states.context.Execution, function($v, $k) { $k != 'Input' })}]); $merge([$sfnContext, $ddTraceContext, {'serverless-version': 'v1'}])) %}"
      }
    }
  }
}

Merge Step Functions traces with nested Step Functions traces

With JSONata

Edit the Step Functions Task to set _datadog in the Input field as follows:

"Step Functions StartExecution": {
  "Type": "Task",
  "Resource": "arn:aws:states:::states:startExecution.sync:2",
  "Arguments": {
    "StateMachineArn": "arn:aws:states:<REGION>:<ACCOUNT_ID>:stateMachine:<STATE_MACHINE_NAME>",
    "Input": {
      "_datadog": "{% ($execInput := $states.context.Execution.Input; $hasDatadogTraceId := $exists($execInput._datadog.`x-datadog-trace-id`); $hasDatadogRootExecutionId := $exists($execInput._datadog.RootExecutionId); $ddTraceContext := $hasDatadogTraceId ? {'x-datadog-trace-id': $execInput._datadog.`x-datadog-trace-id`, 'x-datadog-tags': $execInput._datadog.`x-datadog-tags`} : {'RootExecutionId': $hasDatadogRootExecutionId ?  $execInput._datadog.RootExecutionId : $states.context.Execution.Id}; $sfnContext := $merge([$states.context, {'Execution': $sift($states.context.Execution, function($v, $k) { $k != 'Input' })}]); $merge([$sfnContext, $ddTraceContext, {'serverless-version': 'v1'}])) %}"
    }
  }
}

With JSONPath

Configure your task according to the following example:

"Step Functions StartExecution": {
  "Type": "Task",
  "Resource": "arn:aws:states:::states:startExecution",
  "Parameters": {
    "StateMachineArn": "${stateMachineArn}",
    "Input": {
      "StatePayload": "Hello from Step Functions!",
      "CONTEXT": {
        "Execution.$": "$$.Execution",
        "State.$": "$$.State",
        "StateMachine.$": "$$.StateMachine"
      }
    }
  },
  "End": true
}

Merge Lambda traces with downstream Step Functions traces

Follow these instructions if a Lambda function directly invokes a Step Function using StartExecution or StartSyncExecution.

Requirements

Runtime	Requirement
Node.js	Datadog Lambda Library for Node.js layer v112+ or `dd-trace-js` v3.58.0, v4.37.0, or v5.13.0+
Python	Datadog Lambda Library for Python layer v99+ or `dd-trace-py` v2.13.0+
Java	`dd-trace-java` v1.47.0+
.NET	`dd-trace-dotnet` v3.11.0+

Setup

If the layer or tracer version requirements are fulfilled, no further setup is required.

To ensure proper trace merging, provide input to the Step Functions Start Execution command, even if the input is an empty JSON object.