Merge pull request #19563 from ahrtr/3.5_learner_20250310

[release-3.5] Fix the learner promotion changes not being persisted into v3store (bbolt)

Author: ahrtr

server/etcdserver/api/membership/cluster.go

@@ -696,6 +696,10 @@ func (c *RaftCluster) IsReadyToPromoteMember(id uint64) bool {
 	return true
 }
 
+func (c *RaftCluster) MembersFromStore() (map[types.ID]*Member, map[types.ID]bool) {
+	return membersFromStore(c.lg, c.v2store)
+}
+
 func membersFromStore(lg *zap.Logger, st v2store.Store) (map[types.ID]*Member, map[types.ID]bool) {
 	members := make(map[types.ID]*Member)
 	removed := make(map[types.ID]bool)
@@ -732,6 +736,10 @@ func membersFromStore(lg *zap.Logger, st v2store.Store) (map[types.ID]*Member, m
 	return members, removed
 }
 
+func (c *RaftCluster) MembersFromBackend() (map[types.ID]*Member, map[types.ID]bool) {
+	return membersFromBackend(c.lg, c.be)
+}
+
 func membersFromBackend(lg *zap.Logger, be backend.Backend) (map[types.ID]*Member, map[types.ID]bool) {
 	return mustReadMembersFromBackend(lg, be)
 }

server/etcdserver/api/membership/cluster_test.go

@@ -1049,10 +1049,9 @@ func TestAddMemberSyncsBackendAndStoreV2(t *testing.T) {
 			backendMembers: []*Member{alice},
 		},
 		{
-			name:           "Adding member should fail if it exists in both",
+			name:           "Adding member should success if it exists in both",
 			storeV2Members: []*Member{alice},
 			backendMembers: []*Member{alice},
-			expectPanics:   true,
 		},
 		{
 			name:           "Adding member should fail if it exists in storeV2 and backend is nil",

server/etcdserver/api/membership/store.go

@@ -54,9 +54,6 @@ func unsafeSaveMemberToBackend(lg *zap.Logger, be backend.Backend, m *Member) er
 	tx := be.BatchTx()
 	tx.LockInsideApply()
 	defer tx.Unlock()
-	if unsafeMemberExists(tx, mkey) {
-		return errMemberAlreadyExist
-	}
 	tx.UnsafePut(buckets.Members, mkey, mvalue)
 	return nil
 }

server/etcdserver/server.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"os"
 	"path"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -67,6 +68,7 @@ import (
 	"go.etcd.io/etcd/server/v3/lease/leasehttp"
 	"go.etcd.io/etcd/server/v3/mvcc"
 	"go.etcd.io/etcd/server/v3/mvcc/backend"
+	"go.etcd.io/etcd/server/v3/verify"
 	"go.etcd.io/etcd/server/v3/wal"
 )
 
@@ -2449,9 +2451,51 @@ func (s *EtcdServer) applyConfChange(cc raftpb.ConfChange, confState *raftpb.Con
 			s.r.transport.UpdatePeer(m.ID, m.PeerURLs)
 		}
 	}
+
+	s.verifyV3StoreInSyncWithV2Store(shouldApplyV3)
+
 	return false, nil
 }
 
+func (s *EtcdServer) verifyV3StoreInSyncWithV2Store(shouldApplyV3 membership.ShouldApplyV3) {
+	if !verify.VerifyEnabled() {
+		return
+	}
+
+	// If shouldApplyV3 == false, then it means v2store hasn't caught up with v3store.
+	if !shouldApplyV3 {
+		return
+	}
+
+	// clean up the Attributes, and we only care about the RaftAttributes
+	cleanAttributesFunc := func(members map[types.ID]*membership.Member) map[types.ID]*membership.Member {
+		processedMembers := make(map[types.ID]*membership.Member)
+		for id, m := range members {
+			clonedMember := m.Clone()
+			clonedMember.Attributes = membership.Attributes{}
+			processedMembers[id] = clonedMember
+		}
+
+		return processedMembers
+	}
+
+	v2Members, _ := s.cluster.MembersFromStore()
+	v3Members, _ := s.cluster.MembersFromBackend()
+
+	processedV2Members := cleanAttributesFunc(v2Members)
+	processedV3Members := cleanAttributesFunc(v3Members)
+
+	if match := reflect.DeepEqual(processedV2Members, processedV3Members); !match {
+		v2Data, v2Err := json.Marshal(processedV2Members)
+		v3Data, v3Err := json.Marshal(processedV3Members)
+
+		if v2Err != nil || v3Err != nil {
+			panic("members in v2store doesn't match v3store")
+		}
+		panic(fmt.Sprintf("members in v2store doesn't match v3store, v2store: %s, v3store: %s", string(v2Data), string(v3Data)))
+	}
+}
+
 // TODO: non-blocking snapshot
 func (s *EtcdServer) snapshot(snapi uint64, confState raftpb.ConfState) {
 	clone := s.v2store.Clone()

server/etcdserver/server_test.go

@@ -655,6 +655,7 @@ func TestApplyConfigChangeUpdatesConsistIndex(t *testing.T) {
 
 	be, _ := betesting.NewDefaultTmpBackend(t)
 	defer betesting.Close(t, be)
+	cl.SetBackend(be)
 	cindex.CreateMetaBucket(be.BatchTx())
 
 	cl.AddMember(&membership.Member{ID: types.ID(1)}, true)

server/verify/verify.go

@@ -90,12 +90,37 @@ func Verify(cfg Config) error {
 // VerifyIfEnabled performs verification according to ETCD_VERIFY env settings.
 // See Verify for more information.
 func VerifyIfEnabled(cfg Config) error {
-	if os.Getenv(ENV_VERIFY) == ENV_VERIFY_ALL_VALUE {
+	if VerifyEnabled() {
 		return Verify(cfg)
 	}
 	return nil
 }
 
+// VerifyEnabled returns `true` if verification is enabled.
+func VerifyEnabled() bool {
+	return os.Getenv(ENV_VERIFY) == ENV_VERIFY_ALL_VALUE
+}
+
+// EnableVerification enables the verification and returns a function that
+// can be used to bring the original settings.
+func EnableVerification() func() {
+	previousEnv := os.Getenv(ENV_VERIFY)
+	os.Setenv(ENV_VERIFY, ENV_VERIFY_ALL_VALUE)
+	return func() {
+		os.Setenv(ENV_VERIFY, previousEnv)
+	}
+}
+
+// DisableVerification disables the verification and returns a function that
+// can be used to bring the original settings.
+func DisableVerification() func() {
+	previousEnv := os.Getenv(ENV_VERIFY)
+	os.Unsetenv(ENV_VERIFY)
+	return func() {
+		os.Setenv(ENV_VERIFY, previousEnv)
+	}
+}
+
 // MustVerifyIfEnabled performs verification according to ETCD_VERIFY env settings
 // and exits in case of found problems.
 // See Verify for more information.

tests/e2e/ctl_v2_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"go.etcd.io/etcd/server/v3/verify"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 )
 
@@ -229,6 +230,10 @@ func TestUtlCtlV2Backup(t *testing.T) {
 func testUtlCtlV2Backup(t *testing.T, snapCount int, v3 bool, utl bool) {
 	BeforeTestV2(t)
 
+	// disable the verification because this case updated the db offline
+	revertFunc := verify.DisableVerification()
+	defer revertFunc()
+
 	backupDir, err := ioutil.TempDir(t.TempDir(), "testbackup0.etcd")
 	if err != nil {
 		t.Fatal(err)