README
¶
pkg/k8s/watcher
Two complementary Kubernetes watchers built on top of k8s.io/client-go informers.
Overview
| Watcher | Use case |
|---|---|
*Watcher (created via watcher.New) |
Bulk: subscribe to every object of a GVR (optionally filtered by label/namespace selectors), index them with pkg/k8s/indexer, hand them to a pkg/k8s/store backend, and call back per debounced batch of changes |
*SingleWatcher (created via watcher.NewSingle) |
Targeted: watch one specific named resource (the controller's HAProxyTemplateConfig CRD or the credentials Secret) and fire an immediate callback per change — no debouncing, no store |
Both types take their config via the structs in pkg/k8s/types so they can be wired uniformly from the controller.
Quick Start
Bulk watcher
import (
"gitlab.com/haproxy-haptic/haptic/pkg/k8s/client"
"gitlab.com/haproxy-haptic/haptic/pkg/k8s/types"
"gitlab.com/haproxy-haptic/haptic/pkg/k8s/watcher"
)
c, _ := client.New(client.Config{})
cfg := types.WatcherConfig{
GVR: schema.GroupVersionResource{Group: "networking.k8s.io", Version: "v1", Resource: "ingresses"},
IndexBy: []string{"metadata.namespace", "metadata.name"},
OnChange: func(store types.Store, stats types.ChangeStats) {
if stats.IsInitialSync { return } // skip the bulk-load event
// react to real changes
},
OnSyncComplete: func(store types.Store, count int) {
log.Info("ingresses initial sync complete", "count", count)
},
// DebounceInterval defaults to types.DefaultDebounceInterval (2s)
}
w, err := watcher.New(cfg, c, slog.Default())
if err != nil { /* ... */ }
go w.Start(ctx)
n, err := w.WaitForSync(ctx) // returns the initial-list count
Single watcher
cfg := &types.SingleWatcherConfig{
GVR: schema.GroupVersionResource{Version: "v1", Resource: "secrets"},
Namespace: "haptic",
Name: "haproxy-credentials",
OnChange: func(obj any) error { // typed as types.OnResourceChangeCallback
// obj is the live *unstructured.Unstructured (or nil on delete)
return nil
},
}
sw, err := watcher.NewSingle(cfg, c)
if err != nil { /* ... */ }
go sw.Start(ctx)
err = sw.WaitForSync(ctx)
Behavioural Notes
- Debouncing is leading-edge with a refractory period (the first change in a quiet period fires immediately; further changes inside the window are batched). See
pkg/controller/reconciler/CLAUDE.mdfor why this matters during rolling deploys. - Initial sync behaviour is controlled by
CallOnChangeDuringSync(defaultfalse): with the default,OnChangeis suppressed during the bulk load and the consumer learns the load is finished from the parallelOnSyncCompletecallback. WithCallOnChangeDuringSync: true,OnChangefires for every change during the initial list — each call'sstats.IsInitialSyncistrueso consumers canif stats.IsInitialSync { return }to skip them when they only care about post-sync deltas. SingleWatcheris not debounced — itsOnChangecallback (typed asOnResourceChangeCallback, distinct from the bulk watcher'sOnChangeCallback) runs on every event, intentionally, because credential and CRD updates need to take effect immediately.
See Also
pkg/k8s/client— supplies the typed + dynamic clients required by both constructorspkg/k8s/store— bulk-watcher backing storagepkg/k8s/indexer— JSONPath key extraction + field filteringpkg/k8s/types—WatcherConfig/SingleWatcherConfig/Storeinterface
License
Apache-2.0 — see root LICENSE.
Documentation
¶
Overview ¶
Package watcher provides Kubernetes resource watching with indexing, field filtering, and debounced callbacks.
Package watcher provides Kubernetes resource watching with indexing, field filtering, and debounced callbacks.
This package integrates all k8s subpackages to provide a high-level interface for watching Kubernetes resources and reacting to changes.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Debouncer ¶
type Debouncer struct {
// contains filtered or unexported fields
}
Debouncer batches rapid resource changes into a single callback invocation.
Uses leading-edge triggering with a refractory period: if no callback has fired recently (within interval), the first change triggers an immediate callback. Subsequent changes within the refractory period are batched until the interval expires.
This ensures fast response to isolated changes while still batching rapid successive changes (e.g., during initial sync or cluster restarts).
Thread-safe for concurrent access.
func NewDebouncer ¶
func NewDebouncer(interval time.Duration, callback types.OnChangeCallback, store types.Store, suppressDuringSync bool) *Debouncer
NewDebouncer creates a new debouncer with the specified interval and callback.
The callback will be invoked at most once per interval, with aggregated statistics about all changes that occurred during that interval.
Parameters:
- interval: Minimum time between callback invocations
- callback: Function to call with aggregated changes
- store: Store to pass to callback
- suppressDuringSync: If true, callbacks are suppressed during initial sync
func (*Debouncer) Flush ¶
func (d *Debouncer) Flush()
Flush immediately invokes the callback with current statistics.
This is useful during shutdown or sync completion to ensure pending changes are processed. Flush always invokes the callback regardless of suppressDuringSync setting.
func (*Debouncer) GetInitialCount ¶
GetInitialCount returns the number of resources created during initial sync.
This should be called after SetSyncMode(false) to get the accurate count of pre-existing resources that were loaded.
func (*Debouncer) RecordCreate ¶
func (d *Debouncer) RecordCreate()
RecordCreate records a resource creation.
func (*Debouncer) RecordDelete ¶
func (d *Debouncer) RecordDelete()
RecordDelete records a resource deletion.
func (*Debouncer) RecordUpdate ¶
func (d *Debouncer) RecordUpdate()
RecordUpdate records a resource update.
func (*Debouncer) SetSyncMode ¶
SetSyncMode enables or disables initial sync mode.
When sync mode is enabled, callbacks receive IsInitialSync=true. When disabled, callbacks receive IsInitialSync=false (real-time changes).
When transitioning from sync mode to normal mode, any pending changes are flushed to ensure ADD events accumulated during sync are delivered.
type SingleWatcher ¶
type SingleWatcher struct {
// contains filtered or unexported fields
}
SingleWatcher watches a single named Kubernetes resource and invokes callbacks on changes.
Unlike the bulk Watcher (which watches collections of resources with indexing and debouncing), SingleWatcher is optimized for watching one specific resource:
- No indexing or store (just the current resource)
- No debouncing (immediate callbacks)
- Simpler API (resource object in callback, not Store)
Callback Behavior:
- During initial sync: All events (Add/Update/Delete) are suppressed (no callbacks)
- After sync completes: All events trigger callbacks immediately
- Use WaitForSync() to ensure initial state is loaded before relying on callbacks
Thread Safety:
- Safe for concurrent use
- Callbacks may be invoked from multiple goroutines
- Users should ensure callbacks are thread-safe
This is ideal for watching configuration or credentials in a specific ConfigMap or Secret.
func NewSingle ¶
func NewSingle(cfg *types.SingleWatcherConfig, k8sClient *client.Client) (*SingleWatcher, error)
NewSingle creates a new single-resource watcher with the provided configuration.
Returns an error if:
- Configuration validation fails
- Client is nil
- Informer creation fails
Example:
watcher, err := watcher.NewSingle(types.SingleWatcherConfig{
GVR: schema.GroupVersionResource{
Group: "",
Version: "v1",
Resource: "configmaps",
},
Namespace: "default",
Name: "haproxy-config",
OnChange: func(obj any) error {
cm := obj.(*corev1.ConfigMap)
// Process configuration
return nil
},
}, k8sClient)
func (*SingleWatcher) LastWatchError ¶
func (w *SingleWatcher) LastWatchError() time.Time
LastWatchError returns the time of the most recent watch-connection error, or the zero time if none has occurred. The Reflector retries automatically; this accessor exists purely for observability / health reporting.
func (*SingleWatcher) Start ¶
func (w *SingleWatcher) Start(ctx context.Context) error
Start begins watching the resource.
This method blocks until the context is cancelled or an error occurs. Initial sync is performed before continuing.
This method is idempotent - calling it multiple times is safe. The initialization logic (starting informer, syncing) only runs once, but each call will block until the context is cancelled.
func (*SingleWatcher) Stop ¶
func (w *SingleWatcher) Stop() error
Stop stops watching the resource.
This method is idempotent and safe to call multiple times.
func (*SingleWatcher) WaitForSync ¶
func (w *SingleWatcher) WaitForSync(ctx context.Context) error
WaitForSync blocks until initial synchronization is complete.
This is useful when you need to wait for the watcher to have the current state of the resource before performing operations that depend on it.
Returns an error if sync fails or context is cancelled.
Example:
watcher, _ := watcher.NewSingle(cfg, client)
go watcher.Start(ctx)
err := watcher.WaitForSync(ctx)
if err != nil {
slog.Error("Watcher sync failed", "error", err)
os.Exit(1)
}
slog.Info("Watcher synced, resource is available")
type Watcher ¶
type Watcher struct {
// contains filtered or unexported fields
}
Watcher watches Kubernetes resources and maintains an indexed store.
Resources are: - Filtered by namespace and label selector - Filtered by field selector (client-side JSONPath evaluation) - Indexed using JSONPath expressions for O(1) lookups - Filtered to remove unnecessary fields - Stored in memory or API-backed cache
Changes are debounced and delivered via callback with aggregated statistics.
func (*Watcher) ForceSync ¶
func (w *Watcher) ForceSync()
ForceSync forces an immediate callback with current statistics.
func (*Watcher) IsSynced ¶
IsSynced returns true if initial synchronization has completed.
This provides a non-blocking way to check if the store is fully populated.
func (*Watcher) LastWatchError ¶
LastWatchError returns the time of the most recent watch-connection error, or the zero time if none has occurred. Observability only — retry is the Reflector's job.
func (*Watcher) Start ¶
Start begins watching resources.
This method blocks until the context is cancelled or an error occurs. Initial sync is performed before continuing, and OnSyncComplete is called if configured.
func (*Watcher) Stop ¶
Stop stops watching resources.
This method is idempotent and safe to call multiple times (mirrors SingleWatcher.Stop) — a double close of stopCh would otherwise panic.
func (*Watcher) WaitForSync ¶
WaitForSync blocks until initial synchronization is complete.
This is useful when you need to wait for the store to be fully populated before performing operations that depend on complete data.
Returns:
- The number of resources loaded during initial sync
- An error if sync fails or context is cancelled
Example:
watcher, _ := watcher.New(cfg, client)
go watcher.Start(ctx)
count, err := watcher.WaitForSync(ctx)
if err != nil {
slog.Error("Watcher sync failed", "error", err)
os.Exit(1)
}
slog.Info("Watcher synced", "resource_count", count)
Source Files