feat(storage): add OpenTelemetry tracing support for Zonal Buckets

[shradhakatyal]

This PR adds OpenTelemetry tracing support for Zonal Buckets (Rapid Storage) across the asynchronous gRPC client and streaming classes (AsyncGrpcClient, AsyncAppendableObjectWriter, and AsyncMultiRangeDownloader).

The implementation aligns with existing regional bucket tracing conventions:

• Instruments T2 (Client API) spans using the naming format Storage.<Class>.<methodName>.
• Reuses existing opt-in environment variables (ENABLE_GCS_PYTHON_CLIENT_OTEL_TRACES and DISABLE_GCS_PYTHON_CLIENT_OTEL_BUCKET_METADATA).
• Sets rpc.system="grpc" and attaches App-Centric Observability (ACO) attributes (gcp.resource.destination.id and gcp.resource.destination.location).
• Injects W3C traceparent headers directly into gRPC request metadata tuples.

Fixes: b/489239033

[gemini-code-assist]

Code Review

This pull request introduces OpenTelemetry tracing support for zonal buckets (Rapid Storage) in async gRPC classes. It refactors tracing span creation to support both synchronous and asynchronous context managers, injects traceparent metadata into gRPC requests, and updates the bucket metadata cache to support asynchronous gRPC fetches. The review feedback recommends defensively checking for __aenter__ and __aexit__ in _TraceSpanHelperContext to ensure compatibility with synchronous context managers (such as mocked spans in unit tests) and refactoring the project number extraction in update_from_bucket to use getattr more idiomatically.

[gemini-code-assist]

To ensure robustness against synchronous context managers (which are commonly returned when mocking create_trace_span in existing unit tests), we should defensively check if the underlying context manager supports __aenter__ and __aexit__. If it does not, we should fall back to calling __enter__ and __exit__ synchronously.

    async def __aenter__(self):
        self._prepare_base_cm()
        if hasattr(self._base_cm, "__aenter__"):
            return await self._base_cm.__aenter__()
        return self._base_cm.__enter__()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_val is not None and isinstance(
            exc_val, (NotFound, api_exceptions.NotFound)
        ):
            self._handle_not_found()
        if self._base_cm is not None:
            if hasattr(self._base_cm, "__aexit__"):
                return await self._base_cm.__aexit__(exc_type, exc_val, exc_tb)
            return self._base_cm.__exit__(exc_type, exc_val, exc_tb)
        return False