Merge #148704

148704: failureinjection: add disk stall cycle option r=xinhaoz a=DarrylWong

Some usages of disk stalls wish to repeatedly stall and unstall a disk in a short period of time. The current implementation requires an ssh connection per stall/unstall call, which can easily lead to flakes. e.g. the unstall call runs into an ssh flake and the node fatals before it can be recovered.

This change adds a disk stall cycle option which will do this in one single ssh connection until Recover is called.

Fixes: #148345
Release note: none

Co-authored-by: DarrylWong <[email protected]>

Author: craig[bot]

pkg/cmd/roachtest/operations/disk_stall.go

@@ -24,7 +24,9 @@ type cleanupDiskStall struct {
 }
 
 func (cl *cleanupDiskStall) Cleanup(ctx context.Context, o operation.Operation, c cluster.Cluster) {
-	cl.staller.Unstall(ctx, cl.nodes)
+	if err := cl.staller.Unstall(ctx, cl.nodes); err != nil {
+		o.Fatalf("failed to unstall disk: %v", err)
+	}
 	o.Status("unstalled nodes; waiting 10 seconds before restarting")
 	time.Sleep(10 * time.Second)
 	// We might need to restart the node if it isn't live.

pkg/cmd/roachtest/roachtestutil/disk_stall.go

@@ -7,12 +7,14 @@ package roachtestutil
 
 import (
 	"context"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/spec"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/failureinjection/failures"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
+	"github.com/cockroachdb/errors"
 )
 
 // TODO(darryl): Once the failure injection library is a first class citizen of roachtest,
@@ -22,8 +24,9 @@ type DiskStaller interface {
 	Setup(ctx context.Context)
 	Cleanup(ctx context.Context)
 	Stall(ctx context.Context, nodes option.NodeListOption)
+	StallCycle(ctx context.Context, nodes option.NodeListOption, stallDuration, unstallDuration time.Duration)
 	Slow(ctx context.Context, nodes option.NodeListOption, bytesPerSecond int)
-	Unstall(ctx context.Context, nodes option.NodeListOption)
+	Unstall(ctx context.Context, nodes option.NodeListOption) error
 	DataDir() string
 	LogDir() string
 }
@@ -38,7 +41,11 @@ func (n NoopDiskStaller) LogDir() string
 func (n NoopDiskStaller) Setup(ctx context.Context)                              {}
 func (n NoopDiskStaller) Slow(_ context.Context, _ option.NodeListOption, _ int) {}
 func (n NoopDiskStaller) Stall(_ context.Context, _ option.NodeListOption)       {}
-func (n NoopDiskStaller) Unstall(_ context.Context, _ option.NodeListOption)     {}
+func (n NoopDiskStaller) StallCycle(
+	_ context.Context, _ option.NodeListOption, _, _ time.Duration,
+) {
+}
+func (n NoopDiskStaller) Unstall(_ context.Context, _ option.NodeListOption) error { return nil }
 
 type Fataler interface {
 	Fatal(args ...interface{})
@@ -103,6 +110,23 @@ func (s *cgroupDiskStaller) Stall(ctx context.Context, nodes option.NodeListOpti
 	}
 }
 
+func (s *cgroupDiskStaller) StallCycle(
+	ctx context.Context, nodes option.NodeListOption, stallDuration, unstallDuration time.Duration,
+) {
+	l := newDiskStallLogger(s.f.L(), nodes, "Stall")
+	if err := s.Failer.Inject(ctx, l, failures.DiskStallArgs{
+		StallLogs:            s.stallLogs,
+		StallWrites:          true,
+		StallReads:           s.stallReads,
+		Nodes:                nodes.InstallNodes(),
+		Cycle:                true,
+		CycleStallDuration:   stallDuration,
+		CycleUnstallDuration: unstallDuration,
+	}); err != nil {
+		s.f.Fatalf("failed to stall disk: %s", err)
+	}
+}
+
 func (s *cgroupDiskStaller) Slow(
 	ctx context.Context, nodes option.NodeListOption, bytesPerSecond int,
 ) {
@@ -118,11 +142,12 @@ func (s *cgroupDiskStaller) Slow(
 	}
 }
 
-func (s *cgroupDiskStaller) Unstall(ctx context.Context, nodes option.NodeListOption) {
+func (s *cgroupDiskStaller) Unstall(ctx context.Context, nodes option.NodeListOption) error {
 	l := newDiskStallLogger(s.f.L(), nodes, "Unstall")
-	if err := s.Failer.Recover(ctx, l); err != nil {
-		s.f.Fatalf("failed to unstall disk: %s", err)
-	}
+	// cgroup may fail when unstalling the disk, usually because the node already
+	// fataled and the cgroup is no longer available. Return the error and let
+	// the caller decide if a node fatal is expected or not.
+	return errors.Wrap(s.Failer.Recover(ctx, l), "failed to unstall disk")
 }
 
 type dmsetupDiskStaller struct {
@@ -168,18 +193,34 @@ func (s *dmsetupDiskStaller) Stall(ctx context.Context, nodes option.NodeListOpt
 	}
 }
 
+func (s *dmsetupDiskStaller) StallCycle(
+	ctx context.Context, nodes option.NodeListOption, stallDuration, unstallDuration time.Duration,
+) {
+	l := newDiskStallLogger(s.f.L(), nodes, "Stall")
+	if err := s.Failer.Inject(ctx, l, failures.DiskStallArgs{
+		Nodes:                nodes.InstallNodes(),
+		Cycle:                true,
+		CycleStallDuration:   stallDuration,
+		CycleUnstallDuration: unstallDuration,
+	}); err != nil {
+		s.f.Fatalf("failed to stall disk: %s", err)
+	}
+}
+
 func (s *dmsetupDiskStaller) Slow(
 	ctx context.Context, nodes option.NodeListOption, bytesPerSecond int,
 ) {
 	// TODO(baptist): Consider https://github.com/kawamuray/ddi.
 	s.f.Fatal("Slow is not supported for dmsetupDiskStaller")
 }
 
-func (s *dmsetupDiskStaller) Unstall(ctx context.Context, nodes option.NodeListOption) {
+func (s *dmsetupDiskStaller) Unstall(ctx context.Context, nodes option.NodeListOption) error {
 	l := newDiskStallLogger(s.f.L(), nodes, "Unstall")
+	// Any unstall error for dmsetup is unexpected and should fail the test.
 	if err := s.Failer.Recover(ctx, l); err != nil {
 		s.f.Fatalf("failed to unstall disk: %s", err)
 	}
+	return nil
 }
 
 func (s *dmsetupDiskStaller) DataDir() string { return "{store-dir}" }

pkg/cmd/roachtest/tests/disk_stall.go

@@ -129,7 +129,9 @@ func runDiskStalledWALFailover(ctx context.Context, t test.Test, c cluster.Clust
 					ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 					defer cancel()
 					t.Status("Unstalling disk on n1")
-					s.Unstall(ctx, c.Node(1))
+					if err = s.Unstall(ctx, c.Node(1)); err != nil {
+						t.Fatal(err)
+					}
 					t.Status("Unstalled disk on n1")
 				}()
 
@@ -378,7 +380,11 @@ func runDiskStalledDetection(
 	}
 
 	// Unstall the stalled node. It should be able to be reaped.
-	s.Unstall(ctx, c.Node(1))
+	// Note we only log errors since cgroup unstall is expected to fail due to
+	// nodes panicking from a detected disk stall.
+	if err = s.Unstall(ctx, c.Node(1)); err != nil {
+		t.L().Printf("failed to unstall disk: %v", err)
+	}
 	time.Sleep(1 * time.Second)
 	exit, ok = getProcessExitMonotonic(ctx, t, c, 1)
 	if doStall {
@@ -462,7 +468,11 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 	// Use CgroupDiskStaller with readsToo=false to only stall writes.
 	s := roachtestutil.MakeCgroupDiskStaller(t, c, false /* readsToo */, false /* logsToo */)
 	s.Setup(ctx)
-	defer s.Cleanup(ctx)
+	// NB: We use a background context in the defer'ed cleanup command,
+	// otherwise on test failure our c.Run calls will be ignored. Leaving
+	// the disk stalled will prevent artifact collection, making debugging
+	// difficult.
+	defer s.Cleanup(context.Background())
 
 	t.Status("starting cluster")
 	startOpts := option.DefaultStartOpts()
@@ -612,36 +622,17 @@ func runDiskStalledWALFailoverWithProgress(ctx context.Context, t test.Test, c c
 				case <-time.After(diskStallWaitDur):
 					t.Status("starting disk stall")
 				}
-				stallStart := timeutil.Now()
-				// Execute short 200ms stalls every 5s.
-				for timeutil.Since(stallStart) < operationDur {
-					select {
-					case <-ctx.Done():
-						t.Fatalf("context done while stall induced: %s", ctx.Err())
-					case <-time.After(stallInterval):
-						func() {
-							t.Status("short disk stall on n1")
-							s.Stall(ctx, c.Node(1))
-							defer func() {
-								// NB: We use a background context in the defer'ed unstall command,
-								// otherwise on test failure our Unstall calls will be ignored. Leaving
-								// the disk stalled will prevent artifact collection, making debugging
-								// difficult.
-								ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
-								defer cancel()
-								s.Unstall(ctx, c.Node(1))
-								t.Status("unstalled disk on n1")
-							}()
-							select {
-							case <-ctx.Done():
-								t.Fatalf("context done while stall induced: %s", ctx.Err())
-							case <-time.After(shortStallDur):
-								return
-							}
-						}()
+				// Execute short 200ms stalls every 5s for 3 minutes.
+				s.StallCycle(ctx, c.Node(1), shortStallDur, stallInterval)
+				select {
+				case <-ctx.Done():
+					t.Fatalf("context done while stall induced: %s", ctx.Err())
+				case <-time.After(operationDur):
+					if err = s.Unstall(ctx, c.Node(1)); err != nil {
+						t.Fatal(err)
 					}
+					t.Status("disk stalls stopped")
 				}
-
 				return nil
 			}, task.Name("disk-stall-phase"))
 		} else {

pkg/cmd/roachtest/tests/failover.go

@@ -1635,7 +1635,9 @@ func (f *diskStallFailer) Fail(ctx context.Context, nodeID int) {
 }
 
 func (f *diskStallFailer) Recover(ctx context.Context, nodeID int) {
-	f.staller.Unstall(ctx, f.c.Node(nodeID))
+	if err := f.staller.Unstall(ctx, f.c.Node(nodeID)); err != nil {
+		f.t.Fatalf("failed to unstall disk %v", err)
+	}
 	// Pebble's disk stall detector should have terminated the node, but in case
 	// it didn't, we explicitly stop it first.
 	f.c.Stop(ctx, f.t.L(), option.DefaultStopOpts(), f.c.Node(nodeID))

pkg/cmd/roachtest/tests/perturbation/slow_disk.go

@@ -81,7 +81,9 @@ func (s *slowDisk) startPerturbation(ctx context.Context, t test.Test, v variati
 
 // endPerturbation implements perturbation.
 func (s *slowDisk) endPerturbation(ctx context.Context, t test.Test, v variations) time.Duration {
-	s.staller.Unstall(ctx, v.targetNodes())
+	if err := s.staller.Unstall(ctx, v.targetNodes()); err != nil {
+		t.Fatal("failed to unstall disk:", err)
+	}
 	waitDuration(ctx, v.validationDuration)
 	return v.validationDuration
 }