feat(storage): add resource span attributes for ACO ( App Centric Observability ) for async client

google/cloud/storage/google_cloud_cpp_storage.bzl

@@ -49,6 +49,7 @@ google_cloud_cpp_storage_hdrs = [
     "internal/binary_data_as_debug_string.h",
     "internal/bucket_access_control_parser.h",
     "internal/bucket_acl_requests.h",
+    "internal/bucket_metadata_cache.h",
     "internal/bucket_metadata_parser.h",
     "internal/bucket_requests.h",
     "internal/complex_option.h",
@@ -164,6 +165,7 @@ google_cloud_cpp_storage_srcs = [
     "internal/base64.cc",
     "internal/bucket_access_control_parser.cc",
     "internal/bucket_acl_requests.cc",
+    "internal/bucket_metadata_cache.cc",
     "internal/bucket_metadata_parser.cc",
     "internal/bucket_requests.cc",
     "internal/compute_engine_util.cc",

google/cloud/storage/google_cloud_cpp_storage.cmake

@@ -74,6 +74,8 @@ add_library(
     internal/bucket_access_control_parser.h
     internal/bucket_acl_requests.cc
     internal/bucket_acl_requests.h
+    internal/bucket_metadata_cache.cc
+    internal/bucket_metadata_cache.h
     internal/bucket_metadata_parser.cc
     internal/bucket_metadata_parser.h
     internal/bucket_requests.cc
@@ -420,6 +422,7 @@ if (BUILD_TESTING)
         idempotency_policy_test.cc
         internal/base64_test.cc
         internal/bucket_acl_requests_test.cc
+        internal/bucket_metadata_cache_test.cc
         internal/bucket_requests_test.cc
         internal/complex_option_test.cc
         internal/compute_engine_util_test.cc

google/cloud/storage/internal/async/connection_tracing.cc

@@ -18,12 +18,20 @@
 #include "google/cloud/storage/internal/async/reader_connection_tracing.h"
 #include "google/cloud/storage/internal/async/rewriter_connection_tracing.h"
 #include "google/cloud/storage/internal/async/writer_connection_tracing.h"
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
+#include "google/cloud/storage/internal/connection_factory.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/version.h"
+#include <algorithm>
+#include <chrono>
+#include <future>
 #include <memory>
+#include <mutex>
+#include <vector>
 
 namespace google {
 namespace cloud {
+
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
@@ -33,13 +41,21 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
  public:
   explicit AsyncConnectionTracing(
       std::shared_ptr<storage::AsyncConnection> impl)
-      : impl_(std::move(impl)) {}
+      : impl_(std::move(impl)), cache_(BucketMetadataCache::Singleton()) {}
+
+  ~AsyncConnectionTracing() override {
+    for (auto& f : bg_tasks_) {
+      if (f.valid()) f.wait();
+    }
+  }
 
   Options options() const override { return impl_->options(); }
 
   future<StatusOr<google::storage::v2::Object>> InsertObject(
       InsertObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::InsertObject");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return internal::EndSpan(std::move(span),
                              impl_->InsertObject(std::move(p)));
@@ -48,6 +64,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>>> Open(
       OpenParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::Open");
+    EnrichSpan(*span, p.options, p.read_spec.bucket());
     internal::OTelScope scope(span);
     return impl_->Open(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -67,6 +84,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<std::unique_ptr<storage::AsyncReaderConnection>>> ReadObject(
       ReadObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ReadObject");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
     auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
                  span = std::move(span)](auto f)
@@ -82,6 +100,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<storage::ReadPayload>> ReadObjectRange(
       ReadObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ReadObjectRange");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
     return impl_->ReadObjectRange(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -96,6 +115,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartAppendableObjectUpload(AppendableUploadParams p) override {
     auto span = internal::MakeSpan(
         "storage::AsyncConnection::StartAppendableObjectUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartAppendableObjectUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -112,6 +133,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   ResumeAppendableObjectUpload(AppendableUploadParams p) override {
     auto span = internal::MakeSpan(
         "storage::AsyncConnection::ResumeAppendableObjectUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->ResumeAppendableObjectUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -128,6 +151,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartUnbufferedUpload(UploadParams p) override {
     auto span =
         internal::MakeSpan("storage::AsyncConnection::StartUnbufferedUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartUnbufferedUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -144,6 +169,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   StartBufferedUpload(UploadParams p) override {
     auto span =
         internal::MakeSpan("storage::AsyncConnection::StartBufferedUpload");
+    EnrichSpan(*span, p.options,
+               p.request.write_object_spec().resource().bucket());
     internal::OTelScope scope(span);
     return impl_->StartBufferedUpload(std::move(p))
         .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
@@ -191,13 +218,15 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   future<StatusOr<google::storage::v2::Object>> ComposeObject(
       ComposeObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ComposeObject");
+    EnrichSpan(*span, p.options, p.request.destination().bucket());
     internal::OTelScope scope(span);
     return internal::EndSpan(std::move(span),
                              impl_->ComposeObject(std::move(p)));
   }
 
   future<Status> DeleteObject(DeleteObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::DeleteObject");
+    EnrichSpan(*span, p.options, p.request.bucket());
     internal::OTelScope scope(span);
     return internal::EndSpan(std::move(span),
                              impl_->DeleteObject(std::move(p)));
@@ -211,7 +240,70 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
   }
 
  private:
+  void CleanupCompletedTasks() {
+    std::unique_lock<std::mutex> lk(mu_);
+    bg_tasks_.erase(
+        std::remove_if(bg_tasks_.begin(), bg_tasks_.end(),
+                       [](std::future<void> const& f) {
+                         return f.wait_for(std::chrono::seconds(0)) ==
+                                std::future_status::ready;
+                       }),
+        bg_tasks_.end());
+  }
+
+  void MaybeTriggerBackgroundFetch(Options const& options,
+                                   std::string const& bucket_name) {
+    CleanupCompletedTasks();
+
+    if (!cache_.StartFetch(bucket_name)) {
+      return;
+    }
+
+    std::shared_ptr<storage::internal::StorageConnection> conn;
+    {
+      std::lock_guard<std::mutex> lk(mu_);
+      if (!sync_conn_) {
+        sync_conn_ = MakeStorageConnection(impl_->options());
+      }
+      conn = sync_conn_;
+    }
+
+    auto f =
+        std::async(std::launch::async, [bucket_name, options, conn, this]() {
+          google::cloud::internal::OptionsSpan span(options);
+          auto const normalized_bucket_name =
+              BucketMetadataCache::NormalizeBucketName(bucket_name);
+          storage::internal::GetBucketMetadataRequest request(
+              normalized_bucket_name);
+          auto metadata = conn->GetBucketMetadata(request);
+          auto entry = BucketCacheEntry::Create(bucket_name, metadata);
+          if (entry) {
+            cache_.Put(bucket_name, std::move(*entry));
+          }
+          cache_.EndFetch(bucket_name);
+        });
+
+    std::unique_lock<std::mutex> lk(mu_);
+    bg_tasks_.push_back(std::move(f));
+  }
+
+  void EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
+                  std::string const& bucket_name) {
+    if (bucket_name.empty()) return;
+    auto entry = cache_.Get(bucket_name);
+    if (entry.has_value()) {
+      span.SetAttribute("gcp.resource.destination.id", entry->id);
+      span.SetAttribute("gcp.resource.destination.location", entry->location);
+    } else {
+      MaybeTriggerBackgroundFetch(options, bucket_name);
+    }
+  }
+
   std::shared_ptr<storage::AsyncConnection> impl_;
+  BucketMetadataCache& cache_;
+  std::shared_ptr<storage::internal::StorageConnection> sync_conn_;
+  std::vector<std::future<void>> bg_tasks_;
+  std::mutex mu_;
 };
 
 }  // namespace

google/cloud/storage/internal/bucket_metadata_cache.cc

@@ -0,0 +1,130 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "google/cloud/storage/internal/bucket_metadata_cache.h"
+#include "google/cloud/storage/bucket_metadata.h"
+#include "google/cloud/status.h"
+#include <mutex>
+#include <utility>
+
+namespace google {
+namespace cloud {
+namespace storage_internal {
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
+
+BucketMetadataCache& BucketMetadataCache::Singleton() {
+  static BucketMetadataCache instance(10000);
+  return instance;
+}
+
+BucketCacheEntry BucketCacheEntry::FromMetadata(
+    storage::BucketMetadata const& m) {
+  std::string loc = m.location();
+  if (m.location_type() == "multi-region" ||
+      m.location_type() == "dual-region") {
+    loc = "global";
+  }
+  return {
+      "projects/" + std::to_string(m.project_number()) + "/buckets/" + m.name(),
+      std::move(loc)};
+}
+
+absl::optional<BucketCacheEntry> BucketCacheEntry::Create(
+    std::string const& bucket_name,
+    StatusOr<storage::BucketMetadata> const& metadata) {
+  if (metadata.ok()) {
+    return FromMetadata(*metadata);
+  }
+  if (metadata.status().code() == StatusCode::kPermissionDenied) {
+    return BucketCacheEntry{
+        "projects/_/buckets/" +
+            BucketMetadataCache::NormalizeBucketName(bucket_name),
+        "global"};
+  }
+  return absl::nullopt;
+}
+
+void BucketMetadataCache::MoveToFront(std::list<std::string>::iterator it) {
+  list_.splice(list_.begin(), list_, it);
+}
+
+absl::optional<BucketCacheEntry> BucketMetadataCache::Get(
+    std::string const& raw_bucket_name) {
+  auto const bucket_name = NormalizeBucketName(raw_bucket_name);
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it == map_.end()) return absl::nullopt;
+
+  MoveToFront(it->second.second);
+  return it->second.first;
+}
+
+void BucketMetadataCache::Put(std::string const& raw_bucket_name,
+                              BucketCacheEntry entry) {
+  auto const bucket_name = NormalizeBucketName(raw_bucket_name);
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it != map_.end()) {
+    it->second.first = std::move(entry);
+    MoveToFront(it->second.second);
+    return;
+  }
+
+  if (map_.size() >= max_size_) {
+    auto oldest = list_.back();
+    list_.pop_back();
+    map_.erase(oldest);
+  }
+
+  list_.push_front(bucket_name);
+  map_[bucket_name] = {std::move(entry), list_.begin()};
+}
+
+void BucketMetadataCache::Invalidate(std::string const& raw_bucket_name) {
+  auto const bucket_name = NormalizeBucketName(raw_bucket_name);
+  std::unique_lock<std::mutex> lk(mu_);
+  auto it = map_.find(bucket_name);
+  if (it != map_.end()) {
+    list_.erase(it->second.second);
+    map_.erase(it);
+  }
+}
+
+void BucketMetadataCache::Clear() {
+  std::unique_lock<std::mutex> lk(mu_);
+  map_.clear();
+  list_.clear();
+  in_flight_fetch_.clear();
+}
+
+bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) {
+  auto const bucket_name = NormalizeBucketName(raw_bucket_name);
+  std::unique_lock<std::mutex> lk(mu_);
+  if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) {
+    return false;
+  }
+  in_flight_fetch_.insert(bucket_name);
+  return true;
+}
+
+void BucketMetadataCache::EndFetch(std::string const& raw_bucket_name) {
+  auto const bucket_name = NormalizeBucketName(raw_bucket_name);
+  std::unique_lock<std::mutex> lk(mu_);
+  in_flight_fetch_.erase(bucket_name);
+}
+
+GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
+}  // namespace storage_internal
+}  // namespace cloud
+}  // namespace google