Observability

Overview

A bus that “feels fine” in dev frequently hides issues in production: a slow downstream dragging consumer latency, a producer that’s silently retrying, a queue depth slowly climbing because of one bad partition. Observability turns those into signals you can alert on. ServiceConnect leans on the standard four pillars — logs, metrics, traces, health — and the framework either ships the integration or stays out of the way.

The problem

By default, a handler is a black box: it ran, it acked, it returned. When something goes wrong — latency spikes, retries climb, a single message kicks off a chain of side effects across three services — you need to see across the whole flow, not just one node. The framework can’t choose your telemetry stack, but it can give you the surfaces to plug one in without rewriting the bus.

The solution

Logs

The Logger interface from @serviceconnect/core is intentionally small: it requires six methods — trace, debug, info, warn, error, fatal — each taking a message string and an optional structured-fields object (only child is optional). The valid levels are exported as the LogLevel union ('trace' | 'debug' | 'info' | 'warn' | 'error' | 'fatal') for callers that want to thread the level through their own config without re-typing it. Use the shipped consoleLogger(level: LogLevel) for stdout, or write a thin adapter for Pino, Bunyan, or whatever the rest of the service uses:

import type { Logger } from '@serviceconnect/core';
import pino from 'pino';

const root = pino();
const pinoLogger: Logger = {
  trace: (msg, fields) => root.trace(fields, msg),
  debug: (msg, fields) => root.debug(fields, msg),
  info:  (msg, fields) => root.info(fields, msg),
  warn:  (msg, fields) => root.warn(fields, msg),
  error: (msg, fields) => root.error(fields, msg),
  fatal: (msg, fields) => root.fatal(fields, msg),
};

Pass it via BusOptions.logger. Inside a handler, ctx.logger is a child logger pre-populated with messageType, messageId, and correlationId, so every line a handler emits is automatically traceable back to the originating message without manually copying ids around.

Metrics

@serviceconnect/telemetry emits counters and histograms through the OpenTelemetry metrics SDK, under the ServiceConnect.Bus instrumentation scope and using the OTel messaging.* semantic conventions (so dashboards built against any standard messaging system work without translation). There are four instruments: a published-messages counter (messaging.client.published.messages, incremented on a successful publish/send), a consumed-messages counter (messaging.client.consumed.messages, tagged by messaging.outcome — success, error, or retry), a publish-duration histogram (messaging.publish.duration, recorded for every publish/send and tagged with error.type when it throws), and a process-duration histogram (messaging.process.duration). A failed consume is therefore visible as a metric — the consumed-messages counter carries messaging.outcome=error — not only via the span’s ERROR status. The instrument names match the .NET stack’s meter exactly, so a mixed Node/.NET deployment aggregates onto one series. Pass a custom Meter via TelemetryOptions.meter to override the default global meter.

Traces

telemetryProducer(producer) wraps the transport producer so each publish or send opens a PRODUCER span and injects W3C trace context into the outgoing headers. telemetryConsumeWrapper() extracts that context on the consumer side and opens a CONSUMER span around the handler. Producer spans are named <destination> <operation> — <type> publish for publish and <endpoint> send for send/sendBytes — and consumer spans are named <messageType> process, per OTel messaging conventions.

For spans to flow correctly through await, set a global propagator and a context manager once at boot. Without AsyncHooksContextManager, the active span context is lost on the first await and every consume span becomes a root — the trace tree falls apart.

Health checks

@serviceconnect/healthchecks ships three probes:

producerConnectivity(bus) — verifies the transport producer side is connected.
consumerConnectivity(bus) — verifies the consumer channel is established.
consumerBusy(bus, { graceMs }) — reports degraded when no message has been processed within the grace window, healthy otherwise.

Map them to your platform’s liveness/readiness endpoints — see Hosting for the wiring shapes.

End-to-end wiring

The full setup — propagator, context manager, tracer provider, producer wrap, consume wrapper, logger — is one block at boot:

import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
import { context, propagation } from '@opentelemetry/api';
import { W3CTraceContextPropagator } from '@opentelemetry/core';
import { createBus, consoleLogger } from '@serviceconnect/core';
import { createRabbitMQTransport } from '@serviceconnect/rabbitmq';
import { telemetryProducer, telemetryConsumeWrapper } from '@serviceconnect/telemetry';

context.setGlobalContextManager(new AsyncHooksContextManager().enable());
propagation.setGlobalPropagator(new W3CTraceContextPropagator());

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

const transport = createRabbitMQTransport({ url: 'amqp://localhost' });

const bus = createBus({
  queue: { name: 'orders' },
  transport: {
    producer: telemetryProducer(transport.producer),
    consumer: transport.consumer,
  },
  logger: consoleLogger('info'),
  consumeWrapper: telemetryConsumeWrapper(),
});

@serviceconnect/telemetry depends only on @opentelemetry/api as a peer; you bring the SDK (@opentelemetry/sdk-trace-node, exporters, etc.) so the framework doesn’t pin the wider OTel surface.

Common pitfalls

Forgetting AsyncHooksContextManager — active span context doesn’t propagate through await, every consume span looks like a root, and the trace tree falls apart.
Trying to mutate the transport in place: transport.producer is readonly, so transport.producer = telemetryProducer(...) won’t compile. Wrap the producer when assembling the transport you pass to createBus — transport: { producer: telemetryProducer(transport.producer), consumer: transport.consumer }. The consumer side uses BusOptions.consumeWrapper with telemetryConsumeWrapper(), not direct wrapping.
Treating logs as the only signal — at scale, full-text search across millions of lines doesn’t scale and doesn’t show causality. Combine structured logs with traces so you can pivot from one slow request to the chain of messages it kicked off.
Letting consumerBusy drive restarts — degraded on an idle queue is correct behaviour, not a fault.