WIP DMCM

On-behalf-of: @SAP [email protected]

Author: xrstf

Makefile

@@ -174,7 +174,7 @@ install-yq:
 	@UNCOMPRESSED=true hack/uget.sh https://github.com/mikefarah/yq/releases/download/v{VERSION}/yq_{GOOS}_{GOARCH} yq $(YQ_VERSION) yq_*
 
 .PHONY: install-kcp
-install-kcp: UGET_CHECKSUMS=false # do not checksum because the version regularly gets overwritten in CI jobs
+install-kcp: UGET_CHECKSUMS= # do not checksum because the version regularly gets overwritten in CI jobs
 install-kcp:
 	@hack/uget.sh https://github.com/kcp-dev/kcp/releases/download/v{VERSION}/kcp_{VERSION}_{GOOS}_{GOARCH}.tar.gz kcp $(KCP_VERSION)
 

cmd/api-syncagent/kcp.go

@@ -23,13 +23,12 @@ import (
 	"regexp"
 
 	"github.com/kcp-dev/logicalcluster/v3"
-
-	"github.com/kcp-dev/api-syncagent/internal/kcp"
-
 	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
 	kcpdevcore "github.com/kcp-dev/sdk/apis/core"
 	kcpdevcorev1alpha1 "github.com/kcp-dev/sdk/apis/core/v1alpha1"
 
+	"github.com/kcp-dev/api-syncagent/internal/kcp"
+
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -39,25 +38,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cluster"
 )
 
-// The agent has two potentially different kcp clusters:
-//
-//   endpointCluster - this is where the source of the virtual workspace URLs
-//                     live, i.e. where the APIExport/EndpointSlice.
-//   managedCluster  - this is where the APIExport and APIResourceSchemas
-//                     exist that are meant to be reconciled.
-//
-// The managedCluster always exists, the endpointCluster only if the workspace
-// for the virtual workspace source is different from the managed cluster.
-
 // setupEndpointKcpCluster sets up a plain, non-kcp-aware ctrl-runtime Cluster object
 // that is solvely used to watch whichever object holds the virtual workspace URLs,
 // either the APIExport or the APIExportEndpointSlice.
-func setupEndpointKcpCluster(endpoint *syncEndpoint) (cluster.Cluster, error) {
-	// no need for a dedicated endpoint cluster
-	if endpoint.EndpointSlice == nil || endpoint.EndpointSlice.Cluster == endpoint.APIExport.Cluster {
-		return nil, nil
-	}
-
+func setupEndpointKcpCluster(endpointSlice qualifiedAPIExportEndpointSlice) (cluster.Cluster, error) {
 	scheme := runtime.NewScheme()
 
 	if err := kcpdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -72,11 +56,11 @@ func setupEndpointKcpCluster(endpoint *syncEndpoint) (cluster.Cluster, error) {
 	// restrict the cache's selectors accordingly so we can still make use of caching.
 	byObject := map[ctrlruntimeclient.Object]cache.ByObject{
 		&kcpdevv1alpha1.APIExportEndpointSlice{}: {
-			Field: fields.SelectorFromSet(fields.Set{"metadata.name": endpoint.EndpointSlice.Name}),
+			Field: fields.SelectorFromSet(fields.Set{"metadata.name": endpointSlice.Name}),
 		},
 	}
 
-	return cluster.New(endpoint.EndpointSlice.Config, func(o *cluster.Options) {
+	return cluster.New(endpointSlice.Config, func(o *cluster.Options) {
 		o.Scheme = scheme
 		o.Cache = cache.Options{
 			Scheme:   scheme,
@@ -87,7 +71,7 @@ func setupEndpointKcpCluster(endpoint *syncEndpoint) (cluster.Cluster, error) {
 
 // setupManagedKcpCluster sets up a plain, non-kcp-aware ctrl-runtime Cluster object
 // that is solvely used to manage the APIExport and APIResourceSchemas.
-func setupManagedKcpCluster(endpoint *syncEndpoint) (cluster.Cluster, error) {
+func setupManagedKcpCluster(apiExport qualifiedAPIExport) (cluster.Cluster, error) {
 	scheme := runtime.NewScheme()
 
 	if err := kcpdevv1alpha1.AddToScheme(scheme); err != nil {
@@ -102,11 +86,11 @@ func setupManagedKcpCluster(endpoint *syncEndpoint) (cluster.Cluster, error) {
 	// restrict the cache's selectors accordingly so we can still make use of caching.
 	byObject := map[ctrlruntimeclient.Object]cache.ByObject{
 		&kcpdevv1alpha1.APIExport{}: {
-			Field: fields.SelectorFromSet(fields.Set{"metadata.name": endpoint.APIExport.Name}),
+			Field: fields.SelectorFromSet(fields.Set{"metadata.name": apiExport.Name}),
 		},
 	}
 
-	return cluster.New(endpoint.APIExport.Config, func(o *cluster.Options) {
+	return cluster.New(apiExport.Config, func(o *cluster.Options) {
 		o.Scheme = scheme
 		o.Cache = cache.Options{
 			Scheme:   scheme,
@@ -133,7 +117,7 @@ type qualifiedAPIExportEndpointSlice struct {
 
 type syncEndpoint struct {
 	APIExport     qualifiedAPIExport
-	EndpointSlice *qualifiedAPIExportEndpointSlice
+	EndpointSlice qualifiedAPIExportEndpointSlice
 }
 
 // resolveSyncEndpoint takes the user provided (usually via CLI flags) APIExportEndpointSliceRef and
@@ -142,7 +126,7 @@ type syncEndpoint struct {
 // must point to the cluster where the APIExport lives, and vice versa for the endpoint slice;
 // however the endpoint slice references an APIExport in potentially another cluster, and for this
 // case the initialRestConfig will be rewritten accordingly).
-func resolveSyncEndpoint(ctx context.Context, initialRestConfig *rest.Config, endpointSliceRef string, apiExportRef string) (*syncEndpoint, error) {
+func resolveSyncEndpoint(ctx context.Context, initialRestConfig *rest.Config, endpointSliceRef string) (*syncEndpoint, error) {
 	// construct temporary, uncached client
 	scheme := runtime.NewScheme()
 	if err := kcpdevcorev1alpha1.AddToScheme(scheme); err != nil {
@@ -160,52 +144,32 @@ func resolveSyncEndpoint(ctx context.Context, initialRestConfig *rest.Config, en
 
 	se := &syncEndpoint{}
 
-	// When an endpoint ref is given, both the APIExportEndpointSlice and the APIExport must exist.
-	if endpointSliceRef != "" {
-		endpointSlice, err := resolveAPIExportEndpointSlice(ctx, client, endpointSliceRef)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve APIExportEndpointSlice: %w", err)
-		}
-		endpointSlice.Config = initialRestConfig
-
-		// find the APIExport referenced not by the user (can't: both ref parameters to this function
-		// are mutually exclusive), but in the APIExportEndpointSlice.
-		restConfig, err := retargetRestConfig(initialRestConfig, endpointSlice.Spec.APIExport.Path)
-		if err != nil {
-			return nil, fmt.Errorf("failed to re-target the given kubeconfig to cluster %q: %w", endpointSlice.Spec.APIExport.Path, err)
-		}
-
-		client, err := ctrlruntimeclient.New(restConfig, clientOpts)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create service reader: %w", err)
-		}
+	// First we find the APIExportEndpointSlice.
+	endpointSlice, err := resolveAPIExportEndpointSlice(ctx, client, endpointSliceRef)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve APIExportEndpointSlice: %w", err)
+	}
+	endpointSlice.Config = initialRestConfig
 
-		apiExport, err := resolveAPIExport(ctx, client, endpointSlice.Spec.APIExport.Name)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve APIExport: %w", err)
-		}
-		apiExport.Config = restConfig
-
-		se.APIExport = apiExport
-		se.EndpointSlice = &endpointSlice
-	} else { // if an export ref is given, the endpoint slice is optional (for compat with kcp <0.28)
-		apiExport, err := resolveAPIExport(ctx, client, apiExportRef)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve APIExport: %w", err)
-		}
-		apiExport.Config = initialRestConfig
+	// Now we find the APIExport referenced in the APIExportEndpointSlice.
+	restConfig, err := retargetRestConfig(initialRestConfig, endpointSlice.Spec.APIExport.Path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to re-target the given kubeconfig to cluster %q: %w", endpointSlice.Spec.APIExport.Path, err)
+	}
 
-		se.APIExport = apiExport
+	client, err = ctrlruntimeclient.New(restConfig, clientOpts)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create service reader: %w", err)
+	}
 
-		// try to find an endpoint slice in the same workspace with the same name as the APIExport
-		endpointSlice, err := resolveAPIExportEndpointSlice(ctx, client, apiExportRef)
-		if ctrlruntimeclient.IgnoreNotFound(err) != nil {
-			return nil, fmt.Errorf("failed to resolve APIExportEndpointSlice: %w", err)
-		} else if err == nil {
-			apiExport.Config = initialRestConfig
-			se.EndpointSlice = &endpointSlice
-		}
+	apiExport, err := resolveAPIExport(ctx, client, endpointSlice.Spec.APIExport.Name)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve APIExport: %w", err)
 	}
+	apiExport.Config = restConfig
+
+	se.APIExport = apiExport
+	se.EndpointSlice = endpointSlice
 
 	return se, nil
 }

cmd/api-syncagent/main.go

@@ -32,12 +32,11 @@ import (
 	"github.com/kcp-dev/api-syncagent/internal/controller/apiexport"
 	"github.com/kcp-dev/api-syncagent/internal/controller/apiresourceschema"
 	"github.com/kcp-dev/api-syncagent/internal/controller/syncmanager"
+	"github.com/kcp-dev/api-syncagent/internal/kcp"
 	syncagentlog "github.com/kcp-dev/api-syncagent/internal/log"
 	"github.com/kcp-dev/api-syncagent/internal/version"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
 
-	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
-
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -85,13 +84,11 @@ func main() {
 
 func run(ctx context.Context, log *zap.SugaredLogger, opts *Options) error {
 	v := version.NewAppVersion()
-	hello := log.With("version", v.GitVersion, "name", opts.AgentName)
-
-	if opts.APIExportEndpointSliceRef != "" {
-		hello = hello.With("apiexportendpointslice", opts.APIExportEndpointSliceRef)
-	} else {
-		hello = hello.With("apiexport", opts.APIExportRef)
-	}
+	hello := log.With(
+		"version", v.GitVersion,
+		"name", opts.AgentName,
+		"apiexportendpointslice", opts.APIExportEndpointSliceRef,
+	)
 
 	hello.Info("Moin, I'm the kcp Sync Agent")
 
@@ -112,22 +109,20 @@ func run(ctx context.Context, log *zap.SugaredLogger, opts *Options) error {
 		return fmt.Errorf("kcp kubeconfig does not point to a specific workspace")
 	}
 
-	// We check if the APIExport/APIExportEndpointSlice exists and extract information we need to set up our kcpCluster.
-	endpoint, err := resolveSyncEndpoint(ctx, kcpRestConfig, opts.APIExportEndpointSliceRef, opts.APIExportRef)
+	// We check if the APIExportEndpointSlice exists and extract information we need to set up our kcpCluster.
+	endpoint, err := resolveSyncEndpoint(ctx, kcpRestConfig, opts.APIExportEndpointSliceRef)
 	if err != nil {
 		return fmt.Errorf("failed to resolve APIExport/EndpointSlice: %w", err)
 	}
 
 	log.Infow("Resolved APIExport", "name", endpoint.APIExport.Name, "workspace", endpoint.APIExport.Path, "logicalcluster", endpoint.APIExport.Cluster)
-
-	if s := endpoint.EndpointSlice; s != nil {
-		log.Infow("Using APIExportEndpointSlice", "name", endpoint.EndpointSlice.Name, "workspace", s.Path, "logicalcluster", s.Cluster)
-	}
+	log.Infow("Using APIExportEndpointSlice", "name", endpoint.EndpointSlice.Name, "workspace", endpoint.EndpointSlice.Path, "logicalcluster", endpoint.EndpointSlice.Cluster)
 
 	// init the "permanent" kcp cluster connections
 
-	// always need the managedKcpCluster
-	managedKcpCluster, err := setupManagedKcpCluster(endpoint)
+	// always need the managedKcpCluster, this is where we will manage the APIExport and
+	// its resource schemas.
+	managedKcpCluster, err := setupManagedKcpCluster(endpoint.APIExport)
 	if err != nil {
 		return fmt.Errorf("failed to initialize managed kcp cluster: %w", err)
 	}
@@ -138,18 +133,32 @@ func run(ctx context.Context, log *zap.SugaredLogger, opts *Options) error {
 		return fmt.Errorf("failed to add managed kcp cluster runnable: %w", err)
 	}
 
-	// the endpoint cluster can be nil
-	endpointKcpCluster, err := setupEndpointKcpCluster(endpoint)
-	if err != nil {
-		return fmt.Errorf("failed to initialize endpoint kcp cluster: %w", err)
-	}
+	// If needed, start an additional cluster for the endpoint workspace, where
+	// the EndpointSlice lives.
+	if endpoint.EndpointSlice.Cluster != endpoint.APIExport.Cluster {
+		endpointKcpCluster, err := setupEndpointKcpCluster(endpoint.EndpointSlice)
+		if err != nil {
+			return fmt.Errorf("failed to initialize endpoint kcp cluster: %w", err)
+		}
 
-	if endpointKcpCluster != nil {
 		if err := mgr.Add(endpointKcpCluster); err != nil {
 			return fmt.Errorf("failed to add endpoint kcp cluster runnable: %w", err)
 		}
 	}
 
+	// Setup the magical dynamic multicluster manager. It's a dynamic version of the
+	// regular mcmanager, capable of starting new controllers at any later time and
+	// allowing them to be also stopped at any time. The syncmanager needs it to
+	// start/stop sync controllers for each PublishedResource.
+	dmcm, err := kcp.NewDynamicMultiClusterManager(endpoint.EndpointSlice.Config, endpoint.EndpointSlice.Name)
+	if err != nil {
+		return fmt.Errorf("failed to start dynamic multi cluster manager: %w", err)
+	}
+
+	if err := mgr.Add(dmcm); err != nil {
+		return fmt.Errorf("failed to add endpoint kcp cluster runnable: %w", err)
+	}
+
 	startController := func(name string, creator func() error) error {
 		if slices.Contains(opts.DisabledControllers, name) {
 			log.Infof("Not starting %s controller because it is disabled.", name)
@@ -178,20 +187,7 @@ func run(ctx context.Context, log *zap.SugaredLogger, opts *Options) error {
 	// This controller is called "sync" because it makes the most sense to the users, even though internally the relevant
 	// controller is the syncmanager (which in turn would start/stop the sync controllers).
 	if err := startController("sync", func() error {
-		cluster := endpointKcpCluster
-		if cluster == nil {
-			cluster = managedKcpCluster
-		}
-
-		var endpointSlice *kcpdevv1alpha1.APIExportEndpointSlice
-		if endpoint.EndpointSlice != nil {
-			endpointSlice = endpoint.EndpointSlice.APIExportEndpointSlice
-		}
-
-		// It doesn't matter which rest config we specify, as the URL will be overwritten with the
-		// virtual workspace URL anyway.
-
-		return syncmanager.Add(ctx, mgr, cluster, kcpRestConfig, log, endpoint.APIExport.APIExport, endpointSlice, opts.PublishedResourceSelector, opts.Namespace, opts.AgentName)
+		return syncmanager.Add(ctx, mgr, dmcm, log, opts.PublishedResourceSelector, opts.Namespace, opts.AgentName)
 	}); err != nil {
 		return err
 	}

cmd/api-syncagent/options.go

@@ -54,15 +54,6 @@ type Options struct {
 	// If not given, defaults to "<service ref>-syncagent".
 	AgentName string
 
-	// APIExportRef references the APIExport within a kcp workspace that this
-	// Sync Agent should work with by name. The APIExport has to already exist, but it must not have
-	// any pre-existing resource schemas configured, the agent will fill them in based on
-	// PublishedResources.
-	//
-	// Deprecated: Use APIExportEndpointSliceRef instead. If an APIExport is referenced, the agent
-	// will attempt to find and use an endpoint slice of the same name.
-	APIExportRef string
-
 	// APIExportEndpointSliceRef references the APIExportEndpointSlice within a kcp workspace that this
 	// Sync Agent should work with by name. The agent will automatically manage the resource schemas
 	// in the APIExport referenced by this endpoint slice.
@@ -96,7 +87,6 @@ func (o *Options) AddFlags(flags *pflag.FlagSet) {
 	flags.StringVar(&o.KcpKubeconfig, "kcp-kubeconfig", o.KcpKubeconfig, "kubeconfig file of kcp")
 	flags.StringVar(&o.Namespace, "namespace", o.Namespace, "Kubernetes namespace the Sync Agent is running in")
 	flags.StringVar(&o.AgentName, "agent-name", o.AgentName, "name of this Sync Agent, must not be changed after the first run, can be left blank to auto-generate a name")
-	flags.StringVar(&o.APIExportRef, "apiexport-ref", o.APIExportRef, "name of the APIExport in kcp that this Sync Agent is powering (deprecated, use --apiexportendpointslice-ref instead)")
 	flags.StringVar(&o.APIExportEndpointSliceRef, "apiexportendpointslice-ref", o.APIExportEndpointSliceRef, "name of the APIExportEndpointSlice in kcp that this Sync Agent is powering")
 	flags.StringVar(&o.PublishedResourceSelectorString, "published-resource-selector", o.PublishedResourceSelectorString, "restrict this Sync Agent to only process PublishedResources matching this label selector (optional)")
 	flags.BoolVar(&o.EnableLeaderElection, "enable-leader-election", o.EnableLeaderElection, "whether to perform leader election")
@@ -124,12 +114,8 @@ func (o *Options) Validate() error {
 		}
 	}
 
-	if len(o.APIExportRef) == 0 && len(o.APIExportEndpointSliceRef) == 0 {
-		errs = append(errs, errors.New("either --apiexportendpointslice-ref or --apiexport-ref is required"))
-	}
-
-	if len(o.APIExportRef) != 0 && len(o.APIExportEndpointSliceRef) != 0 {
-		errs = append(errs, errors.New("--apiexportendpointslice-ref and --apiexport-ref are mutually exclusive"))
+	if len(o.APIExportEndpointSliceRef) == 0 {
+		errs = append(errs, errors.New("--apiexportendpointslice-ref is required"))
 	}
 
 	if len(o.KcpKubeconfig) == 0 {
@@ -156,7 +142,7 @@ func (o *Options) Complete() error {
 	errs := []error{}
 
 	if len(o.AgentName) == 0 {
-		o.AgentName = o.APIExportRef + "-syncagent"
+		o.AgentName = o.APIExportEndpointSliceRef + "-syncagent"
 	}
 
 	if s := o.PublishedResourceSelectorString; len(s) > 0 {

internal/controller/apiexport/controller.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 
 	"github.com/kcp-dev/logicalcluster/v3"
+	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
 	"go.uber.org/zap"
 
 	"github.com/kcp-dev/api-syncagent/internal/controllerutil"
@@ -31,8 +32,6 @@ import (
 	"github.com/kcp-dev/api-syncagent/internal/resources/reconciling"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
 
-	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
-
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"

internal/controller/apiexport/reconciler.go

@@ -21,11 +21,11 @@ import (
 	"slices"
 	"strings"
 
+	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
+
 	"github.com/kcp-dev/api-syncagent/internal/resources/reconciling"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
 
-	kcpdevv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
-
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"