feat: support to datastore migration w/ the same driver

api/v1alpha1/tenantcontrolplane_status.go

@@ -187,12 +187,13 @@ type KubernetesStatus struct {
 	Ingress    *KubernetesIngressStatus   `json:"ingress,omitempty"`
 }
 
-// +kubebuilder:validation:Enum=Provisioning;Upgrading;Ready;NotReady
+// +kubebuilder:validation:Enum=Provisioning;Upgrading;Migrating;Ready;NotReady
 type KubernetesVersionStatus string
 
 var (
 	VersionProvisioning KubernetesVersionStatus = "Provisioning"
 	VersionUpgrading    KubernetesVersionStatus = "Upgrading"
+	VersionMigrating    KubernetesVersionStatus = "Migrating"
 	VersionReady        KubernetesVersionStatus = "Ready"
 	VersionNotReady     KubernetesVersionStatus = "NotReady"
 )

cmd/manager/cmd.go

@@ -7,6 +7,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"os"
 	goRuntime "runtime"
 
 	"github.com/spf13/cobra"
@@ -32,17 +33,28 @@ func NewCmd(scheme *runtime.Scheme) *cobra.Command {
 		tmpDirectory           string
 		kineImage              string
 		datastore              string
+		managerNamespace       string
+		serviceAccountName     string
 	)
 
 	cmd := &cobra.Command{
 		Use:           "manager",
 		Short:         "Start the Kamaji Kubernetes Operator",
-		SilenceErrors: true,
+		SilenceErrors: false,
 		SilenceUsage:  true,
-		PreRun: func(cmd *cobra.Command, args []string) {
+		PreRunE: func(cmd *cobra.Command, args []string) error {
 			// Avoid to pollute Kamaji stdout with useless details by the underlying klog implementations
 			klog.SetOutput(io.Discard)
 			klog.LogToStderr(false)
+
+			for _, arg := range []string{"pod-namespace", "serviceaccount-name", "datastore", "kine-image", "tmp-directory"} {
+				v, _ := cmd.Flags().GetString(arg)
+				if len(v) == 0 {
+					return fmt.Errorf("expecting a value for --%s arg", arg)
+				}
+			}
+
+			return nil
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
 			setupLog := ctrl.Log.WithName("setup")
@@ -56,12 +68,13 @@ func NewCmd(scheme *runtime.Scheme) *cobra.Command {
 			setupLog.Info(fmt.Sprintf("Go OS/Arch: %s/%s", goRuntime.GOOS, goRuntime.GOARCH))
 
 			mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
-				Scheme:                 scheme,
-				MetricsBindAddress:     metricsBindAddress,
-				Port:                   9443,
-				HealthProbeBindAddress: healthProbeBindAddress,
-				LeaderElection:         leaderElect,
-				LeaderElectionID:       "799b98bc.clastix.io",
+				Scheme:                  scheme,
+				MetricsBindAddress:      metricsBindAddress,
+				Port:                    9443,
+				HealthProbeBindAddress:  healthProbeBindAddress,
+				LeaderElection:          leaderElect,
+				LeaderElectionNamespace: managerNamespace,
+				LeaderElectionID:        "799b98bc.clastix.io",
 			})
 			if err != nil {
 				setupLog.Error(err, "unable to start manager")
@@ -78,14 +91,17 @@ func NewCmd(scheme *runtime.Scheme) *cobra.Command {
 			}
 
 			reconciler := &controllers.TenantControlPlaneReconciler{
-				Client: mgr.GetClient(),
-				Scheme: mgr.GetScheme(),
+				Client:    mgr.GetClient(),
+				APIReader: mgr.GetAPIReader(),
+				Scheme:    mgr.GetScheme(),
 				Config: controllers.TenantControlPlaneReconcilerConfig{
 					DefaultDataStoreName: datastore,
 					KineContainerImage:   kineImage,
 					TmpBaseDirectory:     tmpDirectory,
 				},
-				TriggerChan: tcpChannel,
+				TriggerChan:          tcpChannel,
+				KamajiNamespace:      managerNamespace,
+				KamajiServiceAccount: serviceAccountName,
 			}
 
 			if err = reconciler.SetupWithManager(mgr); err != nil {
@@ -145,8 +161,10 @@ func NewCmd(scheme *runtime.Scheme) *cobra.Command {
 	cmd.Flags().StringVar(&healthProbeBindAddress, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
 	cmd.Flags().BoolVar(&leaderElect, "leader-elect", true, "Enable leader election for controller manager. Enabling this will ensure there is only one active controller manager.")
 	cmd.Flags().StringVar(&tmpDirectory, "tmp-directory", "/tmp/kamaji", "Directory which will be used to work with temporary files.")
-	cmd.Flags().StringVar(&kineImage, "kine-image", "rancher/kine:v0.9.2-amd64", "Container image along with tag to use for the Kine sidecar container (used only if etcd-storage-type is set to one of kine strategies)")
-	cmd.Flags().StringVar(&datastore, "datastore", "etcd", "The default DataStore that should be used by Kamaji to setup the required storage")
+	cmd.Flags().StringVar(&kineImage, "kine-image", "rancher/kine:v0.9.2-amd64", "Container image along with tag to use for the Kine sidecar container (used only if etcd-storage-type is set to one of kine strategies).")
+	cmd.Flags().StringVar(&datastore, "datastore", "etcd", "The default DataStore that should be used by Kamaji to setup the required storage.")
+	cmd.Flags().StringVar(&managerNamespace, "pod-namespace", os.Getenv("POD_NAMESPACE"), "The Kubernetes Namespace on which the Operator is running in, required for the TenantControlPlane migration jobs.")
+	cmd.Flags().StringVar(&serviceAccountName, "serviceaccount-name", os.Getenv("SERVICE_ACCOUNT"), "The Kubernetes Namespace on which the Operator is running in, required for the TenantControlPlane migration jobs.")
 
 	cobra.OnInitialize(func() {
 		viper.AutomaticEnv()

cmd/migrate/cmd.go

@@ -0,0 +1,119 @@
+// Copyright 2022 Clastix Labs
+// SPDX-License-Identifier: Apache-2.0
+
+package migrate
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	kamajiv1alpha1 "github.com/clastix/kamaji/api/v1alpha1"
+	"github.com/clastix/kamaji/internal/datastore"
+)
+
+func NewCmd(scheme *runtime.Scheme) *cobra.Command {
+	// CLI flags
+	var (
+		tenantControlPlane string
+		targetDataStore    string
+		timeout            time.Duration
+	)
+
+	cmd := &cobra.Command{
+		Use:          "migrate",
+		Short:        "Migrate the data of a TenantControlPlane to another compatible DataStore",
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx, cancelFn := context.WithTimeout(context.Background(), timeout)
+			defer cancelFn()
+
+			log := ctrl.Log
+
+			log.Info("generating the controller-runtime client")
+
+			client, err := ctrlclient.New(ctrl.GetConfigOrDie(), ctrlclient.Options{
+				Scheme: scheme,
+			})
+			if err != nil {
+				return err
+			}
+
+			parts := strings.Split(tenantControlPlane, string(types.Separator))
+			if len(parts) != 2 {
+				return fmt.Errorf("non well-formed namespaced name for the tenant control plane, expected <NAMESPACE>/NAME, fot %s", tenantControlPlane)
+			}
+
+			log.Info("retrieving the TenantControlPlane")
+
+			tcp := &kamajiv1alpha1.TenantControlPlane{}
+			if err = client.Get(ctx, types.NamespacedName{Namespace: parts[0], Name: parts[1]}, tcp); err != nil {
+				return err
+			}
+
+			log.Info("retrieving the TenantControlPlane used DataStore")
+
+			originDs := &kamajiv1alpha1.DataStore{}
+			if err = client.Get(ctx, types.NamespacedName{Name: tcp.Status.Storage.DataStoreName}, originDs); err != nil {
+				return err
+			}
+
+			log.Info("retrieving the target DataStore")
+
+			targetDs := &kamajiv1alpha1.DataStore{}
+			if err = client.Get(ctx, types.NamespacedName{Name: targetDataStore}, targetDs); err != nil {
+				return err
+			}
+
+			if tcp.Status.Storage.Driver != string(targetDs.Spec.Driver) {
+				return fmt.Errorf("migration between DataStore with different driver is not supported")
+			}
+
+			if tcp.Status.Storage.DataStoreName == targetDs.GetName() {
+				return fmt.Errorf("cannot migrate to the same DataStore")
+			}
+
+			log.Info("generating the origin storage connection")
+
+			originConnection, err := datastore.NewStorageConnection(ctx, client, *originDs)
+			if err != nil {
+				return err
+			}
+			defer originConnection.Close()
+
+			log.Info("generating the target storage connection")
+
+			targetConnection, err := datastore.NewStorageConnection(ctx, client, *targetDs)
+			if err != nil {
+				return err
+			}
+			defer targetConnection.Close()
+			// Start migrating from the old Datastore to the new one
+			log.Info("migration from origin to target started")
+
+			if err = originConnection.Migrate(ctx, *tcp, targetConnection); err != nil {
+				return fmt.Errorf("unable to migrate data from %s to %s: %w", originDs.GetName(), targetDs.GetName(), err)
+			}
+
+			log.Info("migration completed")
+
+			return nil
+		},
+	}
+
+	cmd.Flags().StringVar(&tenantControlPlane, "tenant-control-plane", "", "Namespaced-name of the TenantControlPlane that must be migrated (e.g.: default/test)")
+	cmd.Flags().StringVar(&targetDataStore, "target-datastore", "", "Name of the Datastore to which the TenantControlPlane will be migrated")
+	cmd.Flags().DurationVar(&timeout, "timeout", 5*time.Minute, "Amount of time for the context timeout")
+
+	_ = cmd.MarkFlagRequired("tenant-control-plane")
+	_ = cmd.MarkFlagRequired("target-datastore")
+
+	return cmd
+}

controllers/resources.go

@@ -21,12 +21,14 @@ import (
 )
 
 type GroupResourceBuilderConfiguration struct {
-	client              client.Client
-	log                 logr.Logger
-	tcpReconcilerConfig TenantControlPlaneReconcilerConfig
-	tenantControlPlane  kamajiv1alpha1.TenantControlPlane
-	Connection          datastore.Connection
-	DataStore           kamajiv1alpha1.DataStore
+	client               client.Client
+	log                  logr.Logger
+	tcpReconcilerConfig  TenantControlPlaneReconcilerConfig
+	tenantControlPlane   kamajiv1alpha1.TenantControlPlane
+	Connection           datastore.Connection
+	DataStore            kamajiv1alpha1.DataStore
+	KamajiNamespace      string
+	KamajiServiceAccount string
 }
 
 type GroupDeletableResourceBuilderConfiguration struct {
@@ -61,13 +63,16 @@ func GetDeletableResources(tcp *kamajiv1alpha1.TenantControlPlane, config GroupD
 }
 
 func getDefaultResources(config GroupResourceBuilderConfiguration) []resources.Resource {
-	resources := append(getUpgradeResources(config.client), getKubernetesServiceResources(config.client)...)
+	resources := getDataStoreMigratingResources(config.client, config.KamajiNamespace, config.KamajiServiceAccount)
+	resources = append(resources, getUpgradeResources(config.client)...)
+	resources = append(resources, getKubernetesServiceResources(config.client)...)
 	resources = append(resources, getKubeadmConfigResources(config.client, getTmpDirectory(config.tcpReconcilerConfig.TmpBaseDirectory, config.tenantControlPlane), config.DataStore)...)
 	resources = append(resources, getKubernetesCertificatesResources(config.client, config.tcpReconcilerConfig, config.tenantControlPlane)...)
 	resources = append(resources, getKubeconfigResources(config.client, config.tcpReconcilerConfig, config.tenantControlPlane)...)
 	resources = append(resources, getKubernetesStorageResources(config.client, config.Connection, config.DataStore)...)
 	resources = append(resources, getInternalKonnectivityResources(config.client)...)
 	resources = append(resources, getKubernetesDeploymentResources(config.client, config.tcpReconcilerConfig, config.DataStore)...)
+	resources = append(resources, getDataStoreMigratingCleanup(config.client, config.KamajiNamespace)...)
 	resources = append(resources, getKubernetesIngressResources(config.client)...)
 	resources = append(resources, getKubeadmPhaseResources(config.client)...)
 	resources = append(resources, getKubeadmAddonResources(config.client)...)
@@ -76,6 +81,26 @@ func getDefaultResources(config GroupResourceBuilderConfiguration) []resources.R
 	return resources
 }
 
+func getDataStoreMigratingCleanup(c client.Client, kamajiNamespace string) []resources.Resource {
+	return []resources.Resource{
+		&resources.DatastoreMigrate{
+			Client:          c,
+			KamajiNamespace: kamajiNamespace,
+			ShouldCleanUp:   true,
+		},
+	}
+}
+
+func getDataStoreMigratingResources(c client.Client, kamajiNamespace, kamajiServiceAccount string) []resources.Resource {
+	return []resources.Resource{
+		&resources.DatastoreMigrate{
+			Client:               c,
+			KamajiNamespace:      kamajiNamespace,
+			KamajiServiceAccount: kamajiServiceAccount,
+		},
+	}
+}
+
 func getUpgradeResources(c client.Client) []resources.Resource {
 	return []resources.Resource{
 		&resources.KubernetesUpgrade{

controllers/tenantcontrolplane_controller.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/pkg/errors"
 	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -17,11 +18,14 @@ import (
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	kamajiv1alpha1 "github.com/clastix/kamaji/api/v1alpha1"
@@ -33,11 +37,13 @@ import (
 
 // TenantControlPlaneReconciler reconciles a TenantControlPlane object.
 type TenantControlPlaneReconciler struct {
-	Client      client.Client
-	APIReader   client.Reader
-	Scheme      *runtime.Scheme
-	Config      TenantControlPlaneReconcilerConfig
-	TriggerChan TenantControlPlaneChannel
+	Client               client.Client
+	APIReader            client.Reader
+	Scheme               *runtime.Scheme
+	Config               TenantControlPlaneReconcilerConfig
+	TriggerChan          TenantControlPlaneChannel
+	KamajiNamespace      string
+	KamajiServiceAccount string
 }
 
 // TenantControlPlaneReconcilerConfig gives the necessary configuration for TenantControlPlaneReconciler.
@@ -55,6 +61,7 @@ type TenantControlPlaneReconcilerConfig struct {
 //+kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=networking.k8s.io,resources=ingresses,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;delete
 
 func (r *TenantControlPlaneReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := log.FromContext(ctx)
@@ -116,12 +123,14 @@ func (r *TenantControlPlaneReconciler) Reconcile(ctx context.Context, req ctrl.R
 	}
 
 	groupResourceBuilderConfiguration := GroupResourceBuilderConfiguration{
-		client:              r.Client,
-		log:                 log,
-		tcpReconcilerConfig: r.Config,
-		tenantControlPlane:  *tenantControlPlane,
-		DataStore:           *ds,
-		Connection:          dsConnection,
+		client:               r.Client,
+		log:                  log,
+		tcpReconcilerConfig:  r.Config,
+		tenantControlPlane:   *tenantControlPlane,
+		DataStore:            *ds,
+		Connection:           dsConnection,
+		KamajiNamespace:      r.KamajiNamespace,
+		KamajiServiceAccount: r.KamajiServiceAccount,
 	}
 	registeredResources := GetResources(groupResourceBuilderConfiguration)
 
@@ -176,6 +185,34 @@ func (r *TenantControlPlaneReconciler) SetupWithManager(mgr ctrl.Manager) error
 		Owns(&appsv1.Deployment{}).
 		Owns(&corev1.Service{}).
 		Owns(&networkingv1.Ingress{}).
+		Watches(&source.Kind{Type: &batchv1.Job{}}, handler.EnqueueRequestsFromMapFunc(func(object client.Object) []reconcile.Request {
+			labels := object.GetLabels()
+
+			name, namespace := labels["tcp.kamaji.clastix.io/name"], labels["tcp.kamaji.clastix.io/namespace"]
+
+			return []reconcile.Request{
+				{
+					NamespacedName: k8stypes.NamespacedName{
+						Namespace: namespace,
+						Name:      name,
+					},
+				},
+			}
+		}), builder.WithPredicates(predicate.NewPredicateFuncs(func(object client.Object) bool {
+			if object.GetNamespace() != r.KamajiNamespace {
+				return false
+			}
+
+			labels := object.GetLabels()
+
+			if labels == nil {
+				return false
+			}
+
+			v, ok := labels["kamaji.clastix.io/component"]
+
+			return ok && v == "migrate"
+		}))).
 		Complete(r)
 }
 
@@ -232,10 +269,6 @@ func (r *TenantControlPlaneReconciler) dataStore(ctx context.Context, tenantCont
 		dataStoreName = r.Config.DefaultDataStoreName
 	}
 
-	if statusDataStore := tenantControlPlane.Status.Storage.DataStoreName; len(statusDataStore) > 0 && dataStoreName != statusDataStore {
-		return nil, fmt.Errorf("migration from a DataStore (current: %s) to another one (desired: %s) is not yet supported", statusDataStore, dataStoreName)
-	}
-
 	ds := &kamajiv1alpha1.DataStore{}
 	if err := r.Client.Get(ctx, k8stypes.NamespacedName{Name: dataStoreName}, ds); err != nil {
 		return nil, errors.Wrap(err, "cannot retrieve *kamajiv1alpha.DataStore object")

internal/builders/controlplane/deployment.go

@@ -566,7 +566,7 @@ func (d *Deployment) buildKubeAPIServerCommand(tenantControlPlane *kamajiv1alpha
 		}
 
 		desiredArgs["--etcd-compaction-interval"] = "0"
-		desiredArgs["--etcd-prefix"] = fmt.Sprintf("/%s", tenantControlPlane.GetName())
+		desiredArgs["--etcd-prefix"] = fmt.Sprintf("/%s_%s", tenantControlPlane.GetNamespace(), tenantControlPlane.GetName())
 		desiredArgs["--etcd-servers"] = strings.Join(httpsEndpoints, ",")
 		desiredArgs["--etcd-cafile"] = "/etc/kubernetes/pki/etcd/ca.crt"
 		desiredArgs["--etcd-certfile"] = "/etc/kubernetes/pki/etcd/server.crt"

internal/resources/datastore_migrate.go

@@ -0,0 +1,159 @@
+// Copyright 2022 Clastix Labs
+// SPDX-License-Identifier: Apache-2.0
+
+package resources
+
+import (
+	"context"
+	"fmt"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	kamajiv1alpha1 "github.com/clastix/kamaji/api/v1alpha1"
+	"github.com/clastix/kamaji/internal/utilities"
+)
+
+type DatastoreMigrate struct {
+	Client               client.Client
+	KamajiNamespace      string
+	KamajiServiceAccount string
+	ShouldCleanUp        bool
+
+	actualDatastore  *kamajiv1alpha1.DataStore
+	desiredDatastore *kamajiv1alpha1.DataStore
+	job              *batchv1.Job
+
+	inProgress bool
+}
+
+func (d *DatastoreMigrate) Define(ctx context.Context, tenantControlPlane *kamajiv1alpha1.TenantControlPlane) error {
+	if len(tenantControlPlane.Status.Storage.DataStoreName) == 0 {
+		return nil
+	}
+
+	d.job = &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("migrate-%s-%s", tenantControlPlane.GetNamespace(), tenantControlPlane.GetName()),
+			Namespace: d.KamajiNamespace,
+		},
+	}
+
+	if d.ShouldCleanUp {
+		return nil
+	}
+
+	if err := d.Client.Get(ctx, types.NamespacedName{Name: d.job.GetName(), Namespace: d.job.GetNamespace()}, d.job); err != nil {
+		if !errors.IsNotFound(err) {
+			return err
+		}
+	}
+
+	d.actualDatastore = &kamajiv1alpha1.DataStore{}
+	if err := d.Client.Get(ctx, types.NamespacedName{Name: tenantControlPlane.Status.Storage.DataStoreName}, d.actualDatastore); err != nil {
+		return err
+	}
+
+	d.desiredDatastore = &kamajiv1alpha1.DataStore{}
+	if err := d.Client.Get(ctx, types.NamespacedName{Name: tenantControlPlane.Spec.DataStore}, d.desiredDatastore); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (d *DatastoreMigrate) ShouldCleanup(*kamajiv1alpha1.TenantControlPlane) bool {
+	return d.ShouldCleanUp
+}
+
+func (d *DatastoreMigrate) CleanUp(ctx context.Context, _ *kamajiv1alpha1.TenantControlPlane) (bool, error) {
+	if err := d.Client.Get(ctx, types.NamespacedName{Name: d.job.GetName(), Namespace: d.job.GetNamespace()}, d.job); err != nil && errors.IsNotFound(err) {
+		return false, nil
+	}
+
+	err := d.Client.Delete(ctx, d.job)
+
+	return err == nil, err
+}
+
+func (d *DatastoreMigrate) CreateOrUpdate(ctx context.Context, tenantControlPlane *kamajiv1alpha1.TenantControlPlane) (controllerutil.OperationResult, error) {
+	if d.desiredDatastore == nil {
+		return controllerutil.OperationResultNone, nil
+	}
+
+	if d.actualDatastore.GetName() == d.desiredDatastore.GetName() {
+		return controllerutil.OperationResultNone, nil
+	}
+
+	res, err := utilities.CreateOrUpdateWithConflict(ctx, d.Client, d.job, func() error {
+		d.job.SetLabels(map[string]string{
+			"tcp.kamaji.clastix.io/name":      tenantControlPlane.GetName(),
+			"tcp.kamaji.clastix.io/namespace": tenantControlPlane.GetNamespace(),
+			"kamaji.clastix.io/component":     "migrate",
+		})
+
+		d.job.Spec.Template.ObjectMeta.Labels = utilities.MergeMaps(d.job.Spec.Template.ObjectMeta.Labels, d.job.Spec.Template.ObjectMeta.Labels)
+		d.job.Spec.Template.Spec.ServiceAccountName = d.KamajiServiceAccount
+		d.job.Spec.Template.Spec.RestartPolicy = corev1.RestartPolicyOnFailure
+		if len(d.job.Spec.Template.Spec.Containers) == 0 {
+			d.job.Spec.Template.Spec.Containers = append(d.job.Spec.Template.Spec.Containers, corev1.Container{})
+		}
+		d.job.Spec.Template.Spec.Containers[0].Name = "migrate"
+		d.job.Spec.Template.Spec.Containers[0].Image = "clastix/kamaji:v0.1.1"
+		d.job.Spec.Template.Spec.Containers[0].Command = []string{"/kamaji"}
+		d.job.Spec.Template.Spec.Containers[0].Args = []string{
+			"migrate",
+			fmt.Sprintf("--tenant-control-plane=%s/%s", tenantControlPlane.GetNamespace(), tenantControlPlane.GetName()),
+			fmt.Sprintf("--target-datastore=%s", tenantControlPlane.Spec.DataStore),
+		}
+
+		return nil
+	})
+	if err != nil {
+		return res, err
+	}
+
+	switch res {
+	case controllerutil.OperationResultNone:
+		if len(d.job.Status.Conditions) == 0 {
+			break
+		}
+
+		condition := d.job.Status.Conditions[0]
+		if condition.Type == batchv1.JobComplete && condition.Status == corev1.ConditionTrue {
+			return controllerutil.OperationResultNone, nil
+		}
+
+		log.FromContext(ctx).Info("migration job not yet completed", "reason", condition.Reason, "message", condition.Message)
+	case controllerutil.OperationResultCreated:
+		break
+	default:
+		return "", fmt.Errorf("unexpected status %s from the migration job", res)
+	}
+
+	d.inProgress = true
+
+	return controllerutil.OperationResultNone, nil
+}
+
+func (d *DatastoreMigrate) GetName() string {
+	return "migrate"
+}
+
+func (d *DatastoreMigrate) ShouldStatusBeUpdated(context.Context, *kamajiv1alpha1.TenantControlPlane) bool {
+	return d.inProgress
+}
+
+func (d *DatastoreMigrate) UpdateTenantControlPlaneStatus(_ context.Context, tenantControlPlane *kamajiv1alpha1.TenantControlPlane) error {
+	if d.inProgress {
+		tenantControlPlane.Status.Kubernetes.Version.Status = &kamajiv1alpha1.VersionMigrating
+	}
+
+	return nil
+}

internal/resources/k8s_deployment_resource.go

@@ -143,6 +143,7 @@ func (r *KubernetesDeploymentResource) deploymentTemplateLabels(ctx context.Cont
 		"component.kamaji.clastix.io/front-proxy-client-certificate":        hash(ctx, tenantControlPlane.GetNamespace(), tenantControlPlane.Status.Certificates.FrontProxyClient.SecretName),
 		"component.kamaji.clastix.io/service-account":                       hash(ctx, tenantControlPlane.GetNamespace(), tenantControlPlane.Status.Certificates.SA.SecretName),
 		"component.kamaji.clastix.io/scheduler-kubeconfig":                  hash(ctx, tenantControlPlane.GetNamespace(), tenantControlPlane.Status.KubeConfig.Scheduler.SecretName),
+		"component.kamaji.clastix.io/datastore":                             tenantControlPlane.Spec.DataStore,
 	}
 
 	return labels

main.go

@@ -10,13 +10,15 @@ import (
 
 	"github.com/clastix/kamaji/cmd"
 	"github.com/clastix/kamaji/cmd/manager"
+	"github.com/clastix/kamaji/cmd/migrate"
 )
 
 func main() {
 	scheme := runtime.NewScheme()
 
-	root, mgr := cmd.NewCmd(scheme), manager.NewCmd(scheme)
+	root, mgr, migrator := cmd.NewCmd(scheme), manager.NewCmd(scheme), migrate.NewCmd(scheme)
 	root.AddCommand(mgr)
+	root.AddCommand(migrator)
 
 	if err := root.Execute(); err != nil {
 		os.Exit(1)