add MultiPart download, test refactoring, fix #1028

Author: Slach

ChangeLog.md

@@ -1,6 +1,7 @@
 # v2.8.0
 NEW FEATURES
-- add `gcs.parallel_upload` (env `GCS_PARALLEL_UPLOAD`, default `false`, experimental) with `gcs.parallel_upload_part_size`, `gcs.parallel_upload_max_concurrency` and `gcs.parallel_upload_min_size` (default 1GB) — files bigger than `parallel_upload_min_size` upload to GCS as multiple parallel parts 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_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,13 +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_PARALLEL_UPLOAD, experimental: upload files bigger than `parallel_upload_min_size` as multiple parts in parallel and compose them into the final object, see https://github.com/Altinity/clickhouse-backup/issues/1028
+  # 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
-  parallel_upload: false
-  parallel_upload_part_size: 0 # GCS_PARALLEL_UPLOAD_PART_SIZE, size in bytes of each part uploaded in parallel, default 16MiB, minimum 5MiB
-  parallel_upload_max_concurrency: 0 # GCS_PARALLEL_UPLOAD_MAX_CONCURRENCY, how many parts of one file upload in parallel, default min(4 + CPU/2, 16); total concurrent streams = upload_concurrency * parallel_upload_max_concurrency
-  parallel_upload_min_size: 1073741824 # GCS_PARALLEL_UPLOAD_MIN_SIZE, files smaller than this size in bytes use the regular single-stream upload, default 1GB
+  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,16 +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"`
-	// ParallelUpload enables experimental parallel composite uploads via a gRPC client for files
-	// bigger than ParallelUploadMinSize, see https://github.com/Altinity/clickhouse-backup/issues/1028.
+	// 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`.
-	ParallelUpload bool `yaml:"parallel_upload" envconfig:"GCS_PARALLEL_UPLOAD"`
-	// ParallelUploadPartSize - size of each part uploaded in parallel, 0 means SDK default 16MiB, minimum 5MiB
-	ParallelUploadPartSize int `yaml:"parallel_upload_part_size" envconfig:"GCS_PARALLEL_UPLOAD_PART_SIZE"`
-	// ParallelUploadMaxConcurrency - how many parts of one file upload in parallel, 0 means SDK default min(4 + NumCPU/2, 16)
-	ParallelUploadMaxConcurrency int `yaml:"parallel_upload_max_concurrency" envconfig:"GCS_PARALLEL_UPLOAD_MAX_CONCURRENCY"`
-	// ParallelUploadMinSize - files smaller than this size use the regular single-stream upload
-	ParallelUploadMinSize int64 `yaml:"parallel_upload_min_size" envconfig:"GCS_PARALLEL_UPLOAD_MIN_SIZE"`
+	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
@@ -618,19 +623,25 @@ func ValidateConfig(cfg *Config) error {
 			}
 		}
 	}
-	if cfg.General.RemoteStorage == "gcs" && cfg.GCS.ParallelUpload {
+	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 `parallel_upload` requires gRPC client and is not compatible with `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 `parallel_upload` requires gRPC client and is not compatible with `force_http`")
+			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 `parallel_upload` is not compatible with `encryption_key`")
+			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)
 	}
@@ -907,10 +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,
-			ParallelUploadMinSize: 1024 * 1024 * 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,25 +387,34 @@ func TestValidateConfigErrors(t *testing.T) {
 			cfg.General.RemoteStorage = "s3"
 			cfg.S3.HTTPIdleConnTimeout = "1parsec"
 		}, "invalid s3 http_idle_conn_timeout"},
-		{"gcs parallel_upload with default config", func(cfg *Config) {
+		{"gcs allow_multipart_upload with default config", func(cfg *Config) {
 			cfg.General.RemoteStorage = "gcs"
-			cfg.GCS.ParallelUpload = true
+			cfg.GCS.AllowMultipartUpload = true
 		}, ""},
-		{"gcs parallel_upload incompatible with endpoint", func(cfg *Config) {
+		{"gcs allow_multipart_upload incompatible with endpoint", func(cfg *Config) {
 			cfg.General.RemoteStorage = "gcs"
-			cfg.GCS.ParallelUpload = true
+			cfg.GCS.AllowMultipartUpload = true
 			cfg.GCS.Endpoint = "http://gcs:8080/storage/v1/"
-		}, "gcs `parallel_upload` requires gRPC client and is not compatible with `endpoint`"},
-		{"gcs parallel_upload incompatible with force_http", func(cfg *Config) {
+		}, "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.ParallelUpload = true
+			cfg.GCS.AllowMultipartUpload = true
 			cfg.GCS.ForceHttp = true
-		}, "gcs `parallel_upload` requires gRPC client and is not compatible with `force_http`"},
-		{"gcs parallel_upload incompatible with encryption_key", func(cfg *Config) {
+		}, "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.ParallelUpload = true
+			cfg.GCS.AllowMultipartUpload = true
 			cfg.GCS.EncryptionKey = "dGVzdC1rZXktdGVzdC1rZXktdGVzdC1rZXktdGVzdA=="
-		}, "gcs `parallel_upload` is not compatible with `encryption_key`"},
+		}, "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"
@@ -240,7 +241,7 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 		return errors.Wrap(err, "GCS Connect storage.NewClient")
 	}
 
-	if gcs.Config.ParallelUpload {
+	if gcs.Config.AllowMultipartUpload {
 		grpcOptions := []option.ClientOption{option.WithTelemetryDisabled()}
 		if credOption != nil {
 			grpcOptions = append(grpcOptions, credOption)
@@ -393,17 +394,75 @@ 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, localSize int64) error {
-	if gcs.grpcClient != nil && localSize >= gcs.Config.ParallelUploadMinSize {
-		return gcs.putFileParallel(ctx, key, r, localSize)
+	if gcs.grpcClient != nil && localSize >= gcs.Config.MultipartUploadMinSize {
+		return gcs.putFileMultipart(ctx, key, r, localSize)
 	}
 	pClientObj, err := gcs.clientPool.BorrowObject(ctx)
 	if err != nil {
@@ -448,18 +507,19 @@ func (gcs *GCS) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser
 	return nil
 }
 
-// putFileParallel uploads a large file with the experimental parallel composite upload,
+// 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 parallel_upload in config validation, so no CSEK handling here.
-func (gcs *GCS) putFileParallel(ctx context.Context, key string, r io.ReadCloser, localSize int64) error {
-	log.Debug().Msgf("GCS->putFileParallel %s, size=%d", key, localSize)
+// 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.ParallelUploadPartSize,
-		MaxConcurrency: gcs.Config.ParallelUploadMaxConcurrency,
+		PartSize:       gcs.Config.ChunkSize,
+		MaxConcurrency: gcs.Config.UploadConcurrency,
 	}
 	writer.StorageClass = gcs.Config.StorageClass
 	if len(gcs.Config.ObjectLabels) > 0 {
@@ -471,12 +531,12 @@ func (gcs *GCS) putFileParallel(ctx context.Context, key string, r io.ReadCloser
 	}
 	buffer := make([]byte, uploadBufferSize)
 	if _, err := io.CopyBuffer(writer, r, buffer); err != nil {
-		log.Warn().Msgf("gcs.putFileParallel: can't copy buffer: %+v", err)
-		return errors.Wrap(err, "GCS putFileParallel CopyBuffer")
+		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.putFileParallel: can't close writer: %+v", err)
-		return errors.Wrap(err, "GCS putFileParallel writer.Close")
+		log.Warn().Msgf("gcs.putFileMultipart: can't close writer: %+v", err)
+		return errors.Wrap(err, "GCS putFileMultipart writer.Close")
 	}
 	return nil
 }

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

@@ -18,8 +18,10 @@ gcs:
   object_disk_path: object_disks/{cluster}/{shard}/{version}
   credentials_file: /etc/clickhouse-backup/credentials.json
   compression_format: tar
-  # experimental parallel composite upload, see https://github.com/Altinity/clickhouse-backup/issues/1028
-  # low thresholds so the ~30MB test archive splits into multiple parallel parts
-  parallel_upload: true
-  parallel_upload_min_size: 1048576
-  parallel_upload_part_size: 5242880
+  # 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