Concrètement, un contrôleur classique se décompose en deux briques distinctes :

• l'Informer : il écoute l'API Server, maintient un cache et produit des événements de modification ;
• le contrôleur proprement dit : il consomme ces événements et exécute la logique de réconciliation (Reconcile).

Vue d'ensemble des composants

Le module Informer de client-go (version 0.31) repose sur cinq composants principaux :

1. Reflector : dialogue avec l'API Server via List puis Watch ;
2. DeltaFIFO : file d'attente ordonnée des deltas par clé d'objet ;
3. Indexer : cache local avec indexation des ressources ;
4. Controller (aussi appelé Informer en pratique) : boucle de consommation de la file ;
5. WorkQueue (côté utilisateur) : tampon de traitement, découplage et régulation du flux.

Reflector : synchronisation avec l'API Server

Un Reflector est attaché à un seul type de ressource. Il reçoit un ListerWatcher dont le rôle est d'encapsuler les appels List et Watch pour ce type précis.

type ResourceWatcher struct {
    List  func(metav1.ListOptions) (runtime.Object, error)
    Watch func(metav1.ListOptions) (watch.Interface, error)
}

func NewReflector(opts ReflectorOptions, lw ListerWatcher, typ reflect.Type, store Store) *Reflector {
    return &Reflector{
        listerWatcher: lw,
        expectedType:  typ,
        store:         store,
        options:       opts,
    }
}

Le constructeur attend donc une façon d'énumérer et de surveiller les objets, ainsi que le cache local (store) qui recevra les modifications.

Deux stratégies d'initialisation

Depuit les versions récentes, client-go distingue deux approches pour l'obtention de l'état initial :

• List-Watch classique : une requête List paginée, fermeture de connexion, puis ouverture d'un Watch continu.
• Watch streaming : une seule connexion Watch qui commence par envoyer tous les objets existants comme événements Added, marque le point de départ par un bookmark, puis continue avec les changements incrémentaux.

Le mode streaming réduit la pression sur l'API Server en évitant le gros List initial. Il repose sur l'option SendInitialEvents=true et une resource version d'ancrage.

// Initialisation par watch streaming
opts := metav1.ListOptions{
    ResourceVersion:      lastKnownRV,
    ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan,
    SendInitialEvents:    pointer.Bool(true),
    AllowWatchBookmarks:  true,
}

w, err := r.watcher.Watch(opts)
if err != nil {
    // bascule vers le list-watch classique
    return r.fallbackList(stopCh)
}

Le mode classique et la pagination

Quand le mode streaming n'est pas disponible, le reflector effectue un List paginé via un pager. Par défaut, chaque page contient jusqu'à 500 objets. La première requête utilise généralement ResourceVersion="0" pour lire depuis le cache de l'API Server ; si celui-ci est absent, le système bascule vers le stockage etcd et active la pagination complète.

p := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
    return r.watcher.List(opts)
}))

if r.WatchListPageSize != 0 {
    p.PageSize = r.WatchListPageSize
}

list, paginated, err := p.ListWithAlloc(ctx, opts)

Une fois le snapshot reçu, les éléments remplacent l'ancien contenu du cache par un Replace, puis un Watch permanent prend le relais.

Watch et resync

Après l'initialisation, le reflector boucle sur le flux Watch. Il détecte les ajouts, modifications et suppressions et les transmet à DeltaFIFO. En parallèle, un mécanisme de resync périodique ré-émet les objets du cache local sous forme d'événements Sync, permettant au contrôleur de vérifier l'état réel même si certains changements ont été manqués.

DeltaFIFO : file des différences

DeltaFIFO est la file d'attente intermédiaire entre le reflector et le consommateur. Elle ne stocke pas directement des objets isolés, mais des suites de deltas par clé.

type EventType string

const (
    EventAdded    EventType = "Added"
    EventUpdated  EventType = "Updated"
    EventDeleted  EventType = "Deleted"
    EventReplaced EventType = "Replaced"
    EventSync     EventType = "Sync"
)

type Delta struct {
    Action EventType
    Object interface{}
}

type Deltas []Delta

Quand on retire un élément de la file, on obtient tous les deltas d'une même clé, ce qui permet de traiter l'historique récent d'une ressource en une seule opération et d'éviter les traitements intermédiaires inutiles.

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()

    for {
        id := f.queue[0]
        f.queue = f.queue[1:]

        item := f.items[id]
        delete(f.items, id)

        err := process(item)
        return item, err
    }
}

Indexer : cache local et indexation

L'Indexer conserve le dernier état connu de chaque ressource et offre des index permettant de retrouver rapidement les objets selon des critères tels que le namespace, le nom du nœud, etc.

type Indexer interface {
    Store
    IndexKeys(indexName, value string) ([]string, error)
    ByIndex(indexName, value string) ([]interface{}, error)
    ListIndexFuncValues(indexName string) []string
    AddIndexers(newIndexers Indexers) error
}

En interne, l'implémentation par défaut (cache) s'appuie sur un threadSafeMap qui maintient la carte clé → objet et les différents index inversés. L'indexation peut être comparée à un index inversé simple : pour chaque propriété, on conserve la liste des clés d'objets correspondant à chaque valeur.

Controller : la boucle de consommation

Dans client-go, l'interface Controller est ce que l'on appelle couramment un Informer. Son rôle est double : lancer le Reflector qui alimente la file, puis consommer cette file en continu.

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

La méthode Run démarre le reflector et la boucle de traitement. Cette dernière retire les deltas de DeltaFIFO, met à jour le cache local via l'Indexer, puis invoque les gestionnaires d'événements utilisateur.

func (c *controller) Run(stopCh <-chan struct{}) {
    r := NewReflector(c.config.Options, c.config.ListerWatcher, c.config.Type, c.config.Queue)
    go r.Run(stopCh)

    wait.Until(c.processLoop, time.Second, stopCh)
}

La fonction processDeltas distingue les événements et applique les mises à jour au cache avant de notifier les handlers.

func processDeltas(handler ResourceEventHandler, store Store, deltas Deltas) error {
    for _, d := range deltas {
        switch d.Action {
        case EventAdded, EventUpdated, EventSync, EventReplaced:
            old, exists, _ := store.Get(d.Object)
            if exists {
                store.Update(d.Object)
                handler.OnUpdate(old, d.Object)
            } else {
                store.Add(d.Object)
                handler.OnAdd(d.Object)
            }
        case EventDeleted:
            store.Delete(d.Object)
            handler.OnDelete(d.Object)
        }
    }
    return nil
}

SharedIndexInformer : mutualisation du cache et des connexions

Créer un Informer distinct pour chaque besoin d'écoute d'une même ressource multiplierait les connexions vers l'API Server et dupliquerait le cache. Pour cette raison, client-go fournit le SharedIndexInformer.

Le principe est le fan-out : un seul reflector et un seul cache alimentent plusieurs gestionnaires d'événements. L'implémentation utilise le patron de conception proxy : le SharedIndexInformer s'enregistre comme unique ResourceEventHandler auprès du contrôleur interne, puis redistribue chaque événement à tous les gestionnaires ajoutés par l'utilisateur.

type SharedInformer struct {
    handlers []ResourceEventHandler
}

func (s *SharedInformer) AddEventHandler(h ResourceEventHandler) {
    s.handlers = append(s.handlers, h)
}

func (s *SharedInformer) OnAdd(obj interface{}) {
    for _, h := range s.handlers {
        h.OnAdd(obj)
    }
}

func (s *SharedInformer) OnUpdate(oldObj, newObj interface{}) {
    for _, h := range s.handlers {
        h.OnUpdate(oldObj, newObj)
    }
}

func (s *SharedInformer) OnDelete(obj interface{}) {
    for _, h := range s.handlers {
        h.OnDelete(obj)
    }
}

WorkQueue : réguler avant de réconcilier

Dans un contrôleur, on ne traite pas les événements directement dans le handler de l'Informer. On les place dans une WorkQueue qui offre plusieurs propriétés utiles :

• Déduplication : un même objet peut être ajouté plusieurs fois, mais traité une seule fois.
• Tampon : absorption des pics de trafic.
• Limitation de débit : évitement de tempêtes de requêtes vers l'API Server.
• Retards progressifs : en cas d'échec de réconciliation, le traitement d'une clé peut être reporté selon une exponentielle.

queue := workqueue.NewRateLimitingQueue(
    workqueue.DefaultControllerRateLimiter(),
)

Le limiteur par défaut combine deux mécanismes : un limiteur par clé à retour exponentiel et un seau de tokens global. Le premier retarde une clé problématique de plus en plus longtemps, tandis que le second plafonne le débit global de l'ensemble des événements. Cette combinaison protège à la fois l'API Server et les ressources externes manipulées par le contrôleur.

Le Reconciler d'un contrôleur moderne, tel que généré par controller-runtime ou kubebuilder, s'alimente finalement depuis cette même chaîne Informer → WorkQueue → traitement, ce qui explique la réactivité et la fiabilité des opérateurs Kubernetes.