feat: add kafkajs instrumentation

.ci/tav.json

@@ -24,6 +24,7 @@
     { "name": "generic-pool", "minMajorVersion": 8 },
     { "name": "graphql", "minMajorVersion": 8 },
     { "name": "ioredis", "minMajorVersion": 8 },
+    { "name": "kafkajs", "minMajorVersion": 14 },
     { "name": "knex", "minMajorVersion": 8 },
     { "name": "memcached", "minMajorVersion": 8 },
     { "name": "mongodb", "minMajorVersion": 8 },
@@ -39,7 +40,6 @@
     { "name": "ws", "minMajorVersion": 8 },
     { "name": "@koa/router,koa-router", "minMajorVersion": 8 },
     { "name": "handlebars,pug", "minMajorVersion": 8 },
-    { "name": "bluebird,got", "minMajorVersion": 8 },
-    { "name": "kafkajs", "minMajorVersion": 14 }
+    { "name": "bluebird,got", "minMajorVersion": 8 }
   ]
 }

lib/instrumentation/modules/kafkajs.js

@@ -12,6 +12,7 @@ const semver = require('semver');
 
 const constants = require('../../constants');
 const shimmer = require('../shimmer');
+const { redactKeysFromObject } = require('../../filters/sanitize-field-names');
 
 const NAME = 'Kafka';
 const TYPE = 'messaging';
@@ -103,6 +104,10 @@ module.exports = function (mod, agent, { version, enabled }) {
         return origEachMessage.apply(this, arguments);
       }
 
+      // For distributed tracing this instrumentation is going to check
+      // the headers defined by opentelemetry and ignore the propietary
+      // `elasticaapmtraceparent` header
+      // https://github.com/elastic/apm/blob/main/specs/agents/tracing-distributed-tracing.md#binary-fields
       const traceparent = message.headers && message.headers.traceparent;
       const tracestate = message.headers && message.headers.tracestate;
       const opts = {};
@@ -136,19 +141,19 @@ module.exports = function (mod, agent, { version, enabled }) {
         messageCtx.body = message.value.toString();
       }
 
-      if (message.headers) {
-        messageCtx.headers = Object.keys(message.headers).reduce(
-          (acc, name) => {
-            const value = message.headers[name];
-            if (value instanceof Buffer) {
-              acc[name] = value.toString('utf-8');
-            } else {
-              acc[name] = value;
-            }
-            return acc;
-          },
-          {},
+      if (message.headers && config.captureHeaders) {
+        // Make sure there is no sensitive data
+        // and transform non-redacted buffers
+        messageCtx.headers = redactKeysFromObject(
+          message.headers,
+          config.sanitizeFieldNamesRegExp,
         );
+        Object.keys(messageCtx.headers).forEach((key) => {
+          const value = messageCtx.headers[key];
+          if (value instanceof Buffer) {
+            messageCtx.headers[key] = value.toString('utf-8');
+          }
+        });
       }
 
       if (message.timestamp) {
@@ -192,10 +197,10 @@ module.exports = function (mod, agent, { version, enabled }) {
         return origEachBatch.apply(this, arguments);
       }
 
-      const trans = ins.startTransaction(`${NAME} RECEIVE from batch`, TYPE);
-
-      // TODO: not sure if this should be here but we got batches for only one topic
-      // https://github.com/elastic/apm/blob/main/specs/agents/tracing-instrumentation-messaging.md#context-fields
+      const trans = ins.startTransaction(
+        `${NAME} RECEIVE from ${batch.topic}`,
+        TYPE,
+      );
       const messageCtx = { queue: { name: batch.topic } };
       trans.setMessageContext(messageCtx);
 
@@ -210,6 +215,7 @@ module.exports = function (mod, agent, { version, enabled }) {
       const messages = batch && batch.messages;
 
       if (messages) {
+        const traceparentsSeen = new Set();
         const links = [];
         const limit = Math.min(
           messages.length,
@@ -218,8 +224,14 @@ module.exports = function (mod, agent, { version, enabled }) {
 
         for (let i = 0; i < messages.length; i++) {
           const msg = messages[i];
-          if (msg.headers && msg.headers.traceparent) {
-            links.push({ context: msg.headers.traceparent.toString() });
+          const traceparent =
+            msg.headers &&
+            msg.headers.traceparent &&
+            msg.headers.traceparent.toString();
+
+          if (traceparent && !traceparentsSeen.has(traceparent)) {
+            links.push({ context: traceparent });
+            traceparentsSeen.add(traceparent);
 
             if (links.length >= limit) {
               break;
@@ -299,7 +311,6 @@ module.exports = function (mod, agent, { version, enabled }) {
           parentSpan.propagateTraceContextHeaders(
             newHeaders,
             function (carrier, name, value) {
-              // TODO: why doing it everywhere???
               if (name.startsWith('elastic-')) {
                 return;
               }
@@ -356,13 +367,15 @@ module.exports = function (mod, agent, { version, enabled }) {
    */
   function wrapProducerSendBatch(origSendBatch) {
     return async function (batch) {
-      // TODO: discuss this. if all topics are ignored should the agent avoid
-      // creating a span?
       let span;
-      const topics =
-        batch.topicMessages && batch.topicMessages.map((tm) => tm.topic);
-      const shouldIgnoreBatch =
-        topics && topics.every((t) => shouldIgnoreTopic(t, config));
+      let shouldIgnoreBatch = true;
+      const messages = batch.topicMessages || [];
+      for (const msg of messages) {
+        if (!shouldIgnoreTopic(msg.topic, config)) {
+          shouldIgnoreBatch = false;
+          break;
+        }
+      }
 
       if (!shouldIgnoreBatch) {
         span = ins.createSpan(
@@ -386,7 +399,6 @@ module.exports = function (mod, agent, { version, enabled }) {
             parentSpan.propagateTraceContextHeaders(
               newHeaders,
               function (carrier, name, value) {
-                // TODO: why doing it everywhere???
                 if (name.startsWith('elastic-')) {
                   return;
                 }
@@ -405,6 +417,7 @@ module.exports = function (mod, agent, { version, enabled }) {
       const destCtx = {
         // `service.name` and `service.resource` cannot be set since `producer.sendBatch`
         // is meant to send to several topics at a time
+        // TODO: check what the Java agent is doing
         service: { type: SUBTYPE },
         target: { type: SUBTYPE },
       };

test/docker-compose.yml

@@ -144,6 +144,7 @@ services:
 
   kafka:
     # https://hub.docker.com/r/bitnami/kafka/tags
+    # Config ref: https://github.com/bitnami/containers/tree/main/bitnami/kafka#how-to-use-this-image
     image: bitnami/kafka:3.3.2
     ports:
       - "9093:9093"

test/instrumentation/modules/kafkajs/fixtures/use-kafkajs-ctx-propagation.js

@@ -74,7 +74,7 @@ async function useKafkajsClient(kafkaClient, options) {
     await producer.disconnect();
     log.info('messages sent');
   } else if (mode === 'consume') {
-    // On this mode we consumen the already sent messsages. This time they are
+    // On this mode we consume the already sent messsages. This time they are
     // instrumented (not ignored) and trace context should be added to transactions
     // this must be executed 2nd
     consumer = kafkaClient.consumer({ groupId });

test/instrumentation/modules/kafkajs/fixtures/use-kafkajs-each-message.js

@@ -15,8 +15,6 @@ const apm = require('../../../../..').start({
   stackTraceLimit: 4, // get it smaller for reviewing output
   logLevel: 'info',
   ignoreMessageQueues: ['*-ignore'],
-  captureBody: process.env.TEST_CAPTURE_BODY || 'off',
-  captureHeaders: process.env.TEST_CAPTURE_HEADERS || false,
 });
 
 const { Buffer } = require('buffer');
@@ -80,7 +78,7 @@ async function useKafkajsClient(kafkaClient, options) {
     messages: [
       { value: 'each message 1', headers: { foo: 'string' } },
       { value: 'each message 2', headers: { foo: Buffer.from('buffer') } },
-      { value: 'each message 3' },
+      { value: 'each message 3', headers: { auth: 'secret_token' } },
     ],
   });
   log.info({ data }, 'messages sent');