*: handle more errors properly (#1405)

* handle errors properly

Signed-off-by: rleungx <rleungx@gmail.com>

client/client.go

@@ -23,7 +23,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/opentracing/opentracing-go"
+	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pkg/errors"
@@ -363,6 +363,7 @@ func (c *client) tsLoop() {
 			if err != nil {
 				select {
 				case <-loopCtx.Done():
+					cancel()
 					return
 				default:
 				}
@@ -395,6 +396,7 @@ func (c *client) tsLoop() {
 			select {
 			case c.tsDeadlineCh <- dl:
 			case <-loopCtx.Done():
+				cancel()
 				return
 			}
 			opts = extractSpanReference(requests, opts[:0])
@@ -409,6 +411,7 @@ func (c *client) tsLoop() {
 		if err != nil {
 			select {
 			case <-loopCtx.Done():
+				cancel()
 				return
 			default:
 			}

client/client_test.go

@@ -63,7 +63,7 @@ type testClientSuite struct {
 
 func (s *testClientSuite) SetUpSuite(c *C) {
 	var err error
-	_, s.srv, s.cleanup, err = server.NewTestServer()
+	_, s.srv, s.cleanup, err = server.NewTestServer(c)
 	c.Assert(err, IsNil)
 	s.grpcPDClient = mustNewGrpcClient(c, s.srv.GetAddr())
 

server/api/cluster_test.go

@@ -50,7 +50,7 @@ func (s *testClusterInfo) TestCluster(c *C) {
 
 	c2 := &metapb.Cluster{}
 	r := server.ReplicationConfig{MaxReplicas: 6}
-	s.svr.SetReplicationConfig(r)
+	c.Assert(s.svr.SetReplicationConfig(r), IsNil)
 	err = readJSONWithURL(url, c2)
 	c.Assert(err, IsNil)
 

server/api/config.go

@@ -104,7 +104,10 @@ func (h *confHandler) SetReplication(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	h.svr.SetReplicationConfig(*config)
+	if err := h.svr.SetReplicationConfig(*config); err != nil {
+		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
+		return
+	}
 	h.rd.JSON(w, http.StatusOK, nil)
 }
 
@@ -136,7 +139,10 @@ func (h *confHandler) SetNamespace(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	h.svr.SetNamespaceConfig(name, *config)
+	if err := h.svr.SetNamespaceConfig(name, *config); err != nil {
+		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
+		return
+	}
 	h.rd.JSON(w, http.StatusOK, nil)
 }
 
@@ -148,8 +154,11 @@ func (h *confHandler) DeleteNamespace(w http.ResponseWriter, r *http.Request) {
 		h.rd.JSON(w, http.StatusNotFound, fmt.Sprintf("invalid namespace Name %s, not found", name))
 		return
 	}
-	h.svr.DeleteNamespaceConfig(name)
 
+	if err := h.svr.DeleteNamespaceConfig(name); err != nil {
+		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
+		return
+	}
 	h.rd.JSON(w, http.StatusOK, nil)
 }
 

server/api/config_test.go

@@ -76,7 +76,7 @@ func (s *testConfigSuite) TestConfigSchedule(c *C) {
 	resp, err := doGet(addr)
 	c.Assert(err, IsNil)
 	sc := &server.ScheduleConfig{}
-	readJSON(resp.Body, sc)
+	c.Assert(readJSON(resp.Body, sc), IsNil)
 
 	sc.MaxStoreDownTime.Duration = time.Second
 	postData, err := json.Marshal(sc)
@@ -88,7 +88,7 @@ func (s *testConfigSuite) TestConfigSchedule(c *C) {
 	resp, err = doGet(addr)
 	c.Assert(err, IsNil)
 	sc1 := &server.ScheduleConfig{}
-	readJSON(resp.Body, sc1)
+	c.Assert(readJSON(resp.Body, sc1), IsNil)
 
 	c.Assert(*sc, DeepEquals, *sc1)
 }

server/api/diagnose.go

@@ -38,19 +38,15 @@ type Recommendation struct {
 //lint:file-ignore U1000 document available levels and modules
 const (
 	// analyze levels
-	levelNormal   = "Normal"
 	levelWarning  = "Warning"
 	levelMinor    = "Minor"
 	levelMajor    = "Major"
 	levelCritical = "Critical"
-	levelFatal    = "Fatal"
 
 	// analyze modules
-	modMember   = "member"
-	modTiKV     = "TiKV"
-	modReplica  = "Replic"
-	modSchedule = "Schedule"
-	modDefault  = "Default"
+	modMember  = "member"
+	modTiKV    = "TiKV"
+	modDefault = "Default"
 
 	memberOneInstance diagnoseType = iota
 	memberEvenInstance

server/api/diagnose_test.go

@@ -34,7 +34,7 @@ func (s *testDiagnoseAPISuite) SetUpSuite(c *C) {
 
 func checkDiagnoseResponse(c *C, body []byte) {
 	got := []Recommendation{}
-	json.Unmarshal(body, &got)
+	c.Assert(json.Unmarshal(body, &got), IsNil)
 	for _, r := range got {
 		c.Assert(len(r.Module) != 0, IsTrue)
 		c.Assert(len(r.Level) != 0, IsTrue)

server/api/health_test.go

@@ -35,7 +35,7 @@ func (s *testHealthAPISuite) SetUpSuite(c *C) {
 
 func checkSliceResponse(c *C, body []byte, cfgs []*server.Config, unhealth string) {
 	got := []Health{}
-	json.Unmarshal(body, &got)
+	c.Assert(json.Unmarshal(body, &got), IsNil)
 
 	c.Assert(len(got), Equals, len(cfgs))
 

server/api/member_test.go

@@ -93,7 +93,7 @@ func (s *testMemberAPISuite) TestMemberLeader(c *C) {
 	c.Assert(err, IsNil)
 
 	var got pdpb.Member
-	json.Unmarshal(buf, &got)
+	c.Assert(json.Unmarshal(buf, &got), IsNil)
 	c.Assert(got.GetClientUrls(), DeepEquals, leader.GetClientUrls())
 	c.Assert(got.GetMemberId(), Equals, leader.GetMemberId())
 }

server/api/server_test.go

@@ -75,7 +75,7 @@ func mustNewServer(c *C) (*server.Server, cleanUpFunc) {
 
 func mustNewCluster(c *C, num int) ([]*server.Config, []*server.Server, cleanUpFunc) {
 	svrs := make([]*server.Server, 0, num)
-	cfgs := server.NewTestMultiConfig(num)
+	cfgs := server.NewTestMultiConfig(c, num)
 
 	ch := make(chan *server.Server, num)
 	for _, cfg := range cfgs {

server/api/status_test.go

@@ -34,7 +34,7 @@ func (s *testStatusAPISuite) SetUpSuite(c *C) {
 
 func checkStatusResponse(c *C, body []byte, cfgs []*server.Config) {
 	got := status{}
-	json.Unmarshal(body, &got)
+	c.Assert(json.Unmarshal(body, &got), IsNil)
 
 	c.Assert(got.BuildTS, Equals, server.PDBuildTS)
 	c.Assert(got.GitHash, Equals, server.PDGitHash)

server/api/store_ns_test.go

@@ -60,7 +60,7 @@ func (s *testStoreNsSuite) SetUpSuite(c *C) {
 		},
 	}
 
-	cfg := server.NewTestSingleConfig()
+	cfg := server.NewTestSingleConfig(c)
 	cfg.NamespaceClassifier = "table"
 	svr, err := server.CreateServer(cfg, NewHandler)
 	c.Assert(err, IsNil)

server/api/trend_test.go

@@ -46,9 +46,9 @@ func (s *testTrendSuite) TestTrend(c *C) {
 	mustRegionHeartbeat(c, svr, region6)
 
 	// Create 3 operators that transfers leader, moves follower, moves leader.
-	svr.GetHandler().AddTransferLeaderOperator(4, 2)
-	svr.GetHandler().AddTransferPeerOperator(5, 2, 3)
-	svr.GetHandler().AddTransferPeerOperator(6, 1, 3)
+	c.Assert(svr.GetHandler().AddTransferLeaderOperator(4, 2), IsNil)
+	c.Assert(svr.GetHandler().AddTransferPeerOperator(5, 2, 3), IsNil)
+	c.Assert(svr.GetHandler().AddTransferPeerOperator(6, 1, 3), IsNil)
 
 	// Complete the operators.
 	mustRegionHeartbeat(c, svr, region4.Clone(core.WithLeader(region4.GetStorePeer(2))))

server/cluster_info.go

@@ -190,7 +190,8 @@ func (c *clusterInfo) putStoreLocked(store *core.StoreInfo) error {
 			return err
 		}
 	}
-	return c.core.PutStore(store)
+	c.core.PutStore(store)
+	return nil
 }
 
 // BlockStore stops balancer from selecting the store.
@@ -313,7 +314,8 @@ func (c *clusterInfo) putRegionLocked(region *core.RegionInfo) error {
 			return err
 		}
 	}
-	return c.core.PutRegion(region)
+	c.core.PutRegion(region)
+	return nil
 }
 
 func (c *clusterInfo) getRegions() []*core.RegionInfo {

server/cluster_info_test.go

@@ -264,7 +264,8 @@ func (s *testClusterInfoSuite) TestLoadClusterInfo(c *C) {
 	defer cleanup()
 
 	kv := server.kv
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 
 	// Cluster is not bootstrapped.
 	cluster, err := loadClusterInfo(server.idAlloc, kv, opt)
@@ -295,7 +296,8 @@ func (s *testClusterInfoSuite) TestLoadClusterInfo(c *C) {
 }
 
 func (s *testClusterInfoSuite) TestStoreHeartbeat(c *C) {
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 	cluster := newClusterInfo(core.NewMockIDAllocator(), opt, core.NewKV(core.NewMemoryKV()))
 
 	n, np := uint64(3), uint64(3)
@@ -340,7 +342,8 @@ func (s *testClusterInfoSuite) TestStoreHeartbeat(c *C) {
 }
 
 func (s *testClusterInfoSuite) TestRegionHeartbeat(c *C) {
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 	cluster := newClusterInfo(core.NewMockIDAllocator(), opt, core.NewKV(core.NewMemoryKV()))
 
 	n, np := uint64(3), uint64(3)
@@ -349,7 +352,7 @@ func (s *testClusterInfoSuite) TestRegionHeartbeat(c *C) {
 	regions := newTestRegions(n, np)
 
 	for _, store := range stores {
-		cluster.putStore(store)
+		c.Assert(cluster.putStore(store), IsNil)
 	}
 
 	for i, region := range regions {
@@ -554,7 +557,8 @@ func heartbeatRegions(c *C, cluster *clusterInfo, regions []*metapb.Region) {
 }
 
 func (s *testClusterInfoSuite) TestHeartbeatSplit(c *C) {
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 	cluster := newClusterInfo(core.NewMockIDAllocator(), opt, nil)
 
 	// 1: [nil, nil)
@@ -592,7 +596,8 @@ func (s *testClusterInfoSuite) TestHeartbeatSplit(c *C) {
 }
 
 func (s *testClusterInfoSuite) TestRegionSplitAndMerge(c *C) {
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 	cluster := newClusterInfo(core.NewMockIDAllocator(), opt, nil)
 
 	regions := []*metapb.Region{
@@ -631,11 +636,12 @@ func (s *testClusterInfoSuite) TestRegionSplitAndMerge(c *C) {
 }
 
 func (s *testClusterInfoSuite) TestUpdateStorePendingPeerCount(c *C) {
-	_, opt := newTestScheduleConfig()
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
 	tc := newTestClusterInfo(opt)
 	stores := newTestStores(5)
 	for _, s := range stores {
-		tc.putStore(s)
+		c.Assert(tc.putStore(s), IsNil)
 	}
 	peers := []*metapb.Peer{
 		{
@@ -656,10 +662,10 @@ func (s *testClusterInfoSuite) TestUpdateStorePendingPeerCount(c *C) {
 		},
 	}
 	origin := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: peers[:3]}, peers[0], core.WithPendingPeers(peers[1:3]))
-	tc.handleRegionHeartbeat(origin)
+	c.Assert(tc.handleRegionHeartbeat(origin), IsNil)
 	checkPendingPeerCount([]int{0, 1, 1, 0}, tc.clusterInfo, c)
 	newRegion := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: peers[1:]}, peers[1], core.WithPendingPeers(peers[3:4]))
-	tc.handleRegionHeartbeat(newRegion)
+	c.Assert(tc.handleRegionHeartbeat(newRegion), IsNil)
 	checkPendingPeerCount([]int{0, 0, 0, 1}, tc.clusterInfo, c)
 }