Merge pull request #20071 from ahrtr/20250530_force_new_cluster_3.6

[release-3.6] Fix the issue that `--force-new-cluster` can't remove all other members in a corner case

Author: ahrtr

server/etcdserver/bootstrap.go

@@ -87,7 +87,8 @@ func bootstrap(cfg config.ServerConfig) (b *bootstrappedServer, err error) {
 		if err = fileutil.IsDirWriteable(cfg.WALDir()); err != nil {
 			return nil, fmt.Errorf("cannot write to WAL directory: %w", err)
 		}
-		bwal = bootstrapWALFromSnapshot(cfg, backend.snapshot)
+		cfg.Logger.Info("Bootstrapping WAL from snapshot")
+		bwal = bootstrapWALFromSnapshot(cfg, backend.snapshot, backend.ci)
 	}
 
 	cfg.Logger.Info("bootstrapping cluster")
@@ -556,7 +557,7 @@ func (b *bootstrappedRaft) newRaftNode(ss *snap.Snapshotter, wal *wal.WAL, cl *m
 	)
 }
 
-func bootstrapWALFromSnapshot(cfg config.ServerConfig, snapshot *raftpb.Snapshot) *bootstrappedWAL {
+func bootstrapWALFromSnapshot(cfg config.ServerConfig, snapshot *raftpb.Snapshot, ci cindex.ConsistentIndexer) *bootstrappedWAL {
 	wal, st, ents, snap, meta := openWALFromSnapshot(cfg, snapshot)
 	bwal := &bootstrappedWAL{
 		lg:       cfg.Logger,
@@ -569,6 +570,19 @@ func bootstrapWALFromSnapshot(cfg config.ServerConfig, snapshot *raftpb.Snapshot
 	}
 
 	if cfg.ForceNewCluster {
+		consistentIndex := ci.ConsistentIndex()
+		oldCommitIndex := bwal.st.Commit
+		// If only `HardState.Commit` increases, HardState won't be persisted
+		// to disk, even though the committed entries might have already been
+		// applied. This can result in consistent_index > CommitIndex.
+		//
+		// When restarting etcd with `--force-new-cluster`, all uncommitted
+		// entries are dropped. To avoid losing entries that were actually
+		// committed, we reset Commit to max(HardState.Commit, consistent_index).
+		//
+		// See: https://github.com/etcd-io/raft/pull/300 for more details.
+		bwal.st.Commit = max(oldCommitIndex, consistentIndex)
+
 		// discard the previously uncommitted entries
 		bwal.ents = bwal.CommitedEntries()
 		entries := bwal.NewConfigChangeEntries()
@@ -578,6 +592,7 @@ func bootstrapWALFromSnapshot(cfg config.ServerConfig, snapshot *raftpb.Snapshot
 			"forcing restart member",
 			zap.String("cluster-id", meta.clusterID.String()),
 			zap.String("local-member-id", meta.nodeID.String()),
+			zap.Uint64("wal-commit-index", oldCommitIndex),
 			zap.Uint64("commit-index", bwal.st.Commit),
 		)
 	} else {

tests/e2e/force_new_cluster_test.go

@@ -12,14 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//go:build !cluster_proxy
+
 package e2e
 
 import (
 	"context"
+	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
+	"go.etcd.io/bbolt"
+	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
+	"go.etcd.io/etcd/server/v3/storage/datadir"
+	"go.etcd.io/etcd/server/v3/storage/schema"
 	"go.etcd.io/etcd/tests/v3/framework/config"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 )
@@ -70,3 +78,63 @@ func TestForceNewCluster(t *testing.T) {
 		})
 	}
 }
+
+func TestForceNewCluster_MemberCount(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	ctx := context.Background()
+
+	epc, promotedMembers := mustCreateNewClusterByPromotingMembers(t, e2e.CurrentVersion, 3, e2e.WithKeepDataDir(true))
+	require.Len(t, promotedMembers, 2)
+
+	// Wait for the backend TXN to sync/commit the data to disk, to ensure
+	// the consistent-index is persisted. Another way is to issue a snapshot
+	// command to forcibly commit the backend TXN.
+	time.Sleep(time.Second)
+
+	t.Log("Killing all the members")
+	require.NoError(t, epc.Kill())
+	require.NoError(t, epc.Wait(ctx))
+
+	m := epc.Procs[0]
+	t.Logf("Forcibly create a one-member cluster with member: %s", m.Config().Name)
+	m.Config().Args = append(m.Config().Args, "--force-new-cluster")
+	require.NoError(t, m.Start(ctx))
+
+	t.Log("Online checking the member count")
+	mresp, merr := m.Etcdctl().MemberList(ctx, false)
+	require.NoError(t, merr)
+	require.Len(t, mresp.Members, 1)
+
+	t.Log("Closing the member")
+	require.NoError(t, m.Close())
+	require.NoError(t, m.Wait(ctx))
+
+	t.Log("Offline checking the member count")
+	members := mustReadMembersFromBoltDB(t, m.Config().DataDirPath)
+	require.Len(t, members, 1)
+}
+
+func mustReadMembersFromBoltDB(t *testing.T, dataDir string) []*membership.Member {
+	dbPath := datadir.ToBackendFileName(dataDir)
+	db, err := bbolt.Open(dbPath, 0o400, &bbolt.Options{ReadOnly: true})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	var members []*membership.Member
+	_ = db.View(func(tx *bbolt.Tx) error {
+		b := tx.Bucket(schema.Members.Name())
+		_ = b.ForEach(func(k, v []byte) error {
+			m := membership.Member{}
+			err := json.Unmarshal(v, &m)
+			require.NoError(t, err)
+			members = append(members, &m)
+			return nil
+		})
+		return nil
+	})
+
+	return members
+}

tests/framework/e2e/cluster.go

@@ -1002,6 +1002,38 @@ func (epc *EtcdProcessCluster) rollingStart(f func(ep EtcdProcess) error) error
 	return nil
 }
 
+func (epc *EtcdProcessCluster) Kill() (err error) {
+	for _, p := range epc.Procs {
+		if p == nil {
+			continue
+		}
+		if curErr := p.Kill(); curErr != nil {
+			if err != nil {
+				err = fmt.Errorf("%w; %w", err, curErr)
+			} else {
+				err = curErr
+			}
+		}
+	}
+	return err
+}
+
+func (epc *EtcdProcessCluster) Wait(ctx context.Context) error {
+	closedC := make(chan error, len(epc.Procs))
+	for i := range epc.Procs {
+		go func(n int) {
+			epc.Procs[n].Wait(ctx)
+			closedC <- epc.Procs[n].Wait(ctx)
+		}(i)
+	}
+	for range epc.Procs {
+		if err := <-closedC; err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (epc *EtcdProcessCluster) Stop() (err error) {
 	for _, p := range epc.Procs {
 		if p == nil {