tests: Allow failpoint requests to fail assuming that process exists within 1 second

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>

tests/framework/e2e/cluster_proxy.go

@@ -116,8 +116,12 @@ func (p *proxyEtcdProcess) Kill() error {
 	return p.etcdProc.Kill()
 }
 
-func (p *proxyEtcdProcess) Wait() error {
+func (p *proxyEtcdProcess) IsRunning() bool {
-	return p.etcdProc.Wait()
+	return p.etcdProc.IsRunning()
+}
+
+func (p *proxyEtcdProcess) Wait(ctx context.Context) error {
+	return p.etcdProc.Wait(ctx)
 }
 
 type proxyProc struct {

tests/framework/e2e/etcd_process.go

@@ -42,7 +42,8 @@ type EtcdProcess interface {
 	EndpointsMetrics() []string
 	Client(opts ...config.ClientOption) *EtcdctlV3
 
-	Wait() error
+	IsRunning() bool
+	Wait(ctx context.Context) error
 	Start(ctx context.Context) error
 	Restart(ctx context.Context) error
 	Stop() error
@@ -201,11 +202,35 @@ func (ep *EtcdServerProcess) Kill() error {
 	return ep.proc.Signal(syscall.SIGKILL)
 }
 
-func (ep *EtcdServerProcess) Wait() error {
+func (ep *EtcdServerProcess) Wait(ctx context.Context) error {
-	ep.proc.Wait()
+	ch := make(chan struct{})
+	go func() {
+		defer close(ch)
+		if ep.proc != nil {
+			ep.proc.Wait()
+			ep.cfg.lg.Info("server exited", zap.String("name", ep.cfg.Name))
+		}
+	}()
+	select {
+	case <-ch:
+		ep.proc = nil
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (ep *EtcdServerProcess) IsRunning() bool {
+	if ep.proc == nil {
+		return false
+	}
+	_, err := ep.proc.ExitCode()
+	if err == expect.ErrProcessRunning {
+		return true
+	}
 	ep.cfg.lg.Info("server exited", zap.String("name", ep.cfg.Name))
 	ep.proc = nil
-	return nil
+	return false
 }
 
 func AssertProcessLogs(t *testing.T, ep EtcdProcess, expectLog string) {

tests/linearizability/failpoints.go

@@ -31,6 +31,10 @@ import (
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 )
 
+const (
+	triggerTimeout = time.Second
+)
+
 var (
 	KillFailpoint                            Failpoint = killFailpoint{}
 	DefragBeforeCopyPanic                    Failpoint = goFailpoint{"backend/defragBeforeCopy", "panic", triggerDefrag, AnyMember}
@@ -81,15 +85,21 @@ type killFailpoint struct{}
 
 func (f killFailpoint) Trigger(t *testing.T, ctx context.Context, clus *e2e.EtcdProcessCluster) error {
 	member := clus.Procs[rand.Int()%len(clus.Procs)]
-	err := member.Kill()
+
-	if err != nil {
+	killCtx, cancel := context.WithTimeout(ctx, triggerTimeout)
-		return err
+	defer cancel()
+	for member.IsRunning() {
+		err := member.Kill()
+		if err != nil {
+			t.Logf("sending kill signal failed: %v", err)
+		}
+		err = member.Wait(killCtx)
+		if err != nil && !strings.Contains(err.Error(), "unexpected exit code") {
+			return fmt.Errorf("failed to kill the process within %s, err: %w", triggerTimeout, err)
+		}
 	}
-	err = member.Wait()
+
-	if err != nil && !strings.Contains(err.Error(), "unexpected exit code") {
+	err := member.Start(ctx)
-		return err
-	}
-	err = member.Start(ctx)
 	if err != nil {
 		return err
 	}
@@ -118,21 +128,27 @@ func (f goFailpoint) Trigger(t *testing.T, ctx context.Context, clus *e2e.EtcdPr
 	member := f.pickMember(t, clus)
 	address := fmt.Sprintf("127.0.0.1:%d", member.Config().GoFailPort)
 
-	err := setupGoFailpoint(address, f.failpoint, f.payload)
+	triggerCtx, cancel := context.WithTimeout(ctx, triggerTimeout)
-	if err != nil {
+	defer cancel()
-		return fmt.Errorf("gofailpoint setup failed: %w", err)
+
-	}
+	for member.IsRunning() {
-	if f.trigger != nil {
+		err := setupGoFailpoint(triggerCtx, address, f.failpoint, f.payload)
-		err = f.trigger(ctx, member)
 		if err != nil {
-			return fmt.Errorf("triggering gofailpoint failed: %w", err)
+			t.Logf("gofailpoint setup failed: %v", err)
+		}
+		if f.trigger != nil {
+			err = f.trigger(triggerCtx, member)
+			if err != nil {
+				t.Logf("triggering gofailpoint failed: %v", err)
+			}
+		}
+		err = member.Wait(triggerCtx)
+		if err != nil && !strings.Contains(err.Error(), "unexpected exit code") {
+			return fmt.Errorf("failed to trigger a process panic within %s, err: %w", triggerTimeout, err)
 		}
 	}
-	err = member.Wait()
+
-	if err != nil && !strings.Contains(err.Error(), "unexpected exit code") {
+	err := member.Start(ctx)
-		return err
-	}
-	err = member.Start(ctx)
 	if err != nil {
 		return err
 	}
@@ -150,13 +166,13 @@ func (f goFailpoint) pickMember(t *testing.T, clus *e2e.EtcdProcessCluster) e2e.
 	}
 }
 
-func setupGoFailpoint(host, failpoint, payload string) error {
+func setupGoFailpoint(ctx context.Context, host, failpoint, payload string) error {
 	failpointUrl := url.URL{
 		Scheme: "http",
 		Host:   host,
 		Path:   failpoint,
 	}
-	r, err := http.NewRequest("PUT", failpointUrl.String(), bytes.NewBuffer([]byte(payload)))
+	r, err := http.NewRequestWithContext(ctx, "PUT", failpointUrl.String(), bytes.NewBuffer([]byte(payload)))
 	if err != nil {
 		return err
 	}