[shradhakatyal]

Summary

Implements OpenTelemetry Standard Metrics instrumentation for the Async gRPC client (AsyncGrpcClient) and Bidi streaming abstractions in google-cloud-storage:

• Core Metrics Infrastructure (_opentelemetry_metrics.py):
  • Defines all 9 OpenTelemetry standard metric instruments (gcp.storage.operation.count, gcp.storage.operation.duration, gcp.storage.operation.retry.count, gcp.storage.operation.transfer_size, gcp.storage.api.request.count, gcp.storage.api.request.duration, gcp.storage.api.request.body_size, gcp.storage.api.response.body_size, gcp.storage.stream.idle_duration), histogram bucket views, and attribute keys.
  • Adds gRPC transport endpoint parsing (extract_grpc_server_address_and_network_path), gRPC status/error classification (extract_grpc_status_code, classify_error_type), and Bidi stream metrics (_GrpcBidiStreamMetricsContext).
• AsyncGrpcClient (async_grpc_client.py):
  • Initializes _metrics_manager in AsyncGrpcClient.__init__ using the configured gRPC transport endpoint and user options (enable_metrics, meter_provider, metrics_exporter).
  • Wraps unary operations (delete_object, get_object_metadata, compose_object, restore_object, move_object, update_object_metadata, lock_bucket_retention_policy, start_resumable_write, query_write_status) with start_operation and start_api_request, wiring GAPIC AsyncRetry(on_error=...) callbacks to record retry counts and per-attempt API request metrics.
  • Wraps download_object (server-streaming ReadObject) with operation and API request contexts, recording record_response_bytes per chunk.
  • Wraps upload_object (BidiWriteObject) with operation and Bidi stream metrics, recording record_request_bytes and stream.idle_duration.
• Bidi Streaming Abstractions (async_multi_range_downloader.py, async_appendable_object_writer.py, async_read_object_stream.py, async_write_object_stream.py, _stream_multiplexer.py):
  • Instruments AsyncMultiRangeDownloader (open, download_ranges, _reconnect), AsyncAppendableObjectWriter (open, append, flush, close, simple_flush, _reconnect_and_recover), AsyncReadObjectStream, AsyncWriteObjectStream, and _StreamMultiplexer with operation, Bidi stream, transfer size, retry, and idle duration metrics.
• Unit Tests:
  • Comprehensive unit tests across test__opentelemetry_metrics.py and tests/unit/asyncio/ (348 unit tests passing).

[Generated-by: AI]

[gemini-code-assist]

Code Review

This pull request implements OpenTelemetry metrics support for the Google Cloud Storage asyncio client using gRPC. It introduces standard metric instruments, custom histogram bucket boundaries, canonical error classification, and route normalization. Additionally, it adds gRPC client interceptors and call tracking proxies to record physical attempt metrics, and integrates logical operation tracking across various asyncio stream and writer classes. The feedback suggests implementing __getattr__ in _TrackedCallBase to delegate unhandled attribute access to the underlying call object, ensuring robust compatibility with gRPC's internal implementation details.

[gemini-code-assist]

To ensure full compatibility with gRPC's internal implementation details, other interceptors, or future gRPC updates that might access private or undocumented attributes on the call object, implement __getattr__ in _TrackedCallBase to transparently delegate any unhandled attribute or method access to the underlying self._call object.

        self._closing_cleanly = False

    def __getattr__(self, name: str) -> Any:
        return getattr(self._call, name)

[cpriti-os]

Please address the inline comments on network.path detection, gcp.storage.operation vs rpc.method, meter_provider support, and bidi stream attempt/retry metrics.

Nit: The PR description also references outdated metric names (gcp.storage.operation.count, gcp.storage.stream.idle_duration, _GrpcBidiStreamMetricsContext) and methods (download_object, upload_object, compose_object, restore_object, move_object) that aren't in this PR—please update the description to match the current implementation.

[cpriti-os]

In _create_async_grpc_client, network_path is initialized to "directpath" whenever attempt_direct_path=True (which is the default on AsyncGrpcClient). Because grpc.aio.Call in Cython (_AioCall) does not bind grpc_call_get_peer, _resolve_network_path always falls back to that default—meaning every CFE call on a default AsyncGrpcClient (e.g. outside GCE or when DirectPath falls back to CFE) is misreported as network.path="directpath".

Since grpc.aio.Call doesn't expose peer(), we can resolve network.path per call in Python using the metadata already captured on the call:

1. On completed calls (or errors with initial metadata), check initial metadata (call._cython_call._initial_metadata or exc.initial_metadata()) for GFE headers such as x-goog-gfe-service-time—GFE always injects this on CloudPath ("cloudpath"), whereas DirectPath bypasses GFE and never carries it.
2. On failed calls (AioRpcError), exc.debug_error_string() from C-core includes the peer IP (e.g. ipv4:34.126.... or ipv6:...2001:4860:8040:...), which can be matched against the DirectPath prefixes (34.126.0.0/18 and 2001:4860:8040::/42).

[cpriti-os]

_extract_rpc_operation and start_operation pass the full gRPC method path (e.g. google.storage.v2.Storage/DeleteObject, google.storage.v2.Storage/BidiReadObject) as gcp.storage.operation. Per the design doc and Go implementation:

• rpc.method (on rpc.client.call.duration and gcp.client.request.duration) should carry the full gRPC method path (google.storage.v2.Storage/...).
• gcp.storage.operation (on all gcp.storage.client.* metrics and gcp.client.request.duration) should carry the short logical operation name (DeleteObject, ReadObject, WriteObject, etc.).

[cpriti-os]

AsyncGrpcClient.__init__ does not accept meter_provider or pass it to _MetricsRecorder(...). Without that, callers cannot provide a custom MeterProvider, and self._metrics_recorder._meter_provider is always None (so close() -> self._metrics_recorder.shutdown() never flushes a custom provider).

[cpriti-os]

For BidiReadObject (AsyncMultiRangeDownloader) and BidiWriteObject (AsyncAppendableObjectWriter), _MetricsStreamStreamInterceptor only runs once when the underlying stream is opened in open(), and _TrackedStreamStreamCall._record_once only fires when the stream closes or errors.

1. If a stream stays open across multiple download_ranges or append/flush calls and then fails mid-stream during a later operation, _TrackedStreamStreamCall records the attempt with the start_time and op_state from the original open() call rather than the active operation.
2. When open, download_ranges, append, close, or finalize retries after a failure (AsyncRetry or _BidiStreamRetryManager), gcp.storage.client.retries is not recorded.