refactor(instrumentation-kafkajs): move client duration metric to separate method for testability

Author: brianphillips

plugins/node/instrumentation-kafkajs/src/instrumentation.ts

@@ -251,19 +251,26 @@ export class KafkaJsInstrumentation extends InstrumentationBase<KafkaJsInstrumen
   private _setKafkaEventListeners(kafkaObj: KafkaEventEmitter) {
     if (kafkaObj[EVENT_LISTENERS_SET]) return;
 
-    kafkaObj.on(kafkaObj.events.REQUEST, event => {
-      const [address, port] = event.payload.broker.split(':');
-      this._clientDuration.record(event.payload.duration / 1000, {
-        [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
-        [ATTR_MESSAGING_OPERATION_NAME]: `${event.payload.apiName}`, // potentially suffix with @v${event.payload.apiVersion}?
-        [ATTR_SERVER_ADDRESS]: address,
-        [ATTR_SERVER_PORT]: Number.parseInt(port, 10),
-      });
-    });
+    kafkaObj.on(
+      kafkaObj.events.REQUEST,
+      this._recordClientDurationMetric.bind(this)
+    );
 
     kafkaObj[EVENT_LISTENERS_SET] = true;
   }
 
+  private _recordClientDurationMetric(
+    event: Pick<kafkaJs.RequestEvent, 'payload'>
+  ) {
+    const [address, port] = event.payload.broker.split(':');
+    this._clientDuration.record(event.payload.duration / 1000, {
+      [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
+      [ATTR_MESSAGING_OPERATION_NAME]: `${event.payload.apiName}`, // potentially suffix with @v${event.payload.apiVersion}?
+      [ATTR_SERVER_ADDRESS]: address,
+      [ATTR_SERVER_PORT]: Number.parseInt(port, 10),
+    });
+  }
+
   private _getProducerPatch() {
     const instrumentation = this;
     return (original: kafkaJs.Kafka['producer']) => {

plugins/node/instrumentation-kafkajs/test/kafkajs.test.ts

@@ -38,6 +38,7 @@ import {
   MESSAGING_SYSTEM_VALUE_KAFKA,
   METRIC_MESSAGING_PROCESS_DURATION,
   METRIC_MESSAGING_CLIENT_CONSUMED_MESSAGES,
+  METRIC_MESSAGING_CLIENT_OPERATION_DURATION,
 } from '../src/semconv';
 import {
   getTestSpans,
@@ -68,7 +69,11 @@ import {
 } from 'kafkajs';
 import { DummyPropagation } from './DummyPropagation';
 import { bufferTextMapGetter } from '../src/propagator';
-import { ATTR_ERROR_TYPE } from '@opentelemetry/semantic-conventions';
+import {
+  ATTR_ERROR_TYPE,
+  ATTR_SERVER_ADDRESS,
+  ATTR_SERVER_PORT,
+} from '@opentelemetry/semantic-conventions';
 
 function assertMetricCollection(
   { errors, resourceMetrics }: CollectionResult,
@@ -1334,6 +1339,39 @@ describe('instrumentation-kafkajs', () => {
       );
     });
   });
+  describe('client duration metric', () => {
+    it('records the metric', async () => {
+      instrumentation['_recordClientDurationMetric']({
+        payload: {
+          broker: 'kafka.host:4789',
+          duration: 242,
+          apiName: 'some-operation',
+          apiKey: 123,
+          apiVersion: 1,
+          clientId: 'client-id',
+          correlationId: 456,
+          createdAt: Date.now(),
+          pendingDuration: 0,
+          sentAt: Date.now(),
+          size: 1024,
+        },
+      });
+      assertMetricCollection(await metricReader.collect(), {
+        [METRIC_MESSAGING_CLIENT_OPERATION_DURATION]: [
+          {
+            count: 1,
+            value: 0.232,
+            attributes: {
+              [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
+              [ATTR_MESSAGING_OPERATION_NAME]: 'some-operation',
+              [ATTR_SERVER_ADDRESS]: 'kafka.host',
+              [ATTR_SERVER_PORT]: 4789,
+            },
+          },
+        ],
+      });
+    });
+  });
 
   describe('bufferTextMapGetter', () => {
     it('is possible to retrieve keys case insensitively', () => {