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

plugins/node/opentelemetry-instrumentation-aws-sdk/README.md

@@ -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.                                                                                                           |
-| `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`. |

plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.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';
 import {
@@ -33,7 +31,6 @@ import {
 } from '../types';
 import {
   MESSAGINGDESTINATIONKINDVALUES_QUEUE,
-  MESSAGINGOPERATIONVALUES_PROCESS,
   MESSAGINGOPERATIONVALUES_RECEIVE,
   SEMATTRS_MESSAGING_DESTINATION,
   SEMATTRS_MESSAGING_DESTINATION_KIND,
@@ -134,7 +131,7 @@ export class SqsServiceExtension implements ServiceExtension {
   responseHook = (
     response: NormalizedResponse,
     span: Span,
-    tracer: Tracer,
+    _tracer: Tracer,
     config: AwsSdkInstrumentationConfig
   ) => {
     switch (response.request.commandName) {
@@ -150,45 +147,30 @@ export class SqsServiceExtension implements ServiceExtension {
         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('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,
               },
-            }),
-            processHook: (span: Span, message: SQS.Message) =>
-              config.sqsProcessHook?.(span, { message }),
-          });
-
-          pubsubPropagation.patchArrayForProcessSpans(
-            messages,
-            tracer,
-            context.active()
-          );
+            });
+          }
         }
         break;
       }

plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts

@@ -15,7 +15,6 @@
  */
 import { Span } from '@opentelemetry/api';
 import { InstrumentationConfig } from '@opentelemetry/instrumentation';
-import { SQS } from './aws-sdk.types';
 
 export type CommandInput = Record<string, any>;
 
@@ -57,13 +56,6 @@ export interface AwsSdkResponseCustomAttributeFunction {
   (span: Span, responseInfo: AwsSdkResponseHookInformation): void;
 }
 
-export interface AwsSdkSqsProcessHookInformation {
-  message: SQS.Message;
-}
-export interface AwsSdkSqsProcessCustomAttributeFunction {
-  (span: Span, sqsProcessInfo: AwsSdkSqsProcessHookInformation): void;
-}
-
 export type AwsSdkDynamoDBStatementSerializer = (
   operation: string,
   commandInput: CommandInput
@@ -76,9 +68,6 @@ export interface AwsSdkInstrumentationConfig extends InstrumentationConfig {
   /** hook for adding custom attributes when response is received from aws */
   responseHook?: AwsSdkResponseCustomAttributeFunction;
 
-  /** hook for adding custom attribute when an sqs process span is started */
-  sqsProcessHook?: AwsSdkSqsProcessCustomAttributeFunction;
-
   /** custom serializer function for the db.statement attribute in DynamoDB spans */
   dynamoDBStatementSerializer?: AwsSdkDynamoDBStatementSerializer;