Watches and Informers

Keep a local in-memory copy of cluster objects with one shared watch instead of polling the API server.

Syntaxfactory := informers.NewSharedInformerFactory(client, 10*time.Minute)

Calling List in a loop works for one program and breaks a cluster when fifty do it. Each call makes the API server read from its database and serialise every object. The Kubernetes answer is a watch: one long-lived HTTP request that stays open and streams a small event every time a matching object is added, changed or deleted.

A raw watch is awkward. It disconnects, it needs a resume point, and it tells you about changes without telling you the current state. An informer wraps it. An informer does an initial List, keeps every object in a map in your process's memory, opens a watch to keep that map current, reconnects when the connection drops, and calls your functions on each change.

Reading from the cache

Once an informer has caught up, reads cost nothing. A lister reads from the in-memory map, not from the network. pods.Lister().Pods("shop").Get("web-1") does no HTTP at all.

Shared factories

A SharedInformerFactory makes one informer per resource type and hands it to everyone who asks. Ten parts of your program watching pods still means one watch on the API server. Call factory.Start(stopCh) to begin, then wait for the caches to fill before reading. Reading an unsynced cache returns an empty list, which looks exactly like an empty cluster.

Resync is not a re-read

The duration you pass the factory is a resync period. It re-delivers update events from the cache. It does not re-fetch from the API server.

The same thing, written two ways

factory.WaitForCacheSync(stopCh) waits for every informer the factory started. The longhand, one informer at a time, is cache.WaitForCacheSync(stopCh, pods.Informer().HasSynced). Same wait.

Example

Example · go
package main

import (
	"fmt"
	"sort"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes/fake"
	"k8s.io/client-go/tools/cache"
)

func main() {
	client := fake.NewSimpleClientset(
		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "web-2", Namespace: "shop"}},
		&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "web-1", Namespace: "shop"}},
	)

	// 10 minutes is the resync period, not a poll interval.
	factory := informers.NewSharedInformerFactory(client, 10*time.Minute)
	pods := factory.Core().V1().Pods() // every caller shares this one watch

	// Asking for the informer is what registers it with the factory, and Start
	// only runs informers that are already registered. So this line must come
	// before Start, or nothing is ever started and the wait below never ends.
	informer := pods.Informer()

	stop := make(chan struct{})
	defer close(stop) // closing the channel stops every informer the factory started
	factory.Start(stop)

	// Reading before this returns gives an empty cache, not an error.
	if !cache.WaitForCacheSync(stop, informer.HasSynced) {
		fmt.Println("cache did not sync")
		return
	}

	// ListKeys reads the in-memory store. No request leaves the process.
	keys := informer.GetIndexer().ListKeys()
	sort.Strings(keys) // store order is not defined
	fmt.Println(keys)
}

// Output:
// [shop/web-1 shop/web-2]

When to use it

  • Keep a live in-process list of healthy backend pods for a custom load balancer without hammering the API server.
  • React within a second to a Secret changing, instead of polling it every minute.
  • Feed a controller's work queue from pod, Deployment and ConfigMap changes at once.
  • Cut API server load in a tool that currently calls List on a timer.

More examples

Event handlers that only queue work

This is the standard bridge between an informer and a controller. Queueing a key rather than the object means the worker always reads the latest version from the cache, and the queue collapses a burst of changes to one object into a single unit of work.

Example · go
package main

import (
	"fmt"

	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

// register attaches handlers to an informer. They do no work themselves: they
// turn the object into a "namespace/name" key and put it on a queue. Handlers
// run on the informer's own goroutine, so anything slow in here stalls the watch.
func register(inf cache.SharedIndexInformer, queue workqueue.TypedInterface[string]) error {
	// AddEventHandler returns a registration handle and an error in current
	// client-go releases. Older code ignored both because it returned neither.
	_, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj any) {
			if key, kerr := cache.MetaNamespaceKeyFunc(obj); kerr == nil {
				queue.Add(key)
			}
		},
		UpdateFunc: func(oldObj, newObj any) {
			if key, kerr := cache.MetaNamespaceKeyFunc(newObj); kerr == nil {
				queue.Add(key)
			}
		},
		DeleteFunc: func(obj any) {
			// On delete the object may be wrapped in a DeletedFinalStateUnknown
			// tombstone. This key function unwraps it; MetaNamespaceKeyFunc would fail.
			if key, kerr := cache.DeletionHandlingMetaNamespaceKeyFunc(obj); kerr == nil {
				queue.Add(key)
			}
		},
	})
	return err
}

func main() {
	fmt.Println("handlers enqueue keys; workers do the work")
}

Watching one namespace and one label only

Use this whenever your program cares about a subset of objects. Note that the lister on a filtered factory only ever returns the filtered set, which is a trap if another part of the program expects to see everything.

Example · go
package main

import (
	"fmt"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
)

// scopedFactory watches only pods labelled app=web in the shop namespace. The
// filter is applied by the API server, so the cache stays small and the watch
// stream stays quiet. A cluster-wide pod informer in a large cluster can hold
// hundreds of megabytes in memory.
func scopedFactory(client kubernetes.Interface) informers.SharedInformerFactory {
	return informers.NewSharedInformerFactoryWithOptions(
		client,
		10*time.Minute,
		informers.WithNamespace("shop"),
		informers.WithTweakListOptions(func(opts *metav1.ListOptions) {
			opts.LabelSelector = "app=web"
		}),
	)
}

func main() {
	fmt.Println("narrow the watch before you widen the memory limit")
}

Copy before you change

The division of labour in every controller: reads come from the lister because they are free, writes go through the clientset because only the API server owns the real object. Skipping DeepCopy produces bugs that appear in unrelated code and are close to impossible to trace.

Example · go
package main

import (
	"context"
	"fmt"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	listers "k8s.io/client-go/listers/core/v1"
	"k8s.io/client-go/kubernetes"
)

// annotate adds an annotation to a pod. The lister hands back the cache's own
// pointer, shared with every other reader, so it must be copied first.
func annotate(ctx context.Context, client kubernetes.Interface, lister listers.PodLister, ns, name string) error {
	cached, err := lister.Pods(ns).Get(name)
	if err != nil {
		return err
	}

	pod := cached.DeepCopy() // never write through the cached pointer
	if pod.Annotations == nil {
		pod.Annotations = map[string]string{}
	}
	pod.Annotations["checked-by"] = "inventory-controller"

	_, err = client.CoreV1().Pods(ns).Update(ctx, pod, metav1.UpdateOptions{})
	return err
}

func main() {
	fmt.Println("read from the lister, write through the client")
}

Related

go client gogo k8s operatorsgo channel basicsgo mutexgo job queues

Discussion

  • Be the first to comment on this lesson.