Repository: pingcap/tidb-binlog
Pull request #696 (2019-07-30 08:38:40)

Add several UTs to improve drainer's coverage
<!--
Thank you for contributing to TiDB! Please read TiDB's [CONTRIBUTING](https://github.com/pingcap/tidb/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
[TOOL-1044](https://internal.pingcap.net/jira/browse/TOOL-1044)

### What is changed and how it works?
Five UTs are added in this PR. Coverage for `/drainer` is improved to around 77.1%.
1. [commit 1](https://github.com/lichunzhu/tidb-binlog/commit/8763780a6404243b972a7fe6e418314722a2c166) adds a rename table test in `schema_test.go`. Command `DROP TABLE` will use the new table name `RT`.
2. [commit 2](https://github.com/lichunzhu/tidb-binlog/commit/76731a0efc7e9eb1be1778ab9a0504c98dd7540e) adds close caller for merger and adds a UT check whether merger can be closed successfully in `merger.go` & `merger_test.go`.
3. [commit 3](https://github.com/lichunzhu/tidb-binlog/commit/d1141ac1a453d2dbe7ee2fcab9d452929b289b14) adds UT to test whether collector can update status successfully in `collector_test.go`.
4. [commit 4](https://github.com/lichunzhu/tidb-binlog/commit/9d948aa63e71fa76e80191eef91f1ef07a903d0d) adds UT to test whether config can generate correct kafka configs with zookeeper addrs or without zookeeper addrs in `config_test.go`.
5. [commit 5](https://github.com/lichunzhu/tidb-binlog/commit/09d241a965975ca65cb005f3c83f2a594ebd157d) adds UT to test whether `pump.PullBinlog` can successfully get binlog item and omit disordered items with wrong commitTS in `pump_test.go`.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

Code changes

 - Has interface methods change
   Use `newZKFromConnectionString = zk.NewFromConnectionString` which changes the function usage in `/drainer/config.go`

Side effects

 - Increased code complexity

Related changes


july2993: /run-unit-test
lichunzhu: /run-unit-test
lichunzhu: All comments have been accepted and relative parts have been revealed. PTAL. @suzaku 
july2993: /run-all-tests

Patch:
diff --git a/drainer/collector.go b/drainer/collector.go
index 1fe39e7e..23fb7d18 100644
--- a/drainer/collector.go
+++ b/drainer/collector.go
@@ -128,7 +128,10 @@ func (c *Collector) publishBinlogs(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case mergeItem := <-c.merger.Output():
+		case mergeItem, ok := <-c.merger.Output():
+			if !ok {
+				return
+			}
 			item := mergeItem.(*binlogItem)
 			if err := c.syncBinlog(item); err != nil {
 				c.reportErr(ctx, err)
@@ -155,6 +158,7 @@ func (c *Collector) Start(ctx context.Context) {
 	if err := c.reg.Close(); err != nil {
 		log.Error(err.Error())
 	}
+	c.merger.Close()
 
 	wg.Wait()
 }
diff --git a/drainer/config.go b/drainer/config.go
index 32704366..b9025138 100644
--- a/drainer/config.go
+++ b/drainer/config.go
@@ -51,9 +51,10 @@ const (
 )
 
 var (
-	maxBinlogItemCount     int
-	defaultBinlogItemCount = 16 << 12
-	supportedCompressors   = [...]string{"gzip"}
+	maxBinlogItemCount        int
+	defaultBinlogItemCount    = 16 << 12
+	supportedCompressors      = [...]string{"gzip"}
+	newZKFromConnectionString = zk.NewFromConnectionString
 )
 
 // SyncerConfig is the Syncer's configuration.
@@ -278,7 +279,7 @@ func (cfg *Config) adjustConfig() error {
 	if cfg.SyncerCfg.DestDBType == "kafka" {
 		// get KafkaAddrs from zookeeper if ZkAddrs is setted
 		if cfg.SyncerCfg.To.ZKAddrs != "" {
-			zkClient, err := zk.NewFromConnectionString(cfg.SyncerCfg.To.ZKAddrs, time.Second*5, time.Second*60)
+			zkClient, err := newZKFromConnectionString(cfg.SyncerCfg.To.ZKAddrs, time.Second*5, time.Second*60)
 			if err != nil {
 				return errors.Trace(err)
 			}
diff --git a/drainer/merge.go b/drainer/merge.go
index ae843cf5..fe29322f 100644
--- a/drainer/merge.go
+++ b/drainer/merge.go
@@ -242,6 +242,7 @@ func (m *Merger) run() {
 		m.resetSourceChanged()
 
 		if m.isClosed() {
+			log.Info("Merger is closed successfully")
 			return
 		}
 


Test patch:
diff --git a/drainer/collector_test.go b/drainer/collector_test.go
index c86bb434..575c820a 100644
--- a/drainer/collector_test.go
+++ b/drainer/collector_test.go
@@ -46,6 +46,10 @@ type dummyStore struct {
 	kv.Storage
 }
 
+func (ds dummyStore) CurrentVersion() (kv.Version, error) {
+	return kv.NewVersion(2), nil
+}
+
 type collectorSuite struct{}
 
 var _ = Suite(&collectorSuite{})
@@ -463,3 +467,38 @@ func (s *keepUpdatingSuite) TestShouldUpdateWhenNotified(c *C) {
 	s.waitForSignal(c, signal, "Notifies don't finish in time")
 	c.Assert(callCount, Equals, 10)
 }
+
+type updateStatusSuite struct{}
+
+var _ = Suite(&updateStatusSuite{})
+
+func (s *updateStatusSuite) TestUpdateBothPumpAndCollectorStatus(c *C) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	etcdClient := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	r := node.NewEtcdRegistry(etcdClient, time.Duration(5)*time.Second)
+	ns := &node.Status{
+		NodeID:  "test",
+		Addr:    "test",
+		State:   node.Online,
+		IsAlive: true,
+	}
+	err := r.UpdateNode(ctx, "pumps", ns)
+	c.Assert(err, IsNil)
+
+	var latestTS int64 = 5
+	col := Collector{
+		merger:     &Merger{latestTS: latestTS, sources: map[string]MergeSource{}},
+		notifyChan: make(chan *notifyResult),
+		interval:   10 * time.Second,
+		reg:        r,
+		pumps:      map[string]*Pump{},
+		tiStore:    dummyStore{},
+	}
+	err = col.updateStatus(ctx)
+	c.Assert(err, IsNil)
+	c.Assert(col.mu.status, NotNil)
+	c.Assert(col.mu.status.LastTS, Equals, latestTS)
+	c.Assert(col.mu.status.Synced, IsFalse)
+	c.Assert(col.mu.status.PumpPos["test"], Equals, latestTS)
+}
diff --git a/drainer/config_test.go b/drainer/config_test.go
index 4749b284..031c4497 100644
--- a/drainer/config_test.go
+++ b/drainer/config_test.go
@@ -18,12 +18,16 @@ import (
 	"io/ioutil"
 	"path"
 	"testing"
+	"time"
 
 	"github.com/BurntSushi/toml"
 	"github.com/coreos/etcd/integration"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser/mysql"
+	dsync "github.com/pingcap/tidb-binlog/drainer/sync"
 	"github.com/pingcap/tidb-binlog/pkg/util"
+	pkgzk "github.com/pingcap/tidb-binlog/pkg/zk"
+	"github.com/samuel/go-zookeeper/zk"
 )
 
 var testEtcdCluster *integration.ClusterV3
@@ -155,3 +159,74 @@ func (t *testDrainerSuite) TestConfigParsingFileWithInvalidOptions(c *C) {
 	err = cfg.Parse(args)
 	c.Assert(err, ErrorMatches, ".*contained unknown configuration options: unrecognized-option-test.*")
 }
+
+var _ = Suite(&testKafkaSuite{})
+
+type testKafkaSuite struct {
+	origNewZKFromConnectionString func(connectionString string, dialTimeout, sessionTimeout time.Duration) (*pkgzk.Client, error)
+}
+
+func (t *testKafkaSuite) SetUpTest(c *C) {
+	t.origNewZKFromConnectionString = newZKFromConnectionString
+}
+
+func (t *testKafkaSuite) TearDownTest(c *C) {
+	newZKFromConnectionString = t.origNewZKFromConnectionString
+}
+
+type MockConn struct{}
+
+func (m *MockConn) Close() {}
+func (m *MockConn) Children(path string) ([]string, *zk.Stat, error) {
+	return []string{"0", "1"}, nil, nil
+}
+func (m *MockConn) Get(path string) ([]byte, *zk.Stat, error) {
+	if path[len(path)-1] == '0' {
+		return []byte(`{"version":2,"host":"192.0.2.1","port":9092}`), nil, nil
+	} else if path[len(path)-1] == '1' {
+		return []byte(`{"version":2,"host":"192.0.2.2","port":9092}`), nil, nil
+	}
+	return nil, nil, nil
+}
+
+func (t *testKafkaSuite) TestConfigDestDBTypeKafka(c *C) {
+	args := []string{
+		"-metrics-addr", "192.168.15.10:9091",
+		"-txn-batch", "1",
+		"-data-dir", "data.drainer",
+		"-dest-db-type", "kafka",
+		"-config", "../cmd/drainer/drainer.toml",
+		"-addr", "192.168.15.10:8257",
+		"-advertise-addr", "192.168.15.10:8257",
+	}
+	newZKFromConnectionString = func(connectionString string, dialTimeout, sessionTimeout time.Duration) (client *pkgzk.Client, e error) {
+		return pkgzk.NewWithConnection(&MockConn{}, nil), nil
+	}
+
+	// Without Zookeeper address
+	cfg := NewConfig()
+	err := cfg.Parse(args)
+	c.Assert(err, IsNil)
+	// kafka asserts
+	c.Assert(cfg.SyncerCfg.To.KafkaAddrs, Matches, defaultKafkaAddrs)
+	c.Assert(cfg.SyncerCfg.To.KafkaVersion, Equals, defaultKafkaVersion)
+	c.Assert(cfg.SyncerCfg.To.KafkaMaxMessages, Equals, 1024)
+
+	// With Zookeeper address
+	cfg = NewConfig()
+	cfg.SyncerCfg.To = new(dsync.DBConfig)
+	cfg.SyncerCfg.To.ZKAddrs = "host1:2181"
+	err = cfg.Parse(args)
+	c.Assert(err, IsNil)
+	c.Assert(cfg.MetricsAddr, Equals, "192.168.15.10:9091")
+	c.Assert(cfg.DataDir, Equals, "data.drainer")
+	c.Assert(cfg.SyncerCfg.TxnBatch, Equals, 1)
+	c.Assert(cfg.SyncerCfg.DestDBType, Equals, "kafka")
+	var strSQLMode *string
+	c.Assert(cfg.SyncerCfg.StrSQLMode, Equals, strSQLMode)
+	c.Assert(cfg.SyncerCfg.SQLMode, Equals, mysql.SQLMode(0))
+	// kafka asserts
+	c.Assert(cfg.SyncerCfg.To.KafkaAddrs, Matches, `(192\.0\.2\.1:9092,192\.0\.2\.2:9092|192\.0\.2\.2:9092,192\.0\.2\.1:9092)`)
+	c.Assert(cfg.SyncerCfg.To.KafkaVersion, Equals, defaultKafkaVersion)
+	c.Assert(cfg.SyncerCfg.To.KafkaMaxMessages, Equals, 1024)
+}
diff --git a/drainer/merge_test.go b/drainer/merge_test.go
index c7775f44..41e4f215 100644
--- a/drainer/merge_test.go
+++ b/drainer/merge_test.go
@@ -135,3 +135,24 @@ func (s *testMergerSuite) merge(c *C, strategy string) {
 		currentTs = ts
 	}
 }
+
+func (s *testMergerSuite) TestCloseMerger(c *C) {
+	sourceNum := 5
+	sources := make([]MergeSource, sourceNum)
+	for i := 0; i < sourceNum; i++ {
+		source := MergeSource{
+			ID:     strconv.Itoa(i),
+			Source: make(chan MergeItem),
+		}
+		sources[i] = source
+	}
+	merger := NewMerger(0, normalStrategy, sources...)
+	merger.Close()
+	c.Assert(merger.isClosed(), IsTrue)
+	select {
+	case _, ok := <-merger.Output():
+		c.Assert(ok, IsFalse)
+	case <-time.After(time.Second * 2):
+		c.Fatal("Fail to close merger's output in 2s")
+	}
+}
diff --git a/drainer/pump_test.go b/drainer/pump_test.go
index 82e083e0..d1e379ae 100644
--- a/drainer/pump_test.go
+++ b/drainer/pump_test.go
@@ -15,7 +15,12 @@ package drainer
 
 import (
 	"context"
+	"time"
+
 	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tipb/go-binlog"
+	"google.golang.org/grpc"
 )
 
 type pumpSuite struct{}
@@ -40,3 +45,64 @@ func (s *pumpSuite) TestGetCompressorName(c *C) {
 	c.Assert(ok, IsTrue)
 	c.Assert(cp, Equals, "gzip")
 }
+
+type mockPumpPullBinlogsClient struct {
+	grpc.ClientStream
+	binlogBytesChan chan []byte
+}
+
+func (x *mockPumpPullBinlogsClient) Recv() (*binlog.PullBinlogResp, error) {
+	payload, ok := <-x.binlogBytesChan
+	if !ok {
+		return nil, errors.Errorf("pump test has ran out of binlog items!")
+	}
+	return &binlog.PullBinlogResp{Entity: binlog.Entity{Payload: payload}}, nil
+}
+
+func (s *pumpSuite) TestPullBinlog(c *C) {
+	errChan := make(chan error, 10)
+	p := NewPump("pump_test", "", 0, 5, errChan)
+	p.grpcConn = &grpc.ClientConn{}
+	binlogBytesChan := make(chan []byte, 10)
+	p.pullCli = &mockPumpPullBinlogsClient{binlogBytesChan: binlogBytesChan}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer func() {
+		cancel()
+		p.grpcConn = nil
+		p.Close()
+		close(binlogBytesChan)
+	}()
+	ret := p.PullBinlog(ctx, 0)
+	// commitTs ascending
+	var commitTsArray = []int64{7, 9, 11, 13, 15, 17, 19, 21, 23, 25}
+	// commitTs disordering
+	var wrongCommitTsArray = []int64{27, 29, 28}
+
+	// ascending commitTs order
+	pullBinlogCommitTSChecker(commitTsArray, ret, binlogBytesChan, c)
+	c.Assert(p.latestTS, Equals, commitTsArray[len(commitTsArray)-1])
+
+	// should omit disorder binlog item, latestTs should be 29
+	pullBinlogCommitTSChecker(wrongCommitTsArray, ret, binlogBytesChan, c)
+	c.Assert(p.latestTS, Equals, wrongCommitTsArray[len(wrongCommitTsArray)-2])
+}
+
+func pullBinlogCommitTSChecker(commitTsArray []int64, ret chan MergeItem, binlogBytesChan chan []byte, c *C) {
+	go func() {
+		for _, commitTs := range commitTsArray {
+			binlogVal := new(binlog.Binlog)
+			binlogVal.CommitTs = commitTs
+			payload, err := binlogVal.Marshal()
+			c.Assert(err, IsNil)
+			binlogBytesChan <- payload
+		}
+	}()
+	for _, commitTs := range commitTsArray {
+		select {
+		case binlogItemEntity := <-ret:
+			c.Assert(binlogItemEntity.GetCommitTs(), Equals, commitTs)
+		case <-time.After(time.Second):
+			c.Fatal("Haven't receive pump binlog item in 1 sec")
+		}
+	}
+}
diff --git a/drainer/schema_test.go b/drainer/schema_test.go
index c65016f3..c61d5505 100644
--- a/drainer/schema_test.go
+++ b/drainer/schema_test.go
@@ -278,6 +278,7 @@ func (t *schemaSuite) TestHandleDDL(c *C) {
 	dbName := model.NewCIStr("Test")
 	colName := model.NewCIStr("A")
 	tbName := model.NewCIStr("T")
+	newTbName := model.NewCIStr("RT")
 
 	// check rollback done job
 	job := &model.Job{ID: 1, State: model.JobStateRollbackDone}
@@ -328,9 +329,10 @@ func (t *schemaSuite) TestHandleDDL(c *C) {
 		{name: "createSchema", jobID: 3, schemaID: 2, tableID: 0, jobType: model.ActionCreateSchema, binlogInfo: &model.HistoryInfo{SchemaVersion: 1, DBInfo: dbInfo, TableInfo: nil, FinishedTS: 123}, query: "create database Test", resultQuery: "create database Test", schemaName: dbInfo.Name.O, tableName: ""},
 		{name: "createTable", jobID: 7, schemaID: 2, tableID: 6, jobType: model.ActionCreateTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 3, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "create table T(id int);", resultQuery: "create table T(id int);", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
 		{name: "addColumn", jobID: 9, schemaID: 2, tableID: 6, jobType: model.ActionAddColumn, binlogInfo: &model.HistoryInfo{SchemaVersion: 4, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "alter table T add a varchar(45);", resultQuery: "alter table T add a varchar(45);", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
-		{name: "truncateTable", jobID: 11, schemaID: 2, tableID: 6, jobType: model.ActionTruncateTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 5, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "truncate table T;", resultQuery: "truncate table T;", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
-		{name: "dropTable", jobID: 12, schemaID: 2, tableID: 10, jobType: model.ActionDropTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 6, DBInfo: nil, TableInfo: nil, FinishedTS: 123}, query: "drop table T;", resultQuery: "drop table T;", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
-		{name: "dropSchema", jobID: 13, schemaID: 2, tableID: 0, jobType: model.ActionDropSchema, binlogInfo: &model.HistoryInfo{SchemaVersion: 7, DBInfo: nil, TableInfo: nil, FinishedTS: 123}, query: "drop database test;", resultQuery: "drop database test;", schemaName: dbInfo.Name.O, tableName: ""},
+		{name: "truncateTable", jobID: 10, schemaID: 2, tableID: 6, jobType: model.ActionTruncateTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 5, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "truncate table T;", resultQuery: "truncate table T;", schemaName: dbInfo.Name.O, tableName: tblInfo.Name.O},
+		{name: "renameTable", jobID: 11, schemaID: 2, tableID: 10, jobType: model.ActionRenameTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 6, DBInfo: nil, TableInfo: tblInfo, FinishedTS: 123}, query: "rename table T to RT;", resultQuery: "rename table T to RT;", schemaName: dbInfo.Name.O, tableName: newTbName.O},
+		{name: "dropTable", jobID: 12, schemaID: 2, tableID: 12, jobType: model.ActionDropTable, binlogInfo: &model.HistoryInfo{SchemaVersion: 7, DBInfo: nil, TableInfo: nil, FinishedTS: 123}, query: "drop table RT;", resultQuery: "drop table RT;", schemaName: dbInfo.Name.O, tableName: newTbName.O},
+		{name: "dropSchema", jobID: 13, schemaID: 2, tableID: 0, jobType: model.ActionDropSchema, binlogInfo: &model.HistoryInfo{SchemaVersion: 8, DBInfo: nil, TableInfo: nil, FinishedTS: 123}, query: "drop database test;", resultQuery: "drop database test;", schemaName: dbInfo.Name.O, tableName: ""},
 	}
 
 	for _, testCase := range testCases {
@@ -340,6 +342,9 @@ func (t *schemaSuite) TestHandleDDL(c *C) {
 			tblInfo.Columns = []*model.ColumnInfo{colInfo}
 		case "truncateTable":
 			tblInfo.ID = 10
+		case "renameTable":
+			tblInfo.ID = 12
+			tblInfo.Name = newTbName
 		}
 
 		job = &model.Job{
@@ -361,6 +366,10 @@ func (t *schemaSuite) TestHandleDDL(c *C) {
 		case "createTable":
 			_, ok := schema.TableByID(tblInfo.ID)
 			c.Assert(ok, IsTrue)
+		case "renameTable":
+			tb, ok := schema.TableByID(tblInfo.ID)
+			c.Assert(ok, IsTrue)
+			c.Assert(tblInfo.Name, Equals, tb.Name)
 		case "addColumn", "truncateTable":
 			tb, ok := schema.TableByID(tblInfo.ID)
 			c.Assert(ok, IsTrue)