monitoring: add event broker scanned bytes and count metrics and adjust some code (pingcap#1407)

close pingcap#1413

Author: 3AceShowHand

downstreamadapter/eventcollector/event_collector.go

@@ -125,35 +125,37 @@ type EventCollector struct {
 }
 
 func New(serverId node.ID) *EventCollector {
-	eventCollector := EventCollector{
+	receiveChannels := make([]chan *messaging.TargetMessage, config.DefaultBasicEventHandlerConcurrency)
+	for i := 0; i < config.DefaultBasicEventHandlerConcurrency; i++ {
+		receiveChannels[i] = make(chan *messaging.TargetMessage, receiveChanSize)
+	}
+	eventCollector := &EventCollector{
 		serverId:                             serverId,
 		dispatcherMap:                        sync.Map{},
 		dispatcherRequestChan:                chann.NewAutoDrainChann[DispatcherRequestWithTarget](),
 		logCoordinatorRequestChan:            chann.NewAutoDrainChann[*logservicepb.ReusableEventServiceRequest](),
 		mc:                                   appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter),
 		dispatcherHeartbeatChan:              chann.NewAutoDrainChann[*DispatcherHeartbeatWithTarget](),
-		receiveChannels:                      make([]chan *messaging.TargetMessage, config.DefaultBasicEventHandlerConcurrency),
+		receiveChannels:                      receiveChannels,
 		metricDispatcherReceivedKVEventCount: metrics.DispatcherReceivedEventCount.WithLabelValues("KVEvent"),
 		metricDispatcherReceivedResolvedTsEventCount: metrics.DispatcherReceivedEventCount.WithLabelValues("ResolvedTs"),
 		metricReceiveEventLagDuration:                metrics.EventCollectorReceivedEventLagDuration.WithLabelValues("Msg"),
 	}
-	eventCollector.ds = NewEventDynamicStream(&eventCollector)
+	eventCollector.ds = NewEventDynamicStream(eventCollector)
 	eventCollector.mc.RegisterHandler(messaging.EventCollectorTopic, eventCollector.RecvEventsMessage)
 
-	return &eventCollector
+	return eventCollector
 }
 
 func (c *EventCollector) Run(ctx context.Context) {
 	ctx, cancel := context.WithCancel(ctx)
 	c.cancel = cancel
 
 	for i := 0; i < config.DefaultBasicEventHandlerConcurrency; i++ {
-		ch := make(chan *messaging.TargetMessage, receiveChanSize)
-		c.receiveChannels[i] = ch
 		c.wg.Add(1)
 		go func() {
 			defer c.wg.Done()
-			c.runProcessMessage(ctx, ch)
+			c.runProcessMessage(ctx, c.receiveChannels[i])
 		}()
 	}
 
@@ -498,7 +500,7 @@ func (c *EventCollector) sendDispatcherHeartbeat(heartbeat *DispatcherHeartbeatW
 	return nil
 }
 
-func (c *EventCollector) handleDispatcherHeartbeatResponse(targetMessage *messaging.TargetMessage) error {
+func (c *EventCollector) handleDispatcherHeartbeatResponse(targetMessage *messaging.TargetMessage) {
 	if len(targetMessage.Message) != 1 {
 		log.Panic("invalid dispatcher heartbeat response message", zap.Any("msg", targetMessage))
 	}
@@ -519,7 +521,6 @@ func (c *EventCollector) handleDispatcherHeartbeatResponse(targetMessage *messag
 			}
 		}
 	}
-	return nil
 }
 
 // RecvEventsMessage is the handler for the events message from EventService.

logservice/eventstore/event_store.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/common"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/pingcap/ticdc/pkg/node"
@@ -76,7 +77,7 @@ type EventStore interface {
 
 	GetDispatcherDMLEventState(dispatcherID common.DispatcherID) (bool, DMLEventState)
 
-	// return an iterator which scan the data in ts range (dataRange.StartTs, dataRange.EndTs]
+	// GetIterator return an iterator which scan the data in ts range (dataRange.StartTs, dataRange.EndTs]
 	GetIterator(dispatcherID common.DispatcherID, dataRange common.DataRange) (EventIterator, error)
 }
 
@@ -593,11 +594,12 @@ func (e *eventStore) GetIterator(dispatcherID common.DispatcherID, dataRange com
 		return nil, nil
 	}
 	subscriptionStat := stat.subStat
-	if dataRange.StartTs < subscriptionStat.checkpointTs.Load() {
-		log.Panic("dataRange startTs is larger than subscriptionStat checkpointTs, it should not happen",
+	checkpoint := subscriptionStat.checkpointTs.Load()
+	if dataRange.StartTs < checkpoint {
+		log.Panic("dataRange startTs is smaller than subscriptionStat checkpointTs, it should not happen",
 			zap.Stringer("dispatcherID", dispatcherID),
-			zap.Uint64("checkpointTs", subscriptionStat.checkpointTs.Load()),
-			zap.Uint64("startTs", dataRange.StartTs))
+			zap.Uint64("startTs", dataRange.StartTs),
+			zap.Uint64("checkpointTs", checkpoint))
 	}
 	db := e.dbs[subscriptionStat.dbIndex]
 	e.dispatcherMeta.RUnlock()
@@ -611,10 +613,11 @@ func (e *eventStore) GetIterator(dispatcherID common.DispatcherID, dataRange com
 		UpperBound: end,
 	})
 	if err != nil {
-		return nil, err
+		return nil, errors.Trace(err)
 	}
 	startTime := time.Now()
-	iter.First()
+	// todo: what happens if iter.First() returns false?
+	_ = iter.First()
 	metricEventStoreFirstReadDurationHistogram.Observe(time.Since(startTime).Seconds())
 	metrics.EventStoreScanRequestsCount.Inc()
 
@@ -759,7 +762,6 @@ func (e *eventStore) writeEvents(db *pebble.DB, events []eventWithCallback) erro
 	CounterKv.Add(float64(kvCount))
 	metrics.EventStoreWriteBatchEventsCountHist.Observe(float64(kvCount))
 	metrics.EventStoreWriteBatchSizeHist.Observe(float64(batch.Len()))
-	metrics.EventStoreWriteBytes.Add(float64(batch.Len()))
 	start := time.Now()
 	err := batch.Commit(pebble.NoSync)
 	metrics.EventStoreWriteDurationHistogram.Observe(float64(time.Since(start).Milliseconds()) / 1000)
@@ -790,10 +792,6 @@ type eventStoreIter struct {
 }
 
 func (iter *eventStoreIter) Next() (*common.RawKVEntry, bool, error) {
-	if iter.innerIter == nil {
-		log.Panic("iter is nil")
-	}
-
 	rawKV := &common.RawKVEntry{}
 	for {
 		if !iter.innerIter.Valid() {
@@ -805,20 +803,19 @@ func (iter *eventStoreIter) Next() (*common.RawKVEntry, bool, error) {
 			log.Panic("fail to decode raw kv entry", zap.Error(err))
 		}
 		metrics.EventStoreScanBytes.Add(float64(len(value)))
-		if iter.needCheckSpan {
-			comparableKey := common.ToComparableKey(rawKV.Key)
-			if bytes.Compare(comparableKey, iter.tableSpan.StartKey) >= 0 &&
-				bytes.Compare(comparableKey, iter.tableSpan.EndKey) <= 0 {
-				break
-			}
-			log.Debug("event store iter skip kv not in table span",
-				zap.String("tableSpan", common.FormatTableSpan(iter.tableSpan)),
-				zap.String("key", hex.EncodeToString(rawKV.Key)),
-				zap.Uint64("startTs", rawKV.StartTs),
-				zap.Uint64("commitTs", rawKV.CRTs))
-		} else {
+		if !iter.needCheckSpan {
+			break
+		}
+		comparableKey := common.ToComparableKey(rawKV.Key)
+		if bytes.Compare(comparableKey, iter.tableSpan.StartKey) >= 0 &&
+			bytes.Compare(comparableKey, iter.tableSpan.EndKey) <= 0 {
 			break
 		}
+		log.Debug("event store iter skip kv not in table span",
+			zap.String("tableSpan", common.FormatTableSpan(iter.tableSpan)),
+			zap.String("key", hex.EncodeToString(rawKV.Key)),
+			zap.Uint64("startTs", rawKV.StartTs),
+			zap.Uint64("commitTs", rawKV.CRTs))
 		iter.innerIter.Next()
 	}
 	isNewTxn := false
@@ -830,7 +827,7 @@ func (iter *eventStoreIter) Next() (*common.RawKVEntry, bool, error) {
 	iter.rowCount++
 	startTime := time.Now()
 	iter.innerIter.Next()
-	metricEventStoreNextReadDurationHistogram.Observe(float64(time.Since(startTime).Seconds()))
+	metricEventStoreNextReadDurationHistogram.Observe(time.Since(startTime).Seconds())
 	return rawKV, isNewTxn, nil
 }
 

logservice/schemastore/schema_store.go

@@ -42,7 +42,7 @@ type SchemaStore interface {
 
 	UnregisterTable(tableID int64) error
 
-	// return table info with largest version <= ts
+	// GetTableInfo return table info with the largest version <= ts
 	GetTableInfo(tableID int64, ts uint64) (*common.TableInfo, error)
 
 	// TODO: how to respect tableFilter