add namespace filtering

kubernetes-sigs · Nov 4, 2024 · 84ef63f · 84ef63f
1 parent d1ca160
commit 84ef63f
Show file tree

Hide file tree

Showing 39 changed files with 428 additions and 297 deletions.
diff --git a/api/v1beta1/common_types.go b/api/v1beta1/common_types.go
@@ -107,7 +107,7 @@ const (
 	// older MachineSets when Machines are deleted and add the new replicas to the latest MachineSet.
 	DisableMachineCreateAnnotation = "cluster.x-k8s.io/disable-machine-create"
 
-	// WatchLabel is a label othat can be applied to any Cluster API object.
+	// WatchLabel is a label that can be applied to any Cluster API object.
 	//
 	// Controllers which allow for selective reconciliation may check this label and proceed
 	// with reconciliation of the object only if this label and a configured value is present.

diff --git a/bootstrap/kubeadm/controllers/alias.go b/bootstrap/kubeadm/controllers/alias.go
@@ -26,6 +26,7 @@ import (
 
 	kubeadmbootstrapcontrollers "sigs.k8s.io/cluster-api/bootstrap/kubeadm/internal/controllers"
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
+	watchfilter "sigs.k8s.io/cluster-api/controllers/watchfilter"
 )
 
 // Following types provides access to reconcilers implemented in internal/controllers, thus
@@ -43,8 +44,8 @@ type KubeadmConfigReconciler struct {
 
 	ClusterCache clustercache.ClusterCache
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 
 	// TokenTTL is the amount of time a bootstrap token (and therefore a KubeadmConfig) will be valid.
 	TokenTTL time.Duration
@@ -56,7 +57,7 @@ func (r *KubeadmConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		Client:              r.Client,
 		SecretCachingClient: r.SecretCachingClient,
 		ClusterCache:        r.ClusterCache,
-		WatchFilterValue:    r.WatchFilterValue,
+		WatchFilter:         r.WatchFilter,
 		TokenTTL:            r.TokenTTL,
 	}).SetupWithManager(ctx, mgr, options)
 }
diff --git a/bootstrap/kubeadm/internal/controllers/kubeadmconfig_controller.go b/bootstrap/kubeadm/internal/controllers/kubeadmconfig_controller.go
@@ -51,6 +51,7 @@ import (
 	kubeadmtypes "sigs.k8s.io/cluster-api/bootstrap/kubeadm/types"
 	bsutil "sigs.k8s.io/cluster-api/bootstrap/util"
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
+	watchfilter "sigs.k8s.io/cluster-api/controllers/watchfilter"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	"sigs.k8s.io/cluster-api/feature"
 	"sigs.k8s.io/cluster-api/internal/util/taints"
@@ -86,8 +87,8 @@ type KubeadmConfigReconciler struct {
 	ClusterCache        clustercache.ClusterCache
 	KubeadmInitLock     InitLocker
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 
 	// TokenTTL is the amount of time a bootstrap token (and therefore a KubeadmConfig) will be valid.
 	TokenTTL time.Duration
@@ -121,7 +122,7 @@ func (r *KubeadmConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		Watches(
 			&clusterv1.Machine{},
 			handler.EnqueueRequestsFromMapFunc(r.MachineToBootstrapMapFunc),
-		).WithEventFilter(predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue))
+		).WithEventFilter(predicates.ResourceHasFilter(mgr.GetScheme(), predicateLog, r.WatchFilter))
 
 	if feature.Gates.Enabled(feature.MachinePool) {
 		b = b.Watches(
@@ -136,7 +137,7 @@ func (r *KubeadmConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 		builder.WithPredicates(
 			predicates.All(mgr.GetScheme(), predicateLog,
 				predicates.ClusterPausedTransitionsOrInfrastructureReady(mgr.GetScheme(), predicateLog),
-				predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue),
+				predicates.ResourceHasFilter(mgr.GetScheme(), predicateLog, r.WatchFilter),
 			),
 		),
 	).WatchesRawSource(r.ClusterCache.GetClusterSource("kubeadmconfig", r.ClusterToKubeadmConfigs))

diff --git a/bootstrap/kubeadm/main.go b/bootstrap/kubeadm/main.go
@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"os"
 	goruntime "runtime"
+	"strings"
 	"time"
 
 	"github.com/spf13/pflag"
@@ -49,6 +50,7 @@ import (
 	"sigs.k8s.io/cluster-api/bootstrap/kubeadm/internal/webhooks"
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
 	"sigs.k8s.io/cluster-api/controllers/remote"
+	watchfilter "sigs.k8s.io/cluster-api/controllers/watchfilter"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	"sigs.k8s.io/cluster-api/feature"
 	bootstrapv1alpha3 "sigs.k8s.io/cluster-api/internal/apis/bootstrap/kubeadm/v1alpha3"
@@ -64,26 +66,27 @@ var (
 	controllerName = "cluster-api-kubeadm-bootstrap-manager"
 
 	// flags.
-	enableLeaderElection        bool
-	leaderElectionLeaseDuration time.Duration
-	leaderElectionRenewDeadline time.Duration
-	leaderElectionRetryPeriod   time.Duration
-	watchFilterValue            string
-	watchNamespace              string
-	profilerAddress             string
-	enableContentionProfiling   bool
-	syncPeriod                  time.Duration
-	restConfigQPS               float32
-	restConfigBurst             int
-	clusterCacheClientQPS       float32
-	clusterCacheClientBurst     int
-	webhookPort                 int
-	webhookCertDir              string
-	webhookCertName             string
-	webhookKeyName              string
-	healthAddr                  string
-	managerOptions              = flags.ManagerOptions{}
-	logOptions                  = logs.NewOptions()
+	enableLeaderElection          bool
+	leaderElectionLeaseDuration   time.Duration
+	leaderElectionRenewDeadline   time.Duration
+	leaderElectionRetryPeriod     time.Duration
+	watchFilterValue              string
+	watchFilterExcludedNamespaces string
+	watchNamespace                string
+	profilerAddress               string
+	enableContentionProfiling     bool
+	syncPeriod                    time.Duration
+	restConfigQPS                 float32
+	restConfigBurst               int
+	clusterCacheClientQPS         float32
+	clusterCacheClientBurst       int
+	webhookPort                   int
+	webhookCertDir                string
+	webhookCertName               string
+	webhookKeyName                string
+	healthAddr                    string
+	managerOptions                = flags.ManagerOptions{}
+	logOptions                    = logs.NewOptions()
 	// CABPK specific flags.
 	clusterConcurrency       int
 	clusterCacheConcurrency  int
@@ -122,6 +125,9 @@ func InitFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&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(&watchFilterExcludedNamespaces, "excluded-namespaces", "",
+		fmt.Sprintf("Comma separated list of names. Exclude the namespaces controller watches to reconcile cluster-api objects."))
+
 	fs.StringVar(&profilerAddress, "profiler-address", "",
 		"Bind address to expose the pprof profiler (e.g. localhost:6060)")
 
@@ -313,6 +319,13 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		os.Exit(1)
 	}
 
+	watchFilter := watchfilter.WatchFilter{
+		Value: watchFilterValue,
+	}
+	if watchFilterExcludedNamespaces != "" {
+		watchFilter.ExcludedNamespaces = strings.Split(watchFilterExcludedNamespaces, ",")
+	}
+
 	clusterCache, err := clustercache.SetupWithManager(ctx, mgr, clustercache.Options{
 		SecretClient: secretCachingClient,
 		Cache:        clustercache.CacheOptions{},
@@ -328,7 +341,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 				},
 			},
 		},
-		WatchFilterValue: watchFilterValue,
+		WatchFilter: watchFilter,
 	}, concurrency(clusterCacheConcurrency))
 	if err != nil {
 		setupLog.Error(err, "Unable to create ClusterCache")
@@ -339,7 +352,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		Client:              mgr.GetClient(),
 		SecretCachingClient: secretCachingClient,
 		ClusterCache:        clusterCache,
-		WatchFilterValue:    watchFilterValue,
+		WatchFilter:         watchFilter,
 		TokenTTL:            tokenTTL,
 	}).SetupWithManager(ctx, mgr, concurrency(kubeadmConfigConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "KubeadmConfig")

diff --git a/controllers/alias.go b/controllers/alias.go
@@ -25,6 +25,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
+	watchfilter "sigs.k8s.io/cluster-api/controllers/watchfilter"
 	clustercontroller "sigs.k8s.io/cluster-api/internal/controllers/cluster"
 	clusterclasscontroller "sigs.k8s.io/cluster-api/internal/controllers/clusterclass"
 	machinecontroller "sigs.k8s.io/cluster-api/internal/controllers/machine"
@@ -46,8 +47,8 @@ type ClusterReconciler struct {
 	APIReader    client.Reader
 	ClusterCache clustercache.ClusterCache
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 
 	RemoteConnectionGracePeriod time.Duration
 }
@@ -57,7 +58,7 @@ func (r *ClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manag
 		Client:                      r.Client,
 		APIReader:                   r.APIReader,
 		ClusterCache:                r.ClusterCache,
-		WatchFilterValue:            r.WatchFilterValue,
+		WatchFilter:                 r.WatchFilter,
 		RemoteConnectionGracePeriod: r.RemoteConnectionGracePeriod,
 	}).SetupWithManager(ctx, mgr, options)
 }
@@ -68,8 +69,8 @@ type MachineReconciler struct {
 	APIReader    client.Reader
 	ClusterCache clustercache.ClusterCache
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 
 	RemoteConditionsGracePeriod time.Duration
 }
@@ -79,7 +80,7 @@ func (r *MachineReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manag
 		Client:                      r.Client,
 		APIReader:                   r.APIReader,
 		ClusterCache:                r.ClusterCache,
-		WatchFilterValue:            r.WatchFilterValue,
+		WatchFilter:                 r.WatchFilter,
 		RemoteConditionsGracePeriod: r.RemoteConditionsGracePeriod,
 	}).SetupWithManager(ctx, mgr, options)
 }
@@ -90,8 +91,8 @@ type MachineSetReconciler struct {
 	APIReader    client.Reader
 	ClusterCache clustercache.ClusterCache
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 
 	// Deprecated: DeprecatedInfraMachineNaming. Name the InfraStructureMachines after the InfraMachineTemplate.
 	DeprecatedInfraMachineNaming bool
@@ -102,7 +103,7 @@ func (r *MachineSetReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 		Client:                       r.Client,
 		APIReader:                    r.APIReader,
 		ClusterCache:                 r.ClusterCache,
-		WatchFilterValue:             r.WatchFilterValue,
+		WatchFilter:                  r.WatchFilter,
 		DeprecatedInfraMachineNaming: r.DeprecatedInfraMachineNaming,
 	}).SetupWithManager(ctx, mgr, options)
 }
@@ -112,15 +113,15 @@ type MachineDeploymentReconciler struct {
 	Client    client.Client
 	APIReader client.Reader
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *MachineDeploymentReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&machinedeploymentcontroller.Reconciler{
-		Client:           r.Client,
-		APIReader:        r.APIReader,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:      r.Client,
+		APIReader:   r.APIReader,
+		WatchFilter: r.WatchFilter,
 	}).SetupWithManager(ctx, mgr, options)
 }
 
@@ -129,15 +130,15 @@ type MachineHealthCheckReconciler struct {
 	Client       client.Client
 	ClusterCache clustercache.ClusterCache
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *MachineHealthCheckReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&machinehealthcheckcontroller.Reconciler{
-		Client:           r.Client,
-		ClusterCache:     r.ClusterCache,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:       r.Client,
+		ClusterCache: r.ClusterCache,
+		WatchFilter:  r.WatchFilter,
 	}).SetupWithManager(ctx, mgr, options)
 }
 
@@ -151,17 +152,17 @@ type ClusterTopologyReconciler struct {
 
 	RuntimeClient runtimeclient.Client
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *ClusterTopologyReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&clustertopologycontroller.Reconciler{
-		Client:           r.Client,
-		APIReader:        r.APIReader,
-		ClusterCache:     r.ClusterCache,
-		RuntimeClient:    r.RuntimeClient,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:        r.Client,
+		APIReader:     r.APIReader,
+		ClusterCache:  r.ClusterCache,
+		RuntimeClient: r.RuntimeClient,
+		WatchFilter:   r.WatchFilter,
 	}).SetupWithManager(ctx, mgr, options)
 }
 
@@ -173,15 +174,15 @@ type MachineDeploymentTopologyReconciler struct {
 	Client client.Client
 	// APIReader is used to list MachineSets directly via the API server to avoid
 	// race conditions caused by an outdated cache.
-	APIReader        client.Reader
-	WatchFilterValue string
+	APIReader   client.Reader
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *MachineDeploymentTopologyReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&machinedeploymenttopologycontroller.Reconciler{
-		Client:           r.Client,
-		APIReader:        r.APIReader,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:      r.Client,
+		APIReader:   r.APIReader,
+		WatchFilter: r.WatchFilter,
 	}).SetupWithManager(ctx, mgr, options)
 }
 
@@ -193,15 +194,15 @@ type MachineSetTopologyReconciler struct {
 	Client client.Client
 	// APIReader is used to list MachineSets directly via the API server to avoid
 	// race conditions caused by an outdated cache.
-	APIReader        client.Reader
-	WatchFilterValue string
+	APIReader   client.Reader
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *MachineSetTopologyReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	return (&machinesettopologycontroller.Reconciler{
-		Client:           r.Client,
-		APIReader:        r.APIReader,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:      r.Client,
+		APIReader:   r.APIReader,
+		WatchFilter: r.WatchFilter,
 	}).SetupWithManager(ctx, mgr, options)
 }
 
@@ -216,15 +217,15 @@ type ClusterClassReconciler struct {
 	// RuntimeClient is a client for calling runtime extensions.
 	RuntimeClient runtimeclient.Client
 
-	// WatchFilterValue is the label value used to filter events prior to reconciliation.
-	WatchFilterValue string
+	// WatchFilter is used to filter events prior to reconciliation.
+	WatchFilter watchfilter.WatchFilter
 }
 
 func (r *ClusterClassReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
 	r.internalReconciler = &clusterclasscontroller.Reconciler{
-		Client:           r.Client,
-		RuntimeClient:    r.RuntimeClient,
-		WatchFilterValue: r.WatchFilterValue,
+		Client:        r.Client,
+		RuntimeClient: r.RuntimeClient,
+		WatchFilter:   r.WatchFilter,
 	}
 	return r.internalReconciler.SetupWithManager(ctx, mgr, options)
 }