robustness: add mix version scenario with fixed leader.

Signed-off-by: Siyuan Zhang <[email protected]>

tests/framework/e2e/cluster.go

@@ -27,6 +27,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 
@@ -151,8 +152,11 @@ type EtcdProcessClusterConfig struct {
 
 	// Cluster setup config
 
-	ClusterSize  int
-	RollingStart bool
+	ClusterSize int
+	// InitialLeaderIndex makes sure the leader is the ith proc
+	// when the cluster starts if it is specified (>=0).
+	InitialLeaderIndex int
+	RollingStart       bool
 	// BaseDataDirPath specifies the data-dir for the members. If test cases
 	// do not specify `BaseDataDirPath`, then e2e framework creates a
 	// temporary directory for each member; otherwise, it creates a
@@ -180,10 +184,10 @@ type EtcdProcessClusterConfig struct {
 
 func DefaultConfig() *EtcdProcessClusterConfig {
 	cfg := &EtcdProcessClusterConfig{
-		ClusterSize: 3,
-		CN:          true,
-
-		ServerConfig: *embed.NewConfig(),
+		ClusterSize:        3,
+		CN:                 true,
+		InitialLeaderIndex: -1,
+		ServerConfig:       *embed.NewConfig(),
 	}
 	cfg.ServerConfig.InitialClusterToken = "new"
 	return cfg
@@ -207,6 +211,10 @@ func WithVersion(version ClusterVersion) EPClusterOption {
 	return func(c *EtcdProcessClusterConfig) { c.Version = version }
 }
 
+func WithInitialLeaderIndex(i int) EPClusterOption {
+	return func(c *EtcdProcessClusterConfig) { c.InitialLeaderIndex = i }
+}
+
 func WithDataDirPath(path string) EPClusterOption {
 	return func(c *EtcdProcessClusterConfig) { c.BaseDataDirPath = path }
 }
@@ -437,7 +445,11 @@ func StartEtcdProcessCluster(ctx context.Context, t testing.TB, epc *EtcdProcess
 			t.Skip("please run 'make gofail-enable && make build' before running the test")
 		}
 	}
-
+	if cfg.InitialLeaderIndex >= 0 {
+		if err := epc.MoveLeader(ctx, t, cfg.InitialLeaderIndex); err != nil {
+			return nil, fmt.Errorf("failed to move leader: %v", err)
+		}
+	}
 	return epc, nil
 }
 
@@ -462,6 +474,18 @@ func (cfg *EtcdProcessClusterConfig) EtcdAllServerProcessConfigs(tb testing.TB)
 		cfg.SetInitialOrDiscovery(etcdCfgs[i], initialCluster, "new")
 	}
 
+	// validate SnapshotCatchUpEntries could be set for at least one member
+	if cfg.ServerConfig.SnapshotCatchUpEntries > 0 {
+		expectArgsContain := []string{
+			fmt.Sprintf("--experimental-snapshot-catchup-entries=%d", cfg.ServerConfig.SnapshotCatchUpEntries),
+		}
+		aggArgs := []string{}
+		for _, etcdCfg := range etcdCfgs {
+			aggArgs = append(aggArgs, etcdCfg.Args...)
+		}
+		assert.Subset(tb, aggArgs, expectArgsContain)
+	}
+
 	return etcdCfgs
 }
 
@@ -1050,3 +1074,31 @@ func (epc *EtcdProcessCluster) WaitMembersForLeader(ctx context.Context, t testi
 	t.Fatal("impossible path of execution")
 	return -1
 }
+
+// MoveLeader moves the leader to the ith process.
+func (epc *EtcdProcessCluster) MoveLeader(ctx context.Context, t testing.TB, i int) error {
+	if i < 0 || i >= len(epc.Procs) {
+		return fmt.Errorf("invalid index: %d, must between 0 and %d", i, len(epc.Procs)-1)
+	}
+	t.Logf("moving leader to Procs[%d]", i)
+	oldLeader := epc.WaitMembersForLeader(ctx, t, epc.Procs)
+	if oldLeader == i {
+		t.Logf("Procs[%d] is already the leader", i)
+		return nil
+	}
+	resp, err := epc.Procs[i].Etcdctl().Status(ctx)
+	if err != nil {
+		return err
+	}
+	memberID := resp[0].Header.MemberId
+	err = epc.Procs[oldLeader].Etcdctl().MoveLeader(ctx, memberID)
+	if err != nil {
+		return err
+	}
+	newLeader := epc.WaitMembersForLeader(ctx, t, epc.Procs)
+	if newLeader != i {
+		t.Fatalf("expect new leader to be Procs[%d] but got Procs[%d]", i, newLeader)
+	}
+	t.Logf("moved leader from Procs[%d] to Procs[%d]", oldLeader, i)
+	return nil
+}

tests/framework/e2e/cluster_test.go

@@ -15,17 +15,22 @@
 package e2e
 
 import (
+	"fmt"
 	"testing"
 
+	"github.com/coreos/go-semver/semver"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestEtcdServerProcessConfig(t *testing.T) {
+	v3_5_12 := semver.Version{Major: 3, Minor: 5, Patch: 12}
+	v3_5_13 := semver.Version{Major: 3, Minor: 5, Patch: 13}
 	tcs := []struct {
 		name                 string
 		config               *EtcdProcessClusterConfig
 		expectArgsNotContain []string
 		expectArgsContain    []string
+		mockBinaryVersion    *semver.Version
 	}{
 		{
 			name:   "Default",
@@ -73,10 +78,36 @@ func TestEtcdServerProcessConfig(t *testing.T) {
 			expectArgsContain: []string{
 				"--experimental-snapshot-catchup-entries=100",
 			},
+			mockBinaryVersion: &v3_5_13,
+		},
+		{
+			name:   "CatchUpEntriesNoVersion",
+			config: NewConfig(WithSnapshotCatchUpEntries(100), WithVersion(LastVersion)),
+			expectArgsNotContain: []string{
+				"--experimental-snapshot-catchup-entries=100",
+			},
+		},
+		{
+			name:   "CatchUpEntriesOldVersion",
+			config: NewConfig(WithSnapshotCatchUpEntries(100), WithVersion(LastVersion)),
+			expectArgsNotContain: []string{
+				"--experimental-snapshot-catchup-entries=100",
+			},
+			mockBinaryVersion: &v3_5_12,
 		},
 	}
+	origGetVersionFromBinary := GetVersionFromBinary
 	for _, tc := range tcs {
 		t.Run(tc.name, func(t *testing.T) {
+			if tc.mockBinaryVersion == nil {
+				GetVersionFromBinary = func(binaryPath string) (*semver.Version, error) {
+					return nil, fmt.Errorf("could not get binary version")
+				}
+			} else {
+				GetVersionFromBinary = func(binaryPath string) (*semver.Version, error) {
+					return tc.mockBinaryVersion, nil
+				}
+			}
 			args := tc.config.EtcdServerProcessConfig(t, 0).Args
 			if len(tc.expectArgsContain) != 0 {
 				assert.Subset(t, args, tc.expectArgsContain)
@@ -86,4 +117,5 @@ func TestEtcdServerProcessConfig(t *testing.T) {
 			}
 		})
 	}
+	GetVersionFromBinary = origGetVersionFromBinary
 }

tests/framework/e2e/etcd_process.go

@@ -485,7 +485,10 @@ func parseFailpointsBody(body io.Reader) (map[string]string, error) {
 	return failpoints, nil
 }
 
-func GetVersionFromBinary(binaryPath string) (*semver.Version, error) {
+var GetVersionFromBinary = func(binaryPath string) (*semver.Version, error) {
+	if !fileutil.Exist(binaryPath) {
+		return nil, fmt.Errorf("binary path does not exist: %s", binaryPath)
+	}
 	lines, err := RunUtilCompletion([]string{binaryPath, "--version"}, nil)
 	if err != nil {
 		return nil, fmt.Errorf("could not find binary version from %s, err: %w", binaryPath, err)
@@ -510,14 +513,9 @@ func GetVersionFromBinary(binaryPath string) (*semver.Version, error) {
 }
 
 func CouldSetSnapshotCatchupEntries(execPath string) bool {
-	if !fileutil.Exist(execPath) {
-		// default to true if the binary does not exist, because binary might not exist for unit test,
-		// which does not really matter because "experimental-snapshot-catchup-entries" can be set for v3.6 and v3.5.
-		return true
-	}
 	v, err := GetVersionFromBinary(execPath)
 	if err != nil {
-		panic(err)
+		return false
 	}
 	// snapshot-catchup-entries flag was backported in https://github.com/etcd-io/etcd/pull/17808
 	v3_5_13 := semver.Version{Major: 3, Minor: 5, Patch: 13}

tests/framework/e2e/etcdctl.go

@@ -312,6 +312,13 @@ func (ctl *EtcdctlV3) MemberPromote(ctx context.Context, id uint64) (*clientv3.M
 	return &resp, err
 }
 
+// MoveLeader requests current leader to transfer its leadership to the transferee.
+// Request must be made to the leader.
+func (ctl *EtcdctlV3) MoveLeader(ctx context.Context, transfereeID uint64) error {
+	_, err := SpawnWithExpectLines(ctx, ctl.cmdArgs("move-leader", fmt.Sprintf("%x", transfereeID)), nil, expect.ExpectedResponse{Value: "Leadership transferred"})
+	return err
+}
+
 func (ctl *EtcdctlV3) cmdArgs(args ...string) []string {
 	cmdArgs := []string{BinPath.Etcdctl}
 	for k, v := range ctl.flags() {

tests/robustness/failpoint/cluster.go

@@ -137,7 +137,7 @@ func (f memberReplace) Name() string {
 
 func (f memberReplace) Available(config e2e.EtcdProcessClusterConfig, member e2e.EtcdProcess) bool {
 	// a lower etcd version may not be able to join a cluster with higher cluster version.
-	return config.ClusterSize > 1 && member.Config().ExecPath != e2e.BinPath.EtcdLastRelease
+	return config.ClusterSize > 1 && (config.Version == e2e.QuorumLastVersion || member.Config().ExecPath == e2e.BinPath.Etcd)
 }
 
 func getID(ctx context.Context, cc clientv3.Cluster, name string) (id uint64, found bool, err error) {

tests/robustness/options/server_config_options.go

@@ -53,3 +53,7 @@ func WithExperimentalWatchProgressNotifyInterval(input ...time.Duration) e2e.EPC
 func WithVersion(input ...e2e.ClusterVersion) e2e.EPClusterOption {
 	return func(c *e2e.EtcdProcessClusterConfig) { c.Version = input[internalRand.Intn(len(input))] }
 }
+
+func WithInitialLeaderIndex(input ...int) e2e.EPClusterOption {
+	return func(c *e2e.EtcdProcessClusterConfig) { c.InitialLeaderIndex = input[internalRand.Intn(len(input))] }
+}

tests/robustness/scenarios.go

@@ -69,9 +69,20 @@ func exploratoryScenarios(_ *testing.T) []testScenario {
 			options.ClusterOptions{options.WithTickMs(100), options.WithElectionMs(2000)}),
 	}
 
-	// 66% current version, 33% MinorityLastVersion and QuorumLastVersion
-	mixedVersionOption := options.WithVersion(e2e.CurrentVersion, e2e.CurrentVersion, e2e.CurrentVersion,
-		e2e.CurrentVersion, e2e.MinorityLastVersion, e2e.QuorumLastVersion)
+	mixedVersionOption := options.WithClusterOptionGroups(
+		// 60% with all members of current version
+		options.ClusterOptions{options.WithVersion(e2e.CurrentVersion)},
+		options.ClusterOptions{options.WithVersion(e2e.CurrentVersion)},
+		options.ClusterOptions{options.WithVersion(e2e.CurrentVersion)},
+		// 10% with 2 members of current version, 1 member last version, leader is current version
+		options.ClusterOptions{options.WithVersion(e2e.MinorityLastVersion), options.WithInitialLeaderIndex(0)},
+		// 10% with 2 members of current version, 1 member last version, leader is last version
+		options.ClusterOptions{options.WithVersion(e2e.MinorityLastVersion), options.WithInitialLeaderIndex(2)},
+		// 10% with 2 members of last version, 1 member current version, leader is last version
+		options.ClusterOptions{options.WithVersion(e2e.QuorumLastVersion), options.WithInitialLeaderIndex(0)},
+		// 10% with 2 members of last version, 1 member current version, leader is current version
+		options.ClusterOptions{options.WithVersion(e2e.QuorumLastVersion), options.WithInitialLeaderIndex(2)},
+	)
 
 	baseOptions := []e2e.EPClusterOption{
 		options.WithSnapshotCount(50, 100, 1000),