Sinks

A sink is where events go. The interface is four members, one required:

interface Sink {
  emit(event: LogEvent): void;                 // synchronous; never throw for ordinary failures
  flush?(): Promise<void> | void;              // send what is buffered
  close?(): Promise<void> | void;              // flush, then release resources
  restrictedToMinimumLevel?: LogLevel | LevelSwitch;
  audit?: boolean;                             // failures propagate to the caller
  name?: string;                               // for selfLog messages
}

A logger dispatches to every sink in order. A sink that throws is reported to selfLog (or onSinkError) and the others still run — unless it is an audit sink, whose error is yours.

Console#

import { consoleSink } from '@unhingged/logit';

consoleSink();                                  // pretty on a terminal, JSON on a pipe
consoleSink({ format: 'json' });
consoleSink({ format: 'clef' });
consoleSink({ format: 'text', outputTemplate: '{Timestamp:HH:mm:ss} [{Level:u3}] {Message:lj}{NewLine}{Exception}' });
consoleSink({ stderrFrom: 'error' });           // error and fatal to stderr

Off Node — where there is no process.stdout — the sink writes through console.*. In a real browser (a document exists) the pretty format calls console.info / warn / error with the rendered message, the extra properties as an object and the error as an error, so the devtools keep them expandable. In the Next.js edge runtime, a worker or any other runtime without a document, the default is JSON lines through console.log, which is what those platforms collect. A named or custom format logs the formatted line in either case.

Stream#

Anything with write(string): a file stream, a socket, a child process' stdin.

import { streamSink } from '@unhingged/logit';
streamSink(process.stderr, { format: 'json' });

Same options as the console sink minus stderrFrom. If write() returns false, flush() waits for 'drain'.

File — @unhingged/logit/node#

import { fileSink } from '@unhingged/logit/node';

fileSink({ path: 'logs/app.log' });
fileSink({ path: 'logs/app.log', rolling: 'day', retainedFileCount: 14 });
fileSink({ path: 'logs/app.log', maxBytes: 50 * 1024 * 1024, retainedFileCount: 10 });
fileSink({ path: 'logs/app.log', format: 'text', outputTemplate: '{Timestamp:o} [{Level:u3}] {Message:lj}{NewLine}{Exception}' });

Writes are synchronous (fs.writeSync): nothing is lost when the process dies, at the cost of one syscall per event. That is fine for the usual thousands of events a second; for more, or for a disk you do not trust to be fast, put the file behind a sub-logger with a filter, or wait for the buffered variant on the roadmap. close() closes the descriptor.

Memory and callback#

import { memorySink, callbackSink } from '@unhingged/logit';

const mem = memorySink({ capacity: 1000 }); // keeps the last 1000
mem.events;                                  // LogEvent[]
mem.messages();                              // rendered messages
mem.clear();

callbackSink((event) => metrics.count(event.level));

memorySink is the testing tool, and the "last N events" page every ops team wants eventually.

HTTP#

Batched POSTs to any endpoint:

import { httpSink } from '@unhingged/logit';

httpSink({
  url: 'https://logs.example.com/ingest',
  headers: { authorization: `Bearer ${token}` },   // or () => ({ … }) for rotating tokens
  body: 'ndjson',                                  // 'ndjson' | 'array' | 'clef' | (events) => string
  batchSize: 100,
  flushInterval: 2000,
});

The request is sent with keepalive: true; in a browser the queue is flushed on pagehide, so the last events of a session leave. The batching behaviour is shared with the Seq and OTLP sinks; the timer is unref'd, so it never keeps a process alive — which is why you close before exiting.

Seq#

import { seqSink } from '@unhingged/logit';
seqSink({ serverUrl: 'http://localhost:5341', apiKey: process.env.SEQ_API_KEY });

CLEF over HTTP to /ingest/clef with the X-Seq-ApiKey header; the batching options above apply. See Seq.

OpenTelemetry — @unhingged/logit/otel#

import { otlpHttpSink, otelBridgeSink } from '@unhingged/logit/otel';
otlpHttpSink({ url: 'http://localhost:4318/v1/logs', resource: { 'service.name': 'shop' } });
otelBridgeSink(logs.getLoggerProvider());

See OpenTelemetry.

Wrappers#

import { restricted, conditional, audit, mapSink, Matching } from '@unhingged/logit';

restricted(sink, 'warn');                                   // only warn and above
conditional(Matching.fromSource('billing'), billingSink);   // only matching events
audit(ledgerSink);                                          // its failure is the caller's
mapSink((e) => e.properties.TenantId as string, (tenant) => fileSink({ path: `logs/${tenant}.log` }), { maxSinks: 50 });

• conditional is Serilog's WriteTo.Conditional; any predicate works, the Matching helpers are the usual ones.
• audit is Serilog's AuditTo: a sink whose failure must not be swallowed — an audit ledger, a compliance store. The logging call throws. auditSinks: [...] in the options is the same thing.
• mapSink is Serilog.Sinks.Map: one sink per key, created on first sight, the least recently used closed beyond maxSinks (default 100). keyOf returning undefined skips the event.

Sub-loggers#

A Logger is a Sink. Put one in another's sinks and events are re-dispatched through its own minimum level, overrides, bound properties, enrichers, filters and sinks — Serilog's WriteTo.Logger:

const errorsToSeq = createLogger({
  minimumLevel: 'error',
  filters: [byExcluding(Matching.fromSource('health'))],
  sinks: [seqSink({ serverUrl: '…' })],
});

configure({ sinks: [consoleSink(), errorsToSeq] });

With the fluent builder: .writeTo.logger((lc) => lc.minimumLevel.error().writeTo.seq({ … })). Destructuring options of the inner logger do not apply — it receives events already captured (Serilog's rule too).

Batching, flush and close#

Network sinks queue and send in batches; everything else writes as it goes. logger.flush() asks every sink to send what it holds and waits; logger.close() flushes and then closes each sink (descriptors, timers). Both tolerate a failing sink — reported to selfLog — except an audit sink, whose failure is thrown.

await log.flush();        // before a serverless function returns
await closeAndFlush();    // the global logger, at exit
closeOnExit(log);         // Node: do it on beforeExit / SIGINT / SIGTERM, log crashes first

To build a batching sink of your own, batchingSink(send, options) is exported: give it send(events): Promise<void> and it does the queue, the timer, the retries and the bounds.

Writing a sink#

import type { Sink } from '@unhingged/logit';

export function metricsSink(): Sink {
  return {
    name: 'metrics',
    emit(event) {
      counters.increment(`log.${event.level}`);
      if (event.error) counters.increment(`error.${event.error.name}`);
    },
  };
}

Read event.properties (captured, safe), event.renderMessage({ literal: true }) for text, event.messageTemplate, event.eventId, event.errorProperties for the error as data. Format with any formatter.