Feature: rolling update observers' resource spec (#401)

api/v1alpha1/obcluster_types.go

@@ -17,10 +17,11 @@ limitations under the License.
 package v1alpha1
 
 import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	apitypes "github.com/oceanbase/ob-operator/api/types"
+	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	tasktypes "github.com/oceanbase/ob-operator/pkg/task/types"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
@@ -80,3 +81,7 @@ type OBClusterList struct {
 func init() {
 	SchemeBuilder.Register(&OBCluster{}, &OBClusterList{})
 }
+
+func (c *OBCluster) SupportStaticIP() bool {
+	return c.Annotations[oceanbaseconst.AnnotationsSupportStaticIP] == "true"
+}

api/v1alpha1/obcluster_webhook.go

@@ -179,32 +179,20 @@ func (r *OBCluster) ValidateUpdate(old runtime.Object) (admission.Warnings, erro
 	}
 	oldMode, existOld := oldCluster.GetAnnotations()[oceanbaseconst.AnnotationsMode]
 	mode, exist := r.GetAnnotations()[oceanbaseconst.AnnotationsMode]
+	oldResource := oldCluster.Spec.OBServerTemplate.Resource
+	newResource := r.Spec.OBServerTemplate.Resource
 	if existOld && exist && oldMode != mode {
 		return nil, errors.New("mode cannot be changed")
-	} else if oldMode != oceanbaseconst.ModeStandalone && (oldCluster.Spec.OBServerTemplate.Resource.Cpu != r.Spec.OBServerTemplate.Resource.Cpu || oldCluster.Spec.OBServerTemplate.Resource.Memory != r.Spec.OBServerTemplate.Resource.Memory) {
-		return nil, errors.New("forbid to modify cpu or memory quota of non-standalone cluster")
+	} else if !oldCluster.SupportStaticIP() && (oldResource.Cpu != newResource.Cpu || oldResource.Memory != newResource.Memory) {
+		return nil, errors.New("forbid to modify cpu or memory quota of non-static-ip cluster")
 	}
 	if r.Spec.BackupVolume == nil && oldCluster.Spec.BackupVolume != nil {
 		return nil, errors.New("forbid to remove backup volume")
 	}
 	var err error
 	if r.Spec.BackupVolume != nil && oldCluster.Spec.BackupVolume == nil {
-		if mode != oceanbaseconst.ModeStandalone && mode != oceanbaseconst.ModeService {
-			observerList := &OBServerList{}
-			err = clt.List(context.TODO(), observerList)
-			if err != nil {
-				return nil, err
-			}
-			keepIpWithCNI := false
-			for _, observer := range observerList.Items {
-				if observer.SupportStaticIP() {
-					keepIpWithCNI = true
-					break
-				}
-			}
-			if !keepIpWithCNI {
-				return nil, errors.New("forbid to add backup volume on dynamical-ip cluster")
-			}
+		if !oldCluster.SupportStaticIP() {
+			err = errors.New("forbid to add backup volume to non-static-ip cluster")
 		}
 	}
 	if r.Spec.OBServerTemplate.Storage.DataStorage.Size.Cmp(oldCluster.Spec.OBServerTemplate.Storage.DataStorage.Size) > 0 {

api/v1alpha1/obzone_types.go

@@ -20,6 +20,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
+	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	tasktypes "github.com/oceanbase/ob-operator/pkg/task/types"
 )
 
@@ -79,3 +80,7 @@ type OBZoneList struct {
 func init() {
 	SchemeBuilder.Register(&OBZone{}, &OBZoneList{})
 }
+
+func (z *OBZone) SupportStaticIP() bool {
+	return z.Annotations[oceanbaseconst.AnnotationsSupportStaticIP] == "true"
+}

internal/const/oceanbase/annotations.go

@@ -24,6 +24,7 @@ const (
 	AnnotationsSourceClusterAddress    = "oceanbase.oceanbase.com/source-cluster-address"
 	AnnotationsIgnoreDeletion          = "oceanbase.oceanbase.com/ignore-deletion"
 	AnnotationsPauseReconciling        = "oceanbase.oceanbase.com/pause-reconciling"
+	AnnotationsSupportStaticIP         = "oceanbase.oceanbase.com/support-static-ip"
 )
 
 const (

internal/resource/obcluster/names.go

@@ -64,4 +64,5 @@ const (
 	tExpandPVC                  ttypes.TaskName = "expand pvc"
 	tMountBackupVolume          ttypes.TaskName = "mount backup volume"
 	tCheckEnvironment           ttypes.TaskName = "check environment"
+	tAnnotateOBCluster          ttypes.TaskName = "annotate obcluster"
 )

internal/resource/obcluster/obcluster_flow.go

@@ -21,8 +21,21 @@ import (
 func genMigrateOBClusterFromExistingFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fMigrateOBClusterFromExisting,
-			Tasks:        []tasktypes.TaskName{tCheckMigration, tCheckImageReady, tCheckEnvironment, tCheckClusterMode, tCheckAndCreateUserSecrets, tCreateOBZone, tWaitOBZoneRunning, tCreateUsers, tMaintainOBParameter, tCreateServiceForMonitor, tCreateOBClusterService},
+			Name: fMigrateOBClusterFromExisting,
+			Tasks: []tasktypes.TaskName{
+				tCheckMigration,
+				tCheckImageReady,
+				tCheckEnvironment,
+				tCheckClusterMode,
+				tCheckAndCreateUserSecrets,
+				tCreateOBZone,
+				tWaitOBZoneRunning,
+				tCreateUsers,
+				tMaintainOBParameter,
+				tCreateServiceForMonitor,
+				tCreateOBClusterService,
+				tAnnotateOBCluster,
+			},
 			TargetStatus: clusterstatus.Running,
 			OnFailure: tasktypes.FailureRule{
 				NextTryStatus: clusterstatus.Failed,
@@ -34,8 +47,16 @@ func genMigrateOBClusterFromExistingFlow(_ *OBClusterManager) *tasktypes.TaskFlo
 func genBootstrapOBClusterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fBootstrapOBCluster,
-			Tasks:        []tasktypes.TaskName{tCheckImageReady, tCheckEnvironment, tCheckClusterMode, tCheckAndCreateUserSecrets, tCreateOBZone, tWaitOBZoneBootstrapReady, tBootstrap},
+			Name: fBootstrapOBCluster,
+			Tasks: []tasktypes.TaskName{
+				tCheckImageReady,
+				tCheckEnvironment,
+				tCheckClusterMode,
+				tCheckAndCreateUserSecrets,
+				tCreateOBZone,
+				tWaitOBZoneBootstrapReady,
+				tBootstrap,
+			},
 			TargetStatus: clusterstatus.Bootstrapped,
 			OnFailure: tasktypes.FailureRule{
 				NextTryStatus: clusterstatus.Failed,
@@ -47,8 +68,15 @@ func genBootstrapOBClusterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genMaintainOBClusterAfterBootstrapFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fMaintainOBClusterAfterBootstrap,
-			Tasks:        []tasktypes.TaskName{tWaitOBZoneRunning, tCreateUsers, tMaintainOBParameter, tCreateServiceForMonitor, tCreateOBClusterService},
+			Name: fMaintainOBClusterAfterBootstrap,
+			Tasks: []tasktypes.TaskName{
+				tWaitOBZoneRunning,
+				tCreateUsers,
+				tMaintainOBParameter,
+				tCreateServiceForMonitor,
+				tCreateOBClusterService,
+				tAnnotateOBCluster,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -57,8 +85,12 @@ func genMaintainOBClusterAfterBootstrapFlow(_ *OBClusterManager) *tasktypes.Task
 func genAddOBZoneFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fAddOBZone,
-			Tasks:        []tasktypes.TaskName{tCreateOBZone, tWaitOBZoneRunning, tModifySysTenantReplica},
+			Name: fAddOBZone,
+			Tasks: []tasktypes.TaskName{
+				tCreateOBZone,
+				tWaitOBZoneRunning,
+				tModifySysTenantReplica,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -67,8 +99,12 @@ func genAddOBZoneFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genDeleteOBZoneFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fDeleteOBZone,
-			Tasks:        []tasktypes.TaskName{tModifySysTenantReplica, tDeleteOBZone, tWaitOBZoneDeleted},
+			Name: fDeleteOBZone,
+			Tasks: []tasktypes.TaskName{
+				tModifySysTenantReplica,
+				tDeleteOBZone,
+				tWaitOBZoneDeleted,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -77,8 +113,12 @@ func genDeleteOBZoneFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genModifyOBZoneReplicaFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fModifyOBZoneReplica,
-			Tasks:        []tasktypes.TaskName{tModifyOBZoneReplica, tWaitOBZoneTopologyMatch, tWaitOBZoneRunning},
+			Name: fModifyOBZoneReplica,
+			Tasks: []tasktypes.TaskName{
+				tModifyOBZoneReplica,
+				tWaitOBZoneTopologyMatch,
+				tWaitOBZoneRunning,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -87,8 +127,10 @@ func genModifyOBZoneReplicaFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genMaintainOBParameterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fMaintainOBParameter,
-			Tasks:        []tasktypes.TaskName{tMaintainOBParameter},
+			Name: fMaintainOBParameter,
+			Tasks: []tasktypes.TaskName{
+				tMaintainOBParameter,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -97,8 +139,16 @@ func genMaintainOBParameterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genUpgradeOBClusterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fUpgradeOBCluster,
-			Tasks:        []tasktypes.TaskName{tValidateUpgradeInfo, tBackupEssentialParameters, tUpgradeCheck, tBeginUpgrade, tRollingUpgradeByZone, tFinishUpgrade, tRestoreEssentialParameters},
+			Name: fUpgradeOBCluster,
+			Tasks: []tasktypes.TaskName{
+				tValidateUpgradeInfo,
+				tBackupEssentialParameters,
+				tUpgradeCheck,
+				tBeginUpgrade,
+				tRollingUpgradeByZone,
+				tFinishUpgrade,
+				tRestoreEssentialParameters,
+			},
 			TargetStatus: clusterstatus.Running,
 			OnFailure: tasktypes.FailureRule{
 				Strategy: strategy.Pause,
@@ -110,8 +160,11 @@ func genUpgradeOBClusterFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genScaleUpOBZonesFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fScaleUpOBZones,
-			Tasks:        []tasktypes.TaskName{tScaleUpOBZones, tWaitOBZoneRunning},
+			Name: fScaleUpOBZones,
+			Tasks: []tasktypes.TaskName{
+				tScaleUpOBZones,
+				tWaitOBZoneRunning,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -120,8 +173,11 @@ func genScaleUpOBZonesFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genExpandPVCFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fExpandPVC,
-			Tasks:        []tasktypes.TaskName{tExpandPVC, tWaitOBZoneRunning},
+			Name: fExpandPVC,
+			Tasks: []tasktypes.TaskName{
+				tExpandPVC,
+				tWaitOBZoneRunning,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}
@@ -130,8 +186,11 @@ func genExpandPVCFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 func genMountBackupVolumeFlow(_ *OBClusterManager) *tasktypes.TaskFlow {
 	return &tasktypes.TaskFlow{
 		OperationContext: &tasktypes.OperationContext{
-			Name:         fMountBackupVolume,
-			Tasks:        []tasktypes.TaskName{tMountBackupVolume, tWaitOBZoneRunning},
+			Name: fMountBackupVolume,
+			Tasks: []tasktypes.TaskName{
+				tMountBackupVolume,
+				tWaitOBZoneRunning,
+			},
 			TargetStatus: clusterstatus.Running,
 		},
 	}

internal/resource/obcluster/obcluster_manager.go

@@ -180,10 +180,9 @@ func (m *OBClusterManager) UpdateStatus() error {
 			m.Logger.Info("Compare topology need delete zone")
 			m.OBCluster.Status.Status = clusterstatus.DeleteOBZone
 		} else {
-			modeAnnoVal, modeAnnoExist := resourceutils.GetAnnotationField(m.OBCluster, oceanbaseconst.AnnotationsMode)
 		outer:
 			for _, obzone := range obzoneList.Items {
-				if modeAnnoExist && modeAnnoVal == oceanbaseconst.ModeStandalone && m.checkIfCalcResourceChange(&obzone) {
+				if m.OBCluster.SupportStaticIP() && m.checkIfCalcResourceChange(&obzone) {
 					m.OBCluster.Status.Status = clusterstatus.ScaleUp
 					break outer
 				}

internal/resource/obcluster/obcluster_task.go

@@ -974,7 +974,7 @@ func CheckMigration(m *OBClusterManager) tasktypes.TaskError {
 }
 
 func ScaleUpOBZones(m *OBClusterManager) tasktypes.TaskError {
-	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenScaling, zonestatus.ScaleUp, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
+	return m.rollingUpdateZones(m.changeZonesWhenScaling, zonestatus.ScaleUp, zonestatus.Running, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func ExpandPVC(m *OBClusterManager) tasktypes.TaskError {
@@ -1029,7 +1029,7 @@ func CheckEnvironment(m *OBClusterManager) tasktypes.TaskError {
 	defer func() {
 		err = m.Client.Delete(m.Ctx, pvc)
 		if err != nil {
-			m.Logger.Info("Failed to delete pvc for checking storage")
+			m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Failed to delete pvc for checking storage", "err", err)
 		}
 	}()
 	// Assemble volumeConfigs
@@ -1060,3 +1060,58 @@ func CheckEnvironment(m *OBClusterManager) tasktypes.TaskError {
 	}
 	return nil
 }
+
+func AnnotateOBCluster(m *OBClusterManager) tasktypes.TaskError {
+	// Annotate obcluster with mode
+	supportStaticIP := false
+	mode := m.OBCluster.Annotations[oceanbaseconst.AnnotationsMode]
+	withMode := mode == oceanbaseconst.ModeService || mode == oceanbaseconst.ModeStandalone
+
+	if withMode {
+		supportStaticIP = true
+	} else {
+		serverList := &v1alpha1.OBServerList{}
+		err := m.Client.List(m.Ctx, serverList, client.MatchingLabels{oceanbaseconst.LabelRefOBCluster: m.OBCluster.Name}, client.InNamespace(m.OBCluster.Namespace))
+		if err != nil {
+			return errors.Wrap(err, "List servers of obcluster")
+		}
+		if len(serverList.Items) == 0 {
+			return errors.New("No server found for obcluster")
+		}
+		for _, server := range serverList.Items {
+			if server.Status.CNI != oceanbaseconst.CNIUnknown {
+				supportStaticIP = true
+				break
+			}
+		}
+	}
+
+	if supportStaticIP {
+		copied := m.OBCluster.DeepCopy()
+		if copied.Annotations == nil {
+			copied.Annotations = make(map[string]string)
+		}
+		copied.Annotations[oceanbaseconst.AnnotationsSupportStaticIP] = "true"
+		err := m.Client.Patch(m.Ctx, copied, client.MergeFrom(m.OBCluster))
+		if err != nil {
+			return errors.Wrap(err, "Patch obcluster")
+		}
+		zones, err := m.listOBZones()
+		if err != nil {
+			return errors.Wrap(err, "List obzones")
+		}
+		for _, zone := range zones.Items {
+			copiedZone := zone.DeepCopy()
+			if copiedZone.Annotations == nil {
+				copiedZone.Annotations = make(map[string]string)
+			}
+			copiedZone.Annotations[oceanbaseconst.AnnotationsSupportStaticIP] = "true"
+			err = m.Client.Patch(m.Ctx, copiedZone, client.MergeFrom(&zone))
+			if err != nil {
+				return errors.Wrap(err, "Patch obzone")
+			}
+		}
+	}
+
+	return nil
+}

internal/resource/observer/observer_manager.go

@@ -114,10 +114,13 @@ func (m *OBServerManager) UpdateStatus() error {
 		// 1. Check status of observer in OB database
 		if m.OBServer.Status.Status == serverstatus.Running {
 			m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Check observer in obcluster")
-			if mode, exist := resourceutils.GetAnnotationField(m.OBServer, oceanbaseconst.AnnotationsMode); exist && mode == oceanbaseconst.ModeStandalone {
-				if pod.Spec.Containers[0].Resources.Limits.Cpu().Cmp(m.OBServer.Spec.OBServerTemplate.Resource.Cpu) != 0 ||
-					pod.Spec.Containers[0].Resources.Limits.Memory().Cmp(m.OBServer.Spec.OBServerTemplate.Resource.Memory) != 0 {
-					m.OBServer.Status.Status = serverstatus.ScaleUp
+			if m.OBServer.SupportStaticIP() {
+				if len(pod.Spec.Containers) > 0 {
+					tmplRes := m.OBServer.Spec.OBServerTemplate.Resource
+					containerRes := pod.Spec.Containers[0].Resources.Limits
+					if containerRes.Cpu().Cmp(tmplRes.Cpu) != 0 || containerRes.Memory().Cmp(tmplRes.Memory) != 0 {
+						m.OBServer.Status.Status = serverstatus.ScaleUp
+					}
 				}
 			} else if pvcs != nil && len(pvcs.Items) > 0 && m.checkIfStorageExpand(pvcs) {
 				m.OBServer.Status.Status = serverstatus.ExpandPVC