mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
sever: v2store deprecation: Fix etcdctl snapshot restore to restore
correct 'backend' (bbolt) context in aspect of membership.
Prior to this change the 'restored' backend used to still contain:
- old memberid (mvcc deletion used, why the membership is in bolt
bucket, but not mvcc part):
```
mvs := mvcc.NewStore(s.lg, be, lessor, ci, mvcc.StoreConfig{CompactionBatchLimit: math.MaxInt32})
defer mvs.Close()
txn := mvs.Write(traceutil.TODO())
btx := be.BatchTx()
del := func(k, v []byte) error {
txn.DeleteRange(k, nil)
return nil
}
// delete stored members from old cluster since using new members
btx.UnsafeForEach([]byte("members"), del)
```
- didn't get new members added.
This commit is contained in:
@@ -676,6 +676,10 @@ func membersFromStore(lg *zap.Logger, st v2store.Store) (map[types.ID]*Member, m
|
||||
return members, removed
|
||||
}
|
||||
|
||||
func membersFromBackend(lg *zap.Logger, be backend.Backend) (map[types.ID]*Member, map[types.ID]bool) {
|
||||
return mustReadMembersFromBackend(lg, be)
|
||||
}
|
||||
|
||||
func clusterVersionFromStore(lg *zap.Logger, st v2store.Store) *semver.Version {
|
||||
e, err := st.Get(path.Join(storePrefix, "version"), false, false)
|
||||
if err != nil {
|
||||
|
||||
@@ -67,6 +67,66 @@ func mustDeleteMemberFromBackend(be backend.Backend, id types.ID) {
|
||||
tx.UnsafePut(membersRemovedBucketName, mkey, []byte("removed"))
|
||||
}
|
||||
|
||||
func readMembersFromBackend(lg *zap.Logger, be backend.Backend) (map[types.ID]*Member, map[types.ID]bool, error) {
|
||||
members := make(map[types.ID]*Member)
|
||||
removed := make(map[types.ID]bool)
|
||||
|
||||
tx := be.ReadTx()
|
||||
tx.RLock()
|
||||
defer tx.RUnlock()
|
||||
err := tx.UnsafeForEach(membersBucketName, func(k, v []byte) error {
|
||||
memberId := MustParseMemberIDFromBytes(lg, k)
|
||||
m := &Member{ID: memberId}
|
||||
if err := json.Unmarshal(v, &m); err != nil {
|
||||
return err
|
||||
}
|
||||
members[memberId] = m
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("couldn't read members from backend: %w", err)
|
||||
}
|
||||
|
||||
err = tx.UnsafeForEach(membersRemovedBucketName, func(k, v []byte) error {
|
||||
memberId := MustParseMemberIDFromBytes(lg, k)
|
||||
removed[memberId] = true
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("couldn't read members_removed from backend: %w", err)
|
||||
}
|
||||
return members, removed, nil
|
||||
}
|
||||
|
||||
func mustReadMembersFromBackend(lg *zap.Logger, be backend.Backend) (map[types.ID]*Member, map[types.ID]bool) {
|
||||
members, removed, err := readMembersFromBackend(lg, be)
|
||||
if err != nil {
|
||||
lg.Panic("couldn't read members from backend", zap.Error(err))
|
||||
}
|
||||
return members, removed
|
||||
}
|
||||
|
||||
func TrimMembershipFromBackend(lg *zap.Logger, be backend.Backend) error {
|
||||
tx := be.BatchTx()
|
||||
tx.Lock()
|
||||
defer tx.Unlock()
|
||||
err := tx.UnsafeForEach(membersBucketName, func(k, v []byte) error {
|
||||
tx.UnsafeDelete(membersBucketName, k)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = tx.UnsafeForEach(membersRemovedBucketName, func(k, v []byte) error {
|
||||
tx.UnsafeDelete(membersRemovedBucketName, k)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func mustSaveClusterVersionToBackend(be backend.Backend, ver *semver.Version) {
|
||||
ckey := backendClusterVersionKey()
|
||||
|
||||
@@ -221,10 +281,18 @@ func MemberAttributesStorePath(id types.ID) string {
|
||||
return path.Join(MemberStoreKey(id), attributesSuffix)
|
||||
}
|
||||
|
||||
func MustParseMemberIDFromBytes(lg *zap.Logger, key []byte) types.ID {
|
||||
id, err := types.IDFromString(string(key))
|
||||
if err != nil {
|
||||
lg.Panic("failed to parse member id from key", zap.Error(err))
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func MustParseMemberIDFromKey(lg *zap.Logger, key string) types.ID {
|
||||
id, err := types.IDFromString(path.Base(key))
|
||||
if err != nil {
|
||||
lg.Panic("failed to parse memver id from key", zap.Error(err))
|
||||
lg.Panic("failed to parse member id from key", zap.Error(err))
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
43
server/etcdserver/api/membership/store_test.go
Normal file
43
server/etcdserver/api/membership/store_test.go
Normal file
@@ -0,0 +1,43 @@
|
||||
package membership
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/coreos/go-semver/semver"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"go.etcd.io/etcd/client/pkg/v3/types"
|
||||
betesting "go.etcd.io/etcd/server/v3/mvcc/backend/testing"
|
||||
|
||||
"go.etcd.io/etcd/server/v3/mvcc/backend"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
func TestAddRemoveMember(t *testing.T) {
|
||||
c := newTestCluster(t, nil)
|
||||
be, bepath := betesting.NewDefaultTmpBackend(t)
|
||||
c.SetBackend(be)
|
||||
c.AddMember(newTestMember(17, nil, "node17", nil), true)
|
||||
c.RemoveMember(17, true)
|
||||
c.AddMember(newTestMember(18, nil, "node18", nil), true)
|
||||
|
||||
// Skipping removal of already removed member
|
||||
c.RemoveMember(17, true)
|
||||
err := be.Close()
|
||||
assert.NoError(t, err)
|
||||
|
||||
be2 := backend.NewDefaultBackend(bepath)
|
||||
defer func() {
|
||||
assert.NoError(t, be2.Close())
|
||||
}()
|
||||
|
||||
if false {
|
||||
// TODO: Enable this code when Recover is reading membership from the backend.
|
||||
c2 := newTestCluster(t, nil)
|
||||
c2.SetBackend(be2)
|
||||
c2.Recover(func(*zap.Logger, *semver.Version) {})
|
||||
assert.Equal(t, []*Member{{ID: types.ID(18),
|
||||
Attributes: Attributes{Name: "node18"}}}, c2.Members())
|
||||
assert.Equal(t, true, c2.IsIDRemoved(17))
|
||||
assert.Equal(t, false, c2.IsIDRemoved(18))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user