redo(ticdc): use uuid in redo meta filename (pingcap#8075)

close pingcap#8028, close pingcap#8074

Makefile

@@ -109,8 +109,9 @@ cdc:
 kafka_consumer:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/cdc_kafka_consumer ./cmd/kafka-consumer/main.go
 
-install:
-	go install ./...
+cdc_test_image: 
+	@which docker || (echo "docker not found in ${PATH}"; exit 1)
+	docker build --platform linux/amd64 -f deployments/ticdc/docker/test.Dockerfile -t cdc:test ./ 
 
 unit_test: check_failpoint_ctl generate_mock generate-msgp-code generate-protobuf
 	mkdir -p "$(TEST_DIR)"

cdc/owner/changefeed_test.go

@@ -490,7 +490,7 @@ func TestRemoveChangefeed(t *testing.T) {
 	info.Config.Consistent = &config.ConsistentConfig{
 		Level:             "eventual",
 		Storage:           filepath.Join("nfs://", dir),
-		FlushIntervalInMs: config.MinFlushIntervalInMs,
+		FlushIntervalInMs: config.DefaultFlushIntervalInMs,
 	}
 	ctx = cdcContext.WithChangefeedVars(ctx, &cdcContext.ChangefeedVars{
 		ID:   ctx.ChangefeedVars().ID,

cdc/redo/common/metric.go

@@ -75,6 +75,15 @@ var (
 		Help:      "The latency distributions of flushLog called by redoManager",
 		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 13),
 	}, []string{"namespace", "changefeed"})
+
+	// RedoWorkerBusyRatio records the busy ratio of redo bgUpdateLog worker.
+	RedoWorkerBusyRatio = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "worker_busy_ratio",
+			Help:      "Busy ratio (X ms in 1s) for redo bgUpdateLog worker.",
+		}, []string{"namespace", "changefeed"})
 )
 
 // InitMetrics registers all metrics in this file
@@ -85,4 +94,5 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(RedoFlushAllDurationHistogram)
 	registry.MustRegister(RedoWriteLogDurationHistogram)
 	registry.MustRegister(RedoFlushLogDurationHistogram)
+	registry.MustRegister(RedoWorkerBusyRatio)
 }

cdc/redo/manager.go

@@ -133,8 +133,10 @@ type ManagerImpl struct {
 	flushing      int64
 	lastFlushTime time.Time
 
-	metricWriteLogDuration prometheus.Observer
-	metricFlushLogDuration prometheus.Observer
+	metricWriteLogDuration    prometheus.Observer
+	metricFlushLogDuration    prometheus.Observer
+	metricTotalRowsCount      prometheus.Counter
+	metricRedoWorkerBusyRatio prometheus.Counter
 }
 
 // NewManager creates a new Manager
@@ -170,6 +172,10 @@ func NewManager(ctx context.Context, cfg *config.ConsistentConfig, opts *Manager
 		WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
 	m.metricFlushLogDuration = common.RedoFlushLogDurationHistogram.
 		WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
+	m.metricTotalRowsCount = common.RedoTotalRowsCountGauge.
+		WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
+	m.metricRedoWorkerBusyRatio = common.RedoWorkerBusyRatio.
+		WithLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
 
 	// TODO: better to wait background goroutines after the context is canceled.
 	if m.opts.EnableBgRunner {
@@ -193,7 +199,7 @@ func NewMockManager(ctx context.Context) (*ManagerImpl, error) {
 	cfg := &config.ConsistentConfig{
 		Level:             string(redo.ConsistentLevelEventual),
 		Storage:           "blackhole://",
-		FlushIntervalInMs: config.MinFlushIntervalInMs,
+		FlushIntervalInMs: config.DefaultFlushIntervalInMs,
 	}
 
 	errCh := make(chan error, 1)
@@ -359,6 +365,10 @@ func (m *ManagerImpl) Cleanup(ctx context.Context) error {
 		DeleteLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
 	common.RedoFlushLogDurationHistogram.
 		DeleteLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
+	common.RedoTotalRowsCountGauge.
+		DeleteLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
+	common.RedoWorkerBusyRatio.
+		DeleteLabelValues(m.changeFeedID.Namespace, m.changeFeedID.ID)
 	return m.withLock(func(m *ManagerImpl) error { return m.writer.DeleteAllLogs(ctx) })
 }
 
@@ -422,7 +432,13 @@ func (m *ManagerImpl) postFlushMeta(metaCheckpoint, metaResolved model.Ts) {
 	m.metaCheckpointTs.setFlushed(metaCheckpoint)
 }
 
-func (m *ManagerImpl) flushLog(ctx context.Context, handleErr func(err error)) {
+func (m *ManagerImpl) flushLog(
+	ctx context.Context, handleErr func(err error), workTimeSlice *time.Duration,
+) {
+	start := time.Now()
+	defer func() {
+		*workTimeSlice += time.Since(start)
+	}()
 	if !atomic.CompareAndSwapInt64(&m.flushing, 0, 1) {
 		log.Debug("Fail to update flush flag, " +
 			"the previous flush operation hasn't finished yet")
@@ -473,7 +489,8 @@ func (m *ManagerImpl) bgUpdateLog(
 
 	log.Info("redo manager bgUpdateLog is running",
 		zap.String("namespace", m.changeFeedID.Namespace),
-		zap.String("changefeed", m.changeFeedID.ID))
+		zap.String("changefeed", m.changeFeedID.ID),
+		zap.Int64("flushIntervalInMs", flushIntervalInMs))
 
 	ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond)
 	defer func() {
@@ -499,14 +516,23 @@ func (m *ManagerImpl) bgUpdateLog(
 	}()
 
 	var err error
+	overseerTicker := time.NewTicker(time.Second * 5)
+	defer overseerTicker.Stop()
+	var workTimeSlice time.Duration
+	startToWork := time.Now()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case err = <-logErrCh:
+		case now := <-overseerTicker.C:
+			busyRatio := int(workTimeSlice.Seconds() / now.Sub(startToWork).Seconds() * 1000)
+			m.metricRedoWorkerBusyRatio.Add(float64(busyRatio))
+			startToWork = now
+			workTimeSlice = 0
 		case <-ticker.C:
 			// interpolate tick message to flush writer if needed
-			m.flushLog(ctx, handleErr)
+			m.flushLog(ctx, handleErr, &workTimeSlice)
 		case cache, ok := <-m.logBuffer.Out():
 			if !ok {
 				return // channel closed
@@ -520,6 +546,7 @@ func (m *ManagerImpl) bgUpdateLog(
 				}
 				err = m.writer.WriteLog(ctx, cache.tableID, logs)
 				m.metricWriteLogDuration.Observe(time.Since(start).Seconds())
+				m.metricTotalRowsCount.Add(float64(len(logs)))
 			case model.MessageTypeResolved:
 				m.onResolvedTsMsg(cache.tableID, cache.resolvedTs)
 			default:

cdc/redo/manager_test.go

@@ -100,7 +100,7 @@ func TestLogManagerInProcessor(t *testing.T) {
 	defer logMgr.Cleanup(ctx)
 
 	checkResolvedTs := func(mgr LogManager, expectedRts uint64) {
-		time.Sleep(time.Duration(config.MinFlushIntervalInMs+200) * time.Millisecond)
+		time.Sleep(time.Duration(config.DefaultFlushIntervalInMs+200) * time.Millisecond)
 		resolvedTs := mgr.GetMinResolvedTs()
 		require.Equal(t, expectedRts, resolvedTs)
 	}
@@ -322,7 +322,7 @@ func TestManagerError(t *testing.T) {
 	cfg := &config.ConsistentConfig{
 		Level:             string(redo.ConsistentLevelEventual),
 		Storage:           "blackhole://",
-		FlushIntervalInMs: config.MinFlushIntervalInMs,
+		FlushIntervalInMs: config.DefaultFlushIntervalInMs,
 	}
 
 	errCh := make(chan error, 1)
@@ -387,7 +387,7 @@ func TestReuseWritter(t *testing.T) {
 	cfg := &config.ConsistentConfig{
 		Level:             string(redo.ConsistentLevelEventual),
 		Storage:           "local://" + dir,
-		FlushIntervalInMs: config.MinFlushIntervalInMs,
+		FlushIntervalInMs: config.DefaultFlushIntervalInMs,
 	}
 
 	errCh := make(chan error, 1)
@@ -410,7 +410,8 @@ func TestReuseWritter(t *testing.T) {
 	time.Sleep(time.Duration(100) * time.Millisecond)
 
 	// The another redo manager shouldn't be influenced.
-	mgrs[1].flushLog(ctxs[1], func(err error) { opts.ErrCh <- err })
+	var workTimeSlice time.Duration
+	mgrs[1].flushLog(ctxs[1], func(err error) { opts.ErrCh <- err }, &workTimeSlice)
 	select {
 	case x := <-errCh:
 		log.Panic("shouldn't get an error", zap.Error(x))

cdc/redo/reader/file.go

@@ -134,7 +134,7 @@ func selectDownLoadFile(
 			return nil
 		})
 	if err != nil {
-		return nil, cerror.WrapError(cerror.ErrS3StorageAPI, err)
+		return nil, cerror.WrapError(cerror.ErrExternalStorageAPI, err)
 	}
 
 	return files, nil
@@ -154,7 +154,7 @@ func downLoadToLocal(
 		eg.Go(func() error {
 			data, err := extStorage.ReadFile(eCtx, f)
 			if err != nil {
-				return cerror.WrapError(cerror.ErrS3StorageAPI, err)
+				return cerror.WrapError(cerror.ErrExternalStorageAPI, err)
 			}
 
 			err = os.MkdirAll(dir, redo.DefaultDirMode)

cdc/redo/writer/file.go

@@ -366,7 +366,7 @@ func (w *Writer) close() error {
 		if err != nil {
 			w.file.Close()
 			w.file = nil
-			return cerror.WrapError(cerror.ErrS3StorageAPI, err)
+			return cerror.WrapError(cerror.ErrExternalStorageAPI, err)
 		}
 	}
 
@@ -490,7 +490,7 @@ func (w *Writer) GC(checkPointTs uint64) error {
 				errs = multierr.Append(errs, err)
 			}
 			if errs != nil {
-				errs = cerror.WrapError(cerror.ErrS3StorageAPI, errs)
+				errs = cerror.WrapError(cerror.ErrExternalStorageAPI, errs)
 				log.Warn("delete redo log in s3 fail", zap.Error(errs))
 			}
 		}()
@@ -610,7 +610,7 @@ func (w *Writer) writeToS3(ctx context.Context, name string) error {
 	// Key in s3: aws.String(rs.options.Prefix + name), prefix should be changefeed name
 	err = w.storage.WriteFile(ctx, filepath.Base(name), fileData)
 	if err != nil {
-		return cerror.WrapError(cerror.ErrS3StorageAPI, err)
+		return cerror.WrapError(cerror.ErrExternalStorageAPI, err)
 	}
 
 	// in case the page cache piling up triggered the OS memory reclaming which may cause