feat: allow to configure the namespaces to watch (#1351)

**Description**

Allows configuring the namespaces to watch by the controller, to
minimize the permissions and resources needed. The following properties
have been added to the Helm values:

```yaml
controller:
  # Configuration for how the Kubernetes controllers watch the different resources.
  watch:
    # Namespaces to watch. An empty list means to watch all namespaces.
    # Default is an empty list, to watch all namespaces.
    namespaces: []
    # Sync timeout for the Kubernetes cache. If the cache is not synced within this time, the controller will exit.
    # Default is 2 minutes.
    cacheSyncTimeout: 2m
```

**Related Issues/PRs (if applicable)**

Fixes #1334

**Special notes for reviewers (if applicable)**

N/A

---------

Signed-off-by: Ignasi Barrera <[email protected]>

Author: nacx

.github/workflows/build_and_test.yaml

@@ -321,6 +321,36 @@ jobs:
           # do not depend on the EG version.
           EG_VERSION: v1.5.0
 
+  test_e2e_namespaced:
+    needs: changes
+    if: ${{ needs.changes.outputs.code == 'true' }}
+    name: E2E Test for Namespaced Controller
+    # TODO: make it possible to run this job on macOS as well, which is a bit tricky due to the nested
+    # virtualization is not supported on macOS runners.
+    # E.g. Use https://github.com/douglascamata/setup-docker-macos-action  per the comment in
+    # https://github.com/actions/runner-images/issues/17#issuecomment-1971073406
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          cache: false
+          go-version-file: go.mod
+      - uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/.cache/golangci-lint
+            ~/go/pkg/mod
+            ~/go/bin
+          key: e2e-test-${{ hashFiles('**/go.mod', '**/go.sum', '**/Makefile') }}
+      - uses: docker/setup-buildx-action@v3
+      - run: make test-e2e-namespaced
+        env:
+          # We only need to test with the latest stable version of EG, since these e2e tests
+          # do not depend on the EG version.
+          EG_VERSION: v1.5.0
+
   test_e2e_aigw:
     needs: changes
     name: E2E Test for aigw CLI

.gitignore

@@ -13,6 +13,7 @@ ACCESS_LOG_PATH
 
 tests/e2e/logs/
 tests/e2e-inference-extension/logs/
+tests/e2e-namespaced/logs/
 tests/e2e-upgrade/logs/
 
 # Files and directories to ignore in the site directory

Makefile

@@ -187,6 +187,12 @@ test-e2e-upgrade: build-e2e
 	@echo "Run E2E upgrade tests"
 	@go test -v ./tests/e2e-upgrade/... $(GO_TEST_ARGS) $(GO_TEST_E2E_ARGS)
 
+# This runs the end-to-end namespaced tests for the controller and extproc with a local kind cluster.
+.PHONY: test-e2e-namespaced
+test-e2e-namespaced: build-e2e
+	@echo "Run E2E namespaced tests"
+	@go test -v ./tests/e2e-namespaced/... $(GO_TEST_ARGS) $(GO_TEST_E2E_ARGS)
+
 # This runs the MCP end-to-end tests.
 .PHONY: test-e2e-aigw
 test-e2e-aigw: build.aigw ## Run MCP end-to-end tests.

cmd/controller/main.go

@@ -14,6 +14,8 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strings"
+	"time"
 
 	egextension "github.com/envoyproxy/gateway/proto/extension"
 	"go.uber.org/zap/zapcore"
@@ -22,7 +24,9 @@ import (
 	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
 	corev1 "k8s.io/api/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
@@ -51,7 +55,9 @@ type flags struct {
 	// extProcMaxRecvMsgSize is the maximum message size in bytes that the gRPC server can receive.
 	extProcMaxRecvMsgSize int
 	// maxRecvMsgSize is the maximum message size in bytes that the gRPC extension server can receive.
-	maxRecvMsgSize int
+	maxRecvMsgSize   int
+	watchNamespaces  []string
+	cacheSyncTimeout time.Duration
 }
 
 // parsePullPolicy parses string into a k8s PullPolicy.
@@ -64,6 +70,18 @@ func parsePullPolicy(s string) (corev1.PullPolicy, error) {
 	}
 }
 
+// parseWatchNamespaces parses a comma-separated list of namespaces into a slice of strings.
+func parseWatchNamespaces(s string) []string {
+	var namespaces []string
+	for _, n := range strings.Split(s, ",") {
+		ns := strings.TrimSpace(n)
+		if ns != "" {
+			namespaces = append(namespaces, ns)
+		}
+	}
+	return namespaces
+}
+
 // parseAndValidateFlags parses the command-line arguments provided in args,
 // validates them, and returns the parsed configuration.
 func parseAndValidateFlags(args []string) (flags, error) {
@@ -159,6 +177,16 @@ func parseAndValidateFlags(args []string) (flags, error) {
 		4*1024*1024,
 		"Maximum message size in bytes that the gRPC extension server can receive. Default is 4MB.",
 	)
+	watchNamespaces := fs.String(
+		"watchNamespaces",
+		"",
+		"Comma-separated list of namespaces to watch. If not set, the controller watches all namespaces.",
+	)
+	cacheSyncTimeout := fs.Duration(
+		"cacheSyncTimeout",
+		2*time.Minute, // This is the controller-runtime default
+		"Maximum time to wait for k8s caches to sync",
+	)
 
 	if err := fs.Parse(args); err != nil {
 		err = fmt.Errorf("failed to parse flags: %w", err)
@@ -238,41 +266,47 @@ func parseAndValidateFlags(args []string) (flags, error) {
 		extProcImagePullSecrets:        *extProcImagePullSecrets,
 		extProcMaxRecvMsgSize:          *extProcMaxRecvMsgSize,
 		maxRecvMsgSize:                 *maxRecvMsgSize,
+		watchNamespaces:                parseWatchNamespaces(*watchNamespaces),
+		cacheSyncTimeout:               *cacheSyncTimeout,
 	}, nil
 }
 
 func main() {
 	setupLog := ctrl.Log.WithName("setup")
 
-	flags, err := parseAndValidateFlags(os.Args[1:])
+	parsedFlags, err := parseAndValidateFlags(os.Args[1:])
 	if err != nil {
 		setupLog.Error(err, "failed to parse and validate flags")
 		os.Exit(1)
 	}
 
 	// Warn if deprecated flag is being used.
-	if flags.metricsRequestHeaderLabels != "" {
+	if parsedFlags.metricsRequestHeaderLabels != "" {
 		setupLog.Info("The --metricsRequestHeaderLabels flag is deprecated and will be removed in a future release. Please use --metricsRequestHeaderAttributes instead.")
 	}
 
-	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&zap.Options{Development: true, Level: flags.logLevel})))
+	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&zap.Options{Development: true, Level: parsedFlags.logLevel})))
 	k8sConfig := ctrl.GetConfigOrDie()
 
-	lis, err := net.Listen("tcp", flags.extensionServerPort)
+	lis, err := net.Listen("tcp", parsedFlags.extensionServerPort)
 	if err != nil {
-		setupLog.Error(err, "failed to listen", "port", flags.extensionServerPort)
+		setupLog.Error(err, "failed to listen", "port", parsedFlags.extensionServerPort)
 		os.Exit(1)
 	}
 
+	setupLog.Info("configuring kubernetes cache", "watch-namespaces", parsedFlags.watchNamespaces, "sync-timeout", parsedFlags.cacheSyncTimeout)
+
 	ctx := ctrl.SetupSignalHandler()
 	mgrOpts := ctrl.Options{
+		Cache:            setupCache(parsedFlags),
+		Controller:       config.Controller{CacheSyncTimeout: parsedFlags.cacheSyncTimeout},
 		Scheme:           controller.Scheme,
-		LeaderElection:   flags.enableLeaderElection,
+		LeaderElection:   parsedFlags.enableLeaderElection,
 		LeaderElectionID: "envoy-ai-gateway-controller",
 		WebhookServer: webhook.NewServer(webhook.Options{
-			CertDir:  flags.tlsCertDir,
-			CertName: flags.tlsCertName,
-			KeyName:  flags.tlsKeyName,
+			CertDir:  parsedFlags.tlsCertDir,
+			CertName: parsedFlags.tlsCertName,
+			KeyName:  parsedFlags.tlsKeyName,
 			Port:     9443,
 		}),
 	}
@@ -287,14 +321,14 @@ func main() {
 		setupLog.Error(err, "failed to create client")
 		os.Exit(1)
 	}
-	if err := maybePatchAdmissionWebhook(ctx, cli, filepath.Join(flags.tlsCertDir, flags.caBundleName)); err != nil {
+	if err := maybePatchAdmissionWebhook(ctx, cli, filepath.Join(parsedFlags.tlsCertDir, parsedFlags.caBundleName)); err != nil {
 		setupLog.Error(err, "failed to patch admission webhook")
 		os.Exit(1)
 	}
 
 	// Start the extension server running alongside the controller.
 	const extProcUDSPath = "/etc/ai-gateway-extproc-uds/run.sock"
-	s := grpc.NewServer(grpc.MaxRecvMsgSize(flags.maxRecvMsgSize))
+	s := grpc.NewServer(grpc.MaxRecvMsgSize(parsedFlags.maxRecvMsgSize))
 	extSrv := extensionserver.New(mgr.GetClient(), ctrl.Log, extProcUDSPath, false)
 	egextension.RegisterEnvoyGatewayExtensionServer(s, extSrv)
 	grpc_health_v1.RegisterHealthServer(s, extSrv)
@@ -310,17 +344,17 @@ func main() {
 
 	// Start the controller.
 	if err := controller.StartControllers(ctx, mgr, k8sConfig, ctrl.Log.WithName("controller"), controller.Options{
-		ExtProcImage:                   flags.extProcImage,
-		ExtProcImagePullPolicy:         flags.extProcImagePullPolicy,
-		ExtProcLogLevel:                flags.extProcLogLevel,
-		EnableLeaderElection:           flags.enableLeaderElection,
+		ExtProcImage:                   parsedFlags.extProcImage,
+		ExtProcImagePullPolicy:         parsedFlags.extProcImagePullPolicy,
+		ExtProcLogLevel:                parsedFlags.extProcLogLevel,
+		EnableLeaderElection:           parsedFlags.enableLeaderElection,
 		UDSPath:                        extProcUDSPath,
-		MetricsRequestHeaderAttributes: flags.metricsRequestHeaderAttributes,
-		TracingRequestHeaderAttributes: flags.spanRequestHeaderAttributes,
-		RootPrefix:                     flags.rootPrefix,
-		ExtProcExtraEnvVars:            flags.extProcExtraEnvVars,
-		ExtProcImagePullSecrets:        flags.extProcImagePullSecrets,
-		ExtProcMaxRecvMsgSize:          flags.extProcMaxRecvMsgSize,
+		MetricsRequestHeaderAttributes: parsedFlags.metricsRequestHeaderAttributes,
+		TracingRequestHeaderAttributes: parsedFlags.spanRequestHeaderAttributes,
+		RootPrefix:                     parsedFlags.rootPrefix,
+		ExtProcExtraEnvVars:            parsedFlags.extProcExtraEnvVars,
+		ExtProcImagePullSecrets:        parsedFlags.extProcImagePullSecrets,
+		ExtProcMaxRecvMsgSize:          parsedFlags.extProcMaxRecvMsgSize,
 	}); err != nil {
 		setupLog.Error(err, "failed to start controller")
 	}
@@ -356,3 +390,19 @@ func maybePatchAdmissionWebhook(ctx context.Context, cli client.Client, bundlePa
 	}
 	return nil
 }
+
+// setupCache sets up the cache options based on the provided flags.
+func setupCache(f flags) cache.Options {
+	var namespaceCacheConfig map[string]cache.Config
+	if len(f.watchNamespaces) > 0 {
+		namespaceCacheConfig = make(map[string]cache.Config, len(f.watchNamespaces))
+		for _, ns := range f.watchNamespaces {
+			namespaceCacheConfig[ns] = cache.Config{}
+		}
+	}
+
+	return cache.Options{
+		DefaultNamespaces: namespaceCacheConfig,
+		DefaultTransform:  cache.TransformStripManagedFields(),
+	}
+}

cmd/controller/main_test.go

@@ -8,11 +8,13 @@ package main
 import (
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
@@ -51,6 +53,8 @@ func Test_parseAndValidateFlags(t *testing.T) {
 					tc.dash + "extProcExtraEnvVars=OTEL_SERVICE_NAME=test;OTEL_TRACES_EXPORTER=console",
 					tc.dash + "spanRequestHeaderAttributes=x-session-id:session.id",
 					tc.dash + "maxRecvMsgSize=33554432",
+					tc.dash + "watchNamespaces=default,envoy-ai-gateway-system",
+					tc.dash + "cacheSyncTimeout=5m",
 				}
 				f, err := parseAndValidateFlags(args)
 				require.Equal(t, "debug", f.extProcLogLevel)
@@ -62,6 +66,8 @@ func Test_parseAndValidateFlags(t *testing.T) {
 				require.Equal(t, "OTEL_SERVICE_NAME=test;OTEL_TRACES_EXPORTER=console", f.extProcExtraEnvVars)
 				require.Equal(t, "x-session-id:session.id", f.spanRequestHeaderAttributes)
 				require.Equal(t, 32*1024*1024, f.maxRecvMsgSize)
+				require.Equal(t, []string{"default", "envoy-ai-gateway-system"}, f.watchNamespaces)
+				require.Equal(t, 5*time.Minute, f.cacheSyncTimeout)
 				require.NoError(t, err)
 			})
 		}
@@ -243,3 +249,53 @@ func Test_parseAndValidateFlags_extProcImagePullSecrets(t *testing.T) {
 		})
 	}
 }
+
+func Test_parseAndValidateFlags_watchNamespaces(t *testing.T) {
+	tests := []struct {
+		name     string
+		flags    []string
+		expected []string
+	}{
+		{"no watch namespaces", []string{}, nil},
+		{"single watch namespace", []string{"--watchNamespaces=default"}, []string{"default"}},
+		{"multiple watch namespaces", []string{"--watchNamespaces=default,envoy-ai-gateway-system"}, []string{"default", "envoy-ai-gateway-system"}},
+		{"watch namespaces with spaces", []string{"--watchNamespaces= default , envoy-ai-gateway-system "}, []string{"default", "envoy-ai-gateway-system"}},
+		{"empty string", []string{"--watchNamespaces="}, nil},
+		{"empty namespace names", []string{"--watchNamespaces=default,,envoy-ai-gateway-system"}, []string{"default", "envoy-ai-gateway-system"}},
+		{"only commas", []string{"--watchNamespaces=,,,"}, nil},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f, err := parseAndValidateFlags(tt.flags)
+			require.NoError(t, err)
+			require.Equal(t, tt.expected, f.watchNamespaces)
+		})
+	}
+}
+
+func TestSetupCache(t *testing.T) {
+	t.Run("default", func(t *testing.T) {
+		c := setupCache(flags{})
+
+		require.NotNil(t, c.DefaultTransform)
+		require.Nil(t, c.DefaultNamespaces)
+	})
+
+	t.Run("empty watch namespaces", func(t *testing.T) {
+		c := setupCache(flags{watchNamespaces: []string{}})
+
+		require.NotNil(t, c.DefaultTransform)
+		require.Nil(t, c.DefaultNamespaces)
+	})
+
+	t.Run("watch namespaces", func(t *testing.T) {
+		c := setupCache(flags{watchNamespaces: []string{"default", "envoy-ai-gateway-system"}})
+
+		require.NotNil(t, c.DefaultTransform)
+		require.Equal(t, map[string]cache.Config{
+			"default":                 {},
+			"envoy-ai-gateway-system": {},
+		}, c.DefaultNamespaces)
+	})
+}

manifests/charts/ai-gateway-helm/templates/deployment.yaml

@@ -68,6 +68,8 @@ spec:
             {{- if ne .Values.controller.maxRecvMsgSize nil }}
             - --maxRecvMsgSize={{ .Values.controller.maxRecvMsgSize }}
             {{- end }}
+            - --cacheSyncTimeout={{ .Values.controller.watch.cacheSyncTimeout }}
+            - --watchNamespaces={{ join "," .Values.controller.watch.namespaces }}
           livenessProbe:
             grpc:
               port: 1063

manifests/charts/ai-gateway-helm/values.yaml

@@ -67,6 +67,15 @@ controller:
   leaderElection:
     enabled: true
 
+  # Configuration for how the Kubernetes controllers watch the different resources.
+  watch:
+    # Namespaces to watch. An empty list means to watch all namespaces.
+    # Default is an empty list, to watch all namespaces.
+    namespaces: []
+    # Sync timeout for the Kubernetes cache. If the cache is not synced within this time, the controller will exit.
+    # Default is 2 minutes.
+    cacheSyncTimeout: 2m
+
   # -- Deployment configs --
   image:
     repository: docker.io/envoyproxy/ai-gateway-controller