Fix panic of empty value in filter config

drainer/config.go

@@ -226,6 +226,43 @@ func (cfg *Config) configFromFile(path string) error {
 	return util.StrictDecodeFile(path, "drainer", cfg)
 }
 
+func (cfg *Config) validateFilter() error {
+	for _, db := range cfg.SyncerCfg.DoDBs {
+		if len(db) == 0 {
+			return errors.New("empty schema name in `replicate-do-db` config")
+		}
+	}
+
+	dbs := strings.Split(cfg.SyncerCfg.IgnoreSchemas, ",")
+	for _, db := range dbs {
+		if len(db) == 0 {
+			return errors.New("empty schema name in `ignore-schemas` config")
+		}
+	}
+
+	for _, tb := range cfg.SyncerCfg.DoTables {
+		if len(tb.Schema) == 0 {
+			return errors.New("empty schema name in `replicate-do-table` config")
+		}
+
+		if len(tb.Table) == 0 {
+			return errors.New("empty table name in `replicate-do-table` config")
+		}
+	}
+
+	for _, tb := range cfg.SyncerCfg.IgnoreTables {
+		if len(tb.Schema) == 0 {
+			return errors.New("empty schema name in `ignore-table` config")
+		}
+
+		if len(tb.Table) == 0 {
+			return errors.New("empty table name in `ignore-table` config")
+		}
+	}
+
+	return nil
+}
+
 // validate checks whether the configuration is valid
 func (cfg *Config) validate() error {
 	if err := validateAddr(cfg.ListenAddr); err != nil {
@@ -254,7 +291,7 @@ func (cfg *Config) validate() error {
 		}
 	}
 
-	return nil
+	return cfg.validateFilter()
 }
 
 func (cfg *Config) adjustConfig() error {

drainer/config_test.go

@@ -25,6 +25,7 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser/mysql"
 	dsync "github.com/pingcap/tidb-binlog/drainer/sync"
+	"github.com/pingcap/tidb-binlog/pkg/filter"
 	"github.com/pingcap/tidb-binlog/pkg/util"
 	pkgzk "github.com/pingcap/tidb-binlog/pkg/zk"
 	"github.com/samuel/go-zookeeper/zk"
@@ -67,6 +68,38 @@ func (t *testDrainerSuite) TestConfig(c *C) {
 	c.Assert(cfg.SyncerCfg.SQLMode, Equals, mysql.SQLMode(0))
 }
 
+func (t *testDrainerSuite) TestValidateFilter(c *C) {
+	cfg := NewConfig()
+	c.Assert(cfg.validateFilter(), IsNil)
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.DoDBs = []string{""}
+	c.Assert(cfg.validateFilter(), NotNil)
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.IgnoreSchemas = "a,,c"
+	c.Assert(cfg.validateFilter(), NotNil)
+
+	emptyScheme := []filter.TableName{{Schema: "", Table: "t"}}
+	emptyTable := []filter.TableName{{Schema: "s", Table: ""}}
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.DoTables = emptyScheme
+	c.Assert(cfg.validateFilter(), NotNil)
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.DoTables = emptyTable
+	c.Assert(cfg.validateFilter(), NotNil)
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.IgnoreTables = emptyScheme
+	c.Assert(cfg.validateFilter(), NotNil)
+
+	cfg = NewConfig()
+	cfg.SyncerCfg.IgnoreTables = emptyTable
+	c.Assert(cfg.validateFilter(), NotNil)
+}
+
 func (t *testDrainerSuite) TestValidate(c *C) {
 	cfg := NewConfig()
 

pkg/filter/filter.go

@@ -48,10 +48,10 @@ func NewFilter(ignoreDBs []string, ignoreTables []TableName, doDBs []string, doT
 func (s *Filter) addOneRegex(originStr string) {
 	if _, ok := s.reMap[originStr]; !ok {
 		var re *regexp.Regexp
-		if originStr[0] != '~' {
-			re = regexp.MustCompile(fmt.Sprintf("(?i)^%s$", originStr))
-		} else {
+		if len(originStr) > 0 && originStr[0] == '~' {
 			re = regexp.MustCompile(fmt.Sprintf("(?i)%s", originStr[1:]))
+		} else { // must match completely
+			re = regexp.MustCompile(fmt.Sprintf("(?i)^%s$", originStr))
 		}
 		s.reMap[originStr] = re
 	}

pkg/filter/filter_test.go

@@ -49,4 +49,9 @@ func (t *testFilterSuite) TestFilter(c *C) {
 	filter = NewFilter(nil, ignoreTables, nil, nil)
 	c.Assert(filter.SkipSchemaAndTable("ignore", "ignore"), IsTrue)
 	c.Assert(filter.SkipSchemaAndTable("not_ignore", "not_ignore"), IsFalse)
+
+	// with empty string
+	filter = NewFilter(nil, nil, []string{""} /*doDBs*/, nil)
+	c.Assert(filter.SkipSchemaAndTable("", "any"), IsFalse)
+	c.Assert(filter.SkipSchemaAndTable("any", ""), IsTrue)
 }