feat: add replicas support for metasrv in bare-metal mode (#145)

* add replicas support for metasrv in bare-metal mode Signed-off-by: sh2 <[email protected]> * fix ci test Signed-off-by: sh2 <[email protected]> --------- Signed-off-by: sh2 <[email protected]>
GreptimeTeam · Aug 30, 2023 · 05d9130 · 05d9130
1 parent 6809b1c
commit 05d9130
Show file tree

Hide file tree

Showing 10 changed files with 75 additions and 37 deletions.
diff --git a/examples/bare-metal/cluster-with-local-artifacts.yaml b/examples/bare-metal/cluster-with-local-artifacts.yaml
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:14300
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 0.0.0.0:3002
     httpAddr: 0.0.0.0:14001

diff --git a/examples/bare-metal/cluster.yaml b/examples/bare-metal/cluster.yaml
@@ -1,7 +1,7 @@
 cluster:
   name: mycluster # name of the cluster
   artifact:
-    version: v0.2.0-nightly-20230403
+    version: latest
   frontend:
     replicas: 1
   datanode:
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:14300
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 0.0.0.0:3002
     httpAddr: 0.0.0.0:14001

diff --git a/pkg/cmd/gtctl/cluster/get/get.go b/pkg/cmd/gtctl/cluster/get/get.go
@@ -159,9 +159,8 @@ func collectClusterInfoFromBareMetal(data *bmconfig.MetaConfig) (
 
 	rows(component.Frontend, data.Cluster.Frontend.Replicas)
 	rows(component.DataNode, data.Cluster.Datanode.Replicas)
+	rows(component.MetaSrv, data.Cluster.MetaSrv.Replicas)
 
-	// TODO(shawnh2) add metatsrv and etcd replicas support
-	bulk = append(bulk, []string{component.MetaSrv, pidsMap[component.MetaSrv]})
 	bulk = append(bulk, []string{component.Etcd, pidsMap[component.Etcd]})
 
 	config, err := yaml.Marshal(data.Config)

diff --git a/pkg/deployer/baremetal/component/metasrv.go b/pkg/deployer/baremetal/component/metasrv.go
@@ -20,6 +20,7 @@ import (
 	"net"
 	"net/http"
 	"path"
+	"strconv"
 	"sync"
 	"time"
 
@@ -53,28 +54,37 @@ func (m *metaSrv) Name() string {
 }
 
 func (m *metaSrv) Start(ctx context.Context, binary string) error {
-	var (
-		metaSrvLogDir = path.Join(m.workingDirs.LogsDir, m.Name())
-		metaSrvPidDir = path.Join(m.workingDirs.PidsDir, m.Name())
-		metaSrvDirs   = []string{metaSrvLogDir, metaSrvPidDir}
-	)
-	for _, dir := range metaSrvDirs {
-		if err := fileutils.CreateDirIfNotExists(dir); err != nil {
+	// Default bind address for meta srv.
+	bindAddr := net.JoinHostPort("127.0.0.1", "3002")
+	if len(m.config.BindAddr) > 0 {
+		bindAddr = m.config.BindAddr
+	}
+
+	for i := 0; i < m.config.Replicas; i++ {
+		dirName := fmt.Sprintf("%s.%d", m.Name(), i)
+
+		metaSrvLogDir := path.Join(m.workingDirs.LogsDir, dirName)
+		if err := fileutils.CreateDirIfNotExists(metaSrvLogDir); err != nil {
+			return err
+		}
+		m.logsDirs = append(m.logsDirs, metaSrvLogDir)
+
+		metaSrvPidDir := path.Join(m.workingDirs.PidsDir, dirName)
+		if err := fileutils.CreateDirIfNotExists(metaSrvPidDir); err != nil {
+			return err
+		}
+		m.pidsDirs = append(m.pidsDirs, metaSrvPidDir)
+
+		option := &RunOptions{
+			Binary: binary,
+			Name:   dirName,
+			logDir: metaSrvLogDir,
+			pidDir: metaSrvPidDir,
+			args:   m.BuildArgs(ctx, i, bindAddr),
+		}
+		if err := runBinary(ctx, option, m.wg, m.logger); err != nil {
 			return err
 		}
-	}
-	m.logsDirs = append(m.logsDirs, metaSrvLogDir)
-	m.pidsDirs = append(m.pidsDirs, metaSrvPidDir)
-
-	option := &RunOptions{
-		Binary: binary,
-		Name:   m.Name(),
-		logDir: metaSrvLogDir,
-		pidDir: metaSrvPidDir,
-		args:   m.BuildArgs(ctx),
-	}
-	if err := runBinary(ctx, option, m.wg, m.logger); err != nil {
-		return err
 	}
 
 	// Checking component running status with intervals.
@@ -101,33 +111,47 @@ func (m *metaSrv) BuildArgs(ctx context.Context, params ...interface{}) []string
 	if logLevel == "" {
 		logLevel = config.DefaultLogLevel
 	}
+
+	nodeID_, bindAddr_ := params[0], params[1]
+	nodeID := nodeID_.(int)
+	bindAddr := bindAddr_.(string)
+
 	args := []string{
 		fmt.Sprintf("--log-level=%s", logLevel),
 		m.Name(), "start",
 		"--store-addr", m.config.StoreAddr,
 		"--server-addr", m.config.ServerAddr,
-		"--http-addr", m.config.HTTPAddr,
+		"--http-addr", generateMetaSrvAddr(m.config.HTTPAddr, nodeID),
+		"--bind-addr", generateMetaSrvAddr(bindAddr, nodeID),
 	}
 	return args
 }
 
 func (m *metaSrv) IsRunning(ctx context.Context) bool {
-	_, httpPort, err := net.SplitHostPort(m.config.HTTPAddr)
-	if err != nil {
-		m.logger.V(5).Infof("failed to split host port in %s: %s", m.Name(), err)
-		return false
-	}
+	for i := 0; i < m.config.Replicas; i++ {
+		addr := generateMetaSrvAddr(m.config.HTTPAddr, i)
+		_, httpPort, err := net.SplitHostPort(addr)
+		if err != nil {
+			m.logger.V(5).Infof("failed to split host port in %s: %s", m.Name(), err)
+			return false
+		}
 
-	rsp, err := http.Get(fmt.Sprintf("http://localhost:%s/health", httpPort))
-	if err != nil {
-		m.logger.V(5).Infof("failed to get %s health: %s", m.Name(), err)
-		return false
-	}
-	if err = rsp.Body.Close(); err != nil {
-		return false
+		rsp, err := http.Get(fmt.Sprintf("http://localhost:%s/health", httpPort))
+		if err != nil {
+			m.logger.V(5).Infof("failed to get %s health: %s", m.Name(), err)
+			return false
+		}
+
+		if rsp.StatusCode != http.StatusOK {
+			return false
+		}
+
+		if err = rsp.Body.Close(); err != nil {
+			return false
+		}
 	}
 
-	return rsp.StatusCode == http.StatusOK
+	return true
 }
 
 func (m *metaSrv) Delete(ctx context.Context, option DeleteOptions) error {
@@ -136,3 +160,9 @@ func (m *metaSrv) Delete(ctx context.Context, option DeleteOptions) error {
 	}
 	return nil
 }
+
+func generateMetaSrvAddr(addr string, nodeID int) string {
+	host, port, _ := net.SplitHostPort(addr)
+	portInt, _ := strconv.Atoi(port)
+	return net.JoinHostPort(host, strconv.Itoa(portInt+nodeID))
+}
diff --git a/pkg/deployer/baremetal/config/common.go b/pkg/deployer/baremetal/config/common.go
@@ -79,6 +79,7 @@ func DefaultConfig() *Config {
 				Replicas: 1,
 			},
 			MetaSrv: &MetaSrv{
+				Replicas:   1,
 				StoreAddr:  "127.0.0.1:2379",
 				ServerAddr: "0.0.0.0:3002",
 				HTTPAddr:   "0.0.0.0:14001",

diff --git a/pkg/deployer/baremetal/config/metasrv.go b/pkg/deployer/baremetal/config/metasrv.go
@@ -15,6 +15,8 @@
 package config
 
 type MetaSrv struct {
+	Replicas int `yaml:"replicas" validate:"gt=0"`
+
 	StoreAddr  string `yaml:"storeAddr" validate:"hostname_port"`
 	ServerAddr string `yaml:"serverAddr" validate:"hostname_port"`
 	BindAddr   string `yaml:"bindAddr" validate:"omitempty,hostname_port"`

diff --git a/pkg/deployer/baremetal/test_data/invalid_artifact.yaml b/pkg/deployer/baremetal/test_data/invalid_artifact.yaml
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:14300
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 0.0.0.0:3002
     httpAddr: 0.0.0.0:14001

diff --git a/pkg/deployer/baremetal/test_data/invalid_hostname_port.yaml b/pkg/deployer/baremetal/test_data/invalid_hostname_port.yaml
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:1438000  # invalid port
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 6870.0.0.0:3243002  # invalid hostname and port
     httpAddr: 0.0.0.0:14001

diff --git a/pkg/deployer/baremetal/test_data/invalid_replicas.yaml b/pkg/deployer/baremetal/test_data/invalid_replicas.yaml
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:14300
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 0.0.0.0:3002
     httpAddr: 0.0.0.0:14001

diff --git a/pkg/deployer/baremetal/test_data/valid_config.yaml b/pkg/deployer/baremetal/test_data/valid_config.yaml
@@ -10,6 +10,7 @@ cluster:
     mysqlAddr: 0.0.0.0:14200
     httpAddr: 0.0.0.0:14300
   meta:
+    replicas: 1
     storeAddr: 127.0.0.1:2379
     serverAddr: 0.0.0.0:3002
     httpAddr: 0.0.0.0:14001