downgrade: add import/export-configs command (#1877)

dm/config/source_config.go

@@ -45,7 +45,7 @@ type PurgeConfig struct {
 	RemainSpace int64 `yaml:"remain-space" toml:"remain-space" json:"remain-space"` // if remain space in @RelayBaseDir less than @RemainSpace (GB), then it can be purged
 }
 
-// SourceConfig is the configuration for Worker.
+// SourceConfig is the configuration for source.
 type SourceConfig struct {
 	EnableGTID  bool   `yaml:"enable-gtid" toml:"enable-gtid" json:"enable-gtid"`
 	AutoFixGTID bool   `yaml:"auto-fix-gtid" toml:"auto-fix-gtid" json:"auto-fix-gtid"`
@@ -353,3 +353,86 @@ func (c *SourceConfig) check(metaData *toml.MetaData, err error) error {
 	c.adjust()
 	return nil
 }
+
+// YamlForDowngrade returns YAML format represents of config for downgrade.
+func (c *SourceConfig) YamlForDowngrade() (string, error) {
+	s := NewSourceConfigForDowngrade(c)
+
+	// encrypt password
+	cipher, err := utils.Encrypt(utils.DecryptOrPlaintext(c.From.Password))
+	if err != nil {
+		return "", err
+	}
+	s.From.Password = cipher
+
+	// omit default values, so we can ignore them for later marshal
+	s.omitDefaultVals()
+
+	return s.Yaml()
+}
+
+// SourceConfigForDowngrade is the base configuration for source in v2.0.
+// This config is used for downgrade(config export) from a higher dmctl version.
+// When we add any new config item into SourceConfig, we should update it also.
+type SourceConfigForDowngrade struct {
+	EnableGTID      bool                   `yaml:"enable-gtid"`
+	AutoFixGTID     bool                   `yaml:"auto-fix-gtid"`
+	RelayDir        string                 `yaml:"relay-dir"`
+	MetaDir         string                 `yaml:"meta-dir"`
+	Flavor          string                 `yaml:"flavor"`
+	Charset         string                 `yaml:"charset"`
+	EnableRelay     bool                   `yaml:"enable-relay"`
+	RelayBinLogName string                 `yaml:"relay-binlog-name"`
+	RelayBinlogGTID string                 `yaml:"relay-binlog-gtid"`
+	UUIDSuffix      int                    `yaml:"-"`
+	SourceID        string                 `yaml:"source-id"`
+	From            DBConfig               `yaml:"from"`
+	Purge           PurgeConfig            `yaml:"purge"`
+	Checker         CheckerConfig          `yaml:"checker"`
+	ServerID        uint32                 `yaml:"server-id"`
+	Tracer          map[string]interface{} `yaml:"tracer"`
+	// any new config item, we mark it omitempty
+	CaseSensitive bool                  `yaml:"case-sensitive,omitempty"`
+	Filters       []*bf.BinlogEventRule `yaml:"filters,omitempty"`
+}
+
+// NewSourceConfigForDowngrade creates a new base config for downgrade.
+func NewSourceConfigForDowngrade(sourceCfg *SourceConfig) *SourceConfigForDowngrade {
+	return &SourceConfigForDowngrade{
+		EnableGTID:      sourceCfg.EnableGTID,
+		AutoFixGTID:     sourceCfg.AutoFixGTID,
+		RelayDir:        sourceCfg.RelayDir,
+		MetaDir:         sourceCfg.MetaDir,
+		Flavor:          sourceCfg.Flavor,
+		Charset:         sourceCfg.Charset,
+		EnableRelay:     sourceCfg.EnableRelay,
+		RelayBinLogName: sourceCfg.RelayBinLogName,
+		RelayBinlogGTID: sourceCfg.RelayBinlogGTID,
+		UUIDSuffix:      sourceCfg.UUIDSuffix,
+		SourceID:        sourceCfg.SourceID,
+		From:            sourceCfg.From,
+		Purge:           sourceCfg.Purge,
+		Checker:         sourceCfg.Checker,
+		ServerID:        sourceCfg.ServerID,
+		Tracer:          sourceCfg.Tracer,
+		CaseSensitive:   sourceCfg.CaseSensitive,
+		Filters:         sourceCfg.Filters,
+	}
+}
+
+// omitDefaultVals change default value to empty value for new config item.
+// If any default value for new config item is not empty(0 or false or nil),
+// we should change it to empty.
+func (c *SourceConfigForDowngrade) omitDefaultVals() {
+	if len(c.From.Session) > 0 {
+		if timeZone, ok := c.From.Session["time_zone"]; ok && timeZone == defaultTimeZone {
+			delete(c.From.Session, "time_zone")
+		}
+	}
+}
+
+// Yaml returns YAML format representation of the config.
+func (c *SourceConfigForDowngrade) Yaml() (string, error) {
+	b, err := yaml.Marshal(c)
+	return string(b), err
+}

dm/config/source_config_test.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"path"
+	"reflect"
 	"strings"
 	"time"
 
@@ -235,6 +236,22 @@ func (t *testConfig) TestConfigVerify(c *C) {
 	}
 }
 
+func (t *testConfig) TestSourceConfigForDowngrade(c *C) {
+	cfg, err := LoadFromFile(sourceSampleFile)
+	c.Assert(err, IsNil)
+
+	// make sure all new field were added
+	cfgForDowngrade := NewSourceConfigForDowngrade(cfg)
+	cfgReflect := reflect.Indirect(reflect.ValueOf(cfg))
+	cfgForDowngradeReflect := reflect.Indirect(reflect.ValueOf(cfgForDowngrade))
+	c.Assert(cfgReflect.NumField(), Equals, cfgForDowngradeReflect.NumField())
+
+	// make sure all field were copied
+	cfgForClone := &SourceConfigForDowngrade{}
+	Clone(cfgForClone, cfg)
+	c.Assert(cfgForDowngrade, DeepEquals, cfgForClone)
+}
+
 func subtestFlavor(c *C, cfg *SourceConfig, sqlInfo, expectedFlavor, expectedError string) {
 	cfg.Flavor = ""
 	db, mock, err := sqlmock.New()

dm/config/task.go

@@ -33,6 +33,7 @@ import (
 
 	"github.com/pingcap/dm/pkg/log"
 	"github.com/pingcap/dm/pkg/terror"
+	"github.com/pingcap/dm/pkg/utils"
 )
 
 // Online DDL Scheme.
@@ -935,3 +936,154 @@ func checkValidExpr(expr string) error {
 	_, _, err := defaultParser.Parse(expr, "", "")
 	return err
 }
+
+// YamlForDowngrade returns YAML format represents of config for downgrade.
+func (c *TaskConfig) YamlForDowngrade() (string, error) {
+	t := NewTaskConfigForDowngrade(c)
+
+	// encrypt password
+	cipher, err := utils.Encrypt(utils.DecryptOrPlaintext(t.TargetDB.Password))
+	if err != nil {
+		return "", err
+	}
+	t.TargetDB.Password = cipher
+
+	// omit default values, so we can ignore them for later marshal
+	t.omitDefaultVals()
+
+	return t.Yaml()
+}
+
+// MySQLInstanceForDowngrade represents a sync config of a MySQL instance for downgrade.
+type MySQLInstanceForDowngrade struct {
+	SourceID           string          `yaml:"source-id"`
+	Meta               *Meta           `yaml:"meta"`
+	FilterRules        []string        `yaml:"filter-rules"`
+	ColumnMappingRules []string        `yaml:"column-mapping-rules"`
+	RouteRules         []string        `yaml:"route-rules"`
+	BWListName         string          `yaml:"black-white-list"`
+	BAListName         string          `yaml:"block-allow-list"`
+	MydumperConfigName string          `yaml:"mydumper-config-name"`
+	Mydumper           *MydumperConfig `yaml:"mydumper"`
+	MydumperThread     int             `yaml:"mydumper-thread"`
+	LoaderConfigName   string          `yaml:"loader-config-name"`
+	Loader             *LoaderConfig   `yaml:"loader"`
+	LoaderThread       int             `yaml:"loader-thread"`
+	SyncerConfigName   string          `yaml:"syncer-config-name"`
+	Syncer             *SyncerConfig   `yaml:"syncer"`
+	SyncerThread       int             `yaml:"syncer-thread"`
+	// new config item
+	ExpressionFilters []string `yaml:"expression-filters,omitempty"`
+}
+
+// NewMySQLInstancesForDowngrade creates []* MySQLInstanceForDowngrade.
+func NewMySQLInstancesForDowngrade(mysqlInstances []*MySQLInstance) []*MySQLInstanceForDowngrade {
+	mysqlInstancesForDowngrade := make([]*MySQLInstanceForDowngrade, 0, len(mysqlInstances))
+	for _, m := range mysqlInstances {
+		newMySQLInstance := &MySQLInstanceForDowngrade{
+			SourceID:           m.SourceID,
+			Meta:               m.Meta,
+			FilterRules:        m.FilterRules,
+			ColumnMappingRules: m.ColumnMappingRules,
+			RouteRules:         m.RouteRules,
+			BWListName:         m.BWListName,
+			BAListName:         m.BAListName,
+			MydumperConfigName: m.MydumperConfigName,
+			Mydumper:           m.Mydumper,
+			MydumperThread:     m.MydumperThread,
+			LoaderConfigName:   m.LoaderConfigName,
+			Loader:             m.Loader,
+			LoaderThread:       m.LoaderThread,
+			SyncerConfigName:   m.SyncerConfigName,
+			Syncer:             m.Syncer,
+			SyncerThread:       m.SyncerThread,
+			ExpressionFilters:  m.ExpressionFilters,
+		}
+		mysqlInstancesForDowngrade = append(mysqlInstancesForDowngrade, newMySQLInstance)
+	}
+	return mysqlInstancesForDowngrade
+}
+
+// TaskConfigForDowngrade is the base configuration for task in v2.0.
+// This config is used for downgrade(config export) from a higher dmctl version.
+// When we add any new config item into SourceConfig, we should update it also.
+type TaskConfigForDowngrade struct {
+	Name                    string                         `yaml:"name"`
+	TaskMode                string                         `yaml:"task-mode"`
+	IsSharding              bool                           `yaml:"is-sharding"`
+	ShardMode               string                         `yaml:"shard-mode"`
+	IgnoreCheckingItems     []string                       `yaml:"ignore-checking-items"`
+	MetaSchema              string                         `yaml:"meta-schema"`
+	EnableHeartbeat         bool                           `yaml:"enable-heartbeat"`
+	HeartbeatUpdateInterval int                            `yaml:"heartbeat-update-interval"`
+	HeartbeatReportInterval int                            `yaml:"heartbeat-report-interval"`
+	Timezone                string                         `yaml:"timezone"`
+	CaseSensitive           bool                           `yaml:"case-sensitive"`
+	TargetDB                *DBConfig                      `yaml:"target-database"`
+	OnlineDDLScheme         string                         `yaml:"online-ddl-scheme"`
+	Routes                  map[string]*router.TableRule   `yaml:"routes"`
+	Filters                 map[string]*bf.BinlogEventRule `yaml:"filters"`
+	ColumnMappings          map[string]*column.Rule        `yaml:"column-mappings"`
+	BWList                  map[string]*filter.Rules       `yaml:"black-white-list"`
+	BAList                  map[string]*filter.Rules       `yaml:"block-allow-list"`
+	Mydumpers               map[string]*MydumperConfig     `yaml:"mydumpers"`
+	Loaders                 map[string]*LoaderConfig       `yaml:"loaders"`
+	Syncers                 map[string]*SyncerConfig       `yaml:"syncers"`
+	CleanDumpFile           bool                           `yaml:"clean-dump-file"`
+	EnableANSIQuotes        bool                           `yaml:"ansi-quotes"`
+	RemoveMeta              bool                           `yaml:"remove-meta"`
+	// new config item
+	MySQLInstances []*MySQLInstanceForDowngrade `yaml:"mysql-instances"`
+	ExprFilter     map[string]*ExpressionFilter `yaml:"expression-filter,omitempty"`
+	OnlineDDL      bool                         `yaml:"online-ddl,omitempty"`
+}
+
+// NewTaskConfigForDowngrade create new TaskConfigForDowngrade.
+func NewTaskConfigForDowngrade(taskConfig *TaskConfig) *TaskConfigForDowngrade {
+	return &TaskConfigForDowngrade{
+		Name:                    taskConfig.Name,
+		TaskMode:                taskConfig.TaskMode,
+		IsSharding:              taskConfig.IsSharding,
+		ShardMode:               taskConfig.ShardMode,
+		IgnoreCheckingItems:     taskConfig.IgnoreCheckingItems,
+		MetaSchema:              taskConfig.MetaSchema,
+		EnableHeartbeat:         taskConfig.EnableHeartbeat,
+		HeartbeatUpdateInterval: taskConfig.HeartbeatUpdateInterval,
+		HeartbeatReportInterval: taskConfig.HeartbeatReportInterval,
+		Timezone:                taskConfig.Timezone,
+		CaseSensitive:           taskConfig.CaseSensitive,
+		TargetDB:                taskConfig.TargetDB,
+		OnlineDDLScheme:         taskConfig.OnlineDDLScheme,
+		Routes:                  taskConfig.Routes,
+		Filters:                 taskConfig.Filters,
+		ColumnMappings:          taskConfig.ColumnMappings,
+		BWList:                  taskConfig.BWList,
+		BAList:                  taskConfig.BAList,
+		Mydumpers:               taskConfig.Mydumpers,
+		Loaders:                 taskConfig.Loaders,
+		Syncers:                 taskConfig.Syncers,
+		CleanDumpFile:           taskConfig.CleanDumpFile,
+		EnableANSIQuotes:        taskConfig.EnableANSIQuotes,
+		RemoveMeta:              taskConfig.RemoveMeta,
+		MySQLInstances:          NewMySQLInstancesForDowngrade(taskConfig.MySQLInstances),
+		ExprFilter:              taskConfig.ExprFilter,
+		OnlineDDL:               taskConfig.OnlineDDL,
+	}
+}
+
+// omitDefaultVals change default value to empty value for new config item.
+// If any default value for new config item is not empty(0 or false or nil),
+// we should change it to empty.
+func (c *TaskConfigForDowngrade) omitDefaultVals() {
+	if len(c.TargetDB.Session) > 0 {
+		if timeZone, ok := c.TargetDB.Session["time_zone"]; ok && timeZone == defaultTimeZone {
+			delete(c.TargetDB.Session, "time_zone")
+		}
+	}
+}
+
+// Yaml returns YAML format representation of config.
+func (c *TaskConfigForDowngrade) Yaml() (string, error) {
+	b, err := yaml.Marshal(c)
+	return string(b), err
+}

dm/config/task_test.go

@@ -16,6 +16,7 @@ package config
 import (
 	"io/ioutil"
 	"path"
+	"reflect"
 	"sort"
 	"strings"
 
@@ -29,8 +30,7 @@ import (
 	"github.com/coreos/go-semver/semver"
 )
 
-func (t *testConfig) TestUnusedTaskConfig(c *C) {
-	correctTaskConfig := `---
+var correctTaskConfig = `---
 name: test
 task-mode: all
 shard-mode: "pessimistic"       
@@ -135,6 +135,8 @@ mysql-instances:
     loader-config-name: "global2"
     syncer-config-name: "global2"
 `
+
+func (t *testConfig) TestUnusedTaskConfig(c *C) {
 	taskConfig := NewTaskConfig()
 	err := taskConfig.Decode(correctTaskConfig)
 	c.Assert(err, IsNil)
@@ -1011,3 +1013,78 @@ func (t *testConfig) TestExclusiveAndWrongExprFilterFields(c *C) {
 	err = cfg.adjust()
 	c.Assert(terror.ErrConfigExprFilterWrongGrammar.Equal(err), IsTrue)
 }
+
+func (t *testConfig) TestTaskConfigForDowngrade(c *C) {
+	cfg := NewTaskConfig()
+	err := cfg.Decode(correctTaskConfig)
+	c.Assert(err, IsNil)
+
+	cfgForDowngrade := NewTaskConfigForDowngrade(cfg)
+
+	// make sure all new field were added
+	cfgReflect := reflect.Indirect(reflect.ValueOf(cfg))
+	cfgForDowngradeReflect := reflect.Indirect(reflect.ValueOf(cfgForDowngrade))
+	c.Assert(cfgReflect.NumField(), Equals, cfgForDowngradeReflect.NumField()+1) // without flag
+
+	// make sure all field were copied
+	cfgForClone := &TaskConfigForDowngrade{}
+	Clone(cfgForClone, cfg)
+	c.Assert(cfgForDowngrade, DeepEquals, cfgForClone)
+}
+
+// Clone clones src to dest.
+func Clone(dest, src interface{}) {
+	cloneValues(reflect.ValueOf(dest), reflect.ValueOf(src))
+}
+
+// cloneValues clone src to dest recursively.
+// Note: pointer still use shallow copy.
+func cloneValues(dest, src reflect.Value) {
+	destType := dest.Type()
+	srcType := src.Type()
+	if destType.Kind() == reflect.Ptr {
+		destType = destType.Elem()
+	}
+	if srcType.Kind() == reflect.Ptr {
+		srcType = srcType.Elem()
+	}
+
+	if destType.Kind() == reflect.Slice {
+		slice := reflect.MakeSlice(destType, src.Len(), src.Cap())
+		for i := 0; i < src.Len(); i++ {
+			if slice.Index(i).Type().Kind() == reflect.Ptr {
+				newVal := reflect.New(slice.Index(i).Type().Elem())
+				cloneValues(newVal, src.Index(i))
+				slice.Index(i).Set(newVal)
+			} else {
+				cloneValues(slice.Index(i).Addr(), src.Index(i).Addr())
+			}
+		}
+		dest.Set(slice)
+		return
+	}
+
+	destFieldsMap := map[string]int{}
+	for i := 0; i < destType.NumField(); i++ {
+		destFieldsMap[destType.Field(i).Name] = i
+	}
+	for i := 0; i < srcType.NumField(); i++ {
+		if j, ok := destFieldsMap[srcType.Field(i).Name]; ok {
+			destField := dest.Elem().Field(j)
+			srcField := src.Elem().Field(i)
+			destFieldType := destField.Type()
+			srcFieldType := srcField.Type()
+			if destFieldType.Kind() == reflect.Ptr {
+				destFieldType = destFieldType.Elem()
+			}
+			if srcFieldType.Kind() == reflect.Ptr {
+				srcFieldType = srcFieldType.Elem()
+			}
+			if destFieldType != srcFieldType {
+				cloneValues(destField, srcField)
+			} else {
+				destField.Set(srcField)
+			}
+		}
+	}
+}