System design: aggregate metrics and analytics events

Design a system that collects analytics events or metric counts from lots of services, aggregates them, and lets people query the results. Then use your data model to answer a few product questions.

Say up front what you're covering.

Then agree on what "good" means:

• Never slows the sender: sending an event is a quick in-memory call. If our system is down, the service carries on.
• Takes in a lot: a million events a second at peak, and more next year.
• Fresh: a count shows up in queries within about a minute.
• Fast to query: a dashboard covering the last week loads in a second or two.
• Right where it matters: counts and sums are exact. Unique counts and percentiles can be close estimates, as long as we say so.

These are assumptions, but reasonable ones: a million events a second at peak, about 500 bytes each, and around a million distinct combinations of metric and dimensions active at once.

~1M

events per second at peak

~40 TB

of raw events a day

~1M

active metric and dimension combos

~70 GB

of minute rollups a day

1. Each service uses a small library. It records events in memory, adds up simple counts locally, and sends a batch every second or so in the background.
2. Batches land on an event stream, split into partitions by metric name and dimensions, so every event for the same thing reaches the same aggregator.
3. Aggregators add events up into one-minute buckets and write a row per metric, dimension combo and minute to the metrics store.
4. A copy of the raw stream is also written to file storage as compressed files, one folder per hour, for questions the rollups can't answer.
5. The query API reads the rollups, picks minute, hour or day rows depending on the time range, and does the final math like rates and percentiles.

What the apps need to do

What we store and where

The library must never block the service. Recording an event just adds to an in-memory buffer. A background thread sends batches. If the buffer fills because we're down, it drops the oldest and counts how many it dropped, instead of eating the service's memory.

For plain counts, the library adds them up before sending: 10,000 'payment succeeded in US' events in a second become one message saying 10,000. That's the biggest single cut in volume, and it's free.

Events go onto a stream that's partitioned by a hash of metric and dimensions. That's sharding the stream, so each aggregator owns a slice and can add up in memory without talking to the others. The stream is a message queue that holds a few days, so a slow aggregator just falls behind instead of losing data.

• Cap the dimensions on each metric. Someone adding user ID as a dimension turns one combo into millions. Reject or drop unknown dimensions, and alert the owner.
• Each batch carries an ID, so a retried send can be spotted and skipped.
• For crazy-high event types, sample 1 in 100 and multiply back up when querying. Label those metrics as sampled.

Aggregators group events into one-minute windows by when they happened, not when they arrived. That's stream processing with event-time windows. A window stays open a couple of minutes for stragglers, then gets written out.

What you keep depends on the metric type. Counts and sums just add. For latency you can't average percentiles, so keep a histogram of bucket counts, which adds up across servers and minutes. For unique counts, like unique merchants, keep a small sketch (HyperLogLog) that merges, and accept about 1% error.

Each row has a key of metric, dimensions and minute. Writing a row is an overwrite, not an add. So if an aggregator crashes and replays from its last checkpoint, it writes the same rows again and nothing gets counted twice. Events that arrive really late get added as a correction to the minute they belong to.

The rollup table has one row per metric, dimension combo and time bucket. The columns are the dimensions (like country, payment method and status), plus count, sum, a histogram and a unique-count sketch. A column-oriented database is a good fit here, because queries add up a few columns over many rows. The SQL vs NoSQL question really comes down to that read pattern.

Say the interviewer asks 'what's the payment success rate by country this week?'. Query 'payments attempted' grouped by country and status from the hourly table, then divide succeeded by total. Always divide totals. Averaging each hour's rate gives quiet hours the same weight as busy ones.

'What's p99 checkout latency per endpoint?' merges the histograms for each endpoint, then reads the 99th percentile off the merged one. 'How many distinct merchants used a feature yesterday?' merges the daily sketches. If they want it exact, that's a batch query over the raw files.

Questions about one specific merchant are the hard case, because merchant ID has far too many values to be a normal dimension. Build a separate daily table keyed by merchant for the handful of metrics people ask about that way, and say so when you hit it.

• Stream processing and windowingCount things in data that never stops coming in.
• Message queues and event streamsKeep services from depending on each other, and handle traffic spikes.
• Relational vs non-relational databasesPick a database based on how you'll actually use it.
• Sharding and consistent hashingSplit data across machines and keep it balanced.

We wrote this sample answer ourselves.

Practice Loop is independent and isn't affiliated with or endorsed by Stripe.