Enable OpenTelemetry bridge

bashcopydownload
pip install -U "opentelemetry-api>=1.12" "opentelemetry-instrumentation-boto3sqs>=0.34b0"

Note that at this point, opentelemetry-instrumentation-boto3sqs is a separate package from opentelemetry-instrumentation-botocore. The latter instruments all AWS SDK calls, but lacks enhanced support for SQS.

bashcopydownload
pip install -U "opentelemetry-instrumentation-botocore>=0.36b0"

Example sending SQS messages with Python:

pythoncopydownload
from opentelemetry.instrumentation.boto3sqs import Boto3SQSInstrumentor

Boto3SQSInstrumentor().instrument()

import json
import boto3
from datetime import datetime

QUEUE_URL = "<Your SQS Queue URL>"

sqs = boto3.client("sqs")

def lambda_handler(event, context):
    sent = []
    for i in range(5):
        res = sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=f"hello #{i} at {datetime.now()}")
        sent.append(res["MessageId"])

    return {
        "statusCode": 200,
        "body": json.dumps({"produced_messages": sent})
    }

Example sending SNS messages with Python:

pythoncopydownload
from opentelemetry.instrumentation.botocore import BotocoreInstrumentor

BotocoreInstrumentor().instrument()

import json
import boto3
from datetime import datetime

TOPIC_ARN = "<Your SNS topic ARN>"

sns = boto3.client("sns")

def lambda_handler(event, context):
    res = sns.publish(TopicArn=TOPIC_ARN, Message=f"hello at {datetime.now()}")

    return {
        "statusCode": 200,
        "body": json.dumps({"produced_message": res["MessageId"]})
    }

See below for examples of the following:

• Receive messages with AWS Lambda SQS trigger (configured outside the code)
• Call the SQS receive API in code (using boto3)

Notes:

• If you invoke the sender and have deployed the SQS-triggered example, it will be invoked automatically by SQS.
• If you have deployed the example that uses the receive API in code, you need to invoke it manually and it will attempt to read all messages from the queue.
• If your SQS queue is subscribed to an SNS topic, the examples below might need to be adapted. For details, see Tracing SQS messages forwarded from an SNS topic.

Example receiving messages with AWS Lambda SQS trigger

If you use an AWS Lambda with an SQS trigger monitored with the Dynatrace AWS Lambda extension, receiving messages works out of the box. However, the tracer can select only a single parent, and if your Lambda function receives a batch of multiple messages, you need to manually create spans to process every single message if you want to track them separately and have them linked to the sender.

pythoncopydownload
from pprint import pformat

import boto3
import json
from opentelemetry import trace, propagate
from opentelemetry.semconv.trace import SpanAttributes, MessagingOperationValues

tracer = trace.get_tracer("lambda-sqs-triggered")

def lambda_handler(event, context):
    recvcount = 0
    print("Trigger", pformat(event))
    messages = event.get("Records") or ()
    # Lambda SQS event uses lowerCamelCase in its attribute names
    for msg in messages:
        recvcount += 1
        print("Processing", msg["messageId"])
        parent = _extract_parent(msg, from_sns_payload=False)
        with tracer.start_as_current_span("manual-trigger-process", context=parent, kind=trace.SpanKind.CONSUMER, attributes={
            SpanAttributes.MESSAGING_MESSAGE_ID : msg["messageId"],
            SpanAttributes.MESSAGING_URL : msg["eventSourceARN"],
            SpanAttributes.MESSAGING_SYSTEM : msg["eventSource"],
            SpanAttributes.MESSAGING_OPERATION : MessagingOperationValues.PROCESS.value,
        }):
            # ... Here your actual processing would go...
            pass
    print("Processed", recvcount, "messages")

def _extract_parent(msg, from_sns_payload=False):
    if from_sns_payload:
        try:
            body = json.loads(msg.get("body", "{}"))
        except json.JSONDecodeError:
            body = {}
        carrier = {key: value["Value"] for key, value in body.get("MessageAttributes", {}).items() if "Value" in value}
    else:
        carrier = {key: value["stringValue"] for key, value in msg.get("messageAttributes", {}).items() if "stringValue" in value}
    return propagate.extract(carrier)

For the manual-trigger-process span to work correctly, you need to configure the Dynatrace AWS Lambda extension to allow parent spans to be set manually. This can be done by setting the DT_OPEN_TELEMETRY_ALLOW_EXPLICIT_PARENT environment variable to true:

bashcopydownload
DT_OPEN_TELEMETRY_ALLOW_EXPLICIT_PARENT=true

Alternatively, you can configure it in dtconfig.json by setting the following field to true:

jsoncopydownload
{
  "OpenTelemetry": {
    "AllowExplicitParent": "true"
  }
}

For details on how to configure the Dynatrace AWS Lambda extension, see Trace Python, Node.js, and Java Lambda functions.

The resulting path looks as follows:

The invoked Lambda function is a child of one of the messages by which it's triggered. Since there can only be one parent, the other manual-trigger–process spans aren't linked directly to the Lambda invocation in which they are handled. Often, there's more than one Lambda invocation node for a batch of messages. In those cases, AWS distributed the batch over multiple Lambda invocations. This can happen even if the messages are delivered within your configured batch window time and number less than your configured batch size.

Example calling the SQS receive APIs manually

pythoncopydownload
from opentelemetry.instrumentation.boto3sqs import Boto3SQSInstrumentor

Boto3SQSInstrumentor().instrument()

from pprint import pformat

import boto3
import json
from opentelemetry import trace, propagate
from opentelemetry.semconv.trace import SpanAttributes, MessagingOperationValues

QUEUE_URL = '<Your SQS Queue URL>'
sqs = boto3.client("sqs")
tracer = trace.get_tracer("lambda-receive-function")

def lambda_handler(event, context):
    recvcount = 0
    while True:
        msg_receive_result = sqs.receive_message(
          MaxNumberOfMessages=10,
          QueueUrl=QUEUE_URL,
          WaitTimeSeconds=1, # WaitTime of zero would use sampled receive, may return empty even if there is a message

          # This argument is only required if you do not use the boto3sqs instrumentation:
          #MessageAttributeNames=list(propagate.get_global_textmap().fields)
        )
        print("Received", pformat(msg_receive_result))
        if not msg_receive_result.get('Messages'):
            break
        messages = msg_receive_result.get("Messages")

        # receive result uses PascalCase in its attribute names
        for msg in messages:
            recvcount += 1
            print("Processing", msg["MessageId"])
            parent = _extract_parent(msg, from_sns_payload=False)
            with tracer.start_as_current_span("manual-receive-process", context=parent, kind=trace.SpanKind.CONSUMER, attributes={
                SpanAttributes.MESSAGING_MESSAGE_ID: msg["MessageId"],
                SpanAttributes.MESSAGING_URL: QUEUE_URL,
                SpanAttributes.MESSAGING_SYSTEM: "aws.sqs",
                SpanAttributes.MESSAGING_OPERATION: MessagingOperationValues.PROCESS.value,
            }):
                # ... Here your actual processing would go...

                print("Delete result", sqs.delete_message(
                    QueueUrl=QUEUE_URL,
                    ReceiptHandle=msg['ReceiptHandle'],
                ))
    print("Processed", recvcount, "messages")

def _extract_parent(msg, from_sns_payload=False):
    if from_sns_payload:
        try:
            body = json.loads(msg.get("Body", "{}"))
        except json.JSONDecodeError:
            body = {}
        carrier = {key: value["Value"] for key, value in body.get("MessageAttributes", {}).items() if "Value" in value}
    else:
      carrier = {key: value["StringValue"] for key, value in msg.get("MessageAttributes", {}).items() if "StringValue" in value}

    return propagate.extract(carrier)

This code can also be used outside a Lambda function and monitored with pure OpenTelemetry instead of the Dynatrace AWS Lambda extension.

This example uses the boto3sqs instrumentation. If you don't want to use it, you need to uncomment the MessageAttributeNames argument in the receive_message function, otherwise, SQS will omit data required for linking the message to its sender from the retrieved data.

Creating the manual-receive-process span manually is necessary because the boto3sqs instrumentation doesn't set the sender as a parent for the processing span, but uses OpenTelemetry links, which are currently not supported by Dynatrace. For the linking of the manual-receive-process span to work correctly, you need to configure the Dynatrace AWS Lambda extension to allow setting parent spans manually. See the previous example for guidance.

Invoking first the code that sends SQS messages, then the manual receive code, deployed as Lambda functions, results in two traces:

• The first trace shows the flow of the message from the sender to the processor:

  

  There are additional Requests to public networks nodes because the boto3 package uses HTTP requests to send SQS messages, which are captured by Dynatrace HTTP instrumentation.

  You'll notice that the invocation and receive node of the second Lambda invocation are missing from this trace, even though the manual-receive-process nodes are there. This is because the Lambda function was triggered independently of the message flow, and just happened to receive the message as part of its handler code.

• The second trace in Dynatrace shows the Lambda invocation until it's cut in two by setting the explicit parent:

  

Tracing SQS messages forwarded from an SNS topic

For SNS messages that are forwarded to SQS, the message format depends on the raw message delivery configuration on the SNS subscription.

• If raw message delivery is disabled, the SNS message and its MessageAttributes are delivered as a serialized JSON string in the body of the received SQS message. To correctly link the receive span, the parent needs to be extracted from the MessageAttributes of the serialized SNS message.
  • For both examples (calling the SQS receive APIs manually and receiving messages with AWS Lambda SQS trigger), you need to set the value for the from_sns_payload parameter to True when calling the _extract_parent method.
• If raw message delivery is enabled on the SNS subscription, the SNS message attributes are converted to SQS message attributes and the parent can be directly extracted from the MessageAttributes of the SQS message.
  • No additional settings are needed for the examples above.

AWS Lambda functions that are triggered by SNS are supported out of the box when monitored with the Dynatrace AWS Lambda extension.

SNS topics can be configured via a subscription to forward messages to an SQS queue. Messages in the SQS queue can then be consumed by a Lambda function. Tracing the received messages in the SQS-triggered AWS Lambda works out of the box when AWS Lambda is monitored with the Dynatrace AWS Lambda extension. However, the tracer can only select a single parent, and if your Lambda function receives batches of multiple messages, special handling is required to track each message separately. For details, see Receive an SQS message.

javascriptcopydownload
const AWS = require('aws-sdk');

exports.handler = function (event, context, callback) {
  const sqs = new AWS.SQS();
  const params = {
    DelaySeconds: 10,
    MessageBody: "[your payload]",
    QueueUrl: "[your SQS-queue URL]"
  };

  sqs.sendMessage(params, function (err, data) {
    if (err) {
      context.succeed({
        statusCode: 500,
        body: err,
      });
    } else {
      console.log("SQS-Success", data.MessageId);
      context.succeed({
        statusCode: 200,
        body: "SQS-Success",
      });
    }
  });
}

The resulting distributed trace is similar to the Node.js application example:

javascriptcopydownload
const AWS = require('aws-sdk');

exports.handler = function (event, context, callback) {
  const sns = new AWS.SNS();
  const params = {
    Message: "[your payload]",
    TopicArn: "[your SNS-topic ARN]"
  };

  sns.publish(params, function (err, data) {
    if (err) {
      context.succeed({
        statusCode: 500,
        body: err,
      });
    } else {
      console.log("SNS-Success", data.MessageId);
      context.succeed({
        statusCode: 200,
        body: "SNS-Success",
      });
    }
  });
}

In both SQS send examples, the subscribed AWS Lambda function sqs-minimal-sample-receiver is triggered by AWS SQS. The Dynatrace AWS Lambda extension extracts the parent from the SQS message and creates the Lambda span. For a more complex scenario, where a batch of multiple messages is received, only the last message will be considered and used for parent propagation.

One of the ways to propagate parents from the batch of multiple incoming messages is to manually create spans with the parent from each message. To do that, you first need to configure the Dynatrace AWS Lambda extension to allow setting parent spans manually. This can be done by setting the DT_OPEN_TELEMETRY_ALLOW_EXPLICIT_PARENT environment variable to true:

yamlcopydownload
DT_OPEN_TELEMETRY_ALLOW_EXPLICIT_PARENT: true

Alternatively, you can configure it in dtconfig.json by setting the following field to true.

jsoncopydownload
{
  "OpenTelemetry": {
    "AllowExplicitParent": "true"
  }
}

For more details about how to configure the Dynatrace AWS Lambda extension, see Trace Python, Node.js, and Java Lambda functions.

By setting this option to true, new spans can be started with the parent span extracted from each received SQS message, as demonstrated in the following example:

javascriptcopydownload
const { propagation, ROOT_CONTEXT, trace, SpanKind } = require("@opentelemetry/api");
const { MessagingOperationValues, SemanticAttributes } = require("@opentelemetry/semantic-conventions");

const AWS = require("aws-sdk");
const queueUrl = "[sqs queue url]";
const tracer = trace.getTracer("my-receiver");

exports.handler = async (event) => {
  const sqs = new AWS.SQS();

  await new Promise((resolve, reject) => {
    const receiveParams = {
      MaxNumberOfMessages: 10,
      QueueUrl: queueUrl,
      // MessageAttributeNames not needed if @opentelemetry/instrumentation-aws-sdk is used
      MessageAttributeNames: propagation.fields()
    };
    sqs.receiveMessage(receiveParams, function (err, data) {
      if (err) {
        console.log("ERROR: ", err);
        reject(err);
      } else if (data.Messages?.length) {
        data.Messages.forEach((msg) => {
          console.log("message received:", msg.MessageId)

          // manual span creation
          const ctx = extractParent(msg, /*fromSnsPayload=*/ false);
          const spanAttributes = {
            [SemanticAttributes.MESSAGING_MESSAGE_ID]: msg.MessageId,
            [SemanticAttributes.MESSAGING_URL]: queueUrl,
            [SemanticAttributes.MESSAGING_SYSTEM]: "aws.sqs",
            [SemanticAttributes.MESSAGING_OPERATION]: MessagingOperationValues.PROCESS,
          };
          const span = tracer.startSpan("received message", { kind: SpanKind.CONSUMER, attributes: spanAttributes }, ctx);
          // ... Here your actual processing would go...
          span.end();

          const deleteParams = {
            QueueUrl: queueUrl,
            ReceiptHandle: msg.ReceiptHandle
          };
          sqs.deleteMessage(deleteParams, function (err, data) {
            if (err) {
              console.log("Delete Error", err);
            } else {
              console.log("Message Deleted", data);
            }
          });
        });
      }

      resolve();
    });
  });
};

function extractParent(msg, fromSnsPayload=false) {
    let valueKey = "StringValue"
    if (fromSnsPayload) {
        valueKey = "Value";
        try {
            msg = JSON.parse(msg.Body)
        } catch {
            msg = {}
        }
    }
    const carrier = {};
    Object.keys(msg.MessageAttributes || {}).forEach((attrKey) => {
        carrier[attrKey] = msg.MessageAttributes[attrKey]?.[valueKey];
    });
    return propagation.extract(ROOT_CONTEXT, carrier)
};

In the example above, aws-sdk is used to subscribe and receive messages from an SQS queue. For each incoming message, the parent span is extracted from the message attributes, and a new received message span with the extracted parent is created. If your SQS queue is subscribed to an SNS topic, the example above might need to be adapted. For details, see Tracing SQS messages forwarded from an SNS topic.

The resulting distributed trace with two messages sent to a queue looks as below.

Tracing SQS messages forwarded from an SNS topic

For SNS messages that are forwarded to SQS, the format of the message depends on the raw message delivery configuration on the SNS subscription.

• If raw message delivery is disabled, the SNS message and its MessageAttributes are delivered as a serialized JSON string in the body of the received SQS message. To correctly link the receive span, the parent needs to be extracted from the MessageAttributes of the serialized SNS message.
  • For the receive example, the value for the fromSnsPayload parameter needs to be set to true when calling the extractParent method.
• If raw message delivery is enabled on the SNS subscription, the SNS message attributes are converted to SQS message attributes and the parent can be directly extracted from the MessageAttributes of the SQS message.
  • No additional settings are needed for the receive example.

AWS Lambda functions that are triggered by SNS are supported out of the box when monitored with the OneAgent layer.

SNS topics can be configured via a subscription to forward messages to an SQS queue. Messages in the SQS queue can then be consumed by a Lambda function. Tracing the received messages in the SQS-triggered AWS Lambda works out of the box when AWS Lambda is monitored with the Dynatrace AWS Lambda extension. However, the tracer can only select a single parent, and if your Lambda function receives batches of multiple messages, special handling is required to track each message separately. For details, see Receive an SQS message.