Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions cmd/sharded-test-server/frontproxy.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import (
"k8s.io/klog/v2"

kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster"
kcptesting "github.com/kcp-dev/sdk/testing"
kcptestingserver "github.com/kcp-dev/sdk/testing/server"
"github.com/kcp-dev/sdk/testing/third_party/library-go/crypto"

Expand Down Expand Up @@ -131,6 +132,17 @@ func startFrontProxy(
return nil, fmt.Errorf("failed to write server cert: %w", err)
}

// The front-proxy validates service account tokens itself, as it does in
// production: all shards sign with the one key under .kcp, and the proxy
// resolves the service account on the shard hosting its workspace
// through its index. With service account authentication enabled the
// proxy forwards nothing unauthenticated, so the static e2e users need
// their token file here as well.
tokenAuthFile, err := kcptesting.WriteDefaultTokenAuthFile(filepath.Join(workDirPath, ".kcp-front-proxy"))
if err != nil {
return nil, fmt.Errorf("failed to write token auth file: %w", err)
}

// run front-proxy command
workdir, commandLine := kcptestingserver.Command("kcp-front-proxy", "front-proxy")
commandLine = append(commandLine,
Expand All @@ -155,6 +167,8 @@ func startFrontProxy(
"--requestheader-allowed-names=kcp-front-proxy,kcp-mounts-proxy",
fmt.Sprintf("--tls-cert-file=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "apiserver.crt")),
fmt.Sprintf("--tls-private-key-file=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "apiserver.key")),
fmt.Sprintf("--service-account-key-file=%s", filepath.Join(workDirPath, ".kcp", "service-account.crt")),
fmt.Sprintf("--token-auth-file=%s", tokenAuthFile),
"--secure-port=6443",
"--v=4",
)
Expand Down
6 changes: 3 additions & 3 deletions docs/content/concepts/sharding/shards.md
Original file line number Diff line number Diff line change
Expand Up @@ -132,9 +132,9 @@ front-proxy, add a `/services/` path mapping; any shard can be the backend.
The front-proxy itself discovers shards through the Admin workspace, and
reads the identities of the core APIs from the `apiidentities` registry
there, so it starts and runs without the root shard: the kubeconfigs given
via `--shard-peer-kubeconfig` (falling back to the `--root-kubeconfig`
server) are seeds, used to bootstrap discovery and as a last-resort fallback
only. Every shard observed through the view joins the
via `--shard-peer-kubeconfig`, which is required, are seeds, used to
bootstrap discovery and as a last-resort fallback only; `--root-kubeconfig`
is ignored. Every shard observed through the view joins the
failover peer set at runtime and decommissioned shards leave it; requests are
distributed across the discovered shards, not the seeds. With short-lived
shards the discovery channel therefore keeps working as long as any currently
Expand Down
91 changes: 91 additions & 0 deletions pkg/index/indextest/index.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/*
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 indextest provides a test double for index.Index.
package indextest

import (
"context"
"sync/atomic"

"github.com/kcp-dev/logicalcluster/v3"

"github.com/kcp-dev/kcp/pkg/index"
)

var _ index.Index = &Index{}

// Index is a map-based index.Index: it maps logical clusters to shards and
// shards to base URLs. Lookups resolve logical cluster names only, not
// workspace paths. The maps must not be changed once the Index is in use.
type Index struct {
// Clusters maps a logical cluster to the name of the shard hosting it.
Clusters map[logicalcluster.Name]string
// Shards maps a shard name to its base URL.
Shards map[string]string
// RecentlyMigratedClusters are the clusters RecentlyMigrated reports.
RecentlyMigratedClusters map[logicalcluster.Name]bool

// Registered and Released count the ClusterContext calls and the
// calls of the cancel functions they returned.
Registered atomic.Int32
Released atomic.Int32
}

// Lookup returns the shard hosting the cluster, with the cluster's URL on
// that shard.
func (f *Index) Lookup(path logicalcluster.Path) (index.Result, bool) {
name, ok := path.Name()
if !ok {
return index.Result{}, false
}
shard, ok := f.Clusters[name]
if !ok {
return index.Result{}, false
}
return index.Result{
Shard: shard,
Cluster: name,
URL: f.Shards[shard] + name.Path().RequestPath(),
}, true
}

// LookupURL is Lookup: the fake knows no mounts.
func (f *Index) LookupURL(path logicalcluster.Path) (index.Result, bool) {
return f.Lookup(path)
}

// BaseURL returns the base URL of the shard.
func (f *Index) BaseURL(shardName string) (string, bool) {
url, ok := f.Shards[shardName]
return url, ok
}

// RecentlyMigrated reports whether the cluster is in RecentlyMigratedClusters.
func (f *Index) RecentlyMigrated(cluster logicalcluster.Name) bool {
return f.RecentlyMigratedClusters[cluster]
}

// ClusterContext returns a cancellable child of parent and counts the
// registration and its release.
func (f *Index) ClusterContext(parent context.Context, cluster logicalcluster.Name) (context.Context, context.CancelFunc) {
f.Registered.Add(1)
ctx, cancel := context.WithCancel(parent)
return ctx, func() {
f.Released.Add(1)
cancel()
}
}
44 changes: 15 additions & 29 deletions pkg/proxy/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,15 +53,14 @@ type ExtraConfig struct {
// unknown. It is a readiness check: requests needing an identity fail
// until it arrives, nothing blocks startup on it.
CoreIdentitiesReady func() error
// RootShardConfig points at the root shard and is set only when
// --root-kubeconfig is given. Service account authentication still needs
// it.
RootShardConfig *rest.Config
ShardsConfig *rest.Config
ShardsConfig *rest.Config
// ServiceAccounts resolves service accounts and token secrets on the
// shard hosting them, through the index, for service account
// authentication.
ServiceAccounts *ShardRoutedServiceAccounts
// PeersConfig points at the Admin workspace (/services/admin) of the
// peer shards with round-robin and per-request failover, and is the only
// source for Shard discovery. When no peers are configured, the root
// kubeconfig's server acts as the seed peer.
// source for Shard discovery and the core identities.
PeersConfig *rest.Config
// Peers is the failover set behind PeersConfig: the configured seed
// peers plus every discovered shard, fed at runtime from the shard
Expand Down Expand Up @@ -98,22 +97,10 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error)
var kcpShardIdentityRoundTripper func(rt http.RoundTripper) http.RoundTripper
kcpShardIdentityRoundTripper, c.CoreIdentityHandler, c.CoreIdentitiesReady = bootstrap.NewCoreWildcardIdentities(bootstrap.KcpRootGroupExportNames, bootstrap.KcpRootGroupResourceExportNames)

if c.Options.RootKubeconfig != "" {
rootConfig, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{ExplicitPath: c.Options.RootKubeconfig}, nil).ClientConfig()
if err != nil {
return nil, fmt.Errorf("failed to load root kubeconfig: %w", err)
}
c.RootShardConfig = rest.CopyConfig(rootConfig)
c.RootShardConfig.Wrap(kcpShardIdentityRoundTripper)
}

var loopbackClientConfig *rest.Config
if err := c.Options.SecureServing.ApplyTo(&c.ServingInfo, &loopbackClientConfig); err != nil {
return nil, err
}
if err := c.Options.Authentication.ApplyTo(ctx, &c.AuthenticationInfo, c.ServingInfo, c.RootShardConfig); err != nil {
return nil, err
}

var err error
c.ShardsConfig, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
Expand All @@ -126,17 +113,16 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error)
}
c.ShardsConfig.Wrap(kcpShardIdentityRoundTripper)

// Shard discovery and the core identities always go through the Admin
// workspace of the peers; the root shard is never asked. Without
// configured peers, the root kubeconfig's server is the seed peer.
peerKubeconfigs := c.Options.ShardsPeerKubeconfigs
if len(peerKubeconfigs) == 0 {
if c.Options.RootKubeconfig == "" {
return nil, fmt.Errorf("--shard-peer-kubeconfig is required")
}
peerKubeconfigs = []string{c.Options.RootKubeconfig}
// Service account tokens name their logical cluster; the lookup finds the
// hosting shard through the index and asks it, with the shards credential.
c.ServiceAccounts = NewShardRoutedServiceAccounts(c.ShardsConfig)
if err := c.Options.Authentication.ApplyTo(ctx, &c.AuthenticationInfo, c.ServingInfo, c.ServiceAccounts); err != nil {
return nil, err
}
c.PeersConfig, c.Peers, err = NewPeersConfig(peerKubeconfigs)

// Shard discovery and the core identities always go through the Admin
// workspace of the peers; the root shard is never asked.
c.PeersConfig, c.Peers, err = NewPeersConfig(c.Options.ShardsPeerKubeconfigs)
if err != nil {
return nil, err
}
Expand Down
47 changes: 47 additions & 0 deletions pkg/proxy/index/index_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"time"

"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/labels"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
Expand Down Expand Up @@ -73,6 +74,7 @@ func NewController(
shardWorkspaceInformers: map[string]cache.SharedIndexInformer{},
shardLogicalClusterInformers: map[string]cache.SharedIndexInformer{},
shardWorkspaceStopCh: map[string]chan struct{}{},
shardInformersSynced: map[string][]cache.InformerSynced{},

state: *index.New([]index.PathRewriter{
indexrewriters.UserRewriter,
Expand Down Expand Up @@ -133,6 +135,9 @@ type Controller struct {
shardWorkspaceInformers map[string]cache.SharedIndexInformer
shardLogicalClusterInformers map[string]cache.SharedIndexInformer
shardWorkspaceStopCh map[string]chan struct{}
// shardInformersSynced holds, per shard, the sync state of its
// informers; Ready consults it.
shardInformersSynced map[string][]cache.InformerSynced

state index.State
}
Expand Down Expand Up @@ -274,6 +279,7 @@ func (c *Controller) process(ctx context.Context, key string) error {
c.shardWorkspaceInformers[shard.Name] = wsInformer
c.shardLogicalClusterInformers[shard.Name] = twInformer
c.shardWorkspaceStopCh[shard.Name] = stopCh
c.shardInformersSynced[shard.Name] = []cache.InformerSynced{wsInformer.HasSynced, twInformer.HasSynced}

go wsInformer.Run(stopCh)
go twInformer.Run(stopCh)
Expand All @@ -296,6 +302,47 @@ func (c *Controller) stopShard(shardName string) {
delete(c.shardWorkspaceStopCh, shardName)
delete(c.shardWorkspaceInformers, shardName)
delete(c.shardLogicalClusterInformers, shardName)
delete(c.shardInformersSynced, shardName)
}

// Ready reports whether the index knows the workspaces and logical clusters
// of every Shard: for each Shard the lister knows, the per-shard informers
// have been started and have completed their initial list. Until then a
// lookup of a cluster on a not yet indexed shard answers "not found", which
// the proxy turns into a 404 and the service account authenticator into a
// 401, so this is a readiness check rather than a liveness one. A Shard
// whose informers are not running yet, e.g. because its client cannot be
// built, keeps the index not ready.
//
// Cordoned Shards, those carrying the unschedulable annotation, are left
// out: cordoning is how a shard is put into maintenance, and a shard that
// is down for maintenance must not keep every proxy out of rotation, least
// of all on a proxy restart during the maintenance. Their workspaces are
// routable once their informers have synced.
func (c *Controller) Ready() error {
shards, err := c.shardLister.List(labels.Everything())
if err != nil {
return fmt.Errorf("listing shards: %w", err)
}

c.lock.RLock()
defer c.lock.RUnlock()

for _, shard := range shards {
if _, cordoned := shard.Annotations[corev1alpha1.ShardUnschedulableAnnotationKey]; cordoned {
continue
}
synced, found := c.shardInformersSynced[shard.Name]
if !found {
return fmt.Errorf("informers for shard %q are not started", shard.Name)
}
Comment on lines +335 to +338

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if a known shard is down?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should probably at least exclude shards that are marked for maintenance etcpp instead of taking all shards.

for _, hasSynced := range synced {
if !hasSynced() {
return fmt.Errorf("informers for shard %q have not synced", shard.Name)
}
}
}
return nil
}

func (c *Controller) Lookup(path logicalcluster.Path) (index.Result, bool) {
Expand Down
77 changes: 77 additions & 0 deletions pkg/proxy/index/index_controller_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,19 +28,24 @@ import (

"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"

"github.com/kcp-dev/kcp/pkg/index"
)

func newTestController() *Controller {
shardIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
return &Controller{
queue: workqueue.NewTypedRateLimitingQueueWithConfig(
workqueue.DefaultTypedControllerRateLimiter[string](),
workqueue.TypedRateLimitingQueueConfig[string]{Name: "test"},
),
shardIndexer: shardIndexer,
shardLister: corev1alpha1listers.NewShardLister(shardIndexer),
shardWorkspaceInformers: map[string]cache.SharedIndexInformer{},
shardLogicalClusterInformers: map[string]cache.SharedIndexInformer{},
shardWorkspaceStopCh: map[string]chan struct{}{},
shardInformersSynced: map[string][]cache.InformerSynced{},
state: *index.New(nil),
}
}
Expand Down Expand Up @@ -111,3 +116,75 @@ func TestOnShardDelete(t *testing.T) {
_, found := c.state.LookupURL(logicalcluster.NewPath("myws"))
require.False(t, found, "expected the shard's clusters to be unroutable after delete")
}

// TestReady walks the readiness of the index through the life of a shard:
// unknown, known but informers not started, started but still syncing,
// synced, and gone.
func TestReady(t *testing.T) {
t.Parallel()
c := newTestController()

require.NoError(t, c.Ready(), "an index without shards is ready")

alpha := shardObj("alpha", "https://alpha")
require.NoError(t, c.shardIndexer.Add(alpha))
err := c.Ready()
require.Error(t, err, "a shard without informers keeps the index not ready")
require.Contains(t, err.Error(), `"alpha"`)
require.Contains(t, err.Error(), "not started")

synced := false
c.lock.Lock()
c.shardInformersSynced["alpha"] = []cache.InformerSynced{func() bool { return true }, func() bool { return synced }}
c.lock.Unlock()
err = c.Ready()
require.Error(t, err, "a shard whose informers have not synced keeps the index not ready")
require.Contains(t, err.Error(), "not synced")

synced = true
require.NoError(t, c.Ready(), "all informers synced")

beta := shardObj("beta", "https://beta")
require.NoError(t, c.shardIndexer.Add(beta))
require.Error(t, c.Ready(), "a new shard is not ready until its informers are")

c.stopShard("beta")
require.NoError(t, c.shardIndexer.Delete(beta))
require.NoError(t, c.Ready(), "a deleted shard no longer counts")

c.stopShard("alpha")
require.Error(t, c.Ready(), "stopping the informers of a shard that still exists makes the index not ready")
}

// TestReadyCordonedShard checks that a cordoned shard, i.e. one in
// maintenance, does not hold the index back, and counts again once it is
// uncordoned.
func TestReadyCordonedShard(t *testing.T) {
t.Parallel()
c := newTestController()

alpha := shardObj("alpha", "https://alpha")
require.NoError(t, c.shardIndexer.Add(alpha))
c.lock.Lock()
c.shardInformersSynced["alpha"] = []cache.InformerSynced{func() bool { return true }}
c.lock.Unlock()

beta := shardObj("beta", "https://beta")
require.NoError(t, c.shardIndexer.Add(beta))
require.Error(t, c.Ready(), "an uncordoned shard that is not indexed keeps the index not ready")

cordoned := beta.DeepCopy()
cordoned.Annotations = map[string]string{corev1alpha1.ShardUnschedulableAnnotationKey: "true"}
require.NoError(t, c.shardIndexer.Update(cordoned))
require.NoError(t, c.Ready(), "a cordoned shard does not count")

synced := false
c.lock.Lock()
c.shardInformersSynced["beta"] = []cache.InformerSynced{func() bool { return synced }}
c.lock.Unlock()
require.NoError(t, c.shardIndexer.Update(beta))
require.Error(t, c.Ready(), "an uncordoned shard counts again")

synced = true
require.NoError(t, c.Ready())
}
Loading
Loading