diff --git a/cmd/sharded-test-server/frontproxy.go b/cmd/sharded-test-server/frontproxy.go index b1e1ad0b2f8..414458a9cb6 100644 --- a/cmd/sharded-test-server/frontproxy.go +++ b/cmd/sharded-test-server/frontproxy.go @@ -137,7 +137,6 @@ func startFrontProxy( "--bind-address="+hostIP, fmt.Sprintf("--mapping-file=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "mapping.yaml")), fmt.Sprintf("--root-directory=%s", filepath.Join(workDirPath, ".kcp-front-proxy")), - fmt.Sprintf("--root-kubeconfig=%s", filepath.Join(workDirPath, ".kcp", "root.kubeconfig")), fmt.Sprintf("--shards-kubeconfig=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "shards.kubeconfig")), fmt.Sprintf("--shard-peer-kubeconfig=%s", filepath.Join(workDirPath, ".kcp-front-proxy", "peers.kubeconfig")), fmt.Sprintf("--client-ca-file=%s", filepath.Join(workDirPath, ".kcp", "client-ca.crt")), diff --git a/docs/content/concepts/sharding/shards.md b/docs/content/concepts/sharding/shards.md index d1c77bd5217..fd095707fba 100644 --- a/docs/content/concepts/sharding/shards.md +++ b/docs/content/concepts/sharding/shards.md @@ -129,10 +129,12 @@ Access requires membership in the `system:kcp:admin` group; the serving path does not depend on the root shard. To reach the view through a front-proxy, add a `/services/` path mapping; any shard can be the backend. -The front-proxy itself discovers shards through the Admin workspace: 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 +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 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 diff --git a/docs/content/setup/production/metrics.md b/docs/content/setup/production/metrics.md index a6e17282034..09371ef404d 100644 --- a/docs/content/setup/production/metrics.md +++ b/docs/content/setup/production/metrics.md @@ -134,3 +134,10 @@ scrape_configs: - targets: - cache.kcp.svc:6443 ``` + +## Scraping the front-proxy + +The front-proxy's own `/metrics` is authorized by group: members of +`system:kcp:metrics-reader`, `system:kcp:admin` or `system:masters` may read +it. Give the scraper identity the `system:kcp:metrics-reader` group, for +example as a client certificate organization or an OIDC group claim. diff --git a/pkg/apiidentity/projector.go b/pkg/apiidentity/projector.go index b94e5fe1e78..9f99fc5df0c 100644 --- a/pkg/apiidentity/projector.go +++ b/pkg/apiidentity/projector.go @@ -38,7 +38,6 @@ import ( "github.com/kcp-dev/logicalcluster/v3" adminv1alpha1 "github.com/kcp-dev/sdk/apis/admin/v1alpha1" apisv1alpha2 "github.com/kcp-dev/sdk/apis/apis/v1alpha2" - "github.com/kcp-dev/sdk/apis/core" "github.com/kcp-dev/sdk/apis/third_party/conditions/util/conditions" "github.com/kcp-dev/kcp/pkg/server/bootstrap" @@ -48,20 +47,9 @@ import ( // IsCoreExport reports whether the export is one of the core APIExports in // the root workspace, whose identities every shard needs before it can serve. func IsCoreExport(export *apisv1alpha2.APIExport) bool { - return logicalcluster.From(export) == core.RootCluster && coreExportNames.Has(export.Name) + return bootstrap.IsCoreExport(logicalcluster.From(export), export.Name) } -var coreExportNames = func() sets.Set[string] { - names := sets.New[string]() - for _, name := range bootstrap.KcpRootGroupExportNames { - names.Insert(name) - } - for _, name := range bootstrap.KcpRootGroupResourceExportNames { - names.Insert(name) - } - return names -}() - // Project computes the view of one identity from the exports holding it. // The owner is the oldest export, by creation time, then cluster, then name, // so every reader of the same exports agrees on it; the others are holders. diff --git a/pkg/proxy/config.go b/pkg/proxy/config.go index ae3a94f65f3..8151aefcfe1 100644 --- a/pkg/proxy/config.go +++ b/pkg/proxy/config.go @@ -44,17 +44,20 @@ type completedConfig struct { } type ExtraConfig struct { - // resolveIdentities is to be called on server start until it succeeds. It injects the kcp - // resource identities into the rest.Config used by the client. Only after it succeeds, - // the clients can wildcard-list/watch most kcp resources. - ResolveIdentities func(ctx context.Context) error - // CoreIdentityHandler keeps the identities resolved by ResolveIdentities - // in sync with the APIExports in the root workspace, e.g. across an - // APIExport identity rotation. Register it on an informer of those - // APIExports. + // CoreIdentityHandler feeds the identities of the core APIs, which the + // clients built from ShardsConfig inject into their wildcard requests, + // from the APIIdentity registry that the Admin workspace of the peers + // serves. Register it on an informer of those entries. CoreIdentityHandler cache.ResourceEventHandler - RootShardConfig *rest.Config - ShardsConfig *rest.Config + // CoreIdentitiesReady fails while the identity of a core API is still + // 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 // 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 @@ -89,17 +92,20 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error) Options: opts, } - // get root API identities - //nolint:staticcheck // the deprecated root kubeconfig remains the source for identity resolution until identities are resolvable without the root shard. - nonIdentityRootConfig, 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) - } - + // The identities of the core APIs come from the APIIdentity registry, + // read through the Admin workspace of the peers; the root shard is not + // asked for them. var kcpShardIdentityRoundTripper func(rt http.RoundTripper) http.RoundTripper - kcpShardIdentityRoundTripper, c.ResolveIdentities, c.CoreIdentityHandler = bootstrap.NewWildcardIdentitiesWrappingRoundTripper(bootstrap.KcpRootGroupExportNames, bootstrap.KcpRootGroupResourceExportNames, nonIdentityRootConfig, nil) - c.RootShardConfig = rest.CopyConfig(nonIdentityRootConfig) - c.RootShardConfig.Wrap(kcpShardIdentityRoundTripper) + 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 { @@ -109,6 +115,7 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error) return nil, err } + var err error c.ShardsConfig, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig( &clientcmd.ClientConfigLoadingRules{ExplicitPath: c.Options.ShardsKubeconfig}, // We override the Server here so that the user doesn't have to specify unused server value @@ -119,12 +126,14 @@ func NewConfig(ctx context.Context, opts *proxyoptions.Options) (*Config, error) } c.ShardsConfig.Wrap(kcpShardIdentityRoundTripper) - // Shard discovery always goes through the Admin workspace of the peers; - // the root workspace is no longer read for it. Without configured peers, - // the root kubeconfig's server is the seed peer. + // 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 { - //nolint:staticcheck // the deprecated root kubeconfig doubles as the default seed peer for deployments that predate --shard-peer-kubeconfig. + if c.Options.RootKubeconfig == "" { + return nil, fmt.Errorf("--shard-peer-kubeconfig is required") + } peerKubeconfigs = []string{c.Options.RootKubeconfig} } c.PeersConfig, c.Peers, err = NewPeersConfig(peerKubeconfigs) diff --git a/pkg/proxy/metrics_authorizer.go b/pkg/proxy/metrics_authorizer.go new file mode 100644 index 00000000000..1355637068a --- /dev/null +++ b/pkg/proxy/metrics_authorizer.go @@ -0,0 +1,58 @@ +/* +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 proxy + +import ( + "context" + "slices" + + "k8s.io/apiserver/pkg/authorization/authorizer" + + bootstrappolicy "github.com/kcp-dev/kcp/pkg/authorization/bootstrap" +) + +// metricsReaderGroups may read the front-proxy's /metrics. The proxy has +// no RBAC of its own and must not depend on the root shard, so access is +// decided from the caller's groups: system:kcp:admin, and +// system:kcp:metrics-reader, the name of the ClusterRole that authorizes +// scraping on the shards, used as a group here. system:masters is not +// listed: the proxy drops it on ingress (see --authentication-drop-groups), +// so it never reaches this authorizer. +var metricsReaderGroups = []string{ + bootstrappolicy.SystemKcpAdminGroup, + bootstrappolicy.SystemKcpMetricsReader, +} + +type metricsGroupAuthorizer struct{} + +func newMetricsGroupAuthorizer() authorizer.Authorizer { + return metricsGroupAuthorizer{} +} + +// Authorize allows members of metricsReaderGroups and has no opinion on +// everybody else, so that a further authorizer in a union can still allow. +func (metricsGroupAuthorizer) Authorize(_ context.Context, attr authorizer.Attributes) (authorizer.Decision, string, error) { + if attr.GetUser() == nil { + return authorizer.DecisionNoOpinion, "", nil + } + for _, group := range attr.GetUser().GetGroups() { + if slices.Contains(metricsReaderGroups, group) { + return authorizer.DecisionAllow, "user is a member of the " + group + " group", nil + } + } + return authorizer.DecisionNoOpinion, "", nil +} diff --git a/pkg/proxy/metrics_authorizer_test.go b/pkg/proxy/metrics_authorizer_test.go new file mode 100644 index 00000000000..1458211a9d3 --- /dev/null +++ b/pkg/proxy/metrics_authorizer_test.go @@ -0,0 +1,58 @@ +/* +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 proxy + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "k8s.io/apiserver/pkg/authentication/user" + "k8s.io/apiserver/pkg/authorization/authorizer" + + bootstrappolicy "github.com/kcp-dev/kcp/pkg/authorization/bootstrap" +) + +func TestMetricsGroupAuthorizer(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + groups []string + want authorizer.Decision + }{ + {name: "kcp admin may scrape", groups: []string{bootstrappolicy.SystemKcpAdminGroup, user.AllAuthenticated}, want: authorizer.DecisionAllow}, + {name: "system:masters gets no opinion", groups: []string{bootstrappolicy.SystemMastersGroup}, want: authorizer.DecisionNoOpinion}, + {name: "the metrics-reader group may scrape", groups: []string{bootstrappolicy.SystemKcpMetricsReader}, want: authorizer.DecisionAllow}, + // no opinion, not deny: a further authorizer in the union (the :root + // SubjectAccessReview while --root-kubeconfig is given) may still allow. + {name: "an ordinary user gets no opinion", groups: []string{user.AllAuthenticated}, want: authorizer.DecisionNoOpinion}, + {name: "no groups gets no opinion", groups: nil, want: authorizer.DecisionNoOpinion}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, _, err := newMetricsGroupAuthorizer().Authorize(context.Background(), authorizer.AttributesRecord{ + User: &user.DefaultInfo{Name: "scraper", Groups: tc.groups}, + Verb: "get", + Path: "/metrics", + }) + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} diff --git a/pkg/proxy/options/authentication.go b/pkg/proxy/options/authentication.go index 3da26ca6a0a..973ed863947 100644 --- a/pkg/proxy/options/authentication.go +++ b/pkg/proxy/options/authentication.go @@ -96,6 +96,12 @@ func (c *Authentication) tokenAuthEnabled() bool { return c.BuiltInOptions.TokenFile != nil && c.BuiltInOptions.TokenFile.TokenFile != "" } +// ServiceAccountAuthEnabled reports whether service account tokens are +// accepted, which needs a client to the root shard. +func (c *Authentication) ServiceAccountAuthEnabled() bool { + return c.serviceAccountAuthEnabled() +} + func (c *Authentication) serviceAccountAuthEnabled() bool { return c.BuiltInOptions.ServiceAccounts != nil && len(c.BuiltInOptions.ServiceAccounts.KeyFiles) != 0 } @@ -129,6 +135,9 @@ func (c *Authentication) ApplyTo(ctx context.Context, authenticationInfo *generi // Set for service account auth, if enabled if c.serviceAccountAuthEnabled() { + if rootShardConfig == nil { + return fmt.Errorf("service account authentication needs --root-kubeconfig") + } authenticationInfo.APIAudiences = c.BuiltInOptions.APIAudiences if len(c.BuiltInOptions.ServiceAccounts.Issuers) != 0 && len(c.BuiltInOptions.APIAudiences) == 0 { authenticationInfo.APIAudiences = c.BuiltInOptions.ServiceAccounts.Issuers diff --git a/pkg/proxy/options/options.go b/pkg/proxy/options/options.go index 878b63bd4f4..3f6e64c8529 100644 --- a/pkg/proxy/options/options.go +++ b/pkg/proxy/options/options.go @@ -32,10 +32,8 @@ type Options struct { RootDirectory string // RootKubeconfig points at the root shard. // - // Deprecated: shard discovery uses ShardsPeerKubeconfigs (the Admin - // workspace of the peer shards) instead. The root kubeconfig is still - // required for APIExport identity resolution and will be removed once - // identities are resolvable without the root shard. + // TODO: This is going to be removed once service account + // authentication no longer requires the root shard. RootKubeconfig string ShardsKubeconfig string ShardsPeerKubeconfigs []string @@ -65,12 +63,12 @@ func (o *Options) AddFlags(fss *cliflag.NamedFlagSets) { fs := fss.FlagSet("proxy") fs.StringVar(&o.MappingFile, "mapping-file", o.MappingFile, "Config file mapping paths to backends") fs.StringVar(&o.RootDirectory, "root-directory", o.RootDirectory, "Root directory.") - fs.StringVar(&o.RootKubeconfig, "root-kubeconfig", o.RootKubeconfig, "The path to the kubeconfig of the root shard.") - if err := fs.MarkDeprecated("root-kubeconfig", "use --shard-peer-kubeconfig for shard discovery; --root-kubeconfig is still required for APIExport identity resolution and will be removed once identities are resolvable without the root shard."); err != nil { + fs.StringVar(&o.RootKubeconfig, "root-kubeconfig", o.RootKubeconfig, "The path to the kubeconfig of the root shard. Only needed for service account authentication; without --shard-peer-kubeconfig its server also acts as the seed peer.") + if err := fs.MarkDeprecated("root-kubeconfig", "use --shard-peer-kubeconfig; the root shard is needed neither for shard discovery nor for the identities of the core APIs any more. It is still required for service account authentication."); err != nil { panic(err) } fs.StringVar(&o.ShardsKubeconfig, "shards-kubeconfig", o.ShardsKubeconfig, "The path to the kubeconfig used for communication with all shards. The server name if provided is replaced with a shard's hostname.") - fs.StringSliceVar(&o.ShardsPeerKubeconfigs, "shard-peer-kubeconfig", o.ShardsPeerKubeconfigs, "Repeatable path to a kubeconfig whose named clusters are seed peer shards; all files' clusters are merged into one seed list. The proxy discovers Shards through the Admin workspace (/services/admin) of the peers - requests are distributed round-robin with failover - instead of watching the root shard. If unset, the --root-kubeconfig server acts as the only seed peer. The seeds are used for bootstrapping and as a last-resort fallback only: once shards are discovered, requests go to the discovered shards, which join and leave the peer set at runtime, so the files do not have to track short-lived shards.") + fs.StringSliceVar(&o.ShardsPeerKubeconfigs, "shard-peer-kubeconfig", o.ShardsPeerKubeconfigs, "Repeatable path to a kubeconfig whose named clusters are seed peer shards; all files' clusters are merged into one seed list. The proxy discovers Shards and the identities of the core APIs through the Admin workspace (/services/admin) of the peers - requests are distributed round-robin with failover - and never talks to the root shard for them. If unset, the --root-kubeconfig server acts as the only seed peer. The seeds are used for bootstrapping and as a last-resort fallback only: once shards are discovered, requests go to the discovered shards, which join and leave the peer set at runtime, so the files do not have to track short-lived shards.") fs.StringVar(&o.ProfilerAddress, "profiler-address", "", "[Address]:port to bind the profiler to") fs.StringSliceVar(&o.CorsAllowedOriginList, "cors-allowed-origins", o.CorsAllowedOriginList, "List of allowed origins for CORS, comma separated. An allowed origin can be a regular expression to support subdomain matching. If this list is empty CORS will not be enabled.") } @@ -103,6 +101,12 @@ func (o *Options) Validate() []error { if len(o.ShardsKubeconfig) == 0 { errs = append(errs, fmt.Errorf("--shards-kubeconfig is required")) } + if len(o.ShardsPeerKubeconfigs) == 0 && len(o.RootKubeconfig) == 0 { + errs = append(errs, fmt.Errorf("--shard-peer-kubeconfig is required")) + } + if o.Authentication.ServiceAccountAuthEnabled() && len(o.RootKubeconfig) == 0 { + errs = append(errs, fmt.Errorf("--root-kubeconfig is required for service account authentication")) + } errs = append(errs, o.SecureServing.Validate()...) errs = append(errs, o.Authentication.Validate()...) diff --git a/pkg/proxy/server.go b/pkg/proxy/server.go index 34d5f7cf56a..c5842496836 100644 --- a/pkg/proxy/server.go +++ b/pkg/proxy/server.go @@ -24,12 +24,13 @@ import ( "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/runtime/serializer" - "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apiserver/pkg/authentication/authenticator" "k8s.io/apiserver/pkg/authorization/authorizer" + "k8s.io/apiserver/pkg/authorization/union" genericapifilters "k8s.io/apiserver/pkg/endpoints/filters" "k8s.io/apiserver/pkg/endpoints/request" genericfilters "k8s.io/apiserver/pkg/server/filters" @@ -40,6 +41,7 @@ import ( "k8s.io/klog/v2" kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" + adminv1alpha1 "github.com/kcp-dev/sdk/apis/admin/v1alpha1" "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" @@ -64,10 +66,10 @@ type Server struct { IndexController *index.Controller AuthController *authentication.Controller KcpSharedInformerFactory kcpinformers.SharedScopedInformerFactory - // RootAPIExportInformerFactory watches the APIExports of the root workspace - // on the root shard. Their identities are injected into wildcard - // requests and follow an identity rotation through this informer. - RootAPIExportInformerFactory kcpinformers.SharedScopedInformerFactory + // CoreIdentityInformerFactory watches the APIIdentity registry entries + // owned by the root workspace's APIExports, through the Admin workspace + // of the peers. Their identities are injected into wildcard requests. + CoreIdentityInformerFactory kcpinformers.SharedScopedInformerFactory } func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { @@ -94,22 +96,22 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { // the root workspace is not read for discovery. The Admin workspace // accepts /clusters// paths, so the standard scoped informer // factory works against it unchanged. - rootShardConfigInformerClient, err := kcpclientset.NewForConfig(s.CompletedConfig.PeersConfig) + peersClient, err := kcpclientset.NewForConfig(s.CompletedConfig.PeersConfig) if err != nil { return s, fmt.Errorf("failed to create client for informers: %w", err) } - s.KcpSharedInformerFactory = kcpinformers.NewSharedScopedInformerFactoryWithOptions(rootShardConfigInformerClient.Cluster(core.RootCluster.Path()), 30*time.Minute) + s.KcpSharedInformerFactory = kcpinformers.NewSharedScopedInformerFactoryWithOptions(peersClient.Cluster(core.RootCluster.Path()), 30*time.Minute) // The identities of the core APIs are those of the root workspace's - // APIExports, which only the root shard serves. An identity rotation - // flips an export's status.identityHash; the handler swaps the injected - // identity as the change arrives. - rootShardKcpClient, err := kcpclientset.NewForConfig(s.CompletedConfig.RootShardConfig) - if err != nil { - return s, fmt.Errorf("failed to create root shard client: %w", err) - } - s.RootAPIExportInformerFactory = kcpinformers.NewSharedScopedInformerFactoryWithOptions(rootShardKcpClient.Cluster(core.RootCluster.Path()), 30*time.Minute) - _, _ = s.RootAPIExportInformerFactory.Apis().V1alpha2().APIExports().Informer().AddEventHandler(s.CompletedConfig.CoreIdentityHandler) + // APIExports. They are read from the APIIdentity registry, which the + // Admin workspace of every peer serves from the cache server, so the + // root shard is not needed for them: an own factory, restricted to the + // entries flagged as core, feeds the identity handler. + s.CoreIdentityInformerFactory = kcpinformers.NewSharedScopedInformerFactoryWithOptions(peersClient.Cluster(core.RootCluster.Path()), 30*time.Minute, + kcpinformers.WithTweakListOptions(func(options *metav1.ListOptions) { + options.LabelSelector = labels.SelectorFromSet(labels.Set{adminv1alpha1.APIIdentityCoreLabel: "true"}).String() + })) + _, _ = s.CoreIdentityInformerFactory.Admin().V1alpha1().APIIdentities().Informer().AddEventHandler(s.CompletedConfig.CoreIdentityHandler) // Feed every discovered shard into the peer failover set: the configured // peers are seeds for bootstrapping and last-resort fallback only, and @@ -209,17 +211,22 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { handler = genericfilters.WithCORS(handler, c.Options.CorsAllowedOriginList, nil, nil, nil, "true") mux := http.NewServeMux() - healthz.InstallReadyzHandler(mux, healthz.NewInformerSyncHealthz(s.KcpSharedInformerFactory)) + healthz.InstallReadyzHandler(mux, + healthz.NewInformerSyncHealthz(s.KcpSharedInformerFactory), + healthz.NamedCheck("core-identities", func(*http.Request) error { return s.CompletedConfig.CoreIdentitiesReady() }), + ) healthz.InstallLivezHandler(mux, healthz.PingHealthz) // /metrics is served locally by the front proxy (it is not proxied to a // shard), so it needs its own authentication and authorization: without it // the endpoint is world-readable and leaks shard-wide operational data - // (logical-cluster counts, client-cert expiry, request patterns). Access is - // delegated to :root via SubjectAccessReview, so the same - // system:kcp:metrics-reader binding that authorizes scraping on the shards - // authorizes it here too. Registered before "/" so it takes precedence over - // the proxy handler for this exact path. + // (logical-cluster counts, client-cert expiry, request patterns). Members + // of the system:kcp:admin and system:kcp:metrics-reader groups may + // scrape; with --root-kubeconfig a :root SubjectAccessReview + // is consulted as well, so the system:kcp:metrics-reader binding that + // authorizes scraping on the shards keeps working here during the + // deprecation. Registered before "/" so it takes precedence over the proxy + // handler for this exact path. metricsHandler, err := newProtectedMetricsHandler(c.AuthenticationInfo.Authenticator, c.RootShardConfig, requestInfoFactory) if err != nil { return nil, fmt.Errorf("failed to create protected metrics handler: %w", err) @@ -233,16 +240,20 @@ func NewServer(ctx context.Context, c CompletedConfig) (*Server, error) { } func newProtectedMetricsHandler(auth authenticator.Request, rootShardConfig *restclient.Config, requestInfoResolver request.RequestInfoResolver) (http.Handler, error) { - kubeClusterClient, err := kcpkubernetesclientset.NewForConfig(rootShardConfig) - if err != nil { - return nil, fmt.Errorf("failed to create root shard client for metrics authorization: %w", err) - } - authz, err := delegated.NewDelegatedAuthorizer(core.RootCluster, kubeClusterClient, delegated.Options{}) - if err != nil { - return nil, fmt.Errorf("failed to create metrics authorizer: %w", err) + authorizers := []authorizer.Authorizer{newMetricsGroupAuthorizer()} + if rootShardConfig != nil { + kubeClusterClient, err := kcpkubernetesclientset.NewForConfig(rootShardConfig) + if err != nil { + return nil, fmt.Errorf("failed to create root shard client for metrics authorization: %w", err) + } + authz, err := delegated.NewDelegatedAuthorizer(core.RootCluster, kubeClusterClient, delegated.Options{}) + if err != nil { + return nil, fmt.Errorf("failed to create metrics authorizer: %w", err) + } + authorizers = append(authorizers, authz) } - return withMetricsAuthorization(legacyregistry.Handler(), auth, authz, requestInfoResolver), nil + return withMetricsAuthorization(legacyregistry.Handler(), auth, union.New(authorizers...), requestInfoResolver), nil } func withMetricsAuthorization(h http.Handler, auth authenticator.Request, authz authorizer.Authorizer, requestInfoResolver request.RequestInfoResolver) http.Handler { @@ -268,21 +279,18 @@ func (s *Server) PrepareRun(ctx context.Context) (preparedServer, error) { func (s preparedServer) Run(ctx context.Context) error { logger := klog.FromContext(ctx).WithValues("component", "proxy") - var identityErr error - if err := wait.PollUntilContextCancel(ctx, time.Millisecond*500, true, func(ctx context.Context) (bool, error) { - if err := s.CompletedConfig.ResolveIdentities(ctx); err != nil { - logger.V(3).Info("failed to resolve identities, keeping trying", "err", err) - identityErr = err - return false, nil - } - return true, nil - }); err != nil { - err = fmt.Errorf("failed to get or create identities: %w", err) - if identityErr != nil { - err = fmt.Errorf("%v: %w", err, identityErr) - } - return err + // Both factories read the Admin workspace of the peers: any reachable + // shard suffices, the root shard is not special. The core identities + // arrive through the second one; until they do, requests that need them + // fail and readiness reports it, nothing blocks here on the root shard. + s.KcpSharedInformerFactory.Start(ctx.Done()) + s.CoreIdentityInformerFactory.Start(ctx.Done()) + s.KcpSharedInformerFactory.WaitForCacheSync(ctx.Done()) + s.CoreIdentityInformerFactory.WaitForCacheSync(ctx.Done()) + if err := s.CompletedConfig.CoreIdentitiesReady(); err != nil { + logger.Info("starting without all core identities, waiting for the registry", "reason", err.Error()) } + // start indexes go s.IndexController.Start(ctx, 2) @@ -290,12 +298,6 @@ func (s preparedServer) Run(ctx context.Context) error { go s.AuthController.Start(ctx, 2) } - s.KcpSharedInformerFactory.Start(ctx.Done()) - s.KcpSharedInformerFactory.WaitForCacheSync(ctx.Done()) - // not waited for: the identities are resolved already, and the proxy - // must not depend on the root shard to become ready. - s.RootAPIExportInformerFactory.Start(ctx.Done()) - doneCh, _, err := s.CompletedConfig.ServingInfo.Serve(s.Handler, time.Second*60, ctx.Done()) if err != nil { return err diff --git a/pkg/server/bootstrap/identity.go b/pkg/server/bootstrap/identity.go index c1492739615..d081ec518b0 100644 --- a/pkg/server/bootstrap/identity.go +++ b/pkg/server/bootstrap/identity.go @@ -37,6 +37,7 @@ import ( kcpkubernetesclientset "github.com/kcp-dev/client-go/kubernetes" "github.com/kcp-dev/logicalcluster/v3" + adminv1alpha1 "github.com/kcp-dev/sdk/apis/admin/v1alpha1" 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" @@ -63,6 +64,25 @@ var ( } ) +// IsCoreExport reports whether the APIExport named by cluster and name is +// one of the core APIExports in the root workspace, those listed in +// KcpRootGroupExportNames and KcpRootGroupResourceExportNames, whose +// identities every shard and the front-proxy need before they can serve. +func IsCoreExport(cluster logicalcluster.Name, name string) bool { + return cluster == core.RootCluster && coreExportNames.Has(name) +} + +var coreExportNames = func() sets.Set[string] { + names := sets.New[string]() + for _, name := range KcpRootGroupExportNames { + names.Insert(name) + } + for _, name := range KcpRootGroupResourceExportNames { + names.Insert(name) + } + return names +}() + type identities struct { lock sync.RWMutex @@ -237,6 +257,68 @@ func NewWildcardIdentitiesWrappingRoundTripper(groupExportNames map[string]strin apiExportHandler(ids) } +// NewCoreWildcardIdentities creates the identity-injecting round tripper +// for the core APIExports of the root workspace, the ones named by the +// given maps (see KcpRootGroupExportNames). Their identities come from the +// APIIdentity view, which the front-proxy reads through the Admin workspace +// of its peer shards. No other identity is tracked here: nothing in kcp +// injects a provider's identity into wildcard URLs, consumers reach those +// APIs through bindings. +// +// The returned handler is registered on an informer of the APIIdentity +// entries; it picks the core entries out and feeds their identities as they +// arrive. The returned ready func fails while any core identity is still +// unknown. Requests that need an identity fail until it arrives. +func NewCoreWildcardIdentities(groupExportNames map[string]string, + groupResourceExportNames map[schema.GroupResource]string) (wrap func(rt http.RoundTripper) http.RoundTripper, handler cache.ResourceEventHandler, ready func() error) { + ids := newIdentities(groupExportNames, groupResourceExportNames) + return injectKcpIdentities(ids), coreIdentityHandler(ids), ids.ready +} + +// coreIdentityHandler returns an informer event handler that keeps the +// identities of the core APIExports in sync with the APIIdentity view: an +// entry owned by one of them, in the root workspace, names that export's +// identity. Every other entry is ignored, and so are deletions: an identity +// in use is only ever replaced by a new one. +func coreIdentityHandler(ids *identities) cache.ResourceEventHandler { + update := func(obj interface{}) { + entry, ok := obj.(*adminv1alpha1.APIIdentity) + if !ok { + return + } + owner := entry.Status.Owner + if !IsCoreExport(logicalcluster.Name(owner.Cluster), owner.Name) { + return + } + ids.set(owner.Name, entry.Name) + } + return cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { update(obj) }, + UpdateFunc: func(_, obj interface{}) { update(obj) }, + } +} + +// ready fails while the identity of a core APIExport is unknown. +func (ids *identities) ready() error { + ids.lock.RLock() + defer ids.lock.RUnlock() + missing := sets.New[string]() + for group, id := range ids.groupIdentities { + if id == "" { + missing.Insert(ids.groupExportNames[group]) + } + } + for gr, id := range ids.groupResourceIdentities { + if id == "" { + missing.Insert(ids.groupResourceExportNames[gr]) + } + } + if missing.Len() == 0 { + return nil + } + return fmt.Errorf("the identities of the core APIExports %v are not known yet", sets.List(missing)) +} + // apiExportHandler returns an informer event handler that keeps the // identities in sync with status.identityHash of the APIExports in the root // workspace. An identity rotation flips that hash; the handler swaps the @@ -250,23 +332,18 @@ func NewWildcardIdentitiesWrappingRoundTripper(groupExportNames map[string]strin // APIExports: the shard's local one, one against the cache server, or a // scoped one on the root workspace. func apiExportHandler(ids *identities) cache.ResourceEventHandler { - exportNames := sets.New[string]() - for _, name := range ids.groupExportNames { - exportNames.Insert(name) - } - for _, name := range ids.groupResourceExportNames { - exportNames.Insert(name) - } - update := func(obj interface{}) { export, ok := obj.(*apisv1alpha2.APIExport) if !ok { return } - if cluster := logicalcluster.From(export); !cluster.Empty() && cluster != core.RootCluster { - return + // A scoped informer on the root workspace hands out objects without + // the cluster annotation; those are root's. + cluster := logicalcluster.From(export) + if cluster.Empty() { + cluster = core.RootCluster } - if !exportNames.Has(export.Name) || export.Status.IdentityHash == "" { + if !IsCoreExport(cluster, export.Name) || export.Status.IdentityHash == "" { return } ids.set(export.Name, export.Status.IdentityHash) diff --git a/pkg/server/bootstrap/identity_test.go b/pkg/server/bootstrap/identity_test.go index 6690410d1d7..7c2b897e7a9 100644 --- a/pkg/server/bootstrap/identity_test.go +++ b/pkg/server/bootstrap/identity_test.go @@ -38,6 +38,7 @@ import ( "k8s.io/client-go/tools/cache" "github.com/kcp-dev/logicalcluster/v3" + adminv1alpha1 "github.com/kcp-dev/sdk/apis/admin/v1alpha1" apisv1alpha2 "github.com/kcp-dev/sdk/apis/apis/v1alpha2" ) @@ -162,6 +163,82 @@ func TestAPIExportHandler(t *testing.T) { }) } +func TestCoreIdentityHandler(t *testing.T) { + t.Parallel() + + tenancyPath := "/clusters/*/apis/tenancy.kcp.io/v1alpha1/workspaces" + shardsPath := "/clusters/*/apis/core.kcp.io/v1alpha1/shards" + + t.Run("feeds the identities of the core exports in root and reports readiness", func(t *testing.T) { + t.Parallel() + wrap, handler, ready := NewCoreWildcardIdentities(testGroupExportNames, testGroupResourceExportNames) + require.NotNil(t, wrap) + require.Error(t, ready(), "nothing is known yet") + + handler.OnAdd(entry("root", "tenancy.kcp.io", oldHash), true) + require.Error(t, ready(), "the shards identity is still unknown") + + handler.OnAdd(entry("root", "shards.core.kcp.io", newHash), true) + require.NoError(t, ready()) + }) + + ignored := []struct { + name string + obj interface{} + }{ + {name: "an entry owned by an export of another workspace", obj: entry("root:org", "tenancy.kcp.io", newHash)}, + {name: "an entry owned by an export that serves no core API", obj: entry("root", "cowboys", newHash)}, + {name: "a tombstone", obj: cache.DeletedFinalStateUnknown{Key: newHash, Obj: entry("root", "tenancy.kcp.io", newHash)}}, + {name: "another object", obj: export("root", "tenancy.kcp.io", newHash)}, + } + for _, scenario := range ignored { + t.Run("ignores "+scenario.name, func(t *testing.T) { + t.Parallel() + ids := newIdentities(testGroupExportNames, testGroupResourceExportNames) + coreIdentityHandler(ids).OnAdd(entry("root", "tenancy.kcp.io", oldHash), true) + + handler := coreIdentityHandler(ids) + handler.OnAdd(scenario.obj, false) + handler.OnUpdate(scenario.obj, scenario.obj) + + if got, want := decorate(t, ids, tenancyPath), tenancyPath+":"+oldHash; got != want { + t.Errorf("expected %q, got %q", want, got) + } + }) + } + + t.Run("replaces the identity of the changed export only", func(t *testing.T) { + t.Parallel() + ids := newIdentities(testGroupExportNames, testGroupResourceExportNames) + handler := coreIdentityHandler(ids) + handler.OnAdd(entry("root", "tenancy.kcp.io", oldHash), true) + handler.OnAdd(entry("root", "shards.core.kcp.io", oldHash), true) + watch := newExpirableWatchBody(io.NopCloser(strings.NewReader("")), true) + ids.trackWatch(oldHash, watch) + + handler.OnUpdate(entry("root", "tenancy.kcp.io", oldHash), entry("root", "tenancy.kcp.io", newHash)) + + if got, want := decorate(t, ids, tenancyPath), tenancyPath+":"+newHash; got != want { + t.Errorf("expected %q, got %q", want, got) + } + if got, want := decorate(t, ids, shardsPath), shardsPath+":"+oldHash; got != want { + t.Errorf("an unchanged export must keep its identity: expected %q, got %q", want, got) + } + if !watch.expired { + t.Error("watches using the replaced identity must be expired") + } + }) +} + +// entry is a registry entry named after the identity and owned by the +// export of the given name in the given cluster. +func entry(cluster, exportName, identity string) *adminv1alpha1.APIIdentity { + return &adminv1alpha1.APIIdentity{ + ObjectMeta: metav1.ObjectMeta{Name: identity}, + Status: adminv1alpha1.APIIdentityStatus{Owner: adminv1alpha1.ExportReference{Cluster: cluster, Name: exportName}}, + } +} + // TestWatchExpiresOnIdentityChange runs a real watch through the round // tripper and checks that an identity change ends it with a 410 Expired // error event, which makes reflectors relist.