add runtime reconfiguration tests

Signed-off-by: Chao Chen <chaochn@amazon.com>

tests/e2e/ctl_v3_auth_cluster_test.go

@@ -64,7 +64,7 @@ func TestAuthCluster(t *testing.T) {
 	}
 
 	// start second process
-	if err := epc.StartNewProc(ctx, nil, t, rootUserClientOpts); err != nil {
+	if _, err := epc.StartNewProc(ctx, nil, t, e2e.MemberAdd, rootUserClientOpts); err != nil {
 		t.Fatalf("could not start second etcd process (%v)", err)
 	}
 

tests/e2e/etcd_grpcproxy_test.go

@@ -64,7 +64,7 @@ func TestGrpcProxyAutoSync(t *testing.T) {
 	require.NoError(t, err)
 
 	// Add and start second member
-	err = epc.StartNewProc(ctx, nil, t)
+	_, err = epc.StartNewProc(ctx, nil, t, e2e.MemberAdd)
 	require.NoError(t, err)
 
 	// Wait for auto sync of endpoints

tests/e2e/etcd_mix_versions_test.go

@@ -92,7 +92,7 @@ func mixVersionsSnapshotTestByAddingMember(t *testing.T, clusterVersion, newInst
 	newCfg.Version = newInstanceVersion
 	newCfg.SnapshotCatchUpEntries = 10
 	t.Log("Starting a new etcd instance")
-	err = epc.StartNewProc(context.TODO(), &newCfg, t)
+	_, err = epc.StartNewProc(context.TODO(), &newCfg, t, e2e.MemberAdd)
 	require.NoError(t, err, "failed to start the new etcd instance: %v", err)
 	defer epc.CloseProc(context.TODO(), nil)
 

tests/e2e/runtime_reconfiguration_test.go

@@ -0,0 +1,201 @@
+// Copyright 2023 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.
+
+//go:build !cluster_proxy
+
+package e2e
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"go.etcd.io/etcd/server/v3/etcdserver"
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+)
+
+// TestRuntimeReconfigGrowClusterSize ensures growing cluster size with two phases
+// Phase 1 - Inform cluster of new configuration
+// Phase 2 - Start new member
+func TestRuntimeReconfigGrowClusterSize(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	tcs := []struct {
+		name        string
+		clusterSize int
+		asLearner   bool
+	}{
+		{
+			name:        "grow cluster size from 1 to 3",
+			clusterSize: 1,
+		},
+		{
+			name:        "grow cluster size from 3 to 5",
+			clusterSize: 3,
+		},
+		{
+			name:        "grow cluster size from 1 to 3 with learner",
+			clusterSize: 1,
+			asLearner:   true,
+		},
+		{
+			name:        "grow cluster size from 3 to 5 with learner",
+			clusterSize: 3,
+			asLearner:   true,
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+
+			epc, err := e2e.NewEtcdProcessCluster(ctx, t, e2e.WithClusterSize(tc.clusterSize))
+			require.NoError(t, err)
+			require.NoError(t, epc.Procs[0].Etcdctl().Health(ctx))
+			defer func() {
+				err := epc.Close()
+				require.NoError(t, err, "failed to close etcd cluster: %v", err)
+			}()
+
+			for i := 0; i < 2; i++ {
+				time.Sleep(etcdserver.HealthInterval)
+				if !tc.asLearner {
+					addMember(ctx, t, epc)
+				} else {
+					addMemberAsLearner(ctx, t, epc)
+				}
+			}
+		})
+	}
+}
+
+func TestRuntimeReconfigDecreaseClusterSize(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	tcs := []struct {
+		name        string
+		clusterSize int
+		asLearner   bool
+	}{
+		{
+			name:        "decrease cluster size from 3 to 1",
+			clusterSize: 3,
+		},
+		{
+			name:        "decrease cluster size from 5 to 3",
+			clusterSize: 5,
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+
+			epc, err := e2e.NewEtcdProcessCluster(ctx, t, e2e.WithClusterSize(tc.clusterSize))
+			require.NoError(t, err)
+			require.NoError(t, epc.Procs[0].Etcdctl().Health(ctx))
+			defer func() {
+				err := epc.Close()
+				require.NoError(t, err, "failed to close etcd cluster: %v", err)
+			}()
+
+			for i := 0; i < 2; i++ {
+				time.Sleep(etcdserver.HealthInterval)
+				removeMember(ctx, t, epc)
+			}
+		})
+	}
+}
+
+func TestRuntimeReconfigRollingUpgrade(t *testing.T) {
+	e2e.BeforeTest(t)
+
+	tcs := []struct {
+		name        string
+		withLearner bool
+	}{
+		{
+			name:        "with learner",
+			withLearner: true,
+		},
+		{
+			name:        "without learner",
+			withLearner: false,
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+
+			epc, err := e2e.NewEtcdProcessCluster(ctx, t, e2e.WithClusterSize(3))
+			require.NoError(t, err)
+			require.NoError(t, epc.Procs[0].Etcdctl().Health(ctx))
+			defer func() {
+				err := epc.Close()
+				require.NoError(t, err, "failed to close etcd cluster: %v", err)
+			}()
+
+			for i := 0; i < 2; i++ {
+				time.Sleep(etcdserver.HealthInterval)
+				removeMember(ctx, t, epc)
+				epc.WaitLeader(t)
+				// if we do not wait for leader, without the fix of notify raft Advance,
+				// have to wait 1 sec to pass the test stably.
+				if tc.withLearner {
+					addMemberAsLearner(ctx, t, epc)
+				} else {
+					addMember(ctx, t, epc)
+				}
+			}
+		})
+	}
+}
+
+func addMember(ctx context.Context, t *testing.T, epc *e2e.EtcdProcessCluster) {
+	_, err := epc.StartNewProc(ctx, nil, t, e2e.MemberAdd)
+	require.NoError(t, err)
+	require.NoError(t, epc.Procs[len(epc.Procs)-1].Etcdctl().Health(ctx))
+}
+
+func addMemberAsLearner(ctx context.Context, t *testing.T, epc *e2e.EtcdProcessCluster) {
+	id, err := epc.StartNewProc(ctx, nil, t, e2e.MemberAddAsLearner)
+	require.NoError(t, err)
+	newLearnerMemberProc := epc.Procs[len(epc.Procs)-1]
+	_, err = epc.Etcdctl().MemberPromote(ctx, id)
+	require.NoError(t, err)
+	require.NoError(t, newLearnerMemberProc.Etcdctl().Health(ctx))
+}
+
+func removeMember(ctx context.Context, t *testing.T, epc *e2e.EtcdProcessCluster) {
+	// avoid tearing down the last etcd process
+	if len(epc.Procs) == 1 {
+		return
+	}
+
+	lastProcIndex := len(epc.Procs) - 1
+	lastProc := epc.Procs[lastProcIndex]
+	sts, err := lastProc.Etcdctl().Status(ctx)
+	require.NoError(t, err)
+	toRemoveMemberID := sts[0].Header.MemberId
+
+	epc.Procs = epc.Procs[:lastProcIndex]
+	_, err = epc.Etcdctl().MemberRemove(ctx, toRemoveMemberID)
+	require.NoError(t, err)
+	require.NoError(t, lastProc.Stop())
+	require.NoError(t, lastProc.Close())
+}

tests/framework/e2e/cluster.go

@@ -30,6 +30,7 @@ import (
 	"go.uber.org/zap/zaptest"
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/pkg/v3/proxy"
 	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/tests/v3/framework/config"
@@ -780,7 +781,18 @@ func (epc *EtcdProcessCluster) CloseProc(ctx context.Context, finder func(EtcdPr
 	return proc.Close()
 }
 
-func (epc *EtcdProcessCluster) StartNewProc(ctx context.Context, cfg *EtcdProcessClusterConfig, tb testing.TB, opts ...config.ClientOption) error {
+type MemberAddFunc func(ctx context.Context, ctlv3 *EtcdctlV3, name string, peerAddrs []string) (*clientv3.MemberAddResponse, error)
+
+var (
+	MemberAdd = func(ctx context.Context, ctlv3 *EtcdctlV3, name string, peerAddrs []string) (*clientv3.MemberAddResponse, error) {
+		return ctlv3.MemberAdd(ctx, name, peerAddrs)
+	}
+	MemberAddAsLearner = func(ctx context.Context, ctlv3 *EtcdctlV3, name string, peerAddrs []string) (*clientv3.MemberAddResponse, error) {
+		return ctlv3.MemberAddAsLearner(ctx, name, peerAddrs)
+	}
+)
+
+func (epc *EtcdProcessCluster) StartNewProc(ctx context.Context, cfg *EtcdProcessClusterConfig, tb testing.TB, memberAddFunc MemberAddFunc, opts ...config.ClientOption) (memberID uint64, err error) {
 	var serverCfg *EtcdServerProcessConfig
 	if cfg != nil {
 		serverCfg = cfg.EtcdServerProcessConfig(tb, epc.nextSeq)
@@ -801,21 +813,21 @@ func (epc *EtcdProcessCluster) StartNewProc(ctx context.Context, cfg *EtcdProces
 
 	// First add new member to cluster
 	memberCtl := epc.Etcdctl(opts...)
-	_, err := memberCtl.MemberAdd(ctx, serverCfg.Name, []string{serverCfg.PeerURL.String()})
+	resp, err := memberAddFunc(ctx, memberCtl, serverCfg.Name, []string{serverCfg.PeerURL.String()})
 	if err != nil {
-		return fmt.Errorf("failed to add new member: %w", err)
+		return 0, fmt.Errorf("failed to add new member: %w", err)
 	}
 
 	// Then start process
 	proc, err := NewEtcdProcess(serverCfg)
 	if err != nil {
 		epc.Close()
-		return fmt.Errorf("cannot configure: %v", err)
+		return 0, fmt.Errorf("cannot configure: %v", err)
 	}
 
 	epc.Procs = append(epc.Procs, proc)
 
-	return proc.Start(ctx)
+	return resp.Member.ID, proc.Start(ctx)
 }
 
 // UpdateProcOptions updates the options for a specific process. If no opt is set, then the config is identical

tests/framework/e2e/etcdctl.go

@@ -302,6 +302,12 @@ func (ctl *EtcdctlV3) MemberRemove(ctx context.Context, id uint64) (*clientv3.Me
 	return &resp, err
 }
 
+func (ctl *EtcdctlV3) MemberPromote(ctx context.Context, id uint64) (*clientv3.MemberPromoteResponse, error) {
+	var resp clientv3.MemberPromoteResponse
+	err := ctl.spawnJsonCmd(ctx, &resp, "member", "promote", fmt.Sprintf("%x", id))
+	return &resp, err
+}
+
 func (ctl *EtcdctlV3) cmdArgs(args ...string) []string {
 	cmdArgs := []string{BinPath.Etcdctl}
 	for k, v := range ctl.flags() {