← CatZep/Engineering Guides/System Design/Design Scenarios

Designing a Distributed Log Search Engine

A system design interview guide to building a log search engine that retains tens of petabytes, indexes them for near-real-time search, and answers a query across a fleet in seconds — the machinery behind systems like Elasticsearch, OpenSearch, and Loki.

Logs are the highest-volume, most detailed telemetry a system produces, and searching them is deceptively hard: you want to type a free-text query and instantly find the handful of relevant lines out of trillions, spanning data from seconds ago to years ago. That requires two ideas working together — an inverted index that maps terms to the documents containing them, so a search never scans everything, and a distributed, tiered architecture that shards the index across many machines and moves aging data down a hot–warm–cold ladder onto cheap object storage. This guide builds that engine: the index model borrowed from Lucene, sharding and time-partitioning, the tiered storage that makes petabytes affordable, and the scatter-gather query path that turns a query string into a plan executed across shards.

Concepts to understand first

Each definition includes a concrete example and the condition under which the mechanism is useful. You can read this guide without opening another guide.

1. Requirements & Scope

The engine ingests a firehose of log lines and serves search over them. Pin down the requirements.

• Functional. Ingest structured and unstructured log lines; full-text and field search with filters (service, level, time range); aggregations (counts, histograms over time); and results ranked or ordered by time.
• Near-real-time. A log line should be searchable within seconds of being emitted — incidents happen now, not after a batch job.
• Scale. Tens of petabytes retained (exabytes only at the very largest companies), millions of lines per second ingested, queries fanning across thousands of shards.
• Query latency. Interactive queries on recent data in well under a second; historical queries over cold data in seconds.
• Cost. Retaining petabytes on hot storage is unaffordable; the design must tier aggressively onto cheap object storage while keeping old data searchable.
• Custom query language. A DSL that expresses filters, full-text, and aggregations, compiled to a plan executed across shards.

2. Back-of-the-Envelope Estimates

Assume 5,000,000 log lines/second at 500 bytes each, retained 13 months.

Worked estimate, step by step

The math drives the architecture directly: index overhead can rival the raw data, so you cannot fully index everything; and retaining PBs is only affordable on object storage, so the design must keep cold data searchable from S3-class storage rather than local disk.

3. The Index Model & Segments

An inverted index maps analyzed terms to postings (document IDs). Analysis/tokenization determines what terms mean. Doc values support sorting/aggregation; numeric/spatial point indexes support range filtering. Immutable segments let readers use snapshots while new segments are added.

Refresh makes recently indexed data searchable, often from filesystem cache. Durability comes from the translog/WAL fsync and replica acknowledgement policy or a durable commit. A Lucene commit persists index state; translog/engine flush reclaims no-longer-needed recovery records under its policy. 'Refresh=searchable, flush=durable' alone is misleading because acknowledged translog-backed writes can be durable before flush.

Replica loss, asynchronous durability mode, or corruption can still lose state outside the declared guarantee. State the configuration, not 'a crash loses nothing' universally. Updates/deletes create version/tombstone bookkeeping; merge reclaims space later.

4. API & Data Model

# ingest (batched, e.g. the Elasticsearch _bulk API)
POST /logs-2026-09-25/_bulk
{ "index": {} }
{ "@timestamp": "2026-09-25T10:00:01Z", "service": "api", "level": "ERROR",
  "status": 500, "trace_id": "4bf9...", "message": "timeout calling payments" }

# search
POST /logs-*/_search
{ "query": { "bool": {
    "filter": [ { "term": { "service": "api" } },
                { "range": { "@timestamp": { "gte": "now-15m" } } } ],
    "must":   [ { "match": { "message": "timeout" } } ] } },
  "sort": [ { "@timestamp": "desc" } ], "size": 100,
  "aggs": { "per_min": { "date_histogram": { "field": "@timestamp", "fixed_interval": "1m" } } } }

Field mapping is the data model. service, level and status are keyword fields (exact terms plus doc values). message is analyzed text (tokenized into terms). @timestamp is a date (BKD points for range filters).

5. Sharding & Time-Partitioned Indices

One machine cannot hold petabytes, so the index is partitioned two ways.

Time-partitioned indices

Logs are overwhelmingly queried by recent time, so the primary partitioning is by time: create a new index per day (or hour), e.g. logs-2026-07-06. This is the highest-leverage decision in the design because it makes two things trivial. Queries with a time range touch only the relevant indices and skip the rest entirely (partition pruning). And retention/tiering is a whole-index operation — to expire or move old data you drop or migrate an entire old index, with no per-document deletes.

Shards within an index

Each time index is split into shards — each shard being an independent Lucene index — so ingest and search scale horizontally. A document routes to a shard by hashing a routing key (default the doc id, or a tenant id to co-locate a tenant’s data). Each shard has replicas on other nodes for availability and read throughput. Shard count is a real tradeoff: too few and you cannot parallelize or grow; too many and per-shard overhead and query fan-out costs dominate (the “oversharding” problem).

6. Ingest & Tiered Storage

Affordability at petabyte scale comes from moving data down a cost/latency ladder as it ages, since query frequency drops sharply with age.

• Hot. Recent indices on local SSD, fully indexed, serving near-real-time ingest and interactive queries. Small, fast, expensive — kept to days.
• Warm. Older indices moved to cheaper nodes (more storage, less CPU/RAM), still indexed and searchable but not receiving writes. Read-only and often force-merged into fewer segments for query efficiency.
• Cold / frozen. Old indices whose data lives on object storage (S3-class). The node keeps only lightweight metadata and fetches blocks on demand — higher query latency but a fraction of the cost. This searchable-snapshot pattern is what makes retaining petabytes affordable; storage cost decouples from compute.

Data moves down the ladder automatically by index age via a lifecycle policy, and is eventually deleted when it exceeds retention. Because everything is a whole-index operation, tiering is cheap and does not touch individual documents.

7. Query Execution & Scatter-Gather

A query is expressed in a DSL and executed in three phases: parse the query string into an AST, plan it (resolve which indices and shards it touches, prune by time and tenant, choose whether to use the inverted index, doc values, or points), and execute it across shards with scatter-gather.

The execution model is scatter-gather: a coordinator fans the query to every relevant shard, each shard runs the query against its local segments and returns a partial result (top-K hits, or partial aggregation), and the coordinator merges the partials into the final answer. Two optimizations are essential at scale. Pruning — using the time range to skip whole indices and shards — is what keeps a “last 15 minutes” query from touching a year of data. And caching — of filter results, of frequently hit shards, of aggregation results — absorbs repeated dashboard queries. A two-phase approach (query phase returns just doc ids + scores; fetch phase retrieves the full documents only for the final winners) avoids shipping large documents from shards that will not make the top-K.

function search(dsl, time_range):
  ast     = parse(dsl)                          # -> filters, full-text, aggs
  shards  = plan(ast, time_range)               # prune by time + tenant
  partials = scatter(shards, ast)               # each shard searches locally
  return gather(partials, ast.agg)              # merge top-K / aggregations

8. Columnar Analytics, Caching & Compaction

Three supporting mechanisms round out the engine.

Columnar for analytics. Full-text search wants an inverted index, but analytical queries (“count errors per service per hour over a month”) want to scan a few columns over vast rows — exactly what a columnar format like Parquet excels at. So cold data is stored columnar on object storage: cheap, compressible, and fast to scan for aggregations. Modern systems increasingly split the two — an inverted index for search on hot data, columnar files for analytics on cold data — rather than paying full indexing cost forever.

Caching. Query patterns are heavily repeated (dashboards refresh, alerts re-run), so cache aggressively: OS page cache for hot segments, a query/filter cache for reused predicates, and a results cache for identical dashboard queries. Caching is often the difference between a sub-second and a multi-second dashboard.

Compaction / merging. Because segments are immutable and small ones accumulate, background merging rewrites many small segments into fewer large ones. This bounds the number of segments a query must visit (search cost grows with segment count), and is the only time tombstoned/deleted documents are physically reclaimed. On warm data, an explicit force-merge into a single segment maximizes read efficiency since no more writes are coming.

9. Bottlenecks & Tradeoffs

• Index everything vs index selectively. A full inverted index can exceed the raw data in size. Indexing only high-value fields (and storing the rest for scan/columnar) trades some query flexibility for a large cost saving.
• Shard count. Too few shards caps parallelism and growth; too many multiplies overhead and query fan-out. Size shards to a target (e.g. tens of GB each) and use time-partitioning to bound their lifetime.
• Refresh interval vs cost. Frequent refresh gives lower search latency for new data but creates many small segments and more merge work. Relax it for high-ingest, low-freshness workloads.
• Hot storage cost vs query latency. Keeping more data hot is faster but far more expensive; tiering saves money at the cost of higher latency on old data. Tune the ladder to the query-age distribution.
• Write vs query contention. Heavy ingest and heavy search compete for the same nodes. Separating write (hot, indexing) from read (warm/cold, query) roles keeps a query storm from stalling ingestion.

10. Failure Modes

Worked example: visibility is not durability

A log is acknowledged, refreshed into searchable segments, and later committed.

Run a small, testable model

Download concepts.py. It uses only the Python 3.10+ standard library. From the directory where you saved it, run:

python3 concepts.py search
python3 concepts.py test

The model intersects postings and demonstrates date/column pruning against a small exact dataset.

The program contains deterministic assertions and failure traces. It is a learning model, not a production service: it does not implement network consensus, distributed durability, authentication, or third-party delivery. Read its contracts and try changing an input before extending it.

11. 45-Minute Interview Walkthrough

12. Common Mistakes

13. Summary

A distributed log search engine combines the inverted index (search as lookup, not scan) with time-partitioned, sharded, tiered storage (recent data hot and fully indexed, old data cheap on object storage), executed with scatter-gather across shards.

14. Check Yourself

1. How does level:ERROR AND svc:api run without a scan?
   Answer

   Look up both terms in the term dictionary. Fetch their postings lists. Intersect the two sorted lists.

2. What is the difference between refresh and flush in Elasticsearch?
   Answer

   Refresh writes the buffer as a new searchable segment in the OS cache, with no fsync. Flush is a Lucene commit: it fsyncs segments and trims the translog.

3. Why partition indices by day?
   Answer

   Time-range queries skip whole days. Retention and tiering move or drop whole indices, with no per-document deletes.

4. At 54 TB/day compressed and 50 GB shards, how many primary shards per day?
   Answer

   54 TB ÷ 50 GB ≈ 1,080 shards.

5. Why merge segments?
   Answer

   Each query visits every segment, so fewer segments is faster. Merging is also when deleted documents are really removed.

6. How can data on S3 stay searchable?
   Answer

   A searchable snapshot keeps small metadata on the node. It fetches and caches the needed blocks from object storage on demand.

15. Key Terms

Inverted index

An inverted index maps searchable terms to document postings. A columnar file groups values by column so analytical scans can read only selected fields. They optimize different access patterns. Example and limits above.

Postings list

Sorted doc ids for one term.

Segment

An immutable mini-index inside a shard.

Refresh

Makes buffered docs searchable (~1 s).

Flush

Lucene commit plus translog trim.

Translog

Write-ahead log for crash recovery.

Doc values

Columnar per-field data for sort and aggs.

Shard

One Lucene index; the unit of scale.

Searchable snapshot

Index data read from object storage.

16. Further Reading

• Elasticsearch: near real-time search
• Elasticsearch: translog
• Elasticsearch: searchable snapshots
• Apache Lucene
• Grafana Loki documentation (label index, no full-text index)