Introduce watch-filter flag as predicate filter + filter out paused objects

chrischdi · chrischdi · commit db0fa42320f0 · 2023-08-02T10:13:45.000+02:00
diff --git a/controllers/serviceaccount_controller.go b/controllers/serviceaccount_controller.go
@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
+	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -121,6 +122,7 @@ func AddServiceAccountProviderControllerToManager(ctx *context.ControllerManager
 				return requests
 			}),
 		).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		Complete(r)
 }
 
diff --git a/controllers/servicediscovery_controller.go b/controllers/servicediscovery_controller.go
@@ -37,6 +37,7 @@ import (
 	clusterutilv1 "sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
+	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -120,6 +121,7 @@ func AddServiceDiscoveryControllerToManager(ctx *context.ControllerManagerContex
 				&vmwarev1.VSphereCluster{},
 				handler.OnlyControllerOwner(),
 			)).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		Complete(r)
 }
 
diff --git a/controllers/vspherecluster_controller.go b/controllers/vspherecluster_controller.go
@@ -102,6 +102,7 @@ func AddClusterControllerToManager(ctx *context.ControllerManagerContext, mgr ma
 				&vmwarev1.VSphereMachine{},
 				handler.EnqueueRequestsFromMapFunc(reconciler.VSphereMachineToCluster),
 			).
+			WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 			Complete(reconciler)
 	}
 
@@ -159,6 +160,7 @@ func AddClusterControllerToManager(ctx *context.ControllerManagerContext, mgr ma
 			&source.Channel{Source: ctx.GetGenericEventChannelFor(clusterControlledTypeGVK)},
 			&handler.EnqueueRequestForObject{},
 		).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		WithEventFilter(predicates.ResourceIsNotExternallyManaged(reconciler.Logger)).
 		Build(reconciler)
 	if err != nil {
diff --git a/controllers/vsphereclusteridentity_controller.go b/controllers/vsphereclusteridentity_controller.go
@@ -30,6 +30,7 @@ import (
 	clusterutilv1 "sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
+	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -71,6 +72,7 @@ func AddVsphereClusterIdentityControllerToManager(ctx *context.ControllerManager
 	return ctrl.NewControllerManagedBy(mgr).
 		For(identityControlledType).
 		WithOptions(options).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		Complete(reconciler)
 }
 
diff --git a/controllers/vspheredeploymentzone_controller.go b/controllers/vspheredeploymentzone_controller.go
@@ -32,6 +32,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/collections"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
+	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -88,6 +89,7 @@ func AddVSphereDeploymentZoneControllerToManager(ctx *context.ControllerManagerC
 			&source.Channel{Source: ctx.GetGenericEventChannelFor(controlledTypeGVK)},
 			&handler.EnqueueRequestForObject{},
 		).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		Complete(reconciler)
 }
 
diff --git a/controllers/vspheremachine_controller.go b/controllers/vspheremachine_controller.go
@@ -122,7 +122,8 @@ func AddMachineControllerToManager(ctx *context.ControllerManagerContext, mgr ma
 		WatchesRawSource(
 			&source.Channel{Source: ctx.GetGenericEventChannelFor(controlledTypeGVK)},
 			&handler.EnqueueRequestForObject{},
-		)
+		).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue))
 
 	r := &machineReconciler{
 		ControllerContext: controllerContext,
diff --git a/controllers/vspherevm_controller.go b/controllers/vspherevm_controller.go
@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
+	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -103,6 +104,7 @@ func AddVMControllerToManager(ctx *context.ControllerManagerContext, mgr manager
 			&source.Channel{Source: ctx.GetGenericEventChannelFor(controlledTypeGVK)},
 			&handler.EnqueueRequestForObject{},
 		).
+		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), ctx.WatchFilterValue)).
 		Build(r)
 	if err != nil {
 		return err
diff --git a/main.go b/main.go
@@ -35,6 +35,7 @@ import (
 	logsv1 "k8s.io/component-base/logs/api/v1"
 	_ "k8s.io/component-base/logs/json/register"
 	"k8s.io/klog/v2"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/controllers/remote"
 	"sigs.k8s.io/cluster-api/util/flags"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -173,6 +174,9 @@ func InitFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&watchNamespace, "namespace", "",
 		"Namespace that the controller watches to reconcile cluster-api objects. If unspecified, the controller watches for cluster-api objects across all namespaces.")
 
+	fs.StringVar(&managerOpts.WatchFilterValue, "watch-filter", "",
+		fmt.Sprintf("Label value that the controller watches to reconcile cluster-api objects. Label key is always %s. If unspecified, the controller watches for all cluster-api objects.", clusterv1.WatchLabel))
+
 	fs.StringVar(&managerOpts.PprofBindAddress, "profiler-address", defaultProfilerAddr,
 		"Bind address to expose the pprof profiler (e.g. localhost:6060)")
 
diff --git a/pkg/context/controller_manager_context.go b/pkg/context/controller_manager_context.go
@@ -85,6 +85,9 @@ type ControllerManagerContext struct {
 	// NetworkProvider is the network provider used by Supervisor based clusters
 	NetworkProvider string
 
+	// WatchFilterValue is used to filter incoming objects by label.
+	WatchFilterValue string
+
 	genericEventCache sync.Map
 }
 
diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go
@@ -98,6 +98,7 @@ func New(opts Options) (Manager, error) {
 		EnableKeepAlive:         opts.EnableKeepAlive,
 		KeepAliveDuration:       opts.KeepAliveDuration,
 		NetworkProvider:         opts.NetworkProvider,
+		WatchFilterValue:        opts.WatchFilterValue,
 	}
 
 	// Add the requested items to the manager.
diff --git a/pkg/manager/options.go b/pkg/manager/options.go
@@ -44,8 +44,8 @@ type Options struct {
 	// for better load management on vSphere api server
 	EnableKeepAlive bool
 
-	// LeaderElectionNamespace is the namespace in which the pod running the
-	// controller maintains a leader election lock
+	// PodNamespace is the namespace in which the pod running the controller
+	// maintains a leader election lock.
 	//
 	// Defaults to the eponymous constant in this package.
 	PodNamespace string
@@ -81,6 +81,11 @@ type Options struct {
 	// If not set, it will default to a DummyNetworkProvider which is intended for testing purposes.
 	// VIM based clusters and managers will not need to set this flag.
 	NetworkProvider string
+
+	// WatchFilterValue is used to filter incoming objects by label.
+	//
+	// Defaults to the empty string and by that not filter anything.
+	WatchFilterValue string
 }
 
 func (o *Options) defaults() {

Original file line number	Diff line number	Diff line change
`@@ -85,6 +85,9 @@ type ControllerManagerContext struct {`
`85`	`85`	`// NetworkProvider is the network provider used by Supervisor based clusters`
`86`	`86`	`NetworkProvider string`
`87`	`87`
	`88`	`+ // WatchFilterValue is used to filter incoming objects by label.`
	`89`	`+ WatchFilterValue string`
	`90`	`+`
`88`	`91`	`genericEventCache sync.Map`
`89`	`92`	`}`
`90`	`93`
Original file line number	Diff line number	Diff line change
`@@ -98,6 +98,7 @@ func New(opts Options) (Manager, error) {`
`98`	`98`	`EnableKeepAlive: opts.EnableKeepAlive,`
`99`	`99`	`KeepAliveDuration: opts.KeepAliveDuration,`
`100`	`100`	`NetworkProvider: opts.NetworkProvider,`
	`101`	`+ WatchFilterValue: opts.WatchFilterValue,`
`101`	`102`	`}`
`102`	`103`
`103`	`104`	`// Add the requested items to the manager.`