cluster: support use different component versions

components/cluster/command/upgrade.go

@@ -14,12 +14,14 @@
 package command
 
 import (
+	"github.com/pingcap/tiup/pkg/cluster/spec"
 	"github.com/pingcap/tiup/pkg/utils"
 	"github.com/spf13/cobra"
 )
 
 func newUpgradeCmd() *cobra.Command {
 	offlineMode := false
+	var tidbVer, tikvVer, pdVer, tiflashVer, kvcdcVer, dashboardVer, cdcVer, alertmanagerVer, nodeExporterVer, blackboxExporterVer string
 
 	cmd := &cobra.Command{
 		Use:   "upgrade <cluster-name> <version>",
@@ -38,7 +40,20 @@ func newUpgradeCmd() *cobra.Command {
 			teleCommand = append(teleCommand, scrubClusterName(clusterName))
 			teleCommand = append(teleCommand, version)
 
-			return cm.Upgrade(clusterName, version, gOpt, skipConfirm, offlineMode)
+			componentVersions := map[string]string{
+				spec.ComponentDashboard:        dashboardVer,
+				spec.ComponentAlertmanager:     alertmanagerVer,
+				spec.ComponentTiDB:             tidbVer,
+				spec.ComponentTiKV:             tikvVer,
+				spec.ComponentPD:               pdVer,
+				spec.ComponentTiFlash:          tiflashVer,
+				spec.ComponentTiKVCDC:          kvcdcVer,
+				spec.ComponentCDC:              cdcVer,
+				spec.ComponentBlackboxExporter: blackboxExporterVer,
+				spec.ComponentNodeExporter:     nodeExporterVer,
+			}
+
+			return cm.Upgrade(clusterName, version, componentVersions, gOpt, skipConfirm, offlineMode)
 		},
 		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 			switch len(args) {
@@ -56,5 +71,15 @@ func newUpgradeCmd() *cobra.Command {
 	cmd.Flags().StringVar(&gOpt.SSHCustomScripts.BeforeRestartInstance.Raw, "pre-upgrade-script", "", "(EXPERIMENTAL) Custom script to be executed on each server before the server is upgraded")
 	cmd.Flags().StringVar(&gOpt.SSHCustomScripts.AfterRestartInstance.Raw, "post-upgrade-script", "", "(EXPERIMENTAL) Custom script to be executed on each server after the server is upgraded")
 
+	cmd.Flags().StringVar(&tidbVer, "tidb-version", "", "Specify the version of tidb to upgrade to")
+	cmd.Flags().StringVar(&tikvVer, "tikv-version", "", "Specify the version of tikv to upgrade to")
+	cmd.Flags().StringVar(&pdVer, "pd-version", "", "Specify the version of pd to upgrade to")
+	cmd.Flags().StringVar(&tiflashVer, "tiflash-version", "", "Specify the version of tiflash to upgrade to")
+	cmd.Flags().StringVar(&dashboardVer, "tidb-dashboard-version", "", "Specify the version of tidb-dashboard to upgrade to")
+	cmd.Flags().StringVar(&cdcVer, "cdc-version", "", "Specify the version of cdc to upgrade to")
+	cmd.Flags().StringVar(&kvcdcVer, "tikv-cdc-version", "", "Specify the cersion of tikc-cdc to upgrade to")
+	cmd.Flags().StringVar(&alertmanagerVer, "alertmanager-version", "", "Specify the version of alertmanager to upgrade to")
+	cmd.Flags().StringVar(&nodeExporterVer, "node-exporter-version", "", "Specify the version of node-exporter to upgrade to")
+	cmd.Flags().StringVar(&blackboxExporterVer, "blackbox-exporter-version", "", "Specify the version of blackbox-exporter to upgrade to")
 	return cmd
 }

components/dm/command/upgrade.go

@@ -28,7 +28,7 @@ func newUpgradeCmd() *cobra.Command {
 				return cmd.Help()
 			}
 
-			return cm.Upgrade(args[0], args[1], gOpt, skipConfirm, offlineMode)
+			return cm.Upgrade(args[0], args[1], nil, gOpt, skipConfirm, offlineMode)
 		},
 		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 			switch len(args) {

pkg/cluster/manager/builder.go

@@ -213,7 +213,7 @@ func buildScaleOutTask(
 					inst.ComponentSource(),
 					inst.OS(),
 					inst.Arch(),
-					version,
+					inst.CalculateVersion(version),
 					srcPath,
 					inst.GetManageHost(),
 					deployDir,
@@ -272,7 +272,6 @@ func buildScaleOutTask(
 		noAgentHosts,
 		topo.BaseTopo().GlobalOptions,
 		topo.BaseTopo().MonitoredOptions,
-		base.Version,
 		gOpt,
 		p,
 	)
@@ -429,7 +428,6 @@ func buildMonitoredDeployTask(
 	noAgentHosts set.StringSet, // hosts that do not deploy monitor agents
 	globalOptions *spec.GlobalOptions,
 	monitoredOptions *spec.MonitoredOptions,
-	version string,
 	gOpt operator.Options,
 	p *tui.SSHConnectionProps,
 ) (downloadCompTasks []*task.StepDisplay, deployCompTasks []*task.StepDisplay, err error) {
@@ -440,7 +438,10 @@ func buildMonitoredDeployTask(
 	uniqueCompOSArch := set.NewStringSet()
 	// monitoring agents
 	for _, comp := range []string{spec.ComponentNodeExporter, spec.ComponentBlackboxExporter} {
-		version := m.bindVersion(comp, version)
+		version := monitoredOptions.NodeExporterVersion
+		if comp == spec.ComponentBlackboxExporter {
+			version = monitoredOptions.BlackboxExporterVersion
+		}
 		for host, info := range uniqueHosts {
 			// skip deploying monitoring agents if the instance is marked so
 			if noAgentHosts.Exist(host) {
@@ -698,7 +699,7 @@ func buildDownloadCompTasks(
 				// download spark as dependency of tispark
 				tasks = append(tasks, buildDownloadSparkTask(inst, logger, gOpt))
 			} else {
-				version = bindVersion(inst.ComponentSource(), clusterVersion)
+				version = bindVersion(inst.ComponentSource(), inst.CalculateVersion(clusterVersion))
 			}
 
 			t := task.NewBuilder(logger).

pkg/cluster/manager/check.go

@@ -182,7 +182,7 @@ func checkSystemInfo(
 		downloadTasks      []*task.StepDisplay
 	)
 	logger := ctx.Value(logprinter.ContextKeyLogger).(*logprinter.Logger)
-	insightVer := spec.TiDBComponentVersion(spec.ComponentCheckCollector, "")
+	insightVer := ""
 
 	uniqueHosts := map[string]int{}             // host -> ssh-port
 	uniqueArchList := make(map[string]struct{}) // map["os-arch"]{}

pkg/cluster/manager/deploy.go

@@ -249,7 +249,8 @@ func (m *Manager) Deploy(
 
 	// Deploy components to remote
 	topo.IterInstance(func(inst spec.Instance) {
-		version := m.bindVersion(inst.ComponentSource(), clusterVersion)
+		version := inst.CalculateVersion(clusterVersion)
+		inst.SetVersion(version)
 		deployDir := spec.Abs(globalOptions.User, inst.DeployDir())
 		// data dir would be empty for components which don't need it
 		dataDirs := spec.MultiDirAbs(globalOptions.User, inst.DataDir())
@@ -320,7 +321,6 @@ func (m *Manager) Deploy(
 		noAgentHosts,
 		globalOptions,
 		topo.GetMonitoredOptions(),
-		clusterVersion,
 		gOpt,
 		sshProxyProps,
 	)

pkg/cluster/manager/upgrade.go

@@ -25,6 +25,7 @@ import (
 	perrs "github.com/pingcap/errors"
 	"github.com/pingcap/tiup/pkg/cluster/clusterutil"
 	"github.com/pingcap/tiup/pkg/cluster/ctxt"
+	"github.com/pingcap/tiup/pkg/cluster/executor"
 	operator "github.com/pingcap/tiup/pkg/cluster/operation"
 	"github.com/pingcap/tiup/pkg/cluster/spec"
 	"github.com/pingcap/tiup/pkg/cluster/task"
@@ -37,7 +38,19 @@ import (
 )
 
 // Upgrade the cluster.
-func (m *Manager) Upgrade(name string, clusterVersion string, opt operator.Options, skipConfirm, offline bool) error {
+func (m *Manager) Upgrade(name string, clusterVersion string, componentVersions map[string]string, opt operator.Options, skipConfirm, offline bool) error {
+	if !skipConfirm && strings.ToLower(opt.DisplayMode) != "json" {
+		for _, v := range componentVersions {
+			if v != "" {
+				m.logger.Warnf(color.YellowString("tiup-cluster does not provide compatibility guarantees or version checks for different component versions. Please be aware of the risks or use it with the assistance of PingCAP support."))
+				err := tui.PromptForConfirmOrAbortError("Do you want to continue? [y/N]: ")
+				if err != nil {
+					return err
+				}
+				break
+			}
+		}
+	}
 	if err := clusterutil.ValidateClusterNameOrError(name); err != nil {
 		return err
 	}
@@ -91,16 +104,12 @@ Do you want to continue? [y/N]:`,
 		for _, inst := range comp.Instances() {
 			compName := inst.ComponentName()
 
-			// ignore monitor agents for instances marked as ignore_exporter
-			switch compName {
-			case spec.ComponentNodeExporter,
-				spec.ComponentBlackboxExporter:
-				if inst.IgnoreMonitorAgent() {
-					continue
-				}
+			// if component version is not specified, use the cluster version or latest("")
+			version := componentVersions[compName]
+			if version == "" {
+				version = m.bindVersion(compName, clusterVersion)
 			}
-
-			version := m.bindVersion(inst.ComponentSource(), clusterVersion)
+			inst.SetVersion(version)
 
 			// Download component from repository
 			key := fmt.Sprintf("%s-%s-%s-%s", compName, version, inst.OS(), inst.Arch())
@@ -144,6 +153,7 @@ Do you want to continue? [y/N]:`,
 			// backup files of the old version
 			tb = tb.BackupComponent(inst.ComponentSource(), base.Version, inst.GetManageHost(), deployDir)
 
+			// this interface is not used
 			if deployerInstance, ok := inst.(DeployerInstance); ok {
 				deployerInstance.Deploy(tb, "", deployDir, version, name, clusterVersion)
 			} else {
@@ -174,7 +184,7 @@ Do you want to continue? [y/N]:`,
 
 			tb.InitConfig(
 				name,
-				clusterVersion,
+				version,
 				m.specManager,
 				inst,
 				base.User,
@@ -190,6 +200,52 @@ Do you want to continue? [y/N]:`,
 		}
 	}
 
+	var sshProxyProps *tui.SSHConnectionProps = &tui.SSHConnectionProps{}
+	if opt.SSHType != executor.SSHTypeNone {
+		var err error
+		if len(opt.SSHProxyHost) != 0 {
+			if sshProxyProps, err = tui.ReadIdentityFileOrPassword(opt.SSHProxyIdentity, opt.SSHProxyUsePassword); err != nil {
+				return err
+			}
+		}
+	}
+
+	monitoredOptions := topo.GetMonitoredOptions()
+	if componentVersions[spec.ComponentBlackboxExporter] != "" {
+		monitoredOptions.BlackboxExporterVersion = componentVersions[spec.ComponentBlackboxExporter]
+	}
+	if componentVersions[spec.ComponentNodeExporter] != "" {
+		monitoredOptions.NodeExporterVersion = componentVersions[spec.ComponentNodeExporter]
+	}
+	uniqueHosts, noAgentHosts := getMonitorHosts(topo)
+	// Deploy monitor relevant components to remote
+	dlTasks, dpTasks, err := buildMonitoredDeployTask(
+		m,
+		uniqueHosts,
+		noAgentHosts,
+		topo.BaseTopo().GlobalOptions,
+		monitoredOptions,
+		opt,
+		sshProxyProps,
+	)
+	if err != nil {
+		return err
+	}
+
+	monitorConfigTasks := buildInitMonitoredConfigTasks(
+		m.specManager,
+		name,
+		uniqueHosts,
+		noAgentHosts,
+		*topo.BaseTopo().GlobalOptions,
+		monitoredOptions,
+		m.logger,
+		opt.SSHTimeout,
+		opt.OptTimeout,
+		opt,
+		sshProxyProps,
+	)
+
 	// handle dir scheme changes
 	if hasImported {
 		if err := spec.HandleImportPathMigration(name); err != nil {
@@ -229,7 +285,10 @@ Do you want to continue? [y/N]:`,
 	}
 	t := b.
 		Parallel(false, downloadCompTasks...).
+		ParallelStep("download monitored", false, dlTasks...).
 		Parallel(opt.Force, copyCompTasks...).
+		ParallelStep("deploy monitored", false, dpTasks...).
+		ParallelStep("refresh monitored config", false, monitorConfigTasks...).
 		Func("UpgradeCluster", func(ctx context.Context) error {
 			if offline {
 				return nil
@@ -274,10 +333,10 @@ func versionCompare(curVersion, newVersion string) error {
 	}
 
 	switch semver.Compare(curVersion, newVersion) {
-	case -1:
+	case -1, 0:
 		return nil
-	case 0, 1:
-		return perrs.Errorf("please specify a higher version than %s", curVersion)
+	case 1:
+		return perrs.Errorf("please specify a higher or equle version than %s", curVersion)
 	default:
 		return perrs.Errorf("unreachable")
 	}

pkg/cluster/spec/alertmanager.go

@@ -32,6 +32,7 @@ type AlertmanagerSpec struct {
 	Host            string               `yaml:"host"`
 	ManageHost      string               `yaml:"manage_host,omitempty" validate:"manage_host:editable"`
 	SSHPort         int                  `yaml:"ssh_port,omitempty" validate:"ssh_port:editable"`
+	Version         string               `yaml:"version,omitempty"`
 	Imported        bool                 `yaml:"imported,omitempty"`
 	Patched         bool                 `yaml:"patched,omitempty"`
 	IgnoreExporter  bool                 `yaml:"ignore_exporter,omitempty"`
@@ -208,7 +209,8 @@ func (i *AlertManagerInstance) InitConfig(
 	if err := i.TransferLocalConfigFile(ctx, e, configPath, dst); err != nil {
 		return err
 	}
-	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), clusterVersion, i.OS(), i.Arch(), i.ComponentName()+".yml", paths, nil)
+	// version is not used for alertmanager
+	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), "", i.OS(), i.Arch(), i.ComponentName()+".yml", paths)
 }
 
 // ScaleConfig deploy temporary config on scaling
@@ -231,3 +233,8 @@ func (i *AlertManagerInstance) ScaleConfig(
 func (i *AlertManagerInstance) setTLSConfig(ctx context.Context, enableTLS bool, configs map[string]any, paths meta.DirPaths) (map[string]any, error) {
 	return nil, nil
 }
+
+func (i *AlertManagerInstance) CalculateVersion(_ string) string {
+	// always not follow global version, use ""(latest) by default
+	return i.InstanceSpec.(*AlertmanagerSpec).Version
+}

pkg/cluster/spec/bindversion.go

@@ -25,7 +25,6 @@ func TiDBComponentVersion(comp, version string) string {
 	case ComponentAlertmanager,
 		ComponentBlackboxExporter,
 		ComponentNodeExporter,
-		ComponentPushwaygate,
 		ComponentCheckCollector,
 		ComponentSpark,
 		ComponentTiSpark,

pkg/cluster/spec/cdc.go

@@ -36,6 +36,7 @@ type CDCSpec struct {
 	Host            string               `yaml:"host"`
 	ManageHost      string               `yaml:"manage_host,omitempty" validate:"manage_host:editable"`
 	SSHPort         int                  `yaml:"ssh_port,omitempty" validate:"ssh_port:editable"`
+	Version         string               `yaml:"version,omitempty"`
 	Imported        bool                 `yaml:"imported,omitempty"`
 	Patched         bool                 `yaml:"patched,omitempty"`
 	IgnoreExporter  bool                 `yaml:"ignore_exporter,omitempty"`
@@ -191,14 +192,15 @@ func (i *CDCInstance) InitConfig(
 	spec := i.InstanceSpec.(*CDCSpec)
 	globalConfig := topo.ServerConfigs.CDC
 	instanceConfig := spec.Config
+	version := i.CalculateVersion(clusterVersion)
 
-	if !tidbver.TiCDCSupportConfigFile(clusterVersion) {
+	if !tidbver.TiCDCSupportConfigFile(version) {
 		if len(globalConfig)+len(instanceConfig) > 0 {
 			return errors.New("server_config is only supported with TiCDC version v4.0.13 or later")
 		}
 	}
 
-	if !tidbver.TiCDCSupportClusterID(clusterVersion) && spec.TiCDCClusterID != "" {
+	if !tidbver.TiCDCSupportClusterID(version) && spec.TiCDCClusterID != "" {
 		return errors.New("ticdc_cluster_id is only supported with TiCDC version v6.2.0 or later")
 	}
 
@@ -213,13 +215,13 @@ func (i *CDCInstance) InitConfig(
 		GCTTL:             spec.GCTTL,
 		TZ:                spec.TZ,
 		ClusterID:         spec.TiCDCClusterID,
-		DataDirEnabled:    tidbver.TiCDCSupportDataDir(clusterVersion),
-		ConfigFileEnabled: tidbver.TiCDCSupportConfigFile(clusterVersion),
+		DataDirEnabled:    tidbver.TiCDCSupportDataDir(version),
+		ConfigFileEnabled: tidbver.TiCDCSupportConfigFile(version),
 		TLSEnabled:        enableTLS,
 
 		DeployDir: paths.Deploy,
 		LogDir:    paths.Log,
-		DataDir:   utils.Ternary(tidbver.TiCDCSupportSortOrDataDir(clusterVersion), spec.DataDir, "").(string),
+		DataDir:   utils.Ternary(tidbver.TiCDCSupportSortOrDataDir(version), spec.DataDir, "").(string),
 
 		NumaNode: spec.NumaNode,
 	}

pkg/cluster/spec/dashboard.go

@@ -224,7 +224,7 @@ func (i *DashboardInstance) InitConfig(
 		return err
 	}
 
-	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), clusterVersion, i.OS(), i.Arch(), i.ComponentName()+".toml", paths, nil)
+	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), i.CalculateVersion(clusterVersion), i.OS(), i.Arch(), i.ComponentName()+".toml", paths)
 }
 
 // setTLSConfig set TLS Config to support enable/disable TLS

pkg/cluster/spec/drainer.go

@@ -33,6 +33,7 @@ import (
 type DrainerSpec struct {
 	Host            string               `yaml:"host"`
 	ManageHost      string               `yaml:"manage_host,omitempty" validate:"manage_host:editable"`
+	Version         string               `yaml:"version,omitempty"`
 	SSHPort         int                  `yaml:"ssh_port,omitempty" validate:"ssh_port:editable"`
 	Imported        bool                 `yaml:"imported,omitempty"`
 	Patched         bool                 `yaml:"patched,omitempty"`
@@ -274,7 +275,7 @@ func (i *DrainerInstance) InitConfig(
 		return err
 	}
 
-	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), clusterVersion, i.OS(), i.Arch(), i.ComponentName()+".toml", paths, nil)
+	return checkConfig(ctx, e, i.ComponentName(), i.ComponentSource(), clusterVersion, i.OS(), i.Arch(), i.ComponentName()+".toml", paths)
 }
 
 // setTLSConfig set TLS Config to support enable/disable TLS

pkg/cluster/spec/grafana.go

@@ -37,6 +37,7 @@ type GrafanaSpec struct {
 	Host            string               `yaml:"host"`
 	ManageHost      string               `yaml:"manage_host,omitempty" validate:"manage_host:editable"`
 	SSHPort         int                  `yaml:"ssh_port,omitempty" validate:"ssh_port:editable"`
+	Version         string               `yaml:"version,omitempty"`
 	Imported        bool                 `yaml:"imported,omitempty"`
 	Patched         bool                 `yaml:"patched,omitempty"`
 	IgnoreExporter  bool                 `yaml:"ignore_exporter,omitempty"`