Repository: matrixorigin/matrixone
Pull request #4236 (2022-07-27 14:10:27)

logservice: allow cluster to be bootstrapped using toml config file
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

allow cluster to be bootstrapped using toml config file. 

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index 0bf2b843e..068fb26dd 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -103,13 +103,19 @@ func startDNService(cfg *Config, stopper *stopper.Stopper) error {
 }
 
 func startLogService(cfg *Config, stopper *stopper.Stopper) error {
+	s, err := logservice.NewService(cfg.LogService)
+	if err != nil {
+		panic(err)
+	}
+	if err := s.Start(); err != nil {
+		panic(err)
+	}
 	return stopper.RunNamedTask("log-service", func(ctx context.Context) {
-		s, err := logservice.NewService(cfg.LogService)
-		if err != nil {
-			panic(err)
-		}
-		if err := s.Start(); err != nil {
-			panic(err)
+		if cfg.LogService.BootstrapConfig.BootstrapCluster {
+			fmt.Printf("bootstrapping hakeeper...\n")
+			if err := s.BootstrapHAKeeper(ctx, cfg.LogService); err != nil {
+				panic(err)
+			}
 		}
 
 		<-ctx.Done()
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 4bf9867c7..11db1dbfe 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -15,10 +15,13 @@
 package logservice
 
 import (
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/google/uuid"
+	"github.com/lni/dragonboat/v4"
 	"github.com/lni/vfs"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -34,7 +37,8 @@ const (
 )
 
 var (
-	ErrInvalidConfig = moerr.NewError(moerr.BAD_CONFIGURATION, "invalid log configuration")
+	ErrInvalidBootstrapConfig = moerr.NewError(moerr.BAD_CONFIGURATION, "invalid bootstrap configuration")
+	ErrInvalidConfig          = moerr.NewError(moerr.BAD_CONFIGURATION, "invalid log configuration")
 )
 
 const (
@@ -92,6 +96,40 @@ type Config struct {
 	// cluster health checks.
 	HAKeeperCheckInterval toml.Duration `toml:"hakeeper-check-interval"`
 
+	// BootstrapConfig is the configuration specified for the bootstrapping
+	// procedure. It only need to be specified for Log Stores selected to host
+	// initial HAKeeper replicas during bootstrapping.
+	BootstrapConfig struct {
+		// BootstrapCluster indicates whether the cluster should be bootstrapped.
+		BootstrapCluster bool `toml:"bootstrap-cluster"`
+		// NumOfLogShards defines the number of Log shards in the initial deployment.
+		NumOfLogShards uint64 `toml:"num-of-log-shards"`
+		// NumOfDNShards defines the number of DN shards in the initial deployment.
+		// The count must be the same as NumOfLogShards in the current implementation.
+		NumOfDNShards uint64 `toml:"num-of-dn-shards"`
+		// NumOfLogShardReplicas is the number of replicas for each shard managed by
+		// Log Stores, including Log Service shards and the HAKeeper.
+		NumOfLogShardReplicas uint64 `toml:"num-of-log-shard-replicas"`
+		// InitHAKeeperMembers defines the initial members of the HAKeeper as a list
+		// of semicolon separated HAKeeper replicaID and UUID pairs. For example,
+		// when the initial HAKeeper members are
+		// replica with replica ID 101 running on Log Store uuid1
+		// replica with replica ID 102 running on Log Store uuid2
+		// replica with replica ID 103 running on Log Store uuid3
+		// the InitHAKeeperMembers string value should be
+		// "101:uuid1;102:uuid2;103:uuid3"
+		// Note that these initial HAKeeper replica IDs must be assigned by k8s
+		// from the range [K8SIDRangeStart, K8SIDRangeEnd) as defined in pkg/hakeeper.
+		// All uuid values are assigned by k8s, they are used to uniquely identify
+		// CN/DN/Log stores.
+		InitHAKeeperMembers string `toml:"init-hakeeper-members"`
+		// HAKeeperReplicaID is the replica ID of the HAKeeper replica to be started
+		// on the local Log Store during bootstrapping. Note that this replica is an
+		// initial member of the HAKeeper so the replicaID should exist in
+		// InitHAKeeperMembers.
+		HAKeeperReplicaID uint64 `toml:"hakeeper-replica-id"`
+	}
+
 	HAKeeperConfig struct {
 		// TickPerSecond indicates how many ticks every second.
 		// In HAKeeper, we do not use actual time to measure time elapse.
@@ -145,6 +183,38 @@ func (c *Config) GetHAKeeperClientConfig() HAKeeperClientConfig {
 	}
 }
 
+func (c *Config) GetInitHAKeeperMembers() (map[uint64]dragonboat.Target, error) {
+	v := strings.TrimSpace(c.BootstrapConfig.InitHAKeeperMembers)
+	replicas := strings.Split(v, ";")
+	result := make(map[uint64]dragonboat.Target)
+	for _, pair := range replicas {
+		pair = strings.TrimSpace(pair)
+		parts := strings.Split(pair, ":")
+		if len(parts) == 2 {
+			id := strings.TrimSpace(parts[0])
+			target := strings.TrimSpace(parts[1])
+			if _, err := uuid.Parse(target); err != nil {
+				plog.Errorf("invalid uuid %s", target)
+				return nil, ErrInvalidBootstrapConfig
+			}
+			idn, err := strconv.ParseUint(id, 10, 64)
+			if err != nil {
+				plog.Errorf("invalid replicaID %s", id)
+				return nil, ErrInvalidBootstrapConfig
+			}
+			if idn >= hakeeper.K8SIDRangeEnd || idn < hakeeper.K8SIDRangeStart {
+				plog.Errorf("out of range hakeeper replica id")
+				return nil, ErrInvalidBootstrapConfig
+			}
+			result[idn] = target
+		} else {
+			plog.Errorf("invalid replicaID:target uuid pair, %s", pair)
+			return nil, ErrInvalidBootstrapConfig
+		}
+	}
+	return result, nil
+}
+
 // Validate validates the configuration.
 func (c *Config) Validate() error {
 	if len(c.UUID) == 0 {
@@ -176,6 +246,35 @@ func (c *Config) Validate() error {
 	if c.HAKeeperConfig.DnStoreTimeout.Duration == 0 {
 		return errors.Wrapf(ErrInvalidConfig, "DnStoreTimeout not set")
 	}
+	// validate BootstrapConfig
+	if c.BootstrapConfig.BootstrapCluster {
+		if c.BootstrapConfig.NumOfLogShards == 0 {
+			return errors.Wrapf(ErrInvalidConfig, "NumOfLogShards not set")
+		}
+		if c.BootstrapConfig.NumOfDNShards == 0 {
+			return errors.Wrapf(ErrInvalidConfig, "NumOfDNShards not set")
+		}
+		if c.BootstrapConfig.NumOfLogShardReplicas == 0 {
+			return errors.Wrapf(ErrInvalidConfig, "NumOfLogShardReplica not set")
+		}
+		if c.BootstrapConfig.NumOfDNShards != c.BootstrapConfig.NumOfLogShards {
+			return errors.Wrapf(ErrInvalidConfig, "NumOfDNShards != NumOfLogShards")
+		}
+		members, err := c.GetInitHAKeeperMembers()
+		if err != nil {
+			return err
+		}
+		if len(members) == 0 {
+			return errors.Wrapf(ErrInvalidConfig, "InitHAKeeperMembers not set")
+		}
+		if uint64(len(members)) != c.BootstrapConfig.NumOfLogShardReplicas {
+			return errors.Wrapf(ErrInvalidConfig, "InitHAKeeperMembers and NumOfLogShardReplicas not match")
+		}
+		if _, ok := members[c.BootstrapConfig.HAKeeperReplicaID]; !ok {
+			return errors.Wrapf(ErrInvalidConfig, "HAKeeperReplicaID not a part of InitHAKeeperMembers")
+		}
+	}
+
 	return nil
 }
 
diff --git a/pkg/logservice/hakeeper_checker.go b/pkg/logservice/hakeeper_checker.go
index 2fbecff48..154f7bd6e 100644
--- a/pkg/logservice/hakeeper_checker.go
+++ b/pkg/logservice/hakeeper_checker.go
@@ -28,7 +28,7 @@ const (
 	defaultIDBatchSize uint64 = 1024 * 10
 
 	hakeeperDefaultTimeout = 2 * time.Second
-	checkBootstrapInterval = 100
+	checkBootstrapCycles   = 100
 )
 
 type idAllocator struct {
@@ -82,8 +82,11 @@ func (l *store) setInitialClusterInfo(numOfLogShards uint64,
 		plog.Errorf("failed to propose initial cluster info, %v", err)
 		return err
 	}
+	if result.Value == uint64(pb.HAKeeperBootstrapFailed) {
+		panic("bootstrap failed")
+	}
 	if result.Value != uint64(pb.HAKeeperCreated) {
-		plog.Errorf("tried to set initial cluster info again")
+		plog.Errorf("initial cluster info already set")
 	}
 	return nil
 }
@@ -119,7 +122,7 @@ func (l *store) hakeeperCheck() {
 		}
 		switch state.State {
 		case pb.HAKeeperCreated:
-			plog.Warningf("missing initial cluster info, check skipped")
+			plog.Warningf("waiting for initial cluster info to be set, check skipped")
 			return
 		case pb.HAKeeperBootstrapping:
 			l.bootstrap(term, state)
@@ -188,14 +191,14 @@ func (l *store) bootstrap(term uint64, state *pb.CheckerState) {
 			plog.Infof("failed to add schedule commands, %v", err)
 			return
 		}
-		l.bootstrapCheckInterval = checkBootstrapInterval
+		l.bootstrapCheckCycles = checkBootstrapCycles
 		l.bootstrapMgr = bootstrap.NewBootstrapManager(state.ClusterInfo, nil)
 		l.assertHAKeeperState(pb.HAKeeperBootstrapCommandsReceived)
 	}
 }
 
 func (l *store) checkBootstrap(state *pb.CheckerState) {
-	if l.bootstrapCheckInterval == 0 {
+	if l.bootstrapCheckCycles == 0 {
 		if err := l.setBootstrapState(false); err != nil {
 			panic(err)
 		}
@@ -206,7 +209,7 @@ func (l *store) checkBootstrap(state *pb.CheckerState) {
 		l.bootstrapMgr = bootstrap.NewBootstrapManager(state.ClusterInfo, nil)
 	}
 	if !l.bootstrapMgr.CheckBootstrap(state.LogState) {
-		l.bootstrapCheckInterval--
+		l.bootstrapCheckCycles--
 	} else {
 		if err := l.setBootstrapState(true); err != nil {
 			panic(err)
diff --git a/pkg/logservice/service_bootstrap.go b/pkg/logservice/service_bootstrap.go
new file mode 100644
index 000000000..8fa1d8f6e
--- /dev/null
+++ b/pkg/logservice/service_bootstrap.go
@@ -0,0 +1,52 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"context"
+	"time"
+)
+
+func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
+	members, err := cfg.GetInitHAKeeperMembers()
+	if err != nil {
+		return err
+	}
+	replicaID := cfg.BootstrapConfig.HAKeeperReplicaID
+	if err := s.store.startHAKeeperReplica(replicaID, members, false); err != nil {
+		plog.Errorf("failed to start hakeeper replica, %v", err)
+		return err
+	}
+	numOfLogShards := cfg.BootstrapConfig.NumOfLogShards
+	numOfDNShards := cfg.BootstrapConfig.NumOfDNShards
+	numOfLogReplicas := cfg.BootstrapConfig.NumOfLogShardReplicas
+	for i := 0; i < checkBootstrapCycles; i++ {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+		plog.Infof("trying to set initial cluster info")
+		if err := s.store.setInitialClusterInfo(numOfLogShards,
+			numOfDNShards, numOfLogReplicas); err != nil {
+			plog.Errorf("failed to set initial cluster info, %v", err)
+			time.Sleep(time.Second)
+			continue
+		}
+		plog.Infof("initial cluster info set")
+		break
+	}
+	return nil
+}
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index fc1838e58..c98aea5df 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -112,8 +112,8 @@ type store struct {
 	alloc             hakeeper.IDAllocator
 	stopper           *stopper.Stopper
 
-	bootstrapCheckInterval uint64
-	bootstrapMgr           *bootstrap.Manager
+	bootstrapCheckCycles uint64
+	bootstrapMgr         *bootstrap.Manager
 
 	mu struct {
 		sync.Mutex


Test patch:
diff --git a/etc/log-minimal-test.toml b/etc/log-minimal-test.toml
index bc041eca4..8a3cb3396 100644
--- a/etc/log-minimal-test.toml
+++ b/etc/log-minimal-test.toml
@@ -10,3 +10,14 @@ max-size = 512
 deployment-id = 1
 uuid = "9c4dccb4-4d3c-41f8-b482-5251dc7a41bf"
 gossip-seed-addresses = "127.0.0.1:32002"
+
+[logservice.BootstrapConfig]
+bootstrap-cluster = true
+num-of-log-shards = 1
+num-of-dn-shards = 1
+num-of-log-shard-replicas = 1
+init-hakeeper-members = "131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf"
+hakeeper-replica-id = 131072
+
+[logservice.HAKeeperClientConfig]
+hakeeper-service-addresses = "127.0.0.1:32000"
diff --git a/pkg/logservice/config_test.go b/pkg/logservice/config_test.go
index 5a16cb01a..ca7877cd0 100644
--- a/pkg/logservice/config_test.go
+++ b/pkg/logservice/config_test.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 )
@@ -62,6 +63,45 @@ func TestSplitAddress(t *testing.T) {
 	}
 }
 
+func TestGetInitHAKeeperMembers(t *testing.T) {
+	cfg1 := Config{}
+	cfg1.BootstrapConfig.InitHAKeeperMembers = " 131072 :9c4dccb4-4d3c-41f8-b482-5251dc7a41bf "
+	result, err := cfg1.GetInitHAKeeperMembers()
+	require.NoError(t, err)
+	assert.Equal(t, 1, len(result))
+	v, ok := result[131072]
+	assert.True(t, ok)
+	assert.Equal(t, "9c4dccb4-4d3c-41f8-b482-5251dc7a41bf", v)
+
+	cfg2 := Config{}
+	cfg2.BootstrapConfig.InitHAKeeperMembers = "131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf;131073:9c4dccb4-4d3c-41f8-b482-5251dc7a41be"
+	result, err = cfg2.GetInitHAKeeperMembers()
+	require.NoError(t, err)
+	assert.Equal(t, 2, len(result))
+	v1, ok1 := result[131072]
+	v2, ok2 := result[131073]
+	assert.True(t, ok1)
+	assert.True(t, ok2)
+	assert.Equal(t, "9c4dccb4-4d3c-41f8-b482-5251dc7a41bf", v1)
+	assert.Equal(t, "9c4dccb4-4d3c-41f8-b482-5251dc7a41be", v2)
+
+	tests := []string{
+		"131071:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf",
+		"262144:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf",
+		"262145:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf",
+		"131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41b",
+		"131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf;",
+		"131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf;1:1",
+	}
+
+	for _, v := range tests {
+		cfg := Config{}
+		cfg.BootstrapConfig.InitHAKeeperMembers = v
+		_, err := cfg.GetInitHAKeeperMembers()
+		assert.Equal(t, ErrInvalidBootstrapConfig, err)
+	}
+}
+
 func TestGetGossipSeedAddresses(t *testing.T) {
 	cfg := Config{
 		GossipSeedAddresses: "localhost:9000;localhost:9001 ; localhost:9002 ",
@@ -72,7 +112,7 @@ func TestGetGossipSeedAddresses(t *testing.T) {
 
 func TestConfigCanBeValidated(t *testing.T) {
 	c := getTestConfig()
-	assert.Nil(t, c.Validate())
+	assert.NoError(t, c.Validate())
 
 	c1 := c
 	c1.DeploymentID = 0
@@ -95,6 +135,39 @@ func TestConfigCanBeValidated(t *testing.T) {
 	assert.True(t, errors.Is(c5.Validate(), ErrInvalidConfig))
 }
 
+func TestBootstrapConfigCanBeValidated(t *testing.T) {
+	c := getTestConfig()
+	assert.NoError(t, c.Validate())
+
+	c.BootstrapConfig.BootstrapCluster = true
+	c.BootstrapConfig.NumOfLogShards = 3
+	c.BootstrapConfig.NumOfDNShards = 3
+	c.BootstrapConfig.NumOfLogShardReplicas = 1
+	c.BootstrapConfig.InitHAKeeperMembers = "131072:9c4dccb4-4d3c-41f8-b482-5251dc7a41bf"
+	c.BootstrapConfig.HAKeeperReplicaID = 131072
+	assert.NoError(t, c.Validate())
+
+	c1 := c
+	c1.BootstrapConfig.NumOfLogShards = 0
+	assert.True(t, errors.Is(c1.Validate(), ErrInvalidConfig))
+
+	c2 := c
+	c2.BootstrapConfig.NumOfDNShards = 0
+	assert.True(t, errors.Is(c2.Validate(), ErrInvalidConfig))
+
+	c3 := c
+	c3.BootstrapConfig.NumOfDNShards = 2
+	assert.True(t, errors.Is(c3.Validate(), ErrInvalidConfig))
+
+	c4 := c
+	c4.BootstrapConfig.NumOfLogShardReplicas = 2
+	assert.True(t, errors.Is(c4.Validate(), ErrInvalidConfig))
+
+	c5 := c
+	c5.BootstrapConfig.HAKeeperReplicaID = 2
+	assert.True(t, errors.Is(c5.Validate(), ErrInvalidConfig))
+}
+
 func TestFillConfig(t *testing.T) {
 	c := Config{}
 	c.Fill()
diff --git a/pkg/logservice/hakeeper_checker_test.go b/pkg/logservice/hakeeper_checker_test.go
index b46c11a41..c9c1df901 100644
--- a/pkg/logservice/hakeeper_checker_test.go
+++ b/pkg/logservice/hakeeper_checker_test.go
@@ -283,7 +283,7 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 		state, err = leaderStore.getCheckerState()
 		require.NoError(t, err)
 		assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
-		assert.Equal(t, uint64(checkBootstrapInterval), leaderStore.bootstrapCheckInterval)
+		assert.Equal(t, uint64(checkBootstrapCycles), leaderStore.bootstrapCheckCycles)
 		require.NotNil(t, leaderStore.bootstrapMgr)
 		assert.False(t, leaderStore.bootstrapMgr.CheckBootstrap(state.LogState))
 
@@ -484,13 +484,13 @@ func testBootstrap(t *testing.T, fail bool) {
 		state, err = store.getCheckerState()
 		require.NoError(t, err)
 		assert.Equal(t, pb.HAKeeperBootstrapCommandsReceived, state.State)
-		assert.Equal(t, uint64(checkBootstrapInterval), store.bootstrapCheckInterval)
+		assert.Equal(t, uint64(checkBootstrapCycles), store.bootstrapCheckCycles)
 		require.NotNil(t, store.bootstrapMgr)
 		assert.False(t, store.bootstrapMgr.CheckBootstrap(state.LogState))
 
 		if fail {
 			// keep checking, bootstrap will eventually be set as failed
-			for i := 0; i <= checkBootstrapInterval; i++ {
+			for i := 0; i <= checkBootstrapCycles; i++ {
 				store.checkBootstrap(state)
 			}