sink(ticdc): cleanup expired files by day for storage sink (#10097)

cdc/api/v2/model.go

@@ -381,9 +381,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		var cloudStorageConfig *config.CloudStorageConfig
 		if c.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &config.CloudStorageConfig{
-				WorkerCount:   c.Sink.CloudStorageConfig.WorkerCount,
-				FlushInterval: c.Sink.CloudStorageConfig.FlushInterval,
-				FileSize:      c.Sink.CloudStorageConfig.FileSize,
+				WorkerCount:         c.Sink.CloudStorageConfig.WorkerCount,
+				FlushInterval:       c.Sink.CloudStorageConfig.FlushInterval,
+				FileSize:            c.Sink.CloudStorageConfig.FileSize,
+				OutputColumnID:      c.Sink.CloudStorageConfig.OutputColumnID,
+				FileExpirationDays:  c.Sink.CloudStorageConfig.FileExpirationDays,
+				FileCleanupCronSpec: c.Sink.CloudStorageConfig.FileCleanupCronSpec,
 			}
 		}
 
@@ -601,9 +604,12 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		var cloudStorageConfig *CloudStorageConfig
 		if cloned.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &CloudStorageConfig{
-				WorkerCount:   cloned.Sink.CloudStorageConfig.WorkerCount,
-				FlushInterval: cloned.Sink.CloudStorageConfig.FlushInterval,
-				FileSize:      cloned.Sink.CloudStorageConfig.FileSize,
+				WorkerCount:         cloned.Sink.CloudStorageConfig.WorkerCount,
+				FlushInterval:       cloned.Sink.CloudStorageConfig.FlushInterval,
+				FileSize:            cloned.Sink.CloudStorageConfig.FileSize,
+				OutputColumnID:      cloned.Sink.CloudStorageConfig.OutputColumnID,
+				FileExpirationDays:  cloned.Sink.CloudStorageConfig.FileExpirationDays,
+				FileCleanupCronSpec: cloned.Sink.CloudStorageConfig.FileCleanupCronSpec,
 			}
 		}
 
@@ -1038,9 +1044,12 @@ type MySQLConfig struct {
 
 // CloudStorageConfig represents a cloud storage sink configuration
 type CloudStorageConfig struct {
-	WorkerCount   *int    `json:"worker_count,omitempty"`
-	FlushInterval *string `json:"flush_interval,omitempty"`
-	FileSize      *int    `json:"file_size,omitempty"`
+	WorkerCount         *int    `json:"worker_count,omitempty"`
+	FlushInterval       *string `json:"flush_interval,omitempty"`
+	FileSize            *int    `json:"file_size,omitempty"`
+	OutputColumnID      *bool   `json:"output_column_id,omitempty"`
+	FileExpirationDays  *int    `json:"file_expiration_days,omitempty"`
+	FileCleanupCronSpec *string `json:"file_cleanup_cron_spec,omitempty"`
 }
 
 // ChangefeedStatus holds common information of a changefeed in cdc

cdc/redo/meta_manager_test.go

@@ -313,12 +313,14 @@ func TestGCAndCleanup(t *testing.T) {
 	cancel()
 	require.ErrorIs(t, eg.Wait(), context.Canceled)
 
-	m.Cleanup(ctx)
-	ret, err := extStorage.FileExists(ctx, getDeletedChangefeedMarker(changefeedID))
+	cleanupCtx, cleanupCancel := context.WithCancel(context.Background())
+	defer cleanupCancel()
+	m.Cleanup(cleanupCtx)
+	ret, err := extStorage.FileExists(cleanupCtx, getDeletedChangefeedMarker(changefeedID))
 	require.NoError(t, err)
 	require.True(t, ret)
 	cnt := 0
-	extStorage.WalkDir(ctx, nil, func(path string, size int64) error {
+	extStorage.WalkDir(cleanupCtx, nil, func(path string, size int64) error {
 		cnt++
 		return nil
 	})

cdc/sink/ddlsink/cloudstorage/cloud_storage_ddl_sink.go

@@ -17,19 +17,22 @@ import (
 	"context"
 	"encoding/json"
 	"net/url"
+	"sync/atomic"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tiflow/cdc/contextutil"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
 	"github.com/pingcap/tiflow/cdc/sink/metrics"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink"
 	"github.com/pingcap/tiflow/pkg/sink/cloudstorage"
 	"github.com/pingcap/tiflow/pkg/util"
+	"github.com/robfig/cron"
 	"go.uber.org/zap"
 )
 
@@ -43,25 +46,55 @@ type DDLSink struct {
 	// statistic is used to record the DDL metrics
 	statistics *metrics.Statistics
 	storage    storage.ExternalStorage
+	cfg        *cloudstorage.Config
+	cron       *cron.Cron
 
+	lastCheckpointTs         atomic.Uint64
 	lastSendCheckpointTsTime time.Time
 }
 
 // NewDDLSink creates a ddl sink for cloud storage.
-func NewDDLSink(ctx context.Context, sinkURI *url.URL) (*DDLSink, error) {
+func NewDDLSink(ctx context.Context,
+	sinkURI *url.URL,
+	replicaConfig *config.ReplicaConfig,
+) (*DDLSink, error) {
+	changefeedID := contextutil.ChangefeedIDFromCtx(ctx)
+	return newDDLSink(ctx, changefeedID, sinkURI, replicaConfig, nil)
+}
+
+func newDDLSink(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+	sinkURI *url.URL,
+	replicaConfig *config.ReplicaConfig,
+	cleanupJobs []func(), /* only for test */
+) (*DDLSink, error) {
+	// create cloud storage config and then apply the params of sinkURI to it.
+	cfg := cloudstorage.NewConfig()
+	err := cfg.Apply(ctx, sinkURI, replicaConfig)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	storage, err := util.GetExternalStorageFromURI(ctx, sinkURI.String())
 	if err != nil {
 		return nil, err
 	}
 
-	changefeedID := contextutil.ChangefeedIDFromCtx(ctx)
 	d := &DDLSink{
 		id:                       changefeedID,
 		storage:                  storage,
 		statistics:               metrics.NewStatistics(ctx, sink.TxnSink),
+		cfg:                      cfg,
 		lastSendCheckpointTsTime: time.Now(),
 	}
 
+	if err := d.initCron(ctx, sinkURI, cleanupJobs); err != nil {
+		return nil, errors.Trace(err)
+	}
+	// Note: It is intended to run the cleanup goroutine in the background.
+	// we don't wait for it to finish since the gourotine would be stuck if
+	// the downstream is abnormal, especially when the downstream is a nfs.
+	go d.bgCleanup(ctx)
 	return d, nil
 }
 
@@ -117,6 +150,7 @@ func (d *DDLSink) WriteCheckpointTs(ctx context.Context,
 
 	defer func() {
 		d.lastSendCheckpointTsTime = time.Now()
+		d.lastCheckpointTs.Store(ts)
 	}()
 	ckpt, err := json.Marshal(map[string]uint64{"checkpoint-ts": ts})
 	if err != nil {
@@ -126,6 +160,118 @@ func (d *DDLSink) WriteCheckpointTs(ctx context.Context,
 	return errors.Trace(err)
 }
 
+func (d *DDLSink) initCron(
+	ctx context.Context, sinkURI *url.URL, cleanupJobs []func(),
+) (err error) {
+	if cleanupJobs == nil {
+		cleanupJobs = d.genCleanupJob(ctx, sinkURI)
+	}
+
+	d.cron = cron.New()
+	for _, job := range cleanupJobs {
+		err = d.cron.AddFunc(d.cfg.FileCleanupCronSpec, job)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (d *DDLSink) bgCleanup(ctx context.Context) {
+	if d.cfg.DateSeparator != config.DateSeparatorDay.String() || d.cfg.FileExpirationDays <= 0 {
+		log.Info("skip cleanup expired files for storage sink",
+			zap.String("namespace", d.id.Namespace),
+			zap.String("changefeedID", d.id.ID),
+			zap.String("dateSeparator", d.cfg.DateSeparator),
+			zap.Int("expiredFileTTL", d.cfg.FileExpirationDays))
+		return
+	}
+
+	d.cron.Start()
+	defer d.cron.Stop()
+	log.Info("start schedule cleanup expired files for storage sink",
+		zap.String("namespace", d.id.Namespace),
+		zap.String("changefeedID", d.id.ID),
+		zap.String("dateSeparator", d.cfg.DateSeparator),
+		zap.Int("expiredFileTTL", d.cfg.FileExpirationDays))
+
+	// wait for the context done
+	<-ctx.Done()
+	log.Info("stop schedule cleanup expired files for storage sink",
+		zap.String("namespace", d.id.Namespace),
+		zap.String("changefeedID", d.id.ID),
+		zap.Error(ctx.Err()))
+}
+
+func (d *DDLSink) genCleanupJob(ctx context.Context, uri *url.URL) []func() {
+	ret := []func(){}
+
+	isLocal := uri.Scheme == "file" || uri.Scheme == "local" || uri.Scheme == ""
+	isRemoveEmptyDirsRuning := atomic.Bool{}
+	if isLocal {
+		ret = append(ret, func() {
+			if !isRemoveEmptyDirsRuning.CompareAndSwap(false, true) {
+				log.Warn("remove empty dirs is already running, skip this round",
+					zap.String("namespace", d.id.Namespace),
+					zap.String("changefeedID", d.id.ID))
+				return
+			}
+
+			checkpointTs := d.lastCheckpointTs.Load()
+			start := time.Now()
+			cnt, err := cloudstorage.RemoveEmptyDirs(ctx, d.id, uri.Path)
+			if err != nil {
+				log.Error("failed to remove empty dirs",
+					zap.String("namespace", d.id.Namespace),
+					zap.String("changefeedID", d.id.ID),
+					zap.Uint64("checkpointTs", checkpointTs),
+					zap.Duration("cost", time.Since(start)),
+					zap.Error(err),
+				)
+				return
+			}
+			log.Info("remove empty dirs",
+				zap.String("namespace", d.id.Namespace),
+				zap.String("changefeedID", d.id.ID),
+				zap.Uint64("checkpointTs", checkpointTs),
+				zap.Uint64("count", cnt),
+				zap.Duration("cost", time.Since(start)))
+		})
+	}
+
+	isCleanupRunning := atomic.Bool{}
+	ret = append(ret, func() {
+		if !isCleanupRunning.CompareAndSwap(false, true) {
+			log.Warn("cleanup expired files is already running, skip this round",
+				zap.String("namespace", d.id.Namespace),
+				zap.String("changefeedID", d.id.ID))
+			return
+		}
+
+		defer isCleanupRunning.Store(false)
+		start := time.Now()
+		checkpointTs := d.lastCheckpointTs.Load()
+		cnt, err := cloudstorage.RemoveExpiredFiles(ctx, d.id, d.storage, d.cfg, checkpointTs)
+		if err != nil {
+			log.Error("failed to remove expired files",
+				zap.String("namespace", d.id.Namespace),
+				zap.String("changefeedID", d.id.ID),
+				zap.Uint64("checkpointTs", checkpointTs),
+				zap.Duration("cost", time.Since(start)),
+				zap.Error(err),
+			)
+			return
+		}
+		log.Info("remove expired files",
+			zap.String("namespace", d.id.Namespace),
+			zap.String("changefeedID", d.id.ID),
+			zap.Uint64("checkpointTs", checkpointTs),
+			zap.Uint64("count", cnt),
+			zap.Duration("cost", time.Since(start)))
+	})
+	return ret
+}
+
 // Close closes the sink.
 func (d *DDLSink) Close() {
 	if d.statistics != nil {

cdc/sink/ddlsink/cloudstorage/cloud_storage_ddl_sink_test.go

@@ -19,24 +19,30 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/types"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/stretchr/testify/require"
 )
 
 func TestWriteDDLEvent(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	parentDir := t.TempDir()
-	uri := fmt.Sprintf("file:///%s", parentDir)
+	uri := fmt.Sprintf("file:///%s?protocol=csv", parentDir)
 	sinkURI, err := url.Parse(uri)
 	require.Nil(t, err)
-	sink, err := NewDDLSink(ctx, sinkURI)
+	replicaConfig := config.GetDefaultReplicaConfig()
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.Nil(t, err)
+	sink, err := NewDDLSink(ctx, sinkURI, replicaConfig)
 	require.Nil(t, err)
 
 	ddlEvent := &model.DDLEvent{
@@ -97,10 +103,13 @@ func TestWriteCheckpointTs(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	parentDir := t.TempDir()
-	uri := fmt.Sprintf("file:///%s", parentDir)
+	uri := fmt.Sprintf("file:///%s?protocol=csv", parentDir)
 	sinkURI, err := url.Parse(uri)
 	require.Nil(t, err)
-	sink, err := NewDDLSink(ctx, sinkURI)
+	replicaConfig := config.GetDefaultReplicaConfig()
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.Nil(t, err)
+	sink, err := NewDDLSink(ctx, sinkURI, replicaConfig)
 	require.Nil(t, err)
 	tables := []*model.TableInfo{
 		{
@@ -132,3 +141,34 @@ func TestWriteCheckpointTs(t *testing.T) {
 	require.Nil(t, err)
 	require.JSONEq(t, `{"checkpoint-ts":100}`, string(metadata))
 }
+
+func TestCleanupExpiredFiles(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	parentDir := t.TempDir()
+	uri := fmt.Sprintf("file:///%s?protocol=csv", parentDir)
+	sinkURI, err := url.Parse(uri)
+	require.Nil(t, err)
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Sink.CloudStorageConfig = &config.CloudStorageConfig{
+		FileExpirationDays:  util.AddressOf(1),
+		FileCleanupCronSpec: util.AddressOf("* * * * * *"),
+	}
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.Nil(t, err)
+
+	cnt := atomic.Int64{}
+	cleanupJobs := []func(){
+		func() {
+			cnt.Add(1)
+		},
+	}
+	sink, err := newDDLSink(ctx, model.DefaultChangeFeedID("test"), sinkURI, replicaConfig, cleanupJobs)
+	require.Nil(t, err)
+
+	_ = sink
+	time.Sleep(3 * time.Second)
+	require.LessOrEqual(t, int64(1), cnt.Load())
+}

cdc/sink/ddlsink/factory/factory.go

@@ -55,7 +55,7 @@ func New(
 	case sink.MySQLSSLScheme, sink.MySQLScheme, sink.TiDBScheme, sink.TiDBSSLScheme:
 		return mysql.NewDDLSink(ctx, sinkURI, cfg)
 	case sink.S3Scheme, sink.FileScheme, sink.GCSScheme, sink.GSScheme, sink.AzblobScheme, sink.AzureScheme, sink.CloudStorageNoopScheme:
-		return cloudstorage.NewDDLSink(ctx, sinkURI)
+		return cloudstorage.NewDDLSink(ctx, sinkURI, cfg)
 	default:
 		return nil,
 			cerror.ErrSinkURIInvalid.GenWithStack("the sink scheme (%s) is not supported", scheme)

go.mod

@@ -71,6 +71,7 @@ require (
 	github.com/prometheus/client_model v0.3.0
 	github.com/r3labs/diff v1.1.0
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
+	github.com/robfig/cron v1.2.0
 	github.com/segmentio/kafka-go v0.4.39-0.20230217181906-f6986fb02ee7
 	github.com/shirou/gopsutil/v3 v3.23.3
 	github.com/shopspring/decimal v1.3.0

go.sum

@@ -983,6 +983,8 @@ github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJ
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
 github.com/rivo/uniseg v0.4.4/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ=
+github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=