Introduction

The Kubernetes client-go library provides a powerful mechanism for interacting with the API server. At the heart of this library lies the Informer pattern, which enables developers to efficiently watch and respond to changes in cluster resources. This article explores the internal workings of the Informer mechanism within client-go's tool/cache package.

Architecture Overview

When building custom controllers in Kubernetes, the Informer serves as the primary mechanism for resource synchronization. The architecture consists of several interconnected components that work together to provide efficient event-driven resource monitoring.

The workflow involves the following key components operating in sequence to maintain a local cache and dispatch events to registered handlers.

Reflector Component

The Reflector is responsible for watching specific resource types and populating a local store with the results. It implements the ListAndWatch pattern to maintani synchronization with the API server.

Reflector Structure

Located in tools/cache/reflector.go, the Reflector struct contains the core fields needed for resource synchronization:

type Reflector struct {
    name string
    expectedType reflect.Type
    store Store
    listerWatcher ListerWatcher
    period time.Duration
    resyncPeriod time.Duration
    ShouldResync func() bool
    clock clock.Clock
    resourceVersion string
    resourceVersionMu sync.RWMutex
}

ListAndWatch Implementation

The ListAndWatch method performs an initial list operation to retrieve all resources, then establishes a watch connection starting from that resource version. This approach ensures no events are missed during synchronization.

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    // Execute initial list operation
    resourceList, err := r.listerWatcher.List(listOptions)
    items, err := meta.ExtractList(resourceList)
    
    // Populate local store with current state
    if err := r.syncWith(items, resourceVersion); err != nil {
        return fmt.Errorf("failed to sync list result: %v", err)
    }
    r.setLastSyncResourceVersion(resourceVersion)

    // Establish watch connection
    for {
        select {
        case <-stopCh:
            return nil
        default:
        }
        
        watchOptions := metav1.ListOptions{
            ResourceVersion: resourceVersion,
            TimeoutSeconds: &timeoutSeconds,
        }
        
        watcher, err := r.listerWatcher.Watch(watchOptions)
        if err := r.watchHandler(watcher, &resourceVersion, stopCh); err != nil {
            return nil
        }
    }
}

Watch Handler Processing

The watchHandler processes events received from the watch connection and forwards them to the Delta FIFO queue for further processing.

func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, 
    stopCh <-chan struct{}) error {
    
loop:
    for {
        select {
        case <-stopCh:
            return errorStopRequested
        case event, ok := <-w.ResultChan():
            if !ok {
                break loop
            }
            
            newResourceVersion := meta.GetResourceVersion(event.Object)
            
            // Route events based on type
            switch event.Type {
            case watch.Added:
                err := r.store.Add(event.Object)
            case watch.Modified:
                err := r.store.Update(event.Object)
            case watch.Deleted:
                err := r.store.Delete(event.Object)
            default:
                utilruntime.HandleError(fmt.Errorf("unknown event type"))
            }
            
            *resourceVersion = newResourceVersion
            r.setLastSyncResourceVersion(newResourceVersion)
        }
    }
    return nil
}

Controller Implementation

The Controller orchestrates the entire Informer pipeline by coordinating between the Reflector and the processing logic.

Controller Interface

type Controller interface {
    Run(stopCh <-chan struct{})
    HasSynced() bool
    LastSyncResourceVersion() string
}

type controller struct {
    config Config
    reflector *Reflector
    reflectorMutex sync.RWMutex
    clock clock.Clock
}

Controller Execution Flow

func (c *controller) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    
    // Create Reflector with configuration
    r := NewReflector(
        c.config.ListerWatcher,
        c.config.ObjectType,
        c.config.Queue,
        c.config.FullResyncPeriod,
    )
    
    c.reflectorMutex.Lock()
    c.reflector = r
    c.reflectorMutex.Unlock()

    var wg wait.Group
    defer wg.Wait()

    wg.StartWithChannel(stopCh, r.Run)
    // Start the processing loop
    wait.Until(c.processLoop, time.Second, stopCh)
}

Process Loop

The processLoop continuously pops items from the Delta FIFO queue and processes them through the configured handler:

func (c *controller) processLoop() {
    for {
        obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
        if err != nil {
            // Handle error cases
        }
    }
}

The Process function handles different delta types by updating the local indexer and invoking appropriate event handlers:

Process: func(obj interface{}) error {
    for _, d := range obj.(Deltas) {
        switch d.Type {
        case Sync, Added, Updated:
            if old, exists, err := clientState.Get(d.Object); err == nil && exists {
                if err := clientState.Update(d.Object); err != nil {
                    return err
                }
                handler.OnUpdate(old, d.Object)
            } else {
                if err := clientState.Add(d.Object); err != nil {
                    return err
                }
                handler.OnAdd(d.Object)
            }
        case Deleted:
            if err := clientState.Delete(d.Object); err != nil {
                return err
            }
            handler.OnDelete(d.Object)
        }
    }
    return nil
}

Indexer and Thread-Safe Storage

The Indexer provides a thread-safe storage mechanism with indexing capabilities for efficient object retrieval.

Creating the Indexer

func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
    return &cache{
        cacheStorage: NewThreadSafeStore(indexers, Indices{}),
        keyFunc:      keyFunc,
    }
}

Thread-Safe Store Implementation

The threadSafeMap provides synchronized access to the underlying data structure:

type threadSafeMap struct {
    items    map[string]interface{}
    indexers Indexers
    indices  Indices
    lock     sync.RWMutex
}

func (c *threadSafeMap) Add(key string, obj interface{}) {
    c.lock.Lock()
    defer c.lock.Unlock()
    
    oldObject := c.items[key]
    c.items[key] = obj
    c.updateIndices(oldObject, obj, key)
}

SharedIndexInformer

The SharedIndexInformer extends the basic Controller with shared caching and event distribution capabilities, allowing multiple listeners to receive the same events.

SharedInformer Interface

type SharedInformer interface {
    AddEventHandler(handler ResourceEventHandler)
    AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration)
    GetStore() Store
    GetController() Controller
    Run(stopCh <-chan struct{})
    HasSynced() bool
    LastSyncResourceVersion() string
}

Internal Structure

type sharedIndexInformer struct {
    indexer    Indexer
    controller Controller
    processor             *sharedProcessor
    cacheMutationDetector CacheMutationDetector
    listerWatcher ListerWatcher
    objectType    runtime.Object
    resyncCheckPeriod time.Duration
    defaultEventHandlerResyncPeriod time.Duration
    clock clock.Clock
    started, stopped bool
    startedLock      sync.Mutex
    blockDeltas sync.Mutex
}

Shared Processor

The sharedProcessor manages multiple listeners and distributes events to all registered handlers:

type sharedProcessor struct {
    listenersStarted bool
    listenersLock    sync.RWMutex
    listeners        []*processorListener
    syncingListeners []*processorListener
    clock            clock.Clock
    wg               wait.Group
}

Processor Listener

Each listener maintains its own notification channel and buffer:

type processorListener struct {
    nextCh chan interface{}
    addCh  chan interface{}
    handler ResourceEventHandler
    pendingNotifications buffer.RingGrowing
}

The listener run method processes incoming notifications:

func (p *processorListener) run() {
    stopCh := make(chan struct{})
    wait.Until(func() {
        err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
            for next := range p.nextCh {
                switch notification := next.(type) {
                case updateNotification:
                    p.handler.OnUpdate(notification.oldObj, notification.newObj)
                case addNotification:
                    p.handler.OnAdd(notification.newObj)
                case deleteNotification:
                    p.handler.OnDelete(notification.oldObj)
                default:
                    utilruntime.HandleError(fmt.Errorf("unrecognized notification"))
                }
            }
            return true, nil
        })
        
        if err == nil {
            close(stopCh)
        }
    }, 1*time.Minute, stopCh)
}

The pop method manages the flow of notifications from the sharedProcessor to the listener:

func (p *processorListener) pop() {
    defer utilruntime.HandleCrash()
    defer close(p.nextCh)
    
    var nextCh chan<- interface{}
    var notification interface{}
    
    for {
        select {
        case nextCh <- notification:
            notification, ok := p.pendingNotifications.ReadOne()
            if !ok {
                nextCh = nil
            }
        case notificationToAdd, ok := <-p.addCh:
            if !ok {
                return
            }
            if notification == nil {
                notification = notificationToAdd
                nextCh = p.nextCh
            } else {
                p.pendingNotifications.WriteOne(notificationToAdd)
            }
        }
    }
}

Event Distribution

The distribute method forwards events to registered listeners:

func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
    p.listenersLock.RLock()
    defer p.listenersLock.RUnlock()

    if sync {
        for _, listener := range p.syncingListeners {
            listener.add(obj)
        }
    } else {
        for _, listener := range p.listeners {
            listener.add(obj)
        }
    }
}

Handling Deltas

The HandleDeltas method processes incoming deltas and updates both the indexer and distributes events:

func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
    s.blockDeltas.Lock()
    defer s.blockDeltas.Unlock()

    for _, d := range obj.(Deltas) {
        switch d.Type {
        case Sync, Added, Updated:
            isSync := d.Type == Sync
            s.cacheMutationDetector.AddObject(d.Object)
            
            if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
                if err := s.indexer.Update(d.Object); err != nil {
                    return err
                }
                s.processor.distribute(
                    updateNotification{oldObj: old, newObj: d.Object}, 
                    isSync,
                )
            } else {
                if err := s.indexer.Add(d.Object); err != nil {
                    return err
                }
                s.processor.distribute(
                    addNotification{newObj: d.Object}, 
                    isSync,
                )
            }
        case Deleted:
            if err := s.indexer.Delete(d.Object); err != nil {
                return err
            }
            s.processor.distribute(
                deleteNotification{oldObj: d.Object}, 
                false,
            )
        }
    }
    return nil
}

Running the Informer

The Run method initializes all components and starts the synchronization process:

func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    
    // Initialize DeltaFIFO
    fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)

    cfg := &Config{
        Queue:            fifo,
        ListerWatcher:    s.listerWatcher,
        ObjectType:       s.objectType,
        FullResyncPeriod: s.resyncCheckPeriod,
        RetryOnError:     false,
        ShouldResync:     s.processor.shouldResync,
        Process:          s.HandleDeltas,
    }

    func() {
        s.startedLock.Lock()
        defer s.startedLock.Unlock()
        s.controller = New(cfg)
        s.controller.(*controller).clock = s.clock
        s.started = true
    }()

    processorStopCh := make(chan struct{})
    var wg wait.Group
    defer wg.Wait()
    defer close(processorStopCh)
    
    wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
    wg.StartWithChannel(processorStopCh, s.processor.run)

    defer func() {
        s.startedLock.Lock()
        defer s.startedLock.Unlock()
        s.stopped = true
    }()
    
    s.controller.Run(stopCh)
}

The sharedProcessor.run method enitializes all listener goroutines:

func (p *sharedProcessor) run(stopCh <-chan struct{}) {
    func() {
        p.listenersLock.RLock()
        defer p.listenersLock.RUnlock()
        
        for _, listener := range p.listeners {
            p.wg.Start(listener.run)
            p.wg.Start(listener.pop)
        }
        p.listenersStarted = true
    }()
    <-stopCh
}

Summary

The Informer mechanism in Kubernetes client-go provides a sophisticated yet elegant solution for resource synchronization. From the Reflector's ListAndWatch operations through the Delta FIFO queue, to the Controller's processing loop, and finally to the SharedIndexInformer's event distribution system, each component plays a crucial role in maintaining cluster state consistency while minimizing API server load through efficient caching and event-driven patterns.