gcp,s3,azure: make the storage client upload chunk size configurable

pkg/cloud/amazon/s3_storage.go

@@ -352,7 +352,9 @@ func newClient(
 	sess.Config.Region = aws.String(region)
 
 	c := s3.New(sess)
-	u := s3manager.NewUploader(sess)
+	u := s3manager.NewUploader(sess, func(uploader *s3manager.Uploader) {
+		uploader.PartSize = cloud.WriteChunkSize.Get(&settings.SV)
+	})
 	return s3Client{client: c, uploader: u}, region, nil
 }
 

pkg/cloud/azure/azure_storage.go

@@ -125,7 +125,7 @@ func (s *azureStorage) Writer(ctx context.Context, basename string) (io.WriteClo
 		defer sp.Finish()
 		_, err := azblob.UploadStreamToBlockBlob(
 			ctx, r, blob, azblob.UploadStreamToBlockBlobOptions{
-				BufferSize: 4 << 20,
+				BufferSize: int(cloud.WriteChunkSize.Get(&s.settings.SV)),
 			},
 		)
 		return err

pkg/cloud/cloud_io.go

@@ -49,6 +49,15 @@ var httpCustomCA = settings.RegisterStringSetting(
 	"",
 ).WithPublic()
 
+// WriteChunkSize is used to control the size of each chunk that is buffered and
+// uploaded by the cloud storage client.
+var WriteChunkSize = settings.RegisterByteSizeSetting(
+	settings.TenantWritable,
+	"cloudstorage.write_chunk.size",
+	"controls the size of each file chunk uploaded by the cloud storage client",
+	8<<20,
+)
+
 // HTTPRetryOptions defines the tunable settings which control the retry of HTTP
 // operations.
 var HTTPRetryOptions = retry.Options{

pkg/cloud/gcp/gcs_storage.go

@@ -162,6 +162,7 @@ func (g *gcsStorage) Writer(ctx context.Context, basename string) (io.WriteClose
 		path.Join(g.prefix, basename))})
 
 	w := g.bucket.Object(path.Join(g.prefix, basename)).NewWriter(ctx)
+	w.ChunkSize = int(cloud.WriteChunkSize.Get(&g.settings.SV))
 	if !gcsChunkingEnabled.Get(&g.settings.SV) {
 		w.ChunkSize = 0
 	}