add namespace watch

Author: michaelhtm

pkg/runtime/cache/namespace.go

@@ -38,6 +38,8 @@ type namespaceInfo struct {
 	endpointURL string
 	// {service}.services.k8s.aws/deletion-policy Annotations (keyed by service)
 	deletionPolicies map[string]string
+
+	labels map[string]string
 }
 
 // getDefaultRegion returns the default region value
@@ -226,6 +228,16 @@ func (c *NamespaceCache) GetDeletionPolicy(namespace string, service string) (st
 	return "", false
 }
 
+// GetDeletionPolicy returns the deletion policy if it exists
+func (c *NamespaceCache) GetLabels(namespace string) map[string]string {
+	info, ok := c.getNamespaceInfo(namespace)
+	if ok {
+		return info.labels
+	}
+	return nil
+	
+}
+
 // getNamespaceInfo reads a namespace cached annotations and
 // return a given namespace default aws region, owner account id and endpoint url.
 // This function is thread safe.
@@ -268,6 +280,8 @@ func (c *NamespaceCache) setNamespaceInfoFromK8sObject(ns *corev1.Namespace) {
 		nsInfo.deletionPolicies[service] = elem
 	}
 
+	nsInfo.labels = ns.GetLabels()
+
 	c.Lock()
 	defer c.Unlock()
 	c.namespaceInfos[ns.ObjectMeta.Name] = nsInfo

pkg/runtime/iamroleselector/cache.go

@@ -14,6 +14,7 @@
 package iamroleselector
 
 import (
+	"context"
 	"fmt"
 	"sync"
 
@@ -24,29 +25,33 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
 	ackv1alpha1 "github.com/aws-controllers-k8s/runtime/apis/core/v1alpha1"
+	ackcache "github.com/aws-controllers-k8s/runtime/pkg/runtime/cache"
 )
 
 // Cache wraps the informer for IAMRoleSelector resources
 type Cache struct {
 	sync.RWMutex
-	log       logr.Logger
-	informer  cache.SharedIndexInformer
-	selectors map[string]*ackv1alpha1.IAMRoleSelector // name -> selector
+	namespaces *ackcache.NamespaceCache
+	log        logr.Logger
+	informer   cache.SharedIndexInformer
+	selectors  map[string]*ackv1alpha1.IAMRoleSelector // name -> selector
 }
 
 // NewCache creates a new IAMRoleSelector cache
 func NewCache(log logr.Logger) *Cache {
 	return &Cache{
-		log:       log.WithName("cache.iam-role-selector"),
-		selectors: make(map[string]*ackv1alpha1.IAMRoleSelector),
+		log:        log.WithName("cache.iam-role-selector"),
+		selectors:  make(map[string]*ackv1alpha1.IAMRoleSelector),
+		namespaces: ackcache.NewNamespaceCache(log, nil, nil),
 	}
 }
 
 // Run starts the cache and blocks until stopCh is closed
-func (c *Cache) Run(client dynamic.Interface, stopCh <-chan struct{}) {
+func (c *Cache) Run(client dynamic.Interface, namespaceClient kubernetes.Interface, stopCh <-chan struct{}) {
 	c.log.V(1).Info("Starting IAMRoleSelector cache")
 
 	// Create dynamic informer factory
@@ -74,6 +79,8 @@ func (c *Cache) Run(client dynamic.Interface, stopCh <-chan struct{}) {
 	})
 
 	factory.Start(stopCh)
+
+	c.namespaces.Run(namespaceClient, stopCh)
 }
 
 func (c *Cache) handleAdd(obj interface{}) {
@@ -197,15 +204,15 @@ func (c *Cache) ListSelectors() []*ackv1alpha1.IAMRoleSelector {
 
 // Matches returns a list of IAMRoleSelectors that match the given resource. This function
 // should only be called after the cache has been started and synced.
-func (c *Cache) Matches(resource runtime.Object) ([]*ackv1alpha1.IAMRoleSelector, error) {
+func (c *Cache) Matches(ctx context.Context, resource runtime.Object) ([]*ackv1alpha1.IAMRoleSelector, error) {
 	// Extract metadata from the resource
 	metaObj, err := meta.Accessor(resource)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get metadata from resource: %w", err)
 	}
 
-	namespace := metaObj.GetNamespace()
-
+	namespaceName := metaObj.GetNamespace()
+	namespaceLabels := c.namespaces.GetLabels(namespaceName)
 	// Get GVK - should be set on ACK resources
 	gvk := resource.GetObjectKind().GroupVersionKind()
 	if gvk.Empty() {
@@ -215,5 +222,5 @@ func (c *Cache) Matches(resource runtime.Object) ([]*ackv1alpha1.IAMRoleSelector
 
 	// TODO: get namespace labels from a namespace lister/cache
 	// For now, pass empty namespace labels
-	return c.GetMatchingSelectors(namespace, nil, gvk)
+	return c.GetMatchingSelectors(namespaceName, namespaceLabels, gvk)
 }

pkg/runtime/iamroleselector/cache_test.go

@@ -14,6 +14,7 @@
 package iamroleselector
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -27,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic/fake"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 
 	ackv1alpha1 "github.com/aws-controllers-k8s/runtime/apis/core/v1alpha1"
@@ -53,13 +55,16 @@ func TestCache_Matches(t *testing.T) {
 	client := fake.NewSimpleDynamicClientWithCustomListKinds(scheme, gvrToListKind)
 	client.PrependWatchReactor("iamroleselectors", k8stesting.DefaultWatchReactor(watcher, nil))
 
+	k8sClient := k8sfake.NewSimpleClientset()
+	k8sClient.PrependWatchReactor("production", k8stesting.DefaultWatchReactor(watcher, nil))
+
 	logger := zapr.NewLogger(zap.NewNop())
 	cache := NewCache(logger)
 
 	stopCh := make(chan struct{})
 	t.Cleanup(func() { close(stopCh) })
 
-	go cache.Run(client, stopCh)
+	go cache.Run(client, k8sClient, stopCh)
 
 	// Wait for cache to sync
 	require.Eventually(t, func() bool {
@@ -145,10 +150,11 @@ func TestCache_Matches(t *testing.T) {
 			wantCount: 0,
 		},
 	}
+	ctx := context.TODO()
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			matches, err := cache.Matches(tt.resource)
+			matches, err := cache.Matches(ctx, tt.resource)
 			require.NoError(t, err)
 			require.Len(t, matches, tt.wantCount)
 

pkg/runtime/reconciler.go

@@ -266,10 +266,9 @@ func (r *resourceReconciler) Reconcile(ctx context.Context, req ctrlrt.Request)
 		// If the IAMRoleSelector feature gate is enabled, we need to check if there
 		// are any matching IAMRoleSelectors for this resource. If there are, we
 		// override the roleARN from CARM (if any) with the one from the selector.
-		selectors, err := r.irsCache.GetMatchingSelectors(
-			req.Namespace,
-			nil,
-			r.rd.GroupVersionKind(),
+		selectors, err := r.irsCache.Matches(
+			ctx,
+			desired.RuntimeObject(),
 		)
 		if err != nil {
 			return ctrlrt.Result{}, fmt.Errorf("checking for matching IAMRoleSelectors: %w", err)

pkg/runtime/service_controller.go

@@ -213,18 +213,18 @@ func (c *serviceController) BindControllerManager(mgr ctrlrt.Manager, cfg ackcfg
 		}},
 		cfg.FeatureGates,
 	)
+	clusterConfig := mgr.GetConfig()
+	clientSet, err := kubernetes.NewForConfig(clusterConfig)
+	if err != nil {
+		return err
+	}
 	// The caches are only used for cross account resource management. We
 	// want to run them only when --enable-carm is set to true and
 	// --watch-namespace is set to zero or more than one namespaces.
 	if cfg.EnableCARM {
 		if len(namespaces) == 1 {
 			c.log.V(0).Info("--enable-carm is set to true but --watch-namespace is set to a single namespace. CARM will not be enabled.")
 		} else {
-			clusterConfig := mgr.GetConfig()
-			clientSet, err := kubernetes.NewForConfig(clusterConfig)
-			if err != nil {
-				return err
-			}
 			// Run the caches. This will not block as the caches are run in
 			// separate goroutines.
 			carmCache.Run(clientSet)
@@ -268,11 +268,11 @@ func (c *serviceController) BindControllerManager(mgr ctrlrt.Manager, cfg ackcfg
 	if cfg.FeatureGates.IsEnabled(featuregate.IAMRoleSelector) {
 		// init dynamic client
 		clusterConfig := mgr.GetConfig()
-		clientSet, err := dynamic.NewForConfig(clusterConfig)
+		dynamicClient, err := dynamic.NewForConfig(clusterConfig)
 		if err != nil {
 			return err
 		}
-		irsCache.Run(clientSet, context.TODO().Done())
+		irsCache.Run(dynamicClient, clientSet, context.TODO().Done())
 	}
 
 	// Filter the resource manager factories