test: Implement automatic restart

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

pkg/expect/expect.go

@@ -36,9 +36,13 @@ const DEBUG_LINES_TAIL = 40
 type ExpectProcess struct {
 	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{}
+	wg      sync.WaitGroup
 
 	mu    sync.Mutex // protects lines and err
 	lines []string
@@ -50,18 +54,20 @@ type ExpectProcess struct {
 // 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) {
+func NewExpectWithEnv(name string, args []string, env []string, serverProcessConfigName string, restart bool) (ep *ExpectProcess, err error) {
 	ep = &ExpectProcess{
 		cfg: expectConfig{
-			name: serverProcessConfigName,
-			cmd:  name,
-			args: args,
-			env:  env,
+			name:    serverProcessConfigName,
+			cmd:     name,
+			args:    args,
+			env:     env,
+			restart: restart,
 		},
+		closech: make(chan struct{}),
 	}
 	ep.cmd = commandFromConfig(ep.cfg)
 
@@ -75,10 +81,11 @@ func NewExpectWithEnv(name string, args []string, env []string, serverProcessCon
 }
 
 type expectConfig struct {
-	name string
-	cmd  string
-	args []string
-	env  []string
+	name    string
+	cmd     string
+	args    []string
+	env     []string
+	restart bool
 }
 
 func commandFromConfig(config expectConfig) *exec.Cmd {
@@ -96,21 +103,53 @@ func (ep *ExpectProcess) Pid() int {
 func (ep *ExpectProcess) read() {
 	defer ep.wg.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.cfg.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("%s (%s) (%d): %s", ep.cmd.Path, ep.cfg.name, pid, l)
+				}
+				ep.lines = append(ep.lines, l)
+				ep.count++
 			}
-			ep.lines = append(ep.lines, l)
-			ep.count++
+			if err != nil {
+				ep.err = err
+				ep.mu.Unlock()
+				if ep.cfg.restart {
+					break
+				}
+				return
+			}
+			ep.mu.Unlock()
 		}
-		if err != nil {
+		select {
+		case <-ep.closech:
+			return
+		default:
+		}
+		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()
 	}
@@ -179,7 +218,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.
@@ -188,14 +230,20 @@ 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(syscall.SIGTERM)
 	}
 
-	err := ep.cmd.Wait()
+	err := cmd.Wait()
 	ep.fpty.Close()
 	ep.wg.Wait()
 
@@ -207,8 +255,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

@@ -34,4 +34,5 @@ type ClusterConfig struct {
 	DisableStrictReconfigCheck bool
 	AuthToken                  string
 	SnapshotCount              int
+	Restart                    bool
 }

tests/framework/e2e.go

@@ -52,6 +52,7 @@ func (e e2eRunner) NewCluster(ctx context.Context, t testing.TB, cfg config.Clus
 		DisableStrictReconfigCheck: cfg.DisableStrictReconfigCheck,
 		AuthTokenOpts:              cfg.AuthToken,
 		SnapshotCount:              cfg.SnapshotCount,
+		Restart:                    cfg.Restart,
 	}
 	switch cfg.ClientTLS {
 	case config.NoTLS:

tests/framework/e2e/cluster.go

@@ -178,6 +178,7 @@ type EtcdProcessClusterConfig struct {
 	CorruptCheckTime        time.Duration
 	CompactHashCheckEnabled bool
 	CompactHashCheckTime    time.Duration
+	Restart                 bool
 }
 
 // NewEtcdProcessCluster launches a new cluster from etcd processes, returning
@@ -371,6 +372,7 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfigs(tb testing.TB) []*
 			Acurl:        curl,
 			Murl:         murl,
 			InitialToken: cfg.InitialToken,
+			Restart:      cfg.Restart,
 		}
 	}
 

tests/framework/e2e/cluster_proxy.go

@@ -121,7 +121,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

@@ -80,6 +80,7 @@ type EtcdServerProcessConfig struct {
 
 	InitialToken   string
 	InitialCluster string
+	Restart        bool
 }
 
 func NewEtcdServerProcess(cfg *EtcdServerProcessConfig) (*EtcdServerProcess, error) {
@@ -104,7 +105,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

@@ -65,7 +65,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))
-	return expect.NewExpectWithEnv(cmd, allArgs, env, name)
+	return expect.NewExpectWithEnv(cmd, allArgs, env, name, false)
 }
 
 func getCovArgs() ([]string, error) {

tests/framework/e2e/etcd_spawn_nocov.go

@@ -18,15 +18,16 @@
 package e2e
 
 import (
-	"go.uber.org/zap"
 	"os"
 
+	"go.uber.org/zap"
+
 	"go.etcd.io/etcd/pkg/v3/expect"
 )
 
 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
@@ -36,6 +37,8 @@ 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(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)
 }