This is an automated cherry-pick of #10097

Signed-off-by: ti-chi-bot <[email protected]>

cdc/api/v2/model.go

@@ -377,9 +377,18 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		var cloudStorageConfig *config.CloudStorageConfig
 		if c.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &config.CloudStorageConfig{
+<<<<<<< HEAD
 				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,
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 			}
 		}
 
@@ -596,9 +605,18 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		var cloudStorageConfig *CloudStorageConfig
 		if cloned.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &CloudStorageConfig{
+<<<<<<< HEAD
 				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,
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 			}
 		}
 
@@ -1023,9 +1041,18 @@ type MySQLConfig struct {
 
 // CloudStorageConfig represents a cloud storage sink configuration
 type CloudStorageConfig struct {
+<<<<<<< HEAD
 	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"`
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 }
 
 // ChangefeedStatus holds common information of a changefeed in cdc

cdc/sink/ddlsink/cloudstorage/cloud_storage_ddl_sink.go

@@ -17,19 +17,25 @@ 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"
+<<<<<<< HEAD
+=======
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	"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,12 +49,42 @@ type DDLSink struct {
 	// statistic is used to record the DDL metrics
 	statistics *metrics.Statistics
 	storage    storage.ExternalStorage
+	cfg        *cloudstorage.Config
+	cron       *cron.Cron
 
+<<<<<<< HEAD
+=======
+	lastCheckpointTs         atomic.Uint64
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	lastSendCheckpointTsTime time.Time
 }
 
 // NewDDLSink creates a ddl sink for cloud storage.
+<<<<<<< HEAD
 func NewDDLSink(ctx context.Context, sinkURI *url.URL) (*DDLSink, error) {
+=======
+func NewDDLSink(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+	sinkURI *url.URL,
+	replicaConfig *config.ReplicaConfig,
+) (*DDLSink, error) {
+	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)
+	}
+
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	storage, err := util.GetExternalStorageFromURI(ctx, sinkURI.String())
 	if err != nil {
 		return nil, err
@@ -58,10 +94,25 @@ func NewDDLSink(ctx context.Context, sinkURI *url.URL) (*DDLSink, error) {
 	d := &DDLSink{
 		id:                       changefeedID,
 		storage:                  storage,
+<<<<<<< HEAD
 		statistics:               metrics.NewStatistics(ctx, sink.TxnSink),
 		lastSendCheckpointTsTime: time.Now(),
 	}
 
+=======
+		statistics:               metrics.NewStatistics(ctx, changefeedID, 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)
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	return d, nil
 }
 
@@ -90,15 +141,23 @@ func (d *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error
 	}
 
 	var def cloudstorage.TableDefinition
+<<<<<<< HEAD
 	def.FromDDLEvent(ddl)
+=======
+	def.FromDDLEvent(ddl, d.cfg.OutputColumnID)
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	if err := writeFile(def); err != nil {
 		return errors.Trace(err)
 	}
 
 	if ddl.Type == timodel.ActionExchangeTablePartition {
 		// For exchange partition, we need to write the schema of the source table.
 		var sourceTableDef cloudstorage.TableDefinition
+<<<<<<< HEAD
 		sourceTableDef.FromTableInfo(ddl.PreTableInfo, ddl.TableInfo.Version)
+=======
+		sourceTableDef.FromTableInfo(ddl.PreTableInfo, ddl.TableInfo.Version, d.cfg.OutputColumnID)
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 		return writeFile(sourceTableDef)
 	}
 	return nil
@@ -117,6 +176,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 +186,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,34 @@ 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)
+<<<<<<< HEAD
 	sink, err := NewDDLSink(ctx, sinkURI)
+=======
+	replicaConfig := config.GetDefaultReplicaConfig()
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.Nil(t, err)
+	sink, err := NewDDLSink(ctx, model.DefaultChangeFeedID("test"), sinkURI, replicaConfig)
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	require.Nil(t, err)
 
 	ddlEvent := &model.DDLEvent{
@@ -97,10 +107,17 @@ 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)
+<<<<<<< HEAD
 	sink, err := NewDDLSink(ctx, sinkURI)
+=======
+	replicaConfig := config.GetDefaultReplicaConfig()
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.Nil(t, err)
+	sink, err := NewDDLSink(ctx, model.DefaultChangeFeedID("test"), sinkURI, replicaConfig)
+>>>>>>> dd3f8dfc25 (sink(ticdc): cleanup expired files by day for storage sink (#10097))
 	require.Nil(t, err)
 	tables := []*model.TableInfo{
 		{
@@ -132,3 +149,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())
+}