Repository: matrixorigin/matrixone
Pull request #5375 (2022-09-30 06:58:27)

LogService: modify some log level from info to debug
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [x] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:


Patch:
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index c5121267b..335756dc9 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -171,7 +171,7 @@ func startDNService(
 		s, err := dnservice.NewService(
 			&c,
 			fileService,
-			dnservice.WithLogger(logutil.GetGlobalLogger().Named("dn-service")))
+			dnservice.WithLogger(logutil.GetGlobalLogger().Named("dn-service").With(zap.String("uuid", cfg.DN.UUID))))
 		if err != nil {
 			panic(err)
 		}
diff --git a/pkg/cnservice/server.go b/pkg/cnservice/server.go
index 495d99953..d84f8e8ff 100644
--- a/pkg/cnservice/server.go
+++ b/pkg/cnservice/server.go
@@ -17,6 +17,7 @@ package cnservice
 import (
 	"context"
 	"fmt"
+	"go.uber.org/zap"
 	"sync"
 
 	"github.com/fagongzi/goetty/v2"
@@ -62,7 +63,7 @@ func NewService(
 	}
 
 	srv := &service{
-		logger: logutil.GetGlobalLogger().Named("cnservice"),
+		logger: logutil.GetGlobalLogger().Named("cnservice").With(zap.String("uuid", cfg.UUID)),
 		metadata: metadata.CNStore{
 			UUID: cfg.UUID,
 			Role: metadata.MustParseCNRole(cfg.Role),
diff --git a/pkg/hakeeper/checkers/coordinator.go b/pkg/hakeeper/checkers/coordinator.go
index 753245167..8fc03f4f5 100644
--- a/pkg/hakeeper/checkers/coordinator.go
+++ b/pkg/hakeeper/checkers/coordinator.go
@@ -21,7 +21,9 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/syshealth"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"go.uber.org/zap"
 )
 
 // Coordinator is assumed to be used in synchronous, single-threaded context.
@@ -33,7 +35,8 @@ type Coordinator struct {
 	teardown    bool
 	teardownOps []*operator.Operator
 
-	cfg hakeeper.Config
+	cfg    hakeeper.Config
+	logger *zap.Logger
 }
 
 func NewCoordinator(cfg hakeeper.Config) *Coordinator {
@@ -41,6 +44,7 @@ func NewCoordinator(cfg hakeeper.Config) *Coordinator {
 	return &Coordinator{
 		OperatorController: operator.NewController(),
 		cfg:                cfg,
+		logger:             logutil.GetGlobalLogger().Named("hakeeper"),
 	}
 }
 
@@ -66,7 +70,7 @@ func (c *Coordinator) Check(alloc util.IDAllocator, cluster pb.ClusterInfo,
 
 	operators := make([]*operator.Operator, 0)
 	operators = append(operators, logservice.Check(alloc, c.cfg, cluster, logState, executing, currentTick)...)
-	operators = append(operators, dnservice.Check(alloc, c.cfg, cluster, dnState, currentTick)...)
+	operators = append(operators, dnservice.Check(alloc, c.cfg, cluster, dnState, currentTick, c.logger)...)
 
 	return c.OperatorController.Dispatch(operators, logState, dnState)
 }
diff --git a/pkg/hakeeper/checkers/dnservice/check.go b/pkg/hakeeper/checkers/dnservice/check.go
index 61f7fe39b..4974db897 100644
--- a/pkg/hakeeper/checkers/dnservice/check.go
+++ b/pkg/hakeeper/checkers/dnservice/check.go
@@ -23,16 +23,12 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
-	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
 var (
 	// waitingShards makes check logic stateful.
 	waitingShards *initialShards
-
-	// logger for dnservice checker
-	logger = logutil.GetGlobalLogger().Named("hakeeper")
 )
 
 func init() {
@@ -48,6 +44,7 @@ func Check(
 	cluster pb.ClusterInfo,
 	dnState pb.DNState,
 	currTick uint64,
+	logger *zap.Logger,
 ) []*operator.Operator {
 	logger.Debug("dn checker entrance",
 		zap.Any("cluster information", cluster),
@@ -71,17 +68,12 @@ func Check(
 
 	// 1. check reported dn state
 	operators = append(operators,
-		checkReportedState(
-			reportedShards, mapper, stores.WorkingStores(), idAlloc,
-		)...,
+		checkReportedState(reportedShards, mapper, stores.WorkingStores(), idAlloc, logger)...,
 	)
 
 	// 2. check expected dn state
 	operators = append(operators,
-		checkInitiatingShards(
-			reportedShards, mapper, stores.WorkingStores(), idAlloc,
-			cluster, cfg, currTick,
-		)...,
+		checkInitiatingShards(reportedShards, mapper, stores.WorkingStores(), idAlloc, cluster, cfg, currTick, logger)...,
 	)
 
 	return operators
@@ -90,7 +82,7 @@ func Check(
 // schedule generator operator as much as possible
 // NB: the returned order should be deterministic.
 func checkShard(
-	shard *dnShard, mapper ShardMapper, workingStores []*util.Store, idAlloc util.IDAllocator,
+	shard *dnShard, mapper ShardMapper, workingStores []*util.Store, idAlloc util.IDAllocator, logger *zap.Logger,
 ) []operator.OpStep {
 	switch len(shard.workingReplicas()) {
 	case 0: // need add replica
diff --git a/pkg/hakeeper/checkers/dnservice/parse.go b/pkg/hakeeper/checkers/dnservice/parse.go
index 05ba2bb2a..71f7965a3 100644
--- a/pkg/hakeeper/checkers/dnservice/parse.go
+++ b/pkg/hakeeper/checkers/dnservice/parse.go
@@ -88,7 +88,7 @@ func parseDnState(cfg hakeeper.Config,
 // checkReportedState generates Operators for reported state.
 // NB: the order of list is deterministic.
 func checkReportedState(
-	rs *reportedShards, mapper ShardMapper, workingStores []*util.Store, idAlloc util.IDAllocator,
+	rs *reportedShards, mapper ShardMapper, workingStores []*util.Store, idAlloc util.IDAllocator, logger *zap.Logger,
 ) []*operator.Operator {
 	var ops []*operator.Operator
 
@@ -105,7 +105,7 @@ func checkReportedState(
 			panic(fmt.Sprintf("shard `%d` not register", shardID))
 		}
 
-		steps := checkShard(shard, mapper, workingStores, idAlloc)
+		steps := checkShard(shard, mapper, workingStores, idAlloc, logger)
 		// avoid Operator with nil steps
 		if len(steps) > 0 {
 			ops = append(ops,
@@ -123,10 +123,8 @@ func checkReportedState(
 // NB: the order of list is deterministic.
 func checkInitiatingShards(
 	rs *reportedShards, mapper ShardMapper, workingStores []*util.Store, idAlloc util.IDAllocator,
-	cluster pb.ClusterInfo, cfg hakeeper.Config, currTick uint64,
+	cluster pb.ClusterInfo, cfg hakeeper.Config, currTick uint64, logger *zap.Logger,
 ) []*operator.Operator {
-	logger.Debug("cluster expected information", zap.Any("dn shards expected", cluster.DNShards))
-
 	// update the registered newly-created shards
 	for _, record := range cluster.DNShards {
 		shardID := record.ShardID
@@ -135,31 +133,22 @@ func checkInitiatingShards(
 			if moerr.IsMoErrCode(err, moerr.ErrShardNotReported) {
 				// if a shard not reported, register it,
 				// and launch its replica after a while.
-				logger.Debug("dn shard not reported", zap.Uint64("shard ID", shardID))
 				waitingShards.register(shardID, currTick)
 			}
 			continue
 		}
 		// shard reported via heartbeat, no need to wait
-		logger.Debug("dn shard reported, remove from waiting shards", zap.Uint64("shard ID", shardID))
 		waitingShards.remove(shardID)
 	}
 
-	logger.Debug("cluster initiating shards:", zap.Any("shards", waitingShards.shards))
-
 	// list newly-created shards which had been waiting for a while
 	expired := waitingShards.listEligibleShards(func(start uint64) bool {
-		logger.Debug("check initiating shard expired or not",
-			zap.Uint64("recorded tick", start),
-			zap.Uint64("current tick", currTick),
-			zap.Bool("is expired", cfg.DNStoreExpired(start, currTick)),
-		)
 		return cfg.DNStoreExpired(start, currTick)
 	})
 
 	var ops []*operator.Operator
 	for _, id := range expired {
-		steps := checkShard(newDnShard(id), mapper, workingStores, idAlloc)
+		steps := checkShard(newDnShard(id), mapper, workingStores, idAlloc, logger)
 		if len(steps) > 0 { // avoid Operator with nil steps
 			ops = append(ops,
 				operator.NewOperator("dnservice", id, operator.NoopEpoch, steps...),
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index 6bc44ee62..ec91b80a0 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -26,6 +26,7 @@ import (
 	"github.com/lni/dragonboat/v4"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -222,7 +223,7 @@ func (c *managedClient) resetClient() {
 		cc := c.client
 		c.client = nil
 		if err := cc.close(); err != nil {
-			logger.Error("failed to close client", zap.Error(err))
+			logutil.Error("failed to close client", zap.Error(err))
 		}
 	}
 }
@@ -320,7 +321,7 @@ func connectToLogService(ctx context.Context,
 				return c, nil
 			} else {
 				if err := c.close(); err != nil {
-					logger.Error("failed to close the client", zap.Error(err))
+					logutil.Error("failed to close the client", zap.Error(err))
 				}
 				e = err
 			}
@@ -331,7 +332,7 @@ func connectToLogService(ctx context.Context,
 				return c, nil
 			} else {
 				if err := c.close(); err != nil {
-					logger.Error("failed to close the client", zap.Error(err))
+					logutil.Error("failed to close the client", zap.Error(err))
 				}
 				e = err
 			}
@@ -504,6 +505,7 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 		morpc.WithBackendConnectWhenCreate(),
 		morpc.WithBackendConnectTimeout(time.Second),
 		morpc.WithBackendHasPayloadResponse(),
+		morpc.WithBackendLogger(logutil.GetGlobalLogger().Named("hakeeper-client-backend")),
 	}
 	backendOpts = append(backendOpts, GetBackendOptions(ctx)...)
 
@@ -511,6 +513,7 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 	clientOpts := []morpc.ClientOption{
 		morpc.WithClientInitBackends([]string{target}, []int{1}),
 		morpc.WithClientMaxBackendPerHost(1),
+		morpc.WithClientLogger(logutil.GetGlobalLogger().Named("hakeeper-client")),
 	}
 	clientOpts = append(clientOpts, GetClientOptions(ctx)...)
 
diff --git a/pkg/logservice/errors.go b/pkg/logservice/errors.go
index 16dc146b1..cf69d2ea1 100644
--- a/pkg/logservice/errors.go
+++ b/pkg/logservice/errors.go
@@ -22,6 +22,7 @@ import (
 	"github.com/lni/dragonboat/v4"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -57,7 +58,7 @@ func toErrorCode(err error) (uint32, string) {
 	}
 	for _, rec := range errorToCodeMappings {
 		if errors.Is(err, rec.err) {
-			logger.Error(fmt.Sprintf("error: %v, converted to code %d", err, rec.code))
+			logutil.Error(fmt.Sprintf("error: %v, converted to code %d", err, rec.code))
 			return uint32(rec.code), ""
 		}
 	}
diff --git a/pkg/logservice/hakeeper_client.go b/pkg/logservice/hakeeper_client.go
index 9256f50d1..b24c6c361 100644
--- a/pkg/logservice/hakeeper_client.go
+++ b/pkg/logservice/hakeeper_client.go
@@ -24,6 +24,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -243,7 +244,7 @@ func (c *managedHAKeeperClient) resetClient() {
 		cc := c.client
 		c.client = nil
 		if err := cc.close(); err != nil {
-			logger.Error("failed to close client", zap.Error(err))
+			logutil.Error("failed to close client", zap.Error(err))
 		}
 	}
 }
@@ -343,14 +344,14 @@ func connectToHAKeeper(ctx context.Context,
 		c.addr = addr
 		c.client = cc
 		isHAKeeper, err := c.checkIsHAKeeper(ctx)
-		logger.Info(fmt.Sprintf("isHAKeeper: %t, err: %v", isHAKeeper, err))
+		logutil.Info(fmt.Sprintf("isHAKeeper: %t, err: %v", isHAKeeper, err))
 		if err == nil && isHAKeeper {
 			return c, nil
 		} else if err != nil {
 			e = err
 		}
 		if err := cc.Close(); err != nil {
-			logger.Error("failed to close the client", zap.Error(err))
+			logutil.Error("failed to close the client", zap.Error(err))
 		}
 	}
 	if e == nil {
@@ -423,7 +424,7 @@ func (c *hakeeperClient) sendLogHeartbeat(ctx context.Context,
 		return pb.CommandBatch{}, err
 	}
 	for _, cmd := range cb.Commands {
-		logger.Info("hakeeper client received cmd", zap.String("cmd", cmd.LogString()))
+		logutil.Info("hakeeper client received cmd", zap.String("cmd", cmd.LogString()))
 	}
 	return cb, nil
 }
diff --git a/pkg/logservice/rsm.go b/pkg/logservice/rsm.go
index 29bc9f659..b05693b0a 100644
--- a/pkg/logservice/rsm.go
+++ b/pkg/logservice/rsm.go
@@ -91,7 +91,7 @@ type stateMachine struct {
 	state     pb.RSMState
 }
 
-var _ (sm.IStateMachine) = (*stateMachine)(nil)
+var _ sm.IStateMachine = (*stateMachine)(nil)
 
 func newStateMachine(shardID uint64, replicaID uint64) sm.IStateMachine {
 	state := pb.RSMState{
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index d3fa6df23..227336bf9 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -35,10 +35,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
-var (
-	logger = logutil.GetGlobalLogger().Named("LogService")
-)
-
 const (
 	LogServiceRPCName = "logservice-rpc"
 )
@@ -74,6 +70,8 @@ type Service struct {
 		// morpc client would filter remote backend via this
 		backendFilter func(msg morpc.Message, backendAddr string) bool
 	}
+
+	logger *zap.Logger
 }
 
 func NewService(
@@ -86,7 +84,8 @@ func NewService(
 	if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
-	store, err := newLogStore(cfg, taskService)
+	logger := logutil.GetGlobalLogger().Named("LogService").With(zap.String("uuid", cfg.UUID))
+	store, err := newLogStore(cfg, taskService, logger)
 	if err != nil {
 		logger.Error("failed to create log store", zap.Error(err))
 		return nil, err
@@ -115,7 +114,9 @@ func NewService(
 	server, err := morpc.NewRPCServer(LogServiceRPCName, cfg.ServiceListenAddress, codec,
 		morpc.WithServerGoettyOptions(goetty.WithSessionReleaseMsgFunc(func(i interface{}) {
 			respPool.Put(i.(morpc.RPCMessage).Message)
-		})))
+		})),
+		morpc.WithServerLogger(logger),
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -127,6 +128,8 @@ func NewService(
 		respPool:    respPool,
 		stopper:     stopper.NewStopper("log-service"),
 		fileService: fileService,
+
+		logger: logger,
 	}
 	for _, opt := range opts {
 		opt(service)
diff --git a/pkg/logservice/service_bootstrap.go b/pkg/logservice/service_bootstrap.go
index 3eb2be4d9..565dbb109 100644
--- a/pkg/logservice/service_bootstrap.go
+++ b/pkg/logservice/service_bootstrap.go
@@ -36,7 +36,7 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
 		// running as a result of store.startReplicas(), we just ignore the
 		// dragonboat.ErrShardAlreadyExist error below.
 		if err != dragonboat.ErrShardAlreadyExist {
-			logger.Error("failed to start hakeeper replica", zap.Error(err))
+			s.logger.Error("failed to start hakeeper replica", zap.Error(err))
 			return err
 		}
 	}
@@ -51,14 +51,14 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
 		}
 		if err := s.store.setInitialClusterInfo(numOfLogShards,
 			numOfDNShards, numOfLogReplicas); err != nil {
-			logger.Error("failed to set initial cluster info", zap.Error(err))
+			s.logger.Error("failed to set initial cluster info", zap.Error(err))
 			if err == dragonboat.ErrShardNotFound {
 				return nil
 			}
 			time.Sleep(time.Second)
 			continue
 		}
-		logger.Info("initial cluster info set")
+		s.logger.Info("initial cluster info set")
 		break
 	}
 	return nil
diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go
index 42b16aa80..dbdbd848d 100644
--- a/pkg/logservice/service_commands.go
+++ b/pkg/logservice/service_commands.go
@@ -27,9 +27,9 @@ import (
 
 func (s *Service) handleCommands(cmds []pb.ScheduleCommand) {
 	for _, cmd := range cmds {
-		logger.Info(fmt.Sprintf("%s applying cmd: %s", s.ID(), cmd.LogString()))
+		s.logger.Info(fmt.Sprintf("%s applying cmd: %s", s.ID(), cmd.LogString()))
 		if cmd.GetConfigChange() != nil {
-			logger.Info("applying schedule command:", zap.String("command", cmd.LogString()))
+			s.logger.Debug("applying schedule command:", zap.String("command", cmd.LogString()))
 			switch cmd.ConfigChange.ChangeType {
 			case pb.AddReplica:
 				s.handleAddReplica(cmd)
@@ -58,7 +58,7 @@ func (s *Service) handleAddReplica(cmd pb.ScheduleCommand) {
 	epoch := cmd.ConfigChange.Replica.Epoch
 	target := cmd.ConfigChange.Replica.UUID
 	if err := s.store.addReplica(shardID, replicaID, target, epoch); err != nil {
-		logger.Error("failed to add replica", zap.Error(err))
+		s.logger.Error("failed to add replica", zap.Error(err))
 	}
 }
 
@@ -67,7 +67,7 @@ func (s *Service) handleRemoveReplica(cmd pb.ScheduleCommand) {
 	replicaID := cmd.ConfigChange.Replica.ReplicaID
 	epoch := cmd.ConfigChange.Replica.Epoch
 	if err := s.store.removeReplica(shardID, replicaID, epoch); err != nil {
-		logger.Error("failed to remove replica", zap.Error(err))
+		s.logger.Error("failed to remove replica", zap.Error(err))
 	}
 }
 
@@ -78,12 +78,12 @@ func (s *Service) handleStartReplica(cmd pb.ScheduleCommand) {
 	if shardID == hakeeper.DefaultHAKeeperShardID {
 		if err := s.store.startHAKeeperReplica(replicaID,
 			cmd.ConfigChange.InitialMembers, join); err != nil {
-			logger.Error("failed to start HAKeeper replica", zap.Error(err))
+			s.logger.Error("failed to start HAKeeper replica", zap.Error(err))
 		}
 	} else {
 		if err := s.store.startReplica(shardID,
 			replicaID, cmd.ConfigChange.InitialMembers, join); err != nil {
-			logger.Error("failed to start log replica", zap.Error(err))
+			s.logger.Error("failed to start log replica", zap.Error(err))
 		}
 	}
 }
@@ -92,7 +92,7 @@ func (s *Service) handleStopReplica(cmd pb.ScheduleCommand) {
 	shardID := cmd.ConfigChange.Replica.ShardID
 	replicaID := cmd.ConfigChange.Replica.ReplicaID
 	if err := s.store.stopReplica(shardID, replicaID); err != nil {
-		logger.Error("failed to stop replica", zap.Error(err))
+		s.logger.Error("failed to stop replica", zap.Error(err))
 	}
 }
 
@@ -113,7 +113,7 @@ func (s *Service) heartbeatWorker(ctx context.Context) {
 		panic("invalid heartbeat interval")
 	}
 	defer func() {
-		logger.Info("heartbeat worker stopped")
+		s.logger.Info("heartbeat worker stopped")
 	}()
 	ticker := time.NewTicker(s.cfg.HeartbeatInterval.Duration)
 	defer ticker.Stop()
@@ -145,7 +145,7 @@ func (s *Service) heartbeat(ctx context.Context) {
 		}
 		cc, err := NewLogHAKeeperClient(ctx2, s.cfg.GetHAKeeperClientConfig())
 		if err != nil {
-			logger.Error("failed to create HAKeeper client", zap.Error(err))
+			s.logger.Error("failed to create HAKeeper client", zap.Error(err))
 			return
 		}
 		s.haClient = cc
@@ -154,7 +154,7 @@ func (s *Service) heartbeat(ctx context.Context) {
 	hb := s.store.getHeartbeatMessage()
 	cb, err := s.haClient.SendLogHeartbeat(ctx2, hb)
 	if err != nil {
-		logger.Error("failed to send log service heartbeat", zap.Error(err))
+		s.logger.Error("failed to send log service heartbeat", zap.Error(err))
 		return
 	}
 	s.handleCommands(cb.Commands)
diff --git a/pkg/logservice/shardinfo.go b/pkg/logservice/shardinfo.go
index 7727ed0c8..452488dbd 100644
--- a/pkg/logservice/shardinfo.go
+++ b/pkg/logservice/shardinfo.go
@@ -16,6 +16,7 @@ package logservice
 
 import (
 	"context"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"go.uber.org/zap"
 	"reflect"
 	"sync"
@@ -48,7 +49,7 @@ func GetShardInfo(address string, shardID uint64) (ShardInfo, bool, error) {
 	}
 	defer func() {
 		if err := cc.Close(); err != nil {
-			logger.Error("failed to close client", zap.Error(err))
+			logutil.Error("failed to close client", zap.Error(err))
 		}
 	}()
 	req := pb.Request{
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 2190e8c40..eab141da6 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -37,6 +37,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/bootstrap"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/task"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
@@ -74,7 +75,7 @@ func getNodeHostConfig(cfg Config) config.NodeHostConfig {
 	logdbFactory := (config.LogDBFactory)(nil)
 	logdbFactory = tan.Factory
 	if cfg.UseTeeLogDB {
-		logger.Warn("using tee based logdb backed by pebble and tan, for testing purposes only")
+		logutil.Warn("using tee based logdb backed by pebble and tan, for testing purposes only")
 		logdbFactory = tee.TanPebbleLogDBFactory
 	}
 	return config.NodeHostConfig{
@@ -123,6 +124,7 @@ type store struct {
 	alloc             hakeeper.IDAllocator
 	stopper           *stopper.Stopper
 	tickerStopper     *stopper.Stopper
+	logger            *zap.Logger
 
 	bootstrapCheckCycles uint64
 	bootstrapMgr         *bootstrap.Manager
@@ -137,12 +139,13 @@ type store struct {
 	}
 }
 
-func newLogStore(cfg Config, taskService taskservice.TaskService) (*store, error) {
+func newLogStore(cfg Config, taskService taskservice.TaskService, logger *zap.Logger) (*store, error) {
 	nh, err := dragonboat.NewNodeHost(getNodeHostConfig(cfg))
 	if err != nil {
 		return nil, err
 	}
 	hakeeperConfig := cfg.GetHAKeeperConfig()
+	logger = logutil.Adjust(logger)
 	logger.Info("HAKeeper Timeout Configs",
 		zap.Int64("LogStoreTimeout", int64(hakeeperConfig.LogStoreTimeout)),
 		zap.Int64("DNStoreTimeout", int64(hakeeperConfig.DNStoreTimeout)),
@@ -156,6 +159,7 @@ func newLogStore(cfg Config, taskService taskservice.TaskService) (*store, error
 		alloc:         newIDAllocator(),
 		stopper:       stopper.NewStopper("log-store"),
 		tickerStopper: stopper.NewStopper("hakeeper-ticker"),
+		logger:        logger,
 	}
 	ls.mu.truncateCh = make(chan struct{})
 	ls.mu.pendingTruncate = make(map[uint64]struct{})
@@ -213,7 +217,7 @@ func (l *store) startHAKeeperReplica(replicaID uint64,
 	atomic.StoreUint64(&l.haKeeperReplicaID, replicaID)
 	if !l.cfg.DisableWorkers {
 		if err := l.tickerStopper.RunNamedTask("hakeeper-ticker", func(ctx context.Context) {
-			logger.Info("HAKeeper ticker started")
+			l.logger.Info("HAKeeper ticker started")
 			l.ticker(ctx)
 		}); err != nil {
 			return err
@@ -352,11 +356,11 @@ func (l *store) truncateLog(ctx context.Context,
 	cmd := getSetTruncatedLsnCmd(index)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("propose truncate log cmd failed", zap.Error(err))
+		l.logger.Error("propose truncate log cmd failed", zap.Error(err))
 		return err
 	}
 	if result.Value > 0 {
-		logger.Error(fmt.Sprintf("shardID %d already truncated to index %d", shardID, result.Value))
+		l.logger.Error(fmt.Sprintf("shardID %d already truncated to index %d", shardID, result.Value))
 		return moerr.NewInvalidTruncateLsn(shardID, result.Value)
 	}
 	l.mu.Lock()
@@ -371,11 +375,11 @@ func (l *store) append(ctx context.Context,
 	session := l.nh.GetNoOPSession(shardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("propose failed", zap.Error(err))
+		l.logger.Error("propose failed", zap.Error(err))
 		return 0, err
 	}
 	if len(result.Data) > 0 {
-		logger.Error("not current lease holder", zap.Uint64("data", binaryEnc.Uint64(result.Data)))
+		l.logger.Error("not current lease holder", zap.Uint64("data", binaryEnc.Uint64(result.Data)))
 		return 0, moerr.NewNotLeaseHolder(binaryEnc.Uint64(result.Data))
 	}
 	if result.Value == 0 {
@@ -398,7 +402,7 @@ func (l *store) tsoUpdate(ctx context.Context, count uint64) (uint64, error) {
 	session := l.nh.GetNoOPSession(firstLogShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("failed to propose tso updat", zap.Error(err))
+		l.logger.Error("failed to propose tso updat", zap.Error(err))
 		return 0, err
 	}
 	return result.Value, nil
@@ -420,7 +424,7 @@ func (l *store) addLogStoreHeartbeat(ctx context.Context,
 	cmd := hakeeper.GetLogStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	if result, err := l.propose(ctx, session, cmd); err != nil {
-		logger.Error("propose failed", zap.Error(err))
+		l.logger.Error("propose failed", zap.Error(err))
 		return pb.CommandBatch{}, handleNotHAKeeperError(err)
 	} else {
 		var cb pb.CommandBatch
@@ -435,7 +439,7 @@ func (l *store) addCNStoreHeartbeat(ctx context.Context,
 	cmd := hakeeper.GetCNStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	if _, err := l.propose(ctx, session, cmd); err != nil {
-		logger.Error("propose failed", zap.Error(err))
+		l.logger.Error("propose failed", zap.Error(err))
 		return handleNotHAKeeperError(err)
 	}
 	return nil
@@ -447,7 +451,7 @@ func (l *store) cnAllocateID(ctx context.Context,
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("propose get id failed", zap.Error(err))
+		l.logger.Error("propose get id failed", zap.Error(err))
 		return 0, err
 	}
 	return result.Value, nil
@@ -459,7 +463,7 @@ func (l *store) addDNStoreHeartbeat(ctx context.Context,
 	cmd := hakeeper.GetDNStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	if result, err := l.propose(ctx, session, cmd); err != nil {
-		logger.Error("propose failed", zap.Error(err))
+		l.logger.Error("propose failed", zap.Error(err))
 		return pb.CommandBatch{}, handleNotHAKeeperError(err)
 	} else {
 		var cb pb.CommandBatch
@@ -509,7 +513,7 @@ func (l *store) getLeaseHolderID(ctx context.Context,
 	}
 	v, err := l.read(ctx, shardID, leaseHistoryQuery{lsn: e.Index})
 	if err != nil {
-		logger.Error("failed to read", zap.Error(err))
+		l.logger.Error("failed to read", zap.Error(err))
 		return 0, err
 	}
 	return v.(uint64), nil
@@ -607,7 +611,7 @@ func (l *store) queryLog(ctx context.Context, shardID uint64,
 	// FIXME: check whether lastIndex >= firstIndex
 	rs, err := l.nh.QueryRaftLog(shardID, firstIndex, lastIndex+1, maxSize)
 	if err != nil {
-		logger.Error("QueryRaftLog failed", zap.Error(err))
+		l.logger.Error("QueryRaftLog failed", zap.Error(err))
 		return nil, 0, err
 	}
 	select {
@@ -617,13 +621,13 @@ func (l *store) queryLog(ctx context.Context, shardID uint64,
 			next := getNextIndex(entries, firstIndex, logRange.LastIndex)
 			results, err := l.markEntries(ctx, shardID, entries)
 			if err != nil {
-				logger.Error("markEntries failed", zap.Error(err))
+				l.logger.Error("markEntries failed", zap.Error(err))
 				return nil, 0, err
 			}
 			return results, next, nil
 		} else if v.RequestOutOfRange() {
 			// FIXME: add more details to the log, what is the available range
-			logger.Error("OutOfRange query found")
+			l.logger.Error("OutOfRange query found")
 			return nil, 0, dragonboat.ErrInvalidRange
 		}
 		panic(moerr.NewInvalidState("unexpected rs state"))
@@ -636,7 +640,7 @@ func (l *store) ticker(ctx context.Context) {
 	if l.cfg.HAKeeperTickInterval.Duration == 0 {
 		panic("invalid HAKeeperTickInterval")
 	}
-	logger.Info("Hakeeper interval configs",
+	l.logger.Info("Hakeeper interval configs",
 		zap.Int64("HAKeeperTickInterval", int64(l.cfg.HAKeeperTickInterval.Duration)),
 		zap.Int64("HAKeeperCheckInterval", int64(l.cfg.HAKeeperCheckInterval.Duration)))
 	ticker := time.NewTicker(l.cfg.HAKeeperTickInterval.Duration)
@@ -645,7 +649,7 @@ func (l *store) ticker(ctx context.Context) {
 		panic("invalid HAKeeperCheckInterval")
 	}
 	defer func() {
-		logger.Info("HAKeeper ticker stopped")
+		l.logger.Info("HAKeeper ticker stopped")
 	}()
 	haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration)
 	defer haTicker.Stop()
@@ -670,7 +674,7 @@ func (l *store) ticker(ctx context.Context) {
 
 func (l *store) truncationWorker(ctx context.Context) {
 	defer func() {
-		logger.Info("truncation worker stopped")
+		l.logger.Info("truncation worker stopped")
 	}()
 
 	for {
@@ -703,7 +707,7 @@ func (l *store) isLeaderHAKeeper() (bool, uint64, error) {
 func (l *store) hakeeperTick() {
 	isLeader, _, err := l.isLeaderHAKeeper()
 	if err != nil {
-		logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
+		l.logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
 		return
 	}
 
@@ -713,7 +717,7 @@ func (l *store) hakeeperTick() {
 		defer cancel()
 		session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 		if _, err := l.propose(ctx, session, cmd); err != nil {
-			logger.Error("propose tick failed", zap.Error(err))
+			l.logger.Error("propose tick failed", zap.Error(err))
 			return
 		}
 	}
@@ -738,7 +742,7 @@ func (l *store) processTruncateLog(ctx context.Context) error {
 			defer cancel()
 			lsn, err := l.getTruncatedLsn(ctx, shardID)
 			if err != nil {
-				logger.Error("GetTruncatedIndex failed", zap.Error(err))
+				l.logger.Error("GetTruncatedIndex failed", zap.Error(err))
 				// FIXME: check error type, see whether it is a tmp one
 				return err
 			}
@@ -749,7 +753,7 @@ func (l *store) processTruncateLog(ctx context.Context) error {
 					CompactionIndex:            lsn - 1,
 				}
 				if _, err := l.nh.SyncRequestSnapshot(ctx, shardID, opts); err != nil {
-					logger.Error("SyncRequestSnapshot failed", zap.Error(err))
+					l.logger.Error("SyncRequestSnapshot failed", zap.Error(err))
 					// FIXME: check error type, see whether it is a tmp one
 					return err
 				}
@@ -776,7 +780,7 @@ func (l *store) getHeartbeatMessage() pb.LogStoreHeartbeat {
 	nhi := l.nh.GetNodeHostInfo(opts)
 	for _, ci := range nhi.ShardInfoList {
 		if ci.Pending {
-			logger.Info(fmt.Sprintf("shard %d is pending, not included into the heartbeat",
+			l.logger.Info(fmt.Sprintf("shard %d is pending, not included into the heartbeat",
 				ci.ShardID))
 			continue
 		}
diff --git a/pkg/logservice/store_hakeeper_check.go b/pkg/logservice/store_hakeeper_check.go
index 07fffabf1..fcc0645f2 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -81,14 +81,14 @@ func (l *store) setInitialClusterInfo(numOfLogShards uint64,
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("failed to propose initial cluster info", zap.Error(err))
+		l.logger.Error("failed to propose initial cluster info", zap.Error(err))
 		return err
 	}
 	if result.Value == uint64(pb.HAKeeperBootstrapFailed) {
 		panic("bootstrap failed")
 	}
 	if result.Value != uint64(pb.HAKeeperCreated) {
-		logger.Error("initial cluster info already set")
+		l.logger.Error("initial cluster info already set")
 	}
 	return nil
 }
@@ -100,7 +100,7 @@ func (l *store) updateIDAlloc(count uint64) error {
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		logger.Error("propose get id failed", zap.Error(err))
+		l.logger.Error("propose get id failed", zap.Error(err))
 		return err
 	}
 	// TODO: add a test for this
@@ -111,7 +111,7 @@ func (l *store) updateIDAlloc(count uint64) error {
 func (l *store) hakeeperCheck() {
 	isLeader, term, err := l.isLeaderHAKeeper()
 	if err != nil {
-		logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
+		l.logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
 		return
 	}
 
@@ -123,12 +123,12 @@ func (l *store) hakeeperCheck() {
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
-		logger.Error("failed to get checker state", zap.Error(err))
+		l.logger.Error("failed to get checker state", zap.Error(err))
 		return
 	}
 	switch state.State {
 	case pb.HAKeeperCreated:
-		logger.Warn("waiting for initial cluster info to be set, check skipped")
+		l.logger.Warn("waiting for initial cluster info to be set, check skipped")
 		return
 	case pb.HAKeeperBootstrapping:
 		l.bootstrap(term, state)
@@ -148,11 +148,11 @@ func (l *store) assertHAKeeperState(s pb.HAKeeperState) {
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
-		logger.Error("failed to get checker state", zap.Error(err))
+		l.logger.Error("failed to get checker state", zap.Error(err))
 		return
 	}
 	if state.State != s {
-		logger.Panic("unexpected state",
+		l.logger.Panic("unexpected state",
 			zap.String("expected", s.String()),
 			zap.String("got", state.State.String()))
 	}
@@ -167,19 +167,19 @@ func (l *store) healthCheck(term uint64, state *pb.CheckerState) {
 	defer l.assertHAKeeperState(pb.HAKeeperRunning)
 	cmds, err := l.getScheduleCommand(true, term, state)
 	if err != nil {
-		logger.Error("failed to get check schedule commands", zap.Error(err))
+		l.logger.Error("failed to get check schedule commands", zap.Error(err))
 		return
 	}
-	logger.Info(fmt.Sprintf("cluster health check generated %d schedule commands", len(cmds)))
+	l.logger.Debug(fmt.Sprintf("cluster health check generated %d schedule commands", len(cmds)))
 	if len(cmds) > 0 {
 		ctx, cancel := context.WithTimeout(context.Background(), hakeeperDefaultTimeout)
 		defer cancel()
 		for _, cmd := range cmds {
-			logger.Info("adding schedule command to hakeeper", zap.String("command", cmd.LogString()))
+			l.logger.Debug("adding schedule command to hakeeper", zap.String("command", cmd.LogString()))
 		}
 		if err := l.addScheduleCommands(ctx, term, cmds); err != nil {
 			// TODO: check whether this is temp error
-			logger.Info("failed to add schedule commands", zap.Error(err))
+			l.logger.Debug("failed to add schedule commands", zap.Error(err))
 			return
 		}
 	}
@@ -195,18 +195,18 @@ func (l *store) taskSchedule(state *pb.CheckerState) {
 func (l *store) bootstrap(term uint64, state *pb.CheckerState) {
 	cmds, err := l.getScheduleCommand(false, term, state)
 	if err != nil {
-		logger.Error("failed to get bootstrap schedule commands", zap.Error(err))
+		l.logger.Error("failed to get bootstrap schedule commands", zap.Error(err))
 		return
 	}
 	if len(cmds) > 0 {
 		for _, c := range cmds {
-			logger.Info("bootstrap cmd", zap.String("cmd", c.LogString()))
+			l.logger.Debug("bootstrap cmd", zap.String("cmd", c.LogString()))
 		}
 		ctx, cancel := context.WithTimeout(context.Background(), hakeeperDefaultTimeout)
 		defer cancel()
 		if err := l.addScheduleCommands(ctx, term, cmds); err != nil {
 			// TODO: check whether this is temp error
-			logger.Info("failed to add schedule commands", zap.Error(err))
+			l.logger.Debug("failed to add schedule commands", zap.Error(err))
 			return
 		}
 		l.bootstrapCheckCycles = checkBootstrapCycles
diff --git a/pkg/logservice/store_metadata.go b/pkg/logservice/store_metadata.go
index 97d9f717e..3f6c2aa9f 100644
--- a/pkg/logservice/store_metadata.go
+++ b/pkg/logservice/store_metadata.go
@@ -153,7 +153,7 @@ func exist(name string, fs vfs.FS) (bool, error) {
 }
 
 func createMetadataFile(dir string,
-	filename string, obj Marshaler, fs vfs.FS) (err error) {
+	filename string, obj Marshaller, fs vfs.FS) (err error) {
 	de, err := dirExist(dir, fs)
 	if err != nil {
 		return err
@@ -252,7 +252,7 @@ func (l *store) loadMetadata() error {
 		return err
 	}
 	if expectedUUID != l.mu.metadata.UUID {
-		logger.Panic("unexpected UUID",
+		l.logger.Panic("unexpected UUID",
 			zap.String("on disk UUID", l.mu.metadata.UUID),
 			zap.String("expect", expectedUUID))
 	}
@@ -263,7 +263,7 @@ func (l *store) mustSaveMetadata() {
 	fs := l.cfg.FS
 	dir := l.cfg.DataDir
 	if err := createMetadataFile(dir, logMetadataFilename, &l.mu.metadata, fs); err != nil {
-		logger.Panic("failed to save metadata file", zap.Error(err))
+		l.logger.Panic("failed to save metadata file", zap.Error(err))
 	}
 }
 
@@ -276,7 +276,7 @@ func (l *store) addMetadata(shardID uint64, replicaID uint64) {
 
 	for _, rec := range l.mu.metadata.Shards {
 		if rec.ShardID == shardID && rec.ReplicaID == replicaID {
-			logger.Info(fmt.Sprintf("addMetadata for shardID %d skipped, dupl shard", shardID))
+			l.logger.Info(fmt.Sprintf("addMetadata for shardID %d skipped, dupl shard", shardID))
 			return
 		}
 	}
diff --git a/pkg/logservice/utils.go b/pkg/logservice/utils.go
index ff96f5703..cba74a022 100644
--- a/pkg/logservice/utils.go
+++ b/pkg/logservice/utils.go
@@ -14,7 +14,7 @@
 
 package logservice
 
-type Marshaler interface {
+type Marshaller interface {
 	Marshal() ([]byte, error)
 }
 
@@ -22,7 +22,7 @@ type Unmarshaler interface {
 	Unmarshal([]byte) error
 }
 
-func MustMarshal(m Marshaler) []byte {
+func MustMarshal(m Marshaller) []byte {
 	data, err := m.Marshal()
 	if err != nil {
 		panic(err)


Test patch:
diff --git a/pkg/hakeeper/checkers/dnservice/check_test.go b/pkg/hakeeper/checkers/dnservice/check_test.go
index 549d3b11d..b2a656df3 100644
--- a/pkg/hakeeper/checkers/dnservice/check_test.go
+++ b/pkg/hakeeper/checkers/dnservice/check_test.go
@@ -21,11 +21,14 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/util"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/operator"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"github.com/stretchr/testify/require"
 )
 
+var logger = logutil.GetGlobalLogger().Named("hakeeper-test")
+
 func TestExpiredReplicas(t *testing.T) {
 	replicaIDs := []uint64{11, 13, 12, 14, 15}
 	retFirst := expiredReplicas(mockDnShard(10, nil, replicaIDs))
@@ -114,7 +117,7 @@ func TestCheckShard(t *testing.T) {
 
 		// register an expired replica => should add a new replica
 		shard.register(newReplica(11, shardID, "store11"), true)
-		steps := checkShard(shard, mapper, workingStores, idAlloc)
+		steps := checkShard(shard, mapper, workingStores, idAlloc, logger)
 		require.Equal(t, 1, len(steps))
 		add, ok := (steps[0]).(operator.AddDnReplica)
 		require.True(t, ok)
@@ -124,12 +127,12 @@ func TestCheckShard(t *testing.T) {
 
 		// register a working replica => no more step
 		shard.register(newReplica(12, shardID, "store12"), false)
-		steps = checkShard(shard, mapper, workingStores, idAlloc)
+		steps = checkShard(shard, mapper, workingStores, idAlloc, logger)
 		require.Equal(t, 0, len(steps))
 
 		// register another working replica => should remove extra replicas
 		shard.register(newReplica(13, shardID, "store13"), false)
-		steps = checkShard(shard, mapper, workingStores, idAlloc)
+		steps = checkShard(shard, mapper, workingStores, idAlloc, logger)
 		require.Equal(t, 1, len(steps))
 		remove, ok := (steps[0]).(operator.RemoveDnReplica)
 		require.True(t, ok)
@@ -153,7 +156,7 @@ func TestCheckShard(t *testing.T) {
 		anotherShard := uint64(100)
 		// register another expired replica, should add a new replica
 		shard := mockDnShard(anotherShard, nil, []uint64{101})
-		steps := checkShard(shard, mapper, workingStores, idAlloc)
+		steps := checkShard(shard, mapper, workingStores, idAlloc, logger)
 		require.Equal(t, 0, len(steps))
 	}
 }
@@ -221,7 +224,7 @@ func TestCheck(t *testing.T) {
 
 		clusterInfo := mockClusterInfo(10, 11)
 
-		steps := Check(idAlloc, config, clusterInfo, dnState, currTick)
+		steps := Check(idAlloc, config, clusterInfo, dnState, currTick, logger)
 		require.Equal(t, len(steps), 0)
 	}
 
@@ -265,7 +268,7 @@ func TestCheck(t *testing.T) {
 		//  10 - add replica
 		//  12 - remove two extra replica (16, 13)
 		//  14 - no command
-		operators := Check(idAlloc, config, clusterInfo, dnState, currTick)
+		operators := Check(idAlloc, config, clusterInfo, dnState, currTick, logger)
 		require.Equal(t, 2, len(operators))
 
 		// shard 10 - single operator step
@@ -319,13 +322,13 @@ func TestCheck(t *testing.T) {
 		// at the tick of `staleTick`, shard 14, 20:
 		//  14 - no command
 		//  20 - add replica after a while
-		operators := Check(idAlloc, config, cluster, dnState, staleTick)
+		operators := Check(idAlloc, config, cluster, dnState, staleTick, logger)
 		require.Equal(t, 0, len(operators))
 
 		// at the tick of `currTick`, shard 14, 20:
 		//  14 - add replica
 		//  20 - add replica
-		operators = Check(idAlloc, config, cluster, dnState, currTick)
+		operators = Check(idAlloc, config, cluster, dnState, currTick, logger)
 		require.Equal(t, 2, len(operators))
 
 		// shard 14 - single operator step
diff --git a/pkg/hakeeper/checkers/dnservice/parse_test.go b/pkg/hakeeper/checkers/dnservice/parse_test.go
index db56b03c2..9951e34df 100644
--- a/pkg/hakeeper/checkers/dnservice/parse_test.go
+++ b/pkg/hakeeper/checkers/dnservice/parse_test.go
@@ -66,15 +66,15 @@ func TestCheckInitiatingShards(t *testing.T) {
 	expiredTick := config.ExpiredTick(earliestTick, config.DNStoreTimeout) + 1
 
 	// discover an initial shard => no operators generated
-	ops := checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, earliestTick)
+	ops := checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, earliestTick, logger)
 	require.Equal(t, 0, len(ops))
 
 	// waiting some time, but not long enough
-	ops = checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, expiredTick-1)
+	ops = checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, expiredTick-1, logger)
 	require.Equal(t, 0, len(ops))
 
 	// waiting long enough
-	ops = checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, expiredTick)
+	ops = checkInitiatingShards(rs, mapper, workingStores, idAlloc, cluster, config, expiredTick, logger)
 	require.Equal(t, 1, len(ops))
 }
 
@@ -95,14 +95,14 @@ func TestCheckReportedState(t *testing.T) {
 	// register an expired replica => should add a new replica
 	rs := newReportedShards()
 	rs.registerReplica(newReplica(11, shardID, "store11"), true)
-	ops := checkReportedState(rs, mapper, workingStores, idAlloc)
+	ops := checkReportedState(rs, mapper, workingStores, idAlloc, logger)
 	require.Equal(t, 1, len(ops))
 	require.Equal(t, shardID, ops[0].ShardID())
 
 	// register a working replica => no more step
 	rs = newReportedShards()
 	rs.registerReplica(newReplica(12, shardID, "store12"), false)
-	ops = checkReportedState(rs, mapper, workingStores, idAlloc)
+	ops = checkReportedState(rs, mapper, workingStores, idAlloc, logger)
 	require.Equal(t, 0, len(ops))
 }
 
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 9e6d6ead3..013378b3d 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -31,11 +31,14 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
 
+var testLogger = logutil.GetGlobalLogger().Named("logservice-test")
+
 func runClientTest(t *testing.T,
 	readOnly bool, fn func(*testing.T, *Service, ClientConfig, Client)) {
 	defer leaktest.AfterTest(t)()
@@ -233,7 +236,7 @@ func TestClientAppendAlloc(t *testing.T) {
 			_, err := c.Append(ctx, rec)
 			require.NoError(t, err)
 		})
-		logger.Info(fmt.Sprintf("ac: %f", ac))
+		testLogger.Info(fmt.Sprintf("ac: %f", ac))
 	}
 	runClientTest(t, false, fn)
 }
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index db1aa35c1..85f615eac 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -180,6 +180,8 @@ func TestHandleAddReplica(t *testing.T) {
 
 	service1 := Service{
 		store: store1,
+
+		logger: testLogger,
 	}
 	cmd := pb.ScheduleCommand{
 		ConfigChange: &pb.ConfigChange{
@@ -207,7 +209,8 @@ func TestHandleRemoveReplica(t *testing.T) {
 	}()
 
 	service1 := Service{
-		store: store1,
+		store:  store1,
+		logger: testLogger,
 	}
 	cmd := pb.ScheduleCommand{
 		ConfigChange: &pb.ConfigChange{
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index 33b391a6a..6c74b1110 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -583,7 +583,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	for i := 0; i < 6000; i++ {
 		si1, ok := service1.getShardInfo(1)
 		if !ok || si1.LeaderID != 1 {
-			logger.Error("shard 1 info missing on service 1")
+			testLogger.Error("shard 1 info missing on service 1")
 			time.Sleep(time.Millisecond)
 			continue
 		}
@@ -596,7 +596,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 
 		si2, ok := service1.getShardInfo(2)
 		if !ok || si2.LeaderID != 1 {
-			logger.Error("shard 2 info missing on service 1")
+			testLogger.Error("shard 2 info missing on service 1")
 			time.Sleep(time.Millisecond)
 			continue
 		}
@@ -609,7 +609,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 
 		si1, ok = service2.getShardInfo(1)
 		if !ok || si1.LeaderID != 1 {
-			logger.Error("shard 1 info missing on service 2")
+			testLogger.Error("shard 1 info missing on service 2")
 			time.Sleep(time.Millisecond)
 			continue
 		}
@@ -622,7 +622,7 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 
 		si2, ok = service2.getShardInfo(2)
 		if !ok || si2.LeaderID != 1 {
-			logger.Error("shard 2 info missing on service 2")
+			testLogger.Error("shard 2 info missing on service 2")
 			time.Sleep(time.Millisecond)
 			continue
 		}
@@ -687,7 +687,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 		services = append(services, service)
 	}
 	defer func() {
-		logger.Info("going to close all services")
+		testLogger.Info("going to close all services")
 		var wg sync.WaitGroup
 		for _, s := range services {
 			if s != nil {
@@ -696,7 +696,7 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 				go func() {
 					require.NoError(t, selected.Close())
 					wg.Done()
-					logger.Info("closed a service")
+					testLogger.Info("closed a service")
 				}()
 			}
 		}
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index 9dabbc112..6d160753d 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -416,7 +416,7 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 		// wait for HAKeeper to repair the Log & HAKeeper shards
 		dnRepaired := false
 		for i := 0; i < 5000; i++ {
-			logger.Info(fmt.Sprintf("iteration %d", i))
+			testLogger.Info(fmt.Sprintf("iteration %d", i))
 			tn := func() (bool, error) {
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 				defer cancel()
@@ -470,11 +470,11 @@ func TestHAKeeperCanBootstrapAndRepairShards(t *testing.T) {
 						break
 					}
 				}
-				logger.Info(fmt.Sprintf("dnRepaired %t, logRepaired %t", dnRepaired, logRepaired))
+				testLogger.Info(fmt.Sprintf("dnRepaired %t, logRepaired %t", dnRepaired, logRepaired))
 				if !logRepaired || !dnRepaired {
 					return false, nil
 				} else {
-					logger.Info(fmt.Sprintf("repair completed, i: %d", i))
+					testLogger.Info(fmt.Sprintf("repair completed, i: %d", i))
 					return true, nil
 				}
 			}
@@ -784,7 +784,7 @@ func TestTaskSchedulerCanReScheduleExpiredTasks(t *testing.T) {
 
 		cnUUID2 := uuid.New().String()
 		for i := 0; i < 1000; i++ {
-			logger.Info(fmt.Sprintf("iteration %d", i))
+			testLogger.Info(fmt.Sprintf("iteration %d", i))
 			tn := func() bool {
 				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 				defer cancel()
@@ -797,7 +797,7 @@ func TestTaskSchedulerCanReScheduleExpiredTasks(t *testing.T) {
 				tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID2))
 				assert.NoError(t, err)
 				if len(tasks) == 0 {
-					logger.Info("no task found")
+					testLogger.Info("no task found")
 					time.Sleep(50 * time.Millisecond)
 				} else {
 					tasks, err = taskService.QueryTask(ctx, taskservice.WithTaskRunnerCond(taskservice.EQ, cnUUID1))
diff --git a/pkg/logservice/store_metadata_test.go b/pkg/logservice/store_metadata_test.go
index bfd0226b7..b2bab4956 100644
--- a/pkg/logservice/store_metadata_test.go
+++ b/pkg/logservice/store_metadata_test.go
@@ -59,7 +59,7 @@ func TestAddMetadataRejectDupl(t *testing.T) {
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	cfg.Fill()
-	s := store{cfg: cfg}
+	s := store{cfg: cfg, logger: testLogger}
 	require.NoError(t, mkdirAll(s.cfg.DataDir, cfg.FS))
 	s.addMetadata(10, 1)
 	s.addMetadata(10, 1)
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 04879602f..6bf4be274 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -21,18 +21,18 @@ import (
 	"testing"
 	"time"
 
-	"github.com/matrixorigin/matrixone/pkg/taskservice"
-
 	"github.com/google/uuid"
 	"github.com/lni/dragonboat/v4"
 	"github.com/lni/goutils/leaktest"
 	"github.com/lni/vfs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/taskservice"
 )
 
 var (
@@ -75,17 +75,17 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg, nil)
+	store, err := newLogStore(cfg, nil, testLogger)
 	assert.NoError(t, err)
-	logger.Info("1")
+	testLogger.Info("1")
 	defer func() {
 		assert.NoError(t, store.close())
 	}()
-	logger.Info("2")
+	testLogger.Info("2")
 }
 
 func getTestStore(cfg Config, startLogReplica bool, taskService taskservice.TaskService) (*store, error) {
-	store, err := newLogStore(cfg, taskService)
+	store, err := newLogStore(cfg, taskService, testLogger)
 	if err != nil {
 		return nil, err
 	}
@@ -104,7 +104,7 @@ func TestHAKeeperCanBeStarted(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg, nil)
+	store, err := newLogStore(cfg, nil, nil)
 	assert.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[2] = store.nh.ID()
@@ -429,7 +429,7 @@ func getTestStores() (*store, *store, error) {
 		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
 	}
 	cfg1.Fill()
-	store1, err := newLogStore(cfg1, nil)
+	store1, err := newLogStore(cfg1, nil, testLogger.With(zap.String("store", cfg1.DataDir)))
 	if err != nil {
 		return nil, nil, err
 	}
@@ -444,7 +444,7 @@ func getTestStores() (*store, *store, error) {
 		GossipSeedAddresses: []string{"127.0.0.1:9011", "127.0.0.1:9012"},
 	}
 	cfg2.Fill()
-	store2, err := newLogStore(cfg2, nil)
+	store2, err := newLogStore(cfg2, nil, testLogger.With(zap.String("store", cfg2.DataDir)))
 	if err != nil {
 		return nil, nil, err
 	}
@@ -472,7 +472,7 @@ func getTestStores() (*store, *store, error) {
 		}
 		if ok && leaderID != 1 {
 			if err := store1.requestLeaderTransfer(1, 1); err != nil {
-				logger.Error("failed to transfer leader")
+				testLogger.Error("failed to transfer leader")
 			}
 		}
 		time.Sleep(time.Millisecond)