feat(storage): add OpenTelemetry standard metrics for Async gRPC client

[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)