Merge branch 'master' into add-test

pkg/logutil/log.go

@@ -173,7 +173,8 @@ func (f *textFormatter) Format(entry *log.Entry) ([]byte, error) {
 	return b.Bytes(), nil
 }
 
-func stringToLogFormatter(format string, disableTimestamp bool) log.Formatter {
+// StringToLogFormatter uses the different log formatter according to a given format name.
+func StringToLogFormatter(format string, disableTimestamp bool) log.Formatter {
 	switch strings.ToLower(format) {
 	case "text":
 		return &textFormatter{
@@ -248,7 +249,7 @@ func InitLogger(cfg *LogConfig) error {
 		if cfg.Format == "" {
 			cfg.Format = defaultLogFormat
 		}
-		log.SetFormatter(stringToLogFormatter(cfg.Format, cfg.DisableTimestamp))
+		log.SetFormatter(StringToLogFormatter(cfg.Format, cfg.DisableTimestamp))
 
 		// etcd log
 		capnslog.SetFormatter(&redirectFormatter{})

pkg/logutil/log_test.go

@@ -51,6 +51,22 @@ func (s *testLogSuite) TestStringToLogLevel(c *C) {
 	c.Assert(StringToLogLevel("whatever"), Equals, log.InfoLevel)
 }
 
+func (s *testLogSuite) TestStringToLogFormatter(c *C) {
+	c.Assert(StringToLogFormatter("text", true), DeepEquals, &textFormatter{
+		DisableTimestamp: true,
+	})
+	c.Assert(StringToLogFormatter("json", true), DeepEquals, &log.JSONFormatter{
+		DisableTimestamp: true,
+		TimestampFormat:  defaultLogTimeFormat,
+	})
+	c.Assert(StringToLogFormatter("console", true), DeepEquals, &log.TextFormatter{
+		DisableTimestamp: true,
+		FullTimestamp:    true,
+		TimestampFormat:  defaultLogTimeFormat,
+	})
+	c.Assert(StringToLogFormatter("", true), DeepEquals, &textFormatter{})
+}
+
 // TestLogging assure log format and log redirection works.
 func (s *testLogSuite) TestLogging(c *C) {
 	conf := &LogConfig{Level: "warn", File: FileLogConfig{}}
@@ -76,3 +92,8 @@ func (s *testLogSuite) TestLogging(c *C) {
 	c.Assert(entry, Matches, logPattern)
 	c.Assert(strings.Contains(entry, "log_test.go"), IsTrue)
 }
+
+func (s *testLogSuite) TestFileLog(c *C) {
+	c.Assert(InitFileLog(&FileLogConfig{Filename: "/tmp"}), NotNil)
+	c.Assert(InitFileLog(&FileLogConfig{Filename: "/tmp/test_file_log", MaxSize: 0}), IsNil)
+}

server/api/hot_status.go

@@ -25,7 +25,8 @@ type hotStatusHandler struct {
 	rd *render.Render
 }
 
-type hotStoreStats struct {
+// HotStoreStats is used to record the status of hot stores.
+type HotStoreStats struct {
 	BytesWriteStats map[uint64]uint64 `json:"bytes-write-rate,omitempty"`
 	BytesReadStats  map[uint64]uint64 `json:"bytes-read-rate,omitempty"`
 	KeysWriteStats  map[uint64]uint64 `json:"keys-write-rate,omitempty"`
@@ -53,7 +54,7 @@ func (h *hotStatusHandler) GetHotStores(w http.ResponseWriter, r *http.Request)
 	keysWriteStats := h.GetHotKeysWriteStores()
 	keysReadStats := h.GetHotKeysWriteStores()
 
-	stats := hotStoreStats{
+	stats := HotStoreStats{
 		BytesWriteStats: bytesWriteStats,
 		BytesReadStats:  bytesReadStats,
 		KeysWriteStats:  keysWriteStats,

server/api/hot_status_test.go

@@ -45,7 +45,7 @@ func (s *testHotStatusSuite) TearDownSuite(c *C) {
 }
 
 func (s testHotStatusSuite) TestGetHotStore(c *C) {
-	stat := hotStoreStats{}
+	stat := HotStoreStats{}
 	resp, err := http.Get(s.urlPrefix + "/stores")
 	c.Assert(err, IsNil)
 	err = readJSON(resp.Body, &stat)

server/schedulers/scheduler_test.go

@@ -274,3 +274,58 @@ func (s *testRejectLeaderSuite) TestRejectLeader(c *C) {
 	op = sl.Schedule(tc)
 	testutil.CheckTransferLeader(c, op[0], schedule.OpLeader, 1, 2)
 }
+
+var _ = Suite(&testEvictLeaderSuite{})
+
+type testEvictLeaderSuite struct{}
+
+func (s *testEvictLeaderSuite) TestEvictLeader(c *C) {
+	opt := schedule.NewMockSchedulerOptions()
+	tc := schedule.NewMockCluster(opt)
+
+	// Add stores 1, 2, 3
+	tc.AddLeaderStore(1, 0)
+	tc.AddLeaderStore(2, 0)
+	tc.AddLeaderStore(3, 0)
+	// Add regions 1, 2, 3 with leaders in stores 1, 2, 3
+	tc.AddLeaderRegion(1, 1, 2)
+	tc.AddLeaderRegion(2, 2, 1)
+	tc.AddLeaderRegion(3, 3, 1)
+
+	sl, err := schedule.CreateScheduler("evict-leader", schedule.NewOperatorController(nil, nil), "1")
+	c.Assert(err, IsNil)
+	c.Assert(sl.IsScheduleAllowed(tc), IsTrue)
+	op := sl.Schedule(tc)
+	testutil.CheckTransferLeader(c, op[0], schedule.OpLeader, 1, 2)
+}
+
+var _ = Suite(&testShuffleRegionSuite{})
+
+type testShuffleRegionSuite struct{}
+
+func (s *testShuffleRegionSuite) TestShuffle(c *C) {
+	opt := schedule.NewMockSchedulerOptions()
+	tc := schedule.NewMockCluster(opt)
+
+	sl, err := schedule.CreateScheduler("shuffle-region", schedule.NewOperatorController(nil, nil))
+	c.Assert(err, IsNil)
+	c.Assert(sl.IsScheduleAllowed(tc), IsTrue)
+	c.Assert(sl.Schedule(tc), IsNil)
+
+	// Add stores 1, 2, 3, 4
+	tc.AddRegionStore(1, 6)
+	tc.AddRegionStore(2, 7)
+	tc.AddRegionStore(3, 8)
+	tc.AddRegionStore(4, 9)
+	// Add regions 1, 2, 3, 4 with leaders in stores 1,2,3,4
+	tc.AddLeaderRegion(1, 1, 2, 3)
+	tc.AddLeaderRegion(2, 2, 3, 4)
+	tc.AddLeaderRegion(3, 3, 4, 1)
+	tc.AddLeaderRegion(4, 4, 1, 2)
+
+	for i := 0; i < 4; i++ {
+		op := sl.Schedule(tc)
+		c.Assert(op, NotNil)
+		c.Assert(op[0].Kind(), Equals, schedule.OpRegion|schedule.OpAdmin)
+	}
+}

tests/cluster.go

@@ -195,6 +195,13 @@ func (s *TestServer) GetStores() []*metapb.Store {
 	return s.server.GetRaftCluster().GetStores()
 }
 
+// GetStore returns the store with a given store ID.
+func (s *TestServer) GetStore(storeID uint64) (*core.StoreInfo, error) {
+	s.RLock()
+	defer s.RUnlock()
+	return s.server.GetRaftCluster().GetStore(storeID)
+}
+
 // GetRaftCluster returns Raft cluster.
 // If cluster has not been bootstrapped, return nil.
 func (s *TestServer) GetRaftCluster() *server.RaftCluster {

tests/cmd/pdctl_test.go

@@ -28,6 +28,7 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
+	"github.com/pingcap/pd/pkg/etcdutil"
 	"github.com/pingcap/pd/server"
 	"github.com/pingcap/pd/server/api"
 	"github.com/pingcap/pd/server/core"
@@ -52,7 +53,7 @@ func (s *cmdTestSuite) SetUpSuite(c *C) {
 	server.EnableZap = true
 }
 
-func (s *cmdTestSuite) TestCluster(c *C) {
+func (s *cmdTestSuite) TestClusterAndPing(c *C) {
 	c.Parallel()
 
 	cluster, err := tests.NewTestCluster(1)
@@ -64,20 +65,27 @@ func (s *cmdTestSuite) TestCluster(c *C) {
 	cmd := initCommand()
 	defer cluster.Destroy()
 
-	// cluster command
+	// cluster
 	args := []string{"-u", pdAddr, "cluster"}
 	_, output, err := executeCommandC(cmd, args...)
 	c.Assert(err, IsNil)
 	ci := &metapb.Cluster{}
 	c.Assert(json.Unmarshal(output, ci), IsNil)
 	c.Assert(ci, DeepEquals, cluster.GetCluster())
 
+	// cluster status
 	args = []string{"-u", pdAddr, "cluster", "status"}
 	_, output, err = executeCommandC(cmd, args...)
 	c.Assert(err, IsNil)
 	ci = &metapb.Cluster{}
 	c.Assert(json.Unmarshal(output, ci), IsNil)
 	c.Assert(ci, DeepEquals, cluster.GetCluster())
+
+	// ping
+	args = []string{"-u", pdAddr, "ping"}
+	_, output, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	c.Assert(output, NotNil)
 }
 
 func (s *cmdTestSuite) TestHealth(c *C) {
@@ -875,7 +883,12 @@ func (s *cmdTestSuite) TestOperator(c *C) {
 	c.Parallel()
 
 	var err error
-	cluster, err := tests.NewTestCluster(3, func(conf *server.Config) { conf.Replication.MaxReplicas = 2 })
+	var t time.Time
+	t = t.Add(time.Hour)
+	cluster, err := tests.NewTestCluster(3,
+		func(conf *server.Config) { conf.Replication.MaxReplicas = 2 },
+		func(conf *server.Config) { conf.Schedule.MaxStoreDownTime.Duration = time.Since(t) },
+	)
 	c.Assert(err, IsNil)
 	err = cluster.RunInitialServers()
 	c.Assert(err, IsNil)
@@ -995,6 +1008,136 @@ func (s *cmdTestSuite) TestOperator(c *C) {
 	args = []string{"-u", pdAddr, "operator", "remove", "3"}
 	_, _, err = executeCommandC(cmd, args...)
 	c.Assert(err, IsNil)
+
+	// operator add scatter-region <region_id>
+	args = []string{"-u", pdAddr, "operator", "add", "scatter-region", "3"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	args = []string{"-u", pdAddr, "operator", "add", "scatter-region", "1"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	args = []string{"-u", pdAddr, "operator", "show", "region"}
+	_, output, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	c.Assert(strings.Contains(string(output), "transfer leader from store 0 to store 3"), IsTrue)
+}
+
+func (s *cmdTestSuite) TestMember(c *C) {
+	c.Parallel()
+
+	cluster, err := tests.NewTestCluster(3)
+	c.Assert(err, IsNil)
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+	pdAddr := cluster.GetConfig().GetClientURLs()
+	c.Assert(err, IsNil)
+	cmd := initCommand()
+	svr := cluster.GetServer("pd2")
+	id := svr.GetServerID()
+	name := svr.GetServer().Name()
+	client := cluster.GetServer("pd1").GetEtcdClient()
+	defer cluster.Destroy()
+
+	// member leader show
+	args := []string{"-u", pdAddr, "member", "leader", "show"}
+	_, output, err := executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	leader := pdpb.Member{}
+	c.Assert(json.Unmarshal(output, &leader), IsNil)
+	c.Assert(&leader, DeepEquals, svr.GetLeader())
+
+	// member leader transfer <member_name>
+	args = []string{"-u", pdAddr, "member", "leader", "transfer", "pd2"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	c.Assert("pd2", Equals, svr.GetLeader().GetName())
+
+	// member leader resign
+	args = []string{"-u", pdAddr, "member", "leader", "resign"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	c.Assert("pd2", Not(Equals), svr.GetLeader().GetName())
+
+	// member leader_priority <member_name> <priority>
+	args = []string{"-u", pdAddr, "member", "leader_priority", name, "100"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	priority, err := svr.GetServer().GetMemberLeaderPriority(id)
+	c.Assert(err, IsNil)
+	c.Assert(priority, Equals, 100)
+
+	// member delete name <member_name>
+	svr.Destroy()
+	members, err := etcdutil.ListEtcdMembers(client)
+	c.Assert(err, IsNil)
+	c.Assert(len(members.Members), Equals, 3)
+	args = []string{"-u", pdAddr, "member", "delete", "name", name}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	members, err = etcdutil.ListEtcdMembers(client)
+	c.Assert(err, IsNil)
+	c.Assert(len(members.Members), Equals, 2)
+
+	// member delete id <member_id>
+	args = []string{"-u", pdAddr, "member", "delete", "id", string(id)}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	members, err = etcdutil.ListEtcdMembers(client)
+	c.Assert(err, IsNil)
+	c.Assert(len(members.Members), Equals, 2)
+}
+
+func (s *cmdTestSuite) TestHot(c *C) {
+	c.Parallel()
+
+	cluster, err := tests.NewTestCluster(1)
+	c.Assert(err, IsNil)
+	err = cluster.RunInitialServers()
+	c.Assert(err, IsNil)
+	cluster.WaitLeader()
+	pdAddr := cluster.GetConfig().GetClientURLs()
+	cmd := initCommand()
+
+	store := metapb.Store{
+		Id:      1,
+		Address: "tikv1",
+		State:   metapb.StoreState_Up,
+		Version: "2.0.0",
+	}
+
+	leaderServer := cluster.GetServer(cluster.GetLeader())
+	c.Assert(leaderServer.BootstrapCluster(), IsNil)
+	mustPutStore(c, leaderServer.GetServer(), store.Id, store.State, store.Labels)
+	defer cluster.Destroy()
+
+	ss, err := leaderServer.GetStore(1)
+	c.Assert(err, IsNil)
+	bytesWritten := uint64(8 * 1024 * 1024)
+	ss.Stats.BytesWritten = bytesWritten
+	now := time.Now().Second()
+	interval := &pdpb.TimeInterval{StartTimestamp: uint64(now - 10), EndTimestamp: uint64(now)}
+	ss.Stats.Interval = interval
+	ss.RollingStoreStats.Observe(ss.Stats)
+
+	// TODO: Provide a way to test the result of hot read and hot write commands
+	// hot read
+	args := []string{"-u", pdAddr, "hot", "read"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+
+	// hot write
+	args = []string{"-u", pdAddr, "hot", "write"}
+	_, _, err = executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+
+	// hot store
+	args = []string{"-u", pdAddr, "hot", "store"}
+	_, output, err := executeCommandC(cmd, args...)
+	c.Assert(err, IsNil)
+	hotStores := api.HotStoreStats{}
+	c.Assert(json.Unmarshal(output, &hotStores), IsNil)
+	c.Assert(hotStores.BytesWriteStats[1], Equals, bytesWritten/10)
 }
 
 func initCommand() *cobra.Command {