Merge pull request #12962 from ptabor/20210513-write-conf-state

Save raftpb.ConfState in the backend.

server/etcdserver/api/membership/cluster.go

@@ -694,6 +694,7 @@ func clusterVersionFromStore(lg *zap.Logger, st v2store.Store) *semver.Version {
 	return semver.Must(semver.NewVersion(*e.Node.Value))
 }
 
+// The field is populated since etcd v3.5.
 func clusterVersionFromBackend(lg *zap.Logger, be backend.Backend) *semver.Version {
 	ckey := backendClusterVersionKey()
 	tx := be.ReadTx()
@@ -712,6 +713,7 @@ func clusterVersionFromBackend(lg *zap.Logger, be backend.Backend) *semver.Versi
 	return semver.Must(semver.NewVersion(string(vals[0])))
 }
 
+// The field is populated since etcd v3.5.
 func downgradeInfoFromBackend(lg *zap.Logger, be backend.Backend) *DowngradeInfo {
 	dkey := backendDowngradeKey()
 	tx := be.ReadTx()

server/etcdserver/api/membership/confstate.go

@@ -0,0 +1,63 @@
+// Copyright 2021 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package membership
+
+import (
+	"encoding/json"
+	"log"
+
+	"go.etcd.io/etcd/raft/v3/raftpb"
+	"go.etcd.io/etcd/server/v3/mvcc"
+	"go.etcd.io/etcd/server/v3/mvcc/backend"
+	"go.uber.org/zap"
+)
+
+var (
+	confStateKey = []byte("confState")
+)
+
+// MustUnsafeSaveConfStateToBackend persists confState using given transaction (tx).
+// confState in backend is persisted since etcd v3.5.
+func MustUnsafeSaveConfStateToBackend(lg *zap.Logger, tx backend.BatchTx, confState *raftpb.ConfState) {
+	confStateBytes, err := json.Marshal(confState)
+	if err != nil {
+		lg.Panic("Cannot marshal raftpb.ConfState", zap.Stringer("conf-state", confState), zap.Error(err))
+	}
+
+	tx.UnsafePut(mvcc.MetaBucketName, confStateKey, confStateBytes)
+}
+
+// UnsafeConfStateFromBackend retrieves ConfState from the backend.
+// Returns nil if confState in backend is not persisted (e.g. backend writen by <v3.5).
+func UnsafeConfStateFromBackend(lg *zap.Logger, tx backend.ReadTx) *raftpb.ConfState {
+	keys, vals := tx.UnsafeRange(mvcc.MetaBucketName, confStateKey, nil, 0)
+	if len(keys) == 0 {
+		return nil
+	}
+
+	if len(keys) != 1 {
+		lg.Panic(
+			"unexpected number of key: "+string(confStateKey)+" when getting cluster version from backend",
+			zap.Int("number-of-key", len(keys)),
+		)
+	}
+	var confState raftpb.ConfState
+	if err := json.Unmarshal(vals[0], &confState); err != nil {
+		log.Panic("Cannot unmarshal confState json retrieved from the backend",
+			zap.ByteString("conf-state-json", vals[0]),
+			zap.Error(err))
+	}
+	return &confState
+}

server/etcdserver/api/membership/confstate_test.go

@@ -0,0 +1,79 @@
+// Copyright 2021 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package membership_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/raft/v3/raftpb"
+	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
+	"go.etcd.io/etcd/server/v3/etcdserver/cindex"
+	betesting "go.etcd.io/etcd/server/v3/mvcc/backend/testing"
+	"go.uber.org/zap/zaptest"
+)
+
+func TestConfStateFromBackendInOneTx(t *testing.T) {
+	lg := zaptest.NewLogger(t)
+	be, _ := betesting.NewDefaultTmpBackend(t)
+	defer betesting.Close(t, be)
+
+	tx := be.BatchTx()
+	cindex.CreateMetaBucket(tx)
+	tx.Lock()
+	defer tx.Unlock()
+	assert.Nil(t, membership.UnsafeConfStateFromBackend(lg, tx))
+
+	confState := raftpb.ConfState{Learners: []uint64{1, 2}, Voters: []uint64{3}, AutoLeave: false}
+	membership.MustUnsafeSaveConfStateToBackend(lg, tx, &confState)
+
+	assert.Equal(t, confState, *membership.UnsafeConfStateFromBackend(lg, tx))
+}
+
+func TestMustUnsafeSaveConfStateToBackend(t *testing.T) {
+	lg := zaptest.NewLogger(t)
+	be, _ := betesting.NewDefaultTmpBackend(t)
+	defer betesting.Close(t, be)
+
+	{
+		tx := be.BatchTx()
+		cindex.CreateMetaBucket(tx)
+		tx.Commit()
+	}
+
+	t.Run("missing", func(t *testing.T) {
+		tx := be.ReadTx()
+		tx.Lock()
+		defer tx.Unlock()
+		assert.Nil(t, membership.UnsafeConfStateFromBackend(lg, tx))
+	})
+
+	confState := raftpb.ConfState{Learners: []uint64{1, 2}, Voters: []uint64{3}, AutoLeave: false}
+
+	t.Run("save", func(t *testing.T) {
+		tx := be.BatchTx()
+		tx.Lock()
+		membership.MustUnsafeSaveConfStateToBackend(lg, tx, &confState)
+		tx.Unlock()
+		tx.Commit()
+	})
+
+	t.Run("read", func(t *testing.T) {
+		tx := be.ReadTx()
+		tx.Lock()
+		defer tx.Unlock()
+		assert.Equal(t, confState, *membership.UnsafeConfStateFromBackend(lg, tx))
+	})
+}

server/etcdserver/api/membership/store.go

@@ -161,6 +161,7 @@ func TrimMembershipFromV2Store(lg *zap.Logger, s v2store.Store) error {
 	return nil
 }
 
+// The field is populated since etcd v3.5.
 func mustSaveClusterVersionToBackend(be backend.Backend, ver *semver.Version) {
 	ckey := backendClusterVersionKey()
 
@@ -170,6 +171,7 @@ func mustSaveClusterVersionToBackend(be backend.Backend, ver *semver.Version) {
 	tx.UnsafePut(clusterBucketName, ckey, []byte(ver.String()))
 }
 
+// The field is populated since etcd v3.5.
 func mustSaveDowngradeToBackend(lg *zap.Logger, be backend.Backend, downgrade *DowngradeInfo) {
 	dkey := backendDowngradeKey()
 	dvalue, err := json.Marshal(downgrade)