diff --git a/ci/resources/stemcell-version-bump/go.mod b/ci/resources/stemcell-version-bump/go.mod index b928e2e6..7569a22c 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.25.8 require ( - cloud.google.com/go/storage v1.64.0 + cloud.google.com/go/storage v1.65.0 github.com/stretchr/testify v1.11.1 google.golang.org/api v0.292.0 ) diff --git a/ci/resources/stemcell-version-bump/go.sum b/ci/resources/stemcell-version-bump/go.sum index 356f6257..ab9ed225 100644 --- a/ci/resources/stemcell-version-bump/go.sum +++ b/ci/resources/stemcell-version-bump/go.sum @@ -16,8 +16,8 @@ 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.29.0 h1:AHhDsFaSax1/4k+qlIDX/SDGe6hggnfXJ9dkgD9qBPY= cloud.google.com/go/monitoring v1.29.0/go.mod h1:72NOVjJXHY/HBfoLT0+qlCZBT059+9VXLeAnL2PeeVM= -cloud.google.com/go/storage v1.64.0 h1:KLpxI/oX9LxeRsNqn877d2WyeT3ryiEwnGt8pwcSPZg= -cloud.google.com/go/storage v1.64.0/go.mod h1:lWyAtwvDZHdL3k68WVKbESP6bmWaV23ZJJ/JEVw/ZaQ= +cloud.google.com/go/storage v1.65.0 h1:McbFt5j+hTNx+dkFuzq7teakIKcpqGp/cJZRxMyfvAc= +cloud.google.com/go/storage v1.65.0/go.mod h1:UsS9OgFg/XHOSYakQ8ZtLWWeyGkk1WnmD/GsGfN0BHM= 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.33.0 h1:l7+6kwRMJNwdCvYdDl7Eax+wzEYHSnNY7zrrfbhDdTA= 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 3ddb1f06..29019b4a 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,14 @@ # Changes +## [1.65.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.64.0...storage/v1.65.0) (2026-08-14) + + +### Features + +* Update API sources and regenerate ([#20252](https://github.com/googleapis/google-cloud-go/issues/20252)) ([ac017a0](https://github.com/googleapis/google-cloud-go/commit/ac017a0d30ec2d42db35f7d3b5ea7fb20b4b860a)) +* Update API sources and regenerate ([#20369](https://github.com/googleapis/google-cloud-go/issues/20369)) ([bfe3aaf](https://github.com/googleapis/google-cloud-go/commit/bfe3aafc49e0c13815aa0401b1eba47aef4511df)) + ## [1.64.0](https://github.com/googleapis/google-cloud-go/compare/storage/v1.63.1...storage/v1.64.0) (2026-07-21) diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/env.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/env.go new file mode 100644 index 00000000..b7a186ce --- /dev/null +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/env.go @@ -0,0 +1,34 @@ +// 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 storage + +const ( + // envOtelMetrics is the environment variable that allows the user to + // enable OpenTelemetry metrics. When set to "true", the storage client will + // capture standard operation and transport-level metrics. + envOtelMetrics = "GCP_STORAGE_GO_ENABLE_OTEL_METRICS" + + // envOtelDebugMetrics is the environment variable that allows the user to + // enable advanced OpenTelemetry debug and network metrics. When set to "true", + // the storage client will capture additional connection metrics such as DNS, TCP, and TLS durations. + envOtelDebugMetrics = "GCP_STORAGE_GO_ENABLE_OTEL_DEBUG_METRICS" + + // storageOtelTracingDevVar is the environment variable that enables OTel tracing. + storageOtelTracingDevVar = "GO_STORAGE_DEV_OTEL_TRACING" + + // storageBucketMetadataDisabledVar is the environment variable that disables + // Autorequester Bucket Metadata optimization. + storageBucketMetadataDisabledVar = "GO_OTEL_BUCKETMETADATA_DISABLED" +) 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 dbad7b63..ae77e8d5 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 @@ -122,3 +122,10 @@ func WithDirectConnectivityEnforced() option.ClientOption { func WithOtelMetrics() option.ClientOption { return internal.WithOtelMetrics.(func() option.ClientOption)() } + +// WithOtelDebugMetrics provides an [option.ClientOption] that may be passed to +// [cloud.google.com/go/storage.NewClient] or [cloud.google.com/go/storage.NewGRPCClient]. +// It enables debug client-side OpenTelemetry metrics. +func WithOtelDebugMetrics() option.ClientOption { + return internal.WithOtelDebugMetrics.(func() option.ClientOption)() +} 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 16adf561..3819c49d 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 @@ -26,10 +26,12 @@ import ( "strconv" "strings" + "cloud.google.com/go/auth" "cloud.google.com/go/iam/apiv1/iampb" gapic "cloud.google.com/go/storage/internal/apiv2" "cloud.google.com/go/storage/internal/apiv2/storagepb" "github.com/googleapis/gax-go/v2" + "google.golang.org/api/iterator" "google.golang.org/api/option" "google.golang.org/api/option/internaloption" @@ -178,17 +180,14 @@ func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (client *g var clientMetrics *clientMetrics var metricsCleanup func() if isOtelMetricsEnabled(&config) { - var project string - if c, err := transport.Creds(ctx, s.clientOption...); err == nil { - project = c.ProjectID - } - clientMetrics, metricsCleanup = initClientMetrics(ctx, project, &config) + clientMetrics, metricsCleanup = initGRPCMetricsAndWrapCredentials(ctx, &config, s) if clientMetrics != nil { unaryInt, streamInt := metricsInterceptors(clientMetrics) s.clientOption = append(s.clientOption, option.WithGRPCDialOption(grpc.WithChainUnaryInterceptor(unaryInt)), option.WithGRPCDialOption(grpc.WithChainStreamInterceptor(streamInt)), ) + s.clientOption = append(s.clientOption, grpcNetworkMetricsDialOptions("storage.googleapis.com", clientMetrics)...) } } @@ -222,6 +221,28 @@ func newGRPCStorageClient(ctx context.Context, opts ...storageOption) (client *g return c, nil } +func initGRPCMetricsAndWrapCredentials(ctx context.Context, config *storageConfig, s *settings) (*clientMetrics, func()) { + var project string + var authCreds *auth.Credentials + + credsOpts := append([]option.ClientOption{option.WithScopes(gapic.DefaultAuthScopes()...)}, s.clientOption...) + if c, err := internaloption.AuthCreds(ctx, credsOpts); err == nil { + authCreds = c + project, _ = authCreds.ProjectID(ctx) + } else if c, err := transport.Creds(ctx, credsOpts...); err == nil { + project = c.ProjectID + } + + clientMetrics, metricsCleanup := initClientMetrics(ctx, project, config) + if clientMetrics != nil { + if authCreds != nil { + authCreds = wrapAuthCredentials(authCreds, clientMetrics) + s.clientOption = append(s.clientOption, option.WithAuthCredentials(authCreds)) + } + } + return clientMetrics, metricsCleanup +} + // configureStreamingTimeouts explicitly overrides default call timeouts to 0 (unbounded) // for all generated payload streaming RPCs. This guarantees that long-running data reads // and writes are not prematurely aborted by default transport deadlines, while allowing 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 62971346..97498f7a 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 @@ -35,11 +35,13 @@ import ( "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" "google.golang.org/api/iterator" "google.golang.org/api/option" "google.golang.org/api/option/internaloption" raw "google.golang.org/api/storage/v1" + "google.golang.org/api/transport" htransport "google.golang.org/api/transport/http" ) @@ -70,6 +72,7 @@ func newHTTPStorageClient(ctx context.Context, opts ...storageOption) (client st config := newStorageConfig(o...) var creds *auth.Credentials + var googleCreds *google.Credentials // In general, it is recommended to use raw.NewService instead of htransport.NewClient // since raw.NewService configures the correct default endpoints when initializing the // internal http client. However, in our case, "NewRangeReader" in reader.go needs to @@ -91,6 +94,9 @@ func newHTTPStorageClient(ctx context.Context, opts ...storageOption) (client st if err == nil { creds = c o = append(o, option.WithAuthCredentials(creds)) + } else if gc, err := transport.Creds(ctx, o...); err == nil { + googleCreds = gc + o = append(o, option.WithCredentials(googleCreds)) } } else { var hostURL *url.URL @@ -118,18 +124,14 @@ func newHTTPStorageClient(ctx context.Context, opts ...storageOption) (client st } s.clientOption = o - // htransport selects the correct endpoint among WithEndpoint (user override), WithDefaultEndpointTemplate, and WithDefaultMTLSEndpoint. - hc, ep, err := htransport.NewClient(ctx, s.clientOption...) - if err != nil { - return nil, fmt.Errorf("dialing: %w", err) - } - var clientMetrics *clientMetrics var metricsCleanup func() if isOtelMetricsEnabled(&config) { var project string if creds != nil { project, _ = creds.ProjectID(ctx) + } else if googleCreds != nil { + project = googleCreds.ProjectID } clientMetrics, metricsCleanup = initClientMetrics(ctx, project, &config) } @@ -141,6 +143,17 @@ func newHTTPStorageClient(ctx context.Context, opts ...storageOption) (client st }() } + if clientMetrics != nil && creds != nil { + creds = wrapAuthCredentials(creds, clientMetrics) + s.clientOption = append(s.clientOption, option.WithAuthCredentials(creds)) + } + + // htransport selects the correct endpoint among WithEndpoint (user override), WithDefaultEndpointTemplate, and WithDefaultMTLSEndpoint. + hc, ep, err := htransport.NewClient(ctx, s.clientOption...) + if err != nil { + return nil, fmt.Errorf("dialing: %w", err) + } + // Clone the http.Client to avoid modifying the original one if it was provided by the user. hcClone := *hc c := &httpStorageClient{ @@ -1030,6 +1043,7 @@ func (c *httpStorageClient) newRangeReaderXML(ctx context.Context, params *newRa select { case <-timer: log.Printf("[%s] stalled read-req cancelled after %fs", requestID, stallTimeout.Seconds()) + c.metrics.recordStallDuration(ctx, stallTimeout, "ReadObject", "http", stripPort(req.URL.Host)) cancel() <-done if res != nil && res.Body != nil { diff --git a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go index 3f6d33bd..94eb1d8f 100644 --- a/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go +++ b/ci/resources/stemcell-version-bump/vendor/cloud.google.com/go/storage/internal/experimental.go @@ -51,4 +51,8 @@ var ( // WithOtelMetrics is a function which is implemented by the storage package. // It enables client-side OpenTelemetry metrics. WithOtelMetrics any // func() option.ClientOption + + // WithOtelDebugMetrics is a function which is implemented by the storage package. + // It enables debug client-side OpenTelemetry metrics. + WithOtelDebugMetrics any // func() option.ClientOption ) 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 6aaa3e33..2782c08a 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.64.0" +const Version = "1.65.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 d18045d6..978661e0 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 @@ -16,12 +16,14 @@ package storage import ( "context" + "crypto/tls" "errors" "fmt" "io" "log" "net" "net/http" + "net/http/httptrace" "os" "strconv" "strings" @@ -29,17 +31,23 @@ import ( "sync/atomic" "time" + "cloud.google.com/go/auth" "cloud.google.com/go/iam/apiv1/iampb" "cloud.google.com/go/storage/internal" mexporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric" + "go.opentelemetry.io/contrib/detectors/gcp" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.opentelemetry.io/otel/sdk/resource" + "google.golang.org/api/googleapi" + "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/stats" "google.golang.org/grpc/status" ) @@ -59,6 +67,16 @@ type clientMetrics struct { responseBodySize metric.Int64Histogram ttfb metric.Float64Histogram errors metric.Int64Counter + activeRequests metric.Int64UpDownCounter + gfeHeaderMissing metric.Int64Counter + dnsLookupDuration metric.Float64Histogram + tcpConnectDuration metric.Float64Histogram + tlsHandshakeDuration metric.Float64Histogram + gfeDuration metric.Float64Histogram + credentialRefreshDuration metric.Float64Histogram + networkBytesSent metric.Int64Counter + networkBytesReceived metric.Int64Counter + stallDuration metric.Float64Histogram } func formatMetricWithPrefix(m metricdata.Metrics, prefix string) string { @@ -66,12 +84,14 @@ func formatMetricWithPrefix(m metricdata.Metrics, prefix string) string { } // isOtelMetricsEnabled checks if Otel metrics are enabled. -// The environment variable GCP_STORAGE_GO_ENABLE_OTEL_METRICS takes precedence. +// The environment variable GCP_STORAGE_GO_ENABLE_OTEL_METRICS takes precedence +// over the config option. Note that if metrics are completely disabled via +// disableClientMetrics, this will always return false regardless of the environment variable. func isOtelMetricsEnabled(config *storageConfig) bool { if config.disableClientMetrics { return false } - if valStr, present := os.LookupEnv("GCP_STORAGE_GO_ENABLE_OTEL_METRICS"); present { + if valStr, present := os.LookupEnv(envOtelMetrics); present { v, err := strconv.ParseBool(valStr) if err == nil { return v @@ -80,6 +100,23 @@ func isOtelMetricsEnabled(config *storageConfig) bool { return config.enableOtelMetrics } +// isOtelDebugMetricsEnabled checks if debug Otel metrics are enabled. +// The environment variable GCP_STORAGE_GO_ENABLE_OTEL_DEBUG_METRICS takes precedence +// over the config option. Note that if metrics are completely disabled via +// disableClientMetrics, this will always return false regardless of the environment variable. +func isOtelDebugMetricsEnabled(config *storageConfig) bool { + if config.disableClientMetrics { + return false + } + if valStr, present := os.LookupEnv(envOtelDebugMetrics); present { + v, err := strconv.ParseBool(valStr) + if err == nil { + return v + } + } + return config.enableOtelDebugMetrics +} + // newMetricsGCMExporter creates a Google Cloud Monitoring exporter. func newMetricsGCMExporter(ctx context.Context, projectID string) (sdkmetric.Exporter, error) { exporter, err := mexporter.New( @@ -87,7 +124,36 @@ func newMetricsGCMExporter(ctx context.Context, projectID string) (sdkmetric.Exp mexporter.WithMetricDescriptorTypeFormatter(func(m metricdata.Metrics) string { return formatMetricWithPrefix(m, customMetricPrefix) }), - mexporter.WithCreateServiceTimeSeries(), + // The OTel GCP exporter drops any resource attributes that don't map to the + // target MonitoredResource (currently generic_node). + // We use WithFilteredResourceAttributes returning true to ensure that ANY + // resource attributes not in the MonitoredResource schema (like gcp.client.*, + // or gcp detector attributes not supported by generic_node) are preserved + // as metric labels instead of being dropped. + // + // TODO: When storage_client node is allowlisted in Monarch + // (google3/configs/monitoring/cloud_pulse_monarch/storage/storage_client.proto), + // we should uncomment the following options to export to the custom resource: + // mexporter.WithCreateServiceTimeSeries(), + // mexporter.WithMonitoredResourceDescription("storage.googleapis.com/Client", []string{"project_id", "location", "cloud_platform", "host_id", "instance_id", "api"}), + mexporter.WithFilteredResourceAttributes(func(kv attribute.KeyValue) bool { + key := string(kv.Key) + // Keep our custom gcp.client.* attributes as labels + if strings.HasPrefix(key, "gcp.client.") { + return true + } + // Keep attributes that can later be mapped to the storage.googleapis.com/Client + // MonitoredResource schema in Monarch. + switch key { + case "cloud.platform", + "cloud.region", + "cloud.availability_zone", + "host.id": + return true + } + // Drop the rest to avoid hitting the Cloud Monitoring label limit. + return false + }), ) if err != nil { return nil, fmt.Errorf("storage: creating GCM exporter: %w", err) @@ -123,6 +189,7 @@ func initMetrics(ctx context.Context, projectID string, config *storageConfig) ( // Static common attributes are defined as Resource Attributes. res, err := resource.New(ctx, + resource.WithDetectors(gcp.NewDetector()), resource.WithAttributes( attribute.String("gcp.client.version", internal.Version), attribute.String("gcp.client.service", "storage"), @@ -162,6 +229,26 @@ func initMetrics(ctx context.Context, projectID string, config *storageConfig) ( sdkmetric.Instrument{Name: "gcp.storage.client.response.body.size", Kind: sdkmetric.InstrumentKindHistogram}, sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: sizeHistogramBoundaries()}}, ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "gcp.storage.client.network.dns.lookup.duration", Kind: sdkmetric.InstrumentKindHistogram}, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: latencyHistogramBoundaries()}}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "gcp.storage.client.network.tcp.connect.duration", Kind: sdkmetric.InstrumentKindHistogram}, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: latencyHistogramBoundaries()}}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "gcp.storage.client.network.tls.handshake.duration", Kind: sdkmetric.InstrumentKindHistogram}, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: latencyHistogramBoundaries()}}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "gcp.storage.client.gfe.duration", Kind: sdkmetric.InstrumentKindHistogram}, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: latencyHistogramBoundaries()}}, + ), + sdkmetric.NewView( + sdkmetric.Instrument{Name: "gcp.storage.client.auth.credential_refresh.duration", Kind: sdkmetric.InstrumentKindHistogram}, + sdkmetric.Stream{Aggregation: sdkmetric.AggregationExplicitBucketHistogram{Boundaries: latencyHistogramBoundaries()}}, + ), ), ) ownProvider = true @@ -250,6 +337,109 @@ func initMetrics(ctx context.Context, projectID string, config *storageConfig) ( return nil, nil, err } + var activeRequests metric.Int64UpDownCounter + var gfeHeaderMissing metric.Int64Counter + var dnsLookupDuration metric.Float64Histogram + var tcpConnectDuration metric.Float64Histogram + var tlsHandshakeDuration metric.Float64Histogram + var gfeDuration metric.Float64Histogram + var credentialRefreshDuration metric.Float64Histogram + + var networkBytesSent metric.Int64Counter + var networkBytesReceived metric.Int64Counter + var stallDuration metric.Float64Histogram + + if isOtelDebugMetricsEnabled(config) { + networkBytesSent, err = meter.Int64Counter( + "gcp.storage.client.network.egress_bytes_count", + metric.WithDescription("Total physical bytes sent over the wire socket (gRPC only)."), + metric.WithUnit("By"), + ) + if err != nil { + return nil, nil, err + } + + networkBytesReceived, err = meter.Int64Counter( + "gcp.storage.client.network.ingress_bytes_count", + metric.WithDescription("Total physical bytes received over the wire socket (gRPC only)."), + metric.WithUnit("By"), + ) + if err != nil { + return nil, nil, err + } + + 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.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + + credentialRefreshDuration, err = meter.Float64Histogram( + "gcp.storage.client.auth.credential_refresh.duration", + metric.WithDescription("Duration of the background API/network calls made to refresh OAuth2/JWT access credentials."), + metric.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + activeRequests, err = meter.Int64UpDownCounter( + "gcp.storage.client.active_requests", + metric.WithDescription("Number of active GCS client requests"), + metric.WithUnit("1"), + ) + if err != nil { + return nil, nil, err + } + + gfeHeaderMissing, err = meter.Int64Counter( + "gcp.storage.client.gfe.header_missing", + metric.WithDescription("Number of GCS requests where the X-Goog-Gfe-Service-Time header was missing"), + metric.WithUnit("1"), + ) + if err != nil { + return nil, nil, err + } + + dnsLookupDuration, err = meter.Float64Histogram( + "gcp.storage.client.network.dns.lookup.duration", + metric.WithDescription("Time taken for DNS lookup"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + + tcpConnectDuration, err = meter.Float64Histogram( + "gcp.storage.client.network.tcp.connect.duration", + metric.WithDescription("Time taken for TCP connection"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + + tlsHandshakeDuration, err = meter.Float64Histogram( + "gcp.storage.client.network.tls.handshake.duration", + metric.WithDescription("Time taken to perform a TLS handshake"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + + gfeDuration, err = meter.Float64Histogram( + "gcp.storage.client.gfe.duration", + metric.WithDescription("GFE proxy processing time"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, nil, err + } + } + cm := &clientMetrics{ provider: provider, rpcClientCallDuration: rpcDuration, @@ -261,6 +451,16 @@ func initMetrics(ctx context.Context, projectID string, config *storageConfig) ( responseBodySize: responseBodySize, ttfb: ttfb, errors: errors, + activeRequests: activeRequests, + gfeHeaderMissing: gfeHeaderMissing, + dnsLookupDuration: dnsLookupDuration, + tcpConnectDuration: tcpConnectDuration, + tlsHandshakeDuration: tlsHandshakeDuration, + gfeDuration: gfeDuration, + credentialRefreshDuration: credentialRefreshDuration, + networkBytesSent: networkBytesSent, + networkBytesReceived: networkBytesReceived, + stallDuration: stallDuration, } var cleanup func() @@ -348,8 +548,16 @@ func computeErrorType(err error, isHTTP bool, statusCode int64) string { return "AUTHENTICATION_ERROR" } - if strings.Contains(errStr, "connection refused") || strings.Contains(errStr, "dial tcp") || strings.Contains(errStr, "no such host") || strings.Contains(errStr, "broken pipe") || strings.Contains(errStr, "connection reset") || strings.Contains(errStr, "eof") { - return "CONNECTIVITY" + if strings.Contains(errStr, "no such host") || strings.Contains(errStr, "no address") { + return "DNS_FAILURE" + } + + if strings.Contains(errStr, "connection refused") || strings.Contains(errStr, "connection reset") || strings.Contains(errStr, "dial tcp") || strings.Contains(errStr, "broken pipe") || strings.Contains(errStr, "eof") { + return "CONNECTION_ERROR" + } + + if strings.Contains(errStr, "tls") || strings.Contains(errStr, "certificate") || strings.Contains(errStr, "x509") { + return "TLS_FAILURE" } if !isHTTP { @@ -412,12 +620,9 @@ func (cm *clientMetrics) recordRPC(ctx context.Context, method, target string, d service := "google.storage.v2.Storage" methodName := method - if strings.HasPrefix(method, "/") { - parts := strings.Split(strings.TrimPrefix(method, "/"), "/") - if len(parts) >= 2 { - service = parts[0] - methodName = parts[1] - } + if idx := strings.LastIndex(methodName, "/"); idx != -1 && idx > 0 { + service = methodName[1:idx] + methodName = methodName[idx+1:] } errorType := computeErrorType(err, false, statusCode) @@ -432,7 +637,7 @@ func (cm *clientMetrics) recordRPC(ctx context.Context, method, target string, d attribute.String("error.type", errorType), } - cm.rpcClientCallDuration.Record(ctx, duration, metric.WithAttributes(attrs...)) + cm.rpcClientCallDuration.Record(ctx, duration, metric.WithAttributes(injectAPIMethod(ctx, attrs)...)) // Record standard attempt metric: gcp.storage.client.attempts. state := metricsStateFromContext(ctx) @@ -440,27 +645,32 @@ func (cm *clientMetrics) recordRPC(ctx context.Context, method, target string, d if state != nil { logicalMethod = state.method } - attemptAttrs := []attribute.KeyValue{ + attemptAttrs := make([]attribute.KeyValue, 0, 5) + attemptAttrs = append(attemptAttrs, + attribute.String("rpc.system.name", "grpc"), attribute.String("rpc.method", logicalMethod), attribute.Int64("rpc.grpc.status_code", statusCode), attribute.String("error.type", errorType), - } - cm.attempts.Add(ctx, 1, metric.WithAttributes(attemptAttrs...)) + ) + cm.attempts.Add(ctx, 1, metric.WithAttributes(injectAPIMethod(ctx, attemptAttrs)...)) // Record standard error metric: gcp.storage.client.errors. if err != nil && err != io.EOF { - errorAttrs := []attribute.KeyValue{ + errorAttrs := make([]attribute.KeyValue, 0, 5) + errorAttrs = append(errorAttrs, + attribute.String("rpc.system.name", "grpc"), attribute.String("rpc.method", logicalMethod), attribute.String("error.type", errorType), attribute.String("gcp.errors.domain", "storage.googleapis.com"), - } - cm.errors.Add(ctx, 1, metric.WithAttributes(errorAttrs...)) + ) + cm.errors.Add(ctx, 1, metric.WithAttributes(injectAPIMethod(ctx, errorAttrs)...)) } // For unary calls, record TTFB equal to the total attempt latency. isStreaming := methodName == "ReadObject" || methodName == "WriteObject" || methodName == "BidiReadObject" || methodName == "BidiWriteObject" if !isStreaming { - cm.ttfb.Record(ctx, duration, metric.WithAttributes(attribute.String("rpc.method", logicalMethod))) + ttfbAttrs := []attribute.KeyValue{attribute.String("rpc.method", logicalMethod)} + cm.ttfb.Record(ctx, duration, metric.WithAttributes(injectAPIMethod(ctx, ttfbAttrs)...)) } } @@ -483,7 +693,7 @@ func (cm *clientMetrics) recordHTTP(ctx context.Context, req *http.Request, resp attribute.String("error.type", errorType), } - cm.httpClientRequestDuration.Record(ctx, duration, metric.WithAttributes(attrs...)) + cm.httpClientRequestDuration.Record(ctx, duration, metric.WithAttributes(injectAPIMethod(ctx, attrs)...)) } // computeURLTemplate extracts a parameterized template path for a given GCS HTTP request URL path. @@ -547,9 +757,6 @@ type metricsRoundTripper struct { } func (rt *metricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { - startTime := time.Now() - resp, err := rt.base.RoundTrip(req) - state := metricsStateFromContext(req.Context()) var logicalMethod string if state != nil { @@ -558,6 +765,77 @@ func (rt *metricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, err logicalMethod = "Unknown" } + var decrementActiveRequests bool + var netAttrs, rpcAttrs metric.MeasurementOption + if rt.metrics != nil { + host := stripPort(req.URL.Host) + netAttrs = metric.WithAttributes( + attribute.String("rpc.system.name", "http"), + attribute.String("server.address", host), + ) + rpcAttrs = metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "http"), + attribute.String("server.address", host), + ) + if rt.metrics.activeRequests != nil { + rt.metrics.activeRequests.Add(req.Context(), 1, rpcAttrs) + decrementActiveRequests = true + } + } + defer func() { + if decrementActiveRequests && rt.metrics != nil && rt.metrics.activeRequests != nil { + rt.metrics.activeRequests.Add(req.Context(), -1, rpcAttrs) + } + }() + + if rt.metrics != nil && (rt.metrics.dnsLookupDuration != nil || rt.metrics.tcpConnectDuration != nil || rt.metrics.tlsHandshakeDuration != nil) { + var dnsStart, tlsStart time.Time + var tcpStarts sync.Map + + trace := &httptrace.ClientTrace{ + DNSStart: func(info httptrace.DNSStartInfo) { + dnsStart = time.Now() + }, + DNSDone: func(info httptrace.DNSDoneInfo) { + if rt.metrics.dnsLookupDuration != nil { + start := dnsStart + if !start.IsZero() { + duration := time.Since(start).Seconds() + rt.metrics.dnsLookupDuration.Record(req.Context(), duration, netAttrs) + } + } + }, + ConnectStart: func(network, addr string) { + tcpStarts.Store(addr, time.Now()) + }, + ConnectDone: func(network, addr string, err error) { + if err == nil && rt.metrics.tcpConnectDuration != nil { + if startV, ok := tcpStarts.LoadAndDelete(addr); ok { + duration := time.Since(startV.(time.Time)).Seconds() + rt.metrics.tcpConnectDuration.Record(req.Context(), duration, netAttrs) + } + } + }, + TLSHandshakeStart: func() { + tlsStart = time.Now() + }, + TLSHandshakeDone: func(state tls.ConnectionState, err error) { + if err == nil && rt.metrics.tlsHandshakeDuration != nil { + start := tlsStart + if !start.IsZero() { + duration := time.Since(start).Seconds() + rt.metrics.tlsHandshakeDuration.Record(req.Context(), duration, netAttrs) + } + } + }, + } + req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace)) + } + + startTime := time.Now() + resp, err := rt.base.RoundTrip(req) + statusCode := int64(0) if resp != nil { statusCode = int64(resp.StatusCode) @@ -566,21 +844,45 @@ func (rt *metricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, err if rt.metrics != nil { // Record attempt. - attemptAttrs := []attribute.KeyValue{ + attemptAttrs := make([]attribute.KeyValue, 0, 4) + attemptAttrs = append(attemptAttrs, + attribute.String("rpc.system.name", "http"), attribute.String("rpc.method", logicalMethod), attribute.Int64("http.response.status_code", statusCode), attribute.String("error.type", errorType), - } - rt.metrics.attempts.Add(req.Context(), 1, metric.WithAttributes(attemptAttrs...)) + ) + rt.metrics.attempts.Add(req.Context(), 1, metric.WithAttributes(injectAPIMethod(req.Context(), attemptAttrs)...)) // Record error if failed. if err != nil || (resp != nil && resp.StatusCode >= 400) { - errorAttrs := []attribute.KeyValue{ + errorAttrs := make([]attribute.KeyValue, 0, 4) + errorAttrs = append(errorAttrs, + attribute.String("rpc.system.name", "http"), attribute.String("rpc.method", logicalMethod), attribute.String("error.type", errorType), attribute.String("gcp.errors.domain", "storage.googleapis.com"), + ) + rt.metrics.errors.Add(req.Context(), 1, metric.WithAttributes(injectAPIMethod(req.Context(), errorAttrs)...)) + } + + if rt.metrics.gfeHeaderMissing != nil { + headerVal := "" + if resp != nil { + headerVal = resp.Header.Get("X-Goog-Gfe-Service-Time") + } + if resp == nil || headerVal == "" { + missingAttrs := metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "http"), + attribute.String("server.address", stripPort(req.URL.Host)), + attribute.String("error.type", errorType), + ) + rt.metrics.gfeHeaderMissing.Add(req.Context(), 1, missingAttrs) + } else if rt.metrics.gfeDuration != nil { + if ms, parseErr := strconv.ParseFloat(headerVal, 64); parseErr == nil { + rt.metrics.gfeDuration.Record(req.Context(), ms/1000.0, rpcAttrs) + } } - rt.metrics.errors.Add(req.Context(), 1, metric.WithAttributes(errorAttrs...)) } // Record TTFB. @@ -588,7 +890,8 @@ func (rt *metricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, err isResumableInit := req.Method == "POST" && strings.Contains(req.URL.Path, "/upload/") && req.URL.Query().Get("uploadType") == "resumable" if !isDownload || isResumableInit { duration := time.Since(startTime).Seconds() - rt.metrics.ttfb.Record(req.Context(), duration, metric.WithAttributes(attribute.String("rpc.method", logicalMethod))) + ttfbAttrs := []attribute.KeyValue{attribute.String("rpc.method", logicalMethod)} + rt.metrics.ttfb.Record(req.Context(), duration, metric.WithAttributes(injectAPIMethod(req.Context(), ttfbAttrs)...)) } } @@ -601,6 +904,7 @@ func (rt *metricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, err } if resp.Body != nil { + decrementActiveRequests = false resp.Body = &wrappedResponseBody{ ReadCloser: resp.Body, startTime: startTime, @@ -656,31 +960,122 @@ func (w *wrappedResponseBody) record(err error) { if w.recorded.CompareAndSwap(false, true) { duration := time.Since(w.startTime).Seconds() w.metrics.recordHTTP(w.req.Context(), w.req, w.resp, duration, err) + + if w.metrics.activeRequests != nil { + state := metricsStateFromContext(w.req.Context()) + var logicalMethod string + if state != nil { + logicalMethod = state.method + } else { + logicalMethod = "Unknown" + } + w.metrics.activeRequests.Add(w.req.Context(), -1, metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "http"), + attribute.String("server.address", stripPort(w.req.URL.Host)), + )) + } + } +} + +func getLogicalMethod(method string) string { + if idx := strings.LastIndex(method, "/"); idx != -1 { + return method[idx+1:] + } + return method +} + +func (cm *clientMetrics) recordGFEMetrics(ctx context.Context, headerMD, trailerMD metadata.MD, err error, logicalMethod, target string, rpcAttrs metric.MeasurementOption) { + if cm.gfeHeaderMissing == nil { + return + } + + headerVals := headerMD.Get("x-goog-gfe-service-time") + if len(headerVals) == 0 { + headerVals = trailerMD.Get("x-goog-gfe-service-time") + } + headerVal := "" + if len(headerVals) > 0 { + headerVal = headerVals[0] + } + if headerVal == "" { + errType := computeErrorType(err, false, int64(status.Code(err))) + missingAttrs := metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", stripPort(target)), + attribute.String("error.type", errType), + ) + cm.gfeHeaderMissing.Add(ctx, 1, missingAttrs) + } else if cm.gfeDuration != nil { + if ms, parseErr := strconv.ParseFloat(headerVal, 64); parseErr == nil { + cm.gfeDuration.Record(ctx, ms/1000.0, rpcAttrs) + } } } // metricsInterceptors returns gRPC client interceptors. func metricsInterceptors(cm *clientMetrics) (grpc.UnaryClientInterceptor, grpc.StreamClientInterceptor) { unary := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { - startTime := time.Now() - err := invoker(ctx, method, req, reply, cc, opts...) - duration := time.Since(startTime).Seconds() + logicalMethod := getLogicalMethod(method) + target := "" if cc != nil { target = cc.Target() } + + var rpcAttrs metric.MeasurementOption + if cm.activeRequests != nil || cm.gfeHeaderMissing != nil { + rpcAttrs = metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", stripPort(target)), + ) + } + + if cm.activeRequests != nil { + cm.activeRequests.Add(ctx, 1, rpcAttrs) + defer cm.activeRequests.Add(ctx, -1, rpcAttrs) + } + + var headerMD, trailerMD metadata.MD + opts = append(opts, grpc.Header(&headerMD), grpc.Trailer(&trailerMD)) + + startTime := time.Now() + err := invoker(ctx, method, req, reply, cc, opts...) + + cm.recordGFEMetrics(ctx, headerMD, trailerMD, err, logicalMethod, target, rpcAttrs) + + duration := time.Since(startTime).Seconds() cm.recordRPC(ctx, method, target, duration, err) return err } stream := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { - startTime := time.Now() - clientStream, err := streamer(ctx, desc, cc, method, opts...) + logicalMethod := getLogicalMethod(method) + target := "" if cc != nil { target = cc.Target() } + + var rpcAttrs metric.MeasurementOption + if cm.activeRequests != nil { + rpcAttrs = metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", stripPort(target)), + ) + cm.activeRequests.Add(ctx, 1, rpcAttrs) + } + + startTime := time.Now() + clientStream, err := streamer(ctx, desc, cc, method, opts...) + if err != nil { + if cm.activeRequests != nil { + cm.activeRequests.Add(ctx, -1, rpcAttrs) + } duration := time.Since(startTime).Seconds() cm.recordRPC(ctx, method, target, duration, err) return nil, err @@ -740,6 +1135,25 @@ func (w *wrappedClientStream) record(err error) { if w.recorded.CompareAndSwap(false, true) { duration := time.Since(w.startTime).Seconds() w.metrics.recordRPC(w.ctx, w.method, w.target, duration, err) + + logicalMethod := getLogicalMethod(w.method) + + var rpcAttrs metric.MeasurementOption + if w.metrics.activeRequests != nil || w.metrics.gfeHeaderMissing != nil { + rpcAttrs = metric.WithAttributes( + attribute.String("rpc.method", logicalMethod), + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", stripPort(w.target)), + ) + } + + if w.metrics.activeRequests != nil { + w.metrics.activeRequests.Add(w.ctx, -1, rpcAttrs) + } + + headerMD, _ := w.ClientStream.Header() + trailerMD := w.ClientStream.Trailer() + w.metrics.recordGFEMetrics(w.ctx, headerMD, trailerMD, err, logicalMethod, w.target, rpcAttrs) } } @@ -747,13 +1161,7 @@ func (w *wrappedClientStream) recordTTFB(m interface{}) { if w.recordedTTFB.Load() { return } - methodName := w.method - if strings.HasPrefix(methodName, "/") { - parts := strings.Split(strings.TrimPrefix(methodName, "/"), "/") - if len(parts) >= 2 { - methodName = parts[1] - } - } + methodName := getLogicalMethod(w.method) // The first response from the server, whether it contains metadata, // persisted size, or actual content, indicates TTFB. @@ -770,14 +1178,21 @@ func (w *wrappedClientStream) recordTTFB(m interface{}) { type metricsKey struct{} +type apiMethodKey struct{} + +func injectAPIMethod(ctx context.Context, attrs []attribute.KeyValue) []attribute.KeyValue { + if apiMethod, ok := ctx.Value(apiMethodKey{}).(string); ok { + return append(attrs, attribute.String("gcp.client.method", apiMethod)) + } + return attrs +} + type metricsState struct { - method string - startTime time.Time - metrics *clientMetrics - isHTTP bool - ttfbRecorded atomic.Bool - ttfbStart time.Time - record func(error) + method string + startTime time.Time + metrics *clientMetrics + isHTTP bool + record func(error) } func contextWithMetricsState(ctx context.Context, state *metricsState) context.Context { @@ -811,7 +1226,6 @@ func (cm *clientMetrics) startOperation(ctx context.Context, method string, isHT metrics: cm, isHTTP: isHTTP, } - state.ttfbStart = state.startTime var recordOnce sync.Once record := func(err error) { @@ -828,7 +1242,7 @@ func (cm *clientMetrics) startOperation(ctx context.Context, method string, isHT attribute.String("status", statusStr), attribute.String("error.type", errorType), } - opts := metric.WithAttributes(attrs...) + opts := metric.WithAttributes(injectAPIMethod(ctx, attrs)...) cm.duration.Record(ctx, duration, opts) cm.operations.Add(ctx, 1, opts) }) @@ -851,7 +1265,7 @@ func startMetricsOp(ctx context.Context, method string, isHTTP bool) (context.Co // initClientMetrics initializes OpenTelemetry client metrics if enabled in config. // It returns the metrics instance and its cleanup function, or nil if disabled or upon error. func initClientMetrics(ctx context.Context, project string, config *storageConfig) (*clientMetrics, func()) { - if !isOtelMetricsEnabled(config) { + if !isOtelMetricsEnabled(config) && !isOtelDebugMetricsEnabled(config) { return nil, nil } cm, cleanup, err := initMetrics(ctx, project, config) @@ -1139,3 +1553,176 @@ func (mc *metricsStorageClient) Close() error { func (mc *metricsStorageClient) fetchBucketMetadata(ctx context.Context, bucket string) (string, string, error) { return mc.storageClient.fetchBucketMetadata(ctx, bucket) } + +type dialInfo struct { + doneTime time.Time + host string +} + +type dialDoneContextKey struct{} + +// grpcMetricsStatsHandler implements stats.Handler to capture TLS handshake duration. +type grpcMetricsStatsHandler struct { + metrics *clientMetrics + dialTimes *sync.Map + host string +} + +type contextKeyRPCTag struct{} + +func (h *grpcMetricsStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { + method := info.FullMethodName + if idx := strings.LastIndex(method, "/"); idx != -1 { + method = method[idx+1:] + } + return context.WithValue(ctx, contextKeyRPCTag{}, method) +} +func (h *grpcMetricsStatsHandler) HandleRPC(ctx context.Context, s stats.RPCStats) { + if h.metrics == nil { + return + } + method := "" + if v := ctx.Value(contextKeyRPCTag{}); v != nil { + method = v.(string) + } + attrs := metric.WithAttributes( + attribute.String("rpc.system.name", "grpc"), + attribute.String("rpc.method", method), + attribute.String("server.address", h.host), + ) + + switch st := s.(type) { + case *stats.InPayload: + if h.metrics.networkBytesReceived != nil { + h.metrics.networkBytesReceived.Add(ctx, int64(st.WireLength), attrs) + } + case *stats.OutPayload: + if h.metrics.networkBytesSent != nil { + h.metrics.networkBytesSent.Add(ctx, int64(st.WireLength), attrs) + } + } +} +func (h *grpcMetricsStatsHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context { + if info.LocalAddr != nil && h.dialTimes != nil { + if val, ok := h.dialTimes.LoadAndDelete(info.LocalAddr.String()); ok { + return context.WithValue(ctx, dialDoneContextKey{}, val) + } + } + return ctx +} +func (h *grpcMetricsStatsHandler) HandleConn(ctx context.Context, s stats.ConnStats) { + if _, ok := s.(*stats.ConnBegin); ok { + if val := ctx.Value(dialDoneContextKey{}); val != nil { + info := val.(dialInfo) + if h.metrics != nil && h.metrics.tlsHandshakeDuration != nil { + duration := time.Since(info.doneTime).Seconds() + h.metrics.tlsHandshakeDuration.Record(context.Background(), duration, metric.WithAttributes( + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", info.host), + )) + } + } + } +} + +// grpcNetworkMetricsDialOptions returns dial options that instrument TCP and TLS handshake metrics. +func grpcNetworkMetricsDialOptions(host string, metrics *clientMetrics) []option.ClientOption { + if metrics == nil || (metrics.tcpConnectDuration == nil && metrics.tlsHandshakeDuration == nil && metrics.networkBytesSent == nil && metrics.networkBytesReceived == nil) { + return nil + } + var dialTimes sync.Map + + dialer := func(ctx context.Context, addr string) (net.Conn, error) { + var netAttrs metric.MeasurementOption + if metrics.tcpConnectDuration != nil { + netAttrs = metric.WithAttributes( + attribute.String("rpc.system.name", "grpc"), + attribute.String("server.address", host), + ) + } + + tcpStart := time.Now() + var d net.Dialer + conn, err := d.DialContext(ctx, "tcp", addr) + if err == nil { + if metrics.tcpConnectDuration != nil { + tcpDuration := time.Since(tcpStart).Seconds() + metrics.tcpConnectDuration.Record(ctx, tcpDuration, netAttrs) + } + if conn != nil && metrics.tlsHandshakeDuration != nil { + dialTimes.Store(conn.LocalAddr().String(), dialInfo{doneTime: time.Now(), host: host}) + } + } + + return conn, err + } + + sh := &grpcMetricsStatsHandler{ + metrics: metrics, + dialTimes: &dialTimes, + host: host, + } + + return []option.ClientOption{ + option.WithGRPCDialOption(grpc.WithContextDialer(dialer)), + option.WithGRPCDialOption(grpc.WithStatsHandler(sh)), + } +} + +type metricsTokenProvider struct { + base auth.TokenProvider + metrics *clientMetrics +} + +func (p *metricsTokenProvider) Token(ctx context.Context) (*auth.Token, error) { + start := time.Now() + tok, err := p.base.Token(ctx) + p.metrics.recordCredentialRefreshDuration(ctx, time.Since(start), err) + return tok, err +} + +// wrapAuthCredentials wraps an auth.Credentials object to track credential refresh durations. +// Note: We deliberately do not wrap legacy golang.org/x/oauth2/google.Credentials because +// it embeds unexported fields (e.g. universeDomain and its internal Mutex). Wrapping or copying +// it would either trigger go vet lock-copying errors or silently drop those unexported fields, +// which breaks Universe Domain resolution for legacy users. +func wrapAuthCredentials(c *auth.Credentials, m *clientMetrics) *auth.Credentials { + if c == nil || c.TokenProvider == nil { + return c + } + if mtp, ok := c.TokenProvider.(*metricsTokenProvider); ok { + if m == nil || mtp.metrics == m { + return c + } + clone := *c + clone.TokenProvider = &metricsTokenProvider{base: mtp.base, metrics: m} + return &clone + } + clone := *c + clone.TokenProvider = &metricsTokenProvider{base: c.TokenProvider, metrics: m} + return &clone +} + +func (cm *clientMetrics) recordCredentialRefreshDuration(ctx context.Context, duration time.Duration, err error) { + if cm == nil || cm.credentialRefreshDuration == nil { + return + } + errorType := "OK" + if err != nil { + errorType = "CLIENT_AUTHENTICATION_ERROR" + } + attrs := []attribute.KeyValue{attribute.String("error.type", errorType)} + cm.credentialRefreshDuration.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...)) +} + +func (cm *clientMetrics) recordStallDuration(ctx context.Context, duration time.Duration, method string, systemName string, target string) { + if cm == nil || cm.stallDuration == nil { + return + } + attrs := []attribute.KeyValue{ + attribute.String("rpc.system.name", systemName), + attribute.String("rpc.method", method), + attribute.String("server.address", target), + } + cm.stallDuration.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...)) +} 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 c49b224e..758f3ee2 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 @@ -45,6 +45,7 @@ func init() { storageinternal.WithZonalBucketAPIs = withZonalBucketAPIs storageinternal.WithDirectConnectivityEnforced = withDirectConnectivityEnforced storageinternal.WithOtelMetrics = withOtelMetrics + storageinternal.WithOtelDebugMetrics = withOtelDebugMetrics } // getDynamicReadReqIncreaseRateFromEnv returns the value set in the env variable. @@ -83,6 +84,7 @@ type storageConfig struct { readAPIWasSet bool disableClientMetrics bool enableOtelMetrics bool + enableOtelDebugMetrics bool metricExporter *metric.Exporter metricInterval time.Duration meterProvider *metric.MeterProvider @@ -314,3 +316,15 @@ type withOtelMetricsConfig struct { func (w *withOtelMetricsConfig) ApplyStorageOpt(c *storageConfig) { c.enableOtelMetrics = true } + +func withOtelDebugMetrics() option.ClientOption { + return &withOtelDebugMetricsConfig{} +} + +type withOtelDebugMetricsConfig struct { + internaloption.EmbeddableAdapter +} + +func (w *withOtelDebugMetricsConfig) ApplyStorageOpt(c *storageConfig) { + c.enableOtelDebugMetrics = true +} 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 11f0978e..b46fad37 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 @@ -52,11 +52,9 @@ func traceAttributesFromContext(ctx context.Context) ([]attribute.KeyValue, bool } const ( - storageOtelTracingDevVar = "GO_STORAGE_DEV_OTEL_TRACING" - defaultTracerName = "cloud.google.com/go/storage" - gcpClientRepo = "googleapis/google-cloud-go" - gcpClientArtifact = "cloud.google.com/go/storage" - storageBucketMetadataDisabledVar = "GO_OTEL_BUCKETMETADATA_DISABLED" + defaultTracerName = "cloud.google.com/go/storage" + gcpClientRepo = "googleapis/google-cloud-go" + gcpClientArtifact = "cloud.google.com/go/storage" ) // isOTelTracingDevEnabled checks the development flag until experimental feature is launched. @@ -108,6 +106,7 @@ func startSpanWithBucket(ctx context.Context, client *Client, bucket string, nam // If the context.Context provided in `ctx` contains a span then the newly-created // span will be a child of that span, otherwise it will be a root span. func startSpan(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span) { + ctx = context.WithValue(ctx, apiMethodKey{}, name) name = appendPackageName(name) // TODO: Remove internalTrace upon experimental launch. if !isOTelTracingDevEnabled() { diff --git a/ci/resources/stemcell-version-bump/vendor/modules.txt b/ci/resources/stemcell-version-bump/vendor/modules.txt index fccb28a2..359e0d99 100644 --- a/ci/resources/stemcell-version-bump/vendor/modules.txt +++ b/ci/resources/stemcell-version-bump/vendor/modules.txt @@ -44,7 +44,7 @@ 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.64.0 +# cloud.google.com/go/storage v1.65.0 ## explicit; go 1.25.0 cloud.google.com/go/storage cloud.google.com/go/storage/experimental