chore: upgrade controller-runtime to v0.23 and implement client.Apply

cmd/milo/apiserver/config.go

@@ -15,7 +15,7 @@ import (
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/endpoints/filterlatency"
 	genericapifilters "k8s.io/apiserver/pkg/endpoints/filters"
-	impersonationfilters "k8s.io/apiserver/pkg/endpoints/filters/impersonation"
+	impersonationfilter "k8s.io/apiserver/pkg/endpoints/filters/impersonation"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	genericfeatures "k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/server"
@@ -26,7 +26,7 @@ import (
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
 	"k8s.io/component-base/tracing"
-	apiservercompat "k8s.io/apiserver/pkg/util/compatibility"
+	utilversionscompat "k8s.io/apiserver/pkg/util/compatibility"
 	"k8s.io/klog/v2"
 	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
 	aggregatorscheme "k8s.io/kube-aggregator/pkg/apiserver/scheme"
@@ -388,7 +388,7 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		return nil, err
 	}
 	c.ControlPlane = kubeAPIs
-	c.ControlPlane.Generic.EffectiveVersion = apiservercompat.DefaultBuildEffectiveVersion()
+	c.ControlPlane.Generic.EffectiveVersion = utilversionscompat.DefaultBuildEffectiveVersion()
 
 	if kubeAPIs.Generic.LoopbackClientConfig != nil && kubeAPIs.Generic.TracerProvider != nil {
 		kubeAPIs.Generic.LoopbackClientConfig.Wrap(tracing.WrapperFor(kubeAPIs.Generic.TracerProvider))
@@ -438,7 +438,7 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	}
 	c.Aggregator = aggregator
 	c.Aggregator.ExtraConfig.DisableRemoteAvailableConditionController = true
-	c.Aggregator.GenericConfig.EffectiveVersion = apiservercompat.DefaultBuildEffectiveVersion()
+	c.Aggregator.GenericConfig.EffectiveVersion = utilversionscompat.DefaultBuildEffectiveVersion()
 
 	return c, nil
 }
@@ -476,7 +476,7 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *server.Config, loopbac
 	failedHandler = genericapifilters.WithFailedAuthenticationAudit(failedHandler, c.AuditBackend, c.AuditPolicyRuleEvaluator)
 
 	handler = filterlatency.TrackCompleted(handler)
-	handler = impersonationfilters.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
+	handler = impersonationfilter.WithImpersonation(handler, c.Authorization.Authorizer, c.Serializer)
 	handler = filterlatency.TrackStarted(handler, c.TracerProvider, "impersonation")
 
 	failedHandler = filterlatency.TrackCompleted(failedHandler)

cmd/milo/apiserver/server.go

@@ -27,14 +27,14 @@ import (
 	"k8s.io/client-go/rest"
 	cliflag "k8s.io/component-base/cli/flag"
 	"k8s.io/component-base/cli/globalflag"
+	"k8s.io/component-base/compatibility"
 	"k8s.io/component-base/logs"
 	logsapi "k8s.io/component-base/logs/api/v1"
 	_ "k8s.io/component-base/metrics/prometheus/workqueue"
 	"k8s.io/component-base/term"
 	"k8s.io/component-base/version"
 	"k8s.io/component-base/version/verflag"
-	basecompatibility "k8s.io/component-base/compatibility"
-	apiservercompat "k8s.io/apiserver/pkg/util/compatibility"
+	utilversionscompat "k8s.io/apiserver/pkg/util/compatibility"
 	"k8s.io/klog/v2"
 	aggregatorapiserver "k8s.io/kube-aggregator/pkg/apiserver"
 
@@ -132,9 +132,9 @@ func NewCommand() *cobra.Command {
 		},
 	}
 
-	s.GenericServerRunOptions.ComponentGlobalsRegistry = basecompatibility.NewComponentGlobalsRegistry()
+	s.GenericServerRunOptions.ComponentGlobalsRegistry = compatibility.NewComponentGlobalsRegistry()
 	s.GenericServerRunOptions.ComponentGlobalsRegistry.ComponentGlobalsOrRegister(
-		basecompatibility.DefaultKubeComponent, apiservercompat.DefaultBuildEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
+		compatibility.DefaultKubeComponent, utilversionscompat.DefaultBuildEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
 	s.GenericServerRunOptions.AddUniversalFlags(namedFlagSets.FlagSet("generic"))
 	s.Etcd.AddFlags(namedFlagSets.FlagSet("etcd"))
 	s.SecureServing.AddFlags(namedFlagSets.FlagSet("secure serving"))

cmd/milo/controller-manager/controllermanager.go

@@ -43,6 +43,7 @@ import (
 	certutil "k8s.io/client-go/util/cert"
 	cliflag "k8s.io/component-base/cli/flag"
 	"k8s.io/component-base/cli/globalflag"
+	"k8s.io/component-base/compatibility"
 	"k8s.io/component-base/configz"
 	"k8s.io/component-base/featuregate"
 	"k8s.io/component-base/logs"
@@ -53,8 +54,7 @@ import (
 	"k8s.io/component-base/term"
 	"k8s.io/component-base/version"
 	"k8s.io/component-base/version/verflag"
-	basecompatibility "k8s.io/component-base/compatibility"
-	apiservercompat "k8s.io/apiserver/pkg/util/compatibility"
+	utilversionscompat "k8s.io/apiserver/pkg/util/compatibility"
 	genericcontrollermanager "k8s.io/controller-manager/app"
 	"k8s.io/controller-manager/controller"
 	"k8s.io/controller-manager/pkg/clientbuilder"
@@ -206,8 +206,8 @@ const (
 
 // NewCommand creates a *cobra.Command object with default parameters
 func NewCommand() *cobra.Command {
-	_, _ = apiservercompat.DefaultComponentGlobalsRegistry.ComponentGlobalsOrRegister(
-		basecompatibility.DefaultKubeComponent, apiservercompat.DefaultBuildEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
+	_, _ = utilversionscompat.DefaultComponentGlobalsRegistry.ComponentGlobalsOrRegister(
+		compatibility.DefaultKubeComponent, utilversionscompat.DefaultBuildEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
 
 	s, err := NewOptions()
 	if err != nil {
@@ -247,13 +247,13 @@ func NewCommand() *cobra.Command {
 				ProjectOwnerRoleNamespace = SystemNamespace
 			}
 
-			c, err := s.Config(cmd.Context(), KnownControllers(), nil, ControllerAliases())
+			c, err := s.Config(context.Background(), KnownControllers(), nil, ControllerAliases())
 			if err != nil {
 				return err
 			}
 
 			// add feature enablement metrics
-			fg := s.ComponentGlobalsRegistry.FeatureGateFor(basecompatibility.DefaultKubeComponent)
+			fg := s.ComponentGlobalsRegistry.FeatureGateFor(compatibility.DefaultKubeComponent)
 			fg.(featuregate.MutableFeatureGate).AddMetrics()
 			return Run(context.Background(), c.Complete(), s)
 		},
@@ -677,8 +677,15 @@ func Run(ctx context.Context, c *config.CompletedConfig, opts *Options) error {
 				klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 			}
 
-			// The multicluster manager automatically starts the provider because
-			// Provider implements multicluster.ProviderRunnable (Start method).
+			// Start concurrently to resolve circular dependency between provider and manager
+			go func() {
+				logger.Info("Starting Datum cluster provider")
+				if err := provider.Run(ctx, mcMgr); err != nil {
+					logger.Error(err, "Datum cluster provider failed")
+					panic(err)
+				}
+			}()
+
 			go func() {
 				logger.Info("Starting multicluster manager for quota system")
 				if err := mcMgr.Start(ctx); err != nil {
@@ -842,11 +849,11 @@ func Run(ctx context.Context, c *config.CompletedConfig, opts *Options) error {
 	}
 
 	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CoordinatedLeaderElection) {
-		binaryVersion, err := semver.ParseTolerant(apiservercompat.DefaultComponentGlobalsRegistry.EffectiveVersionFor(basecompatibility.DefaultKubeComponent).BinaryVersion().String())
+		binaryVersion, err := semver.ParseTolerant(utilversionscompat.DefaultComponentGlobalsRegistry.EffectiveVersionFor(compatibility.DefaultKubeComponent).BinaryVersion().String())
 		if err != nil {
 			return err
 		}
-		emulationVersion, err := semver.ParseTolerant(apiservercompat.DefaultComponentGlobalsRegistry.EffectiveVersionFor(basecompatibility.DefaultKubeComponent).EmulationVersion().String())
+		emulationVersion, err := semver.ParseTolerant(utilversionscompat.DefaultComponentGlobalsRegistry.EffectiveVersionFor(compatibility.DefaultKubeComponent).EmulationVersion().String())
 		if err != nil {
 			return err
 		}

internal/apiserver/storage/project/mux.go

@@ -173,13 +173,7 @@ func (i *instrumentedStorage) GuaranteedUpdate(ctx context.Context, key string,
 	i.markSuccess()
 	return nil
 }
-func (i *instrumentedStorage) ReadinessCheck() error { return i.inner.ReadinessCheck() }
-func (i *instrumentedStorage) RequestWatchProgress(ctx context.Context) error {
-	if err := i.inner.RequestWatchProgress(ctx); err != nil {
-		return i.markReinit("watch_progress", err)
-	}
-	return nil
-}
+func (i *instrumentedStorage) CompactRevision() int64 { return i.inner.CompactRevision() }
 func (i *instrumentedStorage) Stats(ctx context.Context) (storage.Stats, error) {
 	return i.inner.Stats(ctx)
 }
@@ -189,7 +183,13 @@ func (i *instrumentedStorage) GetCurrentResourceVersion(ctx context.Context) (ui
 func (i *instrumentedStorage) EnableResourceSizeEstimation(fn storage.KeysFunc) error {
 	return i.inner.EnableResourceSizeEstimation(fn)
 }
-func (i *instrumentedStorage) CompactRevision() int64 { return i.inner.CompactRevision() }
+func (i *instrumentedStorage) ReadinessCheck() error { return i.inner.ReadinessCheck() }
+func (i *instrumentedStorage) RequestWatchProgress(ctx context.Context) error {
+	if err := i.inner.RequestWatchProgress(ctx); err != nil {
+		return i.markReinit("watch_progress", err)
+	}
+	return nil
+}
 
 // -------------------- mux --------------------
 
@@ -374,6 +374,22 @@ func (m *projectMux) GuaranteedUpdate(ctx context.Context, key string, out runti
 	return s.GuaranteedUpdate(ctx, key, out, ignoreNotFound, precond, tryUpdate, suggestion)
 }
 
+// CompactRevision proxies to the appropriate child (defaults to the "" project).
+func (m *projectMux) CompactRevision() int64 {
+	m.mu.RLock()
+	c := m.children[""]
+	m.mu.RUnlock()
+	if c == nil {
+		if _, err := m.childForProject(""); err != nil {
+			return 0
+		}
+		m.mu.RLock()
+		c = m.children[""]
+		m.mu.RUnlock()
+	}
+	return c.s.CompactRevision()
+}
+
 // ReadinessCheck proxies to the appropriate child (defaults to the "" project).
 func (m *projectMux) ReadinessCheck() error {
 	m.mu.RLock()
@@ -424,13 +440,3 @@ func (m *projectMux) EnableResourceSizeEstimation(fn storage.KeysFunc) error {
 	}
 	return nil
 }
-
-func (m *projectMux) CompactRevision() int64 {
-	m.mu.RLock()
-	c := m.children[""]
-	m.mu.RUnlock()
-	if c == nil {
-		return 0
-	}
-	return c.s.CompactRevision()
-}

internal/quota/controllers/policy/reconciler_test.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -21,7 +22,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
 	"github.com/go-logr/logr"
-	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 	mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager"
 	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
@@ -50,8 +50,8 @@ func (c *testCluster) GetConfig() *rest.Config                        { return n
 func (c *testCluster) GetCache() cache.Cache                          { return nil }
 func (c *testCluster) GetFieldIndexer() client.FieldIndexer           { return nil }
 func (c *testCluster) GetEventRecorderFor(string) record.EventRecorder { return nil }
-func (c *testCluster) GetEventRecorder(string) events.EventRecorder   { return nil }
-func (c *testCluster) GetRESTMapper() meta.RESTMapper                 { return nil }
+func (c *testCluster) GetEventRecorder(string) events.EventRecorder    { return nil }
+func (c *testCluster) GetRESTMapper() meta.RESTMapper                  { return nil }
 func (c *testCluster) GetAPIReader() client.Reader                    { return nil }
 func (c *testCluster) Start(context.Context) error                    { return nil }
 

internal/webhooks/iam/v1alpha1/platformaccessapproval_webhook.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -52,10 +53,10 @@ func SetupPlatformAccessApprovalWebhooksWithManager(mgr ctrl.Manager) error {
 	}
 
 	return ctrl.NewWebhookManagedBy(mgr, &iamv1alpha1.PlatformAccessApproval{}).
-		WithDefaulter(&PlatformAccessApprovalMutator{
+		WithCustomDefaulter(&PlatformAccessApprovalMutator{
 			client: mgr.GetClient(),
 		}).
-		WithValidator(&PlatformAccessApprovalValidator{
+		WithCustomValidator(&PlatformAccessApprovalValidator{
 			client: mgr.GetClient(),
 		}).
 		Complete()
@@ -68,7 +69,11 @@ type PlatformAccessApprovalMutator struct {
 	client client.Client
 }
 
-func (m *PlatformAccessApprovalMutator) Default(ctx context.Context, paa *iamv1alpha1.PlatformAccessApproval) error {
+func (m *PlatformAccessApprovalMutator) Default(ctx context.Context, obj runtime.Object) error {
+	paa, ok := obj.(*iamv1alpha1.PlatformAccessApproval)
+	if !ok {
+		return errors.NewInternalError(fmt.Errorf("failed to cast object to PlatformAccessApproval"))
+	}
 	log := logf.FromContext(ctx).WithValues("Defaulting PlatformAccessApproval", "name", paa.GetName())
 
 	// Approver is the user who is approving the access request.
@@ -111,7 +116,8 @@ type PlatformAccessApprovalValidator struct {
 	client client.Client
 }
 
-func (v *PlatformAccessApprovalValidator) ValidateCreate(ctx context.Context, paa *iamv1alpha1.PlatformAccessApproval) (admission.Warnings, error) {
+func (v *PlatformAccessApprovalValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
+	paa := obj.(*iamv1alpha1.PlatformAccessApproval)
 	log := logf.FromContext(ctx).WithValues("Validating PlatformAccessApproval", "name", paa.GetName())
 
 	var errs field.ErrorList
@@ -186,10 +192,10 @@ func (v *PlatformAccessApprovalValidator) ValidateCreate(ctx context.Context, pa
 	return nil, nil
 }
 
-func (v *PlatformAccessApprovalValidator) ValidateUpdate(ctx context.Context, oldObj, newObj *iamv1alpha1.PlatformAccessApproval) (admission.Warnings, error) {
+func (v *PlatformAccessApprovalValidator) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
 	return nil, nil
 }
 
-func (v *PlatformAccessApprovalValidator) ValidateDelete(ctx context.Context, obj *iamv1alpha1.PlatformAccessApproval) (admission.Warnings, error) {
+func (v *PlatformAccessApprovalValidator) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
 	return nil, nil
 }

internal/webhooks/iam/v1alpha1/platformaccessarejection_webhook.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -18,10 +19,10 @@ const platformAccessRejectionIndexKey = "iam.miloapis.com/platformaccessrejectio
 
 func SetupPlatformAccessRejectionWebhooksWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewWebhookManagedBy(mgr, &iamv1alpha1.PlatformAccessRejection{}).
-		WithDefaulter(&PlatformAccessRejectionMutator{
+		WithCustomDefaulter(&PlatformAccessRejectionMutator{
 			client: mgr.GetClient(),
 		}).
-		WithValidator(&PlatformAccessRejectionValidator{
+		WithCustomValidator(&PlatformAccessRejectionValidator{
 			client: mgr.GetClient(),
 		}).
 		Complete()
@@ -34,7 +35,11 @@ type PlatformAccessRejectionMutator struct {
 	client client.Client
 }
 
-func (m *PlatformAccessRejectionMutator) Default(ctx context.Context, par *iamv1alpha1.PlatformAccessRejection) error {
+func (m *PlatformAccessRejectionMutator) Default(ctx context.Context, obj runtime.Object) error {
+	par, ok := obj.(*iamv1alpha1.PlatformAccessRejection)
+	if !ok {
+		return errors.NewInternalError(fmt.Errorf("failed to cast object to PlatformAccessRejection"))
+	}
 	log := logf.FromContext(ctx).WithValues("Defaulting PlatformAccessRejection", "name", par.GetName())
 
 	// Rejecter is the user who is rejecting the access request.
@@ -67,7 +72,8 @@ type PlatformAccessRejectionValidator struct {
 	client client.Client
 }
 
-func (v *PlatformAccessRejectionValidator) ValidateCreate(ctx context.Context, par *iamv1alpha1.PlatformAccessRejection) (admission.Warnings, error) {
+func (v *PlatformAccessRejectionValidator) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
+	par := obj.(*iamv1alpha1.PlatformAccessRejection)
 	log := logf.FromContext(ctx).WithValues("Validating PlatformAccessRejection", "name", par.GetName())
 
 	var errs field.ErrorList
@@ -114,10 +120,10 @@ func (v *PlatformAccessRejectionValidator) ValidateCreate(ctx context.Context, p
 	return nil, nil
 }
 
-func (v *PlatformAccessRejectionValidator) ValidateUpdate(ctx context.Context, oldObj, newObj *iamv1alpha1.PlatformAccessRejection) (admission.Warnings, error) {
+func (v *PlatformAccessRejectionValidator) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
 	return nil, nil
 }
 
-func (v *PlatformAccessRejectionValidator) ValidateDelete(ctx context.Context, obj *iamv1alpha1.PlatformAccessRejection) (admission.Warnings, error) {
+func (v *PlatformAccessRejectionValidator) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
 	return nil, nil
 }