Engineering Atlas › System Design Interviews

Design a metrics database

21 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

Requirements

Functional

Non-functional

Out of scope: the alert manager and notification routing, log and trace storage, and the agents that scrape targets. See Prometheus for the single-node engine this design scales out.

Back-of-envelope

Round, assumed numbers for a large company's shared metrics platform.

API

POST /api/v1/push                      (protobuf, snappy; header X-Scope-OrgID: tenant)
  body: { timeseries: [ { labels: [ {name, value} ... ],
                          samples: [ {timestamp_ms, value} ... ] } ... ] }
  -> 200 | 400 { out_of_order | too_old | invalid_labels } | 429 { series_limit | rate_limit }

GET /api/v1/query_range?query=sum by (code)(rate(http_requests_total{job="api"}[5m]))
                       &start=1760572800&end=1760594400&step=60
  -> { resultType: "matrix", result: [ { metric: {code: "500"}, values: [ [t, "v"] ... ] } ] }

GET /api/v1/series?match[]={job="api"}&start=..&end=..   -> [ {labels} ... ]
GET /api/v1/label/{name}/values?start=..&end=..          -> [ "value" ... ]

The write API is a batch of series, each with a few samples, because agents buffer 5 to 10 seconds of scrapes. A 400 response tells the agent to drop the batch, because retrying an out-of-order sample will never succeed. A 429 tells it to back off and retry. Queries are evaluated at fixed step intervals, which is what makes their results cacheable.

Data model

Series identity. A series is its sorted label set, for example {__name__="http_requests_total", job="api", code="500"}. Its ID is a 64-bit hash of the sorted labels. The distributor uses hash(tenant, labels) to place it on the ingester ring, so all samples of one series land on the same 3 ingesters in order.

Head (per ingester, per tenant). A map from series ID to memSeries (labels, last timestamp, chunks of up to 120 samples), plus postings from each (name, value) pair to a sorted list of series IDs, backed by a WAL on local SSD. It covers about 2 to 3 hours, because samples are cut into two-hour blocks (Prometheus docs).

Block (in object storage). tenant/01J9Z.../, named by a time-sortable ULID, holds chunks/000001 (compressed chunks), index (symbol table, series entries with chunk references, postings and a postings offset table) and meta.json (time range, compaction level, sources, downsampling resolution). Blocks are immutable; compaction writes new ones and marks old ones for deletion. A small per-tenant bucket index lists blocks and their time ranges so queriers never list the bucket.

Limits. A per-tenant table of maximum active series, samples per second, label length, series per query and retention, reloaded without restart.

Partitioning by series hash rather than metric name matters: a popular metric like http_requests_total might have 5M series by itself, and hashing by name would put them all on one ingester.

High-level design

flowchart LR
  A["Agents (remote write)"] --> D["Distributors (stateless)"]
  D -->|"hash ring, RF 3"| I["Ingesters (head + WAL)"]
  I -->|"2h blocks"| O[("Object storage")]
  C["Compactor"] <--> O
  Q["Query frontend"] --> R[("Results cache")]
  Q --> QR["Queriers"]
  QR -->|"recent"| I
  QR -->|"older"| S["Store gateways"]
  S --> O

Write path.

  1. A distributor validates the batch, checks the tenant's rate and series limits, and drops samples from the non-elected replica of each high-availability Prometheus pair.
  2. It hashes each series' label set onto the ring and sends the samples to the 3 owning ingesters, one per zone. Mimir's classic architecture works this way, with a replication factor of 3 and zone-aware placement (Mimir ingester docs).
  3. Each ingester appends to its WAL and to the series' open chunk, creating the series and its postings if new.
  4. When 2 of 3 succeed, the distributor returns 200.
  5. Every 2 hours each ingester cuts its head into a block, uploads it and truncates the WAL. The compactor later merges the 3 replica blocks into one.

Read path.

  1. The query frontend aligns the range to the step, splits a 7-day query into 7 one-day pieces, and serves cached days.
  2. For each missing piece, a querier intersects postings to find series and fetches chunks: the last 12 hours or so from ingesters, older data from store gateways reading object storage.
  3. The querier merges overlapping replica samples by timestamp, runs PromQL functions per series, then aggregates. The frontend stitches and caches complete days.

Deep dives

1. Sharding and replicating the write path

Options: shard by tenant (simple, but one big tenant overflows a node), shard by metric name (a hot metric overflows a node), or shard by series hash (load spreads evenly, but every query fans out to many ingesters). I shard by series hash, because writes are continuous and must never hot-spot, while queries can afford the fan-out.

Replication uses a write quorum of 2 out of 3, without consensus. Monitoring data tolerates a little divergence: a sample that reached 2 ingesters but not the third leaves a gap that the merge on read fills in. Running Raft per series would cost a consensus round per batch, and it would block writes during elections in exactly the outage you need to watch. Ring membership is stored in a gossip-based or key-value store. When an ingester leaves, its series move to the next token owners, and the departing ingester flushes its head to object storage so nothing is lost.

A newer option puts a log in front of the ingesters. Mimir 3.0 calls its Kafka-based ingest storage the preferred architecture: distributors write each series to one Kafka partition and acknowledge once Kafka has replicated and persisted it, and each zone runs one ingester per partition that consumes from it. Reads are slightly stale by default, typically under a second, and a query can ask for read-after-write by sending X-Read-Consistency: strong (Mimir docs). I'd choose it when a Kafka platform team already exists, because it lets ingesters crash or scale without failing writes.

Late samples are the subtle problem. A TSDB with Gorilla compression wants samples in order within each series, because the chunk is an append-only bitstream. Agents that buffer during a network partition send old samples late. I reject samples older than the head's lower bound, or older than the series' last timestamp, with out_of_order. For tenants that need it, I enable a small out-of-order window that keeps late samples in a separate uncompressed buffer, which is merged when the block is cut. Prometheus exposes this as out_of_order_time_window, which defaults to 0, so it is off unless you set it (Prometheus config).

2. Cardinality and limits

A developer adds user_id to a request counter, and 5M new series appear within an hour. Each costs about 4 KB on 3 ingesters, so this one change adds 60 GB of RAM and a block index that grows with every query. Options: limit globally (one tenant hurts all the others), limit per tenant at the distributor (fair and cheap), or drop labels on ingest (fast to stop the damage, but it silently changes the data).

I enforce per-tenant limits on active series and on series per metric at the distributor, and return 429 with a clear reason. A cardinality API shows each tenant its top metrics and top label values, so teams can fix the cause. Series churn, where short-lived pods create new series every deploy, is the other half of the problem. A series created and abandoned within an hour still costs a full index entry in every block that covers that hour, so I alert on churn rate, not just on active count.

3. Fast queries over long ranges

A 30-day query over 10,000 series at 15 s resolution touches 1.7 billion samples. The design reduces that in three ways:

Store gateways keep only the header of each block index (symbols and the postings offset table) on local disk, and by default load it into memory only when a query first needs that block (Mimir store-gateway; Mimir configuration reference). They fetch postings and chunks with HTTP range reads, and cache them in a shared memory cache, so a cold 30-day query costs a few hundred object-storage GETs instead of downloading whole blocks.

Down the stack

Layer 1The contract

  • Accepted means replicated. A 200 from push means every sample in the batch is in the WAL of at least 2 ingesters in different zones.
  • In order per series. Samples for a series must have strictly increasing timestamps, except within a configured out-of-order window. Rejected samples are reported per batch with a reason and are never partially applied to a series.
  • Read after write within seconds. A sample is visible to queries as soon as one ingester holds it. Queries merge replicas, so a sample missing from one replica still appears.
  • Staleness. When a scrape stops returning a series, the series is marked stale and drops out of results at once. A series that simply goes quiet is looked up at most 5 minutes back, the default lookback, so it never repeats its last value forever (Prometheus docs).
  • Limits are explicit. A write over a limit fails with 429 and the limit's name. It is never silently dropped.

Layer 2The mechanism

WAL records are Series (ID and labels, once per new series), Samples (ID, timestamp, value) and Tombstones, written in 32 KB pages inside 128 MB segments (Prometheus docs) and checkpointed when the head is truncated. Replaying them is why ingester restarts take minutes at high cardinality, and why closed head chunks are memory-mapped rather than rebuilt.

Chunks use delta-of-delta timestamps and XOR values, as in Time Series Databases. Postings are sorted arrays of series IDs; N matchers intersect N lists in linear time, starting from the shortest. A negative matcher subtracts a list, and a regex expands to the matching values from the sorted value list.

Compaction is a k-way merge: the compactor walks the 3 replica blocks for the same 2 hours in sorted label order, merges samples by timestamp dropping duplicates, and writes one block, then merges 2-hour blocks into 12- and 24-hour ones so long queries open fewer indexes.

Layer 3The machine

  • Memory per series. Label strings are interned, but each series still needs a struct with labels, chunk references and a lock (a few hundred bytes), an open chunk buffer, and 4 to 8 bytes of postings per label. Go's garbage collector scans all of it, so heap size drives CPU too.
  • WAL writes. The WAL calls fsync per segment or on an interval, not per sample. A crash can lose about the last second on one replica, which the other two cover.
  • Decode speed. Gorilla decoding is bit-serial and branchy; a core manages on the order of 100 million samples per second at best, so the 1.7 billion samples in deep dive 3 cost seconds of CPU. Sharding, caching and rollups remove that.
  • Object storage latency. First byte takes 20 to 100 ms, so store gateways merge nearby chunk ranges into one range read and run hundreds of reads in parallel.
  • Cache lines. Intersecting sorted 4-byte IDs streams through memory and suits the prefetcher, which is why postings are arrays rather than trees.

Why it's built this way

  1. Monitoring data arrives continuously from everywhere and must keep flowing during outages, so writes use a sloppy 2-of-3 quorum without consensus.
  2. Recent data is read far more than old data and must be fresh, so it stays in memory in ingesters, and old data goes to immutable blocks.
  3. Immutable blocks fit object storage, which is the cheapest durable store per byte, so the stateful tier is only a few hours deep.
  4. Queries over long ranges would otherwise scan billions of points, so they are split, cached and run against rollups.
  5. Every series costs memory on three machines, so cardinality is limited per tenant at the front door.

Failure modes

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Shard by series hashEven write load, no hot nodesEvery query fans out to many ingestersHigh cardinality, continuous writes
Shard by metric or tenantQueries hit one nodeHot metrics and big tenants overflowSmall scale
2-of-3 quorum without consensusWrites survive failures, low latencyReplicas can briefly differMetrics and other loss-tolerant data
Consensus per shardExactly one historyElection stalls, more latencyFinancial or audit data
Object storage for blocksCheap, durable, stateless queriers20 to 100 ms first byte, needs cachesRetention longer than days
Local disks onlyFast, simpleScaling and recovery move terabytesSingle cluster, short retention
Raw data onlyExact answersYear-long queries scan billions of pointsShort retention
Downsampled rollups20x fewer points for long rangesLoses detail, must keep counter semanticsDashboards over months or years

Leadership lens

Check yourself

  1. Why does the design shard writes by a hash of the series label set rather than by metric name or tenant?

    Show answer

    Writes are continuous and must never hot-spot, and a popular metric like http_requests_total can have 5M series that would all land on one ingester if hashed by name. Series-hash sharding spreads load evenly, at the cost of every query fanning out to many ingesters.

  2. A developer adds user_id to a request counter and 5M new series appear within an hour. What does that cost, and how does the design contain it?

    Show answer

    At about 4 KB per series on 3 ingesters, it adds roughly 60 GB of RAM plus a growing block index. Per-tenant series limits at the distributor return 429 on new series only, and a cardinality report points the team to the offending label while other tenants are unaffected.

  3. Why are writes acknowledged after 2 of 3 ingesters with no consensus protocol, and when would you choose consensus instead?

    Show answer

    Monitoring must keep working during the outages it observes, and a gap on one replica is filled by merging replicas on read, whereas Raft would add a consensus round per batch and stall writes during elections. Consensus per shard fits financial or audit data that needs exactly one history.

  4. Roughly how much ingester RAM does 100M active series need, and why is that the dominant cost rather than object storage?

    Show answer

    About 100M x 4 KB x 3 replicas, or 1.2 TB, which is around 45 ingesters of 64 GB kept at half memory across 3 zones. All 900 TB of object storage costs only around $21,000 a month, so active series in RAM drive the bill.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →