add global component_sources (#2320)

components/dm/spec/logic.go

@@ -68,6 +68,15 @@ func (c *DMMasterComponent) Role() string {
 	return ComponentDMMaster
 }
 
+// Source implements Component interface.
+func (c *DMMasterComponent) Source() string {
+	source := c.Topology.ComponentSources.Master
+	if source != "" {
+		return source
+	}
+	return ComponentDMMaster
+}
+
 // CalculateVersion implements the Component interface
 func (c *DMMasterComponent) CalculateVersion(clusterVersion string) string {
 	return clusterVersion
@@ -93,7 +102,7 @@ func (c *DMMasterComponent) Instances() []Instance {
 				ListenHost:   c.Topology.BaseTopo().GlobalOptions.ListenHost,
 				Port:         s.Port,
 				SSHP:         s.SSHPort,
-				Source:       s.GetSource(),
+				Source:       s.Source,
 
 				Ports: []int{
 					s.Port,
@@ -283,6 +292,15 @@ func (c *DMWorkerComponent) Role() string {
 	return ComponentDMWorker
 }
 
+// Source implements Component interface.
+func (c *DMWorkerComponent) Source() string {
+	source := c.Topology.ComponentSources.Worker
+	if source != "" {
+		return source
+	}
+	return ComponentDMWorker
+}
+
 // CalculateVersion implements the Component interface
 func (c *DMWorkerComponent) CalculateVersion(clusterVersion string) string {
 	return clusterVersion
@@ -308,7 +326,7 @@ func (c *DMWorkerComponent) Instances() []Instance {
 				ListenHost:   c.Topology.BaseTopo().GlobalOptions.ListenHost,
 				Port:         s.Port,
 				SSHP:         s.SSHPort,
-				Source:       s.GetSource(),
+				Source:       s.Source,
 
 				Ports: []int{
 					s.Port,

components/dm/spec/topology_dm.go

@@ -36,9 +36,10 @@ const (
 )
 
 var (
-	globalOptionTypeName  = reflect.TypeOf(GlobalOptions{}).Name()
-	monitorOptionTypeName = reflect.TypeOf(MonitoredOptions{}).Name()
-	serverConfigsTypeName = reflect.TypeOf(DMServerConfigs{}).Name()
+	globalOptionTypeName     = reflect.TypeOf(GlobalOptions{}).Name()
+	monitorOptionTypeName    = reflect.TypeOf(MonitoredOptions{}).Name()
+	serverConfigsTypeName    = reflect.TypeOf(DMServerConfigs{}).Name()
+	componentSourcesTypeName = reflect.TypeOf(ComponentSources{}).Name()
 )
 
 func setDefaultDir(parent, role, port string, field reflect.Value) {
@@ -69,7 +70,7 @@ func isSkipField(field reflect.Value) bool {
 		field = field.Elem()
 	}
 	tp := field.Type().Name()
-	return tp == globalOptionTypeName || tp == monitorOptionTypeName || tp == serverConfigsTypeName
+	return tp == globalOptionTypeName || tp == monitorOptionTypeName || tp == serverConfigsTypeName || tp == componentSourcesTypeName
 }
 
 type (
@@ -95,10 +96,17 @@ type (
 		Grafana map[string]string `yaml:"grafana"`
 	}
 
+	// ComponentSources represents the source of components
+	ComponentSources struct {
+		Master string `yaml:"master,omitempty"`
+		Worker string `yaml:"worker,omitempty"`
+	}
+
 	// Specification represents the specification of topology.yaml
 	Specification struct {
 		GlobalOptions    GlobalOptions            `yaml:"global,omitempty" validate:"global:editable"`
 		MonitoredOptions *MonitoredOptions        `yaml:"monitored,omitempty" validate:"monitored:editable"`
+		ComponentSources ComponentSources         `yaml:"component_sources,omitempty" validate:"component_sources:editable"`
 		ServerConfigs    DMServerConfigs          `yaml:"server_configs,omitempty" validate:"server_configs:ignore"`
 		Masters          []*MasterSpec            `yaml:"master_servers"`
 		Workers          []*WorkerSpec            `yaml:"worker_servers"`
@@ -203,14 +211,6 @@ func (s *MasterSpec) GetAdvertisePeerURL(enableTLS bool) string {
 	return fmt.Sprintf("%s://%s", scheme, utils.JoinHostPort(s.Host, s.PeerPort))
 }
 
-// GetSource returns source to download the component
-func (s *MasterSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentDMMaster
-	}
-	return s.Source
-}
-
 // WorkerSpec represents the Master topology specification in topology.yaml
 type WorkerSpec struct {
 	Host           string `yaml:"host"`
@@ -282,14 +282,6 @@ func (s *WorkerSpec) IgnoreMonitorAgent() bool {
 	return s.IgnoreExporter
 }
 
-// GetSource returns source to download the component
-func (s *WorkerSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentDMWorker
-	}
-	return s.Source
-}
-
 // UnmarshalYAML sets default values when unmarshaling the topology file
 func (s *Specification) UnmarshalYAML(unmarshal func(any) error) error {
 	type topology Specification

pkg/cluster/spec/alertmanager.go

@@ -98,6 +98,11 @@ func (c *AlertManagerComponent) Role() string {
 	return RoleMonitor
 }
 
+// Source implements Component interface.
+func (c *AlertManagerComponent) Source() string {
+	return ComponentAlertmanager
+}
+
 // CalculateVersion implements the Component interface
 func (c *AlertManagerComponent) CalculateVersion(_ string) string {
 	// always not follow cluster version, use ""(latest) by default

pkg/cluster/spec/cdc.go

@@ -92,14 +92,6 @@ func (s *CDCSpec) IgnoreMonitorAgent() bool {
 	return s.IgnoreExporter
 }
 
-// GetSource returns source to download the component
-func (s *CDCSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentCDC
-	}
-	return s.Source
-}
-
 // CDCComponent represents CDC component.
 type CDCComponent struct{ Topology *Specification }
 
@@ -113,6 +105,15 @@ func (c *CDCComponent) Role() string {
 	return ComponentCDC
 }
 
+// Source implements Component interface.
+func (c *CDCComponent) Source() string {
+	source := c.Topology.ComponentSources.CDC
+	if source != "" {
+		return source
+	}
+	return ComponentCDC
+}
+
 // CalculateVersion implements the Component interface
 func (c *CDCComponent) CalculateVersion(clusterVersion string) string {
 	version := c.Topology.ComponentVersions.CDC
@@ -140,7 +141,7 @@ func (c *CDCComponent) Instances() []Instance {
 			ListenHost:   c.Topology.BaseTopo().GlobalOptions.ListenHost,
 			Port:         s.Port,
 			SSHP:         s.SSHPort,
-			Source:       s.GetSource(),
+			Source:       s.Source,
 			NumaNode:     s.NumaNode,
 			NumaCores:    "",
 

pkg/cluster/spec/dashboard.go

@@ -94,14 +94,6 @@ func (s *DashboardSpec) IgnoreMonitorAgent() bool {
 	return s.IgnoreExporter
 }
 
-// GetSource returns source to download the component
-func (s *DashboardSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentDashboard
-	}
-	return s.Source
-}
-
 // DashboardComponent represents Drainer component.
 type DashboardComponent struct{ Topology *Specification }
 
@@ -115,6 +107,15 @@ func (c *DashboardComponent) Role() string {
 	return ComponentDashboard
 }
 
+// Source implements Component interface.
+func (c *DashboardComponent) Source() string {
+	source := c.Topology.ComponentSources.Dashboard
+	if source != "" {
+		return source
+	}
+	return ComponentDashboard
+}
+
 // CalculateVersion implements the Component interface
 func (c *DashboardComponent) CalculateVersion(clusterVersion string) string {
 	version := c.Topology.ComponentVersions.Dashboard

pkg/cluster/spec/drainer.go

@@ -113,14 +113,6 @@ func (s *DrainerSpec) IgnoreMonitorAgent() bool {
 	return s.IgnoreExporter
 }
 
-// GetSource returns source to download the component
-func (s *DrainerSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentDrainer
-	}
-	return s.Source
-}
-
 // DrainerComponent represents Drainer component.
 type DrainerComponent struct{ Topology *Specification }
 
@@ -134,6 +126,15 @@ func (c *DrainerComponent) Role() string {
 	return ComponentDrainer
 }
 
+// Source implements Component interface.
+func (c *DrainerComponent) Source() string {
+	source := c.Topology.ComponentSources.Drainer
+	if source != "" {
+		return source
+	}
+	return ComponentDrainer
+}
+
 // CalculateVersion implements the Component interface
 func (c *DrainerComponent) CalculateVersion(clusterVersion string) string {
 	version := c.Topology.ComponentVersions.Drainer
@@ -161,7 +162,7 @@ func (c *DrainerComponent) Instances() []Instance {
 			ListenHost:   c.Topology.BaseTopo().GlobalOptions.ListenHost,
 			Port:         s.Port,
 			SSHP:         s.SSHPort,
-			Source:       s.GetSource(),
+			Source:       s.Source,
 			NumaNode:     s.NumaNode,
 			NumaCores:    "",
 

pkg/cluster/spec/grafana.go

@@ -107,6 +107,11 @@ func (c *GrafanaComponent) Role() string {
 	return RoleMonitor
 }
 
+// Source implements Component interface.
+func (c *GrafanaComponent) Source() string {
+	return ComponentGrafana
+}
+
 // CalculateVersion implements the Component interface
 func (c *GrafanaComponent) CalculateVersion(clusterVersion string) string {
 	// always not follow cluster version, use ""(latest) by default

pkg/cluster/spec/instance.go

@@ -70,6 +70,7 @@ var (
 type Component interface {
 	Name() string
 	Role() string
+	Source() string
 	Instances() []Instance
 	CalculateVersion(string) string
 	SetVersion(string)
@@ -316,10 +317,12 @@ func (i *BaseInstance) ComponentName() string {
 
 // ComponentSource implements Instance interface
 func (i *BaseInstance) ComponentSource() string {
-	if i.Source == "" {
-		return i.Name
+	if i.Source != "" {
+		return i.Source
+	} else if i.Component.Source() != "" {
+		return i.Component.Source()
 	}
-	return i.Source
+	return i.ComponentName()
 }
 
 // InstanceName implements Instance interface

pkg/cluster/spec/monitoring.go

@@ -124,6 +124,11 @@ func (c *MonitorComponent) Role() string {
 	return RoleMonitor
 }
 
+// Source implements Component interface.
+func (c *MonitorComponent) Source() string {
+	return ComponentPrometheus
+}
+
 // CalculateVersion implements the Component interface
 func (c *MonitorComponent) CalculateVersion(clusterVersion string) string {
 	// always not follow cluster version, use ""(latest) by default

pkg/cluster/spec/pd.go

@@ -138,14 +138,6 @@ func (s *PDSpec) GetAdvertisePeerURL(enableTLS bool) string {
 	return fmt.Sprintf("%s://%s", scheme, utils.JoinHostPort(s.Host, s.PeerPort))
 }
 
-// GetSource returns source to download the component
-func (s *PDSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentPD
-	}
-	return s.Source
-}
-
 // PDComponent represents PD component.
 type PDComponent struct{ Topology *Specification }
 
@@ -159,6 +151,15 @@ func (c *PDComponent) Role() string {
 	return ComponentPD
 }
 
+// Source implements Component interface.
+func (c *PDComponent) Source() string {
+	source := c.Topology.ComponentSources.PD
+	if source != "" {
+		return source
+	}
+	return ComponentPD
+}
+
 // CalculateVersion implements the Component interface
 func (c *PDComponent) CalculateVersion(clusterVersion string) string {
 	version := c.Topology.ComponentVersions.PD
@@ -188,7 +189,7 @@ func (c *PDComponent) Instances() []Instance {
 				ListenHost:   utils.Ternary(s.ListenHost != "", s.ListenHost, c.Topology.BaseTopo().GlobalOptions.ListenHost).(string),
 				Port:         s.ClientPort,
 				SSHP:         s.SSHPort,
-				Source:       s.GetSource(),
+				Source:       s.Source,
 				NumaNode:     s.NumaNode,
 				NumaCores:    "",
 

pkg/cluster/spec/pump.go

@@ -112,14 +112,6 @@ func (s *PumpSpec) IgnoreMonitorAgent() bool {
 	return s.IgnoreExporter
 }
 
-// GetSource returns source to download the component
-func (s *PumpSpec) GetSource() string {
-	if s.Source == "" {
-		return ComponentPump
-	}
-	return s.Source
-}
-
 // PumpComponent represents Pump component.
 type PumpComponent struct{ Topology *Specification }
 
@@ -133,6 +125,15 @@ func (c *PumpComponent) Role() string {
 	return ComponentPump
 }
 
+// Source implements Component interface.
+func (c *PumpComponent) Source() string {
+	source := c.Topology.ComponentSources.Pump
+	if source != "" {
+		return source
+	}
+	return ComponentPump
+}
+
 // CalculateVersion implements the Component interface
 func (c *PumpComponent) CalculateVersion(clusterVersion string) string {
 	version := c.Topology.ComponentVersions.Pump
@@ -160,7 +161,7 @@ func (c *PumpComponent) Instances() []Instance {
 			ListenHost:   c.Topology.BaseTopo().GlobalOptions.ListenHost,
 			Port:         s.Port,
 			SSHP:         s.SSHPort,
-			Source:       s.GetSource(),
+			Source:       s.Source,
 			NumaNode:     s.NumaNode,
 			NumaCores:    "",