Gossip and membership

The problem: who's alive?

Snip's live click counts (07.09 Fan-out patterns) run on 400 gateway servers across two regions. Each needs to know which of the others are alive, to route subscriptions and spread load. Servers come and go all day with deploys and autoscaling, and some fail without saying goodbye.

Two obvious designs struggle:

• A central registry that everyone heartbeats to. Simple, and fine for many systems (it's what etcd or Consul's servers can do), but it's another critical service, and a partition from it makes healthy servers look dead.
• Everyone heartbeats everyone. 400 servers each pinging 399 others every second is about 160,000 messages a second, growing with the square of the cluster size. The SWIM paper opens with exactly this problem: traditional heartbeating protocols "impose network loads that grow quadratically with group size", or detect failures slowly or wrongly.

Gossip: tell a few, who tell a few

In a gossip (or epidemic) protocol, each node periodically picks a few random peers and shares what it knows: "server 117 joined", "server 42 looks dead". Those peers pass it on in their next round, and so on. News spreads like a rumour: the number of nodes that know roughly multiplies each round.

Simulated with each node telling 3 random peers per round (averaged over 30 runs):

The rounds grow with the logarithm of the cluster size: ten times the nodes costs only two or three more rounds. Each node sends the same few messages per round however big the cluster gets. And there's no single point of failure: news routes around dead nodes on its own.

The price is that it's eventually consistent (08.02 Consistency models): for a few rounds, different nodes have different views of the membership. Systems built on gossip must tolerate that, for example by retrying a request on another node if the first turns out to be gone.

Detecting failures: SWIM

SWIM (Scalable Weakly-consistent Infection-style Membership) separates the two jobs:

Failure detection, by probing. Every period, each node pings one random member. If it gets no reply in time, it doesn't declare the member dead; it asks k other members to ping it on its behalf (ping-req). Only if none of them gets a reply is the member suspected. This indirect check filters out problems that are only between two nodes, such as one congested link.

Suspicion before death. A suspected member isn't removed at once. The suspicion is gossiped, and the member has a short time to refute it ("I'm alive, incarnation 8"). Only if it doesn't is it declared dead. This cuts false failures from a node that was briefly slow.

Dissemination, by piggybacking. Membership news (joins, suspicions, deaths) isn't sent separately; it rides along on the ping and ack messages already being sent, infection-style.

The result, from the paper: the expected time to first detect a failure and the expected message load per member don't vary with group size.

Where you'll meet it

• Consul uses a gossip protocol (Serf) for membership and failure detection among all its agents, in a LAN pool per datacenter and a WAN pool between datacenters, alongside Raft among its servers for data that needs consensus (08.06 Consensus and Raft, intuitively).
• HashiCorp's memberlist library, which Serf builds on, implements SWIM with extensions, including Lifeguard, which makes it more robust when a member is slow at processing messages.
• Many Dynamo-style databases and cache clusters gossip membership and ring layout between nodes, so any node can route a request (08.04 Replication and quorums).

A common pattern combines both: gossip for who's alive (large, changing, eventually consistent), consensus for who's in charge and what the configuration is (small, critical, must agree).

Recap

Interview questions