mirror of
https://github.com/agent-substrate/substrate.git
synced 2026-10-02 03:24:42 +08:00
ategcs: build the connection pool with its client's options (#1178)
The pool serves every range past the first and was built with a bare storage.NewClient, so an object opened anonymously had its later ranges fetched with an authenticated token. A public bucket without Uniform Bucket Level Access rejects that with HTTP 412, which left a node with no cached runsc unable to download the gVisor release tarball. Carry the options the client was built with into the pool. Adds a regression test, since CI can't cover this, and I missed this in previous PRs ...
This commit is contained in:
@@ -23,18 +23,25 @@ import (
|
||||
"sync"
|
||||
|
||||
"cloud.google.com/go/storage"
|
||||
"google.golang.org/api/option"
|
||||
)
|
||||
|
||||
type gcsClient struct {
|
||||
client *storage.Client
|
||||
// pool holds extra clients so concurrent upload parts get their own connections;
|
||||
// opts are how client was built, so the pool below is built the same way.
|
||||
// A pooled client that authenticates differently from the one that opened
|
||||
// an object fails partway through reading it.
|
||||
opts []option.ClientOption
|
||||
// pool holds extra clients so concurrent parts get their own connections;
|
||||
// built on first use by uploadClient.
|
||||
poolOnce sync.Once
|
||||
pool []*storage.Client
|
||||
}
|
||||
|
||||
func NewGCSClient(client *storage.Client) ObjectStorage {
|
||||
return &gcsClient{client: client}
|
||||
// NewGCSClient wraps client. opts must be the options client was built with;
|
||||
// the pool of extra connections is built from them.
|
||||
func NewGCSClient(client *storage.Client, opts ...option.ClientOption) ObjectStorage {
|
||||
return &gcsClient{client: client, opts: opts}
|
||||
}
|
||||
|
||||
// supportsStreamingPut is the streamingPutter marker: the GCS client's PutObject
|
||||
|
||||
@@ -40,7 +40,7 @@ func (g *gcsClient) uploadClient(ctx context.Context, i int) *storage.Client {
|
||||
g.poolOnce.Do(func() {
|
||||
for range uploadPoolSize {
|
||||
// The clients outlive this call, so they must not hold its cancellation.
|
||||
c, err := storage.NewClient(context.WithoutCancel(ctx))
|
||||
c, err := storage.NewClient(context.WithoutCancel(ctx), g.opts...)
|
||||
if err != nil {
|
||||
slog.WarnContext(ctx, "Falling back to one client for part uploads", slog.Any("err", err))
|
||||
return
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
// Copyright 2026 Google LLC
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package ategcs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"cloud.google.com/go/storage"
|
||||
"google.golang.org/api/option"
|
||||
)
|
||||
|
||||
// TestPooledClientsAreBuiltLikeTheClientTheyStandIn checks that pooled
|
||||
// connections carry the wrapped client's options. The pool serves ranges past
|
||||
// the first, so a mismatch fails mid-object rather than at open: a public
|
||||
// bucket without Uniform Bucket Level Access rejects an authenticated token
|
||||
// with HTTP 412.
|
||||
//
|
||||
// Credentials are removed so the two cases separate -- an anonymous client
|
||||
// still builds and a default one cannot -- which makes distinct pooled
|
||||
// connections the proof that the options were used.
|
||||
func TestPooledClientsAreBuiltLikeTheClientTheyStandIn(t *testing.T) {
|
||||
t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", "/nonexistent/credentials.json")
|
||||
t.Setenv("GCE_METADATA_HOST", "127.0.0.1:1")
|
||||
|
||||
ctx := context.Background()
|
||||
anon, err := storage.NewClient(ctx, option.WithoutAuthentication())
|
||||
if err != nil {
|
||||
t.Fatalf("an anonymous client must build without credentials: %v", err)
|
||||
}
|
||||
defer anon.Close()
|
||||
|
||||
g, ok := NewGCSClient(anon, option.WithoutAuthentication()).(*gcsClient)
|
||||
if !ok {
|
||||
t.Fatal("NewGCSClient did not return a *gcsClient")
|
||||
}
|
||||
|
||||
pooled := g.uploadClient(ctx, 0)
|
||||
if pooled == nil {
|
||||
t.Fatal("uploadClient returned nil")
|
||||
}
|
||||
if pooled == g.client {
|
||||
t.Fatal("uploadClient fell back to the wrapped client: the pool was not given the client's " +
|
||||
"options, so ranges past the first would go out authenticated")
|
||||
}
|
||||
if len(g.pool) != uploadPoolSize {
|
||||
t.Errorf("pool holds %d clients, want %d", len(g.pool), uploadPoolSize)
|
||||
}
|
||||
for i := range uploadPoolSize {
|
||||
if c := g.uploadClient(ctx, i); c == nil || c == g.client {
|
||||
t.Errorf("uploadClient(%d) did not return a pooled connection", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -245,7 +245,7 @@ func main() {
|
||||
|
||||
var wrappedAnonGCS ategcs.ObjectStorage
|
||||
if anonGCSClient != nil {
|
||||
wrappedAnonGCS = ategcs.NewGCSClient(anonGCSClient)
|
||||
wrappedAnonGCS = ategcs.NewGCSClient(anonGCSClient, option.WithoutAuthentication())
|
||||
}
|
||||
|
||||
var wrappedGCS ategcs.ObjectStorage
|
||||
|
||||
Reference in New Issue
Block a user