Instrument with OpenTelemetry

Instrumentation is built in and lazy-loaded. With @opentelemetry/api absent it costs nothing; install it and spans and metrics start flowing with no code change.

Turn it on

pnpm add @opentelemetry/api @opentelemetry/sdk-node @opentelemetry/auto-instrumentations-node

Set the SDK up before anything else imports the library:

// tracing.ts — import this first
import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { NodeSDK } from "@opentelemetry/sdk-node";

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({ url: "http://localhost:4318/v1/traces" }),
  instrumentations: [getNodeAutoInstrumentations()],
});

sdk.start();

// index.ts
import "./tracing.js";
import { TypedAmqpClient } from "@amqp-contract/client";

There is no telemetry: true flag. Detection is the presence of @opentelemetry/api.

Know what you get

Spans follow the OpenTelemetry messaging conventions.

Publish spans are named {exchange} publish and carry messaging.system (rabbitmq), messaging.destination.name, messaging.destination.kind (exchange), messaging.operation (publish), messaging.rabbitmq.destination.routing_key, messaging.message.id and messaging.message.body.size.

Consume spans are named {queue} process and carry messaging.system, messaging.destination.name, messaging.destination.kind (queue), messaging.operation (process), messaging.rabbitmq.message.delivery_tag and amqp.consumer.name.

On failure a span records the exception, sets status ERROR, and sets error.type to the error class — so RetryableError and NonRetryableError are distinguishable in your backend, which makes "how much of our error rate is permanent?" a query rather than a guess.

Metrics:

Metric	Type	Unit
amqp.client.messages.published	Counter	{message}
amqp.worker.messages.consumed	Counter	{message}
amqp.client.publish.duration	Histogram	ms
amqp.client.rpc.duration	Histogram	ms
amqp.worker.process.duration	Histogram	ms
amqp.client.rpc.late_reply	Counter	{message}

All carry messaging.system, messaging.destination.name and a success boolean (the late-reply counter carries a reason instead). amqp.client.rpc.duration is the full call() round trip — request publish to validated reply — and is kept apart from the publish histogram so a slow handler does not read as a slow broker; RPC calls are not counted in amqp.client.messages.published.

Connect traces across the broker

This is automatic. On publish, the active trace context is injected into the message headers through the propagator your SDK registered (traceparent / tracestate with the default W3C setup); publish() and call() run that injection with their producer span active, so the header carries the producer span. On consume, each delivery runs inside the context extracted from its headers, so the consume span is parented on the producer span and one trace spans producer, broker hop and consumer. The handler (with createContext and the middleware) runs with the consume span active, so a message it publishes continues the same trace.

Without an SDK (or without @opentelemetry/api) nothing is injected and headers are left untouched. A misbehaving propagator degrades to "no propagation" — it never fails a publish or a delivery.

To propagate something other than trace context (a tenant id, a correlation id), stamp it with a publish interceptor and read it in worker middleware — see add middleware.

Sample in production

Tracing every message is rarely affordable:

import { ParentBasedSampler, TraceIdRatioBasedSampler } from "@opentelemetry/sdk-trace-base";

const sdk = new NodeSDK({
  sampler: new ParentBasedSampler({ root: new TraceIdRatioBasedSampler(0.1) }),
  // …
});

ParentBasedSampler keeps a trace whole: once an upstream service samples a trace, this one honours that decision instead of re-rolling and truncating it.

Metrics are unaffected by sampling, so rates and latencies stay accurate at any sampling ratio.

Add your own attributes

The library's spans are not extension points. To attach business context, start your own span inside the handler:

import { trace } from "@opentelemetry/api";

const tracer = trace.getTracer("orders");

processOrder: ({ input: { payload } }) =>
  fromPromise(
    tracer.startActiveSpan("process-order", async (span) => {
      span.setAttribute("order.id", payload.orderId);
      try {
        await handleOrder(payload);
      } finally {
        span.end();
      }
    }),
    qualifyRetryable("processing failed"),
  ).map(() => undefined),

It nests under the consume span automatically: the worker runs the handler with the consume span active.

Supply a custom telemetry provider

To route telemetry somewhere other than OpenTelemetry, or to disable it explicitly, pass a TelemetryProvider:

import type { TelemetryProvider } from "@amqp-contract/core";

const client = await TypedAmqpClient.create({
  contract,
  urls: ["amqp://localhost"],
  telemetry: myProvider,
}).getOrThrow();

Omitting it uses the default provider, which attempts to load OpenTelemetry and no-ops if it is absent. getRpcCallLatencyHistogram is optional on a custom provider; leave it out and RPC durations are not recorded.

Watch the right signals

amqp.worker.process.duration rising while amqp.worker.messages.consumed stays flat means handlers are slowing down — usually a downstream dependency.

Consumed falling below published means the queue is growing. Compare against broker-side queue depth, which is not something this library reports.

success=false on amqp.worker.messages.consumed is the most useful single chart. The metric carries no error class; the failed consume spans do, as error.type, so split by that in your tracing backend: a rise in NonRetryableError means bad data arriving, a rise in RetryableError means infrastructure trouble, and MessageValidationError means a publisher is sending payloads the contract rejects.

Where next

• Add logging — per-message detail traces do not carry.
• Add middleware — propagating other context.
• Tune performance — acting on what the metrics show.