syncer(dm): log sampled unhandled-event counts

Author: GMHDBJD

dm/syncer/syncer.go

@@ -261,6 +261,10 @@ type Syncer struct {
 
 	ddlWorker            *DDLWorker
 	unhandledEventLogger *zap.Logger
+	unhandledEvents      struct {
+		sync.Mutex
+		counts map[string]map[string]int
+	}
 }
 
 // NewSyncer creates a new Syncer.
@@ -317,6 +321,7 @@ func NewSyncer(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, relay rel
 		unhandledEventSampleFirst,
 		logFields...,
 	)()
+	syncer.unhandledEvents.counts = make(map[string]map[string]int)
 
 	return syncer
 }
@@ -354,7 +359,23 @@ func (s *Syncer) closeJobChans() {
 }
 
 func (s *Syncer) recordUnhandledEvent(message string, ev interface{}) {
-	s.unhandledEventLogger.Warn(message, zap.String("type", fmt.Sprintf("%T", ev)))
+	eventType := fmt.Sprintf("%T", ev)
+
+	s.unhandledEvents.Lock()
+	eventCounts, ok := s.unhandledEvents.counts[message]
+	if !ok {
+		eventCounts = make(map[string]int)
+		s.unhandledEvents.counts[message] = eventCounts
+	}
+	eventCounts[eventType]++
+
+	snapshot := make(map[string]int, len(eventCounts))
+	for event, count := range eventCounts {
+		snapshot[event] = count
+	}
+	s.unhandledEvents.Unlock()
+
+	s.unhandledEventLogger.Warn(message, zap.Any("events", snapshot))
 }
 
 // Type implements Unit.Type.

dm/syncer/syncer_test.go

@@ -243,14 +243,28 @@ func (s *testSyncerSuite) TestSampleUnhandledEvents(c *check.C) {
 	}
 
 	c.Assert(seen["unhandled event"], check.HasLen, 1)
-	c.Assert(seen["unhandled event"][0]["type"], check.Equals, "*replication.RowsQueryEvent")
+	c.Assert(seen["unhandled event"][0]["events"], check.DeepEquals, map[string]int{
+		"*replication.RowsQueryEvent": 1,
+	})
 	c.Assert(seen["unhandled event from transaction payload"], check.HasLen, 1)
-	c.Assert(seen["unhandled event from transaction payload"][0]["type"], check.Equals, "*replication.QueryEvent")
+	c.Assert(seen["unhandled event from transaction payload"][0]["events"], check.DeepEquals, map[string]int{
+		"*replication.QueryEvent": 1,
+	})
 
 	syncer.recordUnhandledEvent("unhandled event", &replication.RowsQueryEvent{})
 	syncer.recordUnhandledEvent("unhandled event", &replication.QueryEvent{})
 	syncer.recordUnhandledEvent("unhandled event from transaction payload", &replication.QueryEvent{})
 	c.Assert(logs.All(), check.HasLen, 2)
+
+	syncer.unhandledEvents.Lock()
+	defer syncer.unhandledEvents.Unlock()
+	c.Assert(syncer.unhandledEvents.counts["unhandled event"], check.DeepEquals, map[string]int{
+		"*replication.QueryEvent":     2,
+		"*replication.RowsQueryEvent": 3,
+	})
+	c.Assert(syncer.unhandledEvents.counts["unhandled event from transaction payload"], check.DeepEquals, map[string]int{
+		"*replication.QueryEvent": 2,
+	})
 }
 
 func mockGetServerUnixTS(mock sqlmock.Sqlmock) {