Distributed Metrics Query Engine

Learning outcomes

A metrics backend can ingest billions of samples per minute and still fail the user, because the moment a dashboard opens it issues a single range query that, run naively, must touch a year of blocks and a million series at once. The hard problem at query time is not fetching samples, it is doing so in parallel, bounded in memory, and fair across tenants. That problem forces the read path to split into several cooperating microservices.

After studying this page, you can:

• Name the services on the read path (query-frontend, query-scheduler, querier, store-gateway, ingester) and trace a request through them.
• Specify the proto3 contract for a streaming range query RPC.
• Turn a PromQL expression into a query plan tree and split it by time window and series shard for parallel execution.
• Explain pushdown, sort-merge of partial results, and result caching.
• Diagnose fan-out explosions, slow workers, and multi-tenant starvation, and name their mitigations.
• Reason about garbage collection, memory-mapped reads, and backpressure as the real programming constraints.
• Map the 10x, 100x, and 1000x evolution onto the Grafana Mimir query path.

Before we dive in

Picture a single Prometheus instance asked to evaluate rate(http_requests_total[5m]) over the last twelve months. The expression itself is small. But the engine must resolve every series behind http_requests_total, then for each series walk every chunk that covers a year of samples, then compute the rate at each step. On a large fleet that is a million series times tens of thousands of samples, materialized into one process at once.

The single-node evaluator dies there. Its memory grows with the product of series count and range, because the PromQL engine, the chunk decoder, and the result set all live in one address space, and nothing bounds them independently. The first long-range query OOMs the process and takes every other in-flight query down with it. Scaling the box up only postpones the wall: a query twice as wide needs twice the memory, and a fleet twice as large doubles the series count again.

The fix is not a faster evaluator. It is to stop doing the whole job in one process. Split the range into windows, split the series into shards, send each piece to a different worker, merge the partial answers at the end, and keep the intermediate state streaming so no single process ever holds the whole result. That reframe, from one big evaluation to a fan-out of small ones coordinated by a frontend, is what a distributed metrics query engine is. Every component on the read path exists to make one part of that fan-out cheap, bounded, or fair.

Microservices topology and component interaction

The read path is a pipeline of stateless coordinators in front of stateful leaves. The write path, which produces the immutable blocks the read path scans, is a separate set of services. Nothing on the read path writes data; nothing on the write path serves a range query. That decoupling is what lets the two scale and fail independently.

Five roles matter, each earned by the problem it solves.

The query-frontend is the entry point. It speaks the client’s protocol (a Prometheus-compatible HTTP API on the outside), and it does three jobs the raw evaluator cannot: it splits a long-range query into smaller sub-queries so no worker sees the whole range, it caches results so a repeated query is served from memory, and it merges the partial answers workers return. It holds no data and no queue, which is why you can run several replicas behind a load balancer.

The query-scheduler owns the queue. Once the frontend has split a request into sub-queries, those sub-queries must wait for a free worker somewhere. Putting that wait inside the frontend couples the number of in-flight requests to the number of frontend replicas. Pulling the queue out into a dedicated stateless service lets the frontend scale, the workers scale, and the queue depth grow, all independently. The scheduler also enforces fairness between tenants, so one noisy tenant cannot starve the rest.

The queriers are the workers. Each is a stateless process that picks up a sub-query from the scheduler, fetches the raw samples it needs, runs the PromQL engine over them, and returns the partial result. A worker never holds state between requests, so adding workers linearly raises the cluster’s query throughput.

The store-gateway is the leaf for historical data. It is stateful: it loads the immutable blocks the write path produced, keeps their indexes memory-mapped, and answers sample fetches by scanning the relevant chunks. Blocks are sharded and replicated across store-gateways by a hash ring, so a worker asks the right gateway for the right blocks.

The ingester is the leaf for recent data. It owns the mutable head where the last few hours of samples still live before they are cut into a block, so a query whose range overlaps the present must read from it as well as from the store-gateway.

The wire between these services is gRPC carrying Protocol Buffer messages, and the query RPCs are streaming. Streaming matters because a year-long scan should not buffer its entire result on the worker before sending the first byte; the worker emits partial series as it goes, and the frontend merges streams as they arrive.

The read lifecycle step by step

A single range query moves through the pipeline in a fixed order: receive, parse, split, enqueue, fan out, gather, return.

The lifecycle earns each step. Receive is the only synchronous client contact, so the frontend does as little work as possible before the request enters the pipeline. Parse turns the expression into an AST once, so the split and the workers share one parse. Split is where the OOM defense begins, because every sub-query the frontend emits is bounded in range. Enqueue hands the sub-queries to the scheduler, which is what lets a burst of requests wait instead of being rejected. Fan out is the moment the work becomes parallel, with each worker touching only its slice of the range and series space. Gather is the merge that reconstructs the single answer the client expects. Return streams the result back, and writes it to cache so the next identical query short-circuits at the front.

Concrete API schemas

The query contract must carry everything the frontend needs to split and the workers need to evaluate: a tenant for isolation, a closed time range with a step, the label matchers that select series, and the expression to evaluate over them.

The split between a streaming response and a request that bundles the whole range is deliberate. The request must be one unit because the frontend parses it once; the response must stream because a year of points should not be buffered before the client sees the first one.

Low-level data structures and execution

The query plan tree

A PromQL expression arrives as text and leaves as a tree of operators over series selectors. The frontend never executes this tree directly. It lowers the AST into a query plan whose leaves are sample fetches and whose interior nodes are the PromQL operators, then it cuts that plan into independent pieces.

The plan has two virtues the AST lacks. It makes the boundaries along which the work can be split explicit (every time window is an independent sub-plan, every series shard within it is an independent fetch), and it lets the frontend push operators below the fetch where possible, so the worker does the smallest amount of work that still answers the question.

Splitting, sharding, and parallelization

Splitting happens on two axes. The first is time: a one-year range becomes roughly fifty-two weekly sub-queries, or some smaller configurable interval (a day is the common default in production systems). Each time-window sub-query scans only its slice of blocks, so its memory footprint is bounded by the window, not the whole range.

A worker's peak memory grows with the series count times the number of step points in its window, so shrinking the window caps memory even when the series count is large.

The second axis is series sharding. A single time window that matches a million series still does not fit one worker, so the frontend adds a shard selector to the matchers and emits one sub-query per shard, each handling a disjoint slice of the series space. The two axes compose: a request for a year of data across a million series becomes a grid of time windows times series shards, each cell an independent, bounded sub-query dispatched to a different worker.

This composition is also where query sharding lives in practice: the frontend rewrites each leaf selector with an extra label matcher that picks one of N shards, runs all N in parallel, and the engine treats them as if the user had asked N smaller questions.

Pushdown to the store-gateway

The naive plan pulls every matching raw sample up to the worker and evaluates there. That is wasteful when the matchers alone discard most of a block, or when the operator is a simple sum. The optimization is pushdown: send the matchers, and sometimes the aggregation, down to the store-gateway so it applies them while scanning the block, and only the survivors travel the network.

Pushdown pays twice. It cuts the bytes on the wire, because the store-gateway returns series and points the worker will actually use, and it cuts the worker’s memory, because the worker never materializes the series it would have dropped. The tradeoff is that the store-gateway must understand the matchers and a subset of the operators, so the plan marks each node as pushable or not, and only the pushable ones travel down.

Merging results and caching

When the workers return, the frontend faces a forest of partial series that must become one answer. The merge is a sort-merge keyed by label set and timestamp: series with identical labels are concatenated in time order, and series with different labels are unioned into the result. Because each worker’s output is already sorted within its window and shard, the merge is linear in the number of partial streams rather than the number of points.

Caching sits in front of the split. Before the frontend emits any sub-query, it hashes the sub-query’s parameters and asks the result cache (backed by Memcached in production systems) for a hit. On a full hit the sub-query never reaches a worker; on a partial hit the frontend computes only the missing time slices and fills the gaps. A second cache, the cardinality cache, remembers how many series a given matcher set resolves, so the frontend can size its fan-out and reject explosive queries before they ever enqueue.

Edge cases and chaos engineering

Fan-out explosion and out of memory

The same fan-out that saves the cluster can kill it. A query whose matcher is loose, like http_requests_total with no further filter across a fleet that has labeled every request with a unique trace id, matches a million series. The frontend dutifully splits by time and shard, every shard fans out, and the workers each materialize their slice. The aggregate memory across the cluster is now a million series times a year of points, and the frontend’s merge holds the union. The query-frontend OOMs before the client sees a byte.

The mitigations are all caps. A max series limit rejects a sub-query whose matcher resolves more than a configured number of series. A max bytes or max samples limit truncates a worker’s output and returns a partial error rather than a full result. Splitting caps bound the number of sub-queries the frontend will emit for one request, so a pathological matcher cannot multiply into unbounded fan-out. Crucially, these limits return a partial error with warnings rather than a silent wrong answer, so the client learns the query was too large and narrows it.

Slow workers and cancellation

A worker that hangs, because its store-gateway is slow or its node is starved, must not hold the whole request hostage. Every sub-query carries a deadline, and the scheduler and workers honor cancellation: when the frontend gives up on a request, it cancels its outstanding sub-queries, and the workers abort their scans instead of finishing useless work. Without cancellation, a slow worker leaks memory and CPU for every request that timed out at the client but is still running in the cluster, a slow accumulation that eventually OOMs the workers.

Multi-tenant fairness

A shared cluster serves many tenants, and a single tenant running a heavy batch of queries can fill the scheduler queue and starve everyone else. The fix is per-tenant fairness inside the scheduler, typically a round-robin across tenants that have active queries, so each tenant gets an equal turn regardless of how many sub-queries it has queued. Some systems add per-tenant queues with separate concurrency limits, so a noisy tenant is bounded by its own quota and the rest of the cluster keeps serving.

System hardening and programming realities

Garbage collection and streaming merges

Every partial result a worker returns is a large, short-lived object: thousands of series, each with thousands of points. Materializing the full result before sending guarantees a garbage collection pause proportional to the result size, which is exactly the latency the user sees. The discipline is to stream: emit each series as soon as it is complete, never hold the whole result in memory at once, and let the frontend’s sort-merge consume the stream incrementally. The same principle bounds the frontend’s merge, which must never buffer all partial streams fully before emitting.

Zero-copy memory-mapped reads

The store-gateway’s hot path is reading block indexes and chunks. Loading those into heap memory would double the working set (the file cache the operating system already keeps, plus the process copy) and add a copy on every read. The fix is to memory-map the index header and chunk files, so the store-gateway reads bytes directly from the operating system’s page cache with no copy. Only the portion of the index the query touches is faulted into the process; the rest stays on disk until needed. This is what lets a store-gateway serve terabytes of blocks from a process whose resident memory is a fraction of that.

Backpressure between scheduler and workers

A scheduler queue is a buffer, and an unbounded buffer is an OOM waiting to happen. When the workers are saturated, the queue grows; when the queue grows past a limit, the scheduler must push back on the frontends, which push back on the clients, ideally by rejecting new sub-queries with a clear “busy” error rather than accepting them and timing out later. The number of workers each scheduler feeds is also bounded, matched roughly to the worker’s max-concurrent setting, so a worker is never handed more sub-queries than it can run at once. Backpressure turns a slow degradation into a fast, honest failure.

Architectural evolution: 10x, 100x, 1000x

At 10x: a single querier and a basic cache

At modest scale the read path is one querier process that fetches from local blocks and a tiny result cache. Queries are small enough that a single evaluator handles them, and the cache absorbs repeated dashboard refreshes. The failure mode here is the first wide query: with no splitting, a month-long range evaluates in one process and OOMs under load. The lesson learned is that range and series count must be bounded somewhere other than the evaluator.

At 100x: splitting, scheduler, and store-gateway sharding

Growth forces the split. The query-frontend appears to take over splitting and caching, the query-scheduler appears to hold the queue so the frontend can scale, and the store-gateway becomes a separate sharded service reading from object storage rather than local disk. Blocks are now distributed across many store-gateways by a hash ring, and the querier fans out to the right ones. The read path is now a genuine pipeline, and a wide query becomes many small ones. The new failure mode is cross-tenant interference, which is what drives the scheduler’s fairness.

At 1000x: cells and cross-cell federation

At the largest scale a single cluster hits limits on the hash ring, the object store, and the blast radius of a failure. The architecture splits into cells, each a self-contained cluster with its own frontend, scheduler, workers, and store-gateways. A query that spans cells is federated: a thin layer over the cells issues the sub-query to each and merges, much as the frontend merges across workers within a cell. Asynchronous, decoupled query paths appear, where heavy analytical work runs out of band and the synchronous path returns quickly. The throughline is the same: split, fan out, merge, and bound every stage.

How Grafana Mimir does it

Grafana Mimir is the canonical embodiment of this design, and its query path maps almost one-to-one onto the topology above.

The query-frontend is the stateless entry point of Mimir’s read path. It splits long-range queries into multiple queries over a configurable interval (twenty-four hours by default), runs the parts in parallel across downstream queriers, and combines the results. This splitting is what prevents large multi-day or multi-month queries from causing out-of-memory errors in a querier. The frontend also caches query results, backed by Memcached, and when a cached result is incomplete it computes only the missing partial queries and runs those in parallel.

The query-scheduler is the stateless service that holds the in-memory queue of queries waiting for a querier. Pulling queuing out of the frontend lets the frontend scale, and it isolates queries that only need ingesters from store-gateway degradation. Fairness is enforced with a simple round-robin between all tenants that have active queries, so no single tenant monopolizes the workers.

The querier is a stateless worker. It receives a query range request carrying the expression, start, end, and step, computes which blocks it must consult, sends fetch requests to the matching store-gateways, and if the range overlaps the configured query-ingesters-within duration, also reads recent data from the ingesters. After it has gathered all samples it runs the PromQL engine and returns the result. It runs a consistency check that every expected block was actually queried, and if not, retries missing blocks up to the replication factor (default three) before failing the query rather than returning a wrong answer.

The store-gateway is the stateful leaf over historical blocks. Store-gateways join a hash ring and shard and replicate blocks across the pool, using shuffle-sharding so only a subset of instances load each tenant’s blocks and the blast radius of one tenant’s workload stays inside its shard. Each store-gateway memory-maps the index-header of its blocks to local disk, keeping only a portion in memory and reading the rest on demand, which is what lets it serve terabytes of blocks from bounded resident memory. If a querier asks a store-gateway for a block it has not loaded, the querier retries a different replica, also up to the replication factor.

The ingester holds the mutable head where the most recent samples live before they are cut into a block and uploaded, roughly every two hours. The querier reads from it whenever the query range touches the present, and it skips any ingester the ring reports as unhealthy. The whole read path runs in microservices mode, separately from the write path, which is what lets the two scale and fail on their own schedules.

Mastery Questions

Question

A dashboard issues a one-year range query that OOMs a single-node evaluator. Name the two axes the query-frontend splits along, and why each bounds worker memory.

Question

Why is the query RPC streaming instead of unary, and what goes wrong if it buffers?

Question

A loose matcher matches millions of series and the query-frontend OOMs during the merge. Name three caps that prevent this.

Question

What does pulling the queue out of the query-frontend into a query-scheduler buy, and how is fairness enforced?

Question

Why does the store-gateway memory-map its index header instead of loading blocks into heap memory?

Question

Map the five read-path roles in Grafana Mimir to their jobs, and say what each one does NOT do.

Answer

It splits along time (the range becomes many short-window sub-queries) and along series shards (each window becomes disjoint series slices). Each worker then holds only $N_{\text{series}} \cdot \Delta t_{\text{window}} / \Delta t_{\text{step}}$ worth of points, so shrinking the window and the shard caps memory regardless of the total range or series count.

Answer

Streaming lets the first partial results leave the worker before the scan finishes, so a year-long query returns its first bytes quickly. Buffering the whole result forces a garbage-collection pause proportional to the result size and holds the entire answer in memory at once, which is the OOM the split was meant to prevent.

Answer

A max-series limit rejects a sub-query whose matcher resolves too many series; a max-bytes or max-samples limit truncates a worker's output; and a splitting cap bounds how many sub-queries one request can fan into. Each returns a partial error with warnings rather than a wrong or absent answer.

Answer

It decouples in-flight depth from the number of frontend replicas, so the frontend, the queue, and the workers scale independently, and a store-gateway slowdown cannot stall queries that only need ingesters. Fairness is a round-robin across tenants with active queries, optionally backed by per-tenant queues with separate concurrency limits.

Answer

Memory-mapping reads bytes straight from the operating system page cache with no copy, so the block data lives once in the file cache instead of twice. Only the index portion a query touches is faulted in; the rest stays on disk, which lets a store-gateway serve terabytes of blocks from a resident memory that is a fraction of that.

Answer

The query-frontend splits, caches, and merges but holds no queue or data; the query-scheduler queues and enforces fairness but evaluates nothing; the querier workers evaluate PromQL but hold no state between requests; the store-gateway serves historical blocks but never writes; the ingester serves recent data but is not on the historical read path. Each role is bounded by what it owns.

Question

A dashboard issues a one-year range query that OOMs a single-node evaluator. Name the two axes the query-frontend splits along, and why each bounds worker memory.

Answer

It splits along time (the range becomes many short-window sub-queries) and along series shards (each window becomes disjoint series slices). Each worker then holds only $N_{\text{series}} \cdot \Delta t_{\text{window}} / \Delta t_{\text{step}}$ worth of points, so shrinking the window and the shard caps memory regardless of the total range or series count.

Prev1 / 6Next