eventService: reduce resolvedTs lag

.gitignore

@@ -5,3 +5,4 @@ tools/include
 
 .vscode
 .idea
+*.log

downstreamadapter/dispatchermanager/event_dispatcher_manager.go

@@ -75,7 +75,9 @@ type EventDispatcherManager struct {
 	filter                         filter.Filter
 	metricCreateDispatcherDuration prometheus.Observer
 	metricCheckpointTs             prometheus.Gauge
+	metricCheckpointTsLag          prometheus.Gauge
 	metricResolveTs                prometheus.Gauge
+	metricResolvedTsLag            prometheus.Gauge
 
 	closing bool
 	closed  atomic.Bool
@@ -95,7 +97,9 @@ func NewEventDispatcherManager(changefeedID model.ChangeFeedID, changefeedConfig
 		tableEventDispatcherCount:      metrics.TableEventDispatcherGauge.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricCreateDispatcherDuration: metrics.CreateDispatcherDuration.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricCheckpointTs:             metrics.EventDispatcherManagerCheckpointTsGauge.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
+		metricCheckpointTsLag:          metrics.EventDispatcherManagerCheckpointTsLagGauge.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricResolveTs:                metrics.EventDispatcherManagerResolvedTsGauge.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
+		metricResolvedTsLag:            metrics.EventDispatcherManagerResolvedTsLagGauge.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
 
 	// TODO: 最后去更新一下 filter 的内部 NewFilter 函数，现在是在套壳适配
@@ -383,7 +387,6 @@ func (e *EventDispatcherManager) CollectHeartbeatInfo(needCompleteStatus bool) *
 		// If it's closed successfully, we could clean it up.
 		// TODO: we need to consider how to deal with the checkpointTs of the removed dispatcher if the message will be discarded.
 		dispatcherItem.CollectDispatcherHeartBeatInfo(dispatcherHeartBeatInfo)
-
 		if dispatcherHeartBeatInfo.IsRemoving {
 			watermark, ok := dispatcherItem.TryClose()
 			if ok {
@@ -405,16 +408,22 @@ func (e *EventDispatcherManager) CollectHeartbeatInfo(needCompleteStatus bool) *
 			message.Statuses = append(message.Statuses, &heartbeatpb.TableSpanStatus{
 				Span:            dispatcherHeartBeatInfo.TableSpan.TableSpan,
 				ComponentStatus: dispatcherHeartBeatInfo.ComponentStatus,
+				CheckpointTs:    dispatcherHeartBeatInfo.Watermark.CheckpointTs,
 			})
 		}
 	}
 
 	for _, tableSpan := range toReomveTableSpans {
 		e.cleanTableEventDispatcher(tableSpan)
 	}
-
-	e.metricCheckpointTs.Set(float64(oracle.ExtractPhysical(message.Watermark.CheckpointTs)))
-	e.metricResolveTs.Set(float64(oracle.ExtractPhysical(message.Watermark.ResolvedTs)))
+	ckptTs := oracle.ExtractPhysical(message.Watermark.CheckpointTs)
+	e.metricCheckpointTs.Set(float64(ckptTs))
+	lag := (oracle.GetPhysical(time.Now()) - ckptTs) / 1e3
+	e.metricCheckpointTsLag.Set(float64(lag))
+	resolvedTs := oracle.ExtractPhysical(message.Watermark.ResolvedTs)
+	e.metricResolveTs.Set(float64(resolvedTs))
+	lag = (oracle.GetPhysical(time.Now()) - resolvedTs) / 1e3
+	e.metricResolvedTsLag.Set(float64(lag))
 	return &message
 }
 

downstreamadapter/dispatchermanager/heartbeat_collector.go

@@ -142,7 +142,7 @@ func (c *HeartBeatCollector) RegisterEventDispatcherManager(m *EventDispatcherMa
 func (c *HeartBeatCollector) SendHeartBeatMessages() {
 	for {
 		heartBeatRequestWithTargetID := c.requestQueue.Dequeue()
-		err := appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter).SendEvent(&messaging.TargetMessage{
+		err := appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter).SendCommand(&messaging.TargetMessage{
 			To:      heartBeatRequestWithTargetID.TargetID,
 			Topic:   messaging.MaintainerManagerTopic,
 			Type:    messaging.TypeHeartBeatRequest,

downstreamadapter/eventcollector/event_collector.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/chann"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 )
 
@@ -73,6 +74,9 @@ type EventCollector struct {
 	registerMessageChan                          *chann.DrainableChann[RegisterInfo] // for temp
 	metricDispatcherReceivedKVEventCount         prometheus.Counter
 	metricDispatcherReceivedResolvedTsEventCount prometheus.Counter
+	metricReceiveEventLagDuration                prometheus.Observer
+	metricReceiveResolvedTsEventLagDuration      prometheus.Observer
+	metricResolvedTsLag                          prometheus.Gauge
 }
 
 func NewEventCollector(globalMemoryQuota int64, serverId messaging.ServerId) *EventCollector {
@@ -83,6 +87,9 @@ func NewEventCollector(globalMemoryQuota int64, serverId messaging.ServerId) *Ev
 		registerMessageChan:                  chann.NewAutoDrainChann[RegisterInfo](),
 		metricDispatcherReceivedKVEventCount: metrics.DispatcherReceivedEventCount.WithLabelValues("KVEvent"),
 		metricDispatcherReceivedResolvedTsEventCount: metrics.DispatcherReceivedEventCount.WithLabelValues("ResolvedTs"),
+		metricReceiveEventLagDuration:                metrics.EventCollectorReceivedEventLagDuration.WithLabelValues("KVEvent"),
+		metricReceiveResolvedTsEventLagDuration:      metrics.EventCollectorReceivedEventLagDuration.WithLabelValues("ResolvedTs"),
+		metricResolvedTsLag:                          metrics.EventCollectorResolvedTsLagGauge,
 	}
 	appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter).RegisterHandler(messaging.EventCollectorTopic, eventCollector.RecvEventsMessage)
 
@@ -104,6 +111,8 @@ func NewEventCollector(globalMemoryQuota int64, serverId messaging.ServerId) *Ev
 			}
 		}
 	}()
+	// update metrics
+	eventCollector.updateMetrics(context.Background())
 
 	return &eventCollector
 }
@@ -174,6 +183,7 @@ func (c *EventCollector) RecvEventsMessage(ctx context.Context, msg *messaging.T
 		log.Error("invalid event feed message", zap.Any("msg", msg))
 		return apperror.AppError{Type: apperror.ErrorTypeInvalidMessage, Reason: "invalid heartbeat response message"}
 	}
+	inflightDuration := time.Since(time.Unix(0, msg.CrateAt)).Milliseconds()
 
 	dispatcherID := txnEvent.DispatcherID
 
@@ -207,9 +217,11 @@ func (c *EventCollector) RecvEventsMessage(ctx context.Context, msg *messaging.T
 		if txnEvent.IsDMLEvent() || txnEvent.IsDDLEvent() {
 			dispatcherItem.PushTxnEvent(txnEvent)
 			c.metricDispatcherReceivedKVEventCount.Inc()
+			c.metricReceiveEventLagDuration.Observe(float64(inflightDuration))
 		} else {
 			dispatcherItem.UpdateResolvedTs(txnEvent.ResolvedTs)
 			c.metricDispatcherReceivedResolvedTsEventCount.Inc()
+			c.metricReceiveResolvedTsEventLagDuration.Observe(float64(inflightDuration))
 		}
 
 		// dispatcherItem.UpdateResolvedTs(eventFeeds.ResolvedTs)
@@ -232,3 +244,34 @@ func (c *EventCollector) RecvEventsMessage(ctx context.Context, msg *messaging.T
 	}
 	return nil
 }
+
+func (c *EventCollector) updateMetrics(ctx context.Context) error {
+	ticker := time.NewTicker(10 * time.Second)
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				minResolvedTs := uint64(0)
+				c.dispatcherMap.m.Range(func(key, value interface{}) bool {
+					d, ok := value.(*dispatcher.Dispatcher)
+					if !ok {
+						return true
+					}
+					if minResolvedTs == 0 || d.GetResolvedTs() < minResolvedTs {
+						minResolvedTs = d.GetResolvedTs()
+					}
+					return true
+				})
+				if minResolvedTs == 0 {
+					continue
+				}
+				phyResolvedTs := oracle.ExtractPhysical(minResolvedTs)
+				lag := (oracle.GetPhysical(time.Now()) - phyResolvedTs) / 1e3
+				c.metricResolvedTsLag.Set(float64(lag))
+			}
+		}
+	}()
+	return nil
+}

go.mod

@@ -13,7 +13,6 @@ require (
 	github.com/go-sql-driver/mysql v1.7.1
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/btree v1.1.2
-	github.com/google/martian v2.1.0+incompatible
 	github.com/google/uuid v1.6.0
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/pingcap/errors v0.11.5-0.20240318064555-6bd07397691f