Add cdc metrics and standardized some codes (#27)

Signed-off-by: SimFG <bang.fu@zilliz.com>

core/api/replicate_manager.go

@@ -50,6 +50,21 @@ const (
 	ReplicateError = 100
 )
 
+func (r ReplicateAPIEventType) String() string {
+	switch r {
+	case ReplicateCreateCollection:
+		return "CreateCollection"
+	case ReplicateDropCollection:
+		return "DropCollection"
+	case ReplicateCreatePartition:
+		return "CreatePartition"
+	case ReplicateDropPartition:
+		return "DropPartition"
+	default:
+		return "Unknown"
+	}
+}
+
 type DefaultChannelManager struct{}
 
 var _ ChannelManager = (*DefaultChannelManager)(nil)

core/reader/replicate_channel_manager.go

@@ -47,9 +47,17 @@ type replicateChannelManager struct {
 
 	channelChan  chan string
 	apiEventChan chan *api.ReplicateAPIEvent
+
+	msgPackCallback func(string, *msgstream.MsgPack)
 }
 
-func NewReplicateChannelManager(mqConfig config.MQConfig, factoryCreator FactoryCreator, client api.TargetAPI, messageBufferSize int, metaOp api.MetaOp) (api.ChannelManager, error) {
+func NewReplicateChannelManager(mqConfig config.MQConfig,
+	factoryCreator FactoryCreator,
+	client api.TargetAPI,
+	messageBufferSize int,
+	metaOp api.MetaOp,
+	msgPackCallback func(string, *msgstream.MsgPack),
+) (api.ChannelManager, error) {
 	var factory msgstream.Factory
 	switch {
 	case mqConfig.Pulsar.Address != "":
@@ -72,6 +80,7 @@ func NewReplicateChannelManager(mqConfig config.MQConfig, factoryCreator Factory
 		replicatePartitions:  make(map[int64]map[int64]chan struct{}),
 		channelChan:          make(chan string, 10),
 		apiEventChan:         make(chan *api.ReplicateAPIEvent, 10),
+		msgPackCallback:      msgPackCallback,
 	}, nil
 }
 
@@ -353,6 +362,7 @@ func (r *replicateChannelManager) startReadChannel(sourceInfo *model.SourceColle
 				zap.String("channel_name", sourceInfo.PChannelName), zap.Int64("collection_id", sourceInfo.CollectionID), zap.Error(err))
 			return err
 		}
+		channelHandler.msgPackCallback = r.msgPackCallback
 		r.channelHandlerMap[sourceInfo.PChannelName] = channelHandler
 		r.channelChan <- sourceInfo.PChannelName
 		return nil
@@ -387,6 +397,7 @@ type replicateChannelHandler struct {
 	collectionNames   map[string]int64
 	msgPackChan       chan *msgstream.MsgPack
 	apiEventChan      chan *api.ReplicateAPIEvent
+	msgPackCallback   func(string, *msgstream.MsgPack)
 }
 
 func (r *replicateChannelHandler) AddCollection(collectionID int64, targetInfo *model.TargetCollectionInfo) {
@@ -423,6 +434,7 @@ func (r *replicateChannelHandler) AddPartitionInfo(collectionInfo *pb.Collection
 		return nil
 	}
 	targetInfo.PartitionBarrierChan[partitionID] = barrierChan
+	// TODO use goroutine pool
 	go func() {
 		_ = retry.Do(r.replicateCtx, func() error {
 			id := r.updateTargetPartitionInfo(collectionID, collectionName, partitionName)
@@ -656,6 +668,9 @@ func (r *replicateChannelHandler) handlePack(pack *msgstream.MsgPack) *msgstream
 	for _, position := range newPack.EndPositions {
 		position.ChannelName = pChannel
 	}
+	if r.msgPackCallback != nil {
+		r.msgPackCallback(r.pChannelName, newPack)
+	}
 	needTsMsg = needTsMsg || len(newPack.Msgs) == 0
 	if needTsMsg {
 		timeTickResult := msgpb.TimeTickMsg{

core/reader/replicate_channel_manager_test.go

@@ -23,7 +23,8 @@ import (
 
 func TestNewReplicateChannelManager(t *testing.T) {
 	t.Run("empty config", func(t *testing.T) {
-		_, err := NewReplicateChannelManager(config.MQConfig{}, NewDefaultFactoryCreator(), nil, 10, &api.DefaultMetaOp{})
+		_, err := NewReplicateChannelManager(config.MQConfig{}, NewDefaultFactoryCreator(), nil, 10, &api.DefaultMetaOp{}, func(s string, pack *msgstream.MsgPack) {
+		})
 		assert.Error(t, err)
 	})
 
@@ -35,7 +36,8 @@ func TestNewReplicateChannelManager(t *testing.T) {
 			Pulsar: config.PulsarConfig{
 				Address: "pulsar://localhost:6650",
 			},
-		}, factoryCreator, nil, 10, &api.DefaultMetaOp{})
+		}, factoryCreator, nil, 10, &api.DefaultMetaOp{}, func(s string, pack *msgstream.MsgPack) {
+		})
 		assert.NoError(t, err)
 	})
 }
@@ -95,7 +97,8 @@ func TestStartReadCollection(t *testing.T) {
 		Pulsar: config.PulsarConfig{
 			Address: "pulsar://localhost:6650",
 		},
-	}, factoryCreator, targetClient, 10, &api.DefaultMetaOp{})
+	}, factoryCreator, targetClient, 10, &api.DefaultMetaOp{}, func(s string, pack *msgstream.MsgPack) {
+	})
 	assert.NoError(t, err)
 	manager.SetCtx(context.Background())