test: support mix versions sending snapshot test

tests/e2e/ctl_v3_auth_cluster_test.go

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

tests/e2e/etcd_grpcproxy_test.go

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

tests/e2e/etcd_mix_versions_test.go

@@ -0,0 +1,127 @@
+// Copyright 2022 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 e2e
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.etcd.io/etcd/client/pkg/v3/fileutil"
+	"go.etcd.io/etcd/tests/v3/framework/config"
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+)
+
+// TODO(ahrtr): add network partition scenario to trigger snapshots.
+func TestMixVersionsSendSnapshot(t *testing.T) {
+	cases := []struct {
+		name              string
+		clusterVersion    e2e.ClusterVersion
+		newInstaceVersion e2e.ClusterVersion
+	}{
+		// etcd doesn't support adding a new member of old version into
+		// a cluster with higher version. For example, etcd cluster
+		// version is 3.6.x, then a new member of 3.5.x can't join the
+		// cluster. Please refer to link below,
+		// https://github.com/etcd-io/etcd/blob/3e903d0b12e399519a4013c52d4635ec8bdd6863/server/etcdserver/cluster_util.go#L222-L230
+		/*{
+			name:              "etcd instance with last version receives snapshot from the leader with current version",
+			clusterVersion:    e2e.CurrentVersion,
+			newInstaceVersion: e2e.LastVersion,
+		},*/
+		{
+			name:              "etcd instance with current version receives snapshot from the leader with last version",
+			clusterVersion:    e2e.LastVersion,
+			newInstaceVersion: e2e.CurrentVersion,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			mixVersionsSnapshotTest(t, tc.clusterVersion, tc.newInstaceVersion)
+		})
+	}
+}
+
+func mixVersionsSnapshotTest(t *testing.T, clusterVersion, newInstanceVersion e2e.ClusterVersion) {
+	e2e.BeforeTest(t)
+
+	if !fileutil.Exist(e2e.BinPath.EtcdLastRelease) {
+		t.Skipf("%q does not exist", e2e.BinPath.EtcdLastRelease)
+	}
+
+	// Create an etcd cluster with 1 member
+	cfg := &e2e.EtcdProcessClusterConfig{
+		ClusterSize:   1,
+		InitialToken:  "new",
+		SnapshotCount: 10,
+		Version:       clusterVersion,
+	}
+
+	epc, err := e2e.NewEtcdProcessCluster(context.TODO(), t, cfg)
+	if err != nil {
+		t.Fatalf("failed to start etcd cluster: %v", err)
+	}
+	defer func() {
+		if err := epc.Close(); err != nil {
+			t.Fatalf("failed to close etcd cluster: %v", err)
+		}
+	}()
+
+	// Write more than SnapshotCount entries to trigger at least a snapshot.
+	t.Log("Writing 20 keys to the cluster")
+	for i := 0; i < 20; i++ {
+		key := fmt.Sprintf("key-%d", i)
+		value := fmt.Sprintf("value-%d", i)
+		if err := epc.Client().Put(context.TODO(), key, value, config.PutOptions{}); err != nil {
+			t.Fatalf("failed to put %q, error: %v", key, err)
+		}
+	}
+
+	// start a new etcd instance, which will receive a snapshot from the leader.
+	newCfg := *epc.Cfg
+	newCfg.Version = newInstanceVersion
+	t.Log("Starting a new etcd instance")
+	if err := epc.StartNewProc(context.TODO(), &newCfg, t); err != nil {
+		t.Fatalf("failed to start the new etcd instance: %v", err)
+	}
+	defer epc.CloseProc(context.TODO(), nil)
+
+	// verify all nodes have exact same revision and hash
+	t.Log("Verify all nodes have exact same revision and hash")
+	assert.Eventually(t, func() bool {
+		hashKvs, err := epc.Client().HashKV(context.TODO(), 0)
+		if err != nil {
+			t.Logf("failed to get HashKV: %v", err)
+			return false
+		}
+		if len(hashKvs) != 2 {
+			t.Logf("expected 2 hashkv responses, but got: %d", len(hashKvs))
+			return false
+		}
+
+		if hashKvs[0].Header.Revision != hashKvs[1].Header.Revision {
+			t.Logf("Got different revisions, [%d, %d]", hashKvs[0].Header.Revision, hashKvs[1].Header.Revision)
+			return false
+		}
+
+		assert.Equal(t, hashKvs[0].Hash, hashKvs[1].Hash)
+
+		return true
+	}, 10*time.Second, 500*time.Millisecond)
+}

tests/framework/e2e/cluster.go

@@ -604,11 +604,15 @@ func (epc *EtcdProcessCluster) Endpoints(f func(ep EtcdProcess) []string) (ret [
 
 func (epc *EtcdProcessCluster) CloseProc(ctx context.Context, finder func(EtcdProcess) bool, opts ...config.ClientOption) error {
 	procIndex := -1
-	for i := range epc.Procs {
-		if finder(epc.Procs[i]) {
-			procIndex = i
-			break
+	if finder != nil {
+		for i := range epc.Procs {
+			if finder(epc.Procs[i]) {
+				procIndex = i
+				break
+			}
 		}
+	} else {
+		procIndex = len(epc.Procs) - 1
 	}
 
 	if procIndex == -1 {
@@ -651,8 +655,14 @@ func (epc *EtcdProcessCluster) CloseProc(ctx context.Context, finder func(EtcdPr
 	return proc.Close()
 }
 
-func (epc *EtcdProcessCluster) StartNewProc(ctx context.Context, tb testing.TB, opts ...config.ClientOption) error {
-	serverCfg := epc.Cfg.EtcdServerProcessConfig(tb, epc.nextSeq)
+func (epc *EtcdProcessCluster) StartNewProc(ctx context.Context, cfg *EtcdProcessClusterConfig, tb testing.TB, opts ...config.ClientOption) error {
+	var serverCfg *EtcdServerProcessConfig
+	if cfg != nil {
+		serverCfg = cfg.EtcdServerProcessConfig(tb, epc.nextSeq)
+	} else {
+		serverCfg = epc.Cfg.EtcdServerProcessConfig(tb, epc.nextSeq)
+	}
+
 	epc.nextSeq++
 
 	initialCluster := []string{