Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

mcs: support region http interface in scheduling server #7297

Merged
merged 18 commits into from
Nov 23, 2023
Merged
Show file tree
Hide file tree
Changes from 11 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
214 changes: 214 additions & 0 deletions pkg/mcs/scheduling/server/apis/v1/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,12 @@ package apis

import (
"encoding/hex"
"errors"
"fmt"
"net/http"
"net/url"
"strconv"
"strings"
"sync"

"github.com/gin-contrib/cors"
Expand All @@ -39,6 +42,7 @@ import (
"github.com/tikv/pd/pkg/utils/apiutil"
"github.com/tikv/pd/pkg/utils/apiutil/multiservicesapi"
"github.com/tikv/pd/pkg/utils/logutil"
"github.com/tikv/pd/pkg/utils/typeutil"
"github.com/unrolled/render"
)

Expand Down Expand Up @@ -118,6 +122,7 @@ func NewService(srv *scheserver.Service) *Service {
s.RegisterSchedulersRouter()
s.RegisterCheckersRouter()
s.RegisterHotspotRouter()
s.RegisterRegionsRouter()
return s
}

Expand Down Expand Up @@ -168,6 +173,16 @@ func (s *Service) RegisterOperatorsRouter() {
router.GET("/records", getOperatorRecords)
}

// RegisterRegionsRouter registers the router of the regions handler.
func (s *Service) RegisterRegionsRouter() {
router := s.root.Group("regions")
router.POST("/accelerate-schedule", accelerateRegionsScheduleInRange)
router.POST("/accelerate-schedule/batch", accelerateRegionsScheduleInRanges)
router.POST("/scatter", scatterRegions)
router.POST("/split", splitRegions)
router.GET("/replicated", checkRegionsReplicated)
}

// RegisterConfigRouter registers the router of the config handler.
func (s *Service) RegisterConfigRouter() {
router := s.root.Group("config")
Expand Down Expand Up @@ -1118,3 +1133,202 @@ func getRegionLabelRuleByID(c *gin.Context) {
}
c.IndentedJSON(http.StatusOK, rule)
}

// @Tags region
// @Summary Accelerate regions scheduling a in given range, only receive hex format for keys
// @Accept json
// @Param body body object true "json params"
// @Param limit query integer false "Limit count" default(256)
// @Produce json
// @Success 200 {string} string "Accelerate regions scheduling in a given range [startKey, endKey)"
// @Failure 400 {string} string "The input is invalid."
// @Failure 500 {string} string "PD server failed to proceed the request."
// @Router /regions/accelerate-schedule [post]
func accelerateRegionsScheduleInRange(c *gin.Context) {
handler := c.MustGet(handlerKey).(*handler.Handler)

var input map[string]interface{}
if err := c.BindJSON(&input); err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
rawStartKey, ok1 := input["start_key"].(string)
rawEndKey, ok2 := input["end_key"].(string)
if !ok1 || !ok2 {
c.String(http.StatusBadRequest, "start_key or end_key is not string")
return
}

limitStr, _ := c.GetQuery("limit")
limit, err := handler.AdjustLimit(limitStr, 256 /*default limit*/)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}

err = handler.AccelerateRegionsScheduleInRange(rawStartKey, rawEndKey, limit)
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
c.String(http.StatusOK, fmt.Sprintf("Accelerate regions scheduling in a given range [%s,%s)", rawStartKey, rawEndKey))
}

// @Tags region
// @Summary Accelerate regions scheduling in given ranges, only receive hex format for keys
// @Accept json
// @Param body body object true "json params"
// @Param limit query integer false "Limit count" default(256)
// @Produce json
// @Success 200 {string} string "Accelerate regions scheduling in given ranges [startKey1, endKey1), [startKey2, endKey2), ..."
// @Failure 400 {string} string "The input is invalid."
// @Failure 500 {string} string "PD server failed to proceed the request."
// @Router /regions/accelerate-schedule/batch [post]
func accelerateRegionsScheduleInRanges(c *gin.Context) {
handler := c.MustGet(handlerKey).(*handler.Handler)

var input []map[string]interface{}
if err := c.BindJSON(&input); err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
limitStr, _ := c.GetQuery("limit")
limit, err := handler.AdjustLimit(limitStr, 256 /*default limit*/)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}

var msgBuilder strings.Builder
msgBuilder.Grow(128)
msgBuilder.WriteString("Accelerate regions scheduling in given ranges: ")
var startKeys, endKeys [][]byte
for _, rg := range input {
startKey, rawStartKey, err := apiutil.ParseKey("start_key", rg)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
endKey, rawEndKey, err := apiutil.ParseKey("end_key", rg)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
startKeys = append(startKeys, startKey)
endKeys = append(endKeys, endKey)
msgBuilder.WriteString(fmt.Sprintf("[%s,%s), ", rawStartKey, rawEndKey))
}
err = handler.AccelerateRegionsScheduleInRanges(startKeys, endKeys, limit)
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
c.String(http.StatusOK, msgBuilder.String())
}

// @Tags region
// @Summary Scatter regions by given key ranges or regions id distributed by given group with given retry limit
// @Accept json
// @Param body body object true "json params"
// @Produce json
// @Success 200 {string} string "Scatter regions by given key ranges or regions id distributed by given group with given retry limit"
// @Failure 400 {string} string "The input is invalid."
// @Failure 500 {string} string "PD server failed to proceed the request."
// @Router /regions/scatter [post]
func scatterRegions(c *gin.Context) {
handler := c.MustGet(handlerKey).(*handler.Handler)

var input map[string]interface{}
if err := c.BindJSON(&input); err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
rawStartKey, ok1 := input["start_key"].(string)
rawEndKey, ok2 := input["end_key"].(string)
group, _ := input["group"].(string)
retryLimit := 5
if rl, ok := input["retry_limit"].(float64); ok {
retryLimit = int(rl)
}

opsCount, failures, err := func() (int, map[uint64]error, error) {
if ok1 && ok2 {
return handler.ScatterRegionsByRange(rawStartKey, rawEndKey, group, retryLimit)
}
ids, ok := typeutil.JSONToUint64Slice(input["regions_id"])
if !ok {
return 0, nil, errors.New("regions_id is invalid")
}
return handler.ScatterRegionsByID(ids, group, retryLimit, false)
}()
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
s := handler.BuildScatterRegionsResp(opsCount, failures)
c.IndentedJSON(http.StatusOK, &s)
}

// @Tags region
// @Summary Split regions with given split keys
// @Accept json
// @Param body body object true "json params"
// @Produce json
// @Success 200 {string} string "Split regions with given split keys"
// @Failure 400 {string} string "The input is invalid."
// @Failure 500 {string} string "PD server failed to proceed the request."
// @Router /regions/split [post]
func splitRegions(c *gin.Context) {
handler := c.MustGet(handlerKey).(*handler.Handler)

var input map[string]interface{}
if err := c.BindJSON(&input); err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
s, ok := input["split_keys"]
if !ok {
c.String(http.StatusBadRequest, "split_keys should be provided.")
return
}
rawSplitKeys := s.([]interface{})
if len(rawSplitKeys) < 1 {
c.String(http.StatusBadRequest, "empty split keys.")
return
}
retryLimit := 5
if rl, ok := input["retry_limit"].(float64); ok {
retryLimit = int(rl)
}
s, err := handler.SplitRegions(c.Request.Context(), rawSplitKeys, retryLimit)
if err != nil {
c.String(http.StatusInternalServerError, err.Error())
return
}
c.IndentedJSON(http.StatusOK, &s)
}

// @Tags region
// @Summary Check if regions in the given key ranges are replicated. Returns 'REPLICATED', 'INPROGRESS', or 'PENDING'. 'PENDING' means that there is at least one region pending for scheduling. Similarly, 'INPROGRESS' means there is at least one region in scheduling.
// @Param startKey query string true "Regions start key, hex encoded"
// @Param endKey query string true "Regions end key, hex encoded"
// @Produce plain
// @Success 200 {string} string "INPROGRESS"
// @Failure 400 {string} string "The input is invalid."
// @Router /regions/replicated [get]
func checkRegionsReplicated(c *gin.Context) {
handler := c.MustGet(handlerKey).(*handler.Handler)
rawStartKey, ok1 := c.GetQuery("startKey")
rawEndKey, ok2 := c.GetQuery("endKey")
if !ok1 || !ok2 {
c.String(http.StatusBadRequest, "there is no start_key or end_key")
return
}

state, err := handler.CheckRegionsReplicated(rawStartKey, rawEndKey)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
c.String(http.StatusOK, state)
}