Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

bootstrap default keyspace group in the tso service #6306

Merged
merged 6 commits into from
Apr 14, 2023
Merged

bootstrap default keyspace group in the tso service #6306

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
18fae33
Init default keyspace group in the tso service
binshi-bing Apr 12, 2023
0624827
renaming
binshi-bing Apr 12, 2023
71ccdda
Improve keyspace group membership restriction related logic and fix t…
binshi-bing Apr 12, 2023
c785844
Add logic "If the default keyspace group isn't assigned to any tso no…
binshi-bing Apr 12, 2023
e88e809
Refine keyspace.Bootstrap
binshi-bing Apr 12, 2023
a6fc397
Support upgrade (to multi-keyspace-groups enabled cluster) case.
binshi-bing Apr 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions pkg/keyspace/tso_keyspace_group.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,14 @@ func NewKeyspaceGroupManager(ctx context.Context, store endpoint.KeyspaceGroupSt

// Bootstrap saves default keyspace group info and init group mapping in the memory.
func (m *GroupManager) Bootstrap() error {
// Force the membership restriction that the default keyspace must belong to default keyspace group.
// Have no information to specify the distribution of the default keyspace group replicas, so just
// leave the replica/member list empty. The TSO service will assign the default keyspace group replica
// to every tso node/pod by default.
defaultKeyspaceGroup := &endpoint.KeyspaceGroup{
ID: utils.DefaultKeySpaceGroupID,
UserKind: endpoint.Basic.String(),
ID: utils.DefaultKeyspaceGroupID,
UserKind: endpoint.Basic.String(),
Keyspaces: []uint32{utils.DefaultKeyspaceID},
}

m.Lock()
Expand All @@ -71,11 +76,8 @@ func (m *GroupManager) Bootstrap() error {
return err
}

userKind := endpoint.StringUserKind(defaultKeyspaceGroup.UserKind)
Copy link
Contributor Author

@binshi-bing binshi-bing Apr 12, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

delete line 74 ~ 79 for two reasons:

  1. This is covered by the for loop below in which it loads all keyspace groups, including the default keyspace group, and initialize m.groups.
  2. Theoretically, at line 79, we put defaultKeyspaceGroup into m.groups, but defaultKeyspaceGroup might be different from the latest version persisted in the store, though the current logic of indexedHeap is to overwrite the existing one if put a new one.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rleungx ^

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me know if I missed anything.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right.

m.groups[userKind].Put(defaultKeyspaceGroup)

// Load all the keyspace groups from the storage and add to the respective userKind groups.
groups, err := m.store.LoadKeyspaceGroups(utils.DefaultKeySpaceGroupID, 0)
groups, err := m.store.LoadKeyspaceGroups(utils.DefaultKeyspaceGroupID, 0)
if err != nil {
return err
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/mcs/tso/server/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ func (h *Handler) ResetTS(ts uint64, ignoreSmaller, skipUpperBoundCheck bool) er
zap.Uint64("new-ts", ts),
zap.Bool("ignore-smaller", ignoreSmaller),
zap.Bool("skip-upper-bound-check", skipUpperBoundCheck))
tsoAllocatorManager, err := h.s.GetTSOAllocatorManager(mcsutils.DefaultKeySpaceGroupID)
tsoAllocatorManager, err := h.s.GetTSOAllocatorManager(mcsutils.DefaultKeyspaceGroupID)
if err != nil {
log.Error("failed to get allocator manager", errs.ZapError(err))
return err
Expand Down
9 changes: 4 additions & 5 deletions pkg/mcs/tso/server/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ func (s *Server) IsServing() bool {
}

member, err := s.keyspaceGroupManager.GetElectionMember(
mcsutils.DefaultKeyspaceID, mcsutils.DefaultKeySpaceGroupID)
mcsutils.DefaultKeyspaceID, mcsutils.DefaultKeyspaceGroupID)
if err != nil {
log.Error("failed to get election member", errs.ZapError(err))
return false
Expand All @@ -221,7 +221,7 @@ func (s *Server) IsServing() bool {
// The entry at the index 0 is the primary's service endpoint.
func (s *Server) GetLeaderListenUrls() []string {
member, err := s.keyspaceGroupManager.GetElectionMember(
mcsutils.DefaultKeyspaceID, mcsutils.DefaultKeySpaceGroupID)
mcsutils.DefaultKeyspaceID, mcsutils.DefaultKeyspaceGroupID)
if err != nil {
log.Error("failed to get election member", errs.ZapError(err))
return nil
Expand Down Expand Up @@ -436,15 +436,14 @@ func (s *Server) startServer() (err error) {
return err
}

// Initialize the TSO service.
s.serverLoopCtx, s.serverLoopCancel = context.WithCancel(s.ctx)
legacySvcRootPath := path.Join(pdRootPath, strconv.FormatUint(s.clusterID, 10))
tsoSvcRootPath := fmt.Sprintf(tsoSvcRootPathFormat, s.clusterID)
s.serviceID = &discovery.ServiceRegistryEntry{ServiceAddr: s.cfg.AdvertiseListenAddr}
s.keyspaceGroupManager = tso.NewKeyspaceGroupManager(
s.serverLoopCtx, s.serviceID, s.etcdClient, s.listenURL.Host, legacySvcRootPath, tsoSvcRootPath, s.cfg)
// The param `false` means that we don't initialize the keyspace group manager
// by loading the keyspace group meta from etcd.
if err := s.keyspaceGroupManager.Initialize(false); err != nil {
if err := s.keyspaceGroupManager.Initialize(); err != nil {
return err
}

Expand Down
7 changes: 4 additions & 3 deletions pkg/mcs/utils/constant.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,12 +37,13 @@ const (

// DefaultKeyspaceID is the default key space id.
// Valid keyspace id range is [0, 0xFFFFFF](uint24max, or 16777215)
// ​0 is reserved for default keyspace with the name "DEFAULT", It's initialized when PD bootstrap and reserved for users who haven't been assigned keyspace.
// ​0 is reserved for default keyspace with the name "DEFAULT", It's initialized when PD bootstrap
// and reserved for users who haven't been assigned keyspace.
DefaultKeyspaceID = uint32(0)

// DefaultKeySpaceGroupID is the default key space group id.
// DefaultKeyspaceGroupID is the default key space group id.
// We also reserved 0 for the keyspace group for the same purpose.
DefaultKeySpaceGroupID = uint32(0)
DefaultKeyspaceGroupID = uint32(0)

// APIServiceName is the name of api server.
APIServiceName = "api"
Expand Down
114 changes: 84 additions & 30 deletions pkg/tso/keyspace_group_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,15 @@ func (s *state) getAMWithMembershipCheck(
if kgid, ok := s.keyspaceLookupTable[keyspaceID]; ok {
return nil, kgid, genNotServedErr(errs.ErrGetAllocatorManager, keyspaceGroupID)
}
return nil, keyspaceGroupID, errs.ErrKeyspaceNotAssigned.FastGenByArgs(keyspaceID)

if keyspaceGroupID != mcsutils.DefaultKeyspaceGroupID {
return nil, keyspaceGroupID, errs.ErrKeyspaceNotAssigned.FastGenByArgs(keyspaceID)
}

// The keyspace doesn't belong to any keyspace group, so return the default keyspace group.
// It's for migrating the existing keyspaces which have no keyspace group assigned, so the
// the default keyspace group is used to serve the keyspaces.
return s.ams[mcsutils.DefaultKeyspaceGroupID], mcsutils.DefaultKeyspaceGroupID, nil
}

// KeyspaceGroupManager manages the members of the keyspace groups assigned to this host.
Expand Down Expand Up @@ -222,23 +230,12 @@ func NewKeyspaceGroupManager(
}

// Initialize this KeyspaceGroupManager
func (kgm *KeyspaceGroupManager) Initialize(loadFromStorage bool) error {
// Initialize the default keyspace group if not loading from storage
if !loadFromStorage {
group := &endpoint.KeyspaceGroup{
ID: mcsutils.DefaultKeySpaceGroupID,
Members: []endpoint.KeyspaceGroupMember{{Address: kgm.tsoServiceID.ServiceAddr}},
Keyspaces: []uint32{mcsutils.DefaultKeyspaceID},
}
kgm.updateKeyspaceGroup(group)
return nil
}

func (kgm *KeyspaceGroupManager) Initialize() error {
// Load the initial keyspace group assignment from storage with time limit
done := make(chan struct{}, 1)
ctx, cancel := context.WithCancel(kgm.ctx)
go kgm.checkInitProgress(ctx, cancel, done)
watchStartRevision, err := kgm.initAssignment(ctx)
watchStartRevision, defaultKGConfigured, err := kgm.initAssignment(ctx)
done <- struct{}{}
if err != nil {
log.Error("failed to initialize keyspace group manager", errs.ZapError(err))
Expand All @@ -247,6 +244,12 @@ func (kgm *KeyspaceGroupManager) Initialize(loadFromStorage bool) error {
return err
}

// Initialize the default keyspace group if it isn't configured in the storage.
if !defaultKGConfigured {
keyspaces := []uint32{mcsutils.DefaultKeyspaceID}
kgm.initDefaultKeysapceGroup(keyspaces)
}

// Watch/apply keyspace group membership/distribution meta changes dynamically.
kgm.wg.Add(1)
go kgm.startKeyspaceGroupsMetaWatchLoop(watchStartRevision)
Expand Down Expand Up @@ -284,14 +287,26 @@ func (kgm *KeyspaceGroupManager) checkInitProgress(ctx context.Context, cancel c
<-done
}

func (kgm *KeyspaceGroupManager) initDefaultKeysapceGroup(keyspaces []uint32) {
log.Info("initializing default keyspace group",
zap.Int("keyspaces-length", len(keyspaces)))

group := &endpoint.KeyspaceGroup{
ID: mcsutils.DefaultKeyspaceGroupID,
Members: []endpoint.KeyspaceGroupMember{{Address: kgm.tsoServiceID.ServiceAddr}},
Keyspaces: keyspaces,
}
kgm.updateKeyspaceGroup(group)
}

// initAssignment loads initial keyspace group assignment from storage and initialize the group manager.
func (kgm *KeyspaceGroupManager) initAssignment(ctx context.Context) (int64, error) {
// Return watchStartRevision, the start revision for watching keyspace group membership/distribution change.
func (kgm *KeyspaceGroupManager) initAssignment(
ctx context.Context,
) (watchStartRevision int64, defaultKGConfigured bool, err error) {
var (
// The start revision for watching keyspace group membership/distribution change
watchStartRevision int64
groups []*endpoint.KeyspaceGroup
more bool
err error
keyspaceGroupsLoaded uint32
revision int64
)
Expand All @@ -300,7 +315,7 @@ func (kgm *KeyspaceGroupManager) initAssignment(ctx context.Context) (int64, err
for {
revision, groups, more, err = kgm.loadKeyspaceGroups(ctx, keyspaceGroupsLoaded, kgm.loadKeyspaceGroupsBatchSize)
if err != nil {
return 0, err
return
}

keyspaceGroupsLoaded += uint32(len(groups))
Expand All @@ -313,10 +328,15 @@ func (kgm *KeyspaceGroupManager) initAssignment(ctx context.Context) (int64, err
for _, group := range groups {
select {
case <-ctx.Done():
return watchStartRevision, errs.ErrLoadKeyspaceGroupsTerminated
err = errs.ErrLoadKeyspaceGroupsTerminated
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can return nil here instead of an error.

Copy link
Contributor Author

@binshi-bing binshi-bing Apr 13, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can't return nil. Returning an error here indicates the caller KeyspaceGroupManager.Initialize(), the entire initialization process, should return immediately instead of continuing the other initialization work. Please check the logic after kgm.initAssignment(ctx) returns to see if there is a better way to do it.

func (kgm *KeyspaceGroupManager) Initialize() error {
        ...
	watchStartRevision, defaultKGConfigured, err := kgm.initAssignment(ctx)
	done <- struct{}{}
	if err != nil {
		log.Error("failed to initialize keyspace group manager", errs.ZapError(err))
		return err
	}

	// Initialize the default keyspace group if it isn't configured in the storage.
	if !defaultKGConfigured {
		keyspaces := []uint32{mcsutils.DefaultKeyspaceID}
		kgm.initDefaultKeysapceGroup(keyspaces)
	}

	// Watch/apply keyspace group membership/distribution meta changes dynamically.
	kgm.wg.Add(1)
	go kgm.startKeyspaceGroupsMetaWatchLoop(watchStartRevision)
	@@ -284,14 +279,26 @@
	<-done

return
default:
}

if group.ID == mcsutils.DefaultKeyspaceGroupID {
defaultKGConfigured = true
}

kgm.updateKeyspaceGroup(group)
}

Expand All @@ -326,7 +346,7 @@ func (kgm *KeyspaceGroupManager) initAssignment(ctx context.Context) (int64, err
}

log.Info("loaded keyspace groups", zap.Uint32("keyspace-groups-loaded", keyspaceGroupsLoaded))
return watchStartRevision, nil
return
}

// loadKeyspaceGroups loads keyspace groups from the start ID with limit.
Expand Down Expand Up @@ -441,7 +461,7 @@ func (kgm *KeyspaceGroupManager) watchKeyspaceGroupsMetaChange(revision int64) (
return revision, wresp.Err()
}
for _, event := range wresp.Events {
id, err := endpoint.ExtractKeyspaceGroupIDFromPath(string(event.Kv.Key))
groupID, err := endpoint.ExtractKeyspaceGroupIDFromPath(string(event.Kv.Key))
if err != nil {
log.Warn("failed to extract keyspace group ID from the key path",
zap.String("key-path", string(event.Kv.Key)), zap.Error(err))
Expand All @@ -453,12 +473,20 @@ func (kgm *KeyspaceGroupManager) watchKeyspaceGroupsMetaChange(revision int64) (
group := &endpoint.KeyspaceGroup{}
if err := json.Unmarshal(event.Kv.Value, group); err != nil {
log.Warn("failed to unmarshal keyspace group",
zap.Uint32("keysapce-group-id", id),
zap.Uint32("keysapce-group-id", groupID),
zap.Error(errs.ErrJSONUnmarshal.Wrap(err).FastGenWithCause()))
}
kgm.updateKeyspaceGroup(group)
case clientv3.EventTypeDelete:
kgm.deleteKeyspaceGroup(id)
if groupID == mcsutils.DefaultKeyspaceGroupID {
keyspaces := kgm.kgs[groupID].Keyspaces
kgm.deleteKeyspaceGroup(groupID)
log.Warn("removed default keyspace group meta config from the storage. " +
"now every tso node/pod will initialize it")
kgm.initDefaultKeysapceGroup(keyspaces)
} else {
kgm.deleteKeyspaceGroup(groupID)
}
}
}
revision = wresp.Header.Revision
Expand All @@ -473,6 +501,11 @@ func (kgm *KeyspaceGroupManager) watchKeyspaceGroupsMetaChange(revision int64) (
}

func (kgm *KeyspaceGroupManager) isAssignedToMe(group *endpoint.KeyspaceGroup) bool {
// If the default keyspace group isn't assigned to any tso node/pod, assign it to everyone.
if group.ID == mcsutils.DefaultKeyspaceGroupID && len(group.Members) == 0 {
return true
}

for _, member := range group.Members {
if member.Address == kgm.tsoServiceID.ServiceAddr {
return true
Expand Down Expand Up @@ -516,7 +549,7 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
tsRootPath string
storage *endpoint.StorageEndpoint
)
if group.ID == mcsutils.DefaultKeySpaceGroupID {
if group.ID == mcsutils.DefaultKeyspaceGroupID {
tsRootPath = kgm.legacySvcRootPath
storage = kgm.legacySvcStorage
} else {
Expand All @@ -536,6 +569,10 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro
kgm.ams[group.ID] = am
kgm.Unlock()
} else {
if group.ID == mcsutils.DefaultKeyspaceGroupID {
log.Info("resign default keyspace group membership",
zap.Any("default-keyspace-group", group))
}
// Not assigned to me. If this host/pod owns this keyspace group, it should resign.
kgm.deleteKeyspaceGroup(group.ID)
}
Expand All @@ -554,7 +591,7 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroupMembership(
return newKeyspaces[i] < newKeyspaces[j]
})

// Mostly, the membership has no change, so we optimize for this case.
// Mostly, the membership has no change, so optimize for this case.
sameMembership := true
if oldLen != newLen {
sameMembership = false
Expand All @@ -571,10 +608,10 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroupMembership(
defer kgm.Unlock()

if sameMembership {
// The keyspace group membership is not changed, so we reuse the old one.
// The keyspace group membership is not changed. Reuse the old one.
newGroup.KeyspaceLookupTable = oldGroup.KeyspaceLookupTable
} else {
// The keyspace group membership is changed, so we update the keyspace lookup table.
// The keyspace group membership is changed. Update the keyspace lookup table.
newGroup.KeyspaceLookupTable = make(map[uint32]struct{})
for i, j := 0, 0; i < oldLen || j < newLen; {
if i < oldLen && j < newLen && oldKeyspaces[i] == newKeyspaces[j] {
Expand All @@ -590,12 +627,31 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroupMembership(
j++
}
}
if groupID == mcsutils.DefaultKeyspaceGroupID {
if _, ok := newGroup.KeyspaceLookupTable[mcsutils.DefaultKeyspaceID]; !ok {
log.Warn("default keyspace is not in default keyspace group. add it back")
kgm.keyspaceLookupTable[mcsutils.DefaultKeyspaceID] = groupID
newGroup.KeyspaceLookupTable[mcsutils.DefaultKeyspaceID] = struct{}{}
newGroup.Keyspaces = make([]uint32, 1+len(newKeyspaces))
newGroup.Keyspaces[0] = mcsutils.DefaultKeyspaceID
copy(newGroup.Keyspaces[1:], newKeyspaces)
}
} else {
if _, ok := newGroup.KeyspaceLookupTable[mcsutils.DefaultKeyspaceID]; ok {
log.Warn("default keyspace is in non-default keyspace group. remove it")
kgm.keyspaceLookupTable[mcsutils.DefaultKeyspaceID] = mcsutils.DefaultKeyspaceGroupID
delete(newGroup.KeyspaceLookupTable, mcsutils.DefaultKeyspaceID)
newGroup.Keyspaces = newKeyspaces[1:]
}
}
}
kgm.kgs[groupID] = newGroup
}

// deleteKeyspaceGroup deletes the given keyspace group.
func (kgm *KeyspaceGroupManager) deleteKeyspaceGroup(groupID uint32) {
log.Info("delete keyspace group", zap.Uint32("keyspace-group-id", groupID))

kgm.Lock()
defer kgm.Unlock()

Expand All @@ -618,8 +674,6 @@ func (kgm *KeyspaceGroupManager) deleteKeyspaceGroup(groupID uint32) {
am.close()
kgm.ams[groupID] = nil
}

log.Info("deleted keyspace group", zap.Uint32("keyspace-group-id", groupID))
}

// GetAllocatorManager returns the AllocatorManager of the given keyspace group
Expand Down
Loading