logservice: read code and some tiny refactor (pingcap#2772)

close pingcap#2773

Author: 3AceShowHand

api/middleware/authenticate_middleware.go

@@ -102,7 +102,7 @@ func verify(ctx *gin.Context, etcdCli etcd.Client) error {
 
 // fetchTiDBTopology parses the TiDB topology from etcd.
 func fetchTiDBTopology(ctx context.Context, etcdClient etcd.Client, ks string) ([]upstream.TidbInstance, error) {
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	meta, err := keyspaceManager.LoadKeyspace(ctx, ks)
 	if err != nil {
 		return nil, err

api/middleware/middleware.go

@@ -238,7 +238,7 @@ func KeyspaceCheckerMiddleware() gin.HandlerFunc {
 			return
 		}
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 		meta, err := keyspaceManager.LoadKeyspace(c.Request.Context(), ks)
 		if errors.IsKeyspaceNotExistError(err) {
 			c.IndentedJSON(http.StatusBadRequest, errors.ErrAPIInvalidParam)

api/v2/changefeed.go

@@ -103,7 +103,7 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 		return
 	}
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		_ = c.Error(errors.WrapError(errors.ErrKeyspaceNotFound, err))
@@ -381,7 +381,7 @@ func (h *OpenAPIV2) VerifyTable(c *gin.Context) {
 	}
 	protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(replicaCfg.Sink.Protocol))
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceName := GetKeyspaceValueWithDefault(c)
 	kvStorage, err := keyspaceManager.GetStorage(keyspaceName)
 	if err != nil {
@@ -663,7 +663,7 @@ func (h *OpenAPIV2) ResumeChangefeed(c *gin.Context) {
 		newCheckpointTs = cfg.OverwriteCheckpointTs
 	}
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		_ = c.Error(errors.WrapError(errors.ErrKeyspaceNotFound, err))
@@ -820,7 +820,7 @@ func (h *OpenAPIV2) UpdateChangefeed(c *gin.Context) {
 		}
 		protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(oldCfInfo.Config.Sink.Protocol))
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 
 		kvStorage, err := keyspaceManager.GetStorage(keyspaceName)
 		if err != nil {

api/v2/unsafe.go

@@ -54,7 +54,7 @@ func (h *OpenAPIV2) ResolveLock(c *gin.Context) {
 
 	keyspaceName := GetKeyspaceValueWithDefault(c)
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	// The ctx's lifecycle is the same as the HTTP request.
 	// The schema store may use the context to fetch database information asynchronously.
 	// Therefore, we cannot use the context of the HTTP request.
@@ -93,7 +93,7 @@ func (h *OpenAPIV2) DeleteServiceGcSafePoint(c *gin.Context) {
 	defer pdClient.Close()
 
 	keyspaceName := GetKeyspaceValueWithDefault(c)
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(c.Request.Context(), keyspaceName)
 	if err != nil {
 		_ = c.Error(cerror.WrapError(cerror.ErrKeyspaceNotFound, err))

coordinator/coordinator.go

@@ -269,7 +269,7 @@ func (c *coordinator) handleStateChange(
 	if event.state == config.StateFailed || event.state == config.StateFinished {
 		progress = config.ProgressStopping
 	}
-	if err := c.backend.UpdateChangefeed(context.Background(), cfInfo, cf.GetStatus().CheckpointTs, progress); err != nil {
+	if err = c.backend.UpdateChangefeed(context.Background(), cfInfo, cf.GetStatus().CheckpointTs, progress); err != nil {
 		log.Error("failed to update changefeed state",
 			zap.Error(err))
 		return errors.Trace(err)
@@ -287,7 +287,7 @@ func (c *coordinator) handleStateChange(
 		log.Info("changefeed is resumed or created successfully, try to delete its safeguard gc safepoint",
 			zap.String("changefeed", event.changefeedID.String()))
 
-		keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+		keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 		keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, event.changefeedID.Keyspace())
 		if err != nil {
 			log.Warn("failed to load keyspace", zap.String("keyspace", event.changefeedID.Keyspace()), zap.Error(err))
@@ -473,7 +473,7 @@ func (c *coordinator) updateAllKeyspaceGcBarriers(ctx context.Context) error {
 
 func (c *coordinator) updateKeyspaceGcBarrier(ctx context.Context, barrierMap map[string]uint64, keyspaceName string) error {
 	// Obtain keyspace metadata from PD
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, keyspaceName)
 	if err != nil {
 		return cerror.WrapError(cerror.ErrLoadKeyspaceFailed, err)

downstreamadapter/dispatchermanager/dispatcher_manager.go

@@ -158,7 +158,7 @@ func NewDispatcherManager(
 	ctx, cancel := context.WithCancel(context.Background())
 	pdClock := appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock)
 
-	keyspaceManager := appcontext.GetService[keyspace.KeyspaceManager](appcontext.KeyspaceManager)
+	keyspaceManager := appcontext.GetService[keyspace.Manager](appcontext.KeyspaceManager)
 	keyspaceMeta, err := keyspaceManager.LoadKeyspace(ctx, changefeedID.Keyspace())
 	if err != nil {
 		cancel()

logservice/eventstore/event_store.go

@@ -184,8 +184,8 @@ type eventWithCallback struct {
 
 func eventWithCallbackSizer(e eventWithCallback) int {
 	size := 0
-	for _, e := range e.kvs {
-		size += int(e.KeyLen + e.ValueLen + e.OldValueLen)
+	for _, kv := range e.kvs {
+		size += int(kv.KeyLen + kv.ValueLen + kv.OldValueLen)
 	}
 	return size
 }
@@ -236,7 +236,6 @@ const (
 )
 
 func New(
-	ctx context.Context,
 	root string,
 	subClient logpuller.SubscriptionClient,
 ) EventStore {
@@ -259,6 +258,7 @@ func New(
 		gcManager: newGCManager(),
 
 		subscriptionChangeCh: chann.NewAutoDrainChann[SubscriptionChange](),
+		messageCenter:        appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter),
 
 		decoderPool: &sync.Pool{
 			New: func() any {
@@ -280,6 +280,7 @@ func New(
 	store.dispatcherMeta.dispatcherStats = make(map[common.DispatcherID]*dispatcherStat)
 	store.dispatcherMeta.tableStats = make(map[int64]subscriptionStats)
 
+	store.messageCenter.RegisterHandler(messaging.EventStoreTopic, store.handleMessage)
 	return store
 }
 
@@ -326,11 +327,11 @@ func (p *writeTaskPool) run(ctx context.Context) {
 						return
 					}
 					start := time.Now()
-					if err := p.store.writeEvents(p.db, events, encoder); err != nil {
-						log.Panic("write events failed")
+					if err = p.store.writeEvents(p.db, events, encoder); err != nil {
+						log.Panic("write events failed", zap.Error(err))
 					}
-					for i := range events {
-						events[i].callback()
+					for idx := range events {
+						events[idx].callback()
 					}
 					ioWriteDuration.Observe(time.Since(start).Seconds())
 					totalDuration.Observe(time.Since(totalStart).Seconds())
@@ -362,15 +363,9 @@ func (e *eventStore) Run(ctx context.Context) error {
 	defer func() {
 		log.Info("event store exited")
 	}()
-	eg, ctx := errgroup.WithContext(ctx)
-
-	// recv and handle messages
-	messageCenter := appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter)
-	e.messageCenter = messageCenter
-	messageCenter.RegisterHandler(messaging.EventStoreTopic, e.handleMessage)
 
+	eg, ctx := errgroup.WithContext(ctx)
 	for _, p := range e.writeTaskPools {
-		p := p
 		eg.Go(func() error {
 			p.run(ctx)
 			return nil
@@ -399,7 +394,7 @@ func (e *eventStore) Run(ctx context.Context) error {
 	return eg.Wait()
 }
 
-func (e *eventStore) Close(ctx context.Context) error {
+func (e *eventStore) Close(_ context.Context) error {
 	log.Info("event store start to close")
 	defer log.Info("event store closed")
 
@@ -1340,7 +1335,7 @@ func (e *eventStore) uploadStatePeriodically(ctx context.Context) error {
 			// When the log coordinator resides on the same node, it will receive the same object reference.
 			// To prevent data races, we need to create a clone of the state.
 			message := messaging.NewSingleTargetMessage(coordinatorID, messaging.LogCoordinatorTopic, state.Copy())
-			// just ignore messagees fail to send
+			// just ignore messages fail to send
 			if err := e.messageCenter.SendEvent(message); err != nil {
 				log.Warn("send broadcast message to coordinator failed", zap.Error(err))
 			}

logservice/eventstore/event_store_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/ticdc/logservice/logpuller"
 	"github.com/pingcap/ticdc/pkg/common"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
+	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/pdutil"
 	"github.com/stretchr/testify/require"
 )
@@ -89,11 +90,12 @@ func (s *mockSubscriptionClient) Unsubscribe(subID logpuller.SubscriptionID) {
 }
 
 func newEventStoreForTest(path string) (logpuller.SubscriptionClient, EventStore) {
-	ctx := context.Background()
 	mockPDClock := pdutil.NewClock4Test()
 	appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
+	mc := messaging.NewMockMessageCenter()
+	appcontext.SetService(appcontext.MessageCenter, mc)
 	subClient := NewMockSubscriptionClient()
-	store := New(ctx, path, subClient)
+	store := New(path, subClient)
 	return subClient, store
 }
 
@@ -584,15 +586,12 @@ func TestEventStoreSwitchSubStat(t *testing.T) {
 }
 
 func TestWriteToEventStore(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
 	mockPDClock := pdutil.NewClock4Test()
 	appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
 
 	dir := t.TempDir()
-	store := New(ctx, dir, nil).(*eventStore)
-	defer store.Close(ctx)
+	store := New(dir, nil).(*eventStore)
+	defer store.Close(context.Background())
 
 	smallEntryKey := []byte("small-key")
 	smallEntryValue := []byte("small-value")

logservice/schemastore/ddl_job_fetcher.go

@@ -18,13 +18,12 @@ import (
 	"math"
 	"sync"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/logservice/logpuller"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/common/event"
-	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/utils/heap"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta"
@@ -49,7 +48,7 @@ type ddlJobFetcher struct {
 	}
 
 	// cacheDDLEvent and advanceResolvedTs may be called concurrently
-	// the only gurantee is that when call advanceResolvedTs with ts, all ddl job with commit ts <= ts has been passed to cacheDDLEvent
+	// the only guarantee is that when call advanceResolvedTs with ts, all ddl job with commit ts <= ts has been passed to cacheDDLEvent
 	cacheDDLEvent     func(ddlEvent DDLJobWithCommitTs)
 	advanceResolvedTs func(resolvedTS uint64)
 
@@ -58,7 +57,7 @@ type ddlJobFetcher struct {
 	keyspaceID uint32
 
 	ddlTableInfo         *event.DDLTableInfo
-	onceInitddlTableInfo sync.Once
+	onceInitDDLTableInfo sync.Once
 }
 
 func newDDLJobFetcher(
@@ -69,18 +68,18 @@ func newDDLJobFetcher(
 	cacheDDLEvent func(ddlEvent DDLJobWithCommitTs),
 	advanceResolvedTs func(resolvedTS uint64),
 ) *ddlJobFetcher {
-	ddlJobFetcher := &ddlJobFetcher{
+	fetcher := &ddlJobFetcher{
 		ctx:               ctx,
 		subClient:         subClient,
 		cacheDDLEvent:     cacheDDLEvent,
 		advanceResolvedTs: advanceResolvedTs,
 		kvStorage:         kvStorage,
 		keyspaceID:        keyspaceID,
 	}
-	ddlJobFetcher.resolvedTsTracker.resolvedTsItemMap = make(map[logpuller.SubscriptionID]*resolvedTsItem)
-	ddlJobFetcher.resolvedTsTracker.resolvedTsHeap = heap.NewHeap[*resolvedTsItem]()
+	fetcher.resolvedTsTracker.resolvedTsItemMap = make(map[logpuller.SubscriptionID]*resolvedTsItem)
+	fetcher.resolvedTsTracker.resolvedTsHeap = heap.NewHeap[*resolvedTsItem]()
 
-	return ddlJobFetcher
+	return fetcher
 }
 
 func (p *ddlJobFetcher) run(startTs uint64) error {
@@ -123,16 +122,16 @@ func (p *ddlJobFetcher) tryAdvanceResolvedTs(subID logpuller.SubscriptionID, new
 	if !ok || minResolvedTsItem.resolvedTs == math.MaxUint64 {
 		log.Panic("should not happen")
 	}
-	// minResolvedTsItem may be 0, it it ok to send it because it will be filtered later.
+	// minResolvedTsItem may be 0, it's ok to send it because it will be filtered later.
 	// it is ok to send redundant resolved ts to advanceResolvedTs.
 	p.advanceResolvedTs(minResolvedTsItem.resolvedTs)
 }
 
 func (p *ddlJobFetcher) input(kvs []common.RawKVEntry, _ func()) bool {
-	for _, kv := range kvs {
-		job, err := p.unmarshalDDL(&kv)
+	for _, entry := range kvs {
+		job, err := p.unmarshalDDL(&entry)
 		if err != nil {
-			log.Fatal("unmarshal ddl failed", zap.Any("kv", kv), zap.Error(err))
+			log.Fatal("unmarshal ddl failed", zap.Any("entry", entry), zap.Error(err))
 		}
 
 		if job == nil {
@@ -142,19 +141,19 @@ func (p *ddlJobFetcher) input(kvs []common.RawKVEntry, _ func()) bool {
 		// cache ddl job in memory until the resolve ts pass its commit ts
 		p.cacheDDLEvent(DDLJobWithCommitTs{
 			Job:      job,
-			CommitTs: kv.CRTs,
+			CommitTs: entry.CRTs,
 		})
 	}
 	return false
 }
 
-// unmarshalDDL unmarshals a ddl job from a raw kv entry.
+// unmarshalDDL unmarshal a ddl job from a raw kv entry.
 func (p *ddlJobFetcher) unmarshalDDL(rawKV *common.RawKVEntry) (*model.Job, error) {
 	if rawKV.OpType != common.OpTypePut {
 		return nil, nil
 	}
 	if !event.IsLegacyFormatJob(rawKV) {
-		p.onceInitddlTableInfo.Do(func() {
+		p.onceInitDDLTableInfo.Do(func() {
 			if err := p.initDDLTableInfo(p.ctx, p.kvStorage); err != nil {
 				log.Fatal("init ddl table info failed", zap.Error(err))
 			}
@@ -179,7 +178,7 @@ func (p *ddlJobFetcher) initDDLTableInfo(ctx context.Context, kvStorage kv.Stora
 
 	dbInfos, err := snap.ListDatabases()
 	if err != nil {
-		return cerror.WrapError(cerror.ErrMetaListDatabases, err)
+		return errors.WrapError(errors.ErrMetaListDatabases, err)
 	}
 
 	db, err := findDBByName(dbInfos, mysql.SystemDB)
@@ -231,8 +230,8 @@ func findDBByName(dbs []*model.DBInfo, name string) (*model.DBInfo, error) {
 			return db, nil
 		}
 	}
-	return nil, cerror.WrapError(
-		cerror.ErrDDLSchemaNotFound,
+	return nil, errors.WrapError(
+		errors.ErrDDLSchemaNotFound,
 		errors.Errorf("can't find schema %s", name))
 }
 
@@ -242,8 +241,8 @@ func findTableByName(tbls []*model.TableInfo, name string) (*model.TableInfo, er
 			return t, nil
 		}
 	}
-	return nil, cerror.WrapError(
-		cerror.ErrDDLSchemaNotFound,
+	return nil, errors.WrapError(
+		errors.ErrDDLSchemaNotFound,
 		errors.Errorf("can't find table %s", name))
 }
 
@@ -253,8 +252,8 @@ func findColumnByName(cols []*model.ColumnInfo, name string) (*model.ColumnInfo,
 			return c, nil
 		}
 	}
-	return nil, cerror.WrapError(
-		cerror.ErrDDLSchemaNotFound,
+	return nil, errors.WrapError(
+		errors.ErrDDLSchemaNotFound,
 		errors.Errorf("can't find column %s", name))
 }