cloud/amazon: add option to skip checksums

Our files (SSTs) often have internal checksums anyway, and we have ecplicit checksum
files for those that don't that are cloud agnostic, so we do not generally need the
extra checksums in s3-specific metadata. Computing them has non-zero cost as well, so
users may wish to disable them. Indeed, we almost did so by default a couple years ago
before realizing that they are required by the S3 API if certain other features, like
object locking, are enabled. So while some users may not be able to or want to disable
them, e.g. if they're using object locking, we can provide the option to do so
to those who would want it. This may also be useful to users of certain s3-like
storage services that have different compatibility with different checksums.

Release note: none.
Epic: none.

Author: dt

pkg/cloud/amazon/s3_storage.go

@@ -57,6 +57,8 @@ const (
 	AWSEndpointParam = "AWS_ENDPOINT"
 	// AWSEndpointParam is the query parameter for UsePathStyle in S3 options.
 	AWSUsePathStyle = "AWS_USE_PATH_STYLE"
+	// AWSSkipChecksumParam is the query parameter for SkipChecksum in S3 options.
+	AWSSkipChecksumParam = "AWS_SKIP_CHECKSUM"
 
 	// AWSServerSideEncryptionMode is the query parameter in an AWS URI, for the
 	// mode to be used for server side encryption. It can either be AES256 or
@@ -200,6 +202,7 @@ type s3ClientConfig struct {
 	assumeRoleProvider                                           roleProvider
 	delegateRoleProviders                                        []roleProvider
 
+	skipChecksum bool
 	// log.V(2) decides session init params so include it in key.
 	verbose bool
 }
@@ -232,6 +235,7 @@ func clientConfig(conf *cloudpb.ExternalStorage_S3) s3ClientConfig {
 	return s3ClientConfig{
 		endpoint:              conf.Endpoint,
 		usePathStyle:          conf.UsePathStyle,
+		skipChecksum:          conf.SkipChecksum,
 		region:                conf.Region,
 		bucket:                conf.Bucket,
 		accessKey:             conf.AccessKey,
@@ -280,6 +284,9 @@ func S3URI(bucket, path string, conf *cloudpb.ExternalStorage_S3) string {
 	if conf.UsePathStyle {
 		q.Set(AWSUsePathStyle, "true")
 	}
+	if conf.SkipChecksum {
+		q.Set(AWSSkipChecksumParam, "true")
+	}
 	if conf.AssumeRoleProvider.Role != "" {
 		roleProviderStrings := make([]string, 0, len(conf.DelegateRoleProviders)+1)
 		for _, p := range conf.DelegateRoleProviders {
@@ -325,6 +332,15 @@ func parseS3URL(uri *url.URL) (cloudpb.ExternalStorage, error) {
 			return cloudpb.ExternalStorage{}, errors.Wrapf(err, "cannot parse %s as bool", AWSUsePathStyle)
 		}
 	}
+	skipChecksumStr := s3URL.ConsumeParam(AWSSkipChecksumParam)
+	skipChecksumBool := false
+	if skipChecksumStr != "" {
+		var err error
+		skipChecksumBool, err = strconv.ParseBool(skipChecksumStr)
+		if err != nil {
+			return cloudpb.ExternalStorage{}, errors.Wrapf(err, "cannot parse %s as bool", AWSSkipChecksumParam)
+		}
+	}
 
 	conf.S3Config = &cloudpb.ExternalStorage_S3{
 		Bucket:                s3URL.Host,
@@ -334,6 +350,7 @@ func parseS3URL(uri *url.URL) (cloudpb.ExternalStorage, error) {
 		TempToken:             s3URL.ConsumeParam(AWSTempTokenParam),
 		Endpoint:              s3URL.ConsumeParam(AWSEndpointParam),
 		UsePathStyle:          pathStyleBool,
+		SkipChecksum:          skipChecksumBool,
 		Region:                s3URL.ConsumeParam(S3RegionParam),
 		Auth:                  s3URL.ConsumeParam(cloud.AuthParam),
 		ServerEncMode:         s3URL.ConsumeParam(AWSServerSideEncryptionMode),
@@ -590,6 +607,11 @@ func (s *s3Storage) newClient(ctx context.Context) (s3Client, string, error) {
 		return s3Client{}, "", errors.Wrap(err, "could not initialize an aws config")
 	}
 
+	if s.opts.skipChecksum {
+		cfg.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
+		cfg.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+	}
+
 	var endpointURI string
 	if s.opts.endpoint != "" {
 		var err error
@@ -728,7 +750,7 @@ func (s *s3Storage) putUploader(ctx context.Context, basename string) (io.WriteC
 
 	buf := bytes.NewBuffer(make([]byte, 0, 4<<20))
 
-	return &putUploader{
+	uploader := &putUploader{
 		b: buf,
 		input: &s3.PutObjectInput{
 			Bucket:               s.bucket,
@@ -739,7 +761,11 @@ func (s *s3Storage) putUploader(ctx context.Context, basename string) (io.WriteC
 			ChecksumAlgorithm:    checksumAlgorithm,
 		},
 		client: client,
-	}, nil
+	}
+	if s.conf.SkipChecksum {
+		uploader.input.ChecksumAlgorithm = ""
+	}
+	return uploader, nil
 }
 
 func (s *s3Storage) Writer(ctx context.Context, basename string) (io.WriteCloser, error) {
@@ -757,16 +783,21 @@ func (s *s3Storage) Writer(ctx context.Context, basename string) (io.WriteCloser
 	return cloud.BackgroundPipe(ctx, func(ctx context.Context, r io.Reader) error {
 		defer sp.Finish()
 		// Upload the file to S3.
-		// TODO(dt): test and tune the uploader parameters.
-		_, err := uploader.Upload(ctx, &s3.PutObjectInput{
+		input := &s3.PutObjectInput{
 			Bucket:               s.bucket,
 			Key:                  aws.String(path.Join(s.prefix, basename)),
 			Body:                 r,
 			ServerSideEncryption: types.ServerSideEncryption(s.conf.ServerEncMode),
 			SSEKMSKeyId:          nilIfEmpty(s.conf.ServerKMSID),
 			StorageClass:         types.StorageClass(s.conf.StorageClass),
 			ChecksumAlgorithm:    checksumAlgorithm,
-		})
+		}
+
+		if s.conf.SkipChecksum {
+			input.ChecksumAlgorithm = ""
+		}
+
+		_, err := uploader.Upload(ctx, input)
 		err = interpretAWSError(err)
 		err = errors.Wrap(err, "upload failed")
 		// Mark with ctx's error for upstream code to not interpret this as
@@ -787,7 +818,9 @@ func (s *s3Storage) openStreamAt(
 	if err != nil {
 		return nil, err
 	}
-	req := &s3.GetObjectInput{Bucket: s.bucket, Key: aws.String(path.Join(s.prefix, basename))}
+	req := &s3.GetObjectInput{
+		Bucket: s.bucket, Key: aws.String(path.Join(s.prefix, basename)),
+	}
 	if endPos != 0 {
 		if pos >= endPos {
 			return nil, io.EOF

pkg/cloud/amazon/s3_storage_test.go

@@ -167,6 +167,22 @@ func TestPutS3(t *testing.T) {
 				cloudtestutils.CheckExportStore(t, info)
 			})
 
+			t.Run("skip-checksums", func(t *testing.T) {
+				if locked {
+					skip.IgnoreLint(t, "object-locked buckets do not support skipping checksums")
+				}
+				skipIfNoDefaultConfig(t, ctx)
+				info := cloudtestutils.StoreInfo{
+					URI: fmt.Sprintf(
+						"s3://%s/%s-%d?%s=%s&%s=true",
+						bucket, "backup-test-skip-checksums", testID,
+						cloud.AuthParam, cloud.AuthParamImplicit, AWSSkipChecksumParam,
+					),
+					User: user,
+				}
+				cloudtestutils.CheckExportStore(t, info)
+			})
+
 			t.Run("server-side-encryption-invalid-params", func(t *testing.T) {
 				skipIfNoDefaultConfig(t, ctx)
 				// Unsupported server side encryption option.
@@ -357,6 +373,7 @@ func TestPutS3Endpoint(t *testing.T) {
 		}
 		cloudtestutils.CheckExportStore(t, info)
 	})
+
 	t.Run("use-path-style", func(t *testing.T) {
 		// EngFlow machines have no internet access, and queries even to localhost will time out.
 		// So this test is skipped above.

pkg/cloud/cloudpb/external_storage.proto

@@ -63,6 +63,7 @@ message ExternalStorage {
     string temp_token = 5;
     string endpoint = 6;
     bool use_path_style=16;
+    bool skip_checksum=17;
     string region = 7;
     string auth = 8;
     string server_enc_mode  = 9;
@@ -91,7 +92,10 @@ message ExternalStorage {
     // role chain. These roles will be assumed in the order they appear in the
     // list so that the role specified in AssumeRoleProvider can be assumed.
     repeated AssumeRoleProvider delegate_role_providers = 15 [(gogoproto.nullable) = false];
+
+    // Next ID: 18;
   }
+  
   message GCS {
     string bucket = 1;
     string prefix = 2;