Medium

Design a Metrics Monitoring System (Datadog)

Build a system that ingests time-series metrics from thousands of hosts, stores them efficiently with downsampling, supports fast tag-filtered queries, and evaluates alert rules in near real time.

1Requirements

Functional

• Agents on hosts push counters, gauges, and histograms with tags (host, region, service) at 10 second resolution
• Users query metrics with tag filters and aggregations (avg, sum, p95, max) over arbitrary time ranges
• Users define alert rules (threshold, duration) that fire notifications when breached
• Dashboards render multiple queries with auto-selected resolution based on the time range
• Old data is retained at coarser resolution: raw for 7 days, 1 minute rollups for 30 days, 1 hour rollups for 1 year

Non-functional

• Ingest 10M data points per second at peak without dropping writes
• Query latency under 200ms p99 for dashboard panels over a 24 hour window
• Alert evaluation delay under 30 seconds from data arrival to notification
• Ingestion path must tolerate a storage node failure with no data loss (buffered writes)
• Read availability matters more than perfect freshness; a 1 minute lag in dashboards is acceptable

2Back-of-envelope estimation

Monitored hosts	100,000	each emits ~100 series at 10s resolution
Write throughput	1M points/sec	100k hosts x 100 series / 10s
Raw daily volume	~1.4 TB/day	1M/s x 86,400s x ~16 bytes per point
After compression	~120 GB/day	Gorilla-style delta-of-delta gets ~1.37 bytes/point
Active series (cardinality)	10M series	the real cost driver; each series needs an index entry
1h rollup volume for 1 year	~1.3 TB	10M series x 8,760 hours x ~16 bytes, cheap on object storage

3API design

POST /api/v1/ingest

Batch of points: [{metric, tags, value, ts}]. Returns 202; durability comes from the queue, not the response.

GET /api/v1/query?metric=cpu.util&filter=region:us-east&agg=p95&start=...&end=...&step=60s

Time-series query with tag filter, aggregation, and resolution step.

POST /api/v1/alerts

Create alert rule: {query, threshold, comparator, for_duration, channels}.

GET /api/v1/metrics/search?q=cpu

Metric and tag autocomplete backed by the inverted tag index.

4High-level design

Agents batch points locally and push to an ingestion gateway that validates, normalizes tags into a canonical sorted order, and writes to Kafka partitioned by hash(metric name + tag set). Kafka is the durability boundary: once acked there, a storage node crash cannot lose data.

Storage nodes consume their partitions and write to a time-series store. Each unique (metric, tag set) combination is a series identified by a series_id. Recent data lives in an in-memory write buffer plus a write-ahead log, flushed as compressed immutable blocks (Gorilla encoding) every 2 hours. An inverted index maps each tag key:value pair to the set of series_ids containing it, so a query like region:us-east AND service:api is a set intersection.

A separate rollup pipeline consumes the same Kafka topics and maintains 1 minute and 1 hour pre-aggregates (min, max, sum, count, and a sketch for percentiles). The query planner picks the resolution tier automatically: a 1 hour dashboard reads raw, a 30 day dashboard reads 1h rollups, keeping the number of points scanned roughly constant regardless of range.

The alerting engine is a scheduler that evaluates each rule every 30 seconds by running its query against the hot tier only. Rules track a state machine (ok, pending, firing) so a threshold must be breached for the configured duration before notifying, which suppresses flapping. Notifications go through a dedup and routing layer to Slack, PagerDuty, or webhooks.

Aged blocks migrate down a tiering ladder: hot SSD for 7 days, then 1 minute rollups on cheaper disks for 30 days, then 1 hour rollups on object storage (S3) for a year. Queries fan out across tiers transparently and merge results.

5Data model

series

series_id, metric_name, tags_hash, tags_json, first_seen, last_seen

one row per unique metric + tag combination; this table size is the cardinality

points (columnar blocks)

series_id, block_start_ts, resolution, compressed_timestamps, compressed_values

immutable 2h blocks, Gorilla-compressed

tag_index

tag_key, tag_value, series_ids (posting list)

inverted index; queries intersect posting lists

alert_rules

rule_id, query, comparator, threshold, for_duration_s, state, last_eval_ts, channels

state machine: ok, pending, firing

6Deep dives

Tag cardinality is the silent killer

Storage volume scales with points per second, but memory, index size, and query planning cost all scale with the number of unique series. One engineer adding a user_id or request_id tag can turn 10M series into 500M overnight, blowing out the inverted index and the per-series write buffers. This is the classic cardinality explosion and it is the number one operational incident for real monitoring vendors.

Defenses: enforce a per-metric cardinality budget at the ingestion gateway (track approximate distinct tag sets per metric with a HyperLogLog and reject or drop tags beyond a limit, say 100k series per metric), maintain an allowlist of tag keys, and expose a cardinality dashboard so teams see the cost of their tags. Some systems automatically quarantine high-cardinality tags into logs or traces instead, since those are the right tool for per-request identifiers.

Downsampling and rollups

You cannot answer a 90 day query by scanning raw 10 second points: that is 777,600 points per series, and a dashboard panel touching 200 series would scan 155M points. Rollups keep query cost bounded by pre-aggregating each series into 1 minute and 1 hour buckets as data arrives, so the planner can always choose a tier where points scanned stays in the low thousands.

The subtlety is that you must store decomposable aggregates, not final answers. Store sum and count so any downstream consumer can compute a correct average across merged buckets; storing avg directly makes re-aggregation wrong. Percentiles do not decompose at all, so store a mergeable sketch (t-digest or DDSketch) per bucket. Min and max decompose trivially.

Rollups also solve late data: since a rollup consumer reads from Kafka, a point arriving 5 minutes late simply updates the still-open 1 minute bucket. Buckets seal after a grace period (say 15 minutes), after which late points are counted in a side metric rather than mutating sealed blocks.

Alert evaluation at scale

With 100k alert rules evaluated every 30 seconds you run about 3,300 queries per second just for alerting. Two things make this tractable. First, alert queries only ever touch the hot in-memory tier, which is the cheapest data to read. Second, rules are sharded across evaluator workers by rule_id using consistent hashing, so adding workers scales evaluation linearly and a worker crash only delays its own shard until reassignment.

Correctness details matter more than throughput. The 'for duration' clause requires the rule state machine: a breach moves ok to pending with a timestamp, and only if every subsequent evaluation stays breached until for_duration elapses does it move to firing. Any recovery resets to ok. This suppresses flapping without heuristics. You also need a dead-man switch: a rule that receives no data at all should optionally fire (no data is often the outage), which means the evaluator must distinguish 'query returned empty' from 'query returned values below threshold'.

Why not just use Postgres

A naive row-per-point schema in Postgres dies at this scale for three reasons: 16 bytes of payload carries ~40 bytes of row overhead plus index entries, B-tree indexes on (series_id, ts) suffer constant random inserts, and time-range scans read pages that interleave thousands of series. Write amplification and cache misses dominate.

Purpose-built TSDBs win by exploiting the workload shape: writes are append-only per series, timestamps are nearly regular (delta-of-delta encodes them in ~1 bit), and consecutive values are similar (XOR encoding). Facebook's Gorilla paper reported 1.37 bytes per point versus 16 raw, a 12x reduction, which is what makes keeping 7 days of raw data in a hot tier affordable. For an interview, naming the columnar block layout and the inverted tag index is what distinguishes a real design from 'use InfluxDB'.

7Rapid implementation: build the MVP

Theory is table stakes. Here is how you would stand up a working version fast, on a budget, with the core algorithm in real code.

Stack

Node + Fastify ingest, SQLite for blocks and index, a setInterval alert loop, uPlot dashboard, all on a $6 VPS

01Scaffold a Fastify server with POST /ingest accepting {metric, tags, value, ts} batches
02Canonicalize tags (sort keys, join as k=v,k=v) and hash to a series_id; upsert into a series table in SQLite
03Buffer points in memory per series; every 60s flush each buffer as a compressed block row (series_id, start_ts, JSON-packed deltas)
04Build GET /query that picks blocks by time range, decodes them, applies avg/max/p95, and returns [ts, value] pairs
05Add a rollup pass on flush: write 1-minute sum/count/min/max rows to a rollups table; query uses rollups when range > 6h
06Write the alert loop: every 30s run each rule's query over the last window, drive the ok/pending/firing state machine, POST to a webhook on firing
07Serve a static HTML page with uPlot charts polling /query every 10s
08Load test with a script emitting 10k points/sec of fake CPU metrics and watch p95 query latency

Series canonicalization and delta-encoded block flush

typescript

import { createHash } from "crypto";

function seriesId(metric: string, tags: Record<string, string>): string {
  const canon = Object.keys(tags).sort()
    .map((k) => k + "=" + tags[k]).join(",");
  return createHash("sha1").update(metric + "|" + canon).digest("hex").slice(0, 16);
}

// In-memory buffer per series, flushed as a compact block every 60s.
const buffers = new Map<string, Array<[number, number]>>();

export function ingest(metric: string, tags: Record<string, string>, ts: number, value: number) {
  const id = seriesId(metric, tags);
  if (!buffers.has(id)) buffers.set(id, []);
  buffers.get(id)!.push([ts, value]);
}

export function flushBlock(id: string): { startTs: number; deltas: number[]; values: number[] } | null {
  const pts = buffers.get(id);
  if (!pts || pts.length === 0) return null;
  pts.sort((a, b) => a[0] - b[0]);
  const startTs = pts[0][0];
  const deltas: number[] = [];
  const values: number[] = [];
  let prev = startTs;
  for (const [ts, v] of pts) {
    deltas.push(ts - prev); // mostly 10, compresses to near nothing
    values.push(v);
    prev = ts;
  }
  buffers.set(id, []);
  return { startTs, deltas, values };
}

Rollup query with decomposable aggregates

sql

-- 1-minute rollups store sum and count so avg merges correctly.
CREATE TABLE rollups_1m (
  series_id TEXT NOT NULL,
  bucket_ts INTEGER NOT NULL,  -- epoch seconds floored to 60
  sum REAL NOT NULL,
  count INTEGER NOT NULL,
  min REAL NOT NULL,
  max REAL NOT NULL,
  PRIMARY KEY (series_id, bucket_ts)
);

-- Re-aggregate 1m buckets into 5m buckets at query time.
-- Correct because sum/count decompose; storing avg would not.
SELECT
  (bucket_ts / 300) * 300           AS ts5m,
  SUM(sum) / SUM(count)             AS avg_value,
  MIN(min)                          AS min_value,
  MAX(max)                          AS max_value
FROM rollups_1m
WHERE series_id IN (SELECT series_id FROM tag_index
                    WHERE tag = 'region=us-east')
  AND bucket_ts BETWEEN :start AND :end
GROUP BY ts5m
ORDER BY ts5m;

Alert state machine with for-duration

typescript

type AlertState = "ok" | "pending" | "firing";

interface Rule {
  id: string;
  threshold: number;
  forDurationMs: number;
  state: AlertState;
  breachedSince: number | null;
}

export function evaluate(rule: Rule, latestValue: number | null, now: number): AlertState {
  const breached = latestValue !== null && latestValue > rule.threshold;
  if (!breached) {
    rule.state = "ok";
    rule.breachedSince = null;
    return rule.state;
  }
  if (rule.breachedSince === null) {
    rule.breachedSince = now;
    rule.state = "pending";
  } else if (now - rule.breachedSince >= rule.forDurationMs && rule.state !== "firing") {
    rule.state = "firing"; // notify exactly once on this transition
    notify(rule);
  }
  return rule.state;
}

function notify(rule: Rule) {
  fetch(process.env.WEBHOOK_URL as string, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ rule: rule.id, state: "firing", at: Date.now() }),
  }).catch(() => { /* retry queue in real life */ });
}

Bottlenecks & failure modes

⚠Cardinality explosion from unbounded tag values inflating the index and per-series buffers; needs ingestion-time budgets
⚠Hot Kafka partitions when one metric dominates traffic; partition by series hash, not metric name alone
⚠Query fan-out across storage shards for high-cardinality filters; mitigate with posting-list intersection order (smallest first)
⚠Rollup lag during traffic spikes makes long-range dashboards stale; monitor consumer lag as a first-class SLO
⚠Alert storms during a real outage flooding notification channels; group and dedup by service before paging

Key takeaways

▸Cost and stability scale with series cardinality, not write volume; budget cardinality at the edge
▸Store decomposable aggregates (sum, count, sketches) so rollups can be merged correctly
▸Make the queue the durability boundary so storage nodes can crash without data loss
▸Tier storage by age and resolution so query cost stays roughly constant across time ranges
▸Alerting needs a state machine with a for-duration clause and a no-data path, not just threshold checks

Brush up on the underlying topics

Observability: Metrics, Logs, Traces Message Queues and Streaming Sharding and Partitioning Storage and Search Probabilistic Data Structures