dm: fix empty config cause dm-master panic (#5298)

close #3732

dm/dm/config/task.go

@@ -701,8 +701,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigMydumperCfgNotFound.Generate(i, inst.MydumperConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[mydumperIdx]+inst.MydumperConfigName]++
-			inst.Mydumper = new(MydumperConfig)
-			*inst.Mydumper = *rule // ref mydumper config
+			if rule != nil {
+				inst.Mydumper = new(MydumperConfig)
+				*inst.Mydumper = *rule // ref mydumper config
+			}
 		}
 		if inst.Mydumper == nil {
 			if len(c.Mydumpers) != 0 {
@@ -729,8 +731,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigLoaderCfgNotFound.Generate(i, inst.LoaderConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[loaderIdx]+inst.LoaderConfigName]++
-			inst.Loader = new(LoaderConfig)
-			*inst.Loader = *rule // ref loader config
+			if rule != nil {
+				inst.Loader = new(LoaderConfig)
+				*inst.Loader = *rule // ref loader config
+			}
 		}
 		if inst.Loader == nil {
 			if len(c.Loaders) != 0 {
@@ -749,8 +753,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigSyncerCfgNotFound.Generate(i, inst.SyncerConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[syncerIdx]+inst.SyncerConfigName]++
-			inst.Syncer = new(SyncerConfig)
-			*inst.Syncer = *rule // ref syncer config
+			if rule != nil {
+				inst.Syncer = new(SyncerConfig)
+				*inst.Syncer = *rule // ref syncer config
+			}
 		}
 		if inst.Syncer == nil {
 			if len(c.Syncers) != 0 {
@@ -770,7 +776,9 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrContinuousValidatorCfgNotFound.Generate(i, inst.ContinuousValidatorConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[validatorIdx]+inst.ContinuousValidatorConfigName]++
-			inst.ContinuousValidator = *rule
+			if rule != nil {
+				inst.ContinuousValidator = *rule
+			}
 		}
 
 		// for backward compatible, set global config `ansi-quotes: true` if any syncer is true
@@ -826,9 +834,12 @@ func (c *TaskConfig) adjust() error {
 			unusedConfigs = append(unusedConfigs, mydumper)
 		}
 	}
+
 	for loader, cfg := range c.Loaders {
-		if err1 := cfg.adjust(); err1 != nil {
-			return err1
+		if cfg != nil {
+			if err1 := cfg.adjust(); err1 != nil {
+				return err1
+			}
 		}
 		if globalConfigReferCount[configRefPrefixes[loaderIdx]+loader] == 0 {
 			unusedConfigs = append(unusedConfigs, loader)
@@ -854,6 +865,7 @@ func (c *TaskConfig) adjust() error {
 		sort.Strings(unusedConfigs)
 		return terror.ErrConfigGlobalConfigsUnused.Generate(unusedConfigs)
 	}
+
 	// we postpone default time_zone init in each unit so we won't change the config value in task/sub_task config
 	if c.Timezone != "" {
 		if _, err := utils.ParseTimeZone(c.Timezone); err != nil {

dm/tests/dmctl_basic/check_list/check_task.sh

@@ -87,3 +87,31 @@ function check_task_only_warning() {
 		"check-task $task_conf" \
 		"\"state\": \"warn\"" 1
 }
+
+function check_task_empty_dump_config() {
+	sed "/threads/d" $1 >/tmp/empty-dump.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-dump.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_load_config() {
+	sed "/pool-size/d" $1 >/tmp/empty-load.yaml
+	cat /tmp/empty-load.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-load.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_sync_config() {
+	sed "/worker-count/d" $1 >/tmp/empty-sync.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-sync.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_config() {
+	check_task_empty_dump_config $1
+	check_task_empty_load_config $1
+	check_task_empty_sync_config $1
+}

dm/tests/dmctl_basic/check_list/start_task.sh

@@ -26,3 +26,21 @@ function start_task_not_pass_with_message() {
 		"start-task $task_conf" \
 		"$2" 1
 }
+
+function start_task_empty_config() {
+	cp $1 /tmp/empty-cfg.yaml
+	sed -i "/threads/d" /tmp/empty-cfg.yaml
+	sed -i "/pool-size/d" /tmp/empty-cfg.yaml
+	sed -i "/worker-count/d" /tmp/empty-cfg.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task /tmp/empty-cfg.yaml" \
+		"\"result\": true" 2
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"config task empty-unit-task" \
+		"threads: 4" 1 \
+		"pool-size: 16" 1 \
+		"worker-count: 16" 1
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"stop-task $1" \
+		"\"result\": true" 2
+}

dm/tests/dmctl_basic/conf/empty-unit-task.yaml

@@ -0,0 +1,35 @@
+---
+name: empty-unit-task
+task-mode: all
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+    block-allow-list:  "instance"
+
+mydumpers:
+  global:
+    threads: 4
+
+loaders:
+  global:
+    pool-size: 16
+
+syncers:
+  global:
+    worker-count: 32
+
+block-allow-list:
+  instance:
+    do-dbs: ["dmctl"]
+    do-tables:
+    -  db-name: "dmctl"
+       tbl-name: "~^t_[\\d]+"

dm/tests/dmctl_basic/run.sh

@@ -297,6 +297,11 @@ function run() {
 		"stop-task $cur/conf/only_warning.yaml" \
 		"\"result\": true" 2
 
+	echo "check task with empty unit config"
+	check_task_empty_config $cur/conf/empty-unit-task.yaml
+	echo "start task with empty unit config"
+	start_task_empty_config $cur/conf/empty-unit-task.yaml
+
 	cp $cur/conf/dm-task.yaml $WORK_DIR/dm-task-error-database-config.yaml
 	sed -i "s/password: \"\"/password: \"wrond password\"/g" $WORK_DIR/dm-task-error-database-config.yaml
 	check_task_error_database_config $WORK_DIR/dm-task-error-database-config.yaml