From cf0f09566c33b14c91219ca75b295e94f67e92b3 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 26 May 2026 13:24:34 +0000 Subject: [PATCH 01/20] feat(storage): add resource span attributes ACO ( App Centric Observability ) --- google/cloud/storage/internal/tracing_connection_test.cc | 1 + 1 file changed, 1 insertion(+) diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index 3cc75811ff639..531dc7736d6f0 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -44,6 +44,7 @@ using ::google::cloud::testing_util::SpanIsRoot; using ::google::cloud::testing_util::SpanKindIsClient; using ::google::cloud::testing_util::SpanNamed; using ::google::cloud::testing_util::SpanWithStatus; +using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::StatusIs; using ::google::cloud::testing_util::ThereIsAnActiveSpan; using ::testing::AllOf; From b9f230f2c513cdd5f511a1906009141ae13667ee Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 27 May 2026 10:22:56 +0000 Subject: [PATCH 02/20] Adding LRU cache and span attributes to other bucket and object operations --- .../storage/internal/tracing_connection.cc | 66 +++++++++++++++++++ .../storage/internal/tracing_connection.h | 61 +++++++++++++++++ .../internal/tracing_connection_test.cc | 1 - 3 files changed, 127 insertions(+), 1 deletion(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index ca37f072453a7..5d9ec1ff72cd8 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -53,8 +53,74 @@ TracingConnection::AsyncRunner const& TracingConnection::runner() { BucketMetadataCache& TracingConnection::cache() const { return *cache_; } +TracingConnection::~TracingConnection() { + for (auto& f : bg_tasks_) { + if (f.valid()) f.wait(); + } +} + Options TracingConnection::options() const { return impl_->options(); } +void TracingConnection::CleanupCompletedTasks() { + std::lock_guard lock(mu_); + bg_tasks_.erase(std::remove_if(bg_tasks_.begin(), bg_tasks_.end(), + [](std::future const& f) { + return f.wait_for(std::chrono::seconds(0)) == + std::future_status::ready; + }), + bg_tasks_.end()); +} + +void TracingConnection::MaybeTriggerBackgroundFetch( + std::string const& bucket_name) { + CleanupCompletedTasks(); + + std::lock_guard lock(mu_); + if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { + return; + } + + in_flight_fetch_.insert(bucket_name); + + auto f = std::async(std::launch::async, [this, bucket_name]() { + storage::internal::GetBucketMetadataRequest request(bucket_name); + auto result = impl_->GetBucketMetadata(request); + + BucketCacheEntry entry; + if (result.ok()) { + entry.id = "projects/" + std::to_string(result->project_number()) + + "/buckets/" + result->name(); + entry.location = result->location(); + if (result->location_type() == "multi-region" || + result->location_type() == "dual-region") { + entry.location = "global"; + } + cache_.Put(bucket_name, std::move(entry)); + } else if (result.status().code() == StatusCode::kPermissionDenied) { + entry.id = "projects/_/buckets/" + bucket_name; + entry.location = "global"; + cache_.Put(bucket_name, std::move(entry)); + } + + std::lock_guard lock(mu_); + in_flight_fetch_.erase(bucket_name); + }); + + bg_tasks_.push_back(std::move(f)); +} + +void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, + 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(bucket_name); + } +} + void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, BucketCacheEntry const& entry) { span.SetAttribute("gcp.resource.destination.id", entry.id); diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index a979f9adea585..930a530f9ebbe 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -22,6 +22,7 @@ #include "absl/base/call_once.h" #include #include +#include #include namespace google { @@ -29,6 +30,66 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +struct BucketCacheEntry { + std::string id; + std::string location; +}; + +class BucketMetadataCache { + public: + explicit BucketMetadataCache(std::size_t max_size = 10000) + : max_size_(max_size) {} + + absl::optional Get(std::string const& bucket_name) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it == map_.end()) return absl::nullopt; + + list_.erase(it->second.second); + list_.push_front(bucket_name); + it->second.second = list_.begin(); + return it->second.first; + } + + void Put(std::string const& bucket_name, BucketCacheEntry entry) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it != map_.end()) { + it->second.first = entry; + list_.erase(it->second.second); + list_.push_front(bucket_name); + it->second.second = list_.begin(); + 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 Invalidate(std::string const& bucket_name) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it != map_.end()) { + list_.erase(it->second.second); + map_.erase(it); + } + } + + private: + std::size_t max_size_; + std::mutex mu_; + std::list list_; + std::unordered_map::iterator>> + map_; +}; + class TracingConnection : public storage::internal::StorageConnection { public: using AsyncRunner = std::function)>; diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index 531dc7736d6f0..3cc75811ff639 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -44,7 +44,6 @@ using ::google::cloud::testing_util::SpanIsRoot; using ::google::cloud::testing_util::SpanKindIsClient; using ::google::cloud::testing_util::SpanNamed; using ::google::cloud::testing_util::SpanWithStatus; -using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::StatusIs; using ::google::cloud::testing_util::ThereIsAnActiveSpan; using ::testing::AllOf; From 44576057c9408c8f56d87f8d25cd32d7c967e181 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 2 Jun 2026 14:11:22 +0000 Subject: [PATCH 03/20] fix ci failures --- .../cloud/storage/internal/tracing_connection.cc | 15 ++++++++++++--- .../cloud/storage/internal/tracing_connection.h | 10 ++++++++++ .../storage/internal/tracing_connection_test.cc | 1 + 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 5d9ec1ff72cd8..1d8eee509698f 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -59,6 +59,15 @@ TracingConnection::~TracingConnection() { } } +BucketMetadataCache& TracingConnection::cache() { + static BucketMetadataCache instance(10000); + return instance; +} + +void TracingConnection::ResetCacheForTesting() { + cache().Clear(); +} + Options TracingConnection::options() const { return impl_->options(); } void TracingConnection::CleanupCompletedTasks() { @@ -95,11 +104,11 @@ void TracingConnection::MaybeTriggerBackgroundFetch( result->location_type() == "dual-region") { entry.location = "global"; } - cache_.Put(bucket_name, std::move(entry)); + cache().Put(bucket_name, std::move(entry)); } else if (result.status().code() == StatusCode::kPermissionDenied) { entry.id = "projects/_/buckets/" + bucket_name; entry.location = "global"; - cache_.Put(bucket_name, std::move(entry)); + cache().Put(bucket_name, std::move(entry)); } std::lock_guard lock(mu_); @@ -112,7 +121,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch( void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, std::string const& bucket_name) { if (bucket_name.empty()) return; - auto entry = cache_.Get(bucket_name); + 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); diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index 930a530f9ebbe..789ca50241d1c 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -81,6 +81,12 @@ class BucketMetadataCache { } } + void Clear() { + std::lock_guard lock(mu_); + map_.clear(); + list_.clear(); + } + private: std::size_t max_size_; std::mutex mu_; @@ -97,6 +103,8 @@ class TracingConnection : public storage::internal::StorageConnection { AsyncRunner runner = {}); ~TracingConnection() override; + static void ResetCacheForTesting(); + Options options() const override; StatusOr ListBuckets( @@ -267,6 +275,8 @@ class TracingConnection : public storage::internal::StorageConnection { AsyncRunner const& runner(); + static BucketMetadataCache& cache(); + std::shared_ptr impl_; std::shared_ptr cache_; absl::once_flag once_flag_; diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index 3cc75811ff639..d9210d047506f 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -216,6 +216,7 @@ TEST(TracingClientTest, GetBucketMetadataSuccess) { } TEST(TracingClientTest, BucketMetadataCacheSuccess) { + TracingConnection::ResetCacheForTesting(); auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); From bc689c9a6c3a9d2b18e21b5905590a16db07c249 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 2 Jun 2026 14:24:29 +0000 Subject: [PATCH 04/20] fix ci failures --- google/cloud/storage/internal/tracing_connection.cc | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 1d8eee509698f..5e749a1a705b7 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -64,9 +64,7 @@ BucketMetadataCache& TracingConnection::cache() { return instance; } -void TracingConnection::ResetCacheForTesting() { - cache().Clear(); -} +void TracingConnection::ResetCacheForTesting() { cache().Clear(); } Options TracingConnection::options() const { return impl_->options(); } From 7e66128cbae371b2a83f365b79669775d832f7a2 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 3 Jun 2026 08:09:39 +0000 Subject: [PATCH 05/20] fix ci failures --- .../storage/internal/tracing_connection.cc | 16 +++-- .../storage/internal/tracing_connection.h | 66 ------------------- 2 files changed, 10 insertions(+), 72 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 5e749a1a705b7..e6e54458d94c5 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -82,16 +82,21 @@ void TracingConnection::MaybeTriggerBackgroundFetch( std::string const& bucket_name) { CleanupCompletedTasks(); - std::lock_guard lock(mu_); - if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { + if (!cache().StartFetch(bucket_name)) { return; } - in_flight_fetch_.insert(bucket_name); - auto f = std::async(std::launch::async, [this, bucket_name]() { storage::internal::GetBucketMetadataRequest request(bucket_name); auto result = impl_->GetBucketMetadata(request); + std::cout << "BG Thread: GetBucketMetadata returned ok: " << result.ok() + << "\n"; + if (!result.ok()) { + std::cout << "BG Thread: GetBucketMetadata status: " + << result.status().message() + << " code: " << static_cast(result.status().code()) + << "\n"; + } BucketCacheEntry entry; if (result.ok()) { @@ -109,8 +114,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch( cache().Put(bucket_name, std::move(entry)); } - std::lock_guard lock(mu_); - in_flight_fetch_.erase(bucket_name); + cache().EndFetch(bucket_name); }); bg_tasks_.push_back(std::move(f)); diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index 789ca50241d1c..b5d3da49b2918 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -30,72 +30,6 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN -struct BucketCacheEntry { - std::string id; - std::string location; -}; - -class BucketMetadataCache { - public: - explicit BucketMetadataCache(std::size_t max_size = 10000) - : max_size_(max_size) {} - - absl::optional Get(std::string const& bucket_name) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it == map_.end()) return absl::nullopt; - - list_.erase(it->second.second); - list_.push_front(bucket_name); - it->second.second = list_.begin(); - return it->second.first; - } - - void Put(std::string const& bucket_name, BucketCacheEntry entry) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it != map_.end()) { - it->second.first = entry; - list_.erase(it->second.second); - list_.push_front(bucket_name); - it->second.second = list_.begin(); - 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 Invalidate(std::string const& bucket_name) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it != map_.end()) { - list_.erase(it->second.second); - map_.erase(it); - } - } - - void Clear() { - std::lock_guard lock(mu_); - map_.clear(); - list_.clear(); - } - - private: - std::size_t max_size_; - std::mutex mu_; - std::list list_; - std::unordered_map::iterator>> - map_; -}; - class TracingConnection : public storage::internal::StorageConnection { public: using AsyncRunner = std::function)>; From a21862f44a2251664096d97180974cd5a091dc11 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 3 Jun 2026 10:32:05 +0000 Subject: [PATCH 06/20] fix ci failures --- .../storage/internal/tracing_connection.cc | 63 +++++++------- .../storage/internal/tracing_connection.h | 82 +++++++++++++++++++ 2 files changed, 115 insertions(+), 30 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index e6e54458d94c5..6e8a8f192d2a3 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -86,36 +86,39 @@ void TracingConnection::MaybeTriggerBackgroundFetch( return; } - auto f = std::async(std::launch::async, [this, bucket_name]() { - storage::internal::GetBucketMetadataRequest request(bucket_name); - auto result = impl_->GetBucketMetadata(request); - std::cout << "BG Thread: GetBucketMetadata returned ok: " << result.ok() - << "\n"; - if (!result.ok()) { - std::cout << "BG Thread: GetBucketMetadata status: " - << result.status().message() - << " code: " << static_cast(result.status().code()) - << "\n"; - } - - BucketCacheEntry entry; - if (result.ok()) { - entry.id = "projects/" + std::to_string(result->project_number()) + - "/buckets/" + result->name(); - entry.location = result->location(); - if (result->location_type() == "multi-region" || - result->location_type() == "dual-region") { - entry.location = "global"; - } - cache().Put(bucket_name, std::move(entry)); - } else if (result.status().code() == StatusCode::kPermissionDenied) { - entry.id = "projects/_/buckets/" + bucket_name; - entry.location = "global"; - cache().Put(bucket_name, std::move(entry)); - } - - cache().EndFetch(bucket_name); - }); + auto current_options = google::cloud::internal::SaveCurrentOptions(); + auto f = + std::async(std::launch::async, [this, bucket_name, current_options]() { + google::cloud::internal::OptionsSpan span(current_options); + storage::internal::GetBucketMetadataRequest request(bucket_name); + auto result = impl_->GetBucketMetadata(request); + std::cout << "BG Thread: GetBucketMetadata returned ok: " << result.ok() + << "\n"; + if (!result.ok()) { + std::cout << "BG Thread: GetBucketMetadata status: " + << result.status().message() + << " code: " << static_cast(result.status().code()) + << "\n"; + } + + BucketCacheEntry entry; + if (result.ok()) { + entry.id = "projects/" + std::to_string(result->project_number()) + + "/buckets/" + result->name(); + entry.location = result->location(); + if (result->location_type() == "multi-region" || + result->location_type() == "dual-region") { + entry.location = "global"; + } + cache().Put(bucket_name, std::move(entry)); + } else if (result.status().code() == StatusCode::kPermissionDenied) { + entry.id = "projects/_/buckets/" + bucket_name; + entry.location = "global"; + cache().Put(bucket_name, std::move(entry)); + } + + cache().EndFetch(bucket_name); + }); bg_tasks_.push_back(std::move(f)); } diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index b5d3da49b2918..de7fab7b6ab06 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -30,6 +30,88 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +struct BucketCacheEntry { + std::string id; + std::string location; +}; + +class BucketMetadataCache { + public: + explicit BucketMetadataCache(std::size_t max_size = 10000) + : max_size_(max_size) {} + + absl::optional Get(std::string const& bucket_name) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it == map_.end()) return absl::nullopt; + + list_.erase(it->second.second); + list_.push_front(bucket_name); + it->second.second = list_.begin(); + return it->second.first; + } + + void Put(std::string const& bucket_name, BucketCacheEntry entry) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it != map_.end()) { + it->second.first = entry; + list_.erase(it->second.second); + list_.push_front(bucket_name); + it->second.second = list_.begin(); + 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 Invalidate(std::string const& bucket_name) { + std::lock_guard lock(mu_); + auto it = map_.find(bucket_name); + if (it != map_.end()) { + list_.erase(it->second.second); + map_.erase(it); + } + } + + void Clear() { + std::lock_guard lock(mu_); + map_.clear(); + list_.clear(); + in_flight_fetch_.clear(); + } + + bool StartFetch(std::string const& bucket_name) { + std::lock_guard lock(mu_); + if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { + return false; + } + in_flight_fetch_.insert(bucket_name); + return true; + } + + void EndFetch(std::string const& bucket_name) { + std::lock_guard lock(mu_); + in_flight_fetch_.erase(bucket_name); + } + + private: + std::size_t max_size_; + std::mutex mu_; + std::list list_; + std::unordered_map::iterator>> + map_; + std::unordered_set in_flight_fetch_; +}; + class TracingConnection : public storage::internal::StorageConnection { public: using AsyncRunner = std::function)>; From fb735a997941ad2c81702c0e59565fcbe2bd3704 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Mon, 8 Jun 2026 11:51:19 +0000 Subject: [PATCH 07/20] fix ci failures --- ...lenty_clients_serially_integration_test.cc | 25 ++++++------------- ...clients_simultaneously_integration_test.cc | 18 +++++++------ 2 files changed, 18 insertions(+), 25 deletions(-) diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index 9a551068bc396..ab4d836cd7e85 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -43,26 +43,17 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { // own tests. if (UsingGrpc()) GTEST_SKIP(); - // With the advent of Regional Access Boundaries, connecting to non-regional - // endpoints requires background calls to IAM. These background calls use - // additional file descriptors which causes this test to fail when using the - // default endpoint. - auto regional_bucket = google::cloud::internal::GetEnv( - "GOOGLE_CLOUD_CPP_STORAGE_TEST_DESTINATION"); - if (!regional_bucket.has_value()) GTEST_SKIP(); - bucket_name_ = *regional_bucket; - // The regional_bucket was created in the us-west2 region. - auto options = Options{}.set( - "https://storage.us-west2.rep.googleapis.com"); - - auto client = MakeIntegrationTestClient(options); + auto options = Options{}.set(false); auto object_name = MakeRandomObjectName(); std::string expected = LoremIpsum(); - StatusOr meta = client.InsertObject( - bucket_name_, object_name, expected, IfGenerationMatch(0)); - ASSERT_STATUS_OK(meta); - ScheduleForDelete(*meta); + { + auto client = MakeIntegrationTestClient(options); + StatusOr meta = client.InsertObject( + bucket_name_, object_name, expected, IfGenerationMatch(0)); + ASSERT_STATUS_OK(meta); + ScheduleForDelete(*meta); + } // Track the number of open files to ensure every client creates the same // number of file descriptors and none are leaked. diff --git a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc index 3e489b2e23db0..82dee21ffb9ed 100644 --- a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc @@ -16,6 +16,7 @@ #include "google/cloud/storage/testing/object_integration_test.h" #include "google/cloud/storage/testing/storage_integration_test.h" #include "google/cloud/log.h" +#include "google/cloud/opentelemetry_options.h" #include "google/cloud/status_or.h" #include "google/cloud/testing_util/expect_exception.h" #include "google/cloud/testing_util/status_matchers.h" @@ -45,22 +46,23 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest, // own tests. if (UsingGrpc()) GTEST_SKIP(); - auto client = MakeIntegrationTestClient(); + auto options = Options{}.set(false); auto object_name = MakeRandomObjectName(); - std::string expected = LoremIpsum(); - // Create the object, but only if it does not exist already. - StatusOr meta = client.InsertObject( - bucket_name_, object_name, expected, IfGenerationMatch(0)); - ASSERT_STATUS_OK(meta); - ScheduleForDelete(*meta); + { + auto client = MakeIntegrationTestClient(options); + StatusOr meta = client.InsertObject( + bucket_name_, object_name, expected, IfGenerationMatch(0)); + ASSERT_STATUS_OK(meta); + ScheduleForDelete(*meta); + } auto num_fds_before_test = GetNumOpenFiles(); std::vector read_clients; std::vector read_streams; for (int i = 0; i != 100; ++i) { - auto read_client = MakeIntegrationTestClient(); + auto read_client = MakeIntegrationTestClient(options); auto stream = read_client.ReadObject(bucket_name_, object_name); char c; stream.read(&c, 1); From 872b5888496290004d3f591f8ae23d46c517fd68 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Mon, 8 Jun 2026 14:27:46 +0000 Subject: [PATCH 08/20] fix ci failures --- .../tests/object_plenty_clients_serially_integration_test.cc | 4 +++- .../object_plenty_clients_simultaneously_integration_test.cc | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index ab4d836cd7e85..136d5d0368ca9 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -17,6 +17,7 @@ #include "google/cloud/storage/testing/storage_integration_test.h" #include "google/cloud/internal/getenv.h" #include "google/cloud/log.h" +#include "google/cloud/opentelemetry_options.h" #include "google/cloud/status_or.h" #include "google/cloud/testing_util/expect_exception.h" #include "google/cloud/testing_util/status_matchers.h" @@ -43,7 +44,8 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { // own tests. if (UsingGrpc()) GTEST_SKIP(); - auto options = Options{}.set(false); + auto options = + Options{}.set(false); auto object_name = MakeRandomObjectName(); std::string expected = LoremIpsum(); diff --git a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc index 82dee21ffb9ed..26204cc90c5fe 100644 --- a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc @@ -46,7 +46,8 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest, // own tests. if (UsingGrpc()) GTEST_SKIP(); - auto options = Options{}.set(false); + auto options = + Options{}.set(false); auto object_name = MakeRandomObjectName(); std::string expected = LoremIpsum(); From 5b48aef0ffe710344d01d5a490361649649ab3c2 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 9 Jun 2026 10:56:34 +0000 Subject: [PATCH 09/20] move BucketMetadataCache to separate file --- .../storage/internal/tracing_connection.h | 82 ------------------- 1 file changed, 82 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index de7fab7b6ab06..b5d3da49b2918 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -30,88 +30,6 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN -struct BucketCacheEntry { - std::string id; - std::string location; -}; - -class BucketMetadataCache { - public: - explicit BucketMetadataCache(std::size_t max_size = 10000) - : max_size_(max_size) {} - - absl::optional Get(std::string const& bucket_name) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it == map_.end()) return absl::nullopt; - - list_.erase(it->second.second); - list_.push_front(bucket_name); - it->second.second = list_.begin(); - return it->second.first; - } - - void Put(std::string const& bucket_name, BucketCacheEntry entry) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it != map_.end()) { - it->second.first = entry; - list_.erase(it->second.second); - list_.push_front(bucket_name); - it->second.second = list_.begin(); - 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 Invalidate(std::string const& bucket_name) { - std::lock_guard lock(mu_); - auto it = map_.find(bucket_name); - if (it != map_.end()) { - list_.erase(it->second.second); - map_.erase(it); - } - } - - void Clear() { - std::lock_guard lock(mu_); - map_.clear(); - list_.clear(); - in_flight_fetch_.clear(); - } - - bool StartFetch(std::string const& bucket_name) { - std::lock_guard lock(mu_); - if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { - return false; - } - in_flight_fetch_.insert(bucket_name); - return true; - } - - void EndFetch(std::string const& bucket_name) { - std::lock_guard lock(mu_); - in_flight_fetch_.erase(bucket_name); - } - - private: - std::size_t max_size_; - std::mutex mu_; - std::list list_; - std::unordered_map::iterator>> - map_; - std::unordered_set in_flight_fetch_; -}; - class TracingConnection : public storage::internal::StorageConnection { public: using AsyncRunner = std::function)>; From ea26fad2a017f85392a3c9afac644195c9d6b474 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 9 Jun 2026 13:33:29 +0000 Subject: [PATCH 10/20] more refactoring --- google/cloud/storage/internal/tracing_connection.cc | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 6e8a8f192d2a3..53ca9e778b960 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -69,7 +69,7 @@ void TracingConnection::ResetCacheForTesting() { cache().Clear(); } Options TracingConnection::options() const { return impl_->options(); } void TracingConnection::CleanupCompletedTasks() { - std::lock_guard lock(mu_); + std::unique_lock lk(mu_); bg_tasks_.erase(std::remove_if(bg_tasks_.begin(), bg_tasks_.end(), [](std::future const& f) { return f.wait_for(std::chrono::seconds(0)) == @@ -92,14 +92,6 @@ void TracingConnection::MaybeTriggerBackgroundFetch( google::cloud::internal::OptionsSpan span(current_options); storage::internal::GetBucketMetadataRequest request(bucket_name); auto result = impl_->GetBucketMetadata(request); - std::cout << "BG Thread: GetBucketMetadata returned ok: " << result.ok() - << "\n"; - if (!result.ok()) { - std::cout << "BG Thread: GetBucketMetadata status: " - << result.status().message() - << " code: " << static_cast(result.status().code()) - << "\n"; - } BucketCacheEntry entry; if (result.ok()) { From 11ec323bc85513abccbde52965cb828f41ee1644 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 9 Jun 2026 15:07:03 +0000 Subject: [PATCH 11/20] more refactoring --- .../storage/internal/tracing_connection.cc | 47 +++++++++---------- 1 file changed, 21 insertions(+), 26 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 53ca9e778b960..d509b16afca17 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -78,6 +78,12 @@ void TracingConnection::CleanupCompletedTasks() { bg_tasks_.end()); } +void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, + BucketCacheEntry const& entry) { + span.SetAttribute("gcp.resource.destination.id", entry.id); + span.SetAttribute("gcp.resource.destination.location", entry.location); +} + void TracingConnection::MaybeTriggerBackgroundFetch( std::string const& bucket_name) { CleanupCompletedTasks(); @@ -87,30 +93,20 @@ void TracingConnection::MaybeTriggerBackgroundFetch( } auto current_options = google::cloud::internal::SaveCurrentOptions(); - auto f = - std::async(std::launch::async, [this, bucket_name, current_options]() { - google::cloud::internal::OptionsSpan span(current_options); - storage::internal::GetBucketMetadataRequest request(bucket_name); - auto result = impl_->GetBucketMetadata(request); - - BucketCacheEntry entry; - if (result.ok()) { - entry.id = "projects/" + std::to_string(result->project_number()) + - "/buckets/" + result->name(); - entry.location = result->location(); - if (result->location_type() == "multi-region" || - result->location_type() == "dual-region") { - entry.location = "global"; - } - cache().Put(bucket_name, std::move(entry)); - } else if (result.status().code() == StatusCode::kPermissionDenied) { - entry.id = "projects/_/buckets/" + bucket_name; - entry.location = "global"; - cache().Put(bucket_name, std::move(entry)); - } - - cache().EndFetch(bucket_name); - }); + auto f = std::async(std::launch::async, [this, bucket_name, + current_options]() { + google::cloud::internal::OptionsSpan span(current_options); + storage::internal::GetBucketMetadataRequest request(bucket_name); + auto result = impl_->GetBucketMetadata(request); + + if (result.ok()) { + cache().Put(bucket_name, BucketCacheEntry::FromMetadata(*result)); + } else if (result.status().code() == StatusCode::kPermissionDenied) { + cache().Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"}); + } + + cache().EndFetch(bucket_name); + }); bg_tasks_.push_back(std::move(f)); } @@ -120,8 +116,7 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, 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); + EnrichSpan(span, *entry); } else { MaybeTriggerBackgroundFetch(bucket_name); } From 2fcc7d86f6ec666fb9b3bec4f193b645bf19484d Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 10 Jun 2026 12:30:22 +0000 Subject: [PATCH 12/20] feat(storage): add resource span attributes for ACO ( App Centric Observability ) for async client --- .../internal/async/connection_tracing.cc | 84 +++++++++++++++++++ .../storage/internal/bucket_metadata_cache.cc | 19 +++-- .../storage/internal/bucket_metadata_cache.h | 10 +++ .../storage/internal/tracing_connection.cc | 3 +- 4 files changed, 109 insertions(+), 7 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 2e29660276078..80a416e72d45b 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -14,16 +14,25 @@ #include "google/cloud/storage/internal/async/connection_tracing.h" #include "google/cloud/storage/async/writer_connection.h" +#include "google/cloud/storage/internal/async/default_options.h" #include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h" #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 +#include +#include #include +#include +#include namespace google { namespace cloud { + namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN @@ -35,6 +44,12 @@ class AsyncConnectionTracing : public storage::AsyncConnection { std::shared_ptr impl) : impl_(std::move(impl)) {} + ~AsyncConnectionTracing() override { + for (auto& f : bg_tasks_) { + if (f.valid()) f.wait(); + } + } + Options options() const override { return impl_->options(); } future> GetBucket( @@ -47,6 +62,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future> 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))); @@ -55,6 +72,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future>> 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(), @@ -74,6 +92,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future>> 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) @@ -89,6 +108,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future> 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(), @@ -103,6 +123,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(), @@ -119,6 +141,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(), @@ -135,6 +159,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(), @@ -151,6 +177,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(), @@ -198,6 +226,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future> 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))); @@ -205,6 +234,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future 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))); @@ -218,7 +248,61 @@ class AsyncConnectionTracing : public storage::AsyncConnection { } private: + void CleanupCompletedTasks() { + std::unique_lock lk(mu_); + bg_tasks_.erase( + std::remove_if(bg_tasks_.begin(), bg_tasks_.end(), + [](std::future 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 (!BucketMetadataCache::Singleton().StartFetch(bucket_name)) { + return; + } + + auto f = std::async(std::launch::async, [bucket_name, options]() { + google::cloud::internal::OptionsSpan span(options); + auto conn = MakeStorageConnection(options); + auto const normalized = + BucketMetadataCache::NormalizeBucketName(bucket_name); + storage::internal::GetBucketMetadataRequest request(normalized); + auto metadata = conn->GetBucketMetadata(request); + if (metadata.ok()) { + BucketMetadataCache::Singleton().Put( + bucket_name, BucketCacheEntry::FromMetadata(*metadata)); + } else if (metadata.status().code() == StatusCode::kPermissionDenied) { + BucketMetadataCache::Singleton().Put( + bucket_name, {"projects/_/buckets/" + normalized, "global"}); + } + BucketMetadataCache::Singleton().EndFetch(bucket_name); + }); + + std::unique_lock 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 = BucketMetadataCache::Singleton().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 impl_; + std::vector> bg_tasks_; + std::mutex mu_; }; } // namespace diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc index 09e74119e8041..58fddffdab9bc 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.cc +++ b/google/cloud/storage/internal/bucket_metadata_cache.cc @@ -22,6 +22,11 @@ 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(); @@ -39,7 +44,8 @@ void BucketMetadataCache::MoveToFront(std::list::iterator it) { } absl::optional BucketMetadataCache::Get( - std::string const& bucket_name) { + std::string const& raw_bucket_name) { + auto const bucket_name = NormalizeBucketName(raw_bucket_name); std::unique_lock lk(mu_); auto it = map_.find(bucket_name); if (it == map_.end()) return absl::nullopt; @@ -48,7 +54,7 @@ absl::optional BucketMetadataCache::Get( return it->second.first; } -void BucketMetadataCache::Put(std::string const& bucket_name, +void BucketMetadataCache::Put(std::string const& raw_bucket_name, BucketCacheEntry entry) { if (max_size_ == 0) return; std::unique_lock lk(mu_); @@ -69,7 +75,8 @@ void BucketMetadataCache::Put(std::string const& bucket_name, map_[bucket_name] = {std::move(entry), list_.begin()}; } -void BucketMetadataCache::Invalidate(std::string const& bucket_name) { +void BucketMetadataCache::Invalidate(std::string const& raw_bucket_name) { + auto const bucket_name = NormalizeBucketName(raw_bucket_name); std::unique_lock lk(mu_); auto it = map_.find(bucket_name); if (it != map_.end()) { @@ -84,7 +91,8 @@ void BucketMetadataCache::Clear() { list_.clear(); } -bool BucketMetadataCache::StartFetch(std::string const& bucket_name) { +bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) { + auto const bucket_name = NormalizeBucketName(raw_bucket_name); std::unique_lock lk(mu_); if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { return false; @@ -93,7 +101,8 @@ bool BucketMetadataCache::StartFetch(std::string const& bucket_name) { return true; } -void BucketMetadataCache::EndFetch(std::string const& bucket_name) { +void BucketMetadataCache::EndFetch(std::string const& raw_bucket_name) { + auto const bucket_name = NormalizeBucketName(raw_bucket_name); std::unique_lock lk(mu_); in_flight_fetch_.erase(bucket_name); } diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h index f3023126b2d28..b16abfbfca912 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.h +++ b/google/cloud/storage/internal/bucket_metadata_cache.h @@ -48,6 +48,16 @@ class BucketMetadataCache { explicit BucketMetadataCache(std::size_t max_size = 10000) : max_size_(max_size) {} + static BucketMetadataCache& Singleton(); + + static std::string NormalizeBucketName(std::string const& bucket) { + auto const prefix = std::string("projects/_/buckets/"); + if (bucket.rfind(prefix, 0) == 0) { + return bucket.substr(prefix.size()); + } + return bucket; + } + absl::optional Get(std::string const& bucket_name); void Put(std::string const& bucket_name, BucketCacheEntry entry); void Invalidate(std::string const& bucket_name); diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index d509b16afca17..1b76864fbb9aa 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -60,8 +60,7 @@ TracingConnection::~TracingConnection() { } BucketMetadataCache& TracingConnection::cache() { - static BucketMetadataCache instance(10000); - return instance; + return BucketMetadataCache::Singleton(); } void TracingConnection::ResetCacheForTesting() { cache().Clear(); } From e9dbaa3f17d737908161d07efc6baad46fbfb7ab Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Thu, 11 Jun 2026 05:07:35 +0000 Subject: [PATCH 13/20] fix ci failure --- google/cloud/storage/internal/bucket_metadata_cache.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h index b16abfbfca912..fd1f3d5ad58fa 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.h +++ b/google/cloud/storage/internal/bucket_metadata_cache.h @@ -16,6 +16,7 @@ #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H #include "google/cloud/storage/version.h" +#include "absl/strings/match.h" #include "absl/types/optional.h" #include #include @@ -52,7 +53,7 @@ class BucketMetadataCache { static std::string NormalizeBucketName(std::string const& bucket) { auto const prefix = std::string("projects/_/buckets/"); - if (bucket.rfind(prefix, 0) == 0) { + if (absl::StartsWith(bucket, prefix)) { return bucket.substr(prefix.size()); } return bucket; From e2c9822c0b443accaf6077be27dfc51e6496fef0 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Thu, 11 Jun 2026 05:54:40 +0000 Subject: [PATCH 14/20] Refactoring --- .../internal/async/connection_tracing.cc | 46 +++++++++++-------- .../storage/internal/bucket_metadata_cache.cc | 16 +++++++ .../storage/internal/bucket_metadata_cache.h | 12 ++--- .../storage/internal/tracing_connection.cc | 27 ++++++----- ...lenty_clients_serially_integration_test.cc | 18 ++++++-- 5 files changed, 77 insertions(+), 42 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 80a416e72d45b..77d0e12cdfc7b 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -14,7 +14,6 @@ #include "google/cloud/storage/internal/async/connection_tracing.h" #include "google/cloud/storage/async/writer_connection.h" -#include "google/cloud/storage/internal/async/default_options.h" #include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h" #include "google/cloud/storage/internal/async/reader_connection_tracing.h" #include "google/cloud/storage/internal/async/rewriter_connection_tracing.h" @@ -42,7 +41,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { public: explicit AsyncConnectionTracing( std::shared_ptr impl) - : impl_(std::move(impl)) {} + : impl_(std::move(impl)), cache_(BucketMetadataCache::Singleton()) {} ~AsyncConnectionTracing() override { for (auto& f : bg_tasks_) { @@ -263,26 +262,33 @@ class AsyncConnectionTracing : public storage::AsyncConnection { std::string const& bucket_name) { CleanupCompletedTasks(); - if (!BucketMetadataCache::Singleton().StartFetch(bucket_name)) { + if (!cache_.StartFetch(bucket_name)) { return; } - auto f = std::async(std::launch::async, [bucket_name, options]() { - google::cloud::internal::OptionsSpan span(options); - auto conn = MakeStorageConnection(options); - auto const normalized = - BucketMetadataCache::NormalizeBucketName(bucket_name); - storage::internal::GetBucketMetadataRequest request(normalized); - auto metadata = conn->GetBucketMetadata(request); - if (metadata.ok()) { - BucketMetadataCache::Singleton().Put( - bucket_name, BucketCacheEntry::FromMetadata(*metadata)); - } else if (metadata.status().code() == StatusCode::kPermissionDenied) { - BucketMetadataCache::Singleton().Put( - bucket_name, {"projects/_/buckets/" + normalized, "global"}); + std::shared_ptr conn; + { + std::lock_guard lk(mu_); + if (!sync_conn_) { + sync_conn_ = MakeStorageConnection(impl_->options()); } - BucketMetadataCache::Singleton().EndFetch(bucket_name); - }); + 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 lk(mu_); bg_tasks_.push_back(std::move(f)); @@ -291,7 +297,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { void EnrichSpan(opentelemetry::trace::Span& span, Options const& options, std::string const& bucket_name) { if (bucket_name.empty()) return; - auto entry = BucketMetadataCache::Singleton().Get(bucket_name); + 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); @@ -301,6 +307,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection { } std::shared_ptr impl_; + BucketMetadataCache& cache_; + std::shared_ptr sync_conn_; std::vector> bg_tasks_; std::mutex mu_; }; diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc index 58fddffdab9bc..532f4cea8c5ad 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.cc +++ b/google/cloud/storage/internal/bucket_metadata_cache.cc @@ -14,6 +14,7 @@ #include "google/cloud/storage/internal/bucket_metadata_cache.h" #include "google/cloud/storage/bucket_metadata.h" +#include "google/cloud/status.h" #include #include @@ -39,6 +40,21 @@ BucketCacheEntry BucketCacheEntry::FromMetadata( std::move(loc)}; } +absl::optional BucketCacheEntry::Create( + std::string const& bucket_name, + StatusOr 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::iterator it) { list_.splice(list_.begin(), list_, it); } diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h index fd1f3d5ad58fa..9ef38fe730459 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.h +++ b/google/cloud/storage/internal/bucket_metadata_cache.h @@ -15,7 +15,9 @@ #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H +#include "google/cloud/storage/bucket_metadata.h" #include "google/cloud/storage/version.h" +#include "google/cloud/status_or.h" #include "absl/strings/match.h" #include "absl/types/optional.h" #include @@ -29,11 +31,6 @@ namespace google { namespace cloud { -namespace storage { -GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN -class BucketMetadata; -GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END -} // namespace storage namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN @@ -42,12 +39,15 @@ struct BucketCacheEntry { std::string location; static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m); + static absl::optional Create( + std::string const& bucket_name, + StatusOr const& metadata); }; class BucketMetadataCache { public: explicit BucketMetadataCache(std::size_t max_size = 10000) - : max_size_(max_size) {} + : max_size_(max_size == 0 ? 1 : max_size) {} static BucketMetadataCache& Singleton(); diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 1b76864fbb9aa..3709f4669c180 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -92,20 +92,19 @@ void TracingConnection::MaybeTriggerBackgroundFetch( } auto current_options = google::cloud::internal::SaveCurrentOptions(); - auto f = std::async(std::launch::async, [this, bucket_name, - current_options]() { - google::cloud::internal::OptionsSpan span(current_options); - storage::internal::GetBucketMetadataRequest request(bucket_name); - auto result = impl_->GetBucketMetadata(request); - - if (result.ok()) { - cache().Put(bucket_name, BucketCacheEntry::FromMetadata(*result)); - } else if (result.status().code() == StatusCode::kPermissionDenied) { - cache().Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"}); - } - - cache().EndFetch(bucket_name); - }); + auto f = + std::async(std::launch::async, [this, bucket_name, current_options]() { + google::cloud::internal::OptionsSpan span(current_options); + storage::internal::GetBucketMetadataRequest request(bucket_name); + auto result = impl_->GetBucketMetadata(request); + + auto entry = BucketCacheEntry::Create(bucket_name, result); + if (entry) { + cache().Put(bucket_name, std::move(*entry)); + } + + cache().EndFetch(bucket_name); + }); bg_tasks_.push_back(std::move(f)); } diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index 136d5d0368ca9..0b2f4d4bc1375 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -68,6 +68,21 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { EXPECT_EQ(StatusCode::kUnimplemented, num_fds_before_test.status().code()); } std::size_t delta = 0; + if (track_open_files) { + // Warmup to find the maximum delta, accounting for any asynchronous + // background requests (like IAM AllowedLocations) that might or might + // not have opened their sockets yet when we measure. + for (int i = 0; i != 5; ++i) { + auto read_client = MakeIntegrationTestClient(options); + auto stream = read_client.ReadObject(bucket_name_, object_name); + char c; + stream.read(&c, 1); + auto num_fds_during_test = GetNumOpenFiles(); + ASSERT_STATUS_OK(num_fds_during_test); + delta = (std::max)(delta, *num_fds_during_test - *num_fds_before_test); + } + } + for (int i = 0; i != 100; ++i) { auto read_client = MakeIntegrationTestClient(options); auto stream = read_client.ReadObject(bucket_name_, object_name); @@ -76,9 +91,6 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { if (track_open_files) { auto num_fds_during_test = GetNumOpenFiles(); ASSERT_STATUS_OK(num_fds_during_test); - if (delta == 0) { - delta = *num_fds_during_test - *num_fds_before_test; - } EXPECT_GE(*num_fds_before_test + delta, *num_fds_during_test) << "Expect each client to create the same number of file descriptors" << ", num_fds_before_test=" << *num_fds_before_test From 7a2cfe2e2558174367da3448d3b81ed23ab812e8 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Thu, 16 Jul 2026 14:22:00 +0000 Subject: [PATCH 15/20] sync changes --- .../internal/async/connection_tracing.cc | 221 ++++++++++++------ .../internal/async/connection_tracing_test.cc | 3 + .../storage/internal/bucket_metadata_cache.cc | 35 +-- .../storage/internal/bucket_metadata_cache.h | 23 +- .../storage/internal/tracing_connection.cc | 64 +++-- .../storage/internal/tracing_connection.h | 3 - .../internal/tracing_connection_test.cc | 1 - ...lenty_clients_serially_integration_test.cc | 45 ++-- 8 files changed, 219 insertions(+), 176 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 77d0e12cdfc7b..07d78fde87586 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -19,9 +19,11 @@ #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/storage/options.h" #include "google/cloud/internal/opentelemetry.h" #include "google/cloud/version.h" +#include "absl/strings/match.h" +#include "google/storage/v2/storage.pb.h" #include #include #include @@ -37,17 +39,22 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace { +std::string NormalizeBucketName(std::string const& bucket) { + auto const prefix = std::string("projects/_/buckets/"); + if (absl::StartsWith(bucket, prefix)) { + return bucket.substr(prefix.size()); + } + return bucket; +} + class AsyncConnectionTracing : public storage::AsyncConnection { public: explicit AsyncConnectionTracing( std::shared_ptr impl) - : impl_(std::move(impl)), cache_(BucketMetadataCache::Singleton()) {} + : impl_(std::move(impl)), + cache_(std::make_shared()) {} - ~AsyncConnectionTracing() override { - for (auto& f : bg_tasks_) { - if (f.valid()) f.wait(); - } - } + ~AsyncConnectionTracing() override = default; Options options() const override { return impl_->options(); } @@ -64,8 +71,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection { 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))); + return impl_->InsertObject(std::move(p)) + .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), + span = std::move(span), this, + bucket_name = + p.request.write_object_spec().resource().bucket()](auto f) { + auto result = f.get(); + internal::DetachOTelContext(oc); + MaybeInvalidate(result, bucket_name); + return internal::EndSpan(*span, std::move(result)); + }); } future>> Open( @@ -75,12 +90,14 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->Open(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = p.read_spec.bucket()](auto f) -> StatusOr< std::shared_ptr> { auto result = f.get(); internal::DetachOTelContext(oc); if (!result) { + MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result).status()); } return MakeTracingObjectDescriptorConnection(std::move(span), @@ -94,11 +111,15 @@ class AsyncConnectionTracing : public storage::AsyncConnection { EnrichSpan(*span, p.options, p.request.bucket()); internal::OTelScope scope(span); auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = p.request.bucket()](auto f) -> StatusOr> { auto reader = f.get(); internal::DetachOTelContext(oc); - if (!reader) return internal::EndSpan(*span, std::move(reader).status()); + if (!reader) { + MaybeInvalidate(reader, bucket_name); + return internal::EndSpan(*span, std::move(reader).status()); + } return MakeTracingReaderConnection(std::move(span), *std::move(reader)); }; return impl_->ReadObject(std::move(p)).then(std::move(wrap)); @@ -111,9 +132,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->ReadObjectRange(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) { + span = std::move(span), this, + bucket_name = p.request.bucket()](auto f) { auto result = f.get(); internal::DetachOTelContext(oc); + MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result)); }); } @@ -127,11 +150,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartAppendableObjectUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = + p.request.write_object_spec().resource().bucket()](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); - if (!w) return internal::EndSpan(*span, std::move(w).status()); + if (!w) { + MaybeInvalidate(w, bucket_name); + return internal::EndSpan(*span, std::move(w).status()); + } return MakeTracingWriterConnection(span, *std::move(w)); }); } @@ -145,11 +173,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->ResumeAppendableObjectUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = + p.request.write_object_spec().resource().bucket()](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); - if (!w) return internal::EndSpan(*span, std::move(w).status()); + if (!w) { + MaybeInvalidate(w, bucket_name); + return internal::EndSpan(*span, std::move(w).status()); + } return MakeTracingWriterConnection(span, *std::move(w)); }); } @@ -163,11 +196,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartUnbufferedUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = + p.request.write_object_spec().resource().bucket()](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); - if (!w) return internal::EndSpan(*span, std::move(w).status()); + if (!w) { + MaybeInvalidate(w, bucket_name); + return internal::EndSpan(*span, std::move(w).status()); + } return MakeTracingWriterConnection(span, *std::move(w)); }); } @@ -181,11 +219,16 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartBufferedUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span)](auto f) + span = std::move(span), this, + bucket_name = + p.request.write_object_spec().resource().bucket()](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); - if (!w) return internal::EndSpan(*span, std::move(w).status()); + if (!w) { + MaybeInvalidate(w, bucket_name); + return internal::EndSpan(*span, std::move(w).status()); + } return MakeTracingWriterConnection(span, *std::move(w)); }); } @@ -227,16 +270,30 @@ class AsyncConnectionTracing : public storage::AsyncConnection { 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))); + return impl_->ComposeObject(std::move(p)) + .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), + span = std::move(span), this, + bucket_name = p.request.destination().bucket()](auto f) { + auto result = f.get(); + internal::DetachOTelContext(oc); + MaybeInvalidate(result, bucket_name); + return internal::EndSpan(*span, std::move(result)); + }); } future 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))); + return impl_->DeleteObject(std::move(p)) + .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), + span = std::move(span), this, + bucket_name = p.request.bucket()](auto f) { + auto result = f.get(); + internal::DetachOTelContext(oc); + MaybeInvalidate(result, bucket_name); + return internal::EndSpan(*span, std::move(result)); + }); } std::shared_ptr RewriteObject( @@ -246,58 +303,93 @@ class AsyncConnectionTracing : public storage::AsyncConnection { impl_->RewriteObject(std::move(p)), enabled); } + future> GetBucket( + GetBucketParams p) override { + auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket"); + internal::OTelScope scope(span); + auto const bucket_name = p.request.name(); + return impl_->GetBucket(std::move(p)) + .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), + span = std::move(span), this, + bucket_name](future> f) + -> StatusOr { + auto result = f.get(); + internal::DetachOTelContext(oc); + if (result.ok()) { + std::string loc = result->location(); + if (result->location_type() == "multi-region" || + result->location_type() == "dual-region") { + loc = "global"; + } + BucketCacheEntry entry{result->project() + "/buckets/" + + NormalizeBucketName(bucket_name), + std::move(loc)}; + cache().Put(bucket_name, std::move(entry)); + } else { + MaybeInvalidate(result, bucket_name); + } + return internal::EndSpan(*span, std::move(result)); + }); + } + private: - void CleanupCompletedTasks() { - std::unique_lock lk(mu_); - bg_tasks_.erase( - std::remove_if(bg_tasks_.begin(), bg_tasks_.end(), - [](std::future const& f) { - return f.wait_for(std::chrono::seconds(0)) == - std::future_status::ready; - }), - bg_tasks_.end()); + BucketMetadataCache& cache() const { return *cache_; } + + void MaybeInvalidate(Status const& status, std::string const& bucket_name) { + if (!status.ok() && status.code() == StatusCode::kNotFound) { + cache().Invalidate(bucket_name); + } + } + + template + void MaybeInvalidate(StatusOr const& result, + std::string const& bucket_name) { + MaybeInvalidate(result.status(), bucket_name); } void MaybeTriggerBackgroundFetch(Options const& options, std::string const& bucket_name) { - CleanupCompletedTasks(); - - if (!cache_.StartFetch(bucket_name)) { + if (!cache().StartFetch(bucket_name)) { return; } - std::shared_ptr conn; - { - std::lock_guard 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)); + auto guard = ScopedFetch(cache_, bucket_name); + google::storage::v2::GetBucketRequest request; + auto const normalized_bucket_name = NormalizeBucketName(bucket_name); + request.set_name("projects/_/buckets/" + normalized_bucket_name); + GetBucketParams params{std::move(request), options}; + + impl_->GetBucket(std::move(params)) + .then([cache = cache_, bucket_name, guard = std::move(guard)]( + future> f) { + auto metadata = f.get(); + if (metadata.ok()) { + std::string loc = metadata->location(); + if (metadata->location_type() == "multi-region" || + metadata->location_type() == "dual-region") { + loc = "global"; + } + BucketCacheEntry entry{metadata->project() + "/buckets/" + + NormalizeBucketName(bucket_name), + std::move(loc)}; + cache->Put(bucket_name, std::move(entry)); + } else if (metadata.status().code() == + StatusCode::kPermissionDenied) { + BucketCacheEntry entry{ + "projects/_/buckets/" + NormalizeBucketName(bucket_name), + "global"}; + cache->Put(bucket_name, std::move(entry)); } - cache_.EndFetch(bucket_name); }); - - std::unique_lock 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); + auto const enabled = options.get< + google::cloud::storage_experimental::OTelSpanEnrichmentOption>(); + if (!enabled) 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); @@ -307,10 +399,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { } std::shared_ptr impl_; - BucketMetadataCache& cache_; - std::shared_ptr sync_conn_; - std::vector> bg_tasks_; - std::mutex mu_; + std::shared_ptr cache_; }; } // namespace diff --git a/google/cloud/storage/internal/async/connection_tracing_test.cc b/google/cloud/storage/internal/async/connection_tracing_test.cc index bdc562baef97c..f77af6ddf549f 100644 --- a/google/cloud/storage/internal/async/connection_tracing_test.cc +++ b/google/cloud/storage/internal/async/connection_tracing_test.cc @@ -20,6 +20,7 @@ #include "google/cloud/storage/mocks/mock_async_reader_connection.h" #include "google/cloud/storage/mocks/mock_async_rewriter_connection.h" #include "google/cloud/storage/mocks/mock_async_writer_connection.h" +#include "google/cloud/storage/options.h" #include "google/cloud/storage/testing/canonical_errors.h" #include "google/cloud/opentelemetry_options.h" #include "google/cloud/testing_util/opentelemetry_matchers.h" @@ -44,8 +45,10 @@ using ::google::cloud::testing_util::EventNamed; using ::google::cloud::testing_util::InstallSpanCatcher; using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::IsOkAndHolds; +using ::google::cloud::testing_util::OTelAttribute; using ::google::cloud::testing_util::OTelContextCaptured; using ::google::cloud::testing_util::PromiseWithOTelContext; +using ::google::cloud::testing_util::SpanHasAttributes; using ::google::cloud::testing_util::SpanHasEvents; using ::google::cloud::testing_util::SpanHasInstrumentationScope; using ::google::cloud::testing_util::SpanKindIsClient; diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc index 532f4cea8c5ad..09e74119e8041 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.cc +++ b/google/cloud/storage/internal/bucket_metadata_cache.cc @@ -14,7 +14,6 @@ #include "google/cloud/storage/internal/bucket_metadata_cache.h" #include "google/cloud/storage/bucket_metadata.h" -#include "google/cloud/status.h" #include #include @@ -23,11 +22,6 @@ 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(); @@ -40,28 +34,12 @@ BucketCacheEntry BucketCacheEntry::FromMetadata( std::move(loc)}; } -absl::optional BucketCacheEntry::Create( - std::string const& bucket_name, - StatusOr 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::iterator it) { list_.splice(list_.begin(), list_, it); } absl::optional BucketMetadataCache::Get( - std::string const& raw_bucket_name) { - auto const bucket_name = NormalizeBucketName(raw_bucket_name); + std::string const& bucket_name) { std::unique_lock lk(mu_); auto it = map_.find(bucket_name); if (it == map_.end()) return absl::nullopt; @@ -70,7 +48,7 @@ absl::optional BucketMetadataCache::Get( return it->second.first; } -void BucketMetadataCache::Put(std::string const& raw_bucket_name, +void BucketMetadataCache::Put(std::string const& bucket_name, BucketCacheEntry entry) { if (max_size_ == 0) return; std::unique_lock lk(mu_); @@ -91,8 +69,7 @@ void BucketMetadataCache::Put(std::string const& raw_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); +void BucketMetadataCache::Invalidate(std::string const& bucket_name) { std::unique_lock lk(mu_); auto it = map_.find(bucket_name); if (it != map_.end()) { @@ -107,8 +84,7 @@ void BucketMetadataCache::Clear() { list_.clear(); } -bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) { - auto const bucket_name = NormalizeBucketName(raw_bucket_name); +bool BucketMetadataCache::StartFetch(std::string const& bucket_name) { std::unique_lock lk(mu_); if (in_flight_fetch_.find(bucket_name) != in_flight_fetch_.end()) { return false; @@ -117,8 +93,7 @@ bool BucketMetadataCache::StartFetch(std::string const& raw_bucket_name) { return true; } -void BucketMetadataCache::EndFetch(std::string const& raw_bucket_name) { - auto const bucket_name = NormalizeBucketName(raw_bucket_name); +void BucketMetadataCache::EndFetch(std::string const& bucket_name) { std::unique_lock lk(mu_); in_flight_fetch_.erase(bucket_name); } diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h index 9ef38fe730459..f3023126b2d28 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.h +++ b/google/cloud/storage/internal/bucket_metadata_cache.h @@ -15,10 +15,7 @@ #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H -#include "google/cloud/storage/bucket_metadata.h" #include "google/cloud/storage/version.h" -#include "google/cloud/status_or.h" -#include "absl/strings/match.h" #include "absl/types/optional.h" #include #include @@ -31,6 +28,11 @@ namespace google { namespace cloud { +namespace storage { +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +class BucketMetadata; +GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END +} // namespace storage namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN @@ -39,25 +41,12 @@ struct BucketCacheEntry { std::string location; static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m); - static absl::optional Create( - std::string const& bucket_name, - StatusOr const& metadata); }; class BucketMetadataCache { public: explicit BucketMetadataCache(std::size_t max_size = 10000) - : max_size_(max_size == 0 ? 1 : max_size) {} - - static BucketMetadataCache& Singleton(); - - static std::string NormalizeBucketName(std::string const& bucket) { - auto const prefix = std::string("projects/_/buckets/"); - if (absl::StartsWith(bucket, prefix)) { - return bucket.substr(prefix.size()); - } - return bucket; - } + : max_size_(max_size) {} absl::optional Get(std::string const& bucket_name); void Put(std::string const& bucket_name, BucketCacheEntry entry); diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 3709f4669c180..3df3a0202a19d 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -53,30 +53,27 @@ TracingConnection::AsyncRunner const& TracingConnection::runner() { BucketMetadataCache& TracingConnection::cache() const { return *cache_; } -TracingConnection::~TracingConnection() { - for (auto& f : bg_tasks_) { - if (f.valid()) f.wait(); - } -} +TracingConnection::~TracingConnection() = default; -BucketMetadataCache& TracingConnection::cache() { - return BucketMetadataCache::Singleton(); +TracingConnection::AsyncRunner const& TracingConnection::runner() { + absl::call_once(once_flag_, [this] { + if (!runner_) { + auto threads = + std::make_shared( + 1U); + runner_ = [threads](std::function f) { + threads->cq().RunAsync(std::move(f)); + }; + } + }); + return runner_; } -void TracingConnection::ResetCacheForTesting() { cache().Clear(); } +BucketMetadataCache& TracingConnection::cache() const { return *cache_; } Options TracingConnection::options() const { return impl_->options(); } -void TracingConnection::CleanupCompletedTasks() { - std::unique_lock lk(mu_); - bg_tasks_.erase(std::remove_if(bg_tasks_.begin(), bg_tasks_.end(), - [](std::future const& f) { - return f.wait_for(std::chrono::seconds(0)) == - std::future_status::ready; - }), - bg_tasks_.end()); -} - void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, BucketCacheEntry const& entry) { span.SetAttribute("gcp.resource.destination.id", entry.id); @@ -85,33 +82,32 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, void TracingConnection::MaybeTriggerBackgroundFetch( std::string const& bucket_name) { - CleanupCompletedTasks(); - if (!cache().StartFetch(bucket_name)) { return; } + auto guard = ScopedFetch(cache_, bucket_name); auto current_options = google::cloud::internal::SaveCurrentOptions(); - auto f = - std::async(std::launch::async, [this, bucket_name, current_options]() { - google::cloud::internal::OptionsSpan span(current_options); - storage::internal::GetBucketMetadataRequest request(bucket_name); - auto result = impl_->GetBucketMetadata(request); - - auto entry = BucketCacheEntry::Create(bucket_name, result); - if (entry) { - cache().Put(bucket_name, std::move(*entry)); - } - - cache().EndFetch(bucket_name); - }); + runner()([impl = impl_, cache = cache_, bucket_name, current_options, + guard]() { + google::cloud::internal::OptionsSpan span(current_options); + storage::internal::GetBucketMetadataRequest request(bucket_name); + auto result = impl->GetBucketMetadata(request); - bg_tasks_.push_back(std::move(f)); + if (result.ok()) { + cache->Put(bucket_name, BucketCacheEntry::FromMetadata(*result)); + } else if (result.status().code() == StatusCode::kPermissionDenied) { + cache->Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"}); + } + }); } void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, std::string const& bucket_name) { if (bucket_name.empty()) return; + auto const enabled = + options().get(); + if (!enabled) return; auto entry = cache().Get(bucket_name); if (entry.has_value()) { EnrichSpan(span, *entry); diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index b5d3da49b2918..574769438db46 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -22,7 +22,6 @@ #include "absl/base/call_once.h" #include #include -#include #include namespace google { @@ -37,8 +36,6 @@ class TracingConnection : public storage::internal::StorageConnection { AsyncRunner runner = {}); ~TracingConnection() override; - static void ResetCacheForTesting(); - Options options() const override; StatusOr ListBuckets( diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index d9210d047506f..3cc75811ff639 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -216,7 +216,6 @@ TEST(TracingClientTest, GetBucketMetadataSuccess) { } TEST(TracingClientTest, BucketMetadataCacheSuccess) { - TracingConnection::ResetCacheForTesting(); auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index 0b2f4d4bc1375..9a551068bc396 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -17,7 +17,6 @@ #include "google/cloud/storage/testing/storage_integration_test.h" #include "google/cloud/internal/getenv.h" #include "google/cloud/log.h" -#include "google/cloud/opentelemetry_options.h" #include "google/cloud/status_or.h" #include "google/cloud/testing_util/expect_exception.h" #include "google/cloud/testing_util/status_matchers.h" @@ -44,18 +43,26 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { // own tests. if (UsingGrpc()) GTEST_SKIP(); - auto options = - Options{}.set(false); + // With the advent of Regional Access Boundaries, connecting to non-regional + // endpoints requires background calls to IAM. These background calls use + // additional file descriptors which causes this test to fail when using the + // default endpoint. + auto regional_bucket = google::cloud::internal::GetEnv( + "GOOGLE_CLOUD_CPP_STORAGE_TEST_DESTINATION"); + if (!regional_bucket.has_value()) GTEST_SKIP(); + bucket_name_ = *regional_bucket; + // The regional_bucket was created in the us-west2 region. + auto options = Options{}.set( + "https://storage.us-west2.rep.googleapis.com"); + + auto client = MakeIntegrationTestClient(options); auto object_name = MakeRandomObjectName(); std::string expected = LoremIpsum(); - { - auto client = MakeIntegrationTestClient(options); - StatusOr meta = client.InsertObject( - bucket_name_, object_name, expected, IfGenerationMatch(0)); - ASSERT_STATUS_OK(meta); - ScheduleForDelete(*meta); - } + StatusOr meta = client.InsertObject( + bucket_name_, object_name, expected, IfGenerationMatch(0)); + ASSERT_STATUS_OK(meta); + ScheduleForDelete(*meta); // Track the number of open files to ensure every client creates the same // number of file descriptors and none are leaked. @@ -68,21 +75,6 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { EXPECT_EQ(StatusCode::kUnimplemented, num_fds_before_test.status().code()); } std::size_t delta = 0; - if (track_open_files) { - // Warmup to find the maximum delta, accounting for any asynchronous - // background requests (like IAM AllowedLocations) that might or might - // not have opened their sockets yet when we measure. - for (int i = 0; i != 5; ++i) { - auto read_client = MakeIntegrationTestClient(options); - auto stream = read_client.ReadObject(bucket_name_, object_name); - char c; - stream.read(&c, 1); - auto num_fds_during_test = GetNumOpenFiles(); - ASSERT_STATUS_OK(num_fds_during_test); - delta = (std::max)(delta, *num_fds_during_test - *num_fds_before_test); - } - } - for (int i = 0; i != 100; ++i) { auto read_client = MakeIntegrationTestClient(options); auto stream = read_client.ReadObject(bucket_name_, object_name); @@ -91,6 +83,9 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { if (track_open_files) { auto num_fds_during_test = GetNumOpenFiles(); ASSERT_STATUS_OK(num_fds_during_test); + if (delta == 0) { + delta = *num_fds_during_test - *num_fds_before_test; + } EXPECT_GE(*num_fds_before_test + delta, *num_fds_during_test) << "Expect each client to create the same number of file descriptors" << ", num_fds_before_test=" << *num_fds_before_test From b67e245260820e582535a028a33f358e1b3c1e29 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Thu, 16 Jul 2026 15:31:24 +0000 Subject: [PATCH 16/20] refactor and remove unnecessary code --- .../internal/async/connection_tracing.cc | 124 +++++++----------- .../storage/internal/bucket_metadata_cache.cc | 27 +++- .../storage/internal/bucket_metadata_cache.h | 18 +++ ...lenty_clients_serially_integration_test.cc | 1 + ...clients_simultaneously_integration_test.cc | 18 +-- 5 files changed, 90 insertions(+), 98 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 07d78fde87586..38fb1cd71af3b 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -22,7 +22,6 @@ #include "google/cloud/storage/options.h" #include "google/cloud/internal/opentelemetry.h" #include "google/cloud/version.h" -#include "absl/strings/match.h" #include "google/storage/v2/storage.pb.h" #include #include @@ -39,14 +38,6 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN namespace { -std::string NormalizeBucketName(std::string const& bucket) { - auto const prefix = std::string("projects/_/buckets/"); - if (absl::StartsWith(bucket, prefix)) { - return bucket.substr(prefix.size()); - } - return bucket; -} - class AsyncConnectionTracing : public storage::AsyncConnection { public: explicit AsyncConnectionTracing( @@ -73,12 +64,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->InsertObject(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = - p.request.write_object_spec().resource().bucket()](auto f) { + span = std::move(span)](auto f) { auto result = f.get(); internal::DetachOTelContext(oc); - MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result)); }); } @@ -90,14 +78,12 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->Open(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = p.read_spec.bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr< std::shared_ptr> { auto result = f.get(); internal::DetachOTelContext(oc); if (!result) { - MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result).status()); } return MakeTracingObjectDescriptorConnection(std::move(span), @@ -111,13 +97,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { EnrichSpan(*span, p.options, p.request.bucket()); internal::OTelScope scope(span); auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = p.request.bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr> { auto reader = f.get(); internal::DetachOTelContext(oc); if (!reader) { - MaybeInvalidate(reader, bucket_name); return internal::EndSpan(*span, std::move(reader).status()); } return MakeTracingReaderConnection(std::move(span), *std::move(reader)); @@ -132,11 +116,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->ReadObjectRange(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = p.request.bucket()](auto f) { + span = std::move(span)](auto f) { auto result = f.get(); internal::DetachOTelContext(oc); - MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result)); }); } @@ -150,14 +132,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartAppendableObjectUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = - p.request.write_object_spec().resource().bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); if (!w) { - MaybeInvalidate(w, bucket_name); return internal::EndSpan(*span, std::move(w).status()); } return MakeTracingWriterConnection(span, *std::move(w)); @@ -173,14 +152,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->ResumeAppendableObjectUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = - p.request.write_object_spec().resource().bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); if (!w) { - MaybeInvalidate(w, bucket_name); return internal::EndSpan(*span, std::move(w).status()); } return MakeTracingWriterConnection(span, *std::move(w)); @@ -196,14 +172,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartUnbufferedUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = - p.request.write_object_spec().resource().bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); if (!w) { - MaybeInvalidate(w, bucket_name); return internal::EndSpan(*span, std::move(w).status()); } return MakeTracingWriterConnection(span, *std::move(w)); @@ -219,14 +192,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->StartBufferedUpload(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = - p.request.write_object_spec().resource().bucket()](auto f) + span = std::move(span)](auto f) -> StatusOr> { auto w = f.get(); internal::DetachOTelContext(oc); if (!w) { - MaybeInvalidate(w, bucket_name); return internal::EndSpan(*span, std::move(w).status()); } return MakeTracingWriterConnection(span, *std::move(w)); @@ -272,11 +242,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->ComposeObject(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = p.request.destination().bucket()](auto f) { + span = std::move(span)](auto f) { auto result = f.get(); internal::DetachOTelContext(oc); - MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result)); }); } @@ -287,11 +255,9 @@ class AsyncConnectionTracing : public storage::AsyncConnection { internal::OTelScope scope(span); return impl_->DeleteObject(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name = p.request.bucket()](auto f) { + span = std::move(span)](auto f) { auto result = f.get(); internal::DetachOTelContext(oc); - MaybeInvalidate(result, bucket_name); return internal::EndSpan(*span, std::move(result)); }); } @@ -308,25 +274,18 @@ class AsyncConnectionTracing : public storage::AsyncConnection { auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket"); internal::OTelScope scope(span); auto const bucket_name = p.request.name(); + auto const options = p.options; return impl_->GetBucket(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), - span = std::move(span), this, - bucket_name](future> f) + span = std::move(span), this, bucket_name, + options](future> f) -> StatusOr { auto result = f.get(); internal::DetachOTelContext(oc); if (result.ok()) { - std::string loc = result->location(); - if (result->location_type() == "multi-region" || - result->location_type() == "dual-region") { - loc = "global"; - } - BucketCacheEntry entry{result->project() + "/buckets/" + - NormalizeBucketName(bucket_name), - std::move(loc)}; - cache().Put(bucket_name, std::move(entry)); + EnrichSpan(*span, options, *result, bucket_name); } else { - MaybeInvalidate(result, bucket_name); + cache().MaybeInvalidate(result, bucket_name); } return internal::EndSpan(*span, std::move(result)); }); @@ -335,18 +294,6 @@ class AsyncConnectionTracing : public storage::AsyncConnection { private: BucketMetadataCache& cache() const { return *cache_; } - void MaybeInvalidate(Status const& status, std::string const& bucket_name) { - if (!status.ok() && status.code() == StatusCode::kNotFound) { - cache().Invalidate(bucket_name); - } - } - - template - void MaybeInvalidate(StatusOr const& result, - std::string const& bucket_name) { - MaybeInvalidate(result.status(), bucket_name); - } - void MaybeTriggerBackgroundFetch(Options const& options, std::string const& bucket_name) { if (!cache().StartFetch(bucket_name)) { @@ -355,7 +302,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection { auto guard = ScopedFetch(cache_, bucket_name); google::storage::v2::GetBucketRequest request; - auto const normalized_bucket_name = NormalizeBucketName(bucket_name); + auto const normalized_bucket_name = + BucketMetadataCache::NormalizeBucketName(bucket_name); request.set_name("projects/_/buckets/" + normalized_bucket_name); GetBucketParams params{std::move(request), options}; @@ -364,25 +312,42 @@ class AsyncConnectionTracing : public storage::AsyncConnection { future> f) { auto metadata = f.get(); if (metadata.ok()) { - std::string loc = metadata->location(); - if (metadata->location_type() == "multi-region" || - metadata->location_type() == "dual-region") { - loc = "global"; - } - BucketCacheEntry entry{metadata->project() + "/buckets/" + - NormalizeBucketName(bucket_name), - std::move(loc)}; + BucketCacheEntry entry = BucketCacheEntry::FromLocation( + metadata->project() + "/buckets/" + + BucketMetadataCache::NormalizeBucketName(bucket_name), + metadata->location(), metadata->location_type()); cache->Put(bucket_name, std::move(entry)); } else if (metadata.status().code() == StatusCode::kPermissionDenied) { BucketCacheEntry entry{ - "projects/_/buckets/" + NormalizeBucketName(bucket_name), + "projects/_/buckets/" + + BucketMetadataCache::NormalizeBucketName(bucket_name), "global"}; cache->Put(bucket_name, std::move(entry)); } }); } + void EnrichSpan(opentelemetry::trace::Span& span, + BucketCacheEntry const& entry) { + span.SetAttribute("gcp.resource.destination.id", entry.id); + span.SetAttribute("gcp.resource.destination.location", entry.location); + } + + void EnrichSpan(opentelemetry::trace::Span& span, Options const& options, + google::storage::v2::Bucket const& bucket, + std::string const& bucket_name) { + auto const enabled = options.get< + google::cloud::storage_experimental::OTelSpanEnrichmentOption>(); + if (!enabled) return; + auto entry = BucketCacheEntry::FromLocation( + bucket.project() + "/buckets/" + + BucketMetadataCache::NormalizeBucketName(bucket_name), + bucket.location(), bucket.location_type()); + EnrichSpan(span, entry); + cache().Put(bucket_name, std::move(entry)); + } + void EnrichSpan(opentelemetry::trace::Span& span, Options const& options, std::string const& bucket_name) { if (bucket_name.empty()) return; @@ -391,8 +356,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection { if (!enabled) 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); + EnrichSpan(span, *entry); } else { MaybeTriggerBackgroundFetch(options, bucket_name); } diff --git a/google/cloud/storage/internal/bucket_metadata_cache.cc b/google/cloud/storage/internal/bucket_metadata_cache.cc index 09e74119e8041..583033393967f 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.cc +++ b/google/cloud/storage/internal/bucket_metadata_cache.cc @@ -14,6 +14,7 @@ #include "google/cloud/storage/internal/bucket_metadata_cache.h" #include "google/cloud/storage/bucket_metadata.h" +#include "absl/strings/match.h" #include #include @@ -22,16 +23,28 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +BucketCacheEntry BucketCacheEntry::FromLocation( + std::string id, std::string location, std::string const& location_type) { + if (location_type == "multi-region" || location_type == "dual-region") { + location = "global"; + } + return {std::move(id), std::move(location)}; +} + 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 { + return FromLocation( "projects/" + std::to_string(m.project_number()) + "/buckets/" + m.name(), - std::move(loc)}; + m.location(), m.location_type()); +} + +std::string BucketMetadataCache::NormalizeBucketName( + std::string const& bucket) { + auto const prefix = std::string("projects/_/buckets/"); + if (absl::StartsWith(bucket, prefix)) { + return bucket.substr(prefix.size()); + } + return bucket; } void BucketMetadataCache::MoveToFront(std::list::iterator it) { diff --git a/google/cloud/storage/internal/bucket_metadata_cache.h b/google/cloud/storage/internal/bucket_metadata_cache.h index f3023126b2d28..4292140469ba6 100644 --- a/google/cloud/storage/internal/bucket_metadata_cache.h +++ b/google/cloud/storage/internal/bucket_metadata_cache.h @@ -16,6 +16,8 @@ #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H #include "google/cloud/storage/version.h" +#include "google/cloud/status.h" +#include "google/cloud/status_or.h" #include "absl/types/optional.h" #include #include @@ -40,6 +42,8 @@ struct BucketCacheEntry { std::string id; std::string location; + static BucketCacheEntry FromLocation(std::string id, std::string location, + std::string const& location_type); static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m); }; @@ -48,6 +52,8 @@ class BucketMetadataCache { explicit BucketMetadataCache(std::size_t max_size = 10000) : max_size_(max_size) {} + static std::string NormalizeBucketName(std::string const& bucket); + absl::optional Get(std::string const& bucket_name); void Put(std::string const& bucket_name, BucketCacheEntry entry); void Invalidate(std::string const& bucket_name); @@ -55,6 +61,18 @@ class BucketMetadataCache { bool StartFetch(std::string const& bucket_name); void EndFetch(std::string const& bucket_name); + void MaybeInvalidate(Status const& status, std::string const& bucket_name) { + if (!status.ok() && status.code() == StatusCode::kNotFound) { + Invalidate(bucket_name); + } + } + + template + void MaybeInvalidate(StatusOr const& result, + std::string const& bucket_name) { + MaybeInvalidate(result.status(), bucket_name); + } + private: void MoveToFront(std::list::iterator it); diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index 9a551068bc396..a40f36986c4f6 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -57,6 +57,7 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { auto client = MakeIntegrationTestClient(options); auto object_name = MakeRandomObjectName(); + std::string expected = LoremIpsum(); StatusOr meta = client.InsertObject( diff --git a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc index 26204cc90c5fe..24c54ddb57161 100644 --- a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc @@ -16,7 +16,6 @@ #include "google/cloud/storage/testing/object_integration_test.h" #include "google/cloud/storage/testing/storage_integration_test.h" #include "google/cloud/log.h" -#include "google/cloud/opentelemetry_options.h" #include "google/cloud/status_or.h" #include "google/cloud/testing_util/expect_exception.h" #include "google/cloud/testing_util/status_matchers.h" @@ -46,24 +45,21 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest, // own tests. if (UsingGrpc()) GTEST_SKIP(); - auto options = - Options{}.set(false); auto object_name = MakeRandomObjectName(); + std::string expected = LoremIpsum(); - { - auto client = MakeIntegrationTestClient(options); - StatusOr meta = client.InsertObject( - bucket_name_, object_name, expected, IfGenerationMatch(0)); - ASSERT_STATUS_OK(meta); - ScheduleForDelete(*meta); - } + // Create the object, but only if it does not exist already. + StatusOr meta = client.InsertObject( + bucket_name_, object_name, expected, IfGenerationMatch(0)); + ASSERT_STATUS_OK(meta); + ScheduleForDelete(*meta); auto num_fds_before_test = GetNumOpenFiles(); std::vector read_clients; std::vector read_streams; for (int i = 0; i != 100; ++i) { - auto read_client = MakeIntegrationTestClient(options); + auto read_client = MakeIntegrationTestClient(); auto stream = read_client.ReadObject(bucket_name_, object_name); char c; stream.read(&c, 1); From f05f9bb8cefb733f5f4e934e99e4985d544d1912 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Fri, 17 Jul 2026 02:45:30 +0000 Subject: [PATCH 17/20] fix ci failure --- google/cloud/storage/internal/async/connection_tracing.cc | 4 ++-- .../object_plenty_clients_simultaneously_integration_test.cc | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 38fb1cd71af3b..0ae8410a1211c 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -328,8 +328,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection { }); } - void EnrichSpan(opentelemetry::trace::Span& span, - BucketCacheEntry const& entry) { + static void EnrichSpan(opentelemetry::trace::Span& span, + BucketCacheEntry const& entry) { span.SetAttribute("gcp.resource.destination.id", entry.id); span.SetAttribute("gcp.resource.destination.location", entry.location); } diff --git a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc index 24c54ddb57161..3e489b2e23db0 100644 --- a/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_simultaneously_integration_test.cc @@ -45,6 +45,7 @@ TEST_F(ObjectPlentyClientsSimultaneouslyIntegrationTest, // own tests. if (UsingGrpc()) GTEST_SKIP(); + auto client = MakeIntegrationTestClient(); auto object_name = MakeRandomObjectName(); std::string expected = LoremIpsum(); From c7fa23a5d006a2197c840d2a69853ac8ee2c8b98 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 21 Jul 2026 12:52:10 +0000 Subject: [PATCH 18/20] rebase with main branch --- .../internal/async/connection_tracing.cc | 7 --- .../storage/internal/tracing_connection.cc | 63 +------------------ .../storage/internal/tracing_connection.h | 2 - 3 files changed, 1 insertion(+), 71 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 0ae8410a1211c..af4fa2e940536 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -49,13 +49,6 @@ class AsyncConnectionTracing : public storage::AsyncConnection { Options options() const override { return impl_->options(); } - future> GetBucket( - GetBucketParams p) override { - auto span = internal::MakeSpan("storage::AsyncConnection::GetBucket"); - internal::OTelScope scope(span); - return internal::EndSpan(std::move(span), impl_->GetBucket(std::move(p))); - } - future> InsertObject( InsertObjectParams p) override { auto span = internal::MakeSpan("storage::AsyncConnection::InsertObject"); diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 3df3a0202a19d..1d5419f48f7a8 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -53,25 +53,6 @@ TracingConnection::AsyncRunner const& TracingConnection::runner() { BucketMetadataCache& TracingConnection::cache() const { return *cache_; } -TracingConnection::~TracingConnection() = default; - -TracingConnection::AsyncRunner const& TracingConnection::runner() { - absl::call_once(once_flag_, [this] { - if (!runner_) { - auto threads = - std::make_shared( - 1U); - runner_ = [threads](std::function f) { - threads->cq().RunAsync(std::move(f)); - }; - } - }); - return runner_; -} - -BucketMetadataCache& TracingConnection::cache() const { return *cache_; } - Options TracingConnection::options() const { return impl_->options(); } void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, @@ -89,49 +70,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch( auto guard = ScopedFetch(cache_, bucket_name); auto current_options = google::cloud::internal::SaveCurrentOptions(); runner()([impl = impl_, cache = cache_, bucket_name, current_options, - guard]() { - google::cloud::internal::OptionsSpan span(current_options); - storage::internal::GetBucketMetadataRequest request(bucket_name); - auto result = impl->GetBucketMetadata(request); - - if (result.ok()) { - cache->Put(bucket_name, BucketCacheEntry::FromMetadata(*result)); - } else if (result.status().code() == StatusCode::kPermissionDenied) { - cache->Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"}); - } - }); -} - -void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, - std::string const& bucket_name) { - if (bucket_name.empty()) return; - auto const enabled = - options().get(); - if (!enabled) return; - auto entry = cache().Get(bucket_name); - if (entry.has_value()) { - EnrichSpan(span, *entry); - } else { - MaybeTriggerBackgroundFetch(bucket_name); - } -} - -void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, - BucketCacheEntry const& entry) { - span.SetAttribute("gcp.resource.destination.id", entry.id); - span.SetAttribute("gcp.resource.destination.location", entry.location); -} - -void TracingConnection::MaybeTriggerBackgroundFetch( - std::string const& bucket_name) { - if (!cache().StartFetch(bucket_name)) { - return; - } - - auto guard = ScopedFetch(cache_, bucket_name); - auto current_options = google::cloud::internal::SaveCurrentOptions(); - runner()([impl = impl_, cache = cache_, bucket_name, current_options, - guard]() { + guard = std::move(guard)]() { google::cloud::internal::OptionsSpan span(current_options); storage::internal::GetBucketMetadataRequest request(bucket_name); auto result = impl->GetBucketMetadata(request); diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index 574769438db46..a979f9adea585 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -206,8 +206,6 @@ class TracingConnection : public storage::internal::StorageConnection { AsyncRunner const& runner(); - static BucketMetadataCache& cache(); - std::shared_ptr impl_; std::shared_ptr cache_; absl::once_flag once_flag_; From b9bd172b4657fe6d00f5ae8d0f3fa515fe3412c2 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 21 Jul 2026 14:15:31 +0000 Subject: [PATCH 19/20] fix ci failure --- google/cloud/storage/internal/async/connection_tracing_test.cc | 2 -- google/cloud/storage/internal/tracing_connection.cc | 2 +- .../tests/object_plenty_clients_serially_integration_test.cc | 1 - 3 files changed, 1 insertion(+), 4 deletions(-) diff --git a/google/cloud/storage/internal/async/connection_tracing_test.cc b/google/cloud/storage/internal/async/connection_tracing_test.cc index f77af6ddf549f..3e66ddd8636b4 100644 --- a/google/cloud/storage/internal/async/connection_tracing_test.cc +++ b/google/cloud/storage/internal/async/connection_tracing_test.cc @@ -45,10 +45,8 @@ using ::google::cloud::testing_util::EventNamed; using ::google::cloud::testing_util::InstallSpanCatcher; using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::IsOkAndHolds; -using ::google::cloud::testing_util::OTelAttribute; using ::google::cloud::testing_util::OTelContextCaptured; using ::google::cloud::testing_util::PromiseWithOTelContext; -using ::google::cloud::testing_util::SpanHasAttributes; using ::google::cloud::testing_util::SpanHasEvents; using ::google::cloud::testing_util::SpanHasInstrumentationScope; using ::google::cloud::testing_util::SpanKindIsClient; diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 1d5419f48f7a8..ca37f072453a7 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -70,7 +70,7 @@ void TracingConnection::MaybeTriggerBackgroundFetch( auto guard = ScopedFetch(cache_, bucket_name); auto current_options = google::cloud::internal::SaveCurrentOptions(); runner()([impl = impl_, cache = cache_, bucket_name, current_options, - guard = std::move(guard)]() { + guard]() { google::cloud::internal::OptionsSpan span(current_options); storage::internal::GetBucketMetadataRequest request(bucket_name); auto result = impl->GetBucketMetadata(request); diff --git a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc index a40f36986c4f6..9a551068bc396 100644 --- a/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc +++ b/google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc @@ -57,7 +57,6 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) { auto client = MakeIntegrationTestClient(options); auto object_name = MakeRandomObjectName(); - std::string expected = LoremIpsum(); StatusOr meta = client.InsertObject( From f18bad48d19de96bf0ab5b875ae72a7657148ba7 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 21 Jul 2026 15:20:34 +0000 Subject: [PATCH 20/20] Add tests --- .../internal/async/connection_tracing_test.cc | 205 ++++++++++++++++++ .../storage/internal/async/default_options.cc | 4 +- .../internal/async/default_options_test.cc | 15 ++ 3 files changed, 223 insertions(+), 1 deletion(-) diff --git a/google/cloud/storage/internal/async/connection_tracing_test.cc b/google/cloud/storage/internal/async/connection_tracing_test.cc index 3e66ddd8636b4..f793243f42c05 100644 --- a/google/cloud/storage/internal/async/connection_tracing_test.cc +++ b/google/cloud/storage/internal/async/connection_tracing_test.cc @@ -45,8 +45,10 @@ using ::google::cloud::testing_util::EventNamed; using ::google::cloud::testing_util::InstallSpanCatcher; using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::IsOkAndHolds; +using ::google::cloud::testing_util::OTelAttribute; using ::google::cloud::testing_util::OTelContextCaptured; using ::google::cloud::testing_util::PromiseWithOTelContext; +using ::google::cloud::testing_util::SpanHasAttributes; using ::google::cloud::testing_util::SpanHasEvents; using ::google::cloud::testing_util::SpanHasInstrumentationScope; using ::google::cloud::testing_util::SpanKindIsClient; @@ -763,6 +765,209 @@ TEST(ConnectionTracing, GetBucketError) { SpanHasInstrumentationScope(), SpanKindIsClient()))); } +TEST(ConnectionTracing, GetBucketSpanEnrichment) { + auto span_catcher = InstallSpanCatcher(); + PromiseWithOTelContext> p; + + auto options = + TracingEnabled() + .set( + true); + auto mock = std::make_unique(); + EXPECT_CALL(*mock, options).WillRepeatedly(Return(options)); + EXPECT_CALL(*mock, GetBucket).WillOnce(expect_context(p)); + auto actual = MakeTracingAsyncConnection(std::move(mock)); + + google::storage::v2::GetBucketRequest req; + req.set_name("projects/_/buckets/test-bucket"); + auto result = + actual->GetBucket({std::move(req), options}).then(expect_no_context); + + google::storage::v2::Bucket bucket_meta; + bucket_meta.set_project("projects/123456"); + bucket_meta.set_location("us-east1"); + bucket_meta.set_location_type("regional"); + p.set_value(make_status_or(std::move(bucket_meta))); + ASSERT_STATUS_OK(result.get()); + + auto spans = span_catcher->GetSpans(); + EXPECT_THAT( + spans, + ElementsAre(AllOf( + SpanNamed("storage::AsyncConnection::GetBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); +} + +TEST(ConnectionTracing, BucketMetadataCacheSuccess) { + auto span_catcher = InstallSpanCatcher(); + PromiseWithOTelContext delete_promise_1; + PromiseWithOTelContext> bucket_promise; + PromiseWithOTelContext delete_promise_2; + + auto options = + TracingEnabled() + .set( + true); + + auto mock = std::make_unique(); + EXPECT_CALL(*mock, options).WillRepeatedly(Return(options)); + + // 1st call: DeleteObject on a cache miss triggers background GetBucket + EXPECT_CALL(*mock, DeleteObject) + .WillOnce(expect_context(delete_promise_1)) + .WillOnce(expect_context(delete_promise_2)); + + EXPECT_CALL(*mock, GetBucket).WillOnce([&](auto const&) { + return bucket_promise.get_future(); + }); + + auto actual = MakeTracingAsyncConnection(std::move(mock)); + + google::storage::v2::DeleteObjectRequest req; + req.set_bucket("projects/_/buckets/test-bucket"); + req.set_object("test-object-1"); + + auto res1 = actual->DeleteObject({req, options}).then(expect_no_context); + delete_promise_1.set_value(Status{}); + ASSERT_STATUS_OK(res1.get()); + + // Complete the background GetBucket fetch to populate the cache + google::storage::v2::Bucket bucket_meta; + bucket_meta.set_project("projects/123456"); + bucket_meta.set_location("us-east1"); + bucket_meta.set_location_type("regional"); + bucket_promise.set_value(make_status_or(std::move(bucket_meta))); + + // Clear spans from the first operation and background GetBucket + (void)span_catcher->GetSpans(); + + // 2nd call: DeleteObject hits the cache and span is enriched + req.set_object("test-object-2"); + auto res2 = actual->DeleteObject({req, options}).then(expect_no_context); + delete_promise_2.set_value(Status{}); + ASSERT_STATUS_OK(res2.get()); + + auto spans = span_catcher->GetSpans(); + EXPECT_THAT( + spans, + ElementsAre(AllOf( + SpanNamed("storage::AsyncConnection::DeleteObject"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); +} + +TEST(ConnectionTracing, GetBucketMaybeInvalidateEvicts) { + auto span_catcher = InstallSpanCatcher(); + PromiseWithOTelContext> p1; + PromiseWithOTelContext> p2; + + auto options = + TracingEnabled() + .set( + true); + auto mock = std::make_unique(); + EXPECT_CALL(*mock, options).WillRepeatedly(Return(options)); + + EXPECT_CALL(*mock, GetBucket) + .WillOnce(expect_context(p1)) + .WillOnce(expect_context(p2)); + + auto actual = MakeTracingAsyncConnection(std::move(mock)); + + // 1st call populates the cache + google::storage::v2::GetBucketRequest req; + req.set_name("projects/_/buckets/test-bucket"); + auto res1 = actual->GetBucket({req, options}).then(expect_no_context); + google::storage::v2::Bucket bucket_meta; + bucket_meta.set_project("projects/123456"); + bucket_meta.set_location("us-east1"); + bucket_meta.set_location_type("regional"); + p1.set_value(make_status_or(std::move(bucket_meta))); + ASSERT_STATUS_OK(res1.get()); + + // 2nd call returns kNotFound and evicts the cached metadata + auto res2 = actual->GetBucket({req, options}).then(expect_no_context); + p2.set_value(StatusOr( + Status(StatusCode::kNotFound, "not found"))); + EXPECT_THAT(res2.get(), StatusIs(StatusCode::kNotFound)); +} + +TEST(ConnectionTracing, DeleteObjectNoEvictOnError) { + auto span_catcher = InstallSpanCatcher(); + PromiseWithOTelContext> bucket_p; + PromiseWithOTelContext delete_p1; + PromiseWithOTelContext delete_p2; + + auto options = + TracingEnabled() + .set( + true); + auto mock = std::make_unique(); + EXPECT_CALL(*mock, options).WillRepeatedly(Return(options)); + + EXPECT_CALL(*mock, GetBucket).WillOnce(expect_context(bucket_p)); + EXPECT_CALL(*mock, DeleteObject) + .WillOnce(expect_context(delete_p1)) + .WillOnce(expect_context(delete_p2)); + + auto actual = MakeTracingAsyncConnection(std::move(mock)); + + // 1st call: GetBucket populates the cache + google::storage::v2::GetBucketRequest bucket_req; + bucket_req.set_name("projects/_/buckets/test-bucket"); + auto res_bucket = + actual->GetBucket({bucket_req, options}).then(expect_no_context); + google::storage::v2::Bucket bucket_meta; + bucket_meta.set_project("projects/123456"); + bucket_meta.set_location("us-east1"); + bucket_meta.set_location_type("regional"); + bucket_p.set_value(make_status_or(std::move(bucket_meta))); + ASSERT_STATUS_OK(res_bucket.get()); + + // 2nd call: DeleteObject fails with kNotFound (object missing) + google::storage::v2::DeleteObjectRequest del_req; + del_req.set_bucket("projects/_/buckets/test-bucket"); + del_req.set_object("missing-object"); + auto res_del1 = + actual->DeleteObject({del_req, options}).then(expect_no_context); + delete_p1.set_value(Status(StatusCode::kNotFound, "not found")); + EXPECT_THAT(res_del1.get(), StatusIs(StatusCode::kNotFound)); + + // Clear spans to verify next span enrichment + (void)span_catcher->GetSpans(); + + // 3rd call: DeleteObject proves bucket metadata was NOT evicted by object + // error + del_req.set_object("another-object"); + auto res_del2 = + actual->DeleteObject({del_req, options}).then(expect_no_context); + delete_p2.set_value(Status{}); + ASSERT_STATUS_OK(res_del2.get()); + + auto spans = span_catcher->GetSpans(); + EXPECT_THAT( + spans, + ElementsAre(AllOf( + SpanNamed("storage::AsyncConnection::DeleteObject"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_internal diff --git a/google/cloud/storage/internal/async/default_options.cc b/google/cloud/storage/internal/async/default_options.cc index bee57a17cfe06..c367c6d0a6a02 100644 --- a/google/cloud/storage/internal/async/default_options.cc +++ b/google/cloud/storage/internal/async/default_options.cc @@ -19,6 +19,7 @@ #include "google/cloud/storage/async/retry_policy.h" #include "google/cloud/storage/async/writer_connection.h" #include "google/cloud/storage/internal/grpc/default_options.h" +#include "google/cloud/storage/options.h" #include namespace google { @@ -76,7 +77,8 @@ Options DefaultOptionsAsync(Options opts) { storage::MakeStrictAsyncIdempotencyPolicy) .set(true) .set(128 * 1024 * 1024L) - .set(true)); + .set(true) + .set(true)); return Adjust(DefaultOptionsGrpc(std::move(opts))); } diff --git a/google/cloud/storage/internal/async/default_options_test.cc b/google/cloud/storage/internal/async/default_options_test.cc index 7bdc79553a84b..923bc39d1a33f 100644 --- a/google/cloud/storage/internal/async/default_options_test.cc +++ b/google/cloud/storage/internal/async/default_options_test.cc @@ -21,6 +21,7 @@ #include "google/cloud/storage/async/options.h" #include "google/cloud/storage/async/resume_policy.h" #include "google/cloud/storage/async/writer_connection.h" +#include "google/cloud/storage/options.h" #include "google/cloud/common_options.h" #include @@ -109,6 +110,20 @@ TEST(DefaultOptionsAsync, EnableMultiStreamOptimizationOption) { updated_options.get()); } +TEST(DefaultOptionsAsync, OTelSpanEnrichmentOption) { + auto const options = DefaultOptionsAsync({}); + EXPECT_TRUE(options.get< + google::cloud::storage_experimental::OTelSpanEnrichmentOption>()); + + auto const updated_options = DefaultOptionsAsync( + Options{} + .set( + false)); + EXPECT_FALSE( + updated_options.get< + google::cloud::storage_experimental::OTelSpanEnrichmentOption>()); +} + } // namespace GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_internal