tests: Validate etcd linearizability

Signed-off-by: Marek Siarkowicz <[email protected]>

go.mod

@@ -80,6 +80,7 @@ require (
 	github.com/stretchr/testify v1.7.2 // indirect
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20201229170055-e5319fda7802 // indirect
 	github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 // indirect
+	go.etcd.io/gofail v0.0.0-20220826035847-d0d2a96a6ef0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.32.0 // indirect
 	go.opentelemetry.io/otel v1.7.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.7.0 // indirect

go.sum

@@ -48,6 +48,7 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/anishathalye/porcupine v0.1.2/go.mod h1:/X9OQYnVb7DzfKCQVO4tI1Aq+o56UJW+RvN/5U4EuZA=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=

pkg/expect/expect.go

@@ -34,45 +34,65 @@ import (
 const DEBUG_LINES_TAIL = 40
 
 type ExpectProcess struct {
-	name string
+	cfg expectConfig
 
-	cmd  *exec.Cmd
-	fpty *os.File
-	wg   sync.WaitGroup
+	// StopSignal is the signal Stop sends to the process; defaults to SIGTERM.
+	StopSignal os.Signal
+
+	cmd           *exec.Cmd
+	fpty          *os.File
+	closech       chan struct{}
+	logsCollected sync.WaitGroup
 
 	mu    sync.Mutex // protects lines and err
 	lines []string
 	count int // increment whenever new line gets added
 	cur   int // current read position
 	err   error
+}
 
-	// StopSignal is the signal Stop sends to the process; defaults to SIGTERM.
-	StopSignal os.Signal
+type expectConfig struct {
+	name    string
+	cmd     string
+	args    []string
+	env     []string
+	restart bool
+}
+
+func commandFromConfig(config expectConfig) *exec.Cmd {
+	cmd := exec.Command(config.cmd, config.args...)
+	cmd.Env = config.env
+	cmd.Stderr = cmd.Stdout
+	cmd.Stdin = nil
+	return cmd
 }
 
 // NewExpect creates a new process for expect testing.
 func NewExpect(name string, arg ...string) (ep *ExpectProcess, err error) {
 	// if env[] is nil, use current system env and the default command as name
-	return NewExpectWithEnv(name, arg, nil, name)
+	return NewExpectWithEnv(name, arg, nil, name, false)
 }
 
 // NewExpectWithEnv creates a new process with user defined env variables for expect testing.
-func NewExpectWithEnv(name string, args []string, env []string, serverProcessConfigName string) (ep *ExpectProcess, err error) {
-	cmd := exec.Command(name, args...)
-	cmd.Env = env
+func NewExpectWithEnv(name string, args []string, env []string, serverProcessConfigName string, restart bool) (ep *ExpectProcess, err error) {
 	ep = &ExpectProcess{
-		name:       serverProcessConfigName,
-		cmd:        cmd,
+		closech:    make(chan struct{}),
 		StopSignal: syscall.SIGTERM,
+		cfg: expectConfig{
+			name:    serverProcessConfigName,
+			cmd:     name,
+			args:    args,
+			env:     env,
+			restart: restart,
+		},
 	}
-	ep.cmd.Stderr = ep.cmd.Stdout
-	ep.cmd.Stdin = nil
+	ep.cmd = commandFromConfig(ep.cfg)
 
 	if ep.fpty, err = pty.Start(ep.cmd); err != nil {
 		return nil, err
 	}
 
-	ep.wg.Add(1)
+	ep.logsCollected.Add(1)
 	go ep.read()
 	return ep, nil
 }
@@ -82,23 +102,52 @@ func (ep *ExpectProcess) Pid() int {
 }
 
 func (ep *ExpectProcess) read() {
-	defer ep.wg.Done()
+	defer ep.logsCollected.Done()
 	printDebugLines := os.Getenv("EXPECT_DEBUG") != ""
-	r := bufio.NewReader(ep.fpty)
 	for {
-		l, err := r.ReadString('\n')
 		ep.mu.Lock()
-		if l != "" {
-			if printDebugLines {
-				fmt.Printf("%s (%s) (%d): %s", ep.cmd.Path, ep.name, ep.cmd.Process.Pid, l)
+		cmd := ep.cmd
+		r := bufio.NewReader(ep.fpty)
+		ep.mu.Unlock()
+		if cmd == nil {
+			break
+		}
+		pid := cmd.Process.Pid
+		for {
+			l, err := r.ReadString('\n')
+			ep.mu.Lock()
+			if l != "" {
+				if printDebugLines {
+					fmt.Printf("%d: %s", pid, l)
+				}
+				ep.lines = append(ep.lines, l)
+				ep.count++
+			}
+			if err != nil {
+				ep.err = err
+				ep.mu.Unlock()
+				break
 			}
-			ep.lines = append(ep.lines, l)
-			ep.count++
+			ep.mu.Unlock()
+		}
+		select {
+		case <-ep.closech:
+			return
+		default:
 		}
-		if err != nil {
+		ep.mu.Lock()
+		cmd = ep.cmd
+		ep.mu.Unlock()
+		if cmd != nil {
+			cmd.Wait()
+		}
+		ep.mu.Lock()
+		var err error
+		ep.cmd = commandFromConfig(ep.cfg)
+		if ep.fpty, err = pty.Start(ep.cmd); err != nil {
+			fmt.Printf("Error %s\n", err)
 			ep.err = err
-			ep.mu.Unlock()
-			break
+			ep.cmd = nil
 		}
 		ep.mu.Unlock()
 	}
@@ -167,7 +216,10 @@ func (ep *ExpectProcess) Stop() error { return ep.close(true) }
 
 // Signal sends a signal to the expect process
 func (ep *ExpectProcess) Signal(sig os.Signal) error {
-	return ep.cmd.Process.Signal(sig)
+	ep.mu.Lock()
+	err := ep.cmd.Process.Signal(sig)
+	ep.mu.Unlock()
+	return err
 }
 
 // Close waits for the expect process to exit.
@@ -176,16 +228,22 @@ func (ep *ExpectProcess) Signal(sig os.Signal) error {
 func (ep *ExpectProcess) Close() error { return ep.close(false) }
 
 func (ep *ExpectProcess) close(kill bool) error {
-	if ep.cmd == nil {
+	ep.mu.Lock()
+	cmd := ep.cmd
+	ep.mu.Unlock()
+
+	if cmd == nil {
 		return ep.err
 	}
+	close(ep.closech)
+
 	if kill {
 		ep.Signal(ep.StopSignal)
 	}
 
-	err := ep.cmd.Wait()
+	err := cmd.Wait()
 	ep.fpty.Close()
-	ep.wg.Wait()
+	ep.logsCollected.Wait()
 
 	if err != nil {
 		if !kill && strings.Contains(err.Error(), "exit status") {
@@ -195,8 +253,9 @@ func (ep *ExpectProcess) close(kill bool) error {
 			err = nil
 		}
 	}
-
+	ep.mu.Lock()
 	ep.cmd = nil
+	ep.mu.Unlock()
 	return err
 }
 

tests/framework/config/cluster.go

@@ -31,4 +31,5 @@ type ClusterConfig struct {
 	PeerTLS           TLSConfig
 	ClientTLS         TLSConfig
 	QuotaBackendBytes int64
+	Restart           bool
 }

tests/framework/e2e.go

@@ -46,6 +46,7 @@ func (e e2eRunner) NewCluster(ctx context.Context, t testing.TB, cfg config.Clus
 		InitialToken:      "new",
 		ClusterSize:       cfg.ClusterSize,
 		QuotaBackendBytes: cfg.QuotaBackendBytes,
+		Restart:           cfg.Restart,
 	}
 	switch cfg.ClientTLS {
 	case config.NoTLS:
@@ -87,6 +88,10 @@ func (c *e2eCluster) Client() Client {
 	return e2eClient{e2e.NewEtcdctl(c.Cfg, c.EndpointsV3())}
 }
 
+func (c *e2eCluster) Endpoints() []string {
+	return c.EndpointsV3()
+}
+
 func (c *e2eCluster) Members() (ms []Member) {
 	for _, proc := range c.EtcdProcessCluster.Procs {
 		ms = append(ms, e2eMember{EtcdProcess: proc, Cfg: c.Cfg})

tests/framework/e2e/cluster.go

@@ -187,6 +187,8 @@ type EtcdProcessClusterConfig struct {
 	CorruptCheckTime        time.Duration
 	CompactHashCheckEnabled bool
 	CompactHashCheckTime    time.Duration
+	Restart                 bool
+	GoFailEnabled           bool
 }
 
 // NewEtcdProcessCluster launches a new cluster from etcd processes, returning
@@ -366,12 +368,16 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 		if cfg.CompactHashCheckTime != 0 {
 			args = append(args, "--experimental-compact-hash-check-time", cfg.CompactHashCheckTime.String())
 		}
+		envVars := map[string]string{}
+		if cfg.GoFailEnabled {
+			port = (i+1)*10000 + 2381
+			envVars["GOFAIL_HTTP"] = fmt.Sprintf("127.0.0.1:%d", port)
+		}
 
 		etcdCfgs[i] = &EtcdServerProcessConfig{
 			lg:           lg,
 			ExecPath:     cfg.ExecPath,
 			Args:         args,
-			EnvVars:      cfg.EnvVars,
 			TlsArgs:      cfg.TlsArgs(),
 			DataDirPath:  dataDirPath,
 			KeepDataDir:  cfg.KeepDataDir,
@@ -380,6 +386,8 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 			Acurl:        curl,
 			Murl:         murl,
 			InitialToken: cfg.InitialToken,
+			Restart:      cfg.Restart,
+			EnvVars:      envVars,
 		}
 	}
 

tests/framework/e2e/cluster_proxy.go

@@ -127,7 +127,7 @@ func (pp *proxyProc) start() error {
 	if pp.proc != nil {
 		panic("already started")
 	}
-	proc, err := SpawnCmdWithLogger(pp.lg, append([]string{pp.execPath}, pp.args...), nil, pp.name)
+	proc, err := SpawnCmdWithLogger(pp.lg, append([]string{pp.execPath}, pp.args...), nil, pp.name, false)
 	if err != nil {
 		return err
 	}

tests/framework/e2e/etcd_process.go

@@ -22,10 +22,9 @@ import (
 	"testing"
 	"time"
 
-	"go.uber.org/zap"
-
 	"go.etcd.io/etcd/client/pkg/v3/fileutil"
 	"go.etcd.io/etcd/pkg/v3/expect"
+	"go.uber.org/zap"
 )
 
 var (
@@ -81,6 +80,7 @@ type EtcdServerProcessConfig struct {
 
 	InitialToken   string
 	InitialCluster string
+	Restart        bool
 }
 
 func NewEtcdServerProcess(cfg *EtcdServerProcessConfig) (*EtcdServerProcess, error) {
@@ -104,7 +104,7 @@ func (ep *EtcdServerProcess) Start(ctx context.Context) error {
 		panic("already started")
 	}
 	ep.cfg.lg.Info("starting server...", zap.String("name", ep.cfg.Name))
-	proc, err := SpawnCmdWithLogger(ep.cfg.lg, append([]string{ep.cfg.ExecPath}, ep.cfg.Args...), ep.cfg.EnvVars, ep.cfg.Name)
+	proc, err := SpawnCmdWithLogger(ep.cfg.lg, append([]string{ep.cfg.ExecPath}, ep.cfg.Args...), ep.cfg.EnvVars, ep.cfg.Name, ep.cfg.Restart)
 	if err != nil {
 		return err
 	}

tests/framework/e2e/etcd_spawn.go

@@ -27,5 +27,5 @@ func SpawnCmd(args []string, envVars map[string]string) (*expect.ExpectProcess,
 }
 
 func SpawnNamedCmd(processName string, args []string, envVars map[string]string) (*expect.ExpectProcess, error) {
-	return SpawnCmdWithLogger(zap.NewNop(), args, envVars, processName)
+	return SpawnCmdWithLogger(zap.NewNop(), args, envVars, processName, false)
 }

tests/framework/e2e/etcd_spawn_cov.go

@@ -69,7 +69,7 @@ func SpawnCmdWithLogger(lg *zap.Logger, args []string, envVars map[string]string
 		zap.String("working-dir", wd),
 		zap.String("name", name),
 		zap.Strings("environment-variables", env))
-	ep, err := expect.NewExpectWithEnv(cmd, allArgs, env, name)
+	ep, err := expect.NewExpectWithEnv(cmd, allArgs, env, name, false)
 	if err != nil {
 		return nil, err
 	}

tests/framework/e2e/etcd_spawn_nocov.go

@@ -28,7 +28,7 @@ import (
 
 const noOutputLineCount = 0 // regular binaries emit no extra lines
 
-func SpawnCmdWithLogger(lg *zap.Logger, args []string, envVars map[string]string, name string) (*expect.ExpectProcess, error) {
+func SpawnCmdWithLogger(lg *zap.Logger, args []string, envVars map[string]string, name string, restart bool) (*expect.ExpectProcess, error) {
 	wd, err := os.Getwd()
 	if err != nil {
 		return nil, err
@@ -40,13 +40,17 @@ func SpawnCmdWithLogger(lg *zap.Logger, args []string, envVars map[string]string
 			zap.Strings("args", args),
 			zap.String("working-dir", wd),
 			zap.String("name", name),
-			zap.Strings("environment-variables", env))
-		return expect.NewExpectWithEnv(CtlBinPath, args[1:], env, name)
+			zap.Strings("environment-variables", env),
+			zap.Bool("restart", restart),
+		)
+		return expect.NewExpectWithEnv(CtlBinPath, args[1:], env, name, restart)
 	}
 	lg.Info("spawning process",
 		zap.Strings("args", args),
 		zap.String("working-dir", wd),
 		zap.String("name", name),
-		zap.Strings("environment-variables", env))
-	return expect.NewExpectWithEnv(args[0], args[1:], env, name)
+		zap.Strings("environment-variables", env),
+		zap.Bool("restart", restart),
+	)
+	return expect.NewExpectWithEnv(args[0], args[1:], env, name, restart)
 }

tests/framework/interface.go

@@ -33,6 +33,7 @@ type Cluster interface {
 	Client() Client
 	WaitLeader(t testing.TB) int
 	Close() error
+	Endpoints() []string
 }
 
 type Member interface {

tests/go.mod

@@ -15,6 +15,7 @@ replace (
 )
 
 require (
+	github.com/anishathalye/porcupine v0.1.2
 	github.com/coreos/go-semver v0.3.0
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gogo/protobuf v1.3.2

tests/go.sum

@@ -43,6 +43,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/anishathalye/porcupine v0.1.2 h1:eqWNeLcnTzXt6usipDJ4RFn6XOWqY5wEqBYVG3yFLSE=
+github.com/anishathalye/porcupine v0.1.2/go.mod h1:/X9OQYnVb7DzfKCQVO4tI1Aq+o56UJW+RvN/5U4EuZA=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=