coordinator: improve the log format and code struct (#978)

close #963
pingcap · Feb 11, 2025 · b7a4b1c · b7a4b1c
1 parent 8b3929e
commit b7a4b1c
Show file tree

Hide file tree

Showing 18 changed files with 486 additions and 249 deletions.
diff --git a/coordinator/changefeed/backoff.go b/coordinator/changefeed/backoff.go
@@ -188,15 +188,16 @@ func (m *Backoff) HandleError(errs []*heartbeatpb.RunningError) (bool, *heartbea
 		log.Error("The changefeed won't be restarted as it has been experiencing failures for "+
 			"an extended duration",
 			zap.Duration("maxElapsedTime", m.errBackoff.MaxElapsedTime),
-			zap.String("namespace", m.id.Namespace()),
-			zap.String("changefeed", m.id.Name()),
+			zap.Stringer("changefeed", m.id),
+			zap.Uint64("checkpointTs", m.checkpointTs),
 			zap.Time("nextRetryTime", m.nextRetryTime.Load()),
 		)
 		return true, lastError
 	}
 	// if any error is occurred , we should set the changefeed state to warning and stop the changefeed
 	log.Warn("changefeed meets an error, will be stopped",
-		zap.String("namespace", m.id.Name()),
+		zap.Stringer("changefeed", m.id),
+		zap.Uint64("checkpointTs", m.checkpointTs),
 		zap.Time("nextRetryTime", m.nextRetryTime.Load()),
 		zap.Any("error", errs))
 	// patch the last error to changefeed info

diff --git a/coordinator/changefeed/changefeed.go b/coordinator/changefeed/changefeed.go
@@ -141,7 +141,7 @@ func (c *Changefeed) UpdateStatus(newStatus *heartbeatpb.MaintainerStatus) (bool
 		c.status.Store(newStatus)
 		if old.BootstrapDone != newStatus.BootstrapDone {
 			log.Info("Received changefeed status with bootstrapDone",
-				zap.String("changefeed", c.ID.String()),
+				zap.Stringer("changefeed", c.ID),
 				zap.Bool("bootstrapDone", newStatus.BootstrapDone))
 			return true, model.StateNormal, nil
 		}

diff --git a/coordinator/changefeed/changefeed_db.go b/coordinator/changefeed/changefeed_db.go
@@ -224,7 +224,10 @@ func (db *ChangefeedDB) Resume(id common.ChangeFeedID, resetBackoff bool, overwr
 		delete(db.stopped, id)
 		cf.isNew = overwriteCheckpointTs
 		db.AddAbsentWithoutLock(cf)
-		log.Info("resume changefeed", zap.String("changefeed", id.String()), zap.Any("overwriteCheckpointTs", overwriteCheckpointTs))
+		log.Info("resume changefeed",
+			zap.Stringer("changefeed", id),
+			zap.Uint64("checkpointTs", cf.GetStatus().CheckpointTs),
+			zap.Bool("overwriteCheckpointTs", overwriteCheckpointTs))
 	}
 }
 

diff --git a/coordinator/changefeed/changefeed_db_test.go b/coordinator/changefeed/changefeed_db_test.go
@@ -79,6 +79,9 @@ func TestResume(t *testing.T) {
 	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test")}
 	db.AddStoppedChangefeed(cf)
 	cf.backoff = NewBackoff(cf.ID, 0, 0)
+	cf.status = atomic.NewPointer(&heartbeatpb.MaintainerStatus{
+		CheckpointTs: 100,
+	})
 
 	db.Resume(cf.ID, true, false)