Failure detection and gossip
Read firstPartial failure
A database cluster has 300 nodes. At 3 a.m. one of them loses power. Within a few seconds, every other node should stop sending it reads, start keeping writes meant for it, and agree that it is gone, with no central server (which could itself be the one that died) and without every node pinging every other node 90,000 times a second.
Two questions sit under every cluster: who is in it (membership), and who is alive (failure detection). This chapter is how large clusters answer both cheaply, and what the answers are not good enough for.
A 300 node cluster needs every node to learn within seconds that one node died. A central monitor is a single point of failure, and all-to-all heartbeats are 90,000 messages a round. What else could you do?
The obvious designs, and where they stop#
A central monitor. Every node heartbeats to one place, which declares deaths and tells everyone. Simple and correct, and the monitor is now the thing that must never fail. Make it a three node consensus group and this is a perfectly good design for tens or hundreds of nodes. Kubernetes works this way: each node renews a Lease in the API server every few seconds, and the control plane marks a node NotReady when renewals stop for a grace period. If you already run etcd or ZooKeeper, use their leases and stop here.
All-to-all heartbeats. Every node pings every other node. No central point, and the load is N × (N − 1) messages per round: 90,000 for 300 nodes, a million for 1,000. It also floods the network at exactly the moment it is struggling.
Systems with hundreds or thousands of peers and no central coordinator (Cassandra, Consul's agents, Dynamo-style stores) use a third design.
Gossip#
Gossip spreads information the way a rumour spreads in an office. Every round, every node that knows something tells a few nodes picked at random. Those tell a few more. Play it.
The informed group roughly multiplies every round, so everyone knows in about log N rounds, plus a couple of rounds for the last stragglers. Doubling the cluster adds about one round. Meanwhile each node only ever sends a handful of messages per round, however large the cluster gets. That combination (fast spread, flat per-node cost) is why gossip scales.
It is also hard to break. Messages can be lost and nodes can die mid-rumour, and the news still gets through, because every node hears it from several random directions.
What gets gossiped is usually a table: for each node, a heartbeat counter it increments itself, plus a version number. When two nodes gossip, they merge tables by keeping the higher version of each entry. That merge is commutative, associative and idempotent, the same property that makes CRDTs work, which is why gossip can deliver in any order, twice, or late, and still converge. Cassandra gossips once a second with up to three other nodes.
Detecting a failure: SWIM#
Gossip spreads the news. Something still has to notice the death in the first place, without the N² cost. SWIM (from a 2002 paper by Abhinandan Das, Indranil Gupta and Ashish Motivala) does it with two ideas: probe one random member per period, and ask others for help before accusing anyone.
HashiCorp's memberlist library implements SWIM, with refinements they published in 2017 as Lifeguard: a node that is itself slow (overloaded, paused) becomes less quick to accuse others. It runs inside Consul, Serf and Nomad.
How suspicious to be: phi accrual#
A fixed rule ("three missed heartbeats means dead") is wrong on a network whose delays vary. The phi accrual failure detector (Hayashibara and colleagues, 2004) learns how regularly each node's heartbeats usually arrive, and turns "no heartbeat for this long" into a suspicion level, phi, that rises smoothly. A heartbeat that is 2 seconds late from a node that is normally steady to within 10 ms is alarming. The same delay from a node that is usually jittery is not. Cassandra convicts a node when phi passes a threshold, 8 by default.
Either way, remember what Partial failure said: a failure detector can only ever say "I have not heard from you lately". It suspects. It cannot know.
What gossip is not for#
Gossip converges. It does not agree. Two nodes can hold different views of the cluster for a few seconds, and that is fine for some decisions and fatal for others.
Gossip has one more common job: anti-entropy. Replicas of the same data compare what they hold in the background and copy over whatever the other is missing, so a write that missed a replica during an outage eventually arrives. Dynamo-style stores compare Merkle trees (hashes of key ranges) so two replicas can find the few differing keys without sending everything. The key value store design covers that repair path.
- CassandraGossip every second with up to 3 peers, phi accrual detection
- ConsulSWIM gossip (memberlist) for membership, Raft for the catalogue
- KubernetesNode Leases to the API server instead of gossip
- RedisRedis Cluster nodes gossip slot ownership and failures
| Choice | What you gain | What you pay | Pick it when |
|---|---|---|---|
| Central monitor on a consensus store | One authoritative answer, and simple to reason about. | Load concentrates in one place, and the store must be run carefully. | Up to hundreds of nodes, or whenever you already run etcd or ZooKeeper. |
| Gossip membership with SWIM | Constant load per node, no single point of failure, spreads in about log N rounds. | Views disagree for seconds, and every decision made on them must tolerate that. | Hundreds to thousands of peers with no natural coordinator. |
| All-to-all heartbeats | Every node knows directly, with no protocol to implement. | N² messages per round, which hurts most when the network is already struggling. | Small groups only, a handful of nodes. |
Checkpoint#
1. Roughly how many gossip rounds does it take for news to reach every node in a cluster of 1,000?
2. Why does SWIM ask other nodes to ping a target that did not answer, instead of declaring it dead?
3. Your cluster uses gossip for membership. Can you use the gossiped view to decide which node is the shard primary?
- All-to-all heartbeats cost N²: 90,000 messages a round for 300 nodes.
- Gossip reaches everyone in about log N rounds, at constant cost per node.
- SWIM: ping one, then ask k others before suspecting. Suspect, then dead.
- Gossip converges, consensus agrees. Leaders and locks need consensus.
A cluster needs to know who is a member and who is alive. A central monitor on a consensus store works up to hundreds of nodes, and all-to-all heartbeats cost N squared messages, so large peer-to-peer clusters use gossip: every round each node shares its state with a few random peers, and news reaches everyone in about log N rounds at constant cost per node, surviving lost messages and dead nodes. State merges by keeping the highest version, so order and duplicates do not matter. To detect failures, SWIM pings one random member per period, asks a few others to probe before suspecting, and lets a suspected node refute the rumour before it is declared dead. Phi accrual scales suspicion to how regular a node's heartbeats normally are. Gossip converges but does not agree, so I use it for routing and membership, and consensus for leaders, locks and shard ownership.