From 48e7d86436ec4f8fcadcf77932254c964b4df596 Mon Sep 17 00:00:00 2001 From: Stan Hu Date: Tue, 1 Sep 2026 11:58:49 -0700 Subject: [PATCH 1/2] blob/gcsblob: support the gRPC API and Rapid Storage (zonal) buckets Rapid Storage (zonal) buckets cannot be used through this driver at all. Every write is rejected, on both transports: gs://bucket?grpc=true InvalidArgument: This bucket type only supports appendable objects gs://bucket googleapi: Error 400: This bucket requires appendable objects An appendable object is uploaded over a bidirectional stream. It becomes visible as soon as the first bytes are flushed and stays open for further writes until something finalizes it. Ordinary uploads, one-shot or resumable, produce an object only once the whole payload has been sent. Zonal buckets support the appendable form and nothing else, which is why both transports reject a normal write. Two things are needed. experimental.WithZonalBucketAPIs makes ObjectHandle.NewWriter default Writer.Append to true, selecting the appendable upload path, and switches reads to the bidirectional API. Writer.FinalizeOnClose then makes Close finalize the object; without it Close leaves an unfinalized object exposing only whatever prefix was flushed, which for a small payload is a zero-length object even though Close reported no error. Reads already worked over both transports without any change. The new API is a Dial plus a constructor, following secrets/gcpkms: func DialGRPC(ctx context.Context, ts gcp.TokenSource, opts ...option.ClientOption) (*storage.Client, func(), error) func OpenBucketGRPC(client *storage.Client, bucketName string, opts *Options) (*blob.Bucket, error) c, cleanup, err := gcsblob.DialGRPC(ctx, ts, experimental.WithZonalBucketAPIs()) defer cleanup() b, err := gcsblob.OpenBucketGRPC(c, "my-rapid-bucket", nil) OpenBucketGRPC takes no context, matching gcpkms.OpenKeeper: once a client exists there is nothing left to cancel. Options.Client, which has accepted a *storage.Client since v0.44.0, keeps working. For callers whose whole configuration is a URL string and who therefore cannot supply a client, the URL opener grows two parameters. grpc=true selects the gRPC transport; zonal=true additionally enables the zonal APIs and implies grpc=true. Combining zonal=true with an explicit grpc=false is a contradiction and is rejected rather than silently overridden. URLOpener gains a TokenSource, populated by lazyCredsOpener from the credentials it already resolves, because a gRPC client cannot reuse an HTTP one. A URL that asks for gRPC without a token source and without anonymous=true is now an error instead of quietly producing an unauthenticated client. Buckets that build their own client close it in Close, which was previously a no-op. A gRPC client owns a connection pool, so otherwise every OpenBucketURL leaked one for the process lifetime. Clients passed in by the caller are left alone. FinalizeOnClose is set unconditionally in NewTypedWriter. The storage library reads it only on the appendable write path, which pickBufferSender selects solely when Writer.Append is set, so the JSON/HTTP and plain gRPC paths are unaffected. Emulator support on the gRPC path is left to storage.NewGRPCClient. Reusing STORAGE_EMULATOR_HOST would not work: it is the HTTP endpoint, and a local emulator needs a separate port for gRPC. Passing it to option.WithEndpoint alongside option.WithoutAuthentication would also still dial over TLS, since skipping credentials does not make the transport plaintext. defaultGRPCOptions already reads STORAGE_EMULATOR_HOST_GRPC, strips the scheme, dials insecurely and disables client metrics, and those defaults are merged ahead of caller-supplied options. lazyCredsOpener checks that variable too, so pointing only the gRPC one at an emulator does not trigger an Application Default Credentials lookup. gRPC is not a speedup on its own. Measured from an n2-standard-4 in us-central1-c, 1 KiB objects each read exactly once, n=1000, p50: on a NAM4 dual-region bucket 35.5ms over gRPC against 35.4ms over JSON, and on a US multi-region bucket 59.2ms against 61.2ms. The win comes from the zonal bucket, at 12.0ms. The UseGRPC docs say to measure first. --- blob/gcsblob/gcsblob.go | 245 +++++++++++++++++++++++++++++++---- blob/gcsblob/gcsblob_test.go | 133 ++++++++++++++++++- 2 files changed, 354 insertions(+), 24 deletions(-) diff --git a/blob/gcsblob/gcsblob.go b/blob/gcsblob/gcsblob.go index e282ad2258..c204869ca1 100644 --- a/blob/gcsblob/gcsblob.go +++ b/blob/gcsblob/gcsblob.go @@ -75,6 +75,7 @@ import ( "cloud.google.com/go/compute/metadata" "cloud.google.com/go/storage" + "cloud.google.com/go/storage/experimental" "github.com/google/wire" "golang.org/x/oauth2/google" "google.golang.org/api/googleapi" @@ -155,7 +156,9 @@ func (o *lazyCredsOpener) OpenBucketURL(ctx context.Context, u *url.URL) (*blob. o.init.Do(func() { var opts Options var creds *google.Credentials - if os.Getenv("STORAGE_EMULATOR_HOST") != "" { + // STORAGE_EMULATOR_HOST_GRPC is the gRPC equivalent; a local emulator + // needs separate ports for HTTP and gRPC, so either may be set. + if os.Getenv("STORAGE_EMULATOR_HOST") != "" || os.Getenv("STORAGE_EMULATOR_HOST_GRPC") != "" { creds, _ = google.CredentialsFromJSON(ctx, []byte(`{"type": "service_account", "project_id": "my-project-id"}`)) } else { var err error @@ -205,7 +208,9 @@ func (o *lazyCredsOpener) OpenBucketURL(ctx context.Context, u *url.URL) (*blob. o.err = err return } - o.opener = &URLOpener{Client: client, Options: opts} + // TokenSource is needed for grpc=true and zonal=true URLs, which build a + // separate gRPC client that cannot reuse the HTTP one. + o.opener = &URLOpener{Client: client, TokenSource: gcp.CredentialsTokenSource(creds), Options: opts} }) if o.err != nil { return nil, fmt.Errorf("open bucket %v: %w", u, o.err) @@ -228,47 +233,132 @@ const Scheme = "gs" // a value of "-" forces the use of an unauthenticated client. // - private_key_path: Path to read for Options.PrivateKey; only used in SignedURL. // - universe_domain: Sets the universe domain for the client. +// - grpc: A value of "true" uses the Cloud Storage gRPC API instead of the +// JSON/HTTP API. The bucket is opened with a client from DialGRPC. +// - zonal: A value of "true" additionally enables the zonal bucket APIs, which +// are required for Rapid Storage buckets. Implies grpc=true. +// +// gRPC is not automatically faster than JSON/HTTP. For small-object latency +// against regional, dual-region and multi-region buckets it measured no better, +// so prefer the default unless you have measured a win for your workload. type URLOpener struct { // Client must be set to a non-nil HTTP client authenticated with // Cloud Storage scope or equivalent (unless anonymous=true). Client *gcp.HTTPClient + // TokenSource is used to authenticate the gRPC client built for URLs with + // grpc=true or zonal=true. The gRPC API cannot reuse an HTTP client, so + // Client is not enough on its own. It is not needed for URLs that use the + // default JSON/HTTP API, nor for anonymous=true. + TokenSource gcp.TokenSource + // Options specifies the default options to pass to OpenBucket. Options Options } // OpenBucketURL opens the GCS bucket with the same name as the URL's host. func (o *URLOpener) OpenBucketURL(ctx context.Context, u *url.URL) (*blob.Bucket, error) { - opts, client, err := o.forParams(ctx, u.Query()) + opts, client, params, err := o.forParams(ctx, u.Query()) if err != nil { return nil, fmt.Errorf("open bucket %v: %w", u, err) } - return OpenBucket(ctx, client, u.Host, opts) + if !params.useGRPC { + return OpenBucket(ctx, client, u.Host, opts) + } + + // The gRPC API cannot reuse an HTTP client, so build a dedicated client from + // the same credentials. The bucket owns it and closes it, since nothing else + // holds a reference. + if opts.Client != nil { + return nil, fmt.Errorf("open bucket %v: Options.Client cannot be combined with grpc=true or zonal=true", u) + } + var clientOpts []option.ClientOption + if params.useZonal { + clientOpts = append(clientOpts, experimental.WithZonalBucketAPIs()) + } + clientOpts = append(clientOpts, opts.ClientOptions...) + ts := o.TokenSource + if params.anonymous { + ts = nil + } else if ts == nil { + return nil, fmt.Errorf("open bucket %v: URLOpener.TokenSource is required for grpc=true or zonal=true", u) + } + gc, cleanup, err := DialGRPC(ctx, ts, clientOpts...) + if err != nil { + return nil, fmt.Errorf("open bucket %v: %w", u, err) + } + opts.Client = gc + drv, err := openBucket(ctx, nil, u.Host, opts) + if err != nil { + cleanup() + return nil, err + } + drv.ownedClient = gc + return blob.NewBucket(drv), nil } -func (o *URLOpener) forParams(ctx context.Context, q url.Values) (*Options, *gcp.HTTPClient, error) { +// urlParams holds the transport choices parsed out of a bucket URL. +type urlParams struct { + // useGRPC is set by grpc=true, and implied by zonal=true. + useGRPC bool + // useZonal is set by zonal=true. + useZonal bool + // anonymous is set when the URL asked for an unauthenticated client, via + // anonymous=true or access_id=-. + anonymous bool +} + +func (o *URLOpener) forParams(ctx context.Context, q url.Values) (*Options, *gcp.HTTPClient, urlParams, error) { + var params urlParams for k := range q { - if k != "access_id" && k != "private_key_path" && k != "anonymous" && k != "universe_domain" { - return nil, nil, fmt.Errorf("invalid query parameter %q", k) + if k != "access_id" && k != "private_key_path" && k != "anonymous" && k != "universe_domain" && k != "grpc" && k != "zonal" { + return nil, nil, params, fmt.Errorf("invalid query parameter %q", k) } } opts := new(Options) *opts = o.Options client := o.Client + grpcInURL := false + if g := q.Get("grpc"); g != "" { + useGRPC, err := strconv.ParseBool(g) + if err != nil { + return nil, nil, params, fmt.Errorf("invalid value %q for query parameter \"grpc\": %w", g, err) + } + params.useGRPC = useGRPC + grpcInURL = true + } + if z := q.Get("zonal"); z != "" { + useZonal, err := strconv.ParseBool(z) + if err != nil { + return nil, nil, params, fmt.Errorf("invalid value %q for query parameter \"zonal\": %w", z, err) + } + params.useZonal = useZonal + } + // The zonal APIs exist only on the gRPC client, so zonal=true implies + // grpc=true. Asking for both zonal=true and grpc=false is a contradiction, + // so report it instead of picking a winner. + if params.useZonal { + if grpcInURL && !params.useGRPC { + return nil, nil, params, errors.New("query parameter \"zonal=true\" cannot be combined with \"grpc=false\"") + } + params.useGRPC = true + } if anon := q.Get("anonymous"); anon != "" { isAnon, err := strconv.ParseBool(anon) if err != nil { - return nil, nil, fmt.Errorf("invalid value %q for query parameter \"anonymous\": %w", anon, err) + return nil, nil, params, fmt.Errorf("invalid value %q for query parameter \"anonymous\": %w", anon, err) } if isAnon { opts.clear() client = gcp.NewAnonymousHTTPClient(gcp.DefaultTransport()) + params.anonymous = true } } if accessID := q.Get("access_id"); accessID != "" && accessID != opts.GoogleAccessID { opts.clear() if accessID == "-" { client = gcp.NewAnonymousHTTPClient(gcp.DefaultTransport()) + params.anonymous = true } else { opts.GoogleAccessID = accessID } @@ -276,7 +366,7 @@ func (o *URLOpener) forParams(ctx context.Context, q url.Values) (*Options, *gcp if keyPath := q.Get("private_key_path"); keyPath != "" { pk, err := os.ReadFile(keyPath) if err != nil { - return nil, nil, err + return nil, nil, params, err } opts.PrivateKey = pk } else if _, exists := q["private_key_path"]; exists { @@ -285,7 +375,7 @@ func (o *URLOpener) forParams(ctx context.Context, q url.Values) (*Options, *gcp // is intentional such as for tests or involving a key stored in a HSM/TPM. opts.PrivateKey = nil } - return opts, client, nil + return opts, client, params, nil } // Options sets options for constructing a *blob.Bucket backed by GCS. @@ -315,7 +405,9 @@ type Options struct { // Client provides a *storage.Client to use, instead of constructing one based on // the HTTPClient. When set, you must pass nil as the gcp.HTTPClient to OpenBucket. // - // For example, this can be used to create a Bucket backed by a gRPC client. + // Use this to open a bucket over the Cloud Storage gRPC API, including a + // Rapid Storage (zonal) bucket; see DialGRPC. The caller owns the client and + // is responsible for closing it. Client *storage.Client // ClientOptions are passed when constructing the storage.Client. @@ -352,25 +444,70 @@ func openBucket(ctx context.Context, client *gcp.HTTPClient, bucketName string, return nil, errors.New("gcsblob.OpenBucket: client is required") } - // We wrap the provided http.Client to add a Go CDK User-Agent. - clientOpts := []option.ClientOption{option.WithHTTPClient(useragent.HTTPClient(&client.Client, "blob"))} - if host := os.Getenv("STORAGE_EMULATOR_HOST"); host != "" { - clientOpts = []option.ClientOption{ + clientOpts := append(httpClientOptions(client, os.Getenv("STORAGE_EMULATOR_HOST")), opts.ClientOptions...) + c, err := storage.NewClient(ctx, clientOpts...) + if err != nil { + return nil, err + } + return &bucket{name: bucketName, client: c, opts: opts}, nil +} + +// httpClientOptions returns the client options for the JSON/HTTP storage API. +func httpClientOptions(client *gcp.HTTPClient, emulatorHost string) []option.ClientOption { + if emulatorHost != "" { + return []option.ClientOption{ option.WithoutAuthentication(), - option.WithEndpoint("http://" + host + "/storage/v1/"), + option.WithEndpoint("http://" + emulatorHost + "/storage/v1/"), option.WithHTTPClient(http.DefaultClient), } } - clientOpts = append(clientOpts, opts.ClientOptions...) - c, err := storage.NewClient(ctx, clientOpts...) + // We wrap the provided http.Client to add a Go CDK User-Agent. + return []option.ClientOption{option.WithHTTPClient(useragent.HTTPClient(&client.Client, "blob"))} +} + +// DialGRPC returns a *storage.Client that talks to Cloud Storage over gRPC +// instead of the default JSON/HTTP API. Pass it as Options.Client, along with a +// nil gcp.HTTPClient, to open a bucket that uses it: +// +// c, cleanup, err := gcsblob.DialGRPC(ctx, gcp.CredentialsTokenSource(creds)) +// if err != nil { ... } +// defer cleanup() +// b, err := gcsblob.OpenBucket(ctx, nil, "my-bucket", &gcsblob.Options{Client: c}) +// +// Rapid Storage (zonal) buckets accept only appendable object uploads, and +// reject every other write. They need the zonal bucket APIs: +// +// c, cleanup, err := gcsblob.DialGRPC(ctx, ts, experimental.WithZonalBucketAPIs()) +// +// Do not pass that option for other bucket types, which reject appendable +// uploads in turn. +// +// A nil ts returns an unauthenticated client. The caller owns the returned +// client and must call the returned clean-up function when done with it. +// +// Emulator support is left to storage.NewGRPCClient, which reads +// STORAGE_EMULATOR_HOST_GRPC rather than STORAGE_EMULATOR_HOST, since a local +// emulator needs separate ports for HTTP and gRPC. It also strips the scheme +// from the host and dials insecurely, neither of which could be done correctly +// from here. +func DialGRPC(ctx context.Context, ts gcp.TokenSource, opts ...option.ClientOption) (*storage.Client, func(), error) { + clientOpts := []option.ClientOption{useragent.ClientOption("blob")} + if ts == nil { + clientOpts = append(clientOpts, option.WithoutAuthentication()) + } else { + clientOpts = append(clientOpts, option.WithTokenSource(ts)) + } + clientOpts = append(clientOpts, opts...) + c, err := storage.NewGRPCClient(ctx, clientOpts...) if err != nil { - return nil, err + return nil, nil, err } - return &bucket{name: bucketName, client: c, opts: opts}, nil + return c, func() { _ = c.Close() }, nil } -// OpenBucket returns a *blob.Bucket backed by an existing GCS bucket. See the -// package documentation for an example. +// OpenBucket returns a *blob.Bucket backed by an existing GCS bucket, using the +// JSON/HTTP API. For the gRPC API, including Rapid Storage (zonal) buckets, see +// OpenBucketGRPC. See the package documentation for an example. func OpenBucket(ctx context.Context, client *gcp.HTTPClient, bucketName string, opts *Options) (*blob.Bucket, error) { drv, err := openBucket(ctx, client, bucketName, opts) if err != nil { @@ -379,12 +516,54 @@ func OpenBucket(ctx context.Context, client *gcp.HTTPClient, bucketName string, return blob.NewBucket(drv), nil } +// OpenBucketGRPC returns a *blob.Bucket backed by an existing GCS bucket, using +// a client that talks to Cloud Storage over gRPC rather than the default +// JSON/HTTP API. Use DialGRPC to construct client. +// +// This is the entry point for Rapid Storage (zonal) buckets, which reject every +// write over JSON/HTTP: +// +// c, cleanup, err := gcsblob.DialGRPC(ctx, ts, experimental.WithZonalBucketAPIs()) +// if err != nil { ... } +// defer cleanup() +// b, err := gcsblob.OpenBucketGRPC(c, "my-rapid-bucket", nil) +// +// The caller owns client. Closing the returned bucket does not close it. +// +// It is an error to also set Options.Client. Options.ClientOptions is ignored, +// since those options are applied when client is constructed. +func OpenBucketGRPC(client *storage.Client, bucketName string, opts *Options) (*blob.Bucket, error) { + if client == nil { + return nil, errors.New("gcsblob.OpenBucketGRPC: client is required") + } + if opts == nil { + opts = &Options{} + } + if opts.Client != nil { + return nil, errors.New("gcsblob.OpenBucketGRPC: Options.Client must be nil; pass the client as the first argument") + } + o := *opts + o.Client = client + // ctx is unused once a client exists, so it is not part of the signature; + // compare secrets/gcpkms.OpenKeeper. + drv, err := openBucket(context.Background(), nil, bucketName, &o) + if err != nil { + return nil, err + } + return blob.NewBucket(drv), nil +} + // bucket represents a GCS bucket, which handles read, write and delete operations // on objects within it. type bucket struct { name string client *storage.Client - opts *Options + // ownedClient is non-nil when this bucket created client itself and must + // close it. That happens only for URLs that select the gRPC API, where the + // client owns a connection pool that would otherwise leak. Clients supplied + // through Options.Client belong to the caller and are left alone. + ownedClient *storage.Client + opts *Options } // reader reads a GCS object. It implements driver.Reader. @@ -438,6 +617,9 @@ func (b *bucket) ErrorCode(err error) gcerrors.ErrorCode { } func (b *bucket) Close() error { + if b.ownedClient != nil { + return b.ownedClient.Close() + } return nil } @@ -672,6 +854,23 @@ func (b *bucket) NewTypedWriter(ctx context.Context, key, contentType string, op w.Metadata = opts.Metadata w.MD5 = opts.ContentMD5 w.ForceEmptyContentType = opts.DisableContentTypeDetection + // Zonal buckets upload with "appendable object" semantics: the object + // appears as soon as the first bytes are flushed and stays open for + // more writes until someone finalizes it. Closing the Writer does not + // finalize it by default, which leaves behind an object exposing only + // whatever prefix had been flushed. For a write small enough to fit in + // one buffer that is a zero-length object, even though Close returned + // no error. Finalize on Close so that a closed blob.Writer always + // leaves a complete object, as the blob API promises. + // + // The storage client reads this field only on the appendable write + // path, so it does nothing over JSON/HTTP or plain gRPC. Setting it + // unconditionally is deliberate: it also covers callers who turn on + // appendable uploads by passing experimental.WithZonalBucketAPIs + // through Options.ClientOptions rather than setting UseZonalAPIs. + // Callers who do want an object left open for later appends can set + // this back to false from WriterOptions.BeforeWrite. + w.FinalizeOnClose = true return w } diff --git a/blob/gcsblob/gcsblob_test.go b/blob/gcsblob/gcsblob_test.go index 3b1c2dda46..f0dea5cd23 100644 --- a/blob/gcsblob/gcsblob_test.go +++ b/blob/gcsblob/gcsblob_test.go @@ -25,6 +25,7 @@ import ( "os" "os/user" "path/filepath" + "strings" "testing" "time" @@ -116,6 +117,62 @@ func TestConformance(t *testing.T) { drivertest.RunConformanceTests(t, newHarness, []drivertest.AsTest{verifyContentLanguage{}}) } +func TestOpenBucketGRPC(t *testing.T) { + ctx := context.Background() + client, cleanup, err := DialGRPC(ctx, nil) // unauthenticated; no RPCs are made here + if err != nil { + t.Fatal(err) + } + defer cleanup() + + for _, test := range []struct { + name string + client *storage.Client + bucketName string + opts *Options + wantErr string + }{ + {name: "ok", client: client, bucketName: "mybucket"}, + {name: "ok, nil Options", client: client, bucketName: "mybucket", opts: nil}, + {name: "nil client", bucketName: "mybucket", wantErr: "client is required"}, + {name: "empty bucket name", client: client, wantErr: "bucketName is required"}, + { + name: "Options.Client also set", + client: client, + bucketName: "mybucket", + opts: &Options{Client: client}, + wantErr: "Options.Client must be nil", + }, + } { + t.Run(test.name, func(t *testing.T) { + b, err := OpenBucketGRPC(test.client, test.bucketName, test.opts) + if test.wantErr != "" { + if err == nil { + t.Fatalf("got nil error, want one containing %q", test.wantErr) + } + if !strings.Contains(err.Error(), test.wantErr) { + t.Errorf("got error %q, want it to contain %q", err, test.wantErr) + } + return + } + if err != nil { + t.Fatal(err) + } + // The caller owns the client, so closing the bucket must leave it + // usable rather than closing it. + if err := b.Close(); err != nil { + t.Errorf("Close: %v", err) + } + if _, err := OpenBucketGRPC(client, "mybucket", nil); err != nil { + t.Errorf("client was closed by bucket.Close: %v", err) + } + }) + } +} + +// HTTPClient returns nil: without a GoogleAccessID and a signer, SignedURL +// reports Unimplemented and drivertest skips the checks that would need this. +// SignedURL is signed client-side and is unaffected by the transport. func BenchmarkGcsblob(b *testing.B) { ctx := context.Background() creds, err := gcp.DefaultCredentials(ctx) @@ -561,6 +618,7 @@ func TestURLOpenerForParams(t *testing.T) { currOpts Options query url.Values wantOpts Options + wantParams urlParams wantClient bool wantErr bool }{ @@ -593,6 +651,7 @@ func TestURLOpenerForParams(t *testing.T) { "access_id": {"-"}, }, wantOpts: Options{}, // cleared + wantParams: urlParams{anonymous: true}, wantClient: true, }, { @@ -614,6 +673,7 @@ func TestURLOpenerForParams(t *testing.T) { "anonymous": {"true"}, }, wantOpts: Options{}, // cleared + wantParams: urlParams{anonymous: true}, wantClient: true, }, { @@ -638,6 +698,61 @@ func TestURLOpenerForParams(t *testing.T) { }, wantOpts: Options{}, }, + { + name: "GRPC", + query: url.Values{ + "grpc": {"true"}, + }, + wantOpts: Options{}, + wantParams: urlParams{useGRPC: true}, + }, + { + name: "Invalid value for grpc", + query: url.Values{ + "grpc": {"bad"}, + }, + wantErr: true, + }, + { + name: "Zonal", + query: url.Values{ + "zonal": {"true"}, + }, + wantOpts: Options{}, + // zonal=true implies grpc=true. + wantParams: urlParams{useGRPC: true, useZonal: true}, + }, + { + name: "Zonal with explicit grpc", + query: url.Values{ + "grpc": {"true"}, + "zonal": {"true"}, + }, + wantOpts: Options{}, + wantParams: urlParams{useGRPC: true, useZonal: true}, + }, + { + name: "Zonal false", + query: url.Values{ + "zonal": {"false"}, + }, + wantOpts: Options{}, + }, + { + name: "Zonal conflicts with grpc=false", + query: url.Values{ + "grpc": {"false"}, + "zonal": {"true"}, + }, + wantErr: true, + }, + { + name: "Invalid value for zonal", + query: url.Values{ + "zonal": {"bad"}, + }, + wantErr: true, + }, { name: "AccessID change clears PrivateKey and MakeSignBytes", currOpts: Options{ @@ -659,7 +774,7 @@ func TestURLOpenerForParams(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { o := &URLOpener{Options: test.currOpts} - got, gotClient, err := o.forParams(ctx, test.query) + got, gotClient, gotParams, err := o.forParams(ctx, test.query) if (err != nil) != test.wantErr { t.Errorf("got err %v want error %v", err, test.wantErr) } @@ -669,6 +784,9 @@ func TestURLOpenerForParams(t *testing.T) { if diff := cmp.Diff(got, &test.wantOpts); diff != "" { t.Errorf("opener.forParams(...) diff (-want +got):\n%s", diff) } + if diff := cmp.Diff(gotParams, test.wantParams, cmp.AllowUnexported(urlParams{})); diff != "" { + t.Errorf("opener.forParams(...) params diff (-want +got):\n%s", diff) + } if test.wantClient != (gotClient != nil) { t.Errorf("opener.forParams client return value was unexpected, got %v want %v", gotClient != nil, test.wantClient) } @@ -705,6 +823,19 @@ func TestOpenBucketFromURL(t *testing.T) { {"gs://mybucket?universe_domain=example.com", false}, // OK, universe_domain with empty value. {"gs://mybucket?universe_domain=", false}, + // OK, using the gRPC API. + {"gs://mybucket?grpc=true", false}, + // Invalid value for grpc. + {"gs://mybucket?grpc=bad", true}, + // OK, using the zonal APIs (required for Rapid Storage buckets). + {"gs://mybucket?zonal=true", false}, + // Invalid value for zonal. + {"gs://mybucket?zonal=bad", true}, + // zonal=true cannot be combined with grpc=false. + {"gs://mybucket?zonal=true&grpc=false", true}, + // OK, gRPC without credentials. + {"gs://mybucket?grpc=true&anonymous=true", false}, + {"gs://mybucket?zonal=true&anonymous=true", false}, // Invalid private_key_path. {"gs://mybucket?private_key_path=invalid-path", true}, // Invalid parameter. From 959234e6fe23fb3b2b928f25b63f4ba0954eee45 Mon Sep 17 00:00:00 2001 From: Stan Hu Date: Tue, 1 Sep 2026 11:59:11 -0700 Subject: [PATCH 2/2] blob/gcsblob: add conformance tests for the gRPC API Add TestConformanceGRPC, which runs the full drivertest suite over a gRPC client built by DialGRPC, and TestConformanceGRPCZonal, which does the same with the zonal bucket APIs enabled. Both run against a real bucket named by GCSBLOB_GRPC_TEST_BUCKET and skip when it is unset, rather than using the record/replay harness TestConformance uses. Replay does not work with released grpcreplay. storage reads objects with a zero-copy codec, installed unconditionally in NewRangeReaderReadObject as grpc.ForceCodecV2(bytesCodecReadObject{}), so RecvMsg is handed a *mem.BufferSlice instead of a proto.Message. grpcreplay assumes every message is a proto.Message and panics on the unchecked type assertion in message.set, aborting the test binary on the first gRPC read. Recording writes works; it is reads that cannot be captured. panic: interface conversion: *mem.BufferSlice is not protoreflect.ProtoMessage: missing method ProtoReflect grpcreplay.(*message).set(...) grpcreplay.(*recClientStream).RecvMsg(...) storage.(*grpcStorageClient).NewRangeReaderReadObject... https://github.com/google/go-replayers/pull/70 fixes this by recording such messages as raw wire bytes. Verified against that branch: writes, full reads, out-of-order reads, range reads and unary calls all record and then replay offline with no credentials and no network. Once it is released and go.mod picks it up, these tests should move to the record/replay harness and stop needing a real bucket. Until then they need one, so they skip by default. Against a US multi-region bucket TestConformanceGRPC passes all 88 checks. TestConformanceGRPCZonal is additionally skipped unconditionally. Against a Rapid Storage bucket 55 checks pass and 33 fail, and every failure is a Cloud Storage restriction rather than a driver bug. drivertest cannot express "this driver does not support X": the only opt-out is returning gcerrors.Unimplemented, every place that honors it guards SignedURL, and testCopy treats any error from Copy as a failure. The test is kept, with a comment naming the specific subtests to disable and why, so it can be enabled once that is possible: - TestCopy, TestKeys and TestAs, because Rapid Storage does not support object rewrite. TestKeys accounts for 19 of the 33 failures on its own, since it copies once per key it exercises. - TestListDelimiters/backslash and TestListDelimiters/abc, because a hierarchical namespace, which Rapid Storage requires, only supports "/" as a delimiter. - Four TestWrite subtests that rewrite one object in a tight loop and hit the general per-object mutation rate limit. That is not a Rapid Storage restriction and only appears when the client is close enough to trip it, so it may not need a permanent skip. SignedURL is left unexercised: with no GoogleAccessID the driver reports Unimplemented and drivertest skips those checks, so HTTPClient returns nil. Signing is client-side and does not depend on transport. --- blob/gcsblob/gcsblob_test.go | 111 +++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/blob/gcsblob/gcsblob_test.go b/blob/gcsblob/gcsblob_test.go index f0dea5cd23..99cf9c27ed 100644 --- a/blob/gcsblob/gcsblob_test.go +++ b/blob/gcsblob/gcsblob_test.go @@ -30,6 +30,7 @@ import ( "time" "cloud.google.com/go/storage" + "cloud.google.com/go/storage/experimental" "github.com/google/go-cmp/cmp" "gocloud.dev/blob" "gocloud.dev/blob/driver" @@ -38,6 +39,7 @@ import ( "gocloud.dev/gcp" "gocloud.dev/internal/testing/setup" "google.golang.org/api/googleapi" + "google.golang.org/api/option" ) const ( @@ -170,9 +172,118 @@ func TestOpenBucketGRPC(t *testing.T) { } } +// grpcBucketEnv names a real bucket to run the gRPC conformance tests against. +const grpcBucketEnv = "GCSBLOB_GRPC_TEST_BUCKET" + +// grpcHarness runs the conformance tests over the Cloud Storage gRPC API +// against a real bucket. Unlike newHarness it does not record or replay, so it +// skips unless grpcBucketEnv is set; see TestConformanceGRPC for why. +type grpcHarness struct { + client *storage.Client + cleanup func() + bucket string +} + +func newGRPCHarness(zonal bool) func(ctx context.Context, t *testing.T) (drivertest.Harness, error) { + return func(ctx context.Context, t *testing.T) (drivertest.Harness, error) { + t.Helper() + bkt := os.Getenv(grpcBucketEnv) + if bkt == "" { + t.Skipf("%s not set", grpcBucketEnv) + } + creds, err := gcp.DefaultCredentials(ctx) + if err != nil { + return nil, err + } + var opts []option.ClientOption + if zonal { + opts = append(opts, experimental.WithZonalBucketAPIs()) + } + client, cleanup, err := DialGRPC(ctx, gcp.CredentialsTokenSource(creds), opts...) + if err != nil { + return nil, err + } + return &grpcHarness{client: client, cleanup: cleanup, bucket: bkt}, nil + } +} + +func (h *grpcHarness) MakeDriver(ctx context.Context) (driver.Bucket, error) { + return openBucket(ctx, nil, h.bucket, &Options{Client: h.client}) +} + +func (h *grpcHarness) MakeDriverForNonexistentBucket(ctx context.Context) (driver.Bucket, error) { + return openBucket(ctx, nil, "bucket-does-not-exist", &Options{Client: h.client}) +} + // HTTPClient returns nil: without a GoogleAccessID and a signer, SignedURL // reports Unimplemented and drivertest skips the checks that would need this. // SignedURL is signed client-side and is unaffected by the transport. +func (h *grpcHarness) HTTPClient() *http.Client { return nil } + +func (h *grpcHarness) Close() { h.cleanup() } + +// TestConformanceGRPC runs the conformance suite over the gRPC API. +// +// It talks to a real bucket rather than a recorded one, so it skips unless +// GCSBLOB_GRPC_TEST_BUCKET names a bucket you can write to: +// +// GCSBLOB_GRPC_TEST_BUCKET=my-bucket go test ./blob/gcsblob/ -run TestConformanceGRPC +// +// It cannot yet use the record/replay harness that TestConformance uses. +// cloud.google.com/go/storage reads objects with a zero-copy codec, installed +// unconditionally as grpc.ForceCodecV2(bytesCodecReadObject{}), so RecvMsg is +// handed a *mem.BufferSlice rather than a proto.Message. Released versions of +// grpcreplay assume every message is a proto.Message and panic on the type +// assertion in message.set, so recording any gRPC read aborts the test binary. +// +// https://github.com/google/go-replayers/pull/70 fixes that by recording such +// messages as raw wire bytes. With it applied, storage writes, full reads, +// range reads and unary calls all record and replay correctly. Once it is +// released and go.mod picks it up, this test should move to the record/replay +// harness and stop needing a real bucket. +func TestConformanceGRPC(t *testing.T) { + drivertest.RunConformanceTests(t, newGRPCHarness(false), []drivertest.AsTest{verifyContentLanguage{}}) +} + +// TestConformanceGRPCZonal is TestConformanceGRPC with the zonal bucket APIs +// enabled, which is what a Rapid Storage bucket requires. +// +// It is skipped unconditionally. Against a Rapid Storage bucket 55 checks pass +// and 33 fail, and every failure is a Cloud Storage restriction rather than a +// driver bug. drivertest cannot currently express "this driver does not support +// X": the only opt-out is returning gcerrors.Unimplemented, and every place +// that honors it guards SignedURL, while testCopy treats any error from Copy as +// a failure. +// +// Once specific conformance tests can be disabled, these are the ones to +// disable, and why: +// +// - TestCopy, TestKeys and TestAs. Rapid Storage does not support object +// rewrite, so Copy fails with "Rapid storage class objects do not support +// rewrite". Only TestCopy is about copying; TestKeys and TestAs copy +// incidentally, and TestKeys accounts for 19 of the 33 failures because it +// copies once per key it exercises. +// https://docs.cloud.google.com/storage/docs/rapid/rapid-bucket +// +// - TestListDelimiters/backslash and TestListDelimiters/abc. Rapid Storage +// requires a hierarchical namespace, and such buckets only support "/" as a +// delimiter; anything else fails with "Invalid argument". +// TestListDelimiters/fwdslash passes. +// +// - TestWrite/Content_md5_match, TestWrite/Content_md5_did_not_match,_blob_existed, +// TestWrite/a_small_text_file_gets_a_ContentType and +// TestWrite/write_with_explicit_ContentType_overrides_discovery. These +// rewrite one object in a tight loop and hit "exceeded the rate limit for +// object mutation operations". That is the general Cloud Storage +// per-object mutation limit rather than a Rapid Storage restriction, and it +// only appears when the client is close enough to the bucket to trip it: +// these four failed from a VM in the bucket's zone but not from a laptop. +// They may not need a permanent skip. +func TestConformanceGRPCZonal(t *testing.T) { + t.Skip("Rapid Storage does not support object rewrite or non-\"/\" list delimiters; see the comment above for the tests that need to be skipped") + drivertest.RunConformanceTests(t, newGRPCHarness(true), []drivertest.AsTest{verifyContentLanguage{}}) +} + func BenchmarkGcsblob(b *testing.B) { ctx := context.Background() creds, err := gcp.DefaultCredentials(ctx)