Replace StorageVersion API with aggregated discovery to fetch served resources by a peer for MVP

Co-authored-by: Joe Betz <[email protected]>

Co-authored-by: Jordan Liggitt <[email protected]>

Author: richabanker

pkg/controller/storageversiongc/gc_controller.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/kubernetes/pkg/controlplane"
+	"k8s.io/kubernetes/pkg/controlplane/apiserver"
 
 	"k8s.io/klog/v2"
 )
@@ -214,7 +215,7 @@ func (c *Controller) syncStorageVersion(ctx context.Context, name string) error
 	for _, v := range sv.Status.StorageVersions {
 		lease, err := c.kubeclientset.CoordinationV1().Leases(metav1.NamespaceSystem).Get(ctx, v.APIServerID, metav1.GetOptions{})
 		if err != nil || lease == nil || lease.Labels == nil ||
-			lease.Labels[controlplane.IdentityLeaseComponentLabelKey] != controlplane.KubeAPIServer {
+			lease.Labels[controlplane.IdentityLeaseComponentLabelKey] != apiserver.KubeAPIServer {
 			// We cannot find a corresponding identity lease from apiserver as well.
 			// We need to clean up this storage version.
 			hasInvalidID = true
@@ -243,7 +244,7 @@ func (c *Controller) enqueueStorageVersion(logger klog.Logger, obj *apiserverint
 	for _, sv := range obj.Status.StorageVersions {
 		lease, err := c.leaseLister.Leases(metav1.NamespaceSystem).Get(sv.APIServerID)
 		if err != nil || lease == nil || lease.Labels == nil ||
-			lease.Labels[controlplane.IdentityLeaseComponentLabelKey] != controlplane.KubeAPIServer {
+			lease.Labels[controlplane.IdentityLeaseComponentLabelKey] != apiserver.KubeAPIServer {
 			// we cannot find a corresponding identity lease in cache, enqueue the storageversion
 			logger.V(4).Info("Observed storage version with invalid apiserver entry", "objName", obj.Name)
 			c.storageVersionQueue.Add(obj.Name)
@@ -269,7 +270,7 @@ func (c *Controller) onDeleteLease(logger klog.Logger, obj interface{}) {
 
 	if castObj.Namespace == metav1.NamespaceSystem &&
 		castObj.Labels != nil &&
-		castObj.Labels[controlplane.IdentityLeaseComponentLabelKey] == controlplane.KubeAPIServer {
+		castObj.Labels[controlplane.IdentityLeaseComponentLabelKey] == apiserver.KubeAPIServer {
 		logger.V(4).Info("Observed lease deleted", "castObjName", castObj.Name)
 		c.enqueueLease(castObj)
 	}

pkg/controlplane/apiserver/config.go

@@ -313,10 +313,17 @@ func CreateConfig(
 		if err != nil {
 			return nil, nil, err
 		}
-		// build peer proxy config only if peer ca file exists
 		if opts.PeerCAFile != "" {
-			config.PeerProxy, err = BuildPeerProxy(versionedInformers, genericConfig.StorageVersionManager, opts.ProxyClientCertFile,
-				opts.ProxyClientKeyFile, opts.PeerCAFile, opts.PeerAdvertiseAddress, genericConfig.APIServerID, config.Extra.PeerEndpointLeaseReconciler, config.Generic.Serializer)
+			leaseInformer := versionedInformers.Coordination().V1().Leases()
+			config.PeerProxy, err = BuildPeerProxy(
+				leaseInformer,
+				genericConfig.LoopbackClientConfig,
+				opts.ProxyClientCertFile,
+				opts.ProxyClientKeyFile, opts.PeerCAFile,
+				opts.PeerAdvertiseAddress,
+				genericConfig.APIServerID,
+				config.Extra.PeerEndpointLeaseReconciler,
+				config.Generic.Serializer)
 			if err != nil {
 				return nil, nil, err
 			}

pkg/controlplane/apiserver/options/validation.go

@@ -78,16 +78,6 @@ func validateNodeSelectorAuthorizationFeature() []error {
 	return nil
 }
 
-func validateUnknownVersionInteroperabilityProxyFeature() []error {
-	if utilfeature.DefaultFeatureGate.Enabled(features.UnknownVersionInteroperabilityProxy) {
-		if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) {
-			return nil
-		}
-		return []error{fmt.Errorf("UnknownVersionInteroperabilityProxy feature requires StorageVersionAPI feature flag to be enabled")}
-	}
-	return nil
-}
-
 func validateUnknownVersionInteroperabilityProxyFlags(options *Options) []error {
 	err := []error{}
 	if !utilfeature.DefaultFeatureGate.Enabled(features.UnknownVersionInteroperabilityProxy) {
@@ -142,7 +132,6 @@ func (s *Options) Validate() []error {
 	errs = append(errs, s.APIEnablement.Validate(legacyscheme.Scheme, apiextensionsapiserver.Scheme, aggregatorscheme.Scheme)...)
 	errs = append(errs, validateTokenRequest(s)...)
 	errs = append(errs, s.Metrics.Validate()...)
-	errs = append(errs, validateUnknownVersionInteroperabilityProxyFeature()...)
 	errs = append(errs, validateUnknownVersionInteroperabilityProxyFlags(s)...)
 	errs = append(errs, validateNodeSelectorAuthorizationFeature()...)
 	errs = append(errs, validateServiceAccountTokenSigningConfig(s)...)

pkg/controlplane/apiserver/options/validation_test.go

@@ -26,7 +26,6 @@ import (
 	genericoptions "k8s.io/apiserver/pkg/server/options"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	basecompatibility "k8s.io/component-base/compatibility"
-	"k8s.io/component-base/featuregate"
 	basemetrics "k8s.io/component-base/metrics"
 	"k8s.io/kubernetes/pkg/features"
 
@@ -165,34 +164,6 @@ func TestValidateUnknownVersionInteroperabilityProxy(t *testing.T) {
 	}
 }
 
-func TestValidateUnknownVersionInteroperabilityProxyFeature(t *testing.T) {
-	const conflict = "UnknownVersionInteroperabilityProxy feature requires StorageVersionAPI feature flag to be enabled"
-	tests := []struct {
-		name            string
-		featuresEnabled []featuregate.Feature
-	}{
-		{
-			name:            "enabled: UnknownVersionInteroperabilityProxy, disabled: StorageVersionAPI",
-			featuresEnabled: []featuregate.Feature{features.UnknownVersionInteroperabilityProxy},
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			for _, feature := range test.featuresEnabled {
-				featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, feature, true)
-			}
-			var errMessageGot string
-			if errs := validateUnknownVersionInteroperabilityProxyFeature(); len(errs) > 0 {
-				errMessageGot = errs[0].Error()
-			}
-			if !strings.Contains(errMessageGot, conflict) {
-				t.Errorf("Expected error message to contain: %q, but got: %q", conflict, errMessageGot)
-			}
-		})
-	}
-}
-
 func TestValidateOptions(t *testing.T) {
 	testCases := []struct {
 		name         string

pkg/controlplane/apiserver/peer.go

@@ -24,13 +24,13 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/reconcilers"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/transport"
+
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
-	"k8s.io/apiserver/pkg/storageversion"
 	utilpeerproxy "k8s.io/apiserver/pkg/util/peerproxy"
-	clientgoinformers "k8s.io/client-go/informers"
-	"k8s.io/client-go/transport"
-	"k8s.io/klog/v2"
+	coordinationv1informers "k8s.io/client-go/informers/coordination/v1"
 	api "k8s.io/kubernetes/pkg/apis/core"
 )
 
@@ -43,17 +43,24 @@ const (
 	DefaultPeerEndpointReconcilerTTL = 15 * time.Second
 )
 
-func BuildPeerProxy(versionedInformer clientgoinformers.SharedInformerFactory, svm storageversion.Manager,
-	proxyClientCertFile string, proxyClientKeyFile string, peerCAFile string, peerAdvertiseAddress reconcilers.PeerAdvertiseAddress,
-	apiServerID string, reconciler reconcilers.PeerEndpointLeaseReconciler, serializer runtime.NegotiatedSerializer) (utilpeerproxy.Interface, error) {
+func BuildPeerProxy(
+	leaseInformer coordinationv1informers.LeaseInformer,
+	loopbackClientConfig *rest.Config,
+	proxyClientCertFile string,
+	proxyClientKeyFile string,
+	peerCAFile string,
+	peerAdvertiseAddress reconcilers.PeerAdvertiseAddress,
+	apiServerID string,
+	reconciler reconcilers.PeerEndpointLeaseReconciler,
+	serializer runtime.NegotiatedSerializer) (utilpeerproxy.Interface, error) {
 	if proxyClientCertFile == "" {
 		return nil, fmt.Errorf("error building peer proxy handler, proxy-cert-file not specified")
 	}
 	if proxyClientKeyFile == "" {
 		return nil, fmt.Errorf("error building peer proxy handler, proxy-key-file not specified")
 	}
-	// create proxy client config
-	clientConfig := &transport.Config{
+
+	proxyClientConfig := &transport.Config{
 		TLS: transport.TLSConfig{
 			Insecure:   false,
 			CertFile:   proxyClientCertFile,
@@ -62,20 +69,15 @@ func BuildPeerProxy(versionedInformer clientgoinformers.SharedInformerFactory, s
 			ServerName: "kubernetes.default.svc",
 		}}
 
-	// build proxy transport
-	proxyRoundTripper, transportBuildingError := transport.New(clientConfig)
-	if transportBuildingError != nil {
-		klog.Error(transportBuildingError.Error())
-		return nil, transportBuildingError
-	}
 	return utilpeerproxy.NewPeerProxyHandler(
-		versionedInformer,
-		svm,
-		proxyRoundTripper,
 		apiServerID,
+		IdentityLeaseComponentLabelKey+"="+KubeAPIServer,
+		leaseInformer,
 		reconciler,
 		serializer,
-	), nil
+		loopbackClientConfig,
+		proxyClientConfig,
+	)
 }
 
 // CreatePeerEndpointLeaseReconciler creates a apiserver endpoint lease reconciliation loop

pkg/controlplane/apiserver/server.go

@@ -74,6 +74,8 @@ const (
 	//   1. the lease is an identity lease (different from leader election leases)
 	//   2. which component owns this lease
 	IdentityLeaseComponentLabelKey = "apiserver.kubernetes.io/identity"
+	// KubeAPIServer defines variable used internally when referring to kube-apiserver component
+	KubeAPIServer = "kube-apiserver"
 )
 
 // Server is a struct that contains a generic control plane apiserver instance
@@ -212,7 +214,20 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele
 				return nil
 			})
 		if c.Extra.PeerProxy != nil {
-			s.GenericAPIServer.AddPostStartHookOrDie("unknown-version-proxy-filter", func(context genericapiserver.PostStartHookContext) error {
+			// Run local-discovery sync loop
+			s.GenericAPIServer.AddPostStartHookOrDie("local-discovery-cache-sync", func(context genericapiserver.PostStartHookContext) error {
+				err := c.Extra.PeerProxy.RunLocalDiscoveryCacheSync(context.Done())
+				return err
+			})
+
+			// Run peer-discovery sync loop.
+			s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-cache-sync", func(context genericapiserver.PostStartHookContext) error {
+				go c.Extra.PeerProxy.RunPeerDiscoveryCacheSync(context, 1)
+				return nil
+			})
+
+			// Wait for handler to be ready.
+			s.GenericAPIServer.AddPostStartHookOrDie("mixed-version-proxy-handler", func(context genericapiserver.PostStartHookContext) error {
 				err := c.Extra.PeerProxy.WaitForCacheSync(context.Done())
 				return err
 			})

pkg/controlplane/instance.go

@@ -115,8 +115,6 @@ const (
 	//   2. which component owns this lease
 	// TODO(sttts): remove this indirection
 	IdentityLeaseComponentLabelKey = controlplaneapiserver.IdentityLeaseComponentLabelKey
-	// KubeAPIServer defines variable used internally when referring to kube-apiserver component
-	KubeAPIServer = "kube-apiserver"
 	// repairLoopInterval defines the interval used to run the Services ClusterIP and NodePort repair loops
 	repairLoopInterval = 3 * time.Minute
 )
@@ -312,7 +310,7 @@ func (c CompletedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 		return nil, fmt.Errorf("Master.New() called with empty config.KubeletClientConfig")
 	}
 
-	cp, err := c.ControlPlane.New(KubeAPIServer, delegationTarget)
+	cp, err := c.ControlPlane.New(controlplaneapiserver.KubeAPIServer, delegationTarget)
 	if err != nil {
 		return nil, err
 	}

staging/src/k8s.io/apiserver/pkg/util/peerproxy/metrics/metrics.go

@@ -50,6 +50,11 @@ func Register() {
 	})
 }
 
+// Only used for tests.
+func Reset() {
+	legacyregistry.Reset()
+}
+
 // IncPeerProxiedRequest increments the # of proxied requests to peer kube-apiserver
 func IncPeerProxiedRequest(ctx context.Context, status string) {
 	peerProxiedRequestsTotal.WithContext(ctx).WithLabelValues(status).Add(1)