From 721de25375d6a768b04b40144033d8784d5e1ced Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 26 May 2026 13:24:34 +0000 Subject: [PATCH 1/6] feat(storage): add resource span attributes ACO ( App Centric Observability ) --- .../storage/internal/tracing_connection.cc | 31 ++++- .../storage/internal/tracing_connection.h | 3 + .../internal/tracing_connection_test.cc | 128 ++++++++++++++++++ 3 files changed, 157 insertions(+), 5 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 6d7e20a2d349b..85994dea2ddfa 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -31,6 +31,17 @@ TracingConnection::TracingConnection(std::shared_ptr impl) Options TracingConnection::options() const { return impl_->options(); } +void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, + storage::BucketMetadata const& metadata) { + std::string id = "projects/" + std::to_string(metadata.project_number()) + "/buckets/" + metadata.name(); + std::string location = metadata.location(); + if (metadata.location_type() == "multi-region" || metadata.location_type() == "dual-region") { + location = "global"; + } + span.SetAttribute("gcp.resource.destination.id", id); + span.SetAttribute("gcp.resource.destination.location", location); +} + StatusOr TracingConnection::ListBuckets( storage::internal::ListBucketsRequest const& request) { // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client @@ -43,14 +54,18 @@ StatusOr TracingConnection::CreateBucket( storage::internal::CreateBucketRequest const& request) { auto span = internal::MakeSpan("storage::Client::CreateBucket"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateBucket(request)); + auto result = impl_->CreateBucket(request); + if (result.ok()) EnrichSpan(*span, *result); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetBucketMetadata( storage::internal::GetBucketMetadataRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetBucketMetadata"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetBucketMetadata(request)); + auto result = impl_->GetBucketMetadata(request); + if (result.ok()) EnrichSpan(*span, *result); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::DeleteBucket( @@ -64,14 +79,18 @@ StatusOr TracingConnection::UpdateBucket( storage::internal::UpdateBucketRequest const& request) { auto span = internal::MakeSpan("storage::Client::UpdateBucket"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->UpdateBucket(request)); + auto result = impl_->UpdateBucket(request); + if (result.ok()) EnrichSpan(*span, *result); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::PatchBucket( storage::internal::PatchBucketRequest const& request) { auto span = internal::MakeSpan("storage::Client::PatchBucket"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->PatchBucket(request)); + auto result = impl_->PatchBucket(request); + if (result.ok()) EnrichSpan(*span, *result); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetNativeBucketIamPolicy( @@ -100,7 +119,9 @@ StatusOr TracingConnection::LockBucketRetentionPolicy( storage::internal::LockBucketRetentionPolicyRequest const& request) { auto span = internal::MakeSpan("storage::Client::LockBucketRetentionPolicy"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->LockBucketRetentionPolicy(request)); + auto result = impl_->LockBucketRetentionPolicy(request); + if (result.ok()) EnrichSpan(*span, *result); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::InsertObjectMedia( diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index 45df6239f956d..aab04a6bd8e1a 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -178,6 +178,9 @@ class TracingConnection : public storage::internal::StorageConnection { std::vector InspectStackStructure() const override; private: + void EnrichSpan(opentelemetry::trace::Span& span, + storage::BucketMetadata const& metadata); + std::shared_ptr impl_; }; diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index bda9bc46d0a2d..adbdb253b5dd5 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -41,6 +41,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; @@ -105,6 +106,31 @@ TEST(TracingClientTest, CreateBucket) { "gl-cpp.status_code", code_str))))); } +TEST(TracingClientTest, CreateBucketSuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + EXPECT_CALL(*mock, CreateBucket).WillOnce([](auto const&) { + EXPECT_TRUE(ThereIsAnActiveSpan()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + return metadata; + }); + auto under_test = TracingConnection(mock); + auto actual = under_test.CreateBucket(storage::internal::CreateBucketRequest()); + EXPECT_THAT(actual, IsOk()); + EXPECT_THAT(span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::CreateBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", "us-east1"))))); +} + TEST(TracingClientTest, GetBucketMetadata) { auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); @@ -128,6 +154,32 @@ TEST(TracingClientTest, GetBucketMetadata) { "gl-cpp.status_code", code_str))))); } +TEST(TracingClientTest, GetBucketMetadataSuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + EXPECT_CALL(*mock, GetBucketMetadata).WillOnce([](auto const&) { + EXPECT_TRUE(ThereIsAnActiveSpan()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + return metadata; + }); + auto under_test = TracingConnection(mock); + auto actual = under_test.GetBucketMetadata( + storage::internal::GetBucketMetadataRequest("test-bucket")); + EXPECT_THAT(actual, IsOk()); + EXPECT_THAT(span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::GetBucketMetadata"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", "us-east1"))))); +} + TEST(TracingClientTest, DeleteBucket) { auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); @@ -174,6 +226,31 @@ TEST(TracingClientTest, UpdateBucket) { "gl-cpp.status_code", code_str))))); } +TEST(TracingClientTest, UpdateBucketSuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + EXPECT_CALL(*mock, UpdateBucket).WillOnce([](auto const&) { + EXPECT_TRUE(ThereIsAnActiveSpan()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + return metadata; + }); + auto under_test = TracingConnection(mock); + auto actual = under_test.UpdateBucket(storage::internal::UpdateBucketRequest()); + EXPECT_THAT(actual, IsOk()); + EXPECT_THAT(span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::UpdateBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", "us-east1"))))); +} + TEST(TracingClientTest, PatchBucket) { auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); @@ -196,6 +273,31 @@ TEST(TracingClientTest, PatchBucket) { "gl-cpp.status_code", code_str))))); } +TEST(TracingClientTest, PatchBucketSuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + EXPECT_CALL(*mock, PatchBucket).WillOnce([](auto const&) { + EXPECT_TRUE(ThereIsAnActiveSpan()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + return metadata; + }); + auto under_test = TracingConnection(mock); + auto actual = under_test.PatchBucket(storage::internal::PatchBucketRequest()); + EXPECT_THAT(actual, IsOk()); + EXPECT_THAT(span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::PatchBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", "us-east1"))))); +} + TEST(TracingClientTest, GetNativeBucketIamPolicy) { auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); @@ -290,6 +392,32 @@ TEST(TracingClientTest, LockBucketRetentionPolicy) { "gl-cpp.status_code", code_str))))); } +TEST(TracingClientTest, LockBucketRetentionPolicySuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + EXPECT_CALL(*mock, LockBucketRetentionPolicy).WillOnce([](auto const&) { + EXPECT_TRUE(ThereIsAnActiveSpan()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + return metadata; + }); + auto under_test = TracingConnection(mock); + auto actual = under_test.LockBucketRetentionPolicy( + storage::internal::LockBucketRetentionPolicyRequest()); + EXPECT_THAT(actual, IsOk()); + EXPECT_THAT(span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::LockBucketRetentionPolicy"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", "us-east1"))))); +} + TEST(TracingClientTest, InsertObjectMedia) { auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); From 46d449faeeb984088186815d15946f97f71f4e08 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 27 May 2026 10:22:56 +0000 Subject: [PATCH 2/6] Adding LRU cache and span attributes to other bucket and object operations --- .../storage/internal/tracing_connection.cc | 291 +++++++++++++++--- .../storage/internal/tracing_connection.h | 90 +++++- .../internal/tracing_connection_test.cc | 155 +++++++--- 3 files changed, 445 insertions(+), 91 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 85994dea2ddfa..b29f20156d930 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -16,6 +16,7 @@ #include "google/cloud/storage/internal/tracing_object_read_source.h" #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/internal/opentelemetry.h" +#include #include #include #include @@ -29,17 +30,88 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN TracingConnection::TracingConnection(std::shared_ptr impl) : impl_(std::move(impl)) {} +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, storage::BucketMetadata const& metadata) { - std::string id = "projects/" + std::to_string(metadata.project_number()) + "/buckets/" + metadata.name(); + std::string id = "projects/" + std::to_string(metadata.project_number()) + + "/buckets/" + metadata.name(); std::string location = metadata.location(); - if (metadata.location_type() == "multi-region" || metadata.location_type() == "dual-region") { + if (metadata.location_type() == "multi-region" || + metadata.location_type() == "dual-region") { location = "global"; } span.SetAttribute("gcp.resource.destination.id", id); span.SetAttribute("gcp.resource.destination.location", location); + + // Populate cache since we have metadata! + cache_.Put(metadata.name(), {id, location}); } StatusOr TracingConnection::ListBuckets( @@ -72,7 +144,10 @@ StatusOr TracingConnection::DeleteBucket( storage::internal::DeleteBucketRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteBucket"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteBucket(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteBucket(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::UpdateBucket( @@ -97,14 +172,20 @@ StatusOr TracingConnection::GetNativeBucketIamPolicy( storage::internal::GetBucketIamPolicyRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetNativeBucketIamPolicy"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetNativeBucketIamPolicy(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetNativeBucketIamPolicy(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::SetNativeBucketIamPolicy( storage::internal::SetNativeBucketIamPolicyRequest const& request) { auto span = internal::MakeSpan("storage::Client::SetNativeBucketIamPolicy"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->SetNativeBucketIamPolicy(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->SetNativeBucketIamPolicy(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -112,7 +193,10 @@ TracingConnection::TestBucketIamPermissions( storage::internal::TestBucketIamPermissionsRequest const& request) { auto span = internal::MakeSpan("storage::Client::TestBucketIamPermissions"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->TestBucketIamPermissions(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->TestBucketIamPermissions(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::LockBucketRetentionPolicy( @@ -128,21 +212,30 @@ StatusOr TracingConnection::InsertObjectMedia( storage::internal::InsertObjectMediaRequest const& request) { auto span = internal::MakeSpan("storage::Client::InsertObjectMedia"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->InsertObjectMedia(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->InsertObjectMedia(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::CopyObject( storage::internal::CopyObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::CopyObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CopyObject(request)); + EnrichSpan(*span, request.destination_bucket()); + auto result = impl_->CopyObject(request); + MaybeInvalidate(result, request.destination_bucket()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetObjectMetadata( storage::internal::GetObjectMetadataRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetObjectMetadata"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetObjectMetadata(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetObjectMetadata(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr> @@ -150,8 +243,12 @@ TracingConnection::ReadObject( storage::internal::ReadObjectRangeRequest const& request) { auto span = internal::MakeSpan("storage::Client::ReadObject"); auto scope = opentelemetry::trace::Scope(span); + EnrichSpan(*span, request.bucket_name()); auto reader = impl_->ReadObject(request); - if (!reader) return internal::EndSpan(*span, std::move(reader)); + if (!reader) { + MaybeInvalidate(reader, request.bucket_name()); + return internal::EndSpan(*span, std::move(reader)); + } return std::unique_ptr( std::make_unique(std::move(span), *std::move(reader))); @@ -162,42 +259,60 @@ StatusOr TracingConnection::ListObjects( // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client auto span = internal::MakeSpan("storage::Client::ListObjects"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ListObjects(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ListObjects(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::DeleteObject( storage::internal::DeleteObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::UpdateObject( storage::internal::UpdateObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::UpdateObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->UpdateObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UpdateObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::MoveObject( storage::internal::MoveObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::MoveObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->MoveObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->MoveObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::PatchObject( storage::internal::PatchObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::PatchObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->PatchObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->PatchObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::ComposeObject( storage::internal::ComposeObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::ComposeObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ComposeObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ComposeObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -205,14 +320,20 @@ TracingConnection::RewriteObject( storage::internal::RewriteObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::RewriteObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->RewriteObject(request)); + EnrichSpan(*span, request.destination_bucket()); + auto result = impl_->RewriteObject(request); + MaybeInvalidate(result, request.destination_bucket()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::RestoreObject( storage::internal::RestoreObjectRequest const& request) { auto span = internal::MakeSpan("storage::Client::RestoreObject"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->RestoreObject(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->RestoreObject(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -222,7 +343,10 @@ TracingConnection::CreateResumableUpload( auto span = internal::MakeSpan("storage::Client::WriteObject/CreateResumableUpload"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateResumableUpload(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->CreateResumableUpload(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -258,8 +382,10 @@ StatusOr> TracingConnection::UploadFileSimple( auto span = internal::MakeSpan("storage::Client::UploadFile/UploadFileSimple"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan( - *span, impl_->UploadFileSimple(file_name, file_size, request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UploadFileSimple(file_name, file_size, request); + if (!result) MaybeInvalidate(result.status(), request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr> TracingConnection::UploadFileResumable( @@ -268,8 +394,10 @@ StatusOr> TracingConnection::UploadFileResumable( auto span = internal::MakeSpan("storage::Client::UploadFile/UploadFileResumable"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, - impl_->UploadFileResumable(file_name, request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UploadFileResumable(file_name, request); + if (!result) MaybeInvalidate(result.status(), request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } Status TracingConnection::DownloadStreamToFile( @@ -278,8 +406,11 @@ Status TracingConnection::DownloadStreamToFile( auto span = internal::MakeSpan( "storage::Client::DownloadToFile/DownloadStreamToFile"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DownloadStreamToFile( - std::move(stream), file_name, request)); + EnrichSpan(*span, request.bucket_name()); + auto result = + impl_->DownloadStreamToFile(std::move(stream), file_name, request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, result); } StatusOr TracingConnection::ExecuteParallelUploadFile( @@ -300,42 +431,60 @@ TracingConnection::ListBucketAcl( // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client auto span = internal::MakeSpan("storage::Client::ListBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ListBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ListBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::CreateBucketAcl( storage::internal::CreateBucketAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::CreateBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->CreateBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::DeleteBucketAcl( storage::internal::DeleteBucketAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetBucketAcl( storage::internal::GetBucketAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::UpdateBucketAcl( storage::internal::UpdateBucketAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::UpdateBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->UpdateBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UpdateBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::PatchBucketAcl( storage::internal::PatchBucketAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::PatchBucketAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->PatchBucketAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->PatchBucketAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -344,42 +493,60 @@ TracingConnection::ListObjectAcl( // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client auto span = internal::MakeSpan("storage::Client::ListObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ListObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ListObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::CreateObjectAcl( storage::internal::CreateObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::CreateObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->CreateObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::DeleteObjectAcl( storage::internal::DeleteObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetObjectAcl( storage::internal::GetObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::UpdateObjectAcl( storage::internal::UpdateObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::UpdateObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->UpdateObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UpdateObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::PatchObjectAcl( storage::internal::PatchObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::PatchObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->PatchObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->PatchObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -388,7 +555,10 @@ TracingConnection::ListDefaultObjectAcl( // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client auto span = internal::MakeSpan("storage::Client::ListDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ListDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ListDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -396,7 +566,10 @@ TracingConnection::CreateDefaultObjectAcl( storage::internal::CreateDefaultObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::CreateDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->CreateDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -404,14 +577,20 @@ TracingConnection::DeleteDefaultObjectAcl( storage::internal::DeleteDefaultObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetDefaultObjectAcl( storage::internal::GetDefaultObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -419,14 +598,20 @@ TracingConnection::UpdateDefaultObjectAcl( storage::internal::UpdateDefaultObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::UpdateDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->UpdateDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->UpdateDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::PatchDefaultObjectAcl( storage::internal::PatchDefaultObjectAclRequest const& request) { auto span = internal::MakeSpan("storage::Client::PatchDefaultObjectAcl"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->PatchDefaultObjectAcl(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->PatchDefaultObjectAcl(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetServiceAccount( @@ -487,21 +672,30 @@ TracingConnection::ListNotifications( // TODO(#11395) - use a internal::MakeTracedStreamRange in storage::Client auto span = internal::MakeSpan("storage::Client::ListNotifications"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->ListNotifications(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->ListNotifications(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::CreateNotification( storage::internal::CreateNotificationRequest const& request) { auto span = internal::MakeSpan("storage::Client::CreateNotification"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->CreateNotification(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->CreateNotification(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr TracingConnection::GetNotification( storage::internal::GetNotificationRequest const& request) { auto span = internal::MakeSpan("storage::Client::GetNotification"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->GetNotification(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->GetNotification(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } StatusOr @@ -509,7 +703,10 @@ TracingConnection::DeleteNotification( storage::internal::DeleteNotificationRequest const& request) { auto span = internal::MakeSpan("storage::Client::DeleteNotification"); auto scope = opentelemetry::trace::Scope(span); - return internal::EndSpan(*span, impl_->DeleteNotification(request)); + EnrichSpan(*span, request.bucket_name()); + auto result = impl_->DeleteNotification(request); + MaybeInvalidate(result, request.bucket_name()); + return internal::EndSpan(*span, std::move(result)); } std::vector TracingConnection::InspectStackStructure() const { diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index aab04a6bd8e1a..5142178c549c8 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -18,8 +18,14 @@ #include "google/cloud/storage/internal/storage_connection.h" #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/storage/version.h" +#include "absl/types/optional.h" +#include +#include #include +#include #include +#include +#include #include namespace google { @@ -27,10 +33,70 @@ 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: explicit TracingConnection(std::shared_ptr impl); - ~TracingConnection() override = default; + ~TracingConnection() override; Options options() const override; @@ -178,10 +244,32 @@ class TracingConnection : public storage::internal::StorageConnection { std::vector InspectStackStructure() const override; private: + void EnrichSpan(opentelemetry::trace::Span& span, + std::string const& bucket_name); void EnrichSpan(opentelemetry::trace::Span& span, storage::BucketMetadata const& metadata); + void MaybeTriggerBackgroundFetch(std::string const& bucket_name); + void CleanupCompletedTasks(); + + template + void MaybeInvalidate(StatusOr const& result, + std::string const& bucket_name) { + if (!result.ok() && result.status().code() == StatusCode::kNotFound) { + cache_.Invalidate(bucket_name); + } + } + + void MaybeInvalidate(Status const& status, std::string const& bucket_name) { + if (!status.ok() && status.code() == StatusCode::kNotFound) { + cache_.Invalidate(bucket_name); + } + } std::shared_ptr impl_; + BucketMetadataCache cache_; + std::mutex mu_; + std::unordered_set in_flight_fetch_; + std::vector> bg_tasks_; }; std::shared_ptr MakeTracingClient( diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index adbdb253b5dd5..16cf9159cf093 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -34,6 +34,7 @@ using ::google::cloud::storage::testing::MockClient; using ::google::cloud::storage::testing::MockObjectReadSource; using ::google::cloud::storage::testing::canonical_errors::PermanentError; using ::google::cloud::testing_util::InstallSpanCatcher; +using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::OTelAttribute; using ::google::cloud::testing_util::SpanHasAttributes; using ::google::cloud::testing_util::SpanHasInstrumentationScope; @@ -41,7 +42,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; @@ -119,16 +119,20 @@ TEST(TracingClientTest, CreateBucketSuccess) { return metadata; }); auto under_test = TracingConnection(mock); - auto actual = under_test.CreateBucket(storage::internal::CreateBucketRequest()); + auto actual = + under_test.CreateBucket(storage::internal::CreateBucketRequest()); EXPECT_THAT(actual, IsOk()); - EXPECT_THAT(span_catcher->GetSpans(), - ElementsAre(AllOf( - SpanHasInstrumentationScope(), SpanKindIsClient(), - SpanNamed("storage::Client::CreateBucket"), - SpanWithStatus(opentelemetry::trace::StatusCode::kOk), - SpanHasAttributes( - OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), - OTelAttribute("gcp.resource.destination.location", "us-east1"))))); + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::CreateBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); } TEST(TracingClientTest, GetBucketMetadata) { @@ -170,14 +174,69 @@ TEST(TracingClientTest, GetBucketMetadataSuccess) { auto actual = under_test.GetBucketMetadata( storage::internal::GetBucketMetadataRequest("test-bucket")); EXPECT_THAT(actual, IsOk()); - EXPECT_THAT(span_catcher->GetSpans(), - ElementsAre(AllOf( - SpanHasInstrumentationScope(), SpanKindIsClient(), - SpanNamed("storage::Client::GetBucketMetadata"), - SpanWithStatus(opentelemetry::trace::StatusCode::kOk), - SpanHasAttributes( - OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), - OTelAttribute("gcp.resource.destination.location", "us-east1"))))); + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::GetBucketMetadata"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); +} + +TEST(TracingClientTest, BucketMetadataCacheSuccess) { + auto span_catcher = InstallSpanCatcher(); + auto mock = std::make_shared(); + + std::promise bg_fetch_done; + auto bg_fetch_future = bg_fetch_done.get_future(); + + EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) { + return PermanentError(); + }); + + EXPECT_CALL(*mock, GetBucketMetadata) + .WillOnce([&bg_fetch_done](auto const& request) { + EXPECT_EQ("test-bucket", request.bucket_name()); + storage::BucketMetadata metadata; + metadata.set_name("test-bucket"); + metadata.set_project_number(123456); + metadata.set_location("us-east1"); + metadata.set_location_type("regional"); + bg_fetch_done.set_value(); + return metadata; + }); + + auto under_test = TracingConnection(mock); + + (void)under_test.GetObjectMetadata( + storage::internal::GetObjectMetadataRequest("test-bucket", + "test-object")); + + bg_fetch_future.wait_for(std::chrono::seconds(5)); + + EXPECT_CALL(*mock, DeleteObject).WillOnce([](auto const&) { + return PermanentError(); + }); + + // Clear spans from GetObjectMetadata + (void)span_catcher->GetSpans(); + + (void)under_test.DeleteObject( + storage::internal::DeleteObjectRequest("test-bucket", "test-object")); + + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanNamed("storage::Client::DeleteObject"), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); } TEST(TracingClientTest, DeleteBucket) { @@ -239,16 +298,20 @@ TEST(TracingClientTest, UpdateBucketSuccess) { return metadata; }); auto under_test = TracingConnection(mock); - auto actual = under_test.UpdateBucket(storage::internal::UpdateBucketRequest()); + auto actual = + under_test.UpdateBucket(storage::internal::UpdateBucketRequest()); EXPECT_THAT(actual, IsOk()); - EXPECT_THAT(span_catcher->GetSpans(), - ElementsAre(AllOf( - SpanHasInstrumentationScope(), SpanKindIsClient(), - SpanNamed("storage::Client::UpdateBucket"), - SpanWithStatus(opentelemetry::trace::StatusCode::kOk), - SpanHasAttributes( - OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), - OTelAttribute("gcp.resource.destination.location", "us-east1"))))); + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::UpdateBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); } TEST(TracingClientTest, PatchBucket) { @@ -288,14 +351,17 @@ TEST(TracingClientTest, PatchBucketSuccess) { auto under_test = TracingConnection(mock); auto actual = under_test.PatchBucket(storage::internal::PatchBucketRequest()); EXPECT_THAT(actual, IsOk()); - EXPECT_THAT(span_catcher->GetSpans(), - ElementsAre(AllOf( - SpanHasInstrumentationScope(), SpanKindIsClient(), - SpanNamed("storage::Client::PatchBucket"), - SpanWithStatus(opentelemetry::trace::StatusCode::kOk), - SpanHasAttributes( - OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), - OTelAttribute("gcp.resource.destination.location", "us-east1"))))); + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::PatchBucket"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); } TEST(TracingClientTest, GetNativeBucketIamPolicy) { @@ -408,14 +474,17 @@ TEST(TracingClientTest, LockBucketRetentionPolicySuccess) { auto actual = under_test.LockBucketRetentionPolicy( storage::internal::LockBucketRetentionPolicyRequest()); EXPECT_THAT(actual, IsOk()); - EXPECT_THAT(span_catcher->GetSpans(), - ElementsAre(AllOf( - SpanHasInstrumentationScope(), SpanKindIsClient(), - SpanNamed("storage::Client::LockBucketRetentionPolicy"), - SpanWithStatus(opentelemetry::trace::StatusCode::kOk), - SpanHasAttributes( - OTelAttribute("gcp.resource.destination.id", "projects/123456/buckets/test-bucket"), - OTelAttribute("gcp.resource.destination.location", "us-east1"))))); + EXPECT_THAT( + span_catcher->GetSpans(), + ElementsAre(AllOf( + SpanHasInstrumentationScope(), SpanKindIsClient(), + SpanNamed("storage::Client::LockBucketRetentionPolicy"), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk), + SpanHasAttributes( + OTelAttribute("gcp.resource.destination.id", + "projects/123456/buckets/test-bucket"), + OTelAttribute("gcp.resource.destination.location", + "us-east1"))))); } TEST(TracingClientTest, InsertObjectMedia) { From d26f8a6d4680f59f1dcc805e371def2f62182585 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 2 Jun 2026 14:11:22 +0000 Subject: [PATCH 3/6] fix ci failures --- .../storage/internal/tracing_connection.cc | 17 +++++++++++++---- .../cloud/storage/internal/tracing_connection.h | 15 ++++++++++++--- .../storage/internal/tracing_connection_test.cc | 1 + 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index b29f20156d930..c63633494c1cb 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -36,6 +36,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() { @@ -72,11 +81,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_); @@ -89,7 +98,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); @@ -111,7 +120,7 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span, span.SetAttribute("gcp.resource.destination.location", location); // Populate cache since we have metadata! - cache_.Put(metadata.name(), {id, location}); + cache().Put(metadata.name(), {id, location}); } StatusOr TracingConnection::ListBuckets( diff --git a/google/cloud/storage/internal/tracing_connection.h b/google/cloud/storage/internal/tracing_connection.h index 5142178c549c8..947dba8a27474 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -84,6 +84,12 @@ class BucketMetadataCache { } } + void Clear() { + std::lock_guard lock(mu_); + map_.clear(); + list_.clear(); + } + private: std::size_t max_size_; std::mutex mu_; @@ -98,6 +104,8 @@ class TracingConnection : public storage::internal::StorageConnection { explicit TracingConnection(std::shared_ptr impl); ~TracingConnection() override; + static void ResetCacheForTesting(); + Options options() const override; StatusOr ListBuckets( @@ -255,18 +263,19 @@ class TracingConnection : public storage::internal::StorageConnection { void MaybeInvalidate(StatusOr const& result, std::string const& bucket_name) { if (!result.ok() && result.status().code() == StatusCode::kNotFound) { - cache_.Invalidate(bucket_name); + cache().Invalidate(bucket_name); } } void MaybeInvalidate(Status const& status, std::string const& bucket_name) { if (!status.ok() && status.code() == StatusCode::kNotFound) { - cache_.Invalidate(bucket_name); + cache().Invalidate(bucket_name); } } + static BucketMetadataCache& cache(); + std::shared_ptr impl_; - BucketMetadataCache cache_; std::mutex mu_; std::unordered_set in_flight_fetch_; std::vector> bg_tasks_; diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index 16cf9159cf093..8b402f3245306 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -188,6 +188,7 @@ TEST(TracingClientTest, GetBucketMetadataSuccess) { } TEST(TracingClientTest, BucketMetadataCacheSuccess) { + TracingConnection::ResetCacheForTesting(); auto span_catcher = InstallSpanCatcher(); auto mock = std::make_shared(); From bdb839f323c10e6a8c55a8ed3045c52ed9e52eb1 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Tue, 2 Jun 2026 14:24:29 +0000 Subject: [PATCH 4/6] 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 c63633494c1cb..c6bd8e1b1ff58 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -41,9 +41,7 @@ BucketMetadataCache& TracingConnection::cache() { return instance; } -void TracingConnection::ResetCacheForTesting() { - cache().Clear(); -} +void TracingConnection::ResetCacheForTesting() { cache().Clear(); } Options TracingConnection::options() const { return impl_->options(); } From 53127d486174244ec3e818276bac543358f35d33 Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 3 Jun 2026 08:09:39 +0000 Subject: [PATCH 5/6] fix ci failures --- .../storage/internal/tracing_connection.cc | 17 ++-- .../storage/internal/tracing_connection.h | 95 ------------------- .../internal/tracing_connection_test.cc | 5 +- 3 files changed, 15 insertions(+), 102 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index c6bd8e1b1ff58..7c1feb859a0dd 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -17,6 +17,7 @@ #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/internal/opentelemetry.h" #include +#include #include #include #include @@ -59,16 +60,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()) { @@ -86,8 +92,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 947dba8a27474..bd648b1d68577 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -18,9 +18,6 @@ #include "google/cloud/storage/internal/storage_connection.h" #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/storage/version.h" -#include "absl/types/optional.h" -#include -#include #include #include #include @@ -33,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: explicit TracingConnection(std::shared_ptr impl); @@ -252,33 +183,7 @@ class TracingConnection : public storage::internal::StorageConnection { std::vector InspectStackStructure() const override; private: - void EnrichSpan(opentelemetry::trace::Span& span, - std::string const& bucket_name); - void EnrichSpan(opentelemetry::trace::Span& span, - storage::BucketMetadata const& metadata); - void MaybeTriggerBackgroundFetch(std::string const& bucket_name); - void CleanupCompletedTasks(); - - template - void MaybeInvalidate(StatusOr const& result, - std::string const& bucket_name) { - if (!result.ok() && result.status().code() == StatusCode::kNotFound) { - cache().Invalidate(bucket_name); - } - } - - void MaybeInvalidate(Status const& status, std::string const& bucket_name) { - if (!status.ok() && status.code() == StatusCode::kNotFound) { - cache().Invalidate(bucket_name); - } - } - - static BucketMetadataCache& cache(); - std::shared_ptr impl_; - std::mutex mu_; - std::unordered_set in_flight_fetch_; - std::vector> bg_tasks_; }; std::shared_ptr MakeTracingClient( diff --git a/google/cloud/storage/internal/tracing_connection_test.cc b/google/cloud/storage/internal/tracing_connection_test.cc index 8b402f3245306..0bd1c9308e0d2 100644 --- a/google/cloud/storage/internal/tracing_connection_test.cc +++ b/google/cloud/storage/internal/tracing_connection_test.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include namespace google { @@ -33,6 +34,7 @@ namespace { using ::google::cloud::storage::testing::MockClient; using ::google::cloud::storage::testing::MockObjectReadSource; using ::google::cloud::storage::testing::canonical_errors::PermanentError; +using ::google::cloud::storage::testing::canonical_errors::TransientError; using ::google::cloud::testing_util::InstallSpanCatcher; using ::google::cloud::testing_util::IsOk; using ::google::cloud::testing_util::OTelAttribute; @@ -196,7 +198,7 @@ TEST(TracingClientTest, BucketMetadataCacheSuccess) { auto bg_fetch_future = bg_fetch_done.get_future(); EXPECT_CALL(*mock, GetObjectMetadata).WillOnce([](auto const&) { - return PermanentError(); + return TransientError(); }); EXPECT_CALL(*mock, GetBucketMetadata) @@ -218,6 +220,7 @@ TEST(TracingClientTest, BucketMetadataCacheSuccess) { "test-object")); bg_fetch_future.wait_for(std::chrono::seconds(5)); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); EXPECT_CALL(*mock, DeleteObject).WillOnce([](auto const&) { return PermanentError(); From 897afd759dbfc0917e0acb5fa4c13b426be7e70d Mon Sep 17 00:00:00 2001 From: bajajnehaa Date: Wed, 3 Jun 2026 10:32:05 +0000 Subject: [PATCH 6/6] fix ci failures --- .../storage/internal/tracing_connection.cc | 64 +++++----- .../storage/internal/tracing_connection.h | 111 ++++++++++++++++++ 2 files changed, 145 insertions(+), 30 deletions(-) diff --git a/google/cloud/storage/internal/tracing_connection.cc b/google/cloud/storage/internal/tracing_connection.cc index 7c1feb859a0dd..327466cbef263 100644 --- a/google/cloud/storage/internal/tracing_connection.cc +++ b/google/cloud/storage/internal/tracing_connection.cc @@ -16,6 +16,7 @@ #include "google/cloud/storage/internal/tracing_object_read_source.h" #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/internal/opentelemetry.h" +#include "google/cloud/options.h" #include #include #include @@ -64,36 +65,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 bd648b1d68577..4c0ce96b97846 100644 --- a/google/cloud/storage/internal/tracing_connection.h +++ b/google/cloud/storage/internal/tracing_connection.h @@ -18,6 +18,9 @@ #include "google/cloud/storage/internal/storage_connection.h" #include "google/cloud/storage/parallel_upload.h" #include "google/cloud/storage/version.h" +#include "absl/types/optional.h" +#include +#include #include #include #include @@ -30,6 +33,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: explicit TracingConnection(std::shared_ptr impl); @@ -183,7 +268,33 @@ class TracingConnection : public storage::internal::StorageConnection { std::vector InspectStackStructure() const override; private: + void EnrichSpan(opentelemetry::trace::Span& span, + std::string const& bucket_name); + static void EnrichSpan(opentelemetry::trace::Span& span, + storage::BucketMetadata const& metadata); + void MaybeTriggerBackgroundFetch(std::string const& bucket_name); + void CleanupCompletedTasks(); + + template + static void MaybeInvalidate(StatusOr const& result, + std::string const& bucket_name) { + if (!result.ok() && result.status().code() == StatusCode::kNotFound) { + cache().Invalidate(bucket_name); + } + } + + static void MaybeInvalidate(Status const& status, + std::string const& bucket_name) { + if (!status.ok() && status.code() == StatusCode::kNotFound) { + cache().Invalidate(bucket_name); + } + } + + static BucketMetadataCache& cache(); + std::shared_ptr impl_; + std::mutex mu_; + std::vector> bg_tasks_; }; std::shared_ptr MakeTracingClient(