Repository: pingcap/tidb-binlog
Pull request #601 (2019-05-15 03:43:46)

drainer: Add unit tests
<!--
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-->

[Jira Issue](https://internal.pingcap.net/jira/browse/TOOL-1044)

### What is changed and how it works?

Improve tests coverage: 60% -> 70%

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test

Code changes


Side effects


Related changes

suzaku: /run-unit-tests
suzaku: /run-integration-tests
suzaku: /run-all-tests
suzaku: /run-integration-tests

Patch:
diff --git a/drainer/collector.go b/drainer/collector.go
index 083fe408..3460ec21 100644
--- a/drainer/collector.go
+++ b/drainer/collector.go
@@ -70,7 +70,6 @@ type Collector struct {
 	merger *Merger
 
 	errCh chan error
-	wg    sync.WaitGroup
 }
 
 var (
@@ -123,10 +122,7 @@ func NewCollector(cfg *Config, clusterID uint64, s *Syncer, cpt checkpoint.Check
 }
 
 func (c *Collector) publishBinlogs(ctx context.Context) {
-	defer func() {
-		c.wg.Done()
-		log.Info("publishBinlogs quit")
-	}()
+	defer log.Info("publishBinlogs quit")
 
 	for {
 		select {
@@ -144,44 +140,23 @@ func (c *Collector) publishBinlogs(ctx context.Context) {
 
 // Start run a loop of collecting binlog from pumps online
 func (c *Collector) Start(ctx context.Context) {
-	defer func() {
-		for _, p := range c.pumps {
-			p.Close()
-		}
-		if err := c.reg.Close(); err != nil {
-			log.Error(err.Error())
-		}
-
-		c.wg.Wait()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		c.publishBinlogs(ctx)
+		wg.Done()
 	}()
 
-	c.wg.Add(1)
-	go c.publishBinlogs(ctx)
+	c.keepUpdatingStatus(ctx, c.updateStatus)
 
-	// add all the pump to merger
-	c.merger.Stop()
-	if err := c.updateStatus(ctx); err != nil {
+	for _, p := range c.pumps {
+		p.Close()
+	}
+	if err := c.reg.Close(); err != nil {
 		log.Error(err.Error())
 	}
-	c.merger.Continue()
 
-	// update status when had pump notify or reach wait time
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case nr := <-c.notifyChan:
-			nr.err = c.updateStatus(ctx)
-			nr.wg.Done()
-		case <-time.After(c.interval):
-			if err := c.updateStatus(ctx); err != nil {
-				log.Error(err.Error())
-			}
-		case err := <-c.errCh:
-			log.Error("collector meets error", zap.Error(err))
-			return
-		}
-	}
+	wg.Wait()
 }
 
 // updateCollectStatus updates the http status of the Collector.
@@ -352,3 +327,28 @@ func (c *Collector) handlePumpStatusUpdate(ctx context.Context, n *node.Status)
 		}
 	}
 }
+
+func (c *Collector) keepUpdatingStatus(ctx context.Context, fUpdate func(context.Context) error) {
+	// add all the pump to merger
+	c.merger.Stop()
+	fUpdate(ctx)
+	c.merger.Continue()
+
+	// update status when had pump notify or reach wait time
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case nr := <-c.notifyChan:
+			nr.err = fUpdate(ctx)
+			nr.wg.Done()
+		case <-time.After(c.interval):
+			if err := fUpdate(ctx); err != nil {
+				log.Error("Failed to update collector status", zap.Error(err))
+			}
+		case err := <-c.errCh:
+			log.Error("collector meets error", zap.Error(err))
+			return
+		}
+	}
+}
diff --git a/drainer/server.go b/drainer/server.go
index fe51391b..27a7653a 100644
--- a/drainer/server.go
+++ b/drainer/server.go
@@ -48,6 +48,7 @@ import (
 var (
 	nodePrefix        = "drainers"
 	heartbeatInterval = 1 * time.Second
+	getPdClient       = util.GetPdClient
 )
 
 type drainerKeyType string
@@ -95,7 +96,7 @@ func NewServer(cfg *Config) (*Server, error) {
 	}
 
 	// get pd client and cluster ID
-	pdCli, err := util.GetPdClient(cfg.EtcdURLs, cfg.Security)
+	pdCli, err := getPdClient(cfg.EtcdURLs, cfg.Security)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -236,8 +237,7 @@ func (s *Server) heartbeat(ctx context.Context) <-chan error {
 // Start runs CisternServer to serve the listening addr, and starts to collect binlog
 func (s *Server) Start() error {
 	// register drainer
-	err := s.updateStatus()
-	if err != nil {
+	if err := s.updateStatus(); err != nil {
 		return errors.Trace(err)
 	}
 	log.Info("register success", zap.String("drainer node id", s.ID))
@@ -263,8 +263,7 @@ func (s *Server) Start() error {
 
 	s.tg.GoNoPanic("syncer", func() {
 		defer s.Close()
-		err := s.syncer.Start()
-		if err != nil {
+		if err := s.syncer.Start(); err != nil {
 			log.Error("syncer exited abnormal", zap.Error(err))
 		}
 	})
@@ -286,13 +285,8 @@ func (s *Server) Start() error {
 	binlog.RegisterCisternServer(s.gs, s)
 	go s.gs.Serve(grpcL)
 
-	router := mux.NewRouter()
-	router.HandleFunc("/status", s.collector.Status).Methods("GET")
-	router.HandleFunc("/commit_ts", s.GetLatestTS).Methods("GET")
-	router.HandleFunc("/state/{nodeID}/{action}", s.ApplyAction).Methods("PUT")
+	router := s.initAPIRouter()
 	http.Handle("/", router)
-	prometheus.DefaultGatherer = registry
-	http.Handle("/metrics", promhttp.Handler())
 
 	go http.Serve(httpL, nil)
 
@@ -310,12 +304,12 @@ func (s *Server) ApplyAction(w http.ResponseWriter, r *http.Request) {
 		IndentJSON: true,
 	})
 
-	nodeID := mux.Vars(r)["nodeID"]
-	action := mux.Vars(r)["action"]
+	vars := mux.Vars(r)
+	nodeID, action := vars["nodeID"], vars["action"]
 	log.Info("receive apply action request", zap.String("nodeID", nodeID), zap.String("action", action))
 
 	if nodeID != s.ID {
-		rd.JSON(w, http.StatusOK, util.ErrResponsef("invalide nodeID %s, this pump's nodeID is %s", nodeID, s.ID))
+		rd.JSON(w, http.StatusOK, util.ErrResponsef("invalid nodeID %s, this pump's nodeID is %s", nodeID, s.ID))
 		return
 	}
 
@@ -335,7 +329,7 @@ func (s *Server) ApplyAction(w http.ResponseWriter, r *http.Request) {
 		s.status.State = node.Closing
 	default:
 		s.statusMu.Unlock()
-		rd.JSON(w, http.StatusOK, util.ErrResponsef("invalide action %s", action))
+		rd.JSON(w, http.StatusOK, util.ErrResponsef("invalid action %s", action))
 		return
 	}
 	s.statusMu.Unlock()
@@ -389,6 +383,16 @@ func (s *Server) updateStatus() error {
 	return nil
 }
 
+func (s *Server) initAPIRouter() *mux.Router {
+	router := mux.NewRouter()
+	router.HandleFunc("/status", s.collector.Status).Methods("GET")
+	router.HandleFunc("/commit_ts", s.GetLatestTS).Methods("GET")
+	router.HandleFunc("/state/{nodeID}/{action}", s.ApplyAction).Methods("PUT")
+	prometheus.DefaultGatherer = registry
+	router.Handle("/metrics", promhttp.Handler())
+	return router
+}
+
 // Close stops all goroutines started by drainer server gracefully
 func (s *Server) Close() {
 	if !atomic.CompareAndSwapInt32(&s.isClosed, 0, 1) {
diff --git a/drainer/util.go b/drainer/util.go
index b4e887aa..de590f3a 100644
--- a/drainer/util.go
+++ b/drainer/util.go
@@ -103,15 +103,10 @@ func initializeSaramaGlobalConfig() {
 }
 
 func getDDLJob(tiStore kv.Storage, id int64) (*model.Job, error) {
-	version, err := tiStore.CurrentVersion()
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	snapshot, err := tiStore.GetSnapshot(version)
+	snapMeta, err := getSnapshotMeta(tiStore)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	snapMeta := meta.NewSnapshotMeta(snapshot)
 	job, err := snapMeta.GetHistoryDDLJob(id)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -121,15 +116,10 @@ func getDDLJob(tiStore kv.Storage, id int64) (*model.Job, error) {
 
 // loadHistoryDDLJobs loads all history DDL jobs from TiDB
 func loadHistoryDDLJobs(tiStore kv.Storage) ([]*model.Job, error) {
-	version, err := tiStore.CurrentVersion()
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	snapshot, err := tiStore.GetSnapshot(version)
+	snapMeta, err := getSnapshotMeta(tiStore)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	snapMeta := meta.NewSnapshotMeta(snapshot)
 	jobs, err := snapMeta.GetAllHistoryDDLJobs()
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -143,6 +133,18 @@ func loadHistoryDDLJobs(tiStore kv.Storage) ([]*model.Job, error) {
 	return jobs, nil
 }
 
+func getSnapshotMeta(tiStore kv.Storage) (*meta.Meta, error) {
+	version, err := tiStore.CurrentVersion()
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	snapshot, err := tiStore.GetSnapshot(version)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	return meta.NewSnapshotMeta(snapshot), nil
+}
+
 func genDrainerID(listenAddr string) (string, error) {
 	urllis, err := url.Parse(listenAddr)
 	if err != nil {


Test patch:
diff --git a/drainer/collector_test.go b/drainer/collector_test.go
index 2c8d6566..c86bb434 100644
--- a/drainer/collector_test.go
+++ b/drainer/collector_test.go
@@ -38,6 +38,10 @@ func (cp dummyCheckpoint) TS() int64 {
 	return cp.commitTS
 }
 
+func (cp dummyCheckpoint) Close() error {
+	return nil
+}
+
 type dummyStore struct {
 	kv.Storage
 }
@@ -216,10 +220,8 @@ var _ = Suite(&publishSuite{})
 
 func (s *publishSuite) assertStopInTime(ctx context.Context, col *Collector, c *C) {
 	signal := make(chan struct{})
-	col.wg.Add(1)
 	go func() {
 		col.publishBinlogs(ctx)
-		col.wg.Wait()
 		close(signal)
 	}()
 
@@ -372,3 +374,92 @@ func (s *reportErrSuite) TestErrReported(c *C) {
 	col.reportErr(context.Background(), errors.New("report"))
 	c.Assert(<-col.errCh, ErrorMatches, "report")
 }
+
+type keepUpdatingSuite struct{}
+
+var _ = Suite(&keepUpdatingSuite{})
+
+func (s *keepUpdatingSuite) waitForSignal(c *C, signal <-chan struct{}, msg string) {
+	select {
+	case <-signal:
+	case <-time.After(time.Second):
+		c.Fatal(msg)
+	}
+}
+
+func (s *keepUpdatingSuite) TestShouldStopWhenDone(c *C) {
+	var callCount int
+	ctx, cancel := context.WithCancel(context.Background())
+	col := Collector{
+		merger:   &Merger{},
+		interval: 10 * time.Second,
+	}
+	signal := make(chan struct{})
+	go func() {
+		col.keepUpdatingStatus(ctx, func(ctx context.Context) error {
+			callCount++
+			return nil
+		})
+		close(signal)
+	}()
+
+	cancel()
+
+	s.waitForSignal(c, signal, "Doesn't stop in time after canceled")
+	c.Assert(callCount, Equals, 1)
+}
+
+func (s *keepUpdatingSuite) TestShouldStopOnErr(c *C) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	col := Collector{
+		merger:   &Merger{},
+		interval: time.Second,
+		errCh:    make(chan error, 1),
+	}
+	signal := make(chan struct{})
+	go func() {
+		col.keepUpdatingStatus(ctx, func(ctx context.Context) error {
+			return nil
+		})
+		close(signal)
+	}()
+	col.errCh <- errors.New("Terrible error")
+	s.waitForSignal(c, signal, "Doesn't stop in time after error occurs")
+}
+
+func (s *keepUpdatingSuite) TestShouldUpdateWhenNotified(c *C) {
+	var callCount int
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	col := Collector{
+		merger:     &Merger{},
+		notifyChan: make(chan *notifyResult),
+		interval:   10 * time.Second,
+	}
+	go col.keepUpdatingStatus(ctx, func(ctx context.Context) error {
+		defer func() {
+			callCount++
+		}()
+		if callCount%2 == 0 {
+			return errors.New("testing")
+		}
+		return nil
+	})
+	signal := make(chan struct{})
+	go func() {
+		// Start from 1 since the first call is outside the loop
+		for i := 1; i < 10; i++ {
+			err := col.Notify()
+			if i%2 == 0 {
+				c.Assert(err, ErrorMatches, "testing")
+			} else {
+				c.Assert(err, IsNil)
+			}
+		}
+		close(signal)
+	}()
+	// Wait until all notifies finish or timeout
+	s.waitForSignal(c, signal, "Notifies don't finish in time")
+	c.Assert(callCount, Equals, 10)
+}
diff --git a/drainer/config_test.go b/drainer/config_test.go
index 93513c47..349702f6 100644
--- a/drainer/config_test.go
+++ b/drainer/config_test.go
@@ -16,12 +16,19 @@ package drainer
 import (
 	"testing"
 
+	"github.com/coreos/etcd/integration"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser/mysql"
 )
 
+var testEtcdCluster *integration.ClusterV3
+
 // Hook up gocheck into the "go test" runner.
-func Test(t *testing.T) { TestingT(t) }
+func Test(t *testing.T) {
+	testEtcdCluster = integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer testEtcdCluster.Terminate(t)
+	TestingT(t)
+}
 
 var _ = Suite(&testDrainerSuite{})
 
@@ -68,6 +75,14 @@ func (t *testDrainerSuite) TestValidate(c *C) {
 	cfg.EtcdURLs = "http://127.0.0.1:2379,http://192.168.12.12:2379"
 	err = cfg.validate()
 	c.Assert(err, IsNil)
+
+	cfg.Compressor = "urada"
+	err = cfg.validate()
+	c.Assert(err, ErrorMatches, ".*Invalid compressor.*")
+
+	cfg.Compressor = "gzip"
+	err = cfg.validate()
+	c.Assert(err, IsNil)
 }
 
 func (t *testDrainerSuite) TestAdjustConfig(c *C) {
diff --git a/drainer/server_test.go b/drainer/server_test.go
index 00f2538f..4e36ef5c 100644
--- a/drainer/server_test.go
+++ b/drainer/server_test.go
@@ -14,17 +14,306 @@
 package drainer
 
 import (
+	"encoding/json"
+	"errors"
 	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
 	"os"
+	"path"
+	"testing"
+	"time"
 
+	"github.com/gorilla/mux"
 	. "github.com/pingcap/check"
+	pd "github.com/pingcap/pd/client"
+	"github.com/pingcap/tidb-binlog/pkg/etcd"
+	"github.com/pingcap/tidb-binlog/pkg/node"
+	"github.com/pingcap/tidb-binlog/pkg/security"
+	"github.com/pingcap/tidb-binlog/pkg/util"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
 )
 
 var _ = Suite(&testServerSuite{})
 
 type testServerSuite struct{}
 
-func (t *testServerSuite) TestNewServerShouldReturnErr(c *C) {
+func (t *testServerSuite) TestGetLatestTS(c *C) {
+	cp := dummyCheckpoint{commitTS: 1984}
+	server := Server{
+		syncer: &Syncer{
+			cp: &cp,
+		},
+	}
+	router := server.initAPIRouter()
+
+	req := httptest.NewRequest("GET", "/commit_ts", nil)
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+
+	resp := w.Result()
+	c.Assert(resp.StatusCode, Equals, http.StatusOK)
+
+	body, _ := ioutil.ReadAll(resp.Body)
+	var decoded util.Response
+	err := json.Unmarshal(body, &decoded)
+	c.Assert(err, IsNil)
+	c.Assert(decoded.Code, Equals, 200)
+	data, ok := decoded.Data.(map[string]interface{})
+	c.Assert(ok, IsTrue)
+	c.Assert(data, HasKey, "ts")
+	ts, ok := data["ts"].(float64)
+	c.Assert(ok, IsTrue)
+	c.Assert(int64(ts), Equals, int64(1984))
+}
+
+func (t *testServerSuite) TestNotify(c *C) {
+	server := Server{
+		collector: &Collector{
+			notifyChan: make(chan *notifyResult, 1),
+		},
+	}
+	go func() {
+		_, err := server.Notify(context.Background(), nil)
+		c.Assert(err, IsNil)
+	}()
+	select {
+	case <-server.collector.notifyChan:
+	case <-time.After(time.Second):
+		c.Fatal("Doesn't receive notify in time")
+	}
+}
+
+func (t *testServerSuite) TestClose(c *C) {
+	cli := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	reg := node.NewEtcdRegistry(cli, 5*time.Second)
+	var canceled bool
+	cp := dummyCheckpoint{}
+	server := Server{
+		status:    &node.Status{MaxCommitTS: 1024},
+		collector: &Collector{reg: reg},
+		syncer: &Syncer{
+			shutdown: make(chan struct{}),
+			closed:   make(chan struct{}),
+			cp:       &cp,
+		},
+		cp: &cp,
+		cancel: func() {
+			canceled = true
+		},
+		gs: grpc.NewServer(),
+	}
+	// Close the syncer closed signal channel manually to avoid blocking
+	close(server.syncer.closed)
+	server.Close()
+	c.Assert(canceled, IsTrue)
+	c.Assert(server.isClosed, Equals, int32(1))
+}
+
+func TestCommitStatus(t *testing.T) {
+	cli := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	reg := node.NewEtcdRegistry(cli, 5*time.Second)
+
+	cases := []struct {
+		input    string
+		expected string
+		ts       int64
+	}{
+		{input: node.Pausing, expected: node.Paused, ts: 2019},
+		{input: node.Closing, expected: node.Offline, ts: 2042},
+	}
+	for _, cs := range cases {
+		status := node.Status{State: cs.input}
+		s := Server{
+			status:    &status,
+			collector: &Collector{reg: reg},
+			syncer: &Syncer{
+				cp: &dummyCheckpoint{commitTS: cs.ts},
+			},
+		}
+		s.commitStatus()
+		if s.status.State != cs.expected {
+			t.Fatalf("Invalid state after commit: expect %s, get %s", cs.expected, s.status.State)
+		}
+		if s.status.MaxCommitTS != cs.ts {
+			t.Fatalf("MaxCommitTS not updated correctly: expect %d, get %d", cs.ts, s.status.MaxCommitTS)
+		}
+	}
+}
+
+type applyActionSuite struct {
+	server Server
+	router *mux.Router
+}
+
+var _ = Suite(&applyActionSuite{})
+
+func (s *applyActionSuite) SetUpTest(c *C) {
+	s.server = Server{
+		status: &node.Status{State: node.Online},
+		ID:     "drainer-18",
+		// Set `isClosed` to avoid actually running the `Close` method,
+		// which involves a lot of components that's not directly relevant
+		// to this test and difficult to mock.
+		isClosed: 1,
+	}
+	s.router = s.server.initAPIRouter()
+}
+
+func (s *applyActionSuite) TestShouldCheckNodeID(c *C) {
+	req := httptest.NewRequest("PUT", "/state/drainer-42/pause", nil)
+	w := httptest.NewRecorder()
+
+	s.router.ServeHTTP(w, req)
+
+	resp := w.Result()
+	body, _ := ioutil.ReadAll(resp.Body)
+	c.Assert(resp.StatusCode, Equals, http.StatusOK)
+
+	var decoded util.Response
+	err := json.Unmarshal(body, &decoded)
+	c.Assert(err, IsNil)
+	c.Assert(decoded.Code, Equals, 3)
+	c.Assert(decoded.Message, Matches, ".*invalid nodeID drainer-42.*")
+}
+
+func (s *applyActionSuite) TestShouldCheckState(c *C) {
+	s.server.status.State = node.Paused
+
+	req := httptest.NewRequest("PUT", "/state/drainer-18/close", nil)
+	w := httptest.NewRecorder()
+
+	s.router.ServeHTTP(w, req)
+
+	resp := w.Result()
+	body, _ := ioutil.ReadAll(resp.Body)
+	c.Assert(resp.StatusCode, Equals, http.StatusOK)
+
+	var decoded util.Response
+	err := json.Unmarshal(body, &decoded)
+	c.Assert(err, IsNil)
+	c.Assert(decoded.Code, Equals, 3)
+	c.Assert(decoded.Message, Matches, ".*apply close failed!.*")
+}
+
+func (s *applyActionSuite) TestShouldApplyAction(c *C) {
+	cases := map[string]string{
+		"pause": node.Pausing,
+		"close": node.Closing,
+		"run":   "",
+	}
+	for input, expect := range cases {
+		s.server.status.State = node.Online
+		req := httptest.NewRequest("PUT", "/state/drainer-18/"+input, nil)
+		w := httptest.NewRecorder()
+
+		s.router.ServeHTTP(w, req)
+
+		resp := w.Result()
+		body, _ := ioutil.ReadAll(resp.Body)
+
+		c.Assert(resp.StatusCode, Equals, http.StatusOK)
+
+		var decoded util.Response
+		err := json.Unmarshal(body, &decoded)
+		c.Assert(err, IsNil)
+		if expect != "" {
+			c.Assert(decoded.Code, Equals, 200)
+			c.Assert(decoded.Message, Matches, ".*success.*")
+			c.Assert(s.server.status.State, Equals, expect)
+		} else {
+			c.Assert(decoded.Code, Equals, 3)
+			c.Assert(decoded.Message, Matches, ".*invalid action.*")
+			c.Assert(s.server.status.State, Equals, node.Online) // State unchanged
+		}
+	}
+}
+
+type heartbeatSuite struct{}
+
+var _ = Suite(&heartbeatSuite{})
+
+func (s *heartbeatSuite) TestShouldStopWhenDone(c *C) {
+	cli := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	reg := node.NewEtcdRegistry(cli, 5*time.Second)
+
+	server := Server{
+		status:    &node.Status{},
+		collector: &Collector{reg: reg},
+		syncer: &Syncer{
+			cp: &dummyCheckpoint{},
+		},
+		isClosed: 1, // Hack to avoid actually running close
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	errc := server.heartbeat(ctx)
+	cancel()
+	select {
+	case <-errc:
+	case <-time.After(time.Second):
+		c.Fatal("Doesn't stop in time")
+	}
+}
+
+func (s *heartbeatSuite) TestShouldUpdateStatusPeriodically(c *C) {
+	origInterval := heartbeatInterval
+	heartbeatInterval = time.Millisecond
+	defer func() {
+		heartbeatInterval = origInterval
+	}()
+	cli := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	reg := node.NewEtcdRegistry(cli, 5*time.Second)
+
+	cp := dummyCheckpoint{commitTS: 1024}
+	server := Server{
+		status:    &node.Status{MaxCommitTS: 1024},
+		collector: &Collector{reg: reg},
+		syncer: &Syncer{
+			cp: &cp,
+		},
+		isClosed: 1, // Hack to avoid actually running close
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	_ = server.heartbeat(ctx)
+	c.Assert(server.status.MaxCommitTS, Equals, int64(1024))
+	cp.commitTS = 100200
+	time.Sleep(heartbeatInterval * 3)
+	c.Assert(server.status.MaxCommitTS, Equals, int64(100200))
+}
+
+type mockPdCli struct {
+	pd.Client
+}
+
+func (pc *mockPdCli) GetClusterID(ctx context.Context) uint64 {
+	return 8012
+}
+
+func (pc *mockPdCli) GetTS(ctx context.Context) (int64, int64, error) {
+	return 112233, 12, nil
+}
+
+func (pc *mockPdCli) Close() {}
+
+type newServerSuite struct {
+	origGetPdCli func(string, security.Config) (pd.Client, error)
+}
+
+var _ = Suite(&newServerSuite{})
+
+func (s *newServerSuite) SetUpTest(c *C) {
+	s.origGetPdCli = getPdClient
+}
+
+func (s *newServerSuite) TearDownTest(c *C) {
+	getPdClient = s.origGetPdCli
+}
+
+func (s *newServerSuite) TestCannotCreateDataDir(c *C) {
 	tmpfile, err := ioutil.TempFile("", "test")
 	if err != nil {
 		c.Fatal("Failed to create temp file.")
@@ -37,3 +326,28 @@ func (t *testServerSuite) TestNewServerShouldReturnErr(c *C) {
 	_, err = NewServer(cfg)
 	c.Assert(err, ErrorMatches, ".*mkdir.*")
 }
+
+func (s *newServerSuite) TestCannotGetPdCli(c *C) {
+	getPdClient = func(etcdURLs string, securityConfig security.Config) (pd.Client, error) {
+		return nil, errors.New("Pd")
+	}
+	cfg := NewConfig()
+	cfg.DataDir = path.Join(c.MkDir(), "drainer")
+	cfg.ListenAddr = "http://" + cfg.ListenAddr
+	_, err := NewServer(cfg)
+	c.Assert(err, ErrorMatches, "Pd")
+}
+
+func (s *newServerSuite) TestInvalidDestDBType(c *C) {
+	getPdClient = func(etcdURLs string, securityConfig security.Config) (pd.Client, error) {
+		return &mockPdCli{}, nil
+	}
+	cfg := NewConfig()
+	cfg.DataDir = path.Join(c.MkDir(), "drainer")
+	cfg.ListenAddr = "http://" + cfg.ListenAddr
+	cfg.SyncerCfg.DestDBType = "nothing"
+	cfg.adjustConfig()
+	_, err := NewServer(cfg)
+	c.Assert(err, ErrorMatches, ".*unsupported checkpoint type.*")
+	c.Assert(cfg.SyncerCfg.To.ClusterID, Equals, uint64(8012))
+}
diff --git a/drainer/util_test.go b/drainer/util_test.go
index d0e874ce..26fb4795 100644
--- a/drainer/util_test.go
+++ b/drainer/util_test.go
@@ -14,36 +14,10 @@
 package drainer
 
 import (
-	"errors"
-
-	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
-	"github.com/pingcap/tidb-binlog/pkg/sql"
 	"github.com/pingcap/tidb-binlog/pkg/util"
 )
 
-func (t *testDrainerSuite) TestIgnoreDDLError(c *C) {
-	// test non-mysqltype error
-	err := errors.New("test")
-	ok := sql.IgnoreDDLError(err)
-	c.Assert(ok, IsFalse)
-	// test ignore error
-	err1 := &mysql.MySQLError{
-		Number:  1054,
-		Message: "test",
-	}
-	ok = sql.IgnoreDDLError(err1)
-	c.Assert(ok, IsTrue)
-	// test non-ignore error
-	err2 := &mysql.MySQLError{
-		Number:  1052,
-		Message: "test",
-	}
-	ok = sql.IgnoreDDLError(err2)
-	c.Assert(ok, IsFalse)
-
-}
-
 type taskGroupSuite struct{}
 
 var _ = Suite(&taskGroupSuite{})