@utility/art-trace-logger

A lightweight, distributed trace logger for Node.js services. Streams structured request/response/error trace events to Kafka, with built-in trace context propagation via AsyncLocalStorage and a first-class NestJS integration.

Designed to plug into a Kafka → Logstash → Elasticsearch → Kibana pipeline.

Features

• Structured log entries (LogEntry) with predefined log points (CLIENT_REQUEST, SERVICE_REQUEST, SERVICE_ERROR, …)
• Optional HTTP body capture (inbound Express + outbound http/https patch) with redaction and size limits
• Automatic outbound tracing for got / axios via installOutboundHttpInstrumentation
• Automatic traceId + monotonically-increasing sequence propagation via AsyncLocalStorage
• Fluent LogEntryBuilder API
• Kafka producer (built on kafkajs)
• Optional NestJS DynamicModule with lifecycle-managed connect/disconnect
• Framework-agnostic core — works in plain Node.js, Express, etc.

Installation

npm install @utility/art-trace-logger

Pin an exact version in production:

{
  "dependencies": {
    "@utility/art-trace-logger": "1.0.0"
  }
}

The package ships pre-built (dist/) so no build step is required after install.

Peer dependencies

NestJS and RxJS are optional peers — only required if you use the /nestjs entrypoint:

npm install @nestjs/common @nestjs/core rxjs reflect-metadata

Configuration

All consumers (NestJS or plain) take the same config object:

interface TraceLoggerConfig {
  serviceName: string;     // logical service identity, stamped on every log
  environment: string;     // 'dev' | 'staging' | 'prod' | …
  kafkaBrokers: string[];  // e.g. ['10.0.0.10:9094']
  kafkaTopic: string;      // e.g. 'Middleware-APILogs'
  enableConsole?: boolean; // mirror entries to stdout (useful in dev)
  consoleLogPayload?: boolean;       // include payload on console lines (truncated)
  consoleMaxPayloadBytes?: number;   // console payload limit (default 2048)
  kafkaEnabled?: boolean;  // set false for console-only mode
  ssl?: TraceLoggerSslConfig; // TLS for the Kafka connection — set it to null for local test
  lookup?: LoggerLookupModuleOptions; // service-code lookup cache (see "Service-code lookup" below)
}

Environment variables (Express bootstrap)

Typical flags used by Express services (src/trace/bootstrap.ts):

| Variable | Description | | --- | --- | | TRACE_LOGGER_ENABLED | Master switch (true / false) | | TRACE_SERVICE_NAME | serviceName on every entry (e.g. orders-api) | | TRACE_LOGGER_CONSOLE | Mirror trace lines to stdout | | TRACE_LOGGER_KAFKA | Send to Kafka (false = console-only) | | KAFKA_BROKERS | Comma-separated broker list | | KAFKA_TRACE_TOPIC | Kafka topic (e.g. Middleware-APILogs) | | TRACE_LOGGER_LOG_BODIES | Capture request + response bodies on trace events | | TRACE_LOGGER_LOG_REQUEST_BODY | Request bodies only | | TRACE_LOGGER_LOG_RESPONSE_BODY | Response bodies only | | TRACE_LOGGER_MAX_PAYLOAD_BYTES | Max body size stored on Kafka entries (default 32768) | | TRACE_LOGGER_CONSOLE_LOG_BODIES | Print bodies on console (same line as trace metadata) | | TRACE_LOGGER_CONSOLE_MAX_PAYLOAD_BYTES | Console body truncation (default 2048) |

Helpers: resolveBodyLoggingOptions(process.env), resolveConsolePayloadOptions(process.env).

Kafka shape: each tracer.log() call produces one Kafka message — a single JSON LogEntry. The payload field is part of that object, not a separate message. A full HTTP flow emits multiple messages (CLIENT_REQUEST → SERVICE_REQUEST → …) sharing the same traceId.

Kafka SSL / TLS

The package never ships cert material of its own — supply it from a central runtime source (a mounted Kubernetes secret, a secret manager, or files on disk) via config or env vars.

Config shape

interface TraceLoggerSslConfig {
  // Preferred: filesystem paths (point at a mounted secret / shared volume).
  // If omitted, the matching KAFKA_SSL_*_PATH env var is used as a fallback.
  caPath?: string;
  certPath?: string;
  keyPath?: string;

  // Escape hatch: inline PEM contents (e.g. fetched from a secret manager).
  // Takes precedence over the *Path equivalents.
  ca?: string;
  cert?: string;
  key?: string;

  rejectUnauthorized?: boolean; // default true; set false only for self-signed in non-prod
}

Per cert, precedence is: inline PEM → config *Path → KAFKA_SSL_*_PATH env var.

Via env vars

Point at certs from a mounted secret by setting the env vars (e.g. via a local .env loaded with node --env-file=.env):

KAFKA_SSL_CA_PATH=/etc/kafka-certs/ca.pem
KAFKA_SSL_CERT_PATH=/etc/kafka-certs/client.pem
KAFKA_SSL_KEY_PATH=/etc/kafka-certs/client-key.pem

Via config

Callers that want to control it can pass ssl directly — config values win over env vars:

TraceLogger.create({
  /* …core config… */
  ssl: {
    caPath: '/etc/kafka-certs/ca.pem',
    certPath: '/etc/kafka-certs/client.pem',
    keyPath: '/etc/kafka-certs/client-key.pem',
  },
});

Behavior: a path you configure explicitly (config or env) that can't be read throws at startup rather than being silently skipped — so a real misconfiguration fails loud. When no cert material is configured at all and ssl is omitted, the connection is plaintext.

Service-code lookup

The optional lookup module resolves externalService codes into full service metadata via a Redis-cached lookup table (optionally seeded from an Oracle DB). No endpoints or credentials ship with the package — configure them per deployment, or leave them unset to disable the lookup cache entirely (service codes then pass through unresolved).

Configure via TraceModule.forRoot({ ..., lookup }):

TraceModule.forRoot({
  /* …core config… */
  lookup: {
    redis: {
      // Sentinel mode:
      sentinels: [{ host: 'redis-sentinel.example.internal', port: 26379 }],
      name: 'mymaster',
      // …or standalone mode:
      // host: 'redis.example.internal', port: 6379,
      password: process.env.LOOKUP_REDIS_PASSWORD,
      cacheKey: 'art:trace:logger:lookup',   // default
      cacheExpireSeconds: 31536000,          // default (1 year)
    },
    db: {
      host: 'oracle.example.internal',
      port: 1521,
      username: process.env.LOOKUP_DB_USERNAME,
      password: process.env.LOOKUP_DB_PASSWORD,
      database: 'MYSERVICE',
    },
  },
});

…or via env vars (used when the corresponding config value is omitted):

| Variable | Description | | --- | --- | | LOOKUP_REDIS_SENTINELS | Comma-separated host:port sentinel list | | LOOKUP_REDIS_NAME | Sentinel master group name (default mymaster) | | LOOKUP_REDIS_HOST / LOOKUP_REDIS_PORT | Standalone Redis (when no sentinels) | | LOOKUP_REDIS_PASSWORD / LOOKUP_REDIS_DB | Redis auth / db index | | LOOKUP_REDIS_CACHE_KEY | Cache key (default art:trace:logger:lookup) | | LOOKUP_REDIS_CACHE_EXPIRE | Cache TTL in seconds (default 31536000) | | LOOKUP_DB_HOST / LOOKUP_DB_PORT | Oracle lookup DB endpoint | | LOOKUP_DB_USERNAME / LOOKUP_DB_PASSWORD | Oracle credentials | | LOOKUP_DB_NAME | Service name (contains .) or SID |

Neutrinos / Express service-name lookup

Neutrinos services read the same DB-backed lookup rows from Redis once at startup and keep an in-memory index. Outbound calls are matched by HTTP method and normalized pathname (host, query string, fragment, and trailing slash are ignored). Lookup URI parameters such as {datasetId} match exactly one path segment. Fixed segments remain strict: a version mismatch such as /v2/... versus /v3/... does not match; update LOGGER_LOOKUP with the URI actually called by the service.

import {
  initializeNeutrinosLookup,
  installNeutrinosOutboundInstrumentation,
} from '@alrajhitakaful/art-trace-logger/neutrinos';

const serviceLookup = await initializeNeutrinosLookup({
  onWarning: (message) => console.warn(message),
});

installNeutrinosOutboundInstrumentation({
  tracer,
  serviceLookup,
  // ...body logging options
});

The shared Redis cache remains the runtime source. If its key is empty and the LOOKUP_DB_* variables are configured, the initializer seeds Redis once from LOGGER_LOOKUP; Neutrinos services never query Oracle per request. Set LOOKUP_REFRESH_INTERVAL_MS=0 to disable the default five-minute in-memory refresh. If Redis/Oracle is missing or unavailable, tracing continues and externalService falls back to the outbound hostname.

Quick Start — NestJS

1. Register the module

// src/app.module.ts
import { Module } from '@nestjs/common';
import { TraceModule } from '@utility/art-trace-logger/nestjs';

@Module({
  imports: [
    TraceModule.forRoot({
      serviceName: 'orders-api',
      environment: process.env.NODE_ENV ?? 'dev',
      kafkaBrokers: (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','),
      kafkaTopic: 'Middleware-APILogs',
      enableConsole: true,
      consoleLogPayload: true,
      consoleMaxPayloadBytes: 2048,
    }),
  ],
})
export class AppModule {}

TraceModule is @Global(), so TraceLoggerService is injectable everywhere without re-importing. Kafka connect/disconnect is wired to Nest's lifecycle (OnModuleInit / OnModuleDestroy).

Do not combine createExpressTraceMiddleware and TraceInterceptor on the same app — you will get duplicate CLIENT_* events. Pick one inbound approach.

2. Wrap each request in a trace context

Use a NestJS interceptor to open a trace scope (picking up an upstream X-Trace-Id if present) and log the request and the actual response body / errors via RxJS tap and catchError. This is preferred over a middleware + res.on('finish') because the interceptor's Observable gives you the real response payload and the thrown error object.

// src/trace.interceptor.ts
import {
  CallHandler,
  ExecutionContext,
  Injectable,
  NestInterceptor,
} from '@nestjs/common';
import { Observable, throwError } from 'rxjs';
import { catchError, tap } from 'rxjs/operators';
import {
  LogEntryBuilder,
  TraceLoggerService,
} from '@utility/art-trace-logger/nestjs';
import type { Request, Response } from 'express';

@Injectable()
export class TraceInterceptor implements NestInterceptor {
  constructor(private readonly tracer: TraceLoggerService) {}

  intercept(ctx: ExecutionContext, next: CallHandler): Observable<unknown> {
    const http = ctx.switchToHttp();
    const req = http.getRequest<Request>();
    const res = http.getResponse<Response>();
    const upstream = req.header('x-trace-id');
    const start = Date.now();

    // runInTrace returns whatever the callback returns. The Observable is
    // created *inside* the AsyncLocalStorage scope, so every downstream
    // operator (and any awaited work in the handler) inherits the same traceId.
    return this.tracer.runInTrace(() => {
      this.tracer.log(
        LogEntryBuilder.clientRequest()
          .method(req.method)
          .uri(req.originalUrl)
          .clientIp(req.ip ?? '')
          .headers(req.headers as Record<string, string>)
          .payload(req.body),
      );

      return next.handle().pipe(
        tap((body) =>
          this.tracer.log(
            LogEntryBuilder.clientResponse()
              .method(req.method)
              .uri(req.originalUrl)
              .statusCode(res.statusCode)
              .executionTimeMs(Date.now() - start)
              .payload(body), // ← real response body, available here
          ),
        ),
        catchError((err) => {
          this.tracer.log(
            LogEntryBuilder.serviceError()
              .method(req.method)
              .uri(req.originalUrl)
              .statusCode(err?.status ?? 500)
              .executionTimeMs(Date.now() - start)
              .errorCode(err?.code ?? 'UNHANDLED')
              .errorMessage(err?.message ?? String(err))
              .stackTrace(err?.stack ?? ''),
          );
          return throwError(() => err);
        }),
      );
    }, upstream);
  }
}

Register it globally so every route is traced:

// src/app.module.ts
import { Module } from '@nestjs/common';
import { APP_INTERCEPTOR } from '@nestjs/core';
import { TraceModule } from '@utility/art-trace-logger/nestjs';
import { TraceInterceptor } from './trace.interceptor';

@Module({
  imports: [
    TraceModule.forRoot({
      serviceName: 'orders-api',
      environment: process.env.NODE_ENV ?? 'dev',
      kafkaBrokers: (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','),
      kafkaTopic: 'Middleware-APILogs',
      enableConsole: true,
    }),
  ],
  providers: [
    { provide: APP_INTERCEPTOR, useClass: TraceInterceptor },
  ],
})
export class AppModule {}

Why an interceptor for Nest? tap() receives the controller return value directly. The Express middleware can log req.body and bodies sent via res.json / res.send, but not arbitrary handler return objects — for Nest APIs the interceptor remains the better fit for response payloads.

3. Log inside your services

Because the middleware ran the request inside runInTrace, every log call automatically inherits the same traceId and an incrementing sequence:

// src/orders.service.ts
import { Injectable } from '@nestjs/common';
import { TraceLoggerService, LogEntryBuilder } from '@utility/art-trace-logger/nestjs';

@Injectable()
export class OrdersService {
  constructor(private readonly tracer: TraceLoggerService) {}

  async create(dto: unknown) {
    const started = Date.now();
    try {
      this.tracer.log(
        LogEntryBuilder.serviceRequest()
          .externalService('payments-api')
          .externalSubService('POST /charge')
          .payload(dto),
      );

      const result = await fetch('https://payments/charge', { /* … */ });

      this.tracer.log(
        LogEntryBuilder.serviceResponse()
          .externalService('payments-api')
          .statusCode(result.status)
          .executionTimeMs(Date.now() - started),
      );

      return result;
    } catch (err: any) {
      this.tracer.log(
        LogEntryBuilder.serviceError()
          .errorCode('PAYMENTS_FAILED')
          .errorMessage(err.message)
          .stackTrace(err.stack),
      );
      throw err;
    }
  }
}

Distributed tracing (Express + outbound HTTP)

Use the same traceId on every service by forwarding X-Trace-Id.

Recommended bootstrap:

// src/trace/bootstrap.ts
import {
  TraceLogger,
  createExpressTraceMiddleware,
  installOutboundHttpInstrumentation,
  resolveBodyLoggingOptions,
  resolveConsolePayloadOptions,
} from '@utility/art-trace-logger';

const bodyLoggingOptions = resolveBodyLoggingOptions(process.env);
const consolePayloadOptions = resolveConsolePayloadOptions(process.env);

export const tracer = TraceLogger.create({
  serviceName: 'my-service',
  environment: process.env.NEUTRINOS_APP_ENV ?? 'dev',
  kafkaBrokers: (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','),
  kafkaTopic: process.env.KAFKA_TRACE_TOPIC ?? 'Middleware-APILogs',
  enableConsole: process.env.TRACE_LOGGER_CONSOLE === 'true',
  consoleLogPayload: consolePayloadOptions.consoleLogPayload,
  consoleMaxPayloadBytes: consolePayloadOptions.consoleMaxPayloadBytes,
  kafkaEnabled: process.env.TRACE_LOGGER_KAFKA !== 'false',
});

export async function initTrace() {
  installOutboundHttpInstrumentation(tracer, bodyLoggingOptions);
  await tracer.connect();
}

export function applyTraceMiddleware(app: Express.Application) {
  app.use(createExpressTraceMiddleware(tracer, bodyLoggingOptions));
}

Call initTrace() before routes and applyTraceMiddleware(app) after express.json().

What the middleware / patch capture automatically

| Direction | Metadata | Bodies (when TRACE_LOGGER_LOG_BODIES=true) | | --- | --- | --- | | Inbound | method, uri, status, timing | req.body; response via res.json / res.send | | Outbound (got, axios, …) | method, url, status, timing | http.request options body; response stream (size-limited) |

Manual outbound (if you do not use the patch):

import { injectTraceHeaders, logServiceRequest, logServiceResponse } from '@utility/art-trace-logger';

injectTraceHeaders(headers);
logServiceRequest(tracer, 'POST', url);
// … call …
logServiceResponse(tracer, 'POST', url, statusCode, ms);

Downstream services that use createExpressTraceMiddleware read X-Trace-Id and log CLIENT_REQUEST / CLIENT_RESPONSE with the same id.

Body logging limits: passwords/tokens are redacted; large payloads are truncated. Multipart streams and handlers that only use res.end() may omit response bodies on the Express path — use the Nest interceptor if you need controller return values.

Neutrinos Studio adapter

Neutrinos-generated Express services use the isolated /neutrinos adapter. The core logger and the existing NestJS adapter remain unchanged.

import { TraceLogger } from '@alrajhitakaful/art-trace-logger/core';
import {
  createNeutrinosTraceMiddleware,
  installNeutrinosOutboundInstrumentation,
} from '@alrajhitakaful/art-trace-logger/neutrinos';

const tracer = TraceLogger.create({
  serviceName: 'art-gi-b2c-bff',
  environment: process.env.NEUTRINOS_APP_ENV ?? 'dev',
  kafkaBrokers: (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','),
  kafkaTopic: process.env.KAFKA_TRACE_TOPIC ?? 'Middleware-APILogs',
});

const adapterOptions = {
  tracer,
  logRequestBody: true,
  logResponseBody: true,
};

installNeutrinosOutboundInstrumentation(adapterOptions);
await tracer.connect();

// Register after express.json() and before application routes.
app.use(createNeutrinosTraceMiddleware(adapterOptions));

The adapter automatically propagates:

• x-trace-id: one identifier for the complete distributed journey.
• x-business-trace-id: the configured business identifier.
• x-client-trace-id: the upstream client identifier when supplied.
• x-span-id: a unique identifier for each inbound or outbound operation.
• x-parent-span-id: the calling operation's span identifier.

CLIENT_REQUEST and CLIENT_RESPONSE share the inbound span. Every outbound request gets a new child span; its SERVICE_REQUEST and response/error share that child span, and the receiving service uses it for its client events. This makes parallel calls unambiguous without a distributed sequence counter.

Use traceId + serviceName + spanId + sequence as the event identity in ELK. Map spanId and parentSpanId as Elasticsearch keyword fields.

Quick Start — Plain Node.js (no NestJS)

The core is framework-agnostic. Import from the package root.

1. Bootstrap a single logger instance

// src/logger.ts
import { TraceLogger } from '@utility/art-trace-logger';

export const tracer = TraceLogger.create({
  serviceName: 'orders-api',
  environment: process.env.NODE_ENV ?? 'dev',
  kafkaBrokers: (process.env.KAFKA_BROKERS ?? 'localhost:9094').split(','),
  kafkaTopic: 'Middleware-APILogs',
  enableConsole: true,
});

// Connect once at startup
await tracer.connect();

// Disconnect on shutdown
process.on('SIGTERM', async () => {
  await tracer.disconnect();
  process.exit(0);
});

2. Wrap units of work in a trace context

import { tracer } from './logger';
import { LogEntryBuilder } from '@utility/art-trace-logger';

tracer.runInTrace(() => {
  tracer.log(
    LogEntryBuilder.clientRequest()
      .method('POST')
      .uri('/orders')
      .clientIdentity('partner-xyz'),
  );

  // …business logic…

  tracer.log(
    LogEntryBuilder.clientResponse()
      .method('POST')
      .uri('/orders')
      .statusCode(201)
      .executionTimeMs(42),
  );
});

3. Express example

Prefer the built-in middleware (see Distributed tracing):

import express from 'express';
import { initArtTraceLogger, applyArtTraceMiddleware } from './trace/bootstrap';

const app = express();
app.use(express.json());

await initArtTraceLogger();
applyArtTraceMiddleware(app);

app.listen(3000);

For hand-rolled middleware, call createExpressTraceMiddleware(tracer, resolveBodyLoggingOptions(process.env)) after express.json().

API Reference

TraceLogger (core)

| Method | Description | | --- | --- | | TraceLogger.create(config) | Factory — returns a logger instance. | | connect() | Connects the underlying Kafka producer. Call once at startup. | | disconnect() | Disconnects gracefully. Call on shutdown. | | log(builder \| partial) | Emits a log entry. Accepts a LogEntryBuilder or a Partial<LogEntry>. | | runInTrace(fn, traceId?) | Runs fn inside an AsyncLocalStorage scope where every nested log() shares the same traceId. If traceId is omitted a UUID is generated. |

installOutboundHttpInstrumentation(tracer, bodyOptions?)

Patches Node http / https once per process. While inside runInTrace, outbound calls emit SERVICE_REQUEST / SERVICE_RESPONSE, forward X-Trace-Id, and optionally capture bodies. Call at startup before handling traffic.

resolveBodyLoggingOptions(env?) / resolveConsolePayloadOptions(env?)

Map TRACE_LOGGER_* environment variables to BodyLoggingOptions and console payload settings.

TraceLoggerService (NestJS)

Same surface as TraceLogger, minus connect/disconnect (handled by Nest lifecycle hooks). Import from @utility/art-trace-logger/nestjs (re-exports the core).

LogEntryBuilder

Static factories — one per LogPoint:

• clientRequest() / clientResponse()
• serviceRequest() / serviceResponse()
• serviceError() (alias: internalError() — deprecated)
• custom()

Fluent setters: clientTraceId, clientIp, clientIdentity, method, uri, headers, statusCode, executionTimeMs, externalService, externalSubService, payload, errorCode, errorMessage, errorContext, stackTrace.

TraceContext

Low-level helper around AsyncLocalStorage:

• TraceContext.run(fn, traceId?) — start/inherit a trace scope.
• TraceContext.getTraceId() — current trace id ('no-trace' outside a scope).
• TraceContext.nextSequence() — increment & return the per-trace counter.

The TraceLogger uses these automatically — you rarely call them directly.

LogEntry shape

Each emitted Kafka message is a JSON-serialized LogEntry:

{
  logPoint: 'CLIENT_REQUEST',
  traceId: 'b1a2…',
  sequence: 1,
  serviceName: 'orders-api',
  environment: 'prod',
  machineName: 'pod-7c9f',
  timestamp: '2026-05-04T10:22:15.043Z',
  method: 'POST',
  uri: '/orders',
  statusCode: 201,
  executionTimeMs: 42,
  // …plus any fields you set on the builder
}

Local Infrastructure

For local development, run a Kafka + Logstash + Elasticsearch + Kibana stack (e.g. via Docker Compose) with a Logstash pipeline that consumes the configured trace topic (e.g. Middleware-APILogs) and indexes it into Elasticsearch.

Operational Notes

• log() is fire-and-forget — Kafka send errors are swallowed (and logged to stderr) so they never break a request path.
• A single TraceLogger instance is sufficient per process; share it.
• Outside a runInTrace scope, entries are stamped with traceId: 'no-trace' and sequence: 0 — useful for boot-time logs but propagation is lost.
• To propagate trace ids across services, forward the traceId as an X-Trace-Id header on outbound HTTP calls and pass it to runInTrace(fn, traceId) on the receiving side.

License

Internal — Digital Transformation.