feat(otel-nestjs-instrumentation): add RPC messaging spans and metrics

libs/otel-nestjs-instrumentation/src/internal/index.ts

@@ -2,5 +2,7 @@ export * from './emitter-symbol';
 export * from './get-transaction-name';
 export * from './internal-context';
 export * from './otel-instrumentation';
+export * from './record-messaging-process-duration';
+export * from './resolve-rpc-messaging-metadata';
 export * from './tracer-name';
 export * from './transaction-symbol';

libs/otel-nestjs-instrumentation/src/internal/otel-instrumentation.ts

@@ -1,5 +1,9 @@
 import { ExecutionContext } from '@nestjs/common';
 import * as otel from '@opentelemetry/api';
+import {
+	resolveRpcMessagingMetadata,
+	toMessagingSpanAttributes,
+} from './resolve-rpc-messaging-metadata';
 import { tracerName } from './tracer-name';
 
 /**
@@ -35,11 +39,11 @@
 	 *   (e.g., 'rpc' from interceptor fallback)
 	 * @returns Trace ID or undefined if tracer is unavailable
 	 */
 	create(
 		transactionName: string,
 		context: ExecutionContext,
 		effectiveType?: 'http' | 'rpc',
-	) {
+	): string | undefined {
 		// Create a new span since none exists
 		const tracer = otel.trace.getTracer(tracerName);
 		if (!tracer) return undefined;
@@ -63,8 +67,14 @@
 		// Determine span kind and attributes based on effective type
 		let spanKind = otel.SpanKind.INTERNAL; // default for unknown contexts
 		const attributes: Record<string, string> = {};
+		let messagingMetadata: ReturnType<typeof resolveRpcMessagingMetadata>;
 
-		effectiveType ??= context.getType() as 'http' | 'rpc';
+		if (!effectiveType) {
+			const contextType = context.getType();
+			if (contextType === 'http' || contextType === 'rpc') {
+				effectiveType = contextType;
+			}
+		}
 
 		if (effectiveType === 'http') {
 			spanKind = otel.SpanKind.SERVER;
@@ -83,7 +93,19 @@
 				// Ignore request extraction errors
 			}
 		} else if (effectiveType === 'rpc') {
-			spanKind = otel.SpanKind.SERVER;
+			messagingMetadata = resolveRpcMessagingMetadata(context);
+			if (messagingMetadata) {
+				spanKind = messagingMetadata.spanKind;
+				Object.assign(attributes, toMessagingSpanAttributes(messagingMetadata));
+				if (Object.keys(messagingMetadata.propagationCarrier).length > 0) {
+					spanContext = otel.propagation.extract(
+						spanContext,
+						messagingMetadata.propagationCarrier,
+					);
+				}
+			} else {
+				spanKind = otel.SpanKind.SERVER;
+			}
 			try {
 				attributes['rpc.method'] = context.getHandler()?.name ?? 'Call';
 			} catch {
@@ -101,17 +123,32 @@
 			// Ignore NestJS context extraction errors
 		}
 
+		const spanName =
+			messagingMetadata?.recordMetric === true
+				? `process ${messagingMetadata.operationName}`
+				: transactionName;
+
 		const span = tracer.startSpan(
-			transactionName,
+			spanName,
 			{
 				kind: spanKind,
 				attributes,
 			},
 			spanContext,
 		);
 
-		const spanContextData = span.spanContext();
-		const traceId = spanContextData?.traceId;
+		const traceId = span.spanContext()?.traceId;
+		if (!traceId) {
+			span.end();
+			return undefined;
+		}
+
+		const activeContext = otel.trace.setSpan(spanContext, span);
+		(
+			otel.context as typeof otel.context & {
+				enterWith: (ctx: otel.Context) => void;
+			}
+		).enterWith(activeContext);
 
 		return traceId;
 	},

libs/otel-nestjs-instrumentation/src/internal/record-messaging-process-duration.ts

@@ -0,0 +1,35 @@
+import { metrics, type Histogram } from '@opentelemetry/api';
+import { tracerName } from './tracer-name';
+import {
+	type RpcMessagingMetadata,
+	toMessagingSpanAttributes,
+} from './resolve-rpc-messaging-metadata';
+
+let processDurationHistogram: Histogram | undefined;
+
+function getProcessDurationHistogram(): Histogram {
+	processDurationHistogram ??= metrics
+		.getMeter(tracerName)
+		.createHistogram('messaging.process.duration', {
+			description: 'Measures the duration of inbound messaging operations.',
+			unit: 'ms',
+		});
+	return processDurationHistogram;
+}
+
+/** @internal Test-only reset for module-level meter instrument cache. */
+export function __resetMessagingProcessDurationForTests(): void {
+	processDurationHistogram = undefined;
+}
+
+export function recordMessagingProcessDuration(
+	durationMs: number,
+	metadata: RpcMessagingMetadata,
+): void {
+	if (!metadata.recordMetric) return;
+
+	getProcessDurationHistogram().record(
+		durationMs,
+		toMessagingSpanAttributes(metadata),
+	);
+}

libs/otel-nestjs-instrumentation/src/internal/resolve-rpc-messaging-metadata.ts

@@ -0,0 +1,173 @@
+import { ExecutionContext } from '@nestjs/common';
+import * as otel from '@opentelemetry/api';
+import { getTransactionName } from './get-transaction-name';
+
+type SqsMessageLike = {
+	MessageId?: string;
+	MessageAttributes?: Record<string, { StringValue?: string }>;
+};
+
+type KafkaMessageLike = {
+	headers?: Record<string, Buffer | string | undefined>;
+	offset?: string | number;
+};
+
+type RpcContextLike = {
+	getTopic?: () => string;
+	getPattern?: () => string;
+	getMessage?: () => unknown;
+	getChannelRef?: () => { fields?: { routingKey?: string } };
+};
+
+export type RpcMessagingMetadata = {
+	system: string;
+	destination: string;
+	operationName: string;
+	messageId?: string;
+	propagationCarrier: Record<string, string>;
+	spanKind: otel.SpanKind;
+	recordMetric: boolean;
+};
+
+export function toMessagingSpanAttributes(
+	metadata: RpcMessagingMetadata,
+): Record<string, string> {
+	const attributes: Record<string, string> = {
+		'messaging.system': metadata.system,
+		'messaging.destination.name': metadata.destination,
+		'messaging.operation.type': 'process',
+		'messaging.operation.name': metadata.operationName,
+	};
+	if (metadata.messageId) {
+		attributes['messaging.message.id'] = metadata.messageId;
+	}
+	return attributes;
+}
+
+function buildConsumerMetadata(
+	system: string,
+	destination: string,
+	operationName: string,
+	extras?: Pick<RpcMessagingMetadata, 'messageId' | 'propagationCarrier'>,
+): RpcMessagingMetadata {
+	return {
+		system,
+		destination,
+		operationName,
+		messageId: extras?.messageId,
+		propagationCarrier: extras?.propagationCarrier ?? {},
+		spanKind: otel.SpanKind.CONSUMER,
+		recordMetric: true,
+	};
+}
+
+function normalizeCarrier(
+	entries: Record<string, unknown>,
+	resolveValue: (value: unknown) => string | undefined,
+): Record<string, string> {
+	const carrier: Record<string, string> = {};
+	for (const [key, value] of Object.entries(entries)) {
+		const normalized = resolveValue(value);
+		if (normalized) {
+			carrier[key.toLowerCase()] = normalized;
+		}
+	}
+	return carrier;
+}
+
+function extractSqsCarrier(message: SqsMessageLike): Record<string, string> {
+	return normalizeCarrier(message.MessageAttributes ?? {}, (value) => {
+		if (
+			typeof value === 'object' &&
+			value !== null &&
+			'StringValue' in value &&
+			typeof (value as { StringValue?: string }).StringValue === 'string'
+		) {
+			return (value as { StringValue: string }).StringValue;
+		}
+		return undefined;
+	});
+}
+
+function extractKafkaCarrier(
+	message: KafkaMessageLike | undefined,
+): Record<string, string> {
+	return normalizeCarrier(message?.headers ?? {}, (value) => {
+		if (value === undefined) return undefined;
+		return Buffer.isBuffer(value) ? value.toString('utf8') : String(value);
+	});
+}
+
+function isSqsMessage(message: unknown): message is SqsMessageLike {
+	return (
+		typeof message === 'object' &&
+		message !== null &&
+		'MessageId' in message &&
+		typeof (message as SqsMessageLike).MessageId === 'string'
+	);
+}
+
+function buildGenericRpcMetadata(operationName: string): RpcMessagingMetadata {
+	return {
+		system: 'nestjs',
+		destination: operationName,
+		operationName,
+		propagationCarrier: {},
+		spanKind: otel.SpanKind.SERVER,
+		recordMetric: false,
+	};
+}
+
+/**
+ * Resolves messaging semantic attributes for NestJS RPC/microservice handlers.
+ *
+ * Works across custom and built-in transports (SQS, Kafka, RabbitMQ, etc.)
+ * by duck-typing `context.switchToRpc().getContext()`.
+ */
+export function resolveRpcMessagingMetadata(
+	context: ExecutionContext,
+): RpcMessagingMetadata | undefined {
+	if (context.getType() !== 'rpc') return undefined;
+
+	const operationName = getTransactionName(context);
+	let rpcContext: RpcContextLike;
+	try {
+		rpcContext = context.switchToRpc().getContext();
+	} catch {
+		return buildGenericRpcMetadata(operationName);
+	}
+
+	if (typeof rpcContext.getTopic === 'function') {
+		const message = rpcContext.getMessage?.() as KafkaMessageLike | undefined;
+		return buildConsumerMetadata(
+			'kafka',
+			rpcContext.getTopic(),
+			operationName,
+			{
+				messageId:
+					message?.offset === undefined ? undefined : String(message.offset),
+				propagationCarrier: extractKafkaCarrier(message),
+			},
+		);
+	}
+
+	const message = rpcContext.getMessage?.();
+	if (isSqsMessage(message)) {
+		return buildConsumerMetadata('aws_sqs', operationName, operationName, {
+			messageId: message.MessageId,
+			propagationCarrier: extractSqsCarrier(message),
+		});
+	}
+
+	if (typeof rpcContext.getPattern === 'function') {
+		const pattern = rpcContext.getPattern();
+		const routingKey = rpcContext.getChannelRef?.()?.fields?.routingKey;
+		return buildConsumerMetadata(
+			'rabbitmq',
+			routingKey ?? pattern,
+			operationName,
+		);
+	}
+
+	return buildGenericRpcMetadata(operationName);
+}

libs/otel-nestjs-instrumentation/src/otel.interceptor.ts

@@ -10,11 +10,15 @@ import {
 	emitterSymbol,
 	InternalContext,
 	otelInstrumentation,
+	resolveRpcMessagingMetadata,
+	recordMessagingProcessDuration,
 } from './internal';
 import EventEmitter from 'events';
 import otel, { Span } from '@opentelemetry/api';
 import { startOtelInstrumentationIfAbsent } from './start-otel-instrumentation-if-absent';
 
+const NANOSECONDS_PER_MILLISECOND = 1_000_000;
+
 /**
  * NestJS interceptor that manages OpenTelemetry span lifecycle.
  *
@@ -106,29 +110,56 @@ export class OtelInterceptor implements NestInterceptor {
 	 * @returns Observable that completes when the request is finished
 	 */
 	intercept(context: ExecutionContext, next: CallHandler) {
-		// If no guard ran (e.g. gRPC/microservice), force RPC since HTTP spans
-		// would have been started by the guard already.
-		startOtelInstrumentationIfAbsent(
-			context,
-			this.context,
-			this.emitter,
-			'rpc',
-		);
+		const isRpc = context.getType() === 'rpc';
+		const messagingMetadata = isRpc
+			? resolveRpcMessagingMetadata(context)
+			: undefined;
+		const startedAt =
+			messagingMetadata?.recordMetric === true
+				? process.hrtime.bigint()
+				: undefined;
+
+		if (isRpc) {
+			startOtelInstrumentationIfAbsent(
+				context,
+				this.context,
+				this.emitter,
+				'rpc',
+			);
+		} else {
+			startOtelInstrumentationIfAbsent(context, this.context, this.emitter);
+		}
+
 		const span = otel.trace.getActiveSpan();
 		if (!span) return next.handle();
 		const traceId = span.spanContext().traceId;
 
 		return next.handle().pipe(
 			tap({
-				next: () => this.finishSpan(traceId, span),
+				next: () => {
+					this.recordMessagingMetrics(startedAt, messagingMetadata);
+					this.finishSpan(traceId, span);
+				},
 				error: (error) => {
+					this.recordMessagingMetrics(startedAt, messagingMetadata);
 					this.recordError(error);
 					this.finishSpan(traceId, span);
 				},
 			}),
 		);
 	}
 
+	private recordMessagingMetrics(
+		startedAt: bigint | undefined,
+		messagingMetadata: ReturnType<typeof resolveRpcMessagingMetadata>,
+	): void {
+		if (startedAt === undefined || !messagingMetadata?.recordMetric) return;
+
+		const durationMs =
+			Number(process.hrtime.bigint() - startedAt) / NANOSECONDS_PER_MILLISECOND;
+		recordMessagingProcessDuration(durationMs, messagingMetadata);
+	}
+
 	/**
 	 * Complete the span and emit the appropriate events.
 	 *