Observability

// Chatterbox metrics: every label has a small, fixed set of values.
import client from '@prometheus-io/client';

export const registry = new client.Registry();
registry.setDefaultLabels({ service: 'chatterbox' });
client.collectDefaultMetrics({ register: registry });

const TYPES = new Set(['hello', 'join', 'leave', 'message', 'edit', 'react', 'typing', 'receipt', 'presence', 'history', 'ack', 'error']);
const CODES = new Set([1000, 1001, 1006, 1009, 4400, 4401, 4403, 4408, 4413, 4426, 4429, 4503]);

// Bound the label values so that a hostile client cannot create new series.
export const typeLabel = (t) => (TYPES.has(t) ? t : 'unknown');
export const codeLabel = (c) => (CODES.has(c) ? String(c) : 'other');

export const connections = new client.Gauge({
  name: 'chatterbox_ws_connections', help: 'Open WebSocket connections', registers: [registry],
});
export const opened = new client.Counter({
  name: 'chatterbox_ws_opened_total', help: 'WebSocket connections accepted', registers: [registry],
});
export const closed = new client.Counter({
  name: 'chatterbox_ws_closed_total', help: 'WebSocket connections closed, by close code',
  labelNames: ['code'], registers: [registry],
});
export const messages = new client.Counter({
  name: 'chatterbox_ws_messages_total', help: 'Envelopes by direction and type',
  labelNames: ['direction', 'type'], registers: [registry],
});
// ... the instruments for errors, resumes, fan-out size, ICE and client quality follow the same pattern
export const fanout = new client.Histogram({
  name: 'chatterbox_fanout_seconds',
  help: 'Time from receiving a message to handing the frame to every recipient socket',
  buckets: [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
  registers: [registry],
});

wss.on('connection', (ws) => {
  ws.id = randomUUID();
  sockets.add(ws);
  m.connections.inc();
  m.opened.inc();
  const clog = log.child({ conn: ws.id });
  clog.info({ event: 'ws_open' }, 'connection opened');

  ws.on('message', (data) => {
    const t0 = process.hrtime.bigint();
    let env;
    try { env = JSON.parse(data.toString('utf8')); } catch { return fail(ws, null, 'bad_request'); }
    const type = m.typeLabel(env?.type);
    m.messages.inc({ direction: 'in', type });
    withSpan(`chatterbox ${type}`, { 'messaging.system': 'chatterbox', 'messaging.operation.type': 'process', 'messaging.operation.name': type, 'messaging.destination.name': env.room ?? 'none' }, (span) => {
// ...
      } else if (type === 'message') {
        const set = rooms.get(env.room) ?? new Set();
        let n = 0;
        for (const peer of set) { send(peer, { type: 'message', room: env.room, payload: env.payload }); n++; }
        m.recipients.observe(n);
        m.fanout.observe(Number(process.hrtime.bigint() - t0) / 1e9);
        send(ws, { type: 'ack', room: env.room, payload: { ref: env.id, ok: true } });
        span.setAttribute('chatterbox.recipients', n);
// ...
  ws.on('close', (code) => {
    sockets.delete(ws);
    for (const set of rooms.values()) set.delete(ws);
    m.connections.dec();
    m.closed.inc({ code: m.codeLabel(code) });
    clog.info({ event: 'ws_close', code }, 'connection closed');
  });
// ...
function send(ws, partial) {
  if (ws.readyState !== WebSocket.OPEN) return;
  const env = { v: 1, id: `s_${randomUUID()}`, ts: Date.now(), ...partial };
  m.messages.inc({ direction: 'out', type: m.typeLabel(env.type) });
  ws.send(JSON.stringify(env));
}
// ...
// The metrics listener is separate and bound to loopback: never publish it through the public proxy.
http.createServer(async (req, res) => {
  if (req.url !== '/metrics') return res.writeHead(404).end();
  res.writeHead(200, { 'Content-Type': m.registry.contentType });
  res.end(await m.registry.metrics());
}).listen(METRICS_PORT, process.env.METRICS_HOST ?? '127.0.0.1');

chatterbox_ws_connections{service="chatterbox"} 0
chatterbox_ws_opened_total{service="chatterbox"} 3
chatterbox_ws_closed_total{code="1006",service="chatterbox"} 1
chatterbox_ws_closed_total{code="1000",service="chatterbox"} 1
chatterbox_ws_closed_total{code="4408",service="chatterbox"} 1
chatterbox_ws_messages_total{direction="in",type="message",service="chatterbox"} 20
chatterbox_ws_messages_total{direction="out",type="message",service="chatterbox"} 60
chatterbox_ws_errors_total{code="bad_request",service="chatterbox"} 2
chatterbox_ws_hello_total{result="resumed",service="chatterbox"} 1
chatterbox_fanout_recipients_sum{service="chatterbox"} 60
chatterbox_fanout_recipients_count{service="chatterbox"} 20
chatterbox_client_ice_total{result="failed",path="direct",service="chatterbox"} 1

curl -s http://chatterbox:18602/metrics | promtool check metrics

name: chatterbox-observability
services:
  chatterbox:
    image: node:24-alpine
    working_dir: /app
    command: ["node", "src/server.js"]
    environment:
      APP_HOST: 0.0.0.0
      METRICS_HOST: 0.0.0.0        # reachable on the compose network only; see the published ports below
      OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4318
      OTEL_SERVICE_NAME: chatterbox
    volumes:
      - ./chatterbox:/app:ro
    ports:
      - "127.0.0.1:18601:18601"    # WebSocket endpoint
      # 18602 (metrics) is deliberately not published
  coturn:
    image: coturn/coturn:4.18.0
    command: >
      turnserver -n --realm=example.test --use-auth-secret --static-auth-secret=change-me
      --no-tls --min-port=49160 --max-port=49200
      --prometheus --prometheus-address=0.0.0.0
    ports:
      - "127.0.0.1:18631:3478/udp"
  # livekit and otel-collector: see the sections on LiveKit and tracing
  prometheus:
    image: prom/prometheus:v3.15.1
    command: ["--config.file=/etc/prometheus/prometheus.yml", "--storage.tsdb.retention.time=2d"]
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - ./rules:/etc/prometheus/rules:ro
    ports:
      - "127.0.0.1:18610:9090"

global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - /etc/prometheus/rules/*.yml

scrape_configs:
  - job_name: chatterbox
    static_configs:
      - targets: ["chatterbox:18602"]
  - job_name: coturn
    static_configs:
      - targets: ["coturn:9641"]
  - job_name: livekit
    static_configs:
      - targets: ["livekit:6789"]

groups:
  - name: chatterbox-recording
    rules:
      - record: chatterbox:ws_connections:sum
        expr: sum(chatterbox_ws_connections)
      - record: chatterbox:fanout_seconds:p99_5m
        expr: histogram_quantile(0.99, sum by (le) (rate(chatterbox_fanout_seconds_bucket[5m])))
      - record: chatterbox:ws_abnormal_close_ratio:5m
        expr: |
          sum(rate(chatterbox_ws_closed_total{code=~"1006|4408|4503"}[5m]))
          /
          sum(rate(chatterbox_ws_closed_total[5m]))
      - record: chatterbox:ws_opened:rate1m
        expr: sum(rate(chatterbox_ws_opened_total[1m]))
      # ...
  - name: chatterbox-alerts
    rules:
      # ...
      - alert: ChatterboxFanoutSlow
        expr: chatterbox:fanout_seconds:p99_5m > 0.25
        for: 10m
        labels: { severity: ticket }
        annotations:
          summary: "p99 fan-out latency is {{ $value | humanizeDuration }}"
      - alert: ChatterboxAbnormalCloseRatioHigh
        expr: chatterbox:ws_abnormal_close_ratio:5m > 0.2
        for: 5m
        labels: { severity: page }
        annotations:
          summary: "{{ $value | humanizePercentage }} of closed sockets ended with 1006, 4408 or 4503"

rule_files:
  - ../rules/chatterbox.yml
evaluation_interval: 1m
tests:
  - interval: 1m
    input_series:
      # 40% of the closed sockets end abnormally: 6 of 15 closes per minute
      - series: 'chatterbox_ws_closed_total{code="1006"}'
        values: '0+6x15'
      - series: 'chatterbox_ws_closed_total{code="1000"}'
        values: '0+9x15'
    alert_rule_test:
      - eval_time: 3m
        alertname: ChatterboxAbnormalCloseRatioHigh
        exp_alerts: []
      - eval_time: 12m
        alertname: ChatterboxAbnormalCloseRatioHigh
        exp_alerts:
          - exp_labels: { severity: page }
            exp_annotations:
              summary: "40% of closed sockets ended with 1006, 4408 or 4503"

promtool check config prometheus.yml
promtool check rules rules/chatterbox.yml
promtool test rules tests/chatterbox.test.yml

import { trace, context, propagation, SpanKind, SpanStatusCode } from '@opentelemetry/api';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { BatchSpanProcessor, SimpleSpanProcessor, ConsoleSpanExporter } from '@opentelemetry/sdk-trace';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';

// The exporter reads OTEL_EXPORTER_OTLP_ENDPOINT itself; an empty processor list would leave tracing off.
const spanProcessors = [];
if (process.env.OTEL_EXPORTER_OTLP_ENDPOINT) spanProcessors.push(new BatchSpanProcessor({ exporter: new OTLPTraceExporter() }));
if (process.env.OTEL_CONSOLE === '1') spanProcessors.push(new SimpleSpanProcessor({ exporter: new ConsoleSpanExporter() }));
const sdk = new NodeSDK({
  serviceName: 'chatterbox',
  spanProcessors,
});
sdk.start(); // registers the AsyncLocalStorage context manager and the W3C propagators
process.on('SIGTERM', () => sdk.shutdown()); // flushes the batch processor

export const tracer = trace.getTracer('chatterbox.server');
export function withSpan(name, attributes, fn, options = {}) {
  return tracer.startActiveSpan(name, { kind: SpanKind.SERVER, attributes, ...options }, (span) => {
    try { return fn(span); }
    catch (err) { span.recordException(err); span.setStatus({ code: SpanStatusCode.ERROR }); throw err; }
    finally { span.end(); }
  });
}
export function inject() { const c = {}; propagation.inject(context.active(), c); return c; }
export function extract(carrier) { return propagation.extract(context.active(), carrier); }
export { context, trace };

// Two "nodes" joined by a backplane. An EventEmitter stands in for a Redis channel:
// the trace context travels inside the backplane message, never in the Chatterbox envelope.
import { EventEmitter } from 'node:events';
import { SpanKind } from '@opentelemetry/api';
import { tracer, inject, extract } from './tracing.js';
import { log } from './log.js';

const backplane = new EventEmitter();

// Node B: consumer. The extracted context makes the new span a child of the publisher's span.
backplane.on('chatterbox:general', (raw) => {
  const msg = JSON.parse(raw);
  const parent = extract(msg.traceContext);
  tracer.startActiveSpan('backplane receive', { kind: SpanKind.CONSUMER, attributes: { 'messaging.system': 'redis', 'messaging.destination.name': 'chatterbox:general' } }, parent, (span) => {
    log.info({ event: 'fanout_local', recipients: 3 }, 'delivered to local sockets');
    span.end();
  });
});

// Node A: handles a client message, then publishes it.
tracer.startActiveSpan('chatterbox message', { kind: SpanKind.SERVER }, (handler) => {
  tracer.startActiveSpan('backplane publish', { kind: SpanKind.PRODUCER, attributes: { 'messaging.system': 'redis', 'messaging.destination.name': 'chatterbox:general' } }, (pub) => {
    backplane.emit('chatterbox:general', JSON.stringify({ envelope: { type: 'message' }, traceContext: inject() }));
    pub.end();
  });
  handler.end();
});

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
processors:
  batch:
    timeout: 1s
exporters:
  debug:
    verbosity: detailed
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [debug]

import pino from 'pino';
import { trace, context } from '@opentelemetry/api';

// One JSON object per line; the active span's ids are added to every record.
export const log = pino({
  base: { service: 'chatterbox' },
  timestamp: pino.stdTimeFunctions.isoTime,
  formatters: { level: (label) => ({ level: label }) },
  mixin() {
    const sc = trace.getSpan(context.active())?.spanContext();
    return sc && trace.isSpanContextValid(sc) ? { trace_id: sc.traceId, span_id: sc.spanId } : {};
  },
});

{"level":"info","time":"2026-10-25T00:09:47.589Z","service":"chatterbox","conn":"32f614c4-15de-47f2-b089-a9c1fa12cc01","event":"ws_open","msg":"connection opened"}
{"level":"warn","time":"2026-10-25T00:09:47.800Z","service":"chatterbox","trace_id":"cd657a451ac1c290436d432956d62a1e","span_id":"7026341a9fe1316d","conn":"32f614c4-15de-47f2-b089-a9c1fa12cc01","event":"protocol_error","code":"bad_request","msg":"error envelope sent"}
{"level":"warn","time":"2026-10-25T00:09:47.800Z","service":"chatterbox","conn":"32f614c4-15de-47f2-b089-a9c1fa12cc01","event":"protocol_error","code":"bad_request","msg":"error envelope sent"}
{"level":"info","time":"2026-10-25T00:09:48.103Z","service":"chatterbox","conn":"32f614c4-15de-47f2-b089-a9c1fa12cc01","event":"ws_close","code":1006,"msg":"connection closed"}