cdc: add newarch check for upgrade

components/dm/spec/logic.go

@@ -138,6 +138,7 @@ func (i *MasterInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt spec.InstanceOpt,
 ) error {
 	if err := i.BaseInstance.InitConfig(ctx, e, i.topo.GlobalOptions, deployUser, paths); err != nil {
 		return err
@@ -234,8 +235,9 @@ func (i *MasterInstance) ScaleConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt spec.InstanceOpt,
 ) error {
-	if err := i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths); err != nil {
+	if err := i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt); err != nil {
 		return err
 	}
 
@@ -361,6 +363,7 @@ func (i *WorkerInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt spec.InstanceOpt,
 ) error {
 	if err := i.BaseInstance.InitConfig(ctx, e, i.topo.GlobalOptions, deployUser, paths); err != nil {
 		return err
@@ -455,13 +458,14 @@ func (i *WorkerInstance) ScaleConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt spec.InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() {
 		i.topo = s
 	}()
 	i.topo = topo.(*Specification)
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt)
 }
 
 // GetGlobalOptions returns cluster topology

pkg/cluster/manager/builder.go

@@ -689,6 +689,7 @@ func buildInitConfigTasks(
 					Log:    logDir,
 					Cache:  m.specManager.Path(name, spec.TempConfigPath),
 				},
+				spec.InstanceOpt{},
 			).
 			BuildAsStep(fmt.Sprintf("  - Generate config %s -> %s", compName, instance.ID()))
 		tasks = append(tasks, t)

pkg/cluster/manager/manager_test.go

@@ -23,19 +23,17 @@ import (
 )
 
 func TestVersionCompare(t *testing.T) {
-	var err error
+	res := versionCompare("v4.0.0", "v4.0.1")
+	assert.Equal(t, res, -1)
 
-	err = versionCompare("v4.0.0", "v4.0.1")
-	assert.Nil(t, err)
+	res = versionCompare("v4.0.1", "v4.0.0")
+	assert.Equal(t, res, 1)
 
-	err = versionCompare("v4.0.1", "v4.0.0")
-	assert.NotNil(t, err)
+	res = versionCompare("v4.0.0", "nightly")
+	assert.Equal(t, res, -1)
 
-	err = versionCompare("v4.0.0", "nightly")
-	assert.Nil(t, err)
-
-	err = versionCompare("nightly", "nightly")
-	assert.Nil(t, err)
+	res = versionCompare("nightly", "nightly")
+	assert.Equal(t, res, 0)
 }
 
 func TestValidateNewTopo(t *testing.T) {

pkg/cluster/manager/upgrade.go

@@ -38,6 +38,10 @@ import (
 	"golang.org/x/mod/semver"
 )
 
+const (
+	cdcNewArchVersion = "v9.0.0"
+)
+
 func (m *Manager) upgradePrecheck(name string, componentVersions map[string]string, opt operator.Options, skipConfirm bool) error {
 	if !skipConfirm && strings.ToLower(opt.DisplayMode) != "json" {
 		for _, v := range componentVersions {
@@ -86,9 +90,9 @@ func (m *Manager) Upgrade(name string, clusterVersion string, componentVersions
 		uniqueComps = map[string]struct{}{}
 	)
 
-	if err := versionCompare(base.Version, clusterVersion); err != nil {
+	if versionCompare(base.Version, clusterVersion) == 1 {
 		if !ignoreVersionCheck {
-			return err
+			return perrs.Errorf("please specify a higher or equle version than %s", base.Version)
 		}
 		m.logger.Warnf("%s", color.RedString("There is no guarantee that the cluster can be downgraded. Be careful before you continue."))
 	}
@@ -143,8 +147,8 @@ This operation will upgrade %s %s cluster %s (with a concurrency of %d) to %s:%s
 
 	hasImported := false
 	for _, comp := range components {
+		oldver := comp.CalculateVersion(base.Version)
 		version := comp.CalculateVersion(clusterVersion)
-
 		for _, inst := range comp.Instances() {
 			// Download component from repository
 			key := fmt.Sprintf("%s-%s-%s-%s", inst.ComponentSource(), version, inst.OS(), inst.Arch())
@@ -217,6 +221,10 @@ This operation will upgrade %s %s cluster %s (with a concurrency of %d) to %s:%s
 				}
 			}
 
+			checkCDCNewArch := false
+			if versionCompare(oldver, cdcNewArchVersion) == -1 && versionCompare(cdcNewArchVersion, clusterVersion) > -1 {
+				checkCDCNewArch = true
+			}
 			tb.InitConfig(
 				name,
 				clusterVersion,
@@ -230,6 +238,7 @@ This operation will upgrade %s %s cluster %s (with a concurrency of %d) to %s:%s
 					Log:    logDir,
 					Cache:  m.specManager.Path(name, spec.TempConfigPath),
 				},
+				spec.InstanceOpt{CheckCDCNewArch: checkCDCNewArch},
 			)
 			copyCompTasks = append(copyCompTasks, tb.Build())
 		}
@@ -372,18 +381,16 @@ This operation will upgrade %s %s cluster %s (with a concurrency of %d) to %s:%s
 	return nil
 }
 
-func versionCompare(curVersion, newVersion string) error {
+// versionCompare returns an integer comparing two versions according to semantic version precedence.
+// The result will be 0 if curVersion == newVersion, -1 if curVersion < newVersion, or +1 if curVersion > newVersion.
+func versionCompare(curVersion, newVersion string) int {
+	if newVersion == curVersion {
+		return 0
+	}
 	// Can always upgrade to 'nightly' event the current version is 'nightly'
 	if newVersion == utils.NightlyVersionAlias {
-		return nil
+		return -1
 	}
 
-	switch semver.Compare(curVersion, newVersion) {
-	case -1, 0:
-		return nil
-	case 1:
-		return perrs.Errorf("please specify a higher or equle version than %s", curVersion)
-	default:
-		return perrs.Errorf("unreachable")
-	}
+	return semver.Compare(curVersion, newVersion)
 }

pkg/cluster/spec/alertmanager.go

@@ -175,6 +175,7 @@ func (i *AlertManagerInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	gOpts := *i.topo.BaseTopo().GlobalOptions
 	if err := i.BaseInstance.InitConfig(ctx, e, gOpts, deployUser, paths); err != nil {
@@ -247,11 +248,12 @@ func (i *AlertManagerInstance) ScaleConfig(
 	clusterVersion string,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() { i.topo = s }()
 	i.topo = topo
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt)
 }
 
 // setTLSConfig set TLS Config to support enable/disable TLS

pkg/cluster/spec/cdc.go

@@ -182,14 +182,15 @@ func (i *CDCInstance) ScaleConfig(
 	clusterVersion,
 	user string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() {
 		i.topo = s
 	}()
 	i.topo = mustBeClusterTopo(topo)
 
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths, opt)
 }
 
 // InitConfig implements Instance interface.
@@ -200,6 +201,7 @@ func (i *CDCInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	topo := i.topo.(*Specification)
 	if err := i.BaseInstance.InitConfig(ctx, e, topo.GlobalOptions, deployUser, paths); err != nil {

pkg/cluster/spec/dashboard.go

@@ -178,14 +178,15 @@ func (i *DashboardInstance) ScaleConfig(
 	clusterVersion,
 	user string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() {
 		i.topo = s
 	}()
 	i.topo = mustBeClusterTopo(topo)
 
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths, opt)
 }
 
 // InitConfig implements Instance interface.
@@ -196,6 +197,7 @@ func (i *DashboardInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	topo := i.topo.(*Specification)
 	if err := i.BaseInstance.InitConfig(ctx, e, topo.GlobalOptions, deployUser, paths); err != nil {

pkg/cluster/spec/drainer.go

@@ -197,14 +197,15 @@ func (i *DrainerInstance) ScaleConfig(
 	clusterVersion,
 	user string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() {
 		i.topo = s
 	}()
 	i.topo = mustBeClusterTopo(topo)
 
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, user, paths, opt)
 }
 
 // InitConfig implements Instance interface.
@@ -215,6 +216,7 @@ func (i *DrainerInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	topo := i.topo.(*Specification)
 	if err := i.BaseInstance.InitConfig(ctx, e, topo.GlobalOptions, deployUser, paths); err != nil {

pkg/cluster/spec/grafana.go

@@ -185,6 +185,7 @@ func (i *GrafanaInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	gOpts := *i.topo.BaseTopo().GlobalOptions
 	if err := i.BaseInstance.InitConfig(ctx, e, gOpts, deployUser, paths); err != nil {
@@ -453,11 +454,12 @@ func (i *GrafanaInstance) ScaleConfig(
 	clusterVersion string,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() { i.topo = s }()
 	i.topo = topo.Merge(i.topo)
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt)
 }
 
 func mergeAdditionalGrafanaConf(source string, addition map[string]string) error {

pkg/cluster/spec/grafana_test.go

@@ -225,7 +225,7 @@ func TestGrafanaDatasourceConfig(t *testing.T) {
 
 	// Test datasource configuration
 	clusterName := "test-cluster"
-	err := grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths)
+	err := grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths, InstanceOpt{})
 	require.NoError(t, err)
 
 	// Verify the datasource configuration file
@@ -247,7 +247,7 @@ func TestGrafanaDatasourceConfig(t *testing.T) {
 
 	// Test without VM remote write enabled
 	topo.Monitors[0].PromRemoteWriteToVM = false
-	err = grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths)
+	err = grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), mockExec, clusterName, "v5.4.0", "tidb", paths, InstanceOpt{})
 	require.NoError(t, err)
 
 	// Verify the datasource configuration file again
@@ -400,7 +400,7 @@ func TestVictoriaMetricsDefaultDatasource(t *testing.T) {
 	grafanaInstance := comp.Instances()[0].(*GrafanaInstance)
 
 	// Run InitConfig which will process dashboards
-	err = grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), origExecutor, "test-cluster", "v5.4.0", "tidb", paths)
+	err = grafanaInstance.InitConfig(ctxt.New(ctx, 0, logprinter.NewLogger("")), origExecutor, "test-cluster", "v5.4.0", "tidb", paths, InstanceOpt{})
 	require.NoError(t, err)
 
 	// Check if the dashboard file was created and datasource references were updated

pkg/cluster/spec/instance.go

@@ -97,8 +97,8 @@ type Instance interface {
 	InstanceSpec
 	ID() string
 	Ready(context.Context, ctxt.Executor, uint64, *tls.Config) error
-	InitConfig(ctx context.Context, e ctxt.Executor, clusterName string, clusterVersion string, deployUser string, paths meta.DirPaths) error
-	ScaleConfig(ctx context.Context, e ctxt.Executor, topo Topology, clusterName string, clusterVersion string, deployUser string, paths meta.DirPaths) error
+	InitConfig(ctx context.Context, e ctxt.Executor, clusterName string, clusterVersion string, deployUser string, paths meta.DirPaths, opt InstanceOpt) error
+	ScaleConfig(ctx context.Context, e ctxt.Executor, topo Topology, clusterName string, clusterVersion string, deployUser string, paths meta.DirPaths, opt InstanceOpt) error
 	PrepareStart(ctx context.Context, tlsCfg *tls.Config) error
 	ComponentName() string
 	ComponentSource() string
@@ -149,6 +149,11 @@ func PortStopped(ctx context.Context, e ctxt.Executor, port int, timeout uint64)
 	return w.Execute(ctx, e)
 }
 
+// InstanceOpt can be used to store options when initializing the configuration.
+type InstanceOpt struct {
+	CheckCDCNewArch bool
+}
+
 // BaseInstance implements some method of Instance interface..
 type BaseInstance struct {
 	InstanceSpec
@@ -168,7 +173,8 @@ type BaseInstance struct {
 	StatusFn func(ctx context.Context, timeout time.Duration, tlsCfg *tls.Config, pdHosts ...string) string
 	UptimeFn func(ctx context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration
 
-	Component Component
+	Component   Component
+	InstanceOpt InstanceOpt
 }
 
 // Ready implements Instance interface

pkg/cluster/spec/monitoring.go

@@ -263,6 +263,7 @@ func (i *MonitorInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	gOpts := *i.topo.BaseTopo().GlobalOptions
 	if err := i.BaseInstance.InitConfig(ctx, e, gOpts, deployUser, paths); err != nil {
@@ -651,11 +652,12 @@ func (i *MonitorInstance) ScaleConfig(
 	clusterVersion string,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	s := i.topo
 	defer func() { i.topo = s }()
 	i.topo = topo
-	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths)
+	return i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt)
 }
 
 func mergeAdditionalScrapeConf(source string, addition map[string]any) error {

pkg/cluster/spec/pd.go

@@ -230,6 +230,7 @@ func (i *PDInstance) InitConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	topo := i.topo.(*Specification)
 	if err := i.BaseInstance.InitConfig(ctx, e, topo.GlobalOptions, deployUser, paths); err != nil {
@@ -355,9 +356,10 @@ func (i *PDInstance) ScaleConfig(
 	clusterVersion,
 	deployUser string,
 	paths meta.DirPaths,
+	opt InstanceOpt,
 ) error {
 	// We need pd.toml here, but we don't need to check it
-	if err := i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths); err != nil &&
+	if err := i.InitConfig(ctx, e, clusterName, clusterVersion, deployUser, paths, opt); err != nil &&
 		errors.Cause(err) != ErrorCheckConfig {
 		return err
 	}