Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kalragauri wants to merge 3 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e49ecc7
feat(storage): add fast_open_cache_status telemetry for pre-warmed ra…
kalragauri Aug 5, 2026
b6c50a8
Address feedback from Gemini code assistant
kalragauri Aug 5, 2026
74d6e87
Merge branch 'main' into feat/range-evict
kalragauri Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions google/cloud/storage/internal/async/connection_tracing.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -63,6 +64,10 @@ 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("fast_open_ranges", ranges.size());

@bajajneha27 bajajneha27 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Generally, we follow the naming convention with . or with - .
So, maybe something like gl-cpp.open.ranges.size ?

}
internal::OTelScope scope(span);
return impl_->Open(std::move(p))
.then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
Expand Down
19 changes: 15 additions & 4 deletions google/cloud/storage/internal/async/object_descriptor_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,10 @@ 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);
absl::string_view cache_status = "MISS";

@bajajneha27 bajajneha27 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Consider making it an enum.


if (cache_it != prewarmed_ranges_.end()) {
cache_status = "HIT";
// 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);
Expand All @@ -247,7 +250,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),
cache_status);
}

// If not hit, check if it was evicted earlier due to pacing.
if (evicted_ranges_.erase(cache_key) != 0) {
cache_status = "EVICTED";
}

if (stream_manager_->Empty()) {
Expand All @@ -258,7 +267,7 @@ 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), cache_status);
}

auto it = stream_manager_->GetLeastBusyStream();
Expand All @@ -274,8 +283,7 @@ 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), cache_status);
}

std::shared_ptr<storage::internal::HashFunction>
Expand Down Expand Up @@ -429,6 +437,9 @@ 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;
if (evicted_ranges_.size() < 1000) {

@bajajneha27 bajajneha27 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Can you please put a comment explaining this?

evicted_ranges_.insert(unclaimed_it->second.cache_it->first);
}
prewarmed_ranges_.erase(unclaimed_it->second.cache_it);
unclaimed_ranges_.erase(unclaimed_it);

Expand Down
2 changes: 2 additions & 0 deletions google/cloud/storage/internal/async/object_descriptor_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
17 changes: 13 additions & 4 deletions google/cloud/storage/internal/async/object_descriptor_reader_tracing.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,17 @@ 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("fast_open_cache_status", std::string(cache_status_));

@bajajneha27 bajajneha27 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment as above for attribute name.

}
internal::OTelScope scope(span);
return ObjectDescriptorReader::Read().then(
[span = std::move(span),
Expand All @@ -64,13 +68,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);
}
Comment thread
kalragauri marked this conversation as resolved.

GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
Expand Down
3 changes: 2 additions & 1 deletion google/cloud/storage/internal/async/object_descriptor_reader_tracing.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 4 additions & 2 deletions google/cloud/storage/internal/async/object_descriptor_reader_tracing_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,8 @@ TEST(ObjectDescriptorReaderTracing, Read) {
auto span_catcher = InstallSpanCatcher();

auto impl = std::make_shared<ReadRange>(10000, 30);
auto reader = MakeTracingObjectDescriptorReader(impl);
auto reader =
MakeTracingObjectDescriptorReader(impl, /*cache_status=*/"TEST");

auto data = google::storage::v2::ObjectRangeData{};
auto constexpr kData0 = R"pb(
Expand Down Expand Up @@ -73,7 +74,8 @@ TEST(ObjectDescriptorReaderTracing, Read) {
TEST(ObjectDescriptorReaderTracing, ReadError) {
auto span_catcher = InstallSpanCatcher();
auto impl = std::make_shared<ReadRange>(10000, 30);
auto reader = MakeTracingObjectDescriptorReader(impl);
auto reader =

@bajajneha27 bajajneha27 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we test the attribute that we just set?

MakeTracingObjectDescriptorReader(impl, /*cache_status=*/"TEST");

impl->OnFinish(PermanentError());

Expand Down
Loading