DNM: used for build image

dm/_utils/terror_gen/errors_release.txt

@@ -152,7 +152,7 @@ ErrConfigLoaderCfgConflict,[code=20016:class=config:scope=internal:level=medium]
 ErrConfigSyncerCfgConflict,[code=20017:class=config:scope=internal:level=medium], "Message: syncer-config-name and syncer should only specify one, Workaround: Please check the `syncer-config-name` and `syncer` config in task configuration file."
 ErrConfigReadCfgFromFile,[code=20018:class=config:scope=internal:level=medium], "Message: read config file %v"
 ErrConfigNeedUniqueTaskName,[code=20019:class=config:scope=internal:level=medium], "Message: must specify a unique task name, Workaround: Please check the `name` config in task configuration file."
-ErrConfigInvalidTaskMode,[code=20020:class=config:scope=internal:level=medium], "Message: please specify right task-mode, support `full`, `incremental`, `all`, Workaround: Please check the `task-mode` config in task configuration file."
+ErrConfigInvalidTaskMode,[code=20020:class=config:scope=internal:level=medium], "Message: please specify right task-mode, support `full`, `incremental`, `all`, `dump`, `load`, Workaround: Please check the `task-mode` config in task configuration file."
 ErrConfigNeedTargetDB,[code=20021:class=config:scope=internal:level=medium], "Message: must specify target-database, Workaround: Please check the `target-database` config in task configuration file."
 ErrConfigMetadataNotSet,[code=20022:class=config:scope=internal:level=medium], "Message: mysql-instance(%s) must set meta for task-mode %s, Workaround: Please check the `meta` config in task configuration file."
 ErrConfigRouteRuleNotFound,[code=20023:class=config:scope=internal:level=medium], "Message: mysql-instance(%d)'s route-rules %s not exist in routes, Workaround: Please check the `route-rules` config in task configuration file."

dm/config/helper.go

@@ -26,7 +26,7 @@ func HasDump(taskMode string) bool {
 // HasLoad returns true if taskMode contains load unit.
 func HasLoad(taskMode string) bool {
 	switch taskMode {
-	case ModeAll, ModeFull, ModeLoadSync:
+	case ModeAll, ModeFull, ModeLoad, ModeLoadSync:
 		return true
 	default:
 		return false

dm/config/helper_test.go

@@ -36,6 +36,10 @@ func TestTaskModeHasFunction(t *testing.T) {
 	require.False(t, HasLoad(ModeDump))
 	require.False(t, HasSync(ModeDump))
 
+	require.False(t, HasDump(ModeLoad))
+	require.True(t, HasLoad(ModeLoad))
+	require.False(t, HasSync(ModeLoad))
+
 	require.False(t, HasDump(ModeLoadSync))
 	require.True(t, HasLoad(ModeLoadSync))
 	require.True(t, HasSync(ModeLoadSync))

dm/config/subtask.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/BurntSushi/toml"
+	"github.com/google/uuid"
 	extstorage "github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/util/dbutil"
 	"github.com/pingcap/tidb/pkg/util/filter"
@@ -52,6 +53,7 @@ const (
 	ModeFull      = "full"
 	ModeIncrement = "incremental"
 	ModeDump      = "dump"
+	ModeLoad      = "load"
 	ModeLoadSync  = "load&sync"
 
 	DefaultShadowTableRules = "^_(.+)_(?:new|gho)$"
@@ -178,17 +180,19 @@ type SubTaskConfig struct {
 	ExtStorage      extstorage.ExternalStorage `toml:"-" json:"-"`
 	MetricsFactory  promutil.Factory           `toml:"-" json:"-"`
 	FrameworkLogger *zap.Logger                `toml:"-" json:"-"`
-	// members below are injected by dataflow engine, UUID should be unique in
-	// one go runtime.
+	// members below are injected by dataflow engine
+	// UUID should be unique in one go runtime.
 	// IOTotalBytes is used build TCPConnWithIOCounter and UUID is used to as a
 	// key to let MySQL driver to find the right TCPConnWithIOCounter.
-	UUID         string         `toml:"-" json:"-"`
-	IOTotalBytes *atomic.Uint64 `toml:"-" json:"-"`
+	// It will meter TCP io usage to downstream of the subtask
+	UUID         string         `toml:"uuid" json:"-"`
+	IOTotalBytes *atomic.Uint64 `toml:"io-total-bytes" json:"io-total-bytes"`
 
-	// meter network usage from upstream
+	// DumpUUID as same as UUID
+	// DumpIOTotalBytes meter TCP io usage from upstream of the subtask, other same as IOTotalBytes
 	// e.g., pulling binlog
-	DumpUUID         string         `toml:"-" json:"-"`
-	DumpIOTotalBytes *atomic.Uint64 `toml:"-" json:"-"`
+	DumpUUID         string         `toml:"dump-uuid" json:"-"`
+	DumpIOTotalBytes *atomic.Uint64 `toml:"dump-io-total-bytes" json:"dump-io-total-bytes"`
 }
 
 // SampleSubtaskConfig is the content of subtask.toml in current folder.
@@ -212,6 +216,14 @@ func (c *SubTaskConfig) SetFlagSet(flagSet *flag.FlagSet) {
 	c.flagSet = flagSet
 }
 
+// InitIOCounters init io counter and uuid for syncer.
+func (c *SubTaskConfig) InitIOCounters() {
+	c.IOTotalBytes = atomic.NewUint64(0)
+	c.DumpIOTotalBytes = atomic.NewUint64(0)
+	c.UUID = uuid.NewString()
+	c.DumpUUID = uuid.NewString()
+}
+
 // String returns the config's json string.
 func (c *SubTaskConfig) String() string {
 	cfg, err := json.Marshal(c)
@@ -222,6 +234,10 @@ func (c *SubTaskConfig) String() string {
 }
 
 // Toml returns TOML format representation of config.
+// Note: The atomic.Uint64 fields (IOTotalBytes and DumpIOTotalBytes) are not
+// encoded in the TOML output because they do not implement the necessary
+// marshaling interfaces. As a result, these fields will not be included in
+// the TOML representation.
 func (c *SubTaskConfig) Toml() (string, error) {
 	var b bytes.Buffer
 	enc := toml.NewEncoder(&b)
@@ -242,6 +258,9 @@ func (c *SubTaskConfig) DecodeFile(fpath string, verifyDecryptPassword bool) err
 }
 
 // Decode loads config from file data.
+// Note: The atomic.Uint64 fields (IOTotalBytes and DumpIOTotalBytes) will not
+// be populated from the TOML data since they cannot be decoded by toml.Decode().
+// As a result, these fields will remain uninitialized (zero value) after decoding.
 func (c *SubTaskConfig) Decode(data string, verifyDecryptPassword bool) error {
 	if _, err := toml.Decode(data, c); err != nil {
 		return terror.ErrConfigTomlTransform.Delegate(err, "decode subtask config from data")
@@ -330,6 +349,7 @@ func (c *SubTaskConfig) Adjust(verifyDecryptPassword bool) error {
 	}
 
 	// adjust dir, no need to do for load&sync mode because it needs its own s3 repository
+	// still use dir for standalone load mode (different from the behavior of load&sync mode)
 	if HasLoad(c.Mode) && c.Mode != ModeLoadSync {
 		// check
 		isS3 := storage.IsS3Path(c.LoaderConfig.Dir)
@@ -495,6 +515,12 @@ func (c *SubTaskConfig) Clone() (*SubTaskConfig, error) {
 	if err != nil {
 		return nil, terror.ErrConfigTomlTransform.Delegate(err, "decode subtask config from data")
 	}
-
+	// Manually copy atomic values for atomic.Uint64 doesn't implement TOML marshaling interfaces
+	if c.IOTotalBytes != nil {
+		clone.IOTotalBytes = atomic.NewUint64(c.IOTotalBytes.Load())
+	}
+	if c.DumpIOTotalBytes != nil {
+		clone.DumpIOTotalBytes = atomic.NewUint64(c.DumpIOTotalBytes.Load())
+	}
 	return clone, nil
 }

dm/config/subtask_test.go

@@ -16,7 +16,9 @@ package config
 import (
 	"context"
 	"crypto/rand"
+	"encoding/json"
 	"reflect"
+	"sync"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -27,6 +29,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
 )
 
 func TestSubTask(t *testing.T) {
@@ -345,3 +348,74 @@ func TestFetchTZSetting(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "+01:00", tz)
 }
+
+func TestSubTaskConfigMarshalAtomic(t *testing.T) {
+	var (
+		uuid     = "test-uuid"
+		dumpUUID = "test-dump-uuid"
+	)
+	cfg := &SubTaskConfig{
+		Name:             "test",
+		SourceID:         "source-1",
+		UUID:             uuid,
+		DumpUUID:         dumpUUID,
+		IOTotalBytes:     atomic.NewUint64(100),
+		DumpIOTotalBytes: atomic.NewUint64(200),
+	}
+	require.Equal(t, cfg.IOTotalBytes.Load(), uint64(100))
+	require.Equal(t, cfg.DumpIOTotalBytes.Load(), uint64(200))
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			data, err := json.Marshal(cfg)
+			require.NoError(t, err)
+			jsonMap := make(map[string]interface{})
+			err = json.Unmarshal(data, &jsonMap)
+			require.NoError(t, err)
+
+			// Check atomic values exist and are numbers
+			ioBytes, ok := jsonMap["io-total-bytes"].(float64)
+			require.True(t, ok, "io-total-bytes should be a number")
+			require.GreaterOrEqual(t, ioBytes, float64(100))
+
+			dumpBytes, ok := jsonMap["dump-io-total-bytes"].(float64)
+			require.True(t, ok, "dump-io-total-bytes should be a number")
+			require.GreaterOrEqual(t, dumpBytes, float64(200))
+
+			// UUID fields should not be present in JSON
+			_, hasUUID := jsonMap["uuid"]
+			_, hasDumpUUID := jsonMap["dump-uuid"]
+			require.False(t, hasUUID, "UUID should not be in JSON")
+			require.False(t, hasDumpUUID, "DumpUUID should not be in JSON")
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			newCfg, err := cfg.Clone()
+			require.NoError(t, err)
+
+			// Check atomic values exist and are numbers
+			require.GreaterOrEqual(t, newCfg.IOTotalBytes.Load(), uint64(100))
+			require.GreaterOrEqual(t, newCfg.DumpIOTotalBytes.Load(), uint64(200))
+			require.Equal(t, newCfg.UUID, uuid)
+			require.Equal(t, newCfg.DumpUUID, dumpUUID)
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			cfg.IOTotalBytes.Add(1)
+			cfg.DumpIOTotalBytes.Add(1)
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, cfg.IOTotalBytes.Load(), uint64(110))
+	require.Equal(t, cfg.DumpIOTotalBytes.Load(), uint64(210))
+}