The Controller Pattern
Compare desired state with actual state in a loop, and make the comparison safe to repeat any number of times.
func (r *Reconciler) Reconcile(ctx context.Context, key string) errorA controller is a program with one job: look at what the cluster is supposed to contain, look at what it actually contains, and make the second match the first. Then do it again. Every built-in Kubernetes behaviour works this way. The Deployment controller notices that you asked for three pods and two exist, so it creates one.
The function that does the comparing is called reconcile. It takes the name of one object and returns an error. It does not take an event, and it does not take a list of changes.
Level-triggered, not edge-triggered
This is the central idea. Edge-triggered means reacting to the change: a pod was deleted, so create one. Level-triggered means reacting to the state: three wanted, two present, so create one. Edge-triggered logic breaks the moment a message is lost, duplicated or arrives out of order, and all three happen. Level-triggered logic does not care how many events it missed, because it re-reads the truth every time.
Idempotent means safe to repeat
Reconcile will run again for reasons that have nothing to do with a change: a resync, a restart, a retried error, two events for one edit. So it must reach the same end state whether it runs once or forty times. Create with a deterministic name so the second attempt gets AlreadyExists rather than making a duplicate, and treat that error as success.
The queue
Keys go on a rate-limited work queue. The queue removes duplicates, so a burst of edits to one object becomes one reconcile. A failed key goes back with a growing delay instead of spinning.
The same thing, written two ways
The longhand is informers plus a workqueue plus your own worker goroutines. controller-runtime is the shorthand: ctrl.NewControllerManagedBy(mgr).For(&v1.Database{}).Complete(r) builds exactly that machinery, with the same informers underneath.
Example
package main
import (
"fmt"
"sort"
)
// reconcile compares desired replica counts against actual ones and returns the
// steps needed to close the gap. It looks at state, never at events, so running
// it twice in a row produces no second round of work.
func reconcile(desired, actual map[string]int) []string {
var steps []string
for name, want := range desired {
have, exists := actual[name]
switch {
case !exists:
steps = append(steps, fmt.Sprintf("create %s with %d replicas", name, want))
case have != want:
steps = append(steps, fmt.Sprintf("scale %s from %d to %d", name, have, want))
}
}
// Anything running that nobody asked for is also a gap.
for name := range actual {
if _, wanted := desired[name]; !wanted {
steps = append(steps, "delete "+name)
}
}
sort.Strings(steps) // map order is random; sort so the plan is stable
return steps
}
func main() {
desired := map[string]int{"web": 3, "worker": 2}
actual := map[string]int{"web": 1, "old-cron": 1}
for _, step := range reconcile(desired, actual) {
fmt.Println(step)
}
// Once actual matches desired there is nothing left to do. That is idempotence.
fmt.Println("second pass:", len(reconcile(desired, desired)), "steps")
}
// Output:
// create worker with 2 replicas
// delete old-cron
// scale web from 1 to 3
// second pass: 0 stepsWhen to use it
- Make a Deployment restart automatically whenever the ConfigMap it mounts changes.
- Provision an external resource, such as an S3 bucket, whenever a custom object appears in the cluster.
- Copy a shared TLS Secret into every namespace and keep the copies current.
- Replace a cron job full of kubectl commands with a program that holds the desired state continuously.
More examples
The worker loop around the queue
Get, Done, Forget and AddRateLimited are a set: missing Done deadlocks that key, and missing Forget means a key that once failed keeps a long delay for the rest of the process's life. Write this loop once and reuse it.
package main
import (
"context"
"log/slog"
"k8s.io/client-go/util/workqueue"
)
type controller struct {
queue workqueue.TypedRateLimitingInterface[string]
reconcile func(ctx context.Context, key string) error
}
// runWorker pulls one key at a time until the queue shuts down. Run several of
// these as goroutines; the queue guarantees the same key is never handed to two
// workers at once, so a reconciler needs no locking of its own.
func (c *controller) runWorker(ctx context.Context) {
for {
key, shutdown := c.queue.Get()
if shutdown {
return
}
// Done must always be called, or this key is blocked forever.
func() {
defer c.queue.Done(key)
if err := c.reconcile(ctx, key); err != nil {
slog.Error("reconcile failed", "key", key, "err", err)
c.queue.AddRateLimited(key) // back off and try again
return
}
c.queue.Forget(key) // success resets this key's backoff
}()
}
}
func main() {
// The generic form. Older client-go code writes the same queue without the
// type parameter: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()).
q := workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]())
defer q.ShutDown()
_ = &controller{queue: q}
}A reconciler whose create step is safe to repeat
Two details carry the idempotence: a name computed from the input, and treating AlreadyExists as success. GenerateName would create a new StatefulSet on every reconcile. The owner reference means you never write deletion logic for child objects.
package main
import (
"context"
appsv1 "k8s.io/api/apps/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/kubernetes"
dbv1 "example.com/db/api/v1alpha1"
)
type Reconciler struct {
Client kubernetes.Interface
}
// ensureStatefulSet creates the StatefulSet for a Database, or does nothing if
// it already exists. The name is derived from the Database, so a repeat run
// collides with its own earlier work instead of creating a second copy.
func (r *Reconciler) ensureStatefulSet(ctx context.Context, db *dbv1.Database) error {
want := &appsv1.StatefulSet{
ObjectMeta: metav1.ObjectMeta{
Name: db.Name + "-db", // deterministic, not generated
Namespace: db.Namespace,
// An owner reference makes Kubernetes delete this StatefulSet
// automatically when the Database is deleted.
OwnerReferences: []metav1.OwnerReference{*metav1.NewControllerRef(db,
schema.GroupVersionKind{Group: "db.example.com", Version: "v1alpha1", Kind: "Database"})},
},
Spec: buildSpec(db),
}
_, err := r.Client.AppsV1().StatefulSets(db.Namespace).Create(ctx, want, metav1.CreateOptions{})
if apierrors.IsAlreadyExists(err) {
return nil // the previous run already did this; that is success
}
return err
}
func buildSpec(db *dbv1.Database) appsv1.StatefulSetSpec { return appsv1.StatefulSetSpec{} }Reporting back through status conditions
Conditions are the standard vocabulary for a controller's verdict on an object. Use SetStatusCondition rather than appending, or the list grows one entry per reconcile. Status must be written through the status subresource so a status update does not count as a spec change and trigger another reconcile.
package main
import (
"fmt"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// setReady records the outcome of a reconcile on the object's status. kubectl
// and other controllers read conditions, so this is how a controller explains
// itself instead of only logging.
func setReady(conditions *[]metav1.Condition, generation int64, err error) {
cond := metav1.Condition{
Type: "Ready",
Status: metav1.ConditionTrue,
Reason: "Provisioned",
Message: "database is serving",
// ObservedGeneration records which version of the spec this verdict is about.
// A reader comparing it to metadata.generation can tell a stale status.
ObservedGeneration: generation,
}
if err != nil {
cond.Status = metav1.ConditionFalse
cond.Reason = "ProvisionFailed"
cond.Message = err.Error()
}
// SetStatusCondition replaces the existing condition of this Type, and only
// moves LastTransitionTime when Status actually changed.
apimeta.SetStatusCondition(conditions, cond)
}
func main() {
var conds []metav1.Condition
setReady(&conds, 4, fmt.Errorf("storage class missing"))
setReady(&conds, 5, nil)
fmt.Println(len(conds), conds[0].Status, conds[0].ObservedGeneration)
}
// Output:
// 1 True 5
Discussion