diff --git a/google/cloud/storage/internal/async/connection_tracing.cc b/google/cloud/storage/internal/async/connection_tracing.cc index 469c46468b6e0..4b8b9e1b22859 100644 --- a/google/cloud/storage/internal/async/connection_tracing.cc +++ b/google/cloud/storage/internal/async/connection_tracing.cc @@ -15,6 +15,7 @@ #include "google/cloud/storage/internal/async/connection_tracing.h" #include "google/cloud/storage/async/writer_connection.h" #include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h" +#include "google/cloud/storage/internal/async/options.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" @@ -63,6 +64,11 @@ class AsyncConnectionTracing : public storage::AsyncConnection { OpenParams p) override { auto span = internal::MakeSpan("storage::AsyncConnection::Open"); EnrichSpan(*span, p.options, p.read_spec.bucket()); + if (p.options.has()) { + auto const& ranges = p.options.get(); + span->SetAttribute("gl-cpp.initial-read-ranges.ranges-count", + ranges.size()); + } internal::OTelScope scope(span); return impl_->Open(std::move(p)) .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(), diff --git a/google/cloud/storage/internal/async/connection_tracing_test.cc b/google/cloud/storage/internal/async/connection_tracing_test.cc index c42679cca30bf..632d0e3e443c2 100644 --- a/google/cloud/storage/internal/async/connection_tracing_test.cc +++ b/google/cloud/storage/internal/async/connection_tracing_test.cc @@ -15,6 +15,7 @@ #include "google/cloud/storage/internal/async/connection_tracing.h" #include "google/cloud/storage/async/object_descriptor_connection.h" #include "google/cloud/storage/async/reader_connection.h" +#include "google/cloud/storage/internal/async/options.h" #include "google/cloud/storage/mocks/mock_async_connection.h" #include "google/cloud/storage/mocks/mock_async_object_descriptor_connection.h" #include "google/cloud/storage/mocks/mock_async_reader_connection.h" @@ -647,6 +648,38 @@ TEST(ConnectionTracing, OpenSuccess) { SpanHasInstrumentationScope(), SpanKindIsClient()))); } +TEST(ConnectionTracing, OpenSuccessWithInitialReadRanges) { + auto span_catcher = InstallSpanCatcher(); + PromiseWithOTelContext< + StatusOr>> + p; + auto mock = std::make_unique(); + EXPECT_CALL(*mock, options).WillOnce(Return(TracingEnabled())); + EXPECT_CALL(*mock, Open).WillOnce(expect_context(p)); + + auto actual = MakeTracingAsyncConnection(std::move(mock)); + auto open_params = AsyncConnection::OpenParams{}; + open_params.options.set({{0, 100}, {1000, 200}}); + auto f = actual->Open(std::move(open_params)).then(expect_no_context); + + auto mock_descriptor = + std::make_shared(); + p.set_value(StatusOr>( + std::move(mock_descriptor))); + auto result = f.get(); + ASSERT_STATUS_OK(result); + auto descriptor = *std::move(result); + descriptor.reset(); + + auto spans = span_catcher->GetSpans(); + EXPECT_THAT( + spans, ElementsAre( + AllOf(SpanNamed("storage::AsyncConnection::Open"), + SpanHasAttributes(OTelAttribute( + "gl-cpp.initial-read-ranges.ranges-count", 2)), + SpanWithStatus(opentelemetry::trace::StatusCode::kOk)))); +} + TEST(ConnectionTracing, StartAppendableObjectUploadSuccess) { auto span_catcher = InstallSpanCatcher(); PromiseWithOTelContext< diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.cc b/google/cloud/storage/internal/async/object_descriptor_impl.cc index ecd18be5322ed..5234fc76908aa 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.cc +++ b/google/cloud/storage/internal/async/object_descriptor_impl.cc @@ -40,6 +40,31 @@ namespace cloud { namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN +namespace { + +enum class InitialReadRangesCacheStatus { + kNone, + kMiss, + kHit, + kEvicted, +}; + +absl::string_view CacheStatusToString(InitialReadRangesCacheStatus status) { + switch (status) { + case InitialReadRangesCacheStatus::kHit: + return "HIT"; + case InitialReadRangesCacheStatus::kEvicted: + return "EVICTED"; + case InitialReadRangesCacheStatus::kMiss: + return "MISS"; + case InitialReadRangesCacheStatus::kNone: + return ""; + } + return "INVALID_STATUS"; +} + +} // namespace + ObjectDescriptorImpl::ObjectDescriptorImpl( std::unique_ptr resume_policy, OpenStreamFactory make_stream, @@ -50,6 +75,7 @@ ObjectDescriptorImpl::ObjectDescriptorImpl( make_stream_(std::move(make_stream)), read_object_spec_(std::move(read_object_spec)), options_(std::move(options)), + has_initial_read_ranges_(options_.has()), transport_ok_(std::move(transport_ok)) { stream_manager_ = std::make_unique( []() -> std::shared_ptr { return nullptr; }, // NOLINT @@ -231,7 +257,12 @@ std::unique_ptr ObjectDescriptorImpl::Read( // Check if this range matches a pre-warmed range. auto cache_key = std::make_pair(p.start, p.length); auto cache_it = prewarmed_ranges_.find(cache_key); + auto cache_status = has_initial_read_ranges_ + ? InitialReadRangesCacheStatus::kMiss + : InitialReadRangesCacheStatus::kNone; + if (cache_it != prewarmed_ranges_.end()) { + cache_status = InitialReadRangesCacheStatus::kHit; // Cache hit. Claim the pre-warmed range and return it to the user. auto prewarmed = std::move(cache_it->second); prewarmed_ranges_.erase(cache_it); @@ -247,7 +278,13 @@ std::unique_ptr ObjectDescriptorImpl::Read( return std::unique_ptr( std::make_unique(std::move(prewarmed.range))); } - return MakeTracingObjectDescriptorReader(std::move(prewarmed.range)); + return MakeTracingObjectDescriptorReader(std::move(prewarmed.range), + CacheStatusToString(cache_status)); + } + + // If not hit, check if it was evicted earlier due to pacing. + if (evicted_ranges_.erase(cache_key) != 0) { + cache_status = InitialReadRangesCacheStatus::kEvicted; } if (stream_manager_->Empty()) { @@ -258,7 +295,8 @@ std::unique_ptr ObjectDescriptorImpl::Read( return std::unique_ptr( std::make_unique(std::move(range))); } - return MakeTracingObjectDescriptorReader(std::move(range)); + return MakeTracingObjectDescriptorReader(std::move(range), + CacheStatusToString(cache_status)); } auto it = stream_manager_->GetLeastBusyStream(); @@ -274,8 +312,8 @@ std::unique_ptr ObjectDescriptorImpl::Read( return std::unique_ptr( std::make_unique(std::move(range))); } - - return MakeTracingObjectDescriptorReader(std::move(range)); + return MakeTracingObjectDescriptorReader(std::move(range), + CacheStatusToString(cache_status)); } std::shared_ptr @@ -429,6 +467,11 @@ void ObjectDescriptorImpl::OnRead( max_prewarmed_buffer_size_) { // Evict the range if it exceeds the pacing limit. total_prewarmed_bytes_buffered_ -= unclaimed_it->second.bytes_buffered; + // Cap tombstone set size to prevent unbounded memory growth in long-lived + // descriptors where pre-warmed ranges are evicted but never requested. + if (evicted_ranges_.size() < 1000) { + evicted_ranges_.insert(unclaimed_it->second.cache_it->first); + } prewarmed_ranges_.erase(unclaimed_it->second.cache_it); unclaimed_ranges_.erase(unclaimed_it); diff --git a/google/cloud/storage/internal/async/object_descriptor_impl.h b/google/cloud/storage/internal/async/object_descriptor_impl.h index 0b982b9ca215e..d0cfa98487c05 100644 --- a/google/cloud/storage/internal/async/object_descriptor_impl.h +++ b/google/cloud/storage/internal/async/object_descriptor_impl.h @@ -143,6 +143,8 @@ class ObjectDescriptorImpl // Map of read_id to unclaimed range state (bytes buffered and original key). std::unordered_map unclaimed_ranges_; + // Set capturing tombstones for evicted pre-warmed ranges. + std::set> evicted_ranges_; // Total bytes currently buffered across all unclaimed pre-warmed ranges. std::size_t total_prewarmed_bytes_buffered_ = 0; // Maximum bytes allowed to be buffered across all unclaimed pre-warmed ranges @@ -157,6 +159,7 @@ class ObjectDescriptorImpl google::cloud::StatusOr> pending_stream_; bool cancelled_ = false; + bool has_initial_read_ranges_ = false; std::function transport_ok_; }; diff --git a/google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc b/google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc index 8b95c508c3360..99ce3f4cbb1ec 100644 --- a/google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc +++ b/google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc @@ -31,13 +31,18 @@ namespace sc = ::opentelemetry::semconv; class ObjectDescriptorReaderTracing : public ObjectDescriptorReader { public: - explicit ObjectDescriptorReaderTracing(std::shared_ptr impl) - : ObjectDescriptorReader(std::move(impl)) {} + explicit ObjectDescriptorReaderTracing(std::shared_ptr impl, + absl::string_view cache_status) + : ObjectDescriptorReader(std::move(impl)), cache_status_(cache_status) {} ~ObjectDescriptorReaderTracing() override = default; future Read() override { auto span = internal::MakeSpan("storage::AsyncConnection::ReadRange"); + if (!cache_status_.empty()) { + span->SetAttribute("gl-cpp.initial-read-ranges.cache-status", + std::string(cache_status_)); + } internal::OTelScope scope(span); return ObjectDescriptorReader::Read().then( [span = std::move(span), @@ -64,13 +69,18 @@ class ObjectDescriptorReaderTracing : public ObjectDescriptorReader { return result; }); } + + private: + absl::string_view cache_status_; }; } // namespace std::unique_ptr -MakeTracingObjectDescriptorReader(std::shared_ptr impl) { - return std::make_unique(std::move(impl)); +MakeTracingObjectDescriptorReader(std::shared_ptr impl, + absl::string_view cache_status) { + return std::make_unique(std::move(impl), + cache_status); } GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END diff --git a/google/cloud/storage/internal/async/object_descriptor_reader_tracing.h b/google/cloud/storage/internal/async/object_descriptor_reader_tracing.h index 1a010cdbb366b..41fe54547582a 100644 --- a/google/cloud/storage/internal/async/object_descriptor_reader_tracing.h +++ b/google/cloud/storage/internal/async/object_descriptor_reader_tracing.h @@ -25,7 +25,8 @@ namespace storage_internal { GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN std::unique_ptr -MakeTracingObjectDescriptorReader(std::shared_ptr impl); +MakeTracingObjectDescriptorReader(std::shared_ptr impl, + absl::string_view cache_status); GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END } // namespace storage_internal diff --git a/google/cloud/storage/internal/async/object_descriptor_reader_tracing_test.cc b/google/cloud/storage/internal/async/object_descriptor_reader_tracing_test.cc index 98dc46eb9045a..8115edaf7d402 100644 --- a/google/cloud/storage/internal/async/object_descriptor_reader_tracing_test.cc +++ b/google/cloud/storage/internal/async/object_descriptor_reader_tracing_test.cc @@ -45,7 +45,7 @@ TEST(ObjectDescriptorReaderTracing, Read) { auto span_catcher = InstallSpanCatcher(); auto impl = std::make_shared(10000, 30); - auto reader = MakeTracingObjectDescriptorReader(impl); + auto reader = MakeTracingObjectDescriptorReader(impl, "TEST"); auto data = google::storage::v2::ObjectRangeData{}; auto constexpr kData0 = R"pb( @@ -61,6 +61,8 @@ TEST(ObjectDescriptorReaderTracing, Read) { EXPECT_THAT(spans, ElementsAre(AllOf( SpanNamed("storage::AsyncConnection::ReadRange"), + SpanHasAttributes(OTelAttribute( + "gl-cpp.initial-read-ranges.cache-status", "TEST")), SpanHasEvents(AllOf( EventNamed("gl-cpp.read-range"), SpanEventAttributesAre( @@ -73,23 +75,61 @@ TEST(ObjectDescriptorReaderTracing, Read) { TEST(ObjectDescriptorReaderTracing, ReadError) { auto span_catcher = InstallSpanCatcher(); auto impl = std::make_shared(10000, 30); - auto reader = MakeTracingObjectDescriptorReader(impl); + auto reader = MakeTracingObjectDescriptorReader(impl, "TEST"); impl->OnFinish(PermanentError()); auto actual = reader->Read().get(); auto spans = span_catcher->GetSpans(); - EXPECT_THAT(spans, - ElementsAre(AllOf( - SpanNamed("storage::AsyncConnection::ReadRange"), - SpanHasAttributes(OTelAttribute( - "gl-cpp.status_code", "NOT_FOUND")), - SpanHasEvents(AllOf( - EventNamed("gl-cpp.read-range"), - SpanEventAttributesAre( - OTelAttribute(sc::thread::kThreadId, _), - OTelAttribute("rpc.message.type", - "RECEIVED"))))))); + EXPECT_THAT( + spans, + ElementsAre(AllOf( + SpanNamed("storage::AsyncConnection::ReadRange"), + SpanHasAttributes( + OTelAttribute("gl-cpp.status_code", "NOT_FOUND"), + OTelAttribute( + "gl-cpp.initial-read-ranges.cache-status", "TEST")), + SpanHasEvents( + AllOf(EventNamed("gl-cpp.read-range"), + SpanEventAttributesAre( + OTelAttribute(sc::thread::kThreadId, _), + OTelAttribute("rpc.message.type", + "RECEIVED"))))))); +} + +TEST(ObjectDescriptorReaderTracing, ReadWithoutInitialReadRanges) { + auto span_catcher = InstallSpanCatcher(); + auto impl = std::make_shared(10000, 30); + // Pass empty string for cache_status when initial read ranges were not + // configured. + auto reader = MakeTracingObjectDescriptorReader(impl, ""); + + impl->OnFinish(PermanentError()); + + auto actual = reader->Read().get(); + auto spans = span_catcher->GetSpans(); + ASSERT_EQ(spans.size(), 1); + auto const& attributes = spans[0]->GetAttributes(); + EXPECT_EQ(attributes.find("gl-cpp.initial-read-ranges.cache-status"), + attributes.end()); +} + +TEST(ObjectDescriptorReaderTracing, ReadWithCacheStatuses) { + for (auto const* status : {"HIT", "MISS", "EVICTED"}) { + auto span_catcher = InstallSpanCatcher(); + auto impl = std::make_shared(10000, 30); + auto reader = MakeTracingObjectDescriptorReader(impl, status); + + impl->OnFinish(PermanentError()); + + auto actual = reader->Read().get(); + auto spans = span_catcher->GetSpans(); + EXPECT_THAT(spans, + ElementsAre(AllOf( + SpanNamed("storage::AsyncConnection::ReadRange"), + SpanHasAttributes(OTelAttribute( + "gl-cpp.initial-read-ranges.cache-status", status))))); + } } } // namespace