feat(otel-node): add @opentelemetry/instrumentation-kafkajs

.github/workflows/test-edot.yml

@@ -63,6 +63,25 @@ jobs:
         ports:
           - 9200:9200
 
+      kafka:
+        image: bitnami/kafka:3.4
+        env:
+          KAFKA_CFG_PROCESS_ROLES: 'broker,controller'
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+          KAFKA_CFG_LISTENERS: 'PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094'
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT'
+          KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT://127.0.0.1:9092,EXTERNAL://kafka_b:9094'
+          KAFKA_BROKER_ID: '1'
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '[email protected]:9093'
+          ALLOW_PLAINTEXT_LISTENER: 'yes'
+          KAFKA_CFG_NODE_ID: '1'
+          KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+          BITNAMI_DEBUG: 'yes'
+          KAFKA_CFG_NUM_PARTITIONS: '2'
+        ports:
+          - "9092:9092"
+          - "9094:9094"
+
       mongodb:
         image: mongo:7
         ports:
@@ -103,7 +122,6 @@ jobs:
         ports:
           - 6379:6379
 
-
     steps:
     - uses: actions/checkout@v4
     - uses: actions/setup-node@v4

packages/opentelemetry-node/docs/supported-technologies.md

@@ -47,6 +47,7 @@ requires:
 | `@opentelemetry/instrumentation-hapi` | `@hapi/hapi >=17.0.0 <21` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/opentelemetry-instrumentation-hapi#readme) |
 | `@opentelemetry/instrumentation-http` | `http` module for Node.js `>=14` | [README](https://github.com/open-telemetry/opentelemetry-js/tree/main/experimental/packages/opentelemetry-instrumentation-http#readme) |
 | `@opentelemetry/instrumentation-ioredis` | `ioredis` version range `>=2 <6` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/opentelemetry-instrumentation-ioredis#readme) |
+| `@opentelemetry/instrumentation-kafkajs` | `kafkajs` version range `>=0.1.0 <3` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/instrumentation-kafkajs#readme) |
 | `@opentelemetry/instrumentation-knex` | `knex` version range `>=0.10.0` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/opentelemetry-instrumentation-knex#readme) |
 | `@opentelemetry/instrumentation-koa` | `koa` version range `^2.0.0` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/opentelemetry-instrumentation-koa#readme) |
 | `@opentelemetry/instrumentation-lru-memoizer` | `lru-memoizer` version range `>=1.3 <3` | [README](https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/plugins/node/instrumentation-lru-memoizer#readme) |

packages/opentelemetry-node/lib/instrumentations.js

@@ -40,6 +40,7 @@
  *  "@opentelemetry/instrumentation-hapi": import('@opentelemetry/instrumentation').InstrumentationConfig | InstrumentationFactory,
  *  "@opentelemetry/instrumentation-http": import('@opentelemetry/instrumentation-http').HttpInstrumentationConfig | InstrumentationFactory,
  *  "@opentelemetry/instrumentation-ioredis": import('@opentelemetry/instrumentation-ioredis').IORedisInstrumentationConfig | InstrumentationFactory,
+ *  "@opentelemetry/instrumentation-kafkajs": import('@opentelemetry/instrumentation-kafkajs').KafkaJsInstrumentation | InstrumentationFactory,
  *  "@opentelemetry/instrumentation-knex": import('@opentelemetry/instrumentation-knex').KnexInstrumentationConfig | InstrumentationFactory,
  *  "@opentelemetry/instrumentation-koa": import('@opentelemetry/instrumentation-koa').KoaInstrumentationConfig | InstrumentationFactory,
  *  "@opentelemetry/instrumentation-lru-memoizer": import('@opentelemetry/instrumentation').InstrumentationConfig | InstrumentationFactory,
@@ -82,6 +83,7 @@ const {HapiInstrumentation} = require('@opentelemetry/instrumentation-hapi');
 const {HttpInstrumentation} = require('@opentelemetry/instrumentation-http');
 const {IORedisInstrumentation} = require('@opentelemetry/instrumentation-ioredis');
 const {KnexInstrumentation} = require('@opentelemetry/instrumentation-knex');
+const {KafkaJsInstrumentation} = require('@opentelemetry/instrumentation-kafkajs');
 const {KoaInstrumentation} = require('@opentelemetry/instrumentation-koa');
 const {LruMemoizerInstrumentation} = require('@opentelemetry/instrumentation-lru-memoizer');
 const {MemcachedInstrumentation} = require('@opentelemetry/instrumentation-memcached');
@@ -132,6 +134,7 @@ const INSTRUMENTATIONS = {
     '@opentelemetry/instrumentation-http': (cfg) => new HttpInstrumentation(cfg),
     '@opentelemetry/instrumentation-ioredis': (cfg) => new IORedisInstrumentation(cfg),
     '@opentelemetry/instrumentation-knex': (cfg) => new KnexInstrumentation(cfg),
+    '@opentelemetry/instrumentation-kafkajs': (cfg) => new KafkaJsInstrumentation(cfg),
     '@opentelemetry/instrumentation-koa': (cfg) => new KoaInstrumentation(cfg),
     '@opentelemetry/instrumentation-lru-memoizer': (cfg) => new LruMemoizerInstrumentation(cfg),
     '@opentelemetry/instrumentation-memcached': (cfg) => new MemcachedInstrumentation(cfg),

packages/opentelemetry-node/package.json

@@ -93,6 +93,7 @@
     "@opentelemetry/instrumentation-hapi": "^0.45.1",
     "@opentelemetry/instrumentation-http": "^0.57.1",
     "@opentelemetry/instrumentation-ioredis": "^0.47.0",
+    "@opentelemetry/instrumentation-kafkajs": "^0.7.0",
     "@opentelemetry/instrumentation-knex": "^0.44.0",
     "@opentelemetry/instrumentation-koa": "^0.47.0",
     "@opentelemetry/instrumentation-lru-memoizer": "^0.44.0",
@@ -143,6 +144,7 @@
     "fastify": "^5.1.0",
     "graphql": "^16.10.0",
     "ioredis": "^5.4.1",
+    "kafkajs": "^2.2.4",
     "module-details-from-path": "^1.0.3",
     "mongodb": "^6.9.0",
     "mongoose": "^8.9.4",

packages/opentelemetry-node/test/docker-compose.yaml

@@ -40,6 +40,36 @@ services:
       timeout: 10s
       retries: 30
 
+  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.4
+    ports:
+      - "9092:9092"
+      - "9094:9094"
+    environment:
+      # This enables in KRAFT mode
+      # ref: https://kafka.apache.org/documentation/#kraft
+      # ref: https://github.com/bitnami/containers/blob/283ec43b9dc54ef92a6ec031282e8be3fe280f56/bitnami/kafka/README.md?plain=1#L933
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092,EXTERNAL://kafka_b:9094
+      - KAFKA_BROKER_ID=1
+      - [email protected]:9093
+      - ALLOW_PLAINTEXT_LISTENER=yes
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
+      - BITNAMI_DEBUG=yes
+      - KAFKA_CFG_NUM_PARTITIONS=2
+    healthcheck:
+      # Kafka healthcheck ideas from https://github.com/wurstmeister/kafka-docker/issues/167
+      test: kafka-cluster.sh cluster-id --bootstrap-server localhost:9092 || exit 1
+      interval: 1s
+      timeout: 10s
+      retries: 30
+
   mongodb:
     image: mongo:7
     ports:

packages/opentelemetry-node/test/fixtures/use-kafkajs.js

@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Usage: node --env-file ../test-services.env -r @elastic/opentelemetry-node use-kafkajs.js
+
+const otel = require('@opentelemetry/api');
+const {Kafka} = require('kafkajs');
+
+const host = process.env.KAFKA_HOST;
+const port = process.env.KAFKA_PORT || '9092';
+const broker = `${host}:${port}`;
+const clientId = process.env.TEST_KAFKAJS_CLIENT_ID || 'test-kafkajs-client';
+const topic =
+    process.env.TEST_KAFKAJS_TOPIC ||
+    `test-${Math.floor(Math.random() * 1000)}`;
+
+async function main() {
+    const kafkaClient = new Kafka({clientId, brokers: [broker]});
+
+    const admin = kafkaClient.admin();
+    const producer = kafkaClient.producer();
+
+    // Create the topics & subscribe
+    await admin.connect();
+    await admin.createTopics({
+        waitForLeaders: true,
+        topics: [{topic}],
+    });
+    console.log('topic created');
+
+    await producer.connect();
+    const data = await producer.send({
+        topic,
+        messages: [
+            {value: 'message 1', headers: {foo: 'foo 1'}},
+            {value: 'message 2', headers: {foo: Buffer.from('foo 2')}},
+        ],
+    });
+    console.log({data}, 'messages sent');
+    await producer.disconnect();
+    console.log('producer disconnect');
+    await admin.deleteTopics({topics: [topic]});
+    console.log('topics deleted');
+    await admin.disconnect();
+    console.log('admin disconnect');
+}
+
+const tracer = otel.trace.getTracer('test');
+tracer.startActiveSpan('manual-parent-span', (span) => {
+    main();
+    span.end();
+});

packages/opentelemetry-node/test/instr-kafkajs.test.js

@@ -0,0 +1,67 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Test that instrumentation-kafkajs generates the telemetry we expect.
+
+const test = require('tape');
+const {runTestFixtures, filterOutDnsNetSpans} = require('./testutils');
+
+let skip = process.env.KAFKA_HOST === undefined;
+if (skip) {
+    console.log(
+        '# SKIP kafkajs tests: KAFKA_HOST is not set (try with `KAFKA_HOST=localhost`)'
+    );
+}
+
+/** @type {import('./testutils').TestFixture[]} */
+const testFixtures = [
+    {
+        name: 'use-kafkajs',
+        args: ['./fixtures/use-kafkajs.js'],
+        cwd: __dirname,
+        env: {
+            NODE_OPTIONS: '--require=@elastic/opentelemetry-node',
+            TEST_KAFKAJS_TOPIC: 'edot-test-topic',
+        },
+        // verbose: true,
+        checkTelemetry: (t, col) => {
+            // We expect spans like this
+            // ------ trace 916148 (7 spans) ------
+            //        span 199d7e "manual-parent-span" (3.4ms, SPAN_KIND_INTERNAL)
+            //   -3ms `- span 2da778 "edot-test-topic" (19.6ms, SPAN_KIND_PRODUCER)
+            //   +1ms `- span 6d088b "edot-test-topic" (19.4ms, SPAN_KIND_PRODUCER)
+            const spans = filterOutDnsNetSpans(col.sortedSpans.slice(1));
+            t.equal(spans.length, 2);
+
+            t.ok(
+                spans.every(
+                    (s) => s.scope.name,
+                    '@opentelemetry/instrumentation-kafkajs'
+                )
+            );
+            t.ok(spans.every((s) => s.name, 'edot-test-topic'));
+            t.ok(spans.every((s) => s.kind, 'SPAN_KIND_PRODUCER'));
+        },
+    },
+];
+
+test('kafkajs instrumentation', {skip}, async (suite) => {
+    await runTestFixtures(suite, testFixtures);
+    suite.end();
+});

packages/opentelemetry-node/test/test-services.env

@@ -3,6 +3,7 @@ PGHOST=localhost
 MONGODB_HOST=localhost
 MSSQL_HOST=localhost
 MYSQL_HOST=localhost
+KAFKA_HOST=localhost
 CASSANDRA_HOST=localhost
 ES_URL=http://localhost:9200  # Elasticsearch
 

packages/opentelemetry-node/types/instrumentations.d.ts

@@ -17,6 +17,7 @@ export type InstrumentaionsMap = {
     "@opentelemetry/instrumentation-hapi": import('@opentelemetry/instrumentation').InstrumentationConfig | InstrumentationFactory;
     "@opentelemetry/instrumentation-http": import('@opentelemetry/instrumentation-http').HttpInstrumentationConfig | InstrumentationFactory;
     "@opentelemetry/instrumentation-ioredis": import('@opentelemetry/instrumentation-ioredis').IORedisInstrumentationConfig | InstrumentationFactory;
+    "@opentelemetry/instrumentation-kafkajs": import('@opentelemetry/instrumentation-kafkajs').KafkaJsInstrumentation | InstrumentationFactory;
     "@opentelemetry/instrumentation-knex": import('@opentelemetry/instrumentation-knex').KnexInstrumentationConfig | InstrumentationFactory;
     "@opentelemetry/instrumentation-koa": import('@opentelemetry/instrumentation-koa').KoaInstrumentationConfig | InstrumentationFactory;
     "@opentelemetry/instrumentation-lru-memoizer": import('@opentelemetry/instrumentation').InstrumentationConfig | InstrumentationFactory;