add namespace filtering

kubernetes-sigs · Nov 11, 2024 · 4c5ea6f · 4c5ea6f
1 parent e13abea
commit 4c5ea6f
Show file tree

Hide file tree

Showing 5 changed files with 149 additions and 89 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/main.go b/bootstrap/kubeadm/main.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -64,26 +65,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
+	watchExcludedNamespaces       []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 +124,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.StringSliceVar(&watchExcludedNamespaces, "excluded-namespace", nil,
+		"Comma separated list of namespaces. 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)")
 
@@ -218,6 +223,15 @@ func main() {
 		}
 	}
 
+	var fieldSelector fields.Selector
+	if watchExcludedNamespaces != nil {
+		var conditions []fields.Selector
+		for i := range watchExcludedNamespaces {
+			conditions = append(conditions, fields.OneTermNotEqualSelector("metadata.namespace", watchExcludedNamespaces[i]))
+		}
+		fieldSelector = fields.AndSelectors(conditions...)
+	}
+
 	if enableContentionProfiling {
 		goruntime.SetBlockProfileRate(1)
 	}
@@ -237,8 +251,9 @@ func main() {
 		PprofBindAddress:           profilerAddress,
 		Metrics:                    *metricsOptions,
 		Cache: cache.Options{
-			DefaultNamespaces: watchNamespaces,
-			SyncPeriod:        &syncPeriod,
+			DefaultFieldSelector: fieldSelector,
+			DefaultNamespaces:    watchNamespaces,
+			SyncPeriod:           &syncPeriod,
 			ByObject: map[client.Object]cache.ByObject{
 				// Note: Only Secrets with the cluster name label are cached.
 				// The default client of the manager won't use the cache for secrets at all (see Client.Cache.DisableFor).

diff --git a/controlplane/kubeadm/main.go b/controlplane/kubeadm/main.go
@@ -30,6 +30,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -69,26 +70,27 @@ var (
 	controllerName = "cluster-api-kubeadm-control-plane-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
+	watchExcludedNamespaces       []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()
 	// KCP specific flags.
 	remoteConditionsGracePeriod     time.Duration
 	kubeadmControlPlaneConcurrency  int
@@ -131,6 +133,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.StringSliceVar(&watchExcludedNamespaces, "excluded-namespace", nil,
+		"Comma separated list of namespaces. 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)")
 
@@ -243,6 +248,15 @@ func main() {
 		}
 	}
 
+	var fieldSelector fields.Selector
+	if watchExcludedNamespaces != nil {
+		var conditions []fields.Selector
+		for i := range watchExcludedNamespaces {
+			conditions = append(conditions, fields.OneTermNotEqualSelector("metadata.namespace", watchExcludedNamespaces[i]))
+		}
+		fieldSelector = fields.AndSelectors(conditions...)
+	}
+
 	if enableContentionProfiling {
 		goruntime.SetBlockProfileRate(1)
 	}
@@ -262,8 +276,9 @@ func main() {
 		PprofBindAddress:           profilerAddress,
 		Metrics:                    *metricsOptions,
 		Cache: cache.Options{
-			DefaultNamespaces: watchNamespaces,
-			SyncPeriod:        &syncPeriod,
+			DefaultFieldSelector: fieldSelector,
+			DefaultNamespaces:    watchNamespaces,
+			SyncPeriod:           &syncPeriod,
 			ByObject: map[client.Object]cache.ByObject{
 				// Note: Only Secrets with the cluster name label are cached.
 				// The default client of the manager won't use the cache for secrets at all (see Client.Cache.DisableFor).

diff --git a/main.go b/main.go
@@ -32,6 +32,7 @@ import (
 	storagev1 "k8s.io/api/storage/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -88,26 +89,27 @@ var (
 	controllerName = "cluster-api-controller-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
+	watchExcludedNamespaces       []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()
 	// core Cluster API specific flags.
 	remoteConnectionGracePeriod     time.Duration
 	remoteConditionsGracePeriod     time.Duration
@@ -172,6 +174,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.StringSliceVar(&watchExcludedNamespaces, "excluded-namespace", nil,
+		"Comma separated list of namespaces. 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)")
 
@@ -323,6 +328,15 @@ func main() {
 		}
 	}
 
+	var fieldSelector fields.Selector
+	if watchExcludedNamespaces != nil {
+		var conditions []fields.Selector
+		for i := range watchExcludedNamespaces {
+			conditions = append(conditions, fields.OneTermNotEqualSelector("metadata.namespace", watchExcludedNamespaces[i]))
+		}
+		fieldSelector = fields.AndSelectors(conditions...)
+	}
+
 	if enableContentionProfiling {
 		goruntime.SetBlockProfileRate(1)
 	}
@@ -342,8 +356,9 @@ func main() {
 		PprofBindAddress:           profilerAddress,
 		Metrics:                    *metricsOptions,
 		Cache: cache.Options{
-			DefaultNamespaces: watchNamespaces,
-			SyncPeriod:        &syncPeriod,
+			DefaultFieldSelector: fieldSelector,
+			DefaultNamespaces:    watchNamespaces,
+			SyncPeriod:           &syncPeriod,
 			ByObject: map[client.Object]cache.ByObject{
 				// Note: Only Secrets with the cluster name label are cached.
 				// The default client of the manager won't use the cache for secrets at all (see Client.Cache.DisableFor).

diff --git a/test/infrastructure/docker/main.go b/test/infrastructure/docker/main.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -70,26 +71,27 @@ var (
 	controllerName = "cluster-api-docker-controller-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
+	watchExcludedNamespaces       []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()
 	// CAPD specific flags.
 	concurrency             int
 	clusterCacheConcurrency int
@@ -129,6 +131,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.StringSliceVar(&watchExcludedNamespaces, "excluded-namespaces", nil,
+		"Comma separated list of namespaces. 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)")
 
@@ -222,6 +227,15 @@ func main() {
 		}
 	}
 
+	var fieldSelector fields.Selector
+	if watchExcludedNamespaces != nil {
+		var conditions []fields.Selector
+		for i := range watchExcludedNamespaces {
+			conditions = append(conditions, fields.OneTermNotEqualSelector("metadata.namespace", watchExcludedNamespaces[i]))
+		}
+		fieldSelector = fields.AndSelectors(conditions...)
+	}
+
 	if enableContentionProfiling {
 		goruntime.SetBlockProfileRate(1)
 	}
@@ -241,8 +255,9 @@ func main() {
 		PprofBindAddress:           profilerAddress,
 		Metrics:                    *metricsOptions,
 		Cache: cache.Options{
-			DefaultNamespaces: watchNamespaces,
-			SyncPeriod:        &syncPeriod,
+			DefaultFieldSelector: fieldSelector,
+			DefaultNamespaces:    watchNamespaces,
+			SyncPeriod:           &syncPeriod,
 			ByObject: map[client.Object]cache.ByObject{
 				// Note: Only Secrets with the cluster name label are cached.
 				// The default client of the manager won't use the cache for secrets at all (see Client.Cache.DisableFor).