feat: add kafkajs instrumentation

[david-luna]

PR to add kafkajs instrumentation. Summary of additions

• instrument send& sendBatch methods of kafkajs producers

  • creating a spans if apply (topic not ignored by config)
  • propagating traceparent info into message headers

• instruments eachMessage of kafkajs consumers

  • creating a transaction if apply (topic not ignored by config)
  • setting the trace_id if traceparent present in message headers

• instruments eachBatch of kafkajs consumers

  • creating a transaction if apply (topic not ignored by config)
  • setting the transaction links if traceparent present in message headers

Closes: #2905

Checklist

• [x] Implement code
• [x] Add tests
• [ ] Update TypeScript typings
• [x] Update documentation
• [x] Add CHANGELOG.asciidoc entry
• [x] Commit message follows commit guidelines

[david-luna]

Note: ~concurrency is an issue that make test fail~ turns out it was a timeout problem but it was not clear from the error message.

[oleh-kravchuk]

Hi, wondering about any updates for this MR, when it will be merged approx?

[trentm]

@oleh-kravchuk Thanks for asking. It helps to know there is interest. Currently this is wait on me to review it. We don't have a target date. I don't expect to get there this week, but hopefully soon after that.

[oleh-kravchuk]

@trentm Thanks for the quick reply

We are curious in the new solution (native kafkajs support) cuz currently we reused the next approach:

import { registerInstrumentations } from "@opentelemetry/instrumentation";
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
import { KafkaJsInstrumentation } from "opentelemetry-instrumentation-kafkajs";
import { Span } from "@opentelemetry/api";

const tracerProvider = new NodeTracerProvider();
tracerProvider.register();

const instrumentations = [];
instrumentations.push(
  new KafkaJsInstrumentation({
    producerHook: function (span: Span, _topic: string, message) {
      span.setAttribute("kafka_producer_hook_attribute", customParser(message));
    },
    consumerHook: function (span: Span, _topic: string, message) {
       span.setAttribute("kafka_consumer_hook_attribute", customParser(message));
    }
  })
);

registerInstrumentations({ instrumentations });

We expect a native support will speed up the initialisation phase and simplify the code base, than this one ^.

Could you compare both ways and share with us pros&cons? Thanks in advance

[max-zirkonzahn]

Hi there, are there any news / updates on the state of this MR?

[ImadSai]

I'm interested too about this feature.

[oleh-kravchuk]

@david-luna @trentm

I see that this MR is already at the final stage, isn't it?

Thanks for your efforts for shipping such a valuable changes. We look forward to use native kafkajs instrumentation on our side ;-)

[david-luna]

@trentm I've also updated the test expectations for the latest changes in span names. Now tests are green :)