feat(storage): add internal payload timestamp plumbing for range reads

google/cloud/storage/async/object_responses.h

@@ -22,6 +22,7 @@
 #include "absl/strings/cord.h"
 #include "absl/strings/string_view.h"
 #include "google/storage/v2/storage.pb.h"
+#include <chrono>
 #include <cstdint>
 #include <map>
 #include <optional>
@@ -125,6 +126,9 @@ class ReadPayload {
   storage::HeadersMap headers_;
   // The full object checksums (aka hash values), if known.
   std::optional<storage::internal::HashValues> object_hash_values_;
+  std::chrono::steady_clock::time_point t4_{};
+  std::chrono::steady_clock::time_point t5_{};
+  std::chrono::steady_clock::time_point t6_{};
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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

@@ -305,6 +305,17 @@ void ObjectDescriptorImpl::Flush(std::unique_lock<std::mutex> lk,
   google::storage::v2::BidiReadObjectRequest request;
   request.Swap(&it->stream->next_request);
 
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+  auto t4_stamp = std::chrono::steady_clock::now();
+  for (auto const& rr : request.read_ranges()) {
+    auto const l = it->active_ranges.find(rr.read_id());
+    if (l != it->active_ranges.end()) {
+      l->second->SetT4(t4_stamp);
+    }
+  }
+#endif
+
   // Assign CurrentStream to a temporary variable to prevent
   // lifetime extension which can cause the lock to be held until the
   // end of the block.
@@ -362,10 +373,22 @@ void ObjectDescriptorImpl::OnRead(
   // Release the lock while notifying the ranges. The notifications may trigger
   // application code, and that code may callback on this class.
   lk.unlock();
+
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+  auto t5_stamp = std::chrono::steady_clock::now();
+#endif
+
   for (auto& range_data : *response->mutable_object_data_ranges()) {
     auto id = range_data.read_range().read_id();
     auto const l = copy.find(id);
     if (l == copy.end()) continue;
+
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+    l->second->SetT5(t5_stamp);
+#endif
+
     // TODO(#15104) - Consider returning if the range is done, and then
     // skipping CleanupDoneRanges().
     l->second->OnRead(std::move(range_data), is_transcoded, object_size);

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

@@ -20,6 +20,7 @@
 #include "google/cloud/storage/internal/hash_values.h"
 #include "google/cloud/version.h"
 #include "absl/strings/cord.h"
+#include <chrono>
 #include <optional>
 
 namespace google {
@@ -75,6 +76,27 @@ struct ReadPayloadImpl {
                      storage::ReadPayload new_data) {
     payload.impl_.Append(std::move(new_data.impl_));
   }
+
+  static void SetTimestamps(storage::ReadPayload& payload,
+                            std::chrono::steady_clock::time_point t4,
+                            std::chrono::steady_clock::time_point t5,
+                            std::chrono::steady_clock::time_point t6) {
+    payload.t4_ = t4;
+    payload.t5_ = t5;
+    payload.t6_ = t6;
+  }
+  static std::chrono::steady_clock::time_point GetT4(
+      storage::ReadPayload const& p) {
+    return p.t4_;
+  }
+  static std::chrono::steady_clock::time_point GetT5(
+      storage::ReadPayload const& p) {
+    return p.t5_;
+  }
+  static std::chrono::steady_clock::time_point GetT6(
+      storage::ReadPayload const& p) {
+    return p.t6_;
+  }
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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

@@ -121,6 +121,12 @@ void ReadRange::OnRead(google::storage::v2::ObjectRangeData data,
           GCP_ERROR_INFO());
     }
   }
+
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+  ReadPayloadImpl::SetTimestamps(p, t4_, t5_, std::chrono::steady_clock::now());
+#endif
+
   if (wait_) {
     if (!payload_) return Notify(std::move(lk), std::move(p));
     GCP_LOG(FATAL) << "broken class invariant, `payload_` set when there is an"

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

@@ -22,6 +22,7 @@
 #include "google/cloud/status.h"
 #include "google/cloud/version.h"
 #include "google/storage/v2/storage.pb.h"
+#include <chrono>
 #include <cstdint>
 #include <memory>
 #include <mutex>
@@ -83,6 +84,14 @@ class ReadRange {
   void OnRead(google::storage::v2::ObjectRangeData data,
               bool is_transcoded = false,
               std::optional<std::int64_t> object_size = std::nullopt);
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+  void SetT4(std::chrono::steady_clock::time_point t) { t4_ = t; }
+  void SetT5(std::chrono::steady_clock::time_point t) { t5_ = t; }
+#else
+  void SetT4(std::chrono::steady_clock::time_point) {}
+  void SetT5(std::chrono::steady_clock::time_point) {}
+#endif
 
  private:
   void Notify(std::unique_lock<std::mutex> lk, storage::ReadPayload p);
@@ -103,6 +112,12 @@ class ReadRange {
   std::optional<promise<ReadResponse>> wait_;
   std::shared_ptr<storage::internal::HashFunction> hash_function_;
   std::unique_ptr<storage::internal::HashValidator> hash_validator_;
+
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+  std::chrono::steady_clock::time_point t4_{};
+  std::chrono::steady_clock::time_point t5_{};
+#endif
 };
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/read_range.h"
+#include "google/cloud/storage/internal/async/read_payload_impl.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
 #include "google/cloud/storage/internal/hash_function.h"
 #include "google/cloud/storage/internal/hash_function_impl.h"
@@ -82,10 +83,19 @@ TEST(ReadRange, BasicLifecycle) {
   actual.OnRead(std::move(data));
 
   EXPECT_TRUE(pending.is_ready());
-  EXPECT_THAT(pending.get(),
-              VariantWith<ReadPayload>(ResultOf(
-                  "contents", [](ReadPayload const& p) { return p.contents(); },
-                  ElementsAre("0123456789"))));
+  auto const res0 = pending.get();
+  EXPECT_THAT(
+      res0, VariantWith<ReadPayload>(ResultOf(
+                "contents",
+                [](ReadPayload const& p) {
+#if defined(GOOGLE_CLOUD_CPP_STORAGE_WITH_OTEL_METRICS) || \
+    defined(GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
+                  EXPECT_TRUE(
+                      ReadPayloadImpl::GetT6(p).time_since_epoch().count() > 0);
+#endif
+                  return p.contents();
+                },
+                ElementsAre("0123456789"))));
   range = google::storage::v2::ReadRange{};
   auto constexpr kRange1 = R"pb(
     read_id: 7 read_offset: 10010 read_length: 30