Log Stream

Pluggable structured-log dispatcher — fire-and-forget JSON records from request hot paths to one or many sinks (local file with rotation, generic webhook, OTLP HTTP/JSON, OTLP gRPC, AWS CloudWatch Logs, RFC 5424 syslog over UDP/TCP/TLS). Each sink keeps its own connection / fd / batch buffer, so a slow remote collector backpressures only its own queue — never the request thread that called emit().

Every record passes through b.redact BEFORE any sink sees it. PHI / PCI / JWTs / PEM blocks / AWS access keys / vault-sealed strings / credit-card-shaped digits / SSN-shaped values are stripped on the framework side, not delegated to the operator's sink config — a misnamed field cannot leak sensitive data into operational logs.

Sink failures are drop-silent on the hot path: a captured Promise reroutes the error into audit.system.log.sink_failure so a downed collector never crashes the request that emitted the log line. Pending emits are tracked and drained on shutdown() so records queued just before close still reach disk / the wire.

Bidirectional command channel: onIncoming(handler) registers a handler for inbound events; the operator wires their preferred transport (HTTP route, webhook receiver, SSE subscription, message-queue consumer) to call deliverIncoming(payload), which redacts, audits, and dispatches to every registered handler. The framework provides the dispatch; the operator provides the wire.

Built-in protocols: local — append-only file with size + age rotation webhook — generic HTTP POST (Splunk HEC, Datadog, Loki, Sumo Logic, custom JSON collectors) otlp — OpenTelemetry Protocol over HTTP/JSON otlp-grpc — same Logs Data Model over gRPC (higher throughput; hand-encoded protobuf, no parser dependency) cloudwatch — PutLogEvents over HTTPS with SigV4; honours the 10K-event / 1 MiB / 256 KiB-per-event AWS caps syslog — RFC 5424 octet-counting framing over UDP / TCP / TLS (default ports 514 / 6514)

b.logStream.init(opts) #

{
  sinks:    { [name]: { protocol, minLevel?, ...protocolOpts } },
  minLevel: "debug" | "info" | "warn" | "error",   // default "info"
}

Configure the dispatcher. Call once at boot; subsequent calls are no-ops while the dispatcher is initialized (call shutdown() first to reconfigure). Every named sink resolves a built-in protocol (local / webhook / otlp / otlp-grpc / cloudwatch / syslog) and constructs a per-sink instance from its own typed config block.

Records below minLevel (or a sink's per-sink minLevel override) are dropped before redaction — debug / info chatter on a production deployment costs nothing past the dispatcher.

b.logStream.init({
  minLevel: "info",
  sinks: {
    file:   { protocol: "local",   path: "/var/log/app.log" },
    remote: { protocol: "otlp",
              url:         "https://collector.internal:4318/v1/logs",
              serviceName: "checkout",
              minLevel:    "warn" },
  },
});

b.logStream.emit(level, message, meta?) #

Synchronous, fire-and-forget emit to every registered sink whose level filter accepts level. The record is { ts, level, message, meta }; meta is run through b.redact.redact BEFORE distribution so PHI / credentials / vault-sealed values never reach a sink even on a misnamed field. Sink errors are captured, audited (system.log.sink_failure), and discarded — a downed collector cannot crash the caller. Throws only on an unknown level (config typo at the call site).

// Structured event with sensitive metadata — `apiKey` and
// `cardNumber` are redacted by pattern before any sink sees them.
b.logStream.emit("warn", "checkout retry", {
  orderId:    "ord_01HXYZ",
  attempt:    3,
  apiKey:     "",
  cardNumber: "",
});

b.logStream.debug(message, meta?) #

Convenience wrapper for emit("debug", ...). Records drop below minLevel (default "info") without serialization cost, so leaving debug() calls in production code is cheap.

b.logStream.debug("cache lookup", { key: "user:42", hit: false });

b.logStream.info(message, meta?) #

Convenience wrapper for emit("info", ...). Use for routine lifecycle events worth keeping in the operational log under default filtering.

b.logStream.info("worker ready", { pid: process.pid, queue: "checkout" });

b.logStream.warn(message, meta?) #

Convenience wrapper for emit("warn", ...). Use for recoverable anomalies the operator should notice but that don't fail the request — retry exhaustion below the cap, degraded-mode entry, cache misses on a hot key.

b.logStream.warn("retry succeeded after backoff", {
  route: "POST /checkout", attempts: 4, totalMs: 1820,
});

b.logStream.error(message, meta?) #

Convenience wrapper for emit("error", ...). Use for the failed- request / unhandled-exception class. b.audit remains the authoritative tamper-evident record for privileged actions; the log stream is operational telemetry.

b.logStream.error("dispatcher failure", {
  route: "POST /checkout", err: "ECONNRESET", upstream: "payments",
});

b.logStream.onIncoming(handler) #

Register a handler for inbound command-channel events. Returns an unsubscribe function. Handlers may be async and may return a value; deliverIncoming collects every handler's result and reports per-handler success / failure. Throws on a non-function argument.

var off = b.logStream.onIncoming(async function (payload) {
  if (payload.command === "raise-log-level") return { applied: true };
  return { applied: false };
});
// Later, when teardown is needed:
off();

b.logStream.deliverIncoming(payload, opts?) #

{
  actor:  { userId?, sessionId?, ip?, userAgent? },   // audit context
  source: string,                                     // transport name
}

Dispatch an inbound command-channel payload to every registered handler. The payload is redacted before audit and before handlers run, so even a noisy webhook receiver cannot smuggle secrets into the audit chain. Audit-logs the receipt under system.log.incoming BEFORE invoking handlers — handler exceptions never erase the receipt. Returns a per-handler [{ ok, value? | error? }] array; one handler throwing does not abort the rest.

var results = await b.logStream.deliverIncoming(
  { command: "rotate-sink", sink: "file" },
  { actor: { userId: "ops-42" }, source: "webhook" }
);
// → [{ ok: true, value: { applied: false } }]

b.logStream.shutdown() #

Drain pending fire-and-forget emits, close every sink (file fds, webhook keep-alive sockets, syslog connections, OTLP gRPC streams), and clear registered incoming handlers. Idempotent — safe to call twice. Records queued just before shutdown reach disk / the wire because in-flight Promises are tracked and awaited before close.

process.on("SIGTERM", async function () {
  await b.logStream.shutdown();
  process.exit(0);
});

b.logStream.listSinks() #

Return one descriptor per configured sink: { name, protocol, stats }. Sinks that expose a stats() method (file rotation counters, webhook batch metrics, OTLP queue depth) report through it; those that don't return null. Returns [] before init() runs, so health endpoints can call it unconditionally.

var snapshot = b.logStream.listSinks();
// → [{ name: "file", protocol: "local",
//      stats: { rotations: 2, bytesWritten: 4194304 } }]

b.logStream.bootFromEnv(opts?) #

{
  env: object,   // override process.env (testing / fixtures)
}

Operator-friendly env-driven init. Reads BLAMEJS_LOG_STREAM_* (and standard AWS_*) variables and constructs a single-sink configuration. Returns false and skips silently when BLAMEJS_LOG_STREAM_PROTOCOL is unset, so deployments that wire sinks through init() keep their existing config. Throws on an unknown protocol value.

Recognised variables: BLAMEJS_LOG_STREAM_PROTOCOL (local | webhook | otlp | cloudwatch), BLAMEJS_LOG_STREAM_MIN_LEVEL, BLAMEJS_LOG_STREAM_URL, BLAMEJS_LOG_STREAM_TOKEN, BLAMEJS_LOG_STREAM_SERVICE_NAME, BLAMEJS_LOG_STREAM_PATH, BLAMEJS_LOG_STREAM_CLOUDWATCH_LOG_GROUP, plus AWS_REGION / AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_SESSION_TOKEN.

// Operator sets BLAMEJS_LOG_STREAM_PROTOCOL=otlp and the URL in
// the deployment manifest; the framework wires the sink at boot.
var wired = b.logStream.bootFromEnv({
  env: {
    BLAMEJS_LOG_STREAM_PROTOCOL:     "otlp",
    BLAMEJS_LOG_STREAM_URL:          "https://collector.internal:4318/v1/logs",
    BLAMEJS_LOG_STREAM_SERVICE_NAME: "checkout",
    BLAMEJS_LOG_STREAM_MIN_LEVEL:    "info",
  },
});
// → true   (false when BLAMEJS_LOG_STREAM_PROTOCOL is unset)

Last updated 2026-08-08T16:39:49.652Z by seeder.