feat: Added attribute reconciliation for message producer spans (#2942)

lib/otel/constants.js

@@ -120,6 +120,13 @@ module.exports = {
    */
   ATTR_HTTP_STATUS_TEXT: 'http.status_text',
 
+  /**
+   * The correlation id
+   *
+   * @example MyConversationId
+   */
+  ATTR_MESSAGING_MESSAGE_CONVERSATION_ID: 'messaging.message.conversation_id',
+
   /**
    * The message destination name.
    *
@@ -157,6 +164,13 @@ module.exports = {
    */
   ATTR_MESSAGING_OPERATION_NAME: 'messaging.operation.name',
 
+  /**
+   * RabbitMQ message routing key
+   *
+   * @example myKey
+   */
+  ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY: 'messaging.rabbitmq.destination.routing_key',
+
   /**
    * Target messaging system name.
    *

lib/otel/span-processor.js

@@ -17,6 +17,8 @@ const {
   ATTR_HTTP_ROUTE,
   ATTR_HTTP_STATUS_CODE,
   ATTR_HTTP_STATUS_TEXT,
+  ATTR_MESSAGING_MESSAGE_CONVERSATION_ID,
+  ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY,
   ATTR_NET_PEER_NAME,
   ATTR_NET_PEER_PORT,
   ATTR_NET_HOST_NAME,
@@ -69,6 +71,8 @@ module.exports = class NrSpanProcessor {
       this.reconcileServerAttributes({ segment, span, transaction })
     } else if (span.kind === SpanKind.CLIENT && span.attributes[ATTR_DB_SYSTEM]) {
       this.reconcileDbAttributes({ segment, span })
+    } else if (span.kind === SpanKind.PRODUCER) {
+      this.reconcileProducerAttributes({ segment, span })
     }
     // TODO: add http external checks
   }
@@ -159,4 +163,26 @@ module.exports = class NrSpanProcessor {
       segment.addAttribute(key, sanitized)
     }
   }
+
+  reconcileProducerAttributes({ segment, span }) {
+    for (const [prop, value] of Object.entries(span.attributes)) {
+      let key = prop
+      let sanitized = value
+
+      if (prop === ATTR_SERVER_ADDRESS) {
+        key = 'host'
+        if (urltils.isLocalhost(sanitized)) {
+          sanitized = this.agent.config.getHostnameSafe(sanitized)
+        }
+      } else if (prop === ATTR_SERVER_PORT) {
+        key = 'port'
+      } else if (prop === ATTR_MESSAGING_MESSAGE_CONVERSATION_ID) {
+        key = 'correlation_id'
+      } else if (prop === ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY) {
+        key = 'routing_key'
+      }
+
+      segment.addAttribute(key, sanitized)
+    }
+  }
 }

test/versioned/otel-bridge/span.test.js

@@ -26,7 +26,9 @@ const {
   ATTR_HTTP_URL,
   ATTR_MESSAGING_DESTINATION,
   ATTR_MESSAGING_DESTINATION_KIND,
+  ATTR_MESSAGING_MESSAGE_CONVERSATION_ID,
   ATTR_MESSAGING_OPERATION,
+  ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY,
   ATTR_MESSAGING_SYSTEM,
   ATTR_NET_PEER_NAME,
   ATTR_NET_PEER_PORT,
@@ -375,10 +377,16 @@ test('Otel producer span test', (t, end) => {
   const attributes = {
     [ATTR_MESSAGING_SYSTEM]: 'messaging-lib',
     [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_SYSTEM_KIND_VALUES.QUEUE,
-    [ATTR_MESSAGING_DESTINATION]: 'test-queue'
+    [ATTR_MESSAGING_DESTINATION]: 'test-queue',
+    [ATTR_SERVER_ADDRESS]: 'localhost',
+    [ATTR_SERVER_PORT]: 5672,
+    [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: 'myKey',
+    [ATTR_MESSAGING_MESSAGE_CONVERSATION_ID]: 'MyConversationId'
   }
   helper.runInTransaction(agent, (tx) => {
     tx.name = 'prod-test'
+
+    const expectedHost = agent.config.getHostnameSafe('localhost')
     tracer.startActiveSpan('prod-test', { kind: otel.SpanKind.PRODUCER, attributes }, (span) => {
       const segment = agent.tracer.getSegment()
       assert.equal(segment.name, 'MessageBroker/messaging-lib/queue/Produce/Named/test-queue')
@@ -391,6 +399,14 @@ test('Otel producer span test', (t, end) => {
       const unscopedMetrics = tx.metrics.unscoped
       assert.equal(unscopedMetrics['MessageBroker/messaging-lib/queue/Produce/Named/test-queue'].callCount, 1)
 
+      const attrs = segment.getAttributes()
+      assert.equal(attrs.host, expectedHost)
+      assert.equal(attrs.port, 5672)
+      assert.equal(attrs.correlation_id, 'MyConversationId')
+      assert.equal(attrs.routing_key, 'myKey')
+      assert.equal(attrs[ATTR_MESSAGING_SYSTEM], 'messaging-lib')
+      assert.equal(attrs[ATTR_MESSAGING_DESTINATION], 'test-queue')
+      assert.equal(attrs[ATTR_MESSAGING_DESTINATION_KIND], MESSAGING_SYSTEM_KIND_VALUES.QUEUE)
       end()
     })
   })