Merge pull request #143776 from jpbetz/rbac-bootstrap-list-reconcile

Skip up-to-date bootstrap RBAC objects using the initial Lists

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) {

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")
+	})
 }