The Reflector in Kubernetes' client-go is a core component responsible for synchronizing resources from the API server into a local store—typically a DeltaFIFO queue. It achieves this by performing an initial list operation followed by a continuous watch, ensuring that all changes to watched resources are captured and enqueued for processing by controllers.

Entry Point: Reflector.Run()

The lifecycle of a Reflector begins with its Run() method:

func (r *Reflector) Run(stopCh <-chan struct{}) {
    klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
    wait.BackoffUntil(func() {
        if err := r.ListAndWatch(stopCh); err != nil {
            r.watchErrorHandler(r, err)
        }
    }, r.backoffManager, true, stopCh)
    klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}

This method uses exponential backoff to retry ListAndWatch in case of transient failures, enhancing resilience against temporary API server unavailability.

Core Logic: Reflector.ListAndWatch()

The ListAndWatch method first lists all current instances of a resource type and then initiates a watch starting from the returned resource version. Key steps include:

1. Performing a paginated list using a pager, which handles large result sets efficiently.
2. Handling expired or too-large resource version errors by retrying with a fresh list request.
3. Extracting items and their resource version, then syncing them into the store via syncWith, which enqueues a Sync delta.
4. Starting a background goroutine to handle periodic resyncs if enabled.
5. Entering a loop to establish and maintain a watch connection with randomized timeouts (5–10 minutes) and support for bookmarks to reduce server load.

If the watch disconnects due to non-fatal errors (e.g., 429 Too Many Requests), it backs off and retries. Otherwise, it returns control to Run() for another attempt.

Event Processing: watchHandler()

The watchHandler processes events from the watch stream:

func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
    defer w.Stop()
    eventCount := 0
loop:
    for {
        select {
        case <-stopCh:
            return errorStopRequested
        case err := <-errc:
            return err
        case event, ok := <-w.ResultChan():
            if !ok { break loop }
            if event.Type == watch.Error { return apierrors.FromObject(event.Object) }

            // Validate object type and GVK if specified
            if r.expectedType != nil { /* ... */ }
            if r.expectedGVK != nil { /* ... */ }

            meta, err := meta.Accessor(event.Object)
            if err != nil { /* log and continue */ }

            newRV := meta.GetResourceVersion()
            switch event.Type {
            case watch.Added:    r.store.Add(event.Object)
            case watch.Modified: r.store.Update(event.Object)
            case watch.Deleted:  r.store.Delete(event.Object)
            case watch.Bookmark: // no-op, only updates RV
            default:             // log unknown type
            }

            *resourceVersion = newRV
            r.setLastSyncResourceVersion(newRV)
            if updater, ok := r.store.(ResourceVersionUpdater); ok {
                updater.UpdateResourceVersion(newRV)
            }
            eventCount++
        }
    }

    if r.clock.Since(start) < time.Second && eventCount == 0 {
        return fmt.Errorf("very short watch: %s: Unexpected watch close", r.name)
    }
    return nil
}

Each evant is validated, processed according to its type, and used to update the tracked resource version. Bookmarks are acknowledged but do not trigger store mutations—they only advance the resource version.

Initialization: NewReflector()

A Reflector is constructed with a ListerWatcher, expected object type, backing store (Store interface, usually DeltaFIFO), and optional resync period:

func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
    realClock := &clock.RealClock{}
    r := &Reflector{
        name:                   name,
        listerWatcher:          lw,
        store:                  store,
        backoffManager:         wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),
        initConnBackoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),
        resyncPeriod:           resyncPeriod,
        clock:                  realClock,
        watchErrorHandler:      WatchErrorHandler(DefaultWatchErrorHandler),
    }
    r.setExpectedType(expectedType)
    return r
}

Two independent backoff managers handle genarel retries and initial connection failures, respectively.

The Reflector does not modify existing entries in the store directly. Instead, it appends deltas (Add, Update, Delete, Sync) to the DeltaFIFO, relying on the FIFO’s deduplication and ordering logic. The watch is not persistent—it restarts after each disconnection using the latest known resource version, ensuring continuity while accommodating API server limitations like watch expiration.