diff --git a/pkg/mcs/tso/server/apis/v1/api.go b/pkg/mcs/tso/server/apis/v1/api.go index 6062ebd0a7..b4fa5fa66a 100644 --- a/pkg/mcs/tso/server/apis/v1/api.go +++ b/pkg/mcs/tso/server/apis/v1/api.go @@ -35,6 +35,7 @@ import ( "github.com/tikv/pd/pkg/errs" "github.com/tikv/pd/pkg/keyspace/constant" + "github.com/tikv/pd/pkg/mcs/discovery" tsoserver "github.com/tikv/pd/pkg/mcs/tso/server" "github.com/tikv/pd/pkg/mcs/utils" mcs "github.com/tikv/pd/pkg/mcs/utils/constant" @@ -389,28 +390,58 @@ func transferPrimary(c *gin.Context) { } // evictPrimary transfers away every keyspace group primary currently held by this -// node, moving each to another member of the same group. It is a best-effort, -// node-local operation: groups that this node is not serving as primary are -// skipped, and a failure on one group does not stop the others. +// node, moving each to another member of the same group. Groups that this node is +// not serving as primary are skipped. // -// The request is rejected as a whole if this node is the primary of any -// splitting keyspace group: a split target must campaign on the same TSO node as -// its split source, but eviction transfers each group to an independently chosen -// member, which could break that invariant and leave the target keyspaces -// without a primary. A single TransferPrimary is not atomic and the groups are -// iterated in no particular order, so the split state is checked for every -// candidate group up front before transferring anything; otherwise a normal -// group could be moved before a later splitting group aborts the loop, leaving -// partial side effects. Since splitting is transient, the caller should retry -// once it finishes. +// The request is rejected as a whole, before any group is touched, if this node +// is the primary of any splitting keyspace group or if new_primary does not +// identify a member of every candidate group: a split target must campaign on +// the same TSO node as its split source, but eviction transfers each group to an +// independently chosen member, which could break that invariant and leave the +// target keyspaces without a primary; an invalid new_primary would otherwise only +// be discovered mid-loop, after other groups had already been moved. A single +// TransferPrimary is not atomic and the groups are iterated in no particular +// order, so both conditions are checked for every candidate group up front; +// otherwise a normal group could be moved before a later group aborts the loop, +// leaving partial side effects. Since splitting is transient, the caller should +// retry once it finishes. Once past this pre-check, a transfer failure for an +// individual group (e.g. a transient etcd error) is best-effort and does not stop +// the others. // @Tags primary // @Summary Evict all keyspace group primaries held by this node. // @Produce json -// @Success 200 {object} map[string]string -// @Failure 500 {object} map[string]string +// @Param new_primary body string false "new primary name" +// @Success 200 {object} map[string]string +// @Failure 400 {string} string "invalid request" +// @Failure 500 {object} map[string]string // @Router /primary/evict [post] func evictPrimary(c *gin.Context) { svr := c.MustGet(multiservicesapi.ServiceContextKey).(*tsoserver.Service) + + body, err := io.ReadAll(c.Request.Body) + if err != nil { + c.String(http.StatusBadRequest, err.Error()) + return + } + var input struct { + NewPrimary string `json:"new_primary"` + } + if len(body) > 0 { + if err := json.Unmarshal(body, &input); err != nil { + c.String(http.StatusBadRequest, err.Error()) + return + } + } + // The node being evicted cannot also be its own replacement: TransferPrimary + // treats a target equal to the current primary as a self-transfer and silently + // no-ops, which would report "success" while leaving the node undrained. Match + // on both name and service address since IsValidPrimaryCandidate and + // TransferPrimary accept either as an identifier for new_primary. + if input.NewPrimary != "" && (input.NewPrimary == svr.Name() || input.NewPrimary == svr.GetAdvertiseListenAddr()) { + c.String(http.StatusBadRequest, "new_primary must not be the node being evicted") + return + } + kgm := svr.GetKeyspaceGroupManager() // Collect the keyspace groups this node is currently the primary of. There is @@ -424,15 +455,36 @@ func evictPrimary(c *gin.Context) { primaryGroupIDs = append(primaryGroupIDs, keyspaceGroupID) } + // Resolve the service registry once and reuse it for every candidate group + // below, instead of re-fetching per group. + var entries []discovery.ServiceRegistryEntry + if input.NewPrimary != "" { + entries, err = discovery.GetMSMembers(mcs.TSOServiceName, svr.GetClient()) + if err != nil { + c.AbortWithStatusJSON(http.StatusInternalServerError, err.Error()) + return + } + } + // Pre-check every candidate group before transferring anything so the // operation is all-or-nothing: reject the whole request if any of them is - // splitting, instead of moving some groups first and only then aborting. + // splitting or new_primary is not one of its members, instead of moving some + // groups first and only then aborting. for _, keyspaceGroupID := range primaryGroupIDs { - if group := kgm.GetKeyspaceGroupByID(keyspaceGroupID); group != nil && group.IsSplitting() { + group := kgm.GetKeyspaceGroupByID(keyspaceGroupID) + if group == nil { + continue + } + if group.IsSplitting() { c.AbortWithStatusJSON(http.StatusInternalServerError, errs.ErrKeyspaceGroupInSplit.FastGenByArgs(keyspaceGroupID).Error()) return } + if !utils.IsValidPrimaryCandidate(entries, input.NewPrimary, keyspaceGroupMemberMap(group)) { + c.String(http.StatusBadRequest, + fmt.Sprintf("new_primary %q is not a member of keyspace group %d", input.NewPrimary, keyspaceGroupID)) + return + } } // results maps a keyspace group ID to the transfer outcome ("success" or the @@ -458,10 +510,7 @@ func evictPrimary(c *gin.Context) { } // only members of the specific group are valid primary candidates. - memberMap := make(map[string]bool, len(group.Members)) - for _, member := range group.Members { - memberMap[member.Address] = true - } + memberMap := keyspaceGroupMemberMap(group) participant, ok := allocator.GetMember().(*member.Participant) if !ok { @@ -471,9 +520,8 @@ func evictPrimary(c *gin.Context) { // has a higher priority for the group, the priority checker will move the // primary back to it, so the eviction does not durably drain the node. // Priority handling is being reworked, so revisit this when needed. - // An empty new primary lets TransferPrimary pick a random other member. if err := utils.TransferPrimary(svr.GetClient(), participant, - mcs.TSOServiceName, svr.Name(), "", keyspaceGroupID, memberMap); err != nil { + mcs.TSOServiceName, svr.Name(), input.NewPrimary, keyspaceGroupID, memberMap); err != nil { log.Warn("failed to evict keyspace group primary", zap.Uint32("keyspace-group-id", keyspaceGroupID), errs.ZapError(err)) results[keyspaceGroupID] = err.Error() @@ -490,6 +538,16 @@ func evictPrimary(c *gin.Context) { c.IndentedJSON(http.StatusOK, results) } +// keyspaceGroupMemberMap returns the set of service addresses that are members of +// the group, so callers can filter transfer candidates down to that group. +func keyspaceGroupMemberMap(group *endpoint.KeyspaceGroup) map[string]bool { + memberMap := make(map[string]bool, len(group.Members)) + for _, member := range group.Members { + memberMap[member.Address] = true + } + return memberMap +} + // forwardToGroupPrimary forwards the transfer primary request to the primary of // the given keyspace group, replaying the original request body. The request is // fully handled (forwarded or aborted with an error) when this returns. diff --git a/pkg/mcs/utils/expected_primary.go b/pkg/mcs/utils/expected_primary.go index 36d90dba80..0310d2cda2 100644 --- a/pkg/mcs/utils/expected_primary.go +++ b/pkg/mcs/utils/expected_primary.go @@ -237,3 +237,26 @@ func TransferPrimary(client *clientv3.Client, p *member.Participant, serviceName func isSamePrimary(member discovery.ServiceRegistryEntry, primary string) bool { return primary != "" && (member.Name == primary || member.ServiceAddr == primary) } + +// IsValidPrimaryCandidate reports whether newPrimary identifies a member of the +// group represented by tsoMembersMap, given the already-fetched registry entries +// for the service, so a caller can reject an invalid target up front instead of +// discovering it only when TransferPrimary itself fails. Callers checking multiple +// groups for one request should fetch entries once (e.g. via discovery.GetMSMembers) +// and reuse them here, rather than re-fetching per group. An empty newPrimary +// always matches: it means "let TransferPrimary pick any member", which is valid +// for every group. +func IsValidPrimaryCandidate(entries []discovery.ServiceRegistryEntry, newPrimary string, tsoMembersMap map[string]bool) bool { + if newPrimary == "" { + return true + } + for _, member := range entries { + if tsoMembersMap != nil && !tsoMembersMap[member.ServiceAddr] { + continue + } + if isSamePrimary(member, newPrimary) { + return true + } + } + return false +} diff --git a/pkg/mcs/utils/expected_primary_test.go b/pkg/mcs/utils/expected_primary_test.go index e10b84e5a2..e66a179247 100644 --- a/pkg/mcs/utils/expected_primary_test.go +++ b/pkg/mcs/utils/expected_primary_test.go @@ -117,3 +117,23 @@ func TestIsSamePrimary(t *testing.T) { re.False(isSamePrimary(entry, "http://127.0.0.1:2380")) // different address re.False(isSamePrimary(entry, "")) // empty target never matches } + +// TestIsValidPrimaryCandidate covers the pre-check evictPrimary uses to reject an +// out-of-group new_primary before transferring anything, instead of discovering it +// mid-loop after other groups have already been moved. +func TestIsValidPrimaryCandidate(t *testing.T) { + re := require.New(t) + entries := []discovery.ServiceRegistryEntry{ + {Name: "tso-1", ServiceAddr: "http://127.0.0.1:1"}, + {Name: "tso-2", ServiceAddr: "http://127.0.0.1:2"}, + {Name: "tso-3", ServiceAddr: "http://127.0.0.1:3"}, + } + // Only tso-1 and tso-2 belong to the group under evaluation. + groupMembers := map[string]bool{"http://127.0.0.1:1": true, "http://127.0.0.1:2": true} + + re.True(IsValidPrimaryCandidate(entries, "", groupMembers), "an empty target always matches") + re.True(IsValidPrimaryCandidate(entries, "tso-2", groupMembers), "tso-2 is a group member") + re.True(IsValidPrimaryCandidate(entries, "http://127.0.0.1:2", groupMembers), "matching by service address must also work") + re.False(IsValidPrimaryCandidate(entries, "tso-3", groupMembers), "tso-3 is registered but not a member of this group") + re.False(IsValidPrimaryCandidate(entries, "tso-unknown", groupMembers), "an unregistered name never matches") +} diff --git a/tests/integrations/mcs/members/member_test.go b/tests/integrations/mcs/members/member_test.go index 1c2e1df411..28d75954e0 100644 --- a/tests/integrations/mcs/members/member_test.go +++ b/tests/integrations/mcs/members/member_test.go @@ -422,6 +422,128 @@ func (suite *memberTestSuite) TestEvictPrimary() { return len(served) == len(groupIDs) }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) } + + // Verify that eviction with an explicit new_primary transfers all groups to + // the designated node instead of picking a random member. src and dst are + // both picked from the default keyspace group's members so dst is guaranteed + // a valid transfer target even if src also happens to be the default group's + // primary: unlike the 12 groups created above, the default group only + // replicates on DefaultKeyspaceGroupReplicaCount (2) of the 3 tso nodes, so + // an arbitrary dst could otherwise get rejected as "not a member" for it. + // suite.tsoAvailMembers is captured once in SetupTest right after the nodes + // start, before default group allocation is guaranteed to have converged to + // its full replica count, so poll the current membership here instead of + // trusting that snapshot: check every node and use whichever one first + // reports the default group at its full replica count. + defaultGroupMemberAddrs := make(map[string]bool, mcs.DefaultKeyspaceGroupReplicaCount) + testutil.Eventually(re, func() bool { + for _, node := range nodeList { + g, ok := mustGetKeyspaceGroupMembers(re, node.(*tso.Server))[constant.DefaultKeyspaceGroupID] + if !ok || len(g.Group.Members) != mcs.DefaultKeyspaceGroupReplicaCount { + continue + } + for _, m := range g.Group.Members { + defaultGroupMemberAddrs[m.Address] = true + } + return true + } + return false + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(100*time.Millisecond)) + + var defaultGroupNodes []bs.Server + for _, node := range nodeList { + if defaultGroupMemberAddrs[node.GetAddr()] { + defaultGroupNodes = append(defaultGroupNodes, node) + } + } + re.Len(defaultGroupNodes, mcs.DefaultKeyspaceGroupReplicaCount) + src := defaultGroupNodes[0] + dst := defaultGroupNodes[1] + + // A new_primary that identifies the node being evicted, whether by name or by + // service address, must be rejected outright instead of silently no-oping + // while reporting success. + for _, self := range []string{src.Name(), src.(*tso.Server).GetAdvertiseListenAddr()} { + selfEvictData, err := json.Marshal(map[string]any{"new_primary": self}) + re.NoError(err) + resp, err := tests.TestDialClient.Post(src.GetAddr()+"/tso/api/v1/primary/evict", + "application/json", bytes.NewBuffer(selfEvictData)) + re.NoError(err) + re.Equal(http.StatusBadRequest, resp.StatusCode, "new_primary=%q should be rejected", self) + re.NoError(resp.Body.Close()) + } + + for _, id := range groupIDs { + transferData, err := json.Marshal(map[string]any{ + "new_primary": src.Name(), + "keyspace_group_id": id, + }) + re.NoError(err) + testutil.Eventually(re, func() bool { + resp, err := tests.TestDialClient.Post(src.GetAddr()+"/tso/api/v1/primary/transfer", + "application/json", bytes.NewBuffer(transferData)) + if err != nil { + return false + } + ok := resp.StatusCode == http.StatusOK + return resp.Body.Close() == nil && ok + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + } + testutil.Eventually(re, func() bool { + serving := mustGetKeyspaceGroupMembers(re, src.(*tso.Server)) + for _, id := range groupIDs { + if serving[id] == nil || !serving[id].IsPrimary { + return false + } + } + return true + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + + evictData, err := json.Marshal(map[string]any{ + "new_primary": dst.Name(), + }) + re.NoError(err) + resp, err := tests.TestDialClient.Post(src.GetAddr()+"/tso/api/v1/primary/evict", + "application/json", bytes.NewBuffer(evictData)) + re.NoError(err) + body, err := io.ReadAll(resp.Body) + re.NoError(err) + re.NoError(resp.Body.Close()) + results := make(map[uint32]string) + re.NoError(json.Unmarshal(body, &results), string(body)) + re.Equal(http.StatusOK, resp.StatusCode) + for _, id := range groupIDs { + re.Equalf("success", results[id], "group %d result: %q", id, results[id]) + } + // All primaries must have moved to the designated node. + testutil.Eventually(re, func() bool { + serving := mustGetKeyspaceGroupMembers(re, dst.(*tso.Server)) + for _, id := range groupIDs { + if serving[id] == nil || !serving[id].IsPrimary { + return false + } + } + return true + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + + // The remaining node (neither src nor dst) must not hold the primary of any of + // the groups this eviction covers. Restricted to groupIDs rather than every + // group the node serves: it may independently be the primary of the default + // keyspace group (untouched by this eviction, since src was never its + // primary), and that must not fail this assertion. + var other bs.Server + for _, node := range nodeList { + if node.GetAddr() != src.GetAddr() && node.GetAddr() != dst.GetAddr() { + other = node + break + } + } + re.NotNil(other) + serving := mustGetKeyspaceGroupMembers(re, other.(*tso.Server)) + for _, id := range groupIDs { + m := serving[id] + re.False(m != nil && m.IsPrimary, "node %s should not hold primary of group %d", other.GetAddr(), id) + } } // TestEvictPrimaryRejectedWhileSplitting verifies that /primary/evict refuses to @@ -538,6 +660,186 @@ func (suite *memberTestSuite) TestEvictPrimaryRejectedWhileSplitting() { re.True(g.IsPrimary) } +// TestEvictPrimaryRejectedForInvalidCandidate verifies that /primary/evict +// rejects the whole request, before transferring anything, when new_primary is +// not a member of every candidate group. The candidate groups are iterated in +// indeterminate map order in production, so a broken implementation that folds +// the membership check into the transfer loop would still pass a single trial +// whenever the restricted group happens to be visited first: with 3 normal +// groups alongside the 1 restricted group, that is a 1-in-4 chance per trial. +// Repeat the scenario with fresh group IDs so an undetected regression would +// have to miss on every trial: trials independently reshuffle the map, so the +// miss probability compounds to 0.25^trials. +func (suite *memberTestSuite) TestEvictPrimaryRejectedForInvalidCandidate() { + re := suite.Require() + re.Len(suite.tsoNodes, 3) + + // target must not be a default keyspace group member: that group's primary + // election is not controlled by this test, and if target ended up serving it, + // evictPrimary's pre-check would run against it too, on top of the groups + // created below. suite.tsoAvailMembers is captured once in SetupTest right + // after the nodes start, before default group allocation is guaranteed to + // have converged to its full replica count, so poll the current membership + // here instead of trusting that snapshot: check every node and use whichever + // one first reports the default group at its full replica count. Querying a + // single arbitrarily chosen node would not work here, since a node reports + // group 0 only if it is itself a member of it. + defaultGroupMemberAddrs := make(map[string]bool, mcs.DefaultKeyspaceGroupReplicaCount) + testutil.Eventually(re, func() bool { + for _, node := range suite.tsoNodes { + g, ok := mustGetKeyspaceGroupMembers(re, node.(*tso.Server))[constant.DefaultKeyspaceGroupID] + if !ok || len(g.Group.Members) != mcs.DefaultKeyspaceGroupReplicaCount { + continue + } + for _, m := range g.Group.Members { + defaultGroupMemberAddrs[m.Address] = true + } + return true + } + return false + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(100*time.Millisecond)) + + var target bs.Server + defaultGroupNodes := make([]bs.Server, 0, mcs.DefaultKeyspaceGroupReplicaCount) + for _, node := range suite.tsoNodes { + if defaultGroupMemberAddrs[node.GetAddr()] { + defaultGroupNodes = append(defaultGroupNodes, node) + } else { + target = node + } + } + re.NotNil(target) + re.Len(defaultGroupNodes, mcs.DefaultKeyspaceGroupReplicaCount) + other := defaultGroupNodes[0] + outsider := defaultGroupNodes[1] + + nodeList := []bs.Server{target, other, outsider} + allMembers := make([]endpoint.KeyspaceGroupMember, 0, len(nodeList)) + for _, node := range nodeList { + allMembers = append(allMembers, endpoint.KeyspaceGroupMember{ + Address: node.GetAddr(), + Priority: mcs.DefaultKeyspaceGroupReplicaPriority, + }) + } + restrictedMembers := []endpoint.KeyspaceGroupMember{ + {Address: target.GetAddr(), Priority: mcs.DefaultKeyspaceGroupReplicaPriority}, + {Address: other.GetAddr(), Priority: mcs.DefaultKeyspaceGroupReplicaPriority}, + } + + const ( + trials = 5 + normalPerTrial = 3 + ) + for trial := range trials { + base := uint32(trial*10) + 1 + normalIDs := make([]uint32, normalPerTrial) + for i := range normalIDs { + normalIDs[i] = base + uint32(i) + } + restrictedID := base + normalPerTrial + + // outsider is a member of each of these groups, so it passes their + // pre-check. + for i, id := range normalIDs { + handlersutil.MustCreateKeyspaceGroup(re, suite.server, &handlers.CreateKeyspaceGroupParams{ + KeyspaceGroups: []*endpoint.KeyspaceGroup{ + { + ID: id, + UserKind: endpoint.Standard.String(), + Members: allMembers, + Keyspaces: []uint32{3000 + base + uint32(i)}, + }, + }, + }) + } + // outsider is not a member of this group, so evicting target with + // new_primary=outsider must be rejected for it. + handlersutil.MustCreateKeyspaceGroup(re, suite.server, &handlers.CreateKeyspaceGroupParams{ + KeyspaceGroups: []*endpoint.KeyspaceGroup{ + { + ID: restrictedID, + UserKind: endpoint.Standard.String(), + Members: restrictedMembers, + Keyspaces: []uint32{3000 + base + normalPerTrial}, + }, + }, + }) + + // Concentrate every group's primary on target. + allIDs := append(append([]uint32{}, normalIDs...), restrictedID) + for _, id := range allIDs { + transferData, err := json.Marshal(map[string]any{ + "new_primary": target.Name(), + "keyspace_group_id": id, + }) + re.NoError(err) + testutil.Eventually(re, func() bool { + resp, err := tests.TestDialClient.Post(target.GetAddr()+"/tso/api/v1/primary/transfer", + "application/json", bytes.NewBuffer(transferData)) + if err != nil { + return false + } + ok := resp.StatusCode == http.StatusOK + return resp.Body.Close() == nil && ok + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + } + testutil.Eventually(re, func() bool { + serving := mustGetKeyspaceGroupMembers(re, target.(*tso.Server)) + for _, id := range allIDs { + if serving[id] == nil || !serving[id].IsPrimary { + return false + } + } + return true + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + + // outsider is a valid candidate for every normal group but not for + // restrictedID. + evictData, err := json.Marshal(map[string]any{"new_primary": outsider.Name()}) + re.NoError(err) + resp, err := tests.TestDialClient.Post(target.GetAddr()+"/tso/api/v1/primary/evict", + "application/json", bytes.NewBuffer(evictData)) + re.NoError(err) + body, err := io.ReadAll(resp.Body) + re.NoError(err) + re.NoError(resp.Body.Close()) + re.Equal(http.StatusBadRequest, resp.StatusCode, string(body)) + re.Contains(string(body), fmt.Sprintf("keyspace group %d", restrictedID)) + + // The rejection must happen before any transfer, so every normal + // group's primary is still on target even though outsider would have + // been a valid destination for all of them. + serving := mustGetKeyspaceGroupMembers(re, target.(*tso.Server)) + for _, id := range normalIDs { + g, ok := serving[id] + re.True(ok) + re.True(g.IsPrimary) + } + + // Move every group's primary off target so the next trial's groups are + // the only ones evictPrimary considers: since this trial's eviction was + // rejected, target is still the primary of all of them, and leaving them + // there would make the next trial's error message reference this + // trial's restrictedID instead of its own. + for _, id := range allIDs { + transferData, err := json.Marshal(map[string]any{ + "new_primary": other.Name(), + "keyspace_group_id": id, + }) + re.NoError(err) + testutil.Eventually(re, func() bool { + resp, err := tests.TestDialClient.Post(target.GetAddr()+"/tso/api/v1/primary/transfer", + "application/json", bytes.NewBuffer(transferData)) + if err != nil { + return false + } + ok := resp.StatusCode == http.StatusOK + return resp.Body.Close() == nil && ok + }, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond)) + } + } +} + func (suite *memberTestSuite) TestCampaignPrimaryAfterTransfer() { re := suite.Require() supportedServices := []string{"tso", "scheduling", "resource_manager"}