Migrate obcluster (#234)

api/v1alpha1/obcluster_types.go

@@ -32,7 +32,7 @@ type OBClusterSpec struct {
 	// Important: Run "make" to regenerate code after modifying this file
 
 	ClusterName      string                     `json:"clusterName"`
-	ClusterId        int64                      `json:"clusterId,omitempty"`
+	ClusterId        int64                      `json:"clusterId"`
 	OBServerTemplate *apitypes.OBServerTemplate `json:"observer"`
 	MonitorTemplate  *apitypes.MonitorTemplate  `json:"monitor,omitempty"`
 	BackupVolume     *apitypes.BackupVolumeSpec `json:"backupVolume,omitempty"`

api/v1alpha1/obcluster_webhook.go

@@ -220,6 +220,11 @@ func (r *OBCluster) validateMutation() error {
 		} else if r.Spec.Topology[0].Replica != 1 {
 			allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("topology"), r.Spec.Topology, "standalone mode only support single replica"))
 		}
+		// validate migration
+		migrateAnnoVal, migrateAnnoExist := r.GetAnnotations()[oceanbaseconst.AnnotationsSourceClusterAddress]
+		if migrateAnnoExist {
+			allErrs = append(allErrs, field.Invalid(field.NewPath("metadata").Child("annotations").Child(oceanbaseconst.AnnotationsSourceClusterAddress), migrateAnnoVal, "migrate obcluster into standalone mode is not supported"))
+		}
 	}
 
 	// Validate userSecrets

config/crd/bases/oceanbase.oceanbase.com_obclusters.yaml

@@ -2703,6 +2703,7 @@ spec:
                 - root
                 type: object
             required:
+            - clusterId
             - clusterName
             - observer
             - topology

config/rbac/role.yaml

@@ -19,6 +19,12 @@ rules:
   - get
   - list
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods/log
+  verbs:
+  - get
 - apiGroups:
   - ""
   resources:

deploy/operator.yaml

@@ -2726,6 +2726,7 @@ spec:
                 - root
                 type: object
             required:
+            - clusterId
             - clusterName
             - observer
             - topology
@@ -12038,6 +12039,12 @@ rules:
   - get
   - list
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods/log
+  verbs:
+  - get
 - apiGroups:
   - ""
   resources:

internal/const/oceanbase/oceanbase.go

@@ -64,6 +64,7 @@ const (
 	GetConnectionMaxRetries = 10
 	CheckConnectionInterval = 3
 	CheckJobInterval        = 3
+	CheckJobMaxRetries      = 100
 	CommonCheckInterval     = 5
 )
 
@@ -76,6 +77,7 @@ const (
 	AnnotationsIndependentPVCLifecycle = "oceanbase.oceanbase.com/independent-pvc-lifecycle"
 	AnnotationsSinglePVC               = "oceanbase.oceanbase.com/single-pvc"
 	AnnotationsMode                    = "oceanbase.oceanbase.com/mode"
+	AnnotationsSourceClusterAddress    = "oceanbase.oceanbase.com/source-cluster-address"
 )
 
 const (
@@ -174,3 +176,12 @@ const (
 const (
 	TolerateServerPodNotReadyMinutes = 5
 )
+
+const (
+	ClusterNameParam = "cluster"
+	ClusterIdParam   = "cluster_id"
+)
+
+const (
+	CmdVersion = "rpm -q --queryformat '%{VERSION}-%{RELEASE}' oceanbase-ce | sed 's/\\.[^.]*$//'"
+)

internal/const/status/obcluster/obcluster_status.go

@@ -14,6 +14,7 @@ package obcluster
 
 const (
 	New                 = "new"
+	MigrateFromExisting = "migrate from existing"
 	Running             = "running"
 	AddOBZone           = "add obzone"
 	DeleteOBZone        = "delete obzone"

internal/const/status/obzone/obzone_status.go

@@ -10,19 +10,20 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.
 */
 
-package obcluster
+package obzone
 
 const (
-	New               = "new"
-	Maintaining       = "maintaining"
-	Running           = "running"
-	AddOBServer       = "add observer"
-	DeleteOBServer    = "delete observer"
-	Deleting          = "deleting"
-	Upgrade           = "upgrade"
-	BootstrapReady    = "bootstrap ready"
-	FinalizerFinished = "finalizer finished"
-	ScaleUp           = "scale up"
-	ExpandPVC         = "expand pvc"
-	MountBackupVolume = "mount backup volume"
+	New                 = "new"
+	MigrateFromExisting = "migrate from existing"
+	Maintaining         = "maintaining"
+	Running             = "running"
+	AddOBServer         = "add observer"
+	DeleteOBServer      = "delete observer"
+	Deleting            = "deleting"
+	Upgrade             = "upgrade"
+	BootstrapReady      = "bootstrap ready"
+	FinalizerFinished   = "finalizer finished"
+	ScaleUp             = "scale up"
+	ExpandPVC           = "expand pvc"
+	MountBackupVolume   = "mount backup volume"
 )

internal/controller/obcluster_controller.go

@@ -56,6 +56,7 @@ type OBClusterReconciler struct {
 // +kubebuilder:rbac:groups=batch,resources=jobs/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=batch,resources=jobs/finalizers,verbs=update
 // +kubebuilder:rbac:groups=storage.k8s.io,resources=storageclasses,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=pods/log,verbs=get
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.

internal/resource/obcluster/init.go

@@ -19,6 +19,7 @@ import (
 func init() {
 	// obcluster
 	task.GetRegistry().Register(fBootstrapOBCluster, BootstrapOBCluster)
+	task.GetRegistry().Register(fMigrateOBClusterFromExisting, MigrateOBClusterFromExisting)
 	task.GetRegistry().Register(fMaintainOBClusterAfterBootstrap, MaintainOBClusterAfterBootstrap)
 	task.GetRegistry().Register(fAddOBZone, AddOBZone)
 	task.GetRegistry().Register(fDeleteOBZone, DeleteOBZone)

internal/resource/obcluster/names.go

@@ -18,6 +18,7 @@ import (
 
 // obcluster flows
 const (
+	fMigrateOBClusterFromExisting    ttypes.FlowName = "migrate obcluster from existing"
 	fBootstrapOBCluster              ttypes.FlowName = "bootstrap obcluster"
 	fMaintainOBClusterAfterBootstrap ttypes.FlowName = "maintain obcluster after bootstrap"
 	fAddOBZone                       ttypes.FlowName = "add obzone"
@@ -33,6 +34,7 @@ const (
 
 // obcluster tasks
 const (
+	tCheckMigration            ttypes.TaskName = "check before migration"
 	tCheckImageReady           ttypes.TaskName = "check image ready"
 	tCheckClusterMode          ttypes.TaskName = "check cluster mode"
 	tCheckAndCreateUserSecrets ttypes.TaskName = "check and create user secrets"

internal/resource/obcluster/obcluster_flow.go

@@ -18,6 +18,19 @@ import (
 	tasktypes "github.com/oceanbase/ob-operator/pkg/task/types"
 )
 
+func MigrateOBClusterFromExisting() *tasktypes.TaskFlow {
+	return &tasktypes.TaskFlow{
+		OperationContext: &tasktypes.OperationContext{
+			Name:         fMigrateOBClusterFromExisting,
+			Tasks:        []tasktypes.TaskName{tCheckMigration, tCheckImageReady, tCheckClusterMode, tCheckAndCreateUserSecrets, tCreateOBZone, tWaitOBZoneRunning, tCreateUsers, tMaintainOBParameter, tCreateServiceForMonitor, tCreateOBClusterService},
+			TargetStatus: clusterstatus.Running,
+			OnFailure: tasktypes.FailureRule{
+				NextTryStatus: clusterstatus.Failed,
+			},
+		},
+	}
+}
+
 func BootstrapOBCluster() *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{

internal/resource/obcluster/obcluster_manager.go

@@ -54,9 +54,14 @@ func (m *OBClusterManager) GetStatus() string {
 func (m *OBClusterManager) InitStatus() {
 	m.Logger.Info("Newly created cluster, init status")
 	m.Recorder.Event(m.OBCluster, "Init", "", "newly created cluster, init status")
+	_, migrateAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsSourceClusterAddress)
+	initialStatus := clusterstatus.New
+	if migrateAnnoExist {
+		initialStatus = clusterstatus.MigrateFromExisting
+	}
 	status := v1alpha1.OBClusterStatus{
 		Image:        m.OBCluster.Spec.OBServerTemplate.Image,
-		Status:       clusterstatus.New,
+		Status:       initialStatus,
 		OBZoneStatus: make([]apitypes.OBZoneReplicaStatus, 0, len(m.OBCluster.Spec.Topology)),
 	}
 	m.OBCluster.Status = status
@@ -80,6 +85,8 @@ func (m *OBClusterManager) GetTaskFlow() (*tasktypes.TaskFlow, error) {
 	m.Logger.V(oceanbaseconst.LogLevelTrace).Info("Create task flow according to obcluster status")
 	switch m.OBCluster.Status.Status {
 	// create obcluster, return taskFlow to bootstrap obcluster
+	case clusterstatus.MigrateFromExisting:
+		taskFlow, err = task.GetRegistry().Get(fMigrateOBClusterFromExisting)
 	case clusterstatus.New:
 		taskFlow, err = task.GetRegistry().Get(fBootstrapOBCluster)
 	// after obcluster bootstraped, return taskFlow to maintain obcluster after bootstrap
@@ -277,6 +284,8 @@ func (m *OBClusterManager) HandleFailure() {
 
 func (m *OBClusterManager) GetTaskFunc(name tasktypes.TaskName) (tasktypes.TaskFunc, error) {
 	switch name {
+	case tCheckMigration:
+		return m.CheckMigration, nil
 	case tCheckImageReady:
 		return m.CheckImageReady, nil
 	case tCheckClusterMode:

internal/resource/obcluster/obcluster_task.go

@@ -18,7 +18,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -162,6 +161,7 @@ func (m *OBClusterManager) CreateOBZone() tasktypes.TaskError {
 	independentVolumeAnnoVal, independentVolumeAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsIndependentPVCLifecycle)
 	singlePVCAnnoVal, singlePVCAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsSinglePVC)
 	modeAnnoVal, modeAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsMode)
+	migrateAnnoVal, migrateAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsSourceClusterAddress)
 	for _, zone := range m.OBCluster.Spec.Topology {
 		zoneName := m.generateZoneName(zone.Zone)
 		zoneExists := false
@@ -208,6 +208,9 @@ func (m *OBClusterManager) CreateOBZone() tasktypes.TaskError {
 		if modeAnnoExist {
 			obzone.ObjectMeta.Annotations[oceanbaseconst.AnnotationsMode] = modeAnnoVal
 		}
+		if migrateAnnoExist {
+			obzone.ObjectMeta.Annotations[oceanbaseconst.AnnotationsSourceClusterAddress] = migrateAnnoVal
+		}
 		m.Logger.Info("Create obzone", "zone", zoneName)
 		err := m.Client.Create(m.Ctx, obzone)
 		if err != nil {
@@ -430,9 +433,7 @@ func (m *OBClusterManager) ValidateUpgradeInfo() tasktypes.TaskError {
 		return errors.Wrapf(err, "Failed to get version of obcluster %s", m.OBCluster.Name)
 	}
 	// Get target version and patch
-	parts := strings.Split(uuid.New().String(), "-")
-	suffix := parts[len(parts)-1]
-	jobName := fmt.Sprintf("%s-%s", "oceanbase-upgrade", suffix)
+	jobName := fmt.Sprintf("%s-%s", "oceanbase-upgrade", rand.String(6))
 	var backoffLimit int32
 	var ttl int32 = 300
 	container := corev1.Container{
@@ -714,12 +715,10 @@ func (m *OBClusterManager) CreateServiceForMonitor() tasktypes.TaskError {
 	ownerReferenceList = append(ownerReferenceList, ownerReference)
 	selector := make(map[string]string)
 	selector[oceanbaseconst.LabelRefOBCluster] = m.OBCluster.Name
-	parts := strings.Split(uuid.New().String(), "-")
-	suffix := parts[len(parts)-1]
 	monitorService := corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:       m.OBCluster.Namespace,
-			Name:            fmt.Sprintf("svc-monitor-%s-%s", m.OBCluster.Name, suffix),
+			Name:            fmt.Sprintf("svc-monitor-%s-%s", m.OBCluster.Name, rand.String(6)),
 			OwnerReferences: ownerReferenceList,
 		},
 		Spec: corev1.ServiceSpec{
@@ -1011,3 +1010,76 @@ func (m *OBClusterManager) CheckClusterMode() tasktypes.TaskError {
 	}
 	return nil
 }
+
+func (m *OBClusterManager) CheckMigration() tasktypes.TaskError {
+	m.Logger.Info("Check before migration")
+	manager, err := m.getOceanbaseOperationManager()
+	if err != nil {
+		return errors.Wrap(err, "get operation manager")
+	}
+
+	// check version strictly matches
+	targetVersionStr, err := resourceutils.RunJob(m.Client, m.Logger, m.OBCluster.Namespace, fmt.Sprintf("%s-version", m.OBCluster.Name), m.OBCluster.Spec.OBServerTemplate.Image, oceanbaseconst.CmdVersion)
+	if err != nil {
+		return errors.Wrap(err, "get target oceanbase version")
+	}
+
+	sourceVersion, err := manager.GetVersion()
+	if err != nil {
+		return errors.Wrap(err, "get source oceanbase version")
+	}
+
+	if sourceVersion.String() != targetVersionStr {
+		return errors.Errorf("version mismatch source cluster: %s, target cluster: %s", sourceVersion.String(), targetVersionStr)
+	}
+
+	// check obzone matches topology
+	obzoneList, err := manager.ListZones()
+	if err != nil {
+		return errors.Wrap(err, "list obzones")
+	}
+	zoneMap := make(map[string]struct{})
+	for _, zone := range obzoneList {
+		zoneMap[zone.Name] = struct{}{}
+	}
+
+	extraZones := make([]string, 0)
+	for _, obzone := range m.OBCluster.Spec.Topology {
+		_, found := zoneMap[obzone.Zone]
+		if !found {
+			extraZones = append(extraZones, obzone.Zone)
+		} else {
+			delete(zoneMap, obzone.Zone)
+		}
+	}
+	if len(extraZones) > 0 {
+		return errors.Errorf("obzone %s defined but not in source cluster", strings.Join(extraZones, ","))
+	}
+
+	undefinedZones := make([]string, 0)
+	for zone := range zoneMap {
+		undefinedZones = append(undefinedZones, zone)
+	}
+	if len(undefinedZones) > 0 {
+		return errors.Errorf("obzone %s not defined in obcluster's topology", strings.Join(undefinedZones, ","))
+	}
+
+	// check obcluster name and id
+	obclusterNameParamList, err := manager.GetParameter(oceanbaseconst.ClusterNameParam, nil)
+	if err != nil {
+		return errors.Wrap(err, "get obcluster name failed")
+	}
+	obclusterName := obclusterNameParamList[0].Value
+	obclusterIdParamList, err := manager.GetParameter(oceanbaseconst.ClusterIdParam, nil)
+	if err != nil {
+		return errors.Wrap(err, "get obcluster id failed")
+	}
+	obclusterId := obclusterIdParamList[0].Value
+	if obclusterName != m.OBCluster.Spec.ClusterName {
+		return errors.Errorf("Cluster name mismatch, source cluster: %s, current: %s", obclusterName, m.OBCluster.Spec.ClusterName)
+	}
+	if obclusterId != fmt.Sprintf("%d", m.OBCluster.Spec.ClusterId) {
+		return errors.Errorf("Cluster id mismatch, source cluster: %s, current: %d", obclusterId, m.OBCluster.Spec.ClusterId)
+	}
+	return nil
+}

internal/resource/obzone/init.go

@@ -19,6 +19,7 @@ import (
 func init() {
 	// obzone
 	task.GetRegistry().Register(fCreateOBZone, CreateOBZone)
+	task.GetRegistry().Register(fMigrateOBZoneFromExisting, MigrateOBZoneFromExisting)
 	task.GetRegistry().Register(fAddOBServer, AddOBServer)
 	task.GetRegistry().Register(fDeleteOBServer, DeleteOBServer)
 	task.GetRegistry().Register(fPrepareOBZoneForBootstrap, PrepareOBZoneForBootstrap)

internal/resource/obzone/names.go

@@ -18,6 +18,7 @@ import (
 
 // obzone flows
 const (
+	fMigrateOBZoneFromExisting    ttypes.FlowName = "migrate obzone from existing"
 	fPrepareOBZoneForBootstrap    ttypes.FlowName = "prepare obzone for bootstrap"
 	fMaintainOBZoneAfterBootstrap ttypes.FlowName = "maintain obzone after bootstrap"
 	fAddOBServer                  ttypes.FlowName = "add observer"
@@ -36,6 +37,7 @@ const (
 	tCreateOBServer              ttypes.TaskName = "create observer"
 	tUpgradeOBServer             ttypes.TaskName = "upgrade observer"
 	tWaitOBServerUpgraded        ttypes.TaskName = "wait observer upgraded"
+	tDeleteLegacyOBServers       ttypes.TaskName = "delete legacy observers"
 	tDeleteOBServer              ttypes.TaskName = "delete observer"
 	tDeleteAllOBServer           ttypes.TaskName = "delete all observer"
 	tAddZone                     ttypes.TaskName = "add zone"

internal/resource/obzone/obzone_flow.go

@@ -18,6 +18,16 @@ import (
 	tasktypes "github.com/oceanbase/ob-operator/pkg/task/types"
 )
 
+func MigrateOBZoneFromExisting() *tasktypes.TaskFlow {
+	return &tasktypes.TaskFlow{
+		OperationContext: &tasktypes.OperationContext{
+			Name:         fMigrateOBZoneFromExisting,
+			Tasks:        []tasktypes.TaskName{tCreateOBServer, tWaitOBServerRunning, tDeleteLegacyOBServers},
+			TargetStatus: zonestatus.Running,
+		},
+	}
+}
+
 func PrepareOBZoneForBootstrap() *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{