Design a distributed logging system

It is 8:02 pm on the first night of a Big Billion Days–style sale and checkout errors have jumped from 0.1% to 3%. The metrics dashboard tells you that something is wrong; a log line tells you what: payment timeout calling upi-switch after 3000 ms, order ORD-4711. That line is one of 86 billion your 100,000 containers write today. The logging platform must collect it without slowing the app, keep it safe while half the backend is on fire, and show it to an engineer seconds after a search. The economics are upside down: more than 99% of logs are never read, yet the few that are read matter enormously, so every byte must be cheap to keep but quick to find. Meanwhile one crash-looping service can log 100 times its normal volume, every team invents its own JSON fields, and auditors want a year of history.

Understand the problem

A logging platform is a write-heavy pipeline with an occasional, urgent reader. Say what logs are for before designing anything. A log is one record per event — “this request failed, for this reason, with these details” — so its volume grows with traffic, unlike a metric, which costs the same at 10 or 10 million requests a second. Metrics tell you that the error rate rose; logs and traces tell you which requests and why. The metrics side has its own design in design metrics, monitoring and alerting; this one is about the per-event side. Cloudflare’s SRE team sums up the shape: logs are unpredictable in volume, semi-structured, needed in context, and write-heavy — more than 99% are never read before retention deletes them.

Functional requirements

1. Services and hosts should be able to ship logs without code changes. Anything written to stdout, a file or the system journal is collected and tagged with service, environment, pod, host and region.
2. Engineers should be able to search logs by service, level, time range, field values, free text and trace ID, and see matching lines in seconds.
3. Engineers should be able to live-tail one service’s logs as they are written, with filters.
4. Teams should be able to alert on log patterns, for example “more than 50 payment timeout lines from checkout in 5 minutes”.

Non-functional requirements

• Scale. 20,000 hosts and 100,000 containers running about 3,000 services for 400 teams. About 1M lines/s on average and 3M at peak — sale launches, IPL finals and incidents all multiply logging. Lines average about 500 bytes.
• Freshness. A line is searchable within 30 s at p99 and appears in live tail within 5 s.
• Query latency. One service, last hour: under 1 s at p50 and 5 s at p99. Seven days: under 10 s. Thirty days: under a minute, or a background job.
• No silent loss. Once an agent has read a line, it reaches storage or is dropped by an explicit policy (quota, sampling) and counted. Rare duplicates are acceptable.
• Availability. Ingestion 99.95%, and applications never block or crash because logging is slow. Search 99.9%.
• Retention. 30 days searchable; 1 year archived for audit and restorable within 4 hours.
• Isolation and cost. One team’s flood or monster query must not delay other teams, and every team sees what its logs cost.
• Security. Secrets and personal data are removed on the node; teams see only their own logs; every search is audited.

Out of scope

• Metrics and time-series storage. Numbers sampled on a schedule belong in a TSDB; see design metrics, monitoring and alerting.
• Storing traces. We carry trace and span IDs on every line so logs join traces, but span storage and trace views are a separate system (background in observability: logs, metrics and traces).
• Security analytics (SIEM). Threat-detection rules and long correlation windows have different query patterns and buyers.
• Product analytics. Clickstream and funnels need exact, business-grade counting; see design web analytics.
• Building Kafka itself. We use it as a component; its internals are in design a distributed message queue.
• The log viewer UI. Treated as a client of the query API.

The set-up

Plan the approach

Draw one pipeline and three readers: collect → buffer → store, read by search, live tail and alerts. Build each functional requirement simply, then spend the deep dives on what decides cost and reliability: indexing, not losing data, messy schemas, noisy teams, a year of retention, and fast, fair queries.

Core entities

• Log record. One event, shaped like the OpenTelemetry logs data model: Timestamp (when it happened), ObservedTimestamp (when the agent saw it), SeverityText and SeverityNumber (1–24, from TRACE to FATAL), Body (the message), TraceId and SpanId, Resource (service, pod, host, region) and free-form Attributes.
• Tenant. Usually a team. Owns services and carries quotas (bytes per second, bytes per day), a retention class, an access list and a cost centre.
• Stream. Lines from one tenant, service and environment: the unit of routing, live tail and per-service limits.
• Part. An immutable, compressed block of rows covering one time range of one table, with its own small indexes. Parts are merged, moved between storage tiers and deleted whole.
• Alert rule. A filter, a window, a threshold and where to send the alert.
• Archive file. Raw lines for one tenant, service and hour, compressed, in object storage, plus a manifest entry describing it.

API

There are five surfaces: ingest, search, tail, rules and archive restores. Agents authenticate with mTLS client certificates. Ingest is shown as JSON for readability; in production it is OTLP over gRPC or HTTP, as zstd-compressed protobuf.

POST /v1/logs
Content-Type: application/json
Content-Encoding: zstd

{
  "resource": { "tenant": "payments", "service": "checkout", "env": "prod",
                "region": "ap-south-1", "pod": "checkout-7c9f-x2k4" },
  "records": [
    { "ts": "2026-10-27T14:32:02.118Z", "observed_ts": "2026-10-27T14:32:02.231Z",
      "severity": "ERROR", "severity_number": 17,
      "body": "payment timeout calling upi-switch after 3000 ms",
      "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
      "attributes": { "order_id": "ORD-4711", "bank": "HDFC", "latency_ms": 3000 } }
  ]
}

The gateway answers 200 only after Kafka has the batch, 429 with Retry-After when the tenant is over quota (the agent keeps the batch on disk and retries), and 413 for an oversized batch.

POST /v1/search
{ "tenant": "payments", "from": "2026-10-27T13:30:00Z", "to": "2026-10-27T14:30:00Z",
  "query": "service:checkout level:error \"payment timeout\"",
  "limit": 500, "order": "newest_first" }

{
  "lines": [ { "ts": "2026-10-27T14:29:58.004Z", "service": "checkout",
               "level": "ERROR", "body": "payment timeout calling upi-switch…",
               "trace_id": "4bf92f35…" } ],
  "stats": { "bytes_scanned": 2140000000, "partial": false, "took_ms": 640 },
  "next_cursor": "eyJ0cyI6IjIwMjYtMTAtMjdUMTQ6Mjk6NTgifQ"
}

GET  /v1/tail?tenant=payments&query=service:checkout level:error   upgrades to WebSocket
PUT  /v1/alert-rules/payment-timeouts     body: query, window 5m, threshold > 50, notify
POST /v1/archive/restores                 body: tenant, services, from, to → 202 + job id
GET  /v1/archive/restores/{job_id}        status, bytes restored, scratch table name

The stats block matters: showing engineers how many bytes a query scanned is the cheapest way to teach them to narrow it.

Back-of-envelope estimates

These numbers decide the storage engine, the size of Kafka and the store’s node count (method: back-of-envelope estimation).

Lines. 100,000 containers at about 10 lines/s: 100,000 × 10 = 1M lines/s on average, and 3× at peak: 3M lines/s. Per host: 1M ÷ 20,000 = 50 lines/s, about 25 KB/s — trivial for one node; the hard part is the aggregate.

Bytes. At 500 bytes a line (JSON with a dozen or so fields): 1M × 500 B = 500 MB/s on average, 1.5 GB/s at peak, about 12 Gbit/s.

Per day. 500 MB/s × 86,400 s = 43.2 TB of raw logs a day, or 86.4 billion lines.

Compressed. Reported compression for logs ranges from 3× to 30× (Uber), 5–10× for Loki’s chunks and 10× in Cloudflare’s migration. Plan with 8×: 43.2 ÷ 8 = 5.4 TB/day. Add an assumed 10% for skip indexes, the text index and materialised columns: ≈ 5.9 TB/day. Measure both on your own data.

Searchable for 30 days. The last 7 days on local NVMe with 2 replicas: 7 × 5.9 × 2 ≈ 83 TB of NVMe. Days 8–30 in object storage, which handles its own durability: 23 × 5.9 ≈ 136 TB. Total ≈ 219 TB.

Archive for a year. Raw lines at the same 8×: 365 × 5.4 ≈ 2 PB in an archive storage class.

The full-text alternative. Elastic’s tests found indexes 0.43× to 1.39× the raw size; call it 1× with the message analysed. With one replica for 30 days: 43.2 × 2 × 30 ≈ 2.6 PB of SSD, about 12× the 219 TB above.

Kafka. Agents’ small batches compress worse than big columnar parts; assume 4×: 500 MB/s ÷ 4 = 125 MB/s. For 72 hours: 125 MB/s × 259,200 s ≈ 32.4 TB, ×3 replicas ≈ 97 TB, or 12 brokers of about 8 TB each. Peak replicated writes: 375 MB/s × 3 ≈ 1.1 GB/s, about 94 MB/s per broker.

Store nodes. Uber measured about 300,000 logs/s on one ClickHouse node. Budget half, leaving CPU for merges and queries: 3M ÷ 150k = 20 shards, ×2 replicas = 40 nodes, each holding 83 TB ÷ 40 ≈ 2.1 TB of hot data. CPU, not disk, sets the count.

A typical query. One hour of everything is 43.2 TB ÷ 24 = 1.8 TB raw. A service with 1% of the volume has 18 GB raw in that hour, ≈ 2.25 GB compressed — about 110 MB per shard, a sub-second scan even without index help.

High-level design

Build the pipeline one requirement at a time; by the end it works for one region, before worrying about floods, outages or cost.

Ship logs without code changes

Applications should not know the platform exists. They write to stdout, the container runtime writes that to a file on the node, and an agent — Fluent Bit, Vector or the OpenTelemetry Collector, one per node — tails the files. Logging libraries that post over the network tie the app’s fate to the backend; a local file plus a local agent cuts that cord.

1. The app writes a line to stdout; the runtime appends it to a file on the node.
2. The agent tails the file, joins multi-line stack traces, parses JSON, adds pod, namespace and region from the Kubernetes API, maps the namespace to a tenant, and strips secrets.
3. It batches for about a second, compresses with zstd and sends to the ingest gateway; if the gateway is slow or says no, the batch waits in a disk buffer and is retried with backoff.
4. The stateless gateway authenticates the agent, validates sizes and timestamps, checks the tenant’s quota, normalises records to the OpenTelemetry shape and writes to Kafka with acks=all, answering 200 only once Kafka has the batch.

Kafka sits in the middle because it absorbs bursts the store cannot take, lets several readers consume the same data independently, and allows replay when a later stage mangles a day of logs.

Search logs in seconds

Writers consume Kafka, flatten each record into columns and insert large batches into a log store: a sharded columnar database (justified in the deep dives), sorted by tenant, service and time, in hourly partitions, with a text index on the message and a bloom filter on the trace ID. A query service turns searches into SQL.

1. Writers accumulate rows for 1–5 seconds or about 100,000 rows. Small inserts create many small parts the store must keep merging, which is why Cloudflare calls batch size a key performance lever.
2. They insert each batch and only then commit their Kafka offsets — the at-least-once pattern Uber uses.
3. The query service checks the engineer’s access to the tenant and generates SQL. For the example search:

SELECT ts, service, level, body, trace_id
FROM logs_30d
WHERE tenant = 'payments' AND service = 'checkout'
  AND ts >= '2026-10-27 13:30:00' AND ts < '2026-10-27 14:30:00'
  AND level = 'ERROR'
  AND hasToken(body, 'payment') AND hasToken(body, 'timeout')
ORDER BY ts DESC
LIMIT 500

4. Each shard prunes to the two hourly partitions the range touches, reads only checkout’s rows via the sort key and skips blocks without both words via the text index; the query service merges the newest 500 lines.
5. A second consumer group, the archiver, writes the raw lines to object storage, one compressed file per tenant, service and hour, for audits and restores.

Live-tail a service

Polling the store every second is a poor tail: it is only as fresh as the writers’ batches, and 500 engineers polling would hammer it. Tail from Kafka instead. The gateway keys every record by tenant and service, so checkout’s lines land in a small, known set of partitions. A tail service reads just those from the newest offset, filters, and pushes matches over a WebSocket (see real-time connections).

1. The browser opens GET /v1/tail?query=…, upgraded to a WebSocket; access is checked as for search.
2. The service looks up which partitions hold payments/checkout — two here, more for busy services — and reads them without joining a consumer group, so it never disturbs the writers.
3. Above 200 lines a second it sends a dropped-line count instead: no human reads faster, and a slow browser must not make the server buffer without limit.

Alert on log patterns

The obvious way is a scheduled query per rule: 5,000 rules every minute is 5,000 ÷ 60 ≈ 83 queries a second, each only as fresh as ingestion. But most log alerts are counts of matching lines, which can be computed as lines fly past. So add another Kafka consumer, a rule engine, that matches lines against compiled rules and counts matches per rule per minute.

1. Rules use the search language, are reviewed in CI and compiled into matchers grouped by tenant and service, so each line is tested only against its own service’s rules.
2. The engine emits counters such as log_matches{rule="payment-timeouts"} every minute into the metrics platform, whose alerting handles thresholds, grouping and paging (see design metrics, monitoring and alerting).
3. The rule language is deliberately narrow — fields, tokens, bounded regexes. A rule needing a 7-day window becomes a scheduled query every 15 minutes instead.

You now have a working platform. The deep dives are about what breaks at 43 TB a day.

Deep dives

How should we index 43 TB a day so search is fast but storage stays affordable?

This decision sets the bill. There are three families, and real platforms have moved between them as they grew.

Bad (at this scale): a full-text inverted index on every field. The classic ELK set-up: every token of every field goes into an inverted index (see search data structures), so any word is found in milliseconds. You pay three ways. Disk: Elastic’s storage tests measured indexes from 0.43× to 1.39× the raw size, the high end with the message field analysed — the field you most want searchable. With a replica, 30 days is about 2.6 PB of SSD. CPU: every line is tokenised and merged repeatedly; Uber measured a ClickHouse node ingesting about ten times what an Elasticsearch node could, and before migrating ran 20+ Elasticsearch clusters per region yet could not afford to ingest all its logs. Waste: only 5% of Uber’s indexed fields were ever queried. Below a few TB a day of genuinely free-text searching, it is still a fine choice (see Elasticsearch in depth).

Good: index only labels, brute-force the rest. Grafana Loki’s approach. Index a few low-cardinality labels — tenant, service, environment — and store each stream’s lines in compressed chunks in object storage; the label guidance notes a 1 MB chunk holds about 5–10 MB of logs. The index is tiny and storage is close to compressed size. A query picks streams by label, then decompresses and greps every chunk in range: fast when labels narrow the data (one service, one hour: about 18 GB raw), slow when they do not — “find ORD-4711 anywhere yesterday” scans all 43.2 TB. Aggregations, over 80% of Uber’s queries, re-parse JSON on every read. And a label such as request_id creates a stream per value, exploding the index.

Great: a columnar store with a sort key, skip indexes and a text index on the message. Store each field as a compressed column, sorted by (tenant, service, ts), in hourly partitions — what Cloudflare and Uber both moved to with ClickHouse. A sparse primary index keeps one entry per block of rows (8,192 by default), so a tenant-and-service filter reads only that service’s blocks. Per-block skip indexes — min/max on time, a bloom filter on trace_id — skip blocks that cannot match. For free text, ClickHouse’s per-part inverted text index, generally available from version 26.2, maps tokens to rows, so hasToken(body, 'timeout') reads only matching blocks. You gain compression close to the label approach (Cloudflare went from 600 bytes per Elasticsearch document to 60 per row), needle queries that skip almost everything, and aggregations that read two columns instead of whole documents. You pay with no relevance ranking (logs rarely need it; a web search engine does), weak regex and leading-wildcard search, extra index bytes, and an engine that punishes small inserts.

Pick the columnar store, with a text index on body only and a bloom filter on trace_id: about 12× less disk than full-text indexing, and the queries engineers actually run — one service, a time range, a level, a few words or an ID, and counts — are exactly what sorted columns serve well (background in OLTP vs OLAP and columnar storage).

How do we avoid losing logs when the backend is slow or down?

Logs matter most during incidents, exactly when the backend is most likely to be overloaded. The pipeline has to bend, not break.

Bad: apps send logs straight to the backend, or agents buffer only in memory. A library posting to an HTTP endpoint either blocks request threads when it is slow — a logging outage becomes a checkout outage — or drops lines when its queue fills. A memory-only agent has the same problem one hop later: the Fluent Bit backpressure docs explain that hitting the memory limit pauses the input, and warn that some inputs lose data then. A restart loses whatever was in memory.

Good: tail files, buffer to disk, retry with backoff. The node’s file is the first buffer: a restarted agent resumes from its saved offset. Its disk buffer is the second: unsent batches wait on disk and survive restarts. At 25 KB/s per node, a 2 GB buffer lasts 2 GB ÷ 25 KB/s ≈ 80,000 s, about 22 hours — but only about an hour on a node whose service logs 20× normal. Decide in advance what happens when it fills; Fluent Bit can cap an output’s queue and discard the oldest chunks. You pay with disk on every node, and lag: nothing waiting in 20,000 buffers is searchable.

Great: disk-buffered agents, a replicated Kafka in the middle, and idempotent writes. Kafka with three replicas, acks=all and 72 hours of retention turns “the store is down” from a data-loss event into a freshness event. Agents free their buffers as soon as Kafka has a batch; writers catch up when the store returns; and since consumer groups read independently, a broken store never stalls tail, alerts or the archive. You pay with a 12-broker cluster, a few seconds of latency, and a rule: this Kafka must be logging’s own, or the incident that breaks production also blinds you. Uber notes the flip side: if the write path stays down longer than retention, logs are deleted from Kafka, so the store’s write path gets the strictest reliability budget.

The arithmetic: 1M lines/s × 2,400 s = 2.4 billion lines (1.2 TB raw) behind. Writers sized for a 3M/s peak comfortably insert 2M/s while 1M/s keeps arriving, so the backlog shrinks by 1M lines/s and clears in another 40 minutes. Kafka’s 72 hours hold 1M × 259,200 ≈ 259 billion lines.

Duplicates. A writer that crashes after inserting a batch but before committing offsets will insert it again. Make the retry identical so the store can recognise it: build each batch from a fixed Kafka offset range with a token such as logs.raw/p17/81200000-81299999; replicated ClickHouse tables remember recent insert tokens and skip repeats. As a backstop, the agent stamps each record with an ID from node, file and byte offset, and queries collapse rare duplicates. This is the effectively-once pattern; true exactly-once is not worth its cost here.

How do we store free-form JSON without type conflicts or field explosions?

Three thousand services log whatever fields their authors like — Uber reports 40+ per log on average. One team logs status as the number 200, another as the string "OK"; one service puts user IDs into field names. The schema strategy decides whether that is a non-event or an outage.

Bad: map every new field dynamically. Elasticsearch’s default creates a typed mapping for each new key. The first type seen wins and conflicting lines are rejected — at Uber, type conflicts dropped logs and became a major source of on-call work. Mappings grow without bound, so an index is capped at 1,000 fields by default, and documents that would exceed it fail unless you use a flattened field or ignore extras. Cloudflare saw this mapping explosion eat heap and trigger frequent garbage collection.

Good: a fixed core schema plus one catch-all field. Give the OpenTelemetry top-level fields — timestamp, severity, body, trace and span IDs, service, pod — real columns, and put all other attributes in one JSON string. Nothing explodes or is rejected. You pay at query time: every attribute filter parses JSON for every row. Uber tried this first and found it too slow; a column per field was 50× faster, but with thousands of fields the files and merges overwhelmed the store.

Great: typed key–value arrays, plus materialised columns for hot fields. Uber’s schema-agnostic design flattens each record into key–value pairs grouped by value type, each type stored as two parallel arrays — string_names and string_values, number_names and number_values — so the column count depends on the number of types, not fields. A field seen as both number and string appears in both arrays; a schema store records the types seen per field, and queries resolve them at read time. Cloudflare recommends the same layout beyond 1,000 fields. Array lookups were about 5× faster than parsing JSON in Uber’s tests; for constantly queried fields, add materialised columns filled in the background, chosen by measured query frequency. You pay with a type-aware query planner, slower access to rare fields, and a schema store to run.

As a table definition, in ClickHouse’s dialect and trimmed to the essentials:

CREATE TABLE logs_30d
(
    tenant         LowCardinality(String),
    service        LowCardinality(String),
    ts             DateTime64(3),
    level          LowCardinality(String),
    trace_id       String,
    body           String,
    string_names   Array(LowCardinality(String)),
    string_values  Array(String),
    number_names   Array(LowCardinality(String)),
    number_values  Array(Float64),
    order_id       String MATERIALIZED string_values[indexOf(string_names, 'order_id')],
    INDEX body_text body     TYPE text(tokenizer = splitByNonAlpha),
    INDEX trace_bf  trace_id TYPE bloom_filter(0.01) GRANULARITY 1
)
ENGINE = ReplicatedMergeTree
PARTITION BY toStartOfHour(ts)
ORDER BY (tenant, service, ts)
TTL toDateTime(ts) + INTERVAL 7 DAY TO VOLUME 'object_store',
    toDateTime(ts) + INTERVAL 30 DAY DELETE
SETTINGS storage_policy = 'nvme_then_object_store';

Where to parse. Keep the agent’s work small and universal — join stack traces, detect JSON, add metadata, redact — because agent CPU is stolen from production pods, and a bad rule there needs a fleet-wide rollout to fix. Flatten and type centrally in the writers, where a bug is fixed with one deploy and a Kafka replay, and resolve types at query time. Uber’s reasoning: the write path has almost no error budget and the query path has plenty, so put clever logic where mistakes are cheap.

What happens when one service starts logging 100× normal?

A bad deploy sends checkout into a crash loop. Its 400 pods normally write 4,000 lines/s in total; now each failing request dumps a 30-line stack trace: 400,000 lines/s, 40% of the platform’s average load. On a shared pipe, writers fall behind and every team’s logs go stale — a noisy neighbour inside your observability stack, on the day people need it most.

Bad: one shared pipe, first come, first served. Simple, and fair only in that everyone suffers: freshness degrades for all 400 teams, and the platform pays to store a million copies of one stack trace.

Good: per-tenant quotas at the gateway. Give each tenant a byte rate and burst, enforced with a token bucket (see design a rate limiter), and return 429 when exceeded. Agents keep the batch on disk and retry, so the flood waits at its source while everyone else flows. Loki’s distributors divide a tenant’s global limit by the number of distributors — 10 MB/s across 10 distributors is 1 MB/s each — so it rescales as they are added; Uber rate-limits each tenant too. You pay with bluntness: if checkout stays over its limit, its agent buffers fill and drop the oldest lines, including the errors needed to debug the crash.

Great: layered, severity-aware limits with showback.

• At the agent: join stack traces into one record, collapse identical lines within a second into one line plus a repeat_count, and cap each container (say 200 lines/s, with a burst). The same trace printed a thousand times becomes a handful of records, and the agent counts what it dropped.
• At the gateway, by severity: when a tenant is over quota, drop DEBUG first, then sample INFO — keep 1 in N and write sample_rate = N on the kept lines so counts can be re-weighted — and keep a reserved share for WARN and above. Cloudflare applies a related idea to dashboards, keeping copies of its data at 100%, 10%, 1% and 0.1% and answering each query from the coarsest copy that is accurate enough.
• In Kafka and the writers: separate topics for audit logs, a “gold” tier and best effort, so a flood in best effort cannot delay audit logs; writers pull from tenants round-robin rather than strictly by arrival. This is fairness between tenants applied to a pipeline.
• Showback: publish bytes ingested and stored per team per day, with a rupee figure next to it, and alert a team at 80% of its daily budget. Cloudflare names teams that enable debug logging and forget to turn it off as one reason log volume is unpredictable; a cost dashboard fixes that faster than any limit.

What you pay: per-tenant configuration, sampled data that analysts must remember is sampled, and a policy argument about who sets quotas — better had before the incident than during it.

How do we keep a year of logs without paying for a year of SSDs?

Access to logs falls off a cliff with age: the last hour gets most queries, last month a few, and anything older is read only by auditors or long investigations. Store each age where it is cheapest.

Bad: one tier for everything. Keeping a year of the columnar store on NVMe with two replicas is 365 × 5.9 × 2 ≈ 4.3 PB of NVMe; with a full-text index it would be 365 × 43.2 × 2 ≈ 31.5 PB. You pay premium prices for bytes almost nobody reads.

Good: keep 30 days searchable, dump the rest as raw files. Cheap, and it satisfies “we kept it”. But an auditor’s request for one service’s logs from March becomes a hunt through millions of unindexed files.

Great: tiers by age, moved and deleted as whole parts, with an indexed archive and a restore path.

• Hot, days 0–7, on local NVMe with two replicas: 83 TB. Writes land and most queries run here, and parts are merged into large, well-compressed blocks.
• Warm, days 8–30, in object storage: 136 TB. A table TTL moves whole parts to an object-storage volume after 7 days; queries read through a local cache in seconds to tens of seconds. Object storage provides durability, so no replica is needed — the trick Elasticsearch uses with searchable snapshots on its cold tier (no replicas, about half the disk) and frozen tier (up to 20× more data than warm, with slower searches).
• Delete whole partitions, never rows. Hourly partitions make the 30-day cut-off a cheap metadata operation instead of a rewrite; Elastic gives the same advice for its indices: delete indices, not documents. Teams needing 90 days get a separate logs_90d table: retention is a property of a table, not a filter on rows.
• Archive, days 0–365: ≈ 2 PB. The archiver writes one zstd file per tenant, service and hour into an archive storage class, plus a manifest row per file — tenant, service, time range, line count, and a small bloom filter of trace IDs. Write-once retention (object lock) on this bucket stops audit logs being deleted early, even by administrators.
• Restore on request. POST /v1/archive/restores reads the manifest, fetches only matching files, and loads them into a scratch table that expires after 7 days. Restoring one mid-sized service for three days is about 1.3 TB raw, 160 GB compressed, and fits well inside the 4-hour target; restoring “everything in March” is a planned job with a cost estimate shown first.

What you pay: two copies of every line, a manifest to keep consistent, and a restore that takes hours, not seconds. The blob storage tiers make the 2 PB affordable.

How does a 30-day search come back in seconds — and stay fair?

Most questions are narrow — one service, one hour, errors containing a word. Some are needles — one trace or order ID, service unknown, across a day. Both must be quick, without one person’s "error" over 30 days starving everyone.

Bad: send every query to every shard for the full range. A search for “error” over 30 days touches 162 TB of compressed data. Cloudflare names the lack of per-query limits on documents and memory as one of the reasons it left Elasticsearch: a single bad query degraded the whole cluster, and the damage outlived the query.

Good: prune hard, then fan out with limits. Partitions cut the time range, the sort key cuts to one tenant and service, the text index and bloom filters skip blocks, and only the needed columns are read. Every query carries a byte budget, a row limit and a timeout; every tenant has a cap on concurrent queries. Uber controls the number of in-flight queries per tenant and gives each query a conservative share of threads, leaving room for ingestion.

Great: add time-slicing with early stop, caching, fair queues and a pre-flight estimate.

• Newest first, stop early. A log viewer shows the newest 500 matches, so split a 7-day query into hourly slices, run the newest first, and stop once 500 lines are found — usually within a few slices. Only aggregations visit the whole range.
• Cache immutable slices. A completed hour never changes, so its result can be cached; a dashboard refreshed every 30 seconds recomputes only the current hour. Cache keys include the tenant, so results never leak across teams.
• Fair queues. The query frontend keeps one queue per tenant and hands slices to workers round-robin, so one team’s 30-day scan waits its turn instead of occupying every worker.
• Show the cost first. From partition and part metadata the planner can estimate bytes to scan before running anything. Above, say, 2 TB it asks: “This will scan 4.1 TB and take about two minutes — add a service filter, or run it in the background?”
• Partial answers, labelled. If a shard is slow or down, return what the others found and set partial: true with the missing shards listed. Uber’s platform uses the same skip-unavailable-shards behaviour, preferring a fast answer over a complete one.
• Trace IDs with a time hint. A link from the tracing UI includes the trace’s start time, so the log query covers minutes, not days; the bloom filter on trace_id does the rest.

What you pay: a query frontend, cache and planner to build and run, and the odd engineer surprised by a “partial” banner — far cheaper than a store sized for the worst query anyone could type.

The final design

Put it together. One pipeline writes; four independent readers consume it; a small control plane holds tenant configuration.

1. Collect. On each node the agent tails container files, joins stack traces, collapses repeats, adds Kubernetes metadata, redacts secrets, enforces a per-container cap and ships zstd batches, spilling to a 2 GB disk buffer when the gateway is slow.
2. Admit. The stateless gateway authenticates the agent, applies the tenant’s quota (sampling DEBUG and INFO first, reserving room for errors) and writes to Kafka with acks=all, keyed by tenant and service. It acknowledges only after Kafka does.
3. Index. Writers flatten records into core columns and typed key–value arrays, batch about 100,000 rows, insert with a deduplication token and then commit offsets. The store keeps 7 days on NVMe (20 shards × 2), moves parts to object storage, and drops partitions at 30 days.
4. Archive. The archiver writes raw lines as one zstd file per tenant, service and hour, under object lock, with a manifest row per file, for one year.
5. Alert. The rule engine matches lines against compiled rules and emits per-minute counts to the metrics platform, whose alerting pages the owning team.
6. Tail. The tail service reads a service’s few partitions from the newest offset, filters, and pushes up to 200 lines a second over a WebSocket.
7. Search. The query service checks access, plans with partitions, sort key, text index and bloom filters, runs the newest slices first across 20 shards within byte, row and time limits, caches finished slices and queues fairly per tenant.
8. Restore. For audits and old incidents, a restore job reads the archive manifest and loads just the matching files into a scratch table that expires after a week.

Check it against the requirements. Scale: 3M lines/s fits 20 shards at 150,000 lines/s each and 12 Kafka brokers. Freshness: about a second at the agent and 1–5 s in a writer batch, well inside 30 s at p99; tail reads Kafka directly. Latency: a one-service, one-hour search reads about 2 GB over 20 shards, under a second; needles shrink to gigabytes via bloom filters and the text index. No silent loss: disk buffers, three Kafka replicas and idempotent inserts, with every policy drop counted. Availability: apps write only to local files; Kafka absorbs a store outage of up to 72 hours. Retention and cost: 83 TB of NVMe, 136 TB warm and about 2 PB archived, instead of 2.6 PB of SSD for full-text indexing. Isolation: quotas at three layers, tiered topics, fair writers and fair query queues. Security: redaction on the node, tenant-scoped access on every read path, and an audit log of searches.

What interviewers look for by level

Follow-up questions

What if a developer logs card numbers or Aadhaar numbers?

Stop it as close to the source as possible. The agent runs redaction rules — card numbers that pass a Luhn check, 12-digit national ID formats, phone numbers, emails, tokens and keys — replacing matches with a marker plus a keyed hash, so engineers can still correlate “same value” without seeing it; the gateway catches what agents miss. When something slips through, you need a deletion path: find affected parts by tenant, service and time and rewrite them without the offending lines (a slow background mutation, acceptable because it is rare), and rewrite the matching archive files found via the manifest. Encrypting each tenant’s archive with its own key lets you destroy a whole tenant’s data by destroying the key; see encryption and key management. The platform is itself a sensitive data store, so every search is logged.

What if we run in three regions?

Keep each region’s pipeline self-contained — agents, gateway, Kafka and store — so logs never cross regions on the write path and an outage stays regional. A global query layer fans searches out to each regional query service and merges results, as Uber does across its regions. Data-residency rules can pin a tenant’s logs to one region. Cross-region search is slower and may be partial when a region is unreachable, which the UI must say plainly. More in multi-region and active-active.

What if the logging platform itself is down during an incident?

Design so it degrades in the right order. Agents keep buffering for hours; Kafka keeps three days; alerts come from the stream and keep working without the store. Run the platform on its own Kafka, its own nodes and a separate failure domain from production, and monitor it from the metrics platform rather than from itself, with an alert on ingest lag per tenant. Keep a break-glass path: node-local files are still on disk for a few hours, so kubectl logs and a fleet-wide grep tool can answer urgent questions while the platform recovers.

What if finance wants log costs cut in half?

Most of the bytes come from a few services and a few message patterns, so measure first: top 20 services and top 20 message templates by bytes. Then work through the cheap wins: drop health-check and readiness-probe lines at the agent; sample INFO for the highest-volume services and keep sample_rate on kept lines; turn high-frequency “request served” lines into metrics; and shorten the hot window from 7 to 3 days if few queries reach beyond it. Showback per team usually does the rest — once a team sees a rupee figure next to its debug logs, the volume drops. See cost-aware design.

What if teams want analytics from logs, like p99 latency per endpoint?

The columnar store already does this well — it is why Uber and Cloudflare chose it, and over 80% of Uber’s log queries were aggregations. Materialise the fields involved (endpoint, latency_ms) and let the query service run aggregations over them. For dashboards that refresh constantly, precompute per-minute rollups with a materialised view, or emit the numbers as metrics in the first place. Logs answer “what happened to these requests”; if the question is a standing number on a dashboard, a metric is cheaper and more reliable.

How to say it in the interview

“I’m designing a central logging platform for 100,000 containers: about a million lines a second on average and three million at peak, roughly 43 terabytes of raw logs a day. Logs are write-heavy and rarely read, so the design optimises cost per searchable byte and protects the write path.

Apps just write to stdout. An agent on every node tails the files, adds Kubernetes metadata, joins stack traces, redacts secrets and ships compressed batches, buffering to local disk when the backend is slow. A stateless gateway authenticates the team, applies quotas and writes to a dedicated Kafka cluster with three replicas and 72 hours of retention. Kafka turns a storage outage into a freshness problem instead of data loss.

For storage I’d avoid a full-text index on everything: at about one times raw size plus a replica, 30 days would be 2.6 petabytes of SSD. Instead, writers flatten JSON into typed key–value columns and insert large batches into a columnar store sorted by tenant, service and time, with hourly partitions, a text index on the message and a bloom filter on trace ID. That’s around 8 times compression; a week on NVMe, the rest of 30 days in object storage. An archiver writes a separate one-year audit copy straight from Kafka.

Queries prune by time and sort key, skip blocks with the indexes, fan out to 20 shards and run newest-first with early stop, under per-team limits and fair queues. Live tail reads a service’s Kafka partitions directly, and a streaming rule engine counts pattern matches for alerts, so neither touches the store.

The hard parts are noisy neighbours and schema chaos. For a service logging 100 times normal, I limit at three layers — agent caps, severity-aware quotas at the gateway, tiered Kafka topics — and show every team its daily cost. For schema, typed arrays make type conflicts and new fields a non-event.”

Concepts used in this design

• Observability: logs, metrics and traces — what logs are for, and how they join traces through trace IDs.
• Design metrics, monitoring and alerting — the sibling system that receives the rule engine’s counts and pages people.
• Kafka — the replicated buffer between agents and every reader.
• Backpressure and load shedding — what agents and the gateway do when downstream is slow.
• Effectively-once: idempotent consumers — at-least-once delivery with duplicate-free inserts.
• Search data structures — the inverted index behind full-text search and the text index.
• Elasticsearch in depth — mappings, shards and why indexing everything is expensive at scale.
• OLTP vs OLAP and columnar storage — why sorted columns compress well and scan fast.
• Bloom filters — skipping blocks that cannot hold a trace ID.
• Table partitioning — hourly partitions for pruning and cheap deletion.
• Blob storage (S3) — the warm tier and the year-long archive.
• Noisy neighbours and fairness — quotas, fair queues and reserved shares between teams.
• Design a rate limiter — the token buckets behind per-tenant quotas.
• Real-time connections: SSE and WebSockets — pushing live-tail lines to the browser.
• Encryption and key management — per-tenant keys and deleting data by destroying keys.
• Cost-aware design — sampling, tiering and showback as design tools.

References and further reading

1. Fast and Reliable Schema-Agnostic Log Analytics Platform (Uber Engineering)uber.comWhy Uber left ELK: type conflicts, 20+ clusters per region, 80% aggregation queries, only 5% of indexed fields used; typed key-value arrays in ClickHouse, 300k logs/s per node, at-least-once from Kafka, per-tenant limits, under one minute to searchable.
2. Log analytics using ClickHouse (Cloudflare blog)blog.cloudflare.comOver 99% of logs are never read; mapping explosions and no per-query limits in Elasticsearch; 600 bytes per document down to 60 per row; hourly partitions, big batches, bloom-filter skip indexes, one array per type beyond 1,000 fields; sampled copies for dashboards.
3. Grafana Loki documentation: Componentsgrafana.comStateless distributors that validate pushes and rate-limit each tenant by dividing its global limit across the current number of distributors; ingesters that compress chunks and flush them to object storage.
4. Grafana Loki documentation: Label best practicesgrafana.comIndex only a few static, bounded labels and use filter expressions for the rest; a 1 MB compressed chunk holds about 5–10 MB of logs; IDs such as a request ID must never be labels.
5. Elasticsearch storage: the true story, 2.0 (Elastic blog)elastic.coMeasured index size on disk divided by raw log size: 0.43 to 1.39 depending on mappings and codec; analysing the message field and keeping timestamp strings both inflate it.
6. Elasticsearch reference: Mapping limit settingselastic.coThe default limit of 1,000 mapped fields per index, the error when it is exceeded, and the flattened type and ignore-beyond-limit escape hatches.
7. Elasticsearch data tiers: hot, warm, cold and frozenelastic.coTime-series data moving through tiers; searchable snapshots on cold halve disk by dropping replicas; the frozen tier holds up to 20× more data than warm, with slower searches.
8. Fluent Bit documentation: Backpressuredocs.fluentbit.ioWhat an agent does when its output is slow: memory limits that pause inputs, filesystem buffering to avoid loss, and a per-output size cap that discards the oldest chunks.
9. OpenTelemetry Logs Data Modelopentelemetry.ioStandard log record fields: Timestamp, ObservedTimestamp, TraceId, SpanId, SeverityText and SeverityNumber (1–24), Body, Resource and Attributes.
10. ClickHouse documentation: Full-text search with text indexesclickhouse.comA per-part inverted index mapping tokens to rows, generally available from version 26.2, which replaces token bloom filters for full-text search.