eventcollector, eventservice: add congestion control mechanism (pingcap#1588)

close pingcap#1404

Author: 3AceShowHand

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -317,7 +317,6 @@ func (d *BasicDispatcher) handleEvents(dispatcherEvents []DispatcherEvent, wakeC
 	for _, dispatcherEvent := range dispatcherEvents {
 		log.Debug("dispatcher receive all event",
 			zap.Stringer("dispatcher", d.id), zap.Bool("isRedo", isRedo),
-			zap.String("eventType", commonEvent.TypeToString(dispatcherEvent.Event.GetType())),
 			zap.Any("event", dispatcherEvent.Event))
 		failpoint.Inject("HandleEventsSlowly", func() {
 			lag := time.Duration(rand.Intn(5000)) * time.Millisecond

downstreamadapter/eventcollector/dispatcher_stat.go

@@ -616,13 +616,12 @@ func (d *dispatcherStat) handleHandshakeEvent(event dispatcher.DispatcherEvent)
 }
 
 func (d *dispatcherStat) setRemoteCandidates(nodes []string) {
-	log.Info("set remote candidates",
-		zap.Strings("nodes", nodes),
-		zap.Stringer("dispatcherID", d.getDispatcherID()))
 	if len(nodes) == 0 {
 		return
 	}
 	if d.connState.trySetRemoteCandidates(nodes) {
+		log.Info("set remote candidates", zap.Stringer("dispatcherID", d.getDispatcherID()),
+			zap.Int64("tableID", d.target.GetTableSpan().TableID), zap.Strings("nodes", nodes))
 		candidate := d.connState.getNextRemoteCandidate()
 		d.registerTo(candidate)
 	}

downstreamadapter/eventcollector/event_collector.go

@@ -169,6 +169,10 @@ func (c *EventCollector) Run(ctx context.Context) {
 		return c.processDSFeedback(ctx)
 	})
 
+	g.Go(func() error {
+		return c.controlCongestion(ctx)
+	})
+
 	g.Go(func() error {
 		return c.sendDispatcherRequests(ctx)
 	})
@@ -219,7 +223,9 @@ func (c *EventCollector) PrepareAddDispatcher(
 ) {
 	log.Info("add dispatcher", zap.Stringer("dispatcher", target.GetId()))
 	defer func() {
-		log.Info("add dispatcher done", zap.Stringer("dispatcher", target.GetId()), zap.Int("type", target.GetType()))
+		log.Info("add dispatcher done",
+			zap.Stringer("dispatcherID", target.GetId()), zap.Int64("tableID", target.GetTableSpan().GetTableID()),
+			zap.Uint64("startTs", target.GetStartTs()), zap.Int("type", target.GetType()))
 	}()
 	metrics.EventCollectorRegisteredDispatcherCount.Inc()
 
@@ -238,11 +244,12 @@ func (c *EventCollector) PrepareAddDispatcher(
 
 // CommitAddDispatcher notify local event service that the dispatcher is ready to receive events.
 func (c *EventCollector) CommitAddDispatcher(target dispatcher.Dispatcher, startTs uint64) {
-	log.Info("commit add dispatcher", zap.Stringer("dispatcher", target.GetId()), zap.Uint64("startTs", startTs))
+	log.Info("commit add dispatcher", zap.Stringer("dispatcherID", target.GetId()),
+		zap.Int64("tableID", target.GetTableSpan().GetTableID()), zap.Uint64("startTs", startTs))
 	value, ok := c.dispatcherMap.Load(target.GetId())
 	if !ok {
 		log.Warn("dispatcher not found when commit add dispatcher",
-			zap.Stringer("dispatcher", target.GetId()),
+			zap.Stringer("dispatcherID", target.GetId()), zap.Int64("tableID", target.GetTableSpan().GetTableID()),
 			zap.Uint64("startTs", startTs))
 		return
 	}
@@ -251,9 +258,10 @@ func (c *EventCollector) CommitAddDispatcher(target dispatcher.Dispatcher, start
 }
 
 func (c *EventCollector) RemoveDispatcher(target dispatcher.Dispatcher) {
-	log.Info("remove dispatcher", zap.Stringer("dispatcher", target.GetId()))
+	log.Info("remove dispatcher", zap.Stringer("dispatcherID", target.GetId()))
 	defer func() {
-		log.Info("remove dispatcher done", zap.Stringer("dispatcher", target.GetId()))
+		log.Info("remove dispatcher done", zap.Stringer("dispatcherID", target.GetId()),
+			zap.Int64("tableID", target.GetTableSpan().GetTableID()))
 	}()
 	isRedo := dispatcher.IsRedoDispatcher(target)
 	value, ok := c.dispatcherMap.Load(target.GetId())
@@ -512,6 +520,90 @@ func (c *EventCollector) runDispatchMessage(ctx context.Context, inCh <-chan *me
 	}
 }
 
+func (c *EventCollector) controlCongestion(ctx context.Context) error {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return context.Cause(ctx)
+		case <-ticker.C:
+			messages := c.newCongestionControlMessages()
+			for serverID, m := range messages {
+				if len(m.GetAvailables()) != 0 {
+					msg := messaging.NewSingleTargetMessage(serverID, messaging.EventServiceTopic, m)
+					if err := c.mc.SendCommand(msg); err != nil {
+						log.Warn("send congestion control message failed", zap.Error(err))
+					}
+				}
+			}
+		}
+	}
+}
+
+func (c *EventCollector) newCongestionControlMessages() map[node.ID]*event.CongestionControl {
+	// collect all changefeeds' available memory quota
+	availables := make(map[common.ChangeFeedID]uint64)
+	for _, quota := range c.ds.GetMetrics().MemoryControl.AreaMemoryMetrics {
+		changefeedID, ok := c.changefeedIDMap.Load(quota.Area())
+		if !ok {
+			continue
+		}
+		availables[changefeedID.(common.ChangeFeedID)] = uint64(quota.AvailableMemory())
+	}
+	if len(availables) == 0 {
+		return nil
+	}
+
+	// calculate each changefeed's available memory quota for each node
+	// by the proportion of the dispatcher on each node.
+	// this is not accurate, we should also consider each node's workload distribution.
+	proportions := make(map[common.ChangeFeedID]map[node.ID]uint64)
+	c.dispatcherMap.Range(func(k, v interface{}) bool {
+		stat := v.(*dispatcherStat)
+		eventServiceID := stat.connState.getEventServiceID()
+		if eventServiceID == "" {
+			return true
+		}
+
+		changefeedID := stat.target.GetChangefeedID()
+		holder, ok := proportions[changefeedID]
+		if !ok {
+			holder = make(map[node.ID]uint64)
+			proportions[changefeedID] = holder
+		}
+		holder[eventServiceID]++
+		return true
+	})
+
+	// group the available memory quota by nodeID
+	result := make(map[node.ID]*event.CongestionControl)
+	for changefeedID, total := range availables {
+		proportion := proportions[changefeedID]
+		var sum uint64
+		for _, portion := range proportion {
+			sum += portion
+		}
+		if sum == 0 {
+			continue
+		}
+
+		for nodeID, portion := range proportion {
+			ratio := float64(portion) / float64(sum)
+			quota := uint64(float64(total) * ratio)
+
+			m, ok := result[nodeID]
+			if !ok {
+				m = event.NewCongestionControl()
+				result[nodeID] = m
+			}
+			m.AddAvailableMemory(changefeedID.ID(), quota)
+		}
+	}
+	return result
+}
+
 func (c *EventCollector) updateMetrics(ctx context.Context) error {
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()

downstreamadapter/sink/kafka/sink.go

@@ -362,12 +362,17 @@ func (s *sink) sendMessages(ctx context.Context) error {
 	metricSendMessageDuration := metrics.WorkerSendMessageDuration.WithLabelValues(s.changefeedID.Namespace(), s.changefeedID.Name())
 	defer metrics.WorkerSendMessageDuration.DeleteLabelValues(s.changefeedID.Namespace(), s.changefeedID.Name())
 
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
 	var err error
 	outCh := s.comp.encoderGroup.Output()
 	for {
 		select {
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())
+		case <-ticker.C:
+			s.dmlProducer.Heartbeat()
 		case future, ok := <-outCh:
 			if !ok {
 				log.Info("kafka sink encoder's output channel closed",
@@ -457,6 +462,9 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 		metrics.CheckpointTsMessageCount.DeleteLabelValues(s.changefeedID.Namespace(), s.changefeedID.Name())
 	}()
 
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
 	var (
 		msg          *common.Message
 		partitionNum int32
@@ -466,6 +474,8 @@ func (s *sink) sendCheckpoint(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return errors.Trace(ctx.Err())
+		case <-ticker.C:
+			s.ddlProducer.Heartbeat()
 		case ts, ok := <-s.checkpointChan:
 			if !ok {
 				log.Warn("kafka sink checkpoint channel closed",

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -115,6 +115,8 @@ func (m *kafkaTopicManager) GetPartitionNum(
 }
 
 func (m *kafkaTopicManager) backgroundRefreshMeta(ctx context.Context) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
@@ -123,6 +125,8 @@ func (m *kafkaTopicManager) backgroundRefreshMeta(ctx context.Context) {
 				zap.String("changefeed", m.changefeedID.Name()),
 			)
 			return
+		case <-ticker.C:
+			m.admin.Heartbeat()
 		case <-m.metaRefreshTicker.C:
 			// We ignore the error here, because the error may be caused by the
 			// network problem, and we can try to get the metadata next time.