opentelemetry-ext-js/packages/instrumentation-kafkajs not working

[bekesgy]

Hi @blumamir ,

I am trying to use instrimentation-kafkajs to autoinstrument our Kafka-producer and consumer. After debugging, it seems the relevant methods are not called. I am trying to initialize it as follows:
import { KafkaJsInstrumentation } from 'opentelemetry-instrumentation-kafkajs';

// Enable OpenTelemetry exporters to export traces to Grafan Tempo.
const tracerProvider = new NodeTracerProvider ({
// sampler: new AlwaysOnSampler(),
resource: new Resource({
"service.name": process.env.SERVICE_NAME ?? 'default',
"service.instance.id": process.env.SERVICE_INSTANTE_ID ?? 'default'
}),
plugins: {
kafkajs: { enabled: false, path: 'opentelemetry-plugin-kafkajs' }
}
});
registerInstrumentations({
tracerProvider: tracerProvider,
instrumentations: [
/* new ExpressInstrumentation(),
new HttpInstrumentation(),
new AwsInstrumentation(),
getNodeAutoInstrumentations(),*/
new KafkaJsInstrumentation()
]
});
trace.setGlobalTracerProvider(tracerProvider);

tracerProvider.addSpanProcessor(new SimpleSpanProcessor(new ConsoleSpanExporter()));tracerProvider.register();

export const tracer = opentelemetry.trace.getTracer(process.env.serviceName ?? 'default', '1.0');

package.json:
"dependencies": {
"@aspecto/opentelemetry-instrumentation-express": "^0.0.115",
"@dynatrace/oneagent-sdk": "^1.4.1",
"@opentelemetry/api": "^1.0.3",
"@opentelemetry/auto-instrumentations-node": "^0.24.0",
"@opentelemetry/exporter-collector-proto": "^0.24.0",
"@opentelemetry/exporter-jaeger": "^0.24.0",
"@opentelemetry/instrumentation": "^0.24.0",
"@opentelemetry/instrumentation-http": "^0.24.0",
"@opentelemetry/node": "^0.24.0",
"@types/body-parser": "^1.19.1",
"body-parser": "^1.19.0",
"dotenv": "^10.0.0",
"express": "^4.17.1",
"find-config": "^1.0.0",
"kafkajs": "^1.15.0",
"mappersmith": "^2.34.0",
"opentelemetry-instrumentation-aws-sdk": "^0.24.2",
"opentelemetry-instrumentation-kafkajs": "^0.24.1",
"winston": "^3.3.3"
}

Anything I am missing? Thanks in advance.

[blumamir]

@bekesgy - can you please paste a snippet of the code you are trying to instrument?

BTW, I see you are manually creating the service resource. If you like, you can use the resource-detector-service to automate it, and additional fallback to read the service name and version from package.json if not specified

[bekesgy]

@blumamir simple producer and consumer (this is just a quick POC):

Producer:

export const sendMessage = async (message: Event) => {
await kafkaProducer.connect();

try {
  await kafkaProducer.send({
    topic: configuration.topic,
    messages: [{
        value: JSON.stringify({
          ...message
        })
    },
    {
      value: JSON.stringify({
        ...message
      })
    }]});
} catch (error) {
} finally {

}

kafkaProducer.disconnect();
}

Consumer:
export const messageHandler = async (payload: EachMessagePayload) : Promise => {
logger.info(Consumer #1: Message ${payload.message.value} received from ${payload.topic}:${payload.partition}:${payload.message.offset});
}

export const start = async (): Promise => {
await kafkaConsumer.subscribe({ topic: configuration.topic, fromBeginning: false });
await kafkaConsumer.run({
eachMessage: messageHandler
});
return true;
}

I have tried the resource-detector, which works well... the kafkajs autoinstrumentation is not yet working.

[blumamir]

Copied your code into this simple repo, and it produces spans:
https://github.com/blumamir/otel-kafka-example

Please take a look.
The problem is probably in not in the instrumentation I believe. It is probably somewhere in your otel setup.
If you supply a link to a full working, minimal repo that does not produce any kafkajs spans, I can take a look and try to help you find the issue.