feat(aws-sdk)!: SQS receive: use span links instead of processing spans (#2345)

Co-authored-by: Trent Mick <[email protected]>

Author: seemk

package-lock.json

@@ -53,7 +53,6 @@ aws-sdk instrumentation has few options available to choose from. You can set th
 | `preRequestHook`                          | `AwsSdkRequestCustomAttributeFunction`    | Hook called before request send, which allow to add custom attributes to span.                                                                                                  |
 | `responseHook`                            | `AwsSdkResponseCustomAttributeFunction`   | Hook for adding custom attributes when response is received from aws.                                                                                                           |
 | `exceptionHook`                           | `AwsSdkExceptionCustomAttributeFunction`  | Hook for adding custom attributes when exception is received from aws.                                                                                                          |
-| `sqsProcessHook`                          | `AwsSdkSqsProcessCustomAttributeFunction` | Hook called after starting sqs `process` span (for each sqs received message), which allow to add custom attributes to it.                                                      |
 | `suppressInternalInstrumentation`         | `boolean`                                 | Most aws operation use http requests under the hood. Set this to `true` to hide all underlying http spans.                                                                      |
 | `sqsExtractContextPropagationFromPayload` | `boolean`                                 | Will parse and extract context propagation headers from SQS Payload, false by default. [When should it be used?](./doc/sns.md#integration-with-sqs)                             |
 | `dynamoDBStatementSerializer`             | `AwsSdkDynamoDBStatementSerializer`       | AWS SDK instrumentation will serialize DynamoDB commands to the `db.statement` attribute using the specified function. Defaults to using a serializer that returns `undefined`. |

packages/instrumentation-aws-sdk/README.md

@@ -1,68 +1,19 @@
 # SQS
 
-SQS is amazon's managed message queue. Thus, it should follow the [OpenTelemetry specification for Messaging systems](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md).
+SQS is Amazon's managed message queue. Thus, it should follow the [OpenTelemetry specification for Messaging systems](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/).
 
 ## Specific trace semantic
 
 The following methods are automatically enhanced:
 
 ### sendMessage / sendMessageBatch
 
-- [Messaging Attributes](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#messaging-attributes) are added by this instrumentation according to the spec.
+- [Messaging Attributes](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes) are added by this instrumentation according to the spec.
 - OpenTelemetry trace context is injected as SQS MessageAttributes, so the service receiving the message can link cascading spans to the trace which created the message.
 
 ### receiveMessage
 
-- [Messaging Attributes](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#messaging-attributes) are added by this instrumentation according to the spec.
-- Additional "processing spans" are created for each message received by the application.  
-  If an application invoked `receiveMessage`, and received a 10 messages batch, a single `messaging.operation` = `receive` span will be created for the `receiveMessage` operation, and 10 `messaging.operation` = `process` spans will be created, one for each message.  
-  Those processing spans are created by the library. This behavior is partially implemented, [See discussion below](#processing-spans).
-- Sets the inter process context correctly, so that additional spans created through the process will be linked to parent spans correctly.  
-  This behavior is partially implemented, [See discussion below](#processing-spans).
+- [Messaging Attributes](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes) are added by this instrumentation according to the spec.
+- Sets the inter process context correctly, so that additional spans created through the process will be linked to parent spans correctly.
+  When multiple messages are received, the instrumentation will attach spank links to the receiving span containing the trace context and message ID of each message.
 - Extract trace context from SQS MessageAttributes, and set span's `parent` and `links` correctly according to the spec.
-
-#### Processing Spans
-
-See GH issue [here](https://github.com/open-telemetry/opentelemetry-js-contrib/issues/707)
-
-According to OpenTelemetry specification (and to reasonable expectation for trace structure), user of this library would expect to see one span for the operation of receiving messages batch from SQS, and then, **for each message**, a span with it's own sub-tree for the processing of this specific message.
-
-For example, if a `receiveMessages` returned 2 messages:
-
-- `msg1` resulting in storing something to a DB.
-- `msg2` resulting in calling an external HTTP endpoint.
-
-This will result in a creating a DB span that would be the child of `msg1` process span, and an HTTP span that would be the child of `msg2` process span (in opposed to mixing all those operations under the single `receive` span, or start a new trace for each of them).
-
-Unfortunately, this is not so easy to implement in JS:
-
-1. The SDK is calling a single callback for the messages batch, and it's not straightforward to understand when each individual message processing starts and ends (and set the context correctly for cascading spans).
-2. If async/await is used, context can be lost when returning data from async functions, for example:
-
-```js
-async function asyncRecv() {
-  const data = await sqs.receiveMessage(recvParams).promise();
-  // context of receiveMessage is set here
-  return data;
-}
-
-async function poll() {
-  const result = await asyncRecv();
-  // context is lost when asyncRecv returns. following spans are created with root context.
-  await Promise.all(
-    result.Messages.map((message) => this.processMessage(message))
-  );
-}
-```
-
-Current implementation partially solves this issue by patching the `map` \ `forEach` \ `Filter` functions on the `Messages` array of `receiveMessage` result. This handles issues like the one above, but will not handle situations where the processing is done in other patterns (multiple map\forEach calls, index access to the array, other array operations, etc). This is currently an open issue in the instrumentation.
-
-User can add custom attributes to the `process` span, by setting a function to `sqsProcessHook` in instrumentation config. For example:
-
-```js
-awsInstrumentationConfig = {
-  sqsProcessHook: (span, message) => {
-    span.setAttribute("sqs.receipt_handle", message.params?.ReceiptHandle);
-  },
-};
-```

packages/instrumentation-aws-sdk/doc/sqs.md

@@ -50,7 +50,6 @@
   "dependencies": {
     "@opentelemetry/core": "^2.0.0",
     "@opentelemetry/instrumentation": "^0.203.0",
-    "@opentelemetry/propagation-utils": "^0.31.3",
     "@opentelemetry/semantic-conventions": "^1.34.0"
   },
   "devDependencies": {

packages/instrumentation-aws-sdk/package.json

@@ -21,8 +21,6 @@ export type {
   AwsSdkRequestHookInformation,
   AwsSdkResponseCustomAttributeFunction,
   AwsSdkResponseHookInformation,
-  AwsSdkSqsProcessCustomAttributeFunction,
-  AwsSdkSqsProcessHookInformation,
   CommandInput,
   NormalizedRequest,
   NormalizedResponse,

packages/instrumentation-aws-sdk/src/index.ts

@@ -20,6 +20,25 @@
  * @see https://github.com/open-telemetry/opentelemetry-js/tree/main/semantic-conventions#unstable-semconv
  */
 
+/**
+ * The ARN of the Secret stored in the Secrets Mangger
+ *
+ * @example arn:aws:secretsmanager:us-east-1:123456789012:secret:SecretName-6RandomCharacters
+ *
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_AWS_SECRETSMANAGER_SECRET_ARN =
+  'aws.secretsmanager.secret.arn' as const;
+
+/**
+ * The ARN of the AWS SNS Topic. An Amazon SNS [topic](https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html) is a logical access point that acts as a communication channel.
+ *
+ * @example arn:aws:sns:us-east-1:123456789012:mystack-mytopic-NZJ5JSMVGFIE
+ *
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_AWS_SNS_TOPIC_ARN = 'aws.sns.topic.arn' as const;
+
 /**
  * The name of the operation being performed.
  *
@@ -139,42 +158,84 @@ export const ATTR_GEN_AI_USAGE_INPUT_TOKENS =
 export const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS =
   'gen_ai.usage.output_tokens' as const;
 
+/**
+ * The number of messages sent, received, or processed in the scope of the batching operation.
+ *
+ * @example 0
+ * @example 1
+ * @example 2
+ *
+ * @note Instrumentations **SHOULD NOT** set `messaging.batch.message_count` on spans that operate with a single message. When a messaging client library supports both batch and single-message API for the same operation, instrumentations **SHOULD** use `messaging.batch.message_count` for batching APIs and **SHOULD NOT** use it for single-message APIs.
+ *
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT =
+  'messaging.batch.message_count' as const;
+
+/**
+ * The message destination name
+ *
+ * @example MyQueue
+ * @example MyTopic
+ *
+ * @note Destination name **SHOULD** uniquely identify a specific queue, topic or other entity within the broker. If
+ * the broker doesn't have such notion, the destination name **SHOULD** uniquely identify the broker.
+ *
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_MESSAGING_DESTINATION_NAME =
+  'messaging.destination.name' as const;
+
+/**
+ * A value used by the messaging system as an identifier for the message, represented as a string.
+ *
+ * @example "452a7c7c7c7048c2f887f61572b18fc2"
+ *
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id' as const;
+
+/**
+ * A string identifying the type of the messaging operation.
+ *
+ * @note If a custom value is used, it **MUST** be of low cardinality.
+ * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
+ */
+export const ATTR_MESSAGING_OPERATION_TYPE =
+  'messaging.operation.type' as const;
+
 /**
  * Enum value "chat" for attribute {@link ATTR_GEN_AI_OPERATION_NAME}.
+ *
+ * Chat completion operation such as [OpenAI Chat API](https://platform.openai.com/docs/api-reference/chat)
+ *
+ * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
  */
 export const GEN_AI_OPERATION_NAME_VALUE_CHAT = 'chat' as const;
 
 /**
  * Enum value "aws.bedrock" for attribute {@link ATTR_GEN_AI_SYSTEM}.
+ *
+ * AWS Bedrock
+ *
+ * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
  */
 export const GEN_AI_SYSTEM_VALUE_AWS_BEDROCK = 'aws.bedrock' as const;
 
 /**
  * Enum value "input" for attribute {@link ATTR_GEN_AI_TOKEN_TYPE}.
+ *
+ * Input tokens (prompt, input, etc.)
+ *
+ * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
  */
 export const GEN_AI_TOKEN_TYPE_VALUE_INPUT = 'input' as const;
 
 /**
  * Enum value "output" for attribute {@link ATTR_GEN_AI_TOKEN_TYPE}.
+ *
+ * Output tokens (completion, response, etc.)
+ *
+ * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
  */
 export const GEN_AI_TOKEN_TYPE_VALUE_OUTPUT = 'output' as const;
-
-/**
- * Originally from '@opentelemetry/semantic-conventions/incubating'
- * https://github.com/open-telemetry/semantic-conventions/blob/main/docs/registry/attributes/aws.md#amazon-secrets-manager-attributes
- * The ARN of the Secret stored in the Secrets Mangger
- * @example arn:aws:secretsmanager:us-east-1:123456789012:secret:SecretName-6RandomCharacters
- * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
- */
-export const ATTR_AWS_SECRETSMANAGER_SECRET_ARN =
-  'aws.secretsmanager.secret.arn' as const;
-
-/**
- * Originally from '@opentelemetry/semantic-conventions/incubating'
- * https://github.com/open-telemetry/semantic-conventions/blob/main/docs/registry/attributes/aws.md#amazon-sns-attributes
- * The ARN of the AWS SNS Topic. An Amazon SNS [topic](https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html)
- *  is a logical access point that acts as a communication channel.
- * @example arn:aws:sns:us-east-1:123456789012:mystack-mytopic-NZJ5JSMVGFIE
- * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`.
- */
-export const ATTR_AWS_SNS_TOPIC_ARN = 'aws.sns.topic.arn' as const;

packages/instrumentation-aws-sdk/src/semconv.ts

@@ -19,11 +19,9 @@ import {
   Span,
   propagation,
   trace,
-  context,
   ROOT_CONTEXT,
   Attributes,
 } from '@opentelemetry/api';
-import { pubsubPropagation } from '@opentelemetry/propagation-utils';
 import { RequestMetadata, ServiceExtension } from './ServiceExtension';
 import type { SQS } from '../aws-sdk.types';
 import {
@@ -32,16 +30,15 @@ import {
   NormalizedResponse,
 } from '../types';
 import {
-  MESSAGINGDESTINATIONKINDVALUES_QUEUE,
-  MESSAGINGOPERATIONVALUES_PROCESS,
-  MESSAGINGOPERATIONVALUES_RECEIVE,
-  SEMATTRS_MESSAGING_DESTINATION,
-  SEMATTRS_MESSAGING_DESTINATION_KIND,
-  SEMATTRS_MESSAGING_MESSAGE_ID,
-  SEMATTRS_MESSAGING_OPERATION,
+  ATTR_URL_FULL,
   SEMATTRS_MESSAGING_SYSTEM,
-  SEMATTRS_MESSAGING_URL,
 } from '@opentelemetry/semantic-conventions';
+import {
+  ATTR_MESSAGING_BATCH_MESSAGE_COUNT,
+  ATTR_MESSAGING_DESTINATION_NAME,
+  ATTR_MESSAGING_MESSAGE_ID,
+  ATTR_MESSAGING_OPERATION_TYPE,
+} from '../semconv';
 import {
   contextGetter,
   extractPropagationContext,
@@ -60,11 +57,9 @@ export class SqsServiceExtension implements ServiceExtension {
     let spanName: string | undefined;
 
     const spanAttributes: Attributes = {
-      [SEMATTRS_MESSAGING_SYSTEM]: 'aws.sqs',
-      [SEMATTRS_MESSAGING_DESTINATION_KIND]:
-        MESSAGINGDESTINATIONKINDVALUES_QUEUE,
-      [SEMATTRS_MESSAGING_DESTINATION]: queueName,
-      [SEMATTRS_MESSAGING_URL]: queueUrl,
+      [SEMATTRS_MESSAGING_SYSTEM]: 'aws_sqs',
+      [ATTR_MESSAGING_DESTINATION_NAME]: queueName,
+      [ATTR_URL_FULL]: queueUrl,
     };
 
     let isIncoming = false;
@@ -75,8 +70,7 @@ export class SqsServiceExtension implements ServiceExtension {
           isIncoming = true;
           spanKind = SpanKind.CONSUMER;
           spanName = `${queueName} receive`;
-          spanAttributes[SEMATTRS_MESSAGING_OPERATION] =
-            MESSAGINGOPERATIONVALUES_RECEIVE;
+          spanAttributes[ATTR_MESSAGING_OPERATION_TYPE] = 'receive';
 
           request.commandInput.MessageAttributeNames =
             addPropagationFieldsToAttributeNames(
@@ -136,61 +130,43 @@ export class SqsServiceExtension implements ServiceExtension {
   responseHook = (
     response: NormalizedResponse,
     span: Span,
-    tracer: Tracer,
+    _tracer: Tracer,
     config: AwsSdkInstrumentationConfig
   ) => {
     switch (response.request.commandName) {
       case 'SendMessage':
-        span.setAttribute(
-          SEMATTRS_MESSAGING_MESSAGE_ID,
-          response?.data?.MessageId
-        );
+        span.setAttribute(ATTR_MESSAGING_MESSAGE_ID, response?.data?.MessageId);
         break;
 
       case 'SendMessageBatch':
         // TODO: How should this be handled?
         break;
 
       case 'ReceiveMessage': {
-        const messages: SQS.Message[] = response?.data?.Messages;
-        if (messages) {
-          const queueUrl = this.extractQueueUrl(response.request.commandInput);
-          const queueName = this.extractQueueNameFromUrl(queueUrl);
-
-          pubsubPropagation.patchMessagesArrayToStartProcessSpans<SQS.Message>({
-            messages,
-            parentContext: trace.setSpan(context.active(), span),
-            tracer,
-            messageToSpanDetails: (message: SQS.Message) => ({
-              name: queueName ?? 'unknown',
-              parentContext: propagation.extract(
-                ROOT_CONTEXT,
-                extractPropagationContext(
-                  message,
-                  config.sqsExtractContextPropagationFromPayload
-                ),
-                contextGetter
-              ),
+        const messages: SQS.Message[] = response?.data?.Messages || [];
+
+        span.setAttribute(ATTR_MESSAGING_BATCH_MESSAGE_COUNT, messages.length);
+
+        for (const message of messages) {
+          const propagatedContext = propagation.extract(
+            ROOT_CONTEXT,
+            extractPropagationContext(
+              message,
+              config.sqsExtractContextPropagationFromPayload
+            ),
+            contextGetter
+          );
+
+          const spanContext = trace.getSpanContext(propagatedContext);
+
+          if (spanContext) {
+            span.addLink({
+              context: spanContext,
               attributes: {
-                [SEMATTRS_MESSAGING_SYSTEM]: 'aws.sqs',
-                [SEMATTRS_MESSAGING_DESTINATION]: queueName,
-                [SEMATTRS_MESSAGING_DESTINATION_KIND]:
-                  MESSAGINGDESTINATIONKINDVALUES_QUEUE,
-                [SEMATTRS_MESSAGING_MESSAGE_ID]: message.MessageId,
-                [SEMATTRS_MESSAGING_URL]: queueUrl,
-                [SEMATTRS_MESSAGING_OPERATION]:
-                  MESSAGINGOPERATIONVALUES_PROCESS,
+                [ATTR_MESSAGING_MESSAGE_ID]: message.MessageId,
               },
-            }),
-            processHook: (span: Span, message: SQS.Message) =>
-              config.sqsProcessHook?.(span, { message }),
-          });
-
-          pubsubPropagation.patchArrayForProcessSpans(
-            messages,
-            tracer,
-            context.active()
-          );
+            });
+          }
         }
         break;
       }