feat: Add the ability to use span links when consuming a message amqp plugin

plugins/node/instrumentation-amqplib/README.md

@@ -40,6 +40,7 @@ registerInstrumentations({
       // publishConfirmHook: (span: Span, publishConfirmedInto: PublishConfirmedInfo) => { },
       // consumeHook: (span: Span, consumeInfo: ConsumeInfo) => { },
       // consumeEndHook: (span: Span, consumeEndInfo: ConsumeEndInfo) => { },
+      // useLinksForConsume: boolean,
     }),
   ],
 })
@@ -56,6 +57,7 @@ amqplib instrumentation has few options available to choose from. You can set th
 | `consumeHook`                  | `AmqplibConsumeCustomAttributeFunction`    | hook for adding custom attributes before consumer message is processed.                                             |
 | `consumeEndHook`                  | `AmqplibConsumeEndCustomAttributeFunction`    | hook for adding custom attributes after consumer message is acked to server.                                             |
 | `consumeTimeoutMs`                  | `number`    | read [Consume Timeout](#ConsumeTimeout) below                                             |
+| `useLinksForConsume`                  | `boolean`    | read [Links for Consume](#LinksforConsume) below                                          |
 
 ### Consume Timeout
 
@@ -69,6 +71,12 @@ If timeout is not big enough, span might be closed with 'InstrumentationTimeout'
 
 Default is 1 minute
 
+### Links for Consume
+
+By default, consume spans continue the trace where a message was produced. However, per the [spec](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#consumer-spans), consume spans should be linked to the message's creation context. Setting to true, this will enable the behavior to follow the spec. 
+
+Default is false
+
 ## Migration From opentelemetry-instrumentation-amqplib
 
 This instrumentation was originally published under the name `"opentelemetry-instrumentation-amqplib"` in [this repo](https://github.com/aspecto-io/opentelemetry-ext-js). Few breaking changes were made during porting to the contrib repo to align with conventions:

plugins/node/instrumentation-amqplib/src/amqplib.ts

@@ -22,6 +22,7 @@
   SpanKind,
   SpanStatusCode,
   ROOT_CONTEXT,
+  Link,
 } from '@opentelemetry/api';
 import {
   hrTime,
@@ -409,9 +410,18 @@
         const headers = msg.properties.headers ?? {};
         const parentContext = propagation.extract(ROOT_CONTEXT, headers);
         const exchange = msg.fields?.exchange;
-        const span = self.tracer.startSpan(
-          `${queue} process`,
-          {
+        let span: Span;
+        if (self._config.useLinksForConsume) {
+          const parentSpanContext = trace.getSpan(parentContext)?.spanContext();
+          let links: Link[] | undefined;
+          if (parentSpanContext) {
+            links = [
+              {
+                context: parentSpanContext,
+              },
+            ];
+          }
+          span = self.tracer.startSpan(`${queue} process`, {
             kind: SpanKind.CONSUMER,
             attributes: {
               ...channel?.connection?.[CONNECTION_ATTRIBUTES],
@@ -427,9 +437,31 @@
               [SemanticAttributes.MESSAGING_CONVERSATION_ID]:
                 msg?.properties.correlationId,
             },
-          },
-          parentContext
-        );
+            links: links,
+          });
+        } else {
+          span = self.tracer.startSpan(
+            `${queue} process`,
+            {
+              kind: SpanKind.CONSUMER,
+              attributes: {
+                ...channel?.connection?.[CONNECTION_ATTRIBUTES],
+                [SemanticAttributes.MESSAGING_DESTINATION]: exchange,
+                [SemanticAttributes.MESSAGING_DESTINATION_KIND]:
+                  MessagingDestinationKindValues.TOPIC,
+                [SemanticAttributes.MESSAGING_RABBITMQ_ROUTING_KEY]:
+                  msg.fields?.routingKey,
+                [SemanticAttributes.MESSAGING_OPERATION]:
+                  MessagingOperationValues.PROCESS,
+                [SemanticAttributes.MESSAGING_MESSAGE_ID]:
+                  msg?.properties.messageId,
+                [SemanticAttributes.MESSAGING_CONVERSATION_ID]:
+                  msg?.properties.correlationId,
+              },
+            },
+            parentContext
+          );
+        }
 
         if (self._config.consumeHook) {
           safeExecuteInTheMiddle(

plugins/node/instrumentation-amqplib/src/types.ts

@@ -96,10 +96,14 @@ export interface AmqplibInstrumentationConfig extends InstrumentationConfig {
    * Default is 1 minute
    */
   consumeTimeoutMs?: number;
+
+  /** Used to use a span link for the consume message instead of continuing a trace */
+  useLinksForConsume?: boolean;
 }
 
 export const DEFAULT_CONFIG: AmqplibInstrumentationConfig = {
   consumeTimeoutMs: 1000 * 60, // 1 minute
+  useLinksForConsume: false,
 };
 
 // The following types are vendored from `@types/amqplib@0.10.1` - commit SHA: 4205e03127692a40b4871709a7134fe4e2ed5510