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/BUILD.bazel

@@ -54,6 +54,9 @@ cc_library(
     name = "google_cloud_cpp_common_private",
     srcs = google_cloud_cpp_common_srcs + ["internal/build_info.cc"],
     hdrs = google_cloud_cpp_common_hdrs,
+    defines = [
+        "GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY",
+    ],
     linkopts = select({
         "@platforms//os:windows": [
             "-DEFAULTLIB:bcrypt.lib",

google/cloud/google_cloud_cpp_common.cmake

@@ -196,6 +196,10 @@ if (opentelemetry IN_LIST GOOGLE_CLOUD_CPP_ENABLE)
     if (opentelemetry-cpp_FOUND)
         target_link_libraries(google_cloud_cpp_common
                               PUBLIC opentelemetry-cpp::api)
+        target_compile_definitions(
+            google_cloud_cpp_common
+            PUBLIC # Enable OpenTelemetry features in google-cloud-cpp
+                   GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY)
         set(GOOGLE_CLOUD_CPP_FIND_OPTIONAL_DEPENDENCIES
             "find_dependency(opentelemetry-cpp)")
         set(GOOGLE_CLOUD_CPP_OPENTELEMETRY_API "opentelemetry_api")

google/cloud/storage/google_cloud_cpp_storage.cmake

@@ -498,6 +498,12 @@ if (BUILD_TESTING)
         well_known_headers_test.cc
         well_known_parameters_test.cc)
 
+    if (opentelemetry IN_LIST GOOGLE_CLOUD_CPP_ENABLE)
+        list(APPEND storage_client_unit_tests
+             internal/hedged_object_read_source_otel_test.cc)
+        list(SORT storage_client_unit_tests)
+    endif ()
+
     foreach (fname ${storage_client_unit_tests})
         google_cloud_cpp_add_executable(target "storage" "${fname}")
         target_link_libraries(

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,10 @@
 
 #include "google/cloud/storage/internal/hedged_object_read_source.h"
 #include "google/cloud/internal/make_status.h"
+#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+#include <opentelemetry/metrics/meter.h>
+#include <opentelemetry/metrics/provider.h>
+#endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
 #include <atomic>
 #include <cstring>
 #include <future>
@@ -24,12 +28,44 @@ namespace cloud {
 namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
+
+#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+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);
+}
+#endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+
 namespace {
 
 struct RaceResult {
   StatusOr<ReadSourceResult> result;
   std::unique_ptr<ObjectReadSource> source;
   std::unique_ptr<char[]> buffer;
+  bool is_primary;
 };
 
 struct RaceState {
@@ -44,7 +80,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 +92,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 +111,84 @@ 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();
   }
 }
 
+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;
+    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,
+                             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),
+#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+      metrics_(metrics ? std::move(metrics)
+                       : std::make_shared<HedgedReadMetrics>(nullptr)) {
+}
+#else
+      metrics_(metrics ? std::move(metrics)
+                       : std::make_shared<HedgedReadMetrics>()) {
+}
+#endif
 
 bool HedgedObjectReadSource::IsOpen() const {
   if (active_child_) return active_child_->IsOpen();
@@ -130,48 +219,30 @@ 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_);
+
+  RaceResult race = future.get();
+  if (!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,14 @@
 #include "google/cloud/storage/internal/hedging_thread_pool.h"
 #include "google/cloud/storage/internal/object_read_source.h"
 #include "google/cloud/storage/version.h"
+#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+#include <opentelemetry/metrics/meter_provider.h>
+#include <opentelemetry/metrics/sync_instruments.h>
+#include <opentelemetry/nostd/shared_ptr.h>
+#include <opentelemetry/nostd/unique_ptr.h>
+#endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
 #include <chrono>
+#include <cstdint>
 #include <functional>
 #include <memory>
 
@@ -28,6 +35,56 @@ namespace storage {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace internal {
 
+#ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+/**
+ * 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_;
+};
+#else
+/**
+ * No-op OpenTelemetry metrics fallback for read hedging operations when
+ * OpenTelemetry is disabled.
+ */
+class HedgedReadMetrics {
+ public:
+  HedgedReadMetrics() = default;
+  ~HedgedReadMetrics() = default;
+  HedgedReadMetrics(HedgedReadMetrics const&) = delete;
+  HedgedReadMetrics& operator=(HedgedReadMetrics const&) = delete;
+  HedgedReadMetrics(HedgedReadMetrics&&) = default;
+  HedgedReadMetrics& operator=(HedgedReadMetrics&&) = default;
+
+  void IncrementHedgesDispatched() {}
+  void IncrementHedgeWon() {}
+};
+#endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+
 /**
  * Hedge the *open* of an `ObjectReadSource` to reduce tail latency.
  *
@@ -60,6 +117,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 +137,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;