Design a Distributed Logging System

Collect 10 million log lines a second from 100,000 hosts, make them searchable within seconds, stream them live, and keep a year affordably.

Last updated 2026-10-26. Difficulty: hard. Patterns: log-aggregation, inverted-index, tiered-storage, backpressure, multi-tenancy. Reported at Datadog, Amazon, Google, Microsoft, Uber, Confluent.

Functional requirements

• Collect logs from every host. Application logs (stdout and files, mostly JSON, plenty of free text and stack traces) and infrastructure logs (syslog, Kubernetes, load balancer access logs) from 100 k hosts, each line labelled with service, environment, cluster and host.
• Search by text and fields. For example {service="checkout", env="prod"} |= "card declined" | json | status >= 500. Results come newest first, and a line is searchable within seconds of being written. Most searches name a service and a time range; some hunt for one trace id across everything.
• Live tail. Stream new matching lines for a service as they are written, like tail -f across 2,000 pods, within about two seconds.
• Saved searches that alert. “More than 50 payment timeouts a minute in checkout for 5 minutes” notifies the owning team. The same counts feed dashboards.
• Retention in tiers. 30 days searchable interactively, a year retained for audits and security investigations, with per-tenant policies and legal holds.
• Many tenants, with quotas and cost. Hundreds of teams share the platform. Each has ingest quotas, sees its own volume and cost, and can only read its own logs.
• Parsing and redaction. JSON passes through; known formats (nginx, syslog) are parsed; secrets and personal data are masked before anything is stored.
• Out of scope. Metrics and trace storage (logs link to traces by trace id), the search UI, SIEM correlation rules, and machine-learned anomaly detection beyond a mention.

Non-functional requirements

• Ingest scale (10 M lines/s peak · ~1 PB/day raw). 100 k hosts at ~100 lines a second each, ~1 KB per line. Volume jumps 2 to 3× during incidents, exactly when logs are most needed.
• Never hurt the application (logging never blocks a request). A slow or dead pipeline must cost the application nothing but a growing local buffer. The agent runs under a hard CPU and memory cap.
• Freshness (searchable p99 < 10 s after write). During an incident engineers search for what happened seconds ago. Live tail needs about 2 s.
• Query latency (p95 < 3 s · one service, last hour). Searches over 30 days may take tens of seconds and stream results as they come; searches over data older than 30 days run as background jobs.
• Durability (no acknowledged line lost). At-least-once from the host to storage, with batch ids so retries do not duplicate. Lines are lost only by an explicit, counted policy (dropping debug or sampling info under quota), never silently.
• Cost (storage < 1 cent per raw GB for a year). This is the requirement that drives the hardest trade-off. An inverted index on every field makes any search fast but needs ~60 PB of SSD for 30 days; indexing only labels and scanning compressed chunks is ~25× cheaper for a whole year but makes broad searches slow.
• Tenant isolation (one team’s flood cannot delay another). Per-tenant quotas, shuffle-sharded partitions and fair query queues, so a debug flag left on in one service does not degrade logging for everyone.
• Privacy (no secrets or raw PII at rest). Card numbers, tokens and emails are masked on the host and checked again at ingest; access is per tenant and every search is audited.

Back-of-envelope estimates

• Ingest rate: 10 M lines/s · 10 GB/s raw. 100 k hosts × ~100 lines/s each at peak = 10 M lines/s. At ~1 KB per line that is 10 GB/s (80 Gbit/s) of raw text; zstd at the agent cuts what crosses the network to about 1 GB/s.
• Raw volume per day: ~0.9 PB (plan for 1 PB). 10 GB/s × 86 400 s ≈ 864 TB/day. Assuming the peak holds all day leaves headroom for growth; the rest of the sizing uses 1 PB/day.
• Compressed volume per day: ~100 TB. Log lines repeat the same templates, so zstd over a batch from one stream compresses about 10×: 1 PB ÷ 10 = 100 TB/day, or about 25 M chunks of ~4 MB.
• Index every field for 30 days: ~60 PB of SSD. An inverted index on every field plus the stored line takes roughly the raw size per copy (anywhere from about half the raw size to more than it, depending on mappings and compression): 1 PB × 30 days × 2 copies ≈ 60 PB of SSD, plus hundreds of indexing nodes. This is the number that rules out indexing everything.
• Chunk storage by tier: hot 0.3 PB · warm 3 PB · cold ~34 PB. At 100 TB/day compressed: 3 days on NVMe = 300 TB; 30 days in object storage = 3 PB; days 31 to 365 in an archive class = 335 × 100 TB ≈ 34 PB. At roughly $100, $23 and $4 per TB-month that is about $30 k + $69 k + $134 k ≈ $233 k a month for a full year, against ~$6 M a month for 30 days of SSD index.
• Buffer for a day of catch-up: ~260 TB · 512 partitions. About 1 GB/s compressed into Kafka × 86 400 s ≈ 86 TB, × 3 replicas ≈ 260 TB. At ~10 MB/s per partition the floor is 100 partitions; 512 leaves room for more than a hundred ingesters and per-tenant shuffle sharding.
• Scan for one service-hour: ~360 GB raw · ~1.5 s on 500 cores. A big service such as checkout is ~1 % of volume: 10 GB/s × 1 % × 3 600 s = 360 GB raw (36 GB compressed). Decompressing and substring matching runs at ~0.5 GB/s per core, so 720 core-seconds, ~1.5 s across 500 cores. The same hour across every service is 36 TB: scanning is affordable only when a search names its labels or a bloom filter skips the chunks.
• Agent disk buffer per host: ~28 hours. An average host writes 100 lines/s × 1 KB = 100 KB/s, ~10 KB/s compressed. A 1 GB buffer holds 1 GB ÷ 10 KB/s = 100 000 s ≈ 28 hours, so a pipeline outage of a few hours loses nothing; a noisy host at 10× the average still gets nearly 3 hours.

Components

• Host agents (tail · disk buffer · scrub): One daemon per host (Fluent Bit, Vector or an OpenTelemetry Collector) tails application and system log files, stitches multi-line stack traces, attaches labels, masks secrets and PII, and ships zstd-compressed batches with a batch id. Applications only ever append to local files, so a slow pipeline never blocks them. When the gateway pushes back, batches wait in a capped disk buffer and the agent sheds debug first, then samples info, and counts what it dropped.
• Engineers (UI · CLI): Search, live tail and saved searches from a web UI or CLI, signed in with SSO and authorized per tenant. During an incident hundreds of them run near-identical searches at once, which is the read-side peak to plan for.
• Ingest gateway (auth · quotas · 429): Authenticates each agent to a tenant, checks labels against the tenant’s stream limit, charges its byte quota, and writes the batch to the buffer with acks=all before answering 202. Over quota it returns 429 with a retry-after instead of queueing, so backpressure lands on the agent’s disk, not in gateway memory. Emits bytes accepted and rejected per tenant, service and level as usage metrics.
• Log buffer (Kafka · 24 h · 512 partitions): A partitioned log between collection and indexing, keyed so a stream stays on one partition and each tenant is shuffle-sharded onto a few partitions. Holds 24 hours (~260 TB with three replicas), so indexing can fall behind during an incident and catch up while ingest never slows. Three consumer groups read it independently: ingesters, live tail and the saved-search evaluator.
• Saved-search evaluator (streaming match → counts): A consumer group that matches every line against ~50 k saved searches and counts matches per search per minute. Each line is examined once as it passes, instead of thousands of searches re-scanning the last five minutes every minute. Searches that need parsing and grouping over long windows run as scheduled queries instead.
• Metrics and paging (existing monitoring stack): The company’s metrics system receives log-derived counts and usage meters. Its alert rules, grouping, routing and dead man’s switch page the owning team, so the logging system does not build a second pager.
• Ingesters (head chunks · blooms): Each owns a few partitions. Appends lines to an in-memory head chunk per stream (searchable at once), drops retried batches by their sequence number, and flushes a chunk at ~4 MB compressed or one hour, with a token bloom filter. Kafka is its write-ahead log: the committed offset is the oldest one still in an unflushed chunk, and each flushed chunk records its last offset, so a replay never writes a line twice.
• Chunk storage (object storage · 30 days): The source of truth for every flushed chunk and its bloom filter, at keys like {tenant}/{day}/{hour}/{stream}/{chunk}. About 3 PB for 30 days at roughly $23 per TB-month. Time-partitioned prefixes make retention a prefix delete and per-tenant policies trivial.
• Live tail (own consumer group · WebSocket): Tail workers read every partition once from the head of the buffer and match each line against the active tail sessions, indexed by tenant and labels; matches go to the node holding the engineer’s WebSocket. Independent of indexing, so tail stays live while ingesters are behind. Each session is capped at 500 lines a second and reports how many it skipped.
• Chunk index (labels → chunks, per tenant-day): For each tenant and day: label pairs to streams, and streams to chunk references with time range, size and last offset. About 200 bytes per chunk, ~5 GB a day for the whole company against ~1 PB of raw logs. Partitioned by tenant and day, so retention drops a whole index partition, never individual rows.
• Lifecycle job (tier · expire): Daily, per tenant: moves day partitions to the archive class at day 30, deletes chunk prefixes and index partitions when the tenant’s retention ends, and skips anything under legal hold. Works on whole tenant-days, so a year of retention is a few thousand prefix operations, not billions of row deletes.
• Query frontend (split · budgets · fair queues): Parses the query, splits its time range into hourly sub-queries, estimates the bytes to scan from the index and refuses or defers anything over the tenant’s budget. Dispatches the newest hours first with bounded parallelism, merges results in time order, and stops as soon as it has the requested number of lines. Queues per tenant, so one team’s incident cannot starve another’s.
• Queriers (stateless · scan): Stateless workers that run sub-queries: find chunks in the index, check bloom filters for rare tokens, fetch surviving chunks from the hot cache or object storage, decompress, apply cheap substring filters before any parsing, and stream matches back. About 0.5 GB/s of raw text per core; the fleet autoscales on queue wait, because query load rises 5 to 10× during incidents.
• Hot chunk cache (NVMe · last 3 days): A read-through NVMe cache on the querier nodes, about 300 TB, pre-warmed with newly flushed chunks and blooms. Most searches look at the last hours or days, so they read local SSD in about a millisecond instead of tens of milliseconds from object storage. Losing it costs latency, never data.
• Cold archive (archive class · 1 year): Chunks older than 30 days in an archive storage class at a few dollars per TB-month, about 34 PB. The index and bloom filters for this data stay in standard storage, so a cold search can work out which chunks it needs and restore only those, in minutes to hours.

User flows

1. A log line goes from a host to searchable. The write path that runs 10 million times a second: append locally, batch and buffer on the host, acknowledge only once the line is durable in the buffer, build compressed chunks per stream, and store them in time partitions that can be tiered and dropped whole.
   1. The application writes a JSON line to stdout, and the container runtime appends it to a file on the host. The application never talks to the network to log: an append to a local file takes microseconds and cannot block on a slow pipeline. If the agent is down, the file grows and rotates as usual; the only risk is a file rotated away before the agent reads it, which the agent counts and reports.
   2. The agent tails the file, stitches multi-line stack traces, adds labels, and masks secrets and personal data. It records (inode, offset) per file every second, so a restart resumes where it stopped, and follows a rotated file by inode until its end. Masking runs before the line leaves the host: card numbers (a pattern plus a Luhn check), bearer tokens, API keys and emails are replaced by typed markers such as [REDACTED:card], and user ids by a keyed hash so they can still be correlated.
   3. The agent batches lines per stream for up to 1 s or 1 MB, compresses them, and sends them with a batch id. Batching turns 10 M lines a second into about 100 k requests a second, one per host per second, and zstd shrinks 10 GB/s to 1 GB/s on the wire. Each batch is written to the agent’s disk buffer first and deleted only after a 202, and a stream’s batches are sent in order, so after a crash the agent re-sends the same batch with the same id and sequence.
   4. The gateway authenticates the tenant, charges its quota, and writes the batch to the buffer before acknowledging. The 202 is sent only after Kafka acknowledges with acks=all, so an acknowledged line is on at least two brokers. The partition is chosen inside the tenant’s shuffle shard (a few of the 512 partitions) by hashing the stream, which keeps one stream on one partition, in order, so a single ingester builds its chunks.
   5. An ingester consumes the partition, drops batches it has already stored, and appends lines to the stream’s in-memory head chunk. A retried batch hashes to the same partition, so the ingester that owns it sees both copies. Because each agent sends a stream’s batches in order, the ingester only keeps the highest sequence stored per (agent, stream), about a million pairs and tens of MB across the fleet, and drops anything at or below it. The head chunk is searchable from memory at once: about 3 to 5 seconds from write to searchable. There is no separate write-ahead log; the buffer is the log.
   6. When a chunk reaches about 4 MB compressed or an hour old, the ingester builds a bloom filter of its tokens and uploads both to object storage. The key is time-partitioned per tenant, so retention and tiering act on whole prefixes. The bloom filter holds every distinct token (about 100 k per chunk) at a 1 % false-positive rate: ~120 KB, about 3 % of the chunk, which is what lets a trace-id search skip 99 % of chunks without an inverted index.
   7. The ingester records the chunk in the tenant’s daily index, then commits its offset in the buffer. The committed offset is the oldest offset still sitting in an unflushed chunk on that partition, so everything after it is replayed after a crash. Each chunk ref stores its last offset, and on replay the ingester skips any batch for a stream at or below that stream’s flushed offset: replays never write a line twice.
   8. As the data ages, the lifecycle job moves the tenant’s day partitions to the archive tier and deletes them when retention ends. The SSD cache drops chunks after about 3 days on its own. At day 30 the day’s chunk prefix moves to the archive class while its index and blooms stay in standard storage; at day 365 the prefix and index partition are deleted. A legal hold on a tenant-day simply skips it.
2. Find a trace id in a day of logs. The read path: narrow by labels and time with a tiny index, skip almost every chunk with bloom filters, scan the survivors in parallel from SSD or object storage, include what is still in ingester memory, and stop as soon as the answer is complete.
   1. An engineer searches checkout’s production logs over the last 24 hours for a trace id, newest first, limit 100. The selector in braces is matched against labels and is what the index can answer; the |= filter is a substring match on the line text, which only a scan or a bloom filter can answer.
   2. The frontend splits the range into hourly sub-queries, newest first, and checks the cost against the tenant’s budget. Index stats give the bytes before anything runs: checkout is ~36 GB compressed an hour, ~860 GB for the day. Over the per-query budget (say 10 TB raw) the search is refused with a suggestion or run as a background job. Sub-queries go out a few hours at a time, so a match in the newest hour can end the search early.
   3. Queriers look up which of checkout’s chunks overlap each hour. Label pairs map to streams and streams to chunk refs, so no line text is touched. Checkout writes ~36 GB of compressed chunks an hour; at ~4 MB each that is roughly 9,000 chunks an hour, ~216 k for the day.
   4. Queriers check each chunk’s bloom filter for the trace id; about 99 % of chunks are skipped. Bloom filters are ~3 % of chunk bytes, so checking all 216 k reads ~26 GB, mostly from the SSD cache, instead of 860 GB of chunks. At a 1 % false-positive rate about 2,200 chunks (~9 GB) survive, nearly all of them false positives. A common word such as "error" matches every bloom, which is why level is a label.
   5. Queriers fetch the surviving chunks from the SSD cache, or object storage on a miss, then decompress and grep them in parallel. ~9 GB compressed is ~90 GB of text: 180 core-seconds at 0.5 GB/s per core, under a second on 500 cores. The substring filter runs before any JSON parsing, because a SIMD substring search runs at gigabytes a second and a JSON parser at a few hundred megabytes. Chunks younger than 3 days come from local NVMe in about a millisecond; older ones take tens of milliseconds from object storage.
   6. For the newest hour, queriers also ask the ingesters for lines still in head chunks. Up to an hour of each stream lives only in ingester memory until its chunk is flushed. Only the ingesters that own the tenant’s shuffle shard are asked, a handful rather than all of them. Lines present both in memory and in a just-flushed chunk are deduplicated by (stream, timestamp, line hash).
   7. The frontend merges results newest first and stops once it has 100 lines, cancelling the sub-queries still running. Each chunk is sorted by time, so a k-way merge yields lines in order and can stream them as they arrive. For “latest 100 errors” the newest hour usually suffices and the other 23 never run; a rare trace id needs every hour, which the blooms keep cheap. The stats line shows bytes scanned, which teaches engineers which searches are expensive.
3. One service floods its logs. The scale-breaking case: a deploy turns on debug logging and one service’s volume rises 20×. Quotas, shedding at the source and shuffle sharding keep the damage inside that tenant, and every dropped line is deliberate and counted.
   1. A bad deploy turns on debug logging in checkout, and its volume jumps 20× to 2 GB/s raw. Checkout alone is now a fifth of the whole platform’s normal volume. The application is unaffected, because its writes are local appends. Each agent works through the extra volume within its CPU cap and never takes cores from the application.
   2. The gateway charges the payments tenant’s byte quota and answers 429 once it is used up. Quota is raw bytes a second per tenant (and per service inside it), with a 2× burst for 10 minutes. Each gateway node enforces its share as a local token bucket, so no shared counter sits on the ingest path. Rejecting at the edge keeps the flood out of Kafka, the ingesters and the bill.
   3. Agents keep rejected batches on disk and, as the buffer fills, drop debug first, then sample info, and never drop warn or error. Shedding happens at the source, where the level is known and nothing has been paid for yet. Info is kept 1 in 10 with a sample_rate=10 label, so counts can be scaled back at query time. The agent reports exactly how many lines it dropped or sampled, per service and level.
   4. The gateway records the tenant’s accepted and rejected bytes as usage metrics, and the owning team is alerted. Usage per tenant, service and level is what cost attribution is built on: bytes ingested, GB-days stored per tier and bytes scanned, each with a price. The team sees its own flood on its own dashboard before anyone else notices.
   5. Other tenants’ batches flow normally, because their partitions and ingesters are mostly different ones. Each tenant hashes to 8 of the 512 partitions (big tenants get more). Another random tenant shares at least one of those 8 only about 12 % of the time, and all 8 essentially never, so a neighbour sees at most a slower partition or two rather than an outage. Ingesters also cap active streams and head-chunk memory per tenant.
   6. Searches during the incident are queued per tenant and capped by bytes scanned. Three hundred engineers on one incident open the same searches. Identical in-flight sub-queries are coalesced, completed hours are cached, and each tenant has a concurrency cap, so payments’ incident cannot starve the fleet for everyone else.
   7. After the fix, agents drain their disk buffers at the quota rate, oldest first. Backlogged lines keep their original timestamps and land in the right hourly partitions; ingesters accept lines up to two hours out of order per stream, and older ones go into a separate late chunk for that hour. Draining at the quota rate stops the recovery from becoming a second flood.
4. Indexing falls behind and catches up. The failure path: an ingester crashes with unflushed chunks, and later the whole indexing tier falls half an hour behind. The buffer absorbs both; ingest never slows, nothing is lost or duplicated, and tail and alerts stay live.
   1. An ingester dies with 40 minutes of unflushed head chunks in memory. Those lines are not lost: they are still in Kafka, because the ingester had not committed past them. Unlike the metrics system’s ingesters, these keep no in-memory replicas, which saves two thirds of ingester memory at the cost of a few minutes in which recent searches on these partitions are incomplete.
   2. Its partitions are reassigned, and the new owner resumes from the last committed offset. Four partitions at ~2 MB/s compressed for 40 minutes is about 20 GB, replayed at a few hundred MB/s in a couple of minutes. The committed offset was the oldest offset in any unflushed chunk, so everything that might be missing is after it.
   3. During replay, batches already in flushed chunks are skipped using each stream’s flushed offset. A busy stream may have flushed a chunk after the committed offset was recorded. Its chunk ref carries its last offset, so the replaying ingester skips that stream’s batches up to it while still learning their sequence numbers, so an agent’s retry is still recognised after the crash.
   4. Live tail and saved-search alerts keep running, because they read the buffer with their own consumer groups. This is the main reason tail and alerts are not queries against the index: when indexing is the thing that broke, the incident tooling still sees every line within seconds.
   5. Later, a region-wide incident triples log volume and every ingester falls 30 minutes behind; the buffer absorbs it. The gateway keeps acknowledging, because the write path ends at the buffer, not at the indexer. With 24 hours of retention, indexing could fall a full day behind before any data is at risk. Consumer lag measured in seconds per partition is the alert, not lag in messages.
   6. Ingesters catch up by consuming faster than real time, and more ingesters are added up to the partition count. Run at about 50 % utilisation so they can consume at 2× real time; a 30-minute lag then clears in about 30 minutes once volume returns to normal. Having 512 partitions for ~128 ingesters means the fleet can be scaled out 4× in an emergency without repartitioning.
   7. Until then, search results for the lagging window carry a warning naming the partitions and their lag. A search that silently returns nothing for the last 30 minutes is worse than one that says it is incomplete. The UI shows “indexed up to 13:42” and points to live tail for anything newer.
5. Live tail and a saved-search alert. The streaming paths. Both read the buffer from its head with their own consumer groups and do their work once per line as it passes, instead of polling the index, so they are live within seconds and independent of indexing.
   1. An engineer opens a live tail on checkout’s errors containing “timeout”. A tail session is a standing filter, not a query: it has no time range and never reads stored data.
   2. The session is registered with every tail worker, indexed by tenant and labels. Even during a big incident there are a few thousand sessions, so every worker can hold all of them. Each worker owns some partitions of the buffer in the "tail" consumer group, so between them they see every line exactly once.
   3. Tail workers read new batches from the head of the buffer and match them against their tenant’s sessions. Decompressing the whole 10 GB/s once costs about 20 cores. Labels are already on the stream, so the label match is per batch, and only matching batches have their text checked against the filter. A worker that falls more than ~10 seconds behind skips to the head, because a tail is about now.
   4. Matches are pushed to the engineer’s WebSocket within about two seconds, sampled above the session’s rate cap. Latency is the agent’s 1 s batch plus tens of milliseconds through the gateway, Kafka and the worker. Nobody can read more than a few dozen lines a second, so above 500 lines a second the tail samples and says how many it skipped, suggesting a narrower filter.
   5. Separately, the saved-search evaluator matches every line against 50 k saved searches and counts matches per minute. Running 50 k searches every minute over the last five minutes would scan each line five times and add ~800 searches a second to the query fleet. Streaming evaluation looks at each line once, with searches indexed by tenant and labels and their literal filters compiled into one multi-pattern matcher per tenant.
   6. The counts are written as metrics, and the monitoring system’s alerting pages the owning team. The metrics stack already has “for 5 minutes”, grouping, silences, routing and a dead man’s switch; building a second pager would duplicate all of it badly. The notification links to a search over the matching window, so the on-call engineer lands on the lines.

Deep dives

The theory behind it

• How Elasticsearch works: Elasticsearch and OpenSearch internals for system design: inverted indexes and analyzers, Lucene segments, refresh and near real-time search, shards and replicas, scatter-gather queries, relevance scoring with BM25, indexing pipelines and common scaling mistakes.
• Distributed tracing and structured logging: Following one request through dozens of services: traces, spans and context propagation, OpenTelemetry, sampling strategies including tail-based sampling, correlating logs, metrics and traces, structured logging, log pipelines, cardinality and cost.
• Load shedding and backpressure: Staying up when demand exceeds capacity: why overloaded systems collapse, backpressure in queues and streams, bounded buffers, load shedding by priority, adaptive concurrency limits, deadlines, brownouts and graceful degradation, and retry storms.

Related

• Design Zapier
• Design Google Search
• Design Gmail
• Design a RAG System
• All 83 system design interview questions
• Run this question for a friend: the interviewer kit