Everyone underestimates log aggregation because the demo is trivial: run an agent on a box, ship each line over the network, dump it into a searchable store, done. Then the interviewer sets the actual scale - 100,000 servers, ~10M events per second, retain 30 days, query it all from a dashboard - and every part of the trivial version dies. Ten million events a second is roughly 3 GB/sec on the wire and hundreds of terabytes a day; a single server’s agent hiccuping cannot be allowed to drop your incident forensics; the read pattern is a schizophrenic mix of “aggregate this numeric metric over a range” and “grep for this rare error string across a petabyte in the last 30 days”; and the naive fix - “just index every word so search is fast” - is precisely the decision that makes your index bigger than your data and bankrupts the cluster.

This is a different problem from a pure metrics system, and the difference is the whole interview. Metrics are numbers you aggregate over ranges; logs are unstructured text you search. One wants a time-series engine with columnar compression; the other wants an inverted index or a brute-force scan pipeline, and choosing wrong on the log side is the classic blowup. So the design comes down to four decisions: how you collect from 100K hosts without the agent losing data or melting the network, how you transport and buffer a 10M/sec firehose so a slow storage tier never blocks a producer, how you store and index logs so 30-day search is possible without an index that costs more than the raw data, and how you serve both metric aggregation and log search from one dashboard. All while ingestion never blocks, dashboards stay fast, and the data survives the exact moment - an outage - when everything spikes at once.

Let me do it properly.

Functional Requirements (FR)

In scope:

  • Collect metrics and logs from 100K servers. A lightweight agent on each host tails log files and reads application/host metrics, batches them, and ships them to the pipeline. It must handle the host being busy, the network flaking, and the backend being briefly unavailable without losing data or overwhelming the app it runs beside.
  • Two data shapes, one pipeline. A metric is (name, {labels}, timestamp, value) - numeric, aggregated over ranges. A log event is (timestamp, {labels}, message) where message is arbitrary text (a stack trace, a JSON blob, an access-log line). Both flow through the same collection and transport plane and diverge only at storage.
  • Real-time aggregation. As events stream in, compute rollups on the fly: request rates, error counts per service, p99 latency, log-lines-per-second by level. A dashboard panel for “error rate in the last minute” must reflect data that arrived seconds ago, not minutes ago.
  • Search and filter logs. “Show me all ERROR logs for service=checkout in region=ap-south-1 containing OutOfMemory in the last 6 hours.” Filter by labels (structured) and full-text within the message (unstructured), over arbitrary time ranges up to the 30-day retention.
  • Query metrics over ranges. Aggregation (sum, avg, min, max, quantiles) grouped by labels over time - the standard time-series read.
  • Dashboards and live tail. Serve aggregated results and search hits to a UI, plus a live tail ("tail -f across the fleet") that streams matching new log lines as they arrive.
  • Retention: 30 days. All raw logs and metrics queryable for 30 days, then dropped. Tiering to cheaper storage within that window is allowed.

Explicitly out of scope (own the scope):

  • Distributed tracing. Spans and trace assembly are a third telemetry pillar with their own access pattern (trace-id lookup, service graphs). We note where a trace_id label links a log to a trace and move on - conflating all three in one design is how you lose the interview.
  • Alerting internals. Rules, for-duration debounce, and the pager fan-out are a large subsystem; here they are a consumer of the real-time aggregation stream. We build the pipeline that feeds alerting and hand off.
  • Anomaly detection / ML. Auto-baselining on top of the query layer is a separate system; we build search and aggregation and note where a scorer plugs in.
  • The instrumentation SDK. How app code emits a metric or writes a log line (a logging library, a /metrics endpoint) is assumed. We design from “the agent sees the data” onward.
  • Long-term archival / compliance retention. Beyond 30 days is a downstream cold-storage / SIEM concern. We serve operational observability: recent, high-volume, latency-sensitive.

The one decision that drives everything: this is a write-firehose problem where logs need search (an index or a scan) and metrics need range aggregation, so the pipeline must collect reliably from 100K hosts, decouple ingest from storage with a durable buffer, and - critically - avoid the index-everything trap that makes log search cost more than the logs themselves. The architecture follows: a reliable agent, a partitioned durable buffer, a fork into a metrics TSDB path and a log store path, and a query layer that speaks both.

Non-Functional Requirements (NFR)

  • Scale: 100,000 servers, ~10M events/sec aggregate (roughly 100 events/sec/host averaged - some hosts far noisier). Assume the mix is heavy on logs: say ~8M log lines/sec and ~2M metric samples/sec. Incident spikes push total to ~20-25M/sec (error storms generate the most logs exactly when you need them).
  • Latency: ingest is fire-and-forget - the agent must hand off a batch in single-digit milliseconds and never block the host application. Real-time aggregation visible within ~seconds (data-arrival to dashboard). Log search p95 under ~2-5s for a scoped query (labels + time range narrow it); a broad full-text scan over 30 days may take longer and streams results progressively. Metric dashboard panels under ~1s from rollups.
  • Availability: the collection and transport paths target 99.9%+. If ingest drops during an incident, you lose the forensics that explain the incident. The query path can degrade to slightly stale before it errors.
  • Consistency: eventually consistent, at-least-once. A log line may arrive a few seconds late or, in a crash, be delivered twice - both are acceptable for observability (dedupe on a content hash if needed). We explicitly do not need exactly-once; approximate-but-timely beats exact-but-late.
  • Durability: high for the retention window. Once the agent gets an ack, the event must survive backend node failures for 30 days - losing incident logs is the cardinal sin. The durable buffer plus replicated storage provide this. Data older than 30 days is dropped by design.
  • Backpressure tolerance is a first-class NFR. Load is anti-correlated with health: the firehose is largest when production is on fire. The pipeline must shed or buffer gracefully - never let a slow storage tier propagate back and stall producers, and never let the agent OOM the host it monitors.

Back-of-the-Envelope Estimation (BoE)

Ground it in numbers.

Traffic:

Servers:                        100,000
Aggregate events/sec:           10,000,000     (10M)
  ~ metrics:                     2,000,000/sec
  ~ logs:                        8,000,000/sec
Per-host average:               100 events/sec  (noisy hosts far higher)
Incident spike (~2-2.5x):       20-25M/sec      (size ingest for this)

Sizing for the spike is not waste - it is the incident-survival budget. Autoscaling adds hosts (more agents), error paths log more, and every dashboard gets opened at once, all when production is worst.

Read vs write:

Writes: 10M events/sec sustained, ~25M/sec peak. The firehose.
Reads:
  - Real-time aggregators: continuous stream consumers, not "queries" -
      they process the full firehose once to produce rollups.
  - Dashboards: low thousands of panel-polls/sec, mostly metric rollups + cache.
  - Log searches: low hundreds/sec, but EACH may scan a huge time+label slice.
      This is the expensive, spiky read.

Writes dominate volume by orders of magnitude, but a single broad log search can read more bytes than a thousand metric panels. The two read shapes need different engines.

Storage (the number that scares people):

Avg log line on wire:           ~300 bytes (structured JSON + message)
Raw log volume:  8M/sec * 300B  = 2.4 GB/sec  = ~207 TB/day RAW
Text compresses well (~8-10x):  ~21-26 TB/day compressed
30-day log retention:           ~630-780 TB compressed. (Raw: ~6 PB - infeasible.)

Metrics (compressed TSDB, Gorilla ~1.5 B/point):
  2M/sec * 1.5B = 3 MB/sec = ~260 GB/day = ~7.8 TB over 30 days.
  Metrics are a rounding error next to logs.

Two facts fall out. Compression is mandatory - the difference between 6 PB and ~700 TB is the difference between an impossible and a merely large cluster. And logs dwarf metrics ~100:1 in bytes, so the log storage/index design is where the money and the risk live. That is why the deep dives lean on logs.

The index blowup (the trap):

Full inverted index (index every term, Elasticsearch-style):
  Index size is often 0.5x - 1.5x of the RAW data. On ~6 PB raw/30d that is
  MULTIPLE PETABYTES of index - more expensive to store than the logs, and
  RAM-hungry to keep hot. Indexing 8M lines/sec also burns enormous CPU.

Label-only index (index labels, NOT message body - Loki-style):
  Index only the (small, low-cardinality) label sets = "streams".
  Index is GIGABYTES, not petabytes. Full-text becomes a brute-force scan
  of the compressed chunks the labels+time already narrowed you to.

This is the central log-side decision and the subject of deep dive 2: pay for a giant always-on inverted index and get fast arbitrary search, or index only labels and brute-force-scan the (already narrowed) matching chunks and accept slower rare-term search for a fraction of the storage.

Bandwidth:

Ingest (agent-compressed): 2.4 GB/sec raw -> ~300 MB/sec compressed on wire
   = ~2.4 Gbps inbound, spread across an ingest fleet + regional collectors.
Query responses: search hits and rollups, KB-MB each, low volume. Non-issue.

The scarce resources are sustained write throughput into storage, disk for 30 days of compressed logs (~700 TB), CPU for compression and any indexing, and scan throughput for broad searches. Size for those.

High-Level Design (HLD)

A per-host agent collects and ships; regional collectors terminate connections and push into a partitioned durable buffer; the buffer forks into a metrics path (TSDB) and a log path (object-store chunks + label index), with a real-time aggregation layer tapping the stream; a query layer fronts both for the dashboard.

   100,000 HOSTS
   ┌─────────────────────────────────────────────────────────────────────────┐
   │  [app + logs + /metrics]  <-  AGENT (per host)                            │
   │     - tail files / scrape metrics   - batch + compress                    │
   │     - disk-backed buffer (survive backend outage)  - backpressure         │
   └───────────────────────────────┬───────────────────────────────────────────┘
                                    │ compressed batches (gRPC/HTTP, TLS)
                          ┌─────────▼──────────┐
                          │ Regional Collector │  (stateless fleet, LB'd)
                          │  - authN / tenant  │
                          │  - parse + relabel │  (drop/rewrite hi-card labels)
                          │  - route by key    │
                          └─────────┬──────────┘
                                    │ append (partition by tenant+stream hash)
                     ┌──────────────▼───────────────┐
                     │  KAFKA  (durable buffer)      │  absorbs 25M/sec spikes,
                     │  partitioned, replicated x3   │  decouples ingest/storage,
                     │  short retention (hours)      │  replay on consumer failure
                     └──┬───────────┬────────────┬───┘
        metrics topic   │           │ logs topic │  (same firehose, forked)
             ┌──────────▼───┐  ┌────▼─────────┐ ┌▼──────────────────────┐
             │ TSDB Writers │  │ Log Ingesters│ │ Real-time Aggregators │
             │ (Gorilla,    │  │ - group by   │ │ (stream processors)   │
             │  sharded by  │  │   stream     │ │ - rate/count/quantile │
             │  series)     │  │ - compress   │ │   over tumbling windows│
             └──────┬───────┘  │   chunks     │ │ - emit derived metrics │
                    │          │ - build label│ │   + feed ALERTING      │
                    │          │   index +    │ └──────────┬────────────┘
                    │          │   bloom filt.│            │ derived metrics
                    │          └────┬─────────┘            │ back into TSDB
        ┌───────────▼──────┐   ┌────▼───────────────────┐  │
        │ Metrics store    │   │ Log store              │  │
        │ SSD blocks (2h)  │   │ Object store: chunks   │◄─┘
        │ 30d retention    │   │  (compressed log runs) │
        └───────────┬──────┘   │ + Label index (small)  │
                    │          │ + Bloom filters/chunk  │
                    │          │ 30d retention          │
                    │          └────┬───────────────────┘
                    │               │
              ┌─────▼───────────────▼──────────────────────────┐
              │              QUERY LAYER                        │
              │  - metric range queries (rollups + cache)       │
              │  - log search: label+time prune -> bloom skip   │
              │      -> parallel scan of matching chunks        │
              │  - live tail: subscribe to the log topic        │
              └───────────────────────┬────────────────────────┘
                                ┌──────▼──────┐
                                │  Dashboard  │
                                └─────────────┘

Write path - an event from a host:

  1. The agent tails log files and reads metrics, tags each event with host/service/region labels, batches many events, compresses the batch, and ships it over TLS. If the backend is unreachable, the agent spools to a local disk buffer and retries with backoff - the host may keep logging for minutes without the pipeline, and nothing is lost until the local buffer fills (then it drops oldest, and counts the drop).
  2. A Regional Collector (stateless, load-balanced) authenticates the tenant, parses and normalizes the batch, runs relabeling to drop or collapse high-cardinality labels (never let request_id become a label), and appends events to Kafka, partitioned by a hash of (tenant, stream) so one stream’s events stay ordered on one partition. The collector acks the agent only after the durable append.
  3. Kafka is the shock absorber: replicated x3, it holds hours of data so a spike or a stalled downstream consumer never propagates back to producers. It also lets multiple independent consumer groups read the same firehose.
  4. The stream forks by consumer group: TSDB Writers consume the metrics topic and write compressed time-series blocks (sharded by series); Log Ingesters consume the logs topic, group events by stream, compress them into chunks, and build a small label index plus per-chunk bloom filters; Real-time Aggregators consume the full firehose to produce rollups and feed alerting.
  5. Storage tiers: metrics as SSD-resident time-partitioned blocks; logs as compressed chunks in object storage with a compact label index and bloom filters kept hot. Both enforce 30-day retention by dropping whole immutable segments.

Read path:

  1. Metric query: the query layer resolves label matchers via the series index, picks a rollup resolution for the range, fans out to metric shards, aggregates, and caches common panels. Sub-second.
  2. Log search: the query layer uses labels + time range to prune to a candidate set of chunks (the label index maps {service, level, region} to the streams/chunks that have them). For a full-text term, it consults each candidate chunk’s bloom filter to skip chunks that provably lack the term, then parallel-scans the survivors, decompressing and regex/substring-matching. Results stream back progressively.
  3. Live tail: the query layer subscribes directly to the logs topic (or the ingesters’ tail), filters by the query’s labels, and streams matching new lines to the UI in real time - no storage read at all.

The key insight: collection reliability and buffer decoupling protect the firehose, while the log store deliberately trades a giant inverted index for a tiny label index plus brute-force scan - because at ~700 TB/30d, an index that rivals the data size is the thing that actually bankrupts you. Metrics ride the same pipeline but land in a purpose-built TSDB, because numbers-aggregated-over-ranges and text-you-search are genuinely different workloads.

Component Deep Dive

Four hard parts, each walked from naive to scalable: (1) the collection agent and reliable transport, (2) log storage and the index-everything trap, (3) real-time aggregation over the firehose, and (4) searching 30 days without scanning everything.

1. Collection Agent and Transport: Don’t Drop, Don’t Block, Don’t Melt the Host

100K agents shipping to a backend that will sometimes be slow or down. The agent runs next to production code and must never harm it.

Approach A: Synchronous ship-per-line to the backend (naive)

Each log line is sent to the aggregation backend synchronously over HTTP; the app (or a thin forwarder) blocks until the backend acks.

Where it breaks: three ways at once. First, per-line network round-trips at 100 events/sec/host is 10M tiny requests/sec across the fleet - connection and header overhead alone dwarfs the payload, and TCP/TLS setup costs murder throughput. Second, the app blocks on the backend: when the aggregation tier is slow (which is exactly during an incident), your production service now blocks on logging and its latency spikes - monitoring is now causing the outage. Third, any backend blip loses data - there is no buffer, so a 30-second deploy of the log tier drops 30 seconds of logs from 100K hosts, which is 300M lost events precisely when you might need them. Synchronous, unbuffered, per-line shipping fails the “don’t block, don’t drop” bar completely.

Approach B: Async batching agent with a disk-backed buffer and backpressure

Make the agent an asynchronous, decoupled shipper that treats the host application as sacred:

  • Tail decoupled from the app. The agent reads log files (or a local unix socket / journald), so the app just writes to disk/stdout at memory speed and never waits on the network. The agent tracks a read offset (checkpoint) per file so a restart resumes exactly where it left off and does not re-read or skip.
  • Batch and compress. The agent accumulates events into batches (by size or a short time flush, e.g. 1MB or 1s) and compresses each batch (snappy/zstd). One request now carries thousands of events - amortizing the round-trip and getting the ~10x text compression on the wire.
  • Disk-backed local buffer. Batches go into a bounded on-disk queue before shipping. If the backend is unreachable, the agent keeps buffering to disk and retries with exponential backoff. This is what turns a backend outage from data loss into a delay. The buffer is bounded (say a few GB / N minutes); when full, the agent drops oldest and increments a dropped_events counter it will itself report later - explicit, observable load-shedding, not silent loss.
  • Backpressure and resource caps. The agent runs with hard CPU and memory limits (it must not starve the app it monitors). If it cannot keep up, it drops at the source (sampling low-value debug logs first) rather than growing unbounded. It also honors backend 429 / Retry-After to back off when the pipeline signals overload.
  • Regional collectors terminate the fan-in. 100K persistent connections do not all hit one place. Agents connect to regional collector fleets (behind a load balancer, close to the hosts) that terminate TLS, authenticate, and do the durable Kafka append. This bounds connection count per node and keeps the ingest path horizontally scalable - add collectors to add capacity.
agent loop:
   batch = read_from_offset(files, until 1MB or 1s)
   checkpoint(offset)                 # durable read position
   compressed = zstd(batch)
   enqueue_disk(compressed)           # bounded on-disk queue
   for item in disk_queue:            # separate shipper
        ok = POST /v1/ingest (item)   # with backoff on failure
        if ok: dequeue(item)
        else:  keep, retry later      # backend down -> buffer, don't drop
   if disk_queue.full: drop_oldest(); dropped_events++

The mindset: the agent’s job is to protect the host and never block it, batch hard to survive the firehose, and buffer to disk so a backend outage is a delay not a loss. Everything downstream can be rebuilt or replayed; the events the agent fails to ship are gone forever, so reliability lives here first.

2. Log Storage and Indexing: The Index-Everything Trap

This is the decision that defines the system. We must search ~700 TB of logs by label and by full-text over 30 days.

Approach A: Index every term in a full-text inverted index (the naive “make search fast”)

Feed every log line into an inverted index (Elasticsearch/Lucene-style): tokenize the message, and for every term build a postings list of the documents containing it. Search any word instantly.

Where it breaks: at 8M lines/sec, indexing is brutally expensive on two axes. Storage: a full inverted index over high-entropy log text routinely runs 0.5x to 1.5x the size of the raw data - so on top of ~700 TB of compressed logs you add hundreds of terabytes to petabytes of index, and the index is the part you want hot in RAM/SSD for speed. You are now paying more to make logs searchable than to store them. CPU: tokenizing and indexing 8M lines/sec burns enormous, continuous compute, and it spikes during incidents when log volume spikes - the index falls behind exactly when search matters. Cardinality: log messages contain near-infinite unique tokens (UUIDs, timestamps, memory addresses), so the term dictionary explodes. Full inverted indexing gives you fast arbitrary search at a cost that does not survive contact with a real 10M/sec firehose. It is the right tool for a search product, the wrong default for high-volume operational logs.

Approach B: Index labels only, brute-force-scan the narrowed chunks (the Loki model)

Flip the assumption. Most log searches are already scoped by labels and time (“errors for service=checkout in the last hour”); those two dimensions prune the search space by orders of magnitude before any text matching. So index cheaply and scan smartly:

  • A stream = a unique label set. Group log lines by their label set ({service, host, level, region, ...}); each unique combination is a stream. The only thing indexed is the (small, low-cardinality) label sets - not the message body. This index is gigabytes, not petabytes, and it maps a label matcher to the set of streams (and thus chunks) that satisfy it. High-cardinality fields (request IDs, user IDs) are kept in the message, never promoted to labels - the same admission-control discipline that protects a metrics index protects this one.
  • Chunks: compressed runs of a stream’s lines. Each stream’s lines are appended in time order and packed into chunks - compressed blocks (say a few MB) covering a time range. Chunks are immutable, written to cheap object storage, and named by (stream, time range). Compression on same-stream lines (which are highly repetitive) hits the ~10x. Retention is a matter of deleting chunks older than 30 days - trivial because they are immutable and time-named.
  • Search = prune, then scan. A query resolves its label matchers against the small index to a candidate chunk set, uses the time range to drop non-overlapping chunks, then decompresses and scans the survivors for the full-text term (substring/regex). Because labels + time already cut ~700 TB down to gigabytes for a scoped query, brute-force scanning that residue is fast and massively parallel (fan the chunks across a scanner fleet). No giant text index to build, store, or keep hot.
  • Bloom filters to skip chunks. Brute-scanning is still wasteful if the term is rare and appears in few chunks. So each chunk carries a bloom filter over its tokens. Before decompressing a chunk, the scanner asks the bloom filter “could the term OutOfMemory be here?”; a definite “no” skips the chunk entirely without reading it. Blooms are tiny and turn a rare-term search from “scan every candidate chunk” into “scan only the handful that might match.” This recovers much of the rare-term speed an inverted index would give, at a fraction of the storage.
Elasticsearch model:    index EVERY term  -> fast any-word search,
                        index ~= data size (PB), huge CPU, hot RAM.  $$$$
Loki model (chosen):    index ONLY labels  -> tiny index (GB),
                        text = bloom-skip + parallel scan of pruned chunks.  $

Search "OutOfMemory" for {service=checkout, level=ERROR}, last 6h:
   1. label index -> streams matching {service,level}      (ms)
   2. time range  -> chunks in [now-6h, now]               (ms)
   3. bloom filter per chunk -> drop chunks lacking term    (ms, no I/O)
   4. parallel decompress + substring scan survivors       (the real work)
Dimension Full inverted index (ES) Label index + scan (Loki)
Index size 0.5-1.5x of raw data (PB) GB - only labels
Ingest CPU High (tokenize everything) Low (compress + labels)
Arbitrary-term search Fast anywhere Fast if label/time-scoped; slower if broad
Cost at 10M/sec Very high Low
Best for Search products, low-volume High-volume operational logs

The trade is explicit: we give up instant search over an unscoped full-text term across all 700 TB, in exchange for an index that is thousands of times smaller and an ingest path that keeps up at 10M/sec. For operational logs that is the right call, because real searches are almost always scoped by service, level, and a recent time window - and blooms recover the rare-term case. If a use case genuinely needs Google-over-all-logs, that is a separate, expensive, opt-in indexed store, not the default.

3. Real-Time Aggregation: Rollups From a Moving Firehose

The dashboard wants “error rate per service, last minute” reflecting data from seconds ago. You cannot compute that by querying 700 TB on every panel refresh.

Approach A: Compute aggregations at query time by scanning raw data (naive)

When a panel asks for “errors/sec by service over the last hour,” scan the raw logs/metrics for that window and aggregate on the fly.

Where it breaks: a “last minute error rate” panel refreshing every 10s would rescan the same fresh, huge slice of the firehose repeatedly, and a “last hour rate” rescans an hour of data every refresh. With thousands of open panels this multiplies into a crushing, redundant read load on the storage tier, and it is slow - you cannot decompress and count millions of lines under a 1s panel budget. Query-time aggregation over raw data does not scale to a live dashboard; the work must be done once, as data streams in.

Approach B: A streaming aggregation layer with tumbling windows and pre-materialized rollups

Compute the aggregates once, incrementally, as the firehose flows - then panels read tiny pre-computed results:

  • Stream processors on the firehose. A dedicated consumer group (Flink/Kafka-Streams-style, or purpose-built) reads the full event stream and maintains tumbling window aggregates: per (service, level) count log lines per 10s window; per (service, endpoint) compute request rate and latency histogram per window. Each window, when it closes, emits a small derived metric point. 8M raw lines/sec collapse into a stream of compact rollups.
  • Derived metrics flow back into the TSDB. “Error lines per second by service” is a metric - so the aggregator emits it as a time series and writes it into the same metrics store. Now a dashboard panel for error rate is a cheap TSDB range read of a pre-materialized series, not a scan of raw logs. Logs generate metrics; the two paths join here.
  • Windowing and late data. Windows are keyed by event time (the timestamp on the log), not arrival time, so a batch delayed by the agent’s buffer still lands in the correct window if it arrives within a grace period (say 30s-1m) before the window is finalized. Genuinely late events past the grace are counted into a “late” bucket and dropped from the window - the eventual-consistency relaxation from the NFRs makes this rare misplacement a non-event, not a billing error.
  • This layer feeds alerting. Because it already computes fresh rates and counts every window, the alerting engine consumes these derived streams instead of querying storage - alert evaluation stays real-time and isolated from dashboard read load. The aggregation layer is the shared substrate for both dashboards and the pager.
  • Idempotency under at-least-once. Kafka may redeliver on consumer restart, so window aggregates must be idempotent per event - dedupe on an event id / offset within the window, or use the processor’s exactly-once-within-the-pipeline checkpointing so a replay does not double-count. We accept at-least-once into the pipeline and make the aggregation robust to it.
stream processor, per 10s tumbling window W, keyed by (service, level):
   for each event e in stream with event_time in W:
        counts[(e.service, e.level)] += 1
        if latency metric: hist[(e.service)].observe(e.latency)
   on window_close(W after grace):
        emit metric  log_lines_total{service,level} @ W.end = counts[...]
        emit metric  latency histogram buckets @ W.end
        -> written to TSDB, consumed by dashboards + alerting

The principle: do the aggregation once as data streams past, materialize compact rollups (including metrics derived from logs), and let dashboards and alerts read those - never rescan the raw firehose per panel. This is what makes “last minute error rate across 100K servers” a millisecond read instead of a terabyte scan.

4. Searching 30 Days Without Scanning Everything

Even with the label+scan model, a poorly scoped search (“this rare error anywhere in 30 days”) could touch a huge chunk set. We need it to stay tractable and parallel.

Approach A: A single query node scans all candidate chunks serially (naive)

The query node resolves candidate chunks and decompresses/scans them one after another.

Where it breaks: a 30-day, broad-label search can select tens of thousands of chunks totaling many terabytes; scanning them serially on one node takes minutes to hours and pins one machine while other searches queue behind it. There is no parallelism, no early exit, and one heavy query starves everyone. Serial single-node scan does not meet the search latency bar or isolate tenants.

Approach B: Time-sharded storage + parallel scatter-gather scan + progressive results

Make search embarrassingly parallel and bounded:

  • Shard by time, then by stream hash. Chunks are organized into time buckets (e.g. per-2-hour) and, within a bucket, distributed across storage/scanner nodes by stream hash. A time-ranged query immediately excludes all buckets outside its range (the dominant pruning at 30-day scale - most searches are recent), and within the range the stream-hash spread means the surviving chunks are already fanned across many nodes.
  • Scatter-gather scan. The query layer fans the candidate chunk set out to a scanner fleet in parallel - each scanner blooms-skips, decompresses, and substring/regex-matches its assigned chunks, returning only the matching lines. The query merges partial results. Wall-clock is set by the slowest shard, not the sum, so doubling the fleet roughly halves a big scan.
  • Progressive / streaming results, newest first. Scanners process newest chunks first and stream hits to the UI as they arrive, so the user sees recent matches within seconds even if the full 30-day scan continues in the background. Most searches are answered by the first page of recent hits and the user stops there.
  • Query limits and fairness. Each search gets bounded parallelism, a max-chunks/max-bytes budget, and a timeout; a runaway broad query is capped and told to narrow its labels or time range rather than being allowed to consume the whole scanner fleet. This isolates tenants and keeps one heavy search from starving the rest.
  • Result and metadata caching. The label index and bloom filters are cached hot (they are small); repeated identical searches and dashboard-driven log panels hit a result cache keyed by (query, time range). Metric panels are served from the pre-materialized rollups of deep dive 3, never touching the log scanners.
search over 30d:
   candidate = label_index.match(labels) ∩ time_buckets(range)   # heavy prune
   shards = group(candidate) by scanner-node (stream hash)
   parallel for each shard:                                       # scatter
        for chunk in shard, newest-first:
             if not bloom(chunk).maybe(term): skip                # no I/O
             hits += scan(decompress(chunk), term)
             stream hits to client                                # progressive
   merge + rank + apply limit                                     # gather

Put together: time-bucketing prunes the 30-day range hard, stream-hash sharding fans the residue across a scanner fleet, bloom filters skip chunks without reading them, and progressive newest-first streaming answers most searches from the first page - so a search over ~700 TB touches gigabytes on dozens of nodes in parallel, not terabytes on one.

API Design & Data Schema

Ingest API (write plane - agent to collector)

POST /v1/ingest
  Content-Type: application/x-protobuf   (batched, zstd-compressed)
  Authorization: Bearer <tenant-agent-token>
  body: {
    streams: [
      { labels: { service:"checkout", host:"h-9f2", region:"ap-south-1",
                  level:"ERROR" },
        entries: [                              # logs
          { ts:1752345600000, line:"OutOfMemoryError: heap ... trace_id=abc" },
          ...
        ] },
    ],
    metrics: [                                  # metrics ride the same batch
      { name:"http_requests_total",
        labels:{ service:"checkout", status:"500" },
        samples:[ {ts:1752345600000, value:42} ] }
    ]
  }
  204 No Content        -> durably buffered (Kafka append). Async storage.
  429 Too Many Requests -> tenant over rate/cardinality; Retry-After honored.

Notes:
  - Ack is after the durable buffer append, not after storage (fire-and-forget).
  - Agent retries on 5xx/network by re-sending from its disk buffer (at-least-once).
  - High-cardinality fields (request_id, user_id) stay IN the log line, never labels.

Query API (read plane)

# Log search - LogQL-style: label selector {..} + optional line filter
GET /v1/logs/query
    ?query={service="checkout",level="ERROR"} |= "OutOfMemory"
    &start=2026-07-13T00:00:00Z&end=2026-07-13T06:00:00Z
    &limit=200&direction=backward       # newest-first, progressive
  200 OK  (streamed)
  { "matches":[
      { "ts":..., "labels":{...}, "line":"OutOfMemoryError: ..." }, ... ],
    "stats":{ "chunks_scanned":142, "chunks_skipped_bloom":9310,
              "bytes_scanned":"3.1GB" } }

# Live tail - server-sent stream of new matching lines
GET /v1/logs/tail?query={service="checkout"} |= "timeout"     # SSE / websocket

# Metric range query - PromQL-style over rollups
GET /v1/metrics/query_range
    ?query=sum by(service)(rate(log_lines_total{level="ERROR"}[1m]))
    &start=...&end=...&step=1m
  200 OK  { "resolution":"1m", "series":[ {labels,points[]}, ... ] }

# Metadata / self-observability
GET /v1/labels                     # label names
GET /v1/label/service/values       # values for a label
GET /v1/cardinality?by=stream&top=20   # streams by volume (catch runaway logs)

Data model: two engines behind one API

Log store - label index + immutable chunks (object storage, NoSQL/custom):

stream_id  = hash(sorted(labels))              # a unique label set
label index (inverted, small, hot):
   service="checkout" -> [stream_1, stream_7, ...]     # postings of stream_ids
   level="ERROR"      -> [stream_7, stream_88, ...]    # intersect to resolve
chunk (immutable, in object store):
   { chunk_id, stream_id, ts_min, ts_max,
     data: zstd(concatenated log lines in time order),
     bloom: bloom_filter(tokens in this chunk) }        # skip without reading
   named by (stream_id, time_bucket) -> retention = delete buckets > 30d

Metrics store - compressed time-series blocks (purpose-built TSDB):

series_id = hash(name + sorted(labels))
blocks (immutable, 2h, SSD -> object store):
   per series: timestamps delta-of-delta encoded, values XOR-compressed (Gorilla)
   ~1.5 bytes/point. Retention: 30 days.
series index: (label=value) -> postings of series_ids   (bounded by admission ctrl)

Durable buffer (Kafka) and offsets:

topics: logs, metrics ; partitioned by hash(tenant, stream) ; replicated x3 ;
short retention (hours) as replay/shock-absorber. Consumers track offsets;
a failed ingester replays from last committed offset -> at-least-once.

Why this split: logs are text-you-search, so a tiny label index + compressed immutable chunks + bloom filters on cheap object storage beats a petabyte-scale inverted index; metrics are numbers-you-aggregate, so a columnar compressed TSDB beats both. A single relational database would blow up storage, cannot serve range aggregation at 2M points/sec, and cannot brute-scan 700 TB in parallel. NoSQL/object storage for the log chunks (immutable, append-only, horizontally scalable) and a purpose-built TSDB for metrics - right tool per workload, unified only at the collection pipeline and the query API.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix:

1. The index-everything trap (the number-one log-side killer). Indexing every term over 8M lines/sec makes the index rival the data (petabytes) and burns CPU that falls behind during incidents. Fix: index labels only, brute-force-scan the label+time-pruned chunks, with per-chunk bloom filters to skip non-matching chunks. Index is GB not PB; ingest keeps up. Full-text indexing becomes an expensive opt-in, not the default.

2. Ingest write throughput (10-25M events/sec). No single node accepts it, and synchronous per-line shipping melts both the network and the producing app. Fix: batching, compressing agents with disk buffers, a stateless regional collector fleet, and a partitioned durable buffer (Kafka) decoupling ingest from storage. Scale collectors and partitions to add capacity.

3. Backpressure during incidents (load anti-correlated with health). The firehose is largest when storage is most stressed; naive coupling stalls producers or OOMs agents. Fix: Kafka absorbs spikes (hours of buffer), the agent spools to disk and sheds oldest when its bounded buffer fills (counting drops), and collectors return 429/Retry-After to signal overload. Nothing propagates back to production code.

4. Log cardinality / runaway streams. Promoting a high-cardinality field (request_id) to a label explodes the stream count and the label index, and a chatty service can flood the pipeline. Fix: relabeling drops/collapses dangerous labels at the collector (IDs stay in the message body), per-tenant volume + stream limits shed a runaway tenant rather than the shared system, and self-monitoring surfaces top streams by volume. Same admission-control discipline as the metrics index.

5. Broad log search touching too many chunks. A 30-day, loosely-scoped search selects tens of thousands of chunks and, run serially, pins a node for minutes. Fix: time-bucket sharding prunes the range hard, stream-hash spread + scatter-gather scans survivors across a scanner fleet in parallel, bloom filters skip chunks without I/O, and progressive newest-first streaming answers most searches from the first page. Per-query byte/time budgets isolate tenants.

6. Query-time aggregation crushing storage. Live dashboard panels rescanning raw data per refresh multiply into a huge redundant read load. Fix: a streaming aggregation layer computes tumbling-window rollups once as data flows, materializes derived metrics (including metrics-from-logs) into the TSDB, and dashboards/alerts read those tiny pre-computed series. Never rescan the firehose per panel.

7. Hot partition / hot stream. A single extremely chatty service concentrates load on one Kafka partition and one ingester. Fix: hash-partition by (tenant, stream) to spread load; for a genuinely hot single stream, add a low-cardinality shard label at the source to split it, and merge at query time - the controlled inverse of the cardinality problem.

8. Storage growth (~700 TB / 30 days). Even compressed, logs are enormous and unbounded without retention. Fix: aggressive compression (~10x), immutable time-named chunks in cheap object storage, and 30-day retention by dropping whole time buckets - trivial because chunks are immutable and time-partitioned. Recent chunks stay on faster tiers; older ones on cheapest object storage.

9. Duplicate delivery under at-least-once. A collector or ingester restart replays Kafka offsets, double-writing events. Fix: idempotent aggregation (dedupe by event id / offset within a window) and content-hash dedupe on log storage where it matters. We accept at-least-once into the pipeline because exactly-once is not worth its cost for observability, and make consumers robust to replays.

10. The pipeline’s own availability during an outage. If collection or the buffer dies when production is burning, you lose the forensics and the alerts. Fix: replicate collectors and the buffer x3, reserve headroom for the ~2.5x spike, isolate alerting on the pre-aggregated stream from dashboard read load, and keep the agent’s disk buffer as the last line of defense. The observability pipeline is engineered to be more available than the systems it watches - degrade search before you ever degrade ingest.

Wrap-Up

The trade-offs that define this design:

  • Label index + brute-force scan over a full inverted index. We index only labels (GB) and scan the label+time-pruned, bloom-filtered chunks, giving up instant unscoped full-text search in exchange for an index thousands of times smaller and an ingest path that keeps up at 10M/sec. At ~700 TB/30 days, an index that rivals the data is what actually bankrupts you.
  • Two engines, one pipeline. Logs (text-you-search) land in compressed immutable chunks on object storage with bloom filters; metrics (numbers-you-aggregate) land in a columnar TSDB. They share only the collection agent, the durable buffer, and the query API - because forcing one store to do both does neither well.
  • Reliability at the edge. The agent batches, compresses, and disk-buffers so a backend outage is a delay not a loss and never blocks the host application, and Kafka absorbs the incident spike so backpressure never reaches production code.
  • Aggregate once, read many. A streaming layer materializes rollups (including metrics derived from logs) as data flows, so live dashboards and alerts read tiny pre-computed series instead of rescanning the firehose.
  • At-least-once and eventual consistency over exactly-once. Observability tolerates a late or duplicated event among billions, and we spend that relaxation on timeliness and cost - approximate-but-fresh beats exact-but-late, and it makes the whole pipeline cheaper and more available.

One-line summary: reliable batching agents feed a partitioned durable buffer that forks into a columnar TSDB for metrics and a label-indexed, bloom-filtered, chunked object store for logs, with a streaming layer materializing real-time rollups and a scatter-gather scan serving 30-day search - trading a giant full-text index and exactly-once for a pipeline that ingests 10M events/sec from 100K servers, keeps ~700 TB searchable for 30 days, and stays up when production does not.