Backport 16822 changes and dependencies

tests/framework/config/client.go

@@ -0,0 +1,41 @@
+// 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 config
+
+import (
+	"time"
+
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+// ClientOption configures the client with additional parameter.
+// For example, if Auth is enabled, the common test cases just need to
+// use `WithAuth` to return a ClientOption. Note that the common `WithAuth`
+// function calls `e2e.WithAuth` or `integration.WithAuth`, depending on the
+// build tag (either "e2e" or "integration").
+type ClientOption func(any)
+
+type GetOptions struct {
+	Revision     int
+	End          string
+	CountOnly    bool
+	Serializable bool
+	Prefix       bool
+	FromKey      bool
+	Limit        int
+	Order        clientv3.SortOrder
+	SortBy       clientv3.SortTarget
+	Timeout      time.Duration
+}

tests/framework/config/cluster.go

@@ -0,0 +1,23 @@
+// 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 config
+
+import (
+	"time"
+)
+
+const (
+	TickDuration = 10 * time.Millisecond
+)

tests/framework/e2e/cluster.go

@@ -15,6 +15,7 @@
 package e2e
 
 import (
+	"context"
 	"fmt"
 	"net/url"
 	"os"
@@ -25,6 +26,7 @@ import (
 
 	"go.etcd.io/etcd/pkg/v3/proxy"
 	"go.etcd.io/etcd/server/v3/etcdserver"
+	"go.etcd.io/etcd/tests/v3/framework/config"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 )
@@ -39,6 +41,13 @@ const (
 	ClientTLSAndNonTLS
 )
 
+type ClientConfig struct {
+	ConnectionType ClientConnType
+	CertAuthority  bool
+	AutoTLS        bool
+	RevokeCerts    bool
+}
+
 func NewConfigNoTLS() *EtcdProcessClusterConfig {
 	return &EtcdProcessClusterConfig{ClusterSize: 3,
 		InitialToken: "new",
@@ -150,6 +159,7 @@ type EtcdProcessClusterConfig struct {
 	BasePort   int
 
 	MetricsURLScheme string
+	Client           ClientConfig
 
 	SnapshotCount int // default is 10000
 
@@ -576,3 +586,80 @@ func (epc *EtcdProcessCluster) WithStopSignal(sig os.Signal) (ret os.Signal) {
 	}
 	return ret
 }
+
+func (epc *EtcdProcessCluster) Etcdctl(opts ...config.ClientOption) *EtcdctlV3 {
+	etcdctl, err := NewEtcdctl(epc.Cfg.Client, epc.EndpointsGRPC(), opts...)
+	if err != nil {
+		panic(err)
+	}
+	return etcdctl
+}
+
+func (epc *EtcdProcessCluster) EndpointsGRPC() []string {
+	return epc.Endpoints(func(ep EtcdProcess) []string { return ep.EndpointsGRPC() })
+}
+
+func (epc *EtcdProcessCluster) WaitLeader(t testing.TB) int {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	return epc.WaitMembersForLeader(ctx, t, epc.Procs)
+}
+
+// WaitMembersForLeader waits until given members agree on the same leader,
+// and returns its 'index' in the 'membs' list
+func (epc *EtcdProcessCluster) WaitMembersForLeader(ctx context.Context, t testing.TB, membs []EtcdProcess) int {
+	cc := epc.Etcdctl()
+
+	// ensure leader is up via linearizable get
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal("WaitMembersForLeader timeout")
+		default:
+		}
+		_, err := cc.Get(ctx, "0", config.GetOptions{Timeout: 10*config.TickDuration + time.Second})
+		if err == nil || strings.Contains(err.Error(), "Key not found") {
+			break
+		}
+		t.Logf("WaitMembersForLeader Get err: %v", err)
+	}
+
+	leaders := make(map[uint64]struct{})
+	members := make(map[uint64]int)
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatal("WaitMembersForLeader timeout")
+		default:
+		}
+		for i := range membs {
+			resp, err := membs[i].Etcdctl().Status(ctx)
+			if err != nil {
+				if strings.Contains(err.Error(), "connection refused") {
+					// if member[i] has stopped
+					continue
+				} else {
+					t.Fatal(err)
+				}
+			}
+			members[resp[0].Header.MemberId] = i
+			leaders[resp[0].Leader] = struct{}{}
+		}
+		// members agree on the same leader
+		if len(leaders) == 1 {
+			break
+		}
+		leaders = make(map[uint64]struct{})
+		members = make(map[uint64]int)
+		time.Sleep(10 * config.TickDuration)
+	}
+	for l := range leaders {
+		if index, ok := members[l]; ok {
+			t.Logf("members agree on a leader, members:%v , leader:%v", members, l)
+			return index
+		}
+		t.Fatalf("members agree on a leader which is not one of members, members:%v , leader:%v", members, l)
+	}
+	t.Fatal("impossible path of execution")
+	return -1
+}

tests/framework/e2e/etcd_process.go

@@ -29,6 +29,7 @@ import (
 	"go.etcd.io/etcd/client/pkg/v3/fileutil"
 	"go.etcd.io/etcd/pkg/v3/expect"
 	"go.etcd.io/etcd/pkg/v3/proxy"
+	"go.etcd.io/etcd/tests/v3/framework/config"
 	"go.uber.org/zap"
 )
 
@@ -46,6 +47,7 @@ type EtcdProcess interface {
 	EndpointsGRPC() []string
 	EndpointsHTTP() []string
 	EndpointsMetrics() []string
+	Etcdctl(opts ...config.ClientOption) *EtcdctlV3
 
 	Start() error
 	Restart() error
@@ -80,6 +82,7 @@ type EtcdServerProcessConfig struct {
 	Args     []string
 	TlsArgs  []string
 	EnvVars  map[string]string
+	Client   ClientConfig
 
 	DataDirPath string
 	KeepDataDir bool
@@ -125,6 +128,14 @@ func (ep *EtcdServerProcess) EndpointsHTTP() []string {
 }
 func (ep *EtcdServerProcess) EndpointsMetrics() []string { return []string{ep.cfg.Murl} }
 
+func (ep *EtcdServerProcess) Etcdctl(opts ...config.ClientOption) *EtcdctlV3 {
+	etcdctl, err := NewEtcdctl(ep.Config().Client, ep.EndpointsGRPC(), opts...)
+	if err != nil {
+		panic(err)
+	}
+	return etcdctl
+}
+
 func (ep *EtcdServerProcess) Start() error {
 	if ep.proc != nil {
 		panic("already started")