Merge pull request #13377 from hexfusion/add-learner-limit-flag

Add experimental-max-learners flag

server/etcdserver/api/membership/cluster.go

@@ -40,8 +40,6 @@ import (
 	"go.uber.org/zap"
 )
 
-const maxLearners = 1
-
 // RaftCluster is a list of Members that belong to the same raft cluster
 type RaftCluster struct {
 	lg *zap.Logger
@@ -60,6 +58,7 @@ type RaftCluster struct {
 	removed map[types.ID]bool
 
 	downgradeInfo  *serverversion.DowngradeInfo
+	maxLearners    int
 	versionChanged *notify.Notifier
 }
 
@@ -81,8 +80,8 @@ const (
 
 // NewClusterFromURLsMap creates a new raft cluster using provided urls map. Currently, it does not support creating
 // cluster with raft learner member.
-func NewClusterFromURLsMap(lg *zap.Logger, token string, urlsmap types.URLsMap) (*RaftCluster, error) {
-	c := NewCluster(lg)
+func NewClusterFromURLsMap(lg *zap.Logger, token string, urlsmap types.URLsMap, opts ...ClusterOption) (*RaftCluster, error) {
+	c := NewCluster(lg, opts...)
 	for name, urls := range urlsmap {
 		m := NewMember(name, urls, token, nil)
 		if _, ok := c.members[m.ID]; ok {
@@ -97,8 +96,8 @@ func NewClusterFromURLsMap(lg *zap.Logger, token string, urlsmap types.URLsMap)
 	return c, nil
 }
 
-func NewClusterFromMembers(lg *zap.Logger, id types.ID, membs []*Member) *RaftCluster {
-	c := NewCluster(lg)
+func NewClusterFromMembers(lg *zap.Logger, id types.ID, membs []*Member, opts ...ClusterOption) *RaftCluster {
+	c := NewCluster(lg, opts...)
 	c.cid = id
 	for _, m := range membs {
 		c.members[m.ID] = m
@@ -106,15 +105,18 @@ func NewClusterFromMembers(lg *zap.Logger, id types.ID, membs []*Member) *RaftCl
 	return c
 }
 
-func NewCluster(lg *zap.Logger) *RaftCluster {
+func NewCluster(lg *zap.Logger, opts ...ClusterOption) *RaftCluster {
 	if lg == nil {
 		lg = zap.NewNop()
 	}
+	clOpts := newClusterOpts(opts...)
+
 	return &RaftCluster{
 		lg:            lg,
 		members:       make(map[types.ID]*Member),
 		removed:       make(map[types.ID]bool),
 		downgradeInfo: &serverversion.DowngradeInfo{Enabled: false},
+		maxLearners:   clOpts.maxLearners,
 	}
 }
 
@@ -289,6 +291,7 @@ func (c *RaftCluster) Recover(onSet func(*zap.Logger, *semver.Version)) {
 			zap.String("local-member-id", c.localID.String()),
 			zap.String("recovered-remote-peer-id", m.ID.String()),
 			zap.Strings("recovered-remote-peer-urls", m.PeerURLs),
+			zap.Bool("recovered-remote-peer-is-learner", m.IsLearner),
 		)
 	}
 	if c.version != nil {
@@ -303,9 +306,9 @@ func (c *RaftCluster) Recover(onSet func(*zap.Logger, *semver.Version)) {
 // ensures that it is still valid.
 func (c *RaftCluster) ValidateConfigurationChange(cc raftpb.ConfChange) error {
 	// TODO: this must be switched to backend as well.
-	members, removed := membersFromStore(c.lg, c.v2store)
+	membersMap, removedMap := membersFromStore(c.lg, c.v2store)
 	id := types.ID(cc.NodeID)
-	if removed[id] {
+	if removedMap[id] {
 		return ErrIDRemoved
 	}
 	switch cc.Type {
@@ -316,19 +319,21 @@ func (c *RaftCluster) ValidateConfigurationChange(cc raftpb.ConfChange) error {
 		}
 
 		if confChangeContext.IsPromote { // promoting a learner member to voting member
-			if members[id] == nil {
+			if membersMap[id] == nil {
 				return ErrIDNotFound
 			}
-			if !members[id].IsLearner {
+			if !membersMap[id].IsLearner {
 				return ErrMemberNotLearner
 			}
 		} else { // adding a new member
-			if members[id] != nil {
+			if membersMap[id] != nil {
 				return ErrIDExists
 			}
 
+			var members []*Member
 			urls := make(map[string]bool)
-			for _, m := range members {
+			for _, m := range membersMap {
+				members = append(members, m)
 				for _, u := range m.PeerURLs {
 					urls[u] = true
 				}
@@ -339,29 +344,24 @@ func (c *RaftCluster) ValidateConfigurationChange(cc raftpb.ConfChange) error {
 				}
 			}
 
-			if confChangeContext.Member.IsLearner { // the new member is a learner
-				numLearners := 0
-				for _, m := range members {
-					if m.IsLearner {
-						numLearners++
-					}
-				}
-				if numLearners+1 > maxLearners {
-					return ErrTooManyLearners
+			if confChangeContext.Member.RaftAttributes.IsLearner && cc.Type == raftpb.ConfChangeAddLearnerNode { // the new member is a learner
+				scaleUpLearners := true
+				if err := ValidateMaxLearnerConfig(c.maxLearners, members, scaleUpLearners); err != nil {
+					return err
 				}
 			}
 		}
 	case raftpb.ConfChangeRemoveNode:
-		if members[id] == nil {
+		if membersMap[id] == nil {
 			return ErrIDNotFound
 		}
 
 	case raftpb.ConfChangeUpdateNode:
-		if members[id] == nil {
+		if membersMap[id] == nil {
 			return ErrIDNotFound
 		}
 		urls := make(map[string]bool)
-		for _, m := range members {
+		for _, m := range membersMap {
 			if m.ID == id {
 				continue
 			}
@@ -407,6 +407,7 @@ func (c *RaftCluster) AddMember(m *Member, shouldApplyV3 ShouldApplyV3) {
 		zap.String("local-member-id", c.localID.String()),
 		zap.String("added-peer-id", m.ID.String()),
 		zap.Strings("added-peer-peer-urls", m.PeerURLs),
+		zap.Bool("added-peer-is-learner", m.IsLearner),
 	)
 }
 
@@ -434,6 +435,7 @@ func (c *RaftCluster) RemoveMember(id types.ID, shouldApplyV3 ShouldApplyV3) {
 			zap.String("local-member-id", c.localID.String()),
 			zap.String("removed-remote-peer-id", id.String()),
 			zap.Strings("removed-remote-peer-urls", m.PeerURLs),
+			zap.Bool("removed-remote-peer-is-learner", m.IsLearner),
 		)
 	} else {
 		c.lg.Warn(
@@ -517,6 +519,7 @@ func (c *RaftCluster) UpdateRaftAttributes(id types.ID, raftAttr RaftAttributes,
 		zap.String("local-member-id", c.localID.String()),
 		zap.String("updated-remote-peer-id", id.String()),
 		zap.Strings("updated-remote-peer-urls", raftAttr.PeerURLs),
+		zap.Bool("updated-remote-peer-is-learner", raftAttr.IsLearner),
 	)
 }
 
@@ -831,3 +834,24 @@ func (c *RaftCluster) updateMembershipMetric(peer types.ID, known bool) {
 	}
 	knownPeers.WithLabelValues(c.localID.String(), peer.String()).Set(v)
 }
+
+// ValidateMaxLearnerConfig verifies the existing learner members in the cluster membership and an optional N+1 learner
+// scale up are not more than maxLearners.
+func ValidateMaxLearnerConfig(maxLearners int, members []*Member, scaleUpLearners bool) error {
+	numLearners := 0
+	for _, m := range members {
+		if m.IsLearner {
+			numLearners++
+		}
+	}
+	// Validate config can accommodate scale up.
+	if scaleUpLearners {
+		numLearners++
+	}
+
+	if numLearners > maxLearners {
+		return ErrTooManyLearners
+	}
+
+	return nil
+}

server/etcdserver/api/membership/cluster_opts.go

@@ -0,0 +1,43 @@
+// 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
+
+const DefaultMaxLearners = 1
+
+type ClusterOptions struct {
+	maxLearners int
+}
+
+// ClusterOption are options which can be applied to the raft cluster.
+type ClusterOption func(*ClusterOptions)
+
+func newClusterOpts(opts ...ClusterOption) *ClusterOptions {
+	clOpts := &ClusterOptions{}
+	clOpts.applyOpts(opts)
+	return clOpts
+}
+
+func (co *ClusterOptions) applyOpts(opts []ClusterOption) {
+	for _, opt := range opts {
+		opt(co)
+	}
+}
+
+// WithMaxLearners sets the maximum number of learners that can exist in the cluster membership.
+func WithMaxLearners(max int) ClusterOption {
+	return func(co *ClusterOptions) {
+		co.maxLearners = max
+	}
+}

server/etcdserver/api/membership/cluster_test.go

@@ -278,10 +278,14 @@ func TestClusterValidateAndAssignIDs(t *testing.T) {
 }
 
 func TestClusterValidateConfigurationChange(t *testing.T) {
-	cl := NewCluster(zaptest.NewLogger(t))
+	cl := NewCluster(zaptest.NewLogger(t), WithMaxLearners(1))
 	cl.SetStore(v2store.New())
 	for i := 1; i <= 4; i++ {
-		attr := RaftAttributes{PeerURLs: []string{fmt.Sprintf("http://127.0.0.1:%d", i)}}
+		var isLearner bool
+		if i == 1 {
+			isLearner = true
+		}
+		attr := RaftAttributes{PeerURLs: []string{fmt.Sprintf("http://127.0.0.1:%d", i)}, IsLearner: isLearner}
 		cl.AddMember(&Member{ID: types.ID(i), RaftAttributes: attr}, true)
 	}
 	cl.RemoveMember(4, true)
@@ -326,6 +330,17 @@ func TestClusterValidateConfigurationChange(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	attr = RaftAttributes{PeerURLs: []string{fmt.Sprintf("http://127.0.0.1:%d", 7)}, IsLearner: true}
+	ctx7, err := json.Marshal(&ConfigChangeContext{Member: Member{ID: types.ID(7), RaftAttributes: attr}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	attr = RaftAttributes{PeerURLs: []string{fmt.Sprintf("http://127.0.0.1:%d", 1)}, IsLearner: true}
+	ctx8, err := json.Marshal(&ConfigChangeContext{Member: Member{ID: types.ID(1), RaftAttributes: attr}, IsPromote: true})
+	if err != nil {
+		t.Fatal(err)
+	}
 	tests := []struct {
 		cc   raftpb.ConfChange
 		werr error
@@ -423,6 +438,22 @@ func TestClusterValidateConfigurationChange(t *testing.T) {
 			},
 			ErrIDNotFound,
 		},
+		{
+			raftpb.ConfChange{
+				Type:    raftpb.ConfChangeAddLearnerNode,
+				NodeID:  7,
+				Context: ctx7,
+			},
+			ErrTooManyLearners,
+		},
+		{
+			raftpb.ConfChange{
+				Type:    raftpb.ConfChangeAddNode,
+				NodeID:  1,
+				Context: ctx8,
+			},
+			nil,
+		},
 	}
 	for i, tt := range tests {
 		err := cl.ValidateConfigurationChange(tt.cc)

server/etcdserver/api/membership/membership_test.go

@@ -15,12 +15,14 @@ func TestAddRemoveMember(t *testing.T) {
 	c := newTestCluster(t, nil)
 	be := &backendMock{}
 	c.SetBackend(be)
-	c.AddMember(newTestMember(17, nil, "node17", nil), true)
+	c.AddMember(newTestMemberAsLearner(17, nil, "node17", nil), true)
 	c.RemoveMember(17, true)
 	c.AddMember(newTestMember(18, nil, "node18", nil), true)
+	c.RemoveMember(18, true)
 
 	// Skipping removal of already removed member
 	c.RemoveMember(17, true)
+	c.RemoveMember(18, true)
 
 	if false {
 		// TODO: Enable this code when Recover is reading membership from the backend.