config/task: add check for duplicate config items (#380) (#385)

_utils/terror_gen/errors_release.txt

@@ -141,6 +141,7 @@ ErrConfigMydumperPathNotValid,[code=20028:class=config:scope=internal:level=medi
 ErrConfigLoaderCfgNotFound,[code=20029:class=config:scope=internal:level=medium],"mysql-instance(%d)'s loader config %s not exist in loaders"
 ErrConfigSyncerCfgNotFound,[code=20030:class=config:scope=internal:level=medium],"mysql-instance(%d)'s syncer config %s not exist in syncer"
 ErrConfigSourceIDNotFound,[code=20031:class=config:scope=internal:level=medium],"source %s in deployment configuration not found"
+ErrConfigDuplicateCfgItem,[code=20032:class=config:scope=internal:level=medium],"the following mysql configs have duplicate items, please remove the duplicates:\n%s"
 ErrBinlogExtractPosition,[code=22001:class=binlog-op:scope=internal:level=high],""
 ErrBinlogInvalidFilename,[code=22002:class=binlog-op:scope=internal:level=high],"invalid binlog filename"
 ErrBinlogParsePosFromStr,[code=22003:class=binlog-op:scope=internal:level=high],""

dm/config/task.go

@@ -15,7 +15,9 @@ package config
 
 import (
 	"flag"
+	"fmt"
 	"io/ioutil"
+	"strings"
 	"time"
 
 	"github.com/pingcap/dm/pkg/log"
@@ -348,6 +350,7 @@ func (c *TaskConfig) adjust() error {
 	}
 
 	iids := make(map[string]int) // source-id -> instance-index
+	duplicateErrorStrings := make([]string, 0)
 	for i, inst := range c.MySQLInstances {
 		if err := inst.Verify(); err != nil {
 			return terror.Annotatef(err, "mysql-instance: %s", humanize.Ordinal(i))
@@ -443,6 +446,16 @@ func (c *TaskConfig) adjust() error {
 		if inst.SyncerThread != 0 {
 			inst.Syncer.WorkerCount = inst.SyncerThread
 		}
+
+		if dupeRules := checkDuplicateString(inst.RouteRules); len(dupeRules) > 0 {
+			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s route-rules: %s", i, strings.Join(dupeRules, ", ")))
+		}
+		if dupeRules := checkDuplicateString(inst.FilterRules); len(dupeRules) > 0 {
+			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s filter-rules: %s", i, strings.Join(dupeRules, ", ")))
+		}
+	}
+	if len(duplicateErrorStrings) > 0 {
+		return terror.ErrConfigDuplicateCfgItem.Generate(strings.Join(duplicateErrorStrings, "\n"))
 	}
 
 	if c.Timezone != "" {
@@ -516,3 +529,21 @@ func (c *TaskConfig) SubTaskConfigs(sources map[string]DBConfig) ([]*SubTaskConf
 	}
 	return cfgs, nil
 }
+
+// checkDuplicateString checks whether the given string array has duplicate string item
+// if there is duplicate, it will return **all** the duplicate strings
+func checkDuplicateString(ruleNames []string) []string {
+	mp := make(map[string]bool, len(ruleNames))
+	dupeArray := make([]string, 0)
+	for _, name := range ruleNames {
+		if added, ok := mp[name]; ok {
+			if !added {
+				dupeArray = append(dupeArray, name)
+				mp[name] = true
+			}
+		} else {
+			mp[name] = false
+		}
+	}
+	return dupeArray
+}

dm/config/task_test.go

@@ -14,9 +14,13 @@
 package config
 
 import (
-	. "github.com/pingcap/check"
 	"io/ioutil"
 	"path"
+	"sort"
+
+	"github.com/pingcap/dm/pkg/terror"
+
+	. "github.com/pingcap/check"
 )
 
 func (t *testConfig) TestInvalidTaskConfig(c *C) {
@@ -96,6 +100,8 @@ timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
 `)
 	err = ioutil.WriteFile(filepath, configContent, 0644)
+	c.Assert(err, IsNil)
+	taskConfig = NewTaskConfig()
 	err = taskConfig.DecodeFile(filepath)
 	c.Assert(err, NotNil)
 	c.Assert(err, ErrorMatches, "*line 2: field aaa not found in type config.TaskConfig*")
@@ -113,6 +119,8 @@ timezone: "Asia/Shanghai"
 ignore-checking-items: ["all"]
 `)
 	err = ioutil.WriteFile(filepath, configContent, 0644)
+	c.Assert(err, IsNil)
+	taskConfig = NewTaskConfig()
 	err = taskConfig.DecodeFile(filepath)
 	c.Assert(err, NotNil)
 	c.Assert(err, ErrorMatches, "*line 4: field task-mode already set in type config.TaskConfig*")
@@ -181,6 +189,8 @@ syncers:
 `)
 
 	err = ioutil.WriteFile(filepath, configContent, 0644)
+	c.Assert(err, IsNil)
+	taskConfig = NewTaskConfig()
 	err = taskConfig.DecodeFile(filepath)
 	c.Assert(err, IsNil)
 	c.Assert(taskConfig.MySQLInstances[0].Mydumper.Threads, Equals, 11)
@@ -192,4 +202,67 @@ syncers:
 	c.Assert(taskConfig.MySQLInstances[2].Mydumper.Threads, Equals, 44)
 	c.Assert(taskConfig.MySQLInstances[2].Loader.PoolSize, Equals, 55)
 	c.Assert(taskConfig.MySQLInstances[2].Syncer.WorkerCount, Equals, 66)
+
+	configContent = []byte(`---
+name: test
+task-mode: all
+is-sharding: false
+meta-schema: "dm_meta"
+remove-meta: false
+enable-heartbeat: true
+heartbeat-update-interval: 1
+heartbeat-report-interval: 1
+timezone: "Asia/Shanghai"
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+  - source-id: "mysql-replica-02"
+  - source-id: "mysql-replica-03"
+
+black-white-list:
+  instance:
+    do-dbs: ["test"]
+
+routes: 
+  route-rule-1:
+  route-rule-2:
+  route-rule-3:
+  route-rule-4:
+
+filters:
+  filter-rule-1:
+  filter-rule-2:
+  filter-rule-3:
+  filter-rule-4:
+`)
+
+	err = ioutil.WriteFile(filepath, configContent, 0644)
+	c.Assert(err, IsNil)
+	taskConfig = NewTaskConfig()
+	err = taskConfig.DecodeFile(filepath)
+	c.Assert(err, IsNil)
+	taskConfig.MySQLInstances[0].RouteRules = []string{"route-rule-1", "route-rule-2", "route-rule-1", "route-rule-2"}
+	taskConfig.MySQLInstances[1].FilterRules = []string{"filter-rule-1", "filter-rule-2", "filter-rule-3", "filter-rule-2"}
+	err = taskConfig.adjust()
+	c.Assert(terror.ErrConfigDuplicateCfgItem.Equal(err), IsTrue)
+	c.Assert(err, ErrorMatches, `[\s\S]*mysql-instance\(0\)'s route-rules: route-rule-1, route-rule-2[\s\S]*`)
+	c.Assert(err, ErrorMatches, `[\s\S]*mysql-instance\(1\)'s filter-rules: filter-rule-2[\s\S]*`)
+
+}
+
+func (t *testConfig) TestCheckDuplicateString(c *C) {
+	a := []string{"a", "b", "c", "d"}
+	dupeStrings := checkDuplicateString(a)
+	c.Assert(dupeStrings, HasLen, 0)
+	a = []string{"a", "a", "b", "b", "c", "c"}
+	dupeStrings = checkDuplicateString(a)
+	c.Assert(dupeStrings, HasLen, 3)
+	sort.Strings(dupeStrings)
+	c.Assert(dupeStrings, DeepEquals, []string{"a", "b", "c"})
 }

pkg/terror/error_list.go

@@ -175,6 +175,7 @@ const (
 	codeConfigLoaderCfgNotFound
 	codeConfigSyncerCfgNotFound
 	codeConfigSourceIDNotFound
+	codeConfigDuplicateCfgItem
 )
 
 // Binlog operation error code list
@@ -636,6 +637,7 @@ var (
 	ErrConfigLoaderCfgNotFound      = New(codeConfigLoaderCfgNotFound, ClassConfig, ScopeInternal, LevelMedium, "mysql-instance(%d)'s loader config %s not exist in loaders")
 	ErrConfigSyncerCfgNotFound      = New(codeConfigSyncerCfgNotFound, ClassConfig, ScopeInternal, LevelMedium, "mysql-instance(%d)'s syncer config %s not exist in syncer")
 	ErrConfigSourceIDNotFound       = New(codeConfigSourceIDNotFound, ClassConfig, ScopeInternal, LevelMedium, "source %s in deployment configuration not found")
+	ErrConfigDuplicateCfgItem       = New(codeConfigDuplicateCfgItem, ClassConfig, ScopeInternal, LevelMedium, "the following mysql configs have duplicate items, please remove the duplicates:\n%s")
 
 	// Binlog operation error
 	ErrBinlogExtractPosition = New(codeBinlogExtractPosition, ClassBinlogOp, ScopeInternal, LevelHigh, "")