// Copyright 2018 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package pdapi

import (
	"bytes"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"time"

	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/tidb-operator/pkg/pdapi/placement"
	"github.com/pingcap/tidb-operator/pkg/util/crypto"
	httputil "github.com/pingcap/tidb-operator/pkg/util/http"
	"github.com/tikv/pd/pkg/typeutil"
	corelisterv1 "k8s.io/client-go/listers/core/v1"
	"k8s.io/klog/v2"
)

const (
	DefaultTimeout       = 5 * time.Second
	evictSchedulerLeader = "evict-leader-scheduler"
	tiKVNotBootstrapped  = `TiKV cluster not bootstrapped, please start TiKV first"`
)

// GetTLSConfig returns *tls.Config for given TiDB cluster.
//
// The returned config resolves the client certificate from the secret lister
// on every TLS handshake via GetClientCertificate. When cert-manager rotates
// the underlying Secret, the informer updates its local cache and any newly
// initiated TLS handshake — including gRPC reconnects on long-lived etcd
// clients — will automatically pick up the fresh certificate without
// requiring an operator restart.
//
// Note: RootCAs is still materialized once at config build time. If the CA
// bundle is rotated, callers still need to rebuild the *tls.Config.
func GetTLSConfig(secretLister corelisterv1.SecretLister, namespace Namespace, secretName string) (*tls.Config, error) {
	ns := string(namespace)
	secret, err := secretLister.Secrets(ns).Get(secretName)
	if err != nil {
		return nil, fmt.Errorf("unable to load certificates from secret %s/%s: %v", namespace, secretName, err)
	}

	// Build the base config from the current Secret. This validates the CA
	// pool and the keypair up-front so a misconfigured Secret surfaces here
	// instead of at the first TLS handshake.
	cfg, err := crypto.LoadTlsConfigFromSecret(secret)
	if err != nil {
		return nil, err
	}

	// Replace the static cert with a callback that re-reads the Secret from
	// the informer cache on every handshake, so rotated certificates are
	// picked up automatically.
	cfg.Certificates = nil
	cfg.GetClientCertificate = func(_ *tls.CertificateRequestInfo) (*tls.Certificate, error) {
		s, err := secretLister.Secrets(ns).Get(secretName)
		if err != nil {
			return nil, fmt.Errorf("unable to load certificates from secret %s/%s: %v", ns, secretName, err)
		}
		return crypto.LoadClientCertFromSecret(s)
	}
	return cfg, nil
}

// PDClient provides pd server's api
type PDClient interface {
	// GetHealth returns the PD's health info
	GetHealth() (*HealthInfo, error)
	// GetConfig returns PD's config
	GetConfig() (*PDConfigFromAPI, error)
	// GetCluster returns used when syncing pod labels.
	GetCluster() (*metapb.Cluster, error)
	// GetMembers returns all PD members from cluster
	GetMembers() (*MembersInfo, error)
	// GetStores lists all TiKV/TiFlash stores from cluster
	GetStores() (*StoresInfo, error)
	// GetTiKVStores lists all TiKV stores from cluster
	GetTiKVStores() (*StoresInfo, error)
	// GetTiFlashStores lists all TiFlash stores from cluster
	GetTiFlashStores() (*StoresInfo, error)
	// GetTombStoneStores lists all tombstone stores from cluster
	GetTombStoneStores() (*StoresInfo, error)
	// GetStore gets a TiKV store for a specific store id from cluster
	GetStore(storeID uint64) (*StoreInfo, error)
	// storeLabelsEqualNodeLabels compares store labels with node labels
	// for historic reasons, PD stores TiKV labels as []*StoreLabel which is a key-value pair slice
	SetStoreLabels(storeID uint64, labels map[string]string) (bool, error)
	// UpdateReplicationConfig updates the replication config
	UpdateReplicationConfig(config PDReplicationConfig) error
	// DeleteStore deletes a TiKV store from cluster
	DeleteStore(storeID uint64) error
	// CancelDeleteStore cancel the deletion of TiKV store from cluster
	CancelDeleteStore(storeID uint64) error
	// SetStoreState sets store to specified state.
	SetStoreState(storeID uint64, state string) error
	// SetStoreLimit sets store limit to specified rate.
	SetStoreLimit(storeID uint64, rate float64) error
	// DeleteMember deletes a PD member from cluster
	DeleteMember(name string) error
	// DeleteMemberByID deletes a PD member from cluster
	DeleteMemberByID(memberID uint64) error
	// UpdateMember updates a PD member's peer urls
	UpdateMember(memberID uint64, peerUrls []string) error
	// BeginEvictLeader initiates leader eviction for a storeID.
	// This is used when upgrading a pod.
	BeginEvictLeader(storeID uint64) error
	// EndEvictLeader is used at the end of pod upgrade.
	EndEvictLeader(storeID uint64) error
	// GetEvictLeaderSchedulers gets schedulers of evict leader
	GetEvictLeaderSchedulers() ([]string, error)
	// GetEvictLeaderSchedulersForStores gets schedulers of evict leader for given stores
	GetEvictLeaderSchedulersForStores(storeIDs ...uint64) (map[uint64]string, error)
	// GetPDLeader returns pd leader
	GetPDLeader() (*pdpb.Member, error)
	// TransferPDLeader transfers pd leader to specified member
	TransferPDLeader(name string) error
	// SetPDLeaderPriority sets pd leader priority.
	SetPDLeaderPriority(memberName string, priority int32) error
	// GetAutoscalingPlans returns the scaling plan for the cluster
	GetAutoscalingPlans(strategy Strategy) ([]Plan, error)
	// GetKeyspace gets a keyspace for a specific keyspace name from cluster
	GetKeyspace(keyspaceName string) (*KeyspaceInfo, error)
	// GetRuleByGroupAndID gets placement rule of cluster by group and id.
	GetRuleByGroupAndID(group, id string) (*placement.Rule, error)
	// GetRulesByGroup lists all placement rules of cluster by group.
	GetRulesByGroup(group string) ([]*placement.Rule, error)
	// SetRules sets placement rules of cluster
	SetRules(rules []*placement.Rule) error
	// GetRuleGroup returns a RuleGroup configuration.
	GetRuleGroup(id string) (*placement.RuleGroup, error)
	// SetRuleGroup updates a RuleGroup.
	SetRuleGroup(id string, index int, override bool) error
	// GetRuleGroups returns all RuleGroup configuration.
	GetRuleGroups() ([]*placement.RuleGroup, error)
	// ScanRegions scan regions in a given range [startKey, endKey).
	ScanRegions(startKey, endKey string, limit int) (*RegionsInfo, error)
	// GetRegionByID gets region info by id
	GetRegionByID(regionID uint64) (*RegionInfo, error)
	// GetHotRegions gets hot read and write regions
	GetHotRegions(regionHeartbeatSeconds, storeHeartbeatSeconds, hotRegionSeconds int32) (*HistoryHotRegions, error)
	// GetHotWriteRegions gets hot write regions
	GetHotWriteRegions() (*StoreHotPeersInfos, error)
	// GetHotReadRegions gets hot read regions
	GetHotReadRegions() (*StoreHotPeersInfos, error)
}

var (
	healthPrefix           = "pd/api/v1/health"
	membersPrefix          = "pd/api/v1/members"
	storesPrefix           = "pd/api/v1/stores"
	storePrefix            = "pd/api/v1/store"
	configPrefix           = "pd/api/v1/config"
	clusterIDPrefix        = "pd/api/v1/cluster"
	schedulersPrefix       = "pd/api/v1/schedulers"
	pdLeaderPrefix         = "pd/api/v1/leader"
	pdLeaderTransferPrefix = "pd/api/v1/leader/transfer"
	pdReplicationPrefix    = "pd/api/v1/config/replicate"
	rulePrefix             = "pd/api/v1/config/rule"
	rulesPrefix            = "pd/api/v1/config/rules"
	rulesBatchPrefix       = "pd/api/v1/config/rules/batch"
	ruleGroupPrefix        = "pd/api/v1/config/rule_group"
	ruleGroupsPrefix       = "pd/api/v1/config/rule_groups"
	keyspacesPrefix        = "pd/api/v2/keyspaces"
	etcdMembersPrefix      = "v2/members"
	// evictLeaderSchedulerConfigPrefix is the prefix of evict-leader-scheduler
	// config API, available since PD v3.1.0.
	evictLeaderSchedulerConfigPrefix = "pd/api/v1/scheduler-config/evict-leader-scheduler/list"
	autoscalingPrefix                = "autoscaling"
	regionIDPrefix                   = "pd/api/v1/region/id"
	regionsKeyPrefix                 = "pd/api/v1/regions/key"
	hotReadRegionsPrefix             = "pd/api/v1/hotspot/regions/read"
	hotWriteRegionsPrefix            = "pd/api/v1/hotspot/regions/write"
)

// pdClient is default implementation of PDClient
type pdClient struct {
	url        string
	httpClient *http.Client
}

// NewPDClient returns a new PDClient
func NewPDClient(url string, timeout time.Duration, tlsConfig *tls.Config) PDClient {
	var disableKeepalive bool
	if tlsConfig != nil {
		disableKeepalive = true
	}
	return &pdClient{
		url: url,
		httpClient: &http.Client{
			Timeout:   timeout,
			Transport: &http.Transport{TLSClientConfig: tlsConfig, DisableKeepAlives: disableKeepalive},
		},
	}
}

// following struct definitions are copied from github.com/pingcap/pd/server/api/store
// these are not exported by that package

// HealthInfo define PD's healthy info
type HealthInfo struct {
	Healths []MemberHealth
}

// MemberHealth define a pd member's healthy info
type MemberHealth struct {
	Name       string   `json:"name"`
	MemberID   uint64   `json:"member_id"`
	ClientUrls []string `json:"client_urls"`
	Health     bool     `json:"health"`
}

// MetaStore is TiKV store status defined in protobuf
type MetaStore struct {
	*metapb.Store
	StateName string `json:"state_name"`
}

// StoreStatus is TiKV store status returned from PD RESTful interface
type StoreStatus struct {
	Capacity           typeutil.ByteSize `json:"capacity"`
	Available          typeutil.ByteSize `json:"available"`
	LeaderCount        int               `json:"leader_count"`
	LeaderScore        float64           `json:"leader_score"`
	RegionCount        int               `json:"region_count"`
	RegionScore        float64           `json:"region_score"`
	SendingSnapCount   uint32            `json:"sending_snap_count"`
	ReceivingSnapCount uint32            `json:"receiving_snap_count"`
	ApplyingSnapCount  uint32            `json:"applying_snap_count"`
	IsBusy             bool              `json:"is_busy"`

	StartTS         time.Time         `json:"start_ts"`
	LastHeartbeatTS time.Time         `json:"last_heartbeat_ts"`
	Uptime          typeutil.Duration `json:"uptime"`
}

// StoreInfo is a single store info returned from PD RESTful interface
type StoreInfo struct {
	Store  *MetaStore   `json:"store"`
	Status *StoreStatus `json:"status"`
}

// StoresInfo is stores info returned from PD RESTful interface
type StoresInfo struct {
	Count  int          `json:"count"`
	Stores []*StoreInfo `json:"stores"`
}

// MembersInfo is PD members info returned from PD RESTful interface
// type Members map[string][]*pdpb.Member
type MembersInfo struct {
	Header     *pdpb.ResponseHeader `json:"header,omitempty"`
	Members    []*pdpb.Member       `json:"members,omitempty"`
	Leader     *pdpb.Member         `json:"leader,omitempty"`
	EtcdLeader *pdpb.Member         `json:"etcd_leader,omitempty"`
}

// below copied from github.com/tikv/pd/pkg/autoscaling

// Strategy within a HTTP request provides rules and resources to help make decision for auto scaling.
type Strategy struct {
	Rules     []*Rule     `json:"rules"`
	Resources []*Resource `json:"resources"`
}

// Rule is a set of constraints for a kind of component.
type Rule struct {
	Component   string       `json:"component"`
	CPURule     *CPURule     `json:"cpu_rule,omitempty"`
	StorageRule *StorageRule `json:"storage_rule,omitempty"`
}

// CPURule is the constraints about CPU.
type CPURule struct {
	MaxThreshold  float64  `json:"max_threshold"`
	MinThreshold  float64  `json:"min_threshold"`
	ResourceTypes []string `json:"resource_types"`
}

// StorageRule is the constraints about storage.
type StorageRule struct {
	MinThreshold  float64  `json:"min_threshold"`
	ResourceTypes []string `json:"resource_types"`
}

// Resource represents a kind of resource set including CPU, memory, storage.
type Resource struct {
	ResourceType string `json:"resource_type"`
	// The basic unit of CPU is milli-core.
	CPU uint64 `json:"cpu"`
	// The basic unit of memory is byte.
	Memory uint64 `json:"memory"`
	// The basic unit of storage is byte.
	Storage uint64 `json:"storage"`
	// If count is not set, it indicates no limit.
	Count *uint64 `json:"count,omitempty"`
}

// Plan is the final result of auto scaling, which indicates how to scale in or scale out.
type Plan struct {
	Component    string            `json:"component"`
	Count        uint64            `json:"count"`
	ResourceType string            `json:"resource_type"`
	Labels       map[string]string `json:"labels"`
}

type schedulerInfo struct {
	Name    string `json:"name"`
	StoreID uint64 `json:"store_id"`
}

type KeyspaceInfo struct {
	Id   uint32 `json:"id,omitempty"`
	Name string `json:"name,omitempty"`
}

// RegionsInfo is regions info returned from PD RESTful interface
type RegionsInfo struct {
	Count   int           `json:"count"`
	Regions []*RegionInfo `json:"regions"`
}

// RegionInfo records detail region info for api usage.
// NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
// easyjson:json
type RegionInfo struct {
	ID           uint64              `json:"id"`
	StartKey     string              `json:"start_key"`
	EndKey       string              `json:"end_key"`
	RegionEpoch  *metapb.RegionEpoch `json:"epoch,omitempty"`
	Peers        []MetaPeer          `json:"peers,omitempty"`
	Leader       MetaPeer            `json:"leader,omitempty"`
	PendingPeers []MetaPeer          `json:"pending_peers,omitempty"`
}

type MetaPeer struct {
	*metapb.Peer
	// RoleName is `Role.String()`.
	// Since Role is serialized as int by json by default,
	// introducing it will make the output of pd-ctl easier to identify Role.
	RoleName string `json:"role_name"`
	// IsLearner is `Role == "Learner"`.
	// Since IsLearner was changed to Role in kvproto in 5.0, this field was introduced to ensure api compatibility.
	IsLearner bool `json:"is_learner,omitempty"`
}

// StoreHotPeersInfos is used to get human-readable description for hot regions.
// NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
type StoreHotPeersInfos struct {
	AsPeer   StoreHotPeersStat `json:"as_peer"`
	AsLeader StoreHotPeersStat `json:"as_leader"`
}

func (s StoreHotPeersInfos) String() string {
	b, _ := json.Marshal(s)
	return string(b)
}

func (s StoreHotPeersInfos) SimplifiedString() string {
	var infos = StoreHotPeersInfos{}
	infos.AsPeer = make(StoreHotPeersStat)
	for storeID, ps := range s.AsPeer {
		if len(ps.Stats) > 0 {
			infos.AsPeer[storeID] = ps
		}
	}
	infos.AsLeader = make(StoreHotPeersStat)
	for storeID, ps := range s.AsLeader {
		if len(ps.Stats) > 0 {
			infos.AsLeader[storeID] = ps
		}
	}
	b, _ := json.Marshal(infos)
	return string(b)
}

// StoreHotPeersStat is used to record the hot region statistics group by store.
// NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
type StoreHotPeersStat map[uint64]*HotPeersStat

// HotPeersStat records all hot regions statistics
type HotPeersStat struct {
	StoreByteRate  float64           `json:"store_bytes"`
	StoreKeyRate   float64           `json:"store_keys"`
	StoreQueryRate float64           `json:"store_query"`
	TotalBytesRate float64           `json:"total_flow_bytes"`
	TotalKeysRate  float64           `json:"total_flow_keys"`
	TotalQueryRate float64           `json:"total_flow_query"`
	Count          int               `json:"regions_count"`
	Stats          []HotPeerStatShow `json:"statistics"`
}

// HotPeerStatShow records the hot region statistics for output
type HotPeerStatShow struct {
	StoreID        uint64    `json:"store_id"`
	Stores         []uint64  `json:"stores"`
	IsLeader       bool      `json:"is_leader"`
	IsLearner      bool      `json:"is_learner"`
	RegionID       uint64    `json:"region_id"`
	HotDegree      int       `json:"hot_degree"`
	ByteRate       float64   `json:"flow_bytes"`
	KeyRate        float64   `json:"flow_keys"`
	QueryRate      float64   `json:"flow_query"`
	AntiCount      int       `json:"anti_count"`
	LastUpdateTime time.Time `json:"last_update_time,omitempty"`
}

// HistoryHotRegions wraps historyHotRegion
// it will be returned to TiDB.
//
// NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
type HistoryHotRegions struct {
	HistoryHotRegion []*HistoryHotRegion `json:"history_hot_region"`
	HotRegionsCount  int                 `json:"hot_regions_count"`
}

func (r *HistoryHotRegions) String() string {
	b, _ := json.Marshal(r)
	return string(b)
}

// HistoryHotRegion wraps hot region info
// it is storage format of hot_region_storage
type HistoryHotRegion struct {
	UpdateTime    int64   `json:"update_time"`
	RegionID      uint64  `json:"region_id"`
	StoreID       uint64  `json:"store_id"`
	IsLeader      bool    `json:"is_leader"`
	HotRegionType string  `json:"hot_region_type"`
	HotDegree     int64   `json:"hot_degree"`
	FlowBytes     float64 `json:"flow_bytes"`
	KeyRate       float64 `json:"key_rate"`
	QueryRate     float64 `json:"query_rate"`
	StartKey      string  `json:"start_key"`
	EndKey        string  `json:"end_key"`

	Region *RegionInfo `json:"region"`
}

func (c *pdClient) GetHealth() (*HealthInfo, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, healthPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	healths := []MemberHealth{}
	err = json.Unmarshal(body, &healths)
	if err != nil {
		return nil, err
	}
	return &HealthInfo{
		healths,
	}, nil
}

func (c *pdClient) GetConfig() (*PDConfigFromAPI, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, configPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	config := &PDConfigFromAPI{}
	err = json.Unmarshal(body, config)
	if err != nil {
		return nil, err
	}
	return config, nil
}

func (c *pdClient) GetCluster() (*metapb.Cluster, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, clusterIDPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	cluster := &metapb.Cluster{}
	err = json.Unmarshal(body, cluster)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

func (c *pdClient) GetMembers() (*MembersInfo, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, membersPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	members := &MembersInfo{}
	err = json.Unmarshal(body, members)
	if err != nil {
		return nil, err
	}
	return members, nil
}

func (c *pdClient) UpdateMember(memberID uint64, peerUrls []string) error {
	apiURL := fmt.Sprintf("%s/%s/%x", c.url, etcdMembersPrefix, memberID)
	input := map[string]interface{}{
		"peerURLs": peerUrls,
	}
	b, err := json.Marshal(input)
	if err != nil {
		return err
	}
	req, err := http.NewRequest(http.MethodPut, apiURL, bytes.NewReader(b))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(resp.Body)

	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("fail to update member %x: %v", memberID, string(body))
}

func (c *pdClient) getStores(apiURL string) (*StoresInfo, error) {
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	storesInfo := &StoresInfo{}
	err = json.Unmarshal(body, storesInfo)
	if err != nil {
		return nil, err
	}
	return storesInfo, nil
}

func (c *pdClient) GetStores() (*StoresInfo, error) {
	storesInfo, err := c.getStores(fmt.Sprintf("%s/%s", c.url, storesPrefix))
	if err != nil {
		if strings.HasSuffix(err.Error(), tiKVNotBootstrapped+"\n") {
			err = TiKVNotBootstrappedErrorf(err.Error())
		}
		return nil, err
	}
	return storesInfo, nil
}

func (c *pdClient) getTiKVAndTiFlashStores() (tikvStoresInfo, tiflashStoresInfo *StoresInfo, err error) {
	storesInfo, err := c.GetStores()
	if err != nil {
		return nil, nil, err
	}
	tikvStoresInfo = &StoresInfo{}
	tiflashStoresInfo = &StoresInfo{}
	for _, storeInfo := range storesInfo.Stores {
		var tiflash bool
		for _, label := range storeInfo.Store.Labels {
			if label.Key == "engine" && strings.HasPrefix(label.Value, "tiflash") {
				tiflash = true
				break
			}
		}
		if tiflash {
			tiflashStoresInfo.Stores = append(tiflashStoresInfo.Stores, storeInfo)
		} else {
			tikvStoresInfo.Stores = append(tikvStoresInfo.Stores, storeInfo)
		}
	}
	tikvStoresInfo.Count = len(tikvStoresInfo.Stores)
	tiflashStoresInfo.Count = len(tiflashStoresInfo.Stores)
	return
}

func (c *pdClient) GetTiKVStores() (*StoresInfo, error) {
	tikvStoresInfo, _, err := c.getTiKVAndTiFlashStores()
	if err != nil {
		return nil, err
	}
	return tikvStoresInfo, nil
}

func (c *pdClient) GetTiFlashStores() (*StoresInfo, error) {
	_, tiflashStoresInfo, err := c.getTiKVAndTiFlashStores()
	if err != nil {
		return nil, err
	}
	return tiflashStoresInfo, nil
}

func (c *pdClient) GetTombStoneStores() (*StoresInfo, error) {
	return c.getStores(fmt.Sprintf("%s/%s?state=%d", c.url, storesPrefix, metapb.StoreState_Tombstone))
}

func (c *pdClient) GetStore(storeID uint64) (*StoreInfo, error) {
	apiURL := fmt.Sprintf("%s/%s/%d", c.url, storePrefix, storeID)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	storeInfo := &StoreInfo{}
	err = json.Unmarshal(body, storeInfo)
	if err != nil {
		return nil, err
	}
	return storeInfo, nil
}

func (c *pdClient) DeleteStore(storeID uint64) error {
	var exist bool
	stores, err := c.GetStores()
	if err != nil {
		return err
	}
	for _, store := range stores.Stores {
		if store.Store.GetId() == storeID {
			exist = true
			break
		}
	}
	if !exist {
		return nil
	}
	apiURL := fmt.Sprintf("%s/%s/%d", c.url, storePrefix, storeID)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)

	// Remove an offline store should returns http.StatusOK
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("failed to delete store %d: %v", storeID, string(body))
}

func (c *pdClient) CancelDeleteStore(storeID uint64) error {
	var exist bool
	stores, err := c.GetStores()
	if err != nil {
		return err
	}
	var store *StoreInfo
	for _, store = range stores.Stores {
		if store.Store.GetId() == storeID {
			exist = true
			break
		}
	}
	if !exist {
		klog.Warningf("store %v is not found", storeID)
		return nil
	}
	if store.Store.State != metapb.StoreState_Offline {
		klog.Warningf("store %v is not offline", storeID)
		return nil
	}
	return c.SetStoreState(storeID, "Up")
}

// SetStoreState sets store to specified state.
func (c *pdClient) SetStoreState(storeID uint64, state string) error {
	apiURL := fmt.Sprintf("%s/%s/%d/state?state=%s", c.url, storePrefix, storeID, state)
	req, err := http.NewRequest("POST", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)

	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("failed to set store %d state %s: %v", storeID, state, string(body))
}

// SetStoreLimit sets store limit to specified rate.
func (c *pdClient) SetStoreLimit(storeID uint64, rate float64) error {
	if rate <= 0 {
		return fmt.Errorf("rate should be a number that > 0, but got %v", rate)
	}
	apiURL := fmt.Sprintf("%s/%s/%d/limit", c.url, storePrefix, storeID)
	postInput := map[string]interface{}{
		"rate": rate,
	}
	data, err := json.Marshal(postInput)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)

	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("failed to set store %d limit: %v", storeID, string(body))
}

func (c *pdClient) DeleteMemberByID(memberID uint64) error {
	var exist bool
	members, err := c.GetMembers()
	if err != nil {
		return err
	}
	for _, member := range members.Members {
		if member.MemberId == memberID {
			exist = true
			break
		}
	}
	if !exist {
		return nil
	}
	apiURL := fmt.Sprintf("%s/%s/id/%d", c.url, membersPrefix, memberID)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to delete member %d: %v", res.StatusCode, memberID, err2)
}

func (c *pdClient) DeleteMember(name string) error {
	var exist bool
	members, err := c.GetMembers()
	if err != nil {
		return err
	}
	for _, member := range members.Members {
		if member.Name == name {
			exist = true
			break
		}
	}
	if !exist {
		return nil
	}
	apiURL := fmt.Sprintf("%s/%s/name/%s", c.url, membersPrefix, name)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to delete member %s: %v", res.StatusCode, name, err2)
}

func (c *pdClient) SetStoreLabels(storeID uint64, labels map[string]string) (bool, error) {
	apiURL := fmt.Sprintf("%s/%s/%d/label", c.url, storePrefix, storeID)
	data, err := json.Marshal(labels)
	if err != nil {
		return false, err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return false, err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK {
		return true, nil
	}
	err2 := httputil.ReadErrorBody(res.Body)
	return false, fmt.Errorf("failed %v to set store labels: %v", res.StatusCode, err2)
}

func (c *pdClient) UpdateReplicationConfig(config PDReplicationConfig) error {
	apiURL := fmt.Sprintf("%s/%s", c.url, pdReplicationPrefix)
	data, err := json.Marshal(config)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK {
		return nil
	}
	err = httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to update replication: %v", res.StatusCode, err)
}

func (c *pdClient) BeginEvictLeader(storeID uint64) error {
	leaderEvictInfo := getLeaderEvictSchedulerInfo(storeID)
	apiURL := fmt.Sprintf("%s/%s", c.url, schedulersPrefix)
	data, err := json.Marshal(leaderEvictInfo)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK {
		return nil
	}

	// pd will return an error with the body contains "scheduler existed" if the scheduler already exists
	// this is not the standard response.
	// so these lines are just a workaround for now:
	//   - make a new request to get all schedulers
	//   - return nil if the scheduler already exists
	//
	// when PD returns standard json response, we should get rid of this verbose code.
	evictLeaderSchedulers, err := c.GetEvictLeaderSchedulers()
	if err != nil {
		return err
	}
	for _, s := range evictLeaderSchedulers {
		if s == getLeaderEvictSchedulerStr(storeID) {
			return nil
		}
	}

	err2 := httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to begin evict leader of store:[%d],error: %v", res.StatusCode, storeID, err2)
}

func (c *pdClient) EndEvictLeader(storeID uint64) error {
	sName := getLeaderEvictSchedulerStr(storeID)
	apiURL := fmt.Sprintf("%s/%s/%s", c.url, schedulersPrefix, sName)
	req, err := http.NewRequest("DELETE", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusNotFound {
		return nil
	}
	if res.StatusCode == http.StatusOK {
		klog.Infof("call DELETE method: %s success", apiURL)
	} else {
		err2 := httputil.ReadErrorBody(res.Body)
		klog.Errorf("call DELETE method: %s failed,statusCode: %v,error: %v", apiURL, res.StatusCode, err2)
	}

	// pd will return an error with the body contains "scheduler not found" if the scheduler is not found
	// this is not the standard response.
	// so these lines are just a workaround for now:
	//   - make a new request to get all schedulers
	//   - return nil if the scheduler is not found
	//
	// when PD returns standard json response, we should get rid of this verbose code.
	evictLeaderSchedulers, err := c.GetEvictLeaderSchedulers()
	if err != nil {
		return err
	}
	for _, s := range evictLeaderSchedulers {
		if s == sName {
			return fmt.Errorf("end leader evict scheduler failed,the store:[%d]'s leader evict scheduler is still exist", storeID)
		}
	}

	return nil
}

func (c *pdClient) GetEvictLeaderSchedulers() ([]string, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, schedulersPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	var schedulers []string
	err = json.Unmarshal(body, &schedulers)
	if err != nil {
		return nil, err
	}
	var evicts []string
	for _, scheduler := range schedulers {
		if strings.HasPrefix(scheduler, evictSchedulerLeader) {
			evicts = append(evicts, scheduler)
		}
	}
	evictSchedulers, err := c.filterLeaderEvictScheduler(evicts)
	if err != nil {
		return nil, err
	}
	return evictSchedulers, nil
}

func (c *pdClient) GetEvictLeaderSchedulersForStores(storeIDs ...uint64) (map[uint64]string, error) {
	schedulers, err := c.GetEvictLeaderSchedulers()
	if err != nil {
		return nil, err
	}

	find := func(id uint64) string {
		for _, scheduler := range schedulers {
			sName := getLeaderEvictSchedulerStr(id)
			if scheduler == sName {
				return scheduler
			}
		}
		return ""
	}

	result := make(map[uint64]string)
	for _, id := range storeIDs {
		if scheduler := find(id); scheduler != "" {
			result[id] = scheduler
		}
	}

	return result, nil
}

// getEvictLeaderSchedulerConfig gets the config of PD scheduler "evict-leader-scheduler"
// It's available since PD 3.1.0.
// In the previous versions, PD API returns 404 and this function will return an error.
func (c *pdClient) getEvictLeaderSchedulerConfig() (*evictLeaderSchedulerConfig, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, evictLeaderSchedulerConfigPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	config := &evictLeaderSchedulerConfig{}
	err = json.Unmarshal(body, config)
	if err != nil {
		return nil, err
	}
	return config, nil
}

// This method is to make compatible between old pdapi version and versions after 3.1/4.0.
// To get more detail, see:
// - https://github.com/pingcap/tidb-operator/pull/1831
// - https://github.com/pingcap/pd/issues/2550
func (c *pdClient) filterLeaderEvictScheduler(evictLeaderSchedulers []string) ([]string, error) {
	var schedulerIds []string
	if len(evictLeaderSchedulers) == 1 && evictLeaderSchedulers[0] == evictSchedulerLeader {
		// If there is only one evcit scehduler entry without store ID postfix.
		// We should get the store IDs via scheduler config API and append them
		// to provide consistent results.
		c, err := c.getEvictLeaderSchedulerConfig()
		if err != nil {
			return nil, err
		}
		for k := range c.StoreIDWithRanges {
			schedulerIds = append(schedulerIds, fmt.Sprintf("%s-%v", evictSchedulerLeader, k))
		}
	} else {
		schedulerIds = append(schedulerIds, evictLeaderSchedulers...)
	}
	return schedulerIds, nil
}

func (c *pdClient) GetPDLeader() (*pdpb.Member, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, pdLeaderPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	leader := &pdpb.Member{}
	err = json.Unmarshal(body, leader)
	if err != nil {
		return nil, err
	}
	return leader, nil
}

func (c *pdClient) TransferPDLeader(memberName string) error {
	apiURL := fmt.Sprintf("%s/%s/%s", c.url, pdLeaderTransferPrefix, memberName)
	req, err := http.NewRequest("POST", apiURL, nil)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)
	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	err2 := httputil.ReadErrorBody(res.Body)
	return fmt.Errorf("failed %v to transfer pd leader to %s,error: %v", res.StatusCode, memberName, err2)
}

func (c *pdClient) SetPDLeaderPriority(memberName string, priority int32) error {
	apiURL := fmt.Sprintf("%s/%s/name/%s", c.url, membersPrefix, memberName)
	input := map[string]interface{}{
		"leader-priority": float64(priority),
	}
	b, err := json.Marshal(input)
	if err != nil {
		return err
	}
	req, err := http.NewRequest(http.MethodPost, apiURL, bytes.NewReader(b))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer httputil.DeferClose(resp.Body)

	if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	return fmt.Errorf("fail to set pd %s leader priority %d: %v", memberName, priority, string(body))
}

func (c *pdClient) GetAutoscalingPlans(strategy Strategy) ([]Plan, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, autoscalingPrefix)
	data, err := json.Marshal(strategy)
	if err != nil {
		return nil, err
	}
	body, err := httputil.PostBodyOK(c.httpClient, apiURL, bytes.NewBuffer(data))
	if err != nil {
		return nil, err
	}
	var plans []Plan
	err = json.Unmarshal(body, &plans)
	if err != nil {
		return nil, err
	}
	return plans, nil
}

func (c *pdClient) GetKeyspace(KeyspaceName string) (*KeyspaceInfo, error) {
	apiURL := fmt.Sprintf("%s/%s/%s", c.url, keyspacesPrefix, KeyspaceName)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	keyspaceInfo := &KeyspaceInfo{}
	err = json.Unmarshal(body, keyspaceInfo)
	if err != nil {
		return nil, err
	}
	return keyspaceInfo, nil
}

func (c *pdClient) GetRuleByGroupAndID(group, id string) (*placement.Rule, error) {
	apiURL := fmt.Sprintf("%s/%s/%s/%s", c.url, rulePrefix, group, id)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	rule := &placement.Rule{}
	err = json.Unmarshal(body, rule)
	if err != nil {
		return nil, err
	}
	return rule, nil
}

func (c *pdClient) GetRulesByGroup(group string) ([]*placement.Rule, error) {
	apiURL := fmt.Sprintf("%s/%s/group/%s", c.url, rulesPrefix, group)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	var rules []*placement.Rule
	err = json.Unmarshal(body, &rules)
	if err != nil {
		return nil, err
	}
	return rules, nil
}

func (c *pdClient) SetRules(rules []*placement.Rule) error {
	apiURL := fmt.Sprintf("%s/%s", c.url, rulesBatchPrefix)
	var ops = make([]*placement.RuleOp, 0, len(rules))
	for _, rule := range rules {
		var op = &placement.RuleOp{}
		op.Rule = rule
		if rule.Count > 0 {
			op.Action = placement.RuleOpAdd
			ops = append(ops, op)
		} else if rule.Count == 0 {
			op.Action = placement.RuleOpDel
			ops = append(ops, op)
		}
	}
	data, err := json.Marshal(ops)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)

	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := io.ReadAll(res.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("failed to set rules, msg: %v", string(body))
}

func (c *pdClient) GetRuleGroup(id string) (*placement.RuleGroup, error) {
	apiURL := fmt.Sprintf("%s/%s/%s", c.url, ruleGroupPrefix, id)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	var group placement.RuleGroup
	err = json.Unmarshal(body, &group)
	if err != nil {
		return nil, err
	}
	return &group, nil
}

func (c *pdClient) SetRuleGroup(id string, index int, override bool) error {
	apiURL := fmt.Sprintf("%s/%s", c.url, ruleGroupPrefix)
	postInput := map[string]interface{}{
		"id":       id,
		"index":    index,
		"override": override,
	}
	data, err := json.Marshal(postInput)
	if err != nil {
		return err
	}
	res, err := c.httpClient.Post(apiURL, "application/json", bytes.NewBuffer(data))
	if err != nil {
		return err
	}
	defer httputil.DeferClose(res.Body)

	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
		return nil
	}
	body, err := io.ReadAll(res.Body)
	if err != nil {
		return err
	}

	return fmt.Errorf("failed to set rule group, msg: %v", string(body))
}

func (c *pdClient) GetRuleGroups() ([]*placement.RuleGroup, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, ruleGroupsPrefix)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	var ruleGroups []*placement.RuleGroup
	err = json.Unmarshal(body, &ruleGroups)
	if err != nil {
		return nil, err
	}
	return ruleGroups, nil
}

func (c *pdClient) getHotRegions(apiURL string) (*StoreHotPeersInfos, error) {
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	storeHotPeersInfos := &StoreHotPeersInfos{}
	err = json.Unmarshal(body, storeHotPeersInfos)
	if err != nil {
		return nil, err
	}
	return storeHotPeersInfos, nil
}

func (c *pdClient) GetHotWriteRegions() (*StoreHotPeersInfos, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, hotWriteRegionsPrefix)
	return c.getHotRegions(apiURL)
}

func (c *pdClient) GetHotReadRegions() (*StoreHotPeersInfos, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, hotReadRegionsPrefix)
	return c.getHotRegions(apiURL)
}

func (c *pdClient) ScanRegions(startKey, endKey string, limit int) (*RegionsInfo, error) {
	startBytes, err := hex.DecodeString(startKey)
	if err != nil {
		return nil, err
	}
	startKey = url.QueryEscape(string(startBytes))
	apiURL := fmt.Sprintf("%s/%s?key=%s", c.url, regionsKeyPrefix, startKey)
	if endKey != "" {
		endBytes, err := hex.DecodeString(endKey)
		if err != nil {
			return nil, err
		}
		endKey = url.QueryEscape(string(endBytes))
		apiURL = fmt.Sprintf("%s&end_key=%s", apiURL, endKey)
	}
	if limit >= -1 {
		apiURL = fmt.Sprintf("%s&limit=%d", apiURL, limit)
	}
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	regionsInfo := &RegionsInfo{}
	err = json.Unmarshal(body, regionsInfo)
	if err != nil {
		return nil, err
	}
	return regionsInfo, nil
}

func (c *pdClient) GetRegionByID(regionID uint64) (*RegionInfo, error) {
	apiURL := fmt.Sprintf("%s/%s/%d", c.url, regionIDPrefix, regionID)
	body, err := httputil.GetBodyOK(c.httpClient, apiURL)
	if err != nil {
		return nil, err
	}
	regionInfo := &RegionInfo{}
	err = json.Unmarshal(body, regionInfo)
	if err != nil {
		return nil, err
	}
	return regionInfo, nil
}

func (c *pdClient) GetHotRegions(regionHeartbeatSeconds, storeHeartbeatSeconds, hotRegionSeconds int32) (*HistoryHotRegions, error) {
	// read statistic from store heartbeat and defaults to 10s, write statistic from region heartbeat and defaults to 1m
	ratio := float64(6)
	if regionHeartbeatSeconds > 0 && storeHeartbeatSeconds > 0 {
		ratio = float64(regionHeartbeatSeconds) / float64(storeHeartbeatSeconds)
	}

	hotThreshold := float64(30)
	if hotRegionSeconds > 0 && storeHeartbeatSeconds > 0 {
		hotThreshold = float64(hotRegionSeconds) / float64(storeHeartbeatSeconds)
	}

	hotScore := func(hotRegion *HistoryHotRegion) float64 {
		if hotRegion.HotRegionType == "write" {
			return float64(hotRegion.HotDegree) * ratio
		}
		return float64(hotRegion.HotDegree)
	}

	less := func(i, j *HistoryHotRegion) bool {
		return hotScore(i) < hotScore(j)
	}

	var regions = map[uint64]*HistoryHotRegion{}

	hotWriteRegions, err := c.GetHotWriteRegions()
	if err != nil {
		return nil, err
	}
	klog.Infof("GetHotWriteRegions: %s", hotWriteRegions.SimplifiedString())
	c.packHotRegions(hotWriteRegions.AsPeer, "write", regions, less)

	hotReadRegions, err := c.GetHotReadRegions()
	if err != nil {
		return nil, err
	}
	klog.Infof("GetHotReadRegions: %s", hotReadRegions.SimplifiedString())
	c.packHotRegions(hotReadRegions.AsPeer, "read", regions, less)

	historyHotRegions := make([]*HistoryHotRegion, 0, len(regions))
	for _, hotRegion := range regions {
		historyHotRegions = append(historyHotRegions, hotRegion)
	}
	sort.SliceStable(historyHotRegions, func(i, j int) bool {
		// Reverse sort
		return !less(historyHotRegions[i], historyHotRegions[j])
	})
	cnt := 0
	for idx, hotRegion := range historyHotRegions {
		if hotScore(hotRegion) < hotThreshold {
			break
		}
		cnt = idx + 1
	}
	var hotRegions = &HistoryHotRegions{HistoryHotRegion: historyHotRegions, HotRegionsCount: cnt}

	return hotRegions, nil
}

func (c *pdClient) packHotRegions(storeHotPeersStat StoreHotPeersStat, hotRegionType string, regions map[uint64]*HistoryHotRegion, less func(i, j *HistoryHotRegion) bool) (err error) {
	for _, hotPeersStat := range storeHotPeersStat {
		for _, hotPeerStat := range hotPeersStat.Stats {
			hotRegion := &HistoryHotRegion{
				// store in ms.
				UpdateTime:    hotPeerStat.LastUpdateTime.UnixMilli(),
				RegionID:      hotPeerStat.RegionID,
				StoreID:       hotPeerStat.StoreID,
				IsLeader:      hotPeerStat.IsLeader,
				HotDegree:     int64(hotPeerStat.HotDegree),
				FlowBytes:     hotPeerStat.ByteRate,
				KeyRate:       hotPeerStat.KeyRate,
				QueryRate:     hotPeerStat.QueryRate,
				HotRegionType: hotRegionType,
			}
			if last, ok := regions[hotPeerStat.RegionID]; ok {
				if less(hotRegion, last) {
					continue
				}
				hotRegion.StartKey = last.StartKey
				hotRegion.EndKey = last.EndKey
				hotRegion.Region = last.Region
			} else {
				regionInfo, err := c.GetRegionByID(hotPeerStat.RegionID)
				if err != nil {
					return err
				}
				hotRegion.StartKey = regionInfo.StartKey
				hotRegion.EndKey = regionInfo.EndKey
				hotRegion.Region = regionInfo
			}
			regions[hotPeerStat.RegionID] = hotRegion
		}
	}
	return nil
}

func getLeaderEvictSchedulerInfo(storeID uint64) *schedulerInfo {
	return &schedulerInfo{"evict-leader-scheduler", storeID}
}

func getLeaderEvictSchedulerStr(storeID uint64) string {
	return fmt.Sprintf("%s-%d", "evict-leader-scheduler", storeID)
}

// TiKVNotBootstrappedError represents that TiKV cluster is not bootstrapped yet
type TiKVNotBootstrappedError struct {
	s string
}

func (e *TiKVNotBootstrappedError) Error() string {
	return e.s
}

// TiKVNotBootstrappedErrorf returns a TiKVNotBootstrappedError
func TiKVNotBootstrappedErrorf(format string, a ...interface{}) error {
	return &TiKVNotBootstrappedError{fmt.Sprintf(format, a...)}
}

// IsTiKVNotBootstrappedError returns whether err is a TiKVNotBootstrappedError
func IsTiKVNotBootstrappedError(err error) bool {
	_, ok := err.(*TiKVNotBootstrappedError)
	return ok
}