Skip to content

Commit

Permalink
command:add hot-history-regions command(ref #pingcap/tidb/issues/25281)
Browse files Browse the repository at this point in the history
Signed-off-by: qidi1 <1083369179@qq.com>
  • Loading branch information
qidi1 committed Nov 29, 2021
1 parent eb10652 commit d560865
Show file tree
Hide file tree
Showing 3 changed files with 283 additions and 33 deletions.
81 changes: 81 additions & 0 deletions tests/pdctl/hot/hot_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package hot_test
import (
"context"
"encoding/json"
"strconv"
"testing"
"time"

Expand Down Expand Up @@ -219,3 +220,83 @@ func (s *hotTestSuite) TestHotWithStoreID(c *C) {
c.Assert(hotRegion.AsLeader[1].TotalBytesRate, Equals, float64(200000000))
c.Assert(hotRegion.AsLeader[2].TotalBytesRate, Equals, float64(100000000))
}

func (s *hotTestSuite) TestHistoryHotRegions(c *C) {
statistics.Denoising = false
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cluster, err := tests.NewTestCluster(ctx, 1,
func(cfg *config.Config, serverName string) {
cfg.Schedule.HotRegionCacheHitsThreshold = 0
cfg.Schedule.HotRegionsWriteInterval.Duration = 1000 * time.Millisecond
cfg.Schedule.HotRegionsReservedDays = 1
},
)
c.Assert(err, IsNil)
err = cluster.RunInitialServers()
c.Assert(err, IsNil)
cluster.WaitLeader()
pdAddr := cluster.GetConfig().GetClientURL()
cmd := pdctlCmd.GetRootCmd()

stores := []*metapb.Store{
{
Id: 1,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
{
Id: 2,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
{
Id: 3,
State: metapb.StoreState_Up,
LastHeartbeat: time.Now().UnixNano(),
},
}

leaderServer := cluster.GetServer(cluster.GetLeader())
c.Assert(leaderServer.BootstrapCluster(), IsNil)
for _, store := range stores {
pdctl.MustPutStore(c, leaderServer.GetServer(), store)
}
defer cluster.Destroy()
startTime := time.Now().UnixNano() / int64(time.Millisecond)
pdctl.MustPutRegion(c, cluster, 1, 1, []byte("a"), []byte("b"), core.SetWrittenBytes(3000000000), core.SetReportInterval(statistics.WriteReportInterval))
pdctl.MustPutRegion(c, cluster, 2, 2, []byte("c"), []byte("d"), core.SetWrittenBytes(6000000000), core.SetReportInterval(statistics.WriteReportInterval))
pdctl.MustPutRegion(c, cluster, 3, 1, []byte("e"), []byte("f"), core.SetWrittenBytes(9000000000), core.SetReportInterval(statistics.WriteReportInterval))
pdctl.MustPutRegion(c, cluster, 4, 3, []byte("g"), []byte("h"), core.SetWrittenBytes(9000000000), core.SetReportInterval(statistics.WriteReportInterval))
// wait hot scheduler starts
time.Sleep(5000 * time.Millisecond)
endTime := time.Now().UnixNano() / int64(time.Millisecond)
start := strconv.FormatInt(startTime, 10)
end := strconv.FormatInt(endTime, 10)
args := []string{"-u", pdAddr, "hot", "history",
start, end,
"hot_region_type", "write",
"region_ids", "1,2,3",
"store_ids", "1,4",
}
output, e := pdctl.ExecuteCommand(cmd, args...)
hotRegions := core.HistoryHotRegions{}
c.Assert(e, IsNil)
c.Assert(json.Unmarshal(output, &hotRegions), IsNil)
regions := hotRegions.HistoryHotRegion
c.Assert(len(regions), Equals, 2)
c.Assert(regions[0].RegionID, Equals, uint64(1))
c.Assert(regions[0].StoreID, Equals, uint64(1))
c.Assert(regions[0].HotRegionType, Equals, "write")
c.Assert(regions[1].RegionID, Equals, uint64(3))
c.Assert(regions[1].StoreID, Equals, uint64(1))
c.Assert(regions[1].HotRegionType, Equals, "write")
args = []string{"-u", pdAddr, "hot", "history",
start, end,
"is_leader", "false",
}
output, e = pdctl.ExecuteCommand(cmd, args...)
c.Assert(e, IsNil)
c.Assert(json.Unmarshal(output, &hotRegions), IsNil)
c.Assert(len(hotRegions.HistoryHotRegion), Equals, 0)
}
95 changes: 65 additions & 30 deletions tools/pd-ctl/pdctl/command/global.go
Original file line number Diff line number Diff line change
Expand Up @@ -79,26 +79,21 @@ func doRequest(cmd *cobra.Command, prefix string, method string,

endpoints := getEndpoints(cmd)
err := tryURLs(cmd, endpoints, func(endpoint string) error {
var err error
url := endpoint + "/" + prefix
if method == "" {
method = http.MethodGet
}
var req *http.Request
return doGet(endpoint, prefix, method, &resp, b)
})
return resp, err
}

req, err = http.NewRequest(method, url, b.body)
if err != nil {
return err
}
if b.contentType != "" {
req.Header.Set("Content-Type", b.contentType)
}
// the resp would be returned by the outer function
resp, err = dial(req)
if err != nil {
return err
}
return nil
func doRequestSingleEndpoint(cmd *cobra.Command, endpoint, prefix, method string,
opts ...BodyOption) (string, error) {
b := &bodyOption{}
for _, o := range opts {
o(b)
}
var resp string

err := requestURL(cmd, endpoint, func(endpoint string) error {
return doGet(endpoint, prefix, method, &resp, b)
})
return resp, err
}
Expand Down Expand Up @@ -133,20 +128,11 @@ type DoFunc func(endpoint string) error
func tryURLs(cmd *cobra.Command, endpoints []string, f DoFunc) error {
var err error
for _, endpoint := range endpoints {
var u *url.URL
u, err = url.Parse(endpoint)
endpoint, err := checkURL(endpoint)
if err != nil {
cmd.Println("address format is wrong, should like 'http://127.0.0.1:2379' or '127.0.0.1:2379'")
cmd.Println(err.Error())
os.Exit(1)
}
// tolerate some schemes that will be used by users, the TiKV SDK
// use 'tikv' as the scheme, it is really confused if we do not
// support it by pd-ctl
if u.Scheme == "" || u.Scheme == "pd" || u.Scheme == "tikv" {
u.Scheme = "http"
}

endpoint = u.String()
err = f(endpoint)
if err != nil {
continue
Expand All @@ -159,6 +145,15 @@ func tryURLs(cmd *cobra.Command, endpoints []string, f DoFunc) error {
return err
}

func requestURL(cmd *cobra.Command, endpoint string, f DoFunc) error {
endpoint, err := checkURL(endpoint)
if err != nil {
cmd.Println(err.Error())
os.Exit(1)
}
return f(endpoint)
}

func getEndpoints(cmd *cobra.Command) []string {
addrs, err := cmd.Flags().GetString("pd")
if err != nil {
Expand Down Expand Up @@ -206,3 +201,43 @@ func postJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) {
}
cmd.Println("Success!")
}

// doGet send a get request to server.
func doGet(endpoint, prefix, method string, resp *string, b *bodyOption) error {
var err error
url := endpoint + "/" + prefix
if method == "" {
method = http.MethodGet
}
var req *http.Request

req, err = http.NewRequest(method, url, b.body)
if err != nil {
return err
}
if b.contentType != "" {
req.Header.Set("Content-Type", b.contentType)
}
// the resp would be returned by the outer function
*resp, err = dial(req)
if err != nil {
return err
}
return nil
}

func checkURL(endpoint string) (string, error) {
var u *url.URL
u, err := url.Parse(endpoint)
if err != nil {
return "", errors.Errorf("address format is wrong, should like 'http://127.0.0.1:2379' or '127.0.0.1:2379'")
}
// tolerate some schemes that will be used by users, the TiKV SDK
// use 'tikv' as the scheme, it is really confused if we do not
// support it by pd-ctl
if u.Scheme == "" || u.Scheme == "pd" || u.Scheme == "tikv" {
u.Scheme = "http"
}

return u.String(), nil
}
140 changes: 137 additions & 3 deletions tools/pd-ctl/pdctl/command/hot_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,23 @@
package command

import (
"bytes"
"encoding/json"
"net/http"
"sort"
"strconv"
"strings"

"github.com/pingcap/errors"
"github.com/spf13/cobra"
"github.com/tikv/pd/server/core"
)

const (
hotReadRegionsPrefix = "pd/api/v1/hotspot/regions/read"
hotWriteRegionsPrefix = "pd/api/v1/hotspot/regions/write"
hotStoresPrefix = "pd/api/v1/hotspot/stores"
hotReadRegionsPrefix = "pd/api/v1/hotspot/regions/read"
hotWriteRegionsPrefix = "pd/api/v1/hotspot/regions/write"
hotStoresPrefix = "pd/api/v1/hotspot/stores"
hotRegionsHistoryPrefix = "pd/api/v1/hotspot/regions/history"
)

// NewHotSpotCommand return a hot subcommand of rootCmd
Expand All @@ -36,6 +42,7 @@ func NewHotSpotCommand() *cobra.Command {
cmd.AddCommand(NewHotWriteRegionCommand())
cmd.AddCommand(NewHotReadRegionCommand())
cmd.AddCommand(NewHotStoreCommand())
cmd.AddCommand(NewHotRegionsHistoryCommand())
return cmd
}

Expand Down Expand Up @@ -106,6 +113,55 @@ func showHotStoresCommandFunc(cmd *cobra.Command, args []string) {
cmd.Println(r)
}

// NewHotRegionsHistoryCommand return a hot history regions subcommand of hotSpotCmd
func NewHotRegionsHistoryCommand() *cobra.Command {
cmd := &cobra.Command{
// TODO
// Need a better description.
Use: "history <start_time> <end_time> [<key> <value>]",
Short: "show the hot history regions",
Run: showHotRegionsHistoryCommandFunc,
}
return cmd
}

func showHotRegionsHistoryCommandFunc(cmd *cobra.Command, args []string) {
if len(args) < 2 || len(args)%2 != 0 {
cmd.Println(cmd.UsageString())
}
input, err := parseHotRegionsHistoryArgs(args)
if err != nil {
cmd.Printf("Failed to get history hotspot: %s\n", err)
}
data, _ := json.Marshal(input)
endpoints := getEndpoints(cmd)
hotRegions := &core.HistoryHotRegions{}
for _, endpoint := range endpoints {
tempHotRegions := core.HistoryHotRegions{}
resp, err := doRequestSingleEndpoint(cmd, endpoint, hotRegionsHistoryPrefix,
http.MethodGet, WithBody("application/json", bytes.NewBuffer(data)))
if err != nil {
cmd.Printf("Failed to get history hotspot: %s\n", err)
return
}
err = json.Unmarshal([]byte(resp), &tempHotRegions)
if err != nil {
cmd.Printf("Failed to get history hotspot: %s\n", err)
return
}
hotRegions.HistoryHotRegion = append(hotRegions.HistoryHotRegion, tempHotRegions.HistoryHotRegion...)
}
sort.SliceStable(hotRegions.HistoryHotRegion, func(i, j int) bool {
return hotRegions.HistoryHotRegion[i].UpdateTime > hotRegions.HistoryHotRegion[j].UpdateTime
})
resp, err := json.Marshal(hotRegions)
if err != nil {
cmd.Printf("Failed to get history hotspot: %s\n", err)
return
}
cmd.Println(string(resp))
}

func parseOptionalArgs(cmd *cobra.Command, prefix string, args []string) (string, error) {
argsLen := len(args)
if argsLen > 0 {
Expand All @@ -123,3 +179,81 @@ func parseOptionalArgs(cmd *cobra.Command, prefix string, args []string) (string
}
return prefix, nil
}

func parseHotRegionsHistoryArgs(args []string) (map[string]interface{}, error) {
startTime, err := strconv.ParseInt(args[0], 10, 64)
if err != nil {
return nil, errors.Errorf("start_time should be a number,but got %s", args[0])
}
endTime, err := strconv.ParseInt(args[1], 10, 64)
if err != nil {
return nil, errors.Errorf("end_time should be a number,but got %s", args[1])
}
input := map[string]interface{}{
"start_time": startTime,
"end_time": endTime,
}
stringToIntSlice := func(s string) ([]int64, error) {
results := make([]int64, 0)
args := strings.Split(s, ",")
for _, arg := range args {
result, err := strconv.ParseInt(arg, 10, 64)
if err != nil {
return nil, err
}
results = append(results, result)
}
return results, nil
}
for index := 2; index < len(args); index += 2 {
switch args[index] {
case "hot_region_type":
input["hot_region_type"] = []string{args[index+1]}
case "region_ids":
results, err := stringToIntSlice(args[index+1])
if err != nil {
return nil, errors.Errorf("region_ids should be a number slice,but got %s", args[index+1])
}
input["region_ids"] = results
case "store_ids":
results, err := stringToIntSlice(args[index+1])
if err != nil {
return nil, errors.Errorf("store_ids should be a number slice,but got %s", args[index+1])
}
input["store_ids"] = results
case "peer_ids":
results, err := stringToIntSlice(args[index+1])
if err != nil {
return nil, errors.Errorf("peer_ids should be a number slice,but got %s", args[index+1])
}
input["peer_ids"] = results
case "is_leader":
isLeader, err := strconv.ParseBool(args[index+1])
if err != nil {
return nil, errors.Errorf("is_leader should be a bool,but got %s", args[index+1])
}
input["is_leaders"] = []bool{isLeader}
case "is_learner":
isLearner, err := strconv.ParseBool(args[index+1])
if err != nil {
return nil, errors.Errorf("is_learners should be a bool,but got %s", args[index+1])
}
input["is_learners"] = []bool{isLearner}
default:
return nil, errors.Errorf("key should be one of hot_region_type,region_ids,store_ids,peer_ids,is_leaders,is_learners")
}
}
if _, ok := input["is_leaders"]; !ok {
input["is_leaders"] = []bool{
true,
false,
}
}
if _, ok := input["is_learners"]; !ok {
input["is_learners"] = []bool{
true,
false,
}
}
return input, nil
}

0 comments on commit d560865

Please sign in to comment.