diff --git a/codecov.yml b/codecov.yml index dc58a648479..44a3aac24b7 100644 --- a/codecov.yml +++ b/codecov.yml @@ -10,3 +10,20 @@ coverage: # basic target: auto threshold: 3% + +comment: + layout: "header, diff, flags" + behavior: default + require_changes: false + +flag_management: + default_rules: # the rules that will be followed for any flag added, generally + carryforward: true + statuses: + - type: project + target: 85% + - type: patch + target: 85% + +ignore: + - tests/** # integration test cases or tools. diff --git a/errors.toml b/errors.toml index ed3cd32d52a..43fc6a582aa 100644 --- a/errors.toml +++ b/errors.toml @@ -606,11 +606,6 @@ error = ''' invalid group settings, please check the group name, priority and the number of resources ''' -["PD:resourcemanager:ErrResourceGroupAlreadyExists"] -error = ''' -the %s resource group already exists -''' - ["PD:schedule:ErrCreateOperator"] error = ''' unable to create operator, %s diff --git a/pkg/errs/errno.go b/pkg/errs/errno.go index 0fda57fa7c9..0bd2a57dba5 100644 --- a/pkg/errs/errno.go +++ b/pkg/errs/errno.go @@ -372,8 +372,7 @@ var ( // Resource Manager errors var ( - ErrResourceGroupAlreadyExists = errors.Normalize("the %s resource group already exists", errors.RFCCodeText("PD:resourcemanager:ErrResourceGroupAlreadyExists")) - ErrResourceGroupNotExists = errors.Normalize("the %s resource group does not exist", errors.RFCCodeText("PD:resourcemanager:ErrGroupNotExists")) - ErrDeleteReservedGroup = errors.Normalize("cannot delete reserved group", errors.RFCCodeText("PD:resourcemanager:ErrDeleteReservedGroup")) - ErrInvalidGroup = errors.Normalize("invalid group settings, please check the group name, priority and the number of resources", errors.RFCCodeText("PD:resourcemanager:ErrInvalidGroup")) + ErrResourceGroupNotExists = errors.Normalize("the %s resource group does not exist", errors.RFCCodeText("PD:resourcemanager:ErrGroupNotExists")) + ErrDeleteReservedGroup = errors.Normalize("cannot delete reserved group", errors.RFCCodeText("PD:resourcemanager:ErrDeleteReservedGroup")) + ErrInvalidGroup = errors.Normalize("invalid group settings, please check the group name, priority and the number of resources", errors.RFCCodeText("PD:resourcemanager:ErrInvalidGroup")) ) diff --git a/pkg/keyspace/keyspace.go b/pkg/keyspace/keyspace.go index a763cafd422..64dd1ba8622 100644 --- a/pkg/keyspace/keyspace.go +++ b/pkg/keyspace/keyspace.go @@ -376,7 +376,6 @@ func (manager *Manager) LoadKeyspace(name string) (*keyspacepb.KeyspaceMeta, err if meta == nil { return ErrKeyspaceNotFound } - meta.Id = id return nil }) return meta, err @@ -399,9 +398,6 @@ func (manager *Manager) LoadKeyspaceByID(spaceID uint32) (*keyspacepb.KeyspaceMe } return nil }) - if meta != nil { - meta.Id = spaceID - } return meta, err } diff --git a/pkg/keyspace/tso_keyspace_group.go b/pkg/keyspace/tso_keyspace_group.go index ce9af860039..dd9319e806f 100644 --- a/pkg/keyspace/tso_keyspace_group.go +++ b/pkg/keyspace/tso_keyspace_group.go @@ -622,7 +622,17 @@ func buildSplitKeyspaces( oldSplit = append(oldSplit, keyspace) } } - return oldSplit, new, nil + // If newNum != len(newKeyspaceMap), it means the provided new keyspace list contains + // duplicate keyspaces, and we need to dedup them (https://github.com/tikv/pd/issues/6687); + // otherwise, we can just return the old split and new keyspace list. + if newNum == len(newKeyspaceMap) { + return oldSplit, new, nil + } + newSplit := make([]uint32, 0, len(newKeyspaceMap)) + for keyspace := range newKeyspaceMap { + newSplit = append(newSplit, keyspace) + } + return oldSplit, newSplit, nil } // Split according to the start and end keyspace ID. if startKeyspaceID == 0 && endKeyspaceID == 0 { @@ -634,7 +644,9 @@ func buildSplitKeyspaces( ) for _, keyspace := range old { if keyspace == utils.DefaultKeyspaceID { - return nil, nil, ErrModifyDefaultKeyspace + // The source keyspace group must be the default keyspace group and we always keep the default + // keyspace in the default keyspace group. + continue } if startKeyspaceID <= keyspace && keyspace <= endKeyspaceID { newSplit = append(newSplit, keyspace) diff --git a/pkg/keyspace/tso_keyspace_group_test.go b/pkg/keyspace/tso_keyspace_group_test.go index 40b779382cd..e8a40a839c8 100644 --- a/pkg/keyspace/tso_keyspace_group_test.go +++ b/pkg/keyspace/tso_keyspace_group_test.go @@ -483,6 +483,26 @@ func TestBuildSplitKeyspaces(t *testing.T) { new: []uint32{6}, err: ErrKeyspaceNotInKeyspaceGroup, }, + { + old: []uint32{1, 2}, + new: []uint32{2, 2}, + expectedOld: []uint32{1}, + expectedNew: []uint32{2}, + }, + { + old: []uint32{0, 1, 2, 3, 4, 5}, + startKeyspaceID: 2, + endKeyspaceID: 4, + expectedOld: []uint32{0, 1, 5}, + expectedNew: []uint32{2, 3, 4}, + }, + { + old: []uint32{0, 1, 2, 3, 4, 5}, + startKeyspaceID: 0, + endKeyspaceID: 4, + expectedOld: []uint32{0, 5}, + expectedNew: []uint32{1, 2, 3, 4}, + }, { old: []uint32{1, 2, 3, 4, 5}, startKeyspaceID: 2, @@ -490,6 +510,13 @@ func TestBuildSplitKeyspaces(t *testing.T) { expectedOld: []uint32{1, 5}, expectedNew: []uint32{2, 3, 4}, }, + { + old: []uint32{1, 2, 3, 4, 5}, + startKeyspaceID: 5, + endKeyspaceID: 6, + expectedOld: []uint32{1, 2, 3, 4}, + expectedNew: []uint32{5}, + }, { old: []uint32{1, 2, 3, 4, 5}, startKeyspaceID: 2, @@ -497,6 +524,13 @@ func TestBuildSplitKeyspaces(t *testing.T) { expectedOld: []uint32{1}, expectedNew: []uint32{2, 3, 4, 5}, }, + { + old: []uint32{1, 2, 3, 4, 5}, + startKeyspaceID: 1, + endKeyspaceID: 1, + expectedOld: []uint32{2, 3, 4, 5}, + expectedNew: []uint32{1}, + }, { old: []uint32{1, 2, 3, 4, 5}, startKeyspaceID: 0, @@ -504,6 +538,13 @@ func TestBuildSplitKeyspaces(t *testing.T) { expectedOld: []uint32{}, expectedNew: []uint32{1, 2, 3, 4, 5}, }, + { + old: []uint32{1, 2, 3, 4, 5}, + startKeyspaceID: 7, + endKeyspaceID: 10, + expectedOld: []uint32{1, 2, 3, 4, 5}, + expectedNew: []uint32{}, + }, { old: []uint32{1, 2, 3, 4, 5}, err: ErrKeyspaceNotInKeyspaceGroup, diff --git a/pkg/keyspace/util.go b/pkg/keyspace/util.go index bf4f8413bb1..240306f8124 100644 --- a/pkg/keyspace/util.go +++ b/pkg/keyspace/util.go @@ -135,7 +135,7 @@ func MaskKeyspaceID(id uint32) uint32 { return id & 0xFF } -// makeKeyRanges encodes keyspace ID to correct LabelRule data. +// RegionBound represents the region boundary of the given keyspace. // For a keyspace with id ['a', 'b', 'c'], it has four boundaries: // // Lower bound for raw mode: ['r', 'a', 'b', 'c'] @@ -147,23 +147,38 @@ func MaskKeyspaceID(id uint32) uint32 { // And shares upper bound with keyspace with id ['a', 'b', 'c + 1']. // These repeated bound will not cause any problem, as repetitive bound will be ignored during rangeListBuild, // but provides guard against hole in keyspace allocations should it occur. -func makeKeyRanges(id uint32) []interface{} { +type RegionBound struct { + RawLeftBound []byte + RawRightBound []byte + TxnLeftBound []byte + TxnRightBound []byte +} + +// MakeRegionBound constructs the correct region boundaries of the given keyspace. +func MakeRegionBound(id uint32) *RegionBound { keyspaceIDBytes := make([]byte, 4) nextKeyspaceIDBytes := make([]byte, 4) binary.BigEndian.PutUint32(keyspaceIDBytes, id) binary.BigEndian.PutUint32(nextKeyspaceIDBytes, id+1) - rawLeftBound := hex.EncodeToString(codec.EncodeBytes(append([]byte{'r'}, keyspaceIDBytes[1:]...))) - rawRightBound := hex.EncodeToString(codec.EncodeBytes(append([]byte{'r'}, nextKeyspaceIDBytes[1:]...))) - txnLeftBound := hex.EncodeToString(codec.EncodeBytes(append([]byte{'x'}, keyspaceIDBytes[1:]...))) - txnRightBound := hex.EncodeToString(codec.EncodeBytes(append([]byte{'x'}, nextKeyspaceIDBytes[1:]...))) + return &RegionBound{ + RawLeftBound: codec.EncodeBytes(append([]byte{'r'}, keyspaceIDBytes[1:]...)), + RawRightBound: codec.EncodeBytes(append([]byte{'r'}, nextKeyspaceIDBytes[1:]...)), + TxnLeftBound: codec.EncodeBytes(append([]byte{'x'}, keyspaceIDBytes[1:]...)), + TxnRightBound: codec.EncodeBytes(append([]byte{'x'}, nextKeyspaceIDBytes[1:]...)), + } +} + +// makeKeyRanges encodes keyspace ID to correct LabelRule data. +func makeKeyRanges(id uint32) []interface{} { + regionBound := MakeRegionBound(id) return []interface{}{ map[string]interface{}{ - "start_key": rawLeftBound, - "end_key": rawRightBound, + "start_key": hex.EncodeToString(regionBound.RawLeftBound), + "end_key": hex.EncodeToString(regionBound.RawRightBound), }, map[string]interface{}{ - "start_key": txnLeftBound, - "end_key": txnRightBound, + "start_key": hex.EncodeToString(regionBound.TxnLeftBound), + "end_key": hex.EncodeToString(regionBound.TxnRightBound), }, } } diff --git a/pkg/mcs/resourcemanager/server/manager.go b/pkg/mcs/resourcemanager/server/manager.go index a98d274f506..b054a37e0ac 100644 --- a/pkg/mcs/resourcemanager/server/manager.go +++ b/pkg/mcs/resourcemanager/server/manager.go @@ -154,6 +154,8 @@ func (m *Manager) Init(ctx context.Context) { } // AddResourceGroup puts a resource group. +// NOTE: AddResourceGroup should also be idempotent because tidb depends +// on this retry mechanism. func (m *Manager) AddResourceGroup(grouppb *rmpb.ResourceGroup) error { // Check the name. if len(grouppb.Name) == 0 || len(grouppb.Name) > 32 { @@ -163,12 +165,6 @@ func (m *Manager) AddResourceGroup(grouppb *rmpb.ResourceGroup) error { if grouppb.GetPriority() > 16 { return errs.ErrInvalidGroup } - m.RLock() - _, ok := m.groups[grouppb.Name] - m.RUnlock() - if ok { - return errs.ErrResourceGroupAlreadyExists.FastGenByArgs(grouppb.Name) - } group := FromProtoResourceGroup(grouppb) m.Lock() defer m.Unlock() diff --git a/server/api/region.go b/server/api/region.go index e17b9c8da25..894a85ecd56 100644 --- a/server/api/region.go +++ b/server/api/region.go @@ -31,6 +31,7 @@ import ( "github.com/pingcap/kvproto/pkg/replication_modepb" "github.com/pingcap/log" "github.com/tikv/pd/pkg/core" + "github.com/tikv/pd/pkg/keyspace" "github.com/tikv/pd/pkg/schedule/filter" "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/utils/apiutil" @@ -398,6 +399,56 @@ func (h *regionsHandler) GetStoreRegions(w http.ResponseWriter, r *http.Request) h.rd.JSON(w, http.StatusOK, regionsInfo) } +// @Tags region +// @Summary List regions belongs to the given keyspace ID. +// @Param keyspace_id query string true "Keyspace ID" +// @Param limit query integer false "Limit count" default(16) +// @Produce json +// @Success 200 {object} RegionsInfo +// @Failure 400 {string} string "The input is invalid." +// @Router /regions/keyspace/id/{id} [get] +func (h *regionsHandler) GetKeyspaceRegions(w http.ResponseWriter, r *http.Request) { + rc := getCluster(r) + vars := mux.Vars(r) + keyspaceIDStr := vars["id"] + if keyspaceIDStr == "" { + h.rd.JSON(w, http.StatusBadRequest, "keyspace id is empty") + return + } + + keyspaceID64, err := strconv.ParseUint(keyspaceIDStr, 10, 32) + if err != nil { + h.rd.JSON(w, http.StatusBadRequest, err.Error()) + return + } + keyspaceID := uint32(keyspaceID64) + keyspaceManager := h.svr.GetKeyspaceManager() + if _, err := keyspaceManager.LoadKeyspaceByID(keyspaceID); err != nil { + h.rd.JSON(w, http.StatusBadRequest, err.Error()) + return + } + + limit := defaultRegionLimit + if limitStr := r.URL.Query().Get("limit"); limitStr != "" { + limit, err = strconv.Atoi(limitStr) + if err != nil { + h.rd.JSON(w, http.StatusBadRequest, err.Error()) + return + } + } + if limit > maxRegionLimit { + limit = maxRegionLimit + } + regionBound := keyspace.MakeRegionBound(keyspaceID) + regions := rc.ScanRegions(regionBound.RawLeftBound, regionBound.RawRightBound, limit) + if limit <= 0 || limit > len(regions) { + txnRegion := rc.ScanRegions(regionBound.TxnLeftBound, regionBound.TxnRightBound, limit-len(regions)) + regions = append(regions, txnRegion...) + } + regionsInfo := convertToAPIRegions(regions) + h.rd.JSON(w, http.StatusOK, regionsInfo) +} + // @Tags region // @Summary List all regions that miss peer. // @Produce json diff --git a/server/api/router.go b/server/api/router.go index 0fc77ba6b19..2b030237340 100644 --- a/server/api/router.go +++ b/server/api/router.go @@ -246,6 +246,7 @@ func createRouter(prefix string, svr *server.Server) *mux.Router { registerFunc(clusterRouter, "/regions/key", regionsHandler.ScanRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) registerFunc(clusterRouter, "/regions/count", regionsHandler.GetRegionCount, setMethods(http.MethodGet), setAuditBackend(prometheus)) registerFunc(clusterRouter, "/regions/store/{id}", regionsHandler.GetStoreRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) + registerFunc(clusterRouter, "/regions/keyspace/id/{id}", regionsHandler.GetKeyspaceRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) registerFunc(clusterRouter, "/regions/writeflow", regionsHandler.GetTopWriteFlowRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) registerFunc(clusterRouter, "/regions/readflow", regionsHandler.GetTopReadFlowRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) registerFunc(clusterRouter, "/regions/confver", regionsHandler.GetTopConfVerRegions, setMethods(http.MethodGet), setAuditBackend(prometheus)) diff --git a/server/apiv2/handlers/tso_keyspace_group.go b/server/apiv2/handlers/tso_keyspace_group.go index 5de8fd6a9cc..7030c332406 100644 --- a/server/apiv2/handlers/tso_keyspace_group.go +++ b/server/apiv2/handlers/tso_keyspace_group.go @@ -40,9 +40,9 @@ func RegisterTSOKeyspaceGroup(r *gin.RouterGroup) { router.GET("", GetKeyspaceGroups) router.GET("/:id", GetKeyspaceGroupByID) router.DELETE("/:id", DeleteKeyspaceGroupByID) + router.PATCH("/:id", SetNodesForKeyspaceGroup) // only to support set nodes + router.PATCH("/:id/*node", SetPriorityForKeyspaceGroup) // only to support set priority router.POST("/:id/alloc", AllocNodesForKeyspaceGroup) - router.POST("/:id/nodes", SetNodesForKeyspaceGroup) - router.POST("/:id/priority", SetPriorityForKeyspaceGroup) router.POST("/:id/split", SplitKeyspaceGroupByID) router.DELETE("/:id/split", FinishSplitKeyspaceByID) router.POST("/:id/merge", MergeKeyspaceGroups) @@ -436,8 +436,7 @@ func SetNodesForKeyspaceGroup(c *gin.Context) { // SetPriorityForKeyspaceGroupParams defines the params for setting priority of tso node for the keyspace group. type SetPriorityForKeyspaceGroupParams struct { - Node string `json:"node"` - Priority int `json:"priority"` + Priority int `json:"priority"` } // SetPriorityForKeyspaceGroup sets priority of tso node for the keyspace group. @@ -447,6 +446,11 @@ func SetPriorityForKeyspaceGroup(c *gin.Context) { c.AbortWithStatusJSON(http.StatusBadRequest, "invalid keyspace group id") return } + node, err := parseNodeAddress(c) + if err != nil { + c.AbortWithStatusJSON(http.StatusBadRequest, "invalid node address") + return + } svr := c.MustGet(middlewares.ServerContextKey).(*server.Server) manager := svr.GetKeyspaceGroupManager() if manager == nil { @@ -468,12 +472,12 @@ func SetPriorityForKeyspaceGroup(c *gin.Context) { // check if node exists members := kg.Members if slice.NoneOf(members, func(i int) bool { - return members[i].Address == setParams.Node + return members[i].Address == node }) { c.AbortWithStatusJSON(http.StatusBadRequest, "tso node does not exist in the keyspace group") } // set priority - err = manager.SetPriorityForKeyspaceGroup(id, setParams.Node, setParams.Priority) + err = manager.SetPriorityForKeyspaceGroup(id, node, setParams.Priority) if err != nil { c.AbortWithStatusJSON(http.StatusInternalServerError, err.Error()) return @@ -492,6 +496,19 @@ func validateKeyspaceGroupID(c *gin.Context) (uint32, error) { return uint32(id), nil } +func parseNodeAddress(c *gin.Context) (string, error) { + node := c.Param("node") + if node == "" { + return "", errors.New("invalid node address") + } + // In pd-ctl, we use url.PathEscape to escape the node address and replace the % to \%. + // But in the gin framework, it will unescape the node address automatically. + // So we need to replace the \/ to /. + node = strings.ReplaceAll(node, "\\/", "/") + node = strings.TrimPrefix(node, "/") + return node, nil +} + func isValid(id uint32) bool { return id >= utils.DefaultKeyspaceGroupID && id <= utils.MaxKeyspaceGroupCountInUse } diff --git a/server/keyspace_service.go b/server/keyspace_service.go index 8336b988ece..c5fa7377178 100644 --- a/server/keyspace_service.go +++ b/server/keyspace_service.go @@ -80,6 +80,7 @@ func (s *KeyspaceServer) WatchKeyspaces(request *keyspacepb.WatchKeyspacesReques putFn := func(kv *mvccpb.KeyValue) error { meta := &keyspacepb.KeyspaceMeta{} if err := proto.Unmarshal(kv.Value, meta); err != nil { + defer cancel() // cancel context to stop watcher return err } keyspaces = append(keyspaces, meta) @@ -92,9 +93,14 @@ func (s *KeyspaceServer) WatchKeyspaces(request *keyspacepb.WatchKeyspacesReques defer func() { keyspaces = keyspaces[:0] }() - return stream.Send(&keyspacepb.WatchKeyspacesResponse{ + err := stream.Send(&keyspacepb.WatchKeyspacesResponse{ Header: s.header(), Keyspaces: keyspaces}) + if err != nil { + defer cancel() // cancel context to stop watcher + return err + } + return nil } watcher := etcdutil.NewLoopWatcher( diff --git a/server/server.go b/server/server.go index 355dd834f35..ae1b47d9b40 100644 --- a/server/server.go +++ b/server/server.go @@ -1860,6 +1860,7 @@ func (s *Server) initTSOPrimaryWatcher() { if len(listenUrls) > 0 { // listenUrls[0] is the primary service endpoint of the keyspace group s.servicePrimaryMap.Store(serviceName, listenUrls[0]) + log.Info("update tso primary", zap.String("primary", listenUrls[0])) } return nil } diff --git a/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go b/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go index 41bcba0e90b..dc33016eafb 100644 --- a/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go +++ b/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go @@ -354,7 +354,7 @@ func (suite *keyspaceGroupTestSuite) tryGetKeyspaceGroup(id uint32) (*endpoint.K func (suite *keyspaceGroupTestSuite) trySetNodesForKeyspaceGroup(id int, request *handlers.SetNodesForKeyspaceGroupParams) (*endpoint.KeyspaceGroup, int) { data, err := json.Marshal(request) suite.NoError(err) - httpReq, err := http.NewRequest(http.MethodPost, suite.server.GetAddr()+keyspaceGroupsPrefix+fmt.Sprintf("/%d/nodes", id), bytes.NewBuffer(data)) + httpReq, err := http.NewRequest(http.MethodPatch, suite.server.GetAddr()+keyspaceGroupsPrefix+fmt.Sprintf("/%d", id), bytes.NewBuffer(data)) suite.NoError(err) resp, err := suite.dialClient.Do(httpReq) suite.NoError(err) diff --git a/tests/integrations/mcs/resourcemanager/resource_manager_test.go b/tests/integrations/mcs/resourcemanager/resource_manager_test.go index 467eba6c518..5cbed81d57e 100644 --- a/tests/integrations/mcs/resourcemanager/resource_manager_test.go +++ b/tests/integrations/mcs/resourcemanager/resource_manager_test.go @@ -711,7 +711,7 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() { testCasesSet1 := []struct { name string mode rmpb.GroupMode - addSuccess bool + isNewGroup bool modifySuccess bool expectMarshal string modifySettings func(*rmpb.ResourceGroup) @@ -789,8 +789,8 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() { } // Create Resource Group resp, err := cli.AddResourceGroup(suite.ctx, group) - checkErr(err, tcase.addSuccess) - if tcase.addSuccess { + checkErr(err, true) + if tcase.isNewGroup { finalNum++ re.Contains(resp, "Success!") } @@ -860,11 +860,9 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() { resp, err := http.Post(getAddr(i)+"/resource-manager/api/v1/config/group", "application/json", strings.NewReader(string(createJSON))) re.NoError(err) defer resp.Body.Close() - if tcase.addSuccess { - re.Equal(http.StatusOK, resp.StatusCode) + re.Equal(http.StatusOK, resp.StatusCode) + if tcase.isNewGroup { finalNum++ - } else { - re.Equal(http.StatusInternalServerError, resp.StatusCode) } // Modify Resource Group diff --git a/tools/pd-ctl/pdctl/command/global.go b/tools/pd-ctl/pdctl/command/global.go index 8d888b60b1f..85fe63ac8be 100644 --- a/tools/pd-ctl/pdctl/command/global.go +++ b/tools/pd-ctl/pdctl/command/global.go @@ -165,7 +165,7 @@ func getEndpoints(cmd *cobra.Command) []string { return strings.Split(addrs, ",") } -func postJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) { +func requestJSON(cmd *cobra.Command, method, prefix string, input map[string]interface{}) { data, err := json.Marshal(input) if err != nil { cmd.Println(err) @@ -175,19 +175,31 @@ func postJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) { endpoints := getEndpoints(cmd) err = tryURLs(cmd, endpoints, func(endpoint string) error { var msg []byte - var r *http.Response + var req *http.Request + var resp *http.Response url := endpoint + "/" + prefix - r, err = dialClient.Post(url, "application/json", bytes.NewBuffer(data)) + switch method { + case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete, http.MethodGet: + req, err = http.NewRequest(method, url, bytes.NewBuffer(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err = dialClient.Do(req) + default: + err := errors.Errorf("method %s not supported", method) + return err + } if err != nil { return err } - defer r.Body.Close() - if r.StatusCode != http.StatusOK { - msg, err = io.ReadAll(r.Body) + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + msg, err = io.ReadAll(resp.Body) if err != nil { return err } - return errors.Errorf("[%d] %s", r.StatusCode, msg) + return errors.Errorf("[%d] %s", resp.StatusCode, msg) } return nil }) @@ -198,6 +210,14 @@ func postJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) { cmd.Println("Success!") } +func postJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) { + requestJSON(cmd, http.MethodPost, prefix, input) +} + +func patchJSON(cmd *cobra.Command, prefix string, input map[string]interface{}) { + requestJSON(cmd, http.MethodPatch, prefix, input) +} + // do send a request to server. Default is Get. func do(endpoint, prefix, method string, resp *string, customHeader http.Header, b *bodyOption) error { var err error diff --git a/tools/pd-ctl/pdctl/command/keyspace_group_command.go b/tools/pd-ctl/pdctl/command/keyspace_group_command.go index a4be612a301..b5acf0fa7e8 100644 --- a/tools/pd-ctl/pdctl/command/keyspace_group_command.go +++ b/tools/pd-ctl/pdctl/command/keyspace_group_command.go @@ -288,17 +288,17 @@ func setNodesKeyspaceGroupCommandFunc(cmd *cobra.Command, args []string) { cmd.Printf("Failed to parse the keyspace group ID: %s\n", err) return } - addresses := make([]string, 0, len(args)-1) + nodes := make([]string, 0, len(args)-1) for _, arg := range args[1:] { u, err := url.ParseRequestURI(arg) if u == nil || err != nil { cmd.Printf("Failed to parse the tso node address: %s\n", err) return } - addresses = append(addresses, arg) + nodes = append(nodes, arg) } - postJSON(cmd, fmt.Sprintf("%s/%s/nodes", keyspaceGroupsPrefix, args[0]), map[string]interface{}{ - "Nodes": addresses, + patchJSON(cmd, fmt.Sprintf("%s/%s", keyspaceGroupsPrefix, args[0]), map[string]interface{}{ + "Nodes": nodes, }) } @@ -313,21 +313,26 @@ func setPriorityKeyspaceGroupCommandFunc(cmd *cobra.Command, args []string) { return } - address := args[1] - u, err := url.ParseRequestURI(address) + node := args[1] + u, err := url.ParseRequestURI(node) if u == nil || err != nil { cmd.Printf("Failed to parse the tso node address: %s\n", err) return } + // Escape the node address to avoid the error of parsing the url + // But the url.PathEscape will escape the '/' to '%2F', which % will cause the error of parsing the url + // So we need to replace the % to \% + node = url.PathEscape(node) + node = strings.ReplaceAll(node, "%", "\\%") + priority, err := strconv.ParseInt(args[2], 10, 32) if err != nil { cmd.Printf("Failed to parse the priority: %s\n", err) return } - postJSON(cmd, fmt.Sprintf("%s/%s/priority", keyspaceGroupsPrefix, args[0]), map[string]interface{}{ - "Node": address, + patchJSON(cmd, fmt.Sprintf("%s/%s/%s", keyspaceGroupsPrefix, args[0], node), map[string]interface{}{ "Priority": priority, }) } diff --git a/tools/pd-ctl/pdctl/command/region_command.go b/tools/pd-ctl/pdctl/command/region_command.go index fcebb30e6d8..33191bbe12b 100644 --- a/tools/pd-ctl/pdctl/command/region_command.go +++ b/tools/pd-ctl/pdctl/command/region_command.go @@ -45,6 +45,7 @@ var ( regionsKeyPrefix = "pd/api/v1/regions/key" regionsSiblingPrefix = "pd/api/v1/regions/sibling" regionsRangeHolesPrefix = "pd/api/v1/regions/range-holes" + regionsKeyspacePrefix = "pd/api/v1/regions/keyspace" regionIDPrefix = "pd/api/v1/region/id" regionKeyPrefix = "pd/api/v1/region/key" ) @@ -60,6 +61,7 @@ func NewRegionCommand() *cobra.Command { r.AddCommand(NewRegionWithCheckCommand()) r.AddCommand(NewRegionWithSiblingCommand()) r.AddCommand(NewRegionWithStoreCommand()) + r.AddCommand(NewRegionWithKeyspaceCommand()) r.AddCommand(NewRegionsByKeysCommand()) r.AddCommand(NewRangesWithRangeHolesCommand()) @@ -463,6 +465,43 @@ func showRegionWithStoreCommandFunc(cmd *cobra.Command, args []string) { cmd.Println(r) } +// NewRegionWithKeyspaceCommand returns regions with keyspace subcommand of regionCmd +func NewRegionWithKeyspaceCommand() *cobra.Command { + r := &cobra.Command{ + Use: "keyspace ", + Short: "show region information of the given keyspace", + } + r.AddCommand(&cobra.Command{ + Use: "id ", + Short: "show region information for the given keyspace id", + Run: showRegionWithKeyspaceCommandFunc, + }) + return r +} + +func showRegionWithKeyspaceCommandFunc(cmd *cobra.Command, args []string) { + if len(args) < 1 || len(args) > 2 { + cmd.Println(cmd.UsageString()) + return + } + + keyspaceID := args[0] + prefix := regionsKeyspacePrefix + "/id/" + keyspaceID + if len(args) == 2 { + if _, err := strconv.Atoi(args[1]); err != nil { + cmd.Println("limit should be a number") + return + } + prefix += "?limit=" + args[1] + } + r, err := doRequest(cmd, prefix, http.MethodGet, http.Header{}) + if err != nil { + cmd.Printf("Failed to get regions with the given keyspace: %s\n", err) + return + } + cmd.Println(r) +} + const ( rangeHolesLongDesc = `There are some cases that the region range is not continuous, for example, the region doesn't send the heartbeat to PD after a splitting. This command will output all empty ranges without any region info.`