tiup cluster support router service mcs

components/cluster/command/upgrade.go

@@ -24,7 +24,7 @@ import (
 func newUpgradeCmd() *cobra.Command {
 	offlineMode := false
 	ignoreVersionCheck := false
-	var tidbVer, tikvVer, pdVer, tsoVer, schedulingVer, resourceManagerVer, 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{
@@ -49,6 +49,7 @@ func newUpgradeCmd() *cobra.Command {
 				spec.ComponentPD:               pdVer,
 				spec.ComponentTSO:              tsoVer,
 				spec.ComponentScheduling:       schedulingVer,
+				spec.ComponentRouter:           routerVer,
 				spec.ComponentResourceManager:  resourceManagerVer,
 				spec.ComponentTiFlash:          tiflashVer,
 				spec.ComponentTiKVCDC:          kvcdcVer,

components/playground/playground.go

@@ -1594,12 +1594,12 @@ func (p *Playground) terminate(sig syscall.Signal) {
 	}
 	for _, inst := range p.routers {
 		if inst.Process() != nil && inst.Process().Cmd() != nil && inst.Process().Cmd().Process != nil {
-			kill(inst.Component(), inst.Process().Pid(), inst.Wait)
+			kill(inst.Name(), inst.Process().Pid(), inst.Wait)
 		}
 	}
 	for _, inst := range p.resourceManagers {
 		if inst.Process() != nil && inst.Process().Cmd() != nil && inst.Process().Cmd().Process != nil {
-			kill(inst.Component(), inst.Process().Pid(), inst.Wait)
+			kill(inst.Name(), inst.Process().Pid(), inst.Wait)
 		}
 	}
 	for _, inst := range p.tiproxys {

embed/templates/config/prometheus.yml.tpl

@@ -207,6 +207,21 @@ scrape_configs:
 {{- range .ResourceManagerAddrs}}
       - '{{.}}'
 {{- end}}
+{{- end}}
+  - job_name: "router"
+    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 .RouterAddrs}}
+      - '{{.}}'
 {{- end}}
 {{- if .TiFlashStatusAddrs}}
   - job_name: "tiflash"

embed/templates/scripts/run_router.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 router\
+{{- else}}
+exec env GODEBUG=madvdontneed=1 bin/pd-server services router \
+{{- end}}
+{{- if .Name}}
+    --name="{{.Name}}" \
+{{- end}}
+    --backend-endpoints="{{.BackendEndpoints}}" \
+    --listen-addr="{{.ListenURL}}" \
+    --advertise-listen-addr="{{.AdvertiseListenURL}}" \
+    --config=conf/router.toml \
+    --log-file="{{.LogDir}}/router.log" 2>> "{{.LogDir}}/router_stderr.log"

pkg/cluster/api/pdapi.go

@@ -1051,6 +1051,7 @@ const (
 	tsoStatusURI             = "status"
 	schedulingStatusURI      = "status"
 	resourceManagerStatusURI = "status"
+	routerStatusURI          = "status"
 )
 
 // TSOClient is an HTTP client of the TSO server
@@ -1199,10 +1200,6 @@ func NewSchedulingClient(
 	return cli
 }
 
-// func (tc *SchedulingClient) l() *logprinter.Logger {
-// 	return tc.ctx.Value(logprinter.ContextKeyLogger).(*logprinter.Logger)
-// }
-
 func (tc *SchedulingClient) tryIdentifyVersion() {
 	endpoints := tc.getEndpoints(schedulingStatusURI)
 	response := map[string]string{}
@@ -1338,3 +1335,94 @@ func (tc *ResourceManagerClient) CheckHealth() error {
 	})
 	return err
 }
+
+// RouterClient is an HTTP client of the router server
+type RouterClient struct {
+	version    string
+	addrs      []string
+	tlsEnabled bool
+	httpClient *utils.HTTPClient
+	ctx        context.Context
+}
+
+// NewRouterClient returns a new RouterClient, the context must have
+// a *logprinter.Logger as value of "logger"
+func NewRouterClient(
+	ctx context.Context,
+	addrs []string,
+	timeout time.Duration,
+	tlsConfig *tls.Config,
+) *RouterClient {
+	enableTLS := false
+	if tlsConfig != nil {
+		enableTLS = true
+	}
+
+	if _, ok := ctx.Value(logprinter.ContextKeyLogger).(*logprinter.Logger); !ok {
+		panic("the context must have logger inside")
+	}
+
+	cli := &RouterClient{
+		addrs:      addrs,
+		tlsEnabled: enableTLS,
+		httpClient: utils.NewHTTPClient(timeout, tlsConfig),
+		ctx:        ctx,
+	}
+
+	cli.tryIdentifyVersion()
+	return cli
+}
+
+func (tc *RouterClient) tryIdentifyVersion() {
+	endpoints := tc.getEndpoints(routerStatusURI)
+	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 of PDClient
+func (tc *RouterClient) GetURL(addr string) string {
+	httpPrefix := "http"
+	if tc.tlsEnabled {
+		httpPrefix = "https"
+	}
+	return fmt.Sprintf("%s://%s", httpPrefix, addr)
+}
+
+func (tc *RouterClient) getEndpoints(uri string) (endpoints []string) {
+	for _, addr := range tc.addrs {
+		endpoint := fmt.Sprintf("%s/%s", tc.GetURL(addr), uri)
+		endpoints = append(endpoints, endpoint)
+	}
+
+	return
+}
+
+// CheckHealth checks the health of router node.
+func (tc *RouterClient) CheckHealth() error {
+	endpoints := tc.getEndpoints(routerStatusURI)
+
+	_, 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
+	})
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/cluster/manager/manager_test.go

@@ -90,6 +90,8 @@ tso_servers:
   - host: 172.16.5.53
 scheduling_servers:
   - host: 172.16.5.54
+router_servers:
+  - host: 172.16.5.55
 `), &topo)
 	assert.Nil(err)
 }

pkg/cluster/manager/transfer_test.go

@@ -79,4 +79,18 @@ func TestRenderSpec(t *testing.T) {
 	dir, err = renderSpec("{{.DataDir}}", s, "test-scheduling")
 	assert.Nil(t, err)
 	assert.NotEmpty(t, dir)
+
+	s = &spec.RouterInstance{BaseInstance: spec.BaseInstance{
+		InstanceSpec: &spec.RouterSpec{
+			Host:      "172.16.5.140",
+			SSHPort:   22,
+			Name:      "router-1",
+			DeployDir: "/home/test/deploy/router-3379",
+			DataDir:   "/home/test/deploy/router-3379/data",
+		},
+	}}
+	// s.BaseInstance.InstanceSpec
+	dir, err = renderSpec("{{.DataDir}}", s, "test-scheduling")
+	assert.Nil(t, err)
+	assert.NotEmpty(t, dir)
 }

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.ComponentResourceManager:
+			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 {
@@ -242,6 +242,8 @@ func checkAndDeferPDLeader(ctx context.Context, topo spec.Topology, apiTimeout i
 		isLeader, err = instance.(*spec.SchedulingInstance).IsPrimary(ctx, topo, tlsCfg)
 	case spec.ComponentTSO:
 		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)
 	}

pkg/cluster/spec/monitoring.go

@@ -339,6 +339,13 @@ func (i *MonitorInstance) InitConfig(
 			cfig.AddScheduling(scheduling.Host, uint64(scheduling.Port))
 		}
 	}
+	if servers, found := topoHasField("RouterServers"); found {
+		for i := 0; i < servers.Len(); i++ {
+			router := servers.Index(i).Interface().(*RouterSpec)
+			uniqueHosts.Insert(router.Host)
+			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)