From 000a8410f75987e1c8858cbd37dadec02e52f231 Mon Sep 17 00:00:00 2001 From: Nicolas Hrubec Date: Tue, 19 May 2026 14:54:47 +0200 Subject: [PATCH 1/3] ref(node): Vendor kafkajs instrumentation Closes https://github.com/getsentry/sentry-javascript/issues/20151 Co-Authored-By: Claude Opus 4.6 (1M context) --- .oxlintrc.base.json | 3 +- packages/node/package.json | 1 - .../tracing/{kafka.ts => kafka/index.ts} | 2 +- .../tracing/kafka/vendored/instrumentation.ts | 639 ++++++++++++++++++ .../tracing/kafka/vendored/internal-types.ts | 33 + .../tracing/kafka/vendored/kafkajs-types.ts | 128 ++++ .../tracing/kafka/vendored/propagator.ts | 49 ++ .../tracing/kafka/vendored/semconv.ts | 173 +++++ .../tracing/kafka/vendored/types.ts | 52 ++ yarn.lock | 8 - 10 files changed, 1077 insertions(+), 11 deletions(-) rename packages/node/src/integrations/tracing/{kafka.ts => kafka/index.ts} (93%) create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/internal-types.ts create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/propagator.ts create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/semconv.ts create mode 100644 packages/node/src/integrations/tracing/kafka/vendored/types.ts diff --git a/.oxlintrc.base.json b/.oxlintrc.base.json index 82f9f4ac5d50..c2f5d9db7ebc 100644 --- a/.oxlintrc.base.json +++ b/.oxlintrc.base.json @@ -147,7 +147,8 @@ "**/integrations/fs/vendored/**/*.ts", "**/integrations/tracing/knex/vendored/**/*.ts", "**/integrations/tracing/mongo/vendored/**/*.ts", - "**/integrations/tracing/connect/vendored/**/*.ts" + "**/integrations/tracing/connect/vendored/**/*.ts", + "**/integrations/tracing/kafka/vendored/**/*.ts" ], "rules": { "typescript/no-explicit-any": "off" diff --git a/packages/node/package.json b/packages/node/package.json index e8a57e00a78d..ff9dcf500366 100644 --- a/packages/node/package.json +++ b/packages/node/package.json @@ -72,7 +72,6 @@ "@opentelemetry/instrumentation-graphql": "0.62.0", "@opentelemetry/instrumentation-hapi": "0.60.0", "@opentelemetry/instrumentation-http": "0.214.0", - "@opentelemetry/instrumentation-kafkajs": "0.23.0", "@opentelemetry/instrumentation-koa": "0.62.0", "@opentelemetry/instrumentation-mongoose": "0.60.0", "@opentelemetry/instrumentation-mysql": "0.60.0", diff --git a/packages/node/src/integrations/tracing/kafka.ts b/packages/node/src/integrations/tracing/kafka/index.ts similarity index 93% rename from packages/node/src/integrations/tracing/kafka.ts rename to packages/node/src/integrations/tracing/kafka/index.ts index 94787d92b5b1..08677af4795d 100644 --- a/packages/node/src/integrations/tracing/kafka.ts +++ b/packages/node/src/integrations/tracing/kafka/index.ts @@ -1,4 +1,4 @@ -import { KafkaJsInstrumentation } from '@opentelemetry/instrumentation-kafkajs'; +import { KafkaJsInstrumentation } from './vendored/instrumentation'; import type { IntegrationFn } from '@sentry/core'; import { defineIntegration } from '@sentry/core'; import { addOriginToSpan, generateInstrumentOnce } from '@sentry/node-core'; diff --git a/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts b/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts new file mode 100644 index 000000000000..453a40d3f3b1 --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts @@ -0,0 +1,639 @@ +/* + * Copyright The OpenTelemetry Authors, Aspecto + * + * Licensed 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 + * + * https://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. + * + * NOTICE from the Sentry authors: + * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs + * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 + * - Some types vendored from kafkajs with simplifications + */ +/* eslint-disable */ + +import { + Attributes, + Context, + context, + Counter, + Histogram, + Link, + propagation, + ROOT_CONTEXT, + Span, + SpanKind, + SpanStatusCode, + trace, +} from '@opentelemetry/api'; +import { + InstrumentationBase, + InstrumentationNodeModuleDefinition, + isWrapped, + safeExecuteInTheMiddle, +} from '@opentelemetry/instrumentation'; +import { + ATTR_ERROR_TYPE, + ATTR_SERVER_ADDRESS, + ATTR_SERVER_PORT, + ERROR_TYPE_VALUE_OTHER, +} from '@opentelemetry/semantic-conventions'; +import type { Kafka, Transaction, Producer, ConsumerEvents, ProducerEvents, RequestEvent } from './kafkajs-types'; +import type { + Consumer, + ConsumerRunConfig, + EachBatchHandler, + EachMessageHandler, + KafkaMessage, + Message, + RecordMetadata, +} from './kafkajs-types'; +import { EVENT_LISTENERS_SET } from './internal-types'; +import { bufferTextMapGetter } from './propagator'; +import { + ATTR_MESSAGING_BATCH_MESSAGE_COUNT, + ATTR_MESSAGING_DESTINATION_NAME, + ATTR_MESSAGING_DESTINATION_PARTITION_ID, + ATTR_MESSAGING_KAFKA_MESSAGE_KEY, + ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE, + ATTR_MESSAGING_KAFKA_OFFSET, + ATTR_MESSAGING_OPERATION_NAME, + ATTR_MESSAGING_OPERATION_TYPE, + ATTR_MESSAGING_SYSTEM, + MESSAGING_OPERATION_TYPE_VALUE_PROCESS, + MESSAGING_OPERATION_TYPE_VALUE_RECEIVE, + MESSAGING_OPERATION_TYPE_VALUE_SEND, + MESSAGING_SYSTEM_VALUE_KAFKA, + METRIC_MESSAGING_CLIENT_CONSUMED_MESSAGES, + METRIC_MESSAGING_CLIENT_OPERATION_DURATION, + METRIC_MESSAGING_CLIENT_SENT_MESSAGES, + METRIC_MESSAGING_PROCESS_DURATION, +} from './semconv'; +import { KafkaJsInstrumentationConfig } from './types'; + +interface ConsumerSpanOptions { + topic: string; + message: KafkaMessage | undefined; + operationType: string; + attributes: Attributes; + ctx?: Context | undefined; + link?: Link; +} +import { SDK_VERSION } from '@sentry/core'; + +const PACKAGE_NAME = '@sentry/instrumentation-kafkajs'; + +// This interface acts as a strict subset of the KafkaJS Consumer and +// Producer interfaces (just for the event we're needing) +interface KafkaEventEmitter { + on(eventName: ConsumerEvents['REQUEST'] | ProducerEvents['REQUEST'], listener: (event: RequestEvent) => void): void; + events: { + REQUEST: ConsumerEvents['REQUEST'] | ProducerEvents['REQUEST']; + }; + [EVENT_LISTENERS_SET]?: boolean; +} + +interface StandardAttributes extends Attributes { + [ATTR_MESSAGING_SYSTEM]: string; + [ATTR_MESSAGING_OPERATION_NAME]: OP; + [ATTR_ERROR_TYPE]?: string; +} +interface TopicAttributes { + [ATTR_MESSAGING_DESTINATION_NAME]: string; + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]?: string; +} + +interface ClientDurationAttributes extends StandardAttributes, Partial { + [ATTR_SERVER_ADDRESS]: string; + [ATTR_SERVER_PORT]: number; + [ATTR_MESSAGING_OPERATION_TYPE]?: string; +} +interface SentMessagesAttributes extends StandardAttributes<'send'>, TopicAttributes { + [ATTR_ERROR_TYPE]?: string; +} +type ConsumedMessagesAttributes = StandardAttributes<'receive' | 'process'>; +interface MessageProcessDurationAttributes extends StandardAttributes<'process'>, TopicAttributes { + [ATTR_MESSAGING_SYSTEM]: string; + [ATTR_MESSAGING_OPERATION_NAME]: 'process'; + [ATTR_ERROR_TYPE]?: string; +} +type RecordPendingMetric = (errorType?: string | undefined) => void; + +function prepareCounter(meter: Counter, value: number, attributes: T): RecordPendingMetric { + return (errorType?: string | undefined) => { + meter.add(value, { + ...attributes, + ...(errorType ? { [ATTR_ERROR_TYPE]: errorType } : {}), + }); + }; +} + +function prepareDurationHistogram( + meter: Histogram, + value: number, + attributes: T, +): RecordPendingMetric { + return (errorType?: string | undefined) => { + meter.record((Date.now() - value) / 1000, { + ...attributes, + ...(errorType ? { [ATTR_ERROR_TYPE]: errorType } : {}), + }); + }; +} + +const HISTOGRAM_BUCKET_BOUNDARIES = [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10]; +export class KafkaJsInstrumentation extends InstrumentationBase { + declare private _clientDuration: Histogram; + declare private _sentMessages: Counter; + declare private _consumedMessages: Counter; + declare private _processDuration: Histogram; + + constructor(config: KafkaJsInstrumentationConfig = {}) { + super(PACKAGE_NAME, SDK_VERSION, config); + } + + override _updateMetricInstruments() { + this._clientDuration = this.meter.createHistogram(METRIC_MESSAGING_CLIENT_OPERATION_DURATION, { + advice: { explicitBucketBoundaries: HISTOGRAM_BUCKET_BOUNDARIES }, + }); + this._sentMessages = this.meter.createCounter(METRIC_MESSAGING_CLIENT_SENT_MESSAGES); + this._consumedMessages = this.meter.createCounter(METRIC_MESSAGING_CLIENT_CONSUMED_MESSAGES); + this._processDuration = this.meter.createHistogram(METRIC_MESSAGING_PROCESS_DURATION, { + advice: { explicitBucketBoundaries: HISTOGRAM_BUCKET_BOUNDARIES }, + }); + } + + protected init() { + const unpatch = (moduleExports: any) => { + if (isWrapped(moduleExports?.Kafka?.prototype.producer)) { + this._unwrap(moduleExports.Kafka.prototype, 'producer'); + } + if (isWrapped(moduleExports?.Kafka?.prototype.consumer)) { + this._unwrap(moduleExports.Kafka.prototype, 'consumer'); + } + }; + + const module = new InstrumentationNodeModuleDefinition( + 'kafkajs', + ['>=0.3.0 <3'], + (moduleExports: any) => { + unpatch(moduleExports); + this._wrap(moduleExports?.Kafka?.prototype, 'producer', this._getProducerPatch()); + this._wrap(moduleExports?.Kafka?.prototype, 'consumer', this._getConsumerPatch()); + + return moduleExports; + }, + unpatch, + ); + return module; + } + + private _getConsumerPatch() { + const instrumentation = this; + return (original: Kafka['consumer']) => { + return function consumer(this: Kafka, ...args: Parameters) { + const newConsumer: Consumer = original.apply(this, args); + + if (isWrapped(newConsumer.run)) { + instrumentation._unwrap(newConsumer, 'run'); + } + + instrumentation._wrap(newConsumer, 'run', instrumentation._getConsumerRunPatch()); + + instrumentation._setKafkaEventListeners(newConsumer); + + return newConsumer; + }; + }; + } + + private _setKafkaEventListeners(kafkaObj: KafkaEventEmitter) { + if (kafkaObj[EVENT_LISTENERS_SET]) return; + + // The REQUEST Consumer event was added in kafkajs@1.5.0. + if (kafkaObj.events?.REQUEST) { + kafkaObj.on(kafkaObj.events.REQUEST, this._recordClientDurationMetric.bind(this)); + } + + kafkaObj[EVENT_LISTENERS_SET] = true; + } + + private _recordClientDurationMetric(event: Pick) { + const [address = '', port = '0'] = 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}`, + [ATTR_SERVER_ADDRESS]: address, + [ATTR_SERVER_PORT]: Number.parseInt(port, 10), + }); + } + + private _getProducerPatch() { + const instrumentation = this; + return (original: Kafka['producer']) => { + return function consumer(this: Kafka, ...args: Parameters) { + const newProducer: Producer = original.apply(this, args); + + if (isWrapped(newProducer.sendBatch)) { + instrumentation._unwrap(newProducer, 'sendBatch'); + } + instrumentation._wrap(newProducer, 'sendBatch', instrumentation._getSendBatchPatch()); + + if (isWrapped(newProducer.send)) { + instrumentation._unwrap(newProducer, 'send'); + } + instrumentation._wrap(newProducer, 'send', instrumentation._getSendPatch()); + + if (isWrapped(newProducer.transaction)) { + instrumentation._unwrap(newProducer, 'transaction'); + } + instrumentation._wrap(newProducer, 'transaction', instrumentation._getProducerTransactionPatch()); + + instrumentation._setKafkaEventListeners(newProducer); + + return newProducer; + }; + }; + } + + private _getConsumerRunPatch() { + const instrumentation = this; + return (original: Consumer['run']) => { + return function run(this: Consumer, ...args: Parameters): ReturnType { + const config = args[0]; + if (config?.eachMessage) { + if (isWrapped(config.eachMessage)) { + instrumentation._unwrap(config, 'eachMessage'); + } + instrumentation._wrap(config, 'eachMessage', instrumentation._getConsumerEachMessagePatch()); + } + if (config?.eachBatch) { + if (isWrapped(config.eachBatch)) { + instrumentation._unwrap(config, 'eachBatch'); + } + instrumentation._wrap(config, 'eachBatch', instrumentation._getConsumerEachBatchPatch()); + } + return original.call(this, config); + }; + }; + } + + private _getConsumerEachMessagePatch() { + const instrumentation = this; + return (original: ConsumerRunConfig['eachMessage']) => { + return function eachMessage(this: unknown, ...args: Parameters): Promise { + const payload = args[0]; + const propagatedContext: Context = propagation.extract( + ROOT_CONTEXT, + payload.message.headers, + bufferTextMapGetter, + ); + const span = instrumentation._startConsumerSpan({ + topic: payload.topic, + message: payload.message, + operationType: MESSAGING_OPERATION_TYPE_VALUE_PROCESS, + ctx: propagatedContext, + attributes: { + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.partition), + }, + }); + + const pendingMetrics: RecordPendingMetric[] = [ + prepareDurationHistogram(instrumentation._processDuration, Date.now(), { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'process', + [ATTR_MESSAGING_DESTINATION_NAME]: payload.topic, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.partition), + }), + prepareCounter(instrumentation._consumedMessages, 1, { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'process', + [ATTR_MESSAGING_DESTINATION_NAME]: payload.topic, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.partition), + }), + ]; + + const eachMessagePromise = context.with(trace.setSpan(propagatedContext, span), () => { + return original!.apply(this, args); + }); + return instrumentation._endSpansOnPromise([span], pendingMetrics, eachMessagePromise); + }; + }; + } + + private _getConsumerEachBatchPatch() { + return (original: ConsumerRunConfig['eachBatch']) => { + const instrumentation = this; + return function eachBatch(this: unknown, ...args: Parameters): Promise { + const payload = args[0]; + // https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#topic-with-multiple-consumers + const receivingSpan = instrumentation._startConsumerSpan({ + topic: payload.batch.topic, + message: undefined, + operationType: MESSAGING_OPERATION_TYPE_VALUE_RECEIVE, + ctx: ROOT_CONTEXT, + attributes: { + [ATTR_MESSAGING_BATCH_MESSAGE_COUNT]: payload.batch.messages.length, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.batch.partition), + }, + }); + return context.with(trace.setSpan(context.active(), receivingSpan), () => { + const startTime = Date.now(); + const spans: Span[] = []; + const pendingMetrics: RecordPendingMetric[] = [ + prepareCounter(instrumentation._consumedMessages, payload.batch.messages.length, { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'process', + [ATTR_MESSAGING_DESTINATION_NAME]: payload.batch.topic, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.batch.partition), + }), + ]; + payload.batch.messages.forEach((message: any) => { + const propagatedContext: Context = propagation.extract(ROOT_CONTEXT, message.headers, bufferTextMapGetter); + const spanContext = trace.getSpan(propagatedContext)?.spanContext(); + let origSpanLink: Link | undefined; + if (spanContext) { + origSpanLink = { + context: spanContext, + }; + } + spans.push( + instrumentation._startConsumerSpan({ + topic: payload.batch.topic, + message, + operationType: MESSAGING_OPERATION_TYPE_VALUE_PROCESS, + link: origSpanLink, + attributes: { + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.batch.partition), + }, + }), + ); + pendingMetrics.push( + prepareDurationHistogram(instrumentation._processDuration, startTime, { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'process', + [ATTR_MESSAGING_DESTINATION_NAME]: payload.batch.topic, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(payload.batch.partition), + }), + ); + }); + const batchMessagePromise: Promise = original!.apply(this, args); + spans.unshift(receivingSpan); + return instrumentation._endSpansOnPromise(spans, pendingMetrics, batchMessagePromise); + }); + }; + }; + } + + private _getProducerTransactionPatch() { + const instrumentation = this; + return (original: Producer['transaction']) => { + return function transaction( + this: Producer, + ...args: Parameters + ): ReturnType { + const transactionSpan = instrumentation.tracer.startSpan('transaction'); + + const transactionPromise = original.apply(this, args); + + transactionPromise + .then((transaction: Transaction) => { + const originalSend = transaction.send; + transaction.send = function send(this: Transaction, ...args) { + return context.with(trace.setSpan(context.active(), transactionSpan), () => { + const patched = instrumentation._getSendPatch()(originalSend); + return patched.apply(this, args).catch((err: any) => { + transactionSpan.setStatus({ + code: SpanStatusCode.ERROR, + message: err?.message, + }); + transactionSpan.recordException(err); + throw err; + }); + }); + }; + + const originalSendBatch = transaction.sendBatch; + transaction.sendBatch = function sendBatch(this: Transaction, ...args) { + return context.with(trace.setSpan(context.active(), transactionSpan), () => { + const patched = instrumentation._getSendBatchPatch()(originalSendBatch); + return patched.apply(this, args).catch((err: any) => { + transactionSpan.setStatus({ + code: SpanStatusCode.ERROR, + message: err?.message, + }); + transactionSpan.recordException(err); + throw err; + }); + }); + }; + + const originalCommit = transaction.commit; + transaction.commit = function commit(this: Transaction, ...args) { + const originCommitPromise = originalCommit.apply(this, args).then(() => { + transactionSpan.setStatus({ code: SpanStatusCode.OK }); + }); + return instrumentation._endSpansOnPromise([transactionSpan], [], originCommitPromise); + }; + + const originalAbort = transaction.abort; + transaction.abort = function abort(this: Transaction, ...args) { + const originAbortPromise = originalAbort.apply(this, args); + return instrumentation._endSpansOnPromise([transactionSpan], [], originAbortPromise); + }; + }) + .catch((err: any) => { + transactionSpan.setStatus({ + code: SpanStatusCode.ERROR, + message: err?.message, + }); + transactionSpan.recordException(err); + transactionSpan.end(); + }); + + return transactionPromise; + }; + }; + } + + private _getSendBatchPatch() { + const instrumentation = this; + return (original: Producer['sendBatch'] | Transaction['sendBatch']) => { + return function sendBatch( + this: Producer | Transaction, + ...args: Parameters + ): ReturnType { + const batch = args[0]; + const messages = batch.topicMessages || []; + + const spans: Span[] = []; + const pendingMetrics: RecordPendingMetric[] = []; + + messages.forEach((topicMessage: any) => { + topicMessage.messages.forEach((message: any) => { + spans.push(instrumentation._startProducerSpan(topicMessage.topic, message)); + pendingMetrics.push( + prepareCounter(instrumentation._sentMessages, 1, { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'send', + [ATTR_MESSAGING_DESTINATION_NAME]: topicMessage.topic, + ...(message.partition !== undefined + ? { + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(message.partition), + } + : {}), + }), + ); + }); + }); + const origSendResult: Promise = original.apply(this, args); + return instrumentation._endSpansOnPromise(spans, pendingMetrics, origSendResult); + }; + }; + } + + private _getSendPatch() { + const instrumentation = this; + return (original: Producer['send'] | Transaction['send']) => { + return function send( + this: Producer | Transaction, + ...args: Parameters + ): ReturnType { + const record = args[0]; + const spans: Span[] = record.messages.map((message: any) => { + return instrumentation._startProducerSpan(record.topic, message); + }); + + const pendingMetrics: RecordPendingMetric[] = record.messages.map((m: any) => + prepareCounter(instrumentation._sentMessages, 1, { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_OPERATION_NAME]: 'send', + [ATTR_MESSAGING_DESTINATION_NAME]: record.topic, + ...(m.partition !== undefined + ? { + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: String(m.partition), + } + : {}), + }), + ); + const origSendResult: Promise = original.apply(this, args); + return instrumentation._endSpansOnPromise(spans, pendingMetrics, origSendResult); + }; + }; + } + + private _endSpansOnPromise( + spans: Span[], + pendingMetrics: RecordPendingMetric[], + sendPromise: Promise, + ): Promise { + return Promise.resolve(sendPromise) + .then(result => { + pendingMetrics.forEach(m => m()); + return result; + }) + .catch(reason => { + let errorMessage: string | undefined; + let errorType: string = ERROR_TYPE_VALUE_OTHER; + if (typeof reason === 'string' || reason === undefined) { + errorMessage = reason; + } else if (typeof reason === 'object' && Object.prototype.hasOwnProperty.call(reason, 'message')) { + errorMessage = reason.message; + errorType = reason.constructor.name; + } + pendingMetrics.forEach(m => m(errorType)); + + spans.forEach(span => { + span.setAttribute(ATTR_ERROR_TYPE, errorType); + span.setStatus({ + code: SpanStatusCode.ERROR, + message: errorMessage, + }); + }); + + throw reason; + }) + .finally(() => { + spans.forEach(span => span.end()); + }); + } + + private _startConsumerSpan({ topic, message, operationType, ctx, link, attributes }: ConsumerSpanOptions) { + const operationName = + operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE + ? 'poll' // for batch processing spans + : operationType; // for individual message processing spans + + const span = this.tracer.startSpan( + `${operationName} ${topic}`, + { + kind: operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE ? SpanKind.CLIENT : SpanKind.CONSUMER, + attributes: { + ...attributes, + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_DESTINATION_NAME]: topic, + [ATTR_MESSAGING_OPERATION_TYPE]: operationType, + [ATTR_MESSAGING_OPERATION_NAME]: operationName, + [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: message?.key ? String(message.key) : undefined, + [ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE]: message?.key && message.value === null ? true : undefined, + [ATTR_MESSAGING_KAFKA_OFFSET]: message?.offset, + }, + links: link ? [link] : [], + }, + ctx, + ); + + const { consumerHook } = this.getConfig(); + if (consumerHook && message) { + safeExecuteInTheMiddle( + () => consumerHook(span, { topic, message }), + e => { + if (e) this._diag.error('consumerHook error', e); + }, + true, + ); + } + + return span; + } + + private _startProducerSpan(topic: string, message: Message) { + const span = this.tracer.startSpan(`send ${topic}`, { + kind: SpanKind.PRODUCER, + attributes: { + [ATTR_MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA, + [ATTR_MESSAGING_DESTINATION_NAME]: topic, + [ATTR_MESSAGING_KAFKA_MESSAGE_KEY]: message.key ? String(message.key) : undefined, + [ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE]: message.key && message.value === null ? true : undefined, + [ATTR_MESSAGING_DESTINATION_PARTITION_ID]: + message.partition !== undefined ? String(message.partition) : undefined, + [ATTR_MESSAGING_OPERATION_NAME]: 'send', + [ATTR_MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_TYPE_VALUE_SEND, + }, + }); + + message.headers = message.headers ?? {}; + propagation.inject(trace.setSpan(context.active(), span), message.headers); + + const { producerHook } = this.getConfig(); + if (producerHook) { + safeExecuteInTheMiddle( + () => producerHook(span, { topic, message }), + e => { + if (e) this._diag.error('producerHook error', e); + }, + true, + ); + } + + return span; + } +} diff --git a/packages/node/src/integrations/tracing/kafka/vendored/internal-types.ts b/packages/node/src/integrations/tracing/kafka/vendored/internal-types.ts new file mode 100644 index 000000000000..f42c63917923 --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/internal-types.ts @@ -0,0 +1,33 @@ +/* + * Copyright The OpenTelemetry Authors, Aspecto + * + * Licensed 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 + * + * https://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. + * + * NOTICE from the Sentry authors: + * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs + * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 + * - Some types vendored from kafkajs with simplifications + */ +/* eslint-disable */ + +import type { Consumer, Producer } from './kafkajs-types'; + +export const EVENT_LISTENERS_SET = Symbol('opentelemetry.instrumentation.kafkajs.eventListenersSet'); + +export interface ConsumerExtended extends Consumer { + [EVENT_LISTENERS_SET]?: boolean; // flag to identify if the event listeners for instrumentation have been set +} + +export interface ProducerExtended extends Producer { + [EVENT_LISTENERS_SET]?: boolean; // flag to identify if the event listeners for instrumentation have been set +} diff --git a/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts b/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts new file mode 100644 index 000000000000..583cee2d9cf5 --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts @@ -0,0 +1,128 @@ +/* + * Simplified types inlined from kafkajs/types/index.d.ts. + * Only includes members accessed by this instrumentation. + */ + +export interface InstrumentationEvent { + id: string; + type: string; + timestamp: number; + payload: T; +} + +export type RequestEvent = InstrumentationEvent<{ + apiKey: number; + apiName: string; + apiVersion: number; + broker: string; + clientId: string; + correlationId: number; + createdAt: number; + duration: number; + pendingDuration: number; + sentAt: number; + size: number; +}>; + +export type RemoveInstrumentationEventListener = () => void; + +export type ConsumerEvents = { + REQUEST: 'consumer.network.request'; + [key: string]: string; +}; + +export type ProducerEvents = { + REQUEST: 'producer.network.request'; + [key: string]: string; +}; + +type Sender = { + send(record: any): Promise; + sendBatch(batch: any): Promise; +}; + +export type Producer = Sender & { + connect(): Promise; + disconnect(): Promise; + isIdempotent(): boolean; + readonly events: ProducerEvents; + on(eventName: string, listener: (...args: any[]) => void): RemoveInstrumentationEventListener; + transaction(): Promise; + [key: string]: any; +}; + +export type Transaction = Sender & { + sendOffsets(offsets: any): Promise; + commit(): Promise; + abort(): Promise; + isActive(): boolean; +}; + +export type Consumer = { + connect(): Promise; + disconnect(): Promise; + subscribe(subscription: any): Promise; + run(config?: any): Promise; + readonly events: ConsumerEvents; + on(eventName: string, listener: (...args: any[]) => void): RemoveInstrumentationEventListener; + [key: string]: any; +}; + +export declare class Kafka { + consumer(config: any): Consumer; + producer(config?: any): Producer; + [key: string]: any; +} + +export interface Message { + key?: Buffer | string | null; + value: Buffer | string | null; + partition?: number; + headers?: Record; + timestamp?: string; +} + +export type KafkaMessage = { [key: string]: any } & Message; + +export type RecordMetadata = { + topicName: string; + partition: number; + errorCode: number; + offset?: string; + timestamp?: string; + baseOffset?: string; + logAppendTime?: string; + logStartOffset?: string; +}; + +export interface EachMessagePayload { + topic: string; + partition: number; + message: KafkaMessage; + heartbeat(): Promise; + pause(): () => void; +} + +export interface EachBatchPayload { + batch: any; + resolveOffset(offset: string): void; + heartbeat(): Promise; + pause(): () => void; + commitOffsetsIfNecessary(offsets?: any): Promise; + uncommittedOffsets(): any; + isRunning(): boolean; + isStale(): boolean; +} + +export type EachMessageHandler = (payload: EachMessagePayload) => Promise; +export type EachBatchHandler = (payload: EachBatchPayload) => Promise; + +export type ConsumerRunConfig = { + autoCommit?: boolean; + autoCommitInterval?: number | null; + autoCommitThreshold?: number | null; + eachBatchAutoResolve?: boolean; + partitionsConsumedConcurrently?: number; + eachBatch?: EachBatchHandler; + eachMessage?: EachMessageHandler; +}; diff --git a/packages/node/src/integrations/tracing/kafka/vendored/propagator.ts b/packages/node/src/integrations/tracing/kafka/vendored/propagator.ts new file mode 100644 index 000000000000..bf5226035c4c --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/propagator.ts @@ -0,0 +1,49 @@ +/* + * Copyright The OpenTelemetry Authors, Aspecto + * + * Licensed 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 + * + * https://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. + * + * NOTICE from the Sentry authors: + * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs + * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 + */ +/* eslint-disable */ + +import type { TextMapGetter } from '@opentelemetry/api'; + +/* +same as open telemetry's `defaultTextMapGetter`, +but also handle case where header is buffer, +adding toString() to make sure string is returned +*/ +export const bufferTextMapGetter: TextMapGetter = { + get(carrier, key) { + if (!carrier) { + return undefined; + } + + const keys = Object.keys(carrier); + + for (const carrierKey of keys) { + if (carrierKey === key || carrierKey.toLowerCase() === key) { + return carrier[carrierKey]?.toString(); + } + } + + return undefined; + }, + + keys(carrier) { + return carrier ? Object.keys(carrier) : []; + }, +}; diff --git a/packages/node/src/integrations/tracing/kafka/vendored/semconv.ts b/packages/node/src/integrations/tracing/kafka/vendored/semconv.ts new file mode 100644 index 000000000000..ff20740b12ea --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/semconv.ts @@ -0,0 +1,173 @@ +/* + * Copyright The OpenTelemetry Authors, Aspecto + * + * Licensed 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 + * + * https://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. + * + * NOTICE from the Sentry authors: + * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs + * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 + */ +/* eslint-disable */ + +/* + * This file contains a copy of unstable semantic convention definitions + * used by this package. + * @see https://github.com/open-telemetry/opentelemetry-js/tree/main/semantic-conventions#unstable-semconv + */ + +/** + * The number of messages sent, received, or processed in the scope of the batching operation. + * + * @example 0 + * @example 1 + * @example 2 + * + * @note Instrumentations **SHOULD NOT** set `messaging.batch.message_count` on spans that operate with a single message. When a messaging client library supports both batch and single-message API for the same operation, instrumentations **SHOULD** use `messaging.batch.message_count` for batching APIs and **SHOULD NOT** use it for single-message APIs. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT = 'messaging.batch.message_count' as const; + +/** + * The message destination name + * + * @example MyQueue + * @example MyTopic + * + * @note Destination name **SHOULD** uniquely identify a specific queue, topic or other entity within the broker. If + * the broker doesn't have such notion, the destination name **SHOULD** uniquely identify the broker. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_DESTINATION_NAME = 'messaging.destination.name' as const; + +/** + * The identifier of the partition messages are sent to or received from, unique within the `messaging.destination.name`. + * + * @example "1" + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_DESTINATION_PARTITION_ID = 'messaging.destination.partition.id' as const; + +/** + * Message keys in Kafka are used for grouping alike messages to ensure they're processed on the same partition. They differ from `messaging.message.id` in that they're not unique. If the key is `null`, the attribute **MUST NOT** be set. + * + * @example "myKey" + * + * @note If the key type is not string, it's string representation has to be supplied for the attribute. If the key has no unambiguous, canonical string form, don't include its value. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_KAFKA_MESSAGE_KEY = 'messaging.kafka.message.key' as const; + +/** + * A boolean that is true if the message is a tombstone. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_KAFKA_MESSAGE_TOMBSTONE = 'messaging.kafka.message.tombstone' as const; + +/** + * The offset of a record in the corresponding Kafka partition. + * + * @example 42 + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_KAFKA_OFFSET = 'messaging.kafka.offset' as const; + +/** + * The system-specific name of the messaging operation. + * + * @example ack + * @example nack + * @example send + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_OPERATION_NAME = 'messaging.operation.name' as const; + +/** + * A string identifying the type of the messaging operation. + * + * @note If a custom value is used, it **MUST** be of low cardinality. + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_OPERATION_TYPE = 'messaging.operation.type' as const; + +/** + * The messaging system as identified by the client instrumentation. + * + * @note The actual messaging system may differ from the one known by the client. For example, when using Kafka client libraries to communicate with Azure Event Hubs, the `messaging.system` is set to `kafka` based on the instrumentation's best knowledge. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_SYSTEM = 'messaging.system' as const; + +/** + * Enum value "process" for attribute {@link ATTR_MESSAGING_OPERATION_TYPE}. + */ +export const MESSAGING_OPERATION_TYPE_VALUE_PROCESS = 'process' as const; + +/** + * Enum value "receive" for attribute {@link ATTR_MESSAGING_OPERATION_TYPE}. + */ +export const MESSAGING_OPERATION_TYPE_VALUE_RECEIVE = 'receive' as const; + +/** + * Enum value "send" for attribute {@link ATTR_MESSAGING_OPERATION_TYPE}. + */ +export const MESSAGING_OPERATION_TYPE_VALUE_SEND = 'send' as const; + +/** + * Enum value "kafka" for attribute {@link ATTR_MESSAGING_SYSTEM}. + */ +export const MESSAGING_SYSTEM_VALUE_KAFKA = 'kafka' as const; + +/** + * Number of messages that were delivered to the application. + * + * @note Records the number of messages pulled from the broker or number of messages dispatched to the application in push-based scenarios. + * The metric **SHOULD** be reported once per message delivery. For example, if receiving and processing operations are both instrumented for a single message delivery, this counter is incremented when the message is received and not reported when it is processed. + * + * @experimental This metric is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const METRIC_MESSAGING_CLIENT_CONSUMED_MESSAGES = 'messaging.client.consumed.messages' as const; + +/** + * Duration of messaging operation initiated by a producer or consumer client. + * + * @note This metric **SHOULD NOT** be used to report processing duration - processing duration is reported in `messaging.process.duration` metric. + * + * @experimental This metric is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const METRIC_MESSAGING_CLIENT_OPERATION_DURATION = 'messaging.client.operation.duration' as const; + +/** + * Number of messages producer attempted to send to the broker. + * + * @note This metric **MUST NOT** count messages that were created but haven't yet been sent. + * + * @experimental This metric is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const METRIC_MESSAGING_CLIENT_SENT_MESSAGES = 'messaging.client.sent.messages' as const; + +/** + * Duration of processing operation. + * + * @note This metric **MUST** be reported for operations with `messaging.operation.type` that matches `process`. + * + * @experimental This metric is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const METRIC_MESSAGING_PROCESS_DURATION = 'messaging.process.duration' as const; diff --git a/packages/node/src/integrations/tracing/kafka/vendored/types.ts b/packages/node/src/integrations/tracing/kafka/vendored/types.ts new file mode 100644 index 000000000000..716e986ba61c --- /dev/null +++ b/packages/node/src/integrations/tracing/kafka/vendored/types.ts @@ -0,0 +1,52 @@ +/* + * Copyright The OpenTelemetry Authors, Aspecto + * + * Licensed 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 + * + * https://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. + * + * NOTICE from the Sentry authors: + * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs + * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 + */ +/* eslint-disable */ + +import { Span } from '@opentelemetry/api'; +import { InstrumentationConfig } from '@opentelemetry/instrumentation'; + +export interface KafkajsMessage { + key?: Buffer | string | null; + value: Buffer | string | null; + partition?: number; + headers?: Record; + timestamp?: string; +} + +export interface MessageInfo { + topic: string; + message: T; +} + +export interface KafkaProducerCustomAttributeFunction { + (span: Span, info: MessageInfo): void; +} + +export interface KafkaConsumerCustomAttributeFunction { + (span: Span, info: MessageInfo): void; +} + +export interface KafkaJsInstrumentationConfig extends InstrumentationConfig { + /** hook for adding custom attributes before producer message is sent */ + producerHook?: KafkaProducerCustomAttributeFunction; + + /** hook for adding custom attributes before consumer message is processed */ + consumerHook?: KafkaConsumerCustomAttributeFunction; +} diff --git a/yarn.lock b/yarn.lock index 7337d680cc8f..6edd6e1ee33e 100644 --- a/yarn.lock +++ b/yarn.lock @@ -6102,14 +6102,6 @@ "@opentelemetry/semantic-conventions" "^1.29.0" forwarded-parse "2.1.2" -"@opentelemetry/instrumentation-kafkajs@0.23.0": - version "0.23.0" - resolved "https://registry.yarnpkg.com/@opentelemetry/instrumentation-kafkajs/-/instrumentation-kafkajs-0.23.0.tgz#6b7d449d88d674ddc295a0d0cf2156f0f7d5889f" - integrity sha512-4K+nVo+zI+aDz0Z85SObwbdixIbzS9moIuKJaYsdlzcHYnKOPtB7ya8r8Ezivy/GVIBHiKJVq4tv+BEkgOMLaQ== - dependencies: - "@opentelemetry/instrumentation" "^0.214.0" - "@opentelemetry/semantic-conventions" "^1.30.0" - "@opentelemetry/instrumentation-koa@0.62.0": version "0.62.0" resolved "https://registry.yarnpkg.com/@opentelemetry/instrumentation-koa/-/instrumentation-koa-0.62.0.tgz#65fdf96c1b1ffb382167cd3b7a244631afd0cc1f" From 65e79b1bfb8df59b1878a3070f1651425984f86a Mon Sep 17 00:00:00 2001 From: Nicolas Hrubec Date: Tue, 19 May 2026 14:59:05 +0200 Subject: [PATCH 2/3] ref: Add TS strictness comment to header Co-Authored-By: Claude Opus 4.6 (1M context) --- .../src/integrations/tracing/kafka/vendored/instrumentation.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts b/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts index 453a40d3f3b1..53c90547f551 100644 --- a/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts +++ b/packages/node/src/integrations/tracing/kafka/vendored/instrumentation.ts @@ -17,6 +17,7 @@ * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 * - Some types vendored from kafkajs with simplifications + * - Minor TypeScript strictness adjustments for this repository's compiler settings */ /* eslint-disable */ From bd313802f3285415701456e803f4fec2e1f8ed76 Mon Sep 17 00:00:00 2001 From: Nicolas Hrubec Date: Tue, 19 May 2026 15:00:30 +0200 Subject: [PATCH 3/3] ref: Fix unused type parameter lint warning Co-Authored-By: Claude Opus 4.6 (1M context) --- .../src/integrations/tracing/kafka/vendored/kafkajs-types.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts b/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts index 583cee2d9cf5..1ff955322071 100644 --- a/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts +++ b/packages/node/src/integrations/tracing/kafka/vendored/kafkajs-types.ts @@ -24,7 +24,7 @@ export type RequestEvent = InstrumentationEvent<{ size: number; }>; -export type RemoveInstrumentationEventListener = () => void; +export type RemoveInstrumentationEventListener<_T> = () => void; export type ConsumerEvents = { REQUEST: 'consumer.network.request';