Refactor/operator cfg (#333)

api/v1alpha1/obcluster_webhook.go

@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 )
 
@@ -61,7 +62,7 @@ func (r *OBCluster) Default() {
 	parameterMap := make(map[string]apitypes.Parameter, 0)
 	memorySize, ok := r.Spec.OBServerTemplate.Resource.Memory.AsInt64()
 	if ok {
-		memoryLimit := fmt.Sprintf("%dM", memorySize*oceanbaseconst.DefaultMemoryLimitPercent/100/oceanbaseconst.MegaConverter)
+		memoryLimit := fmt.Sprintf("%dM", memorySize*int64(obcfg.GetConfig().Resource.DefaultMemoryLimitPercent)/100/oceanbaseconst.MegaConverter)
 		parameterMap["memory_limit"] = apitypes.Parameter{
 			Name:  "memory_limit",
 			Value: memoryLimit,
@@ -71,12 +72,12 @@ func (r *OBCluster) Default() {
 	}
 	datafileDiskSize, ok := r.Spec.OBServerTemplate.Storage.DataStorage.Size.AsInt64()
 	if ok {
-		datafileMaxSize := fmt.Sprintf("%dG", datafileDiskSize*oceanbaseconst.DefaultDiskUsePercent/oceanbaseconst.GigaConverter/100)
+		datafileMaxSize := fmt.Sprintf("%dG", datafileDiskSize*int64(obcfg.GetConfig().Resource.DefaultDiskUsePercent)/oceanbaseconst.GigaConverter/100)
 		parameterMap["datafile_maxsize"] = apitypes.Parameter{
 			Name:  "datafile_maxsize",
 			Value: datafileMaxSize,
 		}
-		datafileNextSize := fmt.Sprintf("%dG", datafileDiskSize*oceanbaseconst.DefaultDiskExpandPercent/oceanbaseconst.GigaConverter/100)
+		datafileNextSize := fmt.Sprintf("%dG", datafileDiskSize*int64(obcfg.GetConfig().Resource.DefaultDiskExpandPercent)/oceanbaseconst.GigaConverter/100)
 		parameterMap["datafile_next"] = apitypes.Parameter{
 			Name:  "datafile_next",
 			Value: datafileNextSize,
@@ -92,7 +93,7 @@ func (r *OBCluster) Default() {
 	logSize, ok := r.Spec.OBServerTemplate.Storage.LogStorage.Size.AsInt64()
 	if ok {
 		// observer has 4 types of log and one logfile limits at 256M considering about wf, maximum of 2G will be occupied for 1 syslog count
-		maxSysLogFileCount = logSize * oceanbaseconst.DefaultLogPercent / oceanbaseconst.GigaConverter / 100 / 2
+		maxSysLogFileCount = logSize * int64(obcfg.GetConfig().Resource.DefaultLogPercent) / oceanbaseconst.GigaConverter / 100 / 2
 	}
 	parameterMap["max_syslog_file_count"] = apitypes.Parameter{
 		Name:  "max_syslog_file_count",
@@ -275,18 +276,17 @@ func (r *OBCluster) validateMutation() error {
 	}
 
 	// Validate disk size
-	if r.Spec.OBServerTemplate.Storage.DataStorage.Size.AsApproximateFloat64() < oceanbaseconst.MinDataDiskSize.AsApproximateFloat64() {
+	if r.Spec.OBServerTemplate.Storage.DataStorage.Size.Cmp(resource.MustParse(obcfg.GetConfig().Resource.MinDataDiskSize)) < 0 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("dataStorage").Child("size"), r.Spec.OBServerTemplate.Storage.DataStorage.Size.String(), "The minimum data storage size of OBCluster is "+oceanbaseconst.MinDataDiskSize.String()))
 	}
-	if r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.AsApproximateFloat64() < oceanbaseconst.MinRedoLogDiskSize.AsApproximateFloat64() {
+	if r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.Cmp(resource.MustParse(obcfg.GetConfig().Resource.MinRedoLogDiskSize)) < 0 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("redoLogStorage").Child("size"), r.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.String(), "The minimum redo log storage size of OBCluster is "+oceanbaseconst.MinRedoLogDiskSize.String()))
 	}
-	if r.Spec.OBServerTemplate.Storage.LogStorage.Size.AsApproximateFloat64() < oceanbaseconst.MinLogDiskSize.AsApproximateFloat64() {
+	if r.Spec.OBServerTemplate.Storage.LogStorage.Size.Cmp(resource.MustParse(obcfg.GetConfig().Resource.MinLogDiskSize)) < 0 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("observer").Child("storage").Child("logStorage").Child("size"), r.Spec.OBServerTemplate.Storage.LogStorage.Size.String(), "The minimum log storage size of OBCluster is "+oceanbaseconst.MinLogDiskSize.String()))
 	}
-
 	// Validate memory size
-	if r.Spec.OBServerTemplate.Resource.Memory.AsApproximateFloat64() < oceanbaseconst.MinMemorySize.AsApproximateFloat64() {
+	if r.Spec.OBServerTemplate.Resource.Memory.Cmp(resource.MustParse(obcfg.GetConfig().Resource.MinMemorySize)) < 0 {
 		allErrs = append(allErrs, field.Invalid(field.NewPath("spec").Child("observer").Child("resource").Child("memory"), r.Spec.OBServerTemplate.Resource.Memory.String(), "The minimum memory size of OBCluster is "+oceanbaseconst.MinMemorySize.String()))
 	}
 

cmd/operator/main.go

@@ -33,9 +33,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	"github.com/oceanbase/ob-operator/internal/controller"
 	"github.com/oceanbase/ob-operator/internal/controller/config"
 	"github.com/oceanbase/ob-operator/internal/telemetry"
+	"github.com/oceanbase/ob-operator/pkg/coordinator"
 )
 
 var (
@@ -86,6 +88,10 @@ func main() {
 		},
 	}
 
+	cfg := obcfg.GetConfig()
+	coordinator.SetMaxRetryTimes(cfg.Time.TaskMaxRetryTimes)
+	coordinator.SetRetryBackoffThreshold(cfg.Time.TaskRetryBackoffThreshold)
+
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{

internal/config/operator/types.go

@@ -15,7 +15,8 @@ package operator
 import "github.com/spf13/viper"
 
 type Config struct {
-	v         *viper.Viper
+	v *viper.Viper
+
 	Manager   Manager   `mapstructure:",squash" yaml:"manager"`
 	Database  Database  `mapstructure:"database" yaml:"database"`
 	Task      Task      `mapstructure:"task" yaml:"task"`

internal/const/oceanbase/resource.go

@@ -20,8 +20,11 @@ const (
 	InitialDataDiskUsePercent = 20
 	DefaultDiskUsePercent     = 95
 	DefaultMemoryLimitPercent = 90
-	GigaConverter             = 1 << 30
-	MegaConverter             = 1 << 20
+)
+
+const (
+	GigaConverter = 1 << 30
+	MegaConverter = 1 << 20
 )
 
 const (

internal/resource/obcluster/obcluster_task.go

@@ -33,6 +33,7 @@ import (
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	obagentconst "github.com/oceanbase/ob-operator/internal/const/obagent"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	zonestatus "github.com/oceanbase/ob-operator/internal/const/status/obzone"
@@ -54,7 +55,7 @@ func WaitOBZoneTopologyMatch(_ *OBClusterManager) tasktypes.TaskError {
 
 func WaitOBZoneDeleted(m *OBClusterManager) tasktypes.TaskError {
 	waitSuccess := false
-	for i := 1; i < oceanbaseconst.ServerDeleteTimeoutSeconds; i++ {
+	for i := 1; i < obcfg.GetConfig().Time.ServerDeleteTimeoutSeconds; i++ {
 		obcluster, err := m.getOBCluster()
 		if err != nil {
 			return errors.Wrap(err, "get obcluster failed")
@@ -212,11 +213,11 @@ func Bootstrap(m *OBClusterManager) tasktypes.TaskError {
 		return errors.Wrap(err, "no obzone belongs to this cluster")
 	}
 	var manager *operation.OceanbaseOperationManager
-	for i := 0; i < oceanbaseconst.GetConnectionMaxRetries; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.GetConnectionMaxRetries; i++ {
 		manager, err = m.getOceanbaseOperationManager()
 		if err != nil || manager == nil {
 			m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Get oceanbase operation manager failed")
-			time.Sleep(time.Second * oceanbaseconst.CheckConnectionInterval)
+			time.Sleep(time.Second * time.Duration(obcfg.GetConfig().Time.CheckConnectionInterval))
 		} else {
 			m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Successfully got oceanbase operation manager")
 			break
@@ -393,7 +394,7 @@ func ValidateUpgradeInfo(m *OBClusterManager) tasktypes.TaskError {
 			return false, errors.Wrap(err, "Failed to run validate job")
 		}
 	}
-	err = resourceutils.CheckJobWithTimeout(check, time.Second*oceanbaseconst.WaitForJobTimeoutSeconds)
+	err = resourceutils.CheckJobWithTimeout(check, time.Second*time.Duration(obcfg.GetConfig().Time.WaitForJobTimeoutSeconds))
 	if err != nil {
 		return errors.Wrap(err, "Failed to run validate job")
 	}
@@ -535,7 +536,7 @@ func ModifySysTenantReplica(m *OBClusterManager) tasktypes.TaskError {
 			if err != nil {
 				return errors.Wrapf(err, "Failed to set sys locality to %s", locality)
 			}
-			err = oceanbaseOperationManager.WaitTenantLocalityChangeFinished(oceanbaseconst.SysTenant, oceanbaseconst.LocalityChangeTimeoutSeconds)
+			err = oceanbaseOperationManager.WaitTenantLocalityChangeFinished(oceanbaseconst.SysTenant, obcfg.GetConfig().Time.LocalityChangeTimeoutSeconds)
 			if err != nil {
 				return errors.Wrapf(err, "Locality change to %s not finished after timeout", locality)
 			}
@@ -561,7 +562,7 @@ func ModifySysTenantReplica(m *OBClusterManager) tasktypes.TaskError {
 			if err != nil {
 				return errors.Wrapf(err, "Failed to set sys locality to %s", locality)
 			}
-			err = oceanbaseOperationManager.WaitTenantLocalityChangeFinished(oceanbaseconst.SysTenant, oceanbaseconst.LocalityChangeTimeoutSeconds)
+			err = oceanbaseOperationManager.WaitTenantLocalityChangeFinished(oceanbaseconst.SysTenant, obcfg.GetConfig().Time.LocalityChangeTimeoutSeconds)
 			if err != nil {
 				return errors.Wrapf(err, "Locality change to %s not finished after timeout", locality)
 			}
@@ -791,7 +792,7 @@ outerLoop:
 		}
 		if len(podList.Items) == 0 {
 			m.Logger.V(oceanbaseconst.LogLevelDebug).Info("No pod found for check image pull job")
-			time.Sleep(time.Second * oceanbaseconst.CheckJobInterval)
+			time.Sleep(time.Second * time.Duration(obcfg.GetConfig().Time.CheckJobInterval))
 			continue
 		}
 		pod := podList.Items[0]
@@ -813,7 +814,7 @@ outerLoop:
 					default:
 						m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Container is waiting", "reason", containerStatus.State.Waiting.Reason, "message", containerStatus.State.Waiting.Message)
 					}
-					time.Sleep(time.Second * oceanbaseconst.CheckJobInterval)
+					time.Sleep(time.Second * time.Duration(obcfg.GetConfig().Time.CheckJobInterval))
 					continue outerLoop
 				} else if containerStatus.State.Running != nil || containerStatus.State.Terminated != nil {
 					m.Logger.V(oceanbaseconst.LogLevelDebug).Info("Container is running or terminated")
@@ -929,23 +930,23 @@ func CheckMigration(m *OBClusterManager) tasktypes.TaskError {
 }
 
 func ScaleUpOBZones(m *OBClusterManager) tasktypes.TaskError {
-	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenScaling, zonestatus.ScaleUp, oceanbaseconst.DefaultStateWaitTimeout)()
+	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenScaling, zonestatus.ScaleUp, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func ExpandPVC(m *OBClusterManager) tasktypes.TaskError {
-	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenExpandingPVC, zonestatus.ExpandPVC, oceanbaseconst.DefaultStateWaitTimeout)()
+	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenExpandingPVC, zonestatus.ExpandPVC, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func MountBackupVolume(m *OBClusterManager) tasktypes.TaskError {
-	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenMountingBackupVolume, zonestatus.MountBackupVolume, oceanbaseconst.DefaultStateWaitTimeout)()
+	return m.modifyOBZonesAndCheckStatus(m.changeZonesWhenMountingBackupVolume, zonestatus.MountBackupVolume, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func WaitOBZoneBootstrapReady(m *OBClusterManager) tasktypes.TaskError {
-	return m.generateWaitOBZoneStatusFunc(zonestatus.BootstrapReady, oceanbaseconst.DefaultStateWaitTimeout)()
+	return m.generateWaitOBZoneStatusFunc(zonestatus.BootstrapReady, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func WaitOBZoneRunning(m *OBClusterManager) tasktypes.TaskError {
-	return m.generateWaitOBZoneStatusFunc(zonestatus.Running, oceanbaseconst.DefaultStateWaitTimeout)()
+	return m.generateWaitOBZoneStatusFunc(zonestatus.Running, obcfg.GetConfig().Time.DefaultStateWaitTimeout)()
 }
 
 func CheckEnvironment(m *OBClusterManager) tasktypes.TaskError {

internal/resource/obcluster/utils.go

@@ -25,6 +25,7 @@ import (
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
 	"github.com/oceanbase/ob-operator/api/v1alpha1"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	zonestatus "github.com/oceanbase/ob-operator/internal/const/status/obzone"
 	resourceutils "github.com/oceanbase/ob-operator/internal/resource/utils"
@@ -390,7 +391,7 @@ func (m *OBClusterManager) WaitOBZoneUpgradeFinished(zoneName string) error {
 		}
 		return false, nil
 	}
-	err := resourceutils.CheckJobWithTimeout(check, time.Second*oceanbaseconst.WaitForJobTimeoutSeconds)
+	err := resourceutils.CheckJobWithTimeout(check, time.Second*time.Duration(obcfg.GetConfig().Time.WaitForJobTimeoutSeconds))
 	if err != nil {
 		return errors.Wrap(err, "Timeout to wait obzone upgrade finished")
 	}

internal/resource/observer/observer_task.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	podconst "github.com/oceanbase/ob-operator/internal/const/pod"
 	clusterstatus "github.com/oceanbase/ob-operator/internal/const/status/obcluster"
@@ -82,7 +83,7 @@ func AddServer(m *OBServerManager) tasktypes.TaskError {
 }
 
 func WaitOBClusterBootstrapped(m *OBServerManager) tasktypes.TaskError {
-	for i := 0; i < oceanbaseconst.BootstrapTimeoutSeconds; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.BootstrapTimeoutSeconds; i++ {
 		obcluster, err := m.getOBCluster()
 		if err != nil {
 			return errors.Wrap(err, "Get obcluster from K8s")
@@ -285,7 +286,7 @@ func UpgradeOBServerImage(m *OBServerManager) tasktypes.TaskError {
 
 func WaitOBServerPodReady(m *OBServerManager) tasktypes.TaskError {
 	observerPodRestarted := false
-	for i := 0; i < oceanbaseconst.DefaultStateWaitTimeout; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.DefaultStateWaitTimeout; i++ {
 		observerPod, err := m.getPod()
 		if err != nil {
 			return errors.Wrapf(err, "Failed to get pod of observer %s", m.OBServer.Name)
@@ -320,7 +321,7 @@ func WaitOBServerActiveInCluster(m *OBServerManager) tasktypes.TaskError {
 		Port: oceanbaseconst.RpcPort,
 	}
 	active := false
-	for i := 0; i < oceanbaseconst.DefaultStateWaitTimeout; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.DefaultStateWaitTimeout; i++ {
 		operationManager, err := m.getOceanbaseOperationManager()
 		if err != nil {
 			return errors.Wrapf(err, "Get oceanbase operation manager failed")
@@ -355,7 +356,7 @@ func WaitOBServerDeletedInCluster(m *OBServerManager) tasktypes.TaskError {
 		Port: oceanbaseconst.RpcPort,
 	}
 	deleted := false
-	for i := 0; i < oceanbaseconst.ServerDeleteTimeoutSeconds; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.ServerDeleteTimeoutSeconds; i++ {
 		operationManager, err := m.getOceanbaseOperationManager()
 		if err != nil {
 			return errors.Wrapf(err, "Get oceanbase operation manager failed")
@@ -395,7 +396,7 @@ func DeletePod(m *OBServerManager) tasktypes.TaskError {
 
 func WaitForPodDeleted(m *OBServerManager) tasktypes.TaskError {
 	m.Logger.Info("Wait for observer pod being deleted")
-	for i := 0; i < oceanbaseconst.DefaultStateWaitTimeout; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.DefaultStateWaitTimeout; i++ {
 		time.Sleep(time.Second)
 		err := m.Client.Get(m.Ctx, m.generateNamespacedName(m.OBServer.Name), &corev1.Pod{})
 		if err != nil && kubeerrors.IsNotFound(err) {
@@ -448,7 +449,7 @@ func ExpandPVC(m *OBServerManager) tasktypes.TaskError {
 
 func WaitForPVCResized(m *OBServerManager) tasktypes.TaskError {
 outer:
-	for i := 0; i < oceanbaseconst.DefaultStateWaitTimeout; i++ {
+	for i := 0; i < obcfg.GetConfig().Time.DefaultStateWaitTimeout; i++ {
 		time.Sleep(time.Second)
 
 		observerPVC, err := m.getPVCs()

internal/resource/observer/utils.go

@@ -27,6 +27,7 @@ import (
 
 	apitypes "github.com/oceanbase/ob-operator/api/types"
 	v1alpha1 "github.com/oceanbase/ob-operator/api/v1alpha1"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	obagentconst "github.com/oceanbase/ob-operator/internal/const/obagent"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	secretconst "github.com/oceanbase/ob-operator/internal/const/secret"
@@ -413,8 +414,8 @@ func (m *OBServerManager) createOBServerContainer(obcluster *v1alpha1.OBCluster)
 	readinessProbeTCP.Port = intstr.FromInt(oceanbaseconst.SqlPort)
 	readinessProbe := corev1.Probe{}
 	readinessProbe.ProbeHandler.TCPSocket = &readinessProbeTCP
-	readinessProbe.PeriodSeconds = oceanbaseconst.ProbeCheckPeriodSeconds
-	readinessProbe.InitialDelaySeconds = oceanbaseconst.ProbeCheckDelaySeconds
+	readinessProbe.PeriodSeconds = int32(obcfg.GetConfig().Time.ProbeCheckPeriodSeconds)
+	readinessProbe.InitialDelaySeconds = int32(obcfg.GetConfig().Time.ProbeCheckDelaySeconds)
 	readinessProbe.FailureThreshold = 32
 
 	startOBServerCmd := "/home/admin/oceanbase/bin/oceanbase-helper start"
@@ -445,15 +446,15 @@ func (m *OBServerManager) createOBServerContainer(obcluster *v1alpha1.OBCluster)
 	}
 	envDataFile := corev1.EnvVar{
 		Name:  "DATAFILE_SIZE",
-		Value: fmt.Sprintf("%dG", datafileSize*oceanbaseconst.InitialDataDiskUsePercent/oceanbaseconst.GigaConverter/100),
+		Value: fmt.Sprintf("%dG", datafileSize*int64(obcfg.GetConfig().Resource.InitialDataDiskUsePercent)/oceanbaseconst.GigaConverter/100),
 	}
 	clogDiskSize, ok := m.OBServer.Spec.OBServerTemplate.Storage.RedoLogStorage.Size.AsInt64()
 	if !ok {
 		m.Logger.Error(errors.New("Parse log disk size failed"), "failed to parse log disk size")
 	}
 	envLogDisk := corev1.EnvVar{
 		Name:  "LOG_DISK_SIZE",
-		Value: fmt.Sprintf("%dG", clogDiskSize*oceanbaseconst.DefaultDiskUsePercent/oceanbaseconst.GigaConverter/100),
+		Value: fmt.Sprintf("%dG", clogDiskSize*int64(obcfg.GetConfig().Resource.DefaultDiskUsePercent)/oceanbaseconst.GigaConverter/100),
 	}
 	envClusterName := corev1.EnvVar{
 		Name:  "CLUSTER_NAME",

internal/resource/obtenant/obtenant_task.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/oceanbase/ob-operator/api/constants"
 	"github.com/oceanbase/ob-operator/api/v1alpha1"
+	obcfg "github.com/oceanbase/ob-operator/internal/config/operator"
 	oceanbaseconst "github.com/oceanbase/ob-operator/internal/const/oceanbase"
 	resourceutils "github.com/oceanbase/ob-operator/internal/resource/utils"
 	"github.com/oceanbase/ob-operator/pkg/oceanbase-sdk/const/status/tenant"
@@ -352,7 +353,7 @@ func WatchRestoreJobToFinish(m *OBTenantManager) tasktypes.TaskError {
 		return false, nil
 	}
 	// Tenant restoring is in common quite a slow process, so we need to wait for a longer time
-	err = resourceutils.CheckJobWithTimeout(check, time.Second*oceanbaseconst.LocalityChangeTimeoutSeconds)
+	err = resourceutils.CheckJobWithTimeout(check, time.Second*time.Duration(obcfg.GetConfig().Time.LocalityChangeTimeoutSeconds))
 	if err != nil {
 		return errors.Wrap(err, "Failed to wait for restore job to finish")
 	}
@@ -420,7 +421,7 @@ func UpgradeTenantIfNeeded(m *OBTenantManager) tasktypes.TaskError {
 		if err != nil {
 			return err
 		}
-		maxWait5secTimes := oceanbaseconst.DefaultStateWaitTimeout/5 + 1
+		maxWait5secTimes := obcfg.GetConfig().Time.DefaultStateWaitTimeout/5 + 1
 	outer:
 		for i := 0; i < maxWait5secTimes; i++ {
 			time.Sleep(5 * time.Second)