feat: Added segment synthesis for otel producer spans

lib/otel/rules.js

@@ -45,6 +45,7 @@ const srcJson = require('./rules.json')
 class Rule {
   static OTEL_SPAN_KIND_SERVER = 'server'
   static OTEL_SPAN_KIND_CLIENT = 'client'
+  static OTEL_SPAN_KIND_PRODUCER = 'producer'
 
   #name
   #spanKinds
@@ -82,15 +83,15 @@ class Rule {
   }
 
   get isClientRule() {
-    return this.#spanKinds.includes(Rule.OTEL_SPAN_KIND_CLIENT) || this.isProducer
+    return this.#spanKinds.includes(Rule.OTEL_SPAN_KIND_CLIENT)
   }
 
   get isConsumer() {
     return this.#spanKinds.includes('consumer')
   }
 
-  get isProducer() {
-    return this.#spanKinds.includes('producer')
+  get isProducerRule() {
+    return this.#spanKinds.includes(Rule.OTEL_SPAN_KIND_PRODUCER)
   }
 
   get isServerRule() {
@@ -126,6 +127,7 @@ class RulesEngine {
   #fallbackServerRules = new Map()
   #clientRules = new Map()
   #fallbackClientRules = new Map()
+  #fallbackProducerRules = new Map()
 
   constructor() {
     for (const inputRule of srcJson) {
@@ -136,6 +138,8 @@ class RulesEngine {
           this.#fallbackServerRules.set(rule.name, rule)
         } else if (rule.isClientRule === true) {
           this.#fallbackClientRules.set(rule.name, rule)
+        } else if (rule.isProducerRule === true) {
+          this.#fallbackProducerRules.set(rule.name, rule)
         }
         continue
       }
@@ -178,8 +182,7 @@ class RulesEngine {
         break
       }
 
-      case SpanKind.CLIENT:
-      case SpanKind.PRODUCER: {
+      case SpanKind.CLIENT: {
         for (const rule of this.#clientRules.values()) {
           if (rule.matches(otelSpan) === true) {
             result = rule
@@ -194,6 +197,18 @@ class RulesEngine {
         }
         break
       }
+
+      // there currently are no producer rules, just fallback
+      // if we add new rules they will have to be wired up
+      case SpanKind.PRODUCER: {
+        for (const rule of this.#fallbackProducerRules.values()) {
+          if (rule.matches(otelSpan) === true) {
+            result = rule
+            break
+          }
+        }
+        break
+      }
     }
 
     return result

lib/otel/rules.json

@@ -455,6 +455,7 @@
   },
   {
     "name": "FallbackProducer",
+    "type": "producer",
     "matcher": {
       "required_span_kinds": [
         "producer"

lib/otel/segment-synthesis.js

@@ -6,7 +6,12 @@
 'use strict'
 const { RulesEngine } = require('./rules')
 const defaultLogger = require('../logger').child({ component: 'segment-synthesizer' })
-const { createDbSegment, createHttpExternalSegment, createServerSegment } = require('./segments')
+const {
+  createDbSegment,
+  createHttpExternalSegment,
+  createServerSegment,
+  createProducerSegment
+} = require('./segments')
 
 class SegmentSynthesizer {
   constructor(agent, { logger = defaultLogger } = {}) {
@@ -27,10 +32,12 @@ class SegmentSynthesizer {
     }
 
     switch (rule.type) {
-      case 'external':
-        return createHttpExternalSegment(this.agent, otelSpan)
       case 'db':
         return createDbSegment(this.agent, otelSpan)
+      case 'external':
+        return createHttpExternalSegment(this.agent, otelSpan)
+      case 'producer':
+        return createProducerSegment(this.agent, otelSpan)
       case 'server':
         return createServerSegment(this.agent, otelSpan)
       default:

lib/otel/segments/index.js

@@ -7,9 +7,11 @@
 const createHttpExternalSegment = require('./http-external')
 const createDbSegment = require('./database')
 const createServerSegment = require('./server')
+const createProducerSegment = require('./producer')
 
 module.exports = {
   createDbSegment,
   createHttpExternalSegment,
+  createProducerSegment,
   createServerSegment
 }

lib/otel/segments/producer.js

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+const {
+  SEMATTRS_MESSAGING_SYSTEM,
+  SEMATTRS_MESSAGING_DESTINATION,
+  SEMATTRS_MESSAGING_DESTINATION_KIND
+} = require('@opentelemetry/semantic-conventions')
+
+module.exports = function createProducerSegment(agent, otelSpan) {
+  const context = agent.tracer.getContext()
+  const name = setName(otelSpan)
+  const segment = agent.tracer.createSegment({
+    name,
+    parent: context.segment,
+    transaction: context.transaction
+  })
+  return { segment, transaction: context.transaction }
+}
+
+function setName(otelSpan) {
+  const system = otelSpan.attributes[SEMATTRS_MESSAGING_SYSTEM] || 'Unknown'
+  const destKind = otelSpan.attributes[SEMATTRS_MESSAGING_DESTINATION_KIND] || 'Unknown'
+  const destination = otelSpan.attributes[SEMATTRS_MESSAGING_DESTINATION] || 'Unknown'
+  return `MessageBroker/${system}/${destKind}/Produce/Named/${destination}`
+}

test/unit/lib/otel/fixtures/index.js

@@ -14,6 +14,7 @@ const {
 const createSpan = require('./span')
 const createHttpClientSpan = require('./http-client')
 const { createRpcServerSpan, createHttpServerSpan, createBaseHttpSpan } = require('./server')
+const { createQueueProducerSpan, createTopicProducerSpan } = require('./producer')
 
 module.exports = {
   createBaseHttpSpan,
@@ -23,7 +24,9 @@ module.exports = {
   createHttpServerSpan,
   createMemcachedDbSpan,
   createMongoDbSpan,
+  createQueueProducerSpan,
   createRedisDbSpan,
   createRpcServerSpan,
-  createSpan
+  createSpan,
+  createTopicProducerSpan
 }

test/unit/lib/otel/fixtures/producer.js

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+const {
+  MessagingDestinationKindValues,
+  SEMATTRS_MESSAGING_SYSTEM,
+  SEMATTRS_MESSAGING_DESTINATION,
+  SEMATTRS_MESSAGING_DESTINATION_KIND
+} = require('@opentelemetry/semantic-conventions')
+const { SpanKind } = require('@opentelemetry/api')
+const createSpan = require('./span')
+
+function createTopicProducerSpan({ parentId, tracer, tx, name = 'test-span' }) {
+  const span = createSpan({ name, kind: SpanKind.PRODUCER, parentId, tracer, tx })
+  span.setAttribute(SEMATTRS_MESSAGING_SYSTEM, 'messaging-lib')
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION_KIND, MessagingDestinationKindValues.TOPIC)
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION, 'test-topic')
+  return span
+}
+
+function createQueueProducerSpan({ parentId, tracer, tx, name = 'test-span' }) {
+  const span = createSpan({ name, kind: SpanKind.PRODUCER, parentId, tracer, tx })
+  span.setAttribute(SEMATTRS_MESSAGING_SYSTEM, 'messaging-lib')
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION_KIND, MessagingDestinationKindValues.QUEUE)
+  span.setAttribute(SEMATTRS_MESSAGING_DESTINATION, 'test-queue')
+  return span
+}
+
+module.exports = {
+  createQueueProducerSpan,
+  createTopicProducerSpan
+}

test/unit/lib/otel/rules.test.js

@@ -57,11 +57,22 @@ test('fallback server rule is met', () => {
 
 test('fallback client rule is met', () => {
   const engine = new RulesEngine()
-  const span = new Span(tracer, ROOT_CONTEXT, 'test-span', spanContext, SpanKind.PRODUCER, parentId)
+  const span = new Span(tracer, ROOT_CONTEXT, 'test-span', spanContext, SpanKind.CLIENT, parentId)
   span.setAttribute('foo.bar', 'baz')
   span.end()
 
   const rule = engine.test(span)
   assert.notEqual(rule, undefined)
   assert.equal(rule.name, 'FallbackClient')
 })
+
+test('fallback producer rule is met', () => {
+  const engine = new RulesEngine()
+  const span = new Span(tracer, ROOT_CONTEXT, 'test-span', spanContext, SpanKind.PRODUCER, parentId)
+  span.setAttribute('foo.bar', 'baz')
+  span.end()
+
+  const rule = engine.test(span)
+  assert.notEqual(rule, undefined)
+  assert.equal(rule.name, 'FallbackProducer')
+})

test/unit/lib/otel/segment-synthesizer.test.js

@@ -21,7 +21,9 @@ const {
   createMongoDbSpan,
   createRedisDbSpan,
   createRpcServerSpan,
-  createMemcachedDbSpan
+  createMemcachedDbSpan,
+  createTopicProducerSpan,
+  createQueueProducerSpan
 } = require('./fixtures')
 const { SEMATTRS_DB_SYSTEM } = require('@opentelemetry/semantic-conventions')
 const { SpanKind } = require('@opentelemetry/api')
@@ -186,6 +188,32 @@ test('should create base http server segment', (t) => {
   assert.equal(transaction.name, expectedName)
 })
 
+test('should create topic producer segment', (t, end) => {
+  const { agent, synthesizer, parentId, tracer } = t.nr
+  helper.runInTransaction(agent, (tx) => {
+    const span = createTopicProducerSpan({ tx, parentId, tracer })
+    const { segment, transaction } = synthesizer.synthesize(span)
+    assert.equal(tx.id, transaction.id)
+    assert.equal(segment.name, 'MessageBroker/messaging-lib/topic/Produce/Named/test-topic')
+    assert.equal(segment.parentId, tx.trace.root.id)
+    tx.end()
+    end()
+  })
+})
+
+test('should create queue producer segment', (t, end) => {
+  const { agent, synthesizer, parentId, tracer } = t.nr
+  helper.runInTransaction(agent, (tx) => {
+    const span = createQueueProducerSpan({ tx, parentId, tracer })
+    const { segment, transaction } = synthesizer.synthesize(span)
+    assert.equal(tx.id, transaction.id)
+    assert.equal(segment.name, 'MessageBroker/messaging-lib/queue/Produce/Named/test-queue')
+    assert.equal(segment.parentId, tx.trace.root.id)
+    tx.end()
+    end()
+  })
+})
+
 test('should log warning span does not match a rule', (t, end) => {
   const { agent, synthesizer, loggerMock, parentId, tracer } = t.nr