[DSM] Add messaging.destination tag to spans so they show up in queue pages (#5239)

ericfirth · web-flow · commit c8940851ed2c · 2025-02-24T14:15:48.000-05:00
diff --git a/packages/datadog-plugin-kafkajs/src/consumer.js b/packages/datadog-plugin-kafkajs/src/consumer.js
@@ -7,6 +7,8 @@ const ConsumerPlugin = require('../../dd-trace/src/plugins/consumer')
 const afterStartCh = dc.channel('dd-trace:kafkajs:consumer:afterStart')
 const beforeFinishCh = dc.channel('dd-trace:kafkajs:consumer:beforeFinish')
 
+const MESSAGING_DESTINATION_KEY = 'messaging.destination.name'
+
 class KafkajsConsumerPlugin extends ConsumerPlugin {
   static get id () { return 'kafkajs' }
   static get operation () { return 'consume' }
@@ -72,7 +74,8 @@ class KafkajsConsumerPlugin extends ConsumerPlugin {
         component: 'kafkajs',
         'kafka.topic': topic,
         'kafka.message.offset': message.offset,
-        'kafka.cluster_id': clusterId
+        'kafka.cluster_id': clusterId,
+        [MESSAGING_DESTINATION_KEY]: topic
       },
       metrics: {
         'kafka.partition': partition
diff --git a/packages/datadog-plugin-kafkajs/src/producer.js b/packages/datadog-plugin-kafkajs/src/producer.js
@@ -5,6 +5,7 @@ const { DsmPathwayCodec } = require('../../dd-trace/src/datastreams/pathway')
 const { getMessageSize } = require('../../dd-trace/src/datastreams/processor')
 
 const BOOTSTRAP_SERVERS_KEY = 'messaging.kafka.bootstrap.servers'
+const MESSAGING_DESTINATION_KEY = 'messaging.destination.name'
 
 class KafkajsProducerPlugin extends ProducerPlugin {
   static get id () { return 'kafkajs' }
@@ -72,7 +73,8 @@ class KafkajsProducerPlugin extends ProducerPlugin {
       meta: {
         component: 'kafkajs',
         'kafka.topic': topic,
-        'kafka.cluster_id': clusterId
+        'kafka.cluster_id': clusterId,
+        [MESSAGING_DESTINATION_KEY]: topic
       },
       metrics: {
         'kafka.batch_size': messages.length
diff --git a/packages/datadog-plugin-kafkajs/test/index.spec.js b/packages/datadog-plugin-kafkajs/test/index.spec.js
@@ -73,7 +73,9 @@ describe('Plugin', () => {
             const meta = {
               'span.kind': 'producer',
               component: 'kafkajs',
-              'pathway.hash': expectedProducerHash.readBigUInt64BE(0).toString()
+              'pathway.hash': expectedProducerHash.readBigUInt64BE(0).toString(),
+              'messaging.destination.name': 'test-topic',
+              'messaging.kafka.bootstrap.servers': '127.0.0.1:9092'
             }
             if (clusterIdAvailable) meta['kafka.cluster_id'] = testKafkaClusterId
 
@@ -182,7 +184,8 @@ describe('Plugin', () => {
               meta: {
                 'span.kind': 'consumer',
                 component: 'kafkajs',
-                'pathway.hash': expectedConsumerHash.readBigUInt64BE(0).toString()
+                'pathway.hash': expectedConsumerHash.readBigUInt64BE(0).toString(),
+                'messaging.destination.name': 'test-topic'
               },
               resource: testTopic,
               error: 0,
