use version per component instead

pingcap · Sep 17, 2023 · 9e0a52f · 9e0a52f
1 parent e899ab5
commit 9e0a52f
Show file tree

Hide file tree

Showing 17 changed files with 293 additions and 80 deletions.
diff --git a/components/dm/spec/logic.go b/components/dm/spec/logic.go
@@ -68,6 +68,16 @@ func (c *DMMasterComponent) Role() string {
 	return ComponentDMMaster
 }
 
+// CalculateVersion implements the Component interface
+func (c *DMMasterComponent) CalculateVersion(clusterVersion string) string {
+	return clusterVersion
+}
+
+// SetVersion implements Component interface.
+func (c *DMMasterComponent) SetVersion(version string) {
+	// not supported now
+}
+
 // Instances implements Component interface.
 func (c *DMMasterComponent) Instances() []Instance {
 	ins := make([]Instance, 0)
@@ -271,6 +281,16 @@ func (c *DMWorkerComponent) Role() string {
 	return ComponentDMWorker
 }
 
+// CalculateVersion implements the Component interface
+func (c *DMWorkerComponent) CalculateVersion(clusterVersion string) string {
+	return clusterVersion
+}
+
+// SetVersion implements Component interface.
+func (c *DMWorkerComponent) SetVersion(version string) {
+	// not supported now
+}
+
 // Instances implements Component interface.
 func (c *DMWorkerComponent) Instances() []Instance {
 	ins := make([]Instance, 0)

diff --git a/pkg/cluster/manager/upgrade.go b/pkg/cluster/manager/upgrade.go
@@ -101,16 +101,16 @@ Do you want to continue? [y/N]:`,
 
 	hasImported := false
 	for _, comp := range topo.ComponentsByUpdateOrder(base.Version) {
-		for _, inst := range comp.Instances() {
-			compName := inst.ComponentName()
+		compName := comp.Name()
 
-			// if component version is not specified, use the cluster version or latest("")
-			version := componentVersions[compName]
-			if version != "" {
-				inst.SetVersion(version)
-			}
-			version = inst.CalculateVersion(clusterVersion)
+		// if component version is not specified, use the cluster version or latest("")
+		version := componentVersions[compName]
+		if version != "" {
+			comp.SetVersion(version)
+		}
+		version = comp.CalculateVersion(clusterVersion)
 
+		for _, inst := range comp.Instances() {
 			// Download component from repository
 			key := fmt.Sprintf("%s-%s-%s-%s", compName, version, inst.OS(), inst.Arch())
 			if _, found := uniqueComps[key]; !found {

diff --git a/pkg/cluster/spec/alertmanager.go b/pkg/cluster/spec/alertmanager.go
@@ -32,7 +32,6 @@ 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"`
@@ -99,6 +98,21 @@ func (c *AlertManagerComponent) Role() string {
 	return RoleMonitor
 }
 
+// CalculateVersion implements the Component interface
+func (c *AlertManagerComponent) CalculateVersion(_ string) string {
+	// always not follow cluster version, use ""(latest) by default
+	version := c.Topology.BaseTopo().AlertManagerVersion
+	if version != nil {
+		return *version
+	}
+	return ""
+}
+
+// SetVersion implements Component interface.
+func (c *AlertManagerComponent) SetVersion(version string) {
+	*c.Topology.BaseTopo().AlertManagerVersion = version
+}
+
 // Instances implements Component interface.
 func (c *AlertManagerComponent) Instances() []Instance {
 	alertmanagers := c.Topology.BaseTopo().Alertmanagers
@@ -131,6 +145,7 @@ func (c *AlertManagerComponent) Instances() []Instance {
 				UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 					return UptimeByHost(s.GetManageHost(), s.WebPort, timeout, tlsCfg)
 				},
+				Component: c,
 			},
 			topo: c.Topology,
 		})
@@ -233,9 +248,3 @@ 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
 }
-
-// CalculateVersion implements the Instance interface
-func (i *AlertManagerInstance) CalculateVersion(_ string) string {
-	// always not follow global version, use ""(latest) by default
-	return i.InstanceSpec.(*AlertmanagerSpec).Version
-}
diff --git a/pkg/cluster/spec/cdc.go b/pkg/cluster/spec/cdc.go
@@ -36,7 +36,6 @@ 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"`
@@ -114,6 +113,20 @@ func (c *CDCComponent) Role() string {
 	return ComponentCDC
 }
 
+// CalculateVersion implements the Component interface
+func (c *CDCComponent) CalculateVersion(clusterVersion string) string {
+	version := c.Topology.ComponentVersions.CDC
+	if version == "" {
+		version = clusterVersion
+	}
+	return version
+}
+
+// SetVersion implements Component interface.
+func (c *CDCComponent) SetVersion(version string) {
+	c.Topology.ComponentVersions.CDC = version
+}
+
 // Instances implements Component interface.
 func (c *CDCComponent) Instances() []Instance {
 	ins := make([]Instance, 0, len(c.Topology.CDCServers))
@@ -140,6 +153,7 @@ func (c *CDCComponent) Instances() []Instance {
 			UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 				return UptimeByHost(s.GetManageHost(), s.Port, timeout, tlsCfg)
 			},
+			Component: c,
 		}, c.Topology}
 		if s.DataDir != "" {
 			instance.Dirs = append(instance.Dirs, s.DataDir)

diff --git a/pkg/cluster/spec/dashboard.go b/pkg/cluster/spec/dashboard.go
@@ -31,7 +31,6 @@ type DashboardSpec 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"`
 	Patched         bool                 `yaml:"patched,omitempty"`
 	IgnoreExporter  bool                 `yaml:"ignore_exporter,omitempty"`
 	Port            int                  `yaml:"port" default:"12333"`
@@ -116,6 +115,20 @@ func (c *DashboardComponent) Role() string {
 	return ComponentDashboard
 }
 
+// CalculateVersion implements the Component interface
+func (c *DashboardComponent) CalculateVersion(clusterVersion string) string {
+	version := c.Topology.ComponentVersions.Dashboard
+	if version == "" {
+		version = clusterVersion
+	}
+	return version
+}
+
+// SetVersion implements Component interface.
+func (c *DashboardComponent) SetVersion(version string) {
+	c.Topology.ComponentVersions.Dashboard = version
+}
+
 // Instances implements Component interface.
 func (c *DashboardComponent) Instances() []Instance {
 	ins := make([]Instance, 0, len(c.Topology.Drainers))
@@ -141,6 +154,7 @@ func (c *DashboardComponent) Instances() []Instance {
 			UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 				return UptimeByHost(s.GetManageHost(), s.Port, timeout, tlsCfg)
 			},
+			Component: c,
 		}, c.Topology})
 	}
 	return ins

diff --git a/pkg/cluster/spec/drainer.go b/pkg/cluster/spec/drainer.go
@@ -33,7 +33,6 @@ 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"`
@@ -135,6 +134,20 @@ func (c *DrainerComponent) Role() string {
 	return ComponentDrainer
 }
 
+// CalculateVersion implements the Component interface
+func (c *DrainerComponent) CalculateVersion(clusterVersion string) string {
+	version := c.Topology.ComponentVersions.Drainer
+	if version == "" {
+		version = clusterVersion
+	}
+	return version
+}
+
+// SetVersion implements Component interface.
+func (c *DrainerComponent) SetVersion(version string) {
+	c.Topology.ComponentVersions.Drainer = version
+}
+
 // Instances implements Component interface.
 func (c *DrainerComponent) Instances() []Instance {
 	ins := make([]Instance, 0, len(c.Topology.Drainers))
@@ -160,6 +173,7 @@ func (c *DrainerComponent) Instances() []Instance {
 			UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 				return UptimeByHost(s.GetManageHost(), s.Port, timeout, tlsCfg)
 			},
+			Component: c,
 		}, c.Topology})
 	}
 	return ins

diff --git a/pkg/cluster/spec/grafana.go b/pkg/cluster/spec/grafana.go
@@ -37,7 +37,6 @@ 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"`
@@ -108,6 +107,21 @@ func (c *GrafanaComponent) Role() string {
 	return RoleMonitor
 }
 
+// CalculateVersion implements the Component interface
+func (c *GrafanaComponent) CalculateVersion(_ string) string {
+	// always not follow cluster version, use ""(latest) by default
+	version := c.Topology.BaseTopo().GrafanaVersion
+	if version != nil {
+		return *version
+	}
+	return ""
+}
+
+// SetVersion implements Component interface.
+func (c *GrafanaComponent) SetVersion(version string) {
+	*c.Topology.BaseTopo().GrafanaVersion = version
+}
+
 // Instances implements Component interface.
 func (c *GrafanaComponent) Instances() []Instance {
 	servers := c.BaseTopo().Grafanas
@@ -136,6 +150,7 @@ func (c *GrafanaComponent) Instances() []Instance {
 				UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 					return UptimeByHost(s.GetManageHost(), s.Port, timeout, tlsCfg)
 				},
+				Component: c,
 			},
 			topo: c.Topology,
 		})

diff --git a/pkg/cluster/spec/instance.go b/pkg/cluster/spec/instance.go
@@ -71,6 +71,8 @@ type Component interface {
 	Name() string
 	Role() string
 	Instances() []Instance
+	CalculateVersion(string) string
+	SetVersion(string)
 }
 
 // RollingUpdateInstance represent a instance need to transfer state when restart.
@@ -109,7 +111,7 @@ type Instance interface {
 	IsPatched() bool
 	SetPatched(bool)
 	CalculateVersion(string) string
-	SetVersion(string)
+	// SetVersion(string)
 	setTLSConfig(ctx context.Context, enableTLS bool, configs map[string]any, paths meta.DirPaths) (map[string]any, error)
 }
 
@@ -151,6 +153,8 @@ type BaseInstance struct {
 	Dirs     []string
 	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
 }
 
 // Ready implements Instance interface
@@ -448,23 +452,7 @@ func (i *BaseInstance) SetPatched(p bool) {
 
 // CalculateVersion implements the Instance interface
 func (i *BaseInstance) CalculateVersion(globalVersion string) string {
-	v := reflect.Indirect(reflect.ValueOf(i.InstanceSpec)).FieldByName("Version")
-	if !v.IsValid() {
-		return globalVersion
-	}
-	if v.String() == "" {
-		return globalVersion
-	}
-	return v.String()
-}
-
-// SetVersion implements the Instance interface
-func (i *BaseInstance) SetVersion(version string) {
-	v := reflect.Indirect(reflect.ValueOf(i.InstanceSpec)).FieldByName("Version")
-	if !v.CanSet() {
-		return
-	}
-	v.SetString(version)
+	return i.Component.CalculateVersion(globalVersion)
 }
 
 // PrepareStart checks instance requirements before starting

diff --git a/pkg/cluster/spec/monitoring.go b/pkg/cluster/spec/monitoring.go
@@ -40,7 +40,6 @@ type PrometheusSpec 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"`
@@ -125,6 +124,21 @@ func (c *MonitorComponent) Role() string {
 	return RoleMonitor
 }
 
+// CalculateVersion implements the Component interface
+func (c *MonitorComponent) CalculateVersion(_ string) string {
+	// always not follow cluster version, use ""(latest) by default
+	version := c.Topology.BaseTopo().PrometheusVersion
+	if version != nil {
+		return *version
+	}
+	return ""
+}
+
+// SetVersion implements Component interface.
+func (c *MonitorComponent) SetVersion(version string) {
+	*c.Topology.BaseTopo().PrometheusVersion = version
+}
+
 // Instances implements Component interface.
 func (c *MonitorComponent) Instances() []Instance {
 	servers := c.BaseTopo().Monitors
@@ -153,6 +167,7 @@ func (c *MonitorComponent) Instances() []Instance {
 			UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 				return UptimeByHost(s.GetManageHost(), s.Port, timeout, tlsCfg)
 			},
+			Component: c,
 		}, c.Topology}
 		if s.NgPort > 0 {
 			mi.BaseInstance.Ports = append(mi.BaseInstance.Ports, s.NgPort)

diff --git a/pkg/cluster/spec/pd.go b/pkg/cluster/spec/pd.go
@@ -37,7 +37,6 @@ type PDSpec struct {
 	ListenHost          string `yaml:"listen_host,omitempty"`
 	AdvertiseClientAddr string `yaml:"advertise_client_addr,omitempty"`
 	AdvertisePeerAddr   string `yaml:"advertise_peer_addr,omitempty"`
-	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"`
@@ -160,6 +159,20 @@ func (c *PDComponent) Role() string {
 	return ComponentPD
 }
 
+// CalculateVersion implements the Component interface
+func (c *PDComponent) CalculateVersion(clusterVersion string) string {
+	version := c.Topology.ComponentVersions.PD
+	if version != "" {
+		version = clusterVersion
+	}
+	return version
+}
+
+// SetVersion implements Component interface.
+func (c *PDComponent) SetVersion(version string) {
+	c.Topology.ComponentVersions.PD = version
+}
+
 // Instances implements Component interface.
 func (c *PDComponent) Instances() []Instance {
 	ins := make([]Instance, 0, len(c.Topology.PDServers))
@@ -189,6 +202,7 @@ func (c *PDComponent) Instances() []Instance {
 				UptimeFn: func(_ context.Context, timeout time.Duration, tlsCfg *tls.Config) time.Duration {
 					return UptimeByHost(s.GetManageHost(), s.ClientPort, timeout, tlsCfg)
 				},
+				Component: c,
 			},
 			topo: c.Topology,
 		})