🐛 Fix concurrency bugs in executor cache (#1512)

* 🐛 Fix concurrency bugs in executor cache

- Fix DimensionCaches.remove() using RLock instead of Lock for map
  delete operation, which could cause concurrent map read/write panic
- Fix RemoveByHash accessing len(items) without holding the lock
- Fix getCacheItems returning internal map reference, allowing
  unsynchronized iteration after lock release; return snapshot copies
- Add early return in updateSARCheckResultToCache for clarity

Assisted by Claude

Signed-off-by: zhujian <jiazhu@redhat.com>

* 🐛 Fix wrong index in clusterRoleEnqueueFu causing missed cache refresh

When a ClusterRole changes, the controller should find RoleBindings
referencing it via the byClusterRole index. It was incorrectly using
the byRole index, which indexes by "namespace/name" for Role refs
and never matches a bare ClusterRole name. This caused executor
caches to not refresh when a ClusterRole was modified, leaving
revoked permissions cached as allowed for up to 10 minutes.

Assisted by Claude

Signed-off-by: zhujian <jiazhu@redhat.com>

* 🌱 Make RemoveByHash private as it is only used internally

Assisted by Claude

Signed-off-by: zhujian <jiazhu@redhat.com>

* Add concurrency and index-fix tests for executor cache

- Add concurrent remove/get, getCacheItems, and cleanup tests to verify
  race-free behavior with -race detector
- Add TestCacheControllerClusterRoleWithRoleBindingOnly to verify
  clusterRoleEnqueueFu uses byClusterRole index for RoleBindings

Signed-off-by: zhujian <jiazhu@redhat.com>

---------

Signed-off-by: zhujian <jiazhu@redhat.com>

pkg/work/spoke/auth/cache/auth.go

@@ -145,6 +145,7 @@ func updateSARCheckResultToCache(executorCaches *store.ExecutorCaches, executorK
 	dimension store.Dimension, result error) {
 	if result == nil {
 		executorCaches.Upsert(executorKey, dimension, pointer.Bool(true))
+		return
 	}
 
 	var authError *basic.NotAllowedError

pkg/work/spoke/auth/cache/executor_cache_controller.go

@@ -165,10 +165,10 @@ func (c *CacheController) clusterRoleEnqueueFu(
 		ret := make([]string, 0)
 
 		clusterRoleKey := accessor.GetName()
-		items, err := rbIndexer.ByIndex(byRole, clusterRoleKey)
+		items, err := rbIndexer.ByIndex(byClusterRole, clusterRoleKey)
 		if err != nil {
 			klog.V(4).Infof("RoleBinding indexer get RoleBinding by %s index %s error: %v",
-				byRole, clusterRoleKey, err)
+				byClusterRole, clusterRoleKey, err)
 		} else {
 			for _, item := range items {
 				if rb, ok := item.(*rbacapiv1.RoleBinding); ok {

pkg/work/spoke/auth/cache/executor_cache_controller_test.go

@@ -288,6 +288,128 @@ func countSARRequests(kubeClientActions []clienttesting.Action) int {
 	return len(actualSARActions)
 }
 
+// TestCacheControllerClusterRoleWithRoleBindingOnly verifies that when a ClusterRole changes,
+// the controller finds RoleBindings referencing it via the byClusterRole index (not byRole).
+func TestCacheControllerClusterRoleWithRoleBindingOnly(t *testing.T) {
+	executor := &workapiv1.ManifestWorkExecutor{
+		Subject: workapiv1.ManifestWorkExecutorSubject{
+			Type: workapiv1.ExecutorSubjectTypeServiceAccount,
+			ServiceAccount: &workapiv1.ManifestWorkSubjectServiceAccount{
+				Namespace: "test-ns",
+				Name:      "test-name",
+			},
+		},
+	}
+
+	clusterRoleName := "test-cluster-role"
+	rbNamespace := "test-ns"
+
+	clusterRole := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: clusterRoleName,
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				Verbs:     []string{"create", "update", "patch", "get", "list", "delete"},
+				APIGroups: []string{""},
+				Resources: []string{"configmaps"},
+			},
+		},
+	}
+
+	roleBinding := &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "rb-for-cluster-role",
+			Namespace: rbNamespace,
+		},
+		Subjects: []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: executor.Subject.ServiceAccount.Namespace,
+				Name:      executor.Subject.ServiceAccount.Name,
+			},
+		},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: "rbac.authorization.k8s.io",
+			Kind:     "ClusterRole",
+			Name:     clusterRoleName,
+		},
+	}
+
+	kubeClient := fakekube.NewSimpleClientset(clusterRole, roleBinding)
+	kubeClient.PrependReactor("create", "subjectaccessreviews",
+		func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
+			obj := action.(clienttesting.CreateActionImpl).Object.(*v1.SubjectAccessReview)
+
+			if obj.Spec.ResourceAttributes.Namespace == allowNS {
+				return true, &v1.SubjectAccessReview{
+					Status: v1.SubjectAccessReviewStatus{
+						Allowed: true,
+					},
+				}, nil
+			}
+
+			if obj.Spec.ResourceAttributes.Namespace == denyNS {
+				return true, &v1.SubjectAccessReview{
+					Status: v1.SubjectAccessReviewStatus{
+						Denied: true,
+					},
+				}, nil
+			}
+			return false, nil, nil
+		},
+	)
+
+	ctx := context.TODO()
+
+	work, _ := spoketesting.NewManifestWork(0,
+		testingcommon.NewUnstructured("v1", "Secret", allowNS, "test"),
+		testingcommon.NewUnstructured("v1", "Secret", denyNS, "test"),
+	)
+	work.Spec.Executor = executor
+	work.Spec.DeleteOption = &workapiv1.DeleteOption{
+		PropagationPolicy: workapiv1.DeletePropagationPolicyTypeSelectivelyOrphan,
+		SelectivelyOrphan: &workapiv1.SelectivelyOrphan{
+			OrphaningRules: []workapiv1.OrphaningRule{
+				{
+					Group:     "",
+					Resource:  "secrets",
+					Namespace: allowNS,
+					Name:      "test",
+				},
+			},
+		},
+	}
+
+	initialized := make(chan struct{})
+	cacheController := newExecutorCacheController(t, ctx, clusterName, kubeClient, initialized, work)
+	<-initialized
+
+	err := checkSARCount(kubeClient, 5)
+	if err != nil {
+		t.Error(err)
+	}
+
+	executorKey := fmt.Sprintf("%s/%s",
+		executor.Subject.ServiceAccount.Namespace, executor.Subject.ServiceAccount.Name)
+	actualMapCount := cacheController.bindingExecutorsMapper.count()
+	if actualMapCount != 1 {
+		t.Errorf("Expected 1 map item (RoleBinding only) but got %d", actualMapCount)
+	}
+	checkBindingExecutorMapperInitialized(t, cacheController.bindingExecutorsMapper,
+		fmt.Sprintf("%s/%s", rbNamespace, "rb-for-cluster-role"), executorKey)
+
+	err = kubeClient.RbacV1().ClusterRoles().Delete(ctx, clusterRoleName, metav1.DeleteOptions{})
+	if err != nil {
+		t.Errorf("Expected no error, but got %v", err)
+	}
+
+	err = checkSARCount(kubeClient, 2*5)
+	if err != nil {
+		t.Errorf("ClusterRole deletion did not trigger cache refresh through RoleBinding path: %v", err)
+	}
+}
+
 func checkBindingExecutorMapperInitialized(t *testing.T, m *safeMap, roleKey, executorKey string) {
 	actualExecutors := m.get(roleKey)
 	if len(actualExecutors) != 1 {

pkg/work/spoke/auth/store/cache_store.go

@@ -100,8 +100,8 @@ func (c *ExecutorCaches) Get(executor string, dimension Dimension) (allowed *boo
 	return oldDimensionCaches.get(dimension.Hash())
 }
 
-// RemoveByHash removes an cache item by dimension hash
-func (c *ExecutorCaches) RemoveByHash(executor string, hash string) {
+// removeByHash removes an cache item by dimension hash
+func (c *ExecutorCaches) removeByHash(executor string, hash string) {
 	oldDimensionCaches, ok := c.getDimensionCaches(executor)
 	if !ok {
 		return
@@ -110,7 +110,7 @@ func (c *ExecutorCaches) RemoveByHash(executor string, hash string) {
 	oldDimensionCaches.remove(hash)
 
 	// if the deleted cache is the last element, delete the upper level dimension caches
-	if len(oldDimensionCaches.items) == 0 {
+	if oldDimensionCaches.len() == 0 {
 		c.removeDimensionCaches(executor)
 	}
 }
@@ -126,7 +126,7 @@ func (c *ExecutorCaches) CleanupUnnecessaryCaches(necessaryCaches *ExecutorCache
 
 		for hash := range caches.getCacheItems() {
 			if _, ok := necessaryCaches.getByHash(key, hash); !ok {
-				c.RemoveByHash(key, hash)
+				c.removeByHash(key, hash)
 				klog.V(4).Infof("Remove cache item executor %s dimension %s", key, hash)
 			}
 		}
@@ -197,9 +197,13 @@ func (c *ExecutorCaches) removeDimensionCaches(executor string) {
 }
 
 func (c *ExecutorCaches) getCacheItems() map[string]*DimensionCaches {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-	return c.items
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	copied := make(map[string]*DimensionCaches, len(c.items))
+	for k, v := range c.items {
+		copied[k] = v
+	}
+	return copied
 }
 
 // getByHash gets a cache item value and existence by the dimension hash
@@ -213,17 +217,18 @@ func (c *ExecutorCaches) getByHash(executor string, hash string) (*bool, bool) {
 }
 
 func (c *DimensionCaches) remove(hash string) {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-
-	_, ok := c.items[hash]
-	if !ok {
-		return
-	}
+	c.lock.Lock()
+	defer c.lock.Unlock()
 
 	delete(c.items, hash)
 }
 
+func (c *DimensionCaches) len() int {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	return len(c.items)
+}
+
 func (c *DimensionCaches) get(hash string) (*bool, bool) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
@@ -248,7 +253,11 @@ func (c *DimensionCaches) upsert(dimension Dimension, allowed *bool) {
 func (c *DimensionCaches) getCacheItems() map[string]CacheValue {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
-	return c.items
+	copied := make(map[string]CacheValue, len(c.items))
+	for k, v := range c.items {
+		copied[k] = v
+	}
+	return copied
 }
 
 func (d *Dimension) Hash() string {

pkg/work/spoke/auth/store/cache_store_test.go

@@ -72,7 +72,7 @@ func TestBasic(t *testing.T) {
 	executor := fmt.Sprintf("%v", 0)
 	for j := 0; j < 10; j++ {
 		d := Dimension{Name: fmt.Sprintf("%v", j)}
-		caches.RemoveByHash(executor, d.Hash())
+		caches.removeByHash(executor, d.Hash())
 	}
 
 	exist = caches.DimensionCachesExists(executor)
@@ -95,3 +95,106 @@ func TestBasic(t *testing.T) {
 		t.Errorf("Expected dimension name joining result 45 but got %v", dimensionNameAccumulate)
 	}
 }
+
+func TestConcurrentRemoveAndGet(t *testing.T) {
+	caches := NewExecutorCache()
+	executor := "test-executor"
+
+	for i := 0; i < 100; i++ {
+		d := Dimension{Name: fmt.Sprintf("%d", i)}
+		caches.Upsert(executor, d, nil)
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(200)
+	for i := 0; i < 100; i++ {
+		go func(i int) {
+			defer wg.Done()
+			d := Dimension{Name: fmt.Sprintf("%d", i)}
+			caches.removeByHash(executor, d.Hash())
+		}(i)
+		go func(i int) {
+			defer wg.Done()
+			d := Dimension{Name: fmt.Sprintf("%d", i)}
+			caches.Get(executor, d)
+		}(i)
+	}
+	wg.Wait()
+
+	if caches.Count() != 0 {
+		t.Errorf("Expected all items removed but got %d", caches.Count())
+	}
+}
+
+func TestConcurrentGetCacheItemsWhileModifying(t *testing.T) {
+	caches := NewExecutorCache()
+
+	for i := 0; i < 10; i++ {
+		for j := 0; j < 10; j++ {
+			caches.Upsert(fmt.Sprintf("%d", i), Dimension{Name: fmt.Sprintf("%d", j)}, nil)
+		}
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(30)
+
+	for i := 0; i < 10; i++ {
+		go func() {
+			defer wg.Done()
+			caches.Count()
+		}()
+	}
+
+	for i := 0; i < 10; i++ {
+		go func(i int) {
+			defer wg.Done()
+			caches.Upsert(fmt.Sprintf("new-%d", i), Dimension{Name: "test"}, nil)
+		}(i)
+	}
+
+	for i := 0; i < 10; i++ {
+		go func(i int) {
+			defer wg.Done()
+			d := Dimension{Name: fmt.Sprintf("%d", 0)}
+			caches.removeByHash(fmt.Sprintf("%d", i), d.Hash())
+		}(i)
+	}
+
+	wg.Wait()
+}
+
+func TestConcurrentCleanupUnnecessaryCaches(t *testing.T) {
+	caches := NewExecutorCache()
+
+	for i := 0; i < 20; i++ {
+		for j := 0; j < 10; j++ {
+			caches.Upsert(fmt.Sprintf("%d", i), Dimension{Name: fmt.Sprintf("%d", j)}, nil)
+		}
+	}
+
+	necessaryCaches := NewExecutorCache()
+	for i := 0; i < 10; i++ {
+		for j := 0; j < 5; j++ {
+			necessaryCaches.Upsert(fmt.Sprintf("%d", i), Dimension{Name: fmt.Sprintf("%d", j)}, nil)
+		}
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(11)
+
+	go func() {
+		defer wg.Done()
+		caches.CleanupUnnecessaryCaches(necessaryCaches)
+	}()
+
+	for i := 0; i < 10; i++ {
+		go func(i int) {
+			defer wg.Done()
+			executor := fmt.Sprintf("%d", i)
+			d := Dimension{Name: fmt.Sprintf("%d", i)}
+			caches.Get(executor, d)
+		}(i)
+	}
+
+	wg.Wait()
+}