createDrain()

Create a Drain instance for incremental or streaming log processing, when you need to feed lines in over time instead of all at once.

Create a Drain instance for incremental or streaming log processing. Use this when log lines arrive over time, or when you want to inspect templates between batches.

Signature

function createDrain(options?: DrainOptions): Drain

The Drain class itself is also exported, for instanceof checks and subclassing:

import { Drain, createDrain } from 'logpare';

const drain = new Drain({ depth: 5 });   // equivalent to createDrain({ depth: 5 })

Parameters

`options`

Type: DrainOptions
Required: No

Configuration options for the Drain algorithm. Unlike compress(), these are passed flat — createDrain() takes DrainOptions directly, not the nested { drain: { ... } } shape.

`options.depth`

Type: number
Default: 4

Parse tree depth. Higher values create more specific templates.

`options.simThreshold`

Type: number
Default: 0.4 (from the parsing strategy)
Range: 0.0 to 1.0

Similarity threshold for template matching. When omitted, the parsing strategy stays authoritative and may vary the threshold by depth. When supplied, this value overrides the strategy for every depth.

`options.maxChildren`

Type: number
Default: 100

Maximum children per parse tree node. Once a node is at capacity, further tokens collapse into a single wildcard branch.

`options.maxClusters`

Type: number
Default: 1000

Maximum total templates allowed. Once the cap is reached, lines that do not match an existing template are discarded and counted in stats.droppedLines.

`options.maxSamples`

Type: number
Default: 3

Maximum sample variables per template.

`options.preprocessing`

Type: ParsingStrategy
Default: Built-in strategy

Custom preprocessing strategy. See Custom Preprocessing.

`options.onProgress`

Type: ProgressCallback
Default: undefined

Progress callback. Only addLogLines() emits progress events; addLogLine() does not.

Return Value

Returns a Drain instance with the following public members.

`addLogLine(line: string): LogCluster | null`

Process a single log line. Returns the cluster the line was assigned to, or null when the line was blank, tokenized to nothing, or dropped because maxClusters was reached.

drain.addLogLine('ERROR Connection failed');

LogCluster is an internal class. It is not exported from logpare, and its shape is not part of the public API. Read templates through getTemplates() or getResult() instead of using the returned value.

`addLogLines(lines: string[]): void`

Process multiple log lines at once, emitting progress events if onProgress was supplied.

drain.addLogLines([
  'ERROR Connection failed',
  'INFO Request completed',
]);

`getTemplates(): Template[]`

Get all discovered templates, in discovery order (not sorted by frequency).

for (const template of drain.getTemplates()) {
  console.log(`[${template.occurrences}x] ${template.pattern}`);
}

See the Template interface for every available field.

`getResult(format?: OutputFormat, maxTemplates?: number): CompressionResult`

Get compression results, with templates sorted by occurrence count (descending) and truncated to maxTemplates.

format — 'summary' (default), 'detailed', 'json', or 'json-stable'
maxTemplates — default 50

const result = drain.getResult('detailed');
console.log(result.formatted);

Unlike compress(), getResult() does not populate stats.processingTimeMs — the Drain instance does not own the timing window.

`totalLines: number`

Read-only getter. Number of non-blank lines processed so far.

`totalClusters: number`

Read-only getter. Number of templates discovered so far.

console.log(`${drain.totalClusters} templates from ${drain.totalLines} lines`);

Examples

Incremental Processing

import { createDrain } from 'logpare';

const drain = createDrain({
  depth: 4,
  simThreshold: 0.4,
});

// Process logs one at a time
drain.addLogLine('ERROR Connection to 192.168.1.1 failed');
drain.addLogLine('ERROR Connection to 192.168.1.2 failed');
drain.addLogLine('INFO Request abc123 completed');

// Get results
const result = drain.getResult('summary');
console.log(result.formatted);

Streaming Processing

import { createDrain } from 'logpare';
import { createReadStream } from 'node:fs';
import { createInterface } from 'node:readline';

const drain = createDrain({
  depth: 5,
  simThreshold: 0.4,
  maxClusters: 500,
});

const rl = createInterface({
  input: createReadStream('/var/log/syslog'),
  crlfDelay: Infinity,
});

let lineCount = 0;

rl.on('line', (line) => {
  drain.addLogLine(line);
  lineCount++;

  if (lineCount % 1000 === 0) {
    console.log(`Processed ${lineCount} lines...`);
  }
});

rl.on('close', () => {
  const result = drain.getResult('detailed');
  console.log(result.formatted);
});

Batch Processing with Progress

import { createDrain } from 'logpare';

const drain = createDrain({
  onProgress: (event) => {
    console.log(`${event.currentPhase}: ${event.processedLines} lines`);
    if (event.percentComplete !== undefined) {
      console.log(`Progress: ${event.percentComplete.toFixed(1)}%`);
    }
  }
});

// Process in batches
const batchSize = 1000;
for (let i = 0; i < logs.length; i += batchSize) {
  const batch = logs.slice(i, i + batchSize);
  drain.addLogLines(batch);
}

const result = drain.getResult();

Custom Preprocessing

import { createDrain, defineStrategy, WILDCARD } from 'logpare';

const customStrategy = defineStrategy({
  preprocess(line: string): string {
    // Custom masking for domain-specific patterns
    return line
      .replace(/order-[A-Z0-9]{8}/g, WILDCARD)
      .replace(/user_\d+/g, WILDCARD);
  },

  tokenize(line: string): string[] {
    // Custom tokenization
    return line.split(/\s+/).filter(Boolean);
  },

  getSimThreshold(depth: number): number {
    // Depth-dependent threshold
    return depth <= 2 ? 0.3 : 0.4;
  }
});

const drain = createDrain({
  preprocessing: customStrategy,
  depth: 5,
});

drain.addLogLines(logs);
const result = drain.getResult();

Inspecting Templates Directly

import { createDrain } from 'logpare';

const drain = createDrain();
drain.addLogLines(logs);

// getTemplates() returns discovery order — sort it yourself
const sorted = drain.getTemplates().sort((a, b) => b.occurrences - a.occurrences);

sorted.slice(0, 10).forEach((template, i) => {
  console.log(`${i + 1}. [${template.occurrences}x] ${template.pattern}`);
  console.log(`   Severity: ${template.severity}`);
  console.log(`   URLs: ${template.urlSamples.join(', ')}`);
});

Real-time Monitoring

import { createDrain } from 'logpare';
import { spawn } from 'node:child_process';

const drain = createDrain({
  depth: 4,
  simThreshold: 0.3,
});

// Monitor live logs
const tail = spawn('tail', ['-f', '/var/log/app.log']);

tail.stdout.on('data', (data) => {
  const lines = data.toString().split('\n').filter(Boolean);
  drain.addLogLines(lines);
});

// Report every 10 seconds
setInterval(() => {
  const templates = drain.getTemplates();
  console.log(`\n=== Current State: ${templates.length} templates ===`);

  const errors = templates.filter((t) => t.severity === 'error');

  if (errors.length > 0) {
    console.log('Error templates:');
    errors.slice(0, 5).forEach((t) => {
      console.log(`  [${t.occurrences}x] ${t.pattern}`);
    });
  }
}, 10000);

Watch for dropped lines

The Drain instance is memory-bounded by maxClusters. Once the cap is hit, unmatched lines are thrown away rather than creating a new template, and compressionRatio would otherwise look better than reality. Always check droppedLines:

const result = drain.getResult();

if ((result.stats.droppedLines ?? 0) > 0) {
  console.warn(
    `${result.stats.droppedLines} lines dropped — raise maxClusters for full coverage`
  );
}

The summary and detailed formatters print this warning for you.

When to Use

Use createDrain() when:

Log lines arrive over time (streaming, tailing, polling)
You want to inspect templates between batches
You are feeding a very large file through in chunks
You need totalLines / totalClusters while processing

Use compress() or compressText() when:

You already have the complete set of lines
You want a single call that also reports processingTimeMs