Cluster Specification

Vector Clocks

Vector clocks are a type of data structure and algorithm for generating a partial ordering of events in a distributed system and detecting causality violations.

We use vector clocks to reconcile and merge differences in cluster state during gossiping. A vector clock is a set of (node, counter) pairs. Each update to the cluster state has an accompanying update to the vector clock.

Gossip Convergence

Information about the cluster converges locally at a node at certain points in time. This is when a node can prove that the cluster state he is observing has been observed by all other nodes in the cluster. Convergence is implemented by passing a map from node to current state version during gossip. This information is referred to as the gossip overview. When all versions in the overview are equal there is convergence. Gossip convergence cannot occur while any nodes are unreachable. The nodes need to be moved to the down or removed states (see the Membership Lifecycle section below).

Failure Detector

The failure detector is responsible for trying to detect if a node is unreachable from the rest of the cluster. For this we are using an implementation of The Phi Accrual Failure Detector by Hayashibara et al.

An accrual failure detector decouple monitoring and interpretation. That makes them applicable to a wider area of scenarios and more adequate to build generic failure detection services. The idea is that it is keeping a history of failure statistics, calculated from heartbeats received from other nodes, and is trying to do educated guesses by taking multiple factors, and how they accumulate over time, into account in order to come up with a better guess if a specific node is up or down. Rather than just answering "yes" or "no" to the question "is the node down?" it returns a phi value representing the likelihood that the node is down.

The threshold that is the basis for the calculation is configurable by the user. A low threshold is prone to generate many wrong suspicions but ensures a quick detection in the event of a real crash. Conversely, a high threshold generates fewer mistakes but needs more time to detect actual crashes. The default threshold is 8 and is appropriate for most situations. However in cloud environments, such as Amazon EC2, the value could be increased to 12 in order to account for network issues that sometimes occur on such platforms.

In a cluster each node is monitored by a few (default maximum 5) other nodes, and when any of these detects the node as unreachable that information will spread to the rest of the cluster through the gossip. In other words, only one node needs to mark a node unreachable to have the rest of the cluster mark that node unreachable. Right now there is no way for a node to come back from unreachable. This is planned for the next release of Akka. It also means that the unreachable node needs to be moved to the down or removed states (see the Membership Lifecycle section below).

Leader

After gossip convergence a leader for the cluster can be determined. There is no leader election process, the leader can always be recognised deterministically by any node whenever there is gossip convergence. The leader is simply the first node in sorted order that is able to take the leadership role, where the preferred member states for a leader are up and leaving (see the Membership Lifecycle section below for more information about member states).

The role of the leader is to shift members in and out of the cluster, changing joining members to the up state or exiting members to the removed state. Currently leader actions are only triggered by receiving a new cluster state with gossip convergence.

The leader also has the power, if configured so, to "auto-down" a node that according to the Failure Detector is considered unreachable. This means setting the unreachable node status to down automatically.

Seed Nodes

The seed nodes are configured contact points for initial join of the cluster. When a new node is started it sends a message to all seed nodes and then sends a join command to the seed node that answers first.

It is possible to not use seed nodes and instead join any node in the cluster manually.

Gossip Protocol

A variation of push-pull gossip is used to reduce the amount of gossip information sent around the cluster. In push-pull gossip a digest is sent representing current versions but not actual values; the recipient of the gossip can then send back any values for which it has newer versions and also request values for which it has outdated versions. Akka uses a single shared state with a vector clock for versioning, so the variant of push-pull gossip used in Akka makes use of this version to only push the actual state as needed.

Periodically, the default is every 1 second, each node chooses another random node to initiate a round of gossip with. The choice of node is random but can also include extra gossiping nodes with either newer or older state versions.

The gossip overview contains the current state version for all nodes and also a list of unreachable nodes. This allows any node to easily determine which other nodes have newer or older information, not just the nodes involved in a gossip exchange.

The nodes defined as seed nodes are just regular member nodes whose only "special role" is to function as contact points in the cluster.

During each round of gossip exchange it sends Gossip to random node with newer or older state information, if any, based on the current gossip overview, with some probability. Otherwise Gossip to any random live node.

The gossiper only sends the gossip version to the chosen node. The recipient of the gossip can use the gossip version to determine whether:

1. it has a newer version of the gossip state, in which case it sends that back to the gossiper, or
2. it has an outdated version of the state, in which case the recipient requests the current state from the gossiper

If the recipient and the gossip have the same version then the gossip state is not sent or requested.

State Diagram for the Member States

Member States

• joining

  transient state when joining a cluster

• up

  normal operating state

• leaving / exiting

  states during graceful removal

• down

  marked as down (no longer part of cluster decisions)

• removed

  tombstone state (no longer a member)

User Actions

• join

  join a single node to a cluster - can be explicit or automatic on startup if a node to join have been specified in the configuration

• leave

  tell a node to leave the cluster gracefully

• down

  mark a node as down

Leader Actions

The leader has the following duties:

• shifting members in and out of the cluster
  • joining -> up
  • exiting -> removed

Failure Detection and Unreachability

• fd*

  the failure detector of one of the monitoring nodes has triggered causing the monitored node to be marked as unreachable

• unreachable*

  unreachable is not a real member state but more of a flag in addition to the state signaling that the cluster is unable to talk to this node

Transitions

There are transition times in the handoff process where different approaches can be used to give different guarantees.

Graceful Handoff

A more complete process for graceful handoff would be:

1. the leader sets a pending change for N1 to handoff A to N2
2. N1 notices the pending change and sends an initialization message to N2. Options:
   1. keep A on N1 active and continuing processing messages as normal
   2. N1 forwards all messages for A to N2
   3. N1 drops all messages for A (terminate A with messages becoming dead letters)
3. in response N2 creates A and sends back a ready message. Options:
   1. N2 simply processes messages for A as normal
   2. N2 creates a buffer per sending node for A. Each buffer is opened (flushed and removed) when an acknowledgement for the sending node has been received (via N1)
4. after receiving the ready message N1 marks the change as complete. Options:
   1. N1 forwards all messages for A to N2 during the update transition
   2. N1 drops all messages for A (terminate A with messages becoming dead letters)
5. the leader sees the migration is complete and updates the partition table
6. all nodes eventually see the new partitioning and use N2
   1. each node sends an acknowledgement message to N1
   2. when N1 receives the acknowledgement it can count down the pending acknowledgements and remove forwarding when complete
   3. when N2 receives the acknowledgement it can open the buffer for the sending node (if buffers are used)

The default approach is to take options 2a, 3a, and 4a - allowing A on N1 to continue processing messages during migration and then forwarding any messages during the update transition. This assumes stateless actors that do not have a dependency on message ordering from any given source.

• If an actor has a distributed durable mailbox then nothing needs to be done, other than migrating the actor.
• If message ordering needs to be maintained during the update transition then option 3b can be used, creating buffers per sending node.
• If the actors are robust to message send failures then the dropping messages approach can be used (with no forwarding or buffering needed).
• If an actor is a singleton (only one instance possible throughout the cluster) and state is transferred during the migration initialization, then options 2b and 3b would be required.