diff --git a/pkg/project/auth/cache.go b/pkg/project/auth/cache.go index c4fd41d707..719e86c273 100644 --- a/pkg/project/auth/cache.go +++ b/pkg/project/auth/cache.go @@ -4,6 +4,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" "k8s.io/klog/v2" @@ -173,6 +174,14 @@ func (l syncedClusterRoleBindingLister) LastSyncResourceVersion() string { return l.versioner.LastSyncResourceVersion() } +// authorizationCacheStores groups the three cache stores so they can be +// swapped atomically during full cache invalidation. +type authorizationCacheStores struct { + reviewRecordStore cache.Store + userSubjectRecordStore cache.Store + groupSubjectRecordStore cache.Store +} + // AuthorizationCache maintains a cache on the set of namespaces a user or group can access. type AuthorizationCache struct { // allKnownNamespaces we track all the known namespaces, so we can detect deletes. @@ -187,9 +196,7 @@ type AuthorizationCache struct { roleBindingNamespacer SyncedRoleBindingLister roleLastSyncResourceVersioner LastSyncResourceVersioner - reviewRecordStore cache.Store - userSubjectRecordStore cache.Store - groupSubjectRecordStore cache.Store + stores atomic.Pointer[authorizationCacheStores] clusterBindingResourceVersions sets.String clusterRoleResourceVersions sets.String @@ -199,7 +206,7 @@ type AuthorizationCache struct { reviewer Reviewer - syncHandler func(request *reviewRequest, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) error + syncHandler func(request *reviewRequest, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) error watchers []CacheWatcher watcherLock sync.Mutex @@ -255,10 +262,6 @@ func NewAuthorizationCache( roleBindingNamespacer: srbLister, roleLastSyncResourceVersioner: unionLastSyncResourceVersioner{scrLister, scrbLister, srLister, srbLister}, - reviewRecordStore: cache.NewStore(reviewRecordKeyFn), - userSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), - groupSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), - reviewer: reviewer, skip: &neverSkipSynchronizer{}, @@ -268,6 +271,11 @@ func NewAuthorizationCache( lastCacheInvalidation: realClock.Now(), maxCacheLifespan: defaultMaxCacheLifespan, } + ac.stores.Store(&authorizationCacheStores{ + reviewRecordStore: cache.NewStore(reviewRecordKeyFn), + userSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + groupSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + }) ac.lastSyncResourceVersioner = namespaceLastSyncResourceVersioner ac.syncHandler = ac.syncRequest return ac @@ -315,7 +323,7 @@ func (ac *AuthorizationCache) GetClusterRoleLister() SyncedClusterRoleLister { } // synchronizeNamespaces synchronizes access over each namespace and returns a set of namespace names that were looked at in last sync -func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) sets.String { +func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) sets.String { namespaceSet := sets.NewString() namespaces, err := ac.namespaceLister.List(labels.Everything()) if err != nil { @@ -329,7 +337,7 @@ func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache namespace: namespace.Name, namespaceResourceVersion: namespace.ResourceVersion, } - if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore); err != nil { + if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite); err != nil { utilruntime.HandleError(fmt.Errorf("error synchronizing: %v", err)) } } @@ -337,7 +345,7 @@ func (ac *AuthorizationCache) synchronizeNamespaces(userSubjectRecordStore cache } // synchronizePolicies synchronizes access over each role -func (ac *AuthorizationCache) synchronizePolicies(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) { +func (ac *AuthorizationCache) synchronizePolicies(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) { roleList, err := ac.roleNamespacer.Roles(metav1.NamespaceAll).List(labels.Everything()) if err != nil { utilruntime.HandleError(err) @@ -348,14 +356,14 @@ func (ac *AuthorizationCache) synchronizePolicies(userSubjectRecordStore cache.S namespace: role.Namespace, roleUIDToResourceVersion: map[types.UID]string{role.UID: role.ResourceVersion}, } - if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore); err != nil { + if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite); err != nil { utilruntime.HandleError(fmt.Errorf("error synchronizing: %v", err)) } } } // synchronizeRoleBindings synchronizes access over each role binding -func (ac *AuthorizationCache) synchronizeRoleBindings(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) { +func (ac *AuthorizationCache) synchronizeRoleBindings(userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) { roleBindingList, err := ac.roleBindingNamespacer.RoleBindings(metav1.NamespaceAll).List(labels.Everything()) if err != nil { utilruntime.HandleError(err) @@ -366,20 +374,20 @@ func (ac *AuthorizationCache) synchronizeRoleBindings(userSubjectRecordStore cac namespace: roleBinding.Namespace, roleBindingUIDToResourceVersion: map[types.UID]string{roleBinding.UID: roleBinding.ResourceVersion}, } - if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore); err != nil { + if err := ac.syncHandler(reviewRequest, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite); err != nil { utilruntime.HandleError(fmt.Errorf("error synchronizing: %v", err)) } } } // purgeDeletedNamespaces will remove all namespaces enumerated in a reviewRecordStore that are not in the namespace set -func (ac *AuthorizationCache) purgeDeletedNamespaces(oldNamespaces, newNamespaces sets.String, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) { +func (ac *AuthorizationCache) purgeDeletedNamespaces(oldNamespaces, newNamespaces sets.String, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) { reviewRecordItems := reviewRecordStore.List() for i := range reviewRecordItems { reviewRecord := reviewRecordItems[i].(*reviewRecord) if !newNamespaces.Has(reviewRecord.namespace) { - deleteNamespaceFromSubjects(userSubjectRecordStore, reviewRecord.users, reviewRecord.namespace) - deleteNamespaceFromSubjects(groupSubjectRecordStore, reviewRecord.groups, reviewRecord.namespace) + deleteNamespaceFromSubjects(userSubjectRecordStore, reviewRecord.users, reviewRecord.namespace, copyOnWrite) + deleteNamespaceFromSubjects(groupSubjectRecordStore, reviewRecord.groups, reviewRecord.namespace, copyOnWrite) reviewRecordStore.Delete(reviewRecord) } } @@ -424,7 +432,9 @@ func (ac *AuthorizationCache) invalidateCache(expired bool) bool { return invalidateCache } -// synchronize runs a a full synchronization over the cache data. it must be run in a single-writer model, it's not thread-safe by design. +// synchronize runs a full synchronization over the cache data. Only one +// goroutine may call synchronize at a time, but List() may be called +// concurrently from any number of goroutines. func (ac *AuthorizationCache) synchronize() { expired := ac.cacheHasExpired() // if none of our internal reflectors changed, then we can skip reviewing the cache @@ -434,9 +444,10 @@ func (ac *AuthorizationCache) synchronize() { } // by default, we update our current caches and do an incremental change - userSubjectRecordStore := ac.userSubjectRecordStore - groupSubjectRecordStore := ac.groupSubjectRecordStore - reviewRecordStore := ac.reviewRecordStore + currentStores := ac.stores.Load() + userSubjectRecordStore := currentStores.userSubjectRecordStore + groupSubjectRecordStore := currentStores.groupSubjectRecordStore + reviewRecordStore := currentStores.reviewRecordStore // if there was a global change that forced complete invalidation, we rebuild our cache and do a fast swap at end invalidateCache := ac.invalidateCache(expired) @@ -447,17 +458,25 @@ func (ac *AuthorizationCache) synchronize() { reviewRecordStore = cache.NewStore(reviewRecordKeyFn) } + // During full cache invalidation the stores are private to this + // goroutine, so in-place mutation is safe and avoids the O(n²) + // copy overhead of COW. During incremental updates the stores + // are shared with concurrent List() callers, so COW is required. + copyOnWrite := !invalidateCache + // iterate over caches and synchronize our three caches - newKnownNamespaces := ac.synchronizeNamespaces(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore) - ac.synchronizePolicies(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore) - ac.synchronizeRoleBindings(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore) - ac.purgeDeletedNamespaces(ac.allKnownNamespaces, newKnownNamespaces, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore) + newKnownNamespaces := ac.synchronizeNamespaces(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite) + ac.synchronizePolicies(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite) + ac.synchronizeRoleBindings(userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite) + ac.purgeDeletedNamespaces(ac.allKnownNamespaces, newKnownNamespaces, userSubjectRecordStore, groupSubjectRecordStore, reviewRecordStore, copyOnWrite) - // if we did a full rebuild, now we swap the fully rebuilt cache + // if we did a full rebuild, now we swap the fully rebuilt cache atomically if invalidateCache { - ac.userSubjectRecordStore = userSubjectRecordStore - ac.groupSubjectRecordStore = groupSubjectRecordStore - ac.reviewRecordStore = reviewRecordStore + ac.stores.Store(&authorizationCacheStores{ + userSubjectRecordStore: userSubjectRecordStore, + groupSubjectRecordStore: groupSubjectRecordStore, + reviewRecordStore: reviewRecordStore, + }) } ac.allKnownNamespaces = newKnownNamespaces @@ -465,8 +484,10 @@ func (ac *AuthorizationCache) synchronize() { ac.lastState = currentState } -// syncRequest takes a reviewRequest and determines if it should update the caches supplied, it is not thread-safe -func (ac *AuthorizationCache) syncRequest(request *reviewRequest, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store) error { +// syncRequest takes a reviewRequest and determines if it should update the +// caches supplied. It is only called from synchronize and shares its +// concurrency contract: single writer, concurrent readers via List(). +func (ac *AuthorizationCache) syncRequest(request *reviewRequest, userSubjectRecordStore cache.Store, groupSubjectRecordStore cache.Store, reviewRecordStore cache.Store, copyOnWrite bool) error { lastKnownValue, err := lastKnown(reviewRecordStore, request.namespace) if err != nil { @@ -492,10 +513,10 @@ func (ac *AuthorizationCache) syncRequest(request *reviewRequest, userSubjectRec groupsToRemove.Delete(review.Groups()...) } - deleteNamespaceFromSubjects(userSubjectRecordStore, usersToRemove.List(), namespace) - deleteNamespaceFromSubjects(groupSubjectRecordStore, groupsToRemove.List(), namespace) - addSubjectsToNamespace(userSubjectRecordStore, review.Users(), namespace) - addSubjectsToNamespace(groupSubjectRecordStore, review.Groups(), namespace) + deleteNamespaceFromSubjects(userSubjectRecordStore, usersToRemove.List(), namespace, copyOnWrite) + deleteNamespaceFromSubjects(groupSubjectRecordStore, groupsToRemove.List(), namespace, copyOnWrite) + addSubjectsToNamespace(userSubjectRecordStore, review.Users(), namespace, copyOnWrite) + addSubjectsToNamespace(groupSubjectRecordStore, review.Groups(), namespace, copyOnWrite) cacheReviewRecord(request, lastKnownValue, review, reviewRecordStore) ac.notifyWatchers(namespace, lastKnownValue, sets.NewString(review.Users()...), sets.NewString(review.Groups()...)) @@ -511,14 +532,17 @@ func (ac *AuthorizationCache) List(userInfo user.Info, selector labels.Selector) user := userInfo.GetName() groups := userInfo.GetGroups() - obj, exists, _ := ac.userSubjectRecordStore.GetByKey(user) + // snapshot the stores pointer once so we read from a consistent pair + stores := ac.stores.Load() + + obj, exists, _ := stores.userSubjectRecordStore.GetByKey(user) if exists { subjectRecord := obj.(*subjectRecord) keys.Insert(subjectRecord.namespaces.List()...) } for _, group := range groups { - obj, exists, _ := ac.groupSubjectRecordStore.GetByKey(group) + obj, exists, _ := stores.groupSubjectRecordStore.GetByKey(group) if exists { subjectRecord := obj.(*subjectRecord) keys.Insert(subjectRecord.namespaces.List()...) @@ -594,33 +618,57 @@ func skipReview(request *reviewRequest, lastKnownValue *reviewRecord) bool { return true } -// deleteNamespaceFromSubjects removes the namespace from each subject -// if no other namespaces are active to that subject, it will also delete the subject from the cache entirely -func deleteNamespaceFromSubjects(subjectRecordStore cache.Store, subjects []string, namespace string) { +// deleteNamespaceFromSubjects removes the namespace from each subject. +// When copyOnWrite is true, a new subjectRecord is created so concurrent +// readers iterating the old record's map are not disturbed. When false, +// the map is mutated in place (safe only when the store is not visible +// to readers, e.g. during a full cache rebuild). +func deleteNamespaceFromSubjects(subjectRecordStore cache.Store, subjects []string, namespace string, copyOnWrite bool) { for _, subject := range subjects { obj, exists, _ := subjectRecordStore.GetByKey(subject) if exists { - subjectRecord := obj.(*subjectRecord) - delete(subjectRecord.namespaces, namespace) - if len(subjectRecord.namespaces) == 0 { - subjectRecordStore.Delete(subjectRecord) + sr := obj.(*subjectRecord) + if copyOnWrite { + if !sr.namespaces.Has(namespace) { + continue + } + newNamespaces := sets.NewString(sr.namespaces.UnsortedList()...) + newNamespaces.Delete(namespace) + if len(newNamespaces) == 0 { + subjectRecordStore.Delete(sr) + } else { + subjectRecordStore.Update(&subjectRecord{subject: subject, namespaces: newNamespaces}) + } + } else { + delete(sr.namespaces, namespace) + if len(sr.namespaces) == 0 { + subjectRecordStore.Delete(sr) + } } } } } -// addSubjectsToNamespace adds the specified namespace to each subject -func addSubjectsToNamespace(subjectRecordStore cache.Store, subjects []string, namespace string) { +// addSubjectsToNamespace adds the specified namespace to each subject. +// See deleteNamespaceFromSubjects for the copyOnWrite semantics. +func addSubjectsToNamespace(subjectRecordStore cache.Store, subjects []string, namespace string, copyOnWrite bool) { for _, subject := range subjects { - var item *subjectRecord obj, exists, _ := subjectRecordStore.GetByKey(subject) if exists { - item = obj.(*subjectRecord) + sr := obj.(*subjectRecord) + if copyOnWrite { + if sr.namespaces.Has(namespace) { + continue + } + newNamespaces := sets.NewString(sr.namespaces.UnsortedList()...) + newNamespaces.Insert(namespace) + subjectRecordStore.Update(&subjectRecord{subject: subject, namespaces: newNamespaces}) + } else { + sr.namespaces.Insert(namespace) + } } else { - item = &subjectRecord{subject: subject, namespaces: sets.NewString()} - subjectRecordStore.Add(item) + subjectRecordStore.Add(&subjectRecord{subject: subject, namespaces: sets.NewString(namespace)}) } - item.namespaces.Insert(namespace) } } diff --git a/pkg/project/auth/cache_test.go b/pkg/project/auth/cache_test.go index d1de2d78ff..f25da00714 100644 --- a/pkg/project/auth/cache_test.go +++ b/pkg/project/auth/cache_test.go @@ -3,6 +3,7 @@ package auth import ( "fmt" "strconv" + "sync" "testing" "time" @@ -342,6 +343,188 @@ func TestAuthorizationCache_cacheHasExpired(t *testing.T) { } } +func BenchmarkFullCacheInvalidation(b *testing.B) { + for _, bc := range []struct { + namespaces int + users int + }{ + {namespaces: 10, users: 10}, + {namespaces: 100, users: 10}, + {namespaces: 100, users: 100}, + {namespaces: 1000, users: 100}, + {namespaces: 1000, users: 1000}, + } { + b.Run(fmt.Sprintf("N=%d_U=%d", bc.namespaces, bc.users), func(b *testing.B) { + // Build user names. + userNames := make([]string, bc.users) + for i := range userNames { + userNames[i] = fmt.Sprintf("user-%d", i) + } + + // Build namespaces and reviewer expectations: every user has access to every namespace. + nsIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + expectedResults := make(map[string]*mockReview, bc.namespaces) + for i := 0; i < bc.namespaces; i++ { + name := fmt.Sprintf("ns-%d", i) + nsIndexer.Add(&corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: name, ResourceVersion: "1"}, + }) + expectedResults[name] = &mockReview{users: userNames} + } + + reviewer := &mockReviewer{expectedResults: expectedResults} + nsLister := corev1listers.NewNamespaceLister(nsIndexer) + + crs := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + crbs := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + + // Use a clock that always reports the cache as expired so every + // call to synchronize triggers a full invalidation. + clk := &mockedClock{since: 2 * defaultMaxCacheLifespan} + + ac := &AuthorizationCache{ + allKnownNamespaces: sets.String{}, + namespaceLister: nsLister, + lastSyncResourceVersioner: &fakeVersioner{"v1"}, + clusterRoleLister: syncedClusterRoleLister{ + ClusterRoleLister: rbacv1listers.NewClusterRoleLister(crs), + }, + clusterRoleBindingLister: syncedClusterRoleBindingLister{ + ClusterRoleBindingLister: rbacv1listers.NewClusterRoleBindingLister(crbs), + }, + roleNamespacer: syncedRoleLister{ + RoleLister: rbacv1listers.NewRoleLister(cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})), + }, + roleBindingNamespacer: syncedRoleBindingLister{ + RoleBindingLister: rbacv1listers.NewRoleBindingLister(cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})), + }, + roleLastSyncResourceVersioner: &fakeVersioner{"v1"}, + clusterRoleResourceVersions: sets.NewString(), + clusterBindingResourceVersions: sets.NewString(), + reviewer: reviewer, + skip: &neverSkipSynchronizer{}, + clock: clk, + maxCacheLifespan: defaultMaxCacheLifespan, + } + ac.stores.Store(&authorizationCacheStores{ + reviewRecordStore: cache.NewStore(reviewRecordKeyFn), + userSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + groupSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + }) + ac.syncHandler = ac.syncRequest + + b.ResetTimer() + for i := 0; i < b.N; i++ { + ac.synchronize() + } + }) + } +} + +func BenchmarkIncrementalSyncDuplicateSubjects(b *testing.B) { + for _, bc := range []struct { + namespaces int + users int + dupsPerUser int + }{ + {namespaces: 100, users: 10, dupsPerUser: 1}, + {namespaces: 100, users: 10, dupsPerUser: 10}, + {namespaces: 1000, users: 100, dupsPerUser: 1}, + {namespaces: 1000, users: 100, dupsPerUser: 10}, + } { + b.Run(fmt.Sprintf("N=%d_U=%d_D=%d", bc.namespaces, bc.users, bc.dupsPerUser), func(b *testing.B) { + // Build user names with duplicates to simulate the broken kube dedup. + uniqueUsers := make([]string, bc.users) + for i := range uniqueUsers { + uniqueUsers[i] = fmt.Sprintf("user-%d", i) + } + dupUsers := make([]string, 0, bc.users*bc.dupsPerUser) + for _, u := range uniqueUsers { + for d := 0; d < bc.dupsPerUser; d++ { + dupUsers = append(dupUsers, u) + } + } + + nsIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + expectedResults := make(map[string]*mockReview, bc.namespaces) + for i := 0; i < bc.namespaces; i++ { + name := fmt.Sprintf("ns-%d", i) + nsIndexer.Add(&corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: name, ResourceVersion: "1"}, + }) + expectedResults[name] = &mockReview{users: dupUsers} + } + + reviewer := &mockReviewer{expectedResults: expectedResults} + nsLister := corev1listers.NewNamespaceLister(nsIndexer) + + crs := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + crbs := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + + // Use a clock that reports zero elapsed time so the cache never + // expires — every synchronize after the first hits the incremental + // COW path. + clk := &mockedClock{since: 0} + + ac := &AuthorizationCache{ + allKnownNamespaces: sets.String{}, + namespaceLister: nsLister, + lastSyncResourceVersioner: &fakeVersioner{"v1"}, + clusterRoleLister: syncedClusterRoleLister{ + ClusterRoleLister: rbacv1listers.NewClusterRoleLister(crs), + }, + clusterRoleBindingLister: syncedClusterRoleBindingLister{ + ClusterRoleBindingLister: rbacv1listers.NewClusterRoleBindingLister(crbs), + }, + roleNamespacer: syncedRoleLister{ + RoleLister: rbacv1listers.NewRoleLister(cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})), + }, + roleBindingNamespacer: syncedRoleBindingLister{ + RoleBindingLister: rbacv1listers.NewRoleBindingLister(cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})), + }, + roleLastSyncResourceVersioner: &fakeVersioner{"v1"}, + clusterRoleResourceVersions: sets.NewString(), + clusterBindingResourceVersions: sets.NewString(), + reviewer: reviewer, + skip: &neverSkipSynchronizer{}, + clock: clk, + maxCacheLifespan: defaultMaxCacheLifespan, + } + ac.stores.Store(&authorizationCacheStores{ + reviewRecordStore: cache.NewStore(reviewRecordKeyFn), + userSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + groupSubjectRecordStore: cache.NewStore(subjectRecordKeyFn), + }) + ac.syncHandler = ac.syncRequest + + // Seed the cache with a full rebuild (first call always invalidates). + ac.synchronize() + + // Bump resource versions so subsequent syncs process every namespace + // through the incremental COW path. + for i := 0; i < bc.namespaces; i++ { + name := fmt.Sprintf("ns-%d", i) + nsIndexer.Update(&corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: name, ResourceVersion: "2"}, + }) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + ac.synchronize() + } + }) + } +} + +type fakeVersioner struct { + version string +} + +func (f *fakeVersioner) LastSyncResourceVersion() string { + return f.version +} + func TestAuthorizationCache_invalidateCache(t *testing.T) { for _, tt := range []struct { name string @@ -436,3 +619,133 @@ func TestAuthorizationCache_invalidateCache(t *testing.T) { }) } } + +func TestAuthorizationCacheRace(t *testing.T) { + namespaceList := corev1.NamespaceList{ + Items: []corev1.Namespace{ + {ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1"}}, + {ObjectMeta: metav1.ObjectMeta{Name: "bar", ResourceVersion: "2"}}, + {ObjectMeta: metav1.ObjectMeta{Name: "baz", ResourceVersion: "3"}}, + }, + } + mockKubeClient := fake.NewSimpleClientset(&namespaceList) + + reviewer := &mockReviewer{ + expectedResults: map[string]*mockReview{ + "foo": { + users: []string{alice.GetName(), bob.GetName()}, + groups: eve.GetGroups(), + }, + "bar": { + users: []string{frank.GetName(), eve.GetName()}, + groups: []string{"random"}, + }, + "baz": { + users: []string{alice.GetName()}, + groups: []string{"employee"}, + }, + }, + } + + informers := informers.NewSharedInformerFactory(mockKubeClient, controller.NoResyncPeriodFunc()) + nsIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + nsLister := corev1listers.NewNamespaceLister(nsIndexer) + + authorizationCache := NewAuthorizationCache( + nsLister, + informers.Core().V1().Namespaces().Informer(), + reviewer, + informers.Rbac().V1(), + ) + for i := range namespaceList.Items { + nsIndexer.Add(&namespaceList.Items[i]) + } + + // seed the cache + authorizationCache.synchronize() + + stop := make(chan struct{}) + var wg sync.WaitGroup + + // writer goroutine: continuously synchronize, toggling access + wg.Add(1) + go func() { + defer wg.Done() + rv := 10 + for { + select { + case <-stop: + return + default: + } + // toggle access patterns to force map mutations + if rv%2 == 0 { + reviewer.expectedResults["foo"] = &mockReview{ + users: []string{alice.GetName(), bob.GetName()}, + groups: eve.GetGroups(), + } + reviewer.expectedResults["baz"] = &mockReview{ + users: []string{frank.GetName()}, + groups: []string{}, + } + } else { + reviewer.expectedResults["foo"] = &mockReview{ + users: []string{frank.GetName()}, + groups: []string{"random"}, + } + reviewer.expectedResults["baz"] = &mockReview{ + users: []string{alice.GetName(), eve.GetName()}, + groups: []string{"employee"}, + } + } + rv++ + for i := range namespaceList.Items { + ns := namespaceList.Items[i] + ns.ResourceVersion = strconv.Itoa(rv) + nsIndexer.Update(&ns) + } + authorizationCache.synchronize() + } + }() + + // reader goroutines: continuously call List + users := []user.Info{alice, bob, eve, frank} + for _, u := range users { + u := u + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + _, err := authorizationCache.List(u, labels.Everything()) + if err != nil { + t.Errorf("List(%s) returned error: %v", u.GetName(), err) + return + } + } + }() + } + + time.Sleep(2 * time.Second) + close(stop) + wg.Wait() +} + +func BenchmarkAddSubjectsToNamespace(b *testing.B) { + for _, namespaceCount := range []int{10, 100, 1000} { + b.Run(fmt.Sprintf("namespaces=%d", namespaceCount), func(b *testing.B) { + subjects := []string{"alice", "bob", "eve", "frank", "grace"} + b.ResetTimer() + for i := 0; i < b.N; i++ { + store := cache.NewStore(subjectRecordKeyFn) + for ns := 0; ns < namespaceCount; ns++ { + addSubjectsToNamespace(store, subjects, fmt.Sprintf("namespace-%d", ns), true) + } + } + }) + } +}