Skip to content

Commit

Permalink
mcs: support region http interface in scheduling server
Browse files Browse the repository at this point in the history 
Signed-off-by: lhy1024 <[email protected]>
  • Loading branch information
@lhy1024
lhy1024 committed Nov 1, 2023
1 parent 0452c63 commit 37c3c76
Show file tree
Hide file tree
Showing 6 changed files with 527 additions and 197 deletions.
199 changes: 199 additions & 0 deletions pkg/mcs/scheduling/server/apis/v1/api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,12 @@
package apis

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

"github.com/gin-contrib/cors"
Expand All @@ -38,6 +40,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 @@ -169,6 +172,11 @@ func (s *Service) RegisterRegionsRouter() {
router := s.root.Group("regions")
router.GET("/:id/label/:key", getRegionLabelByKey)
router.GET("/:id/labels", getRegionLabels)
router.POST("/accelerate-schedule", accelerateRegionsScheduleInRange)
router.POST("/accelerate-schedule/batch", accelerateRegionsScheduleInRanges)
router.POST("/scatter", scatterRegions)
router.POST("/split", splitRegions)
router.GET("/replicated", checkRegionsReplicated)
}

// RegisterRegionLabelRouter registers the router of the region label handler.
Expand Down Expand Up @@ -708,3 +716,194 @@ 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."
// @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."
// @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."
// @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."
// @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.([]string)
if len(rawSplitKeys) < 1 {
c.String(http.StatusBadRequest, "empty split keys.")
return
}
fmt.Println(rawSplitKeys)
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, _ := c.GetQuery("start_key")
rawEndKey, _ := c.GetQuery("end_key")
state, err := handler.CheckRegionsReplicated(rawStartKey, rawEndKey)
if err != nil {
c.String(http.StatusBadRequest, err.Error())
return
}
c.String(http.StatusOK, state)
}
Loading

0 comments on commit 37c3c76

Please sign in to comment.