Repository: matrixorigin/matrixone
Pull request #10671 (2023-07-14 12:17:30)

correct the behavior of handle shutdown cmd
## 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 #10658 #10422 #10888 

## What this PR does / why we need it:

when the ha keeper found the log replicas of any shards less than half of the initial setting (num-of-log-shard-replicas), it will issue a shutdown command to log services and dn services to shut down all stores.

however, the shutdown command handler will fall into a deadlock when the log store and dn store received these commands:

heartbeat sends log store and dn store msg to ha keeper, and calls `handleCommands` to handle commands. the shutdown handler then calls service.close which will cancel all ctx and wait tasks exit. but the heartbeat task can detect the cancel signal only when handleCommands is returned.

so, when the shutdown command is generated, the log service and dn service will print these msg respectively
endlessly:
```
 tasks still running in stopper {"stopper": "log-service", "continuous": "3m0.017973416s", "tasks": "log-heartbeat-worker"}

"msg":"tasks still running in stopper","uuid":"dd4dccb4-4d3c-41f8-b482-5251dc7a41bf","stopper":"dn-store","continuous":"2m30.016708291s","tasks":"undefined"
```

the shutdown command needs to shut down all log stores and dn stores because the log shards lose more than half of the replicas, which cannot continue providing log service.



Patch:
diff --git a/cmd/mo-service/launch.go b/cmd/mo-service/launch.go
index ad06bff8c..4fec3cc0d 100644
--- a/cmd/mo-service/launch.go
+++ b/cmd/mo-service/launch.go
@@ -34,7 +34,12 @@ var (
 	cnProxy goetty.Proxy
 )
 
-func startCluster(ctx context.Context, stopper *stopper.Stopper, perfCounterSet *perfcounter.CounterSet) error {
+func startCluster(
+	ctx context.Context,
+	stopper *stopper.Stopper,
+	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
+) error {
 	if *launchFile == "" {
 		panic("launch file not set")
 	}
@@ -44,17 +49,17 @@ func startCluster(ctx context.Context, stopper *stopper.Stopper, perfCounterSet
 		return err
 	}
 
-	if err := startLogServiceCluster(ctx, cfg.LogServiceConfigFiles, stopper, perfCounterSet); err != nil {
+	if err := startLogServiceCluster(ctx, cfg.LogServiceConfigFiles, stopper, perfCounterSet, shutdownC); err != nil {
 		return err
 	}
-	if err := startDNServiceCluster(ctx, cfg.DNServiceConfigsFiles, stopper, perfCounterSet); err != nil {
+	if err := startDNServiceCluster(ctx, cfg.DNServiceConfigsFiles, stopper, perfCounterSet, shutdownC); err != nil {
 		return err
 	}
-	if err := startCNServiceCluster(ctx, cfg.CNServiceConfigsFiles, stopper, perfCounterSet); err != nil {
+	if err := startCNServiceCluster(ctx, cfg.CNServiceConfigsFiles, stopper, perfCounterSet, shutdownC); err != nil {
 		return err
 	}
 	if *withProxy {
-		if err := startProxyServiceCluster(ctx, cfg.ProxyServiceConfigsFiles, stopper, perfCounterSet); err != nil {
+		if err := startProxyServiceCluster(ctx, cfg.ProxyServiceConfigsFiles, stopper, perfCounterSet, shutdownC); err != nil {
 			return err
 		}
 	}
@@ -66,6 +71,7 @@ func startLogServiceCluster(
 	files []string,
 	stopper *stopper.Stopper,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	if len(files) == 0 {
 		return moerr.NewBadConfig(context.Background(), "Log service config not set")
@@ -77,7 +83,7 @@ func startLogServiceCluster(
 		if err := parseConfigFromFile(file, cfg); err != nil {
 			return err
 		}
-		if err := startService(ctx, cfg, stopper, perfCounterSet); err != nil {
+		if err := startService(ctx, cfg, stopper, perfCounterSet, shutdownC); err != nil {
 			return err
 		}
 	}
@@ -89,6 +95,7 @@ func startDNServiceCluster(
 	files []string,
 	stopper *stopper.Stopper,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	if len(files) == 0 {
 		return moerr.NewBadConfig(context.Background(), "DN service config not set")
@@ -99,7 +106,7 @@ func startDNServiceCluster(
 		if err := parseConfigFromFile(file, cfg); err != nil {
 			return err
 		}
-		if err := startService(ctx, cfg, stopper, perfCounterSet); err != nil {
+		if err := startService(ctx, cfg, stopper, perfCounterSet, shutdownC); err != nil {
 			return nil
 		}
 	}
@@ -111,6 +118,7 @@ func startCNServiceCluster(
 	files []string,
 	stopper *stopper.Stopper,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	if len(files) == 0 {
 		return moerr.NewBadConfig(context.Background(), "CN service config not set")
@@ -125,7 +133,7 @@ func startCNServiceCluster(
 			return err
 		}
 		upstreams = append(upstreams, fmt.Sprintf("127.0.0.1:%d", cfg.getCNServiceConfig().Frontend.Port))
-		if err := startService(ctx, cfg, stopper, perfCounterSet); err != nil {
+		if err := startService(ctx, cfg, stopper, perfCounterSet, shutdownC); err != nil {
 			return err
 		}
 	}
@@ -148,6 +156,7 @@ func startProxyServiceCluster(
 	files []string,
 	stopper *stopper.Stopper,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	if len(files) == 0 {
 		return moerr.NewBadConfig(context.Background(), "Proxy service config not set")
@@ -159,7 +168,7 @@ func startProxyServiceCluster(
 		if err := parseConfigFromFile(file, cfg); err != nil {
 			return err
 		}
-		if err := startService(ctx, cfg, stopper, perfCounterSet); err != nil {
+		if err := startService(ctx, cfg, stopper, perfCounterSet, shutdownC); err != nil {
 			return err
 		}
 	}
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index 8331bb46a..8d8626f0c 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 	"sync"
 	"syscall"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/matrixorigin/matrixone/pkg/cnservice"
@@ -83,10 +84,11 @@ func main() {
 	}
 
 	ctx := context.Background()
+	shutdownC := make(chan struct{})
 
 	stopper := stopper.NewStopper("main", stopper.WithLogger(logutil.GetGlobalLogger()))
 	if *launchFile != "" {
-		if err := startCluster(ctx, stopper, globalCounterSet); err != nil {
+		if err := startCluster(ctx, stopper, globalCounterSet, shutdownC); err != nil {
 			panic(err)
 		}
 	} else if *configFile != "" {
@@ -94,22 +96,33 @@ func main() {
 		if err := parseConfigFromFile(*configFile, cfg); err != nil {
 			panic(fmt.Sprintf("failed to parse config from %s, error: %s", *configFile, err.Error()))
 		}
-		if err := startService(ctx, cfg, stopper, globalCounterSet); err != nil {
+		if err := startService(ctx, cfg, stopper, globalCounterSet, shutdownC); err != nil {
 			panic(err)
 		}
 	} else {
 		panic(errors.New("no configuration specified"))
 	}
 
-	waitSignalToStop(stopper)
+	waitSignalToStop(stopper, shutdownC)
 	logutil.GetGlobalLogger().Info("Shutdown complete")
 }
 
-func waitSignalToStop(stopper *stopper.Stopper) {
+func waitSignalToStop(stopper *stopper.Stopper, shutdownC chan struct{}) {
 	sigchan := make(chan os.Signal, 1)
 	signal.Notify(sigchan, syscall.SIGTERM, syscall.SIGINT)
-	sig := <-sigchan
-	logutil.GetGlobalLogger().Info("Starting shutdown...", zap.String("signal", sig.String()))
+
+	detail := "Starting shutdown..."
+	select {
+	case sig := <-sigchan:
+		detail += "signal: " + sig.String()
+	case <-shutdownC:
+		// waiting, give a chance let all log stores and dn stores to get
+		// shutdown cmd from ha keeper
+		time.Sleep(time.Second * 5)
+		detail += "ha keeper issues shutdown command"
+	}
+
+	logutil.GetGlobalLogger().Info(detail)
 	stopper.Stop()
 	if cnProxy != nil {
 		if err := cnProxy.Stop(); err != nil {
@@ -118,7 +131,13 @@ func waitSignalToStop(stopper *stopper.Stopper) {
 	}
 }
 
-func startService(ctx context.Context, cfg *Config, stopper *stopper.Stopper, globalCounterSet *perfcounter.CounterSet) error {
+func startService(
+	ctx context.Context,
+	cfg *Config,
+	stopper *stopper.Stopper,
+	globalCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
+) error {
 	if err := cfg.validate(); err != nil {
 		return err
 	}
@@ -154,9 +173,9 @@ func startService(ctx context.Context, cfg *Config, stopper *stopper.Stopper, gl
 	case metadata.ServiceType_CN:
 		return startCNService(cfg, stopper, fs, globalCounterSet)
 	case metadata.ServiceType_DN:
-		return startDNService(cfg, stopper, fs, globalCounterSet)
+		return startDNService(cfg, stopper, fs, globalCounterSet, shutdownC)
 	case metadata.ServiceType_LOG:
-		return startLogService(cfg, stopper, fs, globalCounterSet)
+		return startLogService(cfg, stopper, fs, globalCounterSet, shutdownC)
 	case metadata.ServiceType_PROXY:
 		return startProxyService(cfg, stopper)
 	default:
@@ -218,6 +237,7 @@ func startDNService(
 	stopper *stopper.Stopper,
 	fileService fileservice.FileService,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	if err := waitClusterCondition(cfg.HAKeeperClient, waitHAKeeperRunning); err != nil {
 		return err
@@ -237,7 +257,8 @@ func startDNService(
 			perfCounterSet,
 			&c,
 			r,
-			fileService)
+			fileService,
+			shutdownC)
 		if err != nil {
 			panic(err)
 		}
@@ -257,9 +278,11 @@ func startLogService(
 	stopper *stopper.Stopper,
 	fileService fileservice.FileService,
 	perfCounterSet *perfcounter.CounterSet,
+	shutdownC chan struct{},
 ) error {
 	lscfg := cfg.getLogServiceConfig()
 	s, err := logservice.NewService(lscfg, fileService,
+		shutdownC,
 		logservice.WithRuntime(runtime.ProcessLevelRuntime()))
 	if err != nil {
 		panic(err)
diff --git a/pkg/dnservice/store.go b/pkg/dnservice/store.go
index 61658ee9d..f15365ddd 100644
--- a/pkg/dnservice/store.go
+++ b/pkg/dnservice/store.go
@@ -96,6 +96,7 @@ type store struct {
 	ctlservice          ctlservice.CtlService
 	replicas            *sync.Map
 	stopper             *stopper.Stopper
+	shutdownC           chan struct{}
 
 	options struct {
 		logServiceClientFactory func(metadata.DNShard) (logservice.Client, error)
@@ -123,6 +124,7 @@ func NewService(
 	cfg *Config,
 	rt runtime.Runtime,
 	fileService fileservice.FileService,
+	shutdownC chan struct{},
 	opts ...Option) (Service, error) {
 	if err := cfg.Validate(); err != nil {
 		return nil, err
@@ -146,6 +148,7 @@ func NewService(
 		rt:                  rt,
 		fileService:         fileService,
 		metadataFileService: metadataFS,
+		shutdownC:           shutdownC,
 	}
 	for _, opt := range opts {
 		opt(s)
diff --git a/pkg/dnservice/store_heartbeat.go b/pkg/dnservice/store_heartbeat.go
index 4e7d19589..d44509b99 100644
--- a/pkg/dnservice/store_heartbeat.go
+++ b/pkg/dnservice/store_heartbeat.go
@@ -118,7 +118,9 @@ func (s *store) handleRemoveReplica(cmd logservicepb.ScheduleCommand) {
 }
 
 func (s *store) handleShutdownStore(_ logservicepb.ScheduleCommand) {
-	if err := s.Close(); err != nil {
-		s.rt.Logger().Error("failed to shutdown store", zap.Error(err))
+	// notify main routine that have received shutdown cmd
+	select {
+	case s.shutdownC <- struct{}{}:
+	default:
 	}
 }
diff --git a/pkg/hakeeper/checkers/syshealth/check.go b/pkg/hakeeper/checkers/syshealth/check.go
index ca84d418f..edee4f67f 100644
--- a/pkg/hakeeper/checkers/syshealth/check.go
+++ b/pkg/hakeeper/checkers/syshealth/check.go
@@ -15,8 +15,11 @@
 package syshealth
 
 import (
+	"fmt"
+
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -56,6 +59,17 @@ func Check(
 		return nil, sysHealthy
 	}
 
+	detail := "Expired logStore info..."
+	for uuid := range logStores.expired {
+		detail += fmt.Sprintf("store %s replicas: [", uuid)
+		for _, replicaInfo := range logState.Stores[uuid].Replicas {
+			detail += fmt.Sprintf("%d-%d, ", replicaInfo.ShardID, replicaInfo.ReplicaID)
+		}
+		detail += "]; "
+	}
+
+	logutil.GetGlobalLogger().Info(detail)
+
 	// parse all dn stores
 	dnStores := parseDnState(cfg, dnState, currTick)
 
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index 453e83728..db38b5331 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -68,6 +68,7 @@ type Service struct {
 	stopper     *stopper.Stopper
 	haClient    LogHAKeeperClient
 	fileService fileservice.FileService
+	shutdownC   chan struct{}
 
 	options struct {
 		// morpc client would filter remote backend via this
@@ -85,6 +86,7 @@ type Service struct {
 func NewService(
 	cfg Config,
 	fileService fileservice.FileService,
+	shutdownC chan struct{},
 	opts ...Option,
 ) (*Service, error) {
 	cfg.Fill()
@@ -96,6 +98,7 @@ func NewService(
 		cfg:         cfg,
 		stopper:     stopper.NewStopper("log-service"),
 		fileService: fileService,
+		shutdownC:   shutdownC,
 	}
 	for _, opt := range opts {
 		opt(service)
diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go
index a15ed92c7..6994b42d0 100644
--- a/pkg/logservice/service_commands.go
+++ b/pkg/logservice/service_commands.go
@@ -109,8 +109,10 @@ func (s *Service) handleKillZombie(cmd pb.ScheduleCommand) {
 }
 
 func (s *Service) handleShutdownStore(_ pb.ScheduleCommand) {
-	if err := s.Close(); err != nil {
-		s.runtime.Logger().Error("failed to shutdown replica", zap.Error(err))
+	// notify main routine that have received shutdown cmd
+	select {
+	case s.shutdownC <- struct{}{}:
+	default:
 	}
 }
 
diff --git a/pkg/logservice/service_wrap.go b/pkg/logservice/service_wrap.go
index 38238aeec..cb42e26da 100644
--- a/pkg/logservice/service_wrap.go
+++ b/pkg/logservice/service_wrap.go
@@ -28,9 +28,10 @@ type WrappedService struct {
 func NewWrappedService(
 	c Config,
 	fileService fileservice.FileService,
+	shutdownC chan struct{},
 	opts ...Option,
 ) (*WrappedService, error) {
-	svc, err := NewService(c, fileService, opts...)
+	svc, err := NewService(c, fileService, shutdownC, opts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/pb/logservice/logservice.go b/pkg/pb/logservice/logservice.go
index 432ba59a5..77011380c 100644
--- a/pkg/pb/logservice/logservice.go
+++ b/pkg/pb/logservice/logservice.go
@@ -223,9 +223,6 @@ func (s *LogState) updateShards(hb LogStoreHeartbeat) {
 // Do not add CN's StartTaskRunner info to log string, because there has user and password.
 func (m *ScheduleCommand) LogString() string {
 	c := func(s string) string {
-		if len(s) > 6 {
-			return s[:6]
-		}
 		return s
 	}
 


Test patch:
diff --git a/pkg/dnservice/store_test.go b/pkg/dnservice/store_test.go
index 532671023..7491f92a2 100644
--- a/pkg/dnservice/store_test.go
+++ b/pkg/dnservice/store_test.go
@@ -60,6 +60,44 @@ func TestAddReplica(t *testing.T) {
 	})
 }
 
+func TestHandleShutdown(t *testing.T) {
+	fn := func(s *store) {
+		cmd := logservicepb.ScheduleCommand{
+			UUID: s.cfg.UUID,
+			ShutdownStore: &logservicepb.ShutdownStore{
+				StoreID: s.cfg.UUID,
+			},
+			ServiceType: logservicepb.DNService,
+		}
+
+		shutdownC := make(chan struct{})
+		exit := atomic.Bool{}
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer func() {
+				cancel()
+				exit.Store(true)
+			}()
+
+			select {
+			case <-ctx.Done():
+				panic("deadline reached")
+			case <-shutdownC:
+				runtime.DefaultRuntime().Logger().Info("received shutdown command")
+			}
+		}()
+
+		s.shutdownC = shutdownC
+
+		for !exit.Load() {
+			s.handleCommands([]logservicepb.ScheduleCommand{cmd})
+			time.Sleep(time.Millisecond)
+		}
+
+	}
+	runDNStoreTest(t, fn)
+}
+
 func TestStartWithReplicas(t *testing.T) {
 	localFS, err := fileservice.NewMemoryFS(defines.LocalFileServiceName, fileservice.DisabledCacheConfig, nil)
 	assert.NoError(t, err)
@@ -263,7 +301,9 @@ func newTestStore(
 		CounterSet,
 		c,
 		rt,
-		fs, options...)
+		fs,
+		nil,
+		options...)
 	assert.NoError(t, err)
 	return s.(*store)
 }
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 95dbf5fb8..c02b894ec 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -59,6 +59,7 @@ func runClientTest(
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -144,6 +145,7 @@ func TestClientCanBeConnectedByReverseProxy(t *testing.T) {
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/hakeeper_client_test.go b/pkg/logservice/hakeeper_client_test.go
index 6c0dfa203..0ab3e2e1d 100644
--- a/pkg/logservice/hakeeper_client_test.go
+++ b/pkg/logservice/hakeeper_client_test.go
@@ -313,6 +313,7 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	cfg2.Fill()
 	service1, err := NewService(cfg1,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -323,6 +324,7 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	}()
 	service2, err := NewService(cfg2,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index a620f6012..81a696432 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -15,6 +15,8 @@
 package logservice
 
 import (
+	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -49,6 +51,7 @@ func TestBackgroundTickAndHeartbeat(t *testing.T) {
 	cfg.Fill()
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -233,3 +236,41 @@ func checkReplicaCount(s *store, shardID uint64) (int, bool) {
 	}
 	return 0, false
 }
+
+func TestHandleShutdown(t *testing.T) {
+	fn := func(t *testing.T, s *Service) {
+		cmd := pb.ScheduleCommand{
+			UUID: s.ID(),
+			ShutdownStore: &pb.ShutdownStore{
+				StoreID: s.ID(),
+			},
+			ServiceType: pb.LogService,
+		}
+
+		shutdownC := make(chan struct{})
+		exit := atomic.Bool{}
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer func() {
+				cancel()
+				exit.Store(true)
+			}()
+
+			select {
+			case <-ctx.Done():
+				panic("deadline reached")
+			case <-shutdownC:
+				runtime.DefaultRuntime().Logger().Info("received shutdown command")
+			}
+		}()
+
+		s.shutdownC = shutdownC
+
+		for !exit.Load() {
+			s.handleCommands([]pb.ScheduleCommand{cmd})
+			time.Sleep(time.Millisecond)
+		}
+
+	}
+	runServiceTest(t, false, true, fn)
+}
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 54db57120..522614856 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -67,6 +67,7 @@ func runServiceTest(t *testing.T,
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -110,6 +111,7 @@ func TestNewService(t *testing.T) {
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -539,6 +541,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	cfg1.Fill()
 	service1, err := NewService(cfg1,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -552,6 +555,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	assert.NoError(t, service1.store.startReplica(1, 1, peers1, false))
 	service2, err := NewService(cfg2,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -668,6 +672,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 		configs = append(configs, cfg)
 		service, err := NewService(cfg,
 			newFS(),
+			nil,
 			WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 				return true
 			}),
@@ -781,6 +786,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 	time.Sleep(2 * time.Second)
 	service, err := NewService(configs[12],
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 32b1f8bd5..b6e94461c 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -224,6 +224,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg4.Fill()
 	service1, err := NewService(cfg1,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -234,6 +235,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}()
 	service2, err := NewService(cfg2,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -244,6 +246,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}()
 	service3, err := NewService(cfg3,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
@@ -254,6 +257,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}()
 	service4, err := NewService(cfg4,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/logservice/testclient.go b/pkg/logservice/testclient.go
index c6ca70c73..2d52b8439 100644
--- a/pkg/logservice/testclient.go
+++ b/pkg/logservice/testclient.go
@@ -37,6 +37,7 @@ func NewTestService(fs vfs.FS) (*Service, ClientConfig, error) {
 	cfg.Fill()
 	service, err := NewService(cfg,
 		newFS(),
+		nil,
 		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
 			return true
 		}),
diff --git a/pkg/pb/logservice/logservice_test.go b/pkg/pb/logservice/logservice_test.go
index 8dbf5f599..82d5efe21 100644
--- a/pkg/pb/logservice/logservice_test.go
+++ b/pkg/pb/logservice/logservice_test.go
@@ -333,7 +333,7 @@ func TestLogString(t *testing.T) {
 				},
 				ServiceType: LogService,
 			},
-			expected: "L/Start storeA storeA:1:1:0 [1:storeA 2:storeB 3:storeC]",
+			expected: "L/Start storeA storeA:1:1:0 [1:storeA123 2:storeB 3:storeC]",
 		},
 	}
 
diff --git a/pkg/tests/service/dnservice.go b/pkg/tests/service/dnservice.go
index bff918103..f372db6b2 100644
--- a/pkg/tests/service/dnservice.go
+++ b/pkg/tests/service/dnservice.go
@@ -146,7 +146,7 @@ func newDNService(
 	opts dnOptions,
 ) (DNService, error) {
 	CounterSet := new(perfcounter.CounterSet)
-	svc, err := dnservice.NewService(CounterSet, cfg, rt, fs, opts...)
+	svc, err := dnservice.NewService(CounterSet, cfg, rt, fs, nil, opts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/tests/service/logservice.go b/pkg/tests/service/logservice.go
index 44fd82d68..20830817d 100644
--- a/pkg/tests/service/logservice.go
+++ b/pkg/tests/service/logservice.go
@@ -147,7 +147,7 @@ func newLogService(
 	fs fileservice.FileService,
 	opts logOptions,
 ) (LogService, error) {
-	svc, err := logservice.NewWrappedService(cfg, fs, opts...)
+	svc, err := logservice.NewWrappedService(cfg, fs, nil, opts...)
 	if err != nil {
 		return nil, err
 	}