Simplify GCS parallel upload config

Author: jcechace

packaging/conf/pbm-conf-reference.yml

@@ -140,20 +140,16 @@
 #      clientType: json
 
 ## The size of data chunks in bytes to be uploaded to the storage bucket in a single request.
-## Applies to standard Google SDK uploads only. It is not used when parallelUpload is enabled.
+## For parallel uploads, it is used as the temporary part size. When undefined,
+## PBM uses 10MB for standard uploads and 16MB for parallel uploads.
 #      chunkSize: 10MB
 
-## Experimental Google SDK parallel upload configuration.
-## This is used only with clientType: grpc. It creates temporary GCS objects and composes
-## them into the final object. The credentials must be allowed to delete temporary objects
-## for cleanup. Failed or interrupted uploads can leave temporary objects behind, so a
-## bucket lifecycle cleanup rule is recommended.
-#      parallelUpload:
-#        enabled: false
-## Size of each temporary part in bytes. When undefined or 0, the Google SDK default is used.
-#        partSize: 16MB
-## Number of parallel upload workers. When undefined or 0, the Google SDK default is used.
-#        maxConcurrency: 4
+## Experimental Google SDK parallel upload concurrency. Parallel upload is used only
+## with clientType: grpc and parallelUploadConcurrency greater than 1. It creates
+## temporary GCS objects and composes them into the final object. The credentials must
+## be allowed to delete temporary objects for cleanup. Failed or interrupted uploads can
+## leave temporary objects behind, so a bucket lifecycle cleanup rule is recommended.
+#      parallelUploadConcurrency: 4
 
 ## GCS access credentials. You can use either HMAC or Service Account keys.
 ## HMAC Example:
@@ -173,7 +169,7 @@
 #        backoffInitial: 1s
 #        backoffMax: 30s
 #        backoffMultiplier: 2
-## Applies to standard Google SDK uploads only. It is not used when parallelUpload is enabled.
+## Applies to standard Google SDK uploads only. It is not used for parallel uploads.
 #        chunkRetryDeadline: 32s
 #
 ## The maximum object size that will be stored on the storage

pbm/config/config_test.go

@@ -406,9 +406,10 @@ func TestConfig(t *testing.T) {
 						ClientEmail: "ce1",
 						PrivateKey:  "pk1",
 					},
-					ClientType:   gcs.ClientTypeJSON,
-					ChunkSize:    100,
-					MaxObjSizeGB: floatPtr(1.1),
+					ClientType:                gcs.ClientTypeJSON,
+					ChunkSize:                 100,
+					ParallelUploadConcurrency: 4,
+					MaxObjSizeGB:              floatPtr(1.1),
 					Retryer: &gcs.Retryer{
 						BackoffInitial:     11 * time.Minute,
 						BackoffMax:         111 * time.Minute,
@@ -450,6 +451,11 @@ func TestConfig(t *testing.T) {
 				param: "storage.gcs.chunkSize",
 				val:   fmt.Sprintf("%d", wantCfg.Storage.GCS.ChunkSize),
 			},
+			{
+				desc:  "parallelUploadConcurrency",
+				param: "storage.gcs.parallelUploadConcurrency",
+				val:   fmt.Sprintf("%d", wantCfg.Storage.GCS.ParallelUploadConcurrency),
+			},
 			{
 				desc:  "maxObjSizeGB",
 				param: "storage.gcs.maxObjSizeGB",
@@ -533,18 +539,8 @@ func TestConfig(t *testing.T) {
 				val:   string(gcs.ClientTypeGRPC),
 			},
 			{
-				desc:  "parallelUpload.enabled",
-				param: "storage.gcs.parallelUpload.enabled",
-				val:   "true",
-			},
-			{
-				desc:  "parallelUpload.partSize",
-				param: "storage.gcs.parallelUpload.partSize",
-				val:   "123",
-			},
-			{
-				desc:  "parallelUpload.maxConcurrency",
-				param: "storage.gcs.parallelUpload.maxConcurrency",
+				desc:  "parallelUploadConcurrency",
+				param: "storage.gcs.parallelUploadConcurrency",
 				val:   "4",
 			},
 		}
@@ -568,17 +564,8 @@ func TestConfig(t *testing.T) {
 		if gotGCS.ClientType != gcs.ClientTypeGRPC {
 			t.Fatalf("clientType: got=%q, want=%q", gotGCS.ClientType, gcs.ClientTypeGRPC)
 		}
-		if gotGCS.ParallelUpload == nil {
-			t.Fatal("expected parallelUpload config")
-		}
-		if !gotGCS.ParallelUpload.Enabled {
-			t.Fatal("expected parallelUpload.enabled=true")
-		}
-		if gotGCS.ParallelUpload.PartSize != 123 {
-			t.Fatalf("parallelUpload.partSize: got=%d, want=123", gotGCS.ParallelUpload.PartSize)
-		}
-		if gotGCS.ParallelUpload.MaxConcurrency != 4 {
-			t.Fatalf("parallelUpload.maxConcurrency: got=%d, want=4", gotGCS.ParallelUpload.MaxConcurrency)
+		if gotGCS.ParallelUploadConcurrency != 4 {
+			t.Fatalf("parallelUploadConcurrency: got=%d, want=4", gotGCS.ParallelUploadConcurrency)
 		}
 	})
 

pbm/storage/gcs/config.go

@@ -25,31 +25,13 @@ type Config struct {
 
 	// The maximum number of bytes that the Writer will attempt to send in a single request.
 	// https://pkg.go.dev/cloud.google.com/go/storage#Writer
-	ChunkSize    int      `bson:"chunkSize,omitempty" json:"chunkSize,omitempty" yaml:"chunkSize,omitempty"`
-	MaxObjSizeGB *float64 `bson:"maxObjSizeGB,omitempty" json:"maxObjSizeGB,omitempty" yaml:"maxObjSizeGB,omitempty"`
-
-	ParallelUpload *ParallelUpload `bson:"parallelUpload,omitempty" json:"parallelUpload,omitempty" yaml:"parallelUpload,omitempty"`
+	ChunkSize                 int      `bson:"chunkSize,omitempty" json:"chunkSize,omitempty" yaml:"chunkSize,omitempty"`
+	ParallelUploadConcurrency int      `bson:"parallelUploadConcurrency,omitempty" json:"parallelUploadConcurrency,omitempty" yaml:"parallelUploadConcurrency,omitempty"`
+	MaxObjSizeGB              *float64 `bson:"maxObjSizeGB,omitempty" json:"maxObjSizeGB,omitempty" yaml:"maxObjSizeGB,omitempty"`
 
 	Retryer *Retryer `bson:"retryer,omitempty" json:"retryer,omitempty" yaml:"retryer,omitempty"`
 }
 
-// ParallelUpload configures the experimental Google Cloud Storage parallel upload feature.
-//
-// Parallel uploads are supported only by the Google SDK gRPC client.
-// HMAC credentials continue to use the MinIO client and are not affected by this setting.
-type ParallelUpload struct {
-	// Enabled enables Writer parallel uploads when clientType is grpc.
-	Enabled bool `bson:"enabled,omitempty" json:"enabled,omitempty" yaml:"enabled,omitempty"`
-
-	// PartSize is the size of each temporary part in bytes.
-	// If unset, PBM uses the computed ChunkSize for the uploaded object.
-	PartSize int `bson:"partSize,omitempty" json:"partSize,omitempty" yaml:"partSize,omitempty"`
-
-	// MaxConcurrency is the number of goroutines used to upload parts in parallel.
-	// If unset, the Google SDK chooses a value based on the number of CPUs.
-	MaxConcurrency int `bson:"maxConcurrency,omitempty" json:"maxConcurrency,omitempty" yaml:"maxConcurrency,omitempty"`
-}
-
 //nolint:lll
 type Credentials struct {
 	// Workload Identity Federation (allows missing JSON/HMAC credentials)
@@ -100,10 +82,6 @@ func (cfg *Config) Clone() *Config {
 		v := *cfg.MaxObjSizeGB
 		rv.MaxObjSizeGB = &v
 	}
-	if cfg.ParallelUpload != nil {
-		v := *cfg.ParallelUpload
-		rv.ParallelUpload = &v
-	}
 	if cfg.Retryer != nil {
 		v := *cfg.Retryer
 		rv.Retryer = &v
@@ -129,10 +107,10 @@ func (cfg *Config) Equal(other *Config) bool {
 	if cfg.ChunkSize != other.ChunkSize {
 		return false
 	}
-	if !reflect.DeepEqual(cfg.MaxObjSizeGB, other.MaxObjSizeGB) {
+	if cfg.ParallelUploadConcurrency != other.ParallelUploadConcurrency {
 		return false
 	}
-	if !reflect.DeepEqual(cfg.ParallelUpload, other.ParallelUpload) {
+	if !reflect.DeepEqual(cfg.MaxObjSizeGB, other.MaxObjSizeGB) {
 		return false
 	}
 	if !reflect.DeepEqual(cfg.Credentials, other.Credentials) {
@@ -173,17 +151,11 @@ func (cfg *Config) Cast() error {
 		return errors.Errorf("invalid clientType %q", cfg.ClientType)
 	}
 
-	if cfg.ChunkSize == 0 {
-		cfg.ChunkSize = defaultChunkSize
+	if cfg.ChunkSize == 0 && cfg.parallelUploadEnabled() {
+		cfg.ChunkSize = defaultParallelUploadChunkSize
 	}
-
-	if cfg.ParallelUpload != nil {
-		if cfg.ParallelUpload.PartSize < 0 {
-			return errors.New("parallelUpload.partSize cannot be negative")
-		}
-		if cfg.ParallelUpload.MaxConcurrency < 0 {
-			return errors.New("parallelUpload.maxConcurrency cannot be negative")
-		}
+	if cfg.ChunkSize == 0 && !cfg.parallelUploadEnabled() {
+		cfg.ChunkSize = defaultChunkSize
 	}
 
 	if cfg.Retryer == nil {
@@ -223,5 +195,5 @@ func (cfg *Config) GetMaxObjSizeGB() float64 {
 }
 
 func (cfg *Config) parallelUploadEnabled() bool {
-	return cfg.ClientType == ClientTypeGRPC && cfg.ParallelUpload != nil && cfg.ParallelUpload.Enabled
+	return cfg.ClientType == ClientTypeGRPC && cfg.ParallelUploadConcurrency > 1
 }

pbm/storage/gcs/config_test.go

@@ -37,22 +37,12 @@ func TestCast(t *testing.T) {
 		t.Fatalf("wrong config after Cast, diff=%s", cmp.Diff(*c, *want))
 	}
 
-	t.Run("parallel upload validation", func(t *testing.T) {
+	t.Run("validation", func(t *testing.T) {
 		for _, tt := range []struct {
 			name string
 			cfg  *Config
 			err  string
 		}{
-			{
-				name: "negative part size",
-				cfg:  &Config{ParallelUpload: &ParallelUpload{PartSize: -1}},
-				err:  "parallelUpload.partSize cannot be negative",
-			},
-			{
-				name: "negative max concurrency",
-				cfg:  &Config{ParallelUpload: &ParallelUpload{MaxConcurrency: -1}},
-				err:  "parallelUpload.maxConcurrency cannot be negative",
-			},
 			{
 				name: "invalid client type",
 				cfg:  &Config{ClientType: ClientType("xml")},
@@ -67,6 +57,35 @@ func TestCast(t *testing.T) {
 			})
 		}
 	})
+
+	t.Run("negative parallel upload concurrency is disabled", func(t *testing.T) {
+		c := &Config{
+			ClientType:                ClientTypeGRPC,
+			ParallelUploadConcurrency: -1,
+		}
+		err := c.Cast()
+		if err != nil {
+			t.Fatalf("got error during Cast: %v", err)
+		}
+		if c.ChunkSize != defaultChunkSize {
+			t.Fatalf("wrong chunk size: got %d, want %d", c.ChunkSize, defaultChunkSize)
+		}
+	})
+
+	t.Run("parallel upload default chunk size", func(t *testing.T) {
+		c := &Config{
+			ClientType:                ClientTypeGRPC,
+			ParallelUploadConcurrency: 2,
+		}
+		err := c.Cast()
+		if err != nil {
+			t.Fatalf("got error during Cast: %v", err)
+		}
+		if c.ChunkSize != defaultParallelUploadChunkSize {
+			t.Fatalf("wrong parallel upload chunk size: got %d, want %d",
+				c.ChunkSize, defaultParallelUploadChunkSize)
+		}
+	})
 }
 
 func TestConfig(t *testing.T) {
@@ -77,12 +96,8 @@ func TestConfig(t *testing.T) {
 			ClientEmail: "email@example.com",
 			PrivateKey:  "-----BEGIN PRIVATE KEY-----\nKey\n-----END PRIVATE KEY-----\n",
 		},
-		ClientType: ClientTypeGRPC,
-		ParallelUpload: &ParallelUpload{
-			Enabled:        true,
-			PartSize:       16 * 1024 * 1024,
-			MaxConcurrency: 4,
-		},
+		ClientType:                ClientTypeGRPC,
+		ParallelUploadConcurrency: 4,
 	}
 
 	t.Run("Clone", func(t *testing.T) {
@@ -95,11 +110,11 @@ func TestConfig(t *testing.T) {
 			t.Error("expected clone to be equal")
 		}
 
-		opts.ParallelUpload.MaxConcurrency = 8
+		opts.ParallelUploadConcurrency = 8
 		if opts.Equal(clone) {
 			t.Error("expected clone to be unchanged when updating original")
 		}
-		opts.ParallelUpload.MaxConcurrency = 4
+		opts.ParallelUploadConcurrency = 4
 
 		opts.Bucket = "updatedName"
 		if opts.Equal(clone) {
@@ -132,9 +147,9 @@ func TestConfig(t *testing.T) {
 		}
 
 		clone = opts.Clone()
-		clone.ParallelUpload.Enabled = false
+		clone.ParallelUploadConcurrency = 8
 		if opts.Equal(clone) {
-			t.Error("expected not to be equal when updating parallel upload")
+			t.Error("expected not to be equal when updating parallel upload concurrency")
 		}
 	})
 

pbm/storage/gcs/gcs.go

@@ -14,8 +14,9 @@ import (
 const (
 	gcsEndpointURL = "storage.googleapis.com"
 
-	defaultChunkSize    = 10 * 1024 * 1024 // 10MiB
-	defaultMaxObjSizeGB = 5018             // 4.9 TB
+	defaultChunkSize               = 10 * 1024 * 1024 // 10MiB
+	defaultParallelUploadChunkSize = 16 * 1024 * 1024 // 16MiB, matches Google SDK PCU default
+	defaultMaxObjSizeGB            = 5018             // 4.9 TB
 
 	defaultMaxAttempts        = 5
 	defaultBackoffInitial     = time.Second

pbm/storage/gcs/google_client.go

@@ -55,7 +55,9 @@ func newGoogleClient(cfg *Config, l log.LogEvent) (*googleClient, error) {
 			return nil, errors.Wrap(merr, "marshal GCS credentials")
 		}
 		if cfg.ClientType == ClientTypeGRPC {
-			cli, err = storagegcs.NewGRPCClient(ctx, option.WithCredentialsJSON(creds))
+			cli, err = storagegcs.NewGRPCClient(ctx,
+				option.WithCredentialsJSON(creds),
+				storagegcs.WithDisabledClientMetrics())
 		} else {
 			cli, err = storagegcs.NewClient(ctx, option.WithCredentialsJSON(creds))
 		}
@@ -71,7 +73,7 @@ func newGoogleClient(cfg *Config, l log.LogEvent) (*googleClient, error) {
 			return nil, fmt.Errorf("validate default credential type: %w", adcErr)
 		}
 		if cfg.ClientType == ClientTypeGRPC {
-			cli, err = storagegcs.NewGRPCClient(ctx)
+			cli, err = storagegcs.NewGRPCClient(ctx, storagegcs.WithDisabledClientMetrics())
 		} else {
 			cli, err = storagegcs.NewClient(ctx)
 		}
@@ -159,28 +161,18 @@ func (g googleClient) save(name string, data io.Reader, options ...storage.Optio
 	ctx := context.Background()
 	w := g.bucketHandle.Object(path.Join(g.cfg.Prefix, name)).NewWriter(ctx)
 	if g.cfg.parallelUploadEnabled() {
-		pu := g.cfg.ParallelUpload
 		if g.log != nil && opts.UseLogger {
-			partSize := "SDK default"
-			if pu.PartSize > 0 {
-				partSize = fmt.Sprintf("%v (%v)", pu.PartSize, storage.PrettySize(int64(pu.PartSize)))
-			}
-
-			concurrency := "SDK default"
-			if pu.MaxConcurrency > 0 {
-				concurrency = fmt.Sprintf("%d", pu.MaxConcurrency)
-			}
-
-			g.log.Debug(`uploading %q [size hint: %v (%v); parallel upload part size: %s; concurrency: %s]`,
+			g.log.Debug(`uploading %q [size hint: %v (%v); parallel upload part size: %v (%v); concurrency: %d]`,
 				name,
 				opts.Size, storage.PrettySize(opts.Size),
-				partSize, concurrency)
+				g.cfg.ChunkSize, storage.PrettySize(int64(g.cfg.ChunkSize)),
+				g.cfg.ParallelUploadConcurrency)
 		}
 
 		w.EnableParallelUpload = true
 		w.ParallelUploadConfig = storagegcs.ParallelUploadConfig{
-			PartSize:       pu.PartSize,
-			MaxConcurrency: pu.MaxConcurrency,
+			PartSize:       g.cfg.ChunkSize,
+			MaxConcurrency: g.cfg.ParallelUploadConcurrency,
 		}
 	} else {
 		const align int64 = 256 << 10 // 256 KiB (both min size and alignment)