resolve conflict

Signed-off-by: bufferflies <[email protected]>

Author: bufferflies

components/cluster/command/upgrade.go

@@ -24,8 +24,7 @@ import (
 func newUpgradeCmd() *cobra.Command {
 	offlineMode := false
 	ignoreVersionCheck := false
-	var tidbVer, tikvVer, pdVer, tsoVer, schedulingVer, routerVer, tiflashVer, kvcdcVer, dashboardVer, cdcVer,
-		alertmanagerVer, nodeExporterVer, blackboxExporterVer, tiproxyVer string
+	var tidbVer, tikvVer, pdVer, tsoVer, schedulingVer, resourceManagerVer, routerVer, tiflashVer, kvcdcVer, dashboardVer, cdcVer, alertmanagerVer, nodeExporterVer, blackboxExporterVer, tiproxyVer string
 	var restartTimeout time.Duration
 
 	cmd := &cobra.Command{
@@ -51,6 +50,7 @@ func newUpgradeCmd() *cobra.Command {
 				spec.ComponentTSO:              tsoVer,
 				spec.ComponentScheduling:       schedulingVer,
 				spec.ComponentRouter:           routerVer,
+				spec.ComponentResourceManager:  resourceManagerVer,
 				spec.ComponentTiFlash:          tiflashVer,
 				spec.ComponentTiKVCDC:          kvcdcVer,
 				spec.ComponentCDC:              cdcVer,
@@ -83,6 +83,7 @@ func newUpgradeCmd() *cobra.Command {
 	cmd.Flags().StringVar(&pdVer, "pd-version", "", "Fix the version of pd and no longer follows the cluster version.")
 	cmd.Flags().StringVar(&tsoVer, "tso-version", "", "Fix the version of tso and no longer follows the cluster version.")
 	cmd.Flags().StringVar(&schedulingVer, "scheduling-version", "", "Fix the version of scheduling and no longer follows the cluster version.")
+	cmd.Flags().StringVar(&resourceManagerVer, "resource-manager-version", "", "Fix the version of resource-manager and no longer follows the cluster version.")
 	cmd.Flags().StringVar(&tiflashVer, "tiflash-version", "", "Fix the version of tiflash and no longer follows the cluster version.")
 	cmd.Flags().StringVar(&dashboardVer, "tidb-dashboard-version", "", "Fix the version of tidb-dashboard and no longer follows the cluster version.")
 	cmd.Flags().StringVar(&cdcVer, "cdc-version", "", "Fix the version of cdc and no longer follows the cluster version.")

components/playground/playground.go

@@ -1423,7 +1423,7 @@ func (p *Playground) bootCluster(ctx context.Context, env *environment.Environme
 			}
 
 			for _, router := range p.routers {
-				routerAddr = append(schedulingAddr, router.Addr())
+				routerAddr = append(routerAddr, router.Addr())
 			}
 
 			fmt.Printf("PD API Endpoints:   ")

embed/templates/config/prometheus.yml.tpl

@@ -190,6 +190,22 @@ scrape_configs:
     - targets:
 {{- range .SchedulingAddrs}}
       - '{{.}}'
+{{- end}}
+{{- if .ResourceManagerAddrs}}
+  - job_name: "resource-manager"
+    honor_labels: true # don't overwrite job & instance labels
+{{- if .TLSEnabled}}
+    scheme: https
+    tls_config:
+      insecure_skip_verify: false
+      ca_file: ../tls/ca.crt
+      cert_file: ../tls/prometheus.crt
+      key_file: ../tls/prometheus.pem
+{{- end}}
+    static_configs:
+    - targets:
+{{- range .ResourceManagerAddrs}}
+      - '{{.}}'
 {{- end}}
   - job_name: "router"
     honor_labels: true # don't overwrite job & instance labels

embed/templates/scripts/run_resource-manager.sh.tpl

@@ -0,0 +1,22 @@
+#!/bin/bash
+set -e
+
+# WARNING: This file was auto-generated. Do not edit!
+#          All your edit might be overwritten!
+DEPLOY_DIR={{.DeployDir}}
+
+cd "${DEPLOY_DIR}" || exit 1
+
+{{- if .NumaNode}}
+exec numactl --cpunodebind={{.NumaNode}} --membind={{.NumaNode}} env GODEBUG=madvdontneed=1 bin/pd-server services resource-manager\
+{{- else}}
+exec env GODEBUG=madvdontneed=1 bin/pd-server services resource-manager \
+{{- end}}
+{{- if .Name}}
+    --name="{{.Name}}" \
+{{- end}}
+    --backend-endpoints="{{.BackendEndpoints}}" \
+    --listen-addr="{{.ListenURL}}" \
+    --advertise-listen-addr="{{.AdvertiseListenURL}}" \
+    --config=conf/resource-manager.toml \
+    --log-file="{{.LogDir}}/resource-manager.log" 2>> "{{.LogDir}}/resource-manager_stderr.log"

pkg/cluster/api/pdapi.go

@@ -1048,9 +1048,10 @@ func (pc *PDClient) SetLeaderPriority(name string, value int32) error {
 }
 
 const (
-	tsoStatusURI        = "status"
-	schedulingStatusURI = "status"
-	routerStatusURI     = "status"
+	tsoStatusURI             = "status"
+	schedulingStatusURI      = "status"
+	resourceManagerStatusURI = "status"
+	routerStatusURI          = "status"
 )
 
 // TSOClient is an HTTP client of the TSO server
@@ -1253,6 +1254,88 @@ func (tc *SchedulingClient) CheckHealth() error {
 	return nil
 }
 
+// ResourceManagerClient is an HTTP client of the resource manager server
+type ResourceManagerClient struct {
+	version    string
+	addrs      []string
+	tlsEnabled bool
+	httpClient *utils.HTTPClient
+	ctx        context.Context
+}
+
+// NewResourceManagerClient returns a new ResourceManagerClient, the context must have
+// a *logprinter.Logger as value of "logger"
+func NewResourceManagerClient(
+	ctx context.Context,
+	addrs []string,
+	timeout time.Duration,
+	tlsConfig *tls.Config,
+) *ResourceManagerClient {
+	enableTLS := false
+	if tlsConfig != nil {
+		enableTLS = true
+	}
+
+	if _, ok := ctx.Value(logprinter.ContextKeyLogger).(*logprinter.Logger); !ok {
+		panic("the context must have logger inside")
+	}
+
+	cli := &ResourceManagerClient{
+		addrs:      addrs,
+		tlsEnabled: enableTLS,
+		httpClient: utils.NewHTTPClient(timeout, tlsConfig),
+		ctx:        ctx,
+	}
+
+	cli.tryIdentifyVersion()
+	return cli
+}
+
+func (tc *ResourceManagerClient) tryIdentifyVersion() {
+	endpoints := tc.getEndpoints(resourceManagerStatusURI)
+	response := map[string]string{}
+	_, err := tryURLs(endpoints, func(endpoint string) ([]byte, error) {
+		body, err := tc.httpClient.Get(tc.ctx, endpoint)
+		if err != nil {
+			return body, err
+		}
+
+		return body, json.Unmarshal(body, &response)
+	})
+	if err == nil {
+		tc.version = response["version"]
+	}
+}
+
+// GetURL builds the client URL
+func (tc *ResourceManagerClient) GetURL(addr string) string {
+	httpPrefix := "http"
+	if tc.tlsEnabled {
+		httpPrefix = "https"
+	}
+	return fmt.Sprintf("%s://%s", httpPrefix, addr)
+}
+
+func (tc *ResourceManagerClient) getEndpoints(uri string) (endpoints []string) {
+	for _, addr := range tc.addrs {
+		endpoints = append(endpoints, fmt.Sprintf("%s/%s", tc.GetURL(addr), uri))
+	}
+	return
+}
+
+// CheckHealth checks the health of resource manager node.
+func (tc *ResourceManagerClient) CheckHealth() error {
+	endpoints := tc.getEndpoints(resourceManagerStatusURI)
+	_, err := tryURLs(endpoints, func(endpoint string) ([]byte, error) {
+		body, err := tc.httpClient.Get(tc.ctx, endpoint)
+		if err != nil {
+			return body, err
+		}
+		return body, nil
+	})
+	return err
+}
+
 // RouterClient is an HTTP client of the router server
 type RouterClient struct {
 	version    string

pkg/cluster/operation/upgrade.go

@@ -143,7 +143,7 @@ func Upgrade(
 
 			// Usage within the switch statement
 			switch component.Name() {
-			case spec.ComponentPD, spec.ComponentTSO, spec.ComponentScheduling, spec.ComponentRouter:
+			case spec.ComponentPD, spec.ComponentTSO, spec.ComponentScheduling, spec.ComponentResourceManager, spec.ComponentRouter:
 				// defer PD related leader/primary to be upgraded after others
 				isLeader, err := checkAndDeferPDLeader(ctx, topo, int(options.APITimeout), tlsCfg, instance)
 				if err != nil {
@@ -244,6 +244,8 @@ func checkAndDeferPDLeader(ctx context.Context, topo spec.Topology, apiTimeout i
 		isLeader, err = instance.(*spec.TSOInstance).IsPrimary(ctx, topo, tlsCfg)
 	case spec.ComponentRouter:
 		isLeader, err = instance.(*spec.RouterInstance).IsPrimary(ctx, topo, tlsCfg)
+	case spec.ComponentResourceManager:
+		isLeader, err = instance.(*spec.ResourceManagerInstance).IsPrimary(ctx, topo, tlsCfg)
 	}
 	if err != nil {
 		return false, err

pkg/cluster/spec/monitoring.go

@@ -346,6 +346,13 @@ func (i *MonitorInstance) InitConfig(
 			cfig.AddRouter(router.Host, uint64(router.Port))
 		}
 	}
+	if servers, found := topoHasField("ResourceManagerServers"); found {
+		for i := 0; i < servers.Len(); i++ {
+			rm := servers.Index(i).Interface().(*ResourceManagerSpec)
+			uniqueHosts.Insert(rm.Host)
+			cfig.AddResourceManager(rm.Host, uint64(rm.Port))
+		}
+	}
 	if servers, found := topoHasField("TiKVServers"); found {
 		for i := 0; i < servers.Len(); i++ {
 			kv := servers.Index(i).Interface().(*TiKVSpec)