diff --git a/ci/resources/stemcell-version-bump/go.mod b/ci/resources/stemcell-version-bump/go.mod index 465ac49f..335b74be 100644 --- a/ci/resources/stemcell-version-bump/go.mod +++ b/ci/resources/stemcell-version-bump/go.mod @@ -3,7 +3,7 @@ module stemcell-version-bump go 1.26.0 require ( - cloud.google.com/go/storage v1.68.0 + cloud.google.com/go/storage v1.69.0 github.com/stretchr/testify v1.12.1 google.golang.org/api v0.299.0 ) @@ -16,7 +16,7 @@ require ( cloud.google.com/go/compute/metadata v0.9.1 // indirect cloud.google.com/go/iam v1.12.0 // indirect cloud.google.com/go/monitoring v1.30.0 // indirect - github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.34.0 // indirect + github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.35.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.57.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect @@ -30,11 +30,11 @@ require ( github.com/google/s2a-go v0.1.10 // indirect github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.22 // indirect - github.com/googleapis/gax-go/v2 v2.24.1 // indirect + github.com/googleapis/gax-go/v2 v2.26.2 // indirect github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect github.com/spiffe/go-spiffe/v2 v2.8.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect + go.opentelemetry.io/contrib/detectors/gcp v1.45.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect go.opentelemetry.io/otel v1.45.0 // indirect diff --git a/ci/resources/stemcell-version-bump/go.sum b/ci/resources/stemcell-version-bump/go.sum index c61c3b21..019292d4 100644 --- a/ci/resources/stemcell-version-bump/go.sum +++ b/ci/resources/stemcell-version-bump/go.sum @@ -16,12 +16,12 @@ cloud.google.com/go/longrunning v1.2.0 h1:WjYH3YHBGCxGJP9M4dWGHBfXr/cFIjMkNgWcJj cloud.google.com/go/longrunning v1.2.0/go.mod h1:5KMQALFGOCtFoi2xSOA1u3H7WKlhmckgiyFw7+LGQp0= cloud.google.com/go/monitoring v1.30.0 h1:r/d+JUbyKmJ8b07iznuKfzVzrIXTWxHQ3lBRm3x2LlY= cloud.google.com/go/monitoring v1.30.0/go.mod h1:htlUR0QWVMrjFzZmN4LGnMAve9xB/eduwjmINxVZ8RM= -cloud.google.com/go/storage v1.68.0 h1:gqrAMJ51OZjYgU6AJ2U60um90YQhSjq8HEIQNtJ4C/8= -cloud.google.com/go/storage v1.68.0/go.mod h1:UsS9OgFg/XHOSYakQ8ZtLWWeyGkk1WnmD/GsGfN0BHM= +cloud.google.com/go/storage v1.69.0 h1:jAAMC1411HEh78nKsU0Zns+eFj3TnhjAWIhg5Ud/XBM= +cloud.google.com/go/storage v1.69.0/go.mod h1:PELYsxTYm2peE4mwLEC1+mS1dA/kUSRUxNv56rOy44g= cloud.google.com/go/trace v1.16.0 h1:GmQovzFc5F0CNfl0VLgL64aoTtu7xsM0YajW2GlG9+E= cloud.google.com/go/trace v1.16.0/go.mod h1:r+bdAn16dKLSV1G2D5v3e58IlQlizfxWrUfjx7kM7X0= -github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.34.0 h1:yzIYdwuro811Z27D3T80Wkd3rqZzb0K43nner7Eh1yE= -github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.34.0/go.mod h1:pJTkW8hEUIIi3Pf65lPZOnn4Y81yCllX6IWk2jNXdkM= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.35.0 h1:bN1gA3of5bXtbnLsRPrwfmbbe7A5UWFlcTHseujLnpc= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.35.0/go.mod h1:Yj5vHEz/aAepZGliRJsA6uvHAVAQyEwajq9ORCHPxzM= github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0 h1:jLdiS1vO+XJFyDSWRHBx56r4s/NNtcl5J6KyCcWUX/w= github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0/go.mod h1:8lmpHY+1VRoteiOwyrQMDt1YGXOrFKCz+1wJW7n3ODY= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.57.0 h1:cSjUzZ7KU8hicTgzaSv9NmSyM9fTVK3y5lsBUl3wOis= @@ -61,8 +61,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/enterprise-certificate-proxy v0.3.22 h1:NU4XpII6jD+Dxcot94fqjE+AfJoE/lQP9q3faYGzC/c= github.com/googleapis/enterprise-certificate-proxy v0.3.22/go.mod h1:L3D/IQExI6LqEjBdXcZQ1WluSgigQmSwBboFstVPM4w= -github.com/googleapis/gax-go/v2 v2.24.1 h1:AtqTN21IXMMWo99LiEVAiBfNNQmO40d8xUfZI640mc0= -github.com/googleapis/gax-go/v2 v2.24.1/go.mod h1:bWeBei0NVwaNZKb2y1HUBS7gLXIF3/Tu3pq7j8D2Tb0= +github.com/googleapis/gax-go/v2 v2.26.2 h1:ydkmNXxj7bEmmeK5AihkKnWxyOyBR9TDebvp5L5izk8= +github.com/googleapis/gax-go/v2 v2.26.2/go.mod h1:sMKqnMesnKH+3wiRJROcttA+cJoZoGbZl1vDQ8XYtGk= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo= github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8= github.com/spiffe/go-spiffe/v2 v2.8.1 h1:eXZMLsu+3MLEPJyGJkolqtVrteZfQdUpOWj6LTiDl/E= @@ -71,16 +71,16 @@ github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWD github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/contrib/detectors/gcp v1.44.0 h1:NmLfL734pJhM0JKaYd2Y28+nY9dPRWYAAbxhRCrKXPw= -go.opentelemetry.io/contrib/detectors/gcp v1.44.0/go.mod h1:tNAsgd8avTGke1+MndXlU5Cru4PQ9Ai/cCNWQv/ZJ/s= +go.opentelemetry.io/contrib/detectors/gcp v1.45.0 h1:9jR0ZPRok9ryaOQ2Wx8rg5F7Aon59mxrqbVI60/vlBk= +go.opentelemetry.io/contrib/detectors/gcp v1.45.0/go.mod h1:VSme3o2fvSg5bVg0dRzyHaj4Z5EVhG+g2Fde6LKzmQA= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 h1:0Qx7VGBacMm9ZENQ7TnNObTYI4ShC+lHI16seduaxZo= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0/go.mod h1:Sje3i3MjSPKTSPvVWCaL8ugBzJwik3u4smCjUeuupqg= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= go.opentelemetry.io/otel v1.45.0/go.mod h1:XZxIqPapzEYnhNSScF5DIqXhm/rYi0FzCe2XddAwZfQ= -go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.44.0 h1:hqxVTu/GtBF+vJ8d1fzW7fRxZFvgoDjWcxwwCaFDYpU= -go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.44.0/go.mod h1:z5fVEF4X5v0ESvlJqBrrFlBVoj5EQuefZpzsu7R+x5Q= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.45.0 h1:dm9iyzn6tioYZtwqaiBSU0TSI8Yu/8dTIbfG0+B49DY= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.45.0/go.mod h1:xAvxYjYK28qvt+yu4BYZ/zMmAjwMXINXD6JiMyeB8iI= go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypRm3br/M= go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= go.opentelemetry.io/otel/metric/x v0.67.0 h1:PcicCNZFkZ4bXfSooXdo3WN7RBOVOtjVdo1wD358Uns= diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md index 102a5dd1..8d0d7793 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/CHANGES.md @@ -1,6 +1,22 @@ # Changes +## [1.69.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.68.0...storage/v1.69.0) (2026-10-01) + + +### Features + +* **all:** Update supported go versions ([#20520](https://github.com/googleapis/google-cloud-go/issues/20520)) ([90413d3](https://github.com/googleapis/google-cloud-go/commit/90413d3e9d381ef4b92e3c496d98b023c5b2d8cc)) +* **storage:** Add read stall retry for gRPC storage client ([#20525](https://github.com/googleapis/google-cloud-go/issues/20525)) ([8bfc486](https://github.com/googleapis/google-cloud-go/commit/8bfc4863da4151976a5e4629ef9dce429edb43d5)) + + +### Bug Fixes + +* **storage:** Add App Hub storage.googleapis.com prefix to destination.id ([#20569](https://github.com/googleapis/google-cloud-go/issues/20569)) ([2bca92a](https://github.com/googleapis/google-cloud-go/commit/2bca92a333bcadaa78b7c3b841e5a293e3a681d7)) +* **storage:** Follow-ups for read stall retry ([#20577](https://github.com/googleapis/google-cloud-go/issues/20577)) ([24eacd5](https://github.com/googleapis/google-cloud-go/commit/24eacd52ffc1697341d5f6a8380abc656292367a)) +* **storage:** Update PartSize documentation and optimize cleanup ([#19930](https://github.com/googleapis/google-cloud-go/issues/19930)) ([3d09abf](https://github.com/googleapis/google-cloud-go/commit/3d09abf7dd357cbf5c1022a7e65c7859d617ccd6)) +* **various:** Address format directive issues ([#20547](https://github.com/googleapis/google-cloud-go/issues/20547)) ([e2e1047](https://github.com/googleapis/google-cloud-go/commit/e2e104731999b3bfc75ed0b8f2d71844ced3b073)) + ## [1.68.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.67.1...storage/v1.68.0) (2026-09-17) diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/bucket_metadata_cache.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/bucket_metadata_cache.go index 258d52b0..6e612c8a 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/bucket_metadata_cache.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/bucket_metadata_cache.go @@ -30,6 +30,13 @@ import ( const ( defaultBucketMetadataCacheLimit = 10000 fetchBackgroundTimeout = 10 * time.Second + + // storageResourceNamePrefix is prepended to the bucket resource name emitted + // in the gcp.resource.destination.id span attribute. Cloud Trace's App Hub + // extractor only accepts full resource names of the form + // "//{service}/{path}"; a bare "projects/.../buckets/..." path is rejected + // as malformed and the span is silently dropped from App Hub enrichment. + storageResourceNamePrefix = "//storage.googleapis.com/" ) type bucketMetadataFetcher interface { @@ -91,11 +98,13 @@ func (c *bucketMetadataCache) evict(bucket string) { c.lru.evict(bucket) } -func (c *bucketMetadataCache) fetchBackground(bucket string) { +func (c *bucketMetadataCache) fetchBackground(ctx context.Context, bucket string) { if c == nil || c.fetcher == nil { return } + detachedCtx := context.WithoutCancel(ctx) + go func() { defer func() { if c.fetchDone != nil { @@ -107,8 +116,8 @@ func (c *bucketMetadataCache) fetchBackground(bucket string) { }() c.muSF.Do(bucket, func() (interface{}, error) { - // Perform the call with context.Background and a timeout so it runs outside request context lifetime but is bounded. - ctx, cancel := context.WithTimeout(context.Background(), fetchBackgroundTimeout) + // Perform the call with detached context and a timeout so it runs outside request context lifetime but is bounded, preserving trace context. + ctx, cancel := context.WithTimeout(detachedCtx, fetchBackgroundTimeout) defer cancel() resource, location, err := c.fetcher.fetchBucketMetadata(ctx, bucket) @@ -126,7 +135,7 @@ func (c *bucketMetadataCache) fetchBackground(bucket string) { } else { if !hit { c.lru.put(bucket, bucketMetadata{ - resource: fmt.Sprintf("projects/_/buckets/%s", bucket), + resource: fmt.Sprintf("%sprojects/_/buckets/%s", storageResourceNamePrefix, bucket), location: "global", placeholder: true, }) @@ -151,11 +160,11 @@ func getMetadataFromAttrs(location, locationType, project, bucket string) (strin finalLocation = strings.ToLower(location) } if strings.HasPrefix(project, "projects/") { - return project + "/buckets/" + bucket, finalLocation + return storageResourceNamePrefix + project + "/buckets/" + bucket, finalLocation } finalProject := "_" if project != "0" && project != "" { finalProject = project } - return fmt.Sprintf("projects/%s/buckets/%s", finalProject, bucket), finalLocation + return fmt.Sprintf("%sprojects/%s/buckets/%s", storageResourceNamePrefix, finalProject, bucket), finalLocation } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/dynamic_delay.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/dynamic_delay.go index 5944f515..e3332bca 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/dynamic_delay.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/dynamic_delay.go @@ -15,6 +15,7 @@ package storage import ( + "context" "fmt" "math" "sync" @@ -235,3 +236,66 @@ func (b *bucketDelayManager) update(bucketName string, latency time.Duration) { func (b *bucketDelayManager) getValue(bucketName string) time.Duration { return b.getDelay(bucketName).getValue() } + +// executeWithReadStallTimeout executes openStream with dynamic delay stall retry tracking. +func executeWithReadStallTimeout( + ctx context.Context, + dm *bucketDelayManager, + bucket string, + openStream func(ctx context.Context) error, + onStall func(stallTimeout time.Duration), +) error { + if dm == nil { + return openStream(ctx) + } + + cancelCtx, cancel := context.WithCancel(ctx) + defer func() { + if cancel != nil { + cancel() + } + }() + var ( + innerErr error + done = make(chan struct{}, 1) + ) + + go func() { + reqStartTime := time.Now() + innerErr = openStream(cancelCtx) + if innerErr == nil { + reqLatency := time.Since(reqStartTime) + dm.update(bucket, reqLatency) + } else if ctx.Err() == nil && cancelCtx.Err() != nil { + dm.increase(bucket) + } + done <- struct{}{} + }() + + stallTimeout := dm.getValue(bucket) + timer := time.NewTimer(stallTimeout) + defer timer.Stop() + + select { + case <-timer.C: + // If openStream completed at the same time the timer fired, select may + // pick either case. Prefer the real result over reporting a stall. + select { + case <-done: + default: + cancel() + <-done + if onStall != nil { + onStall(stallTimeout) + } + return context.DeadlineExceeded + } + case <-done: + } + if innerErr == nil { + // Keep the stream context alive for the caller; it is released when + // the parent context is cancelled. + cancel = nil + } + return innerErr +} diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go index 7ed711a9..7a818b5b 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/experimental/experimental.go @@ -52,7 +52,7 @@ func WithMeterProvider(mp *metric.MeterProvider) option.ClientOption { return internal.WithMeterProvider.(func(*metric.MeterProvider) option.ClientOption)(mp) } -// WithReadStallTimeout provides a [option.ClientOption] that may be passed to [storage.NewClient]. +// WithReadStallTimeout provides a [option.ClientOption] that may be passed to [storage.NewClient] or [storage.NewGRPCClient]. // It enables the client to retry stalled requests when starting a download from // Cloud Storage. If the timeout elapses with no response from the server, the request // is automatically retried. @@ -60,9 +60,8 @@ func WithMeterProvider(mp *metric.MeterProvider) option.ClientOption { // latency across all read requests from the client for each bucket accessed, and can // adjust the timeout higher to the target percentile when latency for request to that // bucket is high. -// Currently, this is supported only for downloads ([storage.NewReader] and -// [storage.NewRangeReader] calls) and only for the XML API. Other read APIs (gRPC & JSON) -// will be supported soon. +// Currently, this is supported for downloads ([storage.NewReader] and +// [storage.NewRangeReader] calls) on the XML API and gRPC API (not implemented for JSON API). func WithReadStallTimeout(rstc *ReadStallTimeoutConfig) option.ClientOption { return internal.WithReadStallTimeout.(func(config *ReadStallTimeoutConfig) option.ClientOption)(rstc) } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go index 330c8705..b216ded0 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_client.go @@ -25,6 +25,7 @@ import ( "os" "strconv" "strings" + "time" "cloud.google.com/go/auth" "cloud.google.com/go/iam/apiv1/iampb" @@ -125,6 +126,7 @@ type grpcStorageClient struct { raw *gapic.Client settings *settings config *storageConfig + readStallMgr *bucketDelayManager dpDiag string metrics *clientMetrics metricsCleanup func() @@ -199,11 +201,26 @@ func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (client *g }() } + var bd *bucketDelayManager + if config.readStallTimeoutConfig != nil { + drrstConfig := config.readStallTimeoutConfig + bd, err = newBucketDelayManager( + drrstConfig.TargetPercentile, + getDynamicReadReqIncreaseRateFromEnv(), + getDynamicReadReqInitialTimeoutSecFromEnv(drrstConfig.Min), + drrstConfig.Min, + defaultDynamicReqdReqMaxTimeout) + if err != nil { + return nil, fmt.Errorf("creating dynamic-delay: %w", err) + } + } + c := &grpcStorageClient{ settings: s, config: &config, metrics: clientMetrics, metricsCleanup: metricsCleanup, + readStallMgr: bd, } // Add routing interceptors to inject headers. ui, si := c.routingInterceptors() @@ -1331,7 +1348,7 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange var err error var decoder *readResponseDecoder - err = run(cc, func(ctx context.Context) error { + openStream := func(ctx context.Context) error { var databufs mem.BufferSlice openAndSendReq := func() error { databufs = mem.BufferSlice{} @@ -1389,6 +1406,18 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange } err = decoder.readFullObjectResponse() return err + } + + err = run(cc, func(ctx context.Context) error { + decoder = nil + return executeWithReadStallTimeout(ctx, c.readStallMgr, params.bucket, openStream, func(stallTimeout time.Duration) { + target := stripPort(metricsStateFromContext(ctx).getTarget()) + c.metrics.recordStallDuration(ctx, stallTimeout, "ReadObject", "grpc", target) + if decoder != nil && decoder.databufs != nil { + decoder.databufs.Free() + decoder = nil + } + }) }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go index c7991c42..d4ee140f 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/grpc_reader.go @@ -21,6 +21,7 @@ import ( "fmt" "hash/crc32" "io" + "time" "cloud.google.com/go/storage/internal/apiv2/storagepb" "github.com/googleapis/gax-go/v2" @@ -134,7 +135,7 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params var err error var decoder *readObjectResponseDecoder - err = run(cc, func(ctx context.Context) error { + openStream := func(ctx context.Context) error { stream, err = c.raw.ReadObject(ctx, req, s.gax...) if err != nil { return err @@ -157,6 +158,18 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params } err = decoder.readFullObjectResponse() return err + } + + err = run(cc, func(ctx context.Context) error { + decoder = nil + return executeWithReadStallTimeout(ctx, c.readStallMgr, params.bucket, openStream, func(stallTimeout time.Duration) { + target := stripPort(metricsStateFromContext(ctx).getTarget()) + c.metrics.recordStallDuration(ctx, stallTimeout, "ReadObject", "grpc", target) + if decoder != nil && decoder.databufs != nil { + decoder.databufs.Free() + decoder = nil + } + }) }, s.retry, s.idempotent, withOperation("ReadObject"), withBucket(params.bucket), withObject(params.object)) if err != nil { // Close the stream context we just created to ensure we don't leak diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go index 97498f7a..2f13ceaa 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/http_client.go @@ -21,11 +21,11 @@ import ( "fmt" "hash/crc32" "io" - "log" "net/http" "net/url" "os" "reflect" + "slices" "strconv" "strings" "time" @@ -33,7 +33,6 @@ import ( "cloud.google.com/go/auth" "cloud.google.com/go/iam/apiv1/iampb" "cloud.google.com/go/internal/optional" - "github.com/google/uuid" "github.com/googleapis/gax-go/v2/callctx" "golang.org/x/oauth2/google" "google.golang.org/api/googleapi" @@ -984,7 +983,6 @@ func (c *httpStorageClient) NewRangeReader(ctx context.Context, params *newRange } func (c *httpStorageClient) newRangeReaderXML(ctx context.Context, params *newRangeReaderParams, s *settings) (r *Reader, err error) { - requestID := uuid.New() u := &url.URL{ Scheme: c.scheme, Host: c.xmlHost, @@ -1008,51 +1006,35 @@ func (c *httpStorageClient) newRangeReaderXML(ctx context.Context, params *newRa return nil, err } + // req is reused across retry attempts and setHeadersFromCtx merges + // x-goog-api-client values into the existing header. Snapshot the value + // set before any attempt and restore it on each attempt so tokens from + // earlier attempts (e.g. gccl-attempt-count/1) do not leak into later ones. + baseXGoogHeader := slices.Clone(req.Header.Values(xGoogHeaderKey)) + reopen := readerReopen(ctx, req.Header, params, s, func(ctx context.Context) (*http.Response, error) { + req.Header.Del(xGoogHeaderKey) + for _, v := range baseXGoogHeader { + req.Header.Add(xGoogHeaderKey, v) + } setHeadersFromCtx(ctx, req.Header) if c.dynamicReadReqStallTimeout == nil { return c.hc.Do(req.WithContext(ctx)) } - cancelCtx, cancel := context.WithCancel(ctx) - var ( - res *http.Response - err error - ) - - done := make(chan bool) - go func() { - reqStartTime := time.Now() - res, err = c.hc.Do(req.WithContext(cancelCtx)) - if err == nil { - reqLatency := time.Since(reqStartTime) - c.dynamicReadReqStallTimeout.update(params.bucket, reqLatency) - } else if errors.Is(err, context.Canceled) { - // context.Canceled means operation took more than current dynamicTimeout, - // hence should be increased. - c.dynamicReadReqStallTimeout.increase(params.bucket) - } - done <- true - }() - - // Wait until stall timeout or request is successful. - stallTimeout := c.dynamicReadReqStallTimeout.getValue(params.bucket) - timer := time.After(stallTimeout) - select { - case <-timer: - log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) + var res *http.Response + err := executeWithReadStallTimeout(ctx, c.dynamicReadReqStallTimeout, params.bucket, func(ctx context.Context) error { + var err error + res, err = c.hc.Do(req.WithContext(ctx)) + return err + }, func(stallTimeout time.Duration) { c.metrics.recordStallDuration(ctx, stallTimeout, "ReadObject", "http", stripPort(req.URL.Host)) - cancel() - <-done if res != nil && res.Body != nil { res.Body.Close() } - return res, context.DeadlineExceeded - case <-done: - cancel = nil - } + }) return res, err }, func() error { return setConditionsHeaders(req.Header, params.conds) }, @@ -1777,14 +1759,18 @@ func setHeadersFromCtx(ctx context.Context, header http.Header) { // Merge x-goog-api-client values into a single space-separated value. if strings.EqualFold(k, xGoogHeaderKey) { alreadySetValues := header.Values(xGoogHeaderKey) - vals = append(vals, alreadySetValues...) - - if len(vals) > 0 { - xGoogHeader := vals[0] - for _, v := range vals[1:] { - xGoogHeader = strings.Join([]string{xGoogHeader, v}, " ") + var uniqueVals []string + for _, list := range [][]string{vals, alreadySetValues} { + for _, v := range list { + for _, part := range strings.Fields(v) { + if !slices.Contains(uniqueVals, part) { + uniqueVals = append(uniqueVals, part) + } + } } - header.Set(k, xGoogHeader) + } + if len(uniqueVals) > 0 { + header.Set(k, strings.Join(uniqueVals, " ")) } } else { for _, v := range vals { diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/apiv2/storage_client.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/apiv2/storage_client.go index aa81b6f0..e4ff9678 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/apiv2/storage_client.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/apiv2/storage_client.go @@ -1054,6 +1054,79 @@ func NewClient(ctx context.Context, opts ...option.ClientOption) (*Client, error client.CallOptions.QueryWriteStatus = append(client.CallOptions.QueryWriteStatus, gax.WithClientMetrics(metrics)) client.CallOptions.MoveObject = append(client.CallOptions.MoveObject, gax.WithClientMetrics(metrics)) } + if gax.IsFeatureEnabled("TRACING") { + tracing := gax.NewClientTracing( + gax.WithTracingAttributes(map[string]string{ + gax.ClientService: "storage", + gax.ClientVersion: getVersionClient(), + gax.ClientArtifact: "cloud.google.com/go/storage/internal/apiv2", + gax.RPCSystem: "grpc", + gax.URLDomain: "storage.googleapis.com", + }), + ) + + client.CallOptions.DeleteBucket = append(client.CallOptions.DeleteBucket, gax.WithClientTracing(tracing)) + client.CallOptions.GetBucket = append(client.CallOptions.GetBucket, gax.WithClientTracing(tracing)) + client.CallOptions.CreateBucket = append(client.CallOptions.CreateBucket, gax.WithClientTracing(tracing)) + client.CallOptions.ListBuckets = append(client.CallOptions.ListBuckets, gax.WithClientTracing(tracing)) + client.CallOptions.LockBucketRetentionPolicy = append(client.CallOptions.LockBucketRetentionPolicy, gax.WithClientTracing(tracing)) + client.CallOptions.GetIamPolicy = append(client.CallOptions.GetIamPolicy, gax.WithClientTracing(tracing)) + client.CallOptions.SetIamPolicy = append(client.CallOptions.SetIamPolicy, gax.WithClientTracing(tracing)) + client.CallOptions.TestIamPermissions = append(client.CallOptions.TestIamPermissions, gax.WithClientTracing(tracing)) + client.CallOptions.UpdateBucket = append(client.CallOptions.UpdateBucket, gax.WithClientTracing(tracing)) + client.CallOptions.ComposeObject = append(client.CallOptions.ComposeObject, gax.WithClientTracing(tracing)) + client.CallOptions.DeleteObject = append(client.CallOptions.DeleteObject, gax.WithClientTracing(tracing)) + client.CallOptions.RestoreObject = append(client.CallOptions.RestoreObject, gax.WithClientTracing(tracing)) + client.CallOptions.CancelResumableWrite = append(client.CallOptions.CancelResumableWrite, gax.WithClientTracing(tracing)) + client.CallOptions.GetObject = append(client.CallOptions.GetObject, gax.WithClientTracing(tracing)) + client.CallOptions.ReadObject = append(client.CallOptions.ReadObject, gax.WithClientTracing(tracing)) + client.CallOptions.BidiReadObject = append(client.CallOptions.BidiReadObject, gax.WithClientTracing(tracing)) + client.CallOptions.UpdateObject = append(client.CallOptions.UpdateObject, gax.WithClientTracing(tracing)) + client.CallOptions.WriteObject = append(client.CallOptions.WriteObject, gax.WithClientTracing(tracing)) + client.CallOptions.BidiWriteObject = append(client.CallOptions.BidiWriteObject, gax.WithClientTracing(tracing)) + client.CallOptions.ListObjects = append(client.CallOptions.ListObjects, gax.WithClientTracing(tracing)) + client.CallOptions.RewriteObject = append(client.CallOptions.RewriteObject, gax.WithClientTracing(tracing)) + client.CallOptions.StartResumableWrite = append(client.CallOptions.StartResumableWrite, gax.WithClientTracing(tracing)) + client.CallOptions.QueryWriteStatus = append(client.CallOptions.QueryWriteStatus, gax.WithClientTracing(tracing)) + client.CallOptions.MoveObject = append(client.CallOptions.MoveObject, gax.WithClientTracing(tracing)) + } + if gax.IsFeatureEnabled("LOGGING") { + logging := gax.NewClientLogging( + gax.WithLoggerProvider(c.logger), + gax.WithLoggingAttributes(map[string]string{ + gax.ClientService: "storage", + gax.ClientVersion: getVersionClient(), + gax.ClientArtifact: "cloud.google.com/go/storage/internal/apiv2", + gax.RPCSystem: "grpc", + gax.URLDomain: "storage.googleapis.com", + }), + ) + + client.CallOptions.DeleteBucket = append(client.CallOptions.DeleteBucket, gax.WithClientLogging(logging)) + client.CallOptions.GetBucket = append(client.CallOptions.GetBucket, gax.WithClientLogging(logging)) + client.CallOptions.CreateBucket = append(client.CallOptions.CreateBucket, gax.WithClientLogging(logging)) + client.CallOptions.ListBuckets = append(client.CallOptions.ListBuckets, gax.WithClientLogging(logging)) + client.CallOptions.LockBucketRetentionPolicy = append(client.CallOptions.LockBucketRetentionPolicy, gax.WithClientLogging(logging)) + client.CallOptions.GetIamPolicy = append(client.CallOptions.GetIamPolicy, gax.WithClientLogging(logging)) + client.CallOptions.SetIamPolicy = append(client.CallOptions.SetIamPolicy, gax.WithClientLogging(logging)) + client.CallOptions.TestIamPermissions = append(client.CallOptions.TestIamPermissions, gax.WithClientLogging(logging)) + client.CallOptions.UpdateBucket = append(client.CallOptions.UpdateBucket, gax.WithClientLogging(logging)) + client.CallOptions.ComposeObject = append(client.CallOptions.ComposeObject, gax.WithClientLogging(logging)) + client.CallOptions.DeleteObject = append(client.CallOptions.DeleteObject, gax.WithClientLogging(logging)) + client.CallOptions.RestoreObject = append(client.CallOptions.RestoreObject, gax.WithClientLogging(logging)) + client.CallOptions.CancelResumableWrite = append(client.CallOptions.CancelResumableWrite, gax.WithClientLogging(logging)) + client.CallOptions.GetObject = append(client.CallOptions.GetObject, gax.WithClientLogging(logging)) + client.CallOptions.ReadObject = append(client.CallOptions.ReadObject, gax.WithClientLogging(logging)) + client.CallOptions.BidiReadObject = append(client.CallOptions.BidiReadObject, gax.WithClientLogging(logging)) + client.CallOptions.UpdateObject = append(client.CallOptions.UpdateObject, gax.WithClientLogging(logging)) + client.CallOptions.WriteObject = append(client.CallOptions.WriteObject, gax.WithClientLogging(logging)) + client.CallOptions.BidiWriteObject = append(client.CallOptions.BidiWriteObject, gax.WithClientLogging(logging)) + client.CallOptions.ListObjects = append(client.CallOptions.ListObjects, gax.WithClientLogging(logging)) + client.CallOptions.RewriteObject = append(client.CallOptions.RewriteObject, gax.WithClientLogging(logging)) + client.CallOptions.StartResumableWrite = append(client.CallOptions.StartResumableWrite, gax.WithClientLogging(logging)) + client.CallOptions.QueryWriteStatus = append(client.CallOptions.QueryWriteStatus, gax.WithClientLogging(logging)) + client.CallOptions.MoveObject = append(client.CallOptions.MoveObject, gax.WithClientLogging(logging)) + } client.internalClient = c @@ -1099,9 +1172,6 @@ func (c *gRPCClient) DeleteBucket(ctx context.Context, req *storagepb.DeleteBuck hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetName())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/DeleteBucket") } @@ -1128,9 +1198,6 @@ func (c *gRPCClient) GetBucket(ctx context.Context, req *storagepb.GetBucketRequ hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetName())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/GetBucket") } @@ -1164,9 +1231,6 @@ func (c *gRPCClient) CreateBucket(ctx context.Context, req *storagepb.CreateBuck hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetParent())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/CreateBucket") } @@ -1197,9 +1261,6 @@ func (c *gRPCClient) ListBuckets(ctx context.Context, req *storagepb.ListBuckets hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetParent())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/ListBuckets") } @@ -1258,9 +1319,6 @@ func (c *gRPCClient) LockBucketRetentionPolicy(ctx context.Context, req *storage hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/LockBucketRetentionPolicy") } @@ -1294,9 +1352,6 @@ func (c *gRPCClient) GetIamPolicy(ctx context.Context, req *iampb.GetIamPolicyRe hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetResource())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/GetIamPolicy") } @@ -1330,9 +1385,6 @@ func (c *gRPCClient) SetIamPolicy(ctx context.Context, req *iampb.SetIamPolicyRe hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetResource())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/SetIamPolicy") } @@ -1369,9 +1421,6 @@ func (c *gRPCClient) TestIamPermissions(ctx context.Context, req *iampb.TestIamP hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetResource())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/TestIamPermissions") } @@ -1432,9 +1481,6 @@ func (c *gRPCClient) ComposeObject(ctx context.Context, req *storagepb.ComposeOb hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetKmsKey())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/ComposeObject") } @@ -1465,9 +1511,6 @@ func (c *gRPCClient) DeleteObject(ctx context.Context, req *storagepb.DeleteObje hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/DeleteObject") } @@ -1494,9 +1537,6 @@ func (c *gRPCClient) RestoreObject(ctx context.Context, req *storagepb.RestoreOb hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/RestoreObject") } @@ -1557,9 +1597,6 @@ func (c *gRPCClient) GetObject(ctx context.Context, req *storagepb.GetObjectRequ hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/GetObject") } @@ -1590,9 +1627,6 @@ func (c *gRPCClient) ReadObject(ctx context.Context, req *storagepb.ReadObjectRe hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/ReadObject") } @@ -1715,9 +1749,6 @@ func (c *gRPCClient) ListObjects(ctx context.Context, req *storagepb.ListObjects hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetParent())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/ListObjects") } @@ -1779,9 +1810,6 @@ func (c *gRPCClient) RewriteObject(ctx context.Context, req *storagepb.RewriteOb hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetDestinationBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/RewriteObject") } @@ -1872,9 +1900,6 @@ func (c *gRPCClient) MoveObject(ctx context.Context, req *storagepb.MoveObjectRe hds = append(c.xGoogHeaders, hds...) ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...) - if gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { - ctx = callctx.WithTelemetryContext(ctx, "resource_name", fmt.Sprintf("//storage.googleapis.com/%v", req.GetBucket())) - } if gax.IsFeatureEnabled("METRICS") || gax.IsFeatureEnabled("TRACING") || gax.IsFeatureEnabled("LOGGING") { ctx = callctx.WithTelemetryContext(ctx, "rpc_method", "google.storage.v2.Storage/MoveObject") } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go index f2949ee9..2f5eb4c9 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/version.go @@ -17,4 +17,4 @@ package internal // Version is the current tagged release of the library. -const Version = "1.68.0" +const Version = "1.69.0" diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/metrics.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/metrics.go index dd68fb22..967c5144 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/metrics.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/metrics.go @@ -371,7 +371,7 @@ func initMetrics(ctx context.Context, projectID string, config *storageConfig) ( stallDuration, err = meter.Float64Histogram( "gcp.storage.client.stall.duration", - metric.WithDescription("Duration a connection was stalled waiting for the first byte before being aborted."), + metric.WithDescription("Stall timeout after which a read attempt was aborted while waiting for the initial response (response headers for HTTP, first response message for gRPC)."), metric.WithUnit("s"), ) if err != nil { diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go index 6d963d12..e54a59b1 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/option.go @@ -245,20 +245,13 @@ func (w *withTestMetricReaderConfig) ApplyStorageOpt(c *storageConfig) { c.manualReader = w.metricReader } -// WithReadStallTimeout is an option that may be passed to [NewClient]. +// WithReadStallTimeout is an option that may be passed to [NewClient] or [NewGRPCClient]. // It enables the client to retry the stalled read request, happens as part of // storage.Reader creation. As the name suggest, timeout is adjusted dynamically // based on past observed read-req latencies. // -// This is only supported for the read operation and that too for http(XML) client. -// Grpc read-operation will be supported soon. +// This is supported for read operations on HTTP (XML) and gRPC clients (not implemented for HTTP JSON client). func withReadStallTimeout(rstc *experimental.ReadStallTimeoutConfig) option.ClientOption { - // TODO (raj-prince): To keep separate dynamicDelay instance for different BucketHandle. - // Currently, dynamicTimeout is kept at the client and hence shared across all the - // BucketHandle, which is not the ideal state. As latency depends on location of VM - // and Bucket, and read latency of different buckets may lie in different range. - // Hence having a separate dynamicTimeout instance at BucketHandle level will - // be better if rstc.Min == time.Duration(0) { rstc.Min = defaultDynamicReadReqMinTimeout } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/pcu.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/pcu.go index 51b15acd..400af194 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/pcu.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/pcu.go @@ -30,7 +30,7 @@ import ( const ( defaultPartSize = 16 * 1024 * 1024 // 16 MiB - minPartSize = 5 * 1024 * 1024 // 5 MiB + minPartSize = 8 * 1024 * 1024 // 8 MiB baseWorkers = 4 maxWorkers = 16 tmpObjectPrefix = "gcs-go-sdk-pu-tmp/" @@ -47,14 +47,12 @@ const ( // // **Note:** This feature is currently experimental and its API surface may change // in future releases. It is not yet recommended for production use. -// -// TODO(b/521239530): Add option to delete source parts after compose operation and remove cleanup logic. type ParallelUploadConfig struct { - // PartSize is the size of each part to be uploaded in parallel. - // Defaults to 16MiB. If a value less than 5MiB is provided, it will be - // automatically increased to 5MiB. The value is automatically rounded up - // to the nearest multiple of 256KiB. + // PartSize is the requested size of each part to be uploaded in parallel. + // The server may determine an actual part size closer to the requested one. + // Defaults to 16MiB. If a value less than 8MiB is provided, it will be + // automatically increased to 8MiB. PartSize int // MaxConcurrency is the number of goroutines to use for uploading parts in parallel. @@ -118,13 +116,14 @@ type pcuState struct { bytesBuffered int64 buffersAlloc int - bufferCh chan []byte - uploadCh chan uploadTask - resultCh chan uploadResult - workerWG sync.WaitGroup - collectorWG sync.WaitGroup - started bool - closeOnce sync.Once + bufferCh chan []byte + uploadCh chan uploadTask + resultCh chan uploadResult + workerWG sync.WaitGroup + collectorWG sync.WaitGroup + started bool + closeOnce sync.Once + finalComposeSucceeded bool // Function to upload a part; can be overridden for testing. uploadPartFn func(s *pcuState, task uploadTask) (*ObjectHandle, *ObjectAttrs, error) @@ -442,9 +441,19 @@ func (s *pcuState) close() error { close(s.resultCh) s.collectorWG.Wait() - // Cleanup is always attempted. We do it in the background to not block returning. + // Manual cleanup is attempted unless the final compose succeeded. On success, + // the compose operations delete the source parts automatically (via + // DeleteSourceObjects = true). The check must not depend on firstErr, since + // setError ignores context.Canceled and a cancelled upload would otherwise + // leak its temporary parts. doCleanup uses a context detached from + // cancellation. We do it in the background to not block returning. defer func() { - go s.doCleanupFn(s) + s.mu.Lock() + finalSucceeded := s.finalComposeSucceeded + s.mu.Unlock() + if !finalSucceeded { + go s.doCleanupFn(s) + } }() s.mu.Lock() @@ -525,6 +534,7 @@ func (s *pcuState) composeParts() error { interHandle := s.w.o.c.Bucket(s.w.o.bucket).Object(compName) composer := interHandle.ComposerFrom(finalComps[start:end]...) + composer.DeleteSourceObjects = true _, err := s.composeFn(s.ctx, composer) if err != nil { @@ -555,12 +565,17 @@ func (s *pcuState) composeParts() error { composer.ObjectAttrs = s.w.ObjectAttrs composer.KMSKeyName = s.w.ObjectAttrs.KMSKeyName composer.SendCRC32C = s.w.SendCRC32C + composer.DeleteSourceObjects = true attrs, err := s.composeFn(s.ctx, composer) if err != nil { return err } + s.mu.Lock() + s.finalComposeSucceeded = true + s.mu.Unlock() + // Perform client-side CRC32C validation if a user-provided checksum was specified. if s.w.SendCRC32C && s.w.CRC32C != 0 && attrs.CRC32C != s.w.CRC32C { return fmt.Errorf("storage: object was uploaded, but its CRC32C (%d) does not match the expected CRC32C (%d)", attrs.CRC32C, s.w.CRC32C) @@ -590,8 +605,10 @@ func (s *pcuState) doCleanup() { cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(s.ctx), 5*time.Minute) defer cancel() // Only log cleanup errors here since its best effort and will rely on bucket - // lifecycle policies if cleanup fails. - if err := s.deleteFn(cleanupCtx, h); err != nil { + // lifecycle policies if cleanup fails. Objects that no longer exist were + // already removed by a successful compose with DeleteSourceObjects, so + // they are not reported. + if err := s.deleteFn(cleanupCtx, h); err != nil && !errors.Is(err, ErrObjectNotExist) { log.Printf("storage: failed to delete temporary part %q during parallel upload cleanup: %v", h.object, err) } } diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go index 594dfc52..b080dbec 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/trace.go @@ -84,12 +84,12 @@ func startSpanWithBucket(ctx context.Context, client *Client, bucket string, nam meta, hit := cache.get(bucket) if !hit { placeholder := bucketMetadata{ - resource: fmt.Sprintf("projects/_/buckets/%s", bucket), + resource: fmt.Sprintf("%sprojects/_/buckets/%s", storageResourceNamePrefix, bucket), location: "global", placeholder: true, } cache.put(bucket, placeholder) - cache.fetchBackground(bucket) + cache.fetchBackground(ctx, bucket) meta = placeholder } attrs := []attribute.KeyValue{ diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go index a660c834..db2161ac 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/writer.go @@ -184,7 +184,7 @@ type Writer struct { // // For parallel uploads, progress is reported when each part is successfully uploaded. // Therefore, the progress may be delayed relative to the standard upload, - // and jump in increments of PartSize (e.g. 16MiB). + // and jump in increments of the part size used (e.g. 16MiB). // // ProgressFunc should return quickly without blocking. ProgressFunc func(int64) diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/.release-please-manifest.json b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/.release-please-manifest.json index 07deff77..c955a3b1 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/.release-please-manifest.json +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/.release-please-manifest.json @@ -1,3 +1,3 @@ { - "v2": "2.24.1" + "v2": "2.26.2" } diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/CHANGES.md b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/CHANGES.md index 29b17d54..03a7df68 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/CHANGES.md +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/CHANGES.md @@ -1,5 +1,37 @@ # Changes +## [2.26.2](https://github.com/googleapis/gax-go/compare/v2.26.1...v2.26.2) (2026-09-26) + + +### Bug Fixes + +* **o11y:** name client request spans as {Service}.{Method} ([#560](https://github.com/googleapis/gax-go/issues/560)) ([1a022c8](https://github.com/googleapis/gax-go/commit/1a022c86141e8e02574dfdfbc979cc95111411e0)) + +## [2.26.1](https://github.com/googleapis/gax-go/compare/v2.26.0...v2.26.1) (2026-09-26) + + +### Bug Fixes + +* correct 1.27 protojson handling ([#558](https://github.com/googleapis/gax-go/issues/558)) ([c1e035d](https://github.com/googleapis/gax-go/commit/c1e035db278d4e60f22da30e5b4772788dd76b20)) +* **o11y:** remove deprioritized gcp.resource.destination.id span attribute ([#559](https://github.com/googleapis/gax-go/issues/559)) ([a6f48a2](https://github.com/googleapis/gax-go/commit/a6f48a2061f4cd09cd54054032a2200d519d0283)) + +## [2.26.0](https://github.com/googleapis/gax-go/compare/v2.25.0...v2.26.0) (2026-09-22) + + +### Features + +* **o11y:** add ClientLogging configuration ([95a5614](https://github.com/googleapis/gax-go/commit/95a5614930f1be8b152b9df8d05fa27e4b1ad9ac)) +* **o11y:** add recordActionableLog helper for L3 error logs ([3d6918a](https://github.com/googleapis/gax-go/commit/3d6918a7e86ee2e55b25f0c09e69f73bf91b6838)) +* **o11y:** add WithClientLogging CallOption ([0393829](https://github.com/googleapis/gax-go/commit/03938291cc30c7d01ad17a885f48e7edfe2623a5)) +* **o11y:** implement actionable error logging in Invoke ([a6867ed](https://github.com/googleapis/gax-go/commit/a6867ed2ea6da971192ba3a89aa9fcb3fa90cdc0)) + +## [2.25.0](https://github.com/googleapis/gax-go/compare/v2.24.1...v2.25.0) (2026-09-21) + + +### Features + +* update go min version to 1.26 ([#548](https://github.com/googleapis/gax-go/issues/548)) ([b943a7d](https://github.com/googleapis/gax-go/commit/b943a7d2b80b8c2acd519311084c92ec0a972b8b)) + ## [2.24.1](https://github.com/googleapis/gax-go/compare/v2.24.0...v2.24.1) (2026-09-03) diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/call_option.go b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/call_option.go index 8f5948eb..b844a5ee 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/call_option.go +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/call_option.go @@ -285,4 +285,44 @@ type CallSettings struct { // clientMetrics holds the pre-allocated OpenTelemetry metrics instruments // to use for this call. clientMetrics *ClientMetrics + + // clientTracing holds the pre-allocated OpenTelemetry tracer + // to use for this call. + clientTracing *ClientTracing + + // clientLogging holds the pre-allocated OpenTelemetry/slog logger + // to use for this call. + clientLogging *ClientLogging +} + +type clientTracingOpt struct { + ct *ClientTracing +} + +// Resolve applies the ClientTracing to the CallSettings. +func (o clientTracingOpt) Resolve(s *CallSettings) { + s.clientTracing = o.ct +} + +// WithClientTracing applies tracing instrumentation to the CallSettings. +// +// This is for internal use only. +func WithClientTracing(ct *ClientTracing) CallOption { + return clientTracingOpt{ct: ct} +} + +type clientLoggingOpt struct { + cl *ClientLogging +} + +// Resolve applies the ClientLogging to the CallSettings. +func (o clientLoggingOpt) Resolve(s *CallSettings) { + s.clientLogging = o.cl +} + +// WithClientLogging applies logging instrumentation to the CallSettings. +// +// This is for internal use only. +func WithClientLogging(cl *ClientLogging) CallOption { + return clientLoggingOpt{cl: cl} } diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/internal/version.go b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/internal/version.go index 845d8697..325b895d 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/internal/version.go +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/internal/version.go @@ -17,4 +17,4 @@ package internal // Version is the current tagged release of the library. -const Version = "2.24.1" +const Version = "2.26.2" diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/invoke.go b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/invoke.go index ea16c3c5..4b1f1a61 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/invoke.go +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/invoke.go @@ -37,6 +37,7 @@ import ( "github.com/googleapis/gax-go/v2/apierror" "github.com/googleapis/gax-go/v2/callctx" + "go.opentelemetry.io/otel/trace" ) // APICall is a user defined call stub. @@ -89,17 +90,42 @@ func invoke(ctx context.Context, call APICall, settings CallSettings, sp sleeper ctx = c } - if IsFeatureEnabled("METRICS") { - start := time.Now() - ctx = InjectTransportTelemetry(ctx, &TransportTelemetryData{}) + metricsEnabled := IsFeatureEnabled("METRICS") + tracingEnabled := IsFeatureEnabled("TRACING") + loggingEnabled := IsFeatureEnabled("LOGGING") + + retryCount := 0 + if metricsEnabled || tracingEnabled || loggingEnabled { + var start time.Time + if metricsEnabled { + start = time.Now() + } + if ExtractTransportTelemetry(ctx) == nil { + ctx = InjectTransportTelemetry(ctx, &TransportTelemetryData{}) + } + + var span trace.Span + if tracingEnabled { + ctx, span = startSpan(ctx, settings.clientTracing) + } + defer func() { - recordMetric(ctx, settings, time.Since(start), err) + if !metricsEnabled && !tracingEnabled && err == nil { + return + } + errInfo := ExtractTelemetryErrorInfo(ctx, err) + if metricsEnabled { + recordMetricWithInfo(ctx, settings, time.Since(start), &errInfo) + } + if tracingEnabled { + endSpan(ctx, span, &errInfo, err) + } + if loggingEnabled && err != nil { + recordActionableLog(ctx, settings.clientLogging, &errInfo, retryCount, err) + } }() } - retryCount := 0 - // Feature gate: GOOGLE_SDK_GO_EXPERIMENTAL_TRACING=true - tracingEnabled := IsFeatureEnabled("TRACING") for { ctxToUse := ctx if tracingEnabled { diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/proto_json_stream_go127.go b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/proto_json_stream_go127.go index cf647c42..93937e4d 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/proto_json_stream_go127.go +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/proto_json_stream_go127.go @@ -34,6 +34,7 @@ package gax import ( "encoding/json/jsontext" "errors" + "fmt" "io" "google.golang.org/protobuf/encoding/protojson" @@ -90,10 +91,14 @@ func (s *ProtoJSONStream) Recv() (proto.Message, error) { if kind == jsontext.KindBeginArray { // Read and discard the start of array. - err := s.stream.SkipValue() + tok, err := s.stream.ReadToken() if err != nil { return nil, err } + // Being pedantic here, because we missed this previously. + if tok.String() != "[" { + return nil, fmt.Errorf("unexpected opening token, expected '[', got %q", tok) + } } else { // stream isn't a JSON array. return nil, errBadOpening @@ -104,10 +109,13 @@ func (s *ProtoJSONStream) Recv() (proto.Message, error) { kind := s.stream.PeekKind() if kind == jsontext.KindEndArray { // We're at the end of the array. - err := s.stream.SkipValue() + tok, err := s.stream.ReadToken() if err != nil { return nil, err } + if tok.String() != "]" { + return nil, fmt.Errorf("unexpected closing token, expected ']', got %q", tok) + } return nil, io.EOF } diff --git a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/telemetry.go b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/telemetry.go index 2490c369..58e389aa 100644 --- a/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/telemetry.go +++ b/ci/resources/stemcell-version-bump/vendor/github.com/googleapis/gax-go/v2/telemetry.go @@ -35,6 +35,7 @@ import ( "fmt" "log/slog" "strconv" + "strings" "sync" "time" @@ -42,7 +43,9 @@ import ( "github.com/googleapis/gax-go/v2/callctx" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" + otelcodes "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/trace" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -122,15 +125,20 @@ const ( ClientVersion = "client_version" // ClientArtifact is the library name. E.g. "cloud.google.com/go/storage". ClientArtifact = "client_artifact" + // ClientRepo is the source repository for the client library. E.g. "googleapis/google-cloud-go". + ClientRepo = "client_repo" // RPCSystem is the RPC system type. E.g. "grpc" or "http". RPCSystem = "rpc_system" // URLDomain is the nominal service domain. E.g. "storage.googleapis.com". URLDomain = "url_domain" // Constants for telemetry attribute keys. - keyGCPClientService = "gcp.client.service" - keyRPCSystemName = "rpc.system.name" - keyURLDomain = "url.domain" + keyGCPClientArtifact = "gcp.client.artifact" + keyGCPClientRepo = "gcp.client.repo" + keyGCPClientService = "gcp.client.service" + keyGCPClientVersion = "gcp.client.version" + keyRPCSystemName = "rpc.system.name" + keyURLDomain = "url.domain" // SchemaURL specifies the OpenTelemetry schema version. schemaURL = "https://opentelemetry.io/schemas/1.39.0" @@ -189,11 +197,39 @@ func (a attrOpt) Resolve(opts *telemetryOptions) { opts.attributes = a.attrs } +func (a attrOpt) ResolveTracing(opts *tracingOptions) { + if opts != nil { + opts.attributes = a.attrs + } +} + +func (a attrOpt) ResolveLogging(opts *loggingOptions) { + if opts == nil || len(a.attrs) == 0 { + return + } + if opts.attributes == nil { + opts.attributes = make(map[string]string, len(a.attrs)) + } + for k, v := range a.attrs { + opts.attributes[k] = v + } +} + // WithTelemetryAttributes specifies the static attributes attachments. func WithTelemetryAttributes(attr map[string]string) TelemetryOption { return &attrOpt{attrs: attr} } +// WithTracingAttributes specifies the static attributes attachments for tracing. +func WithTracingAttributes(attr map[string]string) TracingOption { + return &attrOpt{attrs: attr} +} + +// WithLoggingAttributes specifies the static attributes attachments for logging. +func WithLoggingAttributes(attr map[string]string) LoggingOption { + return &attrOpt{attrs: attr} +} + type boundariesOpt struct { boundaries []float64 } @@ -440,6 +476,12 @@ func ExtractTelemetryErrorInfo(ctx context.Context, err error) TelemetryErrorInf // recordMetric records a duration measurement for the configured metric. func recordMetric(ctx context.Context, settings CallSettings, d time.Duration, err error) { + errInfo := ExtractTelemetryErrorInfo(ctx, err) + recordMetricWithInfo(ctx, settings, d, &errInfo) +} + +// recordMetricWithInfo records a duration measurement using pre-extracted error info. +func recordMetricWithInfo(ctx context.Context, settings CallSettings, d time.Duration, errInfo *TelemetryErrorInfo) { if settings.clientMetrics == nil || settings.clientMetrics.durationHistogram() == nil { return } @@ -452,8 +494,6 @@ func recordMetric(ctx context.Context, settings CallSettings, d time.Duration, e attrs := make([]attribute.KeyValue, 0, len(settings.clientMetrics.attributes())+5) attrs = append(attrs, settings.clientMetrics.attributes()...) - errInfo := ExtractTelemetryErrorInfo(ctx, err) - if td := ExtractTransportTelemetry(ctx); td != nil { if td.ServerAddress() != "" { attrs = append(attrs, attribute.String("server.address", td.ServerAddress())) @@ -481,3 +521,379 @@ func recordMetric(ctx context.Context, settings CallSettings, d time.Duration, e settings.clientMetrics.durationHistogram().Record(recordCtx, d.Seconds(), metric.WithAttributes(attrs...)) } + +// startSpan starts a client request span if ClientTracing and its tracer are configured. +func startSpan(ctx context.Context, ct *ClientTracing) (context.Context, trace.Span) { + if ct == nil { + return ctx, nil + } + tracer := ct.tracer() + if tracer == nil { + return ctx, nil + } + spanName := resolveSpanName(ctx) + staticAttrs := ct.attributes() + attrs := make([]attribute.KeyValue, 0, len(staticAttrs)+1) + attrs = append(attrs, staticAttrs...) + if urlTemplate, ok := callctx.TelemetryFromContext(ctx, "url_template"); ok && urlTemplate != "" { + if sanitized := sanitizeURLTemplate(urlTemplate); sanitized != "" { + attrs = append(attrs, attribute.String("url.template", sanitized)) + } + } + return tracer.Start( + ctx, + spanName, + trace.WithSpanKind(trace.SpanKindClient), + trace.WithAttributes(attrs...), + ) +} + +// endSpan records terminal status, diagnostic error details, and ends the client span. +func endSpan(ctx context.Context, span trace.Span, errInfo *TelemetryErrorInfo, err error) { + if span == nil { + return + } + defer span.End() + + if errInfo == nil { + info := ExtractTelemetryErrorInfo(ctx, err) + errInfo = &info + } + + attrs := make([]attribute.KeyValue, 0, 3+len(errInfo.Metadata)) + + if err != nil { + span.RecordError(err) + span.SetStatus(otelcodes.Error, errInfo.StatusCode) + attrs = append(attrs, + attribute.String("error.type", errInfo.ErrorType), + attribute.String("rpc.response.status_code", errInfo.StatusCode), + ) + if errInfo.Domain != "" { + attrs = append(attrs, attribute.String("gcp.errors.domain", errInfo.Domain)) + } + for k, v := range errInfo.Metadata { + attrs = append(attrs, attribute.String("gcp.errors.metadata."+k, v)) + } + } else { + span.SetStatus(otelcodes.Ok, "") + attrs = append(attrs, attribute.String("rpc.response.status_code", "OK")) + } + + if len(attrs) > 0 { + span.SetAttributes(attrs...) + } +} + +// ClientTracing contains the pre-allocated OpenTelemetry tracer and attributes +// for a specific generated Google Cloud client library. +// There should be exactly one ClientTracing instance instantiated per generated client. +type ClientTracing struct { + get func() clientTracingData +} + +type clientTracingData struct { + tracer trace.Tracer + attr []attribute.KeyValue +} + +type tracingOptions struct { + provider trace.TracerProvider + attributes map[string]string +} + +// TracingOption is an option to configure a ClientTracing instance. +// TracingOption works by modifying relevant fields of tracingOptions. +type TracingOption interface { + // ResolveTracing applies the option by modifying opts. + ResolveTracing(opts *tracingOptions) +} + +type tracerProviderOpt struct { + p trace.TracerProvider +} + +func (p tracerProviderOpt) ResolveTracing(opts *tracingOptions) { + if opts != nil { + opts.provider = p.p + } +} + +// WithTracerProvider specifies the trace.TracerProvider to use for client request tracing. +func WithTracerProvider(p trace.TracerProvider) TracingOption { + return &tracerProviderOpt{p: p} +} + +func (config *tracingOptions) tracerProvider() trace.TracerProvider { + if config != nil && config.provider != nil { + return config.provider + } + // Fall back to global tracer provider to prevent silent no-op bug! + return otel.GetTracerProvider() +} + +// NewClientTracing initializes and returns a new ClientTracing instance. +// It is intended to be called once per generated client during initialization. +func NewClientTracing(opts ...TracingOption) *ClientTracing { + var config tracingOptions + for _, opt := range opts { + if opt != nil { + opt.ResolveTracing(&config) + } + } + + return &ClientTracing{ + get: sync.OnceValue(func() clientTracingData { + provider := config.tracerProvider() + + var tracerOpts []trace.TracerOption + if val, ok := config.attributes[ClientVersion]; ok { + tracerOpts = append(tracerOpts, trace.WithInstrumentationVersion(val)) + } + tracerOpts = append(tracerOpts, trace.WithSchemaURL(schemaURL)) + + tracer := provider.Tracer(config.attributes[ClientArtifact], tracerOpts...) + + var attr []attribute.KeyValue + for _, m := range [...]struct{ attrKey, otelKey string }{ + {URLDomain, keyURLDomain}, + {RPCSystem, keyRPCSystemName}, + {ClientService, keyGCPClientService}, + } { + if val, ok := config.attributes[m.attrKey]; ok { + attr = append(attr, attribute.String(m.otelKey, val)) + } + } + attr = attr[:len(attr):len(attr)] + + return clientTracingData{ + tracer: tracer, + attr: attr, + } + }), + } +} + +func (ct *ClientTracing) tracer() trace.Tracer { + if ct == nil || ct.get == nil { + return nil + } + return ct.get().tracer +} + +func (ct *ClientTracing) attributes() []attribute.KeyValue { + if ct == nil || ct.get == nil { + return nil + } + return ct.get().attr +} + +// formatServiceMethod formats a protobuf RPC name (such as +// "google.cloud.secretmanager.v1.SecretManagerService/GetSecret") +// into a span name formatted as "Service.Method" (such as "SecretManager.GetSecret"). +// If the input does not contain a slash separator, it returns the input unchanged. +func formatServiceMethod(rpcMethod string) string { + slashIdx := strings.LastIndex(rpcMethod, "/") + if slashIdx == -1 { + return rpcMethod + } + + service := rpcMethod[:slashIdx] + if dotIdx := strings.LastIndex(service, "."); dotIdx != -1 { + service = service[dotIdx+1:] + } + if trimmed := strings.TrimSuffix(service, "Service"); trimmed != "" { + service = trimmed + } + + method := rpcMethod[slashIdx+1:] + if service == "" || method == "" { + return rpcMethod + } + + return service + "." + method +} + +func resolveSpanName(ctx context.Context) string { + if clientMethod, ok := callctx.TelemetryFromContext(ctx, "client_method"); ok && clientMethod != "" { + return clientMethod + } + if rpcMethod, ok := callctx.TelemetryFromContext(ctx, "rpc_method"); ok && rpcMethod != "" { + return formatServiceMethod(rpcMethod) + } + httpMethod, okHTTP := callctx.TelemetryFromContext(ctx, "http_method") + urlTemplate, okURL := callctx.TelemetryFromContext(ctx, "url_template") + if okHTTP && httpMethod != "" && okURL && urlTemplate != "" { + if sanitized := sanitizeURLTemplate(urlTemplate); sanitized != "" { + return httpMethod + " " + sanitized + } + } + return "gcp.client.request" +} + +// sanitizeURLTemplate removes query parameters and URL fragments from a raw URL template, +// ensuring only the path template is retained. +func sanitizeURLTemplate(rawURL string) string { + if idx := strings.IndexAny(rawURL, "?#"); idx != -1 { + return rawURL[:idx] + } + return rawURL +} + +// ClientLogging contains the pre-allocated slog.Logger and static attributes +// for a specific generated Google Cloud client library. +// There should be exactly one ClientLogging instance instantiated per generated client. +type ClientLogging struct { + get func() clientLoggingData +} + +type clientLoggingData struct { + logger *slog.Logger + attr []slog.Attr +} + +type loggingOptions struct { + logger *slog.Logger + attributes map[string]string +} + +// LoggingOption is an option to configure a ClientLogging instance. +// LoggingOption works by modifying relevant fields of loggingOptions. +type LoggingOption interface { + // ResolveLogging applies the option by modifying opts. + ResolveLogging(opts *loggingOptions) +} + +type loggerProviderOpt struct { + logger *slog.Logger +} + +func (p loggerProviderOpt) ResolveLogging(opts *loggingOptions) { + if opts != nil { + opts.logger = p.logger + } +} + +// WithLoggerProvider specifies the slog.Logger to use for client request logging. +func WithLoggerProvider(logger *slog.Logger) LoggingOption { + return &loggerProviderOpt{logger: logger} +} + +func (config *loggingOptions) loggerProvider() *slog.Logger { + if config != nil && config.logger != nil { + return config.logger + } + // Fall back to default slog logger to prevent silent no-op bug! + return slog.Default() +} + +// NewClientLogging initializes and returns a new ClientLogging instance. +// It is intended to be called once per generated client during initialization. +func NewClientLogging(opts ...LoggingOption) *ClientLogging { + var config loggingOptions + for _, opt := range opts { + if opt != nil { + opt.ResolveLogging(&config) + } + } + + var attr []slog.Attr + for _, m := range [...]struct{ attrKey, slogKey string }{ + {ClientArtifact, keyGCPClientArtifact}, + {ClientRepo, keyGCPClientRepo}, + {ClientService, keyGCPClientService}, + {ClientVersion, keyGCPClientVersion}, + {RPCSystem, keyRPCSystemName}, + {URLDomain, keyURLDomain}, + } { + if val, ok := config.attributes[m.attrKey]; ok { + attr = append(attr, slog.String(m.slogKey, val)) + } + } + attr = attr[:len(attr):len(attr)] + customLogger := config.logger + + return &ClientLogging{ + get: sync.OnceValue(func() clientLoggingData { + logger := customLogger + if logger == nil { + logger = slog.Default() + } + + return clientLoggingData{ + logger: logger, + attr: attr, + } + }), + } +} + +func (cl *ClientLogging) logger() *slog.Logger { + if cl == nil || cl.get == nil { + return nil + } + return cl.get().logger +} + +func (cl *ClientLogging) attributes() []slog.Attr { + if cl == nil || cl.get == nil { + return nil + } + return cl.get().attr +} + +// recordActionableLog emits a structured warning log strictly upon terminal failure. +func recordActionableLog(ctx context.Context, cl *ClientLogging, errInfo *TelemetryErrorInfo, retries int, err error) { + if cl == nil || err == nil { + return + } + if ctx == nil { + ctx = context.Background() + } + logger := cl.logger() + if logger == nil || !logger.Enabled(ctx, slog.LevelWarn) { + return + } + + if errInfo == nil { + info := ExtractTelemetryErrorInfo(ctx, err) + errInfo = &info + } + + attrs := append([]slog.Attr(nil), cl.attributes()...) + + if rpcMethod, ok := callctx.TelemetryFromContext(ctx, "rpc_method"); ok && rpcMethod != "" { + attrs = append(attrs, slog.String("rpc.method", rpcMethod)) + } + if urlTemplate, ok := callctx.TelemetryFromContext(ctx, "url_template"); ok && urlTemplate != "" { + if sanitized := sanitizeURLTemplate(urlTemplate); sanitized != "" { + attrs = append(attrs, slog.String("url.template", sanitized)) + } + } + if td := ExtractTransportTelemetry(ctx); td != nil { + if td.ServerAddress() != "" { + attrs = append(attrs, slog.String("server.address", td.ServerAddress())) + } + if td.ServerPort() != 0 { + attrs = append(attrs, slog.Int("server.port", td.ServerPort())) + } + } + if errInfo.ErrorType != "" { + attrs = append(attrs, slog.String("error.type", errInfo.ErrorType)) + } + if errInfo.StatusCode != "" { + attrs = append(attrs, slog.String("rpc.response.status_code", errInfo.StatusCode)) + } + attrs = append(attrs, + slog.String("error.message", err.Error()), + slog.Int("resend_count", retries), + ) + if errInfo.Domain != "" { + attrs = append(attrs, slog.String("gcp.errors.domain", errInfo.Domain)) + } + for k, v := range errInfo.Metadata { + attrs = append(attrs, slog.String("gcp.errors.metadata."+k, v)) + } + + logger.LogAttrs(ctx, slog.LevelWarn, "gcp.client.request", attrs...) +} diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-function.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-function.go index 3eea78e5..e783023f 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-function.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-function.go @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 // Package gcp provides a resource detector for GCP Cloud Function. -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import ( "context" @@ -10,7 +10,7 @@ import ( "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.41.0" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) const ( diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-run.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-run.go index 98bcfdb9..b0daeb7a 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-run.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/cloud-run.go @@ -1,7 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import ( "context" @@ -12,7 +12,7 @@ import ( "cloud.google.com/go/compute/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.41.0" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) const serviceNamespace = "cloud-run-managed" diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/detector.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/detector.go index 11b0d673..a5a0a8cf 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/detector.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/detector.go @@ -1,10 +1,11 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import ( "context" + "errors" "fmt" "strconv" @@ -12,7 +13,7 @@ import ( "github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.41.0" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) // NewDetector returns a resource detector which detects resource attributes on: @@ -149,7 +150,7 @@ func (r *resourceBuilder) addZoneOrRegion(detect func() (string, gcp.LocationTyp func (r *resourceBuilder) build() (*resource.Resource, error) { var err error if len(r.errs) > 0 { - err = fmt.Errorf("%w: %s", resource.ErrPartialResource, r.errs) + err = fmt.Errorf("%w: %w", resource.ErrPartialResource, errors.Join(r.errs...)) } return resource.NewWithAttributes(semconv.SchemaURL, r.attrs...), err } diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gce.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gce.go index 800ca29a..fd6be947 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gce.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gce.go @@ -1,7 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import ( "context" @@ -13,7 +13,7 @@ import ( "cloud.google.com/go/compute/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.41.0" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) // GCE collects resource information of GCE computing instances. diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gke.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gke.go index f284f4d2..16ade0e0 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gke.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/gke.go @@ -1,7 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import ( "context" @@ -11,7 +11,7 @@ import ( "cloud.google.com/go/compute/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/resource" - semconv "go.opentelemetry.io/otel/semconv/v1.41.0" + semconv "go.opentelemetry.io/otel/semconv/v1.43.0" ) // GKE collects resource information of GKE computing instances. diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/types.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/types.go index 666d82e6..99ffb437 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/types.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/types.go @@ -1,7 +1,7 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp import "github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp" diff --git a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/version.go b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/version.go index 60344e17..6edd1f27 100644 --- a/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/version.go +++ b/ci/resources/stemcell-version-bump/vendor/go.opentelemetry.io/contrib/detectors/gcp/version.go @@ -1,11 +1,11 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package gcp // import "go.opentelemetry.io/contrib/detectors/gcp" +package gcp // Version is the current release version of the GCP resource detector. func Version() string { - return "1.44.0" + return "1.45.0" // This string is updated by the pre_release.sh script during release } diff --git a/ci/resources/stemcell-version-bump/vendor/modules.txt b/ci/resources/stemcell-version-bump/vendor/modules.txt index 9ec19349..718ad693 100644 --- a/ci/resources/stemcell-version-bump/vendor/modules.txt +++ b/ci/resources/stemcell-version-bump/vendor/modules.txt @@ -44,14 +44,14 @@ cloud.google.com/go/iam/apiv1/iampb cloud.google.com/go/monitoring/apiv3/v2 cloud.google.com/go/monitoring/apiv3/v2/monitoringpb cloud.google.com/go/monitoring/internal -# cloud.google.com/go/storage v1.68.0 -## explicit; go 1.25.0 +# cloud.google.com/go/storage v1.69.0 +## explicit; go 1.26.0 cloud.google.com/go/storage cloud.google.com/go/storage/experimental cloud.google.com/go/storage/internal cloud.google.com/go/storage/internal/apiv2 cloud.google.com/go/storage/internal/apiv2/storagepb -# github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.34.0 +# github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.35.0 ## explicit; go 1.25.0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp # github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.57.0 @@ -166,8 +166,8 @@ github.com/google/uuid ## explicit; go 1.25.0 github.com/googleapis/enterprise-certificate-proxy/client github.com/googleapis/enterprise-certificate-proxy/client/util -# github.com/googleapis/gax-go/v2 v2.24.1 -## explicit; go 1.25.0 +# github.com/googleapis/gax-go/v2 v2.26.2 +## explicit; go 1.26.0 github.com/googleapis/gax-go/v2 github.com/googleapis/gax-go/v2/apierror github.com/googleapis/gax-go/v2/apierror/internal/proto @@ -208,7 +208,7 @@ github.com/stretchr/testify/require ## explicit; go 1.24.0 go.opentelemetry.io/auto/sdk go.opentelemetry.io/auto/sdk/internal/telemetry -# go.opentelemetry.io/contrib/detectors/gcp v1.44.0 +# go.opentelemetry.io/contrib/detectors/gcp v1.45.0 ## explicit; go 1.25.0 go.opentelemetry.io/contrib/detectors/gcp # go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0