1.1 A GPU job controller on a real cluster

In chapter 1, we built a server, a cache, a reflector, a work queue, and a controller, and ran all of them inside one process. In this chapter, we will take that machinery to a real Kubernetes cluster and write a controller that runs GPU jobs on a real GPU. We will first teach the API server a new kind of object, the GPU job, and then watch those objects arrive the way our reflector did. Then we will write the controller, run it, and see a CUDA program run on a GPU inside a pod that our controller created. Finally we will make two clients collide on the same object and watch the API server’s version check stop the second one.

This time we will not write the machinery ourselves. We will use each language’s standard Kubernetes library, which provides the same parts we built by hand:

• In Go, we will use controller-runtime. It is built on client-go, the library whose parts we listed at the end of chapter 1, and it is what Kubebuilder, the usual tool for starting a Go controller project, generates code for.
• In Rust, we will use kube-rs, published as the kube crate.

We will introduce each part of these libraries at the moment we need it, and relate it to the piece we built in chapter 1.

Our cluster is created with kind, which runs each Kubernetes node as a Docker container. We talk to it with kubectl, the Kubernetes command-line client:

$ kind get clusters
gpu-lab
$ kubectl get nodes
NAME                    STATUS   ROLES           AGE   VERSION
gpu-lab-control-plane   Ready    control-plane   22s   v1.37.0

Pods in this cluster can use a GPU. Appendix A shows how we set that up.

The projects

Our Go code lives in the module in code/go, next to the model from chapter 1:

Listing 1.41 The Go module. code/go/go.mod

module gpuk8s

go 1.26.0

require (
	k8s.io/api v0.37.1
	k8s.io/apimachinery v0.37.1
	k8s.io/client-go v0.37.0
	sigs.k8s.io/controller-runtime v0.25.1
)

We depend on four modules. k8s.io/api holds the Go types of the built-in Kubernetes objects, such as Pod. k8s.io/apimachinery holds what all objects share: metadata, resource quantities such as “2 CPUs”, and the errors the API server returns. k8s.io/client-go is the client library, and sigs.k8s.io/controller-runtime is the controller framework built on it. We pick versions that match our cluster’s Kubernetes 1.37.

Our Rust code is the gpujob package in the Cargo workspace in code/rust, next to the model’s informer package:

Listing 1.42 The Cargo workspace. code/rust/Cargo.toml

[workspace]
members = ["informer", "gpujob", "gpuusage", "heartbeats", "phi", "leaseagent", "leasewatch", "fleethealth", "fencing", "provisioner", "migration", "allreduce", "gangsched", "counter", "cas", "gpuctl"]
resolver = "3"

Listing 1.43 The `gpujob` package. code/rust/gpujob/Cargo.toml

[package]
name = "gpujob"
version = "0.1.0"
edition = "2024"
publish = false
default-run = "controller"

[dependencies]
futures = "0.3"
k8s-openapi = { version = "0.28", features = ["v1_36"] }
kube = { version = "4.2", features = ["runtime", "derive"] }
schemars = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }

kube is the client. We turn on its runtime feature for the controller machinery and its derive feature for generating our own object types. k8s-openapi provides the Rust types of the built-in objects; its v1_36 feature selects the newest Kubernetes version it supports. The other dependencies do supporting work: schemars generates JSON schemas, serde and serde_json convert objects to and from JSON, tokio runs our asynchronous code, futures gives us operations on streams, thiserror derives our error type, and tracing writes the controller’s log. Finally, default-run lets us start the controller with a plain cargo run -p gpujob.

Teaching the API server a new kind of object

In chapter 1 our jobs were records with a GPU count. A real cluster has no such kind of object: it knows about pods, services, deployments, and the other built-in kinds, and that is all. Fortunately, the API server can be taught new kinds. We describe the new kind in a CustomResourceDefinition (CRD): its API group and version, its name, and the schema of its fields. Once we install the CRD, the API server stores, validates, lists, and watches objects of our new kind exactly as it does its built-in kinds. That means everything from chapter 1 applies to our jobs too: they get resource versions, clients can list and watch them, and writes can be conditional.

We will call our kind GpuJob, in the API group gpucloud.dev, version v1. Its spec holds a container image, an optional command, and a GPU count. Its status holds a phase, a message, and the generation it describes. In both languages, we write the kind as ordinary types and let a tool generate the CRD from them.

In Go

A Go API package usually has two files. The first one names the group and version, and it registers our types in a scheme. A scheme is a table that maps Go types to the API kinds they represent. A client needs it to know that JSON with "kind": "GpuJob" should become a GpuJob value, and which URL to use when it sends one to the server.

Listing 1.44 The group, the version, and the scheme. code/go/gpujob/api/v1/groupversion.go

// Package v1 defines version v1 of the GpuJob custom resource in the API group
// gpucloud.dev: a request to run one container with a number of GPUs, and the
// status that the controller reports back.
//
// The comments on the types below become the field descriptions in the
// generated CustomResourceDefinition, which kubectl explain prints. The
// +kubebuilder comments are markers that controller-gen reads.
//
// +kubebuilder:object:generate=true
// +groupName=gpucloud.dev
package v1

import (
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/scheme"
)

//go:generate controller-gen object crd paths=. output:crd:dir=../../config/crd

// GroupVersion is the API group and version of the types in this package.
var GroupVersion = schema.GroupVersion{Group: "gpucloud.dev", Version: "v1"}

// SchemeBuilder registers the types in this package with a runtime.Scheme.
var SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}

// AddToScheme adds the types in this package to a runtime.Scheme, so that a
// client can encode and decode them.
var AddToScheme = SchemeBuilder.AddToScheme

The second file defines the types themselves, in three parts: the spec, the status, and the object that holds both.

Listing 1.45 The spec and the phases a job passes through. code/go/gpujob/api/v1/types.go

// GpuJobSpec is a request to run one container with a number of GPUs.
type GpuJobSpec struct {
	// Container image to run.
	Image string `json:"image"`
	// Command to run in the container. When empty, the image's own command runs.
	// +optional
	Command []string `json:"command,omitempty"`
	// Number of GPUs the container requests.
	// +kubebuilder:validation:Minimum=0
	GPUs int32 `json:"gpus"`
}

// Phase is the progress of a job, following the phase of its pod.
// +kubebuilder:validation:Enum=Pending;Running;Succeeded;Failed
type Phase string

// The phases a job moves through.
const (
	// PhasePending means the pod is waiting to be scheduled or for its containers to start.
	PhasePending Phase = "Pending"
	// PhaseRunning means the pod's container is running.
	PhaseRunning Phase = "Running"
	// PhaseSucceeded means the container exited with status 0.
	PhaseSucceeded Phase = "Succeeded"
	// PhaseFailed means the container exited with a non-zero status.
	PhaseFailed Phase = "Failed"
)

Listing 1.46 The status. code/go/gpujob/api/v1/types.go

// GpuJobStatus is the state of a job as the controller last observed it.
type GpuJobStatus struct {
	// Progress of the job's pod.
	Phase Phase `json:"phase"`
	// The scheduler's reason, when the pod cannot be placed on a node.
	// +optional
	Message string `json:"message,omitempty"`
	// The metadata.generation of the spec that this status describes.
	// +optional
	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
}

Listing 1.47 The object, the list, and their registration. code/go/gpujob/api/v1/types.go

// GpuJob runs one container with a number of GPUs.
//
// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:shortName=gj
// +kubebuilder:printcolumn:name="GPUs",type=integer,JSONPath=".spec.gpus"
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=".status.phase"
type GpuJob struct {
	metav1.TypeMeta `json:",inline"`
	// Standard object metadata: name, namespace, labels, resource version.
	metav1.ObjectMeta `json:"metadata,omitempty"`
	// What the user asks for.
	Spec GpuJobSpec `json:"spec"`
	// What the controller has observed.
	// +optional
	Status GpuJobStatus `json:"status,omitempty"`
}

// GpuJobList is a list of GpuJob objects, as the API server returns it.
//
// +kubebuilder:object:root=true
type GpuJobList struct {
	metav1.TypeMeta `json:",inline"`
	// Standard list metadata, including the list's resource version.
	metav1.ListMeta `json:"metadata,omitempty"`
	// The jobs in the list.
	Items []GpuJob `json:"items"`
}

func init() {
	SchemeBuilder.Register(&GpuJob{}, &GpuJobList{})
}

The comments that begin with +kubebuilder: are markers. They are instructions for controller-gen, a code generator from the Kubebuilder project, which reads our source code and writes the CRD for us. Let us go through the ones we use:

• object:root=true marks GpuJob and GpuJobList as complete API objects, as opposed to structs that only appear inside other objects.
• subresource:status turns on the status subresource, which we explain below.
• resource:shortName=gj lets us type kubectl get gj instead of kubectl get gpujobs.
• The two printcolumn markers add the GPU count and the phase to the table that kubectl get prints.
• The validation markers become rules in the schema. The API server enforces them, so a job with a negative GPU count or an unknown phase is rejected before our controller ever sees it.

The ordinary comments on our types and fields matter as well: controller-gen copies them into the CRD as field descriptions, and kubectl explain gpujob.spec shows them to users.

Listing 1.48 Working out a status from a pod. code/go/gpujob/api/v1/types.go

// StatusFromPod derives a job's status from its pod. generation is the job's
// metadata.generation, recorded as the generation that the status describes.
func StatusFromPod(pod *corev1.Pod, generation int64) GpuJobStatus {
	status := GpuJobStatus{Phase: PhasePending, ObservedGeneration: generation}

	switch pod.Status.Phase {
	case corev1.PodRunning:
		status.Phase = PhaseRunning
	case corev1.PodSucceeded:
		status.Phase = PhaseSucceeded
	case corev1.PodFailed:
		status.Phase = PhaseFailed
	}

	// A pod that cannot be placed explains why in its PodScheduled condition.
	for _, c := range pod.Status.Conditions {
		if c.Type == corev1.PodScheduled && c.Status == corev1.ConditionFalse {
			status.Message = c.Message
		}
	}

	return status
}

StatusFromPod works out a job’s status from its pod. The pod’s phase maps directly onto our Phase. There is one extra piece of information worth passing on. When the scheduler cannot find a machine for a pod, it sets the pod’s PodScheduled condition to False and writes the reason in the condition’s message. StatusFromPod copies that message into the job’s status, so that a user who runs kubectl get gpujob can see why the job is still waiting.

The go:generate line in groupversion.go tells Go how to run controller-gen. It writes two files for us. The first is zz_generated.deepcopy.go, with a DeepCopy method for each type. Every Kubernetes API type needs these, because the shared caches described at the end of chapter 1 hand out pointers, and a controller must copy an object before changing it. The second file is the CRD itself, in gpujob/config/crd. Both generated files are checked into the repository, so we only need to run the generator again when we change the types:

$ go install sigs.k8s.io/controller-tools/cmd/controller-gen@v0.22.0
$ go generate ./gpujob/api/...
$ kubectl apply -f gpujob/config/crd/
customresourcedefinition.apiextensions.k8s.io/gpujobs.gpucloud.dev configured

In Rust

In Rust, there is no separate generator to run. kube-rs builds the CRD while our program compiles, from derive macros on our types:

Listing 1.49 The spec, with the attributes that describe the kind. code/rust/gpujob/src/lib.rs

/// A request to run one container with a number of GPUs.
///
/// The `CustomResource` derive generates the `GpuJob` type, which holds this
/// spec together with the object's metadata and an optional `GpuJobStatus`.
#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
#[kube(
    group = "gpucloud.dev",
    version = "v1",
    kind = "GpuJob",
    namespaced,
    status = "GpuJobStatus",
    shortname = "gj",
    printcolumn = r#"{"name":"GPUs","type":"integer","jsonPath":".spec.gpus"}"#,
    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#
)]
pub struct GpuJobSpec {
    /// Container image to run.
    pub image: String,
    /// Command to run in the container. When empty, the image's own command runs.
    #[serde(default)]
    pub command: Vec<String>,
    /// Number of GPUs the container requests.
    // Kubernetes schemas support int32 and int64, so the u32 is published as a
    // non-negative int32.
    #[schemars(with = "i32", range(min = 0))]
    pub gpus: u32,
}

Listing 1.50 The status and the phases. code/rust/gpujob/src/lib.rs

/// The state of a job as the controller last observed it.
#[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct GpuJobStatus {
    /// Progress of the job's pod.
    pub phase: Phase,
    /// The scheduler's reason, when the pod cannot be placed on a node.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    /// The `metadata.generation` of the spec that this status describes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub observed_generation: Option<i64>,
}

/// Progress of a job, following the phase of its pod.
#[derive(Deserialize, Serialize, Clone, Copy, Debug, Default, PartialEq, Eq, JsonSchema)]
pub enum Phase {
    /// The pod is waiting to be scheduled or for its containers to start.
    #[default]
    Pending,
    /// The pod's container is running.
    Running,
    /// The container exited with status 0.
    Succeeded,
    /// The container exited with a non-zero status.
    Failed,
}

Listing 1.51 Working out a status from a pod. code/rust/gpujob/src/lib.rs

impl GpuJobStatus {
    /// Derives a job's status from its pod. `generation` is the job's
    /// `metadata.generation`, recorded as the generation this status describes.
    #[must_use]
    pub fn from_pod(pod: &Pod, generation: Option<i64>) -> Self {
        let status = pod.status.as_ref();
        let phase = match status.and_then(|s| s.phase.as_deref()) {
            Some("Running") => Phase::Running,
            Some("Succeeded") => Phase::Succeeded,
            Some("Failed") => Phase::Failed,
            _ => Phase::Pending,
        };
        // A pod that cannot be placed explains why in its PodScheduled condition.
        let message = status
            .and_then(|s| s.conditions.as_ref())
            .into_iter()
            .flatten()
            .find(|c| c.type_ == "PodScheduled" && c.status == "False")
            .and_then(|c| c.message.clone());
        Self {
            phase,
            message,
            observed_generation: generation,
        }
    }
}

The CustomResource derive generates a GpuJob type for us. It holds our spec together with the object’s metadata and an optional status, the same shape as the Go struct we wrote by hand. The #[kube(...)] attribute does the job of the Go markers. group and version place the kind at gpucloud.dev/v1. namespaced makes each job belong to a namespace, a named partition of the cluster’s objects, so that two teams can each have a job called train. status turns on the status subresource, and shortname and printcolumn do the same as their Go counterparts. The /// comments become the field descriptions, and our Phase enum becomes a string field whose allowed values are the variant names.

One attribute deserves a closer look. The schema that kube-rs generates for a u32 has the format uint32, which the API server does not recognise, and it complains with a warning every time the CRD is applied. The attribute #[schemars(with = "i32", range(min = 0))] publishes the field as a non-negative 32-bit integer instead, which is exactly the schema the Go field of type int32 with its Minimum=0 marker produces, while the Rust field itself stays unsigned. from_pod is the Rust version of StatusFromPod.

A tiny program prints the CRD, and we pipe it straight into kubectl:

Listing 1.52 Printing the CRD. code/rust/gpujob/src/bin/crdgen.rs

//! Prints the `CustomResourceDefinition` for `GpuJob` as JSON, ready for kubectl:
//!
//! ```console
//! cargo run -p gpujob --bin crdgen | kubectl apply -f -
//! ```

use gpujob::GpuJob;
use kube::CustomResourceExt;

/// Generates the definition from the `GpuJob` type and prints it.
fn main() -> serde_json::Result<()> {
    println!("{}", serde_json::to_string_pretty(&GpuJob::crd())?);

    Ok(())
}

$ cargo run -p gpujob --bin crdgen | kubectl apply -f -
customresourcedefinition.apiextensions.k8s.io/gpujobs.gpucloud.dev created

Our Go and Rust definitions describe the same kind with the same fields. We can install either one, and either controller can then serve the jobs.

The status subresource

Why did we turn on the status subresource? Think about two writers touching the same job: a user who edits the spec, and our controller, which writes the status. Without the subresource, both write the whole object, and whoever writes second can overwrite the other’s part with an older copy. With the subresource, the status gets its own endpoint, /status. The API server takes only status fields from writes to /status, and only spec fields from writes to the main endpoint, so the user and the controller can no longer step on each other.

The subresource has a second effect. The API server now increments metadata.generation only when the spec changes. That gives our status field observedGeneration its meaning: our controller copies the generation it acted on into it, and anyone comparing the two numbers can tell whether the latest spec change has been handled yet.

Watching the jobs

In chapter 1, our reflector listed the objects, watched from the list’s version, and listed again when the history ran out. Both libraries give us that loop ready-made. To see exactly what it delivers, we will write a small program in each language that prints every event it receives.

In Go, the loop comes packaged inside an informer, just like in client-go: controller-runtime’s cache package creates one informer per kind of object. To hear about changes, we register an event handler: a set of three functions that the informer calls whenever it adds, updates, or deletes an object in its cache. Our handlers simply print what they receive:

Listing 1.53 Describing an object for the log. code/go/gpujob/cmd/watch-jobs/main.go

// namespace is the namespace whose jobs the program watches.
const namespace = "default"

// describe formats a job as its name, GPU count, and resource version.
func describe(obj any) string {
	// A deletion that a relist discovered arrives wrapped in a tombstone.
	if tombstone, ok := obj.(toolscache.DeletedFinalStateUnknown); ok {
		obj = tombstone.Obj
	}

	job := obj.(*gpujobv1.GpuJob)

	return fmt.Sprintf("%s gpus=%d v%s", job.Name, job.Spec.GPUs, job.ResourceVersion)
}

Listing 1.54 Starting the informer and registering the handlers. code/go/gpujob/cmd/watch-jobs/main.go

// run does the work of main and returns the first error it meets.
func run() error {
	scheme := runtime.NewScheme()

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		return err
	}

	informers, err := cache.New(ctrl.GetConfigOrDie(), cache.Options{
		Scheme:            scheme,
		DefaultNamespaces: map[string]cache.Config{namespace: {}},
	})

	if err != nil {
		return err
	}

	ctx := ctrl.SetupSignalHandler()
	informer, err := informers.GetInformer(ctx, &gpujobv1.GpuJob{})

	if err != nil {
		return err
	}

	registration, err := informer.AddEventHandler(toolscache.ResourceEventHandlerDetailedFuncs{
		AddFunc: func(obj any, isInInitialList bool) {
			if isInInitialList {
				fmt.Println("  listed  ", describe(obj))
			} else {
				fmt.Println("  changed ", describe(obj))
			}
		},
		UpdateFunc: func(_, obj any) { fmt.Println("  changed ", describe(obj)) },
		DeleteFunc: func(obj any) { fmt.Println("  deleted ", describe(obj)) },
	})

	if err != nil {
		return err
	}

	fmt.Println("list started")
	go informers.Start(ctx)

	if !toolscache.WaitForCacheSync(ctx.Done(), registration.HasSynced) {
		return fmt.Errorf("informer did not sync")
	}

	fmt.Println("list complete, watching")
	<-ctx.Done()

	return nil
}

The add handler gets an extra flag, isInInitialList, which is true for objects that came from the first list, so we can tell the objects that already existed from the ones created later. The handler’s registration has a HasSynced method that becomes true once our handler has seen the whole initial list, and WaitForCacheSync waits for that moment before we print “list complete”. The delete handler may receive a DeletedFinalStateUnknown value instead of a job. That is the marker from chapter 1 for a deletion that the informer only discovered by relisting, and describe unwraps it.

In Rust, kube-rs hands us the loop directly as a stream: a sequence of values that arrive over time, which we read one at a time with await. It is the asynchronous counterpart of an iterator. The function watcher returns a stream of Event values for one kind of object:

Listing 1.55 Printing every event, in Rust. code/rust/gpujob/src/bin/watch_jobs.rs

//! Prints the event stream that a watcher produces for `GpuJob` objects in the
//! current namespace: the initial list, one object at a time, and then one
//! event per change. Stop it with Ctrl-C.
//!
//! Run it from `code/rust` with `cargo run -p gpujob --bin watch_jobs`.

use futures::TryStreamExt;
use gpujob::GpuJob;
use kube::runtime::watcher::{self, Event, watcher};
use kube::{Api, Client, ResourceExt};

/// Formats a job as its name, GPU count, and resource version.
fn describe(job: &GpuJob) -> String {
    let version = job.resource_version().unwrap_or_default();

    format!("{} gpus={} v{version}", job.name_any(), job.spec.gpus)
}

/// Watches `GpuJob` objects and prints one line per event.
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let jobs: Api<GpuJob> = Api::default_namespaced(Client::try_default().await?);
    let mut events = std::pin::pin!(watcher(jobs, watcher::Config::default()));

    while let Some(event) = events.try_next().await? {
        match event {
            Event::Init => println!("list started"),
            Event::InitApply(job) => println!("  listed   {}", describe(&job)),
            Event::InitDone => println!("list complete, watching"),
            Event::Apply(job) => println!("  changed  {}", describe(&job)),
            Event::Delete(job) => println!("  deleted  {}", describe(&job)),
        }
    }

    Ok(())
}

The Event enum spells out the list-then-watch sequence we reasoned about in chapter 1. Init, then one InitApply for each object that existed at the time of the list, then InitDone, together make up a list. After that, each event is an Apply for an object that was created or changed, or a Delete. If the watch has to start over, for example after 410 Gone, the stream begins a new Init sequence by itself. A consumer that rebuilds its state at every Init therefore gets the same result as the Replace method of our chapter 1 cache. kube-rs provides exactly such a consumer and calls it a reflector: it applies the events to a store that the rest of the program can read.

We started each program before any jobs existed and left it running while we did everything else in this chapter. The Go program printed:

$ go run ./gpujob/cmd/watch-jobs
list started
list complete, watching
  changed  smoke-test gpus=0 v41900
  changed  train gpus=1 v41901
  changed  smoke-test gpus=0 v41904
  changed  train gpus=1 v41909
  changed  smoke-test gpus=0 v41920
  changed  train gpus=1 v41923
  changed  train gpus=1 v41931
  changed  smoke-test gpus=0 v41960
  changed  train gpus=2 v41972
  changed  train gpus=2 v41977
  changed  train gpus=2 v41979
  deleted  smoke-test gpus=0 v41992

And the Rust program printed:

$ cargo run -p gpujob --bin watch_jobs
list started
list complete, watching
  changed  smoke-test gpus=0 v656
  changed  train gpus=1 v657
  changed  train gpus=1 v661
  changed  smoke-test gpus=0 v662
  changed  train gpus=1 v676
  changed  smoke-test gpus=0 v684
  changed  train gpus=1 v688
  changed  smoke-test gpus=0 v722
  changed  train gpus=2 v731
  changed  train gpus=2 v737
  changed  train gpus=2 v739
  deleted  smoke-test gpus=0 v752

When we started them again later, with train still in the cluster, each one listed the job before it started watching. In Go:

$ go run ./gpujob/cmd/watch-jobs
list started
  listed   train gpus=2 v41979
list complete, watching

In Rust:

$ cargo run -p gpujob --bin watch_jobs
list started
  listed   train gpus=2 v739
list complete, watching

The resource versions are etcd’s revision numbers. Since etcd counts every write in the cluster, including those made by Kubernetes’ own components, the numbers of one job jump by more than one between its changes. The Go run happened later in the cluster’s life, so its numbers are larger. Most of the changed lines are our controller writing a new status: the object changes, even though its spec stays the same.

The controller

Now we can write the controller. It will run each GpuJob as a pod and report the pod’s progress in the job’s status. Both versions follow the same plan, so we walk through them together, one part at a time, and print the complete programs at the end of the chapter.

The pod for a job

Listing 1.56 The pod a job should have, in Go. code/go/gpujob/cmd/controller/main.go

// manager is the field manager name under which the controller writes with
// server-side apply.
const manager = "gpujob-controller"

// gpu is the extended resource through which a container requests GPUs.
const gpu corev1.ResourceName = "nvidia.com/gpu"

// desiredPod builds the pod that should exist for job, as an apply
// configuration: a partial object that lists only the fields this controller
// sets.
//
// The owner reference makes Kubernetes delete the pod together with the job,
// and it lets the controller map each pod event to the job that owns the pod.
func desiredPod(job *gpujobv1.GpuJob) *corev1ac.PodApplyConfiguration {
	gpus := corev1.ResourceList{gpu: *resource.NewQuantity(int64(job.Spec.GPUs), resource.DecimalSI)}
	container := corev1ac.Container().
		WithName("job").
		WithImage(job.Spec.Image).
		WithResources(corev1ac.ResourceRequirements().WithLimits(gpus))
	if len(job.Spec.Command) > 0 {
		container.WithCommand(job.Spec.Command...)
	}

	owner := metav1ac.OwnerReference().
		WithAPIVersion(gpujobv1.GroupVersion.String()).
		WithKind("GpuJob").
		WithName(job.Name).
		WithUID(job.UID).
		WithController(true).
		WithBlockOwnerDeletion(true)
	return corev1ac.Pod(job.Name, job.Namespace).
		WithOwnerReferences(owner).
		WithSpec(corev1ac.PodSpec().
			WithRestartPolicy(corev1.RestartPolicyNever).
			WithContainers(container))
}

// podGPUs returns the number of GPUs that an existing pod's container
// requests, read from its limits, and whether the pod has a GPU limit.
func podGPUs(pod *corev1.Pod) (int32, bool) {
	if len(pod.Spec.Containers) == 0 {
		return 0, false
	}

	quantity, ok := pod.Spec.Containers[0].Resources.Limits[gpu]

	return int32(quantity.Value()), ok
}

Listing 1.57 The pod a job should have, in Rust. code/rust/gpujob/src/bin/controller.rs

/// The field manager name under which the controller writes with server-side apply.
const MANAGER: &str = "gpujob-controller";

/// The extended resource through which a container requests GPUs.
const GPU: &str = "nvidia.com/gpu";

/// Everything a reconcile can fail with.
#[derive(Debug, thiserror::Error)]
enum Error {
    /// A request to the API server failed.
    #[error("kubernetes API: {0}")]
    Kube(#[from] kube::Error),
    /// The status could not be serialized into a patch.
    #[error("serializing status: {0}")]
    Json(#[from] serde_json::Error),
}

/// State shared by every reconcile.
struct Context {
    /// Client for the cluster's API server.
    client: Client,
}

/// Builds the pod that should exist for `job`.
///
/// The owner reference makes Kubernetes delete the pod together with the job,
/// and it lets the controller map each pod event to the job that owns the pod.
fn desired_pod(job: &GpuJob) -> Pod {
    let gpus = BTreeMap::from([(GPU.to_string(), Quantity(job.spec.gpus.to_string()))]);
    Pod {
        metadata: ObjectMeta {
            name: Some(job.name_any()),
            owner_references: job.controller_owner_ref(&()).map(|owner| vec![owner]),
            ..Default::default()
        },
        spec: Some(PodSpec {
            restart_policy: Some("Never".into()),
            containers: vec![Container {
                name: "job".into(),
                image: Some(job.spec.image.clone()),
                command: Some(job.spec.command.clone()).filter(|c| !c.is_empty()),
                resources: Some(ResourceRequirements {
                    limits: Some(gpus),
                    ..Default::default()
                }),
                ..Default::default()
            }],
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// Returns the number of GPUs that an existing pod's container requests, read
/// from its limits, or `None` if the pod has no GPU limit.
fn pod_gpus(pod: &Pod) -> Option<u32> {
    let container = pod.spec.as_ref()?.containers.first()?;
    let limits = container.resources.as_ref()?.limits.as_ref()?;

    limits.get(GPU)?.0.parse().ok()
}

desiredPod (desired_pod in Rust) builds the pod that a job should have. The interesting part is how the pod asks for a GPU. It puts nvidia.com/gpu in its resource limits, just as it would put CPU or memory there. Kubernetes calls a resource like this an extended resource: a name that Kubernetes itself does not understand, but whose quantity a machine can advertise as a whole number, and which the scheduler counts like any other resource. The name nvidia.com/gpu is advertised by the NVIDIA device plugin, a small program on each GPU machine that counts the GPUs and reports them to Kubernetes; Appendix A shows it at work. Extended resources can only be requested in whole units, and they are requested through limits. One unit of nvidia.com/gpu is, by default, one whole physical GPU; chapter 2 examines what that includes and how a GPU can be divided.

The pod also carries an owner reference, a note in its metadata that says “this pod belongs to that job”. It has two uses. When the job is deleted, Kubernetes’ garbage collector, a controller built into every cluster, sees the note and deletes the pod as well, so we never have to clean up after deleted jobs ourselves. And because the note has its controller flag set, both libraries can take any event about the pod and work out which job it belongs to, which is how our controller hears about its pods.

The two languages build the pod differently. The Go version builds an apply configuration, a type from client-go in which every field is optional and is filled in through With methods. The value ends up containing exactly the fields our controller cares about and nothing else, which matters for the kind of write we are about to do. The Rust version builds an ordinary Pod value and leaves everything else at its default; serde leaves the empty fields out when it turns the pod into JSON, which has the same effect.

podGPUs (pod_gpus in Rust) reads the GPU count back out of an existing pod. In Go it returns the count and a flag saying whether the pod had a GPU limit at all. In Rust it returns an Option<u32>, and each ? bails out with None as soon as any field along the way is missing.

Reconciling a job

Listing 1.58 Reconciling one job, in Go. code/go/gpujob/cmd/controller/main.go

// GpuJobReconciler makes the cluster match each GpuJob.
type GpuJobReconciler struct {
	// Client reads objects from the manager's cache and writes to the API server.
	client.Client
}

// Reconcile makes the cluster match one job: it ensures the job's pod exists
// with the requested GPUs, and writes the pod's progress into the job's status.
func (r *GpuJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := ctrl.Log.WithName("controller").WithValues("name", req.Name)

	var job gpujobv1.GpuJob

	if err := r.Get(ctx, req.NamespacedName, &job); err != nil {
		// A job deleted since it was queued needs nothing: its pod goes with it.
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// The resources of a pod are fixed when it is created. When the job asks
	// for a different number of GPUs, delete the pod and come back after it is gone.
	var pod corev1.Pod
	err := r.Get(ctx, req.NamespacedName, &pod)

	switch {
	case err == nil:
		if gpus, ok := podGPUs(&pod); !ok || gpus != job.Spec.GPUs {
			log.Info("GPU count changed, replacing pod")

			if err := r.Delete(ctx, &pod); client.IgnoreNotFound(err) != nil {
				return ctrl.Result{}, err
			}

			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	case !apierrors.IsNotFound(err):
		return ctrl.Result{}, err
	}

	if err := r.Apply(ctx, desiredPod(&job), client.FieldOwner(manager), client.ForceOwnership); err != nil {
		return ctrl.Result{}, err
	}

	// A pod that was just created is not in the cache yet; its zero value
	// yields the Pending phase, and the pod's creation event triggers the
	// reconcile that reports its real progress.
	status := gpujobv1.StatusFromPod(&pod, job.Generation)

	if job.Status != status {
		log.Info("status changed", "phase", status.Phase, "message", status.Message)
		patch := client.MergeFrom(job.DeepCopy())
		job.Status = status

		if err := r.Status().Patch(ctx, &job, patch); err != nil {
			return ctrl.Result{}, err
		}
	}

	// Changes to the pod trigger the next reconcile through Owns.
	return ctrl.Result{}, nil
}

Listing 1.59 Reconciling one job, in Rust. code/rust/gpujob/src/bin/controller.rs

/// Makes the cluster match one job: ensures its pod exists with the requested
/// GPUs, and writes the pod's progress into the job's status.
async fn reconcile(job: Arc<GpuJob>, ctx: Arc<Context>) -> Result<Action, Error> {
    let namespace = job.namespace().unwrap_or_default();
    let name = job.name_any();
    let pods: Api<Pod> = Api::namespaced(ctx.client.clone(), &namespace);
    let jobs: Api<GpuJob> = Api::namespaced(ctx.client.clone(), &namespace);

    // The resources of a pod are fixed when it is created. When the job asks
    // for a different number of GPUs, delete the pod and come back after it is gone.
    if let Some(pod) = pods.get_opt(&name).await?
        && pod_gpus(&pod) != Some(job.spec.gpus)
    {
        tracing::info!(%name, "GPU count changed, replacing pod");
        pods.delete(&name, &DeleteParams::default()).await?;

        return Ok(Action::requeue(Duration::from_secs(2)));
    }

    let apply = PatchParams::apply(MANAGER).force();
    let pod = pods
        .patch(&name, &apply, &Patch::Apply(desired_pod(&job)))
        .await?;

    let status = GpuJobStatus::from_pod(&pod, job.metadata.generation);

    if job.status.as_ref() != Some(&status) {
        let message = status.message.as_deref().unwrap_or("");
        tracing::info!(%name, phase = ?status.phase, message, "status changed");
        let patch = serde_json::json!({
            "apiVersion": GpuJob::api_version(&()),
            "kind": GpuJob::kind(&()),
            "status": status,
        });
        jobs.patch_status(&name, &apply, &Patch::Apply(patch))
            .await?;
    }

    // Changes to the pod trigger the next reconcile through `owns`.
    Ok(Action::await_change())
}

/// Decides what happens after a failed reconcile: log it and retry in five seconds.
#[allow(clippy::needless_pass_by_value)] // the signature is fixed by `Controller::run`
fn error_policy(job: Arc<GpuJob>, error: &Error, _ctx: Arc<Context>) -> Action {
    tracing::warn!(name = %job.name_any(), %error, "reconcile failed");

    Action::requeue(Duration::from_secs(5))
}

Our reconcile function receives a job and brings the cluster into line with it in three steps.

The first step covers a change that cannot be made in place. Kubernetes fixes a pod’s resources when the pod is created, so if the user changes the job’s GPU count, we cannot simply edit the pod. Instead we delete the pod and ask to be called again in two seconds, by which time it should be gone. The Go version asks for that with ctrl.Result{RequeueAfter: 2 * time.Second}, and the Rust version with Action::requeue. The Rust condition uses a let chain, if let ... && ..., available since the 2024 edition, to check for the pod and compare its GPU count in one expression.

The second step creates or updates the pod with a single call, using server-side apply. We send the pod we want, and the API server merges it into whatever it has stored. The server also records which fields each writer set; a writer in this sense is called a field manager, and our controller calls itself gpujob-controller. Applying the same object twice changes nothing, which makes server-side apply a natural fit for a level-triggered controller: we can apply on every single reconcile, without first checking whether the pod exists. We also force ownership, with client.ForceOwnership in Go and force() in Rust. If some other writer has set a field that we also set, we take that field over, which is what we want from a controller that is the only authority over its pods.

The third step works out the job’s status and writes it through the status subresource, but only if it differs from the status already stored. Skipping unchanged writes is more than a saving. Every write produces a watch event, every event causes a reconcile, and a controller that wrote on every reconcile would keep waking itself up forever. The Go version writes the status with a merge patch, which sends only the fields that differ from a copy of the job we saved before changing it. The Rust version applies the status with server-side apply, just as it does the pod.

The two versions also read the pod differently, and the difference is worth understanding. controller-runtime’s client reads from the informer caches that the manager keeps, so the Go Get of the pod is a lookup in memory, exactly like the cache reads in chapter 1. A pod we have only just created is not in the cache yet; in that case the Go version computes the status from an empty pod, which comes out as Pending, and the pod’s creation event soon triggers another reconcile that reports the real state. kube-rs’s Api, on the other hand, talks to the API server directly, so the Rust get_opt is a real GET request. That is fine for our small controller. A Rust controller that looks after thousands of pods would keep a reflector store of pods in its Context and read them from memory, just as the Go version does.

When a reconcile fails, the Go version returns the error, and controller-runtime puts the job back in the queue with the growing delay we described at the end of chapter 1. The Rust version returns the error to kube-rs, which asks our error_policy function what to do; ours logs the error and tries again in five seconds.

Starting the controller

Listing 1.60 Starting the controller, in Go. code/go/gpujob/cmd/controller/main.go

// main builds a manager, registers the reconciler, and runs until Ctrl-C.
func main() {
	ctrl.SetLogger(zap.New(zap.UseDevMode(true))) // readable console output
	setup := ctrl.Log.WithName("setup")

	scheme := runtime.NewScheme()

	if err := clientgoscheme.AddToScheme(scheme); err != nil {
		setup.Error(err, "registering built-in types")
		os.Exit(1)
	}

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		setup.Error(err, "registering GpuJob")
		os.Exit(1)
	}

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:  scheme,
		Metrics: metricsserver.Options{BindAddress: "0"}, // no metrics endpoint
	})

	if err != nil {
		setup.Error(err, "creating manager")
		os.Exit(1)
	}

	err = ctrl.NewControllerManagedBy(mgr).
		For(&gpujobv1.GpuJob{}).
		Owns(&corev1.Pod{}).
		Complete(&GpuJobReconciler{Client: mgr.GetClient()})
	if err != nil {
		setup.Error(err, "creating controller")
		os.Exit(1)
	}

	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setup.Error(err, "running manager")
		os.Exit(1)
	}
}

Listing 1.61 Starting the controller, in Rust. code/rust/gpujob/src/bin/controller.rs

/// Starts the controller and runs it until Ctrl-C.
#[tokio::main]
async fn main() -> Result<(), kube::Error> {
    tracing_subscriber::fmt()
        .with_env_filter("controller=info,kube=warn")
        .with_target(false)
        .with_ansi(std::io::IsTerminal::is_terminal(&std::io::stdout()))
        .init();
    let client = Client::try_default().await?;
    let jobs: Api<GpuJob> = Api::all(client.clone());
    let pods: Api<Pod> = Api::all(client.clone());

    Controller::new(jobs, watcher::Config::default())
        .owns(pods, watcher::Config::default())
        .shutdown_on_signal()
        .run(reconcile, error_policy, Arc::new(Context { client }))
        .for_each(|result| async move {
            if let Err(error) = result {
                tracing::debug!(%error, "reconcile error");
            }
        })
        .await;
    Ok(())
}

In Go, everything a controller process shares lives in a manager: the connection to the cluster, the scheme, the informer caches, and the client that reads from them. ctrl.NewControllerManagedBy(mgr) builds our controller inside the manager, and each call in the chain wires up one piece. For(&GpuJob{}) watches jobs and queues the name of each job that changes. Owns(&corev1.Pod{}) watches pods and queues the name of each pod’s owning job, using the owner reference we set. Complete registers our reconciler. Then mgr.Start starts the caches, waits for their first lists, and runs the workers until the process receives a signal such as Ctrl-C. We switch off the metrics endpoint that a manager starts by default, since we do not need it here.

In Rust, Controller::new starts a watcher and a store for GpuJob objects, and owns adds a second watcher, for pods, that queues each pod’s owning job in the same way. run starts the work queue and the workers and gives us back a stream with one item per finished reconcile, and for_each keeps reading that stream until the program stops. shutdown_on_signal ends the stream when we press Ctrl-C.

In both libraries, the work queue behaves exactly like the one we built in chapter 1: a job waits in it at most once, and a job that one worker is reconciling is held back from all the others until that worker is done.

Running it

Figure 1.2 shows what one reconcile writes for a job named train, and what Kubernetes does with the pod.

We will test the controller with two jobs. smoke-test asks for no GPUs and runs a short shell command, which tells us the basic machinery works. train asks for one GPU and runs gpu-check, a small CUDA program from Appendix A that adds two vectors on the GPU and prints which device it used.

Listing 1.62 The two jobs. code/manifests/gpujobs.yaml

apiVersion: gpucloud.dev/v1
kind: GpuJob
metadata:
  name: smoke-test
spec:
  image: busybox:1.37
  command: ["sh", "-c", "echo hello from the job; sleep 20"]
  gpus: 0
---
apiVersion: gpucloud.dev/v1
kind: GpuJob
metadata:
  name: train
spec:
  image: gpu-check:1
  gpus: 1

We start the Go controller from code/go, create the jobs, and check on them:

$ go run ./gpujob/cmd/controller
$ kubectl apply -f ../manifests/gpujobs.yaml
$ kubectl get gpujobs
NAME         GPUS   PHASE
smoke-test   0      Succeeded
train        1      Succeeded
$ kubectl logs train
device: Orin, compute 8.7, 8 SMs
kernel: no error, out[0] = 3.0, out[n-1] = 3.0

The log of train shows our CUDA kernel running on the cluster’s GPU, inside the pod our controller created. After its startup messages, the Go controller logged the following lines, shown here with the date and time zone trimmed from the timestamps:

15:12:37  INFO  controller  status changed  {"name": "smoke-test", "phase": "Pending", "message": ""}
15:12:37  INFO  controller  status changed  {"name": "train", "phase": "Pending", "message": ""}
15:12:38  INFO  controller  status changed  {"name": "smoke-test", "phase": "Running", "message": ""}
15:12:38  INFO  controller  status changed  {"name": "train", "phase": "Running", "message": ""}
15:12:40  INFO  controller  status changed  {"name": "train", "phase": "Succeeded", "message": ""}
15:13:00  INFO  controller  status changed  {"name": "smoke-test", "phase": "Succeeded", "message": ""}

The Rust controller runs from code/rust, and the same steps give the same result:

$ cargo run -p gpujob
$ kubectl apply -f ../manifests/gpujobs.yaml
$ kubectl get gpujobs
NAME         GPUS   PHASE
smoke-test   0      Succeeded
train        1      Succeeded

14:01:15.949  INFO status changed name=train phase=Pending
14:01:15.954  INFO status changed name=smoke-test phase=Pending
14:01:15.985  INFO status changed name=smoke-test phase=Pending
14:01:17.124  INFO status changed name=train phase=Running
14:01:18.504  INFO status changed name=smoke-test phase=Running
14:01:18.959  INFO status changed name=train phase=Succeeded
14:01:39.957  INFO status changed name=smoke-test phase=Succeeded

In both runs, each job moved from Pending to Running to Succeeded along with its pod, and every step reached our controller through the pod watch that Owns or owns set up. train finished about two seconds after its pod was created, and smoke-test finished after its twenty-second sleep.

Look at the Rust log once more: smoke-test reported Pending twice, 31 milliseconds apart. Here is what happened. The first reconcile wrote the status. The pod’s creation then triggered a second reconcile before our own status write had come back through the watch, so the job in the cache still had no status, the comparison saw a difference, and the controller wrote the same status again. This is the read-your-own-writes gap from the end of chapter 1. Here it costs one redundant write, which server-side apply turns into a no-op.

Two writers, one object

Our controllers never send a resource version when they write, because server-side apply does not need one. Many clients do, though: a client that reads an object, changes it, and sends the whole object back with an update includes the version it read, and the API server applies the same check as our Update in chapter 1. To see that check on a real server, we will write a program in which two clients read the train job and then both try to write it back.

In Go:

Listing 1.63 Two clients and one refused write, in Go. code/go/gpujob/cmd/conflict/main.go

// run does the work of main and returns the first unexpected error.
func run(ctx context.Context) error {
	scheme := runtime.NewScheme()

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		return err
	}

	c, err := client.New(ctrl.GetConfigOrDie(), client.Options{Scheme: scheme})

	if err != nil {
		return err
	}

	var first gpujobv1.GpuJob

	if err := c.Get(ctx, job, &first); err != nil {
		return err
	}

	second := first.DeepCopy()
	fmt.Printf("both clients read %s at v%s\n", job.Name, first.ResourceVersion)

	first.Spec.GPUs++

	if err := c.Update(ctx, &first); err != nil {
		return err
	}

	fmt.Printf("first write stored, now v%s\n", first.ResourceVersion)

	second.Spec.GPUs += 2
	err = c.Update(ctx, second)
	var status *apierrors.StatusError

	if errors.As(err, &status) && status.ErrStatus.Code == http.StatusConflict {
		fmt.Printf("second write refused: %d %s\n", status.ErrStatus.Code, status.ErrStatus.Message)

		return nil
	}

	return fmt.Errorf("unexpected result: %v", err)
}

In Rust:

Listing 1.64 Two clients and one refused write, in Rust. code/rust/gpujob/src/bin/conflict.rs

//! Shows the API server refusing a write with an outdated resource version.
//! Two clients read the `train` job, and both write it back with `replace`.
//! The first write succeeds and changes the version, so the second fails with
//! `409 Conflict`.
//!
//! Run it from `code/rust`, while a `GpuJob` named `train` exists, with
//! `cargo run -p gpujob --bin conflict`.

use gpujob::GpuJob;
use kube::api::PostParams;
use kube::{Api, Client, ResourceExt};

/// The job that both clients read and write.
const JOB: &str = "train";

/// Performs the two writes and reports how the server answered each.
#[tokio::main]
async fn main() -> Result<(), kube::Error> {
    let jobs: Api<GpuJob> = Api::default_namespaced(Client::try_default().await?);
    let mut first = jobs.get(JOB).await?;
    let mut second = first.clone();
    println!(
        "both clients read {JOB} at v{}",
        first.resource_version().unwrap_or_default()
    );

    first.spec.gpus += 1;
    let stored = jobs.replace(JOB, &PostParams::default(), &first).await?;
    println!(
        "first write stored, now v{}",
        stored.resource_version().unwrap_or_default()
    );

    second.spec.gpus += 2;

    match jobs.replace(JOB, &PostParams::default(), &second).await {
        Err(kube::Error::Api(status)) if status.code == 409 => {
            println!("second write refused: {} {}", status.code, status.message);
        }
        other => println!("unexpected: {other:?}"),
    }

    Ok(())
}

The Go version recognises the conflict with errors.As, which digs the API server’s status out of the returned error, and compares its code with http.StatusConflict. The Rust version matches the error against a pattern with a guard, if status.code == 409, and every other outcome falls through to the second arm.

$ go run ./gpujob/cmd/conflict
both clients read train at v41931
first write stored, now v41972
second write refused: 409 Operation cannot be fulfilled on gpujobs.gpucloud.dev "train":
the object has been modified; please apply your changes to the latest version and try again

$ cargo run -p gpujob --bin conflict
both clients read train at v688
first write stored, now v731
second write refused: 409 Operation cannot be fulfilled on gpujobs.gpucloud.dev "train":
the object has been modified; please apply your changes to the latest version and try again

The first write raised train to two GPUs, and our controllers noticed. The Go controller logged:

15:13:07  INFO  controller  GPU count changed, replacing pod  {"name": "train"}
15:13:07  INFO  controller  status changed  {"name": "train", "phase": "Pending", "message": ""}
15:13:07  INFO  controller  status changed  {"name": "train", "phase": "Pending", "message":
          "0/1 nodes are available: 1 Insufficient nvidia.com/gpu. preemption: 0/1 nodes are
          available: 1 Preemption is not helpful for scheduling."}

And the Rust controller:

14:01:46.101  INFO GPU count changed, replacing pod name=train
14:01:46.137  INFO status changed name=train phase=Pending
14:01:46.170  INFO status changed name=train phase=Pending 0/1 nodes are available:
                   1 Insufficient nvidia.com/gpu. preemption: 0/1 nodes are available:
                   1 Preemption is not helpful for scheduling.

Each controller deleted the old pod, because a pod’s resources cannot change, and the deletion event triggered the reconcile that created the replacement. The replacement asks for two GPUs on a machine that has only one, so the scheduler cannot place it anywhere. Thanks to the message we copy from the pod’s PodScheduled condition, the job’s status tells the user exactly why, and kubectl get gpujobs shows the job as Pending until its request is lowered or a machine with two free GPUs joins the cluster.

The complete programs

Listing 1.65 The API types, complete, in Go. code/go/gpujob/api/v1/types.go

package v1

import (
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)


// GpuJobSpec is a request to run one container with a number of GPUs.
type GpuJobSpec struct {
	// Container image to run.
	Image string `json:"image"`
	// Command to run in the container. When empty, the image's own command runs.
	// +optional
	Command []string `json:"command,omitempty"`
	// Number of GPUs the container requests.
	// +kubebuilder:validation:Minimum=0
	GPUs int32 `json:"gpus"`
}

// Phase is the progress of a job, following the phase of its pod.
// +kubebuilder:validation:Enum=Pending;Running;Succeeded;Failed
type Phase string

// The phases a job moves through.
const (
	// PhasePending means the pod is waiting to be scheduled or for its containers to start.
	PhasePending Phase = "Pending"
	// PhaseRunning means the pod's container is running.
	PhaseRunning Phase = "Running"
	// PhaseSucceeded means the container exited with status 0.
	PhaseSucceeded Phase = "Succeeded"
	// PhaseFailed means the container exited with a non-zero status.
	PhaseFailed Phase = "Failed"
)



// GpuJobStatus is the state of a job as the controller last observed it.
type GpuJobStatus struct {
	// Progress of the job's pod.
	Phase Phase `json:"phase"`
	// The scheduler's reason, when the pod cannot be placed on a node.
	// +optional
	Message string `json:"message,omitempty"`
	// The metadata.generation of the spec that this status describes.
	// +optional
	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
}



// GpuJob runs one container with a number of GPUs.
//
// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:shortName=gj
// +kubebuilder:printcolumn:name="GPUs",type=integer,JSONPath=".spec.gpus"
// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=".status.phase"
type GpuJob struct {
	metav1.TypeMeta `json:",inline"`
	// Standard object metadata: name, namespace, labels, resource version.
	metav1.ObjectMeta `json:"metadata,omitempty"`
	// What the user asks for.
	Spec GpuJobSpec `json:"spec"`
	// What the controller has observed.
	// +optional
	Status GpuJobStatus `json:"status,omitempty"`
}

// GpuJobList is a list of GpuJob objects, as the API server returns it.
//
// +kubebuilder:object:root=true
type GpuJobList struct {
	metav1.TypeMeta `json:",inline"`
	// Standard list metadata, including the list's resource version.
	metav1.ListMeta `json:"metadata,omitempty"`
	// The jobs in the list.
	Items []GpuJob `json:"items"`
}

func init() {
	SchemeBuilder.Register(&GpuJob{}, &GpuJobList{})
}



// StatusFromPod derives a job's status from its pod. generation is the job's
// metadata.generation, recorded as the generation that the status describes.
func StatusFromPod(pod *corev1.Pod, generation int64) GpuJobStatus {
	status := GpuJobStatus{Phase: PhasePending, ObservedGeneration: generation}

	switch pod.Status.Phase {
	case corev1.PodRunning:
		status.Phase = PhaseRunning
	case corev1.PodSucceeded:
		status.Phase = PhaseSucceeded
	case corev1.PodFailed:
		status.Phase = PhaseFailed
	}

	// A pod that cannot be placed explains why in its PodScheduled condition.
	for _, c := range pod.Status.Conditions {
		if c.Type == corev1.PodScheduled && c.Status == corev1.ConditionFalse {
			status.Message = c.Message
		}
	}

	return status
}

Listing 1.66 The API types, complete, in Rust. code/rust/gpujob/src/lib.rs

//! The `GpuJob` custom resource: a request to run one container with a number
//! of GPUs, and the status that the controller reports back.
//!
//! The `///` comments on the spec and status types become the field
//! descriptions in the generated `CustomResourceDefinition`, which
//! `kubectl explain gpujob.spec` prints.

use k8s_openapi::api::core::v1::Pod;
use kube::CustomResource;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

/// A request to run one container with a number of GPUs.
///
/// The `CustomResource` derive generates the `GpuJob` type, which holds this
/// spec together with the object's metadata and an optional `GpuJobStatus`.
#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
#[kube(
    group = "gpucloud.dev",
    version = "v1",
    kind = "GpuJob",
    namespaced,
    status = "GpuJobStatus",
    shortname = "gj",
    printcolumn = r#"{"name":"GPUs","type":"integer","jsonPath":".spec.gpus"}"#,
    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#
)]
pub struct GpuJobSpec {
    /// Container image to run.
    pub image: String,
    /// Command to run in the container. When empty, the image's own command runs.
    #[serde(default)]
    pub command: Vec<String>,
    /// Number of GPUs the container requests.
    // Kubernetes schemas support int32 and int64, so the u32 is published as a
    // non-negative int32.
    #[schemars(with = "i32", range(min = 0))]
    pub gpus: u32,
}

/// The state of a job as the controller last observed it.
#[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct GpuJobStatus {
    /// Progress of the job's pod.
    pub phase: Phase,
    /// The scheduler's reason, when the pod cannot be placed on a node.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    /// The `metadata.generation` of the spec that this status describes.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub observed_generation: Option<i64>,
}

/// Progress of a job, following the phase of its pod.
#[derive(Deserialize, Serialize, Clone, Copy, Debug, Default, PartialEq, Eq, JsonSchema)]
pub enum Phase {
    /// The pod is waiting to be scheduled or for its containers to start.
    #[default]
    Pending,
    /// The pod's container is running.
    Running,
    /// The container exited with status 0.
    Succeeded,
    /// The container exited with a non-zero status.
    Failed,
}

impl GpuJobStatus {
    /// Derives a job's status from its pod. `generation` is the job's
    /// `metadata.generation`, recorded as the generation this status describes.
    #[must_use]
    pub fn from_pod(pod: &Pod, generation: Option<i64>) -> Self {
        let status = pod.status.as_ref();
        let phase = match status.and_then(|s| s.phase.as_deref()) {
            Some("Running") => Phase::Running,
            Some("Succeeded") => Phase::Succeeded,
            Some("Failed") => Phase::Failed,
            _ => Phase::Pending,
        };
        // A pod that cannot be placed explains why in its PodScheduled condition.
        let message = status
            .and_then(|s| s.conditions.as_ref())
            .into_iter()
            .flatten()
            .find(|c| c.type_ == "PodScheduled" && c.status == "False")
            .and_then(|c| c.message.clone());
        Self {
            phase,
            message,
            observed_generation: generation,
        }
    }
}

Listing 1.67 The event printer, complete, in Go. code/go/gpujob/cmd/watch-jobs/main.go

// Command watch-jobs prints the events that an informer delivers for GpuJob
// objects in the default namespace: the initial list, one object at a time,
// and then one event per change. Stop it with Ctrl-C.
//
// Run it from code/go with:
//
//	go run ./gpujob/cmd/watch-jobs
package main

import (
	"fmt"
	"os"

	"k8s.io/apimachinery/pkg/runtime"
	toolscache "k8s.io/client-go/tools/cache"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"

	gpujobv1 "gpuk8s/gpujob/api/v1"
)

// namespace is the namespace whose jobs the program watches.
const namespace = "default"

// describe formats a job as its name, GPU count, and resource version.
func describe(obj any) string {
	// A deletion that a relist discovered arrives wrapped in a tombstone.
	if tombstone, ok := obj.(toolscache.DeletedFinalStateUnknown); ok {
		obj = tombstone.Obj
	}

	job := obj.(*gpujobv1.GpuJob)

	return fmt.Sprintf("%s gpus=%d v%s", job.Name, job.Spec.GPUs, job.ResourceVersion)
}


// main starts an informer for GpuJob objects and prints one line per event.
func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// run does the work of main and returns the first error it meets.
func run() error {
	scheme := runtime.NewScheme()

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		return err
	}

	informers, err := cache.New(ctrl.GetConfigOrDie(), cache.Options{
		Scheme:            scheme,
		DefaultNamespaces: map[string]cache.Config{namespace: {}},
	})

	if err != nil {
		return err
	}

	ctx := ctrl.SetupSignalHandler()
	informer, err := informers.GetInformer(ctx, &gpujobv1.GpuJob{})

	if err != nil {
		return err
	}

	registration, err := informer.AddEventHandler(toolscache.ResourceEventHandlerDetailedFuncs{
		AddFunc: func(obj any, isInInitialList bool) {
			if isInInitialList {
				fmt.Println("  listed  ", describe(obj))
			} else {
				fmt.Println("  changed ", describe(obj))
			}
		},
		UpdateFunc: func(_, obj any) { fmt.Println("  changed ", describe(obj)) },
		DeleteFunc: func(obj any) { fmt.Println("  deleted ", describe(obj)) },
	})

	if err != nil {
		return err
	}

	fmt.Println("list started")
	go informers.Start(ctx)

	if !toolscache.WaitForCacheSync(ctx.Done(), registration.HasSynced) {
		return fmt.Errorf("informer did not sync")
	}

	fmt.Println("list complete, watching")
	<-ctx.Done()

	return nil
}

Listing 1.68 The controller, complete, in Go. code/go/gpujob/cmd/controller/main.go

// Command controller is a controller for GpuJob objects. It runs each job as a
// pod that requests the job's GPUs, and reports the pod's progress in the
// job's status.
//
// Run it from code/go, against the cluster in the current kubeconfig context,
// with:
//
//	go run ./gpujob/cmd/controller
package main

import (
	"context"
	"os"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/apimachinery/pkg/runtime"
	corev1ac "k8s.io/client-go/applyconfigurations/core/v1"
	metav1ac "k8s.io/client-go/applyconfigurations/meta/v1"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	gpujobv1 "gpuk8s/gpujob/api/v1"
)

// manager is the field manager name under which the controller writes with
// server-side apply.
const manager = "gpujob-controller"

// gpu is the extended resource through which a container requests GPUs.
const gpu corev1.ResourceName = "nvidia.com/gpu"

// desiredPod builds the pod that should exist for job, as an apply
// configuration: a partial object that lists only the fields this controller
// sets.
//
// The owner reference makes Kubernetes delete the pod together with the job,
// and it lets the controller map each pod event to the job that owns the pod.
func desiredPod(job *gpujobv1.GpuJob) *corev1ac.PodApplyConfiguration {
	gpus := corev1.ResourceList{gpu: *resource.NewQuantity(int64(job.Spec.GPUs), resource.DecimalSI)}
	container := corev1ac.Container().
		WithName("job").
		WithImage(job.Spec.Image).
		WithResources(corev1ac.ResourceRequirements().WithLimits(gpus))
	if len(job.Spec.Command) > 0 {
		container.WithCommand(job.Spec.Command...)
	}

	owner := metav1ac.OwnerReference().
		WithAPIVersion(gpujobv1.GroupVersion.String()).
		WithKind("GpuJob").
		WithName(job.Name).
		WithUID(job.UID).
		WithController(true).
		WithBlockOwnerDeletion(true)
	return corev1ac.Pod(job.Name, job.Namespace).
		WithOwnerReferences(owner).
		WithSpec(corev1ac.PodSpec().
			WithRestartPolicy(corev1.RestartPolicyNever).
			WithContainers(container))
}

// podGPUs returns the number of GPUs that an existing pod's container
// requests, read from its limits, and whether the pod has a GPU limit.
func podGPUs(pod *corev1.Pod) (int32, bool) {
	if len(pod.Spec.Containers) == 0 {
		return 0, false
	}

	quantity, ok := pod.Spec.Containers[0].Resources.Limits[gpu]

	return int32(quantity.Value()), ok
}


// GpuJobReconciler makes the cluster match each GpuJob.
type GpuJobReconciler struct {
	// Client reads objects from the manager's cache and writes to the API server.
	client.Client
}

// Reconcile makes the cluster match one job: it ensures the job's pod exists
// with the requested GPUs, and writes the pod's progress into the job's status.
func (r *GpuJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := ctrl.Log.WithName("controller").WithValues("name", req.Name)

	var job gpujobv1.GpuJob

	if err := r.Get(ctx, req.NamespacedName, &job); err != nil {
		// A job deleted since it was queued needs nothing: its pod goes with it.
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// The resources of a pod are fixed when it is created. When the job asks
	// for a different number of GPUs, delete the pod and come back after it is gone.
	var pod corev1.Pod
	err := r.Get(ctx, req.NamespacedName, &pod)

	switch {
	case err == nil:
		if gpus, ok := podGPUs(&pod); !ok || gpus != job.Spec.GPUs {
			log.Info("GPU count changed, replacing pod")

			if err := r.Delete(ctx, &pod); client.IgnoreNotFound(err) != nil {
				return ctrl.Result{}, err
			}

			return ctrl.Result{RequeueAfter: 2 * time.Second}, nil
		}
	case !apierrors.IsNotFound(err):
		return ctrl.Result{}, err
	}

	if err := r.Apply(ctx, desiredPod(&job), client.FieldOwner(manager), client.ForceOwnership); err != nil {
		return ctrl.Result{}, err
	}

	// A pod that was just created is not in the cache yet; its zero value
	// yields the Pending phase, and the pod's creation event triggers the
	// reconcile that reports its real progress.
	status := gpujobv1.StatusFromPod(&pod, job.Generation)

	if job.Status != status {
		log.Info("status changed", "phase", status.Phase, "message", status.Message)
		patch := client.MergeFrom(job.DeepCopy())
		job.Status = status

		if err := r.Status().Patch(ctx, &job, patch); err != nil {
			return ctrl.Result{}, err
		}
	}

	// Changes to the pod trigger the next reconcile through Owns.
	return ctrl.Result{}, nil
}


// main builds a manager, registers the reconciler, and runs until Ctrl-C.
func main() {
	ctrl.SetLogger(zap.New(zap.UseDevMode(true))) // readable console output
	setup := ctrl.Log.WithName("setup")

	scheme := runtime.NewScheme()

	if err := clientgoscheme.AddToScheme(scheme); err != nil {
		setup.Error(err, "registering built-in types")
		os.Exit(1)
	}

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		setup.Error(err, "registering GpuJob")
		os.Exit(1)
	}

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:  scheme,
		Metrics: metricsserver.Options{BindAddress: "0"}, // no metrics endpoint
	})

	if err != nil {
		setup.Error(err, "creating manager")
		os.Exit(1)
	}

	err = ctrl.NewControllerManagedBy(mgr).
		For(&gpujobv1.GpuJob{}).
		Owns(&corev1.Pod{}).
		Complete(&GpuJobReconciler{Client: mgr.GetClient()})
	if err != nil {
		setup.Error(err, "creating controller")
		os.Exit(1)
	}

	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		setup.Error(err, "running manager")
		os.Exit(1)
	}
}

Listing 1.69 The controller, complete, in Rust. code/rust/gpujob/src/bin/controller.rs

//! A controller for `GpuJob` objects. It runs each job as a pod that requests
//! the job's GPUs, and reports the pod's progress in the job's status.
//!
//! Run it against the cluster in the current kubeconfig context with
//! `cargo run -p gpujob` from `code/rust` (the package's default binary).

use futures::StreamExt;
use gpujob::{GpuJob, GpuJobStatus};
use k8s_openapi::api::core::v1::{Container, Pod, PodSpec, ResourceRequirements};
use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
use kube::api::{DeleteParams, ObjectMeta, Patch, PatchParams};
use kube::runtime::controller::{Action, Controller};
use kube::runtime::watcher;
use kube::{Api, Client, Resource, ResourceExt};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

/// The field manager name under which the controller writes with server-side apply.
const MANAGER: &str = "gpujob-controller";

/// The extended resource through which a container requests GPUs.
const GPU: &str = "nvidia.com/gpu";

/// Everything a reconcile can fail with.
#[derive(Debug, thiserror::Error)]
enum Error {
    /// A request to the API server failed.
    #[error("kubernetes API: {0}")]
    Kube(#[from] kube::Error),
    /// The status could not be serialized into a patch.
    #[error("serializing status: {0}")]
    Json(#[from] serde_json::Error),
}

/// State shared by every reconcile.
struct Context {
    /// Client for the cluster's API server.
    client: Client,
}

/// Builds the pod that should exist for `job`.
///
/// The owner reference makes Kubernetes delete the pod together with the job,
/// and it lets the controller map each pod event to the job that owns the pod.
fn desired_pod(job: &GpuJob) -> Pod {
    let gpus = BTreeMap::from([(GPU.to_string(), Quantity(job.spec.gpus.to_string()))]);
    Pod {
        metadata: ObjectMeta {
            name: Some(job.name_any()),
            owner_references: job.controller_owner_ref(&()).map(|owner| vec![owner]),
            ..Default::default()
        },
        spec: Some(PodSpec {
            restart_policy: Some("Never".into()),
            containers: vec![Container {
                name: "job".into(),
                image: Some(job.spec.image.clone()),
                command: Some(job.spec.command.clone()).filter(|c| !c.is_empty()),
                resources: Some(ResourceRequirements {
                    limits: Some(gpus),
                    ..Default::default()
                }),
                ..Default::default()
            }],
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// Returns the number of GPUs that an existing pod's container requests, read
/// from its limits, or `None` if the pod has no GPU limit.
fn pod_gpus(pod: &Pod) -> Option<u32> {
    let container = pod.spec.as_ref()?.containers.first()?;
    let limits = container.resources.as_ref()?.limits.as_ref()?;

    limits.get(GPU)?.0.parse().ok()
}

/// Makes the cluster match one job: ensures its pod exists with the requested
/// GPUs, and writes the pod's progress into the job's status.
async fn reconcile(job: Arc<GpuJob>, ctx: Arc<Context>) -> Result<Action, Error> {
    let namespace = job.namespace().unwrap_or_default();
    let name = job.name_any();
    let pods: Api<Pod> = Api::namespaced(ctx.client.clone(), &namespace);
    let jobs: Api<GpuJob> = Api::namespaced(ctx.client.clone(), &namespace);

    // The resources of a pod are fixed when it is created. When the job asks
    // for a different number of GPUs, delete the pod and come back after it is gone.
    if let Some(pod) = pods.get_opt(&name).await?
        && pod_gpus(&pod) != Some(job.spec.gpus)
    {
        tracing::info!(%name, "GPU count changed, replacing pod");
        pods.delete(&name, &DeleteParams::default()).await?;

        return Ok(Action::requeue(Duration::from_secs(2)));
    }

    let apply = PatchParams::apply(MANAGER).force();
    let pod = pods
        .patch(&name, &apply, &Patch::Apply(desired_pod(&job)))
        .await?;

    let status = GpuJobStatus::from_pod(&pod, job.metadata.generation);

    if job.status.as_ref() != Some(&status) {
        let message = status.message.as_deref().unwrap_or("");
        tracing::info!(%name, phase = ?status.phase, message, "status changed");
        let patch = serde_json::json!({
            "apiVersion": GpuJob::api_version(&()),
            "kind": GpuJob::kind(&()),
            "status": status,
        });
        jobs.patch_status(&name, &apply, &Patch::Apply(patch))
            .await?;
    }

    // Changes to the pod trigger the next reconcile through `owns`.
    Ok(Action::await_change())
}

/// Decides what happens after a failed reconcile: log it and retry in five seconds.
#[allow(clippy::needless_pass_by_value)] // the signature is fixed by `Controller::run`
fn error_policy(job: Arc<GpuJob>, error: &Error, _ctx: Arc<Context>) -> Action {
    tracing::warn!(name = %job.name_any(), %error, "reconcile failed");

    Action::requeue(Duration::from_secs(5))
}

/// Starts the controller and runs it until Ctrl-C.
#[tokio::main]
async fn main() -> Result<(), kube::Error> {
    tracing_subscriber::fmt()
        .with_env_filter("controller=info,kube=warn")
        .with_target(false)
        .with_ansi(std::io::IsTerminal::is_terminal(&std::io::stdout()))
        .init();
    let client = Client::try_default().await?;
    let jobs: Api<GpuJob> = Api::all(client.clone());
    let pods: Api<Pod> = Api::all(client.clone());

    Controller::new(jobs, watcher::Config::default())
        .owns(pods, watcher::Config::default())
        .shutdown_on_signal()
        .run(reconcile, error_policy, Arc::new(Context { client }))
        .for_each(|result| async move {
            if let Err(error) = result {
                tracing::debug!(%error, "reconcile error");
            }
        })
        .await;
    Ok(())
}

Listing 1.70 The two-client program, complete, in Go. code/go/gpujob/cmd/conflict/main.go

// Command conflict shows the API server refusing a write with an out-of-date
// resource version. Two clients read the train job, and both write it back
// with Update. The first write succeeds and changes the version, so the second
// fails with 409 Conflict.
//
// Run it from code/go, while a GpuJob named train exists, with:
//
//	go run ./gpujob/cmd/conflict
package main

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"os"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	gpujobv1 "gpuk8s/gpujob/api/v1"
)

// job identifies the job that both clients read and write.
var job = client.ObjectKey{Namespace: "default", Name: "train"}

// main performs the two writes and reports how the server answered each.
func main() {
	if err := run(context.Background()); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// run does the work of main and returns the first unexpected error.
func run(ctx context.Context) error {
	scheme := runtime.NewScheme()

	if err := gpujobv1.AddToScheme(scheme); err != nil {
		return err
	}

	c, err := client.New(ctrl.GetConfigOrDie(), client.Options{Scheme: scheme})

	if err != nil {
		return err
	}

	var first gpujobv1.GpuJob

	if err := c.Get(ctx, job, &first); err != nil {
		return err
	}

	second := first.DeepCopy()
	fmt.Printf("both clients read %s at v%s\n", job.Name, first.ResourceVersion)

	first.Spec.GPUs++

	if err := c.Update(ctx, &first); err != nil {
		return err
	}

	fmt.Printf("first write stored, now v%s\n", first.ResourceVersion)

	second.Spec.GPUs += 2
	err = c.Update(ctx, second)
	var status *apierrors.StatusError

	if errors.As(err, &status) && status.ErrStatus.Code == http.StatusConflict {
		fmt.Printf("second write refused: %d %s\n", status.ErrStatus.Code, status.ErrStatus.Message)

		return nil
	}

	return fmt.Errorf("unexpected result: %v", err)
}