Skip to content

Commit c1b5efa

Browse files
kalragaurikalragauri
authored
feat(storage): add internal payload timestamp plumbing for range reads (#16345)
1 parent ee6ae34 commit c1b5efa

6 files changed

Lines changed: 90 additions & 4 deletions

File tree

‎google/cloud/storage/async/object_responses.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include "absl/strings/cord.h"
2323
#include "absl/strings/string_view.h"
2424
#include "google/storage/v2/storage.pb.h"
25+
#include <chrono>
2526
#include <cstdint>
2627
#include <map>
2728
#include <optional>
@@ -125,6 +126,9 @@ class ReadPayload {
125126
storage::HeadersMap headers_;
126127
// The full object checksums (aka hash values), if known.
127128
std::optional<storage::internal::HashValues> object_hash_values_;
129+
std::chrono::steady_clock::time_point t4_{};
130+
std::chrono::steady_clock::time_point t5_{};
131+
std::chrono::steady_clock::time_point t6_{};
128132
};
129133

130134
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

‎google/cloud/storage/internal/async/object_descriptor_impl.cc‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -305,6 +305,17 @@ void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
305305
google::storage::v2::BidiReadObjectRequest request;
306306
request.Swap(&it->stream->next_request);
307307

308+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
309+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
310+
auto t4_stamp = std::chrono::steady_clock::now();
311+
for (auto const& rr : request.read_ranges()) {
312+
auto const l = it->active_ranges.find(rr.read_id());
313+
if (l != it->active_ranges.end()) {
314+
l->second->SetT4(t4_stamp);
315+
}
316+
}
317+
#endif
318+
308319
// Assign CurrentStream to a temporary variable to prevent
309320
// lifetime extension which can cause the lock to be held until the
310321
// end of the block.
@@ -362,10 +373,22 @@ void ObjectDescriptorImpl::OnRead(
362373
// Release the lock while notifying the ranges. The notifications may trigger
363374
// application code, and that code may callback on this class.
364375
lk.unlock();
376+
377+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
378+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
379+
auto t5_stamp = std::chrono::steady_clock::now();
380+
#endif
381+
365382
for (auto& range_data : *response->mutable_object_data_ranges()) {
366383
auto id = range_data.read_range().read_id();
367384
auto const l = copy.find(id);
368385
if (l == copy.end()) continue;
386+
387+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
388+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
389+
l->second->SetT5(t5_stamp);
390+
#endif
391+
369392
// TODO(#15104) - Consider returning if the range is done, and then
370393
// skipping CleanupDoneRanges().
371394
l->second->OnRead(std::move(range_data), is_transcoded, object_size);

‎google/cloud/storage/internal/async/read_payload_impl.h‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include "google/cloud/storage/internal/hash_values.h"
2121
#include "google/cloud/version.h"
2222
#include "absl/strings/cord.h"
23+
#include <chrono>
2324
#include <optional>
2425

2526
namespace google {
@@ -75,6 +76,27 @@ struct ReadPayloadImpl {
7576
storage::ReadPayload new_data) {
7677
payload.impl_.Append(std::move(new_data.impl_));
7778
}
79+
80+
static void SetTimestamps(storage::ReadPayload& payload,
81+
std::chrono::steady_clock::time_point t4,
82+
std::chrono::steady_clock::time_point t5,
83+
std::chrono::steady_clock::time_point t6) {
84+
payload.t4_ = t4;
85+
payload.t5_ = t5;
86+
payload.t6_ = t6;
87+
}
88+
static std::chrono::steady_clock::time_point GetT4(
89+
storage::ReadPayload const& p) {
90+
return p.t4_;
91+
}
92+
static std::chrono::steady_clock::time_point GetT5(
93+
storage::ReadPayload const& p) {
94+
return p.t5_;
95+
}
96+
static std::chrono::steady_clock::time_point GetT6(
97+
storage::ReadPayload const& p) {
98+
return p.t6_;
99+
}
78100
};
79101

80102
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

‎google/cloud/storage/internal/async/read_range.cc‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,12 @@ void ReadRange::OnRead(google::storage::v2::ObjectRangeData data,
121121
GCP_ERROR_INFO());
122122
}
123123
}
124+
125+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
126+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
127+
ReadPayloadImpl::SetTimestamps(p, t4_, t5_, std::chrono::steady_clock::now());
128+
#endif
129+
124130
if (wait_) {
125131
if (!payload_) return Notify(std::move(lk), std::move(p));
126132
GCP_LOG(FATAL) << "broken class invariant, `payload_` set when there is an"

‎google/cloud/storage/internal/async/read_range.h‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include "google/cloud/status.h"
2323
#include "google/cloud/version.h"
2424
#include "google/storage/v2/storage.pb.h"
25+
#include <chrono>
2526
#include <cstdint>
2627
#include <memory>
2728
#include <mutex>
@@ -83,6 +84,20 @@ class ReadRange {
8384
void OnRead(google::storage::v2::ObjectRangeData data,
8485
bool is_transcoded = false,
8586
std::optional<std::int64_t> object_size = std::nullopt);
87+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
88+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
89+
void SetT4(std::chrono::steady_clock::time_point t) {
90+
std::unique_lock<std::mutex> lk(self->mu_);
91+
t4_ = t;
92+
}
93+
void SetT5(std::chrono::steady_clock::time_point t) {
94+
std::unique_lock<std::mutex> lk(self->mu_);
95+
t5_ = t;
96+
}
97+
#else
98+
void SetT4(std::chrono::steady_clock::time_point) {}
99+
void SetT5(std::chrono::steady_clock::time_point) {}
100+
#endif
86101

87102
private:
88103
void Notify(std::unique_lock<std::mutex> lk, storage::ReadPayload p);
@@ -103,6 +118,12 @@ class ReadRange {
103118
std::optional<promise<ReadResponse>> wait_;
104119
std::shared_ptr<storage::internal::HashFunction> hash_function_;
105120
std::unique_ptr<storage::internal::HashValidator> hash_validator_;
121+
122+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
123+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
124+
std::chrono::steady_clock::time_point t4_{};
125+
std::chrono::steady_clock::time_point t5_{};
126+
#endif
106127
};
107128

108129
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

‎google/cloud/storage/internal/async/read_range_test.cc‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
// limitations under the License.
1414

1515
#include "google/cloud/storage/internal/async/read_range.h"
16+
#include "google/cloud/storage/internal/async/read_payload_impl.h"
1617
#include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
1718
#include "google/cloud/storage/internal/hash_function.h"
1819
#include "google/cloud/storage/internal/hash_function_impl.h"
@@ -82,10 +83,19 @@ TEST(ReadRange, BasicLifecycle) {
8283
actual.OnRead(std::move(data));
8384

8485
EXPECT_TRUE(pending.is_ready());
85-
EXPECT_THAT(pending.get(),
86-
VariantWith<ReadPayload>(ResultOf(
87-
"contents", [](ReadPayload const& p) { return p.contents(); },
88-
ElementsAre("0123456789"))));
86+
auto const res0 = pending.get();
87+
EXPECT_THAT(
88+
res0, VariantWith<ReadPayload>(ResultOf(
89+
"contents",
90+
[](ReadPayload const& p) {
91+
#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
92+
defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
93+
EXPECT_TRUE(
94+
ReadPayloadImpl::GetT6(p).time_since_epoch().count() > 0);
95+
#endif
96+
return p.contents();
97+
},
98+
ElementsAre("0123456789"))));
8999
range = google::storage::v2::ReadRange{};
90100
auto constexpr kRange1 = R"pb(
91101
read_id: 7 read_offset: 10010 read_length: 30

0 commit comments

Comments
 (0)