feat(storage): add OpenTelemetry counters for read hedging

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,8 @@
 
 #include "google/cloud/storage/internal/hedged_object_read_source.h"
 #include "google/cloud/internal/make_status.h"
+#include <opentelemetry/metrics/meter.h>
+#include <opentelemetry/metrics/provider.h>
 #include <atomic>
 #include <cstring>
 #include <future>
@@ -24,12 +26,42 @@ 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;
+  opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Meter> 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() {
+  if (hedges_dispatched_) hedges_dispatched_->Add(1);
+}
+
+void HedgedReadMetrics::IncrementHedgeWon() {
+  if (hedge_won_) hedge_won_->Add(1);
+}
+
 namespace {
 
 struct RaceResult {
   StatusOr<ReadSourceResult> result;
   std::unique_ptr<ObjectReadSource> source;
   std::unique_ptr<char[]> buffer;
+  bool is_primary;
 };
 
 struct RaceState {
@@ -39,12 +71,13 @@ struct RaceState {
 
 // Opens a new child and performs its initial read, resolving the race if this
 // attempt finishes first. Losing attempts close their child. Only the primary
-// attempt resolves the race on an open error: a hedge that fails to open must
+// attempt resolves the race on open or read errors: a hedge that fails must
 // not mask a slower, but successful, primary.
 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 +88,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,32 +107,79 @@ 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);
+  if (!result && !resolve_on_open_error) {
+    (*source)->Close();
+    return;
+  }
   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();
   }
 }
 
+void DispatchHedges(std::shared_ptr<RaceState> const& state,
+                    std::future<RaceResult>& future, std::size_t n,
+                    int max_hedges, std::chrono::milliseconds delay,
+                    std::shared_ptr<HedgingThreadPool> const& hedge_pool,
+                    HedgedObjectReadSource::ChildFactory const& child_factory,
+                    HedgedReadMetrics* metrics) {
+  for (int hedges_dispatched = 0; 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
+      // the future instead of busy-spinning if tokens or concurrency slots are
+      // temporarily exhausted.
+      if (delay == std::chrono::milliseconds::zero() &&
+          future.wait_for(std::chrono::milliseconds(10)) !=
+              std::future_status::timeout) {
+        break;
+      }
+      continue;
+    }
+    auto hedge = [state, factory = child_factory, n, pool = hedge_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();
+  }
+}
+
 }  // namespace
 
 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, nullptr) {}
+
+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,48 +210,32 @@ 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_;) {
-    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
-      // the future instead of busy-spinning if tokens or concurrency slots are
-      // temporarily exhausted.
-      if (delay_ == std::chrono::milliseconds::zero()) {
-        if (future.wait_for(std::chrono::milliseconds(10)) !=
-            std::future_status::timeout) {
-          break;
-        }
-      }
-      continue;
-    }
-    auto hedge = [state, factory = child_factory_, n, pool = hedge_pool_] {
-      RunAttempt(state, factory, n, /*resolve_on_open_error=*/false, pool);
-    };
-    if (!hedge_pool_->Enqueue(hedge)) {
-      hedge_pool_->ReleaseHedgeSlot();
-      break;
-    }
-    ++hedges_dispatched;
+  DispatchHedges(state, future, n, max_hedges_, delay_, hedge_pool_,
+                 child_factory_, metrics_.get());
+
+  RaceResult race = future.get();
+  if (metrics_ && !race.is_primary && race.result.ok()) {
+    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,12 @@
 #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_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 +33,37 @@ 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);
+
+  ~HedgedReadMetrics() = default;
+  HedgedReadMetrics(HedgedReadMetrics const&) = delete;
+  HedgedReadMetrics& operator=(HedgedReadMetrics const&) = delete;
+  HedgedReadMetrics(HedgedReadMetrics&&) = default;
+  HedgedReadMetrics& operator=(HedgedReadMetrics&&) = default;
+
+  /// Records that a speculative hedge read attempt was dispatched.
+  void IncrementHedgesDispatched();
+
+  /// Records that a hedged read operation was won by a hedge attempt.
+  void IncrementHedgeWon();
+
+ 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 +96,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 +116,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;