Dynamic Namespaces with restricted license gate

.golangci.yml

@@ -4,6 +4,11 @@ linters:
     exhaustive:
       default-signifies-exhaustive: true
 
+    forbidigo:
+      forbid:
+        - pattern: '^source\.Kind$'
+          msg: "use watches.NamespacedKind (pkg/controller/common/watches) instead of source.Kind directly, so namespace-selector filtering is respected"
+
     goheader:
       template: |-
         Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
@@ -97,6 +102,7 @@ linters:
       - linters:
           - gosec
           - unparam
+          - forbidigo
         path: .*_test\.go
       - linters:
           - staticcheck
@@ -134,6 +140,7 @@ linters:
     - errcheck
     - errorlint
     - exhaustive
+    - forbidigo
     - forcetypeassert
     - gocritic
     - goheader

cmd/manager/main.go

@@ -27,6 +27,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/selection"
@@ -44,6 +45,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	crwebhook "sigs.k8s.io/controller-runtime/pkg/webhook"
+	"sigs.k8s.io/yaml"
 
 	"github.com/elastic/cloud-on-k8s/v3/pkg/about"
 	agentv1alpha1 "github.com/elastic/cloud-on-k8s/v3/pkg/apis/agent/v1alpha1"
@@ -69,6 +71,7 @@ import (
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/container"
 	commonesclient "github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/esclient"
 	commonlicense "github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/license"
+	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/nsmatch"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/operator"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/password"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/reconciler"
@@ -593,7 +596,30 @@ func startOperator(ctx context.Context) error {
 		log.Error(err, "Failed to parse managed namespaces flag")
 		return err
 	}
+
+	// Parse the optional namespace label selector. This field is config-file
+	// only (a metav1.LabelSelector object); it has no CLI flag. When set,
+	// the operator runs in dynamic mode: the cache is cluster-wide and a
+	// Matcher consulted by a predicate on every controller's watches
+	// evaluates the selector against Namespace labels per event.
+	parsedNsSelector, err := parseNSSelector(viper.GetViper(), log)
+	if err != nil {
+		return err
+	}
+	dynamicNSSelector := parsedNsSelector != nil
+
+	// Validate that dynamic ns selector cannot co-exist with the fixed list of managed namespaces.
+	if dynamicNSSelector && len(managedNamespaces) > 0 {
+		err := fmt.Errorf("%q and %q are mutually exclusive", operator.NamespacesFlag, operator.NamespaceSelectorFlag)
+		log.Error(err, "Invalid configuration")
+		return err
+	}
+
 	switch {
+	case dynamicNSSelector:
+		log.Info("Operator configured to manage namespaces dynamically via selector",
+			"selector", parsedNsSelector.String(),
+			"operator_namespace", operatorNamespace)
 	case len(managedNamespaces) == 0:
 		log.Info("Operator configured to manage all namespaces")
 	case len(managedNamespaces) == 1 && managedNamespaces[0] == operatorNamespace:
@@ -611,13 +637,26 @@ func startOperator(ctx context.Context) error {
 		return err
 	}
 
+	namespaceMatcher := nsmatch.NewNamespaceMatcher(parsedNsSelector, operatorNamespace)
+
 	opts.Cache = cache.Options{
 		DefaultNamespaces: map[string]cache.Config{},
 		DefaultTransform:  cache.TransformStripManagedFields(),
 		ByObject:          byObject,
 	}
-	for _, ns := range managedNamespaces {
-		opts.Cache.DefaultNamespaces[ns] = cache.Config{}
+	if !dynamicNSSelector {
+		for _, ns := range managedNamespaces {
+			opts.Cache.DefaultNamespaces[ns] = cache.Config{}
+		}
+	} else {
+		opts.NewClient = func(config *rest.Config, options client.Options) (client.Client, error) {
+			delegate, err := client.New(config, options) // cache-backed; options.Cache already set by mgr
+			if err != nil {
+				return nil, err
+			}
+
+			return nsmatch.NewFilterClient(delegate, namespaceMatcher), nil
+		}
 	}
 
 	// only expose prometheus metrics if provided a non-zero port
@@ -651,6 +690,8 @@ func startOperator(ctx context.Context) error {
 		return err
 	}
 
+	namespaceMatcher.SetCache(mgr.GetCache())
+
 	// Retrieve globally shared CA if any
 	ca, err := readOptionalCA(viper.GetString(operator.CADirFlag))
 	if err != nil {
@@ -752,6 +793,7 @@ func startOperator(ctx context.Context) error {
 		SetDefaultSecurityContext: setDefaultSecurityContext,
 		ValidateStorageClass:      viper.GetBool(operator.ValidateStorageClassFlag),
 		Tracer:                    tracer,
+		NamespaceMatcher:          namespaceMatcher,
 	}
 
 	if viper.GetBool(operator.EnableWebhookFlag) {
@@ -773,7 +815,7 @@ func startOperator(ctx context.Context) error {
 
 	disableTelemetry := viper.GetBool(operator.DisableTelemetryFlag)
 	telemetryInterval := viper.GetDuration(operator.TelemetryIntervalFlag)
-	go asyncTasks(ctx, mgr, cfg, managedNamespaces, operatorNamespace, operatorInfo, disableTelemetry, telemetryInterval, tracer, dialer)
+	go asyncTasks(ctx, mgr, cfg, managedNamespaces, operatorNamespace, namespaceMatcher, operatorInfo, disableTelemetry, telemetryInterval, tracer, dialer)
 
 	log.Info("Starting the manager", "uuid", operatorInfo.OperatorUUID,
 		"namespace", operatorNamespace, "version", operatorInfo.BuildInfo.Version,
@@ -828,6 +870,7 @@ func asyncTasks(
 	cfg *rest.Config,
 	managedNamespaces []string,
 	operatorNamespace string,
+	namespaceMatcher *nsmatch.NamespaceMatcher,
 	operatorInfo about.OperatorInfo,
 	disableTelemetry bool,
 	telemetryInterval time.Duration,
@@ -851,16 +894,25 @@ func asyncTasks(
 	if !disableTelemetry {
 		// Start the telemetry reporter
 		go func() {
-			tr := telemetry.NewReporter(operatorInfo, mgr.GetClient(), operatorNamespace, managedNamespaces, telemetryInterval, tracer)
+			tr := telemetry.NewReporter(operatorInfo, mgr.GetClient(), operatorNamespace, managedNamespaces, namespaceMatcher, telemetryInterval, tracer)
 			tr.Start(ctx)
 		}()
 	}
 
+	namespaces := managedNamespaces
+	if namespaceMatcher.SelectorEnabled() {
+		nsp, err := namespaceMatcher.MatchingNamespaces(ctx)
+		if err != nil {
+			log.Error(err, "exiting due to unrecoverable error while fetching the namespaces")
+			os.Exit(1)
+		}
+		namespaces = nsp
+	}
 	// Garbage collect orphaned secrets leftover from deleted resources while the operator was not running
 	// - association user secrets
 	gcCtx := tracing.NewContextTransaction(ctx, tracer, tracing.RunOnceTxType, "garbage-collection", nil)
 	gcCtx = logconf.AddToContext(gcCtx, logf.Log.WithName("garbage-collection"))
-	err := garbageCollectUsers(gcCtx, cfg, managedNamespaces)
+	err := garbageCollectUsers(gcCtx, cfg, namespaces)
 	if err != nil {
 		log.Error(err, "exiting due to unrecoverable error")
 		os.Exit(1)
@@ -1101,3 +1153,32 @@ func buildByObject(restrictWatchedResources bool) (map[client.Object]cache.ByObj
 
 	return byObject, nil
 }
+
+func parseNSSelector(v *viper.Viper, log logr.Logger) (labels.Selector, error) {
+	if !v.IsSet(operator.NamespaceSelectorFlag) {
+		return nil, nil
+	}
+	raw := v.Get(operator.NamespaceSelectorFlag)
+	yamlBytes, err := yaml.Marshal(raw)
+	if err != nil {
+		log.Error(err, "Failed to serialize namespace-selector for parsing")
+		return nil, err
+	}
+	var ls metav1.LabelSelector
+	if err := yaml.Unmarshal(yamlBytes, &ls); err != nil {
+		log.Error(err, "Failed to parse namespace-selector")
+		return nil, err
+	}
+	if len(ls.MatchLabels) == 0 && len(ls.MatchExpressions) == 0 {
+		// an empty selector ({}) means dynamic namespace selection is disabled,
+		// so fallback to all namespaces.
+		log.Info("namespace-selector is empty: dynamic namespace selection is disabled, all namespaces will be watched")
+		return nil, nil
+	}
+	sel, err := metav1.LabelSelectorAsSelector(&ls)
+	if err != nil {
+		log.Error(err, "Invalid namespace-selector")
+		return nil, err
+	}
+	return sel, nil
+}

cmd/manager/main_test.go

@@ -10,14 +10,18 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/go-logr/logr"
+	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/kubernetes"
@@ -30,6 +34,7 @@ import (
 	esv1 "github.com/elastic/cloud-on-k8s/v3/pkg/apis/elasticsearch/v1"
 	entv1 "github.com/elastic/cloud-on-k8s/v3/pkg/apis/enterprisesearch/v1"
 	kbv1 "github.com/elastic/cloud-on-k8s/v3/pkg/apis/kibana/v1"
+	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/operator"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/reconciler"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/utils/k8s"
 )
@@ -397,3 +402,84 @@ func TestBuildByObject(t *testing.T) {
 		})
 	}
 }
+
+func Test_parseNSSelector(t *testing.T) {
+	tests := []struct {
+		name        string
+		rawSelector any // value set for operator.NamespaceSelectorFlag, unset if nil
+		wantSel     labels.Selector
+		wantErr     bool
+	}{
+		{
+			name:        "flag not set returns nil selector",
+			rawSelector: nil,
+			wantSel:     nil,
+		},
+		{
+			name:        "empty selector ({}) returns nil selector",
+			rawSelector: map[string]any{},
+			wantSel:     nil,
+		},
+		{
+			name: "matchLabels is parsed into a selector",
+			rawSelector: map[string]any{
+				"matchLabels": map[string]any{
+					"environment": "prod",
+				},
+			},
+			wantSel: labels.SelectorFromSet(labels.Set{"environment": "prod"}),
+		},
+		{
+			name: "matchExpressions is parsed into a selector",
+			rawSelector: map[string]any{
+				"matchExpressions": []any{
+					map[string]any{
+						"key":      "environment",
+						"operator": "In",
+						"values":   []any{"prod", "staging"},
+					},
+				},
+			},
+			wantSel: func() labels.Selector {
+				req, err := labels.NewRequirement("environment", selection.In, []string{"prod", "staging"})
+				require.NoError(t, err)
+				return labels.NewSelector().Add(*req)
+			}(),
+		},
+		{
+			name:        "malformed selector value returns an error",
+			rawSelector: "not-a-label-selector",
+			wantErr:     true,
+		},
+		{
+			name: "invalid matchExpressions operator returns an error",
+			rawSelector: map[string]any{
+				"matchExpressions": []any{
+					map[string]any{
+						"key":      "environment",
+						"operator": "NotAValidOperator",
+						"values":   []any{"prod"},
+					},
+				},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			v := viper.New()
+			if tt.rawSelector != nil {
+				v.Set(operator.NamespaceSelectorFlag, tt.rawSelector)
+			}
+
+			got, err := parseNSSelector(v, logr.Discard())
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantSel, got)
+		})
+	}
+}

cmd/manager/validation.go

@@ -34,6 +34,7 @@ import (
 	esavalidation "github.com/elastic/cloud-on-k8s/v3/pkg/controller/autoscaling/elasticsearch/validation"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/certificates"
 	commonlicense "github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/license"
+	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/nsmatch"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/operator"
 	"github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/tracing"
 	commonwebhook "github.com/elastic/cloud-on-k8s/v3/pkg/controller/common/webhook"
@@ -79,34 +80,35 @@ func setupWebhook(
 ) {
 	manageWebhookCerts := viper.GetBool(operator.ManageWebhookCertsFlag)
 	if manageWebhookCerts {
-		if err := reconcileWebhookCertsAndAddController(ctx, mgr, params.CertRotation, clientset, tracer); err != nil {
+		if err := reconcileWebhookCertsAndAddController(ctx, mgr, params.CertRotation, params.NamespaceMatcher, clientset, tracer); err != nil {
 			log.Error(err, "unable to setup the webhook certificates")
 			os.Exit(1)
 		}
 	}
 
 	checker := commonlicense.NewLicenseChecker(mgr.GetClient(), params.OperatorNamespace)
+	matcher := params.NamespaceMatcher
 	// setup webhooks for supported types
-	commonwebhook.RegisterResourceWebhook(mgr, apmv1.WebhookPath, checker, managedNamespaces, apmv1.Validate, "APM Server")
-	commonwebhook.RegisterResourceWebhook(mgr, apmv1beta1.WebhookPath, checker, managedNamespaces, apmv1beta1.Validate, "APM Server")
-	commonwebhook.RegisterResourceWebhook(mgr, beatv1beta1.WebhookPath, checker, managedNamespaces, beatv1beta1.Validate, "Beat")
-	commonwebhook.RegisterResourceWebhook(mgr, entv1.WebhookPath, checker, managedNamespaces, entv1.Validate, "Enterprise Search")
-	commonwebhook.RegisterResourceWebhook(mgr, entv1beta1.WebhookPath, checker, managedNamespaces, entv1beta1.Validate, "Enterprise Search")
-	commonwebhook.RegisterResourceWebhook(mgr, esv1beta1.WebhookPath, checker, managedNamespaces, esv1beta1.Validate, "Elasticsearch")
-	commonwebhook.RegisterResourceWebhook(mgr, kbv1.WebhookPath, checker, managedNamespaces, kbv1.Validate, "Kibana")
-	commonwebhook.RegisterResourceWebhook(mgr, kbv1beta1.WebhookPath, checker, managedNamespaces, kbv1beta1.Validate, "Kibana")
-	commonwebhook.RegisterResourceWebhook(mgr, emsv1alpha1.WebhookPath, checker, managedNamespaces, emsv1alpha1.Validate, "Elastic Maps Server")
-	commonwebhook.RegisterResourceWebhook(mgr, eprv1alpha1.WebhookPath, checker, managedNamespaces, eprv1alpha1.Validate, "Package Registry")
-	scpvalidation.RegisterWebhook(mgr, checker, managedNamespaces, params.OperatorNamespace)
+	commonwebhook.RegisterResourceWebhook(mgr, apmv1.WebhookPath, checker, managedNamespaces, matcher, apmv1.Validate, "APM Server")
+	commonwebhook.RegisterResourceWebhook(mgr, apmv1beta1.WebhookPath, checker, managedNamespaces, matcher, apmv1beta1.Validate, "APM Server")
+	commonwebhook.RegisterResourceWebhook(mgr, beatv1beta1.WebhookPath, checker, managedNamespaces, matcher, beatv1beta1.Validate, "Beat")
+	commonwebhook.RegisterResourceWebhook(mgr, entv1.WebhookPath, checker, managedNamespaces, matcher, entv1.Validate, "Enterprise Search")
+	commonwebhook.RegisterResourceWebhook(mgr, entv1beta1.WebhookPath, checker, managedNamespaces, matcher, entv1beta1.Validate, "Enterprise Search")
+	commonwebhook.RegisterResourceWebhook(mgr, esv1beta1.WebhookPath, checker, managedNamespaces, matcher, esv1beta1.Validate, "Elasticsearch")
+	commonwebhook.RegisterResourceWebhook(mgr, kbv1.WebhookPath, checker, managedNamespaces, matcher, kbv1.Validate, "Kibana")
+	commonwebhook.RegisterResourceWebhook(mgr, kbv1beta1.WebhookPath, checker, managedNamespaces, matcher, kbv1beta1.Validate, "Kibana")
+	commonwebhook.RegisterResourceWebhook(mgr, emsv1alpha1.WebhookPath, checker, managedNamespaces, matcher, emsv1alpha1.Validate, "Elastic Maps Server")
+	commonwebhook.RegisterResourceWebhook(mgr, eprv1alpha1.WebhookPath, checker, managedNamespaces, matcher, eprv1alpha1.Validate, "Package Registry")
+	scpvalidation.RegisterWebhook(mgr, checker, managedNamespaces, params.OperatorNamespace, matcher)
 
 	// Logstash, Elasticsearch v1, ElasticsearchAutoscaling, and AutoOps validating webhooks are wired up
 	// separately so their validators can use the API client and/or embed license checks. Elasticsearch
 	// v1beta1 remains in the RegisterResourceWebhook list above because it only needs a ValidateFunc.
-	esvalidation.RegisterWebhook(mgr, params.ValidateStorageClass, exposedNodeLabels, checker, managedNamespaces)
-	esavalidation.RegisterWebhook(mgr, params.ValidateStorageClass, checker, managedNamespaces)
-	lsvalidation.RegisterWebhook(mgr, params.ValidateStorageClass, managedNamespaces)
-	autoopsvalidation.RegisterWebhook(mgr, checker, managedNamespaces)
-	agentcontroller.RegisterWebhook(mgr, checker, managedNamespaces)
+	esvalidation.RegisterWebhook(mgr, params.ValidateStorageClass, exposedNodeLabels, checker, managedNamespaces, matcher)
+	esavalidation.RegisterWebhook(mgr, params.ValidateStorageClass, checker, managedNamespaces, matcher)
+	lsvalidation.RegisterWebhook(mgr, params.ValidateStorageClass, managedNamespaces, matcher)
+	autoopsvalidation.RegisterWebhook(mgr, checker, managedNamespaces, matcher)
+	agentcontroller.RegisterWebhook(mgr, checker, managedNamespaces, matcher)
 
 	// wait for the secret to be populated in the local filesystem before returning
 	interval := time.Second * 1
@@ -134,7 +136,7 @@ func setupWebhook(
 	}
 }
 
-func reconcileWebhookCertsAndAddController(ctx context.Context, mgr manager.Manager, certRotation certificates.RotationParams, clientset kubernetes.Interface, tracer *apm.Tracer) error {
+func reconcileWebhookCertsAndAddController(ctx context.Context, mgr manager.Manager, certRotation certificates.RotationParams, m *nsmatch.NamespaceMatcher, clientset kubernetes.Interface, tracer *apm.Tracer) error {
 	ctx = tracing.NewContextTransaction(ctx, tracer, tracing.ReconciliationTxType, webhook.ControllerName, nil)
 	defer tracing.EndContextTransaction(ctx)
 	log.Info("Automatic management of the webhook certificates enabled")
@@ -157,5 +159,5 @@ func reconcileWebhookCertsAndAddController(ctx context.Context, mgr manager.Mana
 		return err
 	}
 
-	return webhook.Add(mgr, webhookParams, clientset, wh, tracer)
+	return webhook.Add(mgr, m, webhookParams, clientset, wh, tracer)
 }