From a441ee5a65863a94e07337448995820ffdd1a810 Mon Sep 17 00:00:00 2001 From: autoantohaki Date: Wed, 21 Oct 2026 20:47:20 -0300 Subject: [PATCH 1/2] feat: follow files across rotation and truncation --- CHANGELOG.md | 1 + docs/architecture.md | 4 +- src/cli.tsx | 33 ++++++------ src/cli/options.ts | 43 ++++++++++++++- src/core/follow.ts | 113 ++++++++++++++++++++++++++++++++++++++++ src/core/sources.ts | 94 ++++++++++++++++++++++++++++++++- tests/cli.test.ts | 12 +++++ tests/follow.test.ts | 121 +++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 398 insertions(+), 23 deletions(-) create mode 100644 src/core/follow.ts create mode 100644 tests/follow.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 7867e36..fe22894 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,4 +3,5 @@ ## Unreleased - Added the `relay-logs` command, which prints files or standard input colored by level. +- Added `-f` to follow files across rotation and truncation, and `-n` to start with the last lines of each file. - Added `parseLine`, which recognizes text, JSON, and logfmt lines, and `sanitizeLine`, which removes terminal control sequences. diff --git a/docs/architecture.md b/docs/architecture.md index c4c0ef7..93b9c3d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -3,8 +3,8 @@ Relay Logs turns lines from files or standard input into log entries and displays them. 1. `LineSplitter` and `readLines` split byte streams into complete lines without breaking multibyte characters. -2. `openSources` opens each file, or standard input, as a labeled stream of lines. +2. `openSources` opens each file, or standard input, as a labeled stream of lines. `followFile` polls a file for appended lines and notices truncation and rotation by its size and identity. `mergeSources` interleaves followed files as their lines arrive. 3. `parseLine` sanitizes a line and recognizes its level, timestamp, and message. The original text is kept in `raw`, so nothing is lost when a line is not recognized. 4. `formatPlain` prints an entry for pipes and plain terminals. -Following files, filtering, and the interactive viewer build on these modules. +Filtering and the interactive viewer build on these modules. diff --git a/src/cli.tsx b/src/cli.tsx index 5d98f0d..ef65cba 100644 --- a/src/cli.tsx +++ b/src/cli.tsx @@ -2,7 +2,7 @@ import {helpText, parseCliOptions} from './cli/options.js'; import {parseLine} from './core/entry.js'; -import {openSources} from './core/sources.js'; +import {concatSources, type LogSource, mergeSources, openSources} from './core/sources.js'; import {formatPlain} from './format/plain.js'; import {relayLogsVersion} from './version.js'; @@ -35,23 +35,22 @@ const main = async (): Promise => { color: options.color ?? (process.stdout.isTTY && process.env.NO_COLOR === undefined), showSource: options.files.length > 1, }; + const sources = openSources(options.files, { + follow: options.follow, + ...(options.lines === undefined ? {} : {lines: options.lines}), + }); + const onError = (source: LogSource, error: unknown): void => { + process.stderr.write( + `relay-logs: ${source.label ?? 'stdin'}: ${error instanceof Error ? error.message : String(error)}\n`, + ); + process.exitCode = 1; + }; + const lines = options.follow ? mergeSources(sources, onError) : concatSources(sources, onError); let seq = 0; - for (const source of openSources(options.files)) { - try { - for await (const line of source.lines) { - seq += 1; - const entry = parseLine(line, { - seq, - ...(source.label === undefined ? {} : {source: source.label}), - }); - process.stdout.write(`${formatPlain(entry, format)}\n`); - } - } catch (error) { - process.stderr.write( - `relay-logs: ${source.label ?? 'stdin'}: ${error instanceof Error ? error.message : String(error)}\n`, - ); - process.exitCode = 1; - } + for await (const {label, line} of lines) { + seq += 1; + const entry = parseLine(line, {seq, ...(label === undefined ? {} : {source: label})}); + process.stdout.write(`${formatPlain(entry, format)}\n`); } }; diff --git a/src/cli/options.ts b/src/cli/options.ts index 63da3dd..988e63c 100644 --- a/src/cli/options.ts +++ b/src/cli/options.ts @@ -3,6 +3,10 @@ export interface CliOptions { version: boolean; /** Force color on or off. Undefined means color when standard output is a terminal. */ color: boolean | undefined; + /** Keep reading files as they grow. */ + follow: boolean; + /** Start with only the last lines of each file. */ + lines: number | undefined; files: string[]; } @@ -16,6 +20,8 @@ Reads log files, or standard input when no file is given, and prints each line c level. Options: + -f, --follow Keep printing lines as files grow, across rotation and truncation + -n, --lines Start with the last n lines of each file (10 when following) --color Always color output --no-color Never color output -h, --help Show this help @@ -23,16 +29,49 @@ Options: `; export const parseCliOptions = (arguments_: readonly string[]): CliOptions => { - const options: CliOptions = {help: false, version: false, color: undefined, files: []}; + const options: CliOptions = { + help: false, + version: false, + color: undefined, + follow: false, + lines: undefined, + files: [], + }; let onlyFiles = false; - for (const argument of arguments_) { + for (let index = 0; index < arguments_.length; index += 1) { + const argument = arguments_[index] ?? ''; + const [name, inlineValue] = splitOption(argument); + const value = (): string => { + if (inlineValue !== undefined) return inlineValue; + index += 1; + const next = arguments_[index]; + if (next === undefined) throw new Error(`${name} needs a value`); + return next; + }; if (onlyFiles || argument === '-' || !argument.startsWith('-')) options.files.push(argument); else if (argument === '--') onlyFiles = true; else if (argument === '-h' || argument === '--help') options.help = true; else if (argument === '-v' || argument === '--version') options.version = true; + else if (argument === '-f' || argument === '--follow') options.follow = true; + else if (name === '-n' || name === '--lines') options.lines = parseCount(name, value()); else if (argument === '--color') options.color = true; else if (argument === '--no-color') options.color = false; else throw new Error(`Unknown option: ${argument}`); } return options; }; + +const splitOption = (argument: string): [string, string | undefined] => { + const equals = argument.indexOf('='); + return argument.startsWith('--') && equals !== -1 + ? [argument.slice(0, equals), argument.slice(equals + 1)] + : [argument, undefined]; +}; + +const parseCount = (name: string, text: string): number => { + const count = Number(text); + if (!Number.isInteger(count) || count < 0) { + throw new Error(`${name} must be a whole number, not ${text}`); + } + return count; +}; diff --git a/src/core/follow.ts b/src/core/follow.ts new file mode 100644 index 0000000..ec92174 --- /dev/null +++ b/src/core/follow.ts @@ -0,0 +1,113 @@ +import {open, stat} from 'node:fs/promises'; +import {setTimeout as sleep} from 'node:timers/promises'; +import {LineSplitter} from './lines.js'; + +export interface FollowOptions { + /** Lines already in the file to yield before following. Defaults to 10. */ + initialLines?: number; + /** How often to check the file for changes, in milliseconds. Defaults to 250. */ + interval?: number; + /** Stop following when aborted. */ + signal?: AbortSignal; +} + +interface FileState { + identity: string; + offset: number; +} + +const blockSize = 64 * 1024; + +/** + * Yield the last lines of a file and then every line appended to it, like `tail -F`. + * + * Only complete lines are yielded, so a line still being written is held back until its line + * ending arrives. When the file is truncated the reader starts again from the beginning, and + * when it is replaced, as log rotation does, the new file is read from its first line. A file + * that does not exist yet is waited for. + */ +export async function* followFile( + path: string, + options: FollowOptions = {}, +): AsyncGenerator { + const interval = options.interval ?? 250; + const signal = options.signal; + let state: FileState | undefined; + let splitter = new LineSplitter(); + + while (signal?.aborted !== true) { + const info = await stat(path).catch(() => undefined); + if (info !== undefined) { + const identity = `${info.dev}:${info.ino}:${info.birthtimeMs}`; + if (state === undefined) { + const start = await readLastLines(path, options.initialLines ?? 10, info.size); + state = {identity, offset: start.offset}; + yield* start.lines; + } else if (identity !== state.identity || info.size < state.offset) { + // Rotated or truncated: the rest of the old line will never be completed. + state = {identity, offset: 0}; + splitter = new LineSplitter(); + } + if (info.size > state.offset) { + const chunk = await readRange(path, state.offset, info.size); + state.offset += chunk.length; + yield* splitter.push(chunk); + } + } + await sleep(interval, undefined, signal === undefined ? {} : {signal}).catch(() => undefined); + } +} + +/** + * Read the last complete lines of a file. The returned offset points just past the last line + * ending, so an unterminated final line is read again when it is completed. + */ +export const readLastLines = async ( + path: string, + count: number, + size: number, +): Promise<{lines: string[]; offset: number}> => { + const handle = await open(path, 'r'); + try { + let start = size; + let buffer = Buffer.alloc(0); + // Read backwards until the buffer holds more line endings than lines requested. + while (start > 0 && countNewlines(buffer) <= count) { + const next = Math.max(0, start - blockSize); + const block = Buffer.alloc(start - next); + await handle.read(block, 0, block.length, next); + buffer = Buffer.concat([block, buffer]); + start = next; + } + const lastNewline = buffer.lastIndexOf(0x0a); + if (lastNewline === -1) return {lines: [], offset: start}; + const complete = buffer.subarray(0, lastNewline + 1); + const splitter = new LineSplitter(); + let lines = splitter.push(complete); + // The first line is partial unless the read reached the start of the file. + if (start > 0) lines = lines.slice(1); + return { + lines: count === 0 ? [] : lines.slice(-count), + offset: start + lastNewline + 1, + }; + } finally { + await handle.close(); + } +}; + +const readRange = async (path: string, from: number, to: number): Promise => { + const handle = await open(path, 'r'); + try { + const buffer = Buffer.alloc(to - from); + const {bytesRead} = await handle.read(buffer, 0, buffer.length, from); + return buffer.subarray(0, bytesRead); + } finally { + await handle.close(); + } +}; + +const countNewlines = (buffer: Buffer): number => { + let count = 0; + for (const byte of buffer) if (byte === 0x0a) count += 1; + return count; +}; diff --git a/src/core/sources.ts b/src/core/sources.ts index 875f118..eeab190 100644 --- a/src/core/sources.ts +++ b/src/core/sources.ts @@ -1,5 +1,7 @@ import {createReadStream} from 'node:fs'; +import {stat} from 'node:fs/promises'; import {basename} from 'node:path'; +import {followFile, readLastLines} from './follow.js'; import {readLines} from './lines.js'; export interface LogSource { @@ -8,15 +10,103 @@ export interface LogSource { lines: AsyncIterable; } +export interface OpenSourcesOptions { + /** Keep reading files as they grow. */ + follow?: boolean; + /** Start with only the last lines of each file. Following defaults to 10. */ + lines?: number; + /** Polling interval used when following. */ + interval?: number; + signal?: AbortSignal; + stdin?: AsyncIterable; +} + /** Open files in order, treating `-` as standard input. With no files, read standard input. */ export const openSources = ( files: readonly string[], - stdin: AsyncIterable = process.stdin, + options: OpenSourcesOptions = {}, ): LogSource[] => { + const stdin = options.stdin ?? process.stdin; if (files.length === 0) return [{label: undefined, lines: readLines(stdin)}]; return files.map((file) => file === '-' ? {label: 'stdin', lines: readLines(stdin)} - : {label: basename(file), lines: readLines(createReadStream(file))}, + : {label: basename(file), lines: openFile(file, options)}, ); }; + +const openFile = (file: string, options: OpenSourcesOptions): AsyncIterable => { + if (options.follow === true) { + return followFile(file, { + ...(options.lines === undefined ? {} : {initialLines: options.lines}), + ...(options.interval === undefined ? {} : {interval: options.interval}), + ...(options.signal === undefined ? {} : {signal: options.signal}), + }); + } + if (options.lines !== undefined) return lastLines(file, options.lines); + return readLines(createReadStream(file)); +}; + +async function* lastLines(file: string, count: number): AsyncGenerator { + const {size} = await stat(file); + yield* (await readLastLines(file, count, size)).lines; +} + +export interface SourcedLine { + label: string | undefined; + line: string; +} + +/** + * Read several sources at once and yield their lines as they arrive. Lines from one source keep + * their order. A source that fails is reported through `onError` and the others continue. + */ +export async function* mergeSources( + sources: readonly LogSource[], + onError: (source: LogSource, error: unknown) => void, +): AsyncGenerator { + const iterators = sources.map((source) => source.lines[Symbol.asyncIterator]()); + const pending = new Map}>>(); + const next = (index: number): void => { + const iterator = iterators[index]; + if (iterator === undefined) return; + pending.set( + index, + iterator.next().then( + (result) => ({index, result}), + (error: unknown) => { + const source = sources[index]; + if (source !== undefined) onError(source, error); + return {index, result: {done: true, value: undefined}}; + }, + ), + ); + }; + iterators.forEach((_, index) => next(index)); + try { + while (pending.size > 0) { + const {index, result} = await Promise.race(pending.values()); + pending.delete(index); + if (result.done === true) continue; + yield {label: sources[index]?.label, line: result.value}; + next(index); + } + } finally { + // Do not wait: a source blocked on standard input would never finish returning. + for (const iterator of iterators) void iterator.return?.(); + } +} + +/** Read sources one after another, as `cat` does. A source that fails is reported and skipped. */ +export async function* concatSources( + sources: readonly LogSource[], + onError: (source: LogSource, error: unknown) => void, +): AsyncGenerator { + for (const source of sources) { + try { + for await (const line of source.lines) yield {label: source.label, line}; + } catch (error) { + onError(source, error); + } + } +} diff --git a/tests/cli.test.ts b/tests/cli.test.ts index e30899d..fa5d6d3 100644 --- a/tests/cli.test.ts +++ b/tests/cli.test.ts @@ -19,6 +19,11 @@ describe('CLI options', () => { }); }); + it('reads follow and line counts', () => { + expect(parseCliOptions(['-f', '--lines=5', 'a.log'])).toMatchObject({follow: true, lines: 5}); + expect(() => parseCliOptions(['-n', 'many'])).toThrow('whole number'); + }); + it('rejects unknown options', () => { expect(() => parseCliOptions(['--wat'])).toThrow('Unknown option'); }); @@ -39,6 +44,13 @@ describe('relay-logs command', () => { expect(stdout).toBe('api.log: INFO started\napi.log: ERROR failed\nworker.log: WARN slow\n'); }, 30_000); + it('starts with the last lines when asked', async () => { + const directory = mkdtempSync(join(tmpdir(), 'relay-logs-')); + writeFileSync(join(directory, 'app.log'), 'one\ntwo\nthree\n'); + const {stdout} = await run(process.execPath, [tsx, cli, '-n', '2', join(directory, 'app.log')]); + expect(stdout).toBe('two\nthree\n'); + }, 30_000); + it('colors by level when asked', async () => { const directory = mkdtempSync(join(tmpdir(), 'relay-logs-')); writeFileSync(join(directory, 'app.log'), 'ERROR failed\n'); diff --git a/tests/follow.test.ts b/tests/follow.test.ts new file mode 100644 index 0000000..55b4c09 --- /dev/null +++ b/tests/follow.test.ts @@ -0,0 +1,121 @@ +import {appendFileSync, mkdtempSync, renameSync, writeFileSync} from 'node:fs'; +import {tmpdir} from 'node:os'; +import {join} from 'node:path'; +import {Readable} from 'node:stream'; +import {describe, expect, it} from 'vitest'; +import {followFile, readLastLines} from '../src/core/follow.js'; +import {mergeSources} from '../src/core/sources.js'; + +const directory = (): string => mkdtempSync(join(tmpdir(), 'relay-logs-follow-')); + +/** Collect lines from a follower until `count` have arrived, running `steps` in between. */ +const collect = async ( + path: string, + count: number, + steps: (() => void)[], + initialLines = 10, +): Promise => { + const controller = new AbortController(); + const lines: string[] = []; + const timer = setInterval(() => steps.shift()?.(), 60); + try { + for await (const line of followFile(path, { + interval: 10, + initialLines, + signal: controller.signal, + })) { + lines.push(line); + if (lines.length === count) break; + } + } finally { + clearInterval(timer); + controller.abort(); + } + return lines; +}; + +describe('readLastLines', () => { + it('returns the last complete lines and the offset after them', async () => { + const path = join(directory(), 'app.log'); + writeFileSync(path, 'one\ntwo\nthree\npart'); + const result = await readLastLines(path, 2, 18); + expect(result).toEqual({lines: ['two', 'three'], offset: 14}); + }); + + it('reads files longer than one block', async () => { + const path = join(directory(), 'big.log'); + const lines = Array.from({length: 20_000}, (_, index) => `line ${index}`); + writeFileSync(path, `${lines.join('\n')}\n`); + const size = Buffer.byteLength(`${lines.join('\n')}\n`); + expect((await readLastLines(path, 3, size)).lines).toEqual([ + 'line 19997', + 'line 19998', + 'line 19999', + ]); + }); +}); + +describe('followFile', () => { + it('yields the tail and then appended lines, holding back unfinished ones', async () => { + const path = join(directory(), 'app.log'); + writeFileSync(path, 'old 1\nold 2\nold 3\n'); + const lines = await collect( + path, + 4, + [() => appendFileSync(path, 'new 1\nnew '), () => appendFileSync(path, '2\n')], + 2, + ); + expect(lines).toEqual(['old 2', 'old 3', 'new 1', 'new 2']); + }); + + it('starts over when the file is truncated', async () => { + const path = join(directory(), 'app.log'); + writeFileSync(path, 'a long first line\n'); + const lines = await collect(path, 2, [() => writeFileSync(path, 'short\n')]); + expect(lines).toEqual(['a long first line', 'short']); + }); + + it('reads a replacement file from the beginning after rotation', async () => { + const folder = directory(); + const path = join(folder, 'app.log'); + writeFileSync(path, 'before\n'); + const lines = await collect(path, 3, [ + () => renameSync(path, join(folder, 'app.log.1')), + () => writeFileSync(path, 'after 1\nafter 2 is longer\n'), + ]); + expect(lines).toEqual(['before', 'after 1', 'after 2 is longer']); + }); + + it('waits for a file that does not exist yet', async () => { + const path = join(directory(), 'late.log'); + const lines = await collect(path, 1, [() => writeFileSync(path, 'hello\n')]); + expect(lines).toEqual(['hello']); + }); +}); + +describe('mergeSources', () => { + it('interleaves sources and reports failures without stopping the others', async () => { + async function* slow(): AsyncGenerator { + yield 'a1'; + await new Promise((resolve) => setTimeout(resolve, 20)); + yield 'a2'; + } + const broken: AsyncIterable = { + [Symbol.asyncIterator]: () => ({next: () => Promise.reject(new Error('gone'))}), + }; + const errors: string[] = []; + const seen = []; + for await (const item of mergeSources( + [ + {label: 'a', lines: slow()}, + {label: 'b', lines: broken}, + {label: 'c', lines: Readable.from(['c1'])}, + ], + (source) => errors.push(source.label ?? ''), + )) { + seen.push(`${item.label ?? ''}:${item.line}`); + } + expect(seen).toEqual(['a:a1', 'c:c1', 'a:a2']); + expect(errors).toEqual(['b']); + }); +}); From d5a3c8d165c24fe33061ce9232314ca08ad13579 Mon Sep 17 00:00:00 2001 From: autoantohaki Date: Wed, 21 Oct 2026 20:47:42 -0300 Subject: [PATCH 2/2] docs: describe following files --- README.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/README.md b/README.md index 548526a..27ae7aa 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,17 @@ npm run dev 2>&1 | relay-logs When several files are given, each line is prefixed with the name of the file it came from. Use `-` to read standard input alongside files. +## Follow a log + +```sh +relay-logs -f app.log +relay-logs -f -n 50 api.log worker.log +``` + +`-f` keeps printing lines as files grow, like `tail -F`. It starts with the last 10 lines of each file, or the number given with `-n`. Lines appear only once their line ending has been written, so a line still being written is never split. When a file is truncated, Relay Logs reads it again from the start, and when it is replaced by log rotation, the new file is read from its first line. A file that does not exist yet is waited for. + +When several files are followed, their lines are interleaved in the order they arrive. + ## Recognized lines - Text lines with an optional leading timestamp and a level such as `ERROR`, `[warn]`, ``, or `FATAL:`