Merge pull request #1469 from Altinity/gcs_multipart

add gcs.allow_multipart_upload and gcs.allow_multipart_download for parallel transfers of large files

Author: Slach

ChangeLog.md

@@ -1,5 +1,7 @@
 # v2.8.0
 NEW FEATURES
+- add `gcs.allow_multipart_upload` (env `GCS_ALLOW_MULTIPART_UPLOAD`, default `false`, experimental) with `gcs.upload_concurrency` and `gcs.multipart_upload_min_size` (default 1GB) — files bigger than `multipart_upload_min_size` upload to GCS as multiple parallel parts (part size is `chunk_size`) via gRPC client and compose into the final object; not compatible with `endpoint`, `force_http`, `encryption_key`, fix [#1028](https://github.com/Altinity/clickhouse-backup/issues/1028)
+- add `gcs.allow_multipart_download` (env `GCS_ALLOW_MULTIPART_DOWNLOAD`, default `false`) with `gcs.download_concurrency` — download each file as parallel range reads (part size is `chunk_size`) into a temporary file, mirrors `s3.allow_multipart_download`, fix [#1028](https://github.com/Altinity/clickhouse-backup/issues/1028)
 - add `general.disable_environment_override` (settable ONLY in the config file, no environment variable name on purpose, default `false`) — when `true`, config values come only from the config file: all environment variables and the `--env` CLI flag are ignored during config loading; protects against accidental overrides such as Kubernetes service-discovery variables, fix [#1079](https://github.com/Altinity/clickhouse-backup/issues/1079)
 
 # v2.7.4

ReadMe.md

@@ -367,6 +367,15 @@ gcs:
   client_pool_size: 500        # GCS_CLIENT_POOL_SIZE, default max(upload_concurrency, download concurrency) * 3, should be at least 3 times bigger than `UPLOAD_CONCURRENCY` or `DOWNLOAD_CONCURRENCY` in each upload and download case to avoid stuck
   delete_concurrency: 50       # GCS_DELETE_CONCURRENCY, how many objects delete in parallel during clean/delete operations
   upload_buffer_size: 131072   # GCS_UPLOAD_BUFFER_SIZE, io.CopyBuffer size in bytes feeding the GCS object writer, default 128KB; raise (e.g. 1048576 = 1MB) on high-bandwidth networks, see https://github.com/Altinity/clickhouse-backup/issues/1376
+  # GCS_ALLOW_MULTIPART_UPLOAD, experimental: upload files bigger than `multipart_upload_min_size` as multiple parts (part size is `chunk_size`, minimum 5MiB) in parallel and compose them into the final object, see https://github.com/Altinity/clickhouse-backup/issues/1028
+  # requires direct gRPC connection to storage.googleapis.com, not compatible with `endpoint`, `force_http` and `encryption_key`
+  # temporary parts are written to the `gcs-go-sdk-pu-tmp/` prefix in the bucket root and deleted after compose; setup a bucket lifecycle rule for this prefix to clean up leftovers after crashes
+  allow_multipart_upload: false
+  upload_concurrency: 0        # GCS_UPLOAD_CONCURRENCY, how many parts of one file upload in parallel when `allow_multipart_upload` enabled, default min(4 + CPU/2, 16)
+  multipart_upload_min_size: 1073741824 # GCS_MULTIPART_UPLOAD_MIN_SIZE, files smaller than this size in bytes use the regular single-stream upload, default 1GB
+  # GCS_ALLOW_MULTIPART_DOWNLOAD, download each file as parallel range reads (part size is `chunk_size`) into a temporary file, requires additional disk space, see https://github.com/Altinity/clickhouse-backup/issues/1028
+  allow_multipart_download: false
+  download_concurrency: 1      # GCS_DOWNLOAD_CONCURRENCY, how many parts of one file download in parallel when `allow_multipart_download` enabled, default `max(CPU/2, 1) + 1`
   # GCS_OBJECT_LABELS, allow setup metadata for each object during upload, use {macro_name} from system.macros and {backupName} for current backup name
   # The format for this env variable is "key1:value1,key2:value2". For YAML please continue using map syntax
   object_labels: {}

pkg/config/config.go

@@ -176,6 +176,21 @@ type GCSConfig struct {
 	EncryptionKey string `yaml:"encryption_key" envconfig:"GCS_ENCRYPTION_KEY"`
 	// UploadBufferSize - io.CopyBuffer size feeding the GCS object writer, see https://github.com/Altinity/clickhouse-backup/issues/1376
 	UploadBufferSize int `yaml:"upload_buffer_size" envconfig:"GCS_UPLOAD_BUFFER_SIZE"`
+	// AllowMultipartUpload enables experimental parallel composite uploads via a gRPC client for files
+	// bigger than MultipartUploadMinSize, part size is chunk_size, see https://github.com/Altinity/clickhouse-backup/issues/1028.
+	// Not compatible with `endpoint`, `force_http` and `encryption_key`.
+	AllowMultipartUpload bool `yaml:"allow_multipart_upload" envconfig:"GCS_ALLOW_MULTIPART_UPLOAD"`
+	// UploadConcurrency - how many parts of one file upload in parallel when allow_multipart_upload enabled,
+	// 0 means SDK default min(4 + NumCPU/2, 16)
+	UploadConcurrency int `yaml:"upload_concurrency" envconfig:"GCS_UPLOAD_CONCURRENCY"`
+	// MultipartUploadMinSize - files smaller than this size use the regular single-stream upload
+	MultipartUploadMinSize int64 `yaml:"multipart_upload_min_size" envconfig:"GCS_MULTIPART_UPLOAD_MIN_SIZE"`
+	// AllowMultipartDownload - download each file as parallel range reads into a temporary file
+	// (requires additional disk space), part size is chunk_size, mirrors s3.allow_multipart_download,
+	// see https://github.com/Altinity/clickhouse-backup/issues/1028
+	AllowMultipartDownload bool `yaml:"allow_multipart_download" envconfig:"GCS_ALLOW_MULTIPART_DOWNLOAD"`
+	// DownloadConcurrency - how many parts of one file download in parallel when allow_multipart_download enabled
+	DownloadConcurrency int `yaml:"download_concurrency" envconfig:"GCS_DOWNLOAD_CONCURRENCY"`
 }
 
 // AzureBlobConfig - Azure Blob settings section
@@ -608,6 +623,25 @@ func ValidateConfig(cfg *Config) error {
 			}
 		}
 	}
+	if cfg.General.RemoteStorage == "gcs" && cfg.GCS.AllowMultipartUpload {
+		// parallel composite upload works only via gRPC client to storage.googleapis.com,
+		// and Compose requires the same CSEK for source parts and destination object
+		if cfg.GCS.Endpoint != "" {
+			return errors.New("gcs `allow_multipart_upload` requires gRPC client and is not compatible with `endpoint`")
+		}
+		if cfg.GCS.ForceHttp {
+			return errors.New("gcs `allow_multipart_upload` requires gRPC client and is not compatible with `force_http`")
+		}
+		if cfg.GCS.EncryptionKey != "" {
+			return errors.New("gcs `allow_multipart_upload` is not compatible with `encryption_key`")
+		}
+	}
+	if cfg.General.RemoteStorage == "gcs" && cfg.GCS.AllowMultipartDownload && cfg.GCS.DownloadConcurrency <= 1 {
+		return errors.Errorf(
+			"`allow_multipart_download` require `download_concurrency` in `gcs` section more than 1 (3-4 recommends) current value: %d",
+			cfg.GCS.DownloadConcurrency,
+		)
+	}
 	if cfg.GetCompressionFormat() == "unknown" {
 		return errors.Errorf("'%s' is unknown remote storage", cfg.General.RemoteStorage)
 	}
@@ -884,9 +918,11 @@ func DefaultConfig() *Config {
 			StorageClass:      "STANDARD",
 			ClientPoolSize:    int(max(uploadConcurrency*3, downloadConcurrency*3, objectDiskServerSideCopyConcurrency)),
 			// 16Mb default chunk size, fix https://github.com/Altinity/clickhouse-backup/issues/1292
-			ChunkSize:         16 * 1024 * 1024,
-			DeleteConcurrency: 50,
-			UploadBufferSize:  128 * 1024,
+			ChunkSize:              16 * 1024 * 1024,
+			DeleteConcurrency:      50,
+			UploadBufferSize:       128 * 1024,
+			MultipartUploadMinSize: 1024 * 1024 * 1024,
+			DownloadConcurrency:    int(downloadConcurrency + 1),
 		},
 		COS: COSConfig{
 			RowURL:                 "",

pkg/config/config_test.go

@@ -387,6 +387,34 @@ func TestValidateConfigErrors(t *testing.T) {
 			cfg.General.RemoteStorage = "s3"
 			cfg.S3.HTTPIdleConnTimeout = "1parsec"
 		}, "invalid s3 http_idle_conn_timeout"},
+		{"gcs allow_multipart_upload with default config", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartUpload = true
+		}, ""},
+		{"gcs allow_multipart_upload incompatible with endpoint", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartUpload = true
+			cfg.GCS.Endpoint = "http://gcs:8080/storage/v1/"
+		}, "gcs `allow_multipart_upload` requires gRPC client and is not compatible with `endpoint`"},
+		{"gcs allow_multipart_upload incompatible with force_http", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartUpload = true
+			cfg.GCS.ForceHttp = true
+		}, "gcs `allow_multipart_upload` requires gRPC client and is not compatible with `force_http`"},
+		{"gcs allow_multipart_upload incompatible with encryption_key", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartUpload = true
+			cfg.GCS.EncryptionKey = "dGVzdC1rZXktdGVzdC1rZXktdGVzdC1rZXktdGVzdA=="
+		}, "gcs `allow_multipart_upload` is not compatible with `encryption_key`"},
+		{"gcs allow_multipart_download with default config", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartDownload = true
+		}, ""},
+		{"gcs allow_multipart_download requires download_concurrency > 1", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.AllowMultipartDownload = true
+			cfg.GCS.DownloadConcurrency = 1
+		}, "`allow_multipart_download` require `download_concurrency` in `gcs` section more than 1"},
 		{"unknown remote storage", func(cfg *Config) {
 			cfg.General.RemoteStorage = "banana"
 		}, "'banana' is unknown remote storage"},

pkg/storage/gcs.go

@@ -20,6 +20,7 @@ import (
 	"github.com/pkg/errors"
 
 	"cloud.google.com/go/storage"
+	"cloud.google.com/go/storage/transfermanager"
 	"github.com/rs/zerolog/log"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/api/impersonate"
@@ -31,6 +32,7 @@ import (
 // GCS - presents methods for manipulate data on GCS
 type GCS struct {
 	client        *storage.Client
+	grpcClient    *storage.Client // used only for parallel composite uploads, see https://github.com/Altinity/clickhouse-backup/issues/1028
 	Config        *config.GCSConfig
 	clientPool    *pool.ObjectPool
 	encryptionKey []byte // Customer-Supplied Encryption Key (CSEK)
@@ -239,6 +241,20 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 		return errors.Wrap(err, "GCS Connect storage.NewClient")
 	}
 
+	if gcs.Config.AllowMultipartUpload {
+		grpcOptions := []option.ClientOption{option.WithTelemetryDisabled()}
+		if credOption != nil {
+			grpcOptions = append(grpcOptions, credOption)
+		}
+		gcs.grpcClient, err = storage.NewGRPCClient(ctx, grpcOptions...)
+		if err != nil {
+			return errors.Wrap(err, "GCS Connect storage.NewGRPCClient")
+		}
+		// client-level retry policy, part uploads inside parallel upload create
+		// fresh ObjectHandle's from the client, so per-object Retryer doesn't cover them
+		gcs.grpcClient.SetRetry(storage.WithPolicy(storage.RetryAlways))
+	}
+
 	// Validate and decode the encryption key if provided
 	if gcs.Config.EncryptionKey != "" {
 		key, err := base64.StdEncoding.DecodeString(gcs.Config.EncryptionKey)
@@ -257,6 +273,11 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 
 func (gcs *GCS) Close(ctx context.Context) error {
 	gcs.clientPool.Close(ctx)
+	if gcs.grpcClient != nil {
+		if err := gcs.grpcClient.Close(); err != nil {
+			return errors.Wrap(err, "GCS Close grpcClient")
+		}
+	}
 	if err := gcs.client.Close(); err != nil {
 		return errors.Wrap(err, "GCS Close")
 	}
@@ -373,15 +394,76 @@ func (gcs *GCS) GetFileReaderAbsolute(ctx context.Context, key string) (io.ReadC
 	return reader, nil
 }
 
-func (gcs *GCS) GetFileReaderWithLocalPath(ctx context.Context, key, _ string, _ int64) (io.ReadCloser, error) {
+func (gcs *GCS) GetFileReaderWithLocalPath(ctx context.Context, key, localPath string, remoteSize int64) (io.ReadCloser, error) {
+	/* unfortunately, multipart download require allocate additional disk space
+	and don't allow us to decompress data directly from stream */
+	if gcs.Config.AllowMultipartDownload {
+		log.Debug().Msgf("GCS->GetFileReaderWithLocalPath: multipart download %s, size=%d", key, remoteSize)
+		writer, err := os.CreateTemp(localPath, strings.ReplaceAll(key, "/", "_"))
+		if err != nil {
+			return nil, errors.Wrap(err, "GCS GetFileReaderWithLocalPath CreateTemp")
+		}
+		if err = gcs.downloadMultipart(ctx, key, writer); err != nil {
+			_ = writer.Close()
+			_ = os.Remove(writer.Name())
+			return nil, err
+		}
+		return writer, nil
+	}
 	return gcs.GetFileReader(ctx, key)
 }
 
+// downloadMultipart downloads the object as parallel range reads via transfermanager.Downloader,
+// see https://github.com/Altinity/clickhouse-backup/issues/1028
+func (gcs *GCS) downloadMultipart(ctx context.Context, key string, writer io.WriterAt) error {
+	fullKey := path.Join(gcs.Config.Path, key)
+	partSize := int64(gcs.Config.ChunkSize)
+	if partSize <= 0 {
+		partSize = 16 * 1024 * 1024
+	}
+	attempt := func(encryptionKey []byte) error {
+		downloader, err := transfermanager.NewDownloader(gcs.client,
+			transfermanager.WithWorkers(gcs.Config.DownloadConcurrency),
+			transfermanager.WithPartSize(partSize),
+		)
+		if err != nil {
+			return errors.Wrap(err, "GCS downloadMultipart NewDownloader")
+		}
+		if err = downloader.DownloadObject(ctx, &transfermanager.DownloadObjectInput{
+			Bucket:        gcs.Config.Bucket,
+			Object:        fullKey,
+			Destination:   writer,
+			EncryptionKey: encryptionKey,
+		}); err != nil {
+			return errors.Wrap(err, "GCS downloadMultipart DownloadObject")
+		}
+		_, err = downloader.WaitAndClose()
+		return err
+	}
+	// Do NOT apply encryption for object_disks files - they are not encrypted
+	// because ClickHouse needs to read them directly without encryption key
+	encryptionKey := gcs.encryptionKey
+	if gcs.Config.ObjectDiskPath != "" && strings.HasPrefix(fullKey, gcs.Config.ObjectDiskPath) {
+		encryptionKey = nil
+	}
+	err := attempt(encryptionKey)
+	if err != nil && encryptionKey != nil && gcs.isNotEncryptedError(err) {
+		// If the object is not encrypted but we tried to read it with encryption key,
+		// retry without encryption (for backward compatibility with old backups)
+		log.Warn().Msgf("gcs.downloadMultipart: object %s not encrypted, retrying without encryption key", fullKey)
+		err = attempt(nil)
+	}
+	return errors.Wrap(err, "GCS downloadMultipart WaitAndClose")
+}
+
 func (gcs *GCS) PutFile(ctx context.Context, key string, r io.ReadCloser, localSize int64) error {
 	return gcs.PutFileAbsolute(ctx, path.Join(gcs.Config.Path, key), r, localSize)
 }
 
-func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser, _ int64) error {
+func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser, localSize int64) error {
+	if gcs.grpcClient != nil && localSize >= gcs.Config.MultipartUploadMinSize {
+		return gcs.putFileMultipart(ctx, key, r, localSize)
+	}
 	pClientObj, err := gcs.clientPool.BorrowObject(ctx)
 	if err != nil {
 		log.Error().Msgf("gcs.PutFile: gcs.clientPool.BorrowObject error: %+v", err)
@@ -425,6 +507,40 @@ func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser
 	return nil
 }
 
+// putFileMultipart uploads a large file with the experimental parallel composite upload,
+// temporary parts go to the `gcs-go-sdk-pu-tmp/` prefix in the bucket root and are composed
+// into the final object, see https://github.com/Altinity/clickhouse-backup/issues/1028.
+// encryption_key is rejected together with allow_multipart_upload in config validation, so no CSEK handling here.
+func (gcs *GCS) putFileMultipart(ctx context.Context, key string, r io.ReadCloser, localSize int64) error {
+	log.Debug().Msgf("GCS->putFileMultipart %s, size=%d", key, localSize)
+	obj := gcs.grpcClient.Bucket(gcs.Config.Bucket).Object(key)
+	writer := obj.NewWriter(ctx)
+	writer.EnableParallelUpload = true
+	// PartSize <= 0 falls back to the SDK default 16MiB, values below 5MiB are bumped to 5MiB by the SDK
+	writer.ParallelUploadConfig = storage.ParallelUploadConfig{
+		PartSize:       gcs.Config.ChunkSize,
+		MaxConcurrency: gcs.Config.UploadConcurrency,
+	}
+	writer.StorageClass = gcs.Config.StorageClass
+	if len(gcs.Config.ObjectLabels) > 0 {
+		writer.Metadata = gcs.Config.ObjectLabels
+	}
+	uploadBufferSize := gcs.Config.UploadBufferSize
+	if uploadBufferSize <= 0 {
+		uploadBufferSize = 128 * 1024
+	}
+	buffer := make([]byte, uploadBufferSize)
+	if _, err := io.CopyBuffer(writer, r, buffer); err != nil {
+		log.Warn().Msgf("gcs.putFileMultipart: can't copy buffer: %+v", err)
+		return errors.Wrap(err, "GCS putFileMultipart CopyBuffer")
+	}
+	if err := writer.Close(); err != nil {
+		log.Warn().Msgf("gcs.putFileMultipart: can't close writer: %+v", err)
+		return errors.Wrap(err, "GCS putFileMultipart writer.Close")
+	}
+	return nil
+}
+
 func (gcs *GCS) StatFile(ctx context.Context, key string) (RemoteFile, error) {
 	return gcs.StatFileAbsolute(ctx, path.Join(gcs.Config.Path, key))
 }

pkg/storage/general.go

@@ -785,7 +785,7 @@ func (bd *BackupDestination) UploadPath(ctx context.Context, baseLocalPath strin
 		}
 		retry := retrier.New(retrier.ExponentialBackoff(RetriesOnFailure, common.AddRandomJitter(RetriesDuration, RetriesJitter)), RertierClassifier)
 		err = retry.RunCtx(ctx, func(ctx context.Context) error {
-			return bd.PutFile(ctx, path.Join(remotePath, filename), bwlimit.ReadCloser(ctx, f, limiter), 0)
+			return bd.PutFile(ctx, path.Join(remotePath, filename), bwlimit.ReadCloser(ctx, f, limiter), fInfo.Size())
 		})
 		if err != nil {
 			closeFile()

test/integration/configs/config-gcs-parallel.yml

@@ -0,0 +1,27 @@
+general:
+  remote_storage: gcs
+  upload_concurrency: 4
+  download_concurrency: 4
+  restore_schema_on_cluster: "{cluster}"
+  allow_object_disk_streaming: true
+s3:
+  disable_ssl: false
+  disable_cert_verification: true
+clickhouse:
+  host: clickhouse
+  port: 9000
+  restart_command: bash -c 'echo "FAKE RESTART"'
+  timeout: 10m
+gcs:
+  bucket: altinity-qa-test
+  path: backup/{cluster}/{shard}/{version}
+  object_disk_path: object_disks/{cluster}/{shard}/{version}
+  credentials_file: /etc/clickhouse-backup/credentials.json
+  compression_format: tar
+  # experimental parallel composite upload and multipart download, see https://github.com/Altinity/clickhouse-backup/issues/1028
+  # low thresholds and 5MiB chunk_size so the ~30MB test archive splits into multiple parallel parts
+  allow_multipart_upload: true
+  multipart_upload_min_size: 1048576
+  allow_multipart_download: true
+  download_concurrency: 4
+  chunk_size: 5242880

test/integration/containers.go

@@ -1035,7 +1035,7 @@ func (tc *TestContainers) clickHouseBinds(curDir, configsDir string) []string {
 		"config-custom-kopia.yml", "config-custom-restic.yml", "config-custom-rsync.yml",
 		"config-database-mapping.yml",
 		"config-ftp.yaml", "config-ftp-old.yaml",
-		"config-gcs.yml", "config-gcs-custom-endpoint.yml",
+		"config-gcs.yml", "config-gcs-custom-endpoint.yml", "config-gcs-parallel.yml",
 		"config-s3.yml", "config-s3-embedded.yml", "config-s3-embedded-url.yml",
 		"config-s3-embedded-local.yml", "config-s3-nodelete.yml", "config-s3-plain-embedded.yml",
 		"config-sftp-auth-key.yaml", "config-sftp-auth-password.yaml",