k8s/

pkg/k8s

Pure Kubernetes integration library. Two watchers, two stores, a JSONPath indexer, and a thin client-go wrapper. No EventBus dependency — the controller wraps this package in event adapters (pkg/controller/resourcewatcher) when it wants event-driven behaviour.

Module path: gitlab.com/haproxy-haptic/haptic. The authoritative interface is the source (go doc ./pkg/k8s/...); this README is a short orientation.

Sub-Package Map

Two Watchers

watcher.Watcher — for collections (Ingress, Service, EndpointSlice…). Watches via SharedInformerFactory, indexes by JSONPath into a types.Store, debounces bursts before calling OnChange.

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"

    "k8s.io/apimachinery/pkg/runtime/schema"
)

k8sClient, err := client.New(client.Config{})
if err != nil { /* ... */ }

cfg := types.WatcherConfig{
    GVR: schema.GroupVersionResource{
        Group: "networking.k8s.io", Version: "v1", Resource: "ingresses",
    },
    IndexBy:      []string{"metadata.namespace", "metadata.name"},
    IgnoreFields: []string{"metadata.managedFields"},
    OnChange: func(store types.Store, stats types.ChangeStats) {
        if stats.IsInitialSync { return }
        // react to real-time change…
    },
    OnSyncComplete: func(store types.Store, initialCount int) {
        // initial bulk load finished
    },
}

w, err := watcher.New(cfg, k8sClient, logger)
if err != nil { /* ... */ }
go w.Start(ctx)                                 // Start blocks; run in its own goroutine.
if _, err := w.WaitForSync(ctx); err != nil { /* ... */ }

Key WatcherConfig fields (see pkg/k8s/types/types.go for full docs):

• IndexBy — JSONPath expressions that form the composite lookup key. Order matters: .Get(partial…) does a prefix scan when fewer keys are supplied.
• StoreType — StoreTypeMemory (default) or StoreTypeCached. CacheTTL only applies to the cached store and defaults to ~2 min (auto-derived from drift prevention at the controller layer).
• DebounceInterval — default types.DefaultDebounceInterval (2s). Set to 0 for default, or a positive duration to override.
• CallOnChangeDuringSync — when true, OnChange fires during initial bulk load with stats.IsInitialSync == true; when false, only OnSyncComplete fires for the initial set.
• LabelSelector / FieldSelector — LabelSelector is a server-side filter (*metav1.LabelSelector, pushed to the API server). FieldSelector is a client-side JSONPath filter (evaluated after the list is fetched); it supports any expression but fetches more data than LabelSelector. Two earlier conversions land its value: the CRD's labelSelector is a string like "app=foo,env=prod" (see pkg/apis/haproxytemplate/v1alpha1/types_config.go), pkg/controller/conversion.parseLabelSelector turns it into a map[string]string on the internal config.WatchedResource, and pkg/controller/resourcewatcher/watcher.go:134 finally wraps that map as &metav1.LabelSelector{MatchLabels: m} for this layer. For richer selectors (MatchExpressions) construct the *metav1.LabelSelector directly when bypassing the CRD path.

watcher.SingleWatcher — for one specific named resource (the controller's own HAProxyTemplateConfig, its credentials Secret, etc.). No indexing, no debounce, immediate callbacks.

cfg := &types.SingleWatcherConfig{
    GVR:       schema.GroupVersionResource{Group: "", Version: "v1", Resource: "configmaps"},
    Namespace: "haptic",
    Name:      "haptic-config",
    // OnChange is types.OnResourceChangeCallback — `func(obj any) error`,
    // distinct from the bulk watcher's `func(types.Store, types.ChangeStats)`.
    // Returning an error logs it and keeps the watcher alive.
    OnChange:  func(obj any) error { return nil },
}
w, err := watcher.NewSingle(cfg, k8sClient)

Two Stores

Both implement types.Store. Get(keys...) with fewer keys than IndexBy returns all resources whose composite key starts with those keys (prefix scan) — useful for one-to-many lookups like "all EndpointSlices for service X".

Initial Sync

Every watcher distinguishes the initial bulk load from live changes. Call WaitForSync(ctx) before the first reconciliation; use stats.IsInitialSync inside OnChange to skip premature work. pkg/controller/indextracker coordinates this across all watchers so the reconciliation pipeline only starts after every store is populated.

Common Pitfalls

• Not waiting for sync. Calling store.List() before WaitForSync returns can miss pre-existing resources.
• Field selectors on labels. "metadata.labels['app']=x" is not valid — use LabelSelector instead.
• Long work in callbacks. Callbacks run on the watcher's informer goroutine. Hand work off to a channel / goroutine; don't block.
• List() on a cached store. Force-fetches every reference. Use Get() with the exact composite key instead. (In templates, use resources.X.GetSingle() / resources.X.Fetch() from the render-context layer.)
• Invalid JSONPath. indexer.ValidateJSONPath() is called at watcher construction — invalid expressions fail fast with the field name in the error.

See pkg/k8s/CLAUDE.md for the detailed pitfall catalogue, testing patterns (fake clientset, initial-sync assertions), and the walkthrough for adding a new watched resource type.

Testing

go test ./pkg/k8s/...           # unit tests
go test ./pkg/k8s/... -race     # race detector

Unit tests use k8s.io/client-go/kubernetes/fake; integration tests live under tests/.

See Also

• pkg/k8s/CLAUDE.md — developer context, adding-a-resource walkthrough, pitfalls in depth
• pkg/k8s/leaderelection/README.md — pure leader election component
• pkg/controller/resourcewatcher — the event adapter that wires Watchers into the EventBus
• docs/controller/docs/watching-resources.md — user-facing documentation for the CRD side

License

Apache-2.0 — see root LICENSE.