Skip to content

Commit

Permalink
changed by name
Browse files Browse the repository at this point in the history
Signed-off-by: husharp <[email protected]>
  • Loading branch information
HuSharp committed Jul 25, 2024
1 parent 19ce9d8 commit 4c7f8ac
Show file tree
Hide file tree
Showing 7 changed files with 9 additions and 14 deletions.
2 changes: 1 addition & 1 deletion pkg/election/leadership.go
Original file line number Diff line number Diff line change
Expand Up @@ -397,7 +397,7 @@ func (ls *Leadership) Watch(serverCtx context.Context, revision int64) {
// ONLY `/ms/primary/transfer` API update primary will meet this condition.
if ev.Type == mvccpb.PUT && ls.IsPrimary() {
log.Info("current leadership is updated", zap.Int64("revision", wresp.Header.Revision),
zap.String("leader-key", ls.leaderKey), zap.Any("value", ls.leaderValue),
zap.String("leader-key", ls.leaderKey), zap.ByteString("cur-value", ev.Kv.Value),
zap.String("purpose", ls.purpose))
return
}
Expand Down
7 changes: 1 addition & 6 deletions pkg/mcs/discovery/discover.go
Original file line number Diff line number Diff line change
Expand Up @@ -96,14 +96,9 @@ func TransferPrimary(client *clientv3.Client, serviceName, oldPrimary, newPrimar
}

var primaryIDs []string
var memberValues []string
for _, member := range entries {
if (newPrimary == "" && member.ServiceAddr != oldPrimary) || (newPrimary != "" && member.Name == newPrimary) {
primaryIDs = append(primaryIDs, member.ServiceAddr)
if string(member.MemberValue) == "" {
return errors.New(fmt.Sprintf("member %s value is empty", member.Name))
}
memberValues = append(memberValues, string(member.MemberValue))
}
}
if len(primaryIDs) == 0 {
Expand Down Expand Up @@ -137,7 +132,7 @@ func TransferPrimary(client *clientv3.Client, serviceName, oldPrimary, newPrimar
}
// update primary key to notify old primary server.
putResp, err := kv.NewSlowLogTxn(client).
Then(clientv3.OpPut(primaryKey, memberValues[nextPrimaryID], clientv3.WithLease(grantResp.ID))).
Then(clientv3.OpPut(primaryKey, primaryIDs[nextPrimaryID], clientv3.WithLease(grantResp.ID))).
Commit()
if err != nil || !putResp.Succeeded {
return errors.Errorf("failed to write primary flag for %s, err: %v", serviceName, err)
Expand Down
1 change: 0 additions & 1 deletion pkg/mcs/discovery/registry_entry.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ type ServiceRegistryEntry struct {
GitHash string `json:"git-hash"`
DeployPath string `json:"deploy-path"`
StartTimestamp int64 `json:"start-timestamp"`
MemberValue []byte `json:"member-value"`
}

// Serialize this service registry entry
Expand Down
6 changes: 3 additions & 3 deletions pkg/mcs/scheduling/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
Expand Down Expand Up @@ -252,7 +253,7 @@ func (s *Server) primaryElectionLoop() {
expectedPrimary := utils.AttachExpectedPrimaryFlag(s.GetClient(), s.participant.GetLeaderPath())
// skip campaign the primary if the expected primary is not empty and not equal to the current memberValue.
// expected primary ONLY SET BY `/ms/primary/transfer` API.
if expectedPrimary != "" && expectedPrimary != s.participant.MemberValue() {
if expectedPrimary != "" && !strings.Contains(s.participant.MemberValue(), expectedPrimary) {
log.Info("skip campaigning of scheduling primary and check later",
zap.String("server-name", s.Name()),
zap.String("expected-primary-id", expectedPrimary),
Expand Down Expand Up @@ -356,7 +357,7 @@ func (s *Server) primaryWatch(ctx context.Context, exitPrimary chan struct{}) {
log.Error("scheduling primary getting the leader meets error", errs.ZapError(err))
return
}
if curPrimary != nil && resp.Kvs[0].Value != nil && string(curPrimary) != string(resp.Kvs[0].Value) {
if curPrimary != nil && resp.Kvs[0].Value != nil && !strings.Contains(string(resp.Kvs[0].Value), string(curPrimary)) {
// 1. modify the expected primary flag to the new primary.
utils.MarkExpectedPrimaryFlag(s.participant.Client(), s.participant.GetLeaderPath())
// 2. modify memory status.
Expand Down Expand Up @@ -503,7 +504,6 @@ func (s *Server) startServer() (err error) {
ListenUrls: []string{s.cfg.GetAdvertiseListenAddr()},
}
s.participant.InitInfo(p, endpoint.SchedulingSvcRootPath(s.clusterID), utils.PrimaryKey, "primary election")
s.serviceID.MemberValue = []byte(s.participant.MemberValue())

s.service = &Service{Server: s}
s.AddServiceReadyCallback(s.startCluster)
Expand Down
1 change: 1 addition & 0 deletions pkg/mcs/utils/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ func ClearPrimaryExpectationFlag(client *clientv3.Client, leaderPath string) {
// MarkExpectedPrimaryFlag marks the expected primary flag when the current primary has exited.
func MarkExpectedPrimaryFlag(client *clientv3.Client, leaderPath string) {
log.Info("set expected primary flag", zap.String("leader-path", leaderPath))
// We have updated new primary(server's addr) in `leaderPath` by `/ms/primary/transfer` API.
leaderRaw, err := etcdutil.GetValue(client, leaderPath)
if err != nil {
log.Error("get primary key error", zap.Error(err))
Expand Down
5 changes: 3 additions & 2 deletions pkg/tso/global_allocator.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"errors"
"fmt"
"runtime/trace"
"strings"
"sync"
"sync/atomic"
"time"
Expand Down Expand Up @@ -565,7 +566,7 @@ func (gta *GlobalTSOAllocator) primaryElectionLoop() {
expectedPrimary := mcsutils.AttachExpectedPrimaryFlag(gta.member.Client(), gta.member.GetLeaderPath())
// skip campaign the primary if the expected primary is not empty and not equal to the current memberValue.
// expected primary ONLY SET BY `/ms/primary/transfer` API.
if expectedPrimary != "" && expectedPrimary != gta.member.MemberValue() {
if expectedPrimary != "" && !strings.Contains(gta.member.MemberValue(), expectedPrimary) {
log.Info("skip campaigning of tso primary and check later",
zap.String("server-name", gta.member.Name()),
zap.String("expected-primary-id", expectedPrimary),
Expand Down Expand Up @@ -701,7 +702,7 @@ func (gta *GlobalTSOAllocator) primaryWatch(ctx context.Context, exitPrimary cha
log.Error("tso primary getting the leader meets error", errs.ZapError(err))
return
}
if curPrimary != nil && resp.Kvs[0].Value != nil && string(curPrimary) != string(resp.Kvs[0].Value) {
if curPrimary != nil && resp.Kvs[0].Value != nil && !strings.Contains(string(resp.Kvs[0].Value), string(curPrimary)) {
// 1. modify the expected primary flag to the new primary.
mcsutils.MarkExpectedPrimaryFlag(gta.member.Client(), gta.member.GetLeaderPath())
// 2. modify memory status.
Expand Down
1 change: 0 additions & 1 deletion pkg/tso/keyspace_group_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -737,7 +737,6 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
ListenUrls: []string{kgm.cfg.GetAdvertiseListenAddr()},
}
participant.InitInfo(p, endpoint.KeyspaceGroupsElectionPath(kgm.tsoSvcRootPath, group.ID), mcsutils.PrimaryKey, "keyspace group primary election")
kgm.tsoServiceID.MemberValue = []byte(participant.MemberValue())
// If the keyspace group is in split, we should ensure that the primary elected by the new keyspace group
// is always on the same TSO Server node as the primary of the old keyspace group, and this constraint cannot
// be broken until the entire split process is completed.
Expand Down

0 comments on commit 4c7f8ac

Please sign in to comment.