From b7168e454b465cc58862257b71a1673755f7f138 Mon Sep 17 00:00:00 2001 From: Jian Zhu Date: Thu, 7 May 2026 23:13:45 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Fix=20concurrency=20bugs=20in=20?= =?UTF-8?q?executor=20cache=20(#1512)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 🐛 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 * 🐛 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 * 🌱 Make RemoveByHash private as it is only used internally Assisted by Claude Signed-off-by: zhujian * 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 --------- Signed-off-by: zhujian --- pkg/work/spoke/auth/cache/auth.go | 1 + .../auth/cache/executor_cache_controller.go | 4 +- .../cache/executor_cache_controller_test.go | 122 ++++++++++++++++++ pkg/work/spoke/auth/store/cache_store.go | 39 +++--- pkg/work/spoke/auth/store/cache_store_test.go | 105 ++++++++++++++- 5 files changed, 253 insertions(+), 18 deletions(-) diff --git a/pkg/work/spoke/auth/cache/auth.go b/pkg/work/spoke/auth/cache/auth.go index c9443315f..185f704aa 100644 --- a/pkg/work/spoke/auth/cache/auth.go +++ b/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 diff --git a/pkg/work/spoke/auth/cache/executor_cache_controller.go b/pkg/work/spoke/auth/cache/executor_cache_controller.go index 8ce6bb086..fda0af26c 100644 --- a/pkg/work/spoke/auth/cache/executor_cache_controller.go +++ b/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 { diff --git a/pkg/work/spoke/auth/cache/executor_cache_controller_test.go b/pkg/work/spoke/auth/cache/executor_cache_controller_test.go index cbd19ef37..7182913be 100644 --- a/pkg/work/spoke/auth/cache/executor_cache_controller_test.go +++ b/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 { diff --git a/pkg/work/spoke/auth/store/cache_store.go b/pkg/work/spoke/auth/store/cache_store.go index 5d578737e..8f9b278da 100644 --- a/pkg/work/spoke/auth/store/cache_store.go +++ b/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 { diff --git a/pkg/work/spoke/auth/store/cache_store_test.go b/pkg/work/spoke/auth/store/cache_store_test.go index 6e3bde3f9..1411721af 100644 --- a/pkg/work/spoke/auth/store/cache_store_test.go +++ b/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() +}