Metrics Aggregation System

In the previous two cases, every byte produced was worth storing: an object was kept exactly as written, a video segment was read many times after it was produced. This case removes that assumption. Here records are produced hundreds of thousands of times per second, no single one is meaningful on its own, and storing all of them indefinitely is neither possible nor necessary.

The system is a metrics aggregation system: nodes send numeric samples at regular intervals, the samples are written as a time series, and questions are always asked over a time range. The decision to be measured is the joint effect of downsampling and rollup: stored bytes and scanned rows fall, but the answer read from rolled-up data is not equal to the answer read from raw data. This gap depends on the question type, and the design chooses that gap deliberately.

Which metrics are worth collecting is not this case’s subject; that question was settled in the Performance Anti-Patterns and Monitoring course’s Instrumentation Design lesson. Here the metric set is a given, and the design’s job is to carry it.

Constraints

Functional requirement: accepting samples, reading by series and time range, answering average / max / percentile / threshold-breach questions over a series, and dropping data whose retention period has expired.

Non-functional requirement, with numbers: the write path accepts a peak of 480,000 samples/s without loss; total stored bytes do not exceed 40 TB; a six-hour dashboard query scans fewer than 400 rows per series; over the last 7 days, percentile and threshold-breach questions are answered without error.

Scope narrowing: alerting rules, the dashboard interface, monitoring categories, and distributed trace collection are not designed here — these are the subject of the Performance Anti-Patterns and Monitoring course.

Assumptions

Code	Assumption	Value	Rationale
OT1	monitored node	20,000	sum of running processes
OT2	series per node	120	a few metrics per process, resource, and endpoint
OT3	sampling interval	10 s	the agent’s send frequency
OT4	raw sample record	26 bytes	series ID, time, and value
OT5	rollup record	34 bytes	bucket, count, total, and max
OT6	peak multiplier	2	the metric stream is regular, its peak is shallow
OT7	raw retention	7 days	incident review looks back one week
OT8	one-minute rollup retention	90 days	quarter-over-quarter comparison
OT9	five-minute rollup retention	730 days	capacity trend goes back two years
OT10	dashboard window	6 hours	one shift’s window

OT6 is this case’s distinguishing assumption: the peak multiplier was 3 in earlier cases, here it is 2. The metric stream depends on the agent’s scheduler, not on user behavior, so its peak is shallow.

Scale

// metrics/scale.mjs — scale calculation derived from the OT table and the total across storage tiers
const OT = { node: 20_000, seriesPerNode: 120, intervalS: 10, rawBytes: 26, rollupBytes: 34,
  peak: 2, queryWindowH: 6, rawDays: 7, minuteDays: 90, fiveMinuteDays: 730 };
const series = OT.node * OT.seriesPerNode;
const samplesS = series / OT.intervalS;
const rawDaily = samplesS * 86_400 * OT.rawBytes;
const rollupDaily = (grainS) => (series * (86_400 / grainS)) * OT.rollupBytes;

for (const [name, d] of [
  ["series count", series],
  ["average samples/s", samplesS],
  ["peak samples/s", samplesS * OT.peak],
  ["raw daily GB", rawDaily / 1e9],
  ["1 min rollup daily GB", rollupDaily(60) / 1e9],
  ["5 min rollup daily GB", rollupDaily(300) / 1e9],
]) console.log(name.padEnd(24) + d.toFixed(2).padStart(12));

const tiers = [["raw", rawDaily, OT.rawDays], ["1 min rollup", rollupDaily(60), OT.minuteDays],
  ["5 min rollup", rollupDaily(300), OT.fiveMinuteDays]];
let total = 0;
console.log(`\n${"tier".padEnd(15)}${"days".padStart(6)}${"stored TB".padStart(14)}`);
for (const [name, days, retention] of tiers) {
  total += (days * retention) / 1e12;
  console.log(name.padEnd(15) + String(retention).padStart(6) + ((days * retention) / 1e12).toFixed(2).padStart(14));
}
const allRaw = (rawDaily * OT.fiveMinuteDays) / 1e12;
console.log(`total ${total.toFixed(2)} TB; if everything were kept raw for ${OT.fiveMinuteDays} days it would be ` +
  `${allRaw.toFixed(2)} TB (x${(allRaw / total).toFixed(2)})`);
console.log(`${OT.queryWindowH}-hour query, rows read per series: raw ` +
  `${(OT.queryWindowH * 3600) / OT.intervalS}, 1 min ${OT.queryWindowH * 60}, ` +
  `5 min ${(OT.queryWindowH * 3600) / 300}`);

series count              2400000.00
average samples/s          240000.00
peak samples/s             480000.00
raw daily GB                  539.14
1 min rollup daily GB         117.50
5 min rollup daily GB          23.50

tier             days     stored TB
raw                 7          3.77
1 min rollup       90         10.58
5 min rollup      730         17.16
total 31.50 TB; if everything were kept raw for 730 days it would be 393.57 TB (x12.49)
6-hour query, rows read per series: raw 2160, 1 min 360, 5 min 72

These numbers are in the calculation class. Three of them determine the design. The write path sees 240,000 samples/s on average, 480,000 at peak; this is two orders of magnitude higher than the earlier cases where one request corresponded to one record, and it makes the per-sample cost of every step in the write path decisive. The storage tier total is 31.50 TB, under the 40 TB limit; if the same data were kept raw it would be 393.57 TB, that is, 12.49 times as much, and the limit would be violated. The six-hour query reads 360 rows from the one-minute rollup, 2160 from the raw tier; only the rollup tiers meet the 400-row limit.

The Cost of Downsampling

The table above shows only the gain. The cost is in the answer itself, and it is measured in a real engine. The model below does not stand up a cluster; it generates six hours of a single series inside node:sqlite, rolls it up into two grains, and asks the same four questions of all three formats at once.

// metrics/downsampling.mjs — a single series' raw and rolled-up forms in a real engine via
// node:sqlite. The generator is hand-written, the seed is visible; what is measured is stored
// bytes, scanned rows, and the gap the same four questions get between the two forms. The model
// is written as what it is.
import { DatabaseSync } from "node:sqlite";

const DURATION = 21_600, SEED = 20260801, THRESHOLD = 200;        // 6 hours, one sample per second
let state = SEED;
const rand = () => { state = (state * 1103515245 + 12345) % 2147483648; return state / 2147483648; };
const generate = (t) => {                                    // daily wave + noise + occasional spike
  const base = 100 + 40 * Math.sin((2 * Math.PI * t) / 86_400) + (rand() - 0.5) * 30;
  return base + (rand() < 0.002 ? 300 : 0);
};

const db = new DatabaseSync(":memory:");
db.exec(`CREATE TABLE raw(t INTEGER PRIMARY KEY, value REAL);
CREATE TABLE rollup60(bucket INTEGER PRIMARY KEY, count INTEGER, total REAL, maxValue REAL);
CREATE TABLE rollup300(bucket INTEGER PRIMARY KEY, count INTEGER, total REAL, maxValue REAL);`);
db.exec("BEGIN");
const insert = db.prepare("INSERT INTO raw VALUES(?,?)");
for (let t = 0; t < DURATION; t += 1) insert.run(t, generate(t));
db.exec("COMMIT");
for (const g of [60, 300]) db.exec(`INSERT INTO rollup${g}
  SELECT t / ${g}, count(*), sum(value), max(value) FROM raw GROUP BY t / ${g}`);

const percentile = (a, p) => { const s = [...a].sort((x, y) => x - y); return s[Math.floor(s.length * p)]; };
const rawValues = db.prepare("SELECT value FROM raw").all().map((r) => r.value);
const bytes = (t) => db.prepare("SELECT sum(pgsize) AS b FROM dbstat WHERE name = ?").get(t).b;
const rows = (t) => db.prepare(`SELECT count(*) AS n FROM ${t}`).get().n;

const RAW = { avg: rawValues.reduce((a, b) => a + b) / DURATION, max: Math.max(...rawValues),
  p95: percentile(rawValues, 0.95), over: rawValues.filter((d) => d > THRESHOLD).length };
console.log(`model: ${DURATION} samples (one per second), seed ${SEED}, threshold ${THRESHOLD}\n`);
console.log("format".padEnd(14) + "rows".padStart(7) + "bytes".padStart(9) + "bytes/row".padStart(12) +
  "average".padStart(10) + "max".padStart(10) + "p95".padStart(9) + "over-threshold s".padStart(18));
const line = (name, n, b, avg, max, p, over) => console.log(name.padEnd(14) + String(n).padStart(7) +
  String(b).padStart(9) + (b / n).toFixed(1).padStart(12) + avg.toFixed(2).padStart(10) +
  max.toFixed(2).padStart(10) + p.toFixed(2).padStart(9) + String(over).padStart(18));
line("raw", rows("raw"), bytes("raw"), RAW.avg, RAW.max, RAW.p95, RAW.over);

const ROLLUP = {};
for (const g of [60, 300]) {
  const k = db.prepare(`SELECT count, total, maxValue FROM rollup${g}`).all();
  const bucketAvg = k.map((r) => r.total / r.count);
  const o = { avg: k.reduce((a, r) => a + r.total, 0) / DURATION, max: Math.max(...k.map((r) => r.maxValue)),
    p95: percentile(bucketAvg, 0.95), over: bucketAvg.filter((d) => d > THRESHOLD).length * g };
  ROLLUP[g] = o;
  line(`${g} s rollup`, rows(`rollup${g}`), bytes(`rollup${g}`), o.avg, o.max, o.p95, o.over);
}

console.log(`\ndeviation from the raw answer (percent):`);
for (const g of [60, 300]) {
  const s = (a, b) => (((a - b) / b) * 100).toFixed(2);
  console.log(`  ${g} s rollup -> average ${s(ROLLUP[g].avg, RAW.avg)}, max ` +
    `${s(ROLLUP[g].max, RAW.max)}, p95 ${s(ROLLUP[g].p95, RAW.p95)}, ` +
    `over-threshold time ${s(ROLLUP[g].over, RAW.over)}`);
}
for (const g of [60, 300]) {                                // the same question with the max column
  const n = db.prepare(`SELECT count(*) AS n FROM rollup${g} WHERE maxValue > ?`).get(THRESHOLD).n;
  console.log(`  ${g} s rollup, over-threshold time with max: ${n * g} s (raw ${RAW.over} s, ` +
    `x${((n * g) / RAW.over).toFixed(1)})`);
}
console.log(`scanned-row ratio: 60 s rollup 1/${DURATION / rows("rollup60")}, ` +
  `300 s rollup 1/${DURATION / rows("rollup300")}; byte ratio 1/` +
  `${(bytes("raw") / bytes("rollup60")).toFixed(1)} and 1/${(bytes("raw") / bytes("rollup300")).toFixed(1)}`);

model: 21600 samples (one per second), seed 20260801, threshold 200

format           rows    bytes   bytes/row   average       max      p95  over-threshold s
raw             21600   356352        16.5    125.84    450.17   148.84                41
60 s rollup       360    16384        45.5    125.84    450.17   140.74                 0
300 s rollup       72     4096        56.9    125.84    450.17   139.88                 0

deviation from the raw answer (percent):
  60 s rollup -> average 0.00, max 0.00, p95 -5.44, over-threshold time -100.00
  300 s rollup -> average 0.00, max 0.00, p95 -6.02, over-threshold time -100.00
  60 s rollup, over-threshold time with max: 2400 s (raw 41 s, x58.5)
  300 s rollup, over-threshold time with max: 9600 s (raw 41 s, x234.1)
scanned-row ratio: 60 s rollup 1/60, 300 s rollup 1/300; byte ratio 1/21.8 and 1/87.0

These numbers are in the measurement class; the values depend on this run’s seed, the sign and magnitude of the deviations do not. Because the bytes column depends on the page grain, bytes per row look large in a table with few rows; for comparison, the total byte ratio is what matters.

The four questions show three separate behaviors. Average and max are not distorted at all: the deviation is 0.00 percent at both grains, because the rollup row carries the total, the count, and the max — the average and the max can be reconstructed exactly from these three. The percentile is distorted, but only a little: p95 comes out 5.44 percent low at the one-minute grain, 6.02 percent low at the five-minute grain, because the bucket average flattens the distribution inside the bucket. The threshold breach disappears completely: viewed through the bucket average, the 41-second breach looks like 0 seconds, that is, 100 percent error.

Asking the same question with the max column does not work either, it misses in the opposite direction: 2400 seconds at the one-minute grain, 9600 seconds at the five-minute grain — 58.5 and 234.1 times the raw answer, because a breach in any second of a bucket counts the whole bucket as breaching. A rollup row answers the questions its carried columns can answer; for the others it leaves an uncertainty between a lower bound and an upper bound.

Design

The write path is built on a message broker and competing consumers (the Application Layer and Service Interaction course’s Queues and Workflows topic); queue-based load leveling from the same topic flattens the peak 480,000 samples/s into a steady rate the writers see. On the agent side, backpressure (the Resilience and Reliability course’s Fault Isolation topic) is measured by buffer utilization, and when the buffer fills, the oldest sample is dropped.

The store is a wide-column store (the Scaling the Data Layer course’s Data Distribution topic, Store Types): the row key is the series ID, the columns are time-ordered. The sharding key is the series ID, not the node name — sharding by node name would let one node’s 120 series fall into a single shard and create a hot shard. Partitioning is by time (the same topic, Partitioning Strategies), and the retention policy is enforced by dropping partitions: a raw partition older than 7 days is deleted in a single operation, not row by row.

Rollup is two materialized views (the same topic): the refresh is incremental and writes only the bucket that just closed. The parameters are the grains — 60 seconds and 300 seconds — and their rationale is OT8 and OT9’s retention periods. The tier read for capacity estimation is the five-minute one; the method for going from a time series to a resource estimate was established in the Performance Anti-Patterns and Monitoring course’s Capacity Planning lesson and is not repeated here, only its input comes from this tier.

Deliberately unused pattern: the idempotency key ledger. The ledger from the Resilience and Reliability course’s Idempotent Operations lesson has no place here, because a sample is already addressed by the (series, time) pair, and a duplicate write overwrites the same cell with the same value; a separate ledger would add a record of the same order of magnitude next to every 26-byte sample record, and it would be paid 240,000 times per second. The second is federation (the Data Distribution topic): there is no separate functional domain to split off, all the data is uniform and has a single access pattern.

Eliminated Alternatives

Keeping everything raw answers percentile and threshold questions without error at any date. It requires 393.57 TB, ten times the 40 TB limit; the six-hour query also scans 2160 rows per series and violates the 400-row limit too. It is eliminated by two limits at once.

Keeping everything as a five-minute rollup is the cheapest: 17.16 TB total and 72 rows per query. It answers the threshold-breach question with either 100 percent error (0 seconds) or a 234.1-fold inflation (9600 seconds), and deviates by 6.02 points on the percentile. It is eliminated by the “without error over the last 7 days” limit.

The chosen three-tier layout stays at 31.50 TB, scans 360 rows per query, and answers the sensitive questions without error because raw data sits there for the last seven days. Which limit changes and the alternative wins: if threshold-breach and percentile questions are never asked, the raw tier loses its reason to exist, the 3.77 TB line item drops, and the write path can write straight to the rollup; conversely, if incident review looks back further than a week, raw retention is extended and the total quickly pushes against 40 TB.

Failure Behavior and What Is Given Up

When the message broker goes down, agents accumulate in their buffers; dropping the oldest sample when the buffer fills is graceful degradation (the Fault Isolation topic), because the series stays continuous but its grain temporarily coarsens — the average and max question keeps being answered, the threshold-breach question cannot be. The reverse happens too: if the rollup job falls behind, the dashboard reads from the raw tier and rows scanned per series rise from 360 to 2160, a sixfold increase. When a shard node goes down, only the reads for the series on that shard are affected; since questions are asked per series, the blast radius is one shard.

What is given up: percentile and threshold-breach questions on data older than seven days do not get a reliable answer. How many seconds a fluctuation from a month ago lasted cannot be asked in this design; the answer that can be obtained is either 0 or 58.5 times the true value.

Summary

• The metric stream makes per-sample cost decisive: 2,400,000 series, 240,000 samples/s on average, 480,000 at peak; the peak multiplier is 2, not the 3 of earlier cases, because the stream depends on the agent’s scheduler.
• The storage tier total is 31.50 TB; if the same data were kept raw it would be 393.57 TB, that is, 12.49 times as much.
• Rollup reduces scanned rows by 60- and 300-fold, and stored bytes by 21.8- and 87.0-fold.
• Deviation depends on the question type: 0.00 percent for average and max, 5.44 and 6.02 percent for p95, 100 percent for threshold-breach duration — the same question inflates 58.5- and 234.1-fold when asked with the max column.
• This is why the raw tier is not a performance line item but an answerability line item: seven days of raw data costs 3.77 TB, and in exchange it defines the window in which sensitive questions can be answered.

Next Step

In this case, finding a record was easy: given a series ID and a time range, the rows to read sat in one place, in order. The next case removes that convenience. There, records sit on a two-dimensional plane and the question comes in the form “the ones near this point”; proximity has no sortable key, because every way of sorting two dimensions along a single axis pushes some neighbors far apart. The question becomes: how many candidate records must be examined to answer such a query, and how does that count depend on the chosen cell size.