feat(apiserver): KPEP-0001 — --storage-backend wiring + typed lister NotFound

cmd/apiserver/app/config.go

@@ -18,6 +18,7 @@ package app
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"reflect"
 	"strings"
@@ -32,6 +33,8 @@ import (
 	genericfilters "k8s.io/apiserver/pkg/endpoints/filters"
 	"k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/storage"
+	storagefeature "k8s.io/apiserver/pkg/storage/feature"
 	serverfilters "k8s.io/apiserver/pkg/server/filters"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/apiserver/pkg/util/webhook"
@@ -129,6 +132,38 @@ func NewConfig(opts options.CompletedOptions) (*Config, error) {
 	clientPool := mc.NewClientPool(genericConfig.LoopbackClientConfig, mcOpts.PathPrefix, mcOpts.ControlPlaneSegment)
 	informerRegistry := mc.NewInformerRegistry(wait.ContextForChannel(genericConfig.DrainedNotify()))
 	mcOpts.InformerRegistry = informerRegistry
+
+	// Storage backend dispatch. Options.Complete has already registered the
+	// selected backend with the fork's factory registry (so upstream
+	// EtcdOptions.Validate passed and every internal factory.Create
+	// callsite — master/peer endpoint leases, service IP/NodePort
+	// allocators — dispatches to it at runtime). Here we install the same
+	// backend's CR-storage Factory on the multicluster decorator hook.
+	// spanner.Options.Build reuses the FactoryBackend created in Complete
+	// (one shared spanner.Client per process).
+	backendName := ""
+	if opts.Etcd != nil {
+		backendName = opts.Etcd.StorageConfig.Type
+	}
+	if backendName != "" && backendName != "etcd3" && backendName != "etcd2" {
+		backend, ok := opts.Backends.Get(backendName)
+		if !ok {
+			return nil, fmt.Errorf(
+				"unknown --storage-backend=%q (registered: %v)",
+				backendName, opts.Backends.Names(),
+			)
+		}
+		crFactory, err := backend.Build()
+		if err != nil {
+			return nil, fmt.Errorf("building --storage-backend=%q: %w", backendName, err)
+		}
+		mcOpts.BackendFactory = crFactory
+		// Non-etcd backends typically implement RequestWatchProgress
+		// natively (Spanner via its broadcaster). Tell the feature
+		// checker so the cacher allows SendInitialEvents watches.
+		storagefeature.SetFeatureSupported(storage.RequestWatchProgress, true)
+	}
+
 	var crdRuntimeMgr *mcbootstrap.CRDRuntimeManager
 	systemNamespaceBootstrapper := mcbootstrap.NewSystemNamespaceBootstrapper(mcbootstrap.SystemNamespaceOptions{
 		ClientForCluster: clientPool.KubeClientForCluster,

cmd/apiserver/app/options/completion.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 
 	apiserveroptions "k8s.io/apiserver/pkg/server/options"
+	storagefactory "k8s.io/apiserver/pkg/storage/storagebackend/factory"
 	_ "k8s.io/component-base/metrics/prometheus/workqueue"
 	netutils "k8s.io/utils/net"
 
@@ -49,6 +50,42 @@ func (s *ServerRunOptions) Complete(ctx context.Context) (CompletedOptions, erro
 		return CompletedOptions{completedOptions: &completedOptions{}}, nil
 	}
 
+	// Register the selected non-etcd3 backend with the fork's factory
+	// registry BEFORE downstream Validate runs. EtcdOptions rejects
+	// --storage-backend values it doesn't recognize, but it consults
+	// factory.IsRegistered first — registering here means
+	// "--storage-backend=spanner" passes upstream validation, and every
+	// factory.Create callsite inside upstream (CR storage, master/peer
+	// endpoint leases, service IP/NodePort allocators) dispatches to the
+	// same Spanner backend at runtime. spanner.Options.BuildFactoryBackend
+	// is idempotent (one shared client per Options) so Build/the decorator
+	// hook in config.go reuse the same instance.
+	//
+	// The backend name comes from upstream EtcdOptions's --storage-backend
+	// flag (bound to s.Etcd.StorageConfig.Type). Empty / "etcd3" / "etcd2"
+	// keep the upstream default path.
+	backendName := ""
+	if s.Etcd != nil {
+		backendName = s.Etcd.StorageConfig.Type
+	}
+	if backendName != "" && backendName != "etcd3" && backendName != "etcd2" {
+		backend, ok := s.Backends.Get(backendName)
+		if !ok {
+			return CompletedOptions{}, fmt.Errorf(
+				"unknown --storage-backend=%q (registered: %v)",
+				backendName, s.Backends.Names(),
+			)
+		}
+		if errs := backend.Validate(); len(errs) > 0 {
+			return CompletedOptions{}, fmt.Errorf("validating --storage-backend=%q: %v", backendName, errs)
+		}
+		fb, err := backend.BuildFactoryBackend()
+		if err != nil {
+			return CompletedOptions{}, fmt.Errorf("building factory backend for --storage-backend=%q: %w", backendName, err)
+		}
+		storagefactory.Register(backendName, fb)
+	}
+
 	// process s.ServiceClusterIPRange from list to Primary and Secondary
 	// we process secondary only if provided by user
 	apiServerServiceIP, primaryServiceIPRange, secondaryServiceIPRange, err := getServiceIPAndRanges(s.ServiceClusterIPRanges)

cmd/apiserver/app/options/flags_test.go

@@ -0,0 +1,55 @@
+/*
+Copyright 2026 The Kubernetes 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
+*/
+
+package options
+
+import (
+	"testing"
+
+	"github.com/spf13/pflag"
+)
+
+// TestStorageBackendFlagRoutesToEtcdConfig guards the backend-selection
+// flag wiring. The backend selection comes from upstream EtcdOptions's
+// --storage-backend flag, which binds directly to s.Etcd.StorageConfig.Type.
+// Options.Complete reads that field to decide which backend to register
+// with the fork factory registry.
+//
+// A previous version of kplane's Flags() registered a SECOND --storage-backend
+// pflag pointing at a separate kplane-side field. pflag's NamedFlagSets
+// merge silently collapses duplicate registrations, but which destination
+// pointer "wins" depends on the order flag-sets are added — so half the
+// time the upstream field stayed empty after parse, Complete skipped
+// Register, and upstream Validate then rejected the backend as unknown.
+//
+// This test asserts that after parsing --storage-backend=<name>, the value
+// lands in s.Etcd.StorageConfig.Type (the field Complete actually reads).
+// Re-introducing a duplicate registration anywhere in Flags() fails the
+// test because the new flag's destination wins parsing instead.
+func TestStorageBackendFlagRoutesToEtcdConfig(t *testing.T) {
+	s := NewServerRunOptions()
+
+	fss := s.Flags()
+	merged := pflag.NewFlagSet("merged", pflag.ContinueOnError)
+	for _, fs := range fss.FlagSets {
+		merged.AddFlagSet(fs)
+	}
+
+	if err := merged.Parse([]string{"--storage-backend=spanner"}); err != nil {
+		t.Fatalf("Parse failed: %v", err)
+	}
+
+	if s.Etcd == nil {
+		t.Fatal("s.Etcd is nil after NewServerRunOptions")
+	}
+	if got := s.Etcd.StorageConfig.Type; got != "spanner" {
+		t.Fatalf("after --storage-backend=spanner, s.Etcd.StorageConfig.Type=%q; want \"spanner\"", got)
+	}
+}

cmd/apiserver/app/options/options.go

@@ -23,6 +23,8 @@ import (
 	"time"
 
 	mc "github.com/kplane-dev/apiserver/pkg/multicluster"
+	storagebackends "github.com/kplane-dev/storage/backends"
+	"github.com/kplane-dev/storage/registry"
 	v1 "k8s.io/api/core/v1"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	cliflag "k8s.io/component-base/cli/flag"
@@ -64,6 +66,13 @@ type Extra struct {
 	EndpointReconcilerType string
 
 	MasterCount int
+
+	// Backends is the instance-scoped registry of available storage
+	// backends. The default set is populated in NewServerRunOptions via
+	// storagebackends.RegisterBuiltin. Custom apiservers can call
+	// Backends.Register(<their>.NewOptions()) on the returned value
+	// before Flags() is invoked.
+	Backends *registry.Backends
 }
 
 const (
@@ -75,10 +84,14 @@ const (
 
 // NewServerRunOptions creates and returns ServerRunOptions according to the given featureGate and effectiveVersion of the server binary to run.
 func NewServerRunOptions() *ServerRunOptions {
+	backends := registry.New()
+	storagebackends.RegisterBuiltin(backends)
+
 	s := ServerRunOptions{
 		Options: controlplaneapiserver.NewOptions(),
 
 		Extra: Extra{
+			Backends:               backends,
 			EndpointReconcilerType: string(reconcilers.LeaseEndpointReconcilerType),
 			KubeletConfig: kubeletclient.KubeletClientConfig{
 				Port:         ports.KubeletPort,
@@ -124,6 +137,16 @@ func (s *ServerRunOptions) Flags() (fss cliflag.NamedFlagSets) {
 	mcfs.StringVar(&s.ServiceCIDRSharingMode, "service-cidr-sharing-mode", s.ServiceCIDRSharingMode,
 		"Service CIDR strategy across control planes: 'shared' keeps root-managed defaults only; 'per-cluster' bootstraps default ServiceCIDR in each virtual control plane.")
 
+	// Backend-specific flags (e.g. --spanner-project) are bound by each
+	// backend's AddFlags below. The selection itself reuses upstream
+	// EtcdOptions's --storage-backend flag (bound to
+	// s.Etcd.StorageConfig.Type) — Options.Complete reads that value and
+	// registers the chosen backend with the fork factory registry.
+	// Registering a second --storage-backend pflag here would collide
+	// non-deterministically with the upstream one.
+	storagefs := fss.FlagSet("storage backend")
+	s.Backends.AddFlags(storagefs)
+
 	fs := fss.FlagSet("misc")
 
 	fs.BoolVar(&s.AllowPrivileged, "allow-privileged", s.AllowPrivileged,