1. Keeping a local copy of the cluster

Suppose we want to write a program that runs GPU training jobs on Kubernetes. A user creates a job and says “this job needs one GPU”. Our program has to notice the new job, start its container on a machine with a free GPU, and write down that the job is running. When the user changes the job to ask for two GPUs, our program has to notice that too and replace the container. When the user deletes the job, our program has to stop the container and free the GPU.

In Kubernetes, a program like this is called a controller. In this chapter, we will learn how a controller finds out what is happening in the cluster, which turns out to be the most interesting part of the whole design. We will first build intuition by reasoning about the problems a controller runs into, and quickly prototype the data structures that solve them: a small API server, a cache, a loop that keeps the cache up to date, a work queue, and the controller itself. Then we will run our prototype through a burst of edits, a dropped connection, and a network partition, and finally we will map each piece onto client-go, the library that real Go controllers use.

We will build the model twice, in Go and in Rust, with the same six parts and the same behaviour. The Go files are in code/go/informer, and we run them with go run ./informer from code/go. The Rust files are in code/rust/informer, and we run them with cargo run -p informer from code/rust. Neither version needs anything beyond its language’s standard library.

This chapter covers

objects, and the API server that stores them
resource versions, and the 409 Conflict that protects a write
listing and watching, and the 410 Gone that forces a second list
a cache, a reflector, a work queue, and a controller, each in Go and in Rust
one run through a burst of edits, a dropped watch, and a partition
the parts of client-go that match the model

Objects, specs, and statuses

Kubernetes describes a cluster as a collection of objects. An object is a record with a name and two main parts. The spec holds what a user asked for: “run this image with one GPU”. The status holds what has actually happened so far: “running on one GPU”. A training job in our cluster will be one such object.

All objects live behind a single HTTP service, the API server. Every read and every write of every object goes through it, and it stores the objects in etcd, a replicated key-value database. Users, command-line tools, and controllers are all just clients of the API server.

A controller’s job is to compare each object’s spec with the real world and act on any difference, then record what it did in the status. To do that well, it needs two things from the API server: an accurate view of every object of its kind, and news of every change within a fraction of a second. Let us see how to get both.

Numbering every write

The most obvious approach is to ask the API server for every object every few seconds. It works for a handful of objects, and it falls apart as the cluster grows. A cluster with 50,000 pods stores a few kilobytes of JSON per pod, so a single request for all of them transfers around 200 megabytes. If a dozen controllers each did that every five seconds, the API server would be sending almost half a gigabyte per second, and between two of those requests only a handful of pods would have changed.

What we really want is for the API server to tell us about each change as it happens. That idea brings a problem with it. Connections break: networks fail, servers restart, and proxies close connections they consider idle. When our connection breaks and we open a new one, we need to say “send me everything I missed”, and that requires a way to name the point where we left off.

The API server solves this by numbering its writes. It keeps a single counter for the whole store. Every time any object is created, updated, or deleted, it increments the counter and stamps the new value onto the object that changed. Kubernetes calls this number the object’s resource version. The numbers come straight from etcd, which gives every write a revision number from one counter for the whole database.

The same number helps with a second problem. Imagine two clients read a job at version 7. The first changes the GPU count and writes the job back. The second changes the image and writes the job back a moment later, still based on what it read at version 7. Without any check, the second write would replace the whole object and silently undo the first client’s change. With resource versions, the second client sends “I read this at version 7” along with its write. The object is now at version 8, so the server refuses the write with the HTTP status 409 Conflict, and the second client knows it has to read the job again and redo its change on top of the new version.

Listing, then watching

With numbered writes, a client can build an exact view of the cluster in two steps.

First it asks for every object. This request is called a list, and the response carries the objects together with the resource version at which the server took the snapshot. Then the client opens a watch: a request that stays open and asks the server to send every change made after a given version, one event per change. The client passes the version it got from the list.

Passing that version is the important detail. The list and the watch are two separate requests, and other clients keep writing in between. Suppose our list is taken at version 10, then a user edits a job at version 11, and only then does our watch open, when the store is at version 12:

version:   10          11             12
           list        job-a edited   watch opened
           snapshot                   from version 10
                                      -> the server replays event 11,
                                         then sends new events as they occur

Because the watch asks for everything after version 10, the server first sends us event 11 from its record of recent changes and then carries on with new events. Our view is the snapshot at version 10 plus every change after it: nothing missing, nothing counted twice. If we had opened the watch at the current version, 12, we would have skipped the edit at version 11 and kept an out-of-date copy of job-a until the job happened to change again.

When the history runs out

The server can only replay changes it still has. etcd keeps every revision of every key until it is told to throw old ones away, and that clean-up is called compaction. Without it the database would grow forever. The Kubernetes API server asks etcd to compact every five minutes by default. After a compaction, a watch that asks to resume from an older version cannot be served, because the events it needs have been thrown away. The API server answers with the HTTP status 410 Gone, and the only way for us to get back to an exact view is to list again and watch from the new list’s version.

Watches also end for ordinary reasons. The API server closes every watch after a timeout, chosen at random between 30 and 60 minutes by default, so that thousands of clients do not all reconnect at the same moment. A load balancer between us and the server may close a connection that has been quiet for too long. And the network may fail for a while, so that even new connections fail.

Each of these calls for a different response, and our client has to tell them apart. If the watch simply closed, we reopen it from the last version we received. If we cannot connect at all, we wait and try again. If the server says 410 Gone, we list again.

Building the server

Now we have enough understanding to build a model. We start with the server. It keeps the objects in memory, numbers every write, records recent events, and answers lists and watches. We also give it three extra methods, DropWatches, Partition, and Heal, so that our demonstration at the end of the chapter can cause each of the failures we just discussed.

Low-level design: the model

Part	Holds	Rule it keeps
`Object`	a name, a spec, a status, a version	the version is the number of the write that last changed it
`APIServer`	the objects, the write counter, recent events, the open watches	every write increments the counter and publishes one event
`Cache`	a copy of every object	exact up to the last event applied
`Reflector`	the server, the cache, the queue	updates the cache before it queues a name
`WorkQueue`	the ready names, and the sets `dirty` and `processing`	a name is with at most one worker at a time
`Controller`	the server, the cache, the queue	acts only on the object’s current state

Each part is one source file in each language, and the sections below build them in this order.

The server’s data comes first: the object, the event that reports a change to it, and the two errors a client has to tell apart.

Listing 1.1 Objects, events, and the two errors, in Go. code/go/informer/server.go

// Spec is the part of an object that a user writes: the desired state.
type Spec struct {
	// GPUs is the number of GPUs the job asks for.
	GPUs int
}

// Status is the part of an object that the controller writes: the observed state.
type Status struct {
	// GPUs is the number of GPUs the controller has provided to the job.
	GPUs int
}

// Object is one stored record, such as a training job.
type Object struct {
	// Name identifies the object on the server.
	Name string
	// Spec holds what the user asked for.
	Spec Spec
	// Status holds what the controller has done about it.
	Status Status
	// ResourceVersion is the server's write counter at the object's latest change.
	ResourceVersion int
}

// String formats an object for the demonstration's log lines.
func (o Object) String() string {
	return fmt.Sprintf("%s{spec.GPUs: %d, status.GPUs: %d, v%d}",
		o.Name, o.Spec.GPUs, o.Status.GPUs, o.ResourceVersion)
}

// EventType is the kind of change that an Event records.
type EventType int

// The kinds of change that a watch reports.
const (
	Added    EventType = iota // the object was created
	Modified                  // the object was updated
	Deleted                   // the object was removed
)

// String returns the name that the Kubernetes watch protocol uses for t.
func (t EventType) String() string {
	return [...]string{"ADDED", "MODIFIED", "DELETED"}[t]
}

// Event records one change to one object.
type Event struct {
	// Type is the kind of change.
	Type EventType
	// Object is the object after the change, or its final state if it was deleted.
	Object Object
}

// Errors that the server returns. The first two carry the HTTP status codes
// that the Kubernetes API server uses for the same conditions.
var (
	// ErrGone reports that the history a watch needs has been discarded.
	ErrGone = errors.New("410 Gone: resource version too old")
	// ErrConflict reports a conditional write whose resource version is out of date.
	ErrConflict = errors.New("409 Conflict: object has been modified")
	// ErrUnreachable reports that the client cannot reach the server.
	ErrUnreachable = errors.New("server unreachable")
)

Listing 1.2 Objects, events, and the two errors, in Rust. code/rust/informer/src/server.rs

/// The part of an object that a user writes: the desired state.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Spec {
    /// Number of GPUs the job asks for.
    pub gpus: u32,
}

/// The part of an object that the controller writes: the observed state.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Status {
    /// Number of GPUs the controller has provided to the job.
    pub gpus: u32,
}

/// One stored record, such as a training job.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Object {
    /// Identifies the object on the server.
    pub name: String,
    /// What the user asked for.
    pub spec: Spec,
    /// What the controller has done about it.
    pub status: Status,
    /// The server's write counter at the object's latest change.
    pub resource_version: u64,
}

impl Object {
    /// Returns a new object with the given name and GPU request, an empty
    /// status, and no resource version.
    pub fn new(name: &str, gpus: u32) -> Self {
        Self {
            name: name.to_string(),
            spec: Spec { gpus },
            ..Self::default()
        }
    }
}

impl fmt::Display for Object {
    /// Formats an object for the demonstration's log lines.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{}{{spec.gpus: {}, status.gpus: {}, v{}}}",
            self.name, self.spec.gpus, self.status.gpus, self.resource_version
        )
    }
}

/// The kind of change that an [`Event`] records.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum EventType {
    /// The object was created.
    Added,
    /// The object was updated.
    Modified,
    /// The object was removed.
    Deleted,
}

/// One change to one object.
#[derive(Clone, Debug)]
pub struct Event {
    /// The kind of change.
    pub kind: EventType,
    /// The object after the change, or its final state if it was deleted.
    pub object: Object,
}

/// Errors that the server returns. The first two carry the HTTP status codes
/// that the Kubernetes API server uses for the same conditions.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ApiError {
    /// The history that a watch needs has been discarded.
    Gone,
    /// A conditional write carried an out-of-date resource version.
    Conflict,
    /// The client cannot reach the server.
    Unreachable,
}

impl fmt::Display for ApiError {
    /// Formats the error the way the Kubernetes API server reports it.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Gone => "410 Gone: resource version too old",
            Self::Conflict => "409 Conflict: object has been modified",
            Self::Unreachable => "server unreachable",
        })
    }
}

The model reduces a job to a GPU count in its spec and in its status.

The server itself is a lock around four things: the objects, the counter, the events it still has, and the watches that are open.

Listing 1.3 The server's state, in Go. code/go/informer/server.go

// APIServer stores objects and streams their changes to watchers. Its methods
// are safe for concurrent use.
type APIServer struct {
	// mu guards every field below.
	mu sync.Mutex
	// version is the last resource version the server handed out.
	version int
	// objects holds the current state of every object, keyed by name.
	objects map[string]Object
	// history holds the events newer than compacted, oldest first.
	history []Event
	// compacted is the oldest version from which a watch may resume.
	compacted int
	// watches holds one channel for each open watch.
	watches []chan Event
	// unreachable makes new watches fail, to simulate a network partition.
	unreachable bool
}

// NewAPIServer returns an empty server.
func NewAPIServer() *APIServer {
	return &APIServer{objects: map[string]Object{}}
}

Listing 1.4 The server's state, in Rust. code/rust/informer/src/server.rs

/// Everything the server stores, kept behind one lock.
#[derive(Default)]
struct State {
    /// The last resource version the server handed out.
    version: u64,
    /// The current state of every object, keyed by name.
    objects: HashMap<String, Object>,
    /// The events newer than `compacted`, oldest first.
    history: Vec<Event>,
    /// The oldest version from which a watch may resume.
    compacted: u64,
    /// One sender for each open watch.
    watches: Vec<Sender<Event>>,
    /// Makes new watches fail, to simulate a network partition.
    unreachable: bool,
}

impl State {
    /// Appends `event` to the history and sends it to every open watch.
    /// Watches whose receiver has gone away are forgotten.
    fn publish(&mut self, event: Event) {
        self.watches.retain(|w| w.send(event.clone()).is_ok());
        self.history.push(event);
    }
}

/// Stores objects and streams their changes to watchers. Its methods are safe
/// to call from several threads.
#[derive(Default)]
pub struct ApiServer {
    /// Guards all of the server's data.
    state: Mutex<State>,
}

impl ApiServer {
    /// Locks the server's data.
    fn state(&self) -> std::sync::MutexGuard<'_, State> {
        self.state.lock().expect("server lock poisoned")
    }
}

Two methods change objects, and both end by publishing an event.

Listing 1.5 Writing: update, delete, and publish, in Go. code/go/informer/server.go

// Update creates or replaces an object and returns ErrConflict if the write's
// condition fails. A non-zero o.ResourceVersion is a condition: the caller read
// the object at that version and wants the write applied only if the object is
// still at it. A zero ResourceVersion writes unconditionally.
func (s *APIServer) Update(o Object) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current, exists := s.objects[o.Name]

	if o.ResourceVersion != 0 && o.ResourceVersion != current.ResourceVersion {
		return ErrConflict
	}

	eventType := Modified

	if !exists {
		eventType = Added
	}

	s.version++
	o.ResourceVersion = s.version
	s.objects[o.Name] = o
	s.publish(Event{eventType, o})

	return nil
}

// Delete removes the named object, if it exists.
func (s *APIServer) Delete(name string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	o, exists := s.objects[name]

	if !exists {
		return
	}

	delete(s.objects, name)
	s.version++
	o.ResourceVersion = s.version
	s.publish(Event{Deleted, o})
}

// publish appends e to the history and sends it to every open watch.
// The caller must hold s.mu.
func (s *APIServer) publish(e Event) {
	s.history = append(s.history, e)

	for _, w := range s.watches {
		w <- e
	}
}

Listing 1.6 Writing: update, delete, and publish, in Rust. code/rust/informer/src/server.rs

impl ApiServer {
    /// Creates or replaces an object. A non-zero `resource_version` is a
    /// condition: the caller read the object at that version and wants the
    /// write applied only if the object is still at it. Zero writes
    /// unconditionally.
    pub fn update(&self, mut object: Object) -> Result<(), ApiError> {
        let mut state = self.state();
        let current = state.objects.get(&object.name);
        let current_version = current.map_or(0, |o| o.resource_version);

        if object.resource_version != 0 && object.resource_version != current_version {
            return Err(ApiError::Conflict);
        }

        let kind = if current.is_some() {
            EventType::Modified
        } else {
            EventType::Added
        };
        state.version += 1;
        object.resource_version = state.version;
        state.objects.insert(object.name.clone(), object.clone());
        state.publish(Event { kind, object });

        Ok(())
    }

    /// Removes the named object, if it exists.
    pub fn delete(&self, name: &str) {
        let mut state = self.state();
        let Some(mut object) = state.objects.remove(name) else {
            return;
        };
        state.version += 1;
        object.resource_version = state.version;
        state.publish(Event {
            kind: EventType::Deleted,
            object,
        });
    }
}

Update carries both uses of the resource version. A zero version skips the check, which the demonstration uses to play a user editing a job. Any other version must equal the stored one, and a missing object counts as version 0. Each successful write increments the counter and publishes one event. Delete publishes the object’s final state under a new version.

Two methods read: one for the snapshot and one for the stream of changes after it.

Listing 1.7 Reading: list and watch, in Go. code/go/informer/server.go

// List returns every object, sorted by name, and the version at which the
// snapshot was taken.
func (s *APIServer) List() ([]Object, int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	byName := func(a, b Object) int { return strings.Compare(a.Name, b.Name) }

	return slices.SortedFunc(maps.Values(s.objects), byName), s.version
}

// Watch returns a channel that first delivers the stored events newer than
// version and then every new event, until the server closes it.
func (s *APIServer) Watch(version int) (<-chan Event, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	switch {
	case s.unreachable:
		return nil, ErrUnreachable
	case version < s.compacted:
		return nil, ErrGone
	}

	w := make(chan Event, 1024)

	for _, e := range s.history {
		if e.Object.ResourceVersion > version {
			w <- e
		}
	}

	s.watches = append(s.watches, w)

	return w, nil
}

Listing 1.8 Reading: list and watch, in Rust. code/rust/informer/src/server.rs

impl ApiServer {
    /// Returns every object, sorted by name, and the version at which the
    /// snapshot was taken.
    pub fn list(&self) -> (Vec<Object>, u64) {
        let state = self.state();
        let mut objects: Vec<Object> = state.objects.values().cloned().collect();
        objects.sort_by(|a, b| a.name.cmp(&b.name));

        (objects, state.version)
    }

    /// Returns a receiver that first delivers the stored events newer than
    /// `version` and then every new event, until the server drops the watch.
    pub fn watch(&self, version: u64) -> Result<Receiver<Event>, ApiError> {
        let mut state = self.state();

        if state.unreachable {
            return Err(ApiError::Unreachable);
        }

        if version < state.compacted {
            return Err(ApiError::Gone);
        }

        let (sender, receiver) = channel();

        for event in &state.history {
            if event.object.resource_version > version {
                sender.send(event.clone()).expect("receiver is held here");
            }
        }

        state.watches.push(sender);

        Ok(receiver)
    }
}

publish runs while the server lock is held, and Watch holds the same lock while it replays the history and registers the new watch. Every event is therefore either in the replayed history or delivered after registration: no watch misses an event or receives one twice.

A Go watch buffers 1,024 events. A client that stops reading would eventually block publish while it holds the lock and stall every writer; the real API server closes such a watch. The Rust channel is unbounded. In Go the server ends a watch by closing its channel; in Rust it drops the sender, which ends the receiver’s loop.

The last four methods cause the failures of the previous section on request.

Listing 1.9 Compaction, dropped watches, and partitions, in Go. code/go/informer/server.go

// Compact discards the event history, as etcd discards old versions.
func (s *APIServer) Compact() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.compacted = s.version
	s.history = nil
}

// DropWatches closes every open watch, as a load balancer's idle timeout would.
func (s *APIServer) DropWatches() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.dropWatches()
}

// Partition closes every open watch and makes new watches fail until Heal.
func (s *APIServer) Partition() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.unreachable = true
	s.dropWatches()
}

// Heal ends a partition, so new watches succeed again.
func (s *APIServer) Heal() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.unreachable = false
}

// dropWatches closes and forgets every watch channel. The caller must hold s.mu.
func (s *APIServer) dropWatches() {
	for _, w := range s.watches {
		close(w)
	}

	s.watches = nil
}

Listing 1.10 Compaction, dropped watches, and partitions, in Rust. code/rust/informer/src/server.rs

impl ApiServer {
    /// Discards the event history, as etcd discards old versions.
    pub fn compact(&self) {
        let mut state = self.state();
        state.compacted = state.version;
        state.history.clear();
    }

    /// Closes every open watch, as a load balancer's idle timeout would.
    pub fn drop_watches(&self) {
        self.state().watches.clear();
    }

    /// Closes every open watch and makes new watches fail until [`Self::heal`].
    pub fn partition(&self) {
        let mut state = self.state();
        state.unreachable = true;
        state.watches.clear();
    }

    /// Ends a partition, so new watches succeed again.
    pub fn heal(&self) {
        self.state().unreachable = false;
    }
}

Watch fails with the gone error when the requested version is older than compacted. Compact, DropWatches, Partition, and Heal exist only to reproduce the failures described above.

A local copy of the objects

Once we have listed and are watching, we can keep our own copy of every object in memory. The copy is exact up to the last event we received. Reading an object becomes a map lookup instead of a network request, and the server only has to send each change to us once. We call this copy the cache.

The cache is a map behind a lock, with a method for one event.

Listing 1.11 The cache, a lookup, and applying one event, in Go. code/go/informer/cache.go

// Cache is the client's copy of the server's objects, kept current by a
// Reflector. Its methods are safe for concurrent use.
type Cache struct {
	// mu guards objects.
	mu sync.RWMutex
	// objects holds the latest known state of every object, keyed by name.
	objects map[string]Object
}

// NewCache returns an empty cache.
func NewCache() *Cache {
	return &Cache{objects: map[string]Object{}}
}

// Get returns the cached copy of the named object and whether it exists.
func (c *Cache) Get(name string) (Object, bool) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	o, ok := c.objects[name]

	return o, ok
}

// Apply records one event from a watch.
func (c *Cache) Apply(e Event) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if e.Type == Deleted {
		delete(c.objects, e.Object.Name)

		return
	}

	c.objects[e.Object.Name] = e.Object
}

Listing 1.12 The cache, a lookup, and applying one event, in Rust. code/rust/informer/src/cache.rs

/// The client's copy of the server's objects, kept current by a
/// [`Reflector`](crate::reflector::Reflector). Its methods are safe to call
/// from several threads.
#[derive(Default)]
pub struct Cache {
    /// The latest known state of every object, keyed by name.
    objects: RwLock<HashMap<String, Object>>,
}

impl Cache {
    /// Returns the cached copy of the named object, if it exists.
    pub fn get(&self, name: &str) -> Option<Object> {
        let objects = self.objects.read().expect("cache lock poisoned");

        objects.get(name).cloned()
    }

    /// Records one event from a watch.
    pub fn apply(&self, event: Event) {
        let mut objects = self.objects.write().expect("cache lock poisoned");

        match event.kind {
            EventType::Deleted => objects.remove(&event.object.name),
            EventType::Added | EventType::Modified => {
                objects.insert(event.object.name.clone(), event.object)
            }
        };
    }
}

Both caches use a read-write lock: every reconcile reads the cache, and only incoming events write to it.

Apply records a single event. It stores the object for an addition or a modification and removes it for a deletion.

A second method installs a whole list.

Listing 1.13 Replacing the cache with a fresh list, in Go. code/go/informer/cache.go

// Replace installs the result of a fresh list and returns the names of the
// objects that changed, including objects that disappeared while the client
// was not watching.
func (c *Cache) Replace(list []Object) []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	old := c.objects
	c.objects = make(map[string]Object, len(list))
	var changed []string

	for _, o := range list {
		c.objects[o.Name] = o
		// A name missing from old reads as version 0, so it counts as changed.
		if old[o.Name].ResourceVersion != o.ResourceVersion {
			changed = append(changed, o.Name)
		}
	}

	for name := range old {
		if _, ok := c.objects[name]; !ok {
			changed = append(changed, name)
		}
	}

	return changed
}

Listing 1.14 Replacing the cache with a fresh list, in Rust. code/rust/informer/src/cache.rs

impl Cache {
    /// Installs the result of a fresh list and returns the names of the
    /// objects that changed, including objects that disappeared while the
    /// client was not watching.
    pub fn replace(&self, list: &[Object]) -> Vec<String> {
        let mut objects = self.objects.write().expect("cache lock poisoned");
        let old = std::mem::take(&mut *objects);
        let mut changed = Vec::new();

        for object in list {
            let old_version = old.get(&object.name).map(|o| o.resource_version);

            if old_version != Some(object.resource_version) {
                changed.push(object.name.clone());
            }

            objects.insert(object.name.clone(), object.clone());
        }

        for name in old.keys() {
            if !objects.contains_key(name) {
                changed.push(name.clone());
            }
        }

        changed
    }
}

Replace installs the result of a fresh list, and it has one more job that events alone cannot do. Think about what happens when we relist after 410 Gone. Some objects may have been deleted while we were disconnected, and their deletion events were thrown away by the compaction. Nobody will ever send us those events. The only evidence that the objects are gone is that they are in our old cache and missing from the new list. So Replace builds the new contents from the list, reports every object whose version differs from the one we had, and then reports every name that was in the old cache but is missing from the list. Objects that are new to us have no cached version at all; Go reads the missing entry as version 0 and Rust compares against None, so new objects get reported as changed too.

The reflector

Next we need a loop that keeps the cache in step with the server. It lists, installs the result with Replace, and queues the names of the objects that Replace reported, so that the controller will look at them. Then it watches from the list’s version and applies each event as it arrives. When the history it needs is gone, it lists again. We call this loop the reflector.

The outer loop lists, installs the list, and hands over to the watch.

Listing 1.15 The reflector and its list step, in Go. code/go/informer/reflector.go

// Reflector copies the server's objects into a cache and queues the name of
// every object that changes.
type Reflector struct {
	// server is the source of objects and events.
	server *APIServer
	// cache receives every object the reflector sees.
	cache *Cache
	// queue receives the name of every object that changes.
	queue *WorkQueue
}

// Run lists every object, watches from the list's version, and lists again
// whenever the server can no longer resume the watch. It does not return.
func (r *Reflector) Run() {
	for {
		list, version := r.server.List()
		r.queue.Add(r.cache.Replace(list)...)
		logf("reflector: list at version %d: %v", version, list)
		r.watchFrom(version)
	}
}

Listing 1.16 The reflector and its list step, in Rust. code/rust/informer/src/reflector.rs

/// Copies the server's objects into a cache and queues the name of every
/// object that changes.
pub struct Reflector {
    /// The source of objects and events.
    pub server: Arc<ApiServer>,
    /// Receives every object the reflector sees.
    pub cache: Arc<Cache>,
    /// Receives the name of every object that changes.
    pub queue: Arc<WorkQueue>,
}

impl Reflector {
    /// Lists every object, watches from the list's version, and lists again
    /// whenever the server can no longer resume the watch. It does not return.
    pub fn run(&self) {
        loop {
            let (list, version) = self.server.list();
            self.queue.add(self.cache.replace(&list));
            let objects: Vec<String> = list.iter().map(ToString::to_string).collect();
            log!(
                "reflector: list at version {version}: [{}]",
                objects.join(" ")
            );
            self.watch_from(version);
        }
    }
}

The inner loop watches, and decides what each kind of ending calls for.

Listing 1.17 Watching from a version, in Go. code/go/informer/reflector.go

// watchFrom applies events from version onward, reopens the watch after each
// disconnect, and returns when the server reports that the history is gone.
func (r *Reflector) watchFrom(version int) {
	for {
		w, err := r.server.Watch(version)

		if errors.Is(err, ErrGone) {
			logf("reflector: watch from version %d: %v", version, err)

			return
		}

		if err != nil {
			time.Sleep(10 * time.Millisecond) // client-go backs off exponentially
			continue
		}

		for e := range w {
			// Update the cache before queueing the name, so that a worker
			// that takes the name reads the object as of this event.
			r.cache.Apply(e)
			r.queue.Add(e.Object.Name)
			version = e.Object.ResourceVersion
		}

		logf("reflector: watch closed, resuming from version %d", version)
	}
}

Listing 1.18 Watching from a version, in Rust. code/rust/informer/src/reflector.rs

impl Reflector {
    /// Applies events from `version` onward, reopens the watch after each
    /// disconnect, and returns when the server reports that the history is gone.
    fn watch_from(&self, mut version: u64) {
        loop {
            let events = match self.server.watch(version) {
                Ok(events) => events,
                Err(ApiError::Gone) => {
                    log!(
                        "reflector: watch from version {version}: {}",
                        ApiError::Gone
                    );

                    return;
                }
                Err(_) => {
                    thread::sleep(Duration::from_millis(10)); // client-go backs off exponentially
                    continue;
                }
            };

            for event in events {
                version = event.object.resource_version;
                let name = event.object.name.clone();
                // Update the cache before queueing the name, so that a worker
                // that takes the name reads the object as of this event.
                self.cache.apply(event);
                self.queue.add([name]);
            }

            log!("reflector: watch closed, resuming from version {version}");
        }
    }
}

The heart of the reflector is watchFrom (watch_from in Rust). It keeps a variable, version, that always holds the version of the last event it applied to the cache. That means our cache is complete up to version, so a new watch opened from version picks up exactly where the old one stopped. With that in mind, here is what the function does for each of the three ways a watch can end:

If the watch channel closes because the server dropped the connection, we log the version and open a new watch from it. The server replays anything that happened while we were reconnecting.
If Watch fails because the server is unreachable, we wait 10 milliseconds and try again with the same version. client-go waits a little longer after each failure in a row, so that thousands of clients do not hammer a server that is trying to recover.
If Watch fails with the gone error, our version is older than the compaction point, and no watch can continue from it. watchFrom returns, and Run starts over with a new list.

Look closely at the order of the two calls inside the event loop: we update the cache first, and only then queue the object’s name. Suppose we did it the other way round. A worker could take the name straight away, read the object from the cache before our update landed, see the old state, decide there was nothing to do, and finish. The name would not be queued again, and the change would sit unhandled until the object happened to change once more. Updating the cache first rules that out.

client-go uses the same name for this loop, reflector, because it reflects the server’s state into local memory. A reflector together with its cache, plus the plumbing that tells other code about changes, is what client-go calls an informer.

A queue of names

The reflector knows when an object changes, and the controller needs to hear about it. The obvious design is for the reflector to call the controller once for every event. Let us see why that design breaks down.

The first problem is bursts. A user who edits a job five times in one second produces five events. A controller called five times would start work for four specs that were already out of date before the work finished. The second problem is lost events. After a relist, Replace tells us which names changed, but the individual events that led there are gone for good. A controller that relies on seeing every event would have nothing to react to, and it would miss changes every time the history was compacted.

So we give the controller only the names of objects that changed, and the controller reads the object’s current state from the cache when it gets to the name. Five edits now put the same name in the queue five times, and if the queue simply ignores a name it already holds, they turn into one reconcile that acts on the latest spec. The controller takes names from the queue in one or more workers, each processing one name at a time. Our model runs a single worker.

The queue’s state is a list and two sets behind one lock.

Listing 1.19 The queue's state, in Go. code/go/informer/queue.go

// WorkQueue stores the names of objects that need reconciling. A name waits
// in the queue at most once, and a name that a worker holds is handed to no
// other worker until the first calls Done.
type WorkQueue struct {
	// mu guards every field below.
	mu sync.Mutex
	// cond wakes workers blocked in Get.
	cond *sync.Cond
	// queue holds the names ready for a worker, in arrival order.
	queue []string
	// dirty holds every name that needs processing.
	dirty map[string]bool
	// processing holds every name that a worker currently holds.
	processing map[string]bool
}

// NewWorkQueue returns an empty queue.
func NewWorkQueue() *WorkQueue {
	q := &WorkQueue{dirty: map[string]bool{}, processing: map[string]bool{}}
	q.cond = sync.NewCond(&q.mu)

	return q
}

Listing 1.20 The queue's state, in Rust. code/rust/informer/src/queue.rs

/// The queue's data, kept behind one lock.
#[derive(Default)]
struct State {
    /// The names ready for a worker, in arrival order.
    ready: VecDeque<String>,
    /// Every name that needs processing.
    dirty: HashSet<String>,
    /// Every name that a worker currently holds.
    processing: HashSet<String>,
}

/// Stores the names of objects that need reconciling. A name waits in the
/// queue at most once, and a name that a worker holds is handed to no other
/// worker until the first calls [`WorkQueue::done`].
#[derive(Default)]
pub struct WorkQueue {
    /// Guards the queue's data.
    state: Mutex<State>,
    /// Wakes workers blocked in [`WorkQueue::get`].
    ready: Condvar,
}

impl WorkQueue {
    /// Locks the queue's data.
    fn state(&self) -> MutexGuard<'_, State> {
        self.state.lock().expect("queue lock poisoned")
    }
}

Besides the list of names that are ready for a worker, the queue keeps two sets. dirty holds every name that needs processing. processing holds every name that a worker has taken and not yet handed back.

Three operations work on that state.

Listing 1.21 Add, Get, and Done, in Go. code/go/informer/queue.go

// Add marks each name as needing processing. A name that is already pending
// is left where it is. A name that a worker holds is queued again by Done.
func (q *WorkQueue) Add(names ...string) {
	q.mu.Lock()
	defer q.mu.Unlock()

	for _, name := range names {
		if q.dirty[name] {
			continue
		}

		q.dirty[name] = true

		if !q.processing[name] {
			q.queue = append(q.queue, name)
			q.cond.Signal()
		}
	}
}

// Get blocks until a name is ready and hands it to the caller.
func (q *WorkQueue) Get() string {
	q.mu.Lock()
	defer q.mu.Unlock()

	for len(q.queue) == 0 {
		q.cond.Wait()
	}

	name := q.queue[0]
	q.queue = q.queue[1:]
	delete(q.dirty, name)
	q.processing[name] = true

	return name
}

// Done releases a name that Get handed out, and queues it again if it was
// added while the worker held it.
func (q *WorkQueue) Done(name string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	delete(q.processing, name)

	if q.dirty[name] {
		q.queue = append(q.queue, name)
		q.cond.Signal()
	}
}

Listing 1.22 Add, Get, and Done, in Rust. code/rust/informer/src/queue.rs

impl WorkQueue {
    /// Marks each name as needing processing. A name that is already pending
    /// is left where it is. A name that a worker holds is queued again by `done`.
    pub fn add(&self, names: impl IntoIterator<Item = String>) {
        let mut state = self.state();

        for name in names {
            if !state.dirty.insert(name.clone()) {
                continue;
            }

            if !state.processing.contains(&name) {
                state.ready.push_back(name);
                self.ready.notify_one();
            }
        }
    }

    /// Blocks until a name is ready and hands it to the caller.
    pub fn get(&self) -> String {
        let mut state = self.state();

        loop {
            if let Some(name) = state.ready.pop_front() {
                state.dirty.remove(&name);
                state.processing.insert(name.clone());

                return name;
            }

            state = self.ready.wait(state).expect("queue lock poisoned");
        }
    }

    /// Releases a name that `get` handed out, and queues it again if it was
    /// added while the worker held it.
    pub fn done(&self, name: &str) {
        let mut state = self.state();
        state.processing.remove(name);

        if state.dirty.contains(name) {
            state.ready.push_back(name.to_string());
            self.ready.notify_one();
        }
    }
}

Why the second set? Because an object can change while a worker is in the middle of reconciling it. Without processing, the new event would put the name back in the ready list, a second worker would take it, and two workers would be acting on the same object at once. If both write a status with a version check, one of them fails with a conflict and its work is wasted. If both do something outside Kubernetes, like starting a container, both succeed, and our job ends up with two containers. So the queue holds the name back instead. When Add sees a name that a worker is holding, it marks the name dirty but leaves it out of the ready list. When the worker calls Done, the queue notices the mark and makes the name ready again, and the next reconcile sees the object as it is after the change. The table follows one name through that sequence:

call             ready      dirty      processing   effect
---------------  ---------  ---------  -----------  --------------------------------
Add(job-a)       [job-a]    {job-a}    {}           ready for a worker
Get() -> job-a   []         {}         {job-a}      a worker takes it
Add(job-a)       []         {job-a}    {job-a}      edited during the run: marked
Add(job-a)       []         {job-a}    {job-a}      already marked: no change
Done(job-a)      [job-a]    {job-a}    {}           marked, so ready again

Animation 1.1 plays that sequence with the burst of five edits from the run at the end of this chapter. Its last part repeats the calls on a queue that has no processing set and gives the name to a second worker.

Three boxes, ready, dirty and processing, above a worker. Add puts job-a in ready and dirty. Get moves it to processing and the worker starts a reconcile. Four more Add calls during the reconcile mark the name dirty and leave ready empty. Done moves the name back to ready, and a second reconcile reads a spec of 6 GPUs. In the last part the processing set is removed: an Add during the reconcile makes the name ready, and a second worker takes it while the first is still working. — **Animation 1.1** One name through the work queue. Five edits lead to two reconciles, and the two never overlap.

One more detail: a worker that calls Get on an empty queue has to go to sleep until a name arrives, and it must not hold the lock while it sleeps, or Add could never get in to add one. Both versions solve this with a condition variable, sync.Cond in Go and std::sync::Condvar in Rust. Waiting on a condition variable releases the lock and puts the worker to sleep in one step. Add and Done signal it whenever they make a name ready. The wait sits inside a loop that checks the list again after waking up, because with several workers another one may have taken the name first.

Reconciling

Now we can write the controller itself. It takes a name from the queue, reads the object from the cache, and makes the world match it. Kubernetes calls this reconciling: bringing what actually exists into agreement with what the user asked for.

The part that does the work is one function.

Listing 1.23 Reconciling one name, in Go. code/go/informer/controller.go

// reconcile reads the named object from the cache and, if its status differs
// from its spec, writes a new status conditioned on the version it read.
func (c *Controller) reconcile(name string) error {
	o, ok := c.cache.Get(name)

	if !ok {
		logf("reconcile %s: deleted, releasing its GPUs", name)

		return nil
	}

	if o.Status.GPUs == o.Spec.GPUs {
		return nil
	}

	time.Sleep(20 * time.Millisecond) // slow work, such as starting containers
	o.Status.GPUs = o.Spec.GPUs

	if err := c.server.Update(o); err != nil {
		logf("reconcile %s: write at version %d: %v", name, o.ResourceVersion, err)

		return err
	}

	logf("reconcile %s: status.GPUs = %d", name, o.Status.GPUs)

	return nil
}

Listing 1.24 Reconciling one name, in Rust. code/rust/informer/src/controller.rs

impl Controller {
    /// Reads the named object from the cache and, if its status differs from
    /// its spec, writes a new status conditioned on the version it read.
    fn reconcile(&self, name: &str) -> Result<(), ApiError> {
        let Some(mut object) = self.cache.get(name) else {
            log!("reconcile {name}: deleted, releasing its GPUs");

            return Ok(());
        };

        if object.status.gpus == object.spec.gpus {
            return Ok(());
        }

        thread::sleep(Duration::from_millis(20)); // slow work, such as starting containers
        object.status.gpus = object.spec.gpus;
        let version = object.resource_version;

        match self.server.update(object.clone()) {
            Ok(()) => {
                log!("reconcile {name}: status.gpus = {}", object.status.gpus);

                Ok(())
            }
            Err(error) => {
                log!("reconcile {name}: write at version {version}: {error}");

                Err(error)
            }
        }
    }
}

reconcile first looks the object up in the cache. If the object is missing, it was deleted, and a real controller would now release whatever it had set up for it; our model logs a message. If the status already matches the spec, there is nothing to do. Otherwise the controller does its work, which our model stands in for with a 20-millisecond sleep, sets the status, and writes the object back.

Notice that reconcile looks only at the object’s current state. It does not know, or care, why it was called. A name queued once and a name queued ten times lead to the same action, and so do a change reported by an event and a change discovered by a relist. Engineers call this style level-triggered, a term borrowed from digital electronics, where a level-triggered circuit responds to the present level of a signal and an edge-triggered circuit responds to each change of the signal. A controller that reacted to individual events would be edge-triggered, and it would do wasted work during bursts of edits and miss changes whose events a compaction had discarded.

Around it sits the loop that feeds it names.

Listing 1.25 The controller and its worker loop, in Go. code/go/informer/controller.go

// Controller makes each object's status agree with its spec.
type Controller struct {
	// server receives the controller's status writes.
	server *APIServer
	// cache supplies the current state of each object.
	cache *Cache
	// queue supplies the names of objects to reconcile.
	queue *WorkQueue
}

// Run is a single worker: it takes names from the queue one at a time and
// reconciles them. It does not return.
func (c *Controller) Run() {
	for {
		name := c.queue.Get()
		err := c.reconcile(name)
		c.queue.Done(name)

		if err != nil {
			c.queue.Add(name) // client-go requeues with a backoff
		}
	}
}

Listing 1.26 The controller and its worker loop, in Rust. code/rust/informer/src/controller.rs

/// Makes each object's status agree with its spec.
pub struct Controller {
    /// Receives the controller's status writes.
    pub server: Arc<ApiServer>,
    /// Supplies the current state of each object.
    pub cache: Arc<Cache>,
    /// Supplies the names of objects to reconcile.
    pub queue: Arc<WorkQueue>,
}

impl Controller {
    /// A single worker: takes names from the queue one at a time and
    /// reconciles them. It does not return.
    pub fn run(&self) {
        loop {
            let name = self.queue.get();
            let result = self.reconcile(&name);
            self.queue.done(&name);

            if result.is_err() {
                self.queue.add([name]); // client-go requeues with a backoff
            }
        }
    }
}

Run is the worker loop. It takes a name, reconciles it, and hands it back with Done. If the reconcile failed, it adds the name again so that the next attempt starts from whatever the cache holds by then.

Writing back to the server

Our cache always trails the server by the time it takes an event to arrive. During the 20 milliseconds that reconcile spends working, a user might change the spec. If the controller then wrote its copy of the object back unconditionally, it would store the old spec together with a status computed for the old spec, and the user’s edit would simply disappear.

This is exactly the situation the resource version check was made for. reconcile writes the object with the version it read. If the object changed in the meantime, the server refuses the write with a conflict, Run puts the name back in the queue, and the next reconcile reads the newer object from the cache. This approach is called optimistic concurrency: we take no lock on the object, we assume nobody else will change it, and we rely on the server to catch the cases where that assumption was wrong at the moment we write.

Running the model

Figure 1.1 shows the five parts together and numbers the two steps the reflector takes for each event.

The API server numbers every write and stores objects in etcd. Inside the controller's process, the reflector receives events, applies each one to the cache, and then adds the object's name to the work queue. The controller gets a name from the queue, reads the object from the cache, and writes the status back to the API server with the version it read. — **Figure 1.1** The path of one change: an event from the API server, the cache, the work queue, a reconcile, and a conditional write back to the server.

Time to see it all work together. main creates two jobs, starts the reflector and the controller, and then puts the machinery through each situation from this chapter: a burst of edits that arrives while a reconcile is running, a dropped watch, and a partition during which a job is deleted and the history is compacted. The Go version runs the reflector and the controller as goroutines. The Rust version runs them on threads, each holding an Arc handle to the shared server, cache, and queue. Both write their log to standard error, with the number of milliseconds since the program started; Rust does this with a small log! macro, and Go with a logf function.

Listing 1.27 The demonstration, in Go. code/go/informer/main.go

func main() {
	log.SetFlags(0)
	server := NewAPIServer()
	server.Update(Object{Name: "job-a", Spec: Spec{GPUs: 1}})
	server.Update(Object{Name: "job-b", Spec: Spec{GPUs: 8}})

	cache := NewCache()
	queue := NewWorkQueue()
	go (&Reflector{server, cache, queue}).Run()
	go (&Controller{server, cache, queue}).Run()
	time.Sleep(100 * time.Millisecond)

	logf("--- job-a is edited five times in a burst")

	for gpus := 2; gpus <= 6; gpus++ {
		server.Update(Object{Name: "job-a", Spec: Spec{GPUs: gpus}})

		if gpus == 2 {
			time.Sleep(5 * time.Millisecond) // the controller starts on the first edit
		}
	}

	time.Sleep(150 * time.Millisecond)

	logf("--- the watch connection drops")
	server.DropWatches()
	time.Sleep(50 * time.Millisecond)

	logf("--- partition: job-b is deleted and the history compacted meanwhile")
	server.Partition()
	server.Delete("job-b")
	server.Compact()
	time.Sleep(30 * time.Millisecond)
	server.Heal()
	time.Sleep(150 * time.Millisecond)

	list, version := server.List()
	logf("final state at version %d: %v", version, list)
}

Listing 1.28 The demonstration, in Rust. code/rust/informer/src/main.rs

fn main() {
    START.get_or_init(Instant::now);
    let server = Arc::new(ApiServer::default());
    server
        .update(Object::new("job-a", 1))
        .expect("unconditional write");
    server
        .update(Object::new("job-b", 8))
        .expect("unconditional write");

    let cache = Arc::new(Cache::default());
    let queue = Arc::new(WorkQueue::default());
    let reflector = Reflector {
        server: Arc::clone(&server),
        cache: Arc::clone(&cache),
        queue: Arc::clone(&queue),
    };
    let controller = Controller {
        server: Arc::clone(&server),
        cache,
        queue,
    };
    thread::spawn(move || reflector.run());
    thread::spawn(move || controller.run());
    sleep_ms(100);

    log!("--- job-a is edited five times in a burst");

    for gpus in 2..=6 {
        server
            .update(Object::new("job-a", gpus))
            .expect("unconditional write");
        if gpus == 2 {
            sleep_ms(5); // the controller starts on the first edit
        }
    }

    sleep_ms(150);

    log!("--- the watch connection drops");
    server.drop_watches();
    sleep_ms(50);

    log!("--- partition: job-b is deleted and the history compacted meanwhile");
    server.partition();
    server.delete("job-b");
    server.compact();
    sleep_ms(30);
    server.heal();
    sleep_ms(150);

    let (list, version) = server.list();
    let objects: Vec<String> = list.iter().map(ToString::to_string).collect();
    log!("final state at version {version}: [{}]", objects.join(" "));
}

The Go version prints:

$ go run ./informer
   0ms  reflector: list at version 2: [job-a{spec.GPUs: 1, status.GPUs: 0, v1} job-b{spec.GPUs: 8, status.GPUs: 0, v2}]
  20ms  reconcile job-a: status.GPUs = 1
  40ms  reconcile job-b: status.GPUs = 8
 101ms  --- job-a is edited five times in a burst
 121ms  reconcile job-a: write at version 5: 409 Conflict: object has been modified
 141ms  reconcile job-a: status.GPUs = 6
 257ms  --- the watch connection drops
 257ms  reflector: watch closed, resuming from version 10
 307ms  --- partition: job-b is deleted and the history compacted meanwhile
 307ms  reflector: watch closed, resuming from version 10
 338ms  reflector: watch from version 10: 410 Gone: resource version too old
 338ms  reflector: list at version 11: [job-a{spec.GPUs: 6, status.GPUs: 6, v10}]
 338ms  reconcile job-b: deleted, releasing its GPUs
 489ms  final state at version 11: [job-a{spec.GPUs: 6, status.GPUs: 6, v10}]

The Rust version prints the same sequence of events. Its timings differ by a millisecond or two, and its field names follow Rust’s naming conventions:

$ cargo run -p informer
   0ms  reflector: list at version 2: [job-a{spec.gpus: 1, status.gpus: 0, v1} job-b{spec.gpus: 8, status.gpus: 0, v2}]
  20ms  reconcile job-a: status.gpus = 1
  40ms  reconcile job-b: status.gpus = 8
 100ms  --- job-a is edited five times in a burst
 120ms  reconcile job-a: write at version 5: 409 Conflict: object has been modified
 141ms  reconcile job-a: status.gpus = 6
 255ms  --- the watch connection drops
 256ms  reflector: watch closed, resuming from version 10
 306ms  --- partition: job-b is deleted and the history compacted meanwhile
 306ms  reflector: watch closed, resuming from version 10
 336ms  reflector: watch from version 10: 410 Gone: resource version too old
 337ms  reflector: list at version 11: [job-a{spec.gpus: 6, status.gpus: 6, v10}]
 337ms  reconcile job-b: deleted, releasing its GPUs
 486ms  final state at version 11: [job-a{spec.gpus: 6, status.gpus: 6, v10}]

Let us read the log from the top. The first three lines are the initial list and one reconcile for each job. Each status write is itself a change, so the reflector receives an event for it and queues the name again. The reconcile that follows finds the status equal to the spec and returns quietly, so each job appears only once.

The burst of five edits led to just two reconciles of job-a. The controller took the name right after the first edit, read a spec of 2 GPUs at version 5, and started its 20 milliseconds of work. The other four edits arrived while it was busy, moving the object to version 9, and because the controller was holding the name, the queue marked it dirty. When the controller wrote its status with version 5, the server refused it with a conflict. Done made the name ready again, the next reconcile read a spec of 6 GPUs at version 9, and it wrote a status of 6 at version 10. The specs of 3, 4, and 5 GPUs had been replaced before the controller ever read them, so it did no work for them at all.

When the watch was dropped, the reflector reopened it from version 10, the version of the last event it had applied, and there was no need to list. During the partition, job-b was deleted at version 11 and the history was compacted, which moved the compaction point to 11. When the partition ended, the reflector asked to resume from version 10, the server answered 410 Gone, and the reflector listed again. The new list held only job-a, Replace reported job-b as missing, and the controller found no job-b in the cache and released its GPUs. Our client learned about the deletion by comparing the list with its cache, because the compaction had thrown the deletion event away.

Animation 1.2 replays the log from the list at version 2 to the list at version 11, with the server’s objects on the left and the cache on the right.

The API server on the left with job-a, job-b and its recent events; the reflector's cache on the right. The reflector lists at version 2 and opens a watch. Events 3 to 10 arrive and the cache follows. The watch is closed and reopened from 10 with no list. During a partition job-b is deleted at version 11 and the history is compacted, while the cache still holds job-b. After the partition the server answers a watch from 10 with 410 Gone, the reflector lists at version 11, and job-b, present in the old cache and missing from the list, is removed. — **Animation 1.2** The run of the model: a list, a watch, a dropped watch that resumes, and a partition that ends in `410 Gone` and a second list.

The same design in client-go

Real Go controllers use client-go, either directly or through controller-runtime, a higher-level library built on top of it. Every part of our model has a counterpart there:

model                     client-go
------------------------  ---------------------------------------------------------
APIServer.List, Watch     ListerWatcher, backed by HTTP list and watch requests
Reflector                 Reflector
Cache                     Indexer: a thread-safe store with secondary indexes,
                          for example all pods on one node
Reflector and Cache       SharedIndexInformer
WorkQueue                 workqueue, with the same dirty and processing sets
Controller.reconcile      the Reconcile method of a controller

client-go also has a few mechanisms that our model leaves out, and each one fixes a problem we can now recognize.

Between the reflector and the cache, client-go puts a queue called the DeltaFIFO. For each object it collects the changes in the order they arrived, and the code that updates the cache and notifies controllers takes one object at a time along with all of its pending changes. When a relist shows that an object has disappeared, the DeltaFIFO emits a deletion whose object is wrapped in a DeletedFinalStateUnknown marker. The marker warns event handlers that the object they receive is the last state the cache had seen, which may be older than the state the object was in when it was deleted.

The API server can send bookmark events, which carry a resource version and no object. They solve a problem with quiet watches. A watch on a kind of object that rarely changes can go a long time without any events, so the version it would resume from grows old. After the next compaction, reconnecting from that version would fail with 410 Gone and force a full relist. Bookmarks move the watch’s position forward even when nothing changes, so a reconnect usually succeeds.

An informer can be given a resync period. Every so often it queues the name of every object in its cache again, straight from memory and without asking the server for anything. This is for controllers that manage something outside Kubernetes, such as a cloud load balancer. If someone changes that load balancer by hand, no Kubernetes event tells the controller, and a periodic resync gives it a regular chance to notice and fix the drift.

The work queue waits before retrying a name whose reconcile failed. By default the wait starts at 5 milliseconds and doubles after each failure, up to 1,000 seconds, and the queue also limits how fast it hands out names overall. Our model retries at once. A real controller that did that while its writes kept failing, say because of a bug in how it builds the object, would flood the API server with requests.

Finally, informers are shared within a process. If ten controllers in one program all care about pods, they share a single pod watch and a single cache through a SharedInformerFactory, and each registers its own event handlers and its own work queue. The API server then serves one watch per process instead of one per controller.

Our model’s cache hands out copies, so a caller can change what it gets without affecting anyone else. client-go’s cache hands out pointers into the shared store instead, to avoid copying every object on every read. That has a sharp edge. If a client-go controller changes a field of an object it read from the cache, it is changing the very object every other controller in the process reads, while the server still holds the original. The other controllers now see a state that exists nowhere else. The rule is to call the object’s generated DeepCopy method, which copies every nested slice and map, and to change the copy.

Reading your own writes

A controller’s own writes reach its cache the same way everyone else’s do: the server stores the write, emits an event, and the reflector applies it. Until that event arrives, the cache still shows the state from before the write. Picture a reconcile that creates a pod for a job. If a second event for the same job arrives before the pod’s creation event, the next reconcile finds no pod in the cache and creates a second one.

Kubernetes controllers guard against this in two ways. The ReplicaSet controller, which keeps a requested number of identical pods running, records expectations: after creating three pods, it notes that it expects to see three creation events, and it leaves that ReplicaSet alone until the cache has shown them or a timeout has passed. The second technique uses two fields that every object carries. The API server increments metadata.generation every time the spec changes, and a controller copies the generation it acted on into status.observedGeneration. When the two numbers are equal, the status describes the current spec. When they differ, both the controller and any user reading the object can see that the latest change to the spec has not been handled yet.

The complete program

The model is six source files in each language, printed here in the order the chapter built them.

Listing 1.29 The API server, complete, in Go. code/go/informer/server.go

// server.go models the Kubernetes API server. It stores objects, numbers every
// write, keeps a history of recent changes, and serves lists and watches.

package main

import (
	"errors"
	"fmt"
	"maps"
	"slices"
	"strings"
	"sync"
)

// Spec is the part of an object that a user writes: the desired state.
type Spec struct {
	// GPUs is the number of GPUs the job asks for.
	GPUs int
}

// Status is the part of an object that the controller writes: the observed state.
type Status struct {
	// GPUs is the number of GPUs the controller has provided to the job.
	GPUs int
}

// Object is one stored record, such as a training job.
type Object struct {
	// Name identifies the object on the server.
	Name string
	// Spec holds what the user asked for.
	Spec Spec
	// Status holds what the controller has done about it.
	Status Status
	// ResourceVersion is the server's write counter at the object's latest change.
	ResourceVersion int
}

// String formats an object for the demonstration's log lines.
func (o Object) String() string {
	return fmt.Sprintf("%s{spec.GPUs: %d, status.GPUs: %d, v%d}",
		o.Name, o.Spec.GPUs, o.Status.GPUs, o.ResourceVersion)
}

// EventType is the kind of change that an Event records.
type EventType int

// The kinds of change that a watch reports.
const (
	Added    EventType = iota // the object was created
	Modified                  // the object was updated
	Deleted                   // the object was removed
)

// String returns the name that the Kubernetes watch protocol uses for t.
func (t EventType) String() string {
	return [...]string{"ADDED", "MODIFIED", "DELETED"}[t]
}

// Event records one change to one object.
type Event struct {
	// Type is the kind of change.
	Type EventType
	// Object is the object after the change, or its final state if it was deleted.
	Object Object
}

// Errors that the server returns. The first two carry the HTTP status codes
// that the Kubernetes API server uses for the same conditions.
var (
	// ErrGone reports that the history a watch needs has been discarded.
	ErrGone = errors.New("410 Gone: resource version too old")
	// ErrConflict reports a conditional write whose resource version is out of date.
	ErrConflict = errors.New("409 Conflict: object has been modified")
	// ErrUnreachable reports that the client cannot reach the server.
	ErrUnreachable = errors.New("server unreachable")
)


// APIServer stores objects and streams their changes to watchers. Its methods
// are safe for concurrent use.
type APIServer struct {
	// mu guards every field below.
	mu sync.Mutex
	// version is the last resource version the server handed out.
	version int
	// objects holds the current state of every object, keyed by name.
	objects map[string]Object
	// history holds the events newer than compacted, oldest first.
	history []Event
	// compacted is the oldest version from which a watch may resume.
	compacted int
	// watches holds one channel for each open watch.
	watches []chan Event
	// unreachable makes new watches fail, to simulate a network partition.
	unreachable bool
}

// NewAPIServer returns an empty server.
func NewAPIServer() *APIServer {
	return &APIServer{objects: map[string]Object{}}
}


// Update creates or replaces an object and returns ErrConflict if the write's
// condition fails. A non-zero o.ResourceVersion is a condition: the caller read
// the object at that version and wants the write applied only if the object is
// still at it. A zero ResourceVersion writes unconditionally.
func (s *APIServer) Update(o Object) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current, exists := s.objects[o.Name]

	if o.ResourceVersion != 0 && o.ResourceVersion != current.ResourceVersion {
		return ErrConflict
	}

	eventType := Modified

	if !exists {
		eventType = Added
	}

	s.version++
	o.ResourceVersion = s.version
	s.objects[o.Name] = o
	s.publish(Event{eventType, o})

	return nil
}

// Delete removes the named object, if it exists.
func (s *APIServer) Delete(name string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	o, exists := s.objects[name]

	if !exists {
		return
	}

	delete(s.objects, name)
	s.version++
	o.ResourceVersion = s.version
	s.publish(Event{Deleted, o})
}

// publish appends e to the history and sends it to every open watch.
// The caller must hold s.mu.
func (s *APIServer) publish(e Event) {
	s.history = append(s.history, e)

	for _, w := range s.watches {
		w <- e
	}
}


// List returns every object, sorted by name, and the version at which the
// snapshot was taken.
func (s *APIServer) List() ([]Object, int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	byName := func(a, b Object) int { return strings.Compare(a.Name, b.Name) }

	return slices.SortedFunc(maps.Values(s.objects), byName), s.version
}

// Watch returns a channel that first delivers the stored events newer than
// version and then every new event, until the server closes it.
func (s *APIServer) Watch(version int) (<-chan Event, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	switch {
	case s.unreachable:
		return nil, ErrUnreachable
	case version < s.compacted:
		return nil, ErrGone
	}

	w := make(chan Event, 1024)

	for _, e := range s.history {
		if e.Object.ResourceVersion > version {
			w <- e
		}
	}

	s.watches = append(s.watches, w)

	return w, nil
}


// Compact discards the event history, as etcd discards old versions.
func (s *APIServer) Compact() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.compacted = s.version
	s.history = nil
}

// DropWatches closes every open watch, as a load balancer's idle timeout would.
func (s *APIServer) DropWatches() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.dropWatches()
}

// Partition closes every open watch and makes new watches fail until Heal.
func (s *APIServer) Partition() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.unreachable = true
	s.dropWatches()
}

// Heal ends a partition, so new watches succeed again.
func (s *APIServer) Heal() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.unreachable = false
}

// dropWatches closes and forgets every watch channel. The caller must hold s.mu.
func (s *APIServer) dropWatches() {
	for _, w := range s.watches {
		close(w)
	}

	s.watches = nil
}

Listing 1.30 The API server, complete, in Rust. code/rust/informer/src/server.rs

//! A model of the Kubernetes API server. It stores objects, numbers every
//! write, keeps a history of recent changes, and serves lists and watches.

use std::collections::HashMap;
use std::fmt;
use std::sync::Mutex;
use std::sync::mpsc::{Receiver, Sender, channel};

/// The part of an object that a user writes: the desired state.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Spec {
    /// Number of GPUs the job asks for.
    pub gpus: u32,
}

/// The part of an object that the controller writes: the observed state.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Status {
    /// Number of GPUs the controller has provided to the job.
    pub gpus: u32,
}

/// One stored record, such as a training job.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Object {
    /// Identifies the object on the server.
    pub name: String,
    /// What the user asked for.
    pub spec: Spec,
    /// What the controller has done about it.
    pub status: Status,
    /// The server's write counter at the object's latest change.
    pub resource_version: u64,
}

impl Object {
    /// Returns a new object with the given name and GPU request, an empty
    /// status, and no resource version.
    pub fn new(name: &str, gpus: u32) -> Self {
        Self {
            name: name.to_string(),
            spec: Spec { gpus },
            ..Self::default()
        }
    }
}

impl fmt::Display for Object {
    /// Formats an object for the demonstration's log lines.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{}{{spec.gpus: {}, status.gpus: {}, v{}}}",
            self.name, self.spec.gpus, self.status.gpus, self.resource_version
        )
    }
}

/// The kind of change that an [`Event`] records.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum EventType {
    /// The object was created.
    Added,
    /// The object was updated.
    Modified,
    /// The object was removed.
    Deleted,
}

/// One change to one object.
#[derive(Clone, Debug)]
pub struct Event {
    /// The kind of change.
    pub kind: EventType,
    /// The object after the change, or its final state if it was deleted.
    pub object: Object,
}

/// Errors that the server returns. The first two carry the HTTP status codes
/// that the Kubernetes API server uses for the same conditions.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ApiError {
    /// The history that a watch needs has been discarded.
    Gone,
    /// A conditional write carried an out-of-date resource version.
    Conflict,
    /// The client cannot reach the server.
    Unreachable,
}

impl fmt::Display for ApiError {
    /// Formats the error the way the Kubernetes API server reports it.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Gone => "410 Gone: resource version too old",
            Self::Conflict => "409 Conflict: object has been modified",
            Self::Unreachable => "server unreachable",
        })
    }
}

/// Everything the server stores, kept behind one lock.
#[derive(Default)]
struct State {
    /// The last resource version the server handed out.
    version: u64,
    /// The current state of every object, keyed by name.
    objects: HashMap<String, Object>,
    /// The events newer than `compacted`, oldest first.
    history: Vec<Event>,
    /// The oldest version from which a watch may resume.
    compacted: u64,
    /// One sender for each open watch.
    watches: Vec<Sender<Event>>,
    /// Makes new watches fail, to simulate a network partition.
    unreachable: bool,
}

impl State {
    /// Appends `event` to the history and sends it to every open watch.
    /// Watches whose receiver has gone away are forgotten.
    fn publish(&mut self, event: Event) {
        self.watches.retain(|w| w.send(event.clone()).is_ok());
        self.history.push(event);
    }
}

/// Stores objects and streams their changes to watchers. Its methods are safe
/// to call from several threads.
#[derive(Default)]
pub struct ApiServer {
    /// Guards all of the server's data.
    state: Mutex<State>,
}

impl ApiServer {
    /// Locks the server's data.
    fn state(&self) -> std::sync::MutexGuard<'_, State> {
        self.state.lock().expect("server lock poisoned")
    }
}

impl ApiServer {
    /// Creates or replaces an object. A non-zero `resource_version` is a
    /// condition: the caller read the object at that version and wants the
    /// write applied only if the object is still at it. Zero writes
    /// unconditionally.
    pub fn update(&self, mut object: Object) -> Result<(), ApiError> {
        let mut state = self.state();
        let current = state.objects.get(&object.name);
        let current_version = current.map_or(0, |o| o.resource_version);

        if object.resource_version != 0 && object.resource_version != current_version {
            return Err(ApiError::Conflict);
        }

        let kind = if current.is_some() {
            EventType::Modified
        } else {
            EventType::Added
        };
        state.version += 1;
        object.resource_version = state.version;
        state.objects.insert(object.name.clone(), object.clone());
        state.publish(Event { kind, object });

        Ok(())
    }

    /// Removes the named object, if it exists.
    pub fn delete(&self, name: &str) {
        let mut state = self.state();
        let Some(mut object) = state.objects.remove(name) else {
            return;
        };
        state.version += 1;
        object.resource_version = state.version;
        state.publish(Event {
            kind: EventType::Deleted,
            object,
        });
    }
}

impl ApiServer {
    /// Returns every object, sorted by name, and the version at which the
    /// snapshot was taken.
    pub fn list(&self) -> (Vec<Object>, u64) {
        let state = self.state();
        let mut objects: Vec<Object> = state.objects.values().cloned().collect();
        objects.sort_by(|a, b| a.name.cmp(&b.name));

        (objects, state.version)
    }

    /// Returns a receiver that first delivers the stored events newer than
    /// `version` and then every new event, until the server drops the watch.
    pub fn watch(&self, version: u64) -> Result<Receiver<Event>, ApiError> {
        let mut state = self.state();

        if state.unreachable {
            return Err(ApiError::Unreachable);
        }

        if version < state.compacted {
            return Err(ApiError::Gone);
        }

        let (sender, receiver) = channel();

        for event in &state.history {
            if event.object.resource_version > version {
                sender.send(event.clone()).expect("receiver is held here");
            }
        }

        state.watches.push(sender);

        Ok(receiver)
    }
}

impl ApiServer {
    /// Discards the event history, as etcd discards old versions.
    pub fn compact(&self) {
        let mut state = self.state();
        state.compacted = state.version;
        state.history.clear();
    }

    /// Closes every open watch, as a load balancer's idle timeout would.
    pub fn drop_watches(&self) {
        self.state().watches.clear();
    }

    /// Closes every open watch and makes new watches fail until [`Self::heal`].
    pub fn partition(&self) {
        let mut state = self.state();
        state.unreachable = true;
        state.watches.clear();
    }

    /// Ends a partition, so new watches succeed again.
    pub fn heal(&self) {
        self.state().unreachable = false;
    }
}

Listing 1.31 The cache, complete, in Go. code/go/informer/cache.go

// cache.go holds the client's local copy of the server's objects.

package main

import "sync"

// Cache is the client's copy of the server's objects, kept current by a
// Reflector. Its methods are safe for concurrent use.
type Cache struct {
	// mu guards objects.
	mu sync.RWMutex
	// objects holds the latest known state of every object, keyed by name.
	objects map[string]Object
}

// NewCache returns an empty cache.
func NewCache() *Cache {
	return &Cache{objects: map[string]Object{}}
}

// Get returns the cached copy of the named object and whether it exists.
func (c *Cache) Get(name string) (Object, bool) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	o, ok := c.objects[name]

	return o, ok
}

// Apply records one event from a watch.
func (c *Cache) Apply(e Event) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if e.Type == Deleted {
		delete(c.objects, e.Object.Name)

		return
	}

	c.objects[e.Object.Name] = e.Object
}


// Replace installs the result of a fresh list and returns the names of the
// objects that changed, including objects that disappeared while the client
// was not watching.
func (c *Cache) Replace(list []Object) []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	old := c.objects
	c.objects = make(map[string]Object, len(list))
	var changed []string

	for _, o := range list {
		c.objects[o.Name] = o
		// A name missing from old reads as version 0, so it counts as changed.
		if old[o.Name].ResourceVersion != o.ResourceVersion {
			changed = append(changed, o.Name)
		}
	}

	for name := range old {
		if _, ok := c.objects[name]; !ok {
			changed = append(changed, name)
		}
	}

	return changed
}

Listing 1.32 The cache, complete, in Rust. code/rust/informer/src/cache.rs

//! The client's local copy of the server's objects.

use crate::server::{Event, EventType, Object};
use std::collections::HashMap;
use std::sync::RwLock;

/// The client's copy of the server's objects, kept current by a
/// [`Reflector`](crate::reflector::Reflector). Its methods are safe to call
/// from several threads.
#[derive(Default)]
pub struct Cache {
    /// The latest known state of every object, keyed by name.
    objects: RwLock<HashMap<String, Object>>,
}

impl Cache {
    /// Returns the cached copy of the named object, if it exists.
    pub fn get(&self, name: &str) -> Option<Object> {
        let objects = self.objects.read().expect("cache lock poisoned");

        objects.get(name).cloned()
    }

    /// Records one event from a watch.
    pub fn apply(&self, event: Event) {
        let mut objects = self.objects.write().expect("cache lock poisoned");

        match event.kind {
            EventType::Deleted => objects.remove(&event.object.name),
            EventType::Added | EventType::Modified => {
                objects.insert(event.object.name.clone(), event.object)
            }
        };
    }
}

impl Cache {
    /// Installs the result of a fresh list and returns the names of the
    /// objects that changed, including objects that disappeared while the
    /// client was not watching.
    pub fn replace(&self, list: &[Object]) -> Vec<String> {
        let mut objects = self.objects.write().expect("cache lock poisoned");
        let old = std::mem::take(&mut *objects);
        let mut changed = Vec::new();

        for object in list {
            let old_version = old.get(&object.name).map(|o| o.resource_version);

            if old_version != Some(object.resource_version) {
                changed.push(object.name.clone());
            }

            objects.insert(object.name.clone(), object.clone());
        }

        for name in old.keys() {
            if !objects.contains_key(name) {
                changed.push(name.clone());
            }
        }

        changed
    }
}

Listing 1.33 The reflector, complete, in Go. code/go/informer/reflector.go

// reflector.go keeps a Cache current by listing and watching the server.

package main

import (
	"errors"
	"time"
)

// Reflector copies the server's objects into a cache and queues the name of
// every object that changes.
type Reflector struct {
	// server is the source of objects and events.
	server *APIServer
	// cache receives every object the reflector sees.
	cache *Cache
	// queue receives the name of every object that changes.
	queue *WorkQueue
}

// Run lists every object, watches from the list's version, and lists again
// whenever the server can no longer resume the watch. It does not return.
func (r *Reflector) Run() {
	for {
		list, version := r.server.List()
		r.queue.Add(r.cache.Replace(list)...)
		logf("reflector: list at version %d: %v", version, list)
		r.watchFrom(version)
	}
}


// watchFrom applies events from version onward, reopens the watch after each
// disconnect, and returns when the server reports that the history is gone.
func (r *Reflector) watchFrom(version int) {
	for {
		w, err := r.server.Watch(version)

		if errors.Is(err, ErrGone) {
			logf("reflector: watch from version %d: %v", version, err)

			return
		}

		if err != nil {
			time.Sleep(10 * time.Millisecond) // client-go backs off exponentially
			continue
		}

		for e := range w {
			// Update the cache before queueing the name, so that a worker
			// that takes the name reads the object as of this event.
			r.cache.Apply(e)
			r.queue.Add(e.Object.Name)
			version = e.Object.ResourceVersion
		}

		logf("reflector: watch closed, resuming from version %d", version)
	}
}

Listing 1.34 The reflector, complete, in Rust. code/rust/informer/src/reflector.rs

//! Keeps a [`Cache`] current by listing and watching the server.

use crate::cache::Cache;
use crate::log;
use crate::queue::WorkQueue;
use crate::server::{ApiError, ApiServer};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

/// Copies the server's objects into a cache and queues the name of every
/// object that changes.
pub struct Reflector {
    /// The source of objects and events.
    pub server: Arc<ApiServer>,
    /// Receives every object the reflector sees.
    pub cache: Arc<Cache>,
    /// Receives the name of every object that changes.
    pub queue: Arc<WorkQueue>,
}

impl Reflector {
    /// Lists every object, watches from the list's version, and lists again
    /// whenever the server can no longer resume the watch. It does not return.
    pub fn run(&self) {
        loop {
            let (list, version) = self.server.list();
            self.queue.add(self.cache.replace(&list));
            let objects: Vec<String> = list.iter().map(ToString::to_string).collect();
            log!(
                "reflector: list at version {version}: [{}]",
                objects.join(" ")
            );
            self.watch_from(version);
        }
    }
}

impl Reflector {
    /// Applies events from `version` onward, reopens the watch after each
    /// disconnect, and returns when the server reports that the history is gone.
    fn watch_from(&self, mut version: u64) {
        loop {
            let events = match self.server.watch(version) {
                Ok(events) => events,
                Err(ApiError::Gone) => {
                    log!(
                        "reflector: watch from version {version}: {}",
                        ApiError::Gone
                    );

                    return;
                }
                Err(_) => {
                    thread::sleep(Duration::from_millis(10)); // client-go backs off exponentially
                    continue;
                }
            };

            for event in events {
                version = event.object.resource_version;
                let name = event.object.name.clone();
                // Update the cache before queueing the name, so that a worker
                // that takes the name reads the object as of this event.
                self.cache.apply(event);
                self.queue.add([name]);
            }

            log!("reflector: watch closed, resuming from version {version}");
        }
    }
}

Listing 1.35 The work queue, complete, in Go. code/go/informer/queue.go

// queue.go implements the work queue between the reflector and the controller.

package main

import "sync"

// WorkQueue stores the names of objects that need reconciling. A name waits
// in the queue at most once, and a name that a worker holds is handed to no
// other worker until the first calls Done.
type WorkQueue struct {
	// mu guards every field below.
	mu sync.Mutex
	// cond wakes workers blocked in Get.
	cond *sync.Cond
	// queue holds the names ready for a worker, in arrival order.
	queue []string
	// dirty holds every name that needs processing.
	dirty map[string]bool
	// processing holds every name that a worker currently holds.
	processing map[string]bool
}

// NewWorkQueue returns an empty queue.
func NewWorkQueue() *WorkQueue {
	q := &WorkQueue{dirty: map[string]bool{}, processing: map[string]bool{}}
	q.cond = sync.NewCond(&q.mu)

	return q
}


// Add marks each name as needing processing. A name that is already pending
// is left where it is. A name that a worker holds is queued again by Done.
func (q *WorkQueue) Add(names ...string) {
	q.mu.Lock()
	defer q.mu.Unlock()

	for _, name := range names {
		if q.dirty[name] {
			continue
		}

		q.dirty[name] = true

		if !q.processing[name] {
			q.queue = append(q.queue, name)
			q.cond.Signal()
		}
	}
}

// Get blocks until a name is ready and hands it to the caller.
func (q *WorkQueue) Get() string {
	q.mu.Lock()
	defer q.mu.Unlock()

	for len(q.queue) == 0 {
		q.cond.Wait()
	}

	name := q.queue[0]
	q.queue = q.queue[1:]
	delete(q.dirty, name)
	q.processing[name] = true

	return name
}

// Done releases a name that Get handed out, and queues it again if it was
// added while the worker held it.
func (q *WorkQueue) Done(name string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	delete(q.processing, name)

	if q.dirty[name] {
		q.queue = append(q.queue, name)
		q.cond.Signal()
	}
}

Listing 1.36 The work queue, complete, in Rust. code/rust/informer/src/queue.rs

//! The work queue between the reflector and the controller.

use std::collections::{HashSet, VecDeque};
use std::sync::{Condvar, Mutex, MutexGuard};

/// The queue's data, kept behind one lock.
#[derive(Default)]
struct State {
    /// The names ready for a worker, in arrival order.
    ready: VecDeque<String>,
    /// Every name that needs processing.
    dirty: HashSet<String>,
    /// Every name that a worker currently holds.
    processing: HashSet<String>,
}

/// Stores the names of objects that need reconciling. A name waits in the
/// queue at most once, and a name that a worker holds is handed to no other
/// worker until the first calls [`WorkQueue::done`].
#[derive(Default)]
pub struct WorkQueue {
    /// Guards the queue's data.
    state: Mutex<State>,
    /// Wakes workers blocked in [`WorkQueue::get`].
    ready: Condvar,
}

impl WorkQueue {
    /// Locks the queue's data.
    fn state(&self) -> MutexGuard<'_, State> {
        self.state.lock().expect("queue lock poisoned")
    }
}

impl WorkQueue {
    /// Marks each name as needing processing. A name that is already pending
    /// is left where it is. A name that a worker holds is queued again by `done`.
    pub fn add(&self, names: impl IntoIterator<Item = String>) {
        let mut state = self.state();

        for name in names {
            if !state.dirty.insert(name.clone()) {
                continue;
            }

            if !state.processing.contains(&name) {
                state.ready.push_back(name);
                self.ready.notify_one();
            }
        }
    }

    /// Blocks until a name is ready and hands it to the caller.
    pub fn get(&self) -> String {
        let mut state = self.state();

        loop {
            if let Some(name) = state.ready.pop_front() {
                state.dirty.remove(&name);
                state.processing.insert(name.clone());

                return name;
            }

            state = self.ready.wait(state).expect("queue lock poisoned");
        }
    }

    /// Releases a name that `get` handed out, and queues it again if it was
    /// added while the worker held it.
    pub fn done(&self, name: &str) {
        let mut state = self.state();
        state.processing.remove(name);

        if state.dirty.contains(name) {
            state.ready.push_back(name.to_string());
            self.ready.notify_one();
        }
    }
}

Listing 1.37 The controller, complete, in Go. code/go/informer/controller.go

// controller.go reconciles objects whose names arrive on the work queue.

package main

import "time"

// Controller makes each object's status agree with its spec.
type Controller struct {
	// server receives the controller's status writes.
	server *APIServer
	// cache supplies the current state of each object.
	cache *Cache
	// queue supplies the names of objects to reconcile.
	queue *WorkQueue
}

// Run is a single worker: it takes names from the queue one at a time and
// reconciles them. It does not return.
func (c *Controller) Run() {
	for {
		name := c.queue.Get()
		err := c.reconcile(name)
		c.queue.Done(name)

		if err != nil {
			c.queue.Add(name) // client-go requeues with a backoff
		}
	}
}


// reconcile reads the named object from the cache and, if its status differs
// from its spec, writes a new status conditioned on the version it read.
func (c *Controller) reconcile(name string) error {
	o, ok := c.cache.Get(name)

	if !ok {
		logf("reconcile %s: deleted, releasing its GPUs", name)

		return nil
	}

	if o.Status.GPUs == o.Spec.GPUs {
		return nil
	}

	time.Sleep(20 * time.Millisecond) // slow work, such as starting containers
	o.Status.GPUs = o.Spec.GPUs

	if err := c.server.Update(o); err != nil {
		logf("reconcile %s: write at version %d: %v", name, o.ResourceVersion, err)

		return err
	}

	logf("reconcile %s: status.GPUs = %d", name, o.Status.GPUs)

	return nil
}

Listing 1.38 The controller, complete, in Rust. code/rust/informer/src/controller.rs

//! Reconciles objects whose names arrive on the work queue.

use crate::cache::Cache;
use crate::log;
use crate::queue::WorkQueue;
use crate::server::{ApiError, ApiServer};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

/// Makes each object's status agree with its spec.
pub struct Controller {
    /// Receives the controller's status writes.
    pub server: Arc<ApiServer>,
    /// Supplies the current state of each object.
    pub cache: Arc<Cache>,
    /// Supplies the names of objects to reconcile.
    pub queue: Arc<WorkQueue>,
}

impl Controller {
    /// A single worker: takes names from the queue one at a time and
    /// reconciles them. It does not return.
    pub fn run(&self) {
        loop {
            let name = self.queue.get();
            let result = self.reconcile(&name);
            self.queue.done(&name);

            if result.is_err() {
                self.queue.add([name]); // client-go requeues with a backoff
            }
        }
    }
}

impl Controller {
    /// Reads the named object from the cache and, if its status differs from
    /// its spec, writes a new status conditioned on the version it read.
    fn reconcile(&self, name: &str) -> Result<(), ApiError> {
        let Some(mut object) = self.cache.get(name) else {
            log!("reconcile {name}: deleted, releasing its GPUs");

            return Ok(());
        };

        if object.status.gpus == object.spec.gpus {
            return Ok(());
        }

        thread::sleep(Duration::from_millis(20)); // slow work, such as starting containers
        object.status.gpus = object.spec.gpus;
        let version = object.resource_version;

        match self.server.update(object.clone()) {
            Ok(()) => {
                log!("reconcile {name}: status.gpus = {}", object.status.gpus);

                Ok(())
            }
            Err(error) => {
                log!("reconcile {name}: write at version {version}: {error}");

                Err(error)
            }
        }
    }
}

Listing 1.39 The demonstration, complete, in Go. code/go/informer/main.go

// Command informer models how a Kubernetes controller learns about objects:
// a server that numbers every write, a reflector that lists and watches into
// a local cache, a work queue of names, and a controller that reconciles them.
// It then causes a burst of edits, a dropped connection, and a partition, and
// logs how each part responds.
//
// Run it from code/go with:
//
//	go run ./informer
package main

import (
	"fmt"
	"log"
	"time"
)

// start is the time the program began, for the relative timestamps in the log.
var start = time.Now()

// logf prints one log line prefixed with the milliseconds since start.
func logf(format string, args ...any) {
	log.Printf("%4dms  %s", time.Since(start).Milliseconds(), fmt.Sprintf(format, args...))
}

func main() {
	log.SetFlags(0)
	server := NewAPIServer()
	server.Update(Object{Name: "job-a", Spec: Spec{GPUs: 1}})
	server.Update(Object{Name: "job-b", Spec: Spec{GPUs: 8}})

	cache := NewCache()
	queue := NewWorkQueue()
	go (&Reflector{server, cache, queue}).Run()
	go (&Controller{server, cache, queue}).Run()
	time.Sleep(100 * time.Millisecond)

	logf("--- job-a is edited five times in a burst")

	for gpus := 2; gpus <= 6; gpus++ {
		server.Update(Object{Name: "job-a", Spec: Spec{GPUs: gpus}})

		if gpus == 2 {
			time.Sleep(5 * time.Millisecond) // the controller starts on the first edit
		}
	}

	time.Sleep(150 * time.Millisecond)

	logf("--- the watch connection drops")
	server.DropWatches()
	time.Sleep(50 * time.Millisecond)

	logf("--- partition: job-b is deleted and the history compacted meanwhile")
	server.Partition()
	server.Delete("job-b")
	server.Compact()
	time.Sleep(30 * time.Millisecond)
	server.Heal()
	time.Sleep(150 * time.Millisecond)

	list, version := server.List()
	logf("final state at version %d: %v", version, list)
}

Listing 1.40 The demonstration, complete, in Rust. code/rust/informer/src/main.rs

//! A model of how a Kubernetes controller learns about objects: a server that
//! numbers every write, a reflector that lists and watches into a local cache,
//! a work queue of names, and a controller that reconciles them. It then
//! causes a burst of edits, a dropped connection, and a partition, and logs
//! how each part responds.
//!
//! Run it from `code/rust` with `cargo run -p informer`.

mod cache;
mod controller;
mod queue;
mod reflector;
mod server;

use cache::Cache;
use controller::Controller;
use queue::WorkQueue;
use reflector::Reflector;
use server::{ApiServer, Object};
use std::sync::{Arc, OnceLock};
use std::thread;
use std::time::{Duration, Instant};

/// The time the program began, for the relative timestamps in the log.
static START: OnceLock<Instant> = OnceLock::new();

/// Prints one log line to standard error, prefixed with the milliseconds
/// since the program began.
#[macro_export]
macro_rules! log {
    ($($arg:tt)*) => {
        eprintln!("{:4}ms  {}", $crate::elapsed_ms(), format!($($arg)*))
    };
}

/// Returns the milliseconds since the program began.
pub fn elapsed_ms() -> u128 {
    START.get_or_init(Instant::now).elapsed().as_millis()
}

/// Sleeps for the given number of milliseconds.
fn sleep_ms(ms: u64) {
    thread::sleep(Duration::from_millis(ms));
}

fn main() {
    START.get_or_init(Instant::now);
    let server = Arc::new(ApiServer::default());
    server
        .update(Object::new("job-a", 1))
        .expect("unconditional write");
    server
        .update(Object::new("job-b", 8))
        .expect("unconditional write");

    let cache = Arc::new(Cache::default());
    let queue = Arc::new(WorkQueue::default());
    let reflector = Reflector {
        server: Arc::clone(&server),
        cache: Arc::clone(&cache),
        queue: Arc::clone(&queue),
    };
    let controller = Controller {
        server: Arc::clone(&server),
        cache,
        queue,
    };
    thread::spawn(move || reflector.run());
    thread::spawn(move || controller.run());
    sleep_ms(100);

    log!("--- job-a is edited five times in a burst");

    for gpus in 2..=6 {
        server
            .update(Object::new("job-a", gpus))
            .expect("unconditional write");
        if gpus == 2 {
            sleep_ms(5); // the controller starts on the first edit
        }
    }

    sleep_ms(150);

    log!("--- the watch connection drops");
    server.drop_watches();
    sleep_ms(50);

    log!("--- partition: job-b is deleted and the history compacted meanwhile");
    server.partition();
    server.delete("job-b");
    server.compact();
    sleep_ms(30);
    server.heal();
    sleep_ms(150);

    let (list, version) = server.list();
    let objects: Vec<String> = list.iter().map(ToString::to_string).collect();
    log!("final state at version {version}: [{}]", objects.join(" "));
}

Keyboard shortcuts

GPU-Accelerated Kubernetes - Distributed GPU Kernel & Job Scheduling in a Cloud Native Way