Understanding the Core Workflow of Kubernetes Informer in client-go

The article analyzes the basic process of starting an informer in a Kubernetes controller, emphasizing that both built‑in components and custom controllers need to watch etcd events via the apiserver to implement their control loops.

The client-go informer package offers three main capabilities: a local cache (store), an indexing mechanism (indexer), and event‑handler registration.

Informer Architecture

The upper part of the architecture is provided by client-go, while the lower part is the user‑defined control‑loop logic. The upper part consists of several components:

Reflector

Reflector interacts with the apiserver using list and watch APIs, pushing resource objects into a queue.

DeltaFIFO

DeltaFIFO maintains a FIFO queue ( queue) and a map of items ( items) that store objects together with their change type. The queue holds the order of processing, while items keep the actual objects and their deltas.

type DeltaFIFO struct {
  ...
  // items map[string]Deltas // objects indexed by key
  // queue []string          // FIFO order of keys
  ...
}

Two parts are distinguished:

FIFO – a first‑in‑first‑out queue.

Delta – a map that stores the object together with the operation type (Added, Updated, Deleted, etc.).

Indexer

The indexer is a local store that keeps resource objects and provides indexing functions, keeping the cache in sync with the etcd data.

Basic Example

func main() {
  stopCh := make(chan struct{})
  defer close(stopCh)

  // (1) Build clientset
  masterUrl := "172.27.32.110:8080"
  config, err := clientcmd.BuildConfigFromFlags(masterUrl, "")
  if err != nil { klog.Errorf("BuildConfigFromFlags err: %v", err) }
  clientset, err := k.NewForConfig(config)
  if err != nil { klog.Errorf("Get clientset err: %v", err) }

  // (2) Create shared informer factory
  sharedInformers := informers.NewSharedInformerFactory(clientset, defaultResync)

  // (3) Register a pod informer
  podInformer := sharedInformers.Core().V1().Pods().Informer()

  // (4) Register event handler
  podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: func(obj interface{}) {
      mObj := obj.(v1.Object)
      klog.Infof("Get new obj: %v", mObj)
      klog.Infof("Get new obj name: %s", mObj.GetName())
    },
  })

  // (5) Start all informers
  sharedInformers.Start(stopCh)

  // (6) Wait for cache sync
  if !cache.WaitForCacheSync(stopCh, podInformer.HasSynced) {
    klog.Infof("Cache sync fail!")
  }

  // (7) Use lister to list pods
  podLister := sharedInformers.Core().V1().Pods().Lister()
  pods, err := podLister.List(labels.Everything())
  if err != nil { klog.Infof("err: %v", err) }
  klog.Infof("len(pods), %d", len(pods))
  for _, v := range pods { klog.Infof("pod: %s", v.Name) }

  <-stopCh
}

The example highlights steps (2)–(5): creating a shared informer factory, obtaining a pod informer, adding an event handler, and starting the informer.

Process Analysis

3.1 New a sharedInformers factory

Creating a sharedInformerFactory avoids creating duplicate informers that would overload the apiserver. The factory holds a map of informers and starts them on demand.

sharedInformers := informers.NewSharedInformerFactory(clientset, defaultResync)
factory := &sharedInformerFactory{
  client: client,
  namespace: v1.NamespaceAll,
  defaultResync: defaultResync,
  informers: make(map[reflect.Type]cache.SharedIndexInformer),
  startedInformers: make(map[reflect.Type]bool),
  customResync: make(map[reflect.Type]time.Duration),
}

3.2 Register a informer

The informer is generated and stored in the factory via InformerFor and defaultInformer which ultimately calls NewFilteredPodInformer.

podInformer := sharedInformers.Core().V1().Pods().Informer()
func (f *podInformer) Informer() cache.SharedIndexInformer {
  return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer)
}

func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
  informer := newFunc(f.client, resyncPeriod)
  f.informers[informerType] = informer
  return informer
}

3.3 Register event handler

The handler is added to the shared processor, which creates a listener that forwards events through addCh → nextCh → appropriate callback.

podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { /* ... */ } })

func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
  s.processor.addListener(newProcessorListener(handler, ...))
}

3.4 Start all informers

Calling sharedInformers.Start(stopCh) launches each informer in a separate goroutine.

func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {
  for informerType, informer := range f.informers {
    if !f.startedInformers[informerType] {
      go informer.Run(stopCh)
      f.startedInformers[informerType] = true
    }
  }
}

Informer Run Logic

The run method creates a DeltaFIFO, builds a controller.Config, and starts the reflector and processing loop.

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
  fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{KnownObjects: s.indexer, EmitDeltaTypeReplaced: true})
  cfg := &Config{Queue: fifo, ListerWatcher: s.listerWatcher, ObjectType: s.objectType, FullResyncPeriod: s.resyncCheckPeriod, Process: s.HandleDeltas, ...}
  s.controller = New(cfg)
  s.controller.Run(stopCh)
}

The controller runs the reflector ( ListAndWatch) to fill the queue and a processing loop that pops items, calls the registered handlers, and updates the indexer.

Reflector ListAndWatch

The reflector first performs a list to obtain the current state, stores objects in the DeltaFIFO, then enters a watch loop that converts watch events into deltas and pushes them into the queue.

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
  // List phase
  list, err := r.listerWatcher.List(opts)
  resourceVersion = listMetaInterface.GetResourceVersion()
  items, err := meta.ExtractList(list)
  r.syncWith(items, resourceVersion)

  // Watch phase
  for {
    w, err := r.listerWatcher.Watch(options)
    if err != nil { return err }
    r.watchHandler(start, w, &resourceVersion, errc, stopCh)
  }
}

DeltaFIFO Production & Consumption

Production occurs in Reflector.watchHandler, where events are added to the FIFO via store.Add (which calls queueActionLocked). Consumption happens in controller.processLoop, which pops items, runs HandleDeltas, and distributes notifications to listeners.

func (f *DeltaFIFO) Add(obj interface{}) error { return f.queueActionLocked(Added, obj) }
func (f *DeltaFIFO) queueActionLocked(action DeltaType, obj interface{}) error { /* update items and queue */ }

func (c *controller) processLoop() {
  for {
    obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
    // handle err, requeue if needed
  }
}

Conclusion

The informer mechanism is essentially a producer‑consumer system built on DeltaFIFO with a local cache and indexer, providing a convenient API for registering callbacks and efficiently watching Kubernetes resources. Understanding the upper‑layer workflow—Reflector, DeltaFIFO, Indexer, and the shared informer factory—helps developers build reliable custom controllers.