redo(ticdc): make implementation in manager asynchronously

cdc/metrics.go

@@ -19,7 +19,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/owner"
 	"github.com/pingcap/tiflow/cdc/processor"
 	"github.com/pingcap/tiflow/cdc/puller"
-	redowriter "github.com/pingcap/tiflow/cdc/redo/writer"
+	redo "github.com/pingcap/tiflow/cdc/redo/common"
 	"github.com/pingcap/tiflow/cdc/scheduler"
 	sink "github.com/pingcap/tiflow/cdc/sink/metrics"
 	"github.com/pingcap/tiflow/cdc/sink/mq/producer/kafka"
@@ -59,7 +59,7 @@ func init() {
 	memory.InitMetrics(registry)
 	unified.InitMetrics(registry)
 	leveldb.InitMetrics(registry)
-	redowriter.InitMetrics(registry)
+	redo.InitMetrics(registry)
 	db.InitMetrics(registry)
 	kafka.InitMetrics(registry)
 	scheduler.InitMetrics(registry)

cdc/model/sink.go

@@ -29,18 +29,18 @@ import (
 
 //go:generate msgp
 
-// MqMessageType is the type of message
-type MqMessageType int
+// MessageType is the type of message, which is used by MqSink and RedoLog.
+type MessageType int
 
 const (
-	// MqMessageTypeUnknown is unknown type of message key
-	MqMessageTypeUnknown MqMessageType = iota
-	// MqMessageTypeRow is row type of message key
-	MqMessageTypeRow
-	// MqMessageTypeDDL is ddl type of message key
-	MqMessageTypeDDL
-	// MqMessageTypeResolved is resolved type of message key
-	MqMessageTypeResolved
+	// MessageTypeUnknown is unknown type of message key
+	MessageTypeUnknown MessageType = iota
+	// MessageTypeRow is row type of message key
+	MessageTypeRow
+	// MessageTypeDDL is ddl type of message key
+	MessageTypeDDL
+	// MessageTypeResolved is resolved type of message key
+	MessageTypeResolved
 )
 
 // ColumnFlagType is for encapsulating the flag operations for different flags.

cdc/processor/pipeline/sink.go

@@ -124,17 +124,20 @@ func (n *sinkNode) flushSink(ctx context.Context, resolved model.ResolvedTs) (er
 		}
 	}()
 
-	currentBarrierTs := atomic.LoadUint64(&n.barrierTs)
 	if resolved.Ts > n.targetTs {
 		resolved = model.NewResolvedTs(n.targetTs)
 	}
 
+	currentBarrierTs := atomic.LoadUint64(&n.barrierTs)
 	if n.redoManager != nil && n.redoManager.Enabled() {
 		// redo log do not support batch resolve mode, hence we
 		// use `ResolvedMark` to restore a normal resolved ts
 		resolved = model.NewResolvedTs(resolved.ResolvedMark())
-		err = n.redoManager.FlushLog(ctx, n.tableID, resolved.Ts)
+		err = n.redoManager.UpdateResolvedTs(ctx, n.tableID, resolved.Ts)
 
+		// fail fast check, the happens before relationship is:
+		// 1. sorter resolvedTs >= sink resolvedTs >= table redoTs == tableActor resolvedTs
+		// 2. tableActor resolvedTs >= processor resolvedTs >= global resolvedTs >= barrierTs
 		redoTs := n.redoManager.GetMinResolvedTs()
 		if redoTs < currentBarrierTs {
 			log.Debug("redoTs should not less than current barrierTs",
@@ -143,7 +146,7 @@ func (n *sinkNode) flushSink(ctx context.Context, resolved model.ResolvedTs) (er
 				zap.Uint64("barrierTs", currentBarrierTs))
 		}
 
-		// Fixme(CharlesCheung): remove this check after refactoring redoManager
+		// TODO: remove this check after SchedulerV3 become the first choice.
 		if resolved.Ts > redoTs {
 			resolved = model.NewResolvedTs(redoTs)
 		}

cdc/processor/pipeline/sink_test.go

@@ -15,6 +15,7 @@ package pipeline
 
 import (
 	"context"
+	"math/rand"
 	"sync"
 	"testing"
 	"time"
@@ -138,8 +139,9 @@ func TestState(t *testing.T) {
 
 	state := TableStatePrepared
 	// test stop at targetTs
-	node := newSinkNode(1, &mockSink{}, 0, 10, &mockFlowController{}, redo.NewDisabledManager(),
-		&state, ctx.ChangefeedVars().ID, false)
+	targetTs := model.Ts(10)
+	node := newSinkNode(1, &mockSink{}, 0, targetTs, &mockFlowController{}, redo.NewDisabledManager(),
+		&state, ctx.ChangefeedVars().ID, true)
 	node.initWithReplicaConfig(pipeline.MockNodeContext4Test(ctx, pmessage.Message{}, nil).
 		ChangefeedVars().Info.Config)
 	require.Equal(t, TableStatePrepared, node.State())
@@ -177,6 +179,23 @@ func TestState(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, TableStateReplicating, node.State())
 
+	batchResolved := model.ResolvedTs{
+		Mode:    model.BatchResolvedMode,
+		Ts:      targetTs,
+		BatchID: rand.Uint64() % 10,
+	}
+	msg = pmessage.PolymorphicEventMessage(&model.PolymorphicEvent{
+		CRTs:     targetTs,
+		Resolved: &batchResolved,
+		RawKV:    &model.RawKVEntry{OpType: model.OpTypeResolved},
+		Row:      &model.RowChangedEvent{},
+	})
+	ok, err = node.HandleMessage(ctx, msg)
+	require.Nil(t, err)
+	require.True(t, ok)
+	require.Equal(t, batchResolved, node.getResolvedTs())
+	require.Equal(t, batchResolved, node.getCheckpointTs())
+
 	msg = pmessage.PolymorphicEventMessage(&model.PolymorphicEvent{
 		CRTs: 15, RawKV: &model.RawKVEntry{OpType: model.OpTypeResolved},
 		Row: &model.RowChangedEvent{},
@@ -185,7 +204,7 @@ func TestState(t *testing.T) {
 	require.False(t, ok)
 	require.True(t, cerrors.ErrTableProcessorStoppedSafely.Equal(err))
 	require.Equal(t, TableStateStopped, node.State())
-	require.Equal(t, model.Ts(10), node.CheckpointTs())
+	require.Equal(t, targetTs, node.CheckpointTs())
 
 	// test the stop at ts command
 	state = TableStatePrepared

cdc/processor/pipeline/sorter.go

@@ -168,9 +168,9 @@ func (n *sorterNode) start(
 			if lastCRTs > lastSentResolvedTs && commitTs > lastCRTs {
 				lastSentResolvedTs = lastCRTs
 				lastSendResolvedTsTime = time.Now()
+				msg := model.NewResolvedPolymorphicEvent(0, lastSentResolvedTs)
+				ctx.SendToNextNode(pmessage.PolymorphicEventMessage(msg))
 			}
-			msg := model.NewResolvedPolymorphicEvent(0, lastSentResolvedTs)
-			ctx.SendToNextNode(pmessage.PolymorphicEventMessage(msg))
 		}
 
 		// once receive startTs, which means sink should start replicating data to downstream.

cdc/processor/pipeline/table_actor.go

@@ -396,7 +396,7 @@ func (t *tableActor) ResolvedTs() model.Ts {
 	// another replication barrier for consistent replication instead of reusing
 	// the global resolved-ts.
 	if t.redoManager.Enabled() {
-		return t.sinkNode.ResolvedTs()
+		return t.redoManager.GetResolvedTs(t.tableID)
 	}
 	return t.sortNode.ResolvedTs()
 }

cdc/processor/pipeline/table_actor_test.go

@@ -95,12 +95,15 @@ func TestTableActorInterface(t *testing.T) {
 	require.Equal(t, model.Ts(3), table.CheckpointTs())
 
 	require.Equal(t, model.Ts(5), table.ResolvedTs())
-	table.replicaConfig.Consistent.Level = string(redo.ConsistentLevelEventual)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	table.redoManager, _ = redo.NewMockManager(ctx)
-	table.sinkNode.resolvedTs.Store(model.NewResolvedTs(6))
-	require.Equal(t, model.Ts(6), table.ResolvedTs())
+	table.redoManager.AddTable(table.tableID, 0)
+	require.Equal(t, model.Ts(0), table.ResolvedTs())
+	table.redoManager.UpdateResolvedTs(ctx, table.tableID, model.Ts(6))
+	require.Eventually(t, func() bool { return table.ResolvedTs() == model.Ts(6) },
+		time.Second*5, time.Millisecond*500)
+	table.redoManager.Cleanup(ctx)
 
 	table.sinkNode.state.Store(TableStateStopped)
 	require.Equal(t, TableStateStopped, table.State())

cdc/processor/processor.go

@@ -891,6 +891,10 @@ func (p *processor) createTablePipelineImpl(
 		return nil
 	})
 
+	if p.redoManager.Enabled() {
+		p.redoManager.AddTable(tableID, replicaInfo.StartTs)
+	}
+
 	tableName := p.getTableName(ctx, tableID)
 
 	s, err := sink.NewTableSink(p.sink, tableID, p.metricsTableSinkTotalRows)
@@ -911,10 +915,6 @@ func (p *processor) createTablePipelineImpl(
 		return nil, errors.Trace(err)
 	}
 
-	if p.redoManager.Enabled() {
-		p.redoManager.AddTable(tableID, replicaInfo.StartTs)
-	}
-
 	log.Info("Add table pipeline", zap.Int64("tableID", tableID),
 		cdcContext.ZapFieldChangefeed(ctx),
 		zap.String("name", table.Name()),

cdc/redo/common/metric.go

@@ -0,0 +1,88 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	namespace = "ticdc"
+	subsystem = "redo"
+)
+
+var (
+	// RedoWriteBytesGauge records the total number of bytes written to redo log.
+	RedoWriteBytesGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "write_bytes_total",
+		Help:      "Total number of bytes redo log written",
+	}, []string{"namespace", "changefeed"})
+
+	// RedoFsyncDurationHistogram records the latency distributions of fsync called by redo writer.
+	RedoFsyncDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "fsync_duration_seconds",
+		Help:      "The latency distributions of fsync called by redo writer",
+		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 13),
+	}, []string{"namespace", "changefeed"})
+
+	// RedoFlushAllDurationHistogram records the latency distributions of flushAll
+	// called by redo writer.
+	RedoFlushAllDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "flushall_duration_seconds",
+		Help:      "The latency distributions of flushall called by redo writer",
+		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 13),
+	}, []string{"namespace", "changefeed"})
+
+	// RedoTotalRowsCountGauge records the total number of rows written to redo log.
+	RedoTotalRowsCountGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "total_rows_count",
+		Help:      "The total count of rows that are processed by redo writer",
+	}, []string{"namespace", "changefeed"})
+
+	// RedoWriteLogDurationHistogram records the latency distributions of writeLog.
+	RedoWriteLogDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "write_log_duration_seconds",
+		Help:      "The latency distributions of writeLog called by redoManager",
+		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 13),
+	}, []string{"namespace", "changefeed"})
+
+	// RedoFlushLogDurationHistogram records the latency distributions of flushLog.
+	RedoFlushLogDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Subsystem: subsystem,
+		Name:      "flush_log_duration_seconds",
+		Help:      "The latency distributions of flushLog called by redoManager",
+		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 13),
+	}, []string{"namespace", "changefeed"})
+)
+
+// InitMetrics registers all metrics in this file
+func InitMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(RedoFsyncDurationHistogram)
+	registry.MustRegister(RedoTotalRowsCountGauge)
+	registry.MustRegister(RedoWriteBytesGauge)
+	registry.MustRegister(RedoFlushAllDurationHistogram)
+	registry.MustRegister(RedoWriteLogDurationHistogram)
+	registry.MustRegister(RedoFlushLogDurationHistogram)
+}