阅读中文版本

Build a Log Aggregator with Structured Storage

A log pipeline that ingests newline-delimited JSON from multiple writers, stores events in SQLite with generated columns, and answers filtered tail queries over a CLI and HTTP API.

log-aggregator/
├── src/
│   ├── server.js
│   ├── parser.js
│   ├── store.js
│   └── rotate.js
├── test/
│   ├── parser.test.js
│   └── api.test.js
├── package.json
└── README.md

function createLineSplitter(onLine) {
  let buffer = Buffer.alloc(0);
  return (chunk) => {
    buffer = Buffer.concat([buffer, chunk]);
    let idx;
    while ((idx = buffer.indexOf(10)) !== -1) {
      const line = buffer.subarray(0, idx);
      buffer = buffer.subarray(idx + 1);
      if (line.length) onLine(line);
    }
  };
}

db.exec(`
    CREATE TABLE IF NOT EXISTS events (
      id INTEGER PRIMARY KEY,
      ts TEXT NOT NULL,
      service TEXT NOT NULL,
      level TEXT NOT NULL,
      severity INTEGER GENERATED ALWAYS AS (
        CASE lower(level)
          WHEN 'trace' THEN 0 WHEN 'debug' THEN 1 WHEN 'info' THEN 2
          WHEN 'warn' THEN 3 WHEN 'error' THEN 4 ELSE 5 END),
      body TEXT NOT NULL
    );
    CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts, severity);
  `);

req.on('end', () => {
    const { accepted, rejected } = store.insertBatch(records);
    if (bytesReceived > config.maxBatchBytes) {
      res.writeHead(413).end(JSON.stringify({ error: 'batch too large' }));
      return;
    }
    res.writeHead(202).end(JSON.stringify({ accepted, rejected }));
  });

const rows = db.prepare(`
    SELECT ts, service, level, body FROM events
    WHERE severity >= ? AND service = COALESCE(?, service) AND ts >= ?
    ORDER BY id DESC LIMIT ?`);

Project overview

Structured logging pays off only if something collects it. This project builds the collector: an HTTP endpoint that drinks NDJSON from any number of services, normalizes it into queryable storage, and answers the one query everyone actually runs — show me the newest matching events.

The constraints that shape the design are unglamorous and real: writers send partial lines whenever TCP chunks land mid-line, batches must be all-or-nothing, and the store must stay queryable for weeks without a slow full scan. Each phase attacks exactly one of those.

The finished service delivers five capabilities:

• POST /ingest accepts batched NDJSON from many concurrent writers and responds with accepted and rejected counts.
• A generated column normalizes free-form level strings into a sortable severity integer at insert time.
• GET /tail streams the newest matching events with level, service, and since filters — without scanning unfiltered rows.
• Backpressure with a clear contract: a batch exceeding the configured size cap gets a 413 and nothing is stored.
• Log rotation archives the SQLite file on a size threshold while the server keeps accepting writes.

The architecture is a pipeline with one transactional sink: the server hands raw request chunks to a line splitter that feeds a validating parser, valid records enter a transactional batch insert, and queries run through prepared statements that push filtering into SQL with the timestamp index. Rotation works by moving the file to an archive name under an exclusive transaction boundary, so writers never observe a partially rotated database.

Setup

Verify the prerequisites before starting: Node.js 20 LTS or later, SQL basics including indexes and generated columns, and familiarity with NDJSON and streaming HTTP bodies.

The dependency footprint is one runtime package plus the test runner:

Create the project and install:

mkdir log-aggregator && cd log-aggregator
npm init -y
npm install better-sqlite3@^11.0.0
npm install -D vitest@^2.1.0

Configure the environment in package.json — the server entry and the test script; the database path, ingest port, batch cap, and rotation threshold all arrive as environment variables with sensible defaults:

{
  "name": "log-aggregator",
  "scripts": { "start": "node src/server.js", "test": "vitest run" }
}

The directory structure mirrors the pipeline: parser.js is pure byte-to-record logic, store.js owns SQLite, rotate.js owns the file lifecycle, and server.js is the only module that knows HTTP exists.

Phase 1: NDJSON parser

The parser splits incoming byte chunks on newlines across chunk boundaries, parses each line as JSON, and classifies records as valid or rejected with reasons.

// src/parser.js
export function createLineSplitter(onLine) {
  let buffer = Buffer.alloc(0);
  return (chunk) => {
    buffer = Buffer.concat([buffer, chunk]);
    let idx;
    while ((idx = buffer.indexOf(10)) !== -1) {
      const line = buffer.subarray(0, idx);
      buffer = buffer.subarray(idx + 1);
      if (line.length) onLine(line);
    }
  };
}

export function createParser(onRecord) {
  return createLineSplitter((line) => {
    try {
      const rec = JSON.parse(line.toString('utf8'));
      if (typeof rec.ts !== 'string' || typeof rec.service !== 'string') {
        onRecord({ ok: false, reason: 'missing ts or service' });
        return;
      }
      onRecord({ ok: true, record: rec });
    } catch (err) {
      onRecord({ ok: false, reason: 'malformed JSON: ' + err.message });
    }
  });
}

A naive read of request data assumes chunks align with lines, which TCP never promises — a writer flushing mid-line delivers half a JSON object in one chunk and the rest in the next. The splitter keeps a trailing buffer and only emits complete newline-terminated lines, and the tests split every payload at every single byte offset to prove it: no line lost, no line doubled, across all boundaries.

Multi-byte UTF-8 inside a JSON body is safe because Buffer.indexOf works on bytes — a UTF-8 sequence never contains the byte 0x0A — and JSON.parse always receives whole lines. Malformed lines are counted and skipped rather than aborting the batch, because one buggy producer must not block every other service from logging; the rejected count in the 202 response is the contract that makes producer-side drift visible.

Acceptance for this phase: parser tests feed split multi-byte-safe chunks and assert no line is lost or doubled across boundaries, with malformed lines reported not thrown.

Phase 2: Indexed storage

The store creates the events table with a severity generated column from level, an index on timestamp plus severity, and a transactional batch insert.

// src/store.js
import Database from 'better-sqlite3';

export function createStore(dbPath) {
  const db = new Database(dbPath);
  db.exec(`
    CREATE TABLE IF NOT EXISTS events (
      id INTEGER PRIMARY KEY,
      ts TEXT NOT NULL,
      service TEXT NOT NULL,
      level TEXT NOT NULL,
      severity INTEGER GENERATED ALWAYS AS (
        CASE lower(level)
          WHEN 'trace' THEN 0 WHEN 'debug' THEN 1 WHEN 'info' THEN 2
          WHEN 'warn' THEN 3 WHEN 'error' THEN 4 ELSE 5 END),
      body TEXT NOT NULL
    );
    CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts, severity);
  `);

  const insert = db.prepare(
    'INSERT INTO events (ts, service, level, body) VALUES (?, ?, ?, ?)');

  function insertBatch(records) {
    const insertAll = db.transaction((rows) => {
      let accepted = 0, rejected = 0;
      for (const rec of rows) {
        if (!rec.ok) { rejected += 1; continue; }
        insert.run(rec.record.ts, rec.record.service, rec.record.level ?? 'info',
                   JSON.stringify(rec.record));
        accepted += 1;
      }
      return { accepted, rejected };
    });
    return insertAll(records);
  }

  return { db, insertBatch };
}

The generated severity column turns level strings into sortable integers at insert time, so a WHERE severity >= 3 filter becomes an index range scan rather than a per-row CASE evaluation — and mixed-case inputs (WARN, warn, Warn) all normalize through lower(level) while the original casing stays queryable in the stored row. Unknown levels land at severity 5 rather than erroring, because a producer inventing CRITICAL should still have its logs stored.

The compound index on (ts, severity) matches the tail query shape, and the store test asserts the planner actually uses it via EXPLAIN QUERY PLAN — asserting the index exists is not the same as asserting the planner chooses it, and the latter is the property that survives schema drift. Batch inserts run inside one transaction, which is both the all-or-nothing contract and the performance mechanism: a thousand rows in one transaction is a thousand times cheaper than a thousand transactions.

Acceptance for this phase: inserting mixed-case levels stores once and queries by severity return rows whose original casing is preserved.

Phase 3: Ingest endpoint

The server wires the parser to the store, enforces the batch size cap, and answers 202 with counts or 413 when the body exceeds the cap.

// src/server.js
export function createIngestHandler(store, config) {
  return (req, res) => {
    const records = [];
    const parser = createParser((r) => records.push(r));
    let bytesReceived = 0;

    req.on('data', (chunk) => {
      bytesReceived += chunk.length;
      if (bytesReceived > config.maxBatchBytes) {
        req.destroy();               // stop reading an oversized body
        res.writeHead(413).end(JSON.stringify({ error: 'batch too large' }));
        return;
      }
      parser(chunk);
    });

    req.on('end', () => {
      const { accepted, rejected } = store.insertBatch(records);
      res.writeHead(202).end(JSON.stringify({ accepted, rejected }));
    });
  };
}

The size cap is enforced as bytes arrive, not after the body completes — a 2GB body against a 1MB cap must be refused after 1MB, not buffered in full first, and destroying the request stream is what actually stops the transfer. Nothing from an oversized batch is stored: the parser has run, but the store is never called, so the refusal is clean rather than partial.

The 202 status is deliberate over 200: the request was accepted and processed, and the counts in the body carry the truth about what happened — accepted: 998, rejected: 2 tells the producer exactly which batch to investigate, and monitoring on the rejected counter is the alert that producer-side schema drift has started.

Acceptance for this phase: a batch of 1000 events persists in one transaction and reports exact counts; a 2MB body over a 1MB cap receives 413 and nothing is stored.

Phase 4: Tail query and rotation

The tail endpoint streams filtered newest-first events, and rotation archives the active file at a transaction boundary when it crosses the size threshold.

// src/server.js — the tail query
const tailStmt = db.prepare(`
  SELECT ts, service, level, body FROM events
  WHERE severity >= ? AND service = COALESCE(?, service) AND ts >= ?
  ORDER BY id DESC LIMIT ?`);

export function tail({ minSeverity = 0, service, since, limit = 100 }) {
  return tailStmt.all(minSeverity, service ?? null, since ?? '0000', limit);
}

// src/rotate.js
export function rotateIfNeeded(db, dbPath, maxBytes) {
  const size = statSync(dbPath).size;
  if (size < maxBytes) return false;
  db.pragma('wal_checkpoint(TRUNCATE)');        // flush WAL into the main file
  const archive = dbPath + '.' + new Date().toISOString().replace(/[:.]/g, '-') + '.archive';
  db.exec('BEGIN EXCLUSIVE');                   // transaction boundary
  renameSync(dbPath, archive);
  db.exec('COMMIT');
  db.exec('VACUUM');                            // fresh file for the new epoch
  return true;
}

The tail query pushes every filter into SQL: severity range, service equality via COALESCE (null means no filter, and the parameterization keeps it one prepared statement), and timestamp floor — all served by the compound index. Ordering is by auto-increment id descending rather than timestamp: it is cheaper (the primary key is the row’s physical identity) and correct for event arrival order, while the timestamp filter still prunes by range.

Rotation is the seam where most hand-rolled collectors corrupt data. The sequence here is: checkpoint the WAL first so the archive is self-contained, move the file between transactions under an exclusive boundary so no writer observes a partially rotated database, and continue on a fresh file. Archives are immutable by policy, which makes shipping them to object storage a safe rsync-style operation and keeps the retention question external to the collector.

Acceptance for this phase: tail returns newest-first filtered events without scanning unfiltered rows, and rotation moves the file while ingest continues with zero rejected events.

Design trade-offs

SQLite over a log platform. Loki or Elasticsearch do more; they also cost a cluster. One file, indexed, queryable with plain SQL, and shippable as an archive is the right size for a team’s observability, and the limitation is named: retention is manual, disk grows with archive policy.

Generated severity column over a lookup at query time. Normalizing levels at query time keeps the schema simpler and costs a CASE per row per query, forever. Normalizing at insert time pays once per event and makes the filter an index range — and the mixed-case test pins that original casing survives.

Reject-and-count over reject-and-fail. A strict collector that 400s a whole batch on one malformed line turns one buggy producer into an outage for every producer sharing the endpoint. Counting and skipping isolates the blast radius to the offending lines and surfaces drift as data.

id ordering over ts ordering in tail. Timestamps from different producers skew; arrival order is the truth the collector actually owns, and the id column is free.

Testing strategy

Run npx vitest. Parser tests split payloads at every byte boundary — a helper that emits the same NDJSON input chunked at offset 1, 2, 3, and so on — to prove chunk-safety as a property rather than a set of lucky examples. Store tests insert mixed-case levels and assert severity ordering and index usage via EXPLAIN QUERY PLAN, so a schema change that silently degrades the planner fails a test, not a production query.

API tests drive ingest batches including one over the size cap (asserting the 413 and that nothing was stored) and tail queries with each filter independently and combined. The rotation test is the capstone: it triggers rotation mid-ingest and reconciles counts across the active and archived files — no data loss in the seam is the property the whole phase 4 design exists to deliver, and the test is the only thing that keeps it delivered.

The byte-boundary helper itself deserves a comment in the code: it takes a payload and an offset, emits two chunks split exactly there, and feeds them through the splitter in order. Its simplicity is the point — proving a property at every offset is a loop over a helper, not a matrix of hand-written fixtures, and the same helper serves the ingest endpoint tests unchanged.

Troubleshooting playbook

Lines lost when writers flush mid-line. Every Nth line vanishes from storage. The cause is a splitter that scans for newlines without first concatenating the pending partial line — each chunk is scanned in isolation, so a line spanning two chunks is never emitted. The fix is the buffer-then-scan order in phase 1, and the byte-offset test matrix fails with the exact lost line as soon as the order regresses.

Queries slow down as the file grows. Tail latency climbs from milliseconds to seconds over weeks. The cause is a filter the planner cannot serve from the index — typically a CASE inside WHERE evaluating severity per row, or an index that no longer covers the predicate columns. The fix is the generated column in predicates and the compound index, and the EXPLAIN QUERY PLAN assertion in the store tests is the tripwire that fires when a refactor reintroduces a scan.

Rotation corrupts the database. After a rotation, the archive is unreadable or rows are missing. The cause is copying or moving the file while a write transaction is open — the moved snapshot is mid-write. The fix is the phase 4 sequence: WAL checkpoint first, exclusive transaction boundary around the move, and the mid-ingest rotation test that fails with the exact missing count.

Extending the project

The pipeline stages are the extension points. Structured field extraction — pulling a JSON body’s fields into queryable columns — is one more generated column or a JSON1 virtual table over body. Multiple services’ retention policies are archive-prefix conventions, since archives are immutable and self-contained. A WebSocket tail that pushes new events as they insert is a change-feed on the insert path, and the single-writer design keeps ordering trivial. Alerting is a consumer of the rejected counter, not a feature of the collector.

For deployment, run as a sidecar or central collector on persistent storage. Point every service at the ingest endpoint with batching — 100 to 1000 events per request — to amortize transactions. Configure rotation by max file size and ship archives to object storage; alert on rejected counts, which indicate producer-side schema drift rather than collector failure. The NDJSON choice deserves its closing argument, because it is the contract every producer codes against. One JSON object per line means a malformed line poisons exactly one event — the parser rejects it, counts it, and moves to the next newline — where a single malformed document in a JSON array stream poisons the whole batch and forces the producer to resend everything. Line delimitation also gives partial-tolerance for free: a truncated HTTP body still yields every complete line received, and the rejected count reports the tail. Producers that adopt structured logging for this collector should keep events flat where possible, put the searchable dimensions (service, level, trace ids) in fixed fields, and leave the variable payload in the body — because the generated severity column and the service index reward fixed fields with fast filters, and the body remains greppable with SQLite’s JSON functions when the query was not anticipated.