Add status watch label & field selector filters

Using a label filter significantly cuts down on watch events and memory used by the informer's watch cache. But you'll need to set the labels on the objects yourself before providing them to the applier.

Author: karlkfi

pkg/apply/applier_builder.go

@@ -86,3 +86,8 @@ func (b *ApplierBuilder) WithStatusWatcher(statusWatcher watcher.StatusWatcher)
 b.statusWatcher = statusWatcher
 return b
 }
+
+func (b *ApplierBuilder) WithStatusWatcherFilters(filters *watcher.Filters) *ApplierBuilder {
+b.statusWatcherFilters = filters
+return b
+}

pkg/apply/builder.go

@@ -27,6 +27,7 @@ type commonBuilder struct {
 restConfig *rest.Config
 unstructuredClientForMapping func(*meta.RESTMapping) (resource.RESTClient, error)
 statusWatcher watcher.StatusWatcher
+statusWatcherFilters *watcher.Filters
 }
 
 func (cb *commonBuilder) finalize() (*commonBuilder, error) {
@@ -78,7 +79,15 @@ func (cb *commonBuilder) finalize() (*commonBuilder, error) {
 cx.unstructuredClientForMapping = cx.factory.UnstructuredClientForMapping
 }
 if cx.statusWatcher == nil {
-cx.statusWatcher = watcher.NewDefaultStatusWatcher(cx.client, cx.mapper)
+statusWatcher := watcher.NewDefaultStatusWatcher(cx.client, cx.mapper)
+if cx.statusWatcherFilters != nil {
+statusWatcher.Filters = cx.statusWatcherFilters
+}
+cx.statusWatcher = statusWatcher
+} else if cx.statusWatcherFilters != nil {
+// If you want to use a custom status watcher with a label selector,
+// configure it before injecting the status watcher.
+return nil, errors.New("status watcher and status watcher filters must not both be provided")
 }
 return &cx, nil
 }

pkg/kstatus/watcher/default_status_watcher.go

@@ -11,6 +11,7 @@ import (
 "k8s.io/apimachinery/pkg/api/meta"
 "k8s.io/apimachinery/pkg/runtime/schema"
 "k8s.io/client-go/dynamic"
+"k8s.io/client-go/tools/cache"
 "k8s.io/klog/v2"
 "sigs.k8s.io/cli-utils/pkg/kstatus/polling/clusterreader"
 "sigs.k8s.io/cli-utils/pkg/kstatus/polling/engine"
@@ -43,6 +44,15 @@ type DefaultStatusWatcher struct {
 // required for computing parent object status, to compensate for
 // controllers that aren't following status conventions.
 ClusterReader engine.ClusterReader
+
+// Indexers control how the watch cache is indexed, allowing namespace
+// filtering and field selectors. If you watch at namespace scope, you must
+// provide the namespace indexer. If you specify a field selector filter,
+// you must also provide an indexer for that field.
+Indexers cache.Indexers
+
+// Filters allows filtering the objects being watched.
+Filters *Filters
 }
 
 var _ StatusWatcher = &DefaultStatusWatcher{}
@@ -60,6 +70,7 @@ func NewDefaultStatusWatcher(dynamicClient dynamic.Interface, mapper meta.RESTMa
 DynamicClient: dynamicClient,
 Mapper: mapper,
 },
+Indexers: DefaultIndexers(),
 }
 }
 
@@ -88,13 +99,18 @@ func (w *DefaultStatusWatcher) Watch(ctx context.Context, ids object.ObjMetadata
 }
 
 informer := &ObjectStatusReporter{
-InformerFactory: NewDynamicInformerFactory(w.DynamicClient, w.ResyncPeriod),
-Mapper: w.Mapper,
-StatusReader: w.StatusReader,
-ClusterReader: w.ClusterReader,
-Targets: targets,
-ObjectFilter: &AllowListObjectFilter{AllowList: ids},
-RESTScope: scope,
+InformerFactory: &DynamicInformerFactory{
+Client: w.DynamicClient,
+ResyncPeriod: w.ResyncPeriod,
+Indexers: w.Indexers,
+Filters: w.Filters,
+},
+Mapper: w.Mapper,
+StatusReader: w.StatusReader,
+ClusterReader: w.ClusterReader,
+Targets: targets,
+ObjectFilter: &AllowListObjectFilter{AllowList: ids},
+RESTScope: scope,
 }
 return informer.Start(ctx)
 }