feat(storage): add OpenTelemetry counters for read hedging

Add storage.read_hedging.hedges_dispatched and storage.read_hedging.hedge_won counters to track speculative attempts and secondary hedge wins.

Author: ajayky-os

google/cloud/storage/internal/connection_impl.cc

@@ -159,6 +159,7 @@ StorageConnectionImpl::StorageConnectionImpl(
     : stub_(std::move(stub)),
       options_(MergeOptions(std::move(options), stub_->options())) {
   if (options_.get<storage_experimental::EnableReadHedgingOption>()) {
+    hedged_read_metrics_ = std::make_shared<HedgedReadMetrics>();
     // `DefaultOptions()` normally resolves these, but a connection can be
     // built without it, in which case the option is left at 0 ("automatic").
     // A pool sized 0 would accept reads it never runs, hanging the caller.
@@ -448,9 +449,9 @@ StatusOr<std::unique_ptr<ObjectReadSource>> StorageConnectionImpl::ReadObject(
   // `max_buffer` bounds the size of an individual read, which is only known
   // when the application calls `Read()`; the source applies it there.
   return std::unique_ptr<ObjectReadSource>(
-      std::make_unique<HedgedObjectReadSource>(read_pool_, hedge_pool_,
-                                               std::move(retry_source_factory),
-                                               delay, max_hedges, max_buffer));
+      std::make_unique<HedgedObjectReadSource>(
+          read_pool_, hedge_pool_, std::move(retry_source_factory), delay,
+          max_hedges, max_buffer, hedged_read_metrics_));
 }
 
 StatusOr<ListObjectsResponse> StorageConnectionImpl::ListObjects(

google/cloud/storage/internal/connection_impl.h

@@ -32,6 +32,8 @@ namespace cloud {
 namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
+class HedgedReadMetrics;
+
 /**
  * Decorates a `StorageConnection` to retry each operation.
  */
@@ -190,6 +192,7 @@ class StorageConnectionImpl
   Options options_;
   std::shared_ptr<ThreadPool> read_pool_;
   std::shared_ptr<HedgingThreadPool> hedge_pool_;
+  std::shared_ptr<HedgedReadMetrics> hedged_read_metrics_;
   google::cloud::internal::InvocationIdGenerator invocation_id_generator_;
 };
 

google/cloud/storage/internal/hedged_object_read_source.cc

@@ -14,6 +14,7 @@
 
 #include "google/cloud/storage/internal/hedged_object_read_source.h"
 #include "google/cloud/internal/make_status.h"
+#include <opentelemetry/metrics/provider.h>
 #include <atomic>
 #include <cstring>
 #include <future>
@@ -24,12 +25,45 @@ namespace cloud {
 namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
+
+HedgedReadMetrics::HedgedReadMetrics()
+    : HedgedReadMetrics(opentelemetry::metrics::Provider::GetMeterProvider()) {}
+
+HedgedReadMetrics::HedgedReadMetrics(
+    opentelemetry::nostd::shared_ptr<
+        opentelemetry::metrics::MeterProvider> const& provider) {
+  if (!provider) return;
+  auto meter = provider->GetMeter("gl-cpp", version_string());
+  if (!meter) return;
+
+  hedges_dispatched_ = meter->CreateUInt64Counter(
+      "storage.read_hedging.hedges_dispatched",
+      "Total number of speculative hedge read attempts dispatched", "{hedge}");
+  hedge_won_ = meter->CreateUInt64Counter(
+      "storage.read_hedging.hedge_won",
+      "Total number of hedged read operations won by a secondary hedge attempt",
+      "{request}");
+}
+
+void HedgedReadMetrics::IncrementHedgesDispatched() {
+  IncrementHedgesDispatched(1);
+}
+void HedgedReadMetrics::IncrementHedgesDispatched(std::uint64_t count) {
+  if (hedges_dispatched_) hedges_dispatched_->Add(count);
+}
+
+void HedgedReadMetrics::IncrementHedgeWon() { IncrementHedgeWon(1); }
+void HedgedReadMetrics::IncrementHedgeWon(std::uint64_t count) {
+  if (hedge_won_) hedge_won_->Add(count);
+}
+
 namespace {
 
 struct RaceResult {
   StatusOr<ReadSourceResult> result;
   std::unique_ptr<ObjectReadSource> source;
   std::unique_ptr<char[]> buffer;
+  bool is_primary;
 };
 
 struct RaceState {
@@ -44,7 +78,8 @@ struct RaceState {
 void RunAttempt(std::shared_ptr<RaceState> const& state,
                 HedgedObjectReadSource::ChildFactory const& factory,
                 std::size_t n, bool resolve_on_open_error,
-                std::shared_ptr<HedgingThreadPool> release_slot) {
+                std::shared_ptr<HedgingThreadPool> release_slot,
+                bool is_primary) {
   // Releases the acquired hedge concurrency slot upon function exit across
   // all code paths (early return on open/allocation error, race winner, or
   // race loser). For primary attempts, release_slot is nullptr.
@@ -55,13 +90,13 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
     }
   } guard{std::move(release_slot)};
 
-  auto source = factory();
+  StatusOr<std::unique_ptr<ObjectReadSource>> source = factory();
   if (!source) {
     if (!resolve_on_open_error) return;
     bool expected = false;
     if (state->resolved.compare_exchange_strong(expected, true)) {
       state->promise.set_value(
-          RaceResult{std::move(source).status(), nullptr, {}});
+          RaceResult{std::move(source).status(), nullptr, {}, is_primary});
     }
     return;
   }
@@ -74,15 +109,16 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
           google::cloud::internal::ResourceExhaustedError(
               "Out of memory allocating hedge buffer", GCP_ERROR_INFO()),
           nullptr,
-          {}});
+          {},
+          is_primary});
     }
     return;
   }
-  auto result = (*source)->Read(buffer.get(), n);
+  StatusOr<ReadSourceResult> result = (*source)->Read(buffer.get(), n);
   bool expected = false;
   if (state->resolved.compare_exchange_strong(expected, true)) {
-    state->promise.set_value(
-        RaceResult{std::move(result), *std::move(source), std::move(buffer)});
+    state->promise.set_value(RaceResult{
+        std::move(result), *std::move(source), std::move(buffer), is_primary});
   } else {
     (*source)->Close();
   }
@@ -94,12 +130,23 @@ HedgedObjectReadSource::HedgedObjectReadSource(
     std::shared_ptr<ThreadPool> read_pool,
     std::shared_ptr<HedgingThreadPool> hedge_pool, ChildFactory child_factory,
     std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer)
+    : HedgedObjectReadSource(std::move(read_pool), std::move(hedge_pool),
+                             std::move(child_factory), delay, max_hedges,
+                             max_buffer,
+                             std::make_shared<HedgedReadMetrics>()) {}
+
+HedgedObjectReadSource::HedgedObjectReadSource(
+    std::shared_ptr<ThreadPool> read_pool,
+    std::shared_ptr<HedgingThreadPool> hedge_pool, ChildFactory child_factory,
+    std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer,
+    std::shared_ptr<HedgedReadMetrics> metrics)
     : read_pool_(std::move(read_pool)),
       hedge_pool_(std::move(hedge_pool)),
       child_factory_(std::move(child_factory)),
       delay_(delay),
       max_hedges_(max_hedges),
-      max_buffer_(max_buffer) {}
+      max_buffer_(max_buffer),
+      metrics_(std::move(metrics)) {}
 
 bool HedgedObjectReadSource::IsOpen() const {
   if (active_child_) return active_child_->IsOpen();
@@ -130,24 +177,26 @@ StatusOr<ReadSourceResult> HedgedObjectReadSource::Read(char* buf,
   // the tail latency it avoids, so open the stream without hedging and read
   // straight into the caller's buffer.
   if (n > max_buffer_) {
-    auto child = child_factory_();
+    StatusOr<std::unique_ptr<ObjectReadSource>> child = child_factory_();
     if (!child) return std::move(child).status();
     active_child_ = *std::move(child);
     return active_child_->Read(buf, n);
   }
 
   auto state = std::make_shared<RaceState>();
-  auto future = state->promise.get_future();
+  std::future<RaceResult> future = state->promise.get_future();
 
   auto primary = [state, factory = child_factory_, n] {
-    RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr);
+    RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr,
+               /*is_primary=*/true);
   };
   // The primary attempt is scheduled on the dedicated read pool.
   // If the pool is shutting down run the attempt inline, the read must
   // complete either way.
   if (!read_pool_->Enqueue(primary)) primary();
 
-  for (int hedges_dispatched = 0; hedges_dispatched < max_hedges_;) {
+  int hedges_dispatched = 0;
+  for (; hedges_dispatched < max_hedges_;) {
     if (future.wait_for(delay_) != std::future_status::timeout) break;
     if (!hedge_pool_->TryAcquireHedgeToken()) {
       // When delay_ is 0ms (or token acquisition fails), back off briefly on
@@ -162,16 +211,22 @@ StatusOr<ReadSourceResult> HedgedObjectReadSource::Read(char* buf,
       continue;
     }
     auto hedge = [state, factory = child_factory_, n, pool = hedge_pool_] {
-      RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool);
+      RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool,
+                 /*is_primary=*/false);
     };
     if (!hedge_pool_->Enqueue(hedge)) {
       hedge_pool_->ReleaseHedgeSlot();
       break;
     }
     ++hedges_dispatched;
+    if (metrics_) metrics_->IncrementHedgesDispatched();
+  }
+
+  RaceResult race = future.get();
+  if (metrics_ && !race.is_primary) {
+    metrics_->IncrementHedgeWon();
   }
 
-  auto race = future.get();
   active_child_ = std::move(race.source);
   if (race.result.ok() && race.result->bytes_received > 0) {
     std::memcpy(buf, race.buffer.get(), race.result->bytes_received);

google/cloud/storage/internal/hedged_object_read_source.h

@@ -18,7 +18,13 @@
 #include "google/cloud/storage/internal/hedging_thread_pool.h"
 #include "google/cloud/storage/internal/object_read_source.h"
 #include "google/cloud/storage/version.h"
+#include <opentelemetry/metrics/meter.h>
+#include <opentelemetry/metrics/meter_provider.h>
+#include <opentelemetry/metrics/sync_instruments.h>
+#include <opentelemetry/nostd/shared_ptr.h>
+#include <opentelemetry/nostd/unique_ptr.h>
 #include <chrono>
+#include <cstdint>
 #include <functional>
 #include <memory>
 
@@ -28,6 +34,31 @@ namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 
+/**
+ * OpenTelemetry metrics helper for read hedging operations.
+ */
+class HedgedReadMetrics {
+ public:
+  HedgedReadMetrics();
+  explicit HedgedReadMetrics(
+      opentelemetry::nostd::shared_ptr<
+          opentelemetry::metrics::MeterProvider> const& provider);
+
+  void IncrementHedgesDispatched();
+  void IncrementHedgesDispatched(std::uint64_t count);
+
+  void IncrementHedgeWon();
+  void IncrementHedgeWon(std::uint64_t count);
+
+ private:
+  opentelemetry::nostd::unique_ptr<
+      opentelemetry::metrics::Counter<std::uint64_t>>
+      hedges_dispatched_;
+  opentelemetry::nostd::unique_ptr<
+      opentelemetry::metrics::Counter<std::uint64_t>>
+      hedge_won_;
+};
+
 /**
  * Hedge the *open* of an `ObjectReadSource` to reduce tail latency.
  *
@@ -60,6 +91,13 @@ class HedgedObjectReadSource : public ObjectReadSource {
                          std::chrono::milliseconds delay, int max_hedges,
                          std::size_t max_buffer);
 
+  HedgedObjectReadSource(std::shared_ptr<ThreadPool> read_pool,
+                         std::shared_ptr<HedgingThreadPool> hedge_pool,
+                         ChildFactory child_factory,
+                         std::chrono::milliseconds delay, int max_hedges,
+                         std::size_t max_buffer,
+                         std::shared_ptr<HedgedReadMetrics> metrics);
+
   ~HedgedObjectReadSource() override = default;
 
   bool IsOpen() const override;
@@ -73,6 +111,7 @@ class HedgedObjectReadSource : public ObjectReadSource {
   std::chrono::milliseconds delay_;
   int max_hedges_;
   std::size_t max_buffer_;
+  std::shared_ptr<HedgedReadMetrics> metrics_;
 
   std::unique_ptr<ObjectReadSource> active_child_;
   bool is_closed_ = false;