Merge pull request #2836 from sttts/sttts-server-split-virtual

🌱  tmc: split apart virtual workspaces

cmd/kcp/options/options.go

@@ -22,14 +22,14 @@ import (
 	cliflag "k8s.io/component-base/cli/flag"
 
 	kcpcoreoptions "github.com/kcp-dev/kcp/cmd/kcp-core/options"
-	serveroptions "github.com/kcp-dev/kcp/tmc/pkg/server/options"
+	tmcserveroptions "github.com/kcp-dev/kcp/tmc/pkg/server/options"
 )
 
 type Options struct {
 	Output io.Writer
 
 	Generic kcpcoreoptions.GenericOptions
-	Server  serveroptions.Options
+	Server  tmcserveroptions.Options
 	Extra   ExtraOptions
 }
 
@@ -39,7 +39,7 @@ func NewOptions(rootDir string) *Options {
 	opts := &Options{
 		Output: nil,
 
-		Server:  *serveroptions.NewOptions(rootDir),
+		Server:  *tmcserveroptions.NewOptions(rootDir),
 		Generic: *kcpcoreoptions.NewGeneric(rootDir),
 		Extra:   ExtraOptions{},
 	}
@@ -51,7 +51,7 @@ type completedOptions struct {
 	Output io.Writer
 
 	Generic kcpcoreoptions.GenericOptions
-	Server  serveroptions.CompletedOptions
+	Server  tmcserveroptions.CompletedOptions
 	Extra   ExtraOptions
 }
 

cmd/virtual-workspaces/command/cmd.go

@@ -45,6 +45,7 @@ import (
 	kcpfeatures "github.com/kcp-dev/kcp/pkg/features"
 	"github.com/kcp-dev/kcp/pkg/server/bootstrap"
 	virtualrootapiserver "github.com/kcp-dev/kcp/pkg/virtual/framework/rootapiserver"
+	corevwoptions "github.com/kcp-dev/kcp/pkg/virtual/options"
 )
 
 func NewCommand(ctx context.Context, errout io.Writer) *cobra.Command {
@@ -149,10 +150,6 @@ func Run(ctx context.Context, o *options.Options) error {
 	}
 
 	// create apiserver
-	virtualWorkspaces, err := o.VirtualWorkspaces.NewVirtualWorkspaces(identityConfig, o.RootPathPrefix, wildcardKubeInformers, wildcardKcpInformers, cacheKcpInformers)
-	if err != nil {
-		return err
-	}
 	scheme := runtime.NewScheme()
 	metav1.AddToGroupVersion(scheme, schema.GroupVersion{Group: "", Version: "v1"})
 	codecs := serializer.NewCodecFactory(scheme)
@@ -163,23 +160,36 @@ func Run(ctx context.Context, o *options.Options) error {
 	if err := o.Authentication.ApplyTo(&recommendedConfig.Authentication, recommendedConfig.SecureServing, recommendedConfig.OpenAPIConfig); err != nil {
 		return err
 	}
-	if err := o.Authorization.ApplyTo(&recommendedConfig.Config, virtualWorkspaces); err != nil {
+	if err := o.Audit.ApplyTo(&recommendedConfig.Config); err != nil {
 		return err
 	}
-	if err := o.Audit.ApplyTo(&recommendedConfig.Config); err != nil {
+
+	rootAPIServerConfig, err := virtualrootapiserver.NewConfig(recommendedConfig)
+	if err != nil {
+		return err
+	}
+
+	if err := o.Authorization.ApplyTo(&recommendedConfig.Config, func() []virtualrootapiserver.NamedVirtualWorkspace {
+		return rootAPIServerConfig.Extra.VirtualWorkspaces
+	}); err != nil {
+		return err
+	}
+
+	coreVWs, err := o.CoreVirtualWorkspaces.NewVirtualWorkspaces(identityConfig, o.RootPathPrefix, wildcardKubeInformers, wildcardKcpInformers, cacheKcpInformers)
+	if err != nil {
+		return err
+	}
+	tmcVWs, err := o.TmcVirtualWorkspaces.NewVirtualWorkspaces(identityConfig, o.RootPathPrefix, cacheKcpInformers)
+	if err != nil {
 		return err
 	}
-	rootAPIServerConfig, err := virtualrootapiserver.NewRootAPIConfig(recommendedConfig, []virtualrootapiserver.InformerStart{
-		wildcardKubeInformers.Start,
-		wildcardKcpInformers.Start,
-		cacheKcpInformers.Start,
-	}, virtualWorkspaces)
+	rootAPIServerConfig.Extra.VirtualWorkspaces, err = corevwoptions.Merge(coreVWs, tmcVWs)
 	if err != nil {
 		return err
 	}
 
 	completedRootAPIServerConfig := rootAPIServerConfig.Complete()
-	rootAPIServer, err := completedRootAPIServerConfig.New(genericapiserver.NewEmptyDelegate())
+	rootAPIServer, err := virtualrootapiserver.NewServer(completedRootAPIServerConfig, genericapiserver.NewEmptyDelegate())
 	if err != nil {
 		return err
 	}
@@ -190,8 +200,15 @@ func Run(ctx context.Context, o *options.Options) error {
 		return err
 	}
 
-	logger.Info("Starting virtual workspace apiserver on ", "externalAddress", rootAPIServerConfig.GenericConfig.ExternalAddress, "version", version.Get().String())
+	logger.Info("Starting informers")
+	wildcardKubeInformers.Start(ctx.Done())
+	wildcardKcpInformers.Start(ctx.Done())
+	cacheKcpInformers.Start(ctx.Done())
+	wildcardKubeInformers.WaitForCacheSync(ctx.Done())
+	wildcardKcpInformers.WaitForCacheSync(ctx.Done())
+	cacheKcpInformers.WaitForCacheSync(ctx.Done())
 
+	logger.Info("Starting virtual workspace apiserver on ", "externalAddress", rootAPIServerConfig.Generic.ExternalAddress, "version", version.Get().String())
 	return preparedRootAPIServer.Run(ctx.Done())
 }
 

cmd/virtual-workspaces/options/options.go

@@ -30,7 +30,8 @@ import (
 	"k8s.io/component-base/logs"
 
 	cacheoptions "github.com/kcp-dev/kcp/pkg/cache/client/options"
-	virtualworkspacesoptions "github.com/kcp-dev/kcp/pkg/virtual/options"
+	corevwoptions "github.com/kcp-dev/kcp/pkg/virtual/options"
+	tmcvwoptions "github.com/kcp-dev/kcp/tmc/pkg/virtual/options"
 )
 
 // DefaultRootPathPrefix is basically constant forever, or we risk a breaking change. The
@@ -48,13 +49,15 @@ type Options struct {
 	Cache          cacheoptions.Cache
 	SecureServing  genericapiserveroptions.SecureServingOptions
 	Authentication genericapiserveroptions.DelegatingAuthenticationOptions
-	Authorization  virtualworkspacesoptions.Authorization
+	Authorization  corevwoptions.Authorization
 	Audit          genericapiserveroptions.AuditOptions
 
 	Logs logs.Options
 
-	VirtualWorkspaces virtualworkspacesoptions.Options
-	ProfilerAddress   string
+	CoreVirtualWorkspaces corevwoptions.Options
+	TmcVirtualWorkspaces  tmcvwoptions.Options
+
+	ProfilerAddress string
 }
 
 func NewOptions() *Options {
@@ -66,12 +69,13 @@ func NewOptions() *Options {
 		Cache:          *cacheoptions.NewCache(),
 		SecureServing:  *genericapiserveroptions.NewSecureServingOptions(),
 		Authentication: *genericapiserveroptions.NewDelegatingAuthenticationOptions(),
-		Authorization:  *virtualworkspacesoptions.NewAuthorization(),
+		Authorization:  *corevwoptions.NewAuthorization(),
 		Audit:          *genericapiserveroptions.NewAuditOptions(),
 		Logs:           *logs.NewOptions(),
 
-		VirtualWorkspaces: *virtualworkspacesoptions.NewOptions(),
-		ProfilerAddress:   "",
+		CoreVirtualWorkspaces: *corevwoptions.NewOptions(),
+		TmcVirtualWorkspaces:  *tmcvwoptions.NewOptions(),
+		ProfilerAddress:       "",
 	}
 
 	opts.SecureServing.ServerCert.CertKey.CertFile = filepath.Join(".", ".kcp", "apiserver.crt")
@@ -87,7 +91,8 @@ func (o *Options) AddFlags(flags *pflag.FlagSet) {
 	o.Authentication.AddFlags(flags)
 	o.Audit.AddFlags(flags)
 	o.Logs.AddFlags(flags)
-	o.VirtualWorkspaces.AddFlags(flags)
+	o.CoreVirtualWorkspaces.AddFlags(flags)
+	o.TmcVirtualWorkspaces.AddFlags(flags)
 
 	flags.StringVar(&o.KubeconfigFile, "kubeconfig", o.KubeconfigFile,
 		"The kubeconfig file of the KCP instance that hosts workspaces.")
@@ -102,7 +107,8 @@ func (o *Options) Validate() error {
 	errs = append(errs, o.Cache.Validate()...)
 	errs = append(errs, o.SecureServing.Validate()...)
 	errs = append(errs, o.Authentication.Validate()...)
-	errs = append(errs, o.VirtualWorkspaces.Validate()...)
+	errs = append(errs, o.CoreVirtualWorkspaces.Validate()...)
+	errs = append(errs, o.TmcVirtualWorkspaces.Validate()...)
 
 	if len(o.KubeconfigFile) == 0 {
 		errs = append(errs, fmt.Errorf("--kubeconfig is required for this command"))

pkg/reconciler/workload/synctarget/synctarget_reconcile.go

@@ -30,7 +30,7 @@ import (
 	virtualworkspacesoptions "github.com/kcp-dev/kcp/cmd/virtual-workspaces/options"
 	corev1alpha1 "github.com/kcp-dev/kcp/pkg/apis/core/v1alpha1"
 	workloadv1alpha1 "github.com/kcp-dev/kcp/pkg/apis/workload/v1alpha1"
-	syncerbuilder "github.com/kcp-dev/kcp/pkg/virtual/syncer/builder"
+	syncerbuilder "github.com/kcp-dev/kcp/tmc/pkg/virtual/syncer/builder"
 )
 
 func (c *Controller) reconcile(ctx context.Context, syncTarget *workloadv1alpha1.SyncTarget, workspaceShards []*corev1alpha1.Shard) (*workloadv1alpha1.SyncTarget, error) {

pkg/server/config.go

@@ -74,10 +74,11 @@ type Config struct {
 
 	EmbeddedEtcd *embeddedetcd.Config
 
-	GenericConfig  *genericapiserver.Config // the config embedded into MiniAggregator, the head of the delegation chain
-	MiniAggregator *aggregator.MiniAggregatorConfig
-	Apis           *apis.Config
-	ApiExtensions  *apiextensionsapiserver.Config
+	GenericConfig   *genericapiserver.Config // the config embedded into MiniAggregator, the head of the delegation chain
+	MiniAggregator  *aggregator.MiniAggregatorConfig
+	Apis            *apis.Config
+	ApiExtensions   *apiextensionsapiserver.Config
+	OptionalVirtual *VirtualConfig
 
 	ExtraConfig
 }
@@ -129,11 +130,12 @@ type ExtraConfig struct {
 type completedConfig struct {
 	Options kcpserveroptions.CompletedOptions
 
-	GenericConfig  genericapiserver.CompletedConfig
-	EmbeddedEtcd   embeddedetcd.CompletedConfig
-	MiniAggregator aggregator.CompletedMiniAggregatorConfig
-	Apis           apis.CompletedConfig
-	ApiExtensions  apiextensionsapiserver.CompletedConfig
+	GenericConfig   genericapiserver.CompletedConfig
+	EmbeddedEtcd    embeddedetcd.CompletedConfig
+	MiniAggregator  aggregator.CompletedMiniAggregatorConfig
+	Apis            apis.CompletedConfig
+	ApiExtensions   apiextensionsapiserver.CompletedConfig
+	OptionalVirtual CompletedVirtualConfig
 
 	ExtraConfig
 }
@@ -145,14 +147,21 @@ type CompletedConfig struct {
 
 // Complete fills in any fields not set that are required to have valid data. It's mutating the receiver.
 func (c *Config) Complete() (CompletedConfig, error) {
+	miniAggregator := c.MiniAggregator.Complete()
 	return CompletedConfig{&completedConfig{
 		Options: c.Options,
 
 		GenericConfig:  c.GenericConfig.Complete(informerfactoryhack.Wrap(c.KubeSharedInformerFactory)),
 		EmbeddedEtcd:   c.EmbeddedEtcd.Complete(),
-		MiniAggregator: c.MiniAggregator.Complete(),
+		MiniAggregator: miniAggregator,
 		Apis:           c.Apis.Complete(),
 		ApiExtensions:  c.ApiExtensions.Complete(),
+		OptionalVirtual: c.OptionalVirtual.Complete(
+			miniAggregator.GenericConfig.Authentication,
+			miniAggregator.GenericConfig.AuditPolicyRuleEvaluator,
+			miniAggregator.GenericConfig.AuditBackend,
+			c.GenericConfig.ExternalAddress,
+		),
 
 		ExtraConfig: c.ExtraConfig,
 	}}, nil
@@ -520,5 +529,21 @@ func NewConfig(opts kcpserveroptions.CompletedOptions) (*Config, error) {
 		GenericConfig: c.GenericConfig,
 	}
 
+	if opts.Virtual.Enabled {
+		virtualWorkspacesConfig := rest.CopyConfig(c.GenericConfig.LoopbackClientConfig)
+		virtualWorkspacesConfig = rest.AddUserAgent(virtualWorkspacesConfig, "virtual-workspaces")
+
+		c.OptionalVirtual, err = newVirtualConfig(
+			opts,
+			virtualWorkspacesConfig,
+			c.KubeSharedInformerFactory,
+			c.KcpSharedInformerFactory,
+			c.CacheKcpSharedInformerFactory,
+		)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return c, nil
 }

pkg/server/server.go

@@ -44,6 +44,7 @@ import (
 	"github.com/kcp-dev/kcp/pkg/indexers"
 	"github.com/kcp-dev/kcp/pkg/informer"
 	metadataclient "github.com/kcp-dev/kcp/pkg/metadata"
+	virtualrootapiserver "github.com/kcp-dev/kcp/pkg/virtual/framework/rootapiserver"
 )
 
 const resyncPeriod = 10 * time.Hour
@@ -52,6 +53,7 @@ type Server struct {
 	CompletedConfig
 
 	*genericcontrolplane.ServerChain
+	virtual *virtualrootapiserver.Server
 
 	syncedCh             chan struct{}
 	rootPhase1FinishedCh chan struct{}
@@ -113,6 +115,19 @@ func NewServer(c CompletedConfig) (*Server, error) {
 		return nil, err
 	}
 
+	if c.Options.Virtual.Enabled {
+		s.virtual, err = c.OptionalVirtual.NewServer(s.preHandlerChainMux)
+		if err != nil {
+			return nil, err
+		}
+		if err := s.AddPostStartHook("kcp-start-virtual-workspaces", func(ctx genericapiserver.PostStartHookContext) error {
+			s.virtual.GenericAPIServer.RunPostStartHooks(ctx.StopCh)
+			return nil
+		}); err != nil {
+			return nil, err
+		}
+	}
+
 	return s, nil
 }
 
@@ -472,14 +487,6 @@ func (s *Server) Run(ctx context.Context) error {
 		}
 	}
 
-	if s.Options.Virtual.Enabled {
-		virtualWorkspacesConfig := rest.CopyConfig(s.GenericConfig.LoopbackClientConfig)
-		virtualWorkspacesConfig = rest.AddUserAgent(virtualWorkspacesConfig, "virtual-workspaces")
-		if err := s.installVirtualWorkspaces(ctx, virtualWorkspacesConfig, s.GenericConfig.Authentication, s.GenericConfig.ExternalAddress, s.GenericConfig.AuditPolicyRuleEvaluator, s.preHandlerChainMux); err != nil {
-			return err
-		}
-	}
-
 	if len(s.Options.Cache.Client.KubeconfigFile) == 0 {
 		if err := s.installCacheServer(ctx); err != nil {
 			return err