Skip to content

Commit

Permalink
pdctl: support show keyspace group primary (#6747)
Browse files Browse the repository at this point in the history
close #6746

Signed-off-by: lhy1024 <admin@liudos.us>

Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
  • Loading branch information
lhy1024 and ti-chi-bot[bot] authored Jul 10, 2023
1 parent 5759daf commit 61dfad3
Show file tree
Hide file tree
Showing 5 changed files with 148 additions and 9 deletions.
11 changes: 3 additions & 8 deletions pkg/keyspace/tso_keyspace_group.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ package keyspace
import (
"context"
"encoding/json"
"fmt"
"sort"
"strconv"
"strings"
Expand Down Expand Up @@ -1123,14 +1122,10 @@ func (m *GroupManager) GetKeyspaceGroupPrimaryByID(id uint32) (string, error) {
return "", ErrKeyspaceGroupNotExists(id)
}

// default keyspace group: "/ms/{cluster_id}/tso/00000/primary".
// non-default keyspace group: "/ms/{cluster_id}/tso/keyspace_groups/election/{group}/primary".
path := fmt.Sprintf("/ms/%d/tso/00000/primary", m.clusterID)
if id != utils.DefaultKeyspaceGroupID {
path = fmt.Sprintf("/ms/%d/tso/keyspace_groups/election/%05d/primary", m.clusterID, id)
}
rootPath := endpoint.TSOSvcRootPath(m.clusterID)
primaryPath := endpoint.KeyspaceGroupPrimaryPath(rootPath, id)
leader := &tsopb.Participant{}
ok, _, err := etcdutil.GetProtoMsgWithModRev(m.client, path, leader)
ok, _, err := etcdutil.GetProtoMsgWithModRev(m.client, primaryPath, leader)
if err != nil {
return "", err
}
Expand Down
1 change: 1 addition & 0 deletions pkg/tso/keyspace_group_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -674,6 +674,7 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
}

// If the keyspace group is not initialized, initialize it.
// The format of leader name is address-groupID.
uniqueName := fmt.Sprintf("%s-%05d", kgm.electionNamePrefix, group.ID)
uniqueID := memberutil.GenerateUniqueID(uniqueName)
log.Info("joining primary election",
Expand Down
22 changes: 21 additions & 1 deletion server/apiv2/handlers/tso_keyspace_group.go
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,12 @@ func GetKeyspaceGroups(c *gin.Context) {
c.IndentedJSON(http.StatusOK, kgs)
}

// GetKeyspaceGroupPrimaryResponse defines the response for getting primary node of keyspace group.
type GetKeyspaceGroupPrimaryResponse struct {
ID uint32 `json:"id"`
Primary string `json:"primary"`
}

// GetKeyspaceGroupByID gets keyspace group by ID.
func GetKeyspaceGroupByID(c *gin.Context) {
id, err := validateKeyspaceGroupID(c)
Expand All @@ -149,12 +155,26 @@ func GetKeyspaceGroupByID(c *gin.Context) {
c.AbortWithStatusJSON(http.StatusInternalServerError, groupManagerUninitializedErr)
return
}

fields := c.Query("fields") // split by comma if need to add more fields
if fields == "primary" {
primary, err := manager.GetKeyspaceGroupPrimaryByID(id)
if err != nil {
c.AbortWithStatusJSON(http.StatusInternalServerError, err.Error())
return
}
c.JSON(http.StatusOK, &GetKeyspaceGroupPrimaryResponse{
ID: id,
Primary: primary,
})
return
}

kg, err := manager.GetKeyspaceGroupByID(id)
if err != nil {
c.AbortWithStatusJSON(http.StatusInternalServerError, err.Error())
return
}

c.IndentedJSON(http.StatusOK, kg)
}

Expand Down
95 changes: 95 additions & 0 deletions tests/pdctl/keyspace/keyspace_group_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -450,3 +450,98 @@ func TestKeyspaceGroupState(t *testing.T) {
re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes"))
re.NoError(failpoint.Disable("github.com/tikv/pd/server/delayStartServerLoop"))
}

func TestShowKeyspaceGroupPrimary(t *testing.T) {
re := require.New(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes", `return(true)`))
re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/tso/fastGroupSplitPatroller", `return(true)`))
re.NoError(failpoint.Enable("github.com/tikv/pd/server/delayStartServerLoop", `return(true)`))
keyspaces := make([]string, 0)
for i := 0; i < 10; i++ {
keyspaces = append(keyspaces, fmt.Sprintf("keyspace_%d", i))
}
tc, err := tests.NewTestAPICluster(ctx, 3, func(conf *config.Config, serverName string) {
conf.Keyspace.PreAlloc = keyspaces
})
re.NoError(err)
err = tc.RunInitialServers()
re.NoError(err)
pdAddr := tc.GetConfig().GetClientURL()

s1, tsoServerCleanup1, err := tests.StartSingleTSOTestServer(ctx, re, pdAddr, tempurl.Alloc())
defer tsoServerCleanup1()
re.NoError(err)
s2, tsoServerCleanup2, err := tests.StartSingleTSOTestServer(ctx, re, pdAddr, tempurl.Alloc())
defer tsoServerCleanup2()
re.NoError(err)
cmd := pdctlCmd.GetRootCmd()

tc.WaitLeader()
leaderServer := tc.GetServer(tc.GetLeader())
re.NoError(leaderServer.BootstrapCluster())
defaultKeyspaceGroupID := fmt.Sprintf("%d", utils.DefaultKeyspaceGroupID)

// check keyspace group 0 information.
var keyspaceGroup endpoint.KeyspaceGroup
testutil.Eventually(re, func() bool {
args := []string{"-u", pdAddr, "keyspace-group"}
output, err := pdctl.ExecuteCommand(cmd, append(args, defaultKeyspaceGroupID)...)
re.NoError(err)

err = json.Unmarshal(output, &keyspaceGroup)
re.NoError(err)
re.Equal(utils.DefaultKeyspaceGroupID, keyspaceGroup.ID)
return len(keyspaceGroup.Members) == 2
})
for _, member := range keyspaceGroup.Members {
re.Contains([]string{s1.GetAddr(), s2.GetAddr()}, member.Address)
}

// get primary for keyspace group 0.
testutil.Eventually(re, func() bool {
args := []string{"-u", pdAddr, "keyspace-group", "primary", defaultKeyspaceGroupID}
output, err := pdctl.ExecuteCommand(cmd, args...)
re.NoError(err)
var resp handlers.GetKeyspaceGroupPrimaryResponse
json.Unmarshal(output, &resp)
return s1.GetAddr() == resp.Primary || s2.GetAddr() == resp.Primary
})

// split keyspace group.
testutil.Eventually(re, func() bool {
args := []string{"-u", pdAddr, "keyspace-group", "split", "0", "1", "2"}
output, err := pdctl.ExecuteCommand(cmd, args...)
re.NoError(err)
return strings.Contains(string(output), "Success")
})

// check keyspace group 1 information.
testutil.Eventually(re, func() bool {
args := []string{"-u", pdAddr, "keyspace-group"}
output, err := pdctl.ExecuteCommand(cmd, append(args, "1")...)
re.NoError(err)

err = json.Unmarshal(output, &keyspaceGroup)
re.NoError(err)
return len(keyspaceGroup.Members) == 2
})
for _, member := range keyspaceGroup.Members {
re.Contains([]string{s1.GetAddr(), s2.GetAddr()}, member.Address)
}

// get primary for keyspace group 1.
testutil.Eventually(re, func() bool {
args := []string{"-u", pdAddr, "keyspace-group", "primary", "1"}
output, err := pdctl.ExecuteCommand(cmd, args...)
re.NoError(err)
var resp handlers.GetKeyspaceGroupPrimaryResponse
json.Unmarshal(output, &resp)
return s1.GetAddr() == resp.Primary || s2.GetAddr() == resp.Primary
})

re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes"))
re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/tso/fastGroupSplitPatroller"))
re.NoError(failpoint.Disable("github.com/tikv/pd/server/delayStartServerLoop"))
}
28 changes: 28 additions & 0 deletions tools/pd-ctl/pdctl/command/keyspace_group_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ func NewKeyspaceGroupCommand() *cobra.Command {
cmd.AddCommand(newFinishMergeKeyspaceGroupCommand())
cmd.AddCommand(newSetNodesKeyspaceGroupCommand())
cmd.AddCommand(newSetPriorityKeyspaceGroupCommand())
cmd.AddCommand(newShowKeyspaceGroupPrimaryCommand())
cmd.Flags().String("state", "", "state filter")
return cmd
}
Expand Down Expand Up @@ -111,6 +112,15 @@ func newSetPriorityKeyspaceGroupCommand() *cobra.Command {
return r
}

func newShowKeyspaceGroupPrimaryCommand() *cobra.Command {
r := &cobra.Command{
Use: "primary <keyspace_group_id>",
Short: "show th primary of tso nodes for keyspace group with the given ID.",
Run: showKeyspaceGroupPrimaryCommandFunc,
}
return r
}

func showKeyspaceGroupsCommandFunc(cmd *cobra.Command, args []string) {
prefix := keyspaceGroupsPrefix
if len(args) > 1 {
Expand Down Expand Up @@ -337,6 +347,24 @@ func setPriorityKeyspaceGroupCommandFunc(cmd *cobra.Command, args []string) {
})
}

func showKeyspaceGroupPrimaryCommandFunc(cmd *cobra.Command, args []string) {
if len(args) < 1 {
cmd.Usage()
return
}
_, err := strconv.ParseUint(args[0], 10, 32)
if err != nil {
cmd.Printf("Failed to parse the keyspace group ID: %s\n", err)
return
}
r, err := doRequest(cmd, fmt.Sprintf("%s/%s?fields=primary", keyspaceGroupsPrefix, args[0]), http.MethodGet, http.Header{})
if err != nil {
cmd.Printf("Failed to get the keyspace group primary information: %s\n", err)
return
}
cmd.Println(r)
}

func convertToKeyspaceGroup(content string) string {
kg := endpoint.KeyspaceGroup{}
err := json.Unmarshal([]byte(content), &kg)
Expand Down

0 comments on commit 61dfad3

Please sign in to comment.