cloud: add fault testing to S3 client

This adds a cloud nemesis test suite and uses it to ensure the S3 client
is tolerant of 30 second brown outs. This test is designed to be a
general purpose fault injection test and it is confirmed to reliably
reproduce the token bucket exhaustion errors that are caused by the S3
client.

Release note: none
Informs: #151748

Author: jeffswenson

pkg/cloud/amazon/BUILD.bazel

@@ -67,6 +67,7 @@ go_test(
         "//pkg/testutils",
         "//pkg/testutils/skip",
         "//pkg/util/leaktest",
+        "//pkg/util/log",
         "//pkg/util/uuid",
         "@com_github_aws_aws_sdk_go//aws/awserr",
         "@com_github_aws_aws_sdk_go//aws/request",

pkg/cloud/amazon/s3_storage.go

@@ -114,6 +114,7 @@ type s3Storage struct {
 	bucket         *string
 	conf           *cloudpb.ExternalStorage_S3
 	ioConf         base.ExternalIODirConfig
+	middleware     cloud.HttpMiddleware
 	settings       *cluster.Settings
 	prefix         string
 	metrics        *cloud.Metrics
@@ -514,6 +515,7 @@ func MakeS3Storage(
 		bucket:         aws.String(conf.Bucket),
 		conf:           conf,
 		ioConf:         args.IOConf,
+		middleware:     args.HttpMiddleware,
 		prefix:         conf.Prefix,
 		metrics:        args.MetricsRecorder,
 		settings:       args.Settings,
@@ -611,6 +613,7 @@ func (s *s3Storage) newClient(ctx context.Context) (s3Client, string, error) {
 			Client:             s.storageOptions.ClientName,
 			Cloud:              "aws",
 			InsecureSkipVerify: s.opts.skipTLSVerify,
+			HttpMiddleware:     s.middleware,
 		})
 	if err != nil {
 		return s3Client{}, "", err

pkg/cloud/amazon/s3_storage_test.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
@@ -30,32 +31,34 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
 func makeS3Storage(
-	ctx context.Context, uri string, user username.SQLUsername,
+	ctx context.Context, uri string, user username.SQLUsername, middleware cloud.HttpMiddleware,
 ) (cloud.ExternalStorage, error) {
 	conf, err := cloud.ExternalStorageConfFromURI(uri, user)
 	if err != nil {
 		return nil, err
 	}
 	testSettings := cluster.MakeTestingClusterSettings()
 
-	// Setup a sink for the given args.
-	s, err := cloud.MakeExternalStorage(ctx, conf, base.ExternalIODirConfig{}, testSettings,
-		blobs.TestEmptyBlobClientFactory,
-		nil, /* db */
-		nil, /* limiters */
-		cloud.NilMetrics,
-	)
-	if err != nil {
-		return nil, err
+	// TODO(jeffswenson): remove this once we change the default to false.
+	enableClientRetryTokenBucket.Override(ctx, &testSettings.SV, false)
+
+	args := cloud.EarlyBootExternalStorageContext{
+		IOConf:          base.ExternalIODirConfig{},
+		Settings:        testSettings,
+		Options:         nil,
+		Limiters:        nil,
+		MetricsRecorder: cloud.NilMetrics,
+		HttpMiddleware:  middleware,
 	}
 
-	return s, nil
+	return MakeS3Storage(ctx, args, conf)
 }
 
 // You can create an IAM that can access S3 in the AWS console, then
@@ -70,6 +73,7 @@ func skipIfNoDefaultConfig(t *testing.T, ctx context.Context) {
 	require.NoError(t, err)
 	_, err = cfg.Credentials.Retrieve(ctx)
 	if err != nil {
+		t.Logf("config: %+v", cfg)
 		skip.IgnoreLintf(t, "%s: %s", helpMsg, err)
 	}
 }
@@ -192,7 +196,7 @@ func TestPutS3(t *testing.T) {
 					cloud.AuthParam, cloud.AuthParamImplicit, AWSServerSideEncryptionMode,
 					"unsupported-algorithm")
 
-				_, err = makeS3Storage(ctx, invalidSSEModeURI, user)
+				_, err = makeS3Storage(ctx, invalidSSEModeURI, user, nil)
 				require.True(t, testutils.IsError(err, "unsupported server encryption mode unsupported-algorithm. Supported values are `aws:kms` and `AES256"))
 
 				// Specify aws:kms encryption mode but don't specify kms ID.
@@ -201,13 +205,44 @@ func TestPutS3(t *testing.T) {
 					bucket, "backup-test-sse-256",
 					cloud.AuthParam, cloud.AuthParamImplicit, AWSServerSideEncryptionMode,
 					"aws:kms")
-				_, err = makeS3Storage(ctx, invalidKMSURI, user)
+				_, err = makeS3Storage(ctx, invalidKMSURI, user, nil)
 				require.True(t, testutils.IsError(err, "AWS_SERVER_KMS_ID param must be set when using aws:kms server side encryption mode."))
 			})
 		})
 	}
 }
 
+func TestS3FaultInjection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	skipIfNoDefaultConfig(t, ctx)
+
+	baseBucket := os.Getenv("AWS_S3_BUCKET")
+	if baseBucket == "" {
+		skip.IgnoreLint(t, "AWS_S3_BUCKET env var must be set")
+	}
+
+	// Enable cloud transport logging.
+	defer log.Scope(t).Close(t)
+	prevVModule := log.GetVModule()
+	defer func() { _ = log.SetVModule(prevVModule) }()
+	require.NoError(t, log.SetVModule("cloud_logging_transport=1"))
+
+	uri := fmt.Sprintf(
+		"s3://%s/%d-fault-injection-test?AUTH=implicit",
+		baseBucket, cloudtestutils.NewTestID(),
+	)
+
+	// Inject faults for 15-45 seconds after the storage is opened.
+	middleware := cloudtestutils.BrownoutMiddleware(time.Second*15, time.Second*45)
+	storage, err := makeS3Storage(ctx, uri, username.RootUserName(), middleware)
+	require.NoError(t, err)
+	defer storage.Close()
+
+	cloudtestutils.RunCloudNemesisTest(t, storage)
+}
+
 func TestPutS3AssumeRole(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

pkg/cloud/azure/azure_storage.go

@@ -223,9 +223,10 @@ func makeAzureStorage(
 	options := args.ExternalStorageOptions()
 	t, err := cloud.MakeHTTPClient(args.Settings, args.MetricsRecorder,
 		cloud.HTTPClientConfig{
-			Bucket: dest.AzureConfig.Container,
-			Client: options.ClientName,
-			Cloud:  "azure",
+			Bucket:         dest.AzureConfig.Container,
+			Client:         options.ClientName,
+			Cloud:          "azure",
+			HttpMiddleware: args.HttpMiddleware,
 		})
 	if err != nil {
 		return nil, errors.Wrap(err, "azure: unable to create transport")

pkg/cloud/cloud_io.go

@@ -41,6 +41,8 @@ type HTTPClientConfig struct {
 	// accepts any certificate presented by the server and any host name in that
 	// certificate. In this mode, TLS is susceptible to machine-in-the-middle attacks.
 	InsecureSkipVerify bool
+
+	HttpMiddleware HttpMiddleware
 }
 
 // Timeout is a cluster setting used for cloud storage interactions.
@@ -115,7 +117,7 @@ func MakeHTTPClientForTransport(t http.RoundTripper) (*http.Client, error) {
 // Prefer MakeHTTPClient where possible.
 func MakeTransport(
 	settings *cluster.Settings, metrics *Metrics, config HTTPClientConfig,
-) (*http.Transport, error) {
+) (http.RoundTripper, error) {
 	var tlsConf *tls.Config
 	if config.InsecureSkipVerify {
 		tlsConf = &tls.Config{InsecureSkipVerify: true}
@@ -129,7 +131,6 @@ func MakeTransport(
 		}
 		tlsConf = &tls.Config{RootCAs: roots}
 	}
-
 	t := http.DefaultTransport.(*http.Transport).Clone()
 
 	// Add our custom CA.
@@ -140,7 +141,12 @@ func MakeTransport(
 	if metrics != nil {
 		t.DialContext = metrics.NetMetrics.Wrap(t.DialContext, config.Cloud, config.Bucket, config.Client)
 	}
-	return t, nil
+
+	var roundTripper http.RoundTripper = t
+	if config.HttpMiddleware != nil {
+		roundTripper = config.HttpMiddleware(roundTripper)
+	}
+	return roundTripper, nil
 }
 
 // MaxDelayedRetryAttempts is the number of times the delayedRetry method will

pkg/cloud/cloudtestutils/BUILD.bazel

@@ -2,7 +2,11 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "cloudtestutils",
-    srcs = ["cloud_test_helpers.go"],
+    srcs = [
+        "cloud_nemesis.go",
+        "cloud_test_helpers.go",
+        "http_middleware.go",
+    ],
     importpath = "github.com/cockroachdb/cockroach/pkg/cloud/cloudtestutils",
     visibility = ["//visibility:public"],
     deps = [
@@ -16,8 +20,11 @@ go_library(
         "//pkg/testutils",
         "//pkg/util/ioctx",
         "//pkg/util/randutil",
+        "//pkg/util/syncutil",
         "//pkg/util/sysutil",
+        "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
+        "@org_golang_x_sync//errgroup",
     ],
 )