Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
shradhakatyal wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a8551b1
feat(storage): add OpenTelemetry tracing support for Zonal Buckets (a…
Sep 15, 2026
9781ef6
Update packages/google-cloud-storage/google/cloud/storage/_bucket_met…
shradhakatyal Sep 16, 2026
e572fb0
fix(storage): add sync context manager fallback and prevent trace sta…
Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 101 additions & 20 deletions packages/google-cloud-storage/google/cloud/storage/_bucket_metadata_cache.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

"""In-memory LRU cache for bucket metadata supporting App-centric Observability (ACO)."""

import asyncio
import logging
import threading

Expand Down Expand Up @@ -58,11 +59,18 @@ def get_or_queue_fetch(self, bucket_name):
# bypass starting duplicate fetches.
return None
else:
# fire a background thread and get bucket metadata.
# fire a background fetch and get bucket metadata.
self._inflight_fetches.add(bucket_name)
threading.Thread(
target=self._fetch_background, args=(bucket_name,), daemon=True
).start()
if getattr(self._client, "_is_async_grpc_client", False) is True:
try:
loop = asyncio.get_running_loop()
loop.create_task(self._fetch_background_async(bucket_name))
except RuntimeError:
self._inflight_fetches.discard(bucket_name)
else:
threading.Thread(
target=self._fetch_background, args=(bucket_name,), daemon=True
).start()
return None

def check_and_evict(self, bucket_name):
Expand All @@ -73,11 +81,20 @@ def check_and_evict(self, bucket_name):
if bucket_name in self._inflight_checks:
return
self._inflight_checks.add(bucket_name)
threading.Thread(
target=self._verify_existence_background,
args=(bucket_name,),
daemon=True,
).start()
if getattr(self._client, "_is_async_grpc_client", False) is True:
try:
loop = asyncio.get_running_loop()
loop.create_task(
self._verify_existence_background_async(bucket_name)
)
except RuntimeError:
self._inflight_checks.discard(bucket_name)
else:
threading.Thread(
target=self._verify_existence_background,
args=(bucket_name,),
daemon=True,
).start()

def _verify_existence_background(self, bucket_name):
try:
Expand All @@ -92,6 +109,24 @@ def _verify_existence_background(self, bucket_name):
with self._lock:
self._inflight_checks.discard(bucket_name)

async def _verify_existence_background_async(self, bucket_name):
try:
from google.cloud import _storage_v2 as storage_v2

request = storage_v2.GetBucketRequest(
name=f"projects/_/buckets/{bucket_name}"
)
await self._client.grpc_client.get_bucket(request=request, timeout=10.0)
except (NotFound, api_exceptions.NotFound):
self.evict(bucket_name)
except Exception as e:
logger.debug(
f"Async background verification for bucket existence failed for {bucket_name}: {e}"
)
finally:
with self._lock:
self._inflight_checks.discard(bucket_name)

def _fetch_background(self, bucket_name):
"""Asynchronously fetch bucket metadata and update the cache."""
try:
Expand All @@ -112,26 +147,72 @@ def _fetch_background(self, bucket_name):
with self._lock:
self._inflight_fetches.discard(bucket_name)

def update_from_bucket(self, bucket):
"""Update cache from a Bucket instance."""
if not bucket or not bucket.name:
async def _fetch_background_async(self, bucket_name):
"""Asynchronously fetch bucket metadata via gRPC and update the cache."""
try:
from google.cloud import _storage_v2 as storage_v2

request = storage_v2.GetBucketRequest(
name=f"projects/_/buckets/{bucket_name}"
)
bucket = await self._client.grpc_client.get_bucket(
request=request, timeout=10.0
)
self.update_from_bucket(bucket, bucket_name=bucket_name)
except (NotFound, api_exceptions.NotFound):
self.evict(bucket_name)
except api_exceptions.Forbidden:
self.update_cache(
bucket_name, f"projects/_/buckets/{bucket_name}", "global"
)
except Exception as e:
logger.debug(
f"Async background fetch for bucket metadata failed for {bucket_name}: {e}"
)
finally:
with self._lock:
self._inflight_fetches.discard(bucket_name)

def update_from_bucket(self, bucket, bucket_name=None):
"""Update cache from a Bucket instance or storage_v2.Bucket proto."""
if not bucket:
return
name = bucket_name or getattr(bucket, "name", None)
if not name or not isinstance(name, str):
return
if name.startswith("projects/") and "/buckets/" in name:
name = name.split("/buckets/", 1)[1]

project_number = getattr(bucket, "project_number", None)
location = getattr(bucket, "location", None) or "global"
location = location.lower()
location_type = getattr(bucket, "location_type", None) or "region"
location_type = location_type.lower()
if not project_number:
proj_attr = getattr(bucket, "project", None)
if isinstance(proj_attr, (str, int)):
proj_str = str(proj_attr)
if proj_str.startswith("projects/"):
project_number = proj_str.split("projects/", 1)[1]
elif proj_str:
project_number = proj_str

loc_attr = getattr(bucket, "location", None)
location = (
loc_attr.lower() if isinstance(loc_attr, str) and loc_attr else "global"
)
loc_type_attr = getattr(bucket, "location_type", None)
location_type = (
loc_type_attr.lower()
if isinstance(loc_type_attr, str) and loc_type_attr
else "region"
)

if location_type in ("multi-region", "dual-region"):
location = "global"

if project_number:
destination_id = f"projects/{project_number}/buckets/{bucket.name}"
if project_number and str(project_number) != "_":
destination_id = f"projects/{project_number}/buckets/{name}"
else:
destination_id = f"projects/_/buckets/{bucket.name}"
destination_id = f"projects/_/buckets/{name}"

self.update_cache(bucket.name, destination_id, location)
self.update_cache(name, destination_id, location)

def update_cache(self, bucket_name, destination_id, location):
"""Thread-safely update or insert a cache entry with bounded size."""
Expand Down
174 changes: 116 additions & 58 deletions packages/google-cloud-storage/google/cloud/storage/_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import base64
import datetime
import inspect
import logging
import os
import secrets
Expand All @@ -32,18 +33,15 @@
from google.auth import environment_vars
from google.cloud.exceptions import NotFound

from google.cloud.storage._opentelemetry_tracing import (
_is_bucket_metadata_disabled,
)
from google.cloud.storage._opentelemetry_tracing import (
create_trace_span as _base_create_trace_span,
)
from google.cloud.storage import _opentelemetry_tracing
from google.cloud.storage.constants import _DEFAULT_TIMEOUT
from google.cloud.storage.retry import (
DEFAULT_RETRY,
DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
)

_base_create_trace_span = None

_logger = logging.getLogger(__name__)

STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST" # Despite name, includes scheme.
Expand Down Expand Up @@ -149,61 +147,121 @@ def _validate_name(name):
return name


@contextmanager
def create_trace_span_helper(client, bucket_name, name, attributes=None, **kwargs):
span_attrs = dict(attributes) if attributes else {}

if (
bucket_name
and isinstance(bucket_name, str)
and client
and hasattr(client, "_bucket_metadata_cache")
and client._bucket_metadata_cache
and not _is_bucket_metadata_disabled()
):
try:
if name in (
"Storage.Client.getBucket",
"Storage.Client.lookupBucket",
"Storage.Bucket.reload",
"Storage.Bucket.exists",
):
cached = client._bucket_metadata_cache.get(bucket_name)
else:
cached = client._bucket_metadata_cache.get_or_queue_fetch(bucket_name)

if cached and isinstance(cached, tuple) and len(cached) == 2:
dest_id, loc = cached
span_attrs.update(
{
"gcp.resource.destination.id": dest_id,
"gcp.resource.destination.location": loc,
}
)
except Exception as e:
_logger.debug(f"Failed cache lookup in create_trace_span_helper: {e}")
class _TraceSpanHelperContext:
"""Context manager supporting both sync and async tracing span creation with bucket metadata."""

if "client" not in kwargs and client:
kwargs["client"] = client
def __init__(self, client, bucket_name, name, attributes=None, **kwargs):
self.client = client
self.bucket_name = bucket_name
self.name = name
self.attributes = attributes
self.kwargs = kwargs
self._base_cm = None

def _prepare_base_cm(self):
span_attrs = dict(self.attributes) if self.attributes else {}
client = self.client
bucket_name = self.bucket_name
name = self.name

if (
bucket_name
and isinstance(bucket_name, str)
and client
and hasattr(client, "_bucket_metadata_cache")
and client._bucket_metadata_cache
and _opentelemetry_tracing._is_otel_traces_enabled()
and not _opentelemetry_tracing._is_bucket_metadata_disabled()
):
try:
if name in (
"Storage.Client.getBucket",
"Storage.Client.lookupBucket",
"Storage.Bucket.reload",
"Storage.Bucket.exists",
):
cached = client._bucket_metadata_cache.get(bucket_name)
else:
cached = client._bucket_metadata_cache.get_or_queue_fetch(
bucket_name
)

with _base_create_trace_span(name, attributes=span_attrs, **kwargs) as span:
try:
yield span
except (NotFound, api_exceptions.NotFound):
if (
bucket_name
and isinstance(bucket_name, str)
and client
and hasattr(client, "_bucket_metadata_cache")
and client._bucket_metadata_cache
):
try:
client._bucket_metadata_cache.check_and_evict(bucket_name)
except Exception as e:
_logger.debug(
f"Failed cache eviction on 404 in create_trace_span_helper: {e}"
if cached and isinstance(cached, tuple) and len(cached) == 2:
dest_id, loc = cached
span_attrs.update(
{
"gcp.resource.destination.id": dest_id,
"gcp.resource.destination.location": loc,
}
)
raise
except Exception as e:
_logger.debug(f"Failed cache lookup in create_trace_span_helper: {e}")

kwargs = dict(self.kwargs)
if "client" not in kwargs and client:
kwargs["client"] = client

create_span_fn = (
_base_create_trace_span
if _base_create_trace_span is not None
else _opentelemetry_tracing.create_trace_span
)
self._base_cm = create_span_fn(name, attributes=span_attrs, **kwargs)
return self._base_cm

def _handle_not_found(self):
if (
self.bucket_name
and isinstance(self.bucket_name, str)
and self.client
and hasattr(self.client, "_bucket_metadata_cache")
and self.client._bucket_metadata_cache
and _opentelemetry_tracing._is_otel_traces_enabled()
):
try:
self.client._bucket_metadata_cache.check_and_evict(self.bucket_name)
except Exception as e:
_logger.debug(
f"Failed cache eviction on 404 in create_trace_span_helper: {e}"
)

def __enter__(self):
self._prepare_base_cm()
return self._base_cm.__enter__()

def __exit__(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:
return self._base_cm.__exit__(exc_type, exc_val, exc_tb)
return False

async def __aenter__(self):
self._prepare_base_cm()
if hasattr(self._base_cm, "__aenter__"):
res = self._base_cm.__aenter__()
return await res if inspect.isawaitable(res) else res
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__"):
res = self._base_cm.__aexit__(exc_type, exc_val, exc_tb)
return await res if inspect.isawaitable(res) else res
return self._base_cm.__exit__(exc_type, exc_val, exc_tb)
return False


def create_trace_span_helper(client, bucket_name, name, attributes=None, **kwargs):
return _TraceSpanHelperContext(
client, bucket_name, name, attributes=attributes, **kwargs
)


class _PropertyMixin(object):
Expand Down
Loading
Loading