feat(storage): add telemetry for pre-warmed ranges in ObjectDescriptorImpl

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<ReadRangesOption>()) {
+      auto const& ranges = p.options.get<ReadRangesOption>();
+      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(),

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<std::shared_ptr<storage::ObjectDescriptorConnection>>>
+      p;
+  auto mock = std::make_unique<MockAsyncConnection>();
+  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<ReadRangesOption>({{0, 100}, {1000, 200}});
+  auto f = actual->Open(std::move(open_params)).then(expect_no_context);
+
+  auto mock_descriptor =
+      std::make_shared<MockAsyncObjectDescriptorConnection>();
+  p.set_value(StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>>(
+      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<std::size_t>(
+                           "gl-cpp.initial-read-ranges.ranges-count", 2)),
+                       SpanWithStatus(opentelemetry::trace::StatusCode::kOk))));
+}
+
 TEST(ConnectionTracing, StartAppendableObjectUploadSuccess) {
   auto span_catcher = InstallSpanCatcher();
   PromiseWithOTelContext<

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<storage::ResumePolicy> 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<ReadRangesOption>()),
       transport_ok_(std::move(transport_ok)) {
   stream_manager_ = std::make_unique<StreamManager>(
       []() -> std::shared_ptr<ReadStream> { return nullptr; },  // NOLINT
@@ -231,7 +257,12 @@ std::unique_ptr<storage::AsyncReaderConnection> 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<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
       return std::unique_ptr<storage::AsyncReaderConnection>(
           std::make_unique<ObjectDescriptorReader>(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<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
       return std::unique_ptr<storage::AsyncReaderConnection>(
           std::make_unique<ObjectDescriptorReader>(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<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
     return std::unique_ptr<storage::AsyncReaderConnection>(
         std::make_unique<ObjectDescriptorReader>(std::move(range)));
   }
-
-  return MakeTracingObjectDescriptorReader(std::move(range));
+  return MakeTracingObjectDescriptorReader(std::move(range),
+                                           CacheStatusToString(cache_status));
 }
 
 std::shared_ptr<storage::internal::HashFunction>
@@ -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);
 

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<std::int64_t, UnclaimedRangeState> unclaimed_ranges_;
+  // Set capturing tombstones for evicted pre-warmed ranges.
+  std::set<std::pair<std::int64_t, std::int64_t>> 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<storage_internal::OpenStreamResult>>
       pending_stream_;
   bool cancelled_ = false;
+  bool has_initial_read_ranges_ = false;
   std::function<bool()> transport_ok_;
 };
 

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<ReadRange> impl)
-      : ObjectDescriptorReader(std::move(impl)) {}
+  explicit ObjectDescriptorReaderTracing(std::shared_ptr<ReadRange> impl,
+                                         absl::string_view cache_status)
+      : ObjectDescriptorReader(std::move(impl)), cache_status_(cache_status) {}
 
   ~ObjectDescriptorReaderTracing() override = default;
 
   future<ObjectDescriptorReader::ReadResponse> 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<storage::AsyncReaderConnection>
-MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl) {
-  return std::make_unique<ObjectDescriptorReaderTracing>(std::move(impl));
+MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl,
+                                  absl::string_view cache_status) {
+  return std::make_unique<ObjectDescriptorReaderTracing>(std::move(impl),
+                                                         cache_status);
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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<storage::AsyncReaderConnection>
-MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl);
+MakeTracingObjectDescriptorReader(std::shared_ptr<ReadRange> impl,
+                                  absl::string_view cache_status);
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

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<ReadRange>(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<std::string>(
+                      "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<ReadRange>(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<std::string>(
-                      "gl-cpp.status_code", "NOT_FOUND")),
-                  SpanHasEvents(AllOf(
-                      EventNamed("gl-cpp.read-range"),
-                      SpanEventAttributesAre(
-                          OTelAttribute<std::string>(sc::thread::kThreadId, _),
-                          OTelAttribute<std::string>("rpc.message.type",
-                                                     "RECEIVED")))))));
+  EXPECT_THAT(
+      spans,
+      ElementsAre(AllOf(
+          SpanNamed("storage::AsyncConnection::ReadRange"),
+          SpanHasAttributes(
+              OTelAttribute<std::string>("gl-cpp.status_code", "NOT_FOUND"),
+              OTelAttribute<std::string>(
+                  "gl-cpp.initial-read-ranges.cache-status", "TEST")),
+          SpanHasEvents(
+              AllOf(EventNamed("gl-cpp.read-range"),
+                    SpanEventAttributesAre(
+                        OTelAttribute<std::string>(sc::thread::kThreadId, _),
+                        OTelAttribute<std::string>("rpc.message.type",
+                                                   "RECEIVED")))))));
+}
+
+TEST(ObjectDescriptorReaderTracing, ReadWithoutInitialReadRanges) {
+  auto span_catcher = InstallSpanCatcher();
+  auto impl = std::make_shared<ReadRange>(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<ReadRange>(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<std::string>(
+                        "gl-cpp.initial-read-ranges.cache-status", status)))));
+  }
 }
 
 }  // namespace