Repository: matrixorigin/matrixone
Pull request #4468 (2022-08-11 16:22:26)

logservice: enabled TestGossipInSimulatedCluster
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:

Enabled TestGossipInSimulatedCluster to be always executed, it simulates a 48 nodes cluster to test gossip on log shard info. The test is slightly tweaked to check all nodes except one can get expected shard info in a timely manner. 

Updated go.mod to use a private fork of memberlist for now as the upstream version has known data race issue. 


Patch:
diff --git a/go.mod b/go.mod
index 863813adf..c99ccf52a 100644
--- a/go.mod
+++ b/go.mod
@@ -72,7 +72,7 @@ require (
 	github.com/valyala/fastrand v1.1.0 // indirect
 	github.com/valyala/histogram v1.2.0 // indirect
 	go.etcd.io/etcd/raft/v3 v3.5.1 // indirect
-	golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664 // indirect
+	golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect
 )
 
 require (
@@ -120,6 +120,10 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
+// required until memberlist issue 272 is resolved
+// see https://github.com/hashicorp/memberlist/pull/273 for progress
+replace github.com/hashicorp/memberlist => github.com/lni/memberlist v0.3.2-0.20220811084252-8a89d1ea99cb
+
 replace go.etcd.io/etcd/raft/v3 => github.com/matrixorigin/etcd/raft/v3 v3.5.1-0.20210824022435-0203115049c2
 
 replace go.etcd.io/etcd/v3 => github.com/matrixorigin/etcd/v3 v3.5.1-0.20210824022435-0203115049c2
diff --git a/go.sum b/go.sum
index 023ce440f..73b5ee4d5 100644
--- a/go.sum
+++ b/go.sum
@@ -310,9 +310,6 @@ github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
 github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=
-github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
-github.com/hashicorp/memberlist v0.3.1 h1:MXgUXLqva1QvpVEDQW1IQLG0wivQAtmFlHRQ+1vWZfM=
-github.com/hashicorp/memberlist v0.3.1/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
 github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/hydrogen18/memlistener v0.0.0-20141126152155-54553eb933fb/go.mod h1:qEIFzExnS6016fRpRfxrExeVn2gbClQA99gQhnIcdhE=
@@ -386,6 +383,8 @@ github.com/lni/dragonboat/v4 v4.0.0-20220803152440-a83f853de8b1/go.mod h1:bR6ZGo
 github.com/lni/goutils v1.3.0/go.mod h1:PUPtBAnZlRPUKWUXCsYkIRWubJbtNHpTAee0sczhlf4=
 github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4 h1:6gzI38ZJmbzZ7oZebXz6jII0uVK+MZ3+ds+7mIt1ioI=
 github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4/go.mod h1:LIHvF0fflR+zyXUQFQOiHPpKANf3UIr7DFIv5CBPOoU=
+github.com/lni/memberlist v0.3.2-0.20220811084252-8a89d1ea99cb h1:X1Id1MHzgBNoqNDiOlzsgxjN7CDGzJFzvXPQqR1/Lbc=
+github.com/lni/memberlist v0.3.2-0.20220811084252-8a89d1ea99cb/go.mod h1:yvyXLpo0QaGE59Y7hDTsTzDD25JYBZ4mHgHUZ8lrOI0=
 github.com/lni/vfs v0.2.1-0.20220408085249-8be85be1c3c1/go.mod h1:LOatfyR8Xeej1jbXybwYGVfCccR0u+BQRG9xg7BD7xo=
 github.com/lni/vfs v0.2.1-0.20220616104132-8852fd867376 h1:jX9CoRWNPwrZ2yY3RJFTSwa49qDQqtXglrCByGdQGZg=
 github.com/lni/vfs v0.2.1-0.20220616104132-8852fd867376/go.mod h1:LOatfyR8Xeej1jbXybwYGVfCccR0u+BQRG9xg7BD7xo=
@@ -788,8 +787,8 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664 h1:wEZYwx+kK+KlZ0hpvP2Ls1Xr4+RWnlzGFwPP0aiDjIU=
-golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 h1:WIoqL4EROvwiPdUtaip4VcDdpZ4kha7wBWZrbVKCIZg=
+golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index 36e37f7bc..d9dfb13fe 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -42,7 +42,8 @@ var (
 )
 
 const (
-	defaultHeartbeatInterval = time.Second
+	defaultGossipProbeInterval = 50 * time.Millisecond
+	defaultHeartbeatInterval   = time.Second
 )
 
 // Config defines the Configurations supported by the Log Service.
@@ -78,6 +79,8 @@ type Config struct {
 	// GossipSeedAddresses is list of seed addresses that are used for
 	// introducing the local node into the gossip network.
 	GossipSeedAddresses []string `toml:"gossip-seed-addresses"`
+	// GossipProbeInternval how often gossip nodes probe each other.
+	GossipProbeInterval toml.Duration `toml:"gossip-probe-interval"`
 	// HeartbeatInterval is the interval of how often log service node should be
 	// sending heartbeat message to the HAKeeper.
 	HeartbeatInterval toml.Duration `toml:"logservice-heartbeat-interval"`
@@ -242,6 +245,9 @@ func (c *Config) Validate() error {
 	if c.HAKeeperConfig.DNStoreTimeout.Duration == 0 {
 		return errors.Wrapf(ErrInvalidConfig, "DNStoreTimeout not set")
 	}
+	if c.GossipProbeInterval.Duration == 0 {
+		return errors.Wrapf(ErrInvalidConfig, "GossipProbeInterval not set")
+	}
 	// validate BootstrapConfig
 	if c.BootstrapConfig.BootstrapCluster {
 		if c.BootstrapConfig.NumOfLogShards == 0 {
@@ -317,7 +323,9 @@ func (c *Config) Fill() {
 	if c.HAKeeperCheckInterval.Duration == 0 {
 		c.HAKeeperCheckInterval.Duration = hakeeper.CheckDuration
 	}
-
+	if c.GossipProbeInterval.Duration == 0 {
+		c.GossipProbeInterval.Duration = defaultGossipProbeInterval
+	}
 }
 
 // HAKeeperClientConfig is the config for HAKeeper clients.
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 18aefcbd6..9baeadcb3 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -69,6 +69,9 @@ func getNodeHostConfig(cfg Config) config.NodeHostConfig {
 	meta := storeMeta{
 		serviceAddress: cfg.ServiceAddress,
 	}
+	if cfg.GossipProbeInterval.Duration == 0 {
+		panic("cfg.GossipProbeInterval.Duration is 0")
+	}
 	return config.NodeHostConfig{
 		DeploymentID:        cfg.DeploymentID,
 		NodeHostID:          cfg.UUID,
@@ -82,7 +85,7 @@ func getNodeHostConfig(cfg Config) config.NodeHostConfig {
 			LogDBFactory: tee.TanPebbleLogDBFactory,
 			// FIXME: dragonboat need to be updated to make this field a first class
 			// citizen
-			TestGossipProbeInterval: 50 * time.Millisecond,
+			TestGossipProbeInterval: cfg.GossipProbeInterval.Duration,
 		},
 		Gossip: config.GossipConfig{
 			BindAddress:      cfg.GossipListenAddress,


Test patch:
diff --git a/pkg/logservice/config_test.go b/pkg/logservice/config_test.go
index f69f172a6..1054a6566 100644
--- a/pkg/logservice/config_test.go
+++ b/pkg/logservice/config_test.go
@@ -125,6 +125,10 @@ func TestConfigCanBeValidated(t *testing.T) {
 	c5 := c
 	c5.GossipSeedAddresses = []string{}
 	assert.True(t, errors.Is(c5.Validate(), ErrInvalidConfig))
+
+	c6 := c
+	c6.GossipProbeInterval.Duration = 0
+	assert.True(t, errors.Is(c6.Validate(), ErrInvalidConfig))
 }
 
 func TestBootstrapConfigCanBeValidated(t *testing.T) {
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index d20866ed8..1ea21392b 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -17,7 +17,6 @@ package logservice
 import (
 	"context"
 	"fmt"
-	"os"
 	"sync"
 	"testing"
 	"time"
@@ -594,28 +593,38 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	assert.True(t, done)
 }
 
-func TestGossipConvergeDelay(t *testing.T) {
-	if os.Getenv("LONG_TEST") == "" {
-		// this test will fail on go1.18 when -race is enabled as it requires more
-		// than 8128 goroutines. it works fine on go1.19 beta1.
-		t.Skip("Skipping long test")
-	}
+func TestGossipInSimulatedCluster(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	// start all services
+	nodeCount := 24
+	shardCount := nodeCount / 3
 	configs := make([]Config, 0)
 	services := make([]*Service, 0)
-	for i := 0; i < 48; i++ {
+	for i := 0; i < nodeCount; i++ {
 		cfg := Config{
-			FS:                  vfs.NewStrictMem(),
-			DeploymentID:        1,
-			RTTMillisecond:      5,
-			DataDir:             fmt.Sprintf("data-%d", i),
-			ServiceAddress:      fmt.Sprintf("127.0.0.1:%d", 6000+10*i),
-			RaftAddress:         fmt.Sprintf("127.0.0.1:%d", 6000+10*i+1),
-			GossipAddress:       fmt.Sprintf("127.0.0.1:%d", 6000+10*i+2),
-			GossipSeedAddresses: []string{"127.0.0.1:6002", "127.0.0.1:6012"},
-			DisableWorkers:      true,
+			FS:             vfs.NewStrictMem(),
+			UUID:           uuid.New().String(),
+			DeploymentID:   1,
+			RTTMillisecond: 200,
+			DataDir:        fmt.Sprintf("data-%d", i),
+			ServiceAddress: fmt.Sprintf("127.0.0.1:%d", 6000+10*i),
+			RaftAddress:    fmt.Sprintf("127.0.0.1:%d", 6000+10*i+1),
+			GossipAddress:  fmt.Sprintf("127.0.0.1:%d", 6000+10*i+2),
+			GossipSeedAddresses: []string{
+				"127.0.0.1:6002",
+				"127.0.0.1:6012",
+				"127.0.0.1:6022",
+				"127.0.0.1:6032",
+				"127.0.0.1:6042",
+				"127.0.0.1:6052",
+				"127.0.0.1:6062",
+				"127.0.0.1:6072",
+				"127.0.0.1:6082",
+				"127.0.0.1:6092",
+			},
+			DisableWorkers: true,
 		}
+		cfg.GossipProbeInterval.Duration = 350 * time.Millisecond
 		configs = append(configs, cfg)
 		service, err := NewService(cfg)
 		require.NoError(t, err)
@@ -640,7 +649,7 @@ func TestGossipConvergeDelay(t *testing.T) {
 	// start all replicas
 	// shardID: [1, 16]
 	id := uint64(100)
-	for i := uint64(0); i < 16; i++ {
+	for i := uint64(0); i < uint64(shardCount); i++ {
 		shardID := i + 1
 		r1 := id
 		r2 := id + 1
@@ -655,50 +664,65 @@ func TestGossipConvergeDelay(t *testing.T) {
 		require.NoError(t, services[i*3+2].store.startReplica(shardID, r3, replicas, false))
 	}
 	wait := func() {
-		time.Sleep(10 * time.Millisecond)
+		time.Sleep(50 * time.Millisecond)
 	}
 	// check & wait all leaders to be elected and known to all services
 	cci := uint64(0)
-	for retry := 0; retry < 500; retry++ {
-		done := true
-		for i := 0; i < 48; i++ {
+	iterations := 1000
+	for retry := 0; retry < iterations; retry++ {
+		notReady := 0
+		for i := 0; i < nodeCount; i++ {
 			shardID := uint64(i/3 + 1)
 			service := services[i]
 			info, ok := service.getShardInfo(shardID)
 			if !ok || info.LeaderID == 0 {
-				done = false
+				notReady++
 				wait()
-				break
+				continue
 			}
-			if shardID == 1 {
+			if shardID == 1 && info.Epoch != 0 {
 				cci = info.Epoch
 			}
 		}
-		if done {
+		if notReady <= 1 {
 			break
 		}
-		require.True(t, retry < 499)
+		require.True(t, retry < iterations-1)
 	}
 	require.True(t, cci != 0)
 	// all good now, add a replica to shard 1
 	id += 1
-	require.NoError(t, services[0].store.addReplica(1, id, services[3].ID(), cci))
+
+	for i := 0; i < iterations; i++ {
+		err := services[0].store.addReplica(1, id, services[3].ID(), cci)
+		if err == nil {
+			break
+		} else if err == dragonboat.ErrTimeout || err == dragonboat.ErrSystemBusy ||
+			err == dragonboat.ErrInvalidDeadline {
+			wait()
+			continue
+		} else if err == dragonboat.ErrRejected {
+			break
+		}
+		t.Fatalf("failed to add replica, %v", err)
+	}
+
 	// check the above change can be observed by all services
-	for retry := 0; retry < 500; retry++ {
-		done := true
-		for i := 0; i < 48; i++ {
+	for retry := 0; retry < iterations; retry++ {
+		notReady := 0
+		for i := 0; i < nodeCount; i++ {
 			service := services[i]
 			info, ok := service.getShardInfo(1)
 			if !ok || info.LeaderID == 0 || len(info.Replicas) != 4 {
-				done = false
+				notReady++
 				wait()
-				break
+				continue
 			}
 		}
-		if done {
+		if notReady <= 1 {
 			break
 		}
-		require.True(t, retry < 499)
+		require.True(t, retry < iterations-1)
 	}
 	// restart a service, watch how long will it take to get all required
 	// shard info
@@ -710,20 +734,20 @@ func TestGossipConvergeDelay(t *testing.T) {
 	defer func() {
 		require.NoError(t, service.Close())
 	}()
-	for retry := 0; retry < 500; retry++ {
-		done := true
-		for i := uint64(0); i < 16; i++ {
+	for retry := 0; retry < iterations; retry++ {
+		notReady := 0
+		for i := uint64(0); i < uint64(shardCount); i++ {
 			shardID := i + 1
 			info, ok := service.getShardInfo(shardID)
 			if !ok || info.LeaderID == 0 {
-				done = false
+				notReady++
 				wait()
-				break
+				continue
 			}
 		}
-		if done {
+		if notReady <= 1 {
 			break
 		}
-		require.True(t, retry < 499)
+		require.True(t, retry < iterations-1)
 	}
 }
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 18e9eeddc..9100d59c3 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -42,6 +42,7 @@ func TestNodeHostConfig(t *testing.T) {
 		DeploymentID: 1234,
 		DataDir:      "lalala",
 	}
+	cfg.Fill()
 	nhConfig := getNodeHostConfig(cfg)
 	assert.Equal(t, cfg.DeploymentID, nhConfig.DeploymentID)
 	assert.Equal(t, cfg.DataDir, nhConfig.NodeHostDir)