Distributors, ingesters, queriers (briefly)

Build a distributed logging stack (ELK / Loki) (12 scenes)

Scene 10 · Distributors, ingesters, queriers (briefly)

Both ELK and Loki are sharded write-paths plus sharded read-paths plus a backing store; they differ in the partition key — hash(doc_id) vs hash(label-set).

Previously

We've shaped the data going in, through, and out. Now zoom out to the moving parts of the backend — but only briefly, because the choices we already made determine most of what matters here.

Scene 10

Distributors, ingesters, queriers (briefly)

Watch

Diagram

Two mini-stacks side by side. LEFT is ELK: agents → coordinating node → data nodes, where each data node holds shards (and replica shadows for HA). The shard-routing rule is `hash(doc_id) mod N`. RIGHT is Loki: agents → distributors → a consistent-hash ring → ingesters (replication factor 3 means each stream is held by 3 ingesters) → object store at the bottom. Reads on Loki run through query frontend → query scheduler → queriers, which fan out to BOTH ingesters (for recent data) AND the object store (for old data). A bottom strip surfaces the vocabulary collision: 'shard' (ELK) vs 'stream' (Loki) — same achievement (horizontal scale via consistent hashing), different choice of partition key.

ELK partitions by hash(doc_id) — shards land on data nodes

Two mini-stacks. LEFT (ELK): agents → coordinating node → data nodes; the partition key is hash(doc_id), and each shard has a replica shadow on a neighboring data node. RIGHT (Loki): agents → distributors → consistent-hash ring → ingesters; each stream has 3 replicas (RF=3), and ingesters periodically flush to the object store. Reads on Loki fan out from queriers to BOTH ingesters and the object store. The bottom strip names the vocabulary collision: shard (ELK) vs stream (Loki) — same achievement, different partition-key choice.

Implementation

ELK.coordinator_route

coordinating node hashes doc_id, forwards to data node + replicas

1def on_index(doc):
2    shard_id = hash(doc.id) mod num_shards
3    primary = routing_table.primary(shard_id)
4    replicas = routing_table.replicas(shard_id)  # RF-1
5    primary.write(doc)
6    for r in replicas:
7        r.write(doc)  # in-sync replication

Loki.distributor_route

distributor hashes label-set, sends to RF=3 ingesters on the ring

1def on_push(line):
2    key = hash(canonicalize(line.label_set))
3    owners = ring.successors(key, n=replication_factor)  # RF=3
4    acks = 0
5    for ingester in owners:
6        if ingester.send(line):
7            acks += 1
8    return acks >= quorum  # 2 of 3

Querier.fan_out

queriers ask ingesters for recent data AND object store for old

1def execute(query):
2    plan = frontend.split(query)        # frontend → scheduler
3    ingester_chunks = [
4        i.query(plan) for i in ring.owners(plan.label_set)
5    ]
6    store_chunks = object_store.fetch(plan.time_range)
7    return merge(ingester_chunks, store_chunks)

PreviousSampling — head, tail, and the only error NextDesign your logging stack