Observability stack

End-to-end observability without external SDKs: structured logs with trace correlation, native counters / gauges / histograms, W3C trace propagation across RPC calls, and a store-and-forward relay that ships everything to your sink even through network outages.

Shape

Logs. Structured JSON via LoggerModule, with traceId / spanId attached automatically.
Metrics. Counters / histograms via titan-metrics, with Prometheus exposition + persistent storage.
Traces. W3C traceparent carried through async scopes via @omnitron-dev/titan/tracing.
Health. Standard probes via titan-health.
Shipping. All three pillars buffered locally and shipped via titan-telemetry-relay — survives collector outages without data loss.

Architecture

flowchart LR
  subgraph "Producer pod (your service)"
    Code[Service code]
    Code -- log --> Log[LoggerModule]
    Code -- metric --> Met[titan-metrics]
    Code -- span --> Trace[tracing]
    Health[titan-health]

    Log -- processor adds traceId --> Buf[Telemetry buffer]
    Met -- recordTyped --> Buf
    Trace -- end --> Buf
    Buf -- WAL on backpressure --> WAL[(local WAL)]
  end

  subgraph "Aggregator pod"
    AggBuf[Telemetry buffer]
    AggBuf --> Sink[Sink writer]
  end

  Buf -- Netron transport --> AggBuf
  Sink --> Backend[(OTLP collector / Loki / Tempo)]
  Sink --> Prom[(Prometheus)]
  Health --> Probe[Kubernetes probes]

`AppModule` — producer side

import { Module } from '@omnitron-dev/titan';
import { ConfigModule } from '@omnitron-dev/titan/module/config';
import {
  LoggerModule,
  ConsoleTransport,
  RedactionProcessor,
  type ILogProcessor,
} from '@omnitron-dev/titan/module/logger';
import { currentTrace } from '@omnitron-dev/titan/tracing';

import { TitanMetricsModule } from '@omnitron-dev/titan-metrics';
import { TitanHealthModule }  from '@omnitron-dev/titan-health';

// Custom processor: stamp traceId/spanId on every log line
const TraceContextProcessor: ILogProcessor = {
  process(record) {
    const trace = currentTrace();
    if (trace) {
      record.traceId = trace.traceId;
      record.spanId  = trace.spanId;
    }
    return record;
  },
};

@Module({
  imports: [
    ConfigModule.forRoot({/* ... */}),

    // ── Logging with redaction + trace correlation ─────────────────────
    LoggerModule.forRoot({
      level: process.env.NODE_ENV === 'production' ? 'info' : 'debug',
      transports: [new ConsoleTransport()],
      processors: [
        new RedactionProcessor(['password', 'token', 'headers.authorization']),
        TraceContextProcessor,
      ],
    }),

    // ── Metrics (with auto RPC instrumentation) ────────────────────────
    TitanMetricsModule.forRoot({
      appName: 'my-service',
      collection: {
        enabled:  true,
        interval: 10_000,
        process:  true,      // CPU / RSS / heap
        system:   true,      // load, free mem
        rpc:      true,      // every Netron call auto-counted
        custom:   true,
      },
      storage:   { type: 'memory', batchSize: 500, flushInterval: 10_000 },
      retention: { maxAge: '24h', cleanupInterval: 600_000 },
    }),

    // ── Health probes ──────────────────────────────────────────────────
    TitanHealthModule.forRoot({
      enableMemoryIndicator:    true,
      enableEventLoopIndicator: true,
      memoryThresholds:         { heapDegradedThreshold: 0.8, heapUnhealthyThreshold: 0.95 },
      eventLoopThresholds:      { lagDegradedThreshold: 50, lagUnhealthyThreshold: 200 },
      timeout:                  3_000,
      enableCaching:            true,
      cacheTtl:                 1_000,
      enableRpcService:         true,
      version:                  process.env.APP_VERSION,
    }),

    // ── Your feature modules ───────────────────────────────────────────
    UsersModule,
  ],
})
export class AppModule {}

Wiring the telemetry relay

The relay is not a Titan module — instantiate it in your bootstrap and wire it as a clientModule or via your DI container:

import { TelemetryRelayService } from '@omnitron-dev/titan-telemetry-relay';
import { Application } from '@omnitron-dev/titan';

const relay = new TelemetryRelayService({
  role:   'producer',
  nodeId: process.env.HOSTNAME ?? 'unknown',
  buffer: { maxBufferSize: 10_000, flushIntervalMs: 5_000, maxBatchSize: 500 },
  // `wal: false` disables it; otherwise pass a config (retention is by segment count).
  wal:    { directory: './.wal', maxSizeBytes: 10 * 1024 * 1024, maxSegments: 7 },
});

relay.setTransport(buildNetronTransportToAggregator());
await relay.start();

const app = await Application.create({ modules: [AppModule] });
await app.start();

// Bridge: pipe LoggerService outputs into the relay
// Bridge: pipe MetricsService periodic snapshots into the relay
// (project-specific glue — implement once per app)

process.on('SIGTERM', async () => {
  await app.stop();
  await relay.stop();
});

Aggregator pod

const relay = new TelemetryRelayService({
  role: 'aggregator',
});

relay.setAggregator({
  // The aggregator interface is { ingest(nodeId, entries), query(filter) }.
  // Entry types are 'log' | 'metric' | 'event' | 'health' | 'alert'.
  async ingest(nodeId, entries) {
    // Ship to your backend(s): OTLP, Loki, Tempo, Prometheus remote-write, etc.
    await Promise.all([
      shipLogs(entries.filter(e => e.type === 'log')),
      shipMetrics(entries.filter(e => e.type === 'metric')),
      shipEvents(entries.filter(e => e.type === 'event')),
    ]);
    return entries.length;        // count acknowledged
  },
  async query(filter) { /* … */ return []; },
});

await relay.start();

// Expose a Netron RPC endpoint so producer pods can call relay.receive(nodeId, entries)
const app = await Application.create({ modules: [AggregatorModule] });
await app.start();

Trace propagation in your code

import { startSpan, withTrace } from '@omnitron-dev/titan';

@Public()
async findById(id: string) {
  // startSpan(parent?) returns a new child TraceContext; run work inside
  // withTrace so currentTrace() — and the log processor — see this span.
  const span = startSpan();
  return withTrace(span, () => this.repo.find(id));
}

Netron propagates the traceparent across RPC calls automatically; log lines inside withTrace automatically carry traceId / spanId through the TraceContextProcessor above.

A typical instrumented service

import { Service, startSpan, withTrace } from '@omnitron-dev/titan';
import { Public } from '@omnitron-dev/titan/netron';
import { Metrics } from '@omnitron-dev/titan-metrics';

@Service('users@1.0.0')
class UsersService {
  constructor(
    private readonly logger: ILogger,
    private readonly metrics: MetricsService,
  ) {}

  @Public()
  @Metrics('users.findById')                                    // name only; tracks total + duration + errors
  async findById(id: string) {
    this.logger.info('findById', { id });                       // gets traceId/spanId via processor
    return withTrace(startSpan(), () => this.repo.find(id));
  }
}

Cross-module wiring notes

Concern	Wiring detail
Trace → log correlation	`TraceContextProcessor` reads `currentTrace()` and stamps `traceId` / `spanId` on every log record
Auto-RPC metrics	`collection.rpc: true` auto-counts Netron calls — no per-method `@Metrics` needed
Health → readiness	`/readyz` returns `200` only when every indicator is `healthy` or `degraded` — not `unhealthy`
Relay producer vs aggregator	Producer pods buffer + ship via transport; aggregator receives + writes to sink. Either or both.
WAL crash safety	Provide a `wal` config object (omit or `wal: false` to disable); buffer overflow spills to disk and replays on next start
Metrics retention	`retention.maxAge` is the in-storage window; ship metrics to long-term backend before that elapses

Production checklist

Logs: RedactionProcessor includes every sensitive header / field
Logs: TraceContextProcessor (or equivalent) attached — without it, trace correlation breaks
Metrics: appName set; otherwise cross-pod metrics pile together
Metrics: storage backend chosen — 'memory' is lost on restart
Metrics: scraped by Prometheus (/metrics) OR shipped via relay (pick one)
Traces: trace context propagated across worker thread boundaries via withTrace
Health: enableRpcService: true so the orchestrator can poll via Netron
Relay: a wal config provided (not wal: false) in production
Relay: producer / aggregator topology mirrors your deployment shape
Relay: aggregator's write() is idempotent (handles retries from buffer re-enqueue)
K8s: terminationGracePeriodSeconds ≥ relay's flush interval + buffer drain time