Skip to content

cloudstorage: make all DateSeparator fields use the enum instead of string (#6078) #6149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ti-chi-bot wants to merge 1 commit into
base: release-nextgen-202603
Choose a base branch
from
Open

cloudstorage: make all DateSeparator fields use the enum instead of string (#6078) #6149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions api/v2/model.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1133,6 +1133,7 @@ type Table struct {
// SinkConfig represents sink config for a changefeed
// This is a duplicate of config.SinkConfig
type SinkConfig struct {
<<<<<<< HEAD
Protocol *string `json:"protocol,omitempty"`
SchemaRegistry *string `json:"schema_registry,omitempty"`
CSVConfig *CSVConfig `json:"csv,omitempty"`
Expand All @@ -1144,6 +1145,19 @@ type SinkConfig struct {
DateSeparator *string `json:"date_separator,omitempty"`
EnablePartitionSeparator *bool `json:"enable_partition_separator,omitempty"`
FileIndexWidth *int `json:"file_index_width,omitempty"`
=======
Protocol *string `json:"protocol,omitempty" toml:"protocol,omitempty"`
SchemaRegistry *string `json:"schema_registry,omitempty" toml:"schema-registry,omitempty"`
CSVConfig *CSVConfig `json:"csv,omitempty" toml:"csv,omitempty"`
DispatchRules []*DispatchRule `json:"dispatchers,omitempty" toml:"dispatchers,omitempty"`
ColumnSelectors []*ColumnSelector `json:"column_selectors,omitempty" toml:"column-selectors,omitempty"`
TxnAtomicity *string `json:"transaction_atomicity,omitempty" toml:"transaction-atomicity,omitempty"`
EncoderConcurrency *int `json:"encoder_concurrency,omitempty" toml:"encoder-concurrency,omitempty"`
Terminator *string `json:"terminator,omitempty" toml:"terminator,omitempty"`
DateSeparator *config.DateSeparator `json:"date_separator,omitempty" toml:"date-separator,omitempty"`
EnablePartitionSeparator *bool `json:"enable_partition_separator,omitempty" toml:"enable-partition-separator,omitempty"`
FileIndexWidth *int `json:"file_index_width,omitempty" toml:"file-index-digit,omitempty"`
>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078))
// deprecated: it's become useless since v9.0.0
EnableKafkaSinkV2 *bool `json:"enable_kafka_sink_v2,omitempty"`
OnlyOutputUpdatedColumns *bool `json:"only_output_updated_columns,omitempty"`
Expand Down
63 changes: 63 additions & 0 deletions api/v2/model_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,76 @@
package v2

import (
"encoding/json"
"testing"

"github.com/pingcap/ticdc/pkg/config"
"github.com/pingcap/ticdc/pkg/util"
"github.com/stretchr/testify/require"
)

<<<<<<< HEAD
=======
func TestSinkConfigDateSeparator(t *testing.T) {
t.Parallel()

var sinkConfig SinkConfig
require.NoError(t, json.Unmarshal([]byte(`{"date_separator":"DAY"}`), &sinkConfig))
require.Equal(t, config.DateSeparatorDay, util.GetOrZero(sinkConfig.DateSeparator))

err := json.Unmarshal([]byte(`{"date_separator":"week"}`), &SinkConfig{})
require.Error(t, err)
require.ErrorContains(t, err, "CDC:ErrStorageSinkInvalidConfig")
}

func TestChangeFeedInfoCloneWithMaskedSensitiveData(t *testing.T) {
info := &ChangeFeedInfo{
ID: "test",
SinkURI: "kafka://user:sink-password-sentinel@127.0.0.1:9092/topic?secret=uri-secret-sentinel",
Config: &ReplicaConfig{
Sink: &SinkConfig{
SchemaRegistry: util.AddressOf("https://registry.example.com?access-key=registry-secret-sentinel"),
KafkaConfig: &KafkaConfig{
KafkaClientID: util.AddressOf("visible-client-id"),
SASLPassword: util.AddressOf("plain-password-sentinel"),
SASLGssAPIPassword: util.AddressOf("gssapi-password-sentinel"),
SASLOAuthClientSecret: util.AddressOf("oauth-secret-sentinel"),
SASLOAuthTokenURL: util.AddressOf("https://oauth.example.com/token?client_secret=token-url-secret-sentinel"),
LargeMessageHandle: &LargeMessageHandleConfig{ClaimCheckStorageURI: "s3://bucket/prefix?access-key=claim-check-secret-sentinel"},
GlueSchemaRegistryConfig: &GlueSchemaRegistryConfig{
AccessKey: "glue-access-sentinel",
SecretAccessKey: "glue-secret-sentinel",
Token: "glue-token-sentinel",
},
},
PulsarConfig: &PulsarConfig{
AuthenticationToken: util.AddressOf("pulsar-token-sentinel"),
BasicPassword: util.AddressOf("pulsar-password-sentinel"),
OAuth2: &PulsarOAuth2{OAuth2PrivateKey: "pulsar-private-key-sentinel"},
},
},
Consistent: &ConsistentConfig{Storage: util.AddressOf("s3://bucket/prefix?access-key=consistent-secret-sentinel")},
},
}
original, err := info.Marshal()
require.NoError(t, err)

masked, err := info.CloneWithMaskedSensitiveData()
require.NoError(t, err)
output, err := masked.Marshal()
require.NoError(t, err)
require.NotContains(t, output, "sentinel")
require.NotContains(t, output, "memory_quota")
require.Contains(t, output, "visible-client-id")
require.Nil(t, masked.Config.Sink.KafkaConfig.Key)
after, err := info.Marshal()
require.NoError(t, err)
require.Equal(t, original, after)
}

// TestReplicaConfigConversion verifies API/internal replica config conversion,
// including round-tripping the optional event collector batch overrides.
>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078))
func TestReplicaConfigConversion(t *testing.T) {
t.Parallel()

Expand Down
15 changes: 15 additions & 0 deletions cmd/storage-consumer/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ type indexRange struct {

type consumer struct {
replicationCfg *config.ReplicaConfig
dateSeparator config.DateSeparator
codecCfg *common.Config
columnSelectors *columnselector.ColumnSelectors
externalStorage storeapi.Storage
Expand Down Expand Up @@ -103,6 +104,7 @@ func newConsumer(ctx context.Context) (*consumer, error) {
log.Error("failed to validate replica config", zap.Error(err))
return nil, err
}
dateSeparator := putil.GetOrZero(replicaConfig.Sink.DateSeparator)

switch putil.GetOrZero(replicaConfig.Sink.Protocol) {
case config.ProtocolCsv.String():
Expand Down Expand Up @@ -157,6 +159,7 @@ func newConsumer(ctx context.Context) (*consumer, error) {

return &consumer{
replicationCfg: replicaConfig,
dateSeparator: dateSeparator,
codecCfg: codecConfig,
columnSelectors: columnSelectors,
externalStorage: storage,
Expand Down Expand Up @@ -223,12 +226,24 @@ func (c *consumer) getNewFiles(
origDMLIdxMap[k] = m
}

<<<<<<< HEAD
err := c.externalStorage.WalkDir(ctx, opt, func(path string, size int64) error {
if cloudstorage.IsSchemaFile(path) {
err := c.parseSchemaFilePath(ctx, path)
if err != nil {
log.Error("failed to parse schema file path", zap.Error(err))
// skip handling this file
=======
err := c.externalStorage.WalkDir(ctx, opt, func(path string, _ int64) error {
if cloudstorage.IsSchemaFile(path) {
c.parseSchemaFilePath(ctx, path)
return nil
}
if strings.HasSuffix(path, ".index") {
var dmlkey cloudstorage.DMLPathKey
if err := dmlkey.ParseIndexFilePath(c.dateSeparator, path); err != nil {
log.Debug("ignore handling unsupported dml index file", zap.String("path", path))
>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078))
return nil
}
} else if strings.HasSuffix(path, ".index") {
Expand Down
4 changes: 2 additions & 2 deletions downstreamadapter/sink/cloudstorage/dml_writers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ func TestCloudStorageWriteEventsWithoutDateSeparator(t *testing.T) {
err = replicaConfig.ValidateAndAdjust(sinkURI)
require.NoError(t, err)

replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorNone)
replicaConfig.Sink.FileIndexWidth = putil.AddressOf(6)

ctx, cancel := context.WithCancel(context.Background())
Expand Down Expand Up @@ -279,7 +279,7 @@ func TestCloudStorageWriteEventsWithDateSeparator(t *testing.T) {
err = replicaConfig.ValidateAndAdjust(sinkURI)
require.NoError(t, err)

replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorDay.String())
replicaConfig.Sink.DateSeparator = putil.AddressOf(config.DateSeparatorDay)
replicaConfig.Sink.FileIndexWidth = putil.AddressOf(6)

mockClock := pclock.NewMock()
Expand Down
6 changes: 3 additions & 3 deletions downstreamadapter/sink/cloudstorage/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,11 +405,11 @@ func (s *sink) initCron(
}

func (s *sink) bgCleanup(ctx context.Context) {
if s.cfg.DateSeparator != config.DateSeparatorDay.String() || s.cfg.FileExpirationDays <= 0 {
if s.cfg.DateSeparator != config.DateSeparatorDay || s.cfg.FileExpirationDays <= 0 {
log.Info("skip cleanup expired files for storage sink",
zap.String("keyspace", s.changefeedID.Keyspace()),
zap.String("changefeedID", s.changefeedID.Name()),
zap.String("dateSeparator", s.cfg.DateSeparator),
zap.Stringer("dateSeparator", s.cfg.DateSeparator),
zap.Int("expiredFileTTL", s.cfg.FileExpirationDays))
return
}
Expand All @@ -419,7 +419,7 @@ func (s *sink) bgCleanup(ctx context.Context) {
log.Info("start schedule cleanup expired files for storage sink",
zap.String("keyspace", s.changefeedID.Keyspace()),
zap.String("changefeedID", s.changefeedID.Name()),
zap.String("dateSeparator", s.cfg.DateSeparator),
zap.Stringer("dateSeparator", s.cfg.DateSeparator),
zap.Int("expiredFileTTL", s.cfg.FileExpirationDays))

// wait for the context done
Expand Down
14 changes: 13 additions & 1 deletion downstreamadapter/sink/cloudstorage/sink_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func TestCloudStorageSinkWithColumnSelector(t *testing.T) {
}
err = replicaConfig.ValidateAndAdjust(sinkURI)
require.NoError(t, err)
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone)

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Expand Down Expand Up @@ -646,8 +646,20 @@ func TestCleanupExpiredFiles(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

<<<<<<< HEAD
mockPDClock := pdutil.NewClock4Test()
appcontext.SetService(appcontext.DefaultPDClock, mockPDClock)
=======
cloudStorageSink := &sink{
changefeedID: common.NewChangefeedID4Test("test", "test"),
cfg: &cloudstorage.Config{
DateSeparator: config.DateSeparatorDay,
FileExpirationDays: 1,
FileCleanupCronSpec: util.GetOrZero(replicaConfig.Sink.CloudStorageConfig.FileCleanupCronSpec),
},
}
require.NoError(t, cloudStorageSink.initCron(ctx, sinkURI, cleanupJobs))
>>>>>>> 0697ba0ec (cloudstorage: make all DateSeparator fields use the enum instead of string (#6078))

cloudStorageSink, err := newSinkForTest(ctx, replicaConfig, sinkURI, cleanupJobs)
go cloudStorageSink.Run(ctx)
Expand Down
8 changes: 4 additions & 4 deletions downstreamadapter/sink/cloudstorage/writer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ func testWriter(ctx context.Context, t *testing.T, dir string) *writer {
require.NoError(t, err)
cfg := cloudstorage.NewConfig()
replicaConfig := config.GetDefaultReplicaConfig()
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone)
err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true)
cfg.FileIndexWidth = 6
require.NoError(t, err)
Expand Down Expand Up @@ -470,7 +470,7 @@ func TestWriterStoresPendingMessagesInSpoolBeforeFlush(t *testing.T) {

cfg := cloudstorage.NewConfig()
replicaConfig := config.GetDefaultReplicaConfig()
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone)
replicaConfig.Sink.CloudStorageConfig = &config.CloudStorageConfig{
// Keep the quota larger than this encoded batch so the controller still
// spills it to local spool files instead of taking the oversized in-memory fast path.
Expand Down Expand Up @@ -645,7 +645,7 @@ func TestWriterIndexWriteError(t *testing.T) {
require.NoError(t, err)
cfg := cloudstorage.NewConfig()
replicaConfig := config.GetDefaultReplicaConfig()
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone)
err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true)
require.NoError(t, err)
cfg.FileIndexWidth = 6
Expand Down Expand Up @@ -711,7 +711,7 @@ func TestWriterDataFileCloseError(t *testing.T) {
require.NoError(t, err)
cfg := cloudstorage.NewConfig()
replicaConfig := config.GetDefaultReplicaConfig()
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone)
err = cfg.Apply(context.TODO(), sinkURI, replicaConfig.Sink, true)
require.NoError(t, err)
cfg.FileIndexWidth = 6
Expand Down
Loading
Loading