syncer(dm): debounce repeated unhandled-event warnings (pingcap#12579)

close pingcap#12499

Author: GMHDBJD

dm/syncer/syncer.go

@@ -38,6 +38,7 @@ import (
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/util/dbutil"
 	"github.com/pingcap/tidb/pkg/util/filter"
+	tidblogutil "github.com/pingcap/tidb/pkg/util/logutil"
 	regexprrouter "github.com/pingcap/tidb/pkg/util/regexpr-router"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config"
@@ -97,6 +98,9 @@ const (
 	skipJobIdx = iota
 	ddlJobIdx
 	workerJobTSArrayInitSize // size = skip + ddl
+
+	unhandledEventSampleInterval = 5 * time.Minute
+	unhandledEventSampleFirst    = 1
 )
 
 // waitXIDStatus represents the status for waiting XID event when pause/stop task.
@@ -255,7 +259,8 @@ type Syncer struct {
 	charsetAndDefaultCollation map[string]string
 	idAndCollationMap          map[int]string
 
-	ddlWorker *DDLWorker
+	ddlWorker            *DDLWorker
+	unhandledEventLogger *zap.Logger
 }
 
 // NewSyncer creates a new Syncer.
@@ -307,6 +312,11 @@ func NewSyncer(cfg *config.SubTaskConfig, etcdClient *clientv3.Client, relay rel
 	syncer.lastCheckpointFlushedTime = time.Time{}
 	syncer.relay = relay
 	syncer.safeMode = sm.NewSafeMode()
+	syncer.unhandledEventLogger = tidblogutil.SampleLoggerFactory(
+		unhandledEventSampleInterval,
+		unhandledEventSampleFirst,
+		logFields...,
+	)()
 
 	return syncer
 }
@@ -343,6 +353,10 @@ func (s *Syncer) closeJobChans() {
 	s.jobsClosed.Store(true)
 }
 
+func (s *Syncer) recordUnhandledEvent(message string, ev interface{}) {
+	s.unhandledEventLogger.Warn(message, zap.String("type", fmt.Sprintf("%T", ev)))
+}
+
 // Type implements Unit.Type.
 func (s *Syncer) Type() pb.UnitType {
 	return pb.UnitType_Sync
@@ -2443,7 +2457,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 				case *replication.TableMapEvent:
 				case *replication.FormatDescriptionEvent:
 				default:
-					s.tctx.L().Warn("unhandled event from transaction payload", zap.String("type", fmt.Sprintf("%T", tpevt)))
+					s.recordUnhandledEvent("unhandled event from transaction payload", tpevt)
 				}
 			}
 			if needContinue {
@@ -2452,7 +2466,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		case *replication.TableMapEvent:
 		case *replication.FormatDescriptionEvent:
 		default:
-			s.tctx.L().Warn("unhandled event", zap.String("type", fmt.Sprintf("%T", ev)))
+			s.recordUnhandledEvent("unhandled event", ev)
 		}
 		if err2 != nil {
 			if err := s.handleEventError(err2, startLocation, endLocation, e.Header.EventType == replication.QUERY_EVENT, originSQL); err != nil {

dm/syncer/syncer_test.go

@@ -31,6 +31,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
+	pclog "github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -62,6 +63,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 )
 
 var _ = check.Suite(&testSyncerSuite{})
@@ -219,6 +221,38 @@ func (s *testSyncerSuite) TearDownSuite(c *check.C) {
 	os.RemoveAll(s.cfg.Dir)
 }
 
+func (s *testSyncerSuite) TestSampleUnhandledEvents(c *check.C) {
+	core, logs := observer.New(zap.WarnLevel)
+	restoreGlobal := pclog.ReplaceGlobals(zap.New(core), nil)
+	defer restoreGlobal()
+
+	cfg := genDefaultSubTaskConfig4Test()
+	syncer := NewSyncer(cfg, nil, nil)
+
+	syncer.recordUnhandledEvent("unhandled event", &replication.RowsQueryEvent{})
+	syncer.recordUnhandledEvent("unhandled event", &replication.RowsQueryEvent{})
+	syncer.recordUnhandledEvent("unhandled event", &replication.QueryEvent{})
+	syncer.recordUnhandledEvent("unhandled event from transaction payload", &replication.QueryEvent{})
+
+	entries := logs.All()
+	c.Assert(entries, check.HasLen, 2)
+
+	seen := make(map[string][]map[string]interface{}, len(entries))
+	for _, entry := range entries {
+		seen[entry.Message] = append(seen[entry.Message], entry.ContextMap())
+	}
+
+	c.Assert(seen["unhandled event"], check.HasLen, 1)
+	c.Assert(seen["unhandled event"][0]["type"], check.Equals, "*replication.RowsQueryEvent")
+	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")
+
+	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)
+}
+
 func mockGetServerUnixTS(mock sqlmock.Sqlmock) {
 	ts := time.Now().Unix()
 	rows := sqlmock.NewRows([]string{"UNIX_TIMESTAMP()"}).AddRow(strconv.FormatInt(ts, 10))