feat: Added consumer timeslice metrics from otel

lib/metrics/recorders/message-transaction.js

@@ -7,6 +7,11 @@
 
 const NAMES = require('../../metrics/names.js')
 
+/**
+ * @param {TraceSegment} segment
+ * @param {object} scope
+ * @param {Transaction} tx
+ */
 function recordMessageTransaction(segment, scope, tx) {
   if (tx.type !== 'message' || tx.baseSegment !== segment) {
     return

lib/otel/constants.js

@@ -141,6 +141,22 @@ module.exports = {
    */
   ATTR_MESSAGING_DESTINATION_NAME: 'messaging.destination.name',
 
+  /**
+   * Identifies the type of messaging consumer operation.
+   *
+   * {@link ATTR_MESSAGING_OPERATION_NAME} is newer and should be used.
+   */
+  ATTR_MESSAGING_OPERATION: 'messaging.operation',
+
+  /**
+   * Name of the operation being performed. Value is specific to the
+   * target messaging system.
+   *
+   * @example ack
+   * @example send
+   */
+  ATTR_MESSAGING_OPERATION_NAME: 'messaging.operation.name',
+
   /**
    * Target messaging system name.
    *

lib/otel/segments/consumer.js

@@ -13,6 +13,7 @@ module.exports = createConsumerSegment
 // attributes according to our own internal specs.
 
 const Transaction = require('../../transaction/')
+const recorder = require('../../metrics/recorders/message-transaction')
 const { DESTINATIONS, TYPES } = Transaction
 
 const {
@@ -23,12 +24,12 @@ const {
 
 function createConsumerSegment(agent, otelSpan) {
   const transaction = new Transaction(agent)
-  transaction.type = TYPES.BG
+  transaction.type = TYPES.MESSAGE
 
   const system = otelSpan.attributes[ATTR_MESSAGING_SYSTEM] ?? 'unknown'
   const destination = otelSpan.attributes[ATTR_MESSAGING_DESTINATION] ?? 'unknown'
   const destKind = otelSpan.attributes[ATTR_MESSAGING_DESTINATION_KIND] ?? 'unknown'
-  const segmentName = `OtherTransaction/Message/${system}/${destKind}/Named/${destination}`
+  const segmentName = `${system}/${destKind}/Named/${destination}`
 
   const txAttrs = transaction.trace.attributes
   txAttrs.addAttribute(DESTINATIONS.TRANS_SCOPE, 'message.queueName', destination)
@@ -37,10 +38,11 @@ function createConsumerSegment(agent, otelSpan) {
   //   'host',
   //
   // )
-  transaction.name = segmentName
+  transaction.setPartialName(segmentName)
 
   const segment = agent.tracer.createSegment({
-    name: segmentName,
+    recorder,
+    name: transaction.getFullName(),
     parent: transaction.trace.root,
     transaction
   })

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

@@ -57,15 +57,14 @@ test('should create consumer segment from otel span', (t) => {
 
   const expectedName = 'OtherTransaction/Message/msgqueuer/topic1/Named/dest1'
   const { segment, transaction } = synth.synthesize(span)
+  transaction.end()
   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.type, 'message')
   assert.equal(transaction.baseSegment, segment)
   assert.equal(
     transaction.trace.attributes.get(DESTINATIONS.TRANS_SCOPE)['message.queueName'],
     'dest1'
   )
-
-  transaction.end()
 })

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

@@ -20,12 +20,13 @@ const {
   ATTR_GRPC_STATUS_CODE,
   ATTR_HTTP_HOST,
   ATTR_HTTP_METHOD,
-  ATTR_HTTP_URL,
   ATTR_HTTP_ROUTE,
   ATTR_HTTP_STATUS_CODE,
   ATTR_HTTP_STATUS_TEXT,
+  ATTR_HTTP_URL,
   ATTR_MESSAGING_DESTINATION,
   ATTR_MESSAGING_DESTINATION_KIND,
+  ATTR_MESSAGING_OPERATION,
   ATTR_MESSAGING_SYSTEM,
   ATTR_NET_PEER_NAME,
   ATTR_NET_PEER_PORT,
@@ -394,3 +395,39 @@ test('Otel producer span test', (t, end) => {
     })
   })
 })
+
+test('messaging consumer metrics are bridged correctly', (t, end) => {
+  const { agent, tracer } = t.nr
+  const attributes = {
+    [ATTR_MESSAGING_SYSTEM]: 'kafka',
+    [ATTR_MESSAGING_OPERATION]: 'getMessage',
+    [ATTR_SERVER_ADDRESS]: '127.0.0.1',
+    [ATTR_MESSAGING_DESTINATION]: 'work-queue',
+    [ATTR_MESSAGING_DESTINATION_KIND]: 'queue'
+  }
+
+  tracer.startActiveSpan('consumer-test', { kind: otel.SpanKind.CONSUMER, attributes }, (span) => {
+    const tx = agent.getTransaction()
+    const segment = agent.tracer.getSegment()
+    span.end()
+    const duration = hrTimeToMilliseconds(span.duration)
+    assert.equal(duration, segment.getDurationInMillis())
+    tx.end()
+
+    assert.equal(segment.name, 'OtherTransaction/Message/kafka/queue/Named/work-queue')
+    assert.equal(tx.type, 'message')
+
+    const unscopedMetrics = tx.metrics.unscoped
+    const expectedMetrics = [
+      'OtherTransaction/all',
+      'OtherTransaction/Message/all',
+      'OtherTransaction/Message/kafka/queue/Named/work-queue',
+      'OtherTransactionTotalTime'
+    ]
+    for (const expectedMetric of expectedMetrics) {
+      assert.equal(unscopedMetrics[expectedMetric].callCount, 1, `${expectedMetric}.callCount`)
+    }
+
+    end()
+  })
+})