3. Knowing which machines are alive

A training job can run on a machine for hours or days. If that machine loses power, or its network cable is pulled, the job stops, and nothing on the machine can report the failure. The control plane has to notice on its own, so that it can restart the job somewhere else.

In this chapter, we will learn how a control plane decides which machines are alive. We will first build intuition by reasoning about heartbeats and timeouts, and quickly prototype a liveness registry and a failure detector from scratch. Then we will use the Lease objects that Kubernetes itself uses to track its nodes: we will write an agent that renews a Lease for a host and a watcher that notices when the renewals stop, and run both on our cluster.

As before, every program is written in Go and in Rust. The two models, heartbeats and phi, use only the standard library. The two Kubernetes programs, leaseagent and leasewatch, use client-go and kube-rs. The Go programs are in code/go, and we run them with go run ./<name>; the Rust programs are packages in the workspace in code/rust, and we run them with cargo run -p <name>.

Heartbeats and timeouts

A machine that has crashed cannot announce that it has crashed. The only evidence the control plane gets is silence. So each machine runs a small agent that sends a message, a heartbeat, at a fixed interval, and the control plane treats a long enough silence as a failure.

The difficulty is that silence has more than one cause. The machine may be dead, or it may be alive but overloaded, or the network between it and the control plane may be broken while the machine keeps running its job. From the control plane’s side, the three look the same. Every failure detector therefore makes a bet on how long a silence has to last before it is more likely to mean death than delay, and the bet has a cost in both directions. A short timeout declares healthy machines dead, and their jobs are restarted for nothing, sometimes while the original copy is still running. A long timeout leaves the jobs of truly dead machines stopped for longer.

Keeping heartbeats out of the database

The control plane has to remember the last heartbeat of every machine. The obvious place for that is its database, and the obvious design writes every heartbeat there. A fleet of 10,000 hosts sending one heartbeat a second would then make 10,000 writes a second. In a replicated store such as etcd, every write is sent to a majority of the replicas and flushed to disk before it is acknowledged, and a steady stream of 10,000 writes a second would leave little capacity for the writes that matter, such as creating jobs.

Most heartbeats carry no news: they say that a host that was alive is still alive. The registry can keep the time of each host’s last heartbeat in memory and write to the database only when a host changes state. Our model uses three states. A host is Up while heartbeats arrive. It becomes Suspect after 3 seconds of silence, which lets the control plane stop placing new work on it without yet giving up on the work it has. It becomes Dead after 10 seconds, at which point its jobs would be restarted elsewhere.

The model simulates 10,000 hosts for 60 seconds, without real time passing. Host 42 crashes at second 20, and host 7 misses four heartbeats, from second 30 to second 33.

The types come first. A host’s record is two fields, and a Transition is the only thing the registry writes to the database.

Listing 3.1 The states, the thresholds, and the registry's records, in Go. code/go/heartbeats/main.go

// State is what the registry believes about a host.
type State int

// The states a host moves through.
const (
	Up      State = iota // heartbeats are arriving
	Suspect              // heartbeats stopped recently; the host may be slow
	Dead                 // heartbeats stopped long enough to give up on the host
)

// String returns the state's name for the log.
func (s State) String() string {
	return [...]string{"Up", "Suspect", "Dead"}[s]
}

// Thresholds and the size of the simulation, in seconds and hosts.
const (
	suspectAfter = 3      // seconds without a heartbeat before a host is Suspect
	deadAfter    = 10     // seconds without a heartbeat before a host is Dead
	hostCount    = 10_000 // hosts in the fleet
	duration     = 60     // seconds simulated
)

// entry is the registry's in-memory record of one host.
type entry struct {
	// lastSeen is the second of the host's most recent heartbeat.
	lastSeen int
	// state is the host's current state.
	state State
}

// Transition is one durable write: a host entering a new state.
type Transition struct {
	// At is the second at which the sweep or heartbeat noticed the change.
	At int
	// Host is the host's number.
	Host int
	// State is the state the host entered.
	State State
}

// Registry holds the liveness of every host.
type Registry struct {
	// hosts is indexed by host number.
	hosts []entry
	// heartbeats counts the heartbeats handled in memory.
	heartbeats int
	// writes lists the durable writes, in order.
	writes []Transition
}

Listing 3.2 The states, the thresholds, and the registry's records, in Rust. code/rust/heartbeats/src/main.rs

/// Seconds without a heartbeat before a host is Suspect.
const SUSPECT_AFTER: u32 = 3;
/// Seconds without a heartbeat before a host is Dead.
const DEAD_AFTER: u32 = 10;
/// Hosts in the fleet.
const HOST_COUNT: usize = 10_000;
/// Seconds simulated.
const DURATION: u32 = 60;

/// What the registry believes about a host.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum State {
    /// Heartbeats are arriving.
    #[default]
    Up,
    /// Heartbeats stopped recently; the host may be slow.
    Suspect,
    /// Heartbeats stopped long enough to give up on the host.
    Dead,
}

/// The registry's in-memory record of one host.
#[derive(Clone, Copy, Default)]
struct Entry {
    /// The second of the host's most recent heartbeat.
    last_seen: u32,
    /// The host's current state.
    state: State,
}

/// One durable write: a host entering a new state.
struct Transition {
    /// The second at which the sweep or heartbeat noticed the change.
    at: u32,
    /// The host's number.
    host: usize,
    /// The state the host entered.
    state: State,
}

/// Holds the liveness of every host.
struct Registry {
    /// Indexed by host number.
    hosts: Vec<Entry>,
    /// Counts the heartbeats handled in memory.
    heartbeats: usize,
    /// The durable writes, in order.
    writes: Vec<Transition>,
}

The registry has one operation for an arriving heartbeat and one for the sweep, and a small function that turns a length of silence into a state.

Listing 3.3 A heartbeat, the sweep, and the rule from silence to state, in Go. code/go/heartbeats/main.go

// NewRegistry returns a registry in which every host is Up, as it would be
// after the hosts registered.
func NewRegistry(n int) *Registry {
	return &Registry{hosts: make([]entry, n)}
}

// Heartbeat records a heartbeat from host at second now. Only a host that was
// not Up costs a durable write.
func (r *Registry) Heartbeat(now, host int) {
	r.heartbeats++
	e := &r.hosts[host]
	e.lastSeen = now

	if e.state != Up {
		e.state = Up
		r.writes = append(r.writes, Transition{now, host, Up})
	}
}

// Sweep re-evaluates every host at second now and records each change of state.
func (r *Registry) Sweep(now int) {
	for host := range r.hosts {
		e := &r.hosts[host]

		if next := stateAfter(now - e.lastSeen); next != e.state {
			e.state = next
			r.writes = append(r.writes, Transition{now, host, next})
		}
	}
}

// stateAfter returns the state of a host that has been silent for the given
// number of seconds.
func stateAfter(silent int) State {
	switch {
	case silent >= deadAfter:
		return Dead
	case silent >= suspectAfter:
		return Suspect
	default:
		return Up
	}
}

Listing 3.4 A heartbeat, the sweep, and the rule from silence to state, in Rust. code/rust/heartbeats/src/main.rs

impl Registry {
    /// Returns a registry in which every host is Up, as it would be after the
    /// hosts registered.
    fn new(n: usize) -> Self {
        Self {
            hosts: vec![Entry::default(); n],
            heartbeats: 0,
            writes: Vec::new(),
        }
    }

    /// Records a heartbeat from `host` at second `now`. Only a host that was
    /// not Up costs a durable write.
    fn heartbeat(&mut self, now: u32, host: usize) {
        self.heartbeats += 1;
        let entry = &mut self.hosts[host];
        entry.last_seen = now;

        if entry.state != State::Up {
            entry.state = State::Up;
            self.writes.push(Transition {
                at: now,
                host,
                state: State::Up,
            });
        }
    }

    /// Re-evaluates every host at second `now` and records each change of state.
    fn sweep(&mut self, now: u32) {
        for (host, entry) in self.hosts.iter_mut().enumerate() {
            let next = state_after(now - entry.last_seen);

            if next != entry.state {
                entry.state = next;
                self.writes.push(Transition {
                    at: now,
                    host,
                    state: next,
                });
            }
        }
    }
}

/// Returns the state of a host that has been silent for `silent` seconds.
fn state_after(silent: u32) -> State {
    match silent {
        s if s >= DEAD_AFTER => State::Dead,
        s if s >= SUSPECT_AFTER => State::Suspect,
        _ => State::Up,
    }
}

The run feeds the registry one heartbeat from every host each second, except from the two hosts that the scenario silences.

Listing 3.5 The simulated fleet and the main loop, in Go. code/go/heartbeats/main.go

// sendsHeartbeat reports whether host sends a heartbeat at second now in the
// simulation.
func sendsHeartbeat(host, now int) bool {
	switch host {
	case 42:
		return now < 20 // crashes at second 20
	case 7:
		return now < 30 || now >= 34 // silent from second 30 to 33
	default:
		return true
	}
}

// main runs the simulation and prints the registry's work.
func main() {
	registry := NewRegistry(hostCount)

	for now := 0; now < duration; now++ {
		for host := 0; host < hostCount; host++ {
			if sendsHeartbeat(host, now) {
				registry.Heartbeat(now, host)
			}
		}

		registry.Sweep(now)
	}

	fmt.Printf("heartbeats handled in memory: %d\n", registry.heartbeats)
	fmt.Printf("durable writes:               %d\n", len(registry.writes))

	for _, w := range registry.writes {
		fmt.Printf("  second %2d  host %2d -> %s\n", w.At, w.Host, w.State)
	}
}

Listing 3.6 The simulated fleet and the main loop, in Rust. code/rust/heartbeats/src/main.rs

/// Reports whether `host` sends a heartbeat at second `now` in the simulation.
fn sends_heartbeat(host: usize, now: u32) -> bool {
    match host {
        42 => now < 20,                // crashes at second 20
        7 => !(30..34).contains(&now), // silent from second 30 to 33
        _ => true,
    }
}

/// Runs the simulation and prints the registry's work.
fn main() {
    let mut registry = Registry::new(HOST_COUNT);

    for now in 0..DURATION {
        for host in 0..HOST_COUNT {
            if sends_heartbeat(host, now) {
                registry.heartbeat(now, host);
            }
        }

        registry.sweep(now);
    }

    println!("heartbeats handled in memory: {}", registry.heartbeats);
    println!("durable writes:               {}", registry.writes.len());

    for w in &registry.writes {
        println!("  second {:2}  host {:2} -> {:?}", w.at, w.host, w.state);
    }
}

Heartbeat updates the host’s last-seen time in memory and writes to the database only when the host was not Up. Sweep runs once a second, computes each host’s state from the length of its silence, and writes only the hosts whose state changed. A heartbeat from a Suspect or Dead host brings it back at once; the sweep is what moves hosts the other way.

Both versions print the same result:

$ go run ./heartbeats
heartbeats handled in memory: 599956
durable writes:               4
  second 22  host 42 -> Suspect
  second 29  host 42 -> Dead
  second 32  host  7 -> Suspect
  second 34  host  7 -> Up

The registry handled almost 600,000 heartbeats and made four writes. Host 42 sent its last heartbeat at second 19; it became Suspect at second 22 and Dead at second 29. Host 7 became Suspect at second 32, three seconds after its last heartbeat at second 29, and came back Up with its heartbeat at second 34, before it reached the Dead threshold.

Figure 3.1 shows the two paths, and the three states with the silences that separate them.

Keeping the state in memory has one consequence. When the registry itself restarts, it has lost every host’s last-seen time. Our model starts by marking every host Up with a last-seen time of 0, which gives each host a full timeout to report in before anything is declared dead. A real registry has to do the same after a restart, or it would declare the whole fleet dead at once.

Choosing the timeout: the phi accrual detector

A fixed timeout treats every host alike. Hosts are not alike: one on a quiet network sends heartbeats exactly a second apart, and another on a congested link sends them anywhere from half a second to two and a half seconds apart. Two seconds of silence is alarming for the first host and ordinary for the second.

The phi accrual failure detector, described by Hayashibara, Défago, Yared, and Katayama in 2004 and used today in Apache Cassandra and Akka, makes the timeout depend on each host’s history. It keeps the most recent gaps between a host’s heartbeats, treats them as samples of a normal distribution, and computes the probability that a heartbeat would still arrive after the current silence. It reports that probability as phi, minus its base-10 logarithm, so that each unit of phi is a factor of ten: phi 1 means a 10 percent chance that the host is fine, phi 2 means 1 percent, phi 3 means 0.1 percent. The control plane picks a phi threshold instead of a timeout, and each host’s effective timeout follows from its own behaviour.

A detector belongs to one host. It remembers a window of gaps and the time of the latest heartbeat.

Listing 3.7 The detector and the recording of a heartbeat, in Go. code/go/phi/main.go

// timeout is the fixed timeout that the program compares against, in seconds.
const timeout = 2.0

// minStdDev is the smallest standard deviation the detector assumes, in
// seconds. Without it, a perfectly regular host would make any delay look
// infinitely suspicious.
const minStdDev = 0.2

// Detector is a phi accrual failure detector for one host.
type Detector struct {
	// intervals holds the most recent gaps between heartbeats, in seconds.
	intervals []float64
	// window is the number of gaps the detector remembers.
	window int
	// last is the time of the most recent heartbeat, in seconds.
	last float64
}

// NewDetector returns a detector that remembers window gaps and has seen a
// first heartbeat at time first.
func NewDetector(window int, first float64) *Detector {
	return &Detector{window: window, last: first}
}

// Heartbeat records a heartbeat that arrived at time now.
func (d *Detector) Heartbeat(now float64) {
	d.intervals = append(d.intervals, now-d.last)

	if len(d.intervals) > d.window {
		d.intervals = d.intervals[1:]
	}

	d.last = now
}

Listing 3.8 The detector and the recording of a heartbeat, in Rust. code/rust/phi/src/main.rs

/// The fixed timeout that the program compares against, in seconds.
const TIMEOUT: f64 = 2.0;

/// The smallest standard deviation the detector assumes, in seconds. Without
/// it, a perfectly regular host would make any delay look infinitely
/// suspicious.
const MIN_STD_DEV: f64 = 0.2;

/// A phi accrual failure detector for one host.
struct Detector {
    /// The most recent gaps between heartbeats, in seconds.
    intervals: VecDeque<f64>,
    /// The number of gaps the detector remembers.
    window: usize,
    /// The time of the most recent heartbeat, in seconds.
    last: f64,
}

impl Detector {
    /// Returns a detector that remembers `window` gaps and has seen a first
    /// heartbeat at time `first`.
    fn new(window: usize, first: f64) -> Self {
        Self {
            intervals: VecDeque::new(),
            window,
            last: first,
        }
    }

    /// Records a heartbeat that arrived at time `now`.
    fn heartbeat(&mut self, now: f64) {
        self.intervals.push_back(now - self.last);

        if self.intervals.len() > self.window {
            self.intervals.pop_front();
        }

        self.last = now;
    }
}

Phi turns the window into a number.

Listing 3.9 Computing phi from the recorded gaps, in Go. code/go/phi/main.go

// Phi returns the suspicion level at time now: minus the base-10 logarithm of
// the probability that a heartbeat would still arrive after this much silence,
// given the gaps seen so far. Phi 1 means a 10% chance the host is fine, phi 2
// means 1%, and phi 3 means 0.1%.
func (d *Detector) Phi(now float64) float64 {
	var mean float64

	for _, gap := range d.intervals {
		mean += gap
	}

	mean /= float64(len(d.intervals))
	var variance float64

	for _, gap := range d.intervals {
		variance += (gap - mean) * (gap - mean)
	}

	stdDev := math.Max(math.Sqrt(variance/float64(len(d.intervals))), minStdDev)

	// The probability that a gap is longer than the current silence, from a
	// logistic approximation of the normal distribution.
	silence := now - d.last
	y := (silence - mean) / stdDev
	e := math.Exp(-y * (1.5976 + 0.070566*y*y))
	pLater := 1 - 1/(1+e)

	if silence > mean {
		pLater = e / (1 + e)
	}

	return -math.Log10(pLater)
}

Listing 3.10 Computing phi from the recorded gaps, in Rust. code/rust/phi/src/main.rs

impl Detector {
    /// Returns the suspicion level at time `now`: minus the base-10 logarithm
    /// of the probability that a heartbeat would still arrive after this much
    /// silence, given the gaps seen so far. Phi 1 means a 10% chance the host
    /// is fine, phi 2 means 1%, and phi 3 means 0.1%.
    #[allow(clippy::cast_precision_loss)] // the window holds at most 100 gaps
    fn phi(&self, now: f64) -> f64 {
        let n = self.intervals.len() as f64;
        let mean = self.intervals.iter().sum::<f64>() / n;
        let variance = self
            .intervals
            .iter()
            .map(|gap| (gap - mean).powi(2))
            .sum::<f64>()
            / n;
        let std_dev = variance.sqrt().max(MIN_STD_DEV);

        // The probability that a gap is longer than the current silence, from
        // a logistic approximation of the normal distribution.
        let silence = now - self.last;
        let y = (silence - mean) / std_dev;
        let e = (-y * (1.5976 + 0.070_566 * y * y)).exp();
        let p_later = if silence > mean {
            e / (1.0 + e)
        } else {
            1.0 - 1.0 / (1.0 + e)
        };

        -p_later.log10()
    }
}

The run trains one detector on a steady host and one on a jittery host, then asks each for phi after silences of increasing length.

Listing 3.11 Two hosts and the table of results, in Go. code/go/phi/main.go

// hosts lists the simulated hosts and the gaps between their heartbeats.
var hosts = []struct {
	name string
	gaps []float64
}{
	{"steady", []float64{1.0, 1.02, 0.98, 1.01, 0.99, 1.0, 1.03, 0.97, 1.0, 1.0}},
	{"jittery", []float64{0.4, 1.9, 0.6, 1.7, 0.5, 1.8, 0.7, 1.6, 0.4, 2.4}},
}

// main feeds each host's heartbeats to a detector and prints phi after
// increasing lengths of silence.
func main() {
	for _, host := range hosts {
		detector := NewDetector(100, 0)
		var now float64

		for _, gap := range host.gaps {
			now += gap
			detector.Heartbeat(now)
		}

		fmt.Printf("%s host, silent for:\n", host.name)

		for _, silence := range []float64{1.0, 1.5, 2.0, 2.5, 3.0, 4.0} {
			verdict := "alive"

			if silence > timeout {
				verdict = "dead"
			}

			fmt.Printf("  %.1fs  phi = %6.2f   %.0fs timeout says %s\n",
				silence, detector.Phi(now+silence), timeout, verdict)
		}
	}
}

Listing 3.12 Two hosts and the table of results, in Rust. code/rust/phi/src/main.rs

/// The simulated hosts and the gaps between their heartbeats.
const HOSTS: [(&str, [f64; 10]); 2] = [
    (
        "steady",
        [1.0, 1.02, 0.98, 1.01, 0.99, 1.0, 1.03, 0.97, 1.0, 1.0],
    ),
    (
        "jittery",
        [0.4, 1.9, 0.6, 1.7, 0.5, 1.8, 0.7, 1.6, 0.4, 2.4],
    ),
];

/// Feeds each host's heartbeats to a detector and prints phi after increasing
/// lengths of silence.
fn main() {
    for (name, gaps) in HOSTS {
        let mut detector = Detector::new(100, 0.0);
        let mut now = 0.0;

        for gap in gaps {
            now += gap;
            detector.heartbeat(now);
        }

        println!("{name} host, silent for:");

        for silence in [1.0, 1.5, 2.0, 2.5, 3.0, 4.0] {
            let verdict = if silence > TIMEOUT { "dead" } else { "alive" };
            println!(
                "  {silence:.1}s  phi = {:6.2}   {TIMEOUT:.0}s timeout says {verdict}",
                detector.phi(now + silence)
            );
        }
    }
}

Phi computes the mean and standard deviation of the remembered gaps and measures how many standard deviations the current silence lies beyond the mean. The logistic expression is a standard approximation of the normal distribution’s tail, the same one Akka uses, and it avoids computing the error function. The standard deviation has a floor of 0.2 seconds. Without it, a host whose heartbeats were perfectly regular would have a standard deviation near zero, and the smallest delay would push phi towards infinity.

Both versions print the same table:

$ go run ./phi
steady host, silent for:
  1.0s  phi =   0.30   2s timeout says alive
  1.5s  phi =   2.22   2s timeout says alive
  2.0s  phi =   7.30   2s timeout says alive
  2.5s  phi =  18.13   2s timeout says dead
  3.0s  phi =  37.58   2s timeout says dead
  4.0s  phi = 113.84   2s timeout says dead
jittery host, silent for:
  1.0s  phi =   0.21   2s timeout says alive
  1.5s  phi =   0.47   2s timeout says alive
  2.0s  phi =   0.88   2s timeout says alive
  2.5s  phi =   1.47   2s timeout says dead
  3.0s  phi =   2.25   2s timeout says dead
  4.0s  phi =   4.58   2s timeout says dead

For the steady host, two seconds of silence already gives phi 7.3: a heartbeat that late is roughly a one-in-twenty-million event, and the host is almost certainly gone, while the 2-second timeout still calls it alive. For the jittery host, two and a half seconds gives phi 1.47, about a 3 percent chance that the host is fine, and the timeout has already declared it dead. With a threshold of 3, the detector would give up on the steady host after about 1.6 seconds and on the jittery host only after about 3.4 seconds.

Leases: how Kubernetes tracks its nodes

Kubernetes applies the same ideas to its own nodes. Every node has a Lease, a small object in the coordination.k8s.io API group, in the namespace kube-node-lease. The kubelet renews its node’s Lease every 10 seconds by writing the current time into the Lease’s renewTime field. Our cluster has one node and one node Lease:

$ kubectl get lease -n kube-node-lease
NAME                    HOLDER                  AGE
gpu-lab-control-plane   gpu-lab-control-plane   4d11h
$ kubectl get lease -n kube-node-lease gpu-lab-control-plane -o yaml
...
spec:
  holderIdentity: gpu-lab-control-plane
  leaseDurationSeconds: 40
  renewTime: "2026-10-01T01:31:36.294522Z"

A Lease has five main fields. holderIdentity names whoever holds it, here the node itself. leaseDurationSeconds says how long the Lease stays valid after its last renewal. renewTime is the time of the last renewal, acquireTime the time the holder first took the Lease, and leaseTransitions counts how often the Lease has changed hands. Kubernetes also uses Leases for leader election, so that only one copy of the controller manager or the scheduler is active at a time, and for each API server to publish its identity.

The kubelet used to send its heartbeats by updating the Node object’s status, which holds the node’s capacity, conditions, images, and addresses. The Kubernetes documentation explains the change: “a Lease is a lightweight resource. Using Leases for heartbeats reduces the performance impact of these updates”. The kubelet still updates the Node’s status, but only when something changes or every five minutes.

On the other side, the node lifecycle controller, part of the controller manager, checks every node every 5 seconds. If it has seen no renewal for longer than its grace period, 50 seconds by default, it sets the node’s Ready condition to Unknown. By default, the pods on that node are evicted 5 minutes later, so that their controllers can recreate them elsewhere. The long wait before eviction reflects the cost of a wrong guess: a node that only lost contact with the control plane may still be running its pods.

A lease agent

We will give each simulated GPU host a Lease of its own, in a namespace called fleet:

Listing 3.13 The namespace for the fleet's Leases. code/manifests/fleet-namespace.yaml

# The namespace that holds one Lease per simulated GPU host in chapter 3.
apiVersion: v1
kind: Namespace
metadata:
  name: fleet

The agent plays the part of the kubelet. It writes the Lease named after its host every two seconds, with a lease duration of six seconds, so a host has to miss three renewals in a row before its Lease expires.

The settings are constants.

Listing 3.14 The agent's settings, in Go. code/go/leaseagent/main.go

// Lease settings. The renewal interval is a third of the lease duration, so a
// host has to miss three renewals in a row before its Lease expires.
const (
	namespace     = "fleet"
	renewEvery    = 2 * time.Second
	leaseDuration = 6 // seconds
	fieldManager  = "leaseagent"
)

Listing 3.15 The agent's settings, in Rust. code/rust/leaseagent/src/main.rs

/// The namespace that holds the hosts' Leases.
const NAMESPACE: &str = "fleet";
/// How often the agent renews the Lease: a third of the lease duration, so a
/// host has to miss three renewals in a row before its Lease expires.
const RENEW_EVERY: Duration = Duration::from_secs(2);
/// How long a Lease stays valid after its last renewal, in seconds.
const LEASE_DURATION: i32 = 6;
/// The field manager name for server-side apply.
const FIELD_MANAGER: &str = "leaseagent";

The Go version builds the Lease inside its loop with client-go’s apply configuration types. The Rust version builds it in a function of its own.

Listing 3.16 The renewal loop, in Go. code/go/leaseagent/main.go

// run writes the host's Lease with server-side apply once per renewEvery.
// A renewal that fails is reported and retried at the next tick.
func run(host string) error {
	clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie())

	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	leases := clientset.CoordinationV1().Leases(namespace)
	acquired := metav1.NewMicroTime(time.Now())
	ticker := time.NewTicker(renewEvery)
	defer ticker.Stop()

	for {
		now := metav1.NewMicroTime(time.Now())
		lease := coordinationv1ac.Lease(host, namespace).
			WithSpec(coordinationv1ac.LeaseSpec().
				WithHolderIdentity(host).
				WithLeaseDurationSeconds(leaseDuration).
				WithAcquireTime(acquired).
				WithRenewTime(now))
		_, err := leases.Apply(ctx, lease, metav1.ApplyOptions{FieldManager: fieldManager, Force: true})

		switch {
		case ctx.Err() != nil:
			return nil
		case err != nil:
			fmt.Printf("%s %s renewal failed: %v\n", now.Format(time.TimeOnly), host, err)
		default:
			fmt.Printf("%s %s renewed\n", now.Format(time.TimeOnly), host)
		}

		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

Listing 3.17 Building the Lease, in Rust. code/rust/leaseagent/src/main.rs

/// Builds the Lease that `host` writes at time `now`, having first acquired
/// it at time `acquired`.
fn lease(host: &str, acquired: Timestamp, now: Timestamp) -> Lease {
    Lease {
        metadata: ObjectMeta {
            name: Some(host.to_string()),
            namespace: Some(NAMESPACE.to_string()),
            ..ObjectMeta::default()
        },
        spec: Some(LeaseSpec {
            holder_identity: Some(host.to_string()),
            lease_duration_seconds: Some(LEASE_DURATION),
            acquire_time: Some(MicroTime(acquired)),
            renew_time: Some(MicroTime(now)),
            ..LeaseSpec::default()
        }),
    }
}

Listing 3.18 The renewal loop, in Rust. code/rust/leaseagent/src/main.rs

/// Reads the host name from the command line and renews its Lease with
/// server-side apply until the process receives Ctrl-C. A renewal that fails
/// is reported and retried at the next tick.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let Some(host) = std::env::args().nth(1) else {
        eprintln!("usage: leaseagent <host>");
        std::process::exit(2);
    };
    let leases: Api<Lease> = Api::namespaced(Client::try_default().await?, NAMESPACE);
    let params = PatchParams::apply(FIELD_MANAGER).force();
    let acquired = Timestamp::now();
    let mut ticker = tokio::time::interval(RENEW_EVERY);

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => return Ok(()),
            _ = ticker.tick() => {}
        }
        let now = Timestamp::now();
        match leases
            .patch(&host, &params, &Patch::Apply(lease(&host, acquired, now)))
            .await
        {
            Ok(_) => println!("{} {host} renewed", time_of_day(now)),
            Err(error) => println!("{} {host} renewal failed: {error}", time_of_day(now)),
        }
    }
}

Both agents write with server-side apply, which we met in section 1.1: each renewal sends the whole Lease, and the API server creates it on the first renewal and updates it on the others. A failed renewal, for example while the API server restarts, is logged and retried at the next tick; the Lease simply ages until a renewal succeeds. The renewal time is written in UTC, as every Kubernetes timestamp is, and it comes from the host’s own clock.

A lease watcher

The watcher plays the part of the node lifecycle controller. It has to decide when a Lease has expired, and the obvious rule, “expired when renewTime plus the lease duration is in the past”, has a flaw: renewTime was written with the host’s clock and is compared with the watcher’s clock. If the two clocks disagree by a few seconds, healthy hosts look expired, or dead hosts look alive for longer than they should.

The node lifecycle controller avoids this by never comparing the two clocks. A comment in its source explains that it records its own time of observation “to avoid the problem with time skew across the cluster”: whenever it sees a Lease’s renewTime change, it notes its own current time, and it declares the node unhealthy when its own clock shows that the grace period has passed since that note. The value of renewTime matters only in that it changed. Our watcher works the same way.

An observation is three fields, and the second is the one that matters: the local time at which the watcher first saw the current renewTime.

Listing 3.19 What the watcher remembers about one Lease, in Go. code/go/leasewatch/main.go

// observation is what the program knows about one host's Lease.
type observation struct {
	// renewTime is the latest renewTime seen in the Lease.
	renewTime time.Time
	// observedAt is the local time at which that renewTime was first seen.
	observedAt time.Time
	// expired records whether the host has been reported as expired.
	expired bool
}

Listing 3.20 What the watcher remembers about one Lease, in Rust. code/rust/leasewatch/src/main.rs

/// What the program knows about one host's Lease.
struct Observation {
    /// The latest `renewTime` seen in the Lease.
    renew_time: Timestamp,
    /// The local time at which that `renewTime` was first seen.
    observed_at: Instant,
    /// Whether the host has been reported as expired.
    expired: bool,
}

check holds the rule from the node lifecycle controller.

Listing 3.21 Checking one Lease against the watcher's own clock, in Go. code/go/leasewatch/main.go

// check updates the observation of one Lease at local time now and prints a
// line when the host is first seen, expires, or renews again after expiring.
func check(seen map[string]*observation, lease *coordinationv1.Lease, now time.Time) {
	if lease.Spec.RenewTime == nil || lease.Spec.LeaseDurationSeconds == nil {
		return
	}

	renew := lease.Spec.RenewTime.Time
	duration := time.Duration(*lease.Spec.LeaseDurationSeconds) * time.Second
	stamp := now.Format(time.TimeOnly)

	obs, known := seen[lease.Name]

	switch {
	case !known:
		seen[lease.Name] = &observation{renewTime: renew, observedAt: now}
		fmt.Printf("%s %s seen, lease duration %s\n", stamp, lease.Name, duration)
	case !renew.Equal(obs.renewTime):
		obs.renewTime, obs.observedAt = renew, now

		if obs.expired {
			obs.expired = false
			fmt.Printf("%s %s renewing again\n", stamp, lease.Name)
		}
	case !obs.expired && now.Sub(obs.observedAt) > duration:
		obs.expired = true
		fmt.Printf("%s %s expired: no renewal seen for %s\n",
			stamp, lease.Name, now.Sub(obs.observedAt).Round(time.Second))
	}
}

Listing 3.22 Checking one Lease against the watcher's own clock, in Rust. code/rust/leasewatch/src/main.rs

/// Updates the observation of one Lease at local time `now` and prints a line
/// when the host is first seen, expires, or renews again after expiring.
fn check(seen: &mut HashMap<String, Observation>, lease: &Lease, now: Instant) {
    let Some(spec) = &lease.spec else { return };
    let (Some(renew), Some(seconds)) = (&spec.renew_time, spec.lease_duration_seconds) else {
        return;
    };
    let duration = Duration::from_secs(u64::from(seconds.unsigned_abs()));
    let stamp = Zoned::now().strftime("%H:%M:%S").to_string();
    let name = lease.name_any();

    match seen.get_mut(&name) {
        None => {
            seen.insert(
                name.clone(),
                Observation {
                    renew_time: renew.0,
                    observed_at: now,
                    expired: false,
                },
            );
            println!(
                "{stamp} {name} seen, lease duration {}s",
                duration.as_secs()
            );
        }
        Some(obs) if obs.renew_time != renew.0 => {
            obs.renew_time = renew.0;
            obs.observed_at = now;

            if obs.expired {
                obs.expired = false;
                println!("{stamp} {name} renewing again");
            }
        }
        Some(obs) if !obs.expired && now.duration_since(obs.observed_at) > duration => {
            obs.expired = true;
            let silent = now.duration_since(obs.observed_at).as_secs_f64().round();
            println!("{stamp} {name} expired: no renewal seen for {silent}s");
        }
        Some(_) => {}
    }
}

The rest of the program keeps a cache of Leases and calls check for each of them once a second.

Listing 3.23 The cache of Leases and the once-a-second loop, in Go. code/go/leasewatch/main.go

// run starts an informer for Leases in the namespace, waits for its first
// list, and then checks every Lease in its cache once per checkEvery.
func run() error {
	clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie())

	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	factory := informers.NewSharedInformerFactoryWithOptions(clientset, 0, informers.WithNamespace(namespace))
	lister := factory.Coordination().V1().Leases().Lister().Leases(namespace)
	factory.Start(ctx.Done())
	factory.WaitForCacheSync(ctx.Done())

	seen := map[string]*observation{}
	ticker := time.NewTicker(checkEvery)
	defer ticker.Stop()

	for {
		leases, err := lister.List(labels.Everything())

		if err != nil {
			return err
		}

		now := time.Now()

		for _, lease := range leases {
			check(seen, lease, now)
		}

		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

Listing 3.24 The cache of Leases and the once-a-second loop, in Rust. code/rust/leasewatch/src/main.rs

/// Starts a watcher that fills a reflector store with the namespace's Leases,
/// waits for the first list, and then checks every Lease in the store once
/// per `CHECK_EVERY` until the process receives Ctrl-C.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let leases: Api<Lease> = Api::namespaced(Client::try_default().await?, NAMESPACE);
    let (store, writer) = reflector::store();
    let stream = watcher(leases, watcher::Config::default())
        .default_backoff()
        .reflect(writer)
        .applied_objects();
    tokio::spawn(stream.for_each(|_| std::future::ready(())));
    store.wait_until_ready().await?;

    let mut seen = HashMap::new();
    let mut ticker = tokio::time::interval(CHECK_EVERY);

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => return Ok(()),
            _ = ticker.tick() => {}
        }
        let now = Instant::now();

        for lease in store.state() {
            check(&mut seen, &lease, now);
        }
    }
}

Neither version sends a request per check. The Go version starts an informer for Leases in the fleet namespace from client-go’s shared informer factory and reads its cache through a lister. The Rust version runs a watcher stream into a reflector store in a background task, with default_backoff so that the stream reconnects after errors, and reads the store’s contents. Both are the list-and-watch machinery from chapter 1, and both read the cache once a second.

check keeps one observation per Lease and moves through three cases. A Lease seen for the first time is recorded with the current local time. A Lease whose renewTime has changed gets a new local observation time, and if the host had been reported as expired, the watcher reports that it is renewing again. A Lease whose renewTime has not changed for longer than its duration, measured on the watcher’s clock, is reported as expired, once. The Rust version measures with Instant, a monotonic clock that is unaffected if the system’s wall clock is adjusted.

Running it

We create the namespace, start three agents, two in Go and one in Rust, and then start both watchers. Each command runs in its own terminal:

$ kubectl apply -f code/manifests/fleet-namespace.yaml
$ go run ./leaseagent gpu-host-a          # in code/go
$ cargo run -p leaseagent -- gpu-host-b   # in code/rust
$ go run ./leaseagent gpu-host-c          # in code/go
$ go run ./leasewatch                     # in code/go
$ cargo run -p leasewatch                 # in code/rust

Figure 3.2 shows the five programs and the Lease objects between them.

The Leases appear in the cluster like any other object:

$ kubectl get lease -n fleet
NAME         HOLDER       AGE
gpu-host-a   gpu-host-a   29s
gpu-host-b   gpu-host-b   29s
gpu-host-c   gpu-host-c   29s
$ kubectl get lease -n fleet gpu-host-a -o yaml
...
spec:
  acquireTime: "2026-10-01T01:31:16.509475Z"
  holderIdentity: gpu-host-a
  leaseDurationSeconds: 6
  renewTime: "2026-10-01T01:31:44.511013Z"

At 18:31:25 we kill the agent for gpu-host-b with kill -9, which stops it without giving it a chance to clean up, as a power failure would. Fourteen seconds later, at 18:31:39, we start it again. The agent’s own log shows the gap:

18:31:16 gpu-host-b renewed
18:31:18 gpu-host-b renewed
18:31:20 gpu-host-b renewed
18:31:22 gpu-host-b renewed
18:31:24 gpu-host-b renewed
18:31:39 gpu-host-b renewed
18:31:41 gpu-host-b renewed
18:31:43 gpu-host-b renewed
18:31:45 gpu-host-b renewed

The Go watcher printed:

18:31:19 gpu-host-c seen, lease duration 6s
18:31:19 gpu-host-a seen, lease duration 6s
18:31:19 gpu-host-b seen, lease duration 6s
18:31:31 gpu-host-b expired: no renewal seen for 7s
18:31:39 gpu-host-b renewing again

And the Rust watcher printed:

18:31:19 gpu-host-c seen, lease duration 6s
18:31:19 gpu-host-a seen, lease duration 6s
18:31:19 gpu-host-b seen, lease duration 6s
18:31:30 gpu-host-b expired: no renewal seen for 6s
18:31:40 gpu-host-b renewing again

The last renewal of gpu-host-b happened at 18:31:24. Each watcher sees a renewal only at its next once-a-second check, so its observation time is up to a second later than the renewal itself, and it reports the expiry at the first check that falls more than six seconds after that observation. The two watchers check at slightly different moments, so one reported the expiry at 18:31:30 and the other at 18:31:31. The same one-second granularity explains the second difference: the restarted agent renewed at once, at 18:31:39, and the two watchers noticed at 18:31:39 and 18:31:40.

Animation 3.1 replays the run for gpu-host-b as the Go watcher saw it. Its last part gives the host a clock that runs eight seconds behind and a watcher that subtracts renewTime from its own clock.

Throughout, the Go watchers read a Lease written by the Rust agent and the Rust watcher read Leases written by the Go agents. The Lease in the API server is the contract between them, and neither side needs to know what language the other is written in.

Declaring a host dead is where a control plane’s harder decisions begin. The host’s jobs should run again elsewhere, but a host that only lost its network connection may still be running them, and a GPU job that writes checkpoints or holds a reservation must not run twice at once. Kubernetes answers part of this with the 5-minute delay before eviction. A control plane for GPU jobs needs a firmer answer as well.

The complete programs

Listing 3.25 The complete program in Go. code/go/heartbeats/main.go

// Command heartbeats models how a control plane tracks which machines in a
// fleet are alive. Every host sends a heartbeat once a second. The registry
// keeps each host's last heartbeat in memory, runs a sweep once a second, and
// records a durable write only when a host changes state: Up, Suspect, or Dead.
//
// The program simulates 10,000 hosts for 60 seconds without real time passing.
// Host 42 crashes at second 20, and host 7 misses four heartbeats from second
// 30 to second 33. It prints how many heartbeats the registry handled, how many
// durable writes it made, and each write.
//
// Run it from code/go with:
//
//	go run ./heartbeats
package main

import "fmt"

// State is what the registry believes about a host.
type State int

// The states a host moves through.
const (
	Up      State = iota // heartbeats are arriving
	Suspect              // heartbeats stopped recently; the host may be slow
	Dead                 // heartbeats stopped long enough to give up on the host
)

// String returns the state's name for the log.
func (s State) String() string {
	return [...]string{"Up", "Suspect", "Dead"}[s]
}

// Thresholds and the size of the simulation, in seconds and hosts.
const (
	suspectAfter = 3      // seconds without a heartbeat before a host is Suspect
	deadAfter    = 10     // seconds without a heartbeat before a host is Dead
	hostCount    = 10_000 // hosts in the fleet
	duration     = 60     // seconds simulated
)

// entry is the registry's in-memory record of one host.
type entry struct {
	// lastSeen is the second of the host's most recent heartbeat.
	lastSeen int
	// state is the host's current state.
	state State
}

// Transition is one durable write: a host entering a new state.
type Transition struct {
	// At is the second at which the sweep or heartbeat noticed the change.
	At int
	// Host is the host's number.
	Host int
	// State is the state the host entered.
	State State
}

// Registry holds the liveness of every host.
type Registry struct {
	// hosts is indexed by host number.
	hosts []entry
	// heartbeats counts the heartbeats handled in memory.
	heartbeats int
	// writes lists the durable writes, in order.
	writes []Transition
}


// NewRegistry returns a registry in which every host is Up, as it would be
// after the hosts registered.
func NewRegistry(n int) *Registry {
	return &Registry{hosts: make([]entry, n)}
}

// Heartbeat records a heartbeat from host at second now. Only a host that was
// not Up costs a durable write.
func (r *Registry) Heartbeat(now, host int) {
	r.heartbeats++
	e := &r.hosts[host]
	e.lastSeen = now

	if e.state != Up {
		e.state = Up
		r.writes = append(r.writes, Transition{now, host, Up})
	}
}

// Sweep re-evaluates every host at second now and records each change of state.
func (r *Registry) Sweep(now int) {
	for host := range r.hosts {
		e := &r.hosts[host]

		if next := stateAfter(now - e.lastSeen); next != e.state {
			e.state = next
			r.writes = append(r.writes, Transition{now, host, next})
		}
	}
}

// stateAfter returns the state of a host that has been silent for the given
// number of seconds.
func stateAfter(silent int) State {
	switch {
	case silent >= deadAfter:
		return Dead
	case silent >= suspectAfter:
		return Suspect
	default:
		return Up
	}
}


// sendsHeartbeat reports whether host sends a heartbeat at second now in the
// simulation.
func sendsHeartbeat(host, now int) bool {
	switch host {
	case 42:
		return now < 20 // crashes at second 20
	case 7:
		return now < 30 || now >= 34 // silent from second 30 to 33
	default:
		return true
	}
}

// main runs the simulation and prints the registry's work.
func main() {
	registry := NewRegistry(hostCount)

	for now := 0; now < duration; now++ {
		for host := 0; host < hostCount; host++ {
			if sendsHeartbeat(host, now) {
				registry.Heartbeat(now, host)
			}
		}

		registry.Sweep(now)
	}

	fmt.Printf("heartbeats handled in memory: %d\n", registry.heartbeats)
	fmt.Printf("durable writes:               %d\n", len(registry.writes))

	for _, w := range registry.writes {
		fmt.Printf("  second %2d  host %2d -> %s\n", w.At, w.Host, w.State)
	}
}

Listing 3.26 The complete program in Rust. code/rust/heartbeats/src/main.rs

//! Models how a control plane tracks which machines in a fleet are alive.
//! Every host sends a heartbeat once a second. The registry keeps each host's
//! last heartbeat in memory, runs a sweep once a second, and records a durable
//! write only when a host changes state: Up, Suspect, or Dead.
//!
//! The program simulates 10,000 hosts for 60 seconds without real time
//! passing. Host 42 crashes at second 20, and host 7 misses four heartbeats
//! from second 30 to second 33. It prints how many heartbeats the registry
//! handled, how many durable writes it made, and each write.
//!
//! Run it from `code/rust` with `cargo run -p heartbeats`.

/// Seconds without a heartbeat before a host is Suspect.
const SUSPECT_AFTER: u32 = 3;
/// Seconds without a heartbeat before a host is Dead.
const DEAD_AFTER: u32 = 10;
/// Hosts in the fleet.
const HOST_COUNT: usize = 10_000;
/// Seconds simulated.
const DURATION: u32 = 60;

/// What the registry believes about a host.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum State {
    /// Heartbeats are arriving.
    #[default]
    Up,
    /// Heartbeats stopped recently; the host may be slow.
    Suspect,
    /// Heartbeats stopped long enough to give up on the host.
    Dead,
}

/// The registry's in-memory record of one host.
#[derive(Clone, Copy, Default)]
struct Entry {
    /// The second of the host's most recent heartbeat.
    last_seen: u32,
    /// The host's current state.
    state: State,
}

/// One durable write: a host entering a new state.
struct Transition {
    /// The second at which the sweep or heartbeat noticed the change.
    at: u32,
    /// The host's number.
    host: usize,
    /// The state the host entered.
    state: State,
}

/// Holds the liveness of every host.
struct Registry {
    /// Indexed by host number.
    hosts: Vec<Entry>,
    /// Counts the heartbeats handled in memory.
    heartbeats: usize,
    /// The durable writes, in order.
    writes: Vec<Transition>,
}

impl Registry {
    /// Returns a registry in which every host is Up, as it would be after the
    /// hosts registered.
    fn new(n: usize) -> Self {
        Self {
            hosts: vec![Entry::default(); n],
            heartbeats: 0,
            writes: Vec::new(),
        }
    }

    /// Records a heartbeat from `host` at second `now`. Only a host that was
    /// not Up costs a durable write.
    fn heartbeat(&mut self, now: u32, host: usize) {
        self.heartbeats += 1;
        let entry = &mut self.hosts[host];
        entry.last_seen = now;

        if entry.state != State::Up {
            entry.state = State::Up;
            self.writes.push(Transition {
                at: now,
                host,
                state: State::Up,
            });
        }
    }

    /// Re-evaluates every host at second `now` and records each change of state.
    fn sweep(&mut self, now: u32) {
        for (host, entry) in self.hosts.iter_mut().enumerate() {
            let next = state_after(now - entry.last_seen);

            if next != entry.state {
                entry.state = next;
                self.writes.push(Transition {
                    at: now,
                    host,
                    state: next,
                });
            }
        }
    }
}

/// Returns the state of a host that has been silent for `silent` seconds.
fn state_after(silent: u32) -> State {
    match silent {
        s if s >= DEAD_AFTER => State::Dead,
        s if s >= SUSPECT_AFTER => State::Suspect,
        _ => State::Up,
    }
}

/// Reports whether `host` sends a heartbeat at second `now` in the simulation.
fn sends_heartbeat(host: usize, now: u32) -> bool {
    match host {
        42 => now < 20,                // crashes at second 20
        7 => !(30..34).contains(&now), // silent from second 30 to 33
        _ => true,
    }
}

/// Runs the simulation and prints the registry's work.
fn main() {
    let mut registry = Registry::new(HOST_COUNT);

    for now in 0..DURATION {
        for host in 0..HOST_COUNT {
            if sends_heartbeat(host, now) {
                registry.heartbeat(now, host);
            }
        }

        registry.sweep(now);
    }

    println!("heartbeats handled in memory: {}", registry.heartbeats);
    println!("durable writes:               {}", registry.writes.len());

    for w in &registry.writes {
        println!("  second {:2}  host {:2} -> {:?}", w.at, w.host, w.state);
    }
}

Listing 3.27 The complete program in Go. code/go/phi/main.go

// Command phi compares two ways of deciding that a host has failed: a fixed
// timeout, and the phi accrual failure detector, which learns how regular each
// host's heartbeats are and reports how unlikely the current silence is.
//
// Two simulated hosts both send a heartbeat about once a second. One is
// steady; the other sits on a congested network and its heartbeats arrive
// between 0.4 and 2.4 seconds apart. For each host, the program prints the
// detector's suspicion level, phi, after different lengths of silence, next to
// the verdict of a 2-second timeout.
//
// Run it from code/go with:
//
//	go run ./phi
package main

import (
	"fmt"
	"math"
)

// timeout is the fixed timeout that the program compares against, in seconds.
const timeout = 2.0

// minStdDev is the smallest standard deviation the detector assumes, in
// seconds. Without it, a perfectly regular host would make any delay look
// infinitely suspicious.
const minStdDev = 0.2

// Detector is a phi accrual failure detector for one host.
type Detector struct {
	// intervals holds the most recent gaps between heartbeats, in seconds.
	intervals []float64
	// window is the number of gaps the detector remembers.
	window int
	// last is the time of the most recent heartbeat, in seconds.
	last float64
}

// NewDetector returns a detector that remembers window gaps and has seen a
// first heartbeat at time first.
func NewDetector(window int, first float64) *Detector {
	return &Detector{window: window, last: first}
}

// Heartbeat records a heartbeat that arrived at time now.
func (d *Detector) Heartbeat(now float64) {
	d.intervals = append(d.intervals, now-d.last)

	if len(d.intervals) > d.window {
		d.intervals = d.intervals[1:]
	}

	d.last = now
}


// Phi returns the suspicion level at time now: minus the base-10 logarithm of
// the probability that a heartbeat would still arrive after this much silence,
// given the gaps seen so far. Phi 1 means a 10% chance the host is fine, phi 2
// means 1%, and phi 3 means 0.1%.
func (d *Detector) Phi(now float64) float64 {
	var mean float64

	for _, gap := range d.intervals {
		mean += gap
	}

	mean /= float64(len(d.intervals))
	var variance float64

	for _, gap := range d.intervals {
		variance += (gap - mean) * (gap - mean)
	}

	stdDev := math.Max(math.Sqrt(variance/float64(len(d.intervals))), minStdDev)

	// The probability that a gap is longer than the current silence, from a
	// logistic approximation of the normal distribution.
	silence := now - d.last
	y := (silence - mean) / stdDev
	e := math.Exp(-y * (1.5976 + 0.070566*y*y))
	pLater := 1 - 1/(1+e)

	if silence > mean {
		pLater = e / (1 + e)
	}

	return -math.Log10(pLater)
}


// hosts lists the simulated hosts and the gaps between their heartbeats.
var hosts = []struct {
	name string
	gaps []float64
}{
	{"steady", []float64{1.0, 1.02, 0.98, 1.01, 0.99, 1.0, 1.03, 0.97, 1.0, 1.0}},
	{"jittery", []float64{0.4, 1.9, 0.6, 1.7, 0.5, 1.8, 0.7, 1.6, 0.4, 2.4}},
}

// main feeds each host's heartbeats to a detector and prints phi after
// increasing lengths of silence.
func main() {
	for _, host := range hosts {
		detector := NewDetector(100, 0)
		var now float64

		for _, gap := range host.gaps {
			now += gap
			detector.Heartbeat(now)
		}

		fmt.Printf("%s host, silent for:\n", host.name)

		for _, silence := range []float64{1.0, 1.5, 2.0, 2.5, 3.0, 4.0} {
			verdict := "alive"

			if silence > timeout {
				verdict = "dead"
			}

			fmt.Printf("  %.1fs  phi = %6.2f   %.0fs timeout says %s\n",
				silence, detector.Phi(now+silence), timeout, verdict)
		}
	}
}

Listing 3.28 The complete program in Rust. code/rust/phi/src/main.rs

//! Compares two ways of deciding that a host has failed: a fixed timeout, and
//! the phi accrual failure detector, which learns how regular each host's
//! heartbeats are and reports how unlikely the current silence is.
//!
//! Two simulated hosts both send a heartbeat about once a second. One is
//! steady; the other sits on a congested network and its heartbeats arrive
//! between 0.4 and 2.4 seconds apart. For each host, the program prints the
//! detector's suspicion level, phi, after different lengths of silence, next
//! to the verdict of a 2-second timeout.
//!
//! Run it from `code/rust` with `cargo run -p phi`.

use std::collections::VecDeque;

/// The fixed timeout that the program compares against, in seconds.
const TIMEOUT: f64 = 2.0;

/// The smallest standard deviation the detector assumes, in seconds. Without
/// it, a perfectly regular host would make any delay look infinitely
/// suspicious.
const MIN_STD_DEV: f64 = 0.2;

/// A phi accrual failure detector for one host.
struct Detector {
    /// The most recent gaps between heartbeats, in seconds.
    intervals: VecDeque<f64>,
    /// The number of gaps the detector remembers.
    window: usize,
    /// The time of the most recent heartbeat, in seconds.
    last: f64,
}

impl Detector {
    /// Returns a detector that remembers `window` gaps and has seen a first
    /// heartbeat at time `first`.
    fn new(window: usize, first: f64) -> Self {
        Self {
            intervals: VecDeque::new(),
            window,
            last: first,
        }
    }

    /// Records a heartbeat that arrived at time `now`.
    fn heartbeat(&mut self, now: f64) {
        self.intervals.push_back(now - self.last);

        if self.intervals.len() > self.window {
            self.intervals.pop_front();
        }

        self.last = now;
    }
}

impl Detector {
    /// Returns the suspicion level at time `now`: minus the base-10 logarithm
    /// of the probability that a heartbeat would still arrive after this much
    /// silence, given the gaps seen so far. Phi 1 means a 10% chance the host
    /// is fine, phi 2 means 1%, and phi 3 means 0.1%.
    #[allow(clippy::cast_precision_loss)] // the window holds at most 100 gaps
    fn phi(&self, now: f64) -> f64 {
        let n = self.intervals.len() as f64;
        let mean = self.intervals.iter().sum::<f64>() / n;
        let variance = self
            .intervals
            .iter()
            .map(|gap| (gap - mean).powi(2))
            .sum::<f64>()
            / n;
        let std_dev = variance.sqrt().max(MIN_STD_DEV);

        // The probability that a gap is longer than the current silence, from
        // a logistic approximation of the normal distribution.
        let silence = now - self.last;
        let y = (silence - mean) / std_dev;
        let e = (-y * (1.5976 + 0.070_566 * y * y)).exp();
        let p_later = if silence > mean {
            e / (1.0 + e)
        } else {
            1.0 - 1.0 / (1.0 + e)
        };

        -p_later.log10()
    }
}

/// The simulated hosts and the gaps between their heartbeats.
const HOSTS: [(&str, [f64; 10]); 2] = [
    (
        "steady",
        [1.0, 1.02, 0.98, 1.01, 0.99, 1.0, 1.03, 0.97, 1.0, 1.0],
    ),
    (
        "jittery",
        [0.4, 1.9, 0.6, 1.7, 0.5, 1.8, 0.7, 1.6, 0.4, 2.4],
    ),
];

/// Feeds each host's heartbeats to a detector and prints phi after increasing
/// lengths of silence.
fn main() {
    for (name, gaps) in HOSTS {
        let mut detector = Detector::new(100, 0.0);
        let mut now = 0.0;

        for gap in gaps {
            now += gap;
            detector.heartbeat(now);
        }

        println!("{name} host, silent for:");

        for silence in [1.0, 1.5, 2.0, 2.5, 3.0, 4.0] {
            let verdict = if silence > TIMEOUT { "dead" } else { "alive" };
            println!(
                "  {silence:.1}s  phi = {:6.2}   {TIMEOUT:.0}s timeout says {verdict}",
                detector.phi(now + silence)
            );
        }
    }
}

Listing 3.29 The complete program in Go. code/go/leaseagent/main.go

// Command leaseagent keeps a Kubernetes Lease alive for one host, the way the
// kubelet keeps its node's Lease in the kube-node-lease namespace alive. Every
// two seconds it writes the Lease named after the host in the fleet namespace,
// with the current time as renewTime and a lease duration of six seconds.
// Other programs judge the host alive for as long as these renewals continue.
//
// Run it from code/go, against the cluster in the current kubeconfig context,
// with:
//
//	go run ./leaseagent gpu-host-a
package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	coordinationv1ac "k8s.io/client-go/applyconfigurations/coordination/v1"
	"k8s.io/client-go/kubernetes"
	ctrl "sigs.k8s.io/controller-runtime"
)

// Lease settings. The renewal interval is a third of the lease duration, so a
// host has to miss three renewals in a row before its Lease expires.
const (
	namespace     = "fleet"
	renewEvery    = 2 * time.Second
	leaseDuration = 6 // seconds
	fieldManager  = "leaseagent"
)


// main reads the host name from the command line and renews its Lease until
// the process receives Ctrl-C or SIGTERM.
func main() {
	if len(os.Args) != 2 {
		fmt.Fprintln(os.Stderr, "usage: leaseagent <host>")
		os.Exit(2)
	}

	if err := run(os.Args[1]); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// run writes the host's Lease with server-side apply once per renewEvery.
// A renewal that fails is reported and retried at the next tick.
func run(host string) error {
	clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie())

	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	leases := clientset.CoordinationV1().Leases(namespace)
	acquired := metav1.NewMicroTime(time.Now())
	ticker := time.NewTicker(renewEvery)
	defer ticker.Stop()

	for {
		now := metav1.NewMicroTime(time.Now())
		lease := coordinationv1ac.Lease(host, namespace).
			WithSpec(coordinationv1ac.LeaseSpec().
				WithHolderIdentity(host).
				WithLeaseDurationSeconds(leaseDuration).
				WithAcquireTime(acquired).
				WithRenewTime(now))
		_, err := leases.Apply(ctx, lease, metav1.ApplyOptions{FieldManager: fieldManager, Force: true})

		switch {
		case ctx.Err() != nil:
			return nil
		case err != nil:
			fmt.Printf("%s %s renewal failed: %v\n", now.Format(time.TimeOnly), host, err)
		default:
			fmt.Printf("%s %s renewed\n", now.Format(time.TimeOnly), host)
		}

		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

Listing 3.30 The complete program in Rust. code/rust/leaseagent/src/main.rs

//! Keeps a Kubernetes Lease alive for one host, the way the kubelet keeps its
//! node's Lease in the `kube-node-lease` namespace alive. Every two seconds it
//! writes the Lease named after the host in the `fleet` namespace, with the
//! current time as `renewTime` and a lease duration of six seconds. Other
//! programs judge the host alive for as long as these renewals continue.
//!
//! Run it from `code/rust`, against the cluster in the current kubeconfig
//! context, with `cargo run -p leaseagent -- gpu-host-a`.

use jiff::Timestamp;
use jiff::tz::TimeZone;
use k8s_openapi::api::coordination::v1::{Lease, LeaseSpec};
use k8s_openapi::apimachinery::pkg::apis::meta::v1::MicroTime;
use kube::api::{ObjectMeta, Patch, PatchParams};
use kube::{Api, Client};
use std::error::Error;
use std::time::Duration;

/// The namespace that holds the hosts' Leases.
const NAMESPACE: &str = "fleet";
/// How often the agent renews the Lease: a third of the lease duration, so a
/// host has to miss three renewals in a row before its Lease expires.
const RENEW_EVERY: Duration = Duration::from_secs(2);
/// How long a Lease stays valid after its last renewal, in seconds.
const LEASE_DURATION: i32 = 6;
/// The field manager name for server-side apply.
const FIELD_MANAGER: &str = "leaseagent";

/// Builds the Lease that `host` writes at time `now`, having first acquired
/// it at time `acquired`.
fn lease(host: &str, acquired: Timestamp, now: Timestamp) -> Lease {
    Lease {
        metadata: ObjectMeta {
            name: Some(host.to_string()),
            namespace: Some(NAMESPACE.to_string()),
            ..ObjectMeta::default()
        },
        spec: Some(LeaseSpec {
            holder_identity: Some(host.to_string()),
            lease_duration_seconds: Some(LEASE_DURATION),
            acquire_time: Some(MicroTime(acquired)),
            renew_time: Some(MicroTime(now)),
            ..LeaseSpec::default()
        }),
    }
}

/// Formats a timestamp as local time of day, for the log.
fn time_of_day(t: Timestamp) -> String {
    t.to_zoned(TimeZone::system())
        .strftime("%H:%M:%S")
        .to_string()
}

/// Reads the host name from the command line and renews its Lease with
/// server-side apply until the process receives Ctrl-C. A renewal that fails
/// is reported and retried at the next tick.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let Some(host) = std::env::args().nth(1) else {
        eprintln!("usage: leaseagent <host>");
        std::process::exit(2);
    };
    let leases: Api<Lease> = Api::namespaced(Client::try_default().await?, NAMESPACE);
    let params = PatchParams::apply(FIELD_MANAGER).force();
    let acquired = Timestamp::now();
    let mut ticker = tokio::time::interval(RENEW_EVERY);

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => return Ok(()),
            _ = ticker.tick() => {}
        }
        let now = Timestamp::now();
        match leases
            .patch(&host, &params, &Patch::Apply(lease(&host, acquired, now)))
            .await
        {
            Ok(_) => println!("{} {host} renewed", time_of_day(now)),
            Err(error) => println!("{} {host} renewal failed: {error}", time_of_day(now)),
        }
    }
}

Listing 3.31 The complete program in Go. code/go/leasewatch/main.go

// Command leasewatch reports when a host in the fleet namespace stops renewing
// its Lease, and when it starts again. It keeps an informer cache of the
// Leases and checks it once a second.
//
// It judges a Lease by its own clock, the way the Kubernetes node lifecycle
// controller judges node Leases: it notes the local time at which it first
// sees each new renewTime, and declares the host expired when no new
// renewTime has appeared for longer than the lease duration. The renewTime
// value itself was written with the host's clock, which may differ from ours.
//
// Run it from code/go, against the cluster in the current kubeconfig context,
// with:
//
//	go run ./leasewatch
package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	coordinationv1 "k8s.io/api/coordination/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	ctrl "sigs.k8s.io/controller-runtime"
)

// namespace is the namespace whose Leases the program watches.
const namespace = "fleet"

// checkEvery is how often the program re-evaluates every Lease.
const checkEvery = time.Second

// observation is what the program knows about one host's Lease.
type observation struct {
	// renewTime is the latest renewTime seen in the Lease.
	renewTime time.Time
	// observedAt is the local time at which that renewTime was first seen.
	observedAt time.Time
	// expired records whether the host has been reported as expired.
	expired bool
}


// main runs the watcher until the process receives Ctrl-C or SIGTERM.
func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// run starts an informer for Leases in the namespace, waits for its first
// list, and then checks every Lease in its cache once per checkEvery.
func run() error {
	clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie())

	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	factory := informers.NewSharedInformerFactoryWithOptions(clientset, 0, informers.WithNamespace(namespace))
	lister := factory.Coordination().V1().Leases().Lister().Leases(namespace)
	factory.Start(ctx.Done())
	factory.WaitForCacheSync(ctx.Done())

	seen := map[string]*observation{}
	ticker := time.NewTicker(checkEvery)
	defer ticker.Stop()

	for {
		leases, err := lister.List(labels.Everything())

		if err != nil {
			return err
		}

		now := time.Now()

		for _, lease := range leases {
			check(seen, lease, now)
		}

		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}


// check updates the observation of one Lease at local time now and prints a
// line when the host is first seen, expires, or renews again after expiring.
func check(seen map[string]*observation, lease *coordinationv1.Lease, now time.Time) {
	if lease.Spec.RenewTime == nil || lease.Spec.LeaseDurationSeconds == nil {
		return
	}

	renew := lease.Spec.RenewTime.Time
	duration := time.Duration(*lease.Spec.LeaseDurationSeconds) * time.Second
	stamp := now.Format(time.TimeOnly)

	obs, known := seen[lease.Name]

	switch {
	case !known:
		seen[lease.Name] = &observation{renewTime: renew, observedAt: now}
		fmt.Printf("%s %s seen, lease duration %s\n", stamp, lease.Name, duration)
	case !renew.Equal(obs.renewTime):
		obs.renewTime, obs.observedAt = renew, now

		if obs.expired {
			obs.expired = false
			fmt.Printf("%s %s renewing again\n", stamp, lease.Name)
		}
	case !obs.expired && now.Sub(obs.observedAt) > duration:
		obs.expired = true
		fmt.Printf("%s %s expired: no renewal seen for %s\n",
			stamp, lease.Name, now.Sub(obs.observedAt).Round(time.Second))
	}
}

Listing 3.32 The complete program in Rust. code/rust/leasewatch/src/main.rs

//! Reports when a host in the `fleet` namespace stops renewing its Lease, and
//! when it starts again. It keeps a reflector store of the Leases and checks
//! it once a second.
//!
//! It judges a Lease by its own clock, the way the Kubernetes node lifecycle
//! controller judges node Leases: it notes the local time at which it first
//! sees each new `renewTime`, and declares the host expired when no new
//! `renewTime` has appeared for longer than the lease duration. The
//! `renewTime` value itself was written with the host's clock, which may
//! differ from ours.
//!
//! Run it from `code/rust`, against the cluster in the current kubeconfig
//! context, with `cargo run -p leasewatch`.

use futures::StreamExt;
use jiff::{Timestamp, Zoned};
use k8s_openapi::api::coordination::v1::Lease;
use kube::runtime::{WatchStreamExt, reflector, watcher};
use kube::{Api, Client, ResourceExt};
use std::collections::HashMap;
use std::error::Error;
use std::time::{Duration, Instant};

/// The namespace whose Leases the program watches.
const NAMESPACE: &str = "fleet";
/// How often the program re-evaluates every Lease.
const CHECK_EVERY: Duration = Duration::from_secs(1);

/// What the program knows about one host's Lease.
struct Observation {
    /// The latest `renewTime` seen in the Lease.
    renew_time: Timestamp,
    /// The local time at which that `renewTime` was first seen.
    observed_at: Instant,
    /// Whether the host has been reported as expired.
    expired: bool,
}

/// Updates the observation of one Lease at local time `now` and prints a line
/// when the host is first seen, expires, or renews again after expiring.
fn check(seen: &mut HashMap<String, Observation>, lease: &Lease, now: Instant) {
    let Some(spec) = &lease.spec else { return };
    let (Some(renew), Some(seconds)) = (&spec.renew_time, spec.lease_duration_seconds) else {
        return;
    };
    let duration = Duration::from_secs(u64::from(seconds.unsigned_abs()));
    let stamp = Zoned::now().strftime("%H:%M:%S").to_string();
    let name = lease.name_any();

    match seen.get_mut(&name) {
        None => {
            seen.insert(
                name.clone(),
                Observation {
                    renew_time: renew.0,
                    observed_at: now,
                    expired: false,
                },
            );
            println!(
                "{stamp} {name} seen, lease duration {}s",
                duration.as_secs()
            );
        }
        Some(obs) if obs.renew_time != renew.0 => {
            obs.renew_time = renew.0;
            obs.observed_at = now;

            if obs.expired {
                obs.expired = false;
                println!("{stamp} {name} renewing again");
            }
        }
        Some(obs) if !obs.expired && now.duration_since(obs.observed_at) > duration => {
            obs.expired = true;
            let silent = now.duration_since(obs.observed_at).as_secs_f64().round();
            println!("{stamp} {name} expired: no renewal seen for {silent}s");
        }
        Some(_) => {}
    }
}

/// Starts a watcher that fills a reflector store with the namespace's Leases,
/// waits for the first list, and then checks every Lease in the store once
/// per `CHECK_EVERY` until the process receives Ctrl-C.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let leases: Api<Lease> = Api::namespaced(Client::try_default().await?, NAMESPACE);
    let (store, writer) = reflector::store();
    let stream = watcher(leases, watcher::Config::default())
        .default_backoff()
        .reflect(writer)
        .applied_objects();
    tokio::spawn(stream.for_each(|_| std::future::ready(())));
    store.wait_until_ready().await?;

    let mut seen = HashMap::new();
    let mut ticker = tokio::time::interval(CHECK_EVERY);

    loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => return Ok(()),
            _ = ticker.tick() => {}
        }
        let now = Instant::now();

        for lease in store.state() {
            check(&mut seen, &lease, now);
        }
    }
}

References

• Kubernetes, Leases: https://kubernetes.io/docs/concepts/architecture/leases/
• Kubernetes, Node status and heartbeats: https://kubernetes.io/docs/reference/node/node-status/#heartbeats
• Kubernetes, Nodes and the node controller: https://kubernetes.io/docs/concepts/architecture/nodes/
• Kubernetes source, node lifecycle controller: https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/nodelifecycle/node_lifecycle_controller.go
• N. Hayashibara, X. Défago, R. Yared, T. Katayama, “The φ Accrual Failure Detector”, 23rd IEEE International Symposium on Reliable Distributed Systems (SRDS), 2004.