Repository: matrixorigin/matrixone
Pull request #4435 (2022-08-09 10:35:28)

logservice: fixed various HAKeeper client connectivity issues
## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:

* use constant connection timeout value which is much smaller than the LOG node heartbeat timeout. this ensures we get chance to try most specified HAKeeper nodes. 
* during bootstrap, it is possible for the HAKeeper replica to be removed by a KILL command, terminate the local bootstrap procedure when that happens. 
* always keep the ticker worker even after the HAKeeper replica is stopped.    


Patch:
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index 4a62d23e6..9b723b23c 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -470,14 +470,14 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 	mf := func() morpc.Message {
 		return pool.Get().(*RPCResponse)
 	}
-	timeout, err := getTimeoutFromContext(ctx)
+	/*timeout, err := getTimeoutFromContext(ctx)
 	if err != nil {
 		return nil, err
-	}
+	}*/
 	codec := morpc.NewMessageCodec(mf, defaultWriteSocketSize)
 	bf := morpc.NewGoettyBasedBackendFactory(codec,
 		morpc.WithBackendConnectWhenCreate(),
-		morpc.WithBackendConnectTimeout(timeout))
+		morpc.WithBackendConnectTimeout(time.Second))
 	return morpc.NewClient(bf,
 		morpc.WithClientInitBackends([]string{target}, []int{1}),
 		morpc.WithClientMaxBackendPerHost(1))
diff --git a/pkg/logservice/service_bootstrap.go b/pkg/logservice/service_bootstrap.go
index 274b56486..e5a4226dc 100644
--- a/pkg/logservice/service_bootstrap.go
+++ b/pkg/logservice/service_bootstrap.go
@@ -51,6 +51,9 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
 		if err := s.store.setInitialClusterInfo(numOfLogShards,
 			numOfDNShards, numOfLogReplicas); err != nil {
 			plog.Errorf("failed to set initial cluster info, %v", err)
+			if err == dragonboat.ErrShardNotFound {
+				return nil
+			}
 			time.Sleep(time.Second)
 			continue
 		}
diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go
index 8987b86c5..5af70991a 100644
--- a/pkg/logservice/service_commands.go
+++ b/pkg/logservice/service_commands.go
@@ -134,7 +134,7 @@ func (s *Service) heartbeatWorker(ctx context.Context) {
 }
 
 func (s *Service) heartbeat(ctx context.Context) {
-	ctx2, cancel := context.WithTimeout(ctx, time.Second)
+	ctx2, cancel := context.WithTimeout(ctx, 3*time.Second)
 	defer cancel()
 
 	if s.haClient == nil {
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 0dd77bff6..54e885bbe 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -222,8 +222,9 @@ func (l *store) startReplica(shardID uint64, replicaID uint64,
 
 func (l *store) stopReplica(shardID uint64, replicaID uint64) error {
 	if shardID == hakeeper.DefaultHAKeeperShardID {
-		plog.Infof("going to stop HAKeeper's ticker")
-		l.tickerStopper.Stop()
+		defer func() {
+			atomic.StoreUint64(&l.haKeeperReplicaID, 0)
+		}()
 	}
 	return l.nh.StopReplica(shardID, replicaID)
 }
@@ -665,10 +666,10 @@ func (l *store) truncationWorker(ctx context.Context) {
 func (l *store) isLeaderHAKeeper() (bool, uint64, error) {
 	leaderID, term, ok, err := l.nh.GetLeaderID(hakeeper.DefaultHAKeeperShardID)
 	if err != nil {
-		plog.Errorf("failed to get HAKeeper Leader ID, %v", err)
 		return false, 0, err
 	}
-	return ok && leaderID == l.haKeeperReplicaID, term, nil
+	replicaID := atomic.LoadUint64(&l.haKeeperReplicaID)
+	return ok && replicaID != 0 && leaderID == replicaID, term, nil
 }
 
 // TODO: add test for this


Test patch:
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 585ee8be1..18e9eeddc 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -17,6 +17,7 @@ package logservice
 import (
 	"context"
 	"math"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -498,14 +499,14 @@ func TestRemoveReplica(t *testing.T) {
 	}
 }
 
-func TestStopReplicaCanStopHAKeeperTicker(t *testing.T) {
+func TestStopReplicaCanResetHAKeeperReplicaID(t *testing.T) {
 	fn := func(t *testing.T, store *store) {
 		peers := make(map[uint64]dragonboat.Target)
 		peers[1] = store.id()
 		assert.NoError(t, store.startHAKeeperReplica(1, peers, false))
-		assert.Equal(t, int64(1), store.tickerStopper.GetTaskCount())
+		assert.Equal(t, uint64(1), atomic.LoadUint64(&store.haKeeperReplicaID))
 		assert.NoError(t, store.stopReplica(hakeeper.DefaultHAKeeperShardID, 1))
-		assert.Equal(t, int64(0), store.tickerStopper.GetTaskCount())
+		assert.Equal(t, uint64(0), atomic.LoadUint64(&store.haKeeperReplicaID))
 	}
 	runStoreTest(t, fn)
 }