From 441a89ba51a06f57c1046fcaf5514f3f0c956fa4 Mon Sep 17 00:00:00 2001 From: Mangirdas Judeikis Date: Fri, 21 Aug 2026 11:51:35 +0300 Subject: [PATCH 1/5] Move shards object ownership to shards --- cmd/sharded-test-server/main.go | 54 ---- cmd/sharded-test-server/shard.go | 5 + .../concepts/workspaces/system-workspaces.md | 6 + pkg/indexers/shard.go | 50 ++++ pkg/indexers/shard_test.go | 73 +++++ .../apiexportendpointsliceurls_controller.go | 10 +- ...exportendpointsliceurls_controller_test.go | 10 +- .../apiexportendpointsliceurls_reconcile.go | 4 +- ...hedresourceendpointsliceurls_controller.go | 10 +- .../replication/replication_controller.go | 54 +++- .../shardmirror/shardmirror_controller.go | 267 ++++++++++++++++++ .../shardmirror_controller_test.go | 213 ++++++++++++++ .../logicalclustermigration/datacopy.go | 4 +- .../tenancy/workspace/workspace_reconcile.go | 3 +- pkg/server/config.go | 4 +- pkg/server/controllers.go | 23 ++ pkg/server/options/options.go | 2 + pkg/server/server.go | 33 ++- .../sdk/apis/core/v1alpha1/shard_types.go | 8 + 19 files changed, 736 insertions(+), 97 deletions(-) create mode 100644 pkg/indexers/shard.go create mode 100644 pkg/indexers/shard_test.go create mode 100644 pkg/reconciler/core/shardmirror/shardmirror_controller.go create mode 100644 pkg/reconciler/core/shardmirror/shardmirror_controller_test.go diff --git a/cmd/sharded-test-server/main.go b/cmd/sharded-test-server/main.go index 98ecb15eb9d..1304f961efc 100644 --- a/cmd/sharded-test-server/main.go +++ b/cmd/sharded-test-server/main.go @@ -26,18 +26,12 @@ import ( "strings" "time" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" kuser "k8s.io/apiserver/pkg/authentication/user" genericapiserver "k8s.io/apiserver/pkg/server" - "k8s.io/client-go/tools/clientcmd" - "k8s.io/client-go/util/retry" "k8s.io/klog/v2" - "github.com/kcp-dev/sdk/apis/core" - kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" "github.com/kcp-dev/sdk/testing/third_party/library-go/crypto" testshard "github.com/kcp-dev/kcp/cmd/test-server/kcp" @@ -316,37 +310,6 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb return err } - // Label region of shards - clientConfig, err := loadKubeConfig(filepath.Join(workDirPath, ".kcp", "admin.kubeconfig"), "base") - if err != nil { - return err - } - config, err := clientConfig.ClientConfig() - if err != nil { - return err - } - client, err := kcpclientset.NewForConfig(config) - if err != nil { - return err - } - for i := range shards { - name := fmt.Sprintf("shard-%d", i) - if i == 0 { - name = "root" - } - - if i >= len(regions) { - break - } - patch := fmt.Sprintf(`{"metadata":{"labels":{"region":%q,"shared": "true"}}}`, regions[i]) - if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { - _, err := client.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Patch(ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{}) - return err - }); err != nil { - return err - } - } - readyToTestFile, err := os.Create(filepath.Join(workDirPath, ".kcp", "ready-to-test")) if err != nil { return fmt.Errorf("error creating ready-to-test file: %w", err) @@ -380,23 +343,6 @@ type indexErrTuple struct { error error } -func loadKubeConfig(kubeconfigPath, contextName string) (clientcmd.ClientConfig, error) { - fs, err := os.Stat(kubeconfigPath) - if err != nil { - return nil, err - } - if fs.Size() == 0 { - return nil, fmt.Errorf("%s points to an empty file", kubeconfigPath) - } - - rawConfig, err := clientcmd.LoadFromFile(kubeconfigPath) - if err != nil { - return nil, fmt.Errorf("failed to load admin kubeconfig: %w", err) - } - - return clientcmd.NewNonInteractiveClientConfig(*rawConfig, contextName, nil, nil), nil -} - var regions = []string{ "us-east-2", "us-east-1", diff --git a/cmd/sharded-test-server/shard.go b/cmd/sharded-test-server/shard.go index 11db1fe22ad..5cc026ceae0 100644 --- a/cmd/sharded-test-server/shard.go +++ b/cmd/sharded-test-server/shard.go @@ -112,6 +112,11 @@ func newShard(ctx context.Context, n int, args []string, standaloneVW bool, serv fmt.Sprintf("--service-account-signing-key-file=%s", filepath.Join(kcpDir, "service-account.key")), "--audit-log-path", auditFilePath, fmt.Sprintf("--shard-external-url=https://%s:%d", hostIP, 6443), + ) + if n < len(regions) { + args = append(args, fmt.Sprintf("--shard-labels=region=%s,shared=true", regions[n])) + } + args = append(args, fmt.Sprintf("--tls-cert-file=%s", filepath.Join(shardDir, "apiserver.crt")), fmt.Sprintf("--tls-private-key-file=%s", filepath.Join(shardDir, "apiserver.key")), fmt.Sprintf("--secure-port=%d", 6444+n), diff --git a/docs/content/concepts/workspaces/system-workspaces.md b/docs/content/concepts/workspaces/system-workspaces.md index 2f8bba4a4d4..08f420d1c70 100644 --- a/docs/content/concepts/workspaces/system-workspaces.md +++ b/docs/content/concepts/workspaces/system-workspaces.md @@ -103,6 +103,12 @@ Each bound CRD is: ## `system:shard` The `system:shard` workspace holds essential API bindings that every shard needs to function. +It also holds the shard's own authoritative `Shard` object: each shard registers itself here +at startup, the object is replicated to the cache server, and the root shard mirrors it into +the root workspace as a read-only representation (marked with the +`core.kcp.io/shard-representation` annotation). To change a shard's configuration (URLs, +labels), change the shard's flags/deployment rather than editing the representation. + During shard bootstrap, API bindings for the following API groups are created in this workspace: - `shards.core.kcp.io` diff --git a/pkg/indexers/shard.go b/pkg/indexers/shard.go new file mode 100644 index 00000000000..bfadc58190c --- /dev/null +++ b/pkg/indexers/shard.go @@ -0,0 +1,50 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package indexers + +import ( + "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/labels" + + configshard "github.com/kcp-dev/kcp/config/shard" + "github.com/kcp-dev/logicalcluster/v3" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + corev1alpha1listers "github.com/kcp-dev/sdk/client/listers/core/v1alpha1" +) + +// ShardByName returns the Shard with the given name, regardless of the logical +// cluster it lives in. +func ShardByName(lister corev1alpha1listers.ShardClusterLister, name string) (*corev1alpha1.Shard, error) { + shards, err := lister.List(labels.Everything()) + if err != nil { + return nil, err + } + var found *corev1alpha1.Shard + for _, shard := range shards { + if shard.Name != name { + continue + } + if logicalcluster.From(shard) == configshard.SystemShardCluster { + return shard, nil + } + found = shard + } + if found == nil { + return nil, errors.NewNotFound(corev1alpha1.Resource("shards"), name) + } + return found, nil +} diff --git a/pkg/indexers/shard_test.go b/pkg/indexers/shard_test.go new file mode 100644 index 00000000000..9943e46d4a1 --- /dev/null +++ b/pkg/indexers/shard_test.go @@ -0,0 +1,73 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package indexers + +import ( + "testing" + + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + + kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" + "github.com/kcp-dev/logicalcluster/v3" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + corev1alpha1listers "github.com/kcp-dev/sdk/client/listers/core/v1alpha1" +) + +func shardInCluster(name, cluster string) *corev1alpha1.Shard { + return &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Annotations: map[string]string{logicalcluster.AnnotationKey: cluster}, + }, + } +} + +func TestShardByName(t *testing.T) { + indexer := cache.NewIndexer(kcpcache.MetaClusterNamespaceKeyFunc, cache.Indexers{}) + for _, shard := range []*corev1alpha1.Shard{ + shardInCluster("alpha", "system:shard"), + shardInCluster("alpha", "root"), // legacy copy, authoritative wins + shardInCluster("legacy", "root"), + } { + if err := indexer.Add(shard); err != nil { + t.Fatal(err) + } + } + lister := corev1alpha1listers.NewShardClusterLister(indexer) + + shard, err := ShardByName(lister, "alpha") + if err != nil { + t.Fatal(err) + } + if cluster := logicalcluster.From(shard); cluster != "system:shard" { + t.Errorf("expected the authoritative shard from system:shard, got the one from %q", cluster) + } + + shard, err = ShardByName(lister, "legacy") + if err != nil { + t.Fatal(err) + } + if cluster := logicalcluster.From(shard); cluster != "root" { + t.Errorf("expected the legacy shard from root, got the one from %q", cluster) + } + + if _, err := ShardByName(lister, "missing"); !errors.IsNotFound(err) { + t.Errorf("expected a NotFound error, got %v", err) + } +} diff --git a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller.go b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller.go index 5fc16a1c926..5350e6d541a 100644 --- a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller.go +++ b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller.go @@ -58,7 +58,7 @@ const ( // NewController returns a new controller for APIExportEndpointSlices. // Shards and APIExports are read from the cache server. func NewController( - shardName string, + thisShard string, apiExportEndpointSliceClusterInformer apisv1alpha1informers.APIExportEndpointSliceClusterInformer, apiBindingInformer apisv1alpha2informers.APIBindingClusterInformer, globalAPIExportEndpointSliceClusterInformer apisv1alpha1informers.APIExportEndpointSliceClusterInformer, @@ -67,7 +67,7 @@ func NewController( clusterClient kcpclientset.ClusterInterface, ) (*controller, error) { c := &controller{ - shardName: shardName, + thisShard: thisShard, clusterClient: clusterClient, queue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), @@ -76,7 +76,7 @@ func NewController( }, ), getMyShard: func() (*corev1alpha1.Shard, error) { - return globalShardClusterInformer.Cluster(core.RootCluster).Lister().Get(shardName) + return indexers.ShardByName(globalShardClusterInformer.Lister(), thisShard) }, getAPIExportEndpointSlice: func(path logicalcluster.Path, name string) (*apisv1alpha1.APIExportEndpointSlice, error) { obj, err := indexers.ByPathAndNameWithFallback[*apisv1alpha1.APIExportEndpointSlice](apisv1alpha1.Resource("apiexportendpointslices"), apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), globalAPIExportEndpointSliceClusterInformer.Informer().GetIndexer(), path, name) @@ -121,7 +121,7 @@ func NewController( }, patchAPIExportEndpointSlice: func(ctx context.Context, cluster logicalcluster.Path, patch *apisv1alpha1apply.APIExportEndpointSliceApplyConfiguration) error { _, err := clusterClient.ApisV1alpha1().APIExportEndpointSlices().Cluster(cluster).ApplyStatus(ctx, patch, metav1.ApplyOptions{ - FieldManager: shardName, + FieldManager: thisShard, }) return err }, @@ -174,7 +174,7 @@ func NewController( // in the status of every APIExportEndpointSlices. type controller struct { queue workqueue.TypedRateLimitingInterface[string] - shardName string + thisShard string clusterClient kcpclientset.ClusterInterface getMyShard func() (*corev1alpha1.Shard, error) diff --git a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller_test.go b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller_test.go index 50e5b2c2d34..887268198b4 100644 --- a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller_test.go +++ b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_controller_test.go @@ -145,7 +145,7 @@ func TestReconcile(t *testing.T) { }, }, endpointsReconciler: &endpointsReconciler{ - shardName: "shard2", + thisShard: "shard2", getAPIExport: func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) { return &apisv1alpha2.APIExport{}, nil }, @@ -183,7 +183,7 @@ func TestReconcile(t *testing.T) { }, }, endpointsReconciler: &endpointsReconciler{ - shardName: "shard1", + thisShard: "shard1", getAPIExport: func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) { return &apisv1alpha2.APIExport{}, nil }, @@ -232,7 +232,7 @@ func TestReconcile(t *testing.T) { }, }, endpointsReconciler: &endpointsReconciler{ - shardName: "shard1", + thisShard: "shard1", getAPIExport: func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) { return &apisv1alpha2.APIExport{}, nil }, @@ -276,7 +276,7 @@ func TestReconcile(t *testing.T) { }, }, endpointsReconciler: &endpointsReconciler{ - shardName: "shard1", + thisShard: "shard1", getAPIExport: func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) { return &apisv1alpha2.APIExport{ ObjectMeta: metav1.ObjectMeta{ @@ -325,7 +325,7 @@ func TestReconcile(t *testing.T) { getAPIExport: tc.endpointsReconciler.getAPIExport, listAPIBindingsByAPIExport: tc.endpointsReconciler.listAPIBindingsByAPIExport, patchAPIExportEndpointSlice: tc.endpointsReconciler.patchAPIExportEndpointSlice, - shardName: tc.endpointsReconciler.shardName, + thisShard: tc.endpointsReconciler.thisShard, } input := tc.input.DeepCopy() _, err := c.reconcile(context.Background(), input) diff --git a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_reconcile.go b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_reconcile.go index 4435437eedc..54da7d1f2d6 100644 --- a/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_reconcile.go +++ b/pkg/reconciler/apis/apiexportendpointsliceurls/apiexportendpointsliceurls_reconcile.go @@ -42,7 +42,7 @@ type endpointsReconciler struct { getAPIExport func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) listAPIBindingsByAPIExport func(apiexport *apisv1alpha2.APIExport) ([]*apisv1alpha2.APIBinding, error) patchAPIExportEndpointSlice func(ctx context.Context, cluster logicalcluster.Path, patch *apisv1alpha1apply.APIExportEndpointSliceApplyConfiguration) error - shardName string + thisShard string } type result struct { @@ -55,7 +55,7 @@ func (c *controller) reconcile(ctx context.Context, apiExportEndpointSlice *apis getMyShard: c.getMyShard, getAPIExport: c.getAPIExport, listAPIBindingsByAPIExport: c.listAPIBindingsByAPIExport, - shardName: c.shardName, + thisShard: c.thisShard, patchAPIExportEndpointSlice: c.patchAPIExportEndpointSlice, } diff --git a/pkg/reconciler/cache/clustercachedresourceendpointsliceurls/clustercachedresourceendpointsliceurls_controller.go b/pkg/reconciler/cache/clustercachedresourceendpointsliceurls/clustercachedresourceendpointsliceurls_controller.go index 6d2b265b1cf..73c12a44516 100644 --- a/pkg/reconciler/cache/clustercachedresourceendpointsliceurls/clustercachedresourceendpointsliceurls_controller.go +++ b/pkg/reconciler/cache/clustercachedresourceendpointsliceurls/clustercachedresourceendpointsliceurls_controller.go @@ -89,7 +89,7 @@ func listAPIBindingsByAPIExport(apiBindingInformer apisv1alpha2informers.APIBind } func NewController( - shardName string, + thisShard string, apiBindingInformer apisv1alpha2informers.APIBindingClusterInformer, localClusterCachedResourceEndpointSliceClusterInformer, globalClusterCachedResourceEndpointSliceClusterInformer cachev1alpha1informers.ClusterCachedResourceEndpointSliceClusterInformer, globalShardClusterInformer corev1alpha1informers.ShardClusterInformer, @@ -98,7 +98,7 @@ func NewController( clusterClient kcpclientset.ClusterInterface, ) (*controller, error) { c := &controller{ - shardName: shardName, + thisShard: thisShard, queue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{ @@ -106,7 +106,7 @@ func NewController( }, ), getMyShard: func() (*corev1alpha1.Shard, error) { - return globalShardClusterInformer.Cluster(core.RootCluster).Lister().Get(shardName) + return indexers.ShardByName(globalShardClusterInformer.Lister(), thisShard) }, getClusterCachedResourceEndpointSlice: func(cluster logicalcluster.Name, name string) (*cachev1alpha1.ClusterCachedResourceEndpointSlice, error) { obj, err := indexers.ByPathAndNameWithFallback[*cachev1alpha1.ClusterCachedResourceEndpointSlice](cachev1alpha1.Resource("clustercachedresourceendpointslices"), localClusterCachedResourceEndpointSliceClusterInformer.Informer().GetIndexer(), globalClusterCachedResourceEndpointSliceClusterInformer.Informer().GetIndexer(), cluster.Path(), name) @@ -123,7 +123,7 @@ func NewController( }, patchClusterCachedResourceEndpointSlice: func(ctx context.Context, cluster logicalcluster.Path, patch *cachev1alpha1apply.ClusterCachedResourceEndpointSliceApplyConfiguration) error { _, err := clusterClient.CacheV1alpha1().ClusterCachedResourceEndpointSlices().Cluster(cluster).ApplyStatus(ctx, patch, metav1.ApplyOptions{ - FieldManager: shardName, + FieldManager: thisShard, }) return err }, @@ -251,7 +251,7 @@ func (c *controller) enqueueClusterCachedResourceEndpointSlice(obj *cachev1alpha type controller struct { queue workqueue.TypedRateLimitingInterface[string] - shardName string + thisShard string getAPIExport func(path logicalcluster.Path, name string) (*apisv1alpha2.APIExport, error) listAPIBindingsByAPIExport func(export *apisv1alpha2.APIExport) ([]*apisv1alpha2.APIBinding, error) diff --git a/pkg/reconciler/cache/replication/replication_controller.go b/pkg/reconciler/cache/replication/replication_controller.go index 43802faefab..61d5e3f6e1f 100644 --- a/pkg/reconciler/cache/replication/replication_controller.go +++ b/pkg/reconciler/cache/replication/replication_controller.go @@ -36,6 +36,7 @@ import ( kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" kcpdynamic "github.com/kcp-dev/client-go/dynamic" kcpkubernetesinformers "github.com/kcp-dev/client-go/informers" + "github.com/kcp-dev/logicalcluster/v3" apisv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1" apisv1alpha2 "github.com/kcp-dev/sdk/apis/apis/v1alpha2" cachev1alpha1 "github.com/kcp-dev/sdk/apis/cache/v1alpha1" @@ -45,6 +46,7 @@ import ( tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1" kcpinformers "github.com/kcp-dev/sdk/client/informers/externalversions" + configshard "github.com/kcp-dev/kcp/config/shard" cacheclient "github.com/kcp-dev/kcp/pkg/cache/client" "github.com/kcp-dev/kcp/pkg/cache/client/shard" kcpfeatures "github.com/kcp-dev/kcp/pkg/features" @@ -80,8 +82,13 @@ func NewController( } for gvr, info := range c.Gvrs { + eventFilter := IsNoSystemClusterName + if info.EventFilter != nil { + eventFilter = info.EventFilter + } + _, _ = info.Local.AddEventHandler(cache.FilteringResourceEventHandler{ - FilterFunc: IsNoSystemClusterName, + FilterFunc: eventFilter, Handler: cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { c.enqueueObject(obj, gvr) }, UpdateFunc: func(_, obj interface{}) { c.enqueueObject(obj, gvr) }, @@ -90,7 +97,7 @@ func NewController( }) _, _ = info.Global.AddEventHandler(cache.FilteringResourceEventHandler{ - FilterFunc: IsNoSystemClusterName, // not really needed, but cannot harm + FilterFunc: eventFilter, // not really needed, but cannot harm Handler: cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { c.enqueueCacheObject(obj, gvr) }, UpdateFunc: func(_, obj interface{}) { c.enqueueCacheObject(obj, gvr) }, @@ -183,6 +190,28 @@ func IsNoSystemClusterName(obj interface{}) bool { return true } +// isNoSystemClusterNameExceptSystemShard is like IsNoSystemClusterName, but +// additionally lets objects in the shard-local system:shard logical cluster +// through. Used for resources whose authoritative copy is shard-owned, like +// Shards. +func isNoSystemClusterNameExceptSystemShard(obj interface{}) bool { + key, err := kcpcache.DeletionHandlingMetaClusterNamespaceKeyFunc(obj) + if err != nil { + utilruntime.HandleError(err) + return false + } + + clusterName, _, _, err := kcpcache.SplitMetaClusterNamespaceKey(key) + if err != nil { + utilruntime.HandleError(err) + return false + } + if clusterName == configshard.SystemShardCluster { + return true + } + return !strings.HasPrefix(clusterName.String(), "system:") +} + type controller struct { shardName string queue workqueue.TypedRateLimitingInterface[string] @@ -193,8 +222,12 @@ type controller struct { } type ReplicatedGVR struct { - Kind string - Filter func(u *unstructured.Unstructured) bool + Kind string + Filter func(u *unstructured.Unstructured) bool + // EventFilter overrides the default IsNoSystemClusterName event filter for + // this resource, e.g. to replicate objects living in a system logical + // cluster. + EventFilter func(obj interface{}) bool Global, Local cache.SharedIndexInformer } @@ -256,9 +289,16 @@ func InstallIndexers( Global: globalKcpInformers.Cache().V1alpha1().ClusterCachedResourceEndpointSlices().Informer(), }, corev1alpha1.SchemeGroupVersion.WithResource("shards"): { - Kind: "Shard", - Local: localKcpInformers.Core().V1alpha1().Shards().Informer(), - Global: globalKcpInformers.Core().V1alpha1().Shards().Informer(), + Kind: "Shard", + Filter: func(u *unstructured.Unstructured) bool { + if logicalcluster.From(u) == configshard.SystemShardCluster { + return true + } + return u.GetAnnotations()[corev1alpha1.ShardRepresentationAnnotationKey] == "" + }, + EventFilter: isNoSystemClusterNameExceptSystemShard, + Local: localKcpInformers.Core().V1alpha1().Shards().Informer(), + Global: globalKcpInformers.Core().V1alpha1().Shards().Informer(), }, corev1alpha1.SchemeGroupVersion.WithResource("logicalclusters"): { Kind: "LogicalCluster", diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller.go b/pkg/reconciler/core/shardmirror/shardmirror_controller.go new file mode 100644 index 00000000000..25665c3389a --- /dev/null +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller.go @@ -0,0 +1,267 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package shardmirror mirrors shard-owned Shard objects (living in each +// shard's local system:shard logical cluster and replicated to the cache +// server) into the root workspace as read-only representations. It runs on the +// root shard only. Representations are marked with the +// core.kcp.io/shard-representation annotation, continuously overwritten from +// the authoritative object, and pruned when the authoritative object +// disappears. Shard objects in the root workspace without that annotation +// (e.g. created by tests or old shards) are left alone. +package shardmirror + +import ( + "context" + "fmt" + "time" + + "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + genericrequest "k8s.io/apiserver/pkg/endpoints/request" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + "k8s.io/klog/v2" + + kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" + "github.com/kcp-dev/logicalcluster/v3" + "github.com/kcp-dev/sdk/apis/core" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" + corev1alpha1informers "github.com/kcp-dev/sdk/client/informers/externalversions/core/v1alpha1" + + configshard "github.com/kcp-dev/kcp/config/shard" + "github.com/kcp-dev/kcp/pkg/logging" + "github.com/kcp-dev/kcp/pkg/reconciler/cache/replication" +) + +const ( + ControllerName = "kcp-shard-mirror" +) + +// NewController returns a controller mirroring authoritative Shard objects +// from the cache server into the root workspace as read-only representations. +func NewController( + kcpClusterClient kcpclientset.ClusterInterface, + localShardInformer corev1alpha1informers.ShardClusterInformer, + cacheShardInformer corev1alpha1informers.ShardClusterInformer, +) *Controller { + c := &Controller{ + queue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{ + Name: ControllerName, + }, + ), + getSourceShard: func(name string) (*corev1alpha1.Shard, error) { + return cacheShardInformer.Cluster(configshard.SystemShardCluster).Lister().Get(name) + }, + getLocalShard: func(name string) (*corev1alpha1.Shard, error) { + return localShardInformer.Cluster(core.RootCluster).Lister().Get(name) + }, + createShard: func(ctx context.Context, shard *corev1alpha1.Shard) error { + _, err := kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Create(ctx, shard, metav1.CreateOptions{}) + return err + }, + updateShard: func(ctx context.Context, shard *corev1alpha1.Shard) error { + _, err := kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) + return err + }, + updateShardStatus: func(ctx context.Context, shard *corev1alpha1.Shard) error { + _, err := kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().UpdateStatus(ctx, shard, metav1.UpdateOptions{}) + return err + }, + deleteShard: func(ctx context.Context, name string) error { + return kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Delete(ctx, name, metav1.DeleteOptions{}) + }, + } + + _, _ = cacheShardInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { c.enqueue(obj) }, + UpdateFunc: func(_, obj interface{}) { c.enqueue(obj) }, + DeleteFunc: func(obj interface{}) { c.enqueue(obj) }, + }) + _, _ = localShardInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { c.enqueue(obj) }, + UpdateFunc: func(_, obj interface{}) { c.enqueue(obj) }, + DeleteFunc: func(obj interface{}) { c.enqueue(obj) }, + }) + + return c +} + +// Controller mirrors authoritative Shard objects into the root workspace. +type Controller struct { + queue workqueue.TypedRateLimitingInterface[string] + + getSourceShard func(name string) (*corev1alpha1.Shard, error) + getLocalShard func(name string) (*corev1alpha1.Shard, error) + + createShard func(ctx context.Context, shard *corev1alpha1.Shard) error + updateShard func(ctx context.Context, shard *corev1alpha1.Shard) error + updateShardStatus func(ctx context.Context, shard *corev1alpha1.Shard) error + deleteShard func(ctx context.Context, name string) error +} + +// enqueue maps any Shard event (authoritative copy in the cache, or local copy +// in the root workspace) to the shard name. +func (c *Controller) enqueue(obj interface{}) { + key, err := kcpcache.DeletionHandlingMetaClusterNamespaceKeyFunc(obj) + if err != nil { + utilruntime.HandleError(err) + return + } + clusterName, _, name, err := kcpcache.SplitMetaClusterNamespaceKey(key) + if err != nil { + utilruntime.HandleError(err) + return + } + // only authoritative copies and root-workspace representations are of interest. + if clusterName != configshard.SystemShardCluster && clusterName != core.RootCluster { + return + } + logger := logging.WithQueueKey(logging.WithReconciler(klog.Background(), ControllerName), name) + logger.V(4).Info("queueing Shard") + c.queue.Add(name) +} + +func (c *Controller) Start(ctx context.Context, numThreads int) { + defer utilruntime.HandleCrash() + defer c.queue.ShutDown() + + logger := logging.WithReconciler(klog.FromContext(ctx), ControllerName) + ctx = klog.NewContext(ctx, logger) + logger.Info("Starting controller") + defer logger.Info("Shutting down controller") + + for range numThreads { + go wait.UntilWithContext(ctx, c.startWorker, time.Second) + } + + <-ctx.Done() +} + +func (c *Controller) startWorker(ctx context.Context) { + for c.processNextWorkItem(ctx) { + } +} + +func (c *Controller) processNextWorkItem(ctx context.Context) bool { + name, quit := c.queue.Get() + if quit { + return false + } + defer c.queue.Done(name) + + if err := c.reconcile(ctx, name); err != nil { + utilruntime.HandleError(fmt.Errorf("%q controller failed to sync %q, err: %w", ControllerName, name, err)) + c.queue.AddRateLimited(name) + return true + } + c.queue.Forget(name) + return true +} + +func (c *Controller) reconcile(ctx context.Context, name string) error { + logger := klog.FromContext(ctx) + + source, err := c.getSourceShard(name) + if err != nil && !apierrors.IsNotFound(err) { + return err + } + sourceExists := !apierrors.IsNotFound(err) + + local, err := c.getLocalShard(name) + if err != nil && !apierrors.IsNotFound(err) { + return err + } + localExists := !apierrors.IsNotFound(err) + + if !sourceExists { + // prune representations whose authoritative object is gone. Leave + // shard objects not managed by this controller alone. + if localExists && isRepresentation(local) { + logger.V(2).Info("deleting Shard representation, authoritative object is gone", "shard", name) + if err := c.deleteShard(ctx, name); err != nil && !apierrors.IsNotFound(err) { + return err + } + } + return nil + } + + desired := representationFor(source) + + if !localExists { + logger.V(2).Info("creating Shard representation", "shard", name) + if err := c.createShard(ctx, desired); err != nil && !apierrors.IsAlreadyExists(err) { + return err + } + return nil + } + + updated := local.DeepCopy() + updated.Labels = desired.Labels + updated.Annotations = desired.Annotations + updated.Spec = desired.Spec + if !equality.Semantic.DeepEqual(local.Labels, updated.Labels) || + !equality.Semantic.DeepEqual(local.Annotations, updated.Annotations) || + !equality.Semantic.DeepEqual(local.Spec, updated.Spec) { + logger.V(2).Info("updating Shard representation", "shard", name) + if err := c.updateShard(ctx, updated); err != nil { + return err + } + return nil // status is synced on the next reconcile triggered by the update + } + + if !equality.Semantic.DeepEqual(local.Status, source.Status) { + updated.Status = source.Status + logger.V(2).Info("updating Shard representation status", "shard", name) + return c.updateShardStatus(ctx, updated) + } + + return nil +} + +func isRepresentation(shard *corev1alpha1.Shard) bool { + return shard.Annotations[corev1alpha1.ShardRepresentationAnnotationKey] != "" +} + +// representationFor builds the root-workspace representation of an +// authoritative Shard object from the cache server, stripping cache-server +// bookkeeping annotations. +func representationFor(source *corev1alpha1.Shard) *corev1alpha1.Shard { + shard := &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: source.Name, + Labels: source.Labels, + Annotations: map[string]string{}, + }, + Spec: source.Spec, + Status: source.Status, + } + for k, v := range source.Annotations { + shard.Annotations[k] = v + } + delete(shard.Annotations, logicalcluster.AnnotationKey) + delete(shard.Annotations, genericrequest.ShardAnnotationKey) + delete(shard.Annotations, replication.AnnotationKeyOriginalResourceVersion) + delete(shard.Annotations, replication.AnnotationKeyOriginalResourceUID) + shard.Annotations[corev1alpha1.ShardRepresentationAnnotationKey] = "true" + return shard +} diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go new file mode 100644 index 00000000000..7938019f84b --- /dev/null +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go @@ -0,0 +1,213 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package shardmirror + +import ( + "context" + "testing" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/kcp-dev/logicalcluster/v3" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" +) + +func authoritativeShard(name string) *corev1alpha1.Shard { + return &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Annotations: map[string]string{ + logicalcluster.AnnotationKey: "system:shard", + "kcp.io/shard": name, + "cache.kcp.io/original-resource-version": "42", + "cache.kcp.io/original-resource-UID": "abc", + }, + Labels: map[string]string{"name": name, "region": "us-east-1"}, + }, + Spec: corev1alpha1.ShardSpec{ + BaseURL: "https://" + name, + ExternalURL: "https://" + name, + VirtualWorkspaceURL: "https://" + name, + }, + } +} + +func representation(name string) *corev1alpha1.Shard { + r := representationFor(authoritativeShard(name)) + return r +} + +type fakeActions struct { + created, updated, statusUpdated *corev1alpha1.Shard + deleted string +} + +func newController(source, local *corev1alpha1.Shard, actions *fakeActions) *Controller { + return &Controller{ + getSourceShard: func(name string) (*corev1alpha1.Shard, error) { + if source == nil || source.Name != name { + return nil, apierrors.NewNotFound(corev1alpha1.Resource("shards"), name) + } + return source, nil + }, + getLocalShard: func(name string) (*corev1alpha1.Shard, error) { + if local == nil || local.Name != name { + return nil, apierrors.NewNotFound(corev1alpha1.Resource("shards"), name) + } + return local, nil + }, + createShard: func(_ context.Context, shard *corev1alpha1.Shard) error { + actions.created = shard + return nil + }, + updateShard: func(_ context.Context, shard *corev1alpha1.Shard) error { + actions.updated = shard + return nil + }, + updateShardStatus: func(_ context.Context, shard *corev1alpha1.Shard) error { + actions.statusUpdated = shard + return nil + }, + deleteShard: func(_ context.Context, name string) error { + actions.deleted = name + return nil + }, + } +} + +func TestReconcileCreatesRepresentation(t *testing.T) { + actions := &fakeActions{} + c := newController(authoritativeShard("alpha"), nil, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.created == nil { + t.Fatal("expected a representation to be created") + } + if actions.created.Annotations[corev1alpha1.ShardRepresentationAnnotationKey] == "" { + t.Error("representation must carry the representation annotation") + } + for _, key := range []string{ + logicalcluster.AnnotationKey, + "kcp.io/shard", + "cache.kcp.io/original-resource-version", + "cache.kcp.io/original-resource-UID", + } { + if _, ok := actions.created.Annotations[key]; ok { + t.Errorf("cache bookkeeping annotation %q must be stripped", key) + } + } + if actions.created.Labels["region"] != "us-east-1" { + t.Error("labels must be mirrored") + } + if actions.created.Spec.BaseURL != "https://alpha" { + t.Error("spec must be mirrored") + } +} + +func TestReconcileStompsManualEdit(t *testing.T) { + edited := representation("alpha") + edited.Spec.BaseURL = "https://tampered" + actions := &fakeActions{} + c := newController(authoritativeShard("alpha"), edited, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.updated == nil { + t.Fatal("expected the manual edit to be overwritten") + } + if actions.updated.Spec.BaseURL != "https://alpha" { + t.Errorf("expected spec to be restored, got %q", actions.updated.Spec.BaseURL) + } +} + +func TestReconcileAdoptsLegacyObject(t *testing.T) { + legacy := authoritativeShard("alpha").DeepCopy() // self-registered pre-upgrade, no representation annotation + legacy.Annotations = map[string]string{logicalcluster.AnnotationKey: "root"} + actions := &fakeActions{} + c := newController(authoritativeShard("alpha"), legacy, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.updated == nil { + t.Fatal("expected the legacy object to be adopted") + } + if actions.updated.Annotations[corev1alpha1.ShardRepresentationAnnotationKey] == "" { + t.Error("adopted object must carry the representation annotation") + } +} + +func TestReconcileInSync(t *testing.T) { + actions := &fakeActions{} + c := newController(authoritativeShard("alpha"), representation("alpha"), actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.created != nil || actions.updated != nil || actions.statusUpdated != nil || actions.deleted != "" { + t.Errorf("expected no action, got %+v", actions) + } +} + +func TestReconcileSyncsStatus(t *testing.T) { + source := authoritativeShard("alpha") + source.Status.Capacity = corev1.ResourceList{ + "workspaces": resource.MustParse("5"), + } + local := representation("alpha") + local.Status = corev1alpha1.ShardStatus{} + actions := &fakeActions{} + c := newController(source, local, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.updated != nil { + t.Error("meta/spec are in sync, only status should be updated") + } + if actions.statusUpdated == nil { + t.Fatal("expected a status update") + } + if actions.statusUpdated.Status.Capacity == nil { + t.Error("expected status.capacity to be mirrored") + } +} + +func TestReconcilePrunesOrphanedRepresentation(t *testing.T) { + actions := &fakeActions{} + c := newController(nil, representation("alpha"), actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.deleted != "alpha" { + t.Errorf("expected the orphaned representation to be deleted, got %q", actions.deleted) + } +} + +func TestReconcileLeavesUnmanagedObjectsAlone(t *testing.T) { + synthetic := authoritativeShard("fake") // e.g. created by an e2e test in the root workspace + synthetic.Annotations = map[string]string{logicalcluster.AnnotationKey: "root"} + actions := &fakeActions{} + c := newController(nil, synthetic, actions) + if err := c.reconcile(context.Background(), "fake"); err != nil { + t.Fatal(err) + } + if actions.deleted != "" || actions.updated != nil || actions.created != nil { + t.Errorf("expected no action on an unmanaged shard object, got %+v", actions) + } +} diff --git a/pkg/reconciler/migration/logicalclustermigration/datacopy.go b/pkg/reconciler/migration/logicalclustermigration/datacopy.go index 2008dec6788..4adb0fd8f7b 100644 --- a/pkg/reconciler/migration/logicalclustermigration/datacopy.go +++ b/pkg/reconciler/migration/logicalclustermigration/datacopy.go @@ -26,10 +26,10 @@ import ( "k8s.io/klog/v2" "github.com/kcp-dev/logicalcluster/v3" - "github.com/kcp-dev/sdk/apis/core" migrationv1alpha1 "github.com/kcp-dev/sdk/apis/migration/v1alpha1" kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" + "github.com/kcp-dev/kcp/pkg/indexers" "github.com/kcp-dev/kcp/pkg/virtual/migratingworkspaces" ) @@ -101,7 +101,7 @@ func (c *Controller) acquireOriginClient(lcName logicalcluster.Name, originShard entry, ok := c.originClients[originShardName] if !ok { - originShard, err := c.shardLister.Cluster(core.RootCluster).Get(originShardName) + originShard, err := indexers.ShardByName(c.shardLister, originShardName) if err != nil { return nil, fmt.Errorf("failed to get origin shard %q: %w", originShardName, err) } diff --git a/pkg/reconciler/tenancy/workspace/workspace_reconcile.go b/pkg/reconciler/tenancy/workspace/workspace_reconcile.go index 91ab54e1794..a26905616f3 100644 --- a/pkg/reconciler/tenancy/workspace/workspace_reconcile.go +++ b/pkg/reconciler/tenancy/workspace/workspace_reconcile.go @@ -26,7 +26,6 @@ import ( kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" "github.com/kcp-dev/client-go/kubernetes" "github.com/kcp-dev/logicalcluster/v3" - "github.com/kcp-dev/sdk/apis/core" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1" kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" @@ -48,7 +47,7 @@ type reconciler interface { func (c *Controller) reconcile(ctx context.Context, ws *tenancyv1alpha1.Workspace) (bool, error) { getShard := func(name string) (*corev1alpha1.Shard, error) { - return c.globalShardLister.Cluster(core.RootCluster).Get(name) + return indexers.ShardByName(c.globalShardLister, name) } kcpDirectClientFor := func(shard *corev1alpha1.Shard) (kcpclientset.ClusterInterface, error) { diff --git a/pkg/server/config.go b/pkg/server/config.go index a1620477ea8..7a646ed798c 100644 --- a/pkg/server/config.go +++ b/pkg/server/config.go @@ -60,7 +60,6 @@ import ( "github.com/kcp-dev/embeddedetcd" "github.com/kcp-dev/logicalcluster/v3" apisv1alpha2 "github.com/kcp-dev/sdk/apis/apis/v1alpha2" - "github.com/kcp-dev/sdk/apis/core" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1" kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" @@ -726,8 +725,7 @@ func NewConfig(ctx context.Context, opts kcpserveroptions.CompletedOptions) (*Co // labels itself with its name when it registers, so a selector can name a // shard without the installation having to label anything. thisShardLabels := func() labels.Set { - shard, err := c.CacheKcpSharedInformerFactory.Core().V1alpha1().Shards(). - Lister().Cluster(core.RootCluster).Get(opts.Extra.ShardName) + shard, err := indexers.ShardByName(c.CacheKcpSharedInformerFactory.Core().V1alpha1().Shards().Lister(), opts.Extra.ShardName) if err != nil { // Before the Shard object exists or syncs there is nothing to match // on. Selective endpoints will not match, which fails loudly, and diff --git a/pkg/server/controllers.go b/pkg/server/controllers.go index 81fe1256121..d33ce6c960a 100644 --- a/pkg/server/controllers.go +++ b/pkg/server/controllers.go @@ -96,6 +96,7 @@ import ( coresreplicateclusterrole "github.com/kcp-dev/kcp/pkg/reconciler/core/replicateclusterrole" corereplicateclusterrolebinding "github.com/kcp-dev/kcp/pkg/reconciler/core/replicateclusterrolebinding" "github.com/kcp-dev/kcp/pkg/reconciler/core/shard" + "github.com/kcp-dev/kcp/pkg/reconciler/core/shardmirror" "github.com/kcp-dev/kcp/pkg/reconciler/dynamicrestmapper" "github.com/kcp-dev/kcp/pkg/reconciler/garbagecollector" "github.com/kcp-dev/kcp/pkg/reconciler/kubequota" @@ -625,6 +626,28 @@ func (s *Server) installWorkspaceScheduler(ctx context.Context, config *rest.Con } } + if s.Options.Extra.ShardName == corev1alpha1.RootShard { + shardMirrorController := shardmirror.NewController( + kcpClusterClient, + s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), + s.CacheKcpSharedInformerFactory.Core().V1alpha1().Shards(), + ) + if err := s.registerController(&controllerWrapper{ + Name: shardmirror.ControllerName, + Wait: func(ctx context.Context, s *Server) error { + return wait.PollUntilContextCancel(ctx, waitPollInterval, true, func(ctx context.Context) (bool, error) { + return s.KcpSharedInformerFactory.Core().V1alpha1().Shards().Informer().HasSynced() && + s.CacheKcpSharedInformerFactory.Core().V1alpha1().Shards().Informer().HasSynced(), nil + }) + }, + Runner: func(ctx context.Context) { + shardMirrorController.Start(ctx, 2) + }, + }); err != nil { + return err + } + } + workspaceTypeConfig := rest.CopyConfig(config) workspaceTypeConfig = rest.AddUserAgent(workspaceTypeConfig, workspacetype.ControllerName) kcpClusterClient, err = kcpclientset.NewForConfig(workspaceTypeConfig) diff --git a/pkg/server/options/options.go b/pkg/server/options/options.go index 8ebacf0afe9..8dac7605c30 100644 --- a/pkg/server/options/options.go +++ b/pkg/server/options/options.go @@ -62,6 +62,7 @@ type ExtraOptions struct { ShardBaseURL string ShardExternalURL string ShardName string + ShardLabels map[string]string ShardVirtualWorkspaceURL string ShardClientCertFile string ShardClientKeyFile string @@ -183,6 +184,7 @@ func (o *Options) AddFlags(fss *cliflag.NamedFlagSets) { fs.StringVar(&o.Extra.ShardBaseURL, "shard-base-url", o.Extra.ShardBaseURL, "Base URL to this kcp shard. Defaults to external address.") fs.StringVar(&o.Extra.ShardExternalURL, "shard-external-url", o.Extra.ShardExternalURL, "URL used by outside clients to talk to this kcp shard. Defaults to external address.") fs.StringVar(&o.Extra.ShardName, "shard-name", o.Extra.ShardName, "A name of this kcp shard. Defaults to the \"root\" name.") + fs.StringToStringVar(&o.Extra.ShardLabels, "shard-labels", o.Extra.ShardLabels, "Additional labels to set on this shard's Shard object, e.g. region=us-east-1. The Shard object is owned by the shard; labels are part of its configuration.") fs.StringVar(&o.Extra.ShardVirtualWorkspaceCAFile, "shard-virtual-workspace-ca-file", o.Extra.ShardVirtualWorkspaceCAFile, "Path to a CA certificate file that is valid for the virtual workspace server.") fs.StringVar(&o.Extra.ShardVirtualWorkspaceURL, "shard-virtual-workspace-url", o.Extra.ShardVirtualWorkspaceURL, "An external URL address of a virtual workspace server associated with this shard. Defaults to shard's base address.") fs.StringVar(&o.Extra.ShardClientCertFile, "shard-client-cert-file", o.Extra.ShardClientCertFile, "Path to a client certificate file the shard uses to communicate with other system components.") diff --git a/pkg/server/server.go b/pkg/server/server.go index 9e2e7fc9f64..e93b60e33d5 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -622,14 +622,22 @@ func (s *Server) Run(ctx context.Context) error { s.KcpSharedInformerFactory.WaitForCacheSync(hookCtx.Done()) s.CacheKcpSharedInformerFactory.WaitForCacheSync(hookCtx.Done()) - // create or update shard + // create or update the shard-owned Shard object in the local + // system:shard logical cluster. It is replicated to the cache server + // from there and mirrored into the root workspace as a read-only + // representation, so shard startup does not depend on the root shard + // being reachable. + labels := map[string]string{ + "name": s.Options.Extra.ShardName, + } + for k, v := range s.Options.Extra.ShardLabels { + labels[k] = v + } shard := &corev1alpha1.Shard{ ObjectMeta: metav1.ObjectMeta{ Name: s.Options.Extra.ShardName, - Annotations: map[string]string{logicalcluster.AnnotationKey: core.RootCluster.String()}, - Labels: map[string]string{ - "name": s.Options.Extra.ShardName, - }, + Annotations: map[string]string{logicalcluster.AnnotationKey: configshard.SystemShardCluster.String()}, + Labels: labels, }, Spec: corev1alpha1.ShardSpec{ BaseURL: s.CompletedConfig.ShardBaseURL(), @@ -639,29 +647,30 @@ func (s *Server) Run(ctx context.Context) error { } logger.Info("Creating or updating Shard", "shard", s.Options.Extra.ShardName) if err := wait.PollUntilContextCancel(hookCtx, time.Second, true, func(ctx context.Context) (bool, error) { - existingShard, err := s.RootShardKcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Get(ctx, shard.Name, metav1.GetOptions{}) + existingShard, err := s.KcpClusterClient.Cluster(configshard.SystemShardCluster.Path()).CoreV1alpha1().Shards().Get(ctx, shard.Name, metav1.GetOptions{}) if err != nil && !errors.IsNotFound(err) { - logger.Error(err, "failed getting Shard from the root workspace") + logger.Error(err, "failed getting Shard from the system:shard logical cluster") return false, nil } else if errors.IsNotFound(err) { - if _, err := s.RootShardKcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Create(ctx, shard, metav1.CreateOptions{}); err != nil { - logger.Error(err, "failed creating Shard in the root workspace") + if _, err := s.KcpClusterClient.Cluster(configshard.SystemShardCluster.Path()).CoreV1alpha1().Shards().Create(ctx, shard, metav1.CreateOptions{}); err != nil { + logger.Error(err, "failed creating Shard in the system:shard logical cluster") return false, nil } logger.Info("Created Shard", "shard", s.Options.Extra.ShardName) return true, nil } + existingShard.Labels = shard.Labels existingShard.Spec.BaseURL = shard.Spec.BaseURL existingShard.Spec.ExternalURL = shard.Spec.ExternalURL existingShard.Spec.VirtualWorkspaceURL = shard.Spec.VirtualWorkspaceURL - if _, err := s.RootShardKcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Update(hookCtx, existingShard, metav1.UpdateOptions{}); err != nil { - logger.Error(err, "failed updating Shard in the root workspace") + if _, err := s.KcpClusterClient.Cluster(configshard.SystemShardCluster.Path()).CoreV1alpha1().Shards().Update(hookCtx, existingShard, metav1.UpdateOptions{}); err != nil { + logger.Error(err, "failed updating Shard in the system:shard logical cluster") return false, nil } logger.Info("Updated Shard", "shard", s.Options.Extra.ShardName) return true, nil }); err != nil { - logger.Error(err, "failed reconciling Shard resource in the root workspace") + logger.Error(err, "failed reconciling Shard resource in the system:shard logical cluster") return nil // don't klog.Fatal. This only happens when context is cancelled. } diff --git a/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go b/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go index 70fe8dd16fd..b5bd7961741 100644 --- a/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go +++ b/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go @@ -27,6 +27,14 @@ import ( // RootShard holds a name of the root shard. var RootShard = "root" +// ShardRepresentationAnnotationKey marks a Shard object as a read-only +// representation mirrored from the shard-owned authoritative object living in +// the shard's local system:shard logical cluster. Representations exist for +// discoverability (e.g. in the root workspace) and must not be edited; they +// are continuously overwritten from the authoritative object and are excluded +// from cache replication. +const ShardRepresentationAnnotationKey = "core.kcp.io/shard-representation" + // Shard describes a kcp instance on which a number of logical clusters will live // // +crd From 5d203e5ee8d7b53ed6843b028ee6a82e618757c2 Mon Sep 17 00:00:00 2001 From: Mangirdas Judeikis Date: Fri, 21 Aug 2026 11:56:58 +0300 Subject: [PATCH 2/5] lint --- pkg/indexers/shard.go | 3 ++- pkg/indexers/shard_test.go | 1 + pkg/reconciler/core/shardmirror/shardmirror_controller.go | 6 ++---- .../core/shardmirror/shardmirror_controller_test.go | 7 +++++++ 4 files changed, 12 insertions(+), 5 deletions(-) diff --git a/pkg/indexers/shard.go b/pkg/indexers/shard.go index bfadc58190c..c010a56c579 100644 --- a/pkg/indexers/shard.go +++ b/pkg/indexers/shard.go @@ -20,10 +20,11 @@ import ( "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/labels" - configshard "github.com/kcp-dev/kcp/config/shard" "github.com/kcp-dev/logicalcluster/v3" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" corev1alpha1listers "github.com/kcp-dev/sdk/client/listers/core/v1alpha1" + + configshard "github.com/kcp-dev/kcp/config/shard" ) // ShardByName returns the Shard with the given name, regardless of the logical diff --git a/pkg/indexers/shard_test.go b/pkg/indexers/shard_test.go index 9943e46d4a1..854b568375f 100644 --- a/pkg/indexers/shard_test.go +++ b/pkg/indexers/shard_test.go @@ -39,6 +39,7 @@ func shardInCluster(name, cluster string) *corev1alpha1.Shard { } func TestShardByName(t *testing.T) { + t.Parallel() indexer := cache.NewIndexer(kcpcache.MetaClusterNamespaceKeyFunc, cache.Indexers{}) for _, shard := range []*corev1alpha1.Shard{ shardInCluster("alpha", "system:shard"), diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller.go b/pkg/reconciler/core/shardmirror/shardmirror_controller.go index 25665c3389a..5cb9404e95f 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller.go @@ -223,10 +223,8 @@ func (c *Controller) reconcile(ctx context.Context, name string) error { !equality.Semantic.DeepEqual(local.Annotations, updated.Annotations) || !equality.Semantic.DeepEqual(local.Spec, updated.Spec) { logger.V(2).Info("updating Shard representation", "shard", name) - if err := c.updateShard(ctx, updated); err != nil { - return err - } - return nil // status is synced on the next reconcile triggered by the update + // status is synced on the next reconcile triggered by the update + return c.updateShard(ctx, updated) } if !equality.Semantic.DeepEqual(local.Status, source.Status) { diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go index 7938019f84b..c026e3c3d32 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go @@ -93,6 +93,7 @@ func newController(source, local *corev1alpha1.Shard, actions *fakeActions) *Con } func TestReconcileCreatesRepresentation(t *testing.T) { + t.Parallel() actions := &fakeActions{} c := newController(authoritativeShard("alpha"), nil, actions) if err := c.reconcile(context.Background(), "alpha"); err != nil { @@ -123,6 +124,7 @@ func TestReconcileCreatesRepresentation(t *testing.T) { } func TestReconcileStompsManualEdit(t *testing.T) { + t.Parallel() edited := representation("alpha") edited.Spec.BaseURL = "https://tampered" actions := &fakeActions{} @@ -139,6 +141,7 @@ func TestReconcileStompsManualEdit(t *testing.T) { } func TestReconcileAdoptsLegacyObject(t *testing.T) { + t.Parallel() legacy := authoritativeShard("alpha").DeepCopy() // self-registered pre-upgrade, no representation annotation legacy.Annotations = map[string]string{logicalcluster.AnnotationKey: "root"} actions := &fakeActions{} @@ -155,6 +158,7 @@ func TestReconcileAdoptsLegacyObject(t *testing.T) { } func TestReconcileInSync(t *testing.T) { + t.Parallel() actions := &fakeActions{} c := newController(authoritativeShard("alpha"), representation("alpha"), actions) if err := c.reconcile(context.Background(), "alpha"); err != nil { @@ -166,6 +170,7 @@ func TestReconcileInSync(t *testing.T) { } func TestReconcileSyncsStatus(t *testing.T) { + t.Parallel() source := authoritativeShard("alpha") source.Status.Capacity = corev1.ResourceList{ "workspaces": resource.MustParse("5"), @@ -189,6 +194,7 @@ func TestReconcileSyncsStatus(t *testing.T) { } func TestReconcilePrunesOrphanedRepresentation(t *testing.T) { + t.Parallel() actions := &fakeActions{} c := newController(nil, representation("alpha"), actions) if err := c.reconcile(context.Background(), "alpha"); err != nil { @@ -200,6 +206,7 @@ func TestReconcilePrunesOrphanedRepresentation(t *testing.T) { } func TestReconcileLeavesUnmanagedObjectsAlone(t *testing.T) { + t.Parallel() synthetic := authoritativeShard("fake") // e.g. created by an e2e test in the root workspace synthetic.Annotations = map[string]string{logicalcluster.AnnotationKey: "root"} actions := &fakeActions{} From 0c51198cd73889b3542c6a70fb1a37591c88ccbf Mon Sep 17 00:00:00 2001 From: Mangirdas Judeikis Date: Fri, 21 Aug 2026 13:37:07 +0300 Subject: [PATCH 3/5] Fix test --- docs/content/concepts/sharding/shards.md | 5 ++++- .../reconciler/workspace/controller_test.go | 19 ++++++++++++++----- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/docs/content/concepts/sharding/shards.md b/docs/content/concepts/sharding/shards.md index fcf06fe48db..1c4ac098148 100644 --- a/docs/content/concepts/sharding/shards.md +++ b/docs/content/concepts/sharding/shards.md @@ -264,7 +264,10 @@ inferred only from request errors at the moment of access. Concretely: through the front-proxy until it returns. - **Workspace scheduling** continues to succeed for new workspaces as long as at least one valid shard is reachable; the random selector simply skips - shards annotated `kcp.io/unschedulable`. + shards annotated `experimental.core.kcp.io/unschedulable` (set on the + authoritative Shard object in the shard's `system:shard` logical cluster). + For now there is no way to mark shard unshedulable through the API, but + this is tracked in [kcp-dev/kcp#4336](https://github.com/kcp-dev/kcp/issues/4336) - **Bound APIs** on a workspace whose `APIExport` lives on an unavailable shard keep functioning: the CRD is materialized locally and the binding controller already has the schema cached. New bindings or schema updates diff --git a/test/e2e/reconciler/workspace/controller_test.go b/test/e2e/reconciler/workspace/controller_test.go index 9e585246131..3e4c79a7819 100644 --- a/test/e2e/reconciler/workspace/controller_test.go +++ b/test/e2e/reconciler/workspace/controller_test.go @@ -41,6 +41,7 @@ import ( kcptestinghelpers "github.com/kcp-dev/sdk/testing/helpers" kcptestingserver "github.com/kcp-dev/sdk/testing/server" + configshard "github.com/kcp-dev/kcp/config/shard" "github.com/kcp-dev/kcp/test/e2e/framework" ) @@ -51,6 +52,10 @@ func TestWorkspaceController(t *testing.T) { type runningServer struct { kcptestingserver.RunningServer rootWorkspaceKcpClient, orgWorkspaceKcpClient kcpclientset.Interface + // systemShardKcpClient is a privileged client to the root shard's + // system:shard logical cluster, where the shard-owned authoritative + // Shard object lives. + systemShardKcpClient kcpclientset.Interface } var testCases = []struct { name string @@ -97,8 +102,8 @@ func TestWorkspaceController(t *testing.T) { work: func(ctx context.Context, t *testing.T, server runningServer) { t.Helper() - t.Logf("Get the root shard") - shard, err := server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + t.Logf("Get the root shard's authoritative Shard object from system:shard") + shard, err := server.systemShardKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) require.NoError(t, err) t.Logf("Mark the root shard as unschedulable") @@ -106,7 +111,7 @@ func TestWorkspaceController(t *testing.T) { shard.Annotations = map[string]string{} } shard.Annotations["experimental.core.kcp.io/unschedulable"] = "true" - _, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) + _, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) require.NoError(t, err) // Workspaces cannot become "unschedulable" - once they @@ -157,10 +162,10 @@ func TestWorkspaceController(t *testing.T) { }) t.Logf("Remove unschedulable annotation from the root shard") - shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + shard, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) require.NoError(t, err) delete(shard.Annotations, "experimental.core.kcp.io/unschedulable") - shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) + shard, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) require.NoError(t, err) t.Logf("Expect workspace to be scheduled to the shard and show the external URL") @@ -200,10 +205,14 @@ func TestWorkspaceController(t *testing.T) { kcpClient, err := kcpclusterclientset.NewForConfig(cfg) require.NoError(t, err) + systemShardClient, err := kcpclusterclientset.NewForConfig(server.RootShardSystemMasterBaseConfig(t)) + require.NoError(t, err) + testCase.work(ctx, t, runningServer{ RunningServer: server, rootWorkspaceKcpClient: kcpClient.Cluster(core.RootCluster.Path()), orgWorkspaceKcpClient: kcpClient.Cluster(orgPath), + systemShardKcpClient: systemShardClient.Cluster(configshard.SystemShardCluster.Path()), }) }) } From 6f00c2057fb99bc8b9cfa4596f717da6f3fcfbfc Mon Sep 17 00:00:00 2001 From: Mangirdas Judeikis Date: Fri, 21 Aug 2026 14:14:03 +0300 Subject: [PATCH 4/5] Adjust another test. --- cmd/sharded-test-server/main.go | 47 +++++++++++++++++++ .../shardmirror/shardmirror_controller.go | 6 +++ .../shardmirror_controller_test.go | 3 ++ test/e2e/apibinding/apibinding_test.go | 9 +++- test/e2e/cache/replication_api_cache_test.go | 5 +- test/e2e/reconciler/cache/replication_test.go | 5 +- 6 files changed, 71 insertions(+), 4 deletions(-) diff --git a/cmd/sharded-test-server/main.go b/cmd/sharded-test-server/main.go index 1304f961efc..3f46728892c 100644 --- a/cmd/sharded-test-server/main.go +++ b/cmd/sharded-test-server/main.go @@ -26,12 +26,16 @@ import ( "strings" "time" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" kuser "k8s.io/apiserver/pkg/authentication/user" genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/client-go/tools/clientcmd" "k8s.io/klog/v2" + "github.com/kcp-dev/sdk/apis/core" + kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" "github.com/kcp-dev/sdk/testing/third_party/library-go/crypto" testshard "github.com/kcp-dev/kcp/cmd/test-server/kcp" @@ -310,6 +314,32 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb return err } + // Wait for all shards' Shard objects to be mirrored into the root + // workspace as representations before declaring the environment ready, so + // tests can rely on listing Shards there. + clientConfig, err := loadKubeConfig(filepath.Join(workDirPath, ".kcp", "admin.kubeconfig"), "base") + if err != nil { + return err + } + config, err := clientConfig.ClientConfig() + if err != nil { + return err + } + client, err := kcpclientset.NewForConfig(config) + if err != nil { + return err + } + if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 2*time.Minute, true, func(ctx context.Context) (bool, error) { + shardList, err := client.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().List(ctx, metav1.ListOptions{}) + if err != nil { + // the front-proxy or the root shard may not be fully ready yet, keep polling + return false, nil //nolint:nilerr + } + return len(shardList.Items) >= numberOfShards, nil + }); err != nil { + return fmt.Errorf("failed waiting for %d Shard representations in the root workspace: %w", numberOfShards, err) + } + readyToTestFile, err := os.Create(filepath.Join(workDirPath, ".kcp", "ready-to-test")) if err != nil { return fmt.Errorf("error creating ready-to-test file: %w", err) @@ -343,6 +373,23 @@ type indexErrTuple struct { error error } +func loadKubeConfig(kubeconfigPath, contextName string) (clientcmd.ClientConfig, error) { + fs, err := os.Stat(kubeconfigPath) + if err != nil { + return nil, err + } + if fs.Size() == 0 { + return nil, fmt.Errorf("%s points to an empty file", kubeconfigPath) + } + + rawConfig, err := clientcmd.LoadFromFile(kubeconfigPath) + if err != nil { + return nil, fmt.Errorf("failed to load admin kubeconfig: %w", err) + } + + return clientcmd.NewNonInteractiveClientConfig(*rawConfig, contextName, nil, nil), nil +} + var regions = []string{ "us-east-2", "us-east-1", diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller.go b/pkg/reconciler/core/shardmirror/shardmirror_controller.go index 5cb9404e95f..5330cd33024 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller.go @@ -218,6 +218,12 @@ func (c *Controller) reconcile(ctx context.Context, name string) error { updated := local.DeepCopy() updated.Labels = desired.Labels updated.Annotations = desired.Annotations + // the logical cluster annotation is stamped onto stored objects by the + // server; keep it so the comparison below converges instead of updating + // on every resync. + if cluster, ok := local.Annotations[logicalcluster.AnnotationKey]; ok { + updated.Annotations[logicalcluster.AnnotationKey] = cluster + } updated.Spec = desired.Spec if !equality.Semantic.DeepEqual(local.Labels, updated.Labels) || !equality.Semantic.DeepEqual(local.Annotations, updated.Annotations) || diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go index c026e3c3d32..6d64a371a0e 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go @@ -51,6 +51,9 @@ func authoritativeShard(name string) *corev1alpha1.Shard { func representation(name string) *corev1alpha1.Shard { r := representationFor(authoritativeShard(name)) + // stored objects carry the logical cluster annotation stamped by the + // server; mimic that so tests catch non-converging comparisons. + r.Annotations[logicalcluster.AnnotationKey] = "root" return r } diff --git a/test/e2e/apibinding/apibinding_test.go b/test/e2e/apibinding/apibinding_test.go index 34203b47b5e..13acf9a09e1 100644 --- a/test/e2e/apibinding/apibinding_test.go +++ b/test/e2e/apibinding/apibinding_test.go @@ -170,8 +170,13 @@ func TestAPIBinding(t *testing.T) { kcpClusterClient, err := kcpclientset.NewForConfig(cfg) require.NoError(t, err, "failed to construct kcp cluster client for server") - shards, err = kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().List(t.Context(), metav1.ListOptions{}) - require.NoError(t, err, "failed to list shards") + // The Shard objects in the root workspace are representations mirrored + // from the cache server and appear asynchronously after shard startup. + require.Eventually(t, func() bool { + shards, err = kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().List(t.Context(), metav1.ListOptions{}) + require.NoError(t, err, "failed to list shards") + return len(shards.Items) > 0 + }, wait.ForeverTestTimeout, 100*time.Millisecond, "expected at least one Shard in the root workspace") } t.Logf("Shards: %v", shards.Items) diff --git a/test/e2e/cache/replication_api_cache_test.go b/test/e2e/cache/replication_api_cache_test.go index c18dc727fac..2cd339959aa 100644 --- a/test/e2e/cache/replication_api_cache_test.go +++ b/test/e2e/cache/replication_api_cache_test.go @@ -382,9 +382,12 @@ func createCacheClientConfigForEnvironment(ctx context.Context, t *testing.T, kc require.NoError(t, err) shards, err := kcpRootShardClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().List(t.Context(), metav1.ListOptions{}) require.NoError(t, err) - if len(shards.Items) == 1 { + if len(shards.Items) <= 1 { // Single shard with embedded cache server — use the loopback // bearer token which the embedded cache already trusts. + // Note that the count can be 0 right after startup: the Shard object + // in the root workspace is a representation mirrored from the cache + // server and appears asynchronously. return kcpRootShardConfig } diff --git a/test/e2e/reconciler/cache/replication_test.go b/test/e2e/reconciler/cache/replication_test.go index d6474873106..1bc2f109bbe 100644 --- a/test/e2e/reconciler/cache/replication_test.go +++ b/test/e2e/reconciler/cache/replication_test.go @@ -759,9 +759,12 @@ func createCacheClientConfigForEnvironment(ctx context.Context, t *testing.T, kc require.NoError(t, err) shards, err := kcpRootShardClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().List(t.Context(), metav1.ListOptions{}) require.NoError(t, err) - if len(shards.Items) == 1 { + if len(shards.Items) <= 1 { // Single shard with embedded cache server — use the loopback // bearer token which the embedded cache already trusts. + // Note that the count can be 0 right after startup: the Shard object + // in the root workspace is a representation mirrored from the cache + // server and appears asynchronously. return kcpRootShardConfig } From c93e32cb93b923b5b3302f9080133534568de750 Mon Sep 17 00:00:00 2001 From: Mangirdas Judeikis Date: Mon, 31 Aug 2026 16:22:03 +0300 Subject: [PATCH 5/5] Add BackSyncer for cordoning --- config/crds/core.kcp.io_shards.yaml | 4 + .../apiexport-shards.core.kcp.io.yaml | 2 +- .../apiresourceschema-shards.core.kcp.io.yaml | 6 +- docs/content/concepts/sharding/shards.md | 17 ++- pkg/authorization/bootstrap/policy.go | 4 + .../workspace_content_authorizer.go | 20 ++- pkg/reconciler/core/shard/shard_controller.go | 50 +++++- .../core/shard/shard_controller_test.go | 77 ++++++++++ .../shardmirror/shardmirror_controller.go | 124 +++++++++++++++ .../shardmirror_controller_test.go | 60 ++++++++ .../workspace_reconcile_scheduling.go | 8 +- .../workspace_reconcile_scheduling_test.go | 2 +- pkg/server/controllers.go | 49 +++--- .../cli/cmd/kubectl-kcp/cmd/kubectlKcp.go | 4 + .../kcp-dev/cli/pkg/shard/cmd/cmd.go | 97 ++++++++++++ .../kcp-dev/cli/pkg/shard/plugin/cordon.go | 142 ++++++++++++++++++ .../cli/pkg/shard/plugin/cordon_test.go | 121 +++++++++++++++ .../sdk/apis/core/v1alpha1/shard_types.go | 31 +++- .../reconciler/workspace/controller_test.go | 63 ++++---- 19 files changed, 805 insertions(+), 76 deletions(-) create mode 100644 pkg/reconciler/core/shard/shard_controller_test.go create mode 100644 staging/src/github.com/kcp-dev/cli/pkg/shard/cmd/cmd.go create mode 100644 staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon.go create mode 100644 staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon_test.go diff --git a/config/crds/core.kcp.io_shards.yaml b/config/crds/core.kcp.io_shards.yaml index dc4744e3ae0..90b9896ae46 100644 --- a/config/crds/core.kcp.io_shards.yaml +++ b/config/crds/core.kcp.io_shards.yaml @@ -29,6 +29,10 @@ spec: jsonPath: .spec.externalURL name: External URL type: string + - description: Whether new workspaces are scheduled onto this shard + jsonPath: .status.conditions[?(@.type=="Schedulable")].status + name: Schedulable + type: string - jsonPath: .metadata.creationTimestamp name: Age type: date diff --git a/config/root-phase0/apiexport-shards.core.kcp.io.yaml b/config/root-phase0/apiexport-shards.core.kcp.io.yaml index 212b4b97b6b..1b22329224d 100644 --- a/config/root-phase0/apiexport-shards.core.kcp.io.yaml +++ b/config/root-phase0/apiexport-shards.core.kcp.io.yaml @@ -6,7 +6,7 @@ spec: resources: - group: core.kcp.io name: shards - schema: v260302-ae9fdd9c1.shards.core.kcp.io + schema: v260831-6f00c2057.shards.core.kcp.io storage: crd: {} status: {} diff --git a/config/root-phase0/apiresourceschema-shards.core.kcp.io.yaml b/config/root-phase0/apiresourceschema-shards.core.kcp.io.yaml index 3b7a7cf26ea..bbfc26e2c16 100644 --- a/config/root-phase0/apiresourceschema-shards.core.kcp.io.yaml +++ b/config/root-phase0/apiresourceschema-shards.core.kcp.io.yaml @@ -1,7 +1,7 @@ apiVersion: apis.kcp.io/v1alpha1 kind: APIResourceSchema metadata: - name: v260302-ae9fdd9c1.shards.core.kcp.io + name: v260831-6f00c2057.shards.core.kcp.io spec: group: core.kcp.io names: @@ -26,6 +26,10 @@ spec: jsonPath: .spec.externalURL name: External URL type: string + - description: Whether new workspaces are scheduled onto this shard + jsonPath: .status.conditions[?(@.type=="Schedulable")].status + name: Schedulable + type: string - jsonPath: .metadata.creationTimestamp name: Age type: date diff --git a/docs/content/concepts/sharding/shards.md b/docs/content/concepts/sharding/shards.md index 1c4ac098148..02999579b27 100644 --- a/docs/content/concepts/sharding/shards.md +++ b/docs/content/concepts/sharding/shards.md @@ -264,10 +264,19 @@ inferred only from request errors at the moment of access. Concretely: through the front-proxy until it returns. - **Workspace scheduling** continues to succeed for new workspaces as long as at least one valid shard is reachable; the random selector simply skips - shards annotated `experimental.core.kcp.io/unschedulable` (set on the - authoritative Shard object in the shard's `system:shard` logical cluster). - For now there is no way to mark shard unshedulable through the API, but - this is tracked in [kcp-dev/kcp#4336](https://github.com/kcp-dev/kcp/issues/4336) + shards annotated `experimental.core.kcp.io/unschedulable`. To cordon a + shard, set the annotation on its `Shard` representation in the `root` + workspace: the annotation is on a small allow-list that the shard mirror + syncs back onto the authoritative `Shard` object in the shard's + `system:shard` logical cluster (all other edits to a representation are + overwritten). Removing the annotation from the representation uncordons the + shard the same way. The kcp kubectl plugin wraps this, mirroring + `kubectl cordon`/`uncordon` for nodes: `kubectl kcp shard cordon ` + and `kubectl kcp shard uncordon `. The owning shard acknowledges the signal node-style via + the `Schedulable` status condition on its `Shard` object — `False` with + reason `Cordoned` while cordoned, `True` otherwise — which is mirrored back + to the representation, so `kubectl get shard -o yaml` in `root` shows + whether the shard has actually applied the cordon. - **Bound APIs** on a workspace whose `APIExport` lives on an unavailable shard keep functioning: the CRD is materialized locally and the binding controller already has the schema cached. New bindings or schema updates diff --git a/pkg/authorization/bootstrap/policy.go b/pkg/authorization/bootstrap/policy.go index f43307149dd..2f23f390b0c 100644 --- a/pkg/authorization/bootstrap/policy.go +++ b/pkg/authorization/bootstrap/policy.go @@ -108,6 +108,10 @@ func clusterRoles() []rbacv1.ClusterRole { Rules: []rbacv1.PolicyRule{ rbacv1helpers.NewRule("*").Groups(core.GroupName).Resources("logicalclusters", "logicalclusters/status").RuleOrDie(), rbacv1helpers.NewRule("delete", "update", "get").Groups(tenancy.GroupName).Resources("workspaces").RuleOrDie(), + // the shard mirror on the root shard back-syncs allow-listed + // annotations (e.g. cordoning) onto the shard-owned Shard + // object in each shard's system:shard logical cluster. + rbacv1helpers.NewRule("get", "update", "patch").Groups(core.GroupName).Resources("shards").RuleOrDie(), }, }, { diff --git a/pkg/authorization/workspace_content_authorizer.go b/pkg/authorization/workspace_content_authorizer.go index b9883756a5c..dcd47bdff3f 100644 --- a/pkg/authorization/workspace_content_authorizer.go +++ b/pkg/authorization/workspace_content_authorizer.go @@ -91,10 +91,22 @@ func (a *workspaceContentAuthorizer) Authorize(ctx context.Context, attr authori return DelegateAuthorization("shard-level path", a.delegate).Authorize(ctx, attr) } - // empty or system workspaces have no meaning in the context of authorizing workspace content. - // To access system workspaces, the user must be privileged such that authorization is skipped completely. - if cluster == nil || cluster.Name.Empty() || strings.HasPrefix(cluster.Name.String(), "system:") { - return authorizer.DecisionNoOpinion, "empty or system workspace", nil + // empty workspaces have no meaning in the context of authorizing workspace content. + if cluster == nil || cluster.Name.Empty() { + return authorizer.DecisionNoOpinion, "empty workspace", nil + } + + // System workspaces have no meaning in the context of authorizing workspace + // content either. To access them, the user must be privileged such that + // authorization is skipped completely, with one exception: the + // logical-cluster-admin identity used for direct shard-to-shard requests + // (e.g. the shard mirror back-syncing cordon annotations onto the + // shard-owned Shard object in system:shard) is delegated to RBAC. + if strings.HasPrefix(cluster.Name.String(), "system:") { + if rbacregistryvalidation.EffectiveGroups(ctx, attr.GetUser()).Has(bootstrap.SystemLogicalClusterAdmin) { + return DelegateAuthorization("logical cluster admin access to system workspace", a.delegate).Authorize(ctx, attr) + } + return authorizer.DecisionNoOpinion, "system workspace", nil } isServiceAccount := rbacregistryvalidation.IsServiceAccount(attr.GetUser()) diff --git a/pkg/reconciler/core/shard/shard_controller.go b/pkg/reconciler/core/shard/shard_controller.go index c4501d14d13..7d90382b2f4 100644 --- a/pkg/reconciler/core/shard/shard_controller.go +++ b/pkg/reconciler/core/shard/shard_controller.go @@ -14,6 +14,13 @@ See the License for the specific language governing permissions and limitations under the License. */ +// Package shard runs on every shard and maintains the status of the shard's +// own authoritative Shard object in the local system:shard logical cluster. +// It mimics a Kubernetes node reporting its state: currently it keeps the +// Schedulable condition in sync with the unschedulable (cordon) annotation, +// acknowledging that the shard observed and applied the signal. The status +// replicates to the cache server and is mirrored onto the shard's +// representation in the root workspace, where admins can see the ack. package shard import ( @@ -32,10 +39,13 @@ import ( kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache" "github.com/kcp-dev/logicalcluster/v3" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + conditionsv1alpha1 "github.com/kcp-dev/sdk/apis/third_party/conditions/apis/conditions/v1alpha1" + "github.com/kcp-dev/sdk/apis/third_party/conditions/util/conditions" kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" corev1alpha1client "github.com/kcp-dev/sdk/client/clientset/versioned/typed/core/v1alpha1" corev1alpha1informers "github.com/kcp-dev/sdk/client/informers/externalversions/core/v1alpha1" + configshard "github.com/kcp-dev/kcp/config/shard" "github.com/kcp-dev/kcp/pkg/logging" "github.com/kcp-dev/kcp/pkg/reconciler/committer" ) @@ -45,7 +55,8 @@ const ( ) func NewController( - rootKcpClient kcpclientset.ClusterInterface, + shardName string, + kcpClient kcpclientset.ClusterInterface, shardInformer corev1alpha1informers.ShardClusterInformer, ) (*Controller, error) { c := &Controller{ @@ -55,8 +66,9 @@ func NewController( Name: ControllerName, }, ), - kcpClient: rootKcpClient, - commit: committer.NewCommitter[*Shard, Patcher, *ShardSpec, *ShardStatus](rootKcpClient.CoreV1alpha1().Shards()), + shardName: shardName, + kcpClient: kcpClient, + commit: committer.NewCommitter[*Shard, Patcher, *ShardSpec, *ShardStatus](kcpClient.CoreV1alpha1().Shards()), getShard: func(clusterName logicalcluster.Name, name string) (*corev1alpha1.Shard, error) { return shardInformer.Cluster(clusterName).Lister().Get(name) }, @@ -70,11 +82,13 @@ func NewController( return c, nil } -// Controller watches WorkspaceShards and Secrets in order to make sure every Shard -// has its URL exposed when a valid kubeconfig is connected to it. +// Controller maintains the status of this shard's own authoritative Shard +// object in the local system:shard logical cluster, e.g. the Schedulable +// condition acknowledging cordon/uncordon signals. type Controller struct { queue workqueue.TypedRateLimitingInterface[string] + shardName string kcpClient kcpclientset.ClusterInterface getShard func(clusterName logicalcluster.Name, name string) (*corev1alpha1.Shard, error) @@ -94,6 +108,16 @@ func (c *Controller) enqueue(obj interface{}) { utilruntime.HandleError(err) return } + clusterName, _, name, err := kcpcache.SplitMetaClusterNamespaceKey(key) + if err != nil { + utilruntime.HandleError(err) + return + } + // only this shard's own authoritative object is of interest; leave + // representations and legacy objects in other logical clusters alone. + if clusterName != configshard.SystemShardCluster || name != c.shardName { + return + } logger := logging.WithQueueKey(logging.WithReconciler(klog.Background(), ControllerName), key) logger.V(4).Info("queueing Shard") c.queue.Add(key) @@ -182,6 +206,20 @@ func (c *Controller) process(ctx context.Context, key string) error { return utilerrors.NewAggregate(errs) } -func (c *Controller) reconcile(ctx context.Context, workspaceShard *corev1alpha1.Shard) error { +// reconcile keeps the Schedulable condition in sync with the cordon +// annotation, acknowledging that this shard observed and applied the signal. +func (c *Controller) reconcile(_ context.Context, shard *corev1alpha1.Shard) error { + if _, cordoned := shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; cordoned { + conditions.MarkFalse( + shard, + corev1alpha1.ShardSchedulable, + corev1alpha1.ShardReasonCordoned, + conditionsv1alpha1.ConditionSeverityInfo, + "shard is cordoned via the %s annotation, no new workspaces are scheduled onto it", + corev1alpha1.ShardUnschedulableAnnotationKey, + ) + } else { + conditions.MarkTrue(shard, corev1alpha1.ShardSchedulable) + } return nil } diff --git a/pkg/reconciler/core/shard/shard_controller_test.go b/pkg/reconciler/core/shard/shard_controller_test.go new file mode 100644 index 00000000000..73d6774b8a9 --- /dev/null +++ b/pkg/reconciler/core/shard/shard_controller_test.go @@ -0,0 +1,77 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package shard + +import ( + "context" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + "github.com/kcp-dev/sdk/apis/third_party/conditions/util/conditions" +) + +func TestReconcileMarksCordoned(t *testing.T) { + t.Parallel() + c := &Controller{} + shard := &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: "root", + Annotations: map[string]string{corev1alpha1.ShardUnschedulableAnnotationKey: "true"}, + }, + } + if err := c.reconcile(context.Background(), shard); err != nil { + t.Fatal(err) + } + cond := conditions.Get(shard, corev1alpha1.ShardSchedulable) + if cond == nil { + t.Fatal("expected the Schedulable condition to be set") + } + if cond.Status != corev1.ConditionFalse { + t.Errorf("expected Schedulable=False, got %s", cond.Status) + } + if cond.Reason != corev1alpha1.ShardReasonCordoned { + t.Errorf("expected reason %q, got %q", corev1alpha1.ShardReasonCordoned, cond.Reason) + } +} + +func TestReconcileMarksSchedulableAgain(t *testing.T) { + t.Parallel() + c := &Controller{} + shard := &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: "root", + Annotations: map[string]string{corev1alpha1.ShardUnschedulableAnnotationKey: "true"}, + }, + } + if err := c.reconcile(context.Background(), shard); err != nil { + t.Fatal(err) + } + delete(shard.Annotations, corev1alpha1.ShardUnschedulableAnnotationKey) + if err := c.reconcile(context.Background(), shard); err != nil { + t.Fatal(err) + } + cond := conditions.Get(shard, corev1alpha1.ShardSchedulable) + if cond == nil { + t.Fatal("expected the Schedulable condition to be set") + } + if cond.Status != corev1.ConditionTrue { + t.Errorf("expected Schedulable=True after uncordon, got %s (reason %s)", cond.Status, cond.Reason) + } +} diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller.go b/pkg/reconciler/core/shardmirror/shardmirror_controller.go index 5330cd33024..523749885fa 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller.go @@ -22,11 +22,18 @@ limitations under the License. // the authoritative object, and pruned when the authoritative object // disappears. Shard objects in the root workspace without that annotation // (e.g. created by tests or old shards) are left alone. +// +// The one writable exception on a representation is the allow-list in +// BackSyncedAnnotationKeys (e.g. cordoning a shard via the unschedulable +// annotation): those annotations are owned by the representation, and the +// mirror syncs them back onto the authoritative object through a direct +// connection to the owning shard instead of overwriting them. package shardmirror import ( "context" "fmt" + "sync" "time" "k8s.io/apimachinery/pkg/api/equality" @@ -35,6 +42,7 @@ import ( utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" genericrequest "k8s.io/apiserver/pkg/endpoints/request" + "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" "k8s.io/klog/v2" @@ -55,13 +63,58 @@ const ( ControllerName = "kcp-shard-mirror" ) +// BackSyncedAnnotationKeys lists the annotations an admin may set on a Shard +// representation in the root workspace to act on a shard centrally. The +// mirror copies them back onto the shard-owned authoritative object instead +// of overwriting them; every other part of a representation is read-only. +var BackSyncedAnnotationKeys = []string{corev1alpha1.ShardUnschedulableAnnotationKey} + +// shardClientPool caches per-shard kcp clients that connect directly to a +// shard's base URL with the given (logical-cluster-admin) credentials. +type shardClientPool struct { + mu sync.Mutex + adminConfig *rest.Config + clients map[string]poolEntry +} + +type poolEntry struct { + baseURL string + client kcpclientset.ClusterInterface +} + +func (p *shardClientPool) get(shardName, baseURL string) (kcpclientset.ClusterInterface, error) { + p.mu.Lock() + defer p.mu.Unlock() + if entry, ok := p.clients[shardName]; ok && entry.baseURL == baseURL { + return entry.client, nil + } + config := rest.CopyConfig(p.adminConfig) + config.Host = baseURL + client, err := kcpclientset.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("failed to create shard %q kcp client: %w", shardName, err) + } + p.clients[shardName] = poolEntry{baseURL: baseURL, client: client} + return client, nil +} + // NewController returns a controller mirroring authoritative Shard objects // from the cache server into the root workspace as read-only representations. +// shardName is the name of the shard this controller runs on (the root +// shard); shardAdminConfig holds credentials for direct connections to other +// shards, used to back-sync allow-listed annotations onto their authoritative +// Shard objects. func NewController( + shardName string, + shardAdminConfig *rest.Config, kcpClusterClient kcpclientset.ClusterInterface, localShardInformer corev1alpha1informers.ShardClusterInformer, cacheShardInformer corev1alpha1informers.ShardClusterInformer, ) *Controller { + pool := &shardClientPool{ + adminConfig: shardAdminConfig, + clients: map[string]poolEntry{}, + } c := &Controller{ queue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), @@ -90,6 +143,44 @@ func NewController( deleteShard: func(ctx context.Context, name string) error { return kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Delete(ctx, name, metav1.DeleteOptions{}) }, + backSyncAnnotations: func(ctx context.Context, source *corev1alpha1.Shard, values map[string]*string) error { + client := kcpClusterClient + if source.Name != shardName { + var err error + client, err = pool.get(source.Name, source.Spec.BaseURL) + if err != nil { + return err + } + } + shards := client.Cluster(configshard.SystemShardCluster.Path()).CoreV1alpha1().Shards() + // the informer copy from the cache server may be stale; fetch the + // authoritative object from the owning shard and only write if it + // actually differs. + shard, err := shards.Get(ctx, source.Name, metav1.GetOptions{}) + if err != nil { + return err + } + changed := false + for key, value := range values { + current, ok := shard.Annotations[key] + switch { + case value == nil && ok: + delete(shard.Annotations, key) + changed = true + case value != nil && (!ok || current != *value): + if shard.Annotations == nil { + shard.Annotations = map[string]string{} + } + shard.Annotations[key] = *value + changed = true + } + } + if !changed { + return nil + } + _, err = shards.Update(ctx, shard, metav1.UpdateOptions{}) + return err + }, } _, _ = cacheShardInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ @@ -117,6 +208,11 @@ type Controller struct { updateShard func(ctx context.Context, shard *corev1alpha1.Shard) error updateShardStatus func(ctx context.Context, shard *corev1alpha1.Shard) error deleteShard func(ctx context.Context, name string) error + + // backSyncAnnotations writes the given allow-listed annotation values + // (nil meaning delete) onto the authoritative Shard object in the owning + // shard's system:shard logical cluster. + backSyncAnnotations func(ctx context.Context, source *corev1alpha1.Shard, values map[string]*string) error } // enqueue maps any Shard event (authoritative copy in the cache, or local copy @@ -215,6 +311,34 @@ func (c *Controller) reconcile(ctx context.Context, name string) error { return nil } + // Allow-listed annotations are owned by the representation: an admin sets + // them in the root workspace and the mirror copies them onto the + // authoritative object instead of overwriting them here. Absence on the + // representation means absence on the authoritative object (uncordon). + values := map[string]*string{} + backSyncNeeded := false + for _, key := range BackSyncedAnnotationKeys { + if value, ok := local.Annotations[key]; ok { + desired.Annotations[key] = value + values[key] = &value + if sourceValue, ok := source.Annotations[key]; !ok || sourceValue != value { + backSyncNeeded = true + } + } else { + delete(desired.Annotations, key) + values[key] = nil + if _, ok := source.Annotations[key]; ok { + backSyncNeeded = true + } + } + } + if backSyncNeeded { + logger.V(2).Info("back-syncing annotations to the authoritative Shard object", "shard", name) + if err := c.backSyncAnnotations(ctx, source, values); err != nil { + return err + } + } + updated := local.DeepCopy() updated.Labels = desired.Labels updated.Annotations = desired.Annotations diff --git a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go index 6d64a371a0e..15fcadbd722 100644 --- a/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go +++ b/pkg/reconciler/core/shardmirror/shardmirror_controller_test.go @@ -60,6 +60,7 @@ func representation(name string) *corev1alpha1.Shard { type fakeActions struct { created, updated, statusUpdated *corev1alpha1.Shard deleted string + backSynced map[string]*string } func newController(source, local *corev1alpha1.Shard, actions *fakeActions) *Controller { @@ -92,6 +93,10 @@ func newController(source, local *corev1alpha1.Shard, actions *fakeActions) *Con actions.deleted = name return nil }, + backSyncAnnotations: func(_ context.Context, _ *corev1alpha1.Shard, values map[string]*string) error { + actions.backSynced = values + return nil + }, } } @@ -208,6 +213,61 @@ func TestReconcilePrunesOrphanedRepresentation(t *testing.T) { } } +func TestReconcileBackSyncsCordon(t *testing.T) { + t.Parallel() + cordoned := representation("alpha") + cordoned.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + actions := &fakeActions{} + c := newController(authoritativeShard("alpha"), cordoned, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.updated != nil { + t.Error("the cordon annotation must not be stomped off the representation") + } + value, ok := actions.backSynced[corev1alpha1.ShardUnschedulableAnnotationKey] + if !ok || value == nil || *value != "true" { + t.Errorf("expected the cordon annotation to be back-synced with value %q, got %v", "true", actions.backSynced) + } +} + +func TestReconcileBackSyncsUncordon(t *testing.T) { + t.Parallel() + source := authoritativeShard("alpha") + source.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + actions := &fakeActions{} + c := newController(source, representation("alpha"), actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.updated != nil { + t.Error("expected no representation update, absence of the annotation is authoritative") + } + value, ok := actions.backSynced[corev1alpha1.ShardUnschedulableAnnotationKey] + if !ok || value != nil { + t.Errorf("expected the cordon annotation to be back-synced as a deletion, got %v", actions.backSynced) + } +} + +func TestReconcileCordonInSync(t *testing.T) { + t.Parallel() + source := authoritativeShard("alpha") + source.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + local := representation("alpha") + local.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + actions := &fakeActions{} + c := newController(source, local, actions) + if err := c.reconcile(context.Background(), "alpha"); err != nil { + t.Fatal(err) + } + if actions.backSynced != nil { + t.Errorf("expected no back-sync when annotation values match, got %v", actions.backSynced) + } + if actions.updated != nil || actions.created != nil || actions.deleted != "" { + t.Errorf("expected no action, got %+v", actions) + } +} + func TestReconcileLeavesUnmanagedObjectsAlone(t *testing.T) { t.Parallel() synthetic := authoritativeShard("fake") // e.g. created by an e2e test in the root workspace diff --git a/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling.go b/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling.go index 679d8be2610..10ec7ae7e5b 100644 --- a/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling.go +++ b/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling.go @@ -61,10 +61,6 @@ const ( // workspaceClusterAnnotationKey keeps track of the logical cluster on the shard. workspaceClusterAnnotationKey = "internal.tenancy.kcp.io/cluster" - - // unschedulableAnnotationKey is the annotation key used to indicate that a shard is unschedulable. - // The annotation is meant to be used by e2e tests that otherwise started a private instance of kcp server. - unschedulableAnnotationKey = "experimental.core.kcp.io/unschedulable" ) // applyShardToWorkspaceMetadata updates a workspaces shard metadata. @@ -271,8 +267,8 @@ func (r *schedulingReconciler) chooseShardAndMarkCondition(logger klog.Logger, w reason, message string }{} for _, shard := range shards { - if _, ok := shard.Annotations[unschedulableAnnotationKey]; ok { - logger.V(4).Info("Skipping a shard because it is annotated as unschedulable", "shard", shard.Name, "annotation", unschedulableAnnotationKey) + if _, ok := shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; ok { + logger.V(4).Info("Skipping a shard because it is annotated as unschedulable", "shard", shard.Name, "annotation", corev1alpha1.ShardUnschedulableAnnotationKey) continue } if valid, reason, message := isValidShard(shard); valid { diff --git a/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling_test.go b/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling_test.go index cec26f920da..aae4223f44e 100644 --- a/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling_test.go +++ b/pkg/reconciler/tenancy/workspace/workspace_reconcile_scheduling_test.go @@ -277,7 +277,7 @@ func TestReconcileScheduling(t *testing.T) { name: "only an unschedulable shard is available, the ws is unscheduled", initialShards: []*corev1alpha1.Shard{func() *corev1alpha1.Shard { s := shard("amber") - s.Annotations[unschedulableAnnotationKey] = "true" + s.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" return s }()}, targetWorkspace: workspace("foo"), diff --git a/pkg/server/controllers.go b/pkg/server/controllers.go index d33ce6c960a..e970c07c2d7 100644 --- a/pkg/server/controllers.go +++ b/pkg/server/controllers.go @@ -600,34 +600,37 @@ func (s *Server) installWorkspaceScheduler(ctx context.Context, config *rest.Con return err } - var workspaceShardController *shard.Controller - if s.Options.Extra.ShardName == corev1alpha1.RootShard { - workspaceShardController, err = shard.NewController( - kcpClusterClient, - s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), - ) - if err != nil { - return err - } + // runs on every shard: maintains the status (e.g. the Schedulable + // condition) of this shard's own authoritative Shard object in the local + // system:shard logical cluster. + workspaceShardController, err := shard.NewController( + s.Options.Extra.ShardName, + kcpClusterClient, + s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), + ) + if err != nil { + return err } - if workspaceShardController != nil { - if err := s.registerController(&controllerWrapper{ - Name: shard.ControllerName, - Wait: func(ctx context.Context, s *Server) error { - return wait.PollUntilContextCancel(ctx, waitPollInterval, true, func(ctx context.Context) (bool, error) { - return s.KcpSharedInformerFactory.Core().V1alpha1().Shards().Informer().HasSynced(), nil - }) - }, - Runner: func(ctx context.Context) { - workspaceShardController.Start(ctx, 2) - }, - }); err != nil { - return err - } + if err := s.registerController(&controllerWrapper{ + Name: shard.ControllerName, + Wait: func(ctx context.Context, s *Server) error { + return wait.PollUntilContextCancel(ctx, waitPollInterval, true, func(ctx context.Context) (bool, error) { + return s.KcpSharedInformerFactory.Core().V1alpha1().Shards().Informer().HasSynced(), nil + }) + }, + Runner: func(ctx context.Context) { + workspaceShardController.Start(ctx, 2) + }, + }); err != nil { + return err } if s.Options.Extra.ShardName == corev1alpha1.RootShard { + shardMirrorAdminConfig := rest.CopyConfig(logicalClusterAdminConfig) + shardMirrorAdminConfig = rest.AddUserAgent(shardMirrorAdminConfig, shardmirror.ControllerName) shardMirrorController := shardmirror.NewController( + s.Options.Extra.ShardName, + shardMirrorAdminConfig, kcpClusterClient, s.KcpSharedInformerFactory.Core().V1alpha1().Shards(), s.CacheKcpSharedInformerFactory.Core().V1alpha1().Shards(), diff --git a/staging/src/github.com/kcp-dev/cli/cmd/kubectl-kcp/cmd/kubectlKcp.go b/staging/src/github.com/kcp-dev/cli/cmd/kubectl-kcp/cmd/kubectlKcp.go index 0fc3ba91fe6..98a32a004dd 100644 --- a/staging/src/github.com/kcp-dev/cli/cmd/kubectl-kcp/cmd/kubectlKcp.go +++ b/staging/src/github.com/kcp-dev/cli/cmd/kubectl-kcp/cmd/kubectlKcp.go @@ -31,6 +31,7 @@ import ( claimscmd "github.com/kcp-dev/cli/pkg/claims/cmd" crdcmd "github.com/kcp-dev/cli/pkg/crd/cmd" quickstartcmd "github.com/kcp-dev/cli/pkg/quickstart/cmd" + shardcmd "github.com/kcp-dev/cli/pkg/shard/cmd" workspacecmd "github.com/kcp-dev/cli/pkg/workspace/cmd" "github.com/kcp-dev/sdk/cmd/help" ) @@ -89,5 +90,8 @@ func KubectlKcpCommand() *cobra.Command { quickstartCmd := quickstartcmd.New(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr}) root.AddCommand(quickstartCmd) + shardCmd := shardcmd.New(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr}) + root.AddCommand(shardCmd) + return root } diff --git a/staging/src/github.com/kcp-dev/cli/pkg/shard/cmd/cmd.go b/staging/src/github.com/kcp-dev/cli/pkg/shard/cmd/cmd.go new file mode 100644 index 00000000000..2fe450f432e --- /dev/null +++ b/staging/src/github.com/kcp-dev/cli/pkg/shard/cmd/cmd.go @@ -0,0 +1,97 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cmd + +import ( + "fmt" + + "github.com/spf13/cobra" + + "k8s.io/cli-runtime/pkg/genericclioptions" + + "github.com/kcp-dev/cli/pkg/shard/plugin" +) + +var ( + cordonExample = ` +# Mark shard "alpha" as unschedulable: no new workspaces are scheduled onto it +%[1]s shard cordon alpha +` + uncordonExample = ` +# Mark shard "alpha" as schedulable again +%[1]s shard uncordon alpha +` +) + +// New provides a command for shard operations. +func New(streams genericclioptions.IOStreams) *cobra.Command { + cmd := &cobra.Command{ + Use: "shard", + Short: "Manages kcp shards", + SilenceUsage: true, + TraverseChildren: true, + RunE: func(cmd *cobra.Command, args []string) error { + return cmd.Help() + }, + } + + cordonOptions := plugin.NewCordonOptions(streams) + cordonOptions.Cordon = true + cordonCommand := &cobra.Command{ + Use: "cordon SHARD", + Short: "Mark shard as unschedulable", + Long: "Mark shard as unschedulable: the workspace scheduler will not place new workspaces on it. The shard acknowledges via the Schedulable condition on its Shard object in the root workspace.", + Example: fmt.Sprintf(cordonExample, "kubectl kcp"), + Args: cobra.ExactArgs(1), + SilenceUsage: true, + RunE: func(c *cobra.Command, args []string) error { + if err := cordonOptions.Complete(args); err != nil { + return err + } + if err := cordonOptions.Validate(); err != nil { + return err + } + return cordonOptions.Run(c.Context()) + }, + } + cordonOptions.BindFlags(cordonCommand) + cmd.AddCommand(cordonCommand) + + uncordonOptions := plugin.NewCordonOptions(streams) + uncordonOptions.Cordon = false + uncordonCommand := &cobra.Command{ + Use: "uncordon SHARD", + Short: "Mark shard as schedulable", + Long: "Mark shard as schedulable again, allowing the workspace scheduler to place new workspaces on it.", + Example: fmt.Sprintf(uncordonExample, "kubectl kcp"), + Args: cobra.ExactArgs(1), + SilenceUsage: true, + RunE: func(c *cobra.Command, args []string) error { + if err := uncordonOptions.Complete(args); err != nil { + return err + } + if err := uncordonOptions.Validate(); err != nil { + return err + } + return uncordonOptions.Run(c.Context()) + }, + } + uncordonOptions.BindFlags(uncordonCommand) + cmd.AddCommand(uncordonCommand) + + return cmd +} diff --git a/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon.go b/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon.go new file mode 100644 index 00000000000..0e86f62e9f4 --- /dev/null +++ b/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon.go @@ -0,0 +1,142 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package plugin implements shard operations, e.g. cordoning and uncordoning +// a shard the way kubectl cordon/uncordon act on nodes. +package plugin + +import ( + "context" + "fmt" + "net/url" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/cli-runtime/pkg/genericclioptions" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + + "github.com/kcp-dev/cli/pkg/base" + "github.com/kcp-dev/sdk/apis/core" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" +) + +// CordonOptions holds the options for cordoning or uncordoning a shard. +type CordonOptions struct { + *base.Options + + // Shard is the name of the shard to act on. + Shard string + // Cordon marks the shard unschedulable when true, schedulable again when + // false. + Cordon bool + + // for testing + newKCPClusterClient func(clientConfig clientcmd.ClientConfig) (kcpclientset.ClusterInterface, error) +} + +// NewCordonOptions returns a new CordonOptions. +func NewCordonOptions(streams genericclioptions.IOStreams) *CordonOptions { + options := base.NewOptions(streams) + // Shard objects live in the root workspace only; a --workspace flag would + // be misleading. + options.OptOutOfWorkspaceFlag = true + return &CordonOptions{ + Options: options, + newKCPClusterClient: newKCPClusterClient, + } +} + +// Complete ensures all dynamically populated fields are initialized. +func (o *CordonOptions) Complete(args []string) error { + if err := o.Options.Complete(); err != nil { + return err + } + if len(args) > 0 { + o.Shard = args[0] + } + return nil +} + +// Validate validates the inputs. +func (o *CordonOptions) Validate() error { + if o.Shard == "" { + return fmt.Errorf("shard name is required") + } + return o.Options.Validate() +} + +// Run (un)cordons the shard by updating the unschedulable annotation on the +// Shard object in the root workspace. Like a node's spec.unschedulable in +// Kubernetes, the annotation is the desired state; the owning shard +// acknowledges it via the Schedulable condition on the Shard object. +func (o *CordonOptions) Run(ctx context.Context) error { + kcpClusterClient, err := o.newKCPClusterClient(o.ClientConfig) + if err != nil { + return fmt.Errorf("failed to create kcp client: %w", err) + } + shards := kcpClusterClient.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards() + + shard, err := shards.Get(ctx, o.Shard, metav1.GetOptions{}) + if err != nil { + return err + } + + // the scheduler checks for the presence of the annotation, not its value. + _, cordoned := shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] + + var patch string + switch { + case o.Cordon && cordoned: + fmt.Fprintf(o.Out, "shard/%s already cordoned\n", o.Shard) + return nil + case !o.Cordon && !cordoned: + fmt.Fprintf(o.Out, "shard/%s already uncordoned\n", o.Shard) + return nil + case o.Cordon: + patch = fmt.Sprintf(`{"metadata":{"annotations":{%q:"true"}}}`, corev1alpha1.ShardUnschedulableAnnotationKey) + default: + patch = fmt.Sprintf(`{"metadata":{"annotations":{%q:null}}}`, corev1alpha1.ShardUnschedulableAnnotationKey) + } + + if _, err := shards.Patch(ctx, o.Shard, types.MergePatchType, []byte(patch), metav1.PatchOptions{}); err != nil { + return err + } + + if o.Cordon { + fmt.Fprintf(o.Out, "shard/%s cordoned\n", o.Shard) + } else { + fmt.Fprintf(o.Out, "shard/%s uncordoned\n", o.Shard) + } + return nil +} + +func newKCPClusterClient(clientConfig clientcmd.ClientConfig) (kcpclientset.ClusterInterface, error) { + config, err := clientConfig.ClientConfig() + if err != nil { + return nil, err + } + clusterConfig := rest.CopyConfig(config) + u, err := url.Parse(config.Host) + if err != nil { + return nil, err + } + u.Path = "" + clusterConfig.Host = u.String() + clusterConfig.UserAgent = rest.DefaultKubernetesUserAgent() + return kcpclientset.NewForConfig(clusterConfig) +} diff --git a/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon_test.go b/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon_test.go new file mode 100644 index 00000000000..824374ccc48 --- /dev/null +++ b/staging/src/github.com/kcp-dev/cli/pkg/shard/plugin/cordon_test.go @@ -0,0 +1,121 @@ +/* +Copyright 2026 The kcp Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package plugin + +import ( + "bytes" + "context" + "strings" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/cli-runtime/pkg/genericclioptions" + "k8s.io/client-go/tools/clientcmd" + + "github.com/kcp-dev/logicalcluster/v3" + "github.com/kcp-dev/sdk/apis/core" + corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" + kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster" + kcpfakeclient "github.com/kcp-dev/sdk/client/clientset/versioned/cluster/fake" +) + +func shardObj(name string, cordoned bool) *corev1alpha1.Shard { + shard := &corev1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Annotations: map[string]string{logicalcluster.AnnotationKey: core.RootCluster.String()}, + }, + } + if cordoned { + shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + } + return shard +} + +func run(t *testing.T, shard *corev1alpha1.Shard, cordon bool) (*kcpfakeclient.ClusterClientset, string) { + t.Helper() + client := kcpfakeclient.NewSimpleClientset(shard) //nolint:staticcheck + out := &bytes.Buffer{} + opts := NewCordonOptions(genericclioptions.IOStreams{Out: out, ErrOut: out}) + opts.Shard = shard.Name + opts.Cordon = cordon + opts.newKCPClusterClient = func(clientcmd.ClientConfig) (kcpclientset.ClusterInterface, error) { + return client, nil + } + if err := opts.Validate(); err != nil { + t.Fatal(err) + } + if err := opts.Run(context.Background()); err != nil { + t.Fatal(err) + } + return client, out.String() +} + +func getShard(t *testing.T, client *kcpfakeclient.ClusterClientset, name string) *corev1alpha1.Shard { + t.Helper() + shard, err := client.Cluster(core.RootCluster.Path()).CoreV1alpha1().Shards().Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + return shard +} + +func TestCordon(t *testing.T) { + t.Parallel() + client, out := run(t, shardObj("alpha", false), true) + if !strings.Contains(out, "shard/alpha cordoned") { + t.Errorf("unexpected output: %q", out) + } + if _, ok := getShard(t, client, "alpha").Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; !ok { + t.Error("expected the unschedulable annotation to be set") + } +} + +func TestCordonAlreadyCordoned(t *testing.T) { + t.Parallel() + _, out := run(t, shardObj("alpha", true), true) + if !strings.Contains(out, "shard/alpha already cordoned") { + t.Errorf("unexpected output: %q", out) + } +} + +func TestUncordon(t *testing.T) { + t.Parallel() + client, out := run(t, shardObj("alpha", true), false) + if !strings.Contains(out, "shard/alpha uncordoned") { + t.Errorf("unexpected output: %q", out) + } + if _, ok := getShard(t, client, "alpha").Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; ok { + t.Error("expected the unschedulable annotation to be removed") + } +} + +func TestUncordonAlreadyUncordoned(t *testing.T) { + t.Parallel() + _, out := run(t, shardObj("alpha", false), false) + if !strings.Contains(out, "shard/alpha already uncordoned") { + t.Errorf("unexpected output: %q", out) + } +} + +func TestValidateRequiresShardName(t *testing.T) { + t.Parallel() + opts := NewCordonOptions(genericclioptions.IOStreams{}) + if err := opts.Validate(); err == nil { + t.Error("expected an error when no shard name is given") + } +} diff --git a/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go b/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go index b5bd7961741..22dd07e622b 100644 --- a/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go +++ b/staging/src/github.com/kcp-dev/sdk/apis/core/v1alpha1/shard_types.go @@ -29,12 +29,34 @@ var RootShard = "root" // ShardRepresentationAnnotationKey marks a Shard object as a read-only // representation mirrored from the shard-owned authoritative object living in -// the shard's local system:shard logical cluster. Representations exist for -// discoverability (e.g. in the root workspace) and must not be edited; they -// are continuously overwritten from the authoritative object and are excluded -// from cache replication. +// the shard's local system:shard logical cluster. It is mostly read-only, with +// only exception is a small allow-list of operational annotations (like +// ShardUnschedulableAnnotationKey) that admins may set on the representation; +// those are synced back to the authoritative object instead of being overwritten. const ShardRepresentationAnnotationKey = "core.kcp.io/shard-representation" +// ShardUnschedulableAnnotationKey marks a single shard as unschedulable: the +// workspace scheduler will not place new workspaces on it (cordoning). It +// only affects the shard whose Shard object carries it. +// Admins set it on that shard's representation in the root workspace; the +// shard mirror syncs it back to the shard-owned authoritative object in the +// owning shard's system:shard logical cluster, from where cache replication +// makes it visible to the workspace schedulers running on every shard. +const ShardUnschedulableAnnotationKey = "experimental.core.kcp.io/unschedulable" + +// ShardSchedulable is a condition on the Shard object reflecting the shard's +// scheduling state, similar to a Kubernetes node: True when new workspaces +// may be scheduled onto the shard, False with reason ShardReasonCordoned when +// the shard observed the unschedulable annotation on its authoritative +// object. It is set by the owning shard itself, so seeing it change on the +// representation in the root workspace acknowledges that the shard received +// and applied the cordon/uncordon signal. +const ShardSchedulable v1alpha1.ConditionType = "Schedulable" + +// ShardReasonCordoned is the reason for ShardSchedulable=False when the +// owning shard observed the unschedulable annotation on its Shard object. +const ShardReasonCordoned = "Cordoned" + // Shard describes a kcp instance on which a number of logical clusters will live // // +crd @@ -46,6 +68,7 @@ const ShardRepresentationAnnotationKey = "core.kcp.io/shard-representation" // +kubebuilder:printcolumn:name="Region",type=string,JSONPath=`.metadata.labels['region']`,description="The region this workspace is in" // +kubebuilder:printcolumn:name="URL",type=string,JSONPath=`.spec.baseURL`,description="Type URL to directly connect to the shard" // +kubebuilder:printcolumn:name="External URL",type=string,JSONPath=`.spec.externalURL`,description="The URL exposed in logical clusters created on that shard" +// +kubebuilder:printcolumn:name="Schedulable",type=string,JSONPath=`.status.conditions[?(@.type=="Schedulable")].status`,description="Whether new workspaces are scheduled onto this shard" // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp" type Shard struct { v1.TypeMeta `json:",inline"` diff --git a/test/e2e/reconciler/workspace/controller_test.go b/test/e2e/reconciler/workspace/controller_test.go index 3e4c79a7819..b71a9dbfe22 100644 --- a/test/e2e/reconciler/workspace/controller_test.go +++ b/test/e2e/reconciler/workspace/controller_test.go @@ -30,6 +30,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/util/retry" "github.com/kcp-dev/sdk/apis/core" corev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1" @@ -41,7 +42,6 @@ import ( kcptestinghelpers "github.com/kcp-dev/sdk/testing/helpers" kcptestingserver "github.com/kcp-dev/sdk/testing/server" - configshard "github.com/kcp-dev/kcp/config/shard" "github.com/kcp-dev/kcp/test/e2e/framework" ) @@ -52,10 +52,6 @@ func TestWorkspaceController(t *testing.T) { type runningServer struct { kcptestingserver.RunningServer rootWorkspaceKcpClient, orgWorkspaceKcpClient kcpclientset.Interface - // systemShardKcpClient is a privileged client to the root shard's - // system:shard logical cluster, where the shard-owned authoritative - // Shard object lives. - systemShardKcpClient kcpclientset.Interface } var testCases = []struct { name string @@ -102,17 +98,26 @@ func TestWorkspaceController(t *testing.T) { work: func(ctx context.Context, t *testing.T, server runningServer) { t.Helper() - t.Logf("Get the root shard's authoritative Shard object from system:shard") - shard, err := server.systemShardKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) - require.NoError(t, err) - - t.Logf("Mark the root shard as unschedulable") - if shard.Annotations == nil { - shard.Annotations = map[string]string{} - } - shard.Annotations["experimental.core.kcp.io/unschedulable"] = "true" - _, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) - require.NoError(t, err) + t.Logf("Mark the root shard as unschedulable via its representation in the root workspace") + var shard *corev1alpha1.Shard + require.NoError(t, retry.RetryOnConflict(retry.DefaultBackoff, func() error { + var err error + shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + if err != nil { + return err + } + if shard.Annotations == nil { + shard.Annotations = map[string]string{} + } + shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey] = "true" + shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) + return err + })) + + t.Logf("Wait for the shard to acknowledge the cordon via the Schedulable condition on its representation") + kcptestinghelpers.EventuallyCondition(t, func() (utilconditions.Getter, error) { + return server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + }, kcptestinghelpers.IsNot(corev1alpha1.ShardSchedulable).WithReason(corev1alpha1.ShardReasonCordoned)) // Workspaces cannot become "unschedulable" - once they // are scheduled on a shard this state does not change. @@ -161,12 +166,22 @@ func TestWorkspaceController(t *testing.T) { return server.orgWorkspaceKcpClient.TenancyV1alpha1().Workspaces().Get(ctx, workspaceName, metav1.GetOptions{}) }) - t.Logf("Remove unschedulable annotation from the root shard") - shard, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) - require.NoError(t, err) - delete(shard.Annotations, "experimental.core.kcp.io/unschedulable") - shard, err = server.systemShardKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) - require.NoError(t, err) + t.Logf("Remove unschedulable annotation from the root shard representation") + require.NoError(t, retry.RetryOnConflict(retry.DefaultBackoff, func() error { + var err error + shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + if err != nil { + return err + } + delete(shard.Annotations, corev1alpha1.ShardUnschedulableAnnotationKey) + shard, err = server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Update(ctx, shard, metav1.UpdateOptions{}) + return err + })) + + t.Logf("Wait for the shard to acknowledge the uncordon via the Schedulable condition on its representation") + kcptestinghelpers.EventuallyCondition(t, func() (utilconditions.Getter, error) { + return server.rootWorkspaceKcpClient.CoreV1alpha1().Shards().Get(ctx, "root", metav1.GetOptions{}) + }, kcptestinghelpers.Is(corev1alpha1.ShardSchedulable)) t.Logf("Expect workspace to be scheduled to the shard and show the external URL") kcptestinghelpers.EventuallyCondition(t, func() (utilconditions.Getter, error) { @@ -205,14 +220,10 @@ func TestWorkspaceController(t *testing.T) { kcpClient, err := kcpclusterclientset.NewForConfig(cfg) require.NoError(t, err) - systemShardClient, err := kcpclusterclientset.NewForConfig(server.RootShardSystemMasterBaseConfig(t)) - require.NoError(t, err) - testCase.work(ctx, t, runningServer{ RunningServer: server, rootWorkspaceKcpClient: kcpClient.Cluster(core.RootCluster.Path()), orgWorkspaceKcpClient: kcpClient.Cluster(orgPath), - systemShardKcpClient: systemShardClient.Cluster(configshard.SystemShardCluster.Path()), }) }) }