Repository: matrixorigin/matrixone
Pull request #7149 (2022-12-12 07:12:26)

Support use system-init to search system initialization related logs
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #6989

## What this PR does / why we need it:
Support use system-init to search system initialization related logs

w-zr: > How do you use this? Can you give me an example on how to use/query?
> 
> How about adding a few examples in comment? Actually since bvt must have initialized system we should be able to add a query in bvt.

You can search for init logs by `"process": "system-init"`.

![图片](https://user-images.githubusercontent.com/28624654/207491213-f14dabec-69b5-4ae0-96fd-0aadecbff02a.png)

w-zr: Run 
```sql
select message from system.log_info where extra like '%system-init%';
```
to select system initializing related logs.

Patch:
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index 88276bf3a..68f52ca10 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -34,6 +34,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/cnservice"
 	"github.com/matrixorigin/matrixone/pkg/cnservice/cnclient"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	"github.com/matrixorigin/matrixone/pkg/dnservice"
@@ -228,7 +229,7 @@ func startLogService(
 ) error {
 	lscfg := cfg.getLogServiceConfig()
 	s, err := logservice.NewService(lscfg, fileService,
-		logservice.WithLogger(logutil.GetGlobalLogger().Named("log-service").With(zap.String("uuid", lscfg.UUID))))
+		logservice.WithRuntime(runtime.ProcessLevelRuntime()))
 	if err != nil {
 		panic(err)
 	}
diff --git a/pkg/cnservice/server_task.go b/pkg/cnservice/server_task.go
index 111246a78..0008a7d32 100644
--- a/pkg/cnservice/server_task.go
+++ b/pkg/cnservice/server_task.go
@@ -157,7 +157,7 @@ func (s *service) waitSystemInitCompleted(ctx context.Context) {
 			ts, ok := s.GetTaskService()
 			if ok {
 				tasks, err := ts.QueryTask(ctx,
-					taskservice.WithTaskExecutorCond(taskservice.EQ, uint32(task.TaskCode_SystemInit)),
+					taskservice.WithTaskExecutorCond(taskservice.EQ, task.TaskCode_SystemInit),
 					taskservice.WithTaskStatusCond(taskservice.EQ, task.TaskStatus_Completed))
 				if err != nil {
 					s.logger.Error("wait all init task completed failed", zap.Error(err))
@@ -221,7 +221,7 @@ func (s *service) registerExecutorsLocked() {
 	if !ok {
 		panic(moerr.NewInternalErrorNoCtx("task Service not ok"))
 	}
-	s.task.runner.RegisterExecutor(uint32(task.TaskCode_SystemInit),
+	s.task.runner.RegisterExecutor(task.TaskCode_SystemInit,
 		func(ctx context.Context, t task.Task) error {
 			if err := frontend.InitSysTenant(moServerCtx); err != nil {
 				return err
@@ -245,6 +245,6 @@ func (s *service) registerExecutorsLocked() {
 		})
 
 	// init metric/log merge task executor
-	s.task.runner.RegisterExecutor(uint32(task.TaskCode_MetricLogMerge),
+	s.task.runner.RegisterExecutor(task.TaskCode_MetricLogMerge,
 		export.MergeTaskExecutorFactory(export.WithFileService(s.fileService)))
 }
diff --git a/pkg/common/log/filter.go b/pkg/common/log/filter.go
index d98f52164..0574449d8 100644
--- a/pkg/common/log/filter.go
+++ b/pkg/common/log/filter.go
@@ -24,7 +24,8 @@ var (
 
 	// all SampleType register here.
 	samples = map[SampleType]*sampleValue{
-		ExampleSample: {frequency: 3},
+		ExampleSample:    {frequency: 3},
+		SystemInitSample: {frequency: 5},
 	}
 )
 
diff --git a/pkg/common/log/logger.go b/pkg/common/log/logger.go
index 74f79b079..4a8ad9adf 100644
--- a/pkg/common/log/logger.go
+++ b/pkg/common/log/logger.go
@@ -29,7 +29,9 @@ import (
 // GetServiceLogger returns service logger, it will using the service as the logger name, and
 // append FieldNameServiceUUID field to the logger
 func GetServiceLogger(logger *zap.Logger, service metadata.ServiceType, uuid string) *MOLogger {
-	return wrap(logger.Named(fmt.Sprintf("%s-service", strings.ToLower(service.String()))).With(zap.String(FieldNameServiceUUID, uuid)))
+	return wrap(logger.
+		Named(fmt.Sprintf("%s-service", strings.ToLower(service.String()))).
+		With(zap.String(FieldNameServiceUUID, uuid)))
 }
 
 // GetModuleLogger returns the module logger, it will add ".module" to logger name.
@@ -73,6 +75,13 @@ func newMOLogger(logger *zap.Logger, ctx context.Context) *MOLogger {
 	}
 }
 
+// WithProcess if the current log belongs to a certain process, the process name and process ID
+// can be recorded. When analyzing the log, all related logs can be retrieved according to the
+// process ID.
+func (l *MOLogger) WithProcess(process Process) *MOLogger {
+	return l.With(zap.String(FieldNameProcess, string(process)))
+}
+
 // Enabled returns true if the level is enabled
 func (l *MOLogger) Enabled(level zapcore.Level) bool {
 	return l.logger.Core().Enabled(level)
diff --git a/pkg/common/log/types.go b/pkg/common/log/types.go
index 1ecfa1b31..0859e47a6 100644
--- a/pkg/common/log/types.go
+++ b/pkg/common/log/types.go
@@ -65,10 +65,11 @@ var (
 // need to sample the output.
 type SampleType int
 
-var (
-	noneSample = SampleType(0)
+const (
+	noneSample SampleType = iota
+	SystemInitSample
 	// ExampleSample used in examples
-	ExampleSample = SampleType(math.MaxInt)
+	ExampleSample = math.MaxInt
 )
 
 // MOLogger mo logger based zap.logger. To standardize and standardize the logging
diff --git a/pkg/common/runtime/runtime.go b/pkg/common/runtime/runtime.go
index 61b0da609..59fa6b1eb 100644
--- a/pkg/common/runtime/runtime.go
+++ b/pkg/common/runtime/runtime.go
@@ -30,6 +30,13 @@ var (
 	processLevel atomic.Value
 )
 
+type LoggerName int
+
+const (
+	Default LoggerName = iota
+	SystemInit
+)
+
 // ProcessLevelRuntime returns a process-lelve runtime
 func ProcessLevelRuntime() Runtime {
 	v := processLevel.Load()
@@ -64,6 +71,7 @@ func NewRuntime(service metadata.ServiceType, uuid string, logger *zap.Logger, o
 		opt(rt)
 	}
 	rt.global.logger = log.GetServiceLogger(logutil.Adjust(logger), service, uuid)
+	rt.initSystemInitLogger()
 	return rt
 }
 
@@ -75,6 +83,8 @@ type runtime struct {
 		clock     clock.Clock
 		logger    *log.MOLogger
 		variables sync.Map
+
+		systemInitLogger *log.MOLogger
 	}
 }
 
@@ -82,6 +92,15 @@ func (r *runtime) Logger() *log.MOLogger {
 	return r.global.logger
 }
 
+func (r *runtime) SubLogger(name LoggerName) *log.MOLogger {
+	switch name {
+	case SystemInit:
+		return r.global.systemInitLogger
+	default:
+		return r.Logger()
+	}
+}
+
 func (r *runtime) Clock() clock.Clock {
 	return r.global.clock
 }
@@ -112,3 +131,10 @@ func DefaultRuntime() Runtime {
 			return time.Now().UTC().UnixNano()
 		}, 0)))
 }
+
+func (r *runtime) initSystemInitLogger() {
+	if r.global.logger == nil {
+		r.global.logger = log.GetServiceLogger(logutil.Adjust(nil), r.serviceType, r.serviceUUID)
+	}
+	r.global.systemInitLogger = r.Logger().WithProcess(log.SystemInit)
+}
diff --git a/pkg/common/runtime/types.go b/pkg/common/runtime/types.go
index 166f9f23c..709931427 100644
--- a/pkg/common/runtime/types.go
+++ b/pkg/common/runtime/types.go
@@ -56,6 +56,8 @@ type Runtime interface {
 	// the service type and unique ID of the service; all subsequent loggers must be
 	// built on this logger.
 	Logger() *log.MOLogger
+	// SubLogger returns sub-loggers used for different purposes.
+	SubLogger(LoggerName) *log.MOLogger
 	// Clock returns the Clock instance of the current runtime environment
 	Clock() clock.Clock
 	// SetGlobalVariables set global variables which scope based in runtime.
diff --git a/pkg/dnservice/store.go b/pkg/dnservice/store.go
index 965b0a23c..f3c280857 100644
--- a/pkg/dnservice/store.go
+++ b/pkg/dnservice/store.go
@@ -168,7 +168,7 @@ func (s *store) Start() error {
 	if err := s.server.Start(); err != nil {
 		return err
 	}
-	s.rt.Logger().Info("dn heartbeat task started")
+	s.rt.SubLogger(runtime.SystemInit).Info("dn heartbeat task started")
 	return s.stopper.RunTask(s.heartbeatTask)
 }
 
diff --git a/pkg/hakeeper/bootstrap/bootstrap.go b/pkg/hakeeper/bootstrap/bootstrap.go
index 32db9c96b..15788cd94 100644
--- a/pkg/hakeeper/bootstrap/bootstrap.go
+++ b/pkg/hakeeper/bootstrap/bootstrap.go
@@ -50,10 +50,11 @@ func (bm *Manager) Bootstrap(alloc util.IDAllocator,
 
 	commands := append(logCommands, dnCommands...)
 	for _, command := range commands {
-		runtime.ProcessLevelRuntime().Logger().Info("schedule command generated", zap.String("command", command.LogString()))
+		runtime.ProcessLevelRuntime().SubLogger(runtime.SystemInit).Info("schedule command generated",
+			zap.String("command", command.LogString()))
 	}
 	if len(commands) != 0 {
-		runtime.ProcessLevelRuntime().Logger().Info("bootstrap commands generated")
+		runtime.ProcessLevelRuntime().SubLogger(runtime.SystemInit).Info("schedule command generated")
 	}
 	return commands, nil
 }
diff --git a/pkg/hakeeper/task/task_scheduler.go b/pkg/hakeeper/task/task_scheduler.go
index ce479615f..d5d1b7ed8 100644
--- a/pkg/hakeeper/task/task_scheduler.go
+++ b/pkg/hakeeper/task/task_scheduler.go
@@ -52,9 +52,19 @@ func (s *scheduler) Schedule(cnState logservice.CNState, currentTick uint64) {
 
 	runningTasks := s.queryTasks(task.TaskStatus_Running)
 	createdTasks := s.queryTasks(task.TaskStatus_Created)
+	tasks := append(runningTasks, createdTasks...)
+	for _, task := range tasks {
+		if task.IsInitTask() {
+			runtime.ProcessLevelRuntime().
+				SubLogger(runtime.SystemInit).Debug(
+				"task schedule query init task",
+				zap.String("task", task.Metadata.String()))
+		}
+	}
+
 	runtime.ProcessLevelRuntime().Logger().Debug("task schedule query tasks", zap.Int("created", len(createdTasks)),
 		zap.Int("running", len(runningTasks)))
-	if len(runningTasks) == 0 && len(createdTasks) == 0 {
+	if len(tasks) == 0 {
 		return
 	}
 	orderedCN := getCNOrdered(runningTasks, workingCN)
diff --git a/pkg/logservice/option.go b/pkg/logservice/option.go
index 7f55025d1..27cf6511a 100644
--- a/pkg/logservice/option.go
+++ b/pkg/logservice/option.go
@@ -18,8 +18,8 @@ import (
 	"context"
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
-	"go.uber.org/zap"
 )
 
 // Option is utility that sets callback for Service.
@@ -32,10 +32,10 @@ func WithBackendFilter(filter func(morpc.Message, string) bool) Option {
 	}
 }
 
-// WithLogger sets logger
-func WithLogger(logger *zap.Logger) Option {
+// WithRuntime sets runtime
+func WithRuntime(runtime runtime.Runtime) Option {
 	return func(s *Service) {
-		s.logger = logger
+		s.runtime = runtime
 	}
 }
 
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index beb46510e..f722fca55 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -29,9 +29,9 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
-	"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/util/trace"
@@ -60,7 +60,7 @@ func firstError(err1 error, err2 error) error {
 // be considered as the interface layer of the LogService.
 type Service struct {
 	cfg         Config
-	logger      *zap.Logger
+	runtime     runtime.Runtime
 	store       *store
 	server      morpc.RPCServer
 	pool        *sync.Pool
@@ -100,12 +100,12 @@ func NewService(
 	for _, opt := range opts {
 		opt(service)
 	}
-
-	service.logger = logutil.Adjust(service.logger)
-
-	store, err := newLogStore(cfg, service.getTaskService, service.logger)
+	if service.runtime == nil {
+		service.runtime = runtime.DefaultRuntime()
+	}
+	store, err := newLogStore(cfg, service.getTaskService, service.runtime)
 	if err != nil {
-		service.logger.Error("failed to create log store", zap.Error(err))
+		service.runtime.Logger().Error("failed to create log store", zap.Error(err))
 		return nil, err
 	}
 	if err := store.loadMetadata(); err != nil {
@@ -144,7 +144,7 @@ func NewService(
 		morpc.WithServerGoettyOptions(goetty.WithSessionReleaseMsgFunc(func(i interface{}) {
 			respPool.Put(i.(morpc.RPCMessage).Message)
 		})),
-		morpc.WithServerLogger(service.logger),
+		morpc.WithServerLogger(service.runtime.Logger().RawLogger()),
 	)
 	if err != nil {
 		return nil, err
@@ -159,16 +159,16 @@ func NewService(
 	// TODO: before making the service available to the outside world, restore all
 	// replicas already known to the local store
 	if err := server.Start(); err != nil {
-		service.logger.Error("failed to start the server", zap.Error(err))
+		service.runtime.SubLogger(runtime.SystemInit).Error("failed to start the server", zap.Error(err))
 		if err := store.close(); err != nil {
-			service.logger.Error("failed to close the store", zap.Error(err))
+			service.runtime.SubLogger(runtime.SystemInit).Error("failed to close the store", zap.Error(err))
 		}
 		return nil, err
 	}
 	// start the heartbeat worker
 	if !cfg.DisableWorkers {
 		if err := service.stopper.RunNamedTask("log-heartbeat-worker", func(ctx context.Context) {
-			service.logger.Info("logservice heartbeat worker started")
+			service.runtime.SubLogger(runtime.SystemInit).Info("logservice heartbeat worker started")
 
 			// transfer morpc options via context
 			ctx = SetBackendOptions(ctx, service.getBackendOptions()...)
diff --git a/pkg/logservice/service_bootstrap.go b/pkg/logservice/service_bootstrap.go
index 24a58918b..b57a5ef66 100644
--- a/pkg/logservice/service_bootstrap.go
+++ b/pkg/logservice/service_bootstrap.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/lni/dragonboat/v4"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/pb/task"
 	"go.uber.org/zap"
 )
@@ -37,7 +38,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 {
-			s.logger.Error("failed to start hakeeper replica", zap.Error(err))
+			s.runtime.SubLogger(runtime.SystemInit).Error("failed to start hakeeper replica", zap.Error(err))
 			return err
 		}
 	}
@@ -52,14 +53,14 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
 		}
 		if err := s.store.setInitialClusterInfo(numOfLogShards,
 			numOfDNShards, numOfLogReplicas); err != nil {
-			s.logger.Error("failed to set initial cluster info", zap.Error(err))
+			s.runtime.SubLogger(runtime.SystemInit).Error("failed to set initial cluster info", zap.Error(err))
 			if err == dragonboat.ErrShardNotFound {
 				return nil
 			}
 			time.Sleep(time.Second)
 			continue
 		}
-		s.logger.Info("initial cluster info set")
+		s.runtime.SubLogger(runtime.SystemInit).Info("initial cluster info set")
 		break
 	}
 	for i := 0; i < checkBootstrapCycles; i++ {
@@ -79,11 +80,11 @@ func (s *Service) BootstrapHAKeeper(ctx context.Context, cfg Config) error {
 func (s *Service) createInitTasks(ctx context.Context) error {
 	if err := s.store.taskScheduler.Create(ctx, []task.TaskMetadata{{
 		ID:       task.TaskCode_SystemInit.String(),
-		Executor: uint32(task.TaskCode_SystemInit),
+		Executor: task.TaskCode_SystemInit,
 	}}); err != nil {
-		s.logger.Error("failed to create init tasks", zap.Error(err))
+		s.runtime.SubLogger(runtime.SystemInit).Error("failed to create init tasks", zap.Error(err))
 		return err
 	}
-	s.logger.Info("init tasks created")
+	s.runtime.SubLogger(runtime.SystemInit).Info("init tasks created")
 	return nil
 }
diff --git a/pkg/logservice/service_commands.go b/pkg/logservice/service_commands.go
index abcd01c38..b2e309559 100644
--- a/pkg/logservice/service_commands.go
+++ b/pkg/logservice/service_commands.go
@@ -29,9 +29,9 @@ import (
 
 func (s *Service) handleCommands(cmds []pb.ScheduleCommand) {
 	for _, cmd := range cmds {
-		s.logger.Info(fmt.Sprintf("%s applying cmd: %s", s.ID(), cmd.LogString()))
+		s.runtime.Logger().Info(fmt.Sprintf("%s applying cmd: %s", s.ID(), cmd.LogString()))
 		if cmd.GetConfigChange() != nil {
-			s.logger.Debug("applying schedule command:", zap.String("command", cmd.LogString()))
+			s.runtime.Logger().Debug("applying schedule command:", zap.String("command", cmd.LogString()))
 			switch cmd.ConfigChange.ChangeType {
 			case pb.AddReplica:
 				s.handleAddReplica(cmd)
@@ -62,7 +62,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 {
-		s.logger.Error("failed to add replica", zap.Error(err))
+		s.runtime.Logger().Error("failed to add replica", zap.Error(err))
 	}
 }
 
@@ -71,7 +71,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 {
-		s.logger.Error("failed to remove replica", zap.Error(err))
+		s.runtime.Logger().Error("failed to remove replica", zap.Error(err))
 	}
 }
 
@@ -82,12 +82,12 @@ func (s *Service) handleStartReplica(cmd pb.ScheduleCommand) {
 	if shardID == hakeeper.DefaultHAKeeperShardID {
 		if err := s.store.startHAKeeperReplica(replicaID,
 			cmd.ConfigChange.InitialMembers, join); err != nil {
-			s.logger.Error("failed to start HAKeeper replica", zap.Error(err))
+			s.runtime.Logger().Error("failed to start HAKeeper replica", zap.Error(err))
 		}
 	} else {
 		if err := s.store.startReplica(shardID,
 			replicaID, cmd.ConfigChange.InitialMembers, join); err != nil {
-			s.logger.Error("failed to start log replica", zap.Error(err))
+			s.runtime.Logger().Error("failed to start log replica", zap.Error(err))
 		}
 	}
 }
@@ -96,7 +96,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 {
-		s.logger.Error("failed to stop replica", zap.Error(err))
+		s.runtime.Logger().Error("failed to stop replica", zap.Error(err))
 	}
 }
 
@@ -109,7 +109,7 @@ func (s *Service) handleKillZombie(cmd pb.ScheduleCommand) {
 
 func (s *Service) handleShutdownStore(_ pb.ScheduleCommand) {
 	if err := s.Close(); err != nil {
-		s.logger.Error("failed to shutdown replica", zap.Error(err))
+		s.runtime.Logger().Error("failed to shutdown replica", zap.Error(err))
 	}
 }
 
@@ -119,7 +119,7 @@ func (s *Service) heartbeatWorker(ctx context.Context) {
 		panic("invalid heartbeat interval")
 	}
 	defer func() {
-		s.logger.Info("heartbeat worker stopped")
+		s.runtime.Logger().Info("heartbeat worker stopped")
 	}()
 	ticker := time.NewTicker(s.cfg.HeartbeatInterval.Duration)
 	defer ticker.Stop()
@@ -153,7 +153,7 @@ func (s *Service) heartbeat(ctx context.Context) {
 		}
 		cc, err := NewLogHAKeeperClient(ctx2, s.cfg.GetHAKeeperClientConfig())
 		if err != nil {
-			s.logger.Error("failed to create HAKeeper client", zap.Error(err))
+			s.runtime.Logger().Error("failed to create HAKeeper client", zap.Error(err))
 			return
 		}
 		s.haClient = cc
@@ -163,7 +163,7 @@ func (s *Service) heartbeat(ctx context.Context) {
 	hb.TaskServiceCreated = s.taskServiceCreated()
 	cb, err := s.haClient.SendLogHeartbeat(ctx2, hb)
 	if err != nil {
-		s.logger.Error("failed to send log service heartbeat", zap.Error(err))
+		s.runtime.Logger().Error("failed to send log service heartbeat", zap.Error(err))
 		return
 	}
 	s.handleCommands(cb.Commands)
diff --git a/pkg/logservice/service_task.go b/pkg/logservice/service_task.go
index 5ec96a9f4..36768f103 100644
--- a/pkg/logservice/service_task.go
+++ b/pkg/logservice/service_task.go
@@ -71,7 +71,7 @@ func (s *Service) createTaskService(command *logservicepb.CreateTaskService) {
 		return
 	}
 	if err := s.task.holder.Create(*command); err != nil {
-		s.logger.Error("create task service failed",
+		s.runtime.Logger().Error("create task service failed",
 			zap.Error(err))
 		return
 	}
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 2dc6ba0ec..6c64e1b5e 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -30,6 +30,7 @@ import (
 	"github.com/lni/dragonboat/v4/raftpb"
 	sm "github.com/lni/dragonboat/v4/statemachine"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/common/stopper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper/bootstrap"
@@ -124,7 +125,7 @@ type store struct {
 	alloc             hakeeper.IDAllocator
 	stopper           *stopper.Stopper
 	tickerStopper     *stopper.Stopper
-	logger            *zap.Logger
+	runtime           runtime.Runtime
 
 	bootstrapCheckCycles uint64
 	bootstrapMgr         *bootstrap.Manager
@@ -141,14 +142,13 @@ type store struct {
 
 func newLogStore(cfg Config,
 	taskServiceGetter func() taskservice.TaskService,
-	logger *zap.Logger) (*store, error) {
+	rt runtime.Runtime) (*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",
+	rt.SubLogger(runtime.SystemInit).Info("HAKeeper Timeout Configs",
 		zap.Int64("LogStoreTimeout", int64(hakeeperConfig.LogStoreTimeout)),
 		zap.Int64("DNStoreTimeout", int64(hakeeperConfig.DNStoreTimeout)),
 		zap.Int64("CNStoreTimeout", int64(hakeeperConfig.CNStoreTimeout)),
@@ -161,14 +161,14 @@ func newLogStore(cfg Config,
 		alloc:         newIDAllocator(),
 		stopper:       stopper.NewStopper("log-store"),
 		tickerStopper: stopper.NewStopper("hakeeper-ticker"),
-		logger:        logger,
+		runtime:       rt,
 
 		shardSnapshotInfo: newShardSnapshotInfo(),
 		snapshotMgr:       newSnapshotManager(&cfg),
 	}
 	ls.mu.metadata = metadata.LogStore{UUID: cfg.UUID}
 	if err := ls.stopper.RunNamedTask("truncation-worker", func(ctx context.Context) {
-		logger.Info("logservice truncation worker started")
+		rt.SubLogger(runtime.SystemInit).Info("logservice truncation worker started")
 		ls.truncationWorker(ctx)
 	}); err != nil {
 		return nil, err
@@ -220,7 +220,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) {
-			l.logger.Info("HAKeeper ticker started")
+			l.runtime.SubLogger(runtime.SystemInit).Info("HAKeeper ticker started")
 			l.ticker(ctx)
 		}); err != nil {
 			return err
@@ -366,11 +366,11 @@ func (l *store) truncateLog(ctx context.Context,
 	cmd := getSetTruncatedLsnCmd(index)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("propose truncate log cmd failed", zap.Error(err))
+		l.runtime.Logger().Error("propose truncate log cmd failed", zap.Error(err))
 		return err
 	}
 	if result.Value > 0 {
-		l.logger.Error(fmt.Sprintf("shardID %d already truncated to index %d", shardID, result.Value))
+		l.runtime.Logger().Error(fmt.Sprintf("shardID %d already truncated to index %d", shardID, result.Value))
 		return moerr.NewInvalidTruncateLsn(ctx, shardID, result.Value)
 	}
 	return nil
@@ -381,11 +381,11 @@ func (l *store) append(ctx context.Context,
 	session := l.nh.GetNoOPSession(shardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("propose failed", zap.Error(err))
+		l.runtime.Logger().Error("propose failed", zap.Error(err))
 		return 0, err
 	}
 	if len(result.Data) > 0 {
-		l.logger.Error("not current lease holder", zap.Uint64("data", binaryEnc.Uint64(result.Data)))
+		l.runtime.Logger().Error("not current lease holder", zap.Uint64("data", binaryEnc.Uint64(result.Data)))
 		return 0, moerr.NewNotLeaseHolder(ctx, binaryEnc.Uint64(result.Data))
 	}
 	if result.Value == 0 {
@@ -408,7 +408,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 {
-		l.logger.Error("failed to propose tso updat", zap.Error(err))
+		l.runtime.Logger().Error("failed to propose tso updat", zap.Error(err))
 		return 0, err
 	}
 	return result.Value, nil
@@ -430,7 +430,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 {
-		l.logger.Error("propose failed", zap.Error(err))
+		l.runtime.Logger().Error("propose failed", zap.Error(err))
 		return pb.CommandBatch{}, handleNotHAKeeperError(ctx, err)
 	} else {
 		var cb pb.CommandBatch
@@ -445,7 +445,7 @@ func (l *store) addCNStoreHeartbeat(ctx context.Context,
 	cmd := hakeeper.GetCNStoreHeartbeatCmd(data)
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	if result, err := l.propose(ctx, session, cmd); err != nil {
-		l.logger.Error("propose failed", zap.Error(err))
+		l.runtime.Logger().Error("propose failed", zap.Error(err))
 		return pb.CommandBatch{}, handleNotHAKeeperError(ctx, err)
 	} else {
 		var cb pb.CommandBatch
@@ -460,7 +460,7 @@ func (l *store) cnAllocateID(ctx context.Context,
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("propose get id failed", zap.Error(err))
+		l.runtime.Logger().Error("propose get id failed", zap.Error(err))
 		return 0, err
 	}
 	return result.Value, nil
@@ -472,7 +472,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 {
-		l.logger.Error("propose failed", zap.Error(err))
+		l.runtime.Logger().Error("propose failed", zap.Error(err))
 		return pb.CommandBatch{}, handleNotHAKeeperError(ctx, err)
 	} else {
 		var cb pb.CommandBatch
@@ -522,7 +522,7 @@ func (l *store) getLeaseHolderID(ctx context.Context,
 	}
 	v, err := l.read(ctx, shardID, leaseHistoryQuery{lsn: e.Index})
 	if err != nil {
-		l.logger.Error("failed to read", zap.Error(err))
+		l.runtime.Logger().Error("failed to read", zap.Error(err))
 		return 0, err
 	}
 	return v.(uint64), nil
@@ -620,7 +620,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 {
-		l.logger.Error("QueryRaftLog failed", zap.Error(err))
+		l.runtime.Logger().Error("QueryRaftLog failed", zap.Error(err))
 		return nil, 0, err
 	}
 	select {
@@ -630,13 +630,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 {
-				l.logger.Error("markEntries failed", zap.Error(err))
+				l.runtime.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
-			l.logger.Error("OutOfRange query found")
+			l.runtime.Logger().Error("OutOfRange query found")
 			return nil, 0, dragonboat.ErrInvalidRange
 		}
 		panic(moerr.NewInvalidState(ctx, "unexpected rs state"))
@@ -649,7 +649,7 @@ func (l *store) ticker(ctx context.Context) {
 	if l.cfg.HAKeeperTickInterval.Duration == 0 {
 		panic("invalid HAKeeperTickInterval")
 	}
-	l.logger.Info("Hakeeper interval configs",
+	l.runtime.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)
@@ -658,7 +658,7 @@ func (l *store) ticker(ctx context.Context) {
 		panic("invalid HAKeeperCheckInterval")
 	}
 	defer func() {
-		l.logger.Info("HAKeeper ticker stopped")
+		l.runtime.Logger().Info("HAKeeper ticker stopped")
 	}()
 	haTicker := time.NewTicker(l.cfg.HAKeeperCheckInterval.Duration)
 	defer haTicker.Stop()
@@ -694,7 +694,7 @@ func (l *store) isLeaderHAKeeper() (bool, uint64, error) {
 func (l *store) hakeeperTick() {
 	isLeader, _, err := l.isLeaderHAKeeper()
 	if err != nil {
-		l.logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
+		l.runtime.Logger().Error("failed to get HAKeeper Leader ID", zap.Error(err))
 		return
 	}
 
@@ -704,7 +704,7 @@ func (l *store) hakeeperTick() {
 		defer cancel()
 		session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 		if _, err := l.propose(ctx, session, cmd); err != nil {
-			l.logger.Error("propose tick failed", zap.Error(err))
+			l.runtime.Logger().Error("propose tick failed", zap.Error(err))
 			return
 		}
 	}
@@ -724,7 +724,7 @@ func (l *store) getHeartbeatMessage() pb.LogStoreHeartbeat {
 	nhi := l.nh.GetNodeHostInfo(opts)
 	for _, ci := range nhi.ShardInfoList {
 		if ci.Pending {
-			l.logger.Info(fmt.Sprintf("shard %d is pending, not included into the heartbeat",
+			l.runtime.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 8ae6a6647..c21d571f4 100644
--- a/pkg/logservice/store_hakeeper_check.go
+++ b/pkg/logservice/store_hakeeper_check.go
@@ -84,14 +84,14 @@ func (l *store) setInitialClusterInfo(numOfLogShards uint64,
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("failed to propose initial cluster info", zap.Error(err))
+		l.runtime.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) {
-		l.logger.Error("initial cluster info already set")
+		l.runtime.Logger().Error("initial cluster info already set")
 	}
 	return nil
 }
@@ -103,7 +103,7 @@ func (l *store) updateIDAlloc(count uint64) error {
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("propose get id failed", zap.Error(err))
+		l.runtime.Logger().Error("propose get id failed", zap.Error(err))
 		return err
 	}
 	// TODO: add a test for this
@@ -114,7 +114,7 @@ func (l *store) updateIDAlloc(count uint64) error {
 func (l *store) hakeeperCheck() {
 	isLeader, term, err := l.isLeaderHAKeeper()
 	if err != nil {
-		l.logger.Error("failed to get HAKeeper Leader ID", zap.Error(err))
+		l.runtime.Logger().Error("failed to get HAKeeper Leader ID", zap.Error(err))
 		return
 	}
 
@@ -125,12 +125,12 @@ func (l *store) hakeeperCheck() {
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
-		l.logger.Error("failed to get checker state", zap.Error(err))
+		l.runtime.Logger().Error("failed to get checker state", zap.Error(err))
 		return
 	}
 	switch state.State {
 	case pb.HAKeeperCreated:
-		l.logger.Warn("waiting for initial cluster info to be set, check skipped")
+		l.runtime.Logger().Warn("waiting for initial cluster info to be set, check skipped")
 		return
 	case pb.HAKeeperBootstrapping:
 		l.bootstrap(term, state)
@@ -150,11 +150,11 @@ func (l *store) assertHAKeeperState(s pb.HAKeeperState) {
 	state, err := l.getCheckerState()
 	if err != nil {
 		// TODO: check whether this is temp error
-		l.logger.Error("failed to get checker state", zap.Error(err))
+		l.runtime.Logger().Error("failed to get checker state", zap.Error(err))
 		return
 	}
 	if state.State != s {
-		l.logger.Panic("unexpected state",
+		l.runtime.Logger().Panic("unexpected state",
 			zap.String("expected", s.String()),
 			zap.String("got", state.State.String()))
 	}
@@ -169,19 +169,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 {
-		l.logger.Error("failed to get check schedule commands", zap.Error(err))
+		l.runtime.Logger().Error("failed to get check schedule commands", zap.Error(err))
 		return
 	}
-	l.logger.Debug(fmt.Sprintf("cluster health check generated %d schedule commands", len(cmds)))
+	l.runtime.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 {
-			l.logger.Debug("adding schedule command to hakeeper", zap.String("command", cmd.LogString()))
+			l.runtime.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
-			l.logger.Debug("failed to add schedule commands", zap.Error(err))
+			l.runtime.Logger().Debug("failed to add schedule commands", zap.Error(err))
 			return
 		}
 	}
@@ -211,25 +211,25 @@ func (l *store) registerTaskUser() {
 
 	// TODO: rename TaskTableUser to moadmin
 	if err := l.setTaskTableUser(user); err != nil {
-		l.logger.Error("failed to set task table user", zap.Error(err))
+		l.runtime.Logger().Error("failed to set task table user", zap.Error(err))
 	}
 }
 
 func (l *store) bootstrap(term uint64, state *pb.CheckerState) {
 	cmds, err := l.getScheduleCommand(false, term, state)
 	if err != nil {
-		l.logger.Error("failed to get bootstrap schedule commands", zap.Error(err))
+		l.runtime.Logger().Error("failed to get bootstrap schedule commands", zap.Error(err))
 		return
 	}
 	if len(cmds) > 0 {
 		for _, c := range cmds {
-			l.logger.Debug("bootstrap cmd", zap.String("cmd", c.LogString()))
+			l.runtime.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
-			l.logger.Debug("failed to add schedule commands", zap.Error(err))
+			l.runtime.Logger().Debug("failed to add schedule commands", zap.Error(err))
 			return
 		}
 		l.bootstrapCheckCycles = checkBootstrapCycles
@@ -304,14 +304,14 @@ func (l *store) setTaskTableUser(user pb.TaskTableUser) error {
 	session := l.nh.GetNoOPSession(hakeeper.DefaultHAKeeperShardID)
 	result, err := l.propose(ctx, session, cmd)
 	if err != nil {
-		l.logger.Error("failed to propose task user info", zap.Error(err))
+		l.runtime.Logger().Error("failed to propose task user info", zap.Error(err))
 		return err
 	}
 	if result.Value == uint64(pb.TaskSchedulerStopped) {
 		panic("failed to set task user")
 	}
 	if result.Value != uint64(pb.TaskSchedulerCreated) {
-		l.logger.Error("task user info already set")
+		l.runtime.Logger().Error("task user info already set")
 	}
 	return nil
 }
diff --git a/pkg/logservice/store_metadata.go b/pkg/logservice/store_metadata.go
index 161b98a82..69bcfb8ea 100644
--- a/pkg/logservice/store_metadata.go
+++ b/pkg/logservice/store_metadata.go
@@ -251,7 +251,7 @@ func (l *store) loadMetadata() error {
 		return err
 	}
 	if expectedUUID != l.mu.metadata.UUID {
-		l.logger.Panic("unexpected UUID",
+		l.runtime.Logger().Panic("unexpected UUID",
 			zap.String("on disk UUID", l.mu.metadata.UUID),
 			zap.String("expect", expectedUUID))
 	}
@@ -262,7 +262,7 @@ func (l *store) mustSaveMetadata() {
 	fs := l.cfg.FS
 	dir := l.cfg.DataDir
 	if err := createMetadataFile(dir, logMetadataFilename, &l.mu.metadata, fs); err != nil {
-		l.logger.Panic("failed to save metadata file", zap.Error(err))
+		l.runtime.Logger().Panic("failed to save metadata file", zap.Error(err))
 	}
 }
 
@@ -275,7 +275,7 @@ func (l *store) addMetadata(shardID uint64, replicaID uint64) {
 
 	for _, rec := range l.mu.metadata.Shards {
 		if rec.ShardID == shardID && rec.ReplicaID == replicaID {
-			l.logger.Info(fmt.Sprintf("addMetadata for shardID %d skipped, dupl shard", shardID))
+			l.runtime.Logger().Info(fmt.Sprintf("addMetadata for shardID %d skipped, dupl shard", shardID))
 			return
 		}
 	}
diff --git a/pkg/logservice/truncation.go b/pkg/logservice/truncation.go
index e5d1eda50..ec454cdd1 100644
--- a/pkg/logservice/truncation.go
+++ b/pkg/logservice/truncation.go
@@ -80,7 +80,7 @@ func (s shardSnapshotInfo) getSnapshotIndex(shardID uint64) uint64 {
 
 func (l *store) truncationWorker(ctx context.Context) {
 	defer func() {
-		l.logger.Info("truncation worker stopped")
+		l.runtime.Logger().Info("truncation worker stopped")
 	}()
 
 	if l.cfg.TruncateInterval.Duration == 0 {
@@ -95,7 +95,7 @@ func (l *store) truncationWorker(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := l.processTruncateLog(ctx); err != nil {
-				l.logger.Error("truncate failed", zap.Error(err))
+				l.runtime.Logger().Error("truncate failed", zap.Error(err))
 			}
 			select {
 			case <-ctx.Done():
@@ -114,7 +114,7 @@ func (l *store) processTruncateLog(ctx context.Context) error {
 			continue
 		}
 		if err := l.processShardTruncateLog(ctx, shard.ShardID); err != nil {
-			l.logger.Error("process truncate log failed", zap.Error(err))
+			l.runtime.Logger().Error("process truncate log failed", zap.Error(err))
 		}
 	}
 	return nil
@@ -130,12 +130,12 @@ func (l *store) exportSnapshot(ctx context.Context, shardID uint64, replicaID ui
 	// Just take a snapshot and export it. This snapshot is invisible to system.
 	idx, err := l.nh.SyncRequestSnapshot(ctx, shardID, opts)
 	if err != nil {
-		l.logger.Error("request export snapshot failed", zap.Error(err))
+		l.runtime.Logger().Error("request export snapshot failed", zap.Error(err))
 		return err
 	}
 	// Add the exported snapshot to snapshot manager.
 	if err := l.snapshotMgr.Add(shardID, replicaID, idx); err != nil {
-		l.logger.Error("add exported snapshot failed", zap.Error(err))
+		l.runtime.Logger().Error("add exported snapshot failed", zap.Error(err))
 		return err
 	}
 	// forward the snapshot index.
@@ -148,12 +148,12 @@ func (l *store) importSnapshot(
 ) error {
 	// Import a snapshot to override the snapshot in system.
 	if err := l.nh.SyncRequestImportSnapshot(ctx, shardID, replicaID, dir); err != nil {
-		l.logger.Error("import snapshot failed", zap.Error(err))
+		l.runtime.Logger().Error("import snapshot failed", zap.Error(err))
 		return err
 	}
 	// Then remove the exported snapshot in manager.
 	if err := l.snapshotMgr.Remove(shardID, replicaID, lsn); err != nil {
-		l.logger.Error("remove exported snapshots failed")
+		l.runtime.Logger().Error("remove exported snapshots failed")
 		return err
 	}
 	// forward the truncate lsn.
@@ -200,7 +200,7 @@ func (l *store) processShardTruncateLog(ctx context.Context, shardID uint64) err
 
 	lsnInSM, err := l.getTruncatedLsn(ctx, shardID)
 	if err != nil {
-		l.logger.Error("get truncated lsn in state machine failed",
+		l.runtime.Logger().Error("get truncated lsn in state machine failed",
 			zap.Uint64("shard ID", shardID), zap.Error(err))
 		return err
 	}
@@ -213,7 +213,7 @@ func (l *store) processShardTruncateLog(ctx context.Context, shardID uint64) err
 	dir, lsn := l.snapshotMgr.EvalImportSnapshot(shardID, replicaID, lsnInSM)
 	if l.shouldDoImport(shardID, lsn, dir) {
 		if err := l.importSnapshot(ctx, shardID, replicaID, lsn, dir); err != nil {
-			l.logger.Error("do truncate log failed",
+			l.runtime.Logger().Error("do truncate log failed",
 				zap.Uint64("shard ID", shardID),
 				zap.Uint64("replica ID", replicaID),
 				zap.Uint64("lsn", lsn),
@@ -226,7 +226,7 @@ func (l *store) processShardTruncateLog(ctx context.Context, shardID uint64) err
 
 	if l.shouldDoExport(ctx, shardID, replicaID) {
 		if err := l.exportSnapshot(ctx, shardID, replicaID); err != nil {
-			l.logger.Error("export snapshot failed",
+			l.runtime.Logger().Error("export snapshot failed",
 				zap.Uint64("shard ID", shardID),
 				zap.Uint64("replica ID", replicaID),
 				zap.Uint64("lsn", lsn),
diff --git a/pkg/pb/task/task.go b/pkg/pb/task/task.go
index 8aeacf561..c790e6197 100644
--- a/pkg/pb/task/task.go
+++ b/pkg/pb/task/task.go
@@ -24,6 +24,11 @@ func (m Task) IsDelayTask() bool {
 	return m.Metadata.Options.DelayDuration > 0
 }
 
+// IsInitTask returns true if the task is an init task
+func (m Task) IsInitTask() bool {
+	return m.Metadata.Executor == TaskCode_SystemInit
+}
+
 // GetDelayDuration returns delay duration
 func (m Task) GetDelayDuration() time.Duration {
 	return time.Duration(m.Metadata.Options.DelayDuration)
diff --git a/pkg/pb/task/task.pb.go b/pkg/pb/task/task.pb.go
index 52b404c6c..08dc2f4ec 100644
--- a/pkg/pb/task/task.pb.go
+++ b/pkg/pb/task/task.pb.go
@@ -121,7 +121,7 @@ type TaskMetadata struct {
 	// ID task id, global unique
 	ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"`
 	// Executor used to select a particular task executor to run a task
-	Executor uint32 `protobuf:"varint,2,opt,name=Executor,proto3" json:"Executor,omitempty"`
+	Executor TaskCode `protobuf:"varint,2,opt,name=Executor,proto3,enum=task.TaskCode" json:"Executor,omitempty"`
 	// Context context needed to run the task
 	Context []byte `protobuf:"bytes,3,opt,name=Context,proto3" json:"Context,omitempty"`
 	// Options options for execute task
@@ -171,11 +171,11 @@ func (m *TaskMetadata) GetID() string {
 	return ""
 }
 
-func (m *TaskMetadata) GetExecutor() uint32 {
+func (m *TaskMetadata) GetExecutor() TaskCode {
 	if m != nil {
 		return m.Executor
 	}
-	return 0
+	return TaskCode_TestOnly
 }
 
 func (m *TaskMetadata) GetContext() []byte {
@@ -564,48 +564,48 @@ func init() {
 func init() { proto.RegisterFile("task.proto", fileDescriptor_ce5d8dd45b4a91ff) }
 
 var fileDescriptor_ce5d8dd45b4a91ff = []byte{
-	// 645 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x54, 0x4f, 0x6f, 0xd3, 0x4e,
-	0x10, 0x8d, 0x93, 0x34, 0x7f, 0x26, 0x7f, 0xe4, 0xdf, 0xfe, 0x38, 0x58, 0x39, 0x84, 0x28, 0x2a,
-	0x52, 0x54, 0x89, 0x46, 0x14, 0x38, 0x20, 0x21, 0xa1, 0x36, 0x29, 0x22, 0xa2, 0xa1, 0x68, 0x1b,
-	0x2e, 0xdc, 0x36, 0xc9, 0xe0, 0x5a, 0x4d, 0xbc, 0xd6, 0x7a, 0x8c, 0x92, 0x13, 0x47, 0xbe, 0x5a,
-	0x8f, 0xfd, 0x04, 0x08, 0x22, 0xee, 0x7c, 0x05, 0xb4, 0xbb, 0x89, 0x1b, 0xf7, 0xcc, 0xcd, 0xef,
-	0xbd, 0x99, 0x9d, 0x99, 0x37, 0x23, 0x03, 0x90, 0x88, 0x6f, 0x8e, 0x23, 0x25, 0x49, 0xb2, 0xa2,
-	0xfe, 0x6e, 0x3d, 0xf5, 0x03, 0xba, 0x4e, 0xa6, 0xc7, 0x33, 0xb9, 0xec, 0xfb, 0xd2, 0x97, 0x7d,
-	0x23, 0x4e, 0x93, 0x2f, 0x06, 0x19, 0x60, 0xbe, 0x6c, 0x52, 0xf7, 0xbb, 0x03, 0xf5, 0x89, 0x88,
-	0x6f, 0xc6, 0x48, 0x62, 0x2e, 0x48, 0xb0, 0x26, 0xe4, 0x47, 0x43, 0xcf, 0xe9, 0x38, 0xbd, 0x2a,
-	0xcf, 0x8f, 0x86, 0xac, 0x05, 0x95, 0xf3, 0x15, 0xce, 0x12, 0x92, 0xca, 0xcb, 0x77, 0x9c, 0x5e,
-	0x83, 0xa7, 0x98, 0x79, 0x50, 0x1e, 0xc8, 0x90, 0x70, 0x45, 0x5e, 0xa1, 0xe3, 0xf4, 0xea, 0x7c,
-	0x07, 0xd9, 0x33, 0x28, 0x5f, 0x46, 0x14, 0xc8, 0x30, 0xf6, 0x8a, 0x1d, 0xa7, 0x57, 0x3b, 0xf9,
-	0xef, 0xd8, 0x74, 0xaa, 0x4b, 0x6d, 0x85, 0xb3, 0xe2, 0xed, 0x8f, 0xc7, 0x39, 0xbe, 0x8b, 0xeb,
-	0x7e, 0x83, 0xda, 0x9e, 0xca, 0x0e, 0xa1, 0x31, 0x16, 0x2b, 0x8e, 0xa4, 0xd6, 0x93, 0x60, 0x89,
-	0xb1, 0x69, 0xa9, 0xc1, 0xb3, 0xa4, 0x8e, 0x32, 0x68, 0x14, 0x12, 0xaa, 0xaf, 0x62, 0x61, 0x5a,
-	0x2c, 0xf0, 0x2c, 0xa9, 0xa3, 0x86, 0xb8, 0x10, 0xeb, 0x61, 0xa2, 0x84, 0x7e, 0xdd, 0x74, 0x5b,
-	0xe0, 0x59, 0xb2, 0xfb, 0x1e, 0x1a, 0x76, 0x32, 0xe4, 0x18, 0x27, 0x0b, 0x62, 0x87, 0x50, 0x1c,
-	0xc8, 0x39, 0x9a, 0xca, 0xcd, 0x13, 0xd7, 0x4e, 0x60, 0x35, 0xcd, 0x73, 0xa3, 0xb2, 0x47, 0x70,
-	0x70, 0xae, 0xd4, 0xd6, 0x9d, 0x2a, 0xb7, 0xa0, 0xfb, 0x27, 0x0f, 0x45, 0x3d, 0xce, 0x9e, 0x9f,
-	0x45, 0xe3, 0xe7, 0x0b, 0xa8, 0xec, 0xbc, 0x36, 0x19, 0xb5, 0x13, 0x76, 0x6f, 0xcd, 0x4e, 0xd9,
-	0x7a, 0x93, 0x46, 0xb2, 0x2e, 0xd4, 0x3f, 0x0a, 0x85, 0x21, 0xe9, 0xa8, 0xd1, 0xd0, 0x0c, 0x50,
-	0xe5, 0x19, 0x8e, 0xf5, 0xa0, 0x74, 0x45, 0x82, 0x12, 0x6b, 0x79, 0xda, 0xb0, 0x56, 0x2d, 0xcf,
-	0xb7, 0x3a, 0x6b, 0x03, 0x68, 0x96, 0x27, 0x61, 0x88, 0xca, 0x3b, 0x30, 0x6f, 0xed, 0x31, 0x66,
-	0xa4, 0x48, 0xce, 0xae, 0xbd, 0x92, 0xf1, 0xdc, 0x02, 0xed, 0xe2, 0x85, 0x88, 0xe9, 0x1d, 0x0a,
-	0x45, 0x53, 0x14, 0xe4, 0x95, 0xad, 0x8b, 0x19, 0x52, 0xdf, 0xcb, 0x40, 0xa1, 0x20, 0x3c, 0x25,
-	0xaf, 0x62, 0x02, 0x52, 0xcc, 0x3a, 0x50, 0x1b, 0xc8, 0x65, 0xb4, 0x40, 0xc2, 0xf9, 0x29, 0x79,
-	0x55, 0x23, 0xef, 0x53, 0xec, 0xd5, 0x83, 0x1d, 0x78, 0x60, 0x2c, 0xfa, 0xdf, 0x8e, 0x92, 0x91,
-	0x78, 0x36, 0xb2, 0xfb, 0xdb, 0xd1, 0x95, 0x65, 0xf8, 0x0f, 0x5d, 0x6f, 0xd9, 0x17, 0xcf, 0x57,
-	0x91, 0xda, 0x3a, 0x9e, 0x62, 0xad, 0x7d, 0xc0, 0x15, 0xe9, 0x33, 0x34, 0x7e, 0x17, 0x78, 0x8a,
-	0xf5, 0xb6, 0x26, 0x2a, 0xf0, 0x7d, 0x54, 0xf6, 0x74, 0x0f, 0x4c, 0x1f, 0x19, 0x2e, 0xe3, 0x53,
-	0xe9, 0x81, 0x4f, 0x2d, 0xa8, 0x7c, 0x8a, 0xe6, 0x56, 0xb3, 0x26, 0xa7, 0xf8, 0xe8, 0xa5, 0xdd,
-	0xdd, 0x76, 0x93, 0x35, 0x28, 0xdb, 0xac, 0xb9, 0x9b, 0xd3, 0x40, 0x2f, 0x30, 0x08, 0x7d, 0xd7,
-	0x61, 0x0d, 0xa8, 0xa6, 0xc6, 0xba, 0xf9, 0xa3, 0xd7, 0x50, 0xd1, 0x69, 0xe6, 0x62, 0xeb, 0x50,
-	0x99, 0x60, 0x4c, 0x97, 0xe1, 0x62, 0xed, 0xe6, 0x58, 0x13, 0xe0, 0x6a, 0x1d, 0x13, 0x2e, 0x47,
-	0x61, 0x40, 0xae, 0xc3, 0x18, 0x34, 0xc7, 0x48, 0x2a, 0x98, 0x5d, 0x48, 0x7f, 0x8c, 0xca, 0x47,
-	0x37, 0x7f, 0xf4, 0x04, 0xe0, 0xfe, 0xee, 0x75, 0x9d, 0xab, 0x64, 0x36, 0xc3, 0x38, 0x76, 0x73,
-	0x0c, 0xa0, 0xf4, 0x56, 0x04, 0x0b, 0x9c, 0xbb, 0xce, 0xd9, 0x9b, 0xbb, 0x5f, 0x6d, 0xe7, 0x76,
-	0xd3, 0x76, 0xee, 0x36, 0x6d, 0xe7, 0xe7, 0xa6, 0xed, 0x7c, 0xde, 0xff, 0x1b, 0x2d, 0x05, 0xa9,
-	0x60, 0x25, 0x55, 0xe0, 0x07, 0xe1, 0x0e, 0x84, 0xd8, 0x8f, 0x6e, 0xfc, 0x7e, 0x34, 0xed, 0xeb,
-	0x6d, 0x4c, 0x4b, 0xe6, 0xa7, 0xf4, 0xfc, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x90, 0x7c, 0x5d,
-	0xf0, 0xd7, 0x04, 0x00, 0x00,
+	// 653 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x94, 0xcb, 0x6e, 0x1a, 0x49,
+	0x14, 0x86, 0x69, 0xc0, 0x5c, 0x0e, 0x17, 0xf5, 0xd4, 0xcc, 0xa2, 0xc5, 0x82, 0x41, 0xc8, 0x23,
+	0x21, 0xa4, 0x31, 0x0a, 0x49, 0x16, 0x91, 0x22, 0x45, 0x36, 0x38, 0x0a, 0x8a, 0x89, 0xa3, 0x32,
+	0xd9, 0x64, 0x57, 0xc0, 0x49, 0xbb, 0x65, 0xe8, 0x6a, 0x55, 0x9f, 0x8e, 0x60, 0x95, 0xc7, 0xc8,
+	0x2b, 0x79, 0xe9, 0x27, 0x88, 0x12, 0x2b, 0xfb, 0xbc, 0x42, 0x54, 0x55, 0xd0, 0xa6, 0xbd, 0xce,
+	0x8e, 0xff, 0xff, 0x4f, 0x5d, 0xce, 0x77, 0x8a, 0x06, 0x20, 0x11, 0xdf, 0x9c, 0x44, 0x4a, 0x92,
+	0x64, 0x45, 0xfd, 0xbb, 0xf5, 0xbf, 0x1f, 0xd0, 0x75, 0x32, 0x3f, 0x59, 0xc8, 0xf5, 0xc0, 0x97,
+	0xbe, 0x1c, 0x98, 0x70, 0x9e, 0x7c, 0x32, 0xca, 0x08, 0xf3, 0xcb, 0x2e, 0xea, 0x7e, 0x75, 0xa0,
+	0x3e, 0x13, 0xf1, 0xcd, 0x14, 0x49, 0x2c, 0x05, 0x09, 0xd6, 0x84, 0xfc, 0x64, 0xec, 0x39, 0x1d,
+	0xa7, 0x57, 0xe5, 0xf9, 0xc9, 0x98, 0xf5, 0xa1, 0x72, 0xbe, 0xc1, 0x45, 0x42, 0x52, 0x79, 0xf9,
+	0x8e, 0xd3, 0x6b, 0x0e, 0x9b, 0x27, 0xe6, 0x50, 0xbd, 0x6a, 0x24, 0x97, 0xc8, 0xd3, 0x9c, 0x79,
+	0x50, 0x1e, 0xc9, 0x90, 0x70, 0x43, 0x5e, 0xa1, 0xe3, 0xf4, 0xea, 0x7c, 0x2f, 0xd9, 0x13, 0x28,
+	0x5f, 0x46, 0x14, 0xc8, 0x30, 0xf6, 0x8a, 0x1d, 0xa7, 0x57, 0x1b, 0xfe, 0xf5, 0xb0, 0xc9, 0x2e,
+	0x38, 0x2b, 0xde, 0x7e, 0xfb, 0x37, 0xc7, 0xf7, 0x75, 0xdd, 0x2f, 0x50, 0x3b, 0x48, 0xd9, 0x31,
+	0x34, 0xa6, 0x62, 0xc3, 0x91, 0xd4, 0x76, 0x16, 0xac, 0x31, 0x36, 0x57, 0x6c, 0xf0, 0xac, 0xa9,
+	0xab, 0x8c, 0x9a, 0x84, 0x84, 0xea, 0xb3, 0x58, 0x99, 0x2b, 0x17, 0x78, 0xd6, 0xd4, 0x55, 0x63,
+	0x5c, 0x89, 0xed, 0x38, 0x51, 0x42, 0xef, 0x6e, 0x6e, 0x5b, 0xe0, 0x59, 0xb3, 0xfb, 0x16, 0x1a,
+	0xb6, 0x33, 0xe4, 0x18, 0x27, 0x2b, 0x62, 0xc7, 0x50, 0xd4, 0x0d, 0x9b, 0x93, 0x9b, 0x43, 0xd7,
+	0x76, 0x60, 0x33, 0x03, 0xc2, 0xa4, 0xec, 0x1f, 0x38, 0x3a, 0x57, 0x6a, 0x47, 0xab, 0xca, 0xad,
+	0xe8, 0xfe, 0xca, 0x43, 0x51, 0xb7, 0x73, 0xc0, 0xb7, 0x68, 0xf8, 0x3e, 0x83, 0xca, 0x9e, 0xbd,
+	0x59, 0x51, 0x1b, 0xb2, 0x07, 0x34, 0xfb, 0x64, 0xc7, 0x26, 0xad, 0x64, 0x5d, 0xa8, 0xbf, 0x17,
+	0x0a, 0x43, 0xd2, 0x55, 0x93, 0xb1, 0x69, 0xa0, 0xca, 0x33, 0x1e, 0xeb, 0x41, 0xe9, 0x8a, 0x04,
+	0x25, 0x16, 0x79, 0x7a, 0x61, 0x9d, 0x5a, 0x9f, 0xef, 0x72, 0xd6, 0x06, 0xd0, 0x2e, 0x4f, 0xc2,
+	0x10, 0x95, 0x77, 0x64, 0xf6, 0x3a, 0x70, 0x4c, 0x4b, 0x91, 0x5c, 0x5c, 0x7b, 0x25, 0xc3, 0xdc,
+	0x0a, 0x4d, 0xf1, 0x42, 0xc4, 0xf4, 0x06, 0x85, 0xa2, 0x39, 0x0a, 0xf2, 0xca, 0x96, 0x62, 0xc6,
+	0x64, 0x2d, 0xa8, 0x8c, 0x14, 0x0a, 0xc2, 0x53, 0xf2, 0x2a, 0xa6, 0x20, 0xd5, 0xac, 0x03, 0xb5,
+	0x91, 0x5c, 0x47, 0x2b, 0x24, 0x5c, 0x9e, 0x92, 0x57, 0x35, 0xf1, 0xa1, 0xc5, 0x5e, 0x3c, 0x9a,
+	0x81, 0x07, 0x06, 0xd1, 0xdf, 0xb6, 0x95, 0x4c, 0xc4, 0xb3, 0x95, 0xdd, 0x9f, 0x8e, 0x3e, 0x59,
+	0x86, 0x7f, 0x90, 0x7a, 0xcb, 0xee, 0x78, 0xbe, 0x89, 0xd4, 0x8e, 0x78, 0xaa, 0x75, 0xf6, 0x0e,
+	0x37, 0xa4, 0x9f, 0xa1, 0xe1, 0x5d, 0xe0, 0xa9, 0xd6, 0xd3, 0x9a, 0xa9, 0xc0, 0xf7, 0x51, 0xd9,
+	0xa7, 0x7b, 0x64, 0xee, 0x91, 0xf1, 0x32, 0x9c, 0x4a, 0x8f, 0x38, 0xb5, 0xa0, 0xf2, 0x21, 0x5a,
+	0xda, 0xcc, 0x42, 0x4e, 0x75, 0xff, 0xb9, 0x9d, 0xdd, 0x6e, 0x92, 0x35, 0x28, 0xdb, 0x55, 0x4b,
+	0x37, 0xa7, 0x85, 0x1e, 0x60, 0x10, 0xfa, 0xae, 0xc3, 0x1a, 0x50, 0x4d, 0xc1, 0xba, 0xf9, 0xfe,
+	0x4b, 0xa8, 0xec, 0xff, 0xc0, 0xac, 0x0e, 0x95, 0x19, 0xc6, 0x74, 0x19, 0xae, 0xb6, 0x6e, 0x8e,
+	0x35, 0x01, 0xae, 0xb6, 0x31, 0xe1, 0x7a, 0x12, 0x06, 0xe4, 0x3a, 0x8c, 0x41, 0x73, 0x8a, 0xa4,
+	0x82, 0xc5, 0x85, 0xf4, 0xa7, 0xa8, 0x7c, 0x74, 0xf3, 0xfd, 0xff, 0x00, 0x1e, 0xde, 0xbd, 0x3e,
+	0xe7, 0x2a, 0x59, 0x2c, 0x30, 0x8e, 0xdd, 0x1c, 0x03, 0x28, 0xbd, 0x16, 0xc1, 0x0a, 0x97, 0xae,
+	0x73, 0xf6, 0xea, 0xee, 0x47, 0xdb, 0xb9, 0xbd, 0x6f, 0x3b, 0x77, 0xf7, 0x6d, 0xe7, 0xfb, 0x7d,
+	0xdb, 0xf9, 0x78, 0xf8, 0x75, 0x5a, 0x0b, 0x52, 0xc1, 0x46, 0xaa, 0xc0, 0x0f, 0xc2, 0xbd, 0x08,
+	0x71, 0x10, 0xdd, 0xf8, 0x83, 0x68, 0x3e, 0xd0, 0xd3, 0x98, 0x97, 0xcc, 0x47, 0xea, 0xe9, 0xef,
+	0x00, 0x00, 0x00, 0xff, 0xff, 0x13, 0x58, 0xba, 0x6a, 0xe7, 0x04, 0x00, 0x00,
 }
 
 func (m *TaskMetadata) Marshal() (dAtA []byte, err error) {
@@ -1141,7 +1141,7 @@ func (m *TaskMetadata) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.Executor |= uint32(b&0x7F) << shift
+				m.Executor |= TaskCode(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
diff --git a/pkg/taskservice/task_runner.go b/pkg/taskservice/task_runner.go
index bb5483b6c..f85f0018c 100644
--- a/pkg/taskservice/task_runner.go
+++ b/pkg/taskservice/task_runner.go
@@ -122,7 +122,7 @@ type taskRunner struct {
 	mu struct {
 		sync.RWMutex
 		started      bool
-		executors    map[uint32]TaskExecutor
+		executors    map[task.TaskCode]TaskExecutor
 		runningTasks map[uint64]runningTask
 		retryTasks   []runningTask
 	}
@@ -145,7 +145,7 @@ func NewTaskRunner(runnerID string, service TaskService, opts ...RunnerOption) T
 		runnerID: runnerID,
 		service:  service,
 	}
-	r.mu.executors = make(map[uint32]TaskExecutor)
+	r.mu.executors = make(map[task.TaskCode]TaskExecutor)
 	for _, opt := range opts {
 		opt(r)
 	}
@@ -239,12 +239,12 @@ func (r *taskRunner) Parallelism() int {
 	return r.options.parallelism
 }
 
-func (r *taskRunner) RegisterExecutor(code uint32, executor TaskExecutor) {
+func (r *taskRunner) RegisterExecutor(code task.TaskCode, executor TaskExecutor) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
 	if _, ok := r.mu.executors[code]; !ok {
-		r.logger.Debug("executor registered", zap.Uint32("code", code))
+		r.logger.Debug("executor registered", zap.Any("code", code))
 		r.mu.executors[code] = executor
 	}
 }
@@ -534,7 +534,7 @@ func (r *taskRunner) removeRunningTask(id uint64) {
 	delete(r.mu.runningTasks, id)
 }
 
-func (r *taskRunner) getExecutor(code uint32) (TaskExecutor, error) {
+func (r *taskRunner) getExecutor(code task.TaskCode) (TaskExecutor, error) {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 
diff --git a/pkg/taskservice/types.go b/pkg/taskservice/types.go
index 733673409..9b76c90ce 100644
--- a/pkg/taskservice/types.go
+++ b/pkg/taskservice/types.go
@@ -73,7 +73,7 @@ type conditions struct {
 
 	hasTaskExecutorCond bool
 	taskExecutorOp      Op
-	taskExecutor        uint32
+	taskExecutor        task.TaskCode
 
 	orderByDesc bool
 }
@@ -86,7 +86,7 @@ func WithTaskIDDesc() Condition {
 }
 
 // WithTaskExecutorCond set task executor condition
-func WithTaskExecutorCond(op Op, value uint32) Condition {
+func WithTaskExecutorCond(op Op, value task.TaskCode) Condition {
 	return func(qo *conditions) {
 		qo.hasTaskExecutorCond = true
 		qo.taskExecutorOp = op
@@ -204,7 +204,7 @@ type TaskRunner interface {
 	// Parallelism maximum number of concurrently executing Tasks
 	Parallelism() int
 	// RegisterExecutor register the task executor
-	RegisterExecutor(code uint32, executor TaskExecutor)
+	RegisterExecutor(code task.TaskCode, executor TaskExecutor)
 }
 
 // TaskStorage task storage
diff --git a/pkg/util/export/merge.go b/pkg/util/export/merge.go
index acca74a22..ea0f50b1d 100644
--- a/pkg/util/export/merge.go
+++ b/pkg/util/export/merge.go
@@ -703,7 +703,7 @@ const ParamSeparator = " "
 func MergeTaskMetadata(id task.TaskCode, args ...string) task.TaskMetadata {
 	return task.TaskMetadata{
 		ID:       path.Join("ETL_merge_task", path.Join(args...)),
-		Executor: uint32(id),
+		Executor: id,
 		Context:  []byte(strings.Join(args, ParamSeparator)),
 	}
 }
diff --git a/proto/task.proto b/proto/task.proto
index 9cbdca92e..bba36153d 100644
--- a/proto/task.proto
+++ b/proto/task.proto
@@ -47,7 +47,7 @@
      // ID task id, global unique
      string      ID       = 1;
      // Executor used to select a particular task executor to run a task
-     uint32      Executor = 2;
+     TaskCode    Executor = 2;
      // Context context needed to run the task
      bytes       Context  = 3;
      // Options options for execute task


Test patch:
diff --git a/pkg/common/log/example_log_test.go b/pkg/common/log/example_log_test.go
index c9d64ee2e..01897cd33 100644
--- a/pkg/common/log/example_log_test.go
+++ b/pkg/common/log/example_log_test.go
@@ -95,10 +95,10 @@ func ExampleMOLogger_LogAction() {
 	// 2022/11/17 15:28:16.600754 +0800 INFO cn-service log/example_log_test.go:127 do action {"uuid": "cn0", "cost": "1.001430792s"}
 }
 
-func ExampleLogOptions_WithProcess() {
-	processStep1InCN("txn uuid")
-	processStep2InDN("txn uuid")
-	processStep3InLOG("txn uuid")
+func ExampleMOLogger_WithProcess() {
+	processStep1InCNWithoutID()
+	processStep2InDNWithoutID()
+	processStep3InLOGWithoutID()
 
 	// Output logs:
 	// 2022/11/17 15:36:04.724470 +0800 INFO cn-service log/example_log_test.go:131 step 1 {"uuid": "cn0", "process": "txn", "process-id": "txn uuid"}
@@ -118,12 +118,38 @@ func ExampleLogOptions_WithSample() {
 	// 2022/11/17 15:43:14.645242 +0800 INFO cn-service log/example_log_test.go:116 example sample log {"uuid": "cn0"}
 }
 
+func ExampleLogOptions_WithProcess() {
+	processStep1InCN("txn uuid")
+	processStep2InDN("txn uuid")
+	processStep3InLOG("txn uuid")
+
+	// Output logs:
+	// 2022/11/17 15:36:04.724470 +0800 INFO cn-service log/logger.go:51 step 1 {"uuid": "cn0", "process": "txn", "process-id": "txn uuid"}
+	// 2022/11/17 15:36:04.724797 +0800 INFO dn-service log/logger.go:51 step 2 {"uuid": "dn0", "process": "txn", "process-id": "txn uuid"}
+	// 2022/11/17 15:36:04.724812 +0800 INFO log-service log/logger.go:51 step 3 {"uuid": "log0", "process": "txn", "process-id": "txn uuid"}
+}
+
 func someAction() {
 	logger := getServiceLogger(metadata.ServiceType_CN, "cn0")
 	defer logger.InfoAction("do action")()
 	time.Sleep(time.Second)
 }
 
+func processStep1InCNWithoutID() {
+	logger := getServiceLogger(metadata.ServiceType_CN, "cn0").WithProcess(log.Txn)
+	logger.Log("step 1", log.DefaultLogOptions())
+}
+
+func processStep2InDNWithoutID() {
+	logger := getServiceLogger(metadata.ServiceType_DN, "dn0").WithProcess(log.Txn)
+	logger.Log("step 2", log.DefaultLogOptions())
+}
+
+func processStep3InLOGWithoutID() {
+	logger := getServiceLogger(metadata.ServiceType_LOG, "log0").WithProcess(log.Txn)
+	logger.Log("step 3", log.DefaultLogOptions())
+}
+
 func processStep1InCN(id string) {
 	logger := getServiceLogger(metadata.ServiceType_CN, "cn0")
 	logger.Log("step 1", log.DefaultLogOptions().WithProcess(log.Txn, id))
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index 7cfce5b50..4a412b101 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 )
@@ -177,9 +178,8 @@ func TestHandleAddReplica(t *testing.T) {
 	}()
 
 	service1 := Service{
-		store: store1,
-
-		logger: testLogger,
+		store:   store1,
+		runtime: runtime.DefaultRuntime(),
 	}
 	cmd := pb.ScheduleCommand{
 		ConfigChange: &pb.ConfigChange{
@@ -207,8 +207,8 @@ func TestHandleRemoveReplica(t *testing.T) {
 	}()
 
 	service1 := Service{
-		store:  store1,
-		logger: testLogger,
+		store:   store1,
+		runtime: runtime.DefaultRuntime(),
 	}
 	cmd := pb.ScheduleCommand{
 		ConfigChange: &pb.ConfigChange{
diff --git a/pkg/logservice/store_metadata_test.go b/pkg/logservice/store_metadata_test.go
index b2bab4956..b03b1f0ad 100644
--- a/pkg/logservice/store_metadata_test.go
+++ b/pkg/logservice/store_metadata_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 )
 
@@ -59,7 +60,7 @@ func TestAddMetadataRejectDupl(t *testing.T) {
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	cfg.Fill()
-	s := store{cfg: cfg, logger: testLogger}
+	s := store{cfg: cfg, runtime: runtime.DefaultRuntime()}
 	require.NoError(t, mkdirAll(s.cfg.DataDir, cfg.FS))
 	s.addMetadata(10, 1)
 	s.addMetadata(10, 1)
@@ -71,12 +72,12 @@ func TestRemoveMetadata(t *testing.T) {
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
 	cfg.Fill()
-	s := store{cfg: cfg}
+	s := store{cfg: cfg, runtime: runtime.DefaultRuntime()}
 	require.NoError(t, mkdirAll(s.cfg.DataDir, cfg.FS))
 	s.addMetadata(10, 1)
 	s.addMetadata(20, 2)
 	s.removeMetadata(10, 1)
-	ss := store{cfg: s.cfg}
+	ss := store{cfg: s.cfg, runtime: runtime.DefaultRuntime()}
 	ss.mu.metadata = metadata.LogStore{}
 	assert.NoError(t, ss.loadMetadata())
 	require.Equal(t, 1, len(ss.mu.metadata.Shards))
diff --git a/pkg/logservice/store_test.go b/pkg/logservice/store_test.go
index 76a0990be..8dc12e24a 100644
--- a/pkg/logservice/store_test.go
+++ b/pkg/logservice/store_test.go
@@ -26,14 +26,26 @@ import (
 	"github.com/lni/goutils/leaktest"
 	"github.com/lni/vfs"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"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"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
 )
 
+func TestMain(m *testing.M) {
+	logutil.SetupMOLogger(&logutil.LogConfig{
+		Level:  "debug",
+		Format: "console",
+	})
+
+	runtime.SetupProcessLevelRuntime(runtime.NewRuntime(metadata.ServiceType_LOG, "test", logutil.GetGlobalLogger()))
+	m.Run()
+}
+
 var (
 	testIOTimeout = 5 * time.Second
 )
@@ -74,17 +86,17 @@ func TestStoreCanBeCreatedAndClosed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg, nil, testLogger)
+	store, err := newLogStore(cfg, nil, runtime.DefaultRuntime())
 	assert.NoError(t, err)
-	testLogger.Info("1")
+	runtime.DefaultRuntime().Logger().Info("1")
 	defer func() {
 		assert.NoError(t, store.close())
 	}()
-	testLogger.Info("2")
+	runtime.DefaultRuntime().Logger().Info("2")
 }
 
 func getTestStore(cfg Config, startLogReplica bool, taskService taskservice.TaskService) (*store, error) {
-	store, err := newLogStore(cfg, func() taskservice.TaskService { return taskService }, testLogger)
+	store, err := newLogStore(cfg, func() taskservice.TaskService { return taskService }, runtime.DefaultRuntime())
 	if err != nil {
 		return nil, err
 	}
@@ -103,7 +115,7 @@ func TestHAKeeperCanBeStarted(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getStoreTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	store, err := newLogStore(cfg, nil, nil)
+	store, err := newLogStore(cfg, nil, runtime.DefaultRuntime())
 	assert.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[2] = store.nh.ID()
@@ -429,7 +441,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, testLogger.With(zap.String("store", cfg1.DataDir)))
+	store1, err := newLogStore(cfg1, nil, runtime.DefaultRuntime())
 	if err != nil {
 		return nil, nil, err
 	}
@@ -444,7 +456,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, testLogger.With(zap.String("store", cfg2.DataDir)))
+	store2, err := newLogStore(cfg2, nil, runtime.DefaultRuntime())
 	if err != nil {
 		return nil, nil, err
 	}
@@ -472,7 +484,7 @@ func getTestStores() (*store, *store, error) {
 		}
 		if ok && leaderID != 1 {
 			if err := store1.requestLeaderTransfer(1, 1); err != nil {
-				testLogger.Error("failed to transfer leader")
+				runtime.DefaultRuntime().Logger().Error("failed to transfer leader")
 			}
 		}
 		time.Sleep(time.Millisecond)
diff --git a/pkg/tests/service/service.go b/pkg/tests/service/service.go
index b7a19ec9d..4d61ce0e5 100644
--- a/pkg/tests/service/service.go
+++ b/pkg/tests/service/service.go
@@ -301,7 +301,7 @@ func NewCluster(t *testing.T, opt Options) (Cluster, error) {
 	}
 
 	// TODO: CN and LOG use process level runtime
-	runtime.SetupProcessLevelRuntime(c.newRuntime(metadata.ServiceType_CN, ""))
+	runtime.SetupProcessLevelRuntime(c.newRuntime())
 
 	// build addresses for all services
 	c.network.addresses = c.buildServiceAddresses()
@@ -1295,7 +1295,6 @@ func (c *testCluster) buildLogConfigs(
 
 		localAddr := cfg.ServiceAddress
 		opt := buildLogOptions(cfg, c.backendFilterFactory(localAddr))
-		opt = append(opt, logservice.WithLogger(c.logger))
 		opts = append(opts, opt)
 	}
 	return cfgs, opts
@@ -1341,7 +1340,7 @@ func (c *testCluster) initDNServices(fileservices *fileServices) []DNService {
 		}
 		ds, err := newDNService(
 			cfg,
-			c.newRuntime(metadata.ServiceType_DN, cfg.UUID),
+			c.newRuntime(),
 			fs,
 			opt)
 		require.NoError(c.t, err)
@@ -1614,7 +1613,7 @@ func (c *testCluster) waitSystemInitCompleted(ctx context.Context) error {
 	return nil
 }
 
-func (c *testCluster) newRuntime(st metadata.ServiceType, uuid string) runtime.Runtime {
+func (c *testCluster) newRuntime() runtime.Runtime {
 	return runtime.NewRuntime(metadata.ServiceType_CN, "", c.logger, runtime.WithClock(c.clock))
 }
 
diff --git a/pkg/tests/service/task_test.go b/pkg/tests/service/task_test.go
index 79044e079..4e469c4ad 100644
--- a/pkg/tests/service/task_test.go
+++ b/pkg/tests/service/task_test.go
@@ -204,16 +204,16 @@ func TestTaskSchedulerCanReallocateTask(t *testing.T) {
 
 	cn2, err := c.GetCNServiceIndexed(1)
 	require.NoError(t, err)
-	cn1.GetTaskRunner().RegisterExecutor(uint32(task.TaskCode_TestOnly), taskExecutor)
-	cn2.GetTaskRunner().RegisterExecutor(uint32(task.TaskCode_TestOnly), taskExecutor)
+	cn1.GetTaskRunner().RegisterExecutor(task.TaskCode_TestOnly, taskExecutor)
+	cn2.GetTaskRunner().RegisterExecutor(task.TaskCode_TestOnly, taskExecutor)
 
 	taskService, ok := cn1.GetTaskService()
 	require.True(t, ok)
-	err = taskService.Create(context.TODO(), task.TaskMetadata{ID: "a", Executor: uint32(task.TaskCode_TestOnly)})
+	err = taskService.Create(context.TODO(), task.TaskMetadata{ID: "a", Executor: task.TaskCode_TestOnly})
 	require.NoError(t, err)
 
 	tasks, err := taskService.QueryTask(ctx,
-		taskservice.WithTaskExecutorCond(taskservice.EQ, uint32(task.TaskCode_TestOnly)))
+		taskservice.WithTaskExecutorCond(taskservice.EQ, task.TaskCode_TestOnly))
 	require.NoError(t, err)
 	require.Equal(t, 1, len(tasks))
 
@@ -265,12 +265,12 @@ func TestTaskRunner(t *testing.T) {
 	indexed, err := c.GetCNServiceIndexed(0)
 	require.NoError(t, err)
 
-	indexed.GetTaskRunner().RegisterExecutor(uint32(task.TaskCode_TestOnly), taskExecutor)
+	indexed.GetTaskRunner().RegisterExecutor(task.TaskCode_TestOnly, taskExecutor)
 
 	taskService, ok := indexed.GetTaskService()
 	require.True(t, ok)
 
-	err = taskService.Create(context.TODO(), task.TaskMetadata{ID: "a", Executor: uint32(task.TaskCode_TestOnly)})
+	err = taskService.Create(context.TODO(), task.TaskMetadata{ID: "a", Executor: task.TaskCode_TestOnly})
 	require.NoError(t, err)
 
 	waitTaskScheduled(t, ctx, taskService)
@@ -320,7 +320,7 @@ func TestCronTask(t *testing.T) {
 	indexed, err := c.GetCNServiceIndexed(0)
 	require.NoError(t, err)
 
-	indexed.GetTaskRunner().RegisterExecutor(uint32(task.TaskCode_TestOnly), taskExecutor)
+	indexed.GetTaskRunner().RegisterExecutor(task.TaskCode_TestOnly, taskExecutor)
 
 	taskService, ok := indexed.GetTaskService()
 	require.True(t, ok)
@@ -328,7 +328,7 @@ func TestCronTask(t *testing.T) {
 	require.NoError(t, taskService.CreateCronTask(context.TODO(),
 		task.TaskMetadata{
 			ID:       "a",
-			Executor: uint32(task.TaskCode_TestOnly),
+			Executor: task.TaskCode_TestOnly,
 		},
 		"*/1 * * * * *", // every 1 second
 	))
diff --git a/test/distributed/cases/log/system_init_log.result b/test/distributed/cases/log/system_init_log.result
new file mode 100644
index 000000000..4a8c047d2
--- /dev/null
+++ b/test/distributed/cases/log/system_init_log.result
@@ -0,0 +1,3 @@
+select message from system.log_info where extra like '%system-init%' and message="HAKeeper Timeout Configs";
+message
+HAKeeper Timeout Configs
\ No newline at end of file
diff --git a/test/distributed/cases/log/system_init_log.sql b/test/distributed/cases/log/system_init_log.sql
new file mode 100644
index 000000000..e430ad81f
--- /dev/null
+++ b/test/distributed/cases/log/system_init_log.sql
@@ -0,0 +1,1 @@
+select message from system.log_info where extra like '%system-init%' and message="HAKeeper Timeout Configs";
\ No newline at end of file