add gcs.parallel_upload for parallel composite uploads of large files

Files bigger than gcs.parallel_upload_min_size (default 1GB) upload to GCS
as multiple parallel parts via a dedicated gRPC client and compose into the
final object, using the experimental EnableParallelUpload writer surface
from cloud.google.com/go/storage v1.62+.

Single-stream uploads of 100GB+ parts were the bottleneck for large-table
backups (12h+), see #1028.

- parallel_upload_part_size (0 = SDK default 16MiB, min 5MiB) and
  parallel_upload_max_concurrency (0 = SDK default min(4+CPU/2, 16))
- rejected together with endpoint, force_http and encryption_key: PCU works
  only via gRPC to storage.googleapis.com and Compose requires the same CSEK
  for source parts and destination
- retry policy set client-level: part uploads create fresh ObjectHandles,
  per-object Retryer does not cover them
- UploadPath now passes the real file size to PutFile so the min_size
  threshold works for uncompressed uploads
- TestGCSParallelUpload runs against the real bucket with GCS_ENCRYPTION_KEY
  unset per command (run.sh exports it globally for CSEK coverage in TestGCS)

Author: Slach

ChangeLog.md

@@ -1,5 +1,6 @@
 # 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 `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,13 @@ 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
+  # 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
   # 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,16 @@ 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.
+	// 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"`
 }
 
 // AzureBlobConfig - Azure Blob settings section
@@ -608,6 +618,19 @@ func ValidateConfig(cfg *Config) error {
 			}
 		}
 	}
+	if cfg.General.RemoteStorage == "gcs" && cfg.GCS.ParallelUpload {
+		// 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`")
+		}
+		if cfg.GCS.ForceHttp {
+			return errors.New("gcs `parallel_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`")
+		}
+	}
 	if cfg.GetCompressionFormat() == "unknown" {
 		return errors.Errorf("'%s' is unknown remote storage", cfg.General.RemoteStorage)
 	}
@@ -884,9 +907,10 @@ 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,
+			ParallelUploadMinSize: 1024 * 1024 * 1024,
 		},
 		COS: COSConfig{
 			RowURL:                 "",

pkg/config/config_test.go

@@ -387,6 +387,25 @@ 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) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.ParallelUpload = true
+		}, ""},
+		{"gcs parallel_upload incompatible with endpoint", func(cfg *Config) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.ParallelUpload = 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) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.ParallelUpload = 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) {
+			cfg.General.RemoteStorage = "gcs"
+			cfg.GCS.ParallelUpload = true
+			cfg.GCS.EncryptionKey = "dGVzdC1rZXktdGVzdC1rZXktdGVzdC1rZXktdGVzdA=="
+		}, "gcs `parallel_upload` is not compatible with `encryption_key`"},
 		{"unknown remote storage", func(cfg *Config) {
 			cfg.General.RemoteStorage = "banana"
 		}, "'banana' is unknown remote storage"},

pkg/storage/gcs.go

@@ -31,6 +31,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 +240,20 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 		return errors.Wrap(err, "GCS Connect storage.NewClient")
 	}
 
+	if gcs.Config.ParallelUpload {
+		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 +272,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")
 	}
@@ -381,7 +401,10 @@ func (gcs *GCS) PutFile(ctx context.Context, key string, r io.ReadCloser, localS
 	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.ParallelUploadMinSize {
+		return gcs.putFileParallel(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 +448,39 @@ 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,
+// 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)
+	obj := gcs.grpcClient.Bucket(gcs.Config.Bucket).Object(key)
+	writer := obj.NewWriter(ctx)
+	writer.EnableParallelUpload = true
+	writer.ParallelUploadConfig = storage.ParallelUploadConfig{
+		PartSize:       gcs.Config.ParallelUploadPartSize,
+		MaxConcurrency: gcs.Config.ParallelUploadMaxConcurrency,
+	}
+	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.putFileParallel: can't copy buffer: %+v", err)
+		return errors.Wrap(err, "GCS putFileParallel 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")
+	}
+	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,25 @@
+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, 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

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",

test/integration/gcs_test.go

@@ -2,7 +2,12 @@
 
 package main
 
-import "testing"
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+)
 
 func TestGCS(t *testing.T) {
 	if isTestShouldSkip("GCS_TESTS") {
@@ -13,3 +18,51 @@ func TestGCS(t *testing.T) {
 	defer env.Cleanup(t, r)
 	env.runMainIntegrationScenario(t, "GCS", "config-gcs.yml")
 }
+
+// TestGCSParallelUpload exercises the experimental parallel composite upload path
+// (gcs.parallel_upload), see https://github.com/Altinity/clickhouse-backup/issues/1028.
+// run.sh exports GCS_ENCRYPTION_KEY for all containers, but CSEK is not compatible with
+// parallel_upload, so every command unsets it explicitly.
+func TestGCSParallelUpload(t *testing.T) {
+	if isTestShouldSkip("GCS_TESTS") {
+		t.Skip("Skipping GCS integration tests...")
+		return
+	}
+	env, r := NewTestEnvironment(t)
+	defer env.Cleanup(t, r)
+	env.connectWithWait(t, r, 500*time.Millisecond, 1500*time.Millisecond, 3*time.Minute)
+
+	backupName := fmt.Sprintf("%s_%d", t.Name(), rand.Int())
+	cfgPath := "/etc/clickhouse-backup/config-gcs-parallel.yml"
+	dbName := "test_gcs_parallel_upload"
+
+	execCmd := func(cmd string) string {
+		out, err := env.DockerExecOut("clickhouse-backup", "bash", "-ce", "GCS_ENCRYPTION_KEY= LOG_LEVEL=debug "+cmd)
+		r.NoError(err, "%s\n%s", cmd, out)
+		return out
+	}
+
+	env.queryWithNoError(t, r, "CREATE DATABASE IF NOT EXISTS "+dbName)
+	env.queryWithNoError(t, r, "CREATE TABLE IF NOT EXISTS "+dbName+".big (key UInt64, value String) ENGINE=MergeTree() ORDER BY key")
+	// ~30MB of incompressible data, so the tar stream exceeds parallel_upload_min_size
+	// and splits into multiple parallel_upload_part_size parts
+	env.queryWithNoError(t, r, "INSERT INTO "+dbName+".big SELECT number, randomString(1000) FROM numbers(30000)")
+
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s create --tables=%s.* %s", cfgPath, dbName, backupName))
+	out := execCmd(fmt.Sprintf("clickhouse-backup -c %s upload %s", cfgPath, backupName))
+	r.Contains(out, "putFileParallel", "upload must go through the parallel composite upload path")
+
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s delete local %s", cfgPath, backupName))
+	env.queryWithNoError(t, r, "DROP DATABASE IF EXISTS "+dbName)
+
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s download %s", cfgPath, backupName))
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s restore %s", cfgPath, backupName))
+
+	var count uint64
+	r.NoError(env.ch.SelectSingleRowNoCtx(&count, "SELECT count() FROM "+dbName+".big"))
+	r.Equal(uint64(30000), count)
+
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s delete local %s", cfgPath, backupName))
+	execCmd(fmt.Sprintf("clickhouse-backup -c %s delete remote %s", cfgPath, backupName))
+	env.queryWithNoError(t, r, "DROP DATABASE IF EXISTS "+dbName)
+}