Follow files across rotation and truncation

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.

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]`, `<info>`, or `FATAL:`

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.

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<void> => {
     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`);
   }
 };
 

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,23 +20,58 @@ 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 <n>    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
   -v, --version      Show the version
 `;
 
 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;
+};

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<string> {
+  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<Buffer> => {
+  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;
+};

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<string>;
 }
 
+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<Buffer | string>;
+}
+
 /** Open files in order, treating `-` as standard input. With no files, read standard input. */
 export const openSources = (
   files: readonly string[],
-  stdin: AsyncIterable<Buffer | string> = 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<string> => {
+  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<string> {
+  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<SourcedLine> {
+  const iterators = sources.map((source) => source.lines[Symbol.asyncIterator]());
+  const pending = new Map<number, Promise<{index: number; result: IteratorResult<string>}>>();
+  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<SourcedLine> {
+  for (const source of sources) {
+    try {
+      for await (const line of source.lines) yield {label: source.label, line};
+    } catch (error) {
+      onError(source, error);
+    }
+  }
+}