TiCDC support checking if data is entirely replicated to Downstream

cdc/api/v1/api_test.go

@@ -100,6 +100,13 @@ func (p *mockStatusProvider) GetCaptures(ctx context.Context) ([]*model.CaptureI
 	return args.Get(0).([]*model.CaptureInfo), args.Error(1)
 }
 
+func (p *mockStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	args := p.Called(ctx)
+	return args.Get(0).(*model.ChangeFeedSyncedStatusForAPI), args.Error(1)
+}
+
 func (p *mockStatusProvider) IsHealthy(ctx context.Context) (bool, error) {
 	args := p.Called(ctx)
 	return args.Get(0).(bool), args.Error(1)

cdc/api/v2/api.go

@@ -63,6 +63,7 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	changefeedGroup.POST("/:changefeed_id/resume", changefeedOwnerMiddleware, api.resumeChangefeed)
 	changefeedGroup.POST("/:changefeed_id/pause", changefeedOwnerMiddleware, api.pauseChangefeed)
 	changefeedGroup.GET("/:changefeed_id/status", changefeedOwnerMiddleware, api.status)
+	changefeedGroup.GET("/:changefeed_id/synced", changefeedOwnerMiddleware, api.synced)
 
 	// capture apis
 	captureGroup := v2.Group("/captures")

cdc/api/v2/api_test.go

@@ -49,7 +49,7 @@ func (c *mockPDClient) UpdateServiceGCSafePoint(ctx context.Context,
 
 // GetTS of mockPDClient returns a mock tso
 func (c *mockPDClient) GetTS(ctx context.Context) (int64, int64, error) {
-	return c.logicTime, c.timestamp, nil
+	return c.timestamp, c.logicTime, nil
 }
 
 // GetClusterID of mockPDClient returns a mock ClusterID
@@ -62,13 +62,14 @@ func (c *mockPDClient) Close() {}
 
 type mockStatusProvider struct {
 	owner.StatusProvider
-	changefeedStatus   *model.ChangeFeedStatusForAPI
-	changefeedInfo     *model.ChangeFeedInfo
-	processors         []*model.ProcInfoSnap
-	taskStatus         map[model.CaptureID]*model.TaskStatus
-	changefeedInfos    map[model.ChangeFeedID]*model.ChangeFeedInfo
-	changefeedStatuses map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
-	err                error
+	changefeedStatus       *model.ChangeFeedStatusForAPI
+	changefeedInfo         *model.ChangeFeedInfo
+	processors             []*model.ProcInfoSnap
+	taskStatus             map[model.CaptureID]*model.TaskStatus
+	changefeedInfos        map[model.ChangeFeedID]*model.ChangeFeedInfo
+	changefeedStatuses     map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
+	changeFeedSyncedStatus *model.ChangeFeedSyncedStatusForAPI
+	err                    error
 }
 
 // GetChangeFeedStatus returns a changefeeds' runtime status.
@@ -120,6 +121,14 @@ func (m *mockStatusProvider) GetAllChangeFeedStatuses(_ context.Context) (
 	return m.changefeedStatuses, m.err
 }
 
+// GetChangeFeedSyncedStatus returns a mock changefeed status.
+func (m *mockStatusProvider) GetChangeFeedSyncedStatus(_ context.Context, changefeedID model.ChangeFeedID) (
+	*model.ChangeFeedSyncedStatusForAPI,
+	error,
+) {
+	return m.changeFeedSyncedStatus, m.err
+}
+
 func (m *mockStatusProvider) IsChangefeedOwner(_ context.Context, id model.ChangeFeedID) (bool, error) {
 	return true, nil
 }

cdc/api/v2/changefeed.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
@@ -895,6 +896,135 @@ func (h *OpenAPIV2) status(c *gin.Context) {
 	})
 }
 
+// transformer timestamp to readable format
+func transformerTime(timestamp int64) string {
+	location := time.Local
+	tm := time.Unix((timestamp / 1000), 0).In(location)
+	return tm.Format("2006-01-02 15:04:05")
+}
+
+// synced get the synced status of a changefeed
+// @Summary Get synced status
+// @Description get the synced status of a changefeed
+// @Tags changefeed,v2
+// @Accept json
+// @Produce json
+// @Param changefeed_id  path  string  true  "changefeed_id"
+// @Param namespace query string false "default"
+// @Success 200 {object} SyncedStatus
+// @Failure 500,400 {object} model.HTTPError
+// @Router /api/v2/changefeeds/{changefeed_id}/synced [get]
+func (h *OpenAPIV2) synced(c *gin.Context) {
+	ctx := c.Request.Context()
+
+	namespace := getNamespaceValueWithDefault(c)
+	changefeedID := model.ChangeFeedID{Namespace: namespace, ID: c.Param(apiOpVarChangefeedID)}
+	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
+		_ = c.Error(cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed_id: %s",
+			changefeedID.ID))
+		return
+	}
+
+	status, err := h.capture.StatusProvider().GetChangeFeedSyncedStatus(
+		ctx,
+		changefeedID,
+	)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+
+	log.Info("Get changefeed synced status:", zap.Any("status", status), zap.Any("changefeedID", changefeedID))
+
+	cfg := &ChangefeedConfig{ReplicaConfig: GetDefaultReplicaConfig()}
+	if (status.SyncedCheckInterval != 0) && (status.CheckpointInterval != 0) {
+		cfg.ReplicaConfig.SyncedStatus.CheckpointInterval = status.CheckpointInterval
+		cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval = status.SyncedCheckInterval
+	}
+
+	// get pd client
+	if len(cfg.PDAddrs) == 0 {
+		up, err := getCaptureDefaultUpstream(h.capture)
+		if err != nil {
+			_ = c.Error(err)
+			return
+		}
+		cfg.PDConfig = getUpstreamPDConfig(up)
+	}
+	credential := cfg.PDConfig.toCredential()
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+	if err != nil {
+		// pd is unavailable
+		var message string
+		if (status.PullerResolvedTs - status.CheckpointTs) > cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 { // 5s
+			message = fmt.Sprintf("%s. Besides the data is not finish syncing", terror.Message(err))
+		} else {
+			message = fmt.Sprintf("%s. You can check the pd first, and if pd is available, means we don't finish sync data. "+
+				"If pd is not available, please check the whether we satisfy the condition that "+
+				"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than %v secs. "+
+				"If it's satisfied, means the data syncing is totally finished", err, cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval)
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: transformerTime(status.CheckpointTs),
+			PullerResolvedTs: transformerTime(status.PullerResolvedTs),
+			LastSyncedTs:     transformerTime(status.LastSyncedTs),
+			NowTs:            transformerTime(0),
+			Info:             message,
+		})
+		return
+	}
+	defer pdClient.Close()
+
+	// get time from pd
+	physicalNow, _, _ := pdClient.GetTS(ctx)
+
+	if (physicalNow-status.LastSyncedTs > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000) &&
+		(physicalNow-status.CheckpointTs < cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000) {
+		// reach strict synced  condition
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           true,
+			SinkCheckpointTs: transformerTime(status.CheckpointTs),
+			PullerResolvedTs: transformerTime(status.PullerResolvedTs),
+			LastSyncedTs:     transformerTime(status.LastSyncedTs),
+			NowTs:            transformerTime(physicalNow),
+			Info:             "Data syncing is finished",
+		})
+	} else if physicalNow-status.LastSyncedTs > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000 {
+		// lastSyncedTs reach the synced condition, while checkpoint-ts doesn't
+		var message string
+		if (status.PullerResolvedTs - status.CheckpointTs) < cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+			message = fmt.Sprintf("Please check whether pd is health and tikv region is all available. " +
+				"If pd is not health or tikv region is not available, the data syncing is finished. " +
+				" Otherwise the data syncing is not finished, please wait")
+		} else {
+			message = "The data syncing is not finished, please wait"
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: transformerTime(status.CheckpointTs),
+			PullerResolvedTs: transformerTime(status.PullerResolvedTs),
+			LastSyncedTs:     transformerTime(status.LastSyncedTs),
+			NowTs:            transformerTime(physicalNow),
+			Info:             message,
+		})
+	} else {
+		// lastSyncedTs doesn't reach the synced condition
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: transformerTime(status.CheckpointTs),
+			PullerResolvedTs: transformerTime(status.PullerResolvedTs),
+			LastSyncedTs:     transformerTime(status.LastSyncedTs),
+			NowTs:            transformerTime(physicalNow),
+			Info:             "The data syncing is not finished, please wait",
+		})
+	}
+}
+
 func toAPIModel(
 	info *model.ChangeFeedInfo,
 	resolvedTs uint64,