Implement more HTTP APIs

Signed-off-by: JmPotato <[email protected]>

client/http/api.go

@@ -17,24 +17,58 @@ package http
 import (
 	"fmt"
 	"net/url"
+	"time"
 )
 
 // The following constants are the paths of PD HTTP APIs.
 const (
-	HotRead             = "/pd/api/v1/hotspot/regions/read"
-	HotWrite            = "/pd/api/v1/hotspot/regions/write"
-	Regions             = "/pd/api/v1/regions"
-	regionByID          = "/pd/api/v1/region/id"
-	regionByKey         = "/pd/api/v1/region/key"
-	regionsByKey        = "/pd/api/v1/regions/key"
-	regionsByStoreID    = "/pd/api/v1/regions/store"
-	Stores              = "/pd/api/v1/stores"
+	// Metadata
+	HotRead                = "/pd/api/v1/hotspot/regions/read"
+	HotWrite               = "/pd/api/v1/hotspot/regions/write"
+	HotHistory             = "/pd/api/v1/hotspot/regions/history"
+	RegionByIDPrefix       = "/pd/api/v1/region/id"
+	regionByKey            = "/pd/api/v1/region/key"
+	Regions                = "/pd/api/v1/regions"
+	regionsByKey           = "/pd/api/v1/regions/key"
+	RegionsByStoreIDPrefix = "/pd/api/v1/regions/store"
+	EmptyRegions           = "/pd/api/v1/regions/check/empty-region"
+	accelerateSchedule     = "/pd/api/v1/regions/accelerate-schedule"
+	store                  = "/pd/api/v1/store"
+	Stores                 = "/pd/api/v1/stores"
+	StatsRegion            = "/pd/api/v1/stats/region"
+	// Config
+	Config          = "/pd/api/v1/config"
+	ClusterVersion  = "/pd/api/v1/config/cluster-version"
+	ScheduleConfig  = "/pd/api/v1/config/schedule"
+	ReplicateConfig = "/pd/api/v1/config/replicate"
+	// Rule
+	PlacementRule         = "/pd/api/v1/config/rule"
+	PlacementRules        = "/pd/api/v1/config/rules"
+	placementRulesByGroup = "/pd/api/v1/config/rules/group"
+	RegionLabelRule       = "/pd/api/v1/config/region-label/rule"
+	// Scheduler
+	Schedulers            = "/pd/api/v1/schedulers"
+	scatterRangeScheduler = "/pd/api/v1/schedulers/scatter-range-"
+	// Admin
+	ResetTS                = "/pd/api/v1/admin/reset-ts"
+	BaseAllocID            = "/pd/api/v1/admin/base-alloc-id"
+	SnapshotRecoveringMark = "/pd/api/v1/admin/cluster/markers/snapshot-recovering"
+	// Debug
+	PProfProfile   = "/pd/api/v1/debug/pprof/profile"
+	PProfHeap      = "/pd/api/v1/debug/pprof/heap"
+	PProfMutex     = "/pd/api/v1/debug/pprof/mutex"
+	PProfAllocs    = "/pd/api/v1/debug/pprof/allocs"
+	PProfBlock     = "/pd/api/v1/debug/pprof/block"
+	PProfGoroutine = "/pd/api/v1/debug/pprof/goroutine"
+	// Others
 	MinResolvedTSPrefix = "/pd/api/v1/min-resolved-ts"
+	Status              = "/pd/api/v1/status"
+	Version             = "/pd/api/v1/version"
 )
 
 // RegionByID returns the path of PD HTTP API to get region by ID.
 func RegionByID(regionID uint64) string {
-	return fmt.Sprintf("%s/%d", regionByID, regionID)
+	return fmt.Sprintf("%s/%d", RegionByIDPrefix, regionID)
 }
 
 // RegionByKey returns the path of PD HTTP API to get region by key.
@@ -50,5 +84,55 @@ func RegionsByKey(startKey, endKey []byte, limit int) string {
 
 // RegionsByStoreID returns the path of PD HTTP API to get regions by store ID.
 func RegionsByStoreID(storeID uint64) string {
-	return fmt.Sprintf("%s/%d", regionsByStoreID, storeID)
+	return fmt.Sprintf("%s/%d", RegionsByStoreIDPrefix, storeID)
+}
+
+// RegionStatsByStartEndKey returns the path of PD HTTP API to get region stats by start key and end key.
+func RegionStatsByStartEndKey(startKey, endKey string) string {
+	return fmt.Sprintf("%s?start_key=%s&end_key=%s", StatsRegion, startKey, endKey)
+}
+
+// StoreByID returns the store API with store ID parameter.
+func StoreByID(id uint64) string {
+	return fmt.Sprintf("%s/%d", store, id)
+}
+
+// StoreLabelByID returns the store label API with store ID parameter.
+func StoreLabelByID(id uint64) string {
+	return fmt.Sprintf("%s/%d/label", store, id)
+}
+
+// ConfigWithTTLSeconds returns the config API with the TTL seconds parameter.
+func ConfigWithTTLSeconds(ttlSeconds float64) string {
+	return fmt.Sprintf("%s?ttlSecond=%.0f", Config, ttlSeconds)
+}
+
+// PlacementRulesByGroup returns the path of PD HTTP API to get placement rules by group.
+func PlacementRulesByGroup(group string) string {
+	return fmt.Sprintf("%s/%s", placementRulesByGroup, group)
+}
+
+// PlacementRuleByGroupAndID returns the path of PD HTTP API to get placement rule by group and ID.
+func PlacementRuleByGroupAndID(group, id string) string {
+	return fmt.Sprintf("%s/%s/%s", PlacementRule, group, id)
+}
+
+// SchedulerByName returns the scheduler API with the given scheduler name.
+func SchedulerByName(name string) string {
+	return fmt.Sprintf("%s/%s", Schedulers, name)
+}
+
+// ScatterRangeSchedulerWithName returns the scatter range scheduler API with name parameter.
+func ScatterRangeSchedulerWithName(name string) string {
+	return fmt.Sprintf("%s%s", scatterRangeScheduler, name)
+}
+
+// PProfProfileAPIWithInterval returns the pprof profile API with interval parameter.
+func PProfProfileAPIWithInterval(interval time.Duration) string {
+	return fmt.Sprintf("%s?seconds=%d", PProfProfile, interval/time.Second)
+}
+
+// PProfGoroutineWithDebugLevel returns the pprof goroutine API with debug level parameter.
+func PProfGoroutineWithDebugLevel(level int) string {
+	return fmt.Sprintf("%s?debug=%d", PProfGoroutine, level)
 }

client/http/client.go

@@ -47,8 +47,13 @@ type Client interface {
 	GetRegionsByStoreID(context.Context, uint64) (*RegionsInfo, error)
 	GetHotReadRegions(context.Context) (*StoreHotPeersInfos, error)
 	GetHotWriteRegions(context.Context) (*StoreHotPeersInfos, error)
+	GetRegionStatusByKey(context.Context, []byte, []byte) (*RegionStats, error)
 	GetStores(context.Context) (*StoresInfo, error)
+	GetPlacementRulesByGroup(context.Context, string) ([]*Rule, error)
+	SetPlacementRule(context.Context, *Rule) error
+	DeletePlacementRule(context.Context, string, string) error
 	GetMinResolvedTSByStoresIDs(context.Context, []uint64) (uint64, map[uint64]uint64, error)
+	AccelerateSchedule(context.Context, []byte, []byte) error
 	Close()
 }
 
@@ -154,7 +159,7 @@ func (c *client) execDuration(name string, duration time.Duration) {
 // it consistent with the current implementation of some clients (e.g. TiDB).
 func (c *client) requestWithRetry(
 	ctx context.Context,
-	name, uri string,
+	name, uri, method string,
 	res interface{},
 ) error {
 	var (
@@ -163,7 +168,7 @@ func (c *client) requestWithRetry(
 	)
 	for idx := 0; idx < len(c.pdAddrs); idx++ {
 		addr = c.pdAddrs[idx]
-		err = c.request(ctx, name, addr, uri, res)
+		err = c.request(ctx, name, fmt.Sprintf("%s%s", addr, uri), method, res)
 		if err == nil {
 			break
 		}
@@ -175,16 +180,15 @@ func (c *client) requestWithRetry(
 
 func (c *client) request(
 	ctx context.Context,
-	name, addr, uri string,
+	name, url, method string,
 	res interface{},
 ) error {
-	reqURL := fmt.Sprintf("%s%s", addr, uri)
 	logFields := []zap.Field{
 		zap.String("name", name),
-		zap.String("url", reqURL),
+		zap.String("url", url),
 	}
 	log.Debug("[pd] request the http url", logFields...)
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
+	req, err := http.NewRequestWithContext(ctx, method, url, nil)
 	if err != nil {
 		log.Error("[pd] create http request failed", append(logFields, zap.Error(err))...)
 		return errors.Trace(err)
@@ -229,7 +233,7 @@ func (c *client) request(
 // GetRegionByID gets the region info by ID.
 func (c *client) GetRegionByID(ctx context.Context, regionID uint64) (*RegionInfo, error) {
 	var region RegionInfo
-	err := c.requestWithRetry(ctx, "GetRegionByID", RegionByID(regionID), &region)
+	err := c.requestWithRetry(ctx, "GetRegionByID", RegionByID(regionID), http.MethodGet, &region)
 	if err != nil {
 		return nil, err
 	}
@@ -239,7 +243,7 @@ func (c *client) GetRegionByID(ctx context.Context, regionID uint64) (*RegionInf
 // GetRegionByKey gets the region info by key.
 func (c *client) GetRegionByKey(ctx context.Context, key []byte) (*RegionInfo, error) {
 	var region RegionInfo
-	err := c.requestWithRetry(ctx, "GetRegionByKey", RegionByKey(key), &region)
+	err := c.requestWithRetry(ctx, "GetRegionByKey", RegionByKey(key), http.MethodGet, &region)
 	if err != nil {
 		return nil, err
 	}
@@ -249,7 +253,7 @@ func (c *client) GetRegionByKey(ctx context.Context, key []byte) (*RegionInfo, e
 // GetRegions gets the regions info.
 func (c *client) GetRegions(ctx context.Context) (*RegionsInfo, error) {
 	var regions RegionsInfo
-	err := c.requestWithRetry(ctx, "GetRegions", Regions, &regions)
+	err := c.requestWithRetry(ctx, "GetRegions", Regions, http.MethodGet, &regions)
 	if err != nil {
 		return nil, err
 	}
@@ -259,7 +263,7 @@ func (c *client) GetRegions(ctx context.Context) (*RegionsInfo, error) {
 // GetRegionsByKey gets the regions info by key range. If the limit is -1, it will return all regions within the range.
 func (c *client) GetRegionsByKey(ctx context.Context, startKey, endKey []byte, limit int) (*RegionsInfo, error) {
 	var regions RegionsInfo
-	err := c.requestWithRetry(ctx, "GetRegionsByKey", RegionsByKey(startKey, endKey, limit), &regions)
+	err := c.requestWithRetry(ctx, "GetRegionsByKey", RegionsByKey(startKey, endKey, limit), http.MethodGet, &regions)
 	if err != nil {
 		return nil, err
 	}
@@ -269,7 +273,7 @@ func (c *client) GetRegionsByKey(ctx context.Context, startKey, endKey []byte, l
 // GetRegionsByStoreID gets the regions info by store ID.
 func (c *client) GetRegionsByStoreID(ctx context.Context, storeID uint64) (*RegionsInfo, error) {
 	var regions RegionsInfo
-	err := c.requestWithRetry(ctx, "GetRegionsByStoreID", RegionsByStoreID(storeID), &regions)
+	err := c.requestWithRetry(ctx, "GetRegionsByStoreID", RegionsByStoreID(storeID), http.MethodGet, &regions)
 	if err != nil {
 		return nil, err
 	}
@@ -279,7 +283,7 @@ func (c *client) GetRegionsByStoreID(ctx context.Context, storeID uint64) (*Regi
 // GetHotReadRegions gets the hot read region statistics info.
 func (c *client) GetHotReadRegions(ctx context.Context) (*StoreHotPeersInfos, error) {
 	var hotReadRegions StoreHotPeersInfos
-	err := c.requestWithRetry(ctx, "GetHotReadRegions", HotRead, &hotReadRegions)
+	err := c.requestWithRetry(ctx, "GetHotReadRegions", HotRead, http.MethodGet, &hotReadRegions)
 	if err != nil {
 		return nil, err
 	}
@@ -289,23 +293,57 @@ func (c *client) GetHotReadRegions(ctx context.Context) (*StoreHotPeersInfos, er
 // GetHotWriteRegions gets the hot write region statistics info.
 func (c *client) GetHotWriteRegions(ctx context.Context) (*StoreHotPeersInfos, error) {
 	var hotWriteRegions StoreHotPeersInfos
-	err := c.requestWithRetry(ctx, "GetHotWriteRegions", HotWrite, &hotWriteRegions)
+	err := c.requestWithRetry(ctx, "GetHotWriteRegions", HotWrite, http.MethodGet, &hotWriteRegions)
 	if err != nil {
 		return nil, err
 	}
 	return &hotWriteRegions, nil
 }
 
+// GetRegionStatusByKey gets the region status by key range.
+func (c *client) GetRegionStatusByKey(ctx context.Context, startKey, endKey []byte) (*RegionStats, error) {
+	var regionStats RegionStats
+	err := c.requestWithRetry(
+		ctx, "GetRegionStatusByKey",
+		RegionStatsByStartEndKey(string(startKey), string(endKey)), http.MethodGet,
+		&regionStats,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &regionStats, nil
+}
+
 // GetStores gets the stores info.
 func (c *client) GetStores(ctx context.Context) (*StoresInfo, error) {
 	var stores StoresInfo
-	err := c.requestWithRetry(ctx, "GetStores", Stores, &stores)
+	err := c.requestWithRetry(ctx, "GetStores", Stores, http.MethodGet, &stores)
 	if err != nil {
 		return nil, err
 	}
 	return &stores, nil
 }
 
+// GetPlacementRulesByGroup gets the placement rules by group.
+func (c *client) GetPlacementRulesByGroup(ctx context.Context, group string) ([]*Rule, error) {
+	var rules []*Rule
+	err := c.requestWithRetry(ctx, "GetPlacementRulesByGroup", PlacementRulesByGroup(group), http.MethodGet, &rules)
+	if err != nil {
+		return nil, err
+	}
+	return rules, nil
+}
+
+// SetPlacementRule sets the placement rule.
+func (c *client) SetPlacementRule(ctx context.Context, rule *Rule) error {
+	return c.requestWithRetry(ctx, "SetPlacementRule", PlacementRule, http.MethodPost, nil)
+}
+
+// DeletePlacementRule deletes the placement rule.
+func (c *client) DeletePlacementRule(ctx context.Context, group, id string) error {
+	return c.requestWithRetry(ctx, "DeletePlacementRule", PlacementRuleByGroupAndID(group, id), http.MethodDelete, nil)
+}
+
 // GetMinResolvedTSByStoresIDs get min-resolved-ts by stores IDs.
 func (c *client) GetMinResolvedTSByStoresIDs(ctx context.Context, storeIDs []uint64) (uint64, map[uint64]uint64, error) {
 	uri := MinResolvedTSPrefix
@@ -326,7 +364,7 @@ func (c *client) GetMinResolvedTSByStoresIDs(ctx context.Context, storeIDs []uin
 		IsRealTime          bool              `json:"is_real_time,omitempty"`
 		StoresMinResolvedTS map[uint64]uint64 `json:"stores_min_resolved_ts"`
 	}{}
-	err := c.requestWithRetry(ctx, "GetMinResolvedTSByStoresIDs", uri, &resp)
+	err := c.requestWithRetry(ctx, "GetMinResolvedTSByStoresIDs", uri, http.MethodGet, &resp)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -335,3 +373,8 @@ func (c *client) GetMinResolvedTSByStoresIDs(ctx context.Context, storeIDs []uin
 	}
 	return resp.MinResolvedTS, resp.StoresMinResolvedTS, nil
 }
+
+// AccelerateSchedule accelerates the scheduling of the regions within the given key range.
+func (c *client) AccelerateSchedule(ctx context.Context, startKey, endKey []byte) error {
+	return c.requestWithRetry(ctx, "AccelerateSchedule", accelerateSchedule, http.MethodPost, nil)
+}

client/http/types.go

@@ -176,3 +176,67 @@ type StoreStatus struct {
 	LastHeartbeatTS time.Time `json:"last_heartbeat_ts"`
 	Uptime          string    `json:"uptime"`
 }
+
+// RegionStats stores the statistics of regions.
+type RegionStats struct {
+	Count            int            `json:"count"`
+	EmptyCount       int            `json:"empty_count"`
+	StorageSize      int64          `json:"storage_size"`
+	StorageKeys      int64          `json:"storage_keys"`
+	StoreLeaderCount map[uint64]int `json:"store_leader_count"`
+	StorePeerCount   map[uint64]int `json:"store_peer_count"`
+}
+
+// PeerRoleType is the expected peer type of the placement rule.
+type PeerRoleType string
+
+const (
+	// Voter can either match a leader peer or follower peer
+	Voter PeerRoleType = "voter"
+	// Leader matches a leader.
+	Leader PeerRoleType = "leader"
+	// Follower matches a follower.
+	Follower PeerRoleType = "follower"
+	// Learner matches a learner.
+	Learner PeerRoleType = "learner"
+)
+
+// LabelConstraint is used to filter store when trying to place peer of a region.
+type LabelConstraint struct {
+	Key    string            `json:"key,omitempty"`
+	Op     LabelConstraintOp `json:"op,omitempty"`
+	Values []string          `json:"values,omitempty"`
+}
+
+// LabelConstraintOp defines how a LabelConstraint matches a store. It can be one of
+// 'in', 'notIn', 'exists', or 'notExists'.
+type LabelConstraintOp string
+
+const (
+	// In restricts the store label value should in the value list.
+	// If label does not exist, `in` is always false.
+	In LabelConstraintOp = "in"
+	// NotIn restricts the store label value should not in the value list.
+	// If label does not exist, `notIn` is always true.
+	NotIn LabelConstraintOp = "notIn"
+	// Exists restricts the store should have the label.
+	Exists LabelConstraintOp = "exists"
+	// NotExists restricts the store should not have the label.
+	NotExists LabelConstraintOp = "notExists"
+)
+
+// Rule is the placement rule that can be checked against a region. When
+// applying rules (apply means schedule regions to match selected rules), the
+// apply order is defined by the tuple [GroupIndex, GroupID, Index, ID].
+type Rule struct {
+	GroupID        string          `json:"group_id"`
+	ID             string          `json:"id"`
+	Index          int             `json:"index,omitempty"`
+	Override       bool            `json:"override,omitempty"`
+	StartKeyHex    string          `json:"start_key"`
+	EndKeyHex      string          `json:"end_key"`
+	Role           PeerRoleType    `json:"role"`
+	Count          int             `json:"count"`
+	Constraints    LabelConstraint `json:"label_constraints,omitempty"`
+	LocationLabels []string        `json:"location_labels,omitempty"`
+}