Add CRD migrator

Makefile

@@ -310,6 +310,22 @@ generate-manifests-core: $(CONTROLLER_GEN) $(KUSTOMIZE) ## Generate manifests e.
 		paths=./util/test/builder/... \
 		crd:crdVersions=v1 \
 		output:crd:dir=./util/test/builder/crd
+	$(CONTROLLER_GEN) \
+		paths=./controllers/crdmigrator/test/t1/... \
+		crd:crdVersions=v1 \
+		output:crd:dir=./controllers/crdmigrator/test/t1/crd
+	$(CONTROLLER_GEN) \
+		paths=./controllers/crdmigrator/test/t2/... \
+		crd:crdVersions=v1 \
+		output:crd:dir=./controllers/crdmigrator/test/t2/crd
+	$(CONTROLLER_GEN) \
+		paths=./controllers/crdmigrator/test/t3/... \
+		crd:crdVersions=v1 \
+		output:crd:dir=./controllers/crdmigrator/test/t3/crd
+	$(CONTROLLER_GEN) \
+		paths=./controllers/crdmigrator/test/t4/... \
+		crd:crdVersions=v1 \
+		output:crd:dir=./controllers/crdmigrator/test/t4/crd
 
 .PHONY: generate-manifests-kubeadm-bootstrap
 generate-manifests-kubeadm-bootstrap: $(CONTROLLER_GEN) ## Generate manifests e.g. CRD, RBAC etc. for kubeadm bootstrap
@@ -384,7 +400,8 @@ generate-go-deepcopy-core: $(CONTROLLER_GEN) ## Generate deepcopy go code for co
 		paths=./$(EXP_DIR)/runtime/hooks/api/... \
 		paths=./internal/runtime/test/... \
 		paths=./cmd/clusterctl/... \
-		paths=./util/test/builder/...
+		paths=./util/test/builder/... \
+		paths=./controllers/crdmigrator/test/...
 
 .PHONY: generate-go-deepcopy-kubeadm-bootstrap
 generate-go-deepcopy-kubeadm-bootstrap: $(CONTROLLER_GEN) ## Generate deepcopy go code for kubeadm bootstrap

api/v1beta1/common_types.go

@@ -197,6 +197,9 @@ const (
 	// VariableDefinitionFromInline indicates a patch or variable was defined in the `.spec` of a ClusterClass
 	// rather than from an external patch extension.
 	VariableDefinitionFromInline = "inline"
+
+	// CRDMigrationObservedGenerationAnnotation indicates on a CRD for which generation CRD migration is completed.
+	CRDMigrationObservedGenerationAnnotation = "crd-migration.cluster.x-k8s.io/observed-generation"
 )
 
 // MachineSetPreflightCheck defines a valid MachineSet preflight check.

bootstrap/kubeadm/config/rbac/role.yaml

@@ -24,6 +24,23 @@ rules:
   verbs:
   - create
   - patch
+- apiGroups:
+  - apiextensions.k8s.io
+  resources:
+  - customresourcedefinitions
+  verbs:
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - apiextensions.k8s.io
+  resources:
+  - customresourcedefinitions/status
+  verbs:
+  - patch
+  - update
 - apiGroups:
   - authentication.k8s.io
   resources:
@@ -50,6 +67,16 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - bootstrap.cluster.x-k8s.io
+  resources:
+  - kubeadmconfigtemplates
+  verbs:
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - cluster.x-k8s.io
   resources:

bootstrap/kubeadm/main.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -50,6 +51,7 @@ import (
 	kubeadmbootstrapcontrollers "sigs.k8s.io/cluster-api/bootstrap/kubeadm/controllers"
 	"sigs.k8s.io/cluster-api/bootstrap/kubeadm/internal/webhooks"
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
+	"sigs.k8s.io/cluster-api/controllers/crdmigrator"
 	"sigs.k8s.io/cluster-api/controllers/remote"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	"sigs.k8s.io/cluster-api/feature"
@@ -90,11 +92,13 @@ var (
 	clusterConcurrency       int
 	clusterCacheConcurrency  int
 	kubeadmConfigConcurrency int
+	skipCRDMigrationPhases   string
 	tokenTTL                 time.Duration
 )
 
 func init() {
 	_ = clientgoscheme.AddToScheme(scheme)
+	_ = apiextensionsv1.AddToScheme(scheme)
 	_ = clusterv1.AddToScheme(scheme)
 	_ = expv1.AddToScheme(scheme)
 	_ = bootstrapv1alpha3.AddToScheme(scheme)
@@ -140,6 +144,9 @@ func InitFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&kubeadmConfigConcurrency, "kubeadmconfig-concurrency", 10,
 		"Number of kubeadm configs to process simultaneously")
 
+	fs.StringVar(&skipCRDMigrationPhases, "skip-crd-migration-phases", "",
+		"Comma-separated list of CRD migration phases to skip. Valid values are: All, StorageVersionMigration, CleanupManagedFields.")
+
 	fs.DurationVar(&syncPeriod, "sync-period", 10*time.Minute,
 		"The minimum interval at which watched resources are reconciled (e.g. 15m)")
 
@@ -181,6 +188,10 @@ func InitFlags(fs *pflag.FlagSet) {
 // Add RBAC for the authorized diagnostics endpoint.
 // +kubebuilder:rbac:groups=authentication.k8s.io,resources=tokenreviews,verbs=create
 // +kubebuilder:rbac:groups=authorization.k8s.io,resources=subjectaccessreviews,verbs=create
+// ADD RBAC for CRD Migrator.
+// +kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get;list;watch;update;patch
+// +kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions/status,verbs=update;patch
+// +kubebuilder:rbac:groups=bootstrap.cluster.x-k8s.io,resources=kubeadmconfigtemplates,verbs=get;list;watch;patch;update
 
 func main() {
 	InitFlags(pflag.CommandLine)
@@ -340,6 +351,19 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		os.Exit(1)
 	}
 
+	if err := (&crdmigrator.CRDMigrator{
+		Client:                 mgr.GetClient(),
+		APIReader:              mgr.GetAPIReader(),
+		SkipCRDMigrationPhases: skipCRDMigrationPhases,
+		Config: map[client.Object]crdmigrator.ByObjectConfig{
+			&bootstrapv1.KubeadmConfig{}:         {UseCache: true},
+			&bootstrapv1.KubeadmConfigTemplate{}: {UseCache: false},
+		},
+	}).SetupWithManager(ctx, mgr, concurrency(1)); err != nil {
+		setupLog.Error(err, "Unable to create controller", "controller", "CRDMigrator")
+		os.Exit(1)
+	}
+
 	if err := (&kubeadmbootstrapcontrollers.KubeadmConfigReconciler{
 		Client:              mgr.GetClient(),
 		SecretCachingClient: secretCachingClient,

cmd/clusterctl/client/cluster/upgrader.go

@@ -57,8 +57,9 @@ type UpgradePlan struct {
 
 // UpgradeOptions defines the options used to upgrade installation.
 type UpgradeOptions struct {
-	WaitProviders       bool
-	WaitProviderTimeout time.Duration
+	WaitProviders                    bool
+	WaitProviderTimeout              time.Duration
+	EnableCRDStorageVersionMigration bool
 }
 
 // isPartialUpgrade returns true if at least one upgradeItem in the plan does not have a target version.
@@ -363,28 +364,30 @@ func (u *providerUpgrader) doUpgrade(ctx context.Context, upgradePlan *UpgradePl
 		return providers[a].GetProviderType().Order() < providers[b].GetProviderType().Order()
 	})
 
-	// Migrate CRs to latest CRD storage version, if necessary.
-	// Note: We have to do this before the providers are scaled down or deleted
-	// so conversion webhooks still work.
-	for _, upgradeItem := range providers {
-		// If there is not a specified next version, skip it (we are already up-to-date).
-		if upgradeItem.NextVersion == "" {
-			continue
-		}
+	if opts.EnableCRDStorageVersionMigration {
+		// Migrate CRs to latest CRD storage version, if necessary.
+		// Note: We have to do this before the providers are scaled down or deleted
+		// so conversion webhooks still work.
+		for _, upgradeItem := range providers {
+			// If there is not a specified next version, skip it (we are already up-to-date).
+			if upgradeItem.NextVersion == "" {
+				continue
+			}
 
-		// Gets the provider components for the target version.
-		components, err := u.getUpgradeComponents(ctx, upgradeItem)
-		if err != nil {
-			return err
-		}
+			// Gets the provider components for the target version.
+			components, err := u.getUpgradeComponents(ctx, upgradeItem)
+			if err != nil {
+				return err
+			}
 
-		c, err := u.proxy.NewClient(ctx)
-		if err != nil {
-			return err
-		}
+			c, err := u.proxy.NewClient(ctx)
+			if err != nil {
+				return err
+			}
 
-		if err := NewCRDMigrator(c).Run(ctx, components.Objs()); err != nil {
-			return err
+			if err := NewCRDMigrator(c).Run(ctx, components.Objs()); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -448,7 +451,11 @@ func (u *providerUpgrader) doUpgrade(ctx context.Context, upgradePlan *UpgradePl
 		}
 	}
 
-	return waitForProvidersReady(ctx, InstallOptions(opts), installQueue, u.proxy)
+	installOpts := InstallOptions{
+		WaitProviders:       opts.WaitProviders,
+		WaitProviderTimeout: opts.WaitProviderTimeout,
+	}
+	return waitForProvidersReady(ctx, installOpts, installQueue, u.proxy)
 }
 
 func (u *providerUpgrader) scaleDownProvider(ctx context.Context, provider clusterctlv1.Provider) error {

cmd/clusterctl/client/upgrade.go

@@ -123,6 +123,9 @@ type ApplyUpgradeOptions struct {
 
 	// WaitProviderTimeout sets the timeout per provider upgrade.
 	WaitProviderTimeout time.Duration
+
+	// EnableCRDStorageVersionMigration enables storage version migration of CRDs.
+	EnableCRDStorageVersionMigration bool
 }
 
 func (c *clusterctlClient) ApplyUpgrade(ctx context.Context, options ApplyUpgradeOptions) error {
@@ -171,8 +174,9 @@ func (c *clusterctlClient) ApplyUpgrade(ctx context.Context, options ApplyUpgrad
 		len(options.AddonProviders) > 0
 
 	opts := cluster.UpgradeOptions{
-		WaitProviders:       options.WaitProviders,
-		WaitProviderTimeout: options.WaitProviderTimeout,
+		WaitProviders:                    options.WaitProviders,
+		WaitProviderTimeout:              options.WaitProviderTimeout,
+		EnableCRDStorageVersionMigration: options.EnableCRDStorageVersionMigration,
 	}
 
 	// If we are upgrading a specific set of providers only, process the providers and call ApplyCustomPlan.

cmd/clusterctl/cmd/upgrade_apply.go

@@ -28,18 +28,19 @@ import (
 )
 
 type upgradeApplyOptions struct {
-	kubeconfig                string
-	kubeconfigContext         string
-	contract                  string
-	coreProvider              string
-	bootstrapProviders        []string
-	controlPlaneProviders     []string
-	infrastructureProviders   []string
-	ipamProviders             []string
-	runtimeExtensionProviders []string
-	addonProviders            []string
-	waitProviders             bool
-	waitProviderTimeout       int
+	kubeconfig                       string
+	kubeconfigContext                string
+	contract                         string
+	coreProvider                     string
+	bootstrapProviders               []string
+	controlPlaneProviders            []string
+	infrastructureProviders          []string
+	ipamProviders                    []string
+	runtimeExtensionProviders        []string
+	addonProviders                   []string
+	waitProviders                    bool
+	waitProviderTimeout              int
+	enableCRDStorageVersionMigration bool
 }
 
 var ua = &upgradeApplyOptions{}
@@ -94,6 +95,10 @@ func init() {
 		"Wait for providers to be upgraded.")
 	upgradeApplyCmd.Flags().IntVar(&ua.waitProviderTimeout, "wait-provider-timeout", 5*60,
 		"Wait timeout per provider upgrade in seconds. This value is ignored if --wait-providers is false")
+	upgradeApplyCmd.Flags().BoolVar(&ua.enableCRDStorageVersionMigration, "enable-crd-storage-version-migration", false,
+		"Enable CRD storage version migration")
+	_ = upgradeApplyCmd.Flags().MarkDeprecated("enable-crd-storage-version-migration",
+		"Storage version migration during upgrades has been deprecated and will be removed in Cluster API v1.13")
 }
 
 func runUpgradeApply() error {
@@ -120,16 +125,17 @@ func runUpgradeApply() error {
 	}
 
 	return c.ApplyUpgrade(ctx, client.ApplyUpgradeOptions{
-		Kubeconfig:                client.Kubeconfig{Path: ua.kubeconfig, Context: ua.kubeconfigContext},
-		Contract:                  ua.contract,
-		CoreProvider:              ua.coreProvider,
-		BootstrapProviders:        ua.bootstrapProviders,
-		ControlPlaneProviders:     ua.controlPlaneProviders,
-		InfrastructureProviders:   ua.infrastructureProviders,
-		IPAMProviders:             ua.ipamProviders,
-		RuntimeExtensionProviders: ua.runtimeExtensionProviders,
-		AddonProviders:            ua.addonProviders,
-		WaitProviders:             ua.waitProviders,
-		WaitProviderTimeout:       time.Duration(ua.waitProviderTimeout) * time.Second,
+		Kubeconfig:                       client.Kubeconfig{Path: ua.kubeconfig, Context: ua.kubeconfigContext},
+		Contract:                         ua.contract,
+		CoreProvider:                     ua.coreProvider,
+		BootstrapProviders:               ua.bootstrapProviders,
+		ControlPlaneProviders:            ua.controlPlaneProviders,
+		InfrastructureProviders:          ua.infrastructureProviders,
+		IPAMProviders:                    ua.ipamProviders,
+		RuntimeExtensionProviders:        ua.runtimeExtensionProviders,
+		AddonProviders:                   ua.addonProviders,
+		WaitProviders:                    ua.waitProviders,
+		WaitProviderTimeout:              time.Duration(ua.waitProviderTimeout) * time.Second,
+		EnableCRDStorageVersionMigration: ua.enableCRDStorageVersionMigration,
 	})
 }

config/rbac/role.yaml

@@ -72,7 +72,16 @@ rules:
   verbs:
   - get
   - list
+  - patch
+  - update
   - watch
+- apiGroups:
+  - apiextensions.k8s.io
+  resources:
+  - customresourcedefinitions/status
+  verbs:
+  - patch
+  - update
 - apiGroups:
   - authentication.k8s.io
   resources:
@@ -93,6 +102,7 @@ rules:
   - clusters
   - clusters/finalizers
   - clusters/status
+  - machinedrainrules
   - machinehealthchecks/finalizers
   - machinehealthchecks/status
   verbs:
@@ -125,21 +135,16 @@ rules:
   - patch
   - update
   - watch
-- apiGroups:
-  - cluster.x-k8s.io
-  resources:
-  - machinedrainrules
-  verbs:
-  - get
-  - list
-  - watch
 - apiGroups:
   - ipam.cluster.x-k8s.io
   resources:
   - ipaddressclaims
+  - ipaddresses
   verbs:
   - get
   - list
+  - patch
+  - update
   - watch
 - apiGroups:
   - runtime.cluster.x-k8s.io