Merge #142054

142054: roachprod/failureinjection: add artificial latency failure r=golgeek a=DarrylWong

This change adds a new artificial latency failure mode. This
failure mode uses TC to create traffic filters that can be used
to add latency between nodes.

Informs: #138970
Release note: none

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

Author: craig[bot]

pkg/cmd/roachtest/cluster.go

@@ -2507,7 +2507,7 @@ func (c *clusterImpl) RunE(ctx context.Context, options install.RunOptions, args
 		return errors.New("No command passed")
 	}
 	nodes := option.FromInstallNodes(options.Nodes)
-	l, logFile, err := c.loggerForCmd(nodes, args...)
+	l, logFile, err := roachtestutil.LoggerForCmd(c.l, nodes, args...)
 	if err != nil {
 		return err
 	}
@@ -2575,7 +2575,7 @@ func (c *clusterImpl) RunWithDetails(
 		return nil, errors.New("No command passed")
 	}
 	nodes := option.FromInstallNodes(options.Nodes)
-	l, logFile, err := c.loggerForCmd(nodes, args...)
+	l, logFile, err := roachtestutil.LoggerForCmd(c.l, nodes, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -2659,30 +2659,6 @@ func (c *clusterImpl) Install(
 	return errors.Wrap(roachprod.Install(ctx, l, c.MakeNodes(nodes), software), "cluster.Install")
 }
 
-// cmdLogFileName comes up with a log file to use for the given argument string.
-func cmdLogFileName(t time.Time, nodes option.NodeListOption, args ...string) string {
-	logFile := fmt.Sprintf(
-		"run_%s_n%s_%s",
-		t.Format(`150405.000000000`),
-		nodes.String()[1:],
-		install.GenFilenameFromArgs(20, args...),
-	)
-	return logFile
-}
-
-func (c *clusterImpl) loggerForCmd(
-	node option.NodeListOption, args ...string,
-) (*logger.Logger, string, error) {
-	logFile := cmdLogFileName(timeutil.Now(), node, args...)
-
-	// NB: we set no prefix because it's only going to a file anyway.
-	l, err := c.l.ChildLogger(logFile, logger.QuietStderr, logger.QuietStdout)
-	if err != nil {
-		return nil, "", err
-	}
-	return l, logFile, nil
-}
-
 // pgURLErr returns the Postgres endpoint for the specified nodes. It accepts a
 // flag specifying whether the URL should include the node's internal or
 // external IP address. In general, inter-cluster communication and should use

pkg/cmd/roachtest/cluster_test.go

@@ -11,7 +11,6 @@ import (
 	"strconv"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/task"
@@ -24,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/version"
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -686,21 +684,6 @@ func TestMachineTypes(t *testing.T) {
 	})
 }
 
-func TestCmdLogFileName(t *testing.T) {
-	ts := time.Date(2000, 1, 1, 15, 4, 12, 0, time.Local)
-
-	const exp = `run_150412.000000000_n1,3-4,9_cockroach-bla-foo-ba`
-	nodes := option.NodeListOption{1, 3, 4, 9}
-	assert.Equal(t,
-		exp,
-		cmdLogFileName(ts, nodes, "./cockroach", "bla", "--foo", "bar"),
-	)
-	assert.Equal(t,
-		exp,
-		cmdLogFileName(ts, nodes, "./cockroach bla --foo bar"),
-	)
-}
-
 func TestVerifyLibraries(t *testing.T) {
 	originalLibraryPaths := libraryFilePaths
 	defer func() { libraryFilePaths = originalLibraryPaths }()

pkg/cmd/roachtest/roachtestutil/BUILD.bazel

@@ -53,10 +53,12 @@ go_test(
     srcs = [
         "commandbuilder_test.go",
         "load_group_test.go",
+        "utils_test.go",
     ],
     embed = [":roachtestutil"],
     deps = [
         "//pkg/cmd/roachtest/option",
+        "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/cmd/roachtest/roachtestutil/utils.go

@@ -13,6 +13,7 @@ import (
 	"net/http"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -218,6 +219,31 @@ func IfLocal(c cluster.Cluster, trueVal, falseVal string) string {
 	return falseVal
 }
 
+// LoggerForCmd creates a logger to a file with quiet stdout and stderr.
+func LoggerForCmd(
+	l *logger.Logger, node option.NodeListOption, args ...string,
+) (*logger.Logger, string, error) {
+	logFile := cmdLogFileName(timeutil.Now(), node, args...)
+
+	// NB: we set no prefix because it's only going to a file anyway.
+	l, err := l.ChildLogger(logFile, logger.QuietStderr, logger.QuietStdout)
+	if err != nil {
+		return nil, "", err
+	}
+	return l, logFile, nil
+}
+
+// cmdLogFileName comes up with a log file to use for the given argument string.
+func cmdLogFileName(t time.Time, nodes option.NodeListOption, args ...string) string {
+	logFile := fmt.Sprintf(
+		"run_%s_n%s_%s",
+		t.Format(`150405.000000000`),
+		nodes.String()[1:],
+		install.GenFilenameFromArgs(20, args...),
+	)
+	return logFile
+}
+
 // CheckPortBlocked returns true if a connection from a node to a port on another node
 // can be established. Requires nmap to be installed.
 func CheckPortBlocked(
@@ -237,3 +263,19 @@ func CheckPortBlocked(
 	}
 	return strings.Contains(res.Stdout, "filtered"), nil
 }
+
+// PortLatency returns the latency from one node to another port.
+// Requires nmap to be installed.
+func PortLatency(
+	ctx context.Context, l *logger.Logger, c cluster.Cluster, fromNode, toNode option.NodeListOption,
+) (time.Duration, error) {
+	res, err := c.RunWithDetailsSingleNode(ctx, l, option.WithNodes(fromNode), fmt.Sprintf("nmap -p {pgport%[1]s} -Pn {ip%[1]s} -oG - | grep 'scanned in' | awk '{print $(NF-1)}'", toNode))
+	if err != nil {
+		return 0, err
+	}
+	avgRTT, err := strconv.ParseFloat(strings.TrimSpace(res.Stdout), 64)
+	if err != nil {
+		return 0, err
+	}
+	return time.Duration(avgRTT * float64(time.Second)), nil
+}

pkg/cmd/roachtest/roachtestutil/utils_test.go

@@ -0,0 +1,29 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package roachtestutil
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCmdLogFileName(t *testing.T) {
+	ts := time.Date(2000, 1, 1, 15, 4, 12, 0, time.Local)
+
+	const exp = `run_150412.000000000_n1,3-4,9_cockroach-bla-foo-ba`
+	nodes := option.NodeListOption{1, 3, 4, 9}
+	assert.Equal(t,
+		exp,
+		cmdLogFileName(ts, nodes, "./cockroach", "bla", "--foo", "bar"),
+	)
+	assert.Equal(t,
+		exp,
+		cmdLogFileName(ts, nodes, "./cockroach bla --foo bar"),
+	)
+}

pkg/cmd/roachtest/tests/failure_injection.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
@@ -35,45 +36,79 @@ type failureSmokeTest struct {
 
 func (t *failureSmokeTest) run(
 	ctx context.Context, l *logger.Logger, c cluster.Cluster, fr *failures.FailureRegistry,
-) error {
+) (err error) {
 	// TODO(darryl): In the future, roachtests should interact with the failure injection library
 	// through helper functions in roachtestutil so they don't have to interface with roachprod
 	// directly.
-	failureMode, err := fr.GetFailureMode(c.MakeNodes(), t.failureName, l, c.IsSecure())
+	failureMode, err := fr.GetFailureMode(c.MakeNodes(c.CRDBNodes()), t.failureName, l, c.IsSecure())
 	if err != nil {
 		return err
 	}
-	if err = failureMode.Setup(ctx, l, t.args); err != nil {
+	// Make sure to cleanup the failure mode even if the test fails.
+	defer func() {
+		quietLogger, file, logErr := roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "cleanup")
+		if logErr != nil {
+			l.Printf("failed to create logger for cleanup: %v", logErr)
+			quietLogger = l
+		}
+		l.Printf("%s: Running Cleanup(); details in %s.log", t.failureName, file)
+		err = errors.CombineErrors(err, failureMode.Cleanup(ctx, quietLogger, t.args))
+	}()
+
+	quietLogger, file, err := roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "setup")
+	if err != nil {
+		return err
+	}
+	l.Printf("%s: Running Setup(); details in %s.log", t.failureName, file)
+	if err = failureMode.Setup(ctx, quietLogger, t.args); err != nil {
+		return err
+	}
+
+	quietLogger, file, err = roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "inject")
+	if err != nil {
 		return err
 	}
-	if err = failureMode.Inject(ctx, l, t.args); err != nil {
+	l.Printf("%s: Running Inject(); details in %s.log", t.failureName, file)
+	if err = failureMode.Inject(ctx, quietLogger, t.args); err != nil {
 		return err
 	}
 
 	// Allow the failure to take effect.
-	if err = failureMode.WaitForFailureToPropagate(ctx, l, t.args); err != nil {
+	quietLogger, file, err = roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "wait for propagate")
+	if err != nil {
+		return err
+	}
+	l.Printf("%s: Running WaitForFailureToPropagate(); details in %s.log", t.failureName, file)
+	if err = failureMode.WaitForFailureToPropagate(ctx, quietLogger, t.args); err != nil {
 		return err
 	}
 
+	l.Printf("validating failure was properly injected")
 	if err = t.validateFailure(ctx, l, c); err != nil {
 		return err
 	}
-	if err = failureMode.Restore(ctx, l, t.args); err != nil {
+
+	quietLogger, file, err = roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "restore")
+	if err != nil {
 		return err
 	}
-
-	// Allow the cluster to return to normal.
-	if err = failureMode.WaitForFailureToRestore(ctx, l, t.args); err != nil {
+	l.Printf("%s: Running Restore(); details in %s.log", t.failureName, file)
+	if err = failureMode.Restore(ctx, quietLogger, t.args); err != nil {
 		return err
 	}
 
-	if err = t.validateRestore(ctx, l, c); err != nil {
+	// Allow the cluster to return to normal.
+	quietLogger, file, err = roachtestutil.LoggerForCmd(l, c.CRDBNodes(), t.testName, "wait for restore")
+	if err != nil {
 		return err
 	}
-	if err = failureMode.Cleanup(ctx, l, t.args); err != nil {
+	l.Printf("%s: Running WaitForFailureToRestore(); details in %s.log", t.failureName, file)
+	if err = failureMode.WaitForFailureToRestore(ctx, quietLogger, t.args); err != nil {
 		return err
 	}
-	return nil
+
+	l.Printf("validating failure was properly restored")
+	return t.validateRestore(ctx, l, c)
 }
 
 func (t *failureSmokeTest) noopRun(
@@ -233,6 +268,69 @@ var asymmetricOutgoingNetworkPartitionTest = func(c cluster.Cluster) failureSmok
 	}
 }
 
+var latencyTest = func(c cluster.Cluster) failureSmokeTest {
+	nodes := c.CRDBNodes()
+	rand.Shuffle(len(nodes), func(i, j int) {
+		nodes[i], nodes[j] = nodes[j], nodes[i]
+	})
+	srcNode := nodes[0]
+	destNode := nodes[1]
+	unaffectedNode := nodes[2]
+	return failureSmokeTest{
+		testName:    "Network Latency",
+		failureName: failures.NetworkLatencyName,
+		args: failures.NetworkLatencyArgs{
+			ArtificialLatencies: []failures.ArtificialLatency{
+				{
+					Source:      install.Nodes{install.Node(srcNode)},
+					Destination: install.Nodes{install.Node(destNode)},
+					Delay:       2 * time.Second,
+				},
+				{
+					Source:      install.Nodes{install.Node(destNode)},
+					Destination: install.Nodes{install.Node(srcNode)},
+					Delay:       2 * time.Second,
+				},
+			},
+		},
+		validateFailure: func(ctx context.Context, l *logger.Logger, c cluster.Cluster) error {
+			// Note that this is one way latency, since the sender doesn't have the matching port.
+			delayedLatency, err := roachtestutil.PortLatency(ctx, l, c, c.Nodes(srcNode), c.Nodes(destNode))
+			if err != nil {
+				return err
+			}
+			normalLatency, err := roachtestutil.PortLatency(ctx, l, c, c.Nodes(unaffectedNode), c.Nodes(destNode))
+			if err != nil {
+				return err
+			}
+			if delayedLatency < normalLatency*2 {
+				return errors.Errorf("expected latency between nodes with artificial latency (n%d and n%d) to be much higher than between nodes without (n%d and n%d)", srcNode, destNode, unaffectedNode, destNode)
+			}
+			if delayedLatency < time.Second || delayedLatency > 3*time.Second {
+				return errors.Errorf("expected latency between nodes with artificial latency (n%d and n%d) to be at least within 1s and 3s", srcNode, destNode)
+			}
+			return nil
+		},
+		validateRestore: func(ctx context.Context, l *logger.Logger, c cluster.Cluster) error {
+			delayedLatency, err := roachtestutil.PortLatency(ctx, l, c, c.Nodes(srcNode), c.Nodes(destNode))
+			if err != nil {
+				return err
+			}
+			normalLatency, err := roachtestutil.PortLatency(ctx, l, c, c.Nodes(unaffectedNode), c.Nodes(destNode))
+			if err != nil {
+				return err
+			}
+			if delayedLatency > 2*normalLatency {
+				return errors.Errorf("expected latency between nodes with artificial latency (n%d and n%d) to be close to latency between nodes without (n%d and n%d)", srcNode, destNode, unaffectedNode, destNode)
+			}
+			if delayedLatency > 500*time.Millisecond {
+				return errors.Errorf("expected latency between nodes with artificial latency (n%d and n%d) to have restored to at least less than 500ms", srcNode, destNode)
+			}
+			return nil
+		},
+	}
+}
+
 func setupFailureSmokeTests(ctx context.Context, t test.Test, c cluster.Cluster) error {
 	// Download any dependencies needed.
 	if err := c.Install(ctx, t.L(), c.CRDBNodes(), "nmap"); err != nil {
@@ -258,6 +356,7 @@ func runFailureSmokeTest(ctx context.Context, t test.Test, c cluster.Cluster, no
 		bidirectionalNetworkPartitionTest(c),
 		asymmetricIncomingNetworkPartitionTest(c),
 		asymmetricOutgoingNetworkPartitionTest(c),
+		latencyTest(c),
 	}
 
 	// Randomize the order of the tests in case any of the failures have unexpected side
@@ -284,7 +383,7 @@ func runFailureSmokeTest(ctx context.Context, t test.Test, c cluster.Cluster, no
 
 func registerFISmokeTest(r registry.Registry) {
 	r.Add(registry.TestSpec{
-		Name:             "failure-injection-smoke-test",
+		Name:             "failure-injection/smoke-test",
 		Owner:            registry.OwnerTestEng,
 		Cluster:          r.MakeClusterSpec(4, spec.WorkloadNode(), spec.CPU(2), spec.WorkloadNodeCPU(2), spec.ReuseNone()),
 		CompatibleClouds: registry.OnlyGCE,
@@ -295,7 +394,7 @@ func registerFISmokeTest(r registry.Registry) {
 		},
 	})
 	r.Add(registry.TestSpec{
-		Name:             "failure-injection-noop-smoke-test",
+		Name:             "failure-injection/smoke-test/noop",
 		Owner:            registry.OwnerTestEng,
 		Cluster:          r.MakeClusterSpec(4, spec.WorkloadNode(), spec.CPU(2), spec.WorkloadNodeCPU(2), spec.ReuseNone()),
 		CompatibleClouds: registry.OnlyGCE,