Merge branch 'master' into rustin-patch-sink-hooks

cdc/api/owner.go

@@ -24,13 +24,14 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/tikv/client-go/v2/oracle"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.uber.org/zap"
+
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/logutil"
-	"github.com/tikv/client-go/v2/oracle"
-	"go.etcd.io/etcd/clientv3/concurrency"
-	"go.uber.org/zap"
 )
 
 const (

cdc/api/owner_test.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/stretchr/testify/require"
-	"go.etcd.io/etcd/clientv3/concurrency"
+	"go.etcd.io/etcd/client/v3/concurrency"
 )
 
 func TestHTTPStatus(t *testing.T) {

cdc/api/status.go

@@ -24,7 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/version"
-	"go.etcd.io/etcd/clientv3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 // status of cdc server

cdc/capture/capture.go

@@ -25,6 +25,13 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	tidbkv "github.com/pingcap/tidb/kv"
+	"github.com/tikv/client-go/v2/tikv"
+	pd "github.com/tikv/pd/client"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.etcd.io/etcd/server/v3/mvcc"
+	"go.uber.org/zap"
+	"golang.org/x/time/rate"
+
 	"github.com/pingcap/tiflow/cdc/kv"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/owner"
@@ -39,12 +46,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/pdtime"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/tikv/client-go/v2/tikv"
-	pd "github.com/tikv/pd/client"
-	"go.etcd.io/etcd/clientv3/concurrency"
-	"go.etcd.io/etcd/mvcc"
-	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )
 
 // Capture represents a Capture server, it monitors the changefeed information in etcd and schedules Task on it.

cdc/model/protocol.go

@@ -31,9 +31,10 @@ func DispatchTableTopic(changefeedID ChangeFeedID) p2p.Topic {
 
 // DispatchTableMessage is the message body for dispatching a table.
 type DispatchTableMessage struct {
-	OwnerRev int64   `json:"owner-rev"`
-	ID       TableID `json:"id"`
-	IsDelete bool    `json:"is-delete"`
+	OwnerRev int64          `json:"owner-rev"`
+	Epoch    ProcessorEpoch `json:"epoch"`
+	ID       TableID        `json:"id"`
+	IsDelete bool           `json:"is-delete"`
 }
 
 // DispatchTableResponseTopic returns a message topic for the result of
@@ -44,7 +45,8 @@ func DispatchTableResponseTopic(changefeedID ChangeFeedID) p2p.Topic {
 
 // DispatchTableResponseMessage is the message body for the result of dispatching a table.
 type DispatchTableResponseMessage struct {
-	ID TableID `json:"id"`
+	ID    TableID        `json:"id"`
+	Epoch ProcessorEpoch `json:"epoch"`
 }
 
 // AnnounceTopic returns a message topic for announcing an ownership change.
@@ -64,14 +66,23 @@ func SyncTopic(changefeedID ChangeFeedID) p2p.Topic {
 	return fmt.Sprintf("send-status-resp/%s", changefeedID)
 }
 
+// ProcessorEpoch designates a continuous period of the processor working normally.
+type ProcessorEpoch = string
+
 // SyncMessage is the message body for syncing the current states of a processor.
 // MsgPack serialization has been implemented to minimize the size of the message.
 type SyncMessage struct {
 	// Sends the processor's version for compatibility check
 	ProcessorVersion string
-	Running          []TableID
-	Adding           []TableID
-	Removing         []TableID
+
+	// Epoch is reset to a unique value when the processor has
+	// encountered an internal error or other events so that
+	// it has to re-sync its states with the Owner.
+	Epoch ProcessorEpoch
+
+	Running  []TableID
+	Adding   []TableID
+	Removing []TableID
 }
 
 // Marshal serializes the message into MsgPack format.

cdc/model/protocol_test.go

@@ -65,21 +65,23 @@ func makeVeryLargeSyncMessage() *SyncMessage {
 func TestMarshalDispatchTableMessage(t *testing.T) {
 	msg := &DispatchTableMessage{
 		OwnerRev: 1,
+		Epoch:    "test-epoch",
 		ID:       TableID(1),
 		IsDelete: true,
 	}
 	bytes, err := json.Marshal(msg)
 	require.NoError(t, err)
-	require.Equal(t, `{"owner-rev":1,"id":1,"is-delete":true}`, string(bytes))
+	require.Equal(t, `{"owner-rev":1,"epoch":"test-epoch","id":1,"is-delete":true}`, string(bytes))
 }
 
 func TestMarshalDispatchTableResponseMessage(t *testing.T) {
 	msg := &DispatchTableResponseMessage{
-		ID: TableID(1),
+		ID:    TableID(1),
+		Epoch: "test-epoch",
 	}
 	bytes, err := json.Marshal(msg)
 	require.NoError(t, err)
-	require.Equal(t, `{"id":1}`, string(bytes))
+	require.Equal(t, `{"id":1,"epoch":"test-epoch"}`, string(bytes))
 }
 
 func TestMarshalAnnounceMessage(t *testing.T) {

cdc/owner/metrics.go

@@ -61,7 +61,7 @@ var (
 			Subsystem: "owner",
 			Name:      "maintain_table_num",
 			Help:      "number of replicated tables maintained in owner",
-		}, []string{"changefeed", "type"})
+		}, []string{"changefeed", "capture", "type"})
 	changefeedStatusGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",

cdc/owner/owner.go

@@ -352,23 +352,29 @@ func (o *ownerImpl) updateMetrics(state *orchestrator.GlobalReactorState) {
 			pendingCounts := infoProvider.GetPendingTableCounts()
 
 			for captureID, info := range o.captures {
-				ownerMaintainTableNumGauge.WithLabelValues(
-					cfID, info.AdvertiseAddr, maintainTableTypeTotal).Set(float64(totalCounts[captureID]))
-				ownerMaintainTableNumGauge.WithLabelValues(
-					cfID, info.AdvertiseAddr, maintainTableTypeWip).Set(float64(pendingCounts[captureID]))
+				ownerMaintainTableNumGauge.
+					WithLabelValues(cfID, info.AdvertiseAddr, maintainTableTypeTotal).
+					Set(float64(totalCounts[captureID]))
+				ownerMaintainTableNumGauge.
+					WithLabelValues(cfID, info.AdvertiseAddr, maintainTableTypeWip).
+					Set(float64(pendingCounts[captureID]))
 			}
 		}
 		return
 	}
 
 	for changefeedID, changefeedState := range state.Changefeeds {
-		for captureID := range state.Captures {
+		for captureID, captureInfo := range state.Captures {
 			taskStatus, exist := changefeedState.TaskStatuses[captureID]
 			if !exist {
 				continue
 			}
-			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, maintainTableTypeTotal).Set(float64(len(taskStatus.Tables)))
-			ownerMaintainTableNumGauge.WithLabelValues(changefeedID, maintainTableTypeWip).Set(float64(len(taskStatus.Operation)))
+			ownerMaintainTableNumGauge.
+				WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeTotal).
+				Set(float64(len(taskStatus.Tables)))
+			ownerMaintainTableNumGauge.
+				WithLabelValues(changefeedID, captureInfo.AdvertiseAddr, maintainTableTypeWip).
+				Set(float64(len(taskStatus.Operation)))
 			if changefeedState.Info != nil {
 				changefeedStatusGauge.WithLabelValues(changefeedID).Set(float64(changefeedState.Info.State.ToInt()))
 			}

cdc/owner/scheduler.go

@@ -127,14 +127,27 @@ func (s *schedulerV2) DispatchTable(
 	tableID model.TableID,
 	captureID model.CaptureID,
 	isDelete bool,
+	epoch model.ProcessorEpoch,
 ) (done bool, err error) {
 	topic := model.DispatchTableTopic(changeFeedID)
 	message := &model.DispatchTableMessage{
 		OwnerRev: ctx.GlobalVars().OwnerRevision,
 		ID:       tableID,
 		IsDelete: isDelete,
+		Epoch:    epoch,
 	}
 
+	defer func() {
+		if err != nil {
+			return
+		}
+		log.Info("schedulerV2: DispatchTable",
+			zap.Any("message", message),
+			zap.Any("successful", done),
+			zap.String("changefeedID", changeFeedID),
+			zap.String("captureID", captureID))
+	}()
+
 	ok, err := s.trySendMessage(ctx, captureID, topic, message)
 	if err != nil {
 		return false, errors.Trace(err)
@@ -155,13 +168,24 @@ func (s *schedulerV2) Announce(
 	ctx context.Context,
 	changeFeedID model.ChangeFeedID,
 	captureID model.CaptureID,
-) (bool, error) {
+) (done bool, err error) {
 	topic := model.AnnounceTopic(changeFeedID)
 	message := &model.AnnounceMessage{
 		OwnerRev:     ctx.GlobalVars().OwnerRevision,
 		OwnerVersion: version.ReleaseSemver(),
 	}
 
+	defer func() {
+		if err != nil {
+			return
+		}
+		log.Info("schedulerV2: Announce",
+			zap.Any("message", message),
+			zap.Any("successful", done),
+			zap.String("changefeedID", changeFeedID),
+			zap.String("captureID", captureID))
+	}()
+
 	ok, err := s.trySendMessage(ctx, captureID, topic, message)
 	if err != nil {
 		return false, errors.Trace(err)
@@ -239,7 +263,7 @@ func (s *schedulerV2) registerPeerMessageHandlers(ctx context.Context) (ret erro
 		func(sender string, messageI interface{}) error {
 			message := messageI.(*model.DispatchTableResponseMessage)
 			s.stats.RecordDispatchResponse()
-			s.OnAgentFinishedTableOperation(sender, message.ID)
+			s.OnAgentFinishedTableOperation(sender, message.ID, message.Epoch)
 			return nil
 		})
 	if err != nil {
@@ -256,6 +280,7 @@ func (s *schedulerV2) registerPeerMessageHandlers(ctx context.Context) (ret erro
 			s.stats.RecordSync()
 			s.OnAgentSyncTaskStatuses(
 				sender,
+				message.Epoch,
 				message.Running,
 				message.Adding,
 				message.Removing)