add S3 minimum part size defined by the user (vitessio#17171)

Signed-off-by: Renan Rangel <rrangel@slack-corp.com>

examples/operator/operator.yaml

@@ -511,6 +511,9 @@ spec:
                           maxLength: 256
                           pattern: ^[^\r\n]*$
                           type: string
+                        minPartSize:
+                          format: int64
+                          type: integer
                         region:
                           minLength: 1
                           type: string
@@ -1351,6 +1354,9 @@ spec:
                                 maxLength: 256
                                 pattern: ^[^\r\n]*$
                                 type: string
+                              minPartSize:
+                                format: int64
+                                type: integer
                               region:
                                 minLength: 1
                                 type: string
@@ -3898,6 +3904,9 @@ spec:
                             maxLength: 256
                             pattern: ^[^\r\n]*$
                             type: string
+                          minPartSize:
+                            format: int64
+                            type: integer
                           region:
                             minLength: 1
                             type: string
@@ -5247,6 +5256,9 @@ spec:
                             maxLength: 256
                             pattern: ^[^\r\n]*$
                             type: string
+                          minPartSize:
+                            format: int64
+                            type: integer
                           region:
                             minLength: 1
                             type: string

go.mod

@@ -71,7 +71,7 @@ require (
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/net v0.28.0
 	golang.org/x/oauth2 v0.22.0
-	golang.org/x/sys v0.24.0
+	golang.org/x/sys v0.25.0
 	golang.org/x/term v0.23.0
 	golang.org/x/text v0.17.0 // indirect
 	golang.org/x/time v0.6.0
@@ -96,6 +96,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.60.1
 	github.com/aws/smithy-go v1.20.4
 	github.com/bndr/gotabulate v1.1.2
+	github.com/dustin/go-humanize v1.0.1
 	github.com/gammazero/deque v0.2.1
 	github.com/google/safehtml v0.1.0
 	github.com/hashicorp/go-cleanhttp v0.5.2
@@ -151,9 +152,8 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/dustin/go-humanize v1.0.1 // indirect
-	github.com/ebitengine/purego v0.6.1 // indirect
-	github.com/fatih/color v1.16.0 // indirect
+	github.com/ebitengine/purego v0.8.1 // indirect
+	github.com/fatih/color v1.18.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect

go.sum

@@ -152,8 +152,8 @@ github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUn
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
-github.com/ebitengine/purego v0.6.1 h1:sjN8rfzbhXQ59/pE+wInswbU9aMDHiwlup4p/a07Mkg=
-github.com/ebitengine/purego v0.6.1/go.mod h1:ah1In8AOtksoNK6yk5z1HTJeUkC1Ez4Wk2idgGslMwQ=
+github.com/ebitengine/purego v0.8.1 h1:sdRKd6plj7KYW33EH5As6YKfe8m9zbN9JMrOjNVF/BE=
+github.com/ebitengine/purego v0.8.1/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
@@ -163,8 +163,8 @@ github.com/evanphx/json-patch v5.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLi
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
-github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM=
-github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
+github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
+github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
@@ -682,8 +682,8 @@ golang.org/x/sys v0.0.0-20220627191245-f75cf1eec38b/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
-golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
 golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=

go/flags/endtoend/vtbackup.txt

@@ -196,6 +196,7 @@ Flags:
       --remote_operation_timeout duration                           time to wait for a remote operation (default 15s)
       --restart_before_backup                                       Perform a mysqld clean/full restart after applying binlogs, but before taking the backup. Only makes sense to work around xtrabackup bugs.
       --s3_backup_aws_endpoint string                               endpoint of the S3 backend (region must be provided).
+      --s3_backup_aws_min_partsize int                              Minimum part size to use, defaults to 5MiB but can be increased due to the dataset size. (default 5242880)
       --s3_backup_aws_region string                                 AWS region to use. (default "us-east-1")
       --s3_backup_aws_retries int                                   AWS request retries. (default -1)
       --s3_backup_force_path_style                                  force the s3 path style.

go/flags/endtoend/vtctld.txt

@@ -110,6 +110,7 @@ Flags:
       --purge_logs_interval duration                                     how often try to remove old logs (default 1h0m0s)
       --remote_operation_timeout duration                                time to wait for a remote operation (default 15s)
       --s3_backup_aws_endpoint string                                    endpoint of the S3 backend (region must be provided).
+      --s3_backup_aws_min_partsize int                                   Minimum part size to use, defaults to 5MiB but can be increased due to the dataset size. (default 5242880)
       --s3_backup_aws_region string                                      AWS region to use. (default "us-east-1")
       --s3_backup_aws_retries int                                        AWS request retries. (default -1)
       --s3_backup_force_path_style                                       force the s3 path style.

go/flags/endtoend/vttablet.txt

@@ -312,6 +312,7 @@ Flags:
       --restore_from_backup_ts string                                    (init restore parameter) if set, restore the latest backup taken at or before this timestamp. Example: '2021-04-29.133050'
       --retain_online_ddl_tables duration                                How long should vttablet keep an old migrated table before purging it (default 24h0m0s)
       --s3_backup_aws_endpoint string                                    endpoint of the S3 backend (region must be provided).
+      --s3_backup_aws_min_partsize int                                   Minimum part size to use, defaults to 5MiB but can be increased due to the dataset size. (default 5242880)
       --s3_backup_aws_region string                                      AWS region to use. (default "us-east-1")
       --s3_backup_aws_retries int                                        AWS request retries. (default -1)
       --s3_backup_force_path_style                                       force the s3 path style.

go/vt/mysqlctl/s3backupstorage/s3.go

@@ -44,7 +44,9 @@ import (
 	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	transport "github.com/aws/smithy-go/endpoints"
 	"github.com/aws/smithy-go/middleware"
+	"github.com/dustin/go-humanize"
 	"github.com/spf13/pflag"
 
 	"vitess.io/vitess/go/vt/concurrency"
@@ -54,6 +56,11 @@ import (
 	"vitess.io/vitess/go/vt/servenv"
 )
 
+const (
+	sseCustomerPrefix = "sse_c:"
+	MaxPartSize       = 1024 * 1024 * 1024 * 5 // 5GiB - limited by AWS https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html
+)
+
 var (
 	// AWS API region
 	region string
@@ -83,6 +90,11 @@ var (
 
 	// path component delimiter
 	delimiter = "/"
+
+	// minimum part size
+	minPartSize int64
+
+	ErrPartSize = errors.New("minimum S3 part size must be between 5MiB and 5GiB")
 )
 
 func registerFlags(fs *pflag.FlagSet) {
@@ -95,6 +107,7 @@ func registerFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&tlsSkipVerifyCert, "s3_backup_tls_skip_verify_cert", false, "skip the 'certificate is valid' check for SSL connections.")
 	fs.StringVar(&requiredLogLevel, "s3_backup_log_level", "LogOff", "determine the S3 loglevel to use from LogOff, LogDebug, LogDebugWithSigning, LogDebugWithHTTPBody, LogDebugWithRequestRetries, LogDebugWithRequestErrors.")
 	fs.StringVar(&sse, "s3_backup_server_side_encryption", "", "server-side encryption algorithm (e.g., AES256, aws:kms, sse_c:/path/to/key/file).")
+	fs.Int64Var(&minPartSize, "s3_backup_aws_min_partsize", manager.MinUploadPartSize, "Minimum part size to use, defaults to 5MiB but can be increased due to the dataset size.")
 }
 
 func init() {
@@ -108,7 +121,22 @@ type logNameToLogLevel map[string]aws.ClientLogMode
 
 var logNameMap logNameToLogLevel
 
-const sseCustomerPrefix = "sse_c:"
+type endpointResolver struct {
+	r        s3.EndpointResolverV2
+	endpoint *string
+}
+
+func (er *endpointResolver) ResolveEndpoint(ctx context.Context, params s3.EndpointParameters) (transport.Endpoint, error) {
+	params.Endpoint = er.endpoint
+	return er.r.ResolveEndpoint(ctx, params)
+}
+
+func newEndpointResolver() *endpointResolver {
+	return &endpointResolver{
+		r:        s3.NewDefaultEndpointResolverV2(),
+		endpoint: &endpoint,
+	}
+}
 
 type iClient interface {
 	manager.UploadAPIClient
@@ -161,17 +189,13 @@ func (bh *S3BackupHandle) AddFile(ctx context.Context, filename string, filesize
 		return nil, fmt.Errorf("AddFile cannot be called on read-only backup")
 	}
 
-	// Calculate s3 upload part size using the source filesize
-	partSizeBytes := manager.DefaultUploadPartSize
-	if filesize > 0 {
-		minimumPartSize := float64(filesize) / float64(manager.MaxUploadParts)
-		// Round up to ensure large enough partsize
-		calculatedPartSizeBytes := int64(math.Ceil(minimumPartSize))
-		if calculatedPartSizeBytes > partSizeBytes {
-			partSizeBytes = calculatedPartSizeBytes
-		}
+	partSizeBytes, err := calculateUploadPartSize(filesize)
+	if err != nil {
+		return nil, err
 	}
 
+	bh.bs.params.Logger.Infof("Using S3 upload part size: %s", humanize.IBytes(uint64(partSizeBytes)))
+
 	reader, writer := io.Pipe()
 	bh.waitGroup.Add(1)
 
@@ -212,6 +236,32 @@ func (bh *S3BackupHandle) AddFile(ctx context.Context, filename string, filesize
 	return writer, nil
 }
 
+// calculateUploadPartSize is a helper to calculate the part size, taking into consideration the minimum part size
+// passed in by an operator.
+func calculateUploadPartSize(filesize int64) (partSizeBytes int64, err error) {
+	// Calculate s3 upload part size using the source filesize
+	partSizeBytes = manager.DefaultUploadPartSize
+	if filesize > 0 {
+		minimumPartSize := float64(filesize) / float64(manager.MaxUploadParts)
+		// Round up to ensure large enough partsize
+		calculatedPartSizeBytes := int64(math.Ceil(minimumPartSize))
+		if calculatedPartSizeBytes > partSizeBytes {
+			partSizeBytes = calculatedPartSizeBytes
+		}
+	}
+
+	if minPartSize != 0 && partSizeBytes < minPartSize {
+		if minPartSize > MaxPartSize || minPartSize < manager.MinUploadPartSize { // 5GiB and 5MiB respectively
+			return 0, fmt.Errorf("%w, currently set to %s",
+				ErrPartSize, humanize.IBytes(uint64(minPartSize)),
+			)
+		}
+		partSizeBytes = int64(minPartSize)
+	}
+
+	return
+}
+
 // EndBackup is part of the backupstorage.BackupHandle interface.
 func (bh *S3BackupHandle) EndBackup(ctx context.Context) error {
 	if bh.readOnly {

go/vt/mysqlctl/s3backupstorage/s3_test.go

@@ -328,3 +328,68 @@ func TestWithParams(t *testing.T) {
 	assert.NotNil(t, s3.transport.DialContext)
 	assert.NotNil(t, s3.transport.Proxy)
 }
+
+func TestCalculateUploadPartSize(t *testing.T) {
+	originalMinimum := minPartSize
+	defer func() { minPartSize = originalMinimum }()
+
+	tests := []struct {
+		name            string
+		filesize        int64
+		minimumPartSize int64
+		want            int64
+		err             error
+	}{
+		{
+			name:            "minimum - 10 MiB",
+			filesize:        1024 * 1024 * 10, // 10 MiB
+			minimumPartSize: 1024 * 1024 * 5,  // 5 MiB
+			want:            1024 * 1024 * 5,  // 5 MiB,
+			err:             nil,
+		},
+		{
+			name:            "below minimum - 10 MiB",
+			filesize:        1024 * 1024 * 10, // 10 MiB
+			minimumPartSize: 1024 * 1024 * 8,  // 8 MiB
+			want:            1024 * 1024 * 8,  // 8 MiB,
+			err:             nil,
+		},
+		{
+			name:            "above minimum - 1 TiB",
+			filesize:        1024 * 1024 * 1024 * 1024, // 1 TiB
+			minimumPartSize: 1024 * 1024 * 5,           // 5 MiB
+			want:            109951163,                 // ~104 MiB
+			err:             nil,
+		},
+		{
+			name:            "below minimum - 1 TiB",
+			filesize:        1024 * 1024 * 1024 * 1024, // 1 TiB
+			minimumPartSize: 1024 * 1024 * 200,         // 200 MiB
+			want:            1024 * 1024 * 200,         // 200 MiB
+			err:             nil,
+		},
+		{
+			name:            "below S3 limits - 5 MiB",
+			filesize:        1024 * 1024 * 3, // 3 MiB
+			minimumPartSize: 1024 * 1024 * 4, // 4 MiB
+			want:            1024 * 1024 * 5, // 5 MiB - should always return the minimum
+			err:             nil,
+		},
+		{
+			name:            "above S3 limits - 5 GiB",
+			filesize:        1024 * 1024 * 1024 * 1024, // 1 TiB
+			minimumPartSize: 1024 * 1024 * 1024 * 6,    // 6 GiB
+			want:            0,
+			err:             ErrPartSize,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			minPartSize = tt.minimumPartSize
+			partSize, err := calculateUploadPartSize(tt.filesize)
+			require.ErrorIs(t, err, tt.err)
+			require.Equal(t, tt.want, partSize)
+		})
+	}
+}