From 4ffa763ad2dfa03920c0dc0cc64c6e3133194bc3 Mon Sep 17 00:00:00 2001 From: Joe Betz Date: Tue, 3 Nov 2026 18:01:30 +0000 Subject: [PATCH] Skip up-to-date bootstrap RBAC objects using the initial Lists EnsureRBACPolicy already lists all ClusterRoles and ClusterRoleBindings before reconciling bootstrap RBAC policy, but discarded both lists and then issued 132 sequential Get calls (ResourceVersion="") for the bootstrap ClusterRoles and ClusterRoleBindings, including the priming checks. Use the lists only to skip work. A dry run (Confirm: false) against each object's listed copy skips the object when nothing would change, and priming skips entries the lists show need no migration. Everything else is reconciled exactly as before, starting from a live Get, so a stale list result can delay a repair until the next start but never drives a write. The 14 kube-system and kube-public Roles and RoleBindings are still read with per-object Gets. --- pkg/registry/rbac/rest/storage_rbac.go | 78 +++++++- pkg/registry/rbac/rest/storage_rbac_test.go | 192 +++++++++++++++++++- 2 files changed, 261 insertions(+), 9 deletions(-) diff --git a/pkg/registry/rbac/rest/storage_rbac.go b/pkg/registry/rbac/rest/storage_rbac.go index d4d7a45f860..a5b9f1c00f9 100644 --- a/pkg/registry/rbac/rest/storage_rbac.go +++ b/pkg/registry/rbac/rest/storage_rbac.go @@ -18,6 +18,7 @@ package rest import ( "context" + "errors" "fmt" "time" @@ -27,6 +28,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apiserver/pkg/authorization/authorizer" @@ -180,26 +182,64 @@ func (p *PolicyData) EnsureRBACPolicy() genericapiserver.PostStartHookFunc { } } +// listedObjects is a client optimized for dry-run reconciles. It serves Gets from +// previously listed ClusterRoles or ClusterRoleBindings. +type listedObjects[T any] map[string]T + +func (l listedObjects[T]) Get(_, name string) (T, error) { + obj, ok := l[name] + if !ok { + return obj, apierrors.NewNotFound(schema.GroupResource{}, name) // The reconciler only checks IsNotFound, so no GroupResource is needed. + } + return obj, nil +} + +func (listedObjects[T]) Create(obj T) (T, error) { + return obj, errors.New("dry-run reconcile attempted a write") +} + +func (listedObjects[T]) Update(obj T) (T, error) { + return obj, errors.New("dry-run reconcile attempted a write") +} + +func (listedObjects[T]) Delete(string, string, types.UID) error { + return errors.New("dry-run reconcile attempted a write") +} + func ensureRBACPolicy(p *PolicyData, client clientset.Interface) (done bool, err error) { failedReconciliation := false // Make sure etcd is responding before we start reconciling - if _, err := client.RbacV1().ClusterRoles().List(context.TODO(), metav1.ListOptions{}); err != nil { + clusterRoleList, err := client.RbacV1().ClusterRoles().List(context.TODO(), metav1.ListOptions{}) + if err != nil { utilruntime.HandleError(fmt.Errorf("unable to initialize clusterroles: %v", err)) return false, nil } - if _, err := client.RbacV1().ClusterRoleBindings().List(context.TODO(), metav1.ListOptions{}); err != nil { + clusterRoleBindingList, err := client.RbacV1().ClusterRoleBindings().List(context.TODO(), metav1.ListOptions{}) + if err != nil { utilruntime.HandleError(fmt.Errorf("unable to initialize clusterrolebindings: %v", err)) return false, nil } + // Keep list responses and use them to skip objects that need no write. + // To ensure data consistency, this list data must never be used to + // prepare a write. + listedRoles := listedObjects[reconciliation.RuleOwner]{} + for _, role := range clusterRoleList.Items { + listedRoles[role.Name] = reconciliation.ClusterRoleRuleOwner{ClusterRole: &role} + } + listedBindings := listedObjects[reconciliation.RoleBinding]{} + for _, binding := range clusterRoleBindingList.Items { + listedBindings[binding.Name] = reconciliation.ClusterRoleBindingAdapter{ClusterRoleBinding: &binding} + } + // if the new cluster roles to aggregate do not yet exist, then we need to copy the old roles if they don't exist // in new locations - if err := primeAggregatedClusterRoles(p.ClusterRolesToAggregate, client.RbacV1()); err != nil { + if err := primeAggregatedClusterRoles(p.ClusterRolesToAggregate, listedRoles, client.RbacV1()); err != nil { utilruntime.HandleError(fmt.Errorf("unable to prime aggregated clusterroles: %v", err)) return false, nil } - if err := primeSplitClusterRoleBindings(p.ClusterRoleBindingsToSplit, client.RbacV1()); err != nil { + if err := primeSplitClusterRoleBindings(p.ClusterRoleBindingsToSplit, listedBindings, client.RbacV1()); err != nil { utilruntime.HandleError(fmt.Errorf("unable to prime split ClusterRoleBindings: %v", err)) return false, nil } @@ -211,6 +251,14 @@ func ensureRBACPolicy(p *PolicyData, client clientset.Interface) (done bool, err Client: reconciliation.ClusterRoleModifier{Client: client.RbacV1().ClusterRoles()}, Confirm: true, } + // Skip the role if a dry run against the List finds nothing to change. + dryRun := opts + dryRun.Client = listedRoles + dryRun.Confirm = false + if result, err := dryRun.Run(); err == nil && result.Operation == reconciliation.ReconcileNone { + continue + } + // ServiceUnavailble error is returned when the API server is blocked by storage version updates err := retryOnConflictOrServiceUnavailable(retry.DefaultBackoff, func() error { result, err := opts.Run() @@ -241,6 +289,14 @@ func ensureRBACPolicy(p *PolicyData, client clientset.Interface) (done bool, err Client: reconciliation.ClusterRoleBindingClientAdapter{Client: client.RbacV1().ClusterRoleBindings()}, Confirm: true, } + // Skip the binding if a dry run against the List finds nothing to change. + dryRun := opts + dryRun.Client = listedBindings + dryRun.Confirm = false + if result, err := dryRun.Run(); err == nil && result.Operation == reconciliation.ReconcileNone { + continue + } + // ServiceUnavailble error is returned when the API server is blocked by storage version updates err := retryOnConflictOrServiceUnavailable(retry.DefaultBackoff, func() error { result, err := opts.Run() @@ -345,8 +401,13 @@ func (p RESTStorageProvider) GroupName() string { // primeAggregatedClusterRoles copies roles that have transitioned to aggregated roles and may need to pick up changes // that were done to the legacy roles. -func primeAggregatedClusterRoles(clusterRolesToAggregate map[string]string, clusterRoleClient rbacv1client.ClusterRolesGetter) error { +func primeAggregatedClusterRoles(clusterRolesToAggregate map[string]string, listedRoles map[string]reconciliation.RuleOwner, clusterRoleClient rbacv1client.ClusterRolesGetter) error { for oldName, newName := range clusterRolesToAggregate { + // Skip the live Gets if the List shows nothing to migrate. + if listedRoles[newName] != nil || listedRoles[oldName] == nil { + continue + } + _, err := clusterRoleClient.ClusterRoles().Get(context.TODO(), newName, metav1.GetOptions{}) if err == nil { continue @@ -380,8 +441,13 @@ func primeAggregatedClusterRoles(clusterRolesToAggregate map[string]string, clus // primeSplitClusterRoleBindings ensures the existence of target ClusterRoleBindings // by copying Subjects, Annotations, and Labels from the specified source // ClusterRoleBinding, if present. -func primeSplitClusterRoleBindings(clusterRoleBindingToSplit map[string]rbacapiv1.ClusterRoleBinding, clusterRoleBindingClient rbacv1client.ClusterRoleBindingsGetter) error { +func primeSplitClusterRoleBindings(clusterRoleBindingToSplit map[string]rbacapiv1.ClusterRoleBinding, listedBindings map[string]reconciliation.RoleBinding, clusterRoleBindingClient rbacv1client.ClusterRoleBindingsGetter) error { for existingBindingName, clusterRoleBindingToCreate := range clusterRoleBindingToSplit { + // Skip the live Gets if the List shows nothing to copy. + if listedBindings[existingBindingName] == nil || listedBindings[clusterRoleBindingToCreate.Name] != nil { + continue + } + // If source ClusterRoleBinding does not exist, do nothing. existingRoleBinding, err := clusterRoleBindingClient.ClusterRoleBindings().Get(context.TODO(), existingBindingName, metav1.GetOptions{}) if apierrors.IsNotFound(err) { diff --git a/pkg/registry/rbac/rest/storage_rbac_test.go b/pkg/registry/rbac/rest/storage_rbac_test.go index 11f8ce61752..eaf7157f1bb 100644 --- a/pkg/registry/rbac/rest/storage_rbac_test.go +++ b/pkg/registry/rbac/rest/storage_rbac_test.go @@ -19,13 +19,174 @@ package rest import ( "testing" + "github.com/google/go-cmp/cmp" + + rbacapiv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" + clienttesting "k8s.io/client-go/testing" "k8s.io/kubernetes/plugin/pkg/auth/authorizer/rbac/bootstrappolicy" ) +func TestEnsureRBACPolicy(t *testing.T) { + tests := []struct { + name string + // policy defaults to the cluster-scoped bootstrap policy. + policy *PolicyData + // change edits the cluster before ensureRBACPolicy runs again. + change func(*testing.T, clienttesting.ObjectTracker) + // changeAfterList edits the cluster after ensureRBACPolicy lists it. + changeAfterList func(*testing.T, clienttesting.ObjectTracker) + // wantRequests are the requests ensureRBACPolicy sends, in order. + wantRequests []string + }{ + { + // Without drift, no bootstrap object needs its own Get. + name: "already bootstrapped", + wantRequests: []string{"list clusterroles", "list clusterrolebindings"}, + }, + { + name: "drifted objects are read live and repaired unless protected", + change: func(t *testing.T, tracker clienttesting.ObjectTracker) { + updateObject(t, tracker, "clusterroles", "cluster-admin", func(role *rbacapiv1.ClusterRole) { + role.Rules = nil + }) + updateObject(t, tracker, "clusterroles", "system:monitoring", func(role *rbacapiv1.ClusterRole) { + role.Annotations[rbacapiv1.AutoUpdateAnnotationKey] = "false" + role.Rules = nil + }) + updateObject(t, tracker, "clusterrolebindings", "system:basic-user", func(binding *rbacapiv1.ClusterRoleBinding) { + binding.RoleRef.Name = "custom-role" + }) + }, + wantRequests: []string{ + "list clusterroles", "list clusterrolebindings", + "get clusterroles/cluster-admin", "update clusterroles/cluster-admin", + "get clusterroles/system:monitoring", // protected, so not updated + "get clusterrolebindings/system:basic-user", "delete clusterrolebindings/system:basic-user", "create clusterrolebindings/system:basic-user", + }, + }, + { + name: "objects protected after the List are not written", + change: func(t *testing.T, tracker clienttesting.ObjectTracker) { + updateObject(t, tracker, "clusterroles", "cluster-admin", func(role *rbacapiv1.ClusterRole) { + role.Rules = nil + }) + updateObject(t, tracker, "clusterrolebindings", "system:basic-user", func(binding *rbacapiv1.ClusterRoleBinding) { + binding.RoleRef.Name = "custom-role" + }) + }, + changeAfterList: func(t *testing.T, tracker clienttesting.ObjectTracker) { + updateObject(t, tracker, "clusterroles", "cluster-admin", func(role *rbacapiv1.ClusterRole) { + role.Annotations[rbacapiv1.AutoUpdateAnnotationKey] = "false" + }) + updateObject(t, tracker, "clusterrolebindings", "system:basic-user", func(binding *rbacapiv1.ClusterRoleBinding) { + binding.Annotations[rbacapiv1.AutoUpdateAnnotationKey] = "false" + }) + }, + wantRequests: []string{ + "list clusterroles", "list clusterrolebindings", + "get clusterroles/cluster-admin", + "get clusterrolebindings/system:basic-user", + }, + }, + { + name: "legacy roles and bindings are copied", + policy: &PolicyData{ + ClusterRolesToAggregate: bootstrappolicy.ClusterRolesToAggregate(), + ClusterRoleBindingsToSplit: bootstrappolicy.ClusterRoleBindingsToSplit(), + }, + change: func(t *testing.T, tracker clienttesting.ObjectTracker) { + for _, obj := range []runtime.Object{ + &rbacapiv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: "admin"}}, + &rbacapiv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: "system:discovery"}}, + } { + if err := tracker.Add(obj); err != nil { + t.Fatal(err) + } + } + }, + wantRequests: []string{ + "list clusterroles", "list clusterrolebindings", + "get clusterroles/system:aggregate-to-admin", "get clusterroles/admin", "create clusterroles/system:aggregate-to-admin", + "get clusterrolebindings/system:discovery", "get clusterrolebindings/system:public-info-viewer", "create clusterrolebindings/system:public-info-viewer", + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + policy := tc.policy + if policy == nil { + // Leave out the namespaced Roles and RoleBindings. Every run reads + // each of them with its own Get, so they would add the same + // requests to every case's wantRequests. + policy = &PolicyData{ + ClusterRoles: append(bootstrappolicy.ClusterRoles(), bootstrappolicy.ControllerRoles()...), + ClusterRoleBindings: append(bootstrappolicy.ClusterRoleBindings(), bootstrappolicy.ControllerRoleBindings()...), + ClusterRolesToAggregate: bootstrappolicy.ClusterRolesToAggregate(), + ClusterRoleBindingsToSplit: bootstrappolicy.ClusterRoleBindingsToSplit(), + } + } + client := fake.NewSimpleClientset() + if done, err := ensureRBACPolicy(policy, client); err != nil || !done { + t.Fatalf("bootstrap ensureRBACPolicy() = (%v, %v), want (true, nil)", done, err) + } + if tc.change != nil { + tc.change(t, client.Tracker()) + } + if tc.changeAfterList != nil { + // ensureRBACPolicy lists ClusterRoles, then ClusterRoleBindings, then + // reconciles. Making the change right after the second List leaves + // both List results out of date before any object is reconciled. + // The change goes through the tracker because calling the fake + // client from inside a reactor deadlocks. + client.PrependReactor("list", "clusterrolebindings", func(action clienttesting.Action) (bool, runtime.Object, error) { + handled, list, err := clienttesting.ObjectReaction(client.Tracker())(action) + tc.changeAfterList(t, client.Tracker()) + return handled, list, err + }) + } + + client.ClearActions() + if done, err := ensureRBACPolicy(policy, client); err != nil || !done { + t.Fatalf("ensureRBACPolicy() = (%v, %v), want (true, nil)", done, err) + } + var requests []string + for _, action := range client.Actions() { + request := action.GetVerb() + " " + action.GetResource().Resource + switch action := action.(type) { + case clienttesting.GetAction: // also matches Deletes + request += "/" + action.GetName() + case clienttesting.CreateAction: // also matches Updates + request += "/" + action.GetObject().(metav1.Object).GetName() + } + requests = append(requests, request) + } + if diff := cmp.Diff(tc.wantRequests, requests); diff != "" { + t.Errorf("requests (-want +got):\n%s", diff) + } + }) + } +} + +// updateObject edits the named ClusterRole or ClusterRoleBinding in tracker. +func updateObject[T runtime.Object](t *testing.T, tracker clienttesting.ObjectTracker, resource, name string, change func(T)) { + t.Helper() + gvr := rbacapiv1.SchemeGroupVersion.WithResource(resource) + obj, err := tracker.Get(gvr, "", name) + if err != nil { + t.Fatal(err) + } + change(obj.(T)) + if err := tracker.Update(gvr, obj, ""); err != nil { + t.Fatal(err) + } +} + func BenchmarkEnsureRBACPolicy(b *testing.B) { - for n := 0; n < b.N; n++ { - var policy = &PolicyData{ + newPolicy := func() *PolicyData { + return &PolicyData{ ClusterRoles: append(bootstrappolicy.ClusterRoles(), bootstrappolicy.ControllerRoles()...), ClusterRoleBindings: append(bootstrappolicy.ClusterRoleBindings(), bootstrappolicy.ControllerRoleBindings()...), Roles: bootstrappolicy.NamespaceRoles(), @@ -33,7 +194,32 @@ func BenchmarkEnsureRBACPolicy(b *testing.B) { ClusterRolesToAggregate: bootstrappolicy.ClusterRolesToAggregate(), ClusterRoleBindingsToSplit: bootstrappolicy.ClusterRoleBindingsToSplit(), } + } + b.Run("empty-cluster", func(b *testing.B) { + b.ReportAllocs() + requests := 0 + for n := 0; n < b.N; n++ { + b.StopTimer() + policy := newPolicy() + coreClientSet := fake.NewSimpleClientset() + b.StartTimer() + _, _ = ensureRBACPolicy(policy, coreClientSet) + requests += len(coreClientSet.Actions()) + } + b.ReportMetric(float64(requests)/float64(b.N), "requests/op") + }) + b.Run("already-bootstrapped", func(b *testing.B) { + policy := newPolicy() coreClientSet := fake.NewSimpleClientset() _, _ = ensureRBACPolicy(policy, coreClientSet) - } + b.ReportAllocs() + b.ResetTimer() + requests := 0 + for n := 0; n < b.N; n++ { + coreClientSet.ClearActions() + _, _ = ensureRBACPolicy(policy, coreClientSet) + requests += len(coreClientSet.Actions()) + } + b.ReportMetric(float64(requests)/float64(b.N), "requests/op") + }) }