feat(storage): implement App-centric Observability (ACO) OpenTelemetry tracing

[chandra-siri]

Inject mandatory OpenTelemetry span attributes (gcp.resource.destination.id and gcp.resource.destination.location) into Cloud Storage bucket and object spans to support App-centric Observability (ACO). Implements a highly optimized, lockless in-memory LRU cache on the GCS Client, backed by asynchronous singleflight background fetching to prevent cache miss stampedes and eliminate latency overhead.

[gemini-code-assist]

Code Review

This pull request introduces an in-memory LRU cache for GCS bucket metadata to support App-centric Observability (ACO) within OpenTelemetry tracing. Key changes include the implementation of a thread-safe BucketMetadataCache with background fetching, updates to the Client class to support context management and cache lifecycle, and the re-enabling of system tests in Cloud Build. Feedback identifies a potential StopIteration error in the cache logic if max_size is zero and suggests pre-compiling regular expressions in the tracing module to improve performance.

[gemini-code-assist]

The current implementation of update_cache will raise a StopIteration exception if max_size is set to 0, as next(iter(self._cache)) will be called on an empty dictionary. While the default is 10000, it is safer to validate that max_size is at least 1 in the constructor or handle the zero-size case here.

Suggested change

	"""Thread-safely update or insert a cache entry with bounded size."""
	with self._lock:
	if bucket_name in self._cache:
	self._cache.pop(bucket_name)
	elif len(self._cache) >= self._max_size:
	# Pop the first item (oldest inserted)
	first_key = next(iter(self._cache))
	self._cache.pop(first_key)
	self._cache[bucket_name] = (destination_id, location)
	def update_cache(self, bucket_name, destination_id, location):
	"""Thread-safely update or insert a cache entry with bounded size."""
	if self._max_size <= 0:
	return
	with self._lock:
	if bucket_name in self._cache:
	self._cache.pop(bucket_name)
	elif len(self._cache) >= self._max_size:
	# Pop the first item (oldest inserted)
	first_key = next(iter(self._cache))
	self._cache.pop(first_key)
	self._cache[bucket_name] = (destination_id, location)