feat: cluster context storage rewrite

Author: zachsmith1

README.md

@@ -147,29 +147,14 @@ keyFunc := func(obj runtime.Object) (string, error) {
 }
 ```
 
-#### Admission: server-owned cluster label
-We stamp a server-owned cluster label on persisted objects and validate it on
-create/update to prevent cross-cluster writes and support watchcache keying.
+#### Admission: internal cluster identity
+We store cluster identity in managed fields and validate it on create/update
+to prevent cross-cluster writes and support watchcache keying.
 
 ```go
-lbls := accessor.GetLabels()
-if lbls == nil {
-	lbls = map[string]string{}
-}
-key := m.Options.ClusterAnnotationKey
-if key == "" {
-	key = mcv1.DefaultClusterAnnotation
-}
-lbls[key] = cid
-accessor.SetLabels(lbls)
-```
-Note: the `ClusterAnnotationKey` fallback to `DefaultClusterAnnotation` is
-temporary and not ideal. It exists to preserve compatibility while we migrate
-callers to an explicit label key; avoid relying on this fallback long term.
-
-```go
-if cid := acc.GetLabels()[key]; cid != reqCID {
-	return fmt.Errorf("cluster label %q=%q must match request cluster %q", key, cid, reqCID)
+mc.SetObjectClusterIdentity(obj, reqCID)
+if cid := mc.ObjectClusterIdentity(obj); cid != reqCID {
+	return fmt.Errorf("cluster identity %q must match request cluster %q", cid, reqCID)
 }
 ```
 

cmd/apiserver/app/config.go

@@ -146,7 +146,7 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	})
 	genericConfig.BuildHandlerChainFunc = func(h http.Handler, conf *server.Config) http.Handler {
 		ex := mc.PathExtractor{PathPrefix: mcOpts.PathPrefix, ControlPlaneSegment: mcOpts.ControlPlaneSegment}
-		base := server.DefaultBuildHandlerChain(h, conf)
+		base := withVersionOverride(server.DefaultBuildHandlerChain(h, conf))
 		dispatch := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			cid, _, _ := mc.FromContext(r.Context())
 			if cid != "" && cid != mcOpts.DefaultCluster && crdRuntimeMgr != nil {
@@ -179,6 +179,8 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		BaseLoopbackClientConfig: genericConfig.LoopbackClientConfig,
 		PathPrefix:               mcOpts.PathPrefix,
 		ControlPlaneSegment:      mcOpts.ControlPlaneSegment,
+		EtcdPrefix:               storageFactory.StorageConfig.Prefix,
+		EtcdTransport:            storageFactory.StorageConfig.Transport,
 		Authentication:           opts.Authentication,
 		Authorization:            opts.Authorization,
 		EgressSelector:           genericConfig.EgressSelector,
@@ -190,7 +192,9 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		genericConfig.Authentication.Authenticator = mcauth.NewClusterAuthenticator(mcOpts.DefaultCluster, genericConfig.Authentication.Authenticator, authManager)
 	}
 	if genericConfig.Authorization.Authorizer != nil {
-		clusterAuthorizer := mcauth.NewClusterAuthorizer(mcOpts.DefaultCluster, genericConfig.Authorization.Authorizer, genericConfig.RuleResolver, authManager)
+		// Route root and tenant authorization through the same multicluster
+		// manager-backed path to avoid root-only stale lister divergence.
+		clusterAuthorizer := mcauth.NewClusterAuthorizer(mcOpts.DefaultCluster, nil, nil, authManager)
 		genericConfig.Authorization.Authorizer = clusterAuthorizer
 		genericConfig.RuleResolver = clusterAuthorizer
 	}
@@ -314,6 +318,9 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	if apiExtensions.GenericConfig.RESTOptionsGetter != nil {
 		apiExtensions.GenericConfig.RESTOptionsGetter = decorateRESTOptionsGetter("apiextensions", apiExtensions.GenericConfig.RESTOptionsGetter, mcOpts)
 	}
+	if apiExtensions.ExtraConfig.CRDRESTOptionsGetter != nil {
+		apiExtensions.ExtraConfig.CRDRESTOptionsGetter = decorateRESTOptionsGetter("apiextensions-crd", apiExtensions.ExtraConfig.CRDRESTOptionsGetter, mcOpts)
+	}
 	apiExtensionsClientPool := mc.NewAPIExtensionsClientPool(apiExtensions.GenericConfig.LoopbackClientConfig, mcOpts.PathPrefix, mcOpts.ControlPlaneSegment)
 	apiExtensionsInformerPool := mc.NewAPIExtensionsInformerPoolFromClientPool(apiExtensionsClientPool, 0, genericConfig.DrainedNotify())
 	crdRuntimeMgr = mcbootstrap.NewCRDRuntimeManager(mcbootstrap.CRDRuntimeManagerOptions{
@@ -322,6 +329,8 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		PathPrefix:                mcOpts.PathPrefix,
 		ControlPlaneSegment:       mcOpts.ControlPlaneSegment,
 		DefaultCluster:            mcOpts.DefaultCluster,
+		EtcdPrefix:                storageFactory.StorageConfig.Prefix,
+		EtcdTransport:             storageFactory.StorageConfig.Transport,
 	})
 	setCRDRequestResolvers(apiExtensions, crdRuntimeMgr.CRDGetterForRequest, crdRuntimeMgr.CRDListerForRequest)
 	crdController := mcbootstrap.NewMulticlusterCRDController(crdRuntimeMgr, mcOpts.DefaultCluster)
@@ -364,7 +373,7 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	// Ensure CRDs are also routed through the multicluster handler
 	apiExtensions.GenericConfig.BuildHandlerChainFunc = func(h http.Handler, conf *server.Config) http.Handler {
 		ex := mc.PathExtractor{PathPrefix: mcOpts.PathPrefix, ControlPlaneSegment: mcOpts.ControlPlaneSegment}
-		base := server.DefaultBuildHandlerChain(h, conf)
+		base := withVersionOverride(server.DefaultBuildHandlerChain(h, conf))
 		dispatch := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			cid, _, _ := mc.FromContext(r.Context())
 			if cid != "" && cid != mcOpts.DefaultCluster {
@@ -381,11 +390,11 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		mut := mca.NewMutating(mcOpts)
 		val := mca.NewValidating(mcOpts)
 		base := apiExtensions.GenericConfig.AdmissionControl
-		chain := []admission.Interface{mut, mcNamespaceLifecycle, mcMutatingWebhook}
+		chain := []admission.Interface{mut, mcNamespaceLifecycle}
 		if base != nil {
 			chain = append(chain, base)
 		}
-		chain = append(chain, mcValidatingWebhook, val)
+		chain = append(chain, val)
 		apiExtensions.GenericConfig.AdmissionControl = admission.NewChainHandler(chain...)
 	}
 	c.ApiExtensions = apiExtensions
@@ -400,7 +409,7 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	// Ensure aggregator also receives multicluster routing
 	aggregator.GenericConfig.BuildHandlerChainFunc = func(h http.Handler, conf *server.Config) http.Handler {
 		ex := mc.PathExtractor{PathPrefix: mcOpts.PathPrefix, ControlPlaneSegment: mcOpts.ControlPlaneSegment}
-		base := server.DefaultBuildHandlerChain(h, conf)
+		base := withVersionOverride(server.DefaultBuildHandlerChain(h, conf))
 		dispatch := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			cid, _, _ := mc.FromContext(r.Context())
 			if cid != "" && cid != mcOpts.DefaultCluster && crdRuntimeMgr != nil {
@@ -417,11 +426,11 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 		mut := mca.NewMutating(mcOpts)
 		val := mca.NewValidating(mcOpts)
 		base := aggregator.GenericConfig.AdmissionControl
-		chain := []admission.Interface{mut, mcNamespaceLifecycle, mcMutatingWebhook}
+		chain := []admission.Interface{mut, mcNamespaceLifecycle}
 		if base != nil {
 			chain = append(chain, base)
 		}
-		chain = append(chain, mcValidatingWebhook, val)
+		chain = append(chain, val)
 		aggregator.GenericConfig.AdmissionControl = admission.NewChainHandler(chain...)
 	}
 	c.Aggregator = aggregator

cmd/apiserver/app/server.go

@@ -49,7 +49,6 @@ import (
 	logsapi "k8s.io/component-base/logs/api/v1"
 	_ "k8s.io/component-base/metrics/prometheus/workqueue"
 	"k8s.io/component-base/term"
-	utilversion "k8s.io/component-base/version"
 	"k8s.io/component-base/version/verflag"
 	"k8s.io/component-base/zpages/flagz"
 	"k8s.io/klog/v2"
@@ -148,7 +147,7 @@ cluster's shared state through which all other components interact.`,
 // Run runs the specified APIServer.  This should never exit.
 func Run(ctx context.Context, opts options.CompletedOptions) error {
 	// To help debugging, immediately log version
-	klog.Infof("Version: %+v", utilversion.Get())
+	klog.Infof("Version: %+v", normalizedServerVersion())
 
 	klog.InfoS("Golang settings", "GOGC", os.Getenv("GOGC"), "GOMAXPROCS", os.Getenv("GOMAXPROCS"), "GOTRACEBACK", os.Getenv("GOTRACEBACK"))
 

cmd/apiserver/app/version_override.go

@@ -0,0 +1,51 @@
+package app
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	apimachineryversion "k8s.io/apimachinery/pkg/version"
+	utilversion "k8s.io/component-base/version"
+)
+
+const gitVersionArchivePlaceholder = "v0.0.0-master+$Format:%H$"
+
+func withVersionOverride(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if isVersionPath(r.URL.Path) {
+			info := normalizedServerVersion()
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(info)
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+func isVersionPath(path string) bool {
+	if path == "/version" || path == "/version/" {
+		return true
+	}
+	trimmed := strings.Trim(path, "/")
+	parts := strings.Split(trimmed, "/")
+	return len(parts) == 3 && parts[0] == "clusters" && parts[2] == "version"
+}
+
+func normalizedServerVersion() apimachineryversion.Info {
+	info := utilversion.Get()
+	if info.GitVersion != gitVersionArchivePlaceholder {
+		return info
+	}
+
+	parts := strings.SplitN(utilversion.DefaultKubeBinaryVersion, ".", 2)
+	if len(parts) != 2 {
+		return info
+	}
+
+	info.Major = parts[0]
+	info.Minor = parts[1]
+	info.GitVersion = "v" + utilversion.DefaultKubeBinaryVersion + ".0"
+	return info
+}
+

cmd/apiserver/app/version_override_test.go

@@ -0,0 +1,30 @@
+package app
+
+import "testing"
+
+func TestIsVersionPath(t *testing.T) {
+	tests := []struct {
+		path string
+		want bool
+	}{
+		{path: "/version", want: true},
+		{path: "/version/", want: true},
+		{path: "/clusters/kpt-1/version", want: true},
+		{path: "/clusters/kpt-1/version/", want: true},
+		{path: "/apis", want: false},
+	}
+
+	for _, tt := range tests {
+		if got := isVersionPath(tt.path); got != tt.want {
+			t.Fatalf("isVersionPath(%q)=%v, want %v", tt.path, got, tt.want)
+		}
+	}
+}
+
+func TestNormalizedServerVersion(t *testing.T) {
+	info := normalizedServerVersion()
+	if info.GitVersion == gitVersionArchivePlaceholder {
+		t.Fatalf("expected non-placeholder git version, got %q", info.GitVersion)
+	}
+}
+