roachprod/failure-injection: add artificial latency failure

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.

Author: DarrylWong

pkg/cmd/roachtest/roachtestutil/utils.go

@@ -13,6 +13,7 @@ import (
 	"net/http"
 	"os"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -237,3 +238,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/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,14 +36,18 @@ 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
 	}
+	// Make sure to cleanup the failure mode even if the test fails.
+	defer func() {
+		err = errors.CombineErrors(err, failureMode.Cleanup(ctx, l, t.args))
+	}()
 	if err = failureMode.Setup(ctx, l, t.args); err != nil {
 		return err
 	}
@@ -67,13 +72,7 @@ func (t *failureSmokeTest) run(
 		return err
 	}
 
-	if err = t.validateRestore(ctx, l, c); err != nil {
-		return err
-	}
-	if err = failureMode.Cleanup(ctx, l, t.args); err != nil {
-		return err
-	}
-	return nil
+	return t.validateRestore(ctx, l, c)
 }
 
 func (t *failureSmokeTest) noopRun(
@@ -233,6 +232,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 +320,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 +347,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 +358,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,

pkg/roachprod/failureinjection/failures/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "failures",
     srcs = [
         "failure.go",
+        "latency.go",
         "network_partition.go",
         "registry.go",
     ],
@@ -13,5 +14,6 @@ go_library(
         "//pkg/roachprod",
         "//pkg/roachprod/install",
         "//pkg/roachprod/logger",
+        "@com_github_cockroachdb_errors//:errors",
     ],
 )

pkg/roachprod/failureinjection/failures/failure.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/logger"
+	"github.com/cockroachdb/errors"
 )
 
 // FailureArgs describes the args passed to a failure mode.
@@ -35,7 +36,8 @@ type FailureMode interface {
 	// Inject a failure into the system.
 	Inject(ctx context.Context, l *logger.Logger, args FailureArgs) error
 
-	// Restore reverses the effects of Inject.
+	// Restore reverses the effects of Inject. The same args passed to Inject
+	// must be passed to Restore.
 	Restore(ctx context.Context, l *logger.Logger, args FailureArgs) error
 
 	// Cleanup uninstalls any dependencies that were installed by Setup.
@@ -52,9 +54,11 @@ type FailureMode interface {
 // provide commonly used functionality that doesn't differ between failure modes,
 // e.g. running remote commands on the cluster.
 type GenericFailure struct {
+	// TODO(Darryl): support specifying virtual clusters
 	c *install.SyncedCluster
 	// runTitle is the title to prefix command output with.
-	runTitle string
+	runTitle          string
+	networkInterfaces []string
 }
 
 func (f *GenericFailure) Run(
@@ -84,3 +88,22 @@ func (f *GenericFailure) RunWithDetails(
 	}
 	return res[0], nil
 }
+
+// NetworkInterfaces returns the network interfaces used by the VMs in the cluster.
+// Assumes that all VMs are using the same machine type and will have the same
+// network interfaces.
+func (f *GenericFailure) NetworkInterfaces(
+	ctx context.Context, l *logger.Logger,
+) ([]string, error) {
+	if f.networkInterfaces == nil {
+		res, err := f.c.RunWithDetails(ctx, l, install.WithNodes(f.c.Nodes[:1]), "Get Network Interfaces", "ip -o link show | awk -F ': ' '{print $2}'")
+		if err != nil {
+			return nil, errors.Wrapf(err, "error when determining network interfaces")
+		}
+		interfaces := strings.Split(strings.TrimSpace(res[0].Stdout), "\n")
+		for _, iface := range interfaces {
+			f.networkInterfaces = append(f.networkInterfaces, strings.TrimSpace(iface))
+		}
+	}
+	return f.networkInterfaces, nil
+}