Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | 24x 24x 24x 24x 24x 24x 2x 5x 1x 24x 24x 24x 24x 24x 24x 31x 31x 24x 24x 24x 23x 1x 16x 16x 16x 3x 3x 3x 15x 1x 4x 4x 4x 4x | /**
* Streaming log template miner — Node.js Transform stream wrapper.
*
* Extends TemplateMiner with a `Transform` stream interface for processing
* large log files, stdin, or HTTP response streams in real time.
*
* Features:
* - Backpressure-aware: respects downstream consumer speed
* - Line-delimited: splits input on newlines
* - JSON output: each chunk is a JSON object with template + metadata
* - Inherits all TemplateMiner capabilities (masking, persistence, profiling)
*
* Usage:
* ```typescript
* import { DrainStream } from "@agentix-e/drain-ts";
* import { createReadStream } from "node:fs";
*
* createReadStream("app.log")
* .pipe(new DrainStream())
* .on("data", (result) => console.log(result.templateMined));
* ```
*
* @module
*/
import { Transform, type TransformCallback } from "node:stream";
import { TemplateMiner } from "./TemplateMiner.js";
import type { TemplateMinerConfig } from "./TemplateMinerConfig.js";
import type { PersistenceHandler } from "./persistence/PersistenceHandler.js";
/**
* Options for `DrainStream`.
*/
export interface DrainStreamOptions {
/** TemplateMiner configuration (defaults used if omitted). */
config?: TemplateMinerConfig;
/** Optional persistence handler. */
persistenceHandler?: PersistenceHandler | null;
/** Whether to emit JSON objects (default) or strings. */
objectMode?: boolean;
}
/**
* A Node.js Transform stream that processes log lines through Drain
* and outputs structured results.
*
* Each input chunk is treated as one or more log lines (split on `\n`).
* Each output chunk is a JSON-serializable `AddLogResult` object
* (in objectMode) or a JSON string.
*
* The stream handles backpressure naturally through Node.js stream
* mechanics — it won't read more input until the consumer is ready.
*
* @example
* ```typescript
* // Pipe a log file through DrainStream
* import { createReadStream, createWriteStream } from "node:fs";
* import { DrainStream } from "@agentix-e/drain-ts";
*
* const stream = new DrainStream({
* config: TemplateMinerConfig.from({ simTh: 0.5 }),
* });
*
* createReadStream("huge.log")
* .pipe(new SplitStream()) // optional: chunk into lines
* .pipe(stream)
* .pipe(createWriteStream("templates.jsonl"));
* ```
*/
export class DrainStream extends Transform {
private readonly _miner: TemplateMiner;
private _buffer: string = "";
private _lineCount: number = 0;
/**
* Creates a DrainStream.
*
* @param options - Configuration for the underlying TemplateMiner.
*/
constructor(options: DrainStreamOptions = {}) {
super({
readableObjectMode: options.objectMode !== false,
writableObjectMode: false, // always accept Buffer/string input
});
const minerOpts: { config?: TemplateMinerConfig; persistenceHandler?: PersistenceHandler | null } = {
persistenceHandler: options.persistenceHandler ?? null,
};
if (options.config) minerOpts.config = options.config;
this._miner = new TemplateMiner(minerOpts);
}
/**
* Access the underlying TemplateMiner for direct API calls
* (e.g., `match()`, `extractParameters()`).
*/
get miner(): TemplateMiner {
return this._miner;
}
/**
* Total log lines processed so far.
*/
get lineCount(): number {
return this._lineCount;
}
/**
* Current cluster count in the model.
*/
get clusterCount(): number {
return this._miner.drain.idToCluster.size;
}
// ============================================================
// Transform implementation
// ============================================================
override _transform(
chunk: Buffer | string,
_encoding: BufferEncoding,
callback: TransformCallback,
): void {
try {
const data = typeof chunk === "string" ? chunk : chunk.toString("utf-8");
this._buffer += data;
// Process complete lines
const lines = this._buffer.split("\n");
// Last element may be incomplete — keep in buffer
this._buffer = lines.pop() ?? "";
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed) continue; // skip empty lines
const result = this._miner.addLogMessage(trimmed);
this._lineCount++;
this.push(result);
}
callback();
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
override _flush(callback: TransformCallback): void {
try {
// Process any remaining partial line
const trimmed = this._buffer.trim();
if (trimmed) {
const result = this._miner.addLogMessage(trimmed);
this._lineCount++;
this.push(result);
}
callback();
} catch (err) {
callback(err instanceof Error ? err : new Error(String(err)));
}
}
}
/**
* Convenience: creates a DrainStream from a config and optional persistence.
*
* Equivalent to `new DrainStream({ config, persistenceHandler })`.
*/
export function createDrainStream(
config?: TemplateMinerConfig,
persistenceHandler?: PersistenceHandler | null,
): DrainStream {
const opts: DrainStreamOptions = {};
if (config) opts.config = config;
if (persistenceHandler !== undefined) opts.persistenceHandler = persistenceHandler;
return new DrainStream(opts);
}
|