feat: Added otel consumer span processing (#2854)

lib/otel/rules.json

@@ -165,6 +165,7 @@
   },
   {
     "name": "OtelMessagingConsumer1_24",
+    "type": "consumer",
     "matcher": {
       "required_span_kinds": [
         "consumer"
@@ -200,6 +201,7 @@
   },
   {
     "name": "FallbackConsumer",
+    "type": "consumer",
     "matcher": {
       "required_span_kinds": [
         "consumer"

lib/otel/segment-synthesis.js

@@ -7,11 +7,12 @@
 const { RulesEngine } = require('./rules')
 const defaultLogger = require('../logger').child({ component: 'segment-synthesizer' })
 const {
+  createConsumerSegment,
   createDbSegment,
   createHttpExternalSegment,
-  createServerSegment,
+  createInternalSegment,
   createProducerSegment,
-  createInternalSegment
+  createServerSegment
 } = require('./segments')
 
 class SegmentSynthesizer {
@@ -33,18 +34,33 @@ class SegmentSynthesizer {
     }
 
     switch (rule.type) {
-      case 'db':
+      case 'consumer': {
+        return createConsumerSegment(this.agent, otelSpan)
+      }
+
+      case 'db': {
         return createDbSegment(this.agent, otelSpan)
-      case 'external':
+      }
+
+      case 'external': {
         return createHttpExternalSegment(this.agent, otelSpan)
-      case 'internal':
+      }
+
+      case 'internal': {
         return createInternalSegment(this.agent, otelSpan)
-      case 'producer':
+      }
+
+      case 'producer': {
         return createProducerSegment(this.agent, otelSpan)
-      case 'server':
+      }
+
+      case 'server': {
         return createServerSegment(this.agent, otelSpan)
-      default:
+      }
+
+      default: {
         this.logger.debug('Found type: %s, no synthesis rule currently built', rule.type)
+      }
     }
   }
 }

lib/otel/segments/consumer.js

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+module.exports = createConsumerSegment
+
+// Notes:
+// + https://github.com/open-telemetry/semantic-conventions/blob/v1.24.0/docs/messaging/messaging-spans.md
+// + We probably want to inspect `messaging.system` so that we can generate
+// attributes according to our own internal specs.
+
+const Transaction = require('../../transaction/')
+const { DESTINATIONS, TYPES } = Transaction
+
+const {
+  SEMATTRS_MESSAGING_SYSTEM,
+  SEMATTRS_MESSAGING_DESTINATION,
+  SEMATTRS_MESSAGING_DESTINATION_KIND
+} = require('@opentelemetry/semantic-conventions')
+
+function createConsumerSegment(agent, otelSpan) {
+  const transaction = new Transaction(agent)
+  transaction.type = TYPES.BG
+
+  const system = otelSpan.attributes[SEMATTRS_MESSAGING_SYSTEM] ?? 'unknown'
+  const destination = otelSpan.attributes[SEMATTRS_MESSAGING_DESTINATION] ?? 'unknown'
+  const destKind = otelSpan.attributes[SEMATTRS_MESSAGING_DESTINATION_KIND] ?? 'unknown'
+  const segmentName = `OtherTransaction/Message/${system}/${destKind}/Named/${destination}`
+
+  const txAttrs = transaction.trace.attributes
+  txAttrs.addAttribute(DESTINATIONS.TRANS_SCOPE, 'message.queueName', destination)
+  // txAttrs.addAttribute(
+  //   DESTINATIONS.TRANS_SCOPE,
+  //   'host',
+  //
+  // )
+  transaction.name = segmentName
+
+  const segment = agent.tracer.createSegment({
+    name: segmentName,
+    parent: transaction.trace.root,
+    transaction
+  })
+  transaction.baseSegment = segment
+
+  return { segment, transaction }
+}

lib/otel/segments/index.js

@@ -4,13 +4,16 @@
  */
 
 'use strict'
-const createHttpExternalSegment = require('./http-external')
+
+const createConsumerSegment = require('./consumer')
 const createDbSegment = require('./database')
-const createServerSegment = require('./server')
+const createHttpExternalSegment = require('./http-external')
 const createProducerSegment = require('./producer')
+const createServerSegment = require('./server')
 const createInternalSegment = require('./internal')
 
 module.exports = {
+  createConsumerSegment,
   createDbSegment,
   createHttpExternalSegment,
   createInternalSegment,

lib/transaction/index.js

@@ -157,6 +157,7 @@ function Transaction(agent) {
   agent.emit('transactionStarted', this)
 }
 
+Transaction.DESTINATIONS = DESTS
 Transaction.TYPES = TYPES
 Transaction.TYPES_SET = TYPES_SET
 Transaction.TRANSPORT_TYPES = TRANSPORT_TYPES

test/unit/lib/otel/consumer.test.js

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+// Tests to verify that we can map OTEL "consumer" spans to NR segments.
+
+const test = require('node:test')
+const assert = require('node:assert')
+
+const { BasicTracerProvider } = require('@opentelemetry/sdk-trace-base')
+const { SpanKind } = require('@opentelemetry/api')
+const {
+  SEMATTRS_MESSAGING_SYSTEM,
+  SEMATTRS_MESSAGING_DESTINATION,
+  SEMATTRS_MESSAGING_DESTINATION_KIND
+} = require('@opentelemetry/semantic-conventions')
+
+const { DESTINATIONS } = require('../../../../lib/transaction')
+const helper = require('../../../lib/agent_helper')
+const createSpan = require('./fixtures/span')
+const SegmentSynthesizer = require('../../../../lib/otel/segment-synthesis')
+
+test.beforeEach((ctx) => {
+  const logs = []
+  const logger = {
+    debug(...args) {
+      logs.push(args)
+    }
+  }
+  const agent = helper.loadMockedAgent()
+  const synth = new SegmentSynthesizer(agent, { logger })
+  const tracer = new BasicTracerProvider().getTracer('default')
+
+  ctx.nr = {
+    agent,
+    logger,
+    logs,
+    synth,
+    tracer
+  }
+})
+
+test.afterEach((ctx) => {
+  helper.unloadAgent(ctx.nr.agent)
+})
+
+test('should create consumer segment from otel span', (t) => {
+  const { synth, tracer } = t.nr
+  const span = createSpan({ tracer, kind: SpanKind.CONSUMER })
+  span.setAttribute('messaging.operation', 'receive')
+  span.setAttribute(SEMATTRS_MESSAGING_SYSTEM, 'msgqueuer')
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION, 'dest1')
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION_KIND, 'topic1')
+
+  const expectedName = 'OtherTransaction/Message/msgqueuer/topic1/Named/dest1'
+  const { segment, transaction } = synth.synthesize(span)
+  assert.equal(segment.name, expectedName)
+  assert.equal(segment.parentId, segment.root.id)
+  assert.equal(transaction.name, expectedName)
+  assert.equal(transaction.type, 'bg')
+  assert.equal(transaction.baseSegment, segment)
+  assert.equal(
+    transaction.trace.attributes.get(DESTINATIONS.TRANS_SCOPE)['message.queueName'],
+    'dest1'
+  )
+
+  transaction.end()
+})