[core][telemetry/11] support histogram metric on worker side

Signed-off-by: can <can@anyscale.com>
Signed-off-by: Cuong Nguyen <can@anyscale.com>

Author: can-anyscale

src/ray/stats/metric.cc

@@ -119,29 +119,27 @@ void Metric::Record(double value, TagsType tags) {
     // construction time to avoid static initialization order issues. This function
     // is thread-safe.
     RegisterOpenTelemetryMetric();
-    if (OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered(name_)) {
-      // Collect tags from both the metric-specific tags and the global tags.
-      absl::flat_hash_map<std::string, std::string> open_telemetry_tags;
-      std::unordered_set<std::string> tag_keys_set;
-      for (const auto &tag_key : tag_keys_) {
-        tag_keys_set.insert(tag_key.name());
-      }
-      // Insert metric-specific tags that match the expected keys.
-      for (const auto &tag : tags) {
-        const std::string &key = tag.first.name();
-        if (tag_keys_set.count(key)) {
-          open_telemetry_tags[key] = tag.second;
-        }
-      }
-      // Add global tags, overwriting any existing tag keys.
-      for (const auto &tag : StatsConfig::instance().GetGlobalTags()) {
-        open_telemetry_tags[tag.first.name()] = tag.second;
+    // Collect tags from both the metric-specific tags and the global tags.
+    absl::flat_hash_map<std::string, std::string> open_telemetry_tags;
+    std::unordered_set<std::string> tag_keys_set;
+    for (const auto &tag_key : tag_keys_) {
+      tag_keys_set.insert(tag_key.name());
+    }
+    // Insert metric-specific tags that match the expected keys.
+    for (const auto &tag : tags) {
+      const std::string &key = tag.first.name();
+      if (tag_keys_set.count(key)) {
+        open_telemetry_tags[key] = tag.second;
       }
-      OpenTelemetryMetricRecorder::GetInstance().SetMetricValue(
-          name_, std::move(open_telemetry_tags), value);
-
-      return;
     }
+    // Add global tags, overwriting any existing tag keys.
+    for (const auto &tag : StatsConfig::instance().GetGlobalTags()) {
+      open_telemetry_tags[tag.first.name()] = tag.second;
+    }
+    OpenTelemetryMetricRecorder::GetInstance().SetMetricValue(
+        name_, std::move(open_telemetry_tags), value);
+
+    return;
   }
 
   absl::MutexLock lock(&registration_mutex_);
@@ -196,11 +194,6 @@ void Gauge::RegisterOpenTelemetryMetric() {
 }
 
 void Gauge::RegisterView() {
-  if (::RayConfig::instance().experimental_enable_open_telemetry_on_core()) {
-    // Register the metric in OpenTelemetry.
-    OpenTelemetryMetricRecorder::GetInstance().RegisterGaugeMetric(name_, description_);
-    return;
-  }
   opencensus::stats::ViewDescriptor view_descriptor =
       opencensus::stats::ViewDescriptor()
           .set_name(name_)
@@ -211,14 +204,11 @@ void Gauge::RegisterView() {
 }
 
 void Histogram::RegisterOpenTelemetryMetric() {
-  // Histogram is not supported in OpenTelemetry.
-  return;
+  OpenTelemetryMetricRecorder::GetInstance().RegisterHistogramMetric(
+      name_, description_, boundaries_);
 }
 
 void Histogram::RegisterView() {
-  if (measure_ == nullptr) {
-    return;
-  }
   opencensus::stats::ViewDescriptor view_descriptor =
       opencensus::stats::ViewDescriptor()
           .set_name(name_)
@@ -235,11 +225,6 @@ void Count::RegisterOpenTelemetryMetric() {
 }
 
 void Count::RegisterView() {
-  if (::RayConfig::instance().experimental_enable_open_telemetry_on_core()) {
-    // Register the metric in OpenTelemetry.
-    OpenTelemetryMetricRecorder::GetInstance().RegisterCounterMetric(name_, description_);
-    return;
-  }
   opencensus::stats::ViewDescriptor view_descriptor =
       opencensus::stats::ViewDescriptor()
           .set_name(name_)
@@ -255,11 +240,6 @@ void Sum::RegisterOpenTelemetryMetric() {
 }
 
 void Sum::RegisterView() {
-  if (::RayConfig::instance().experimental_enable_open_telemetry_on_core()) {
-    // Register the metric in OpenTelemetry.
-    OpenTelemetryMetricRecorder::GetInstance().RegisterSumMetric(name_, description_);
-    return;
-  }
   opencensus::stats::ViewDescriptor view_descriptor =
       opencensus::stats::ViewDescriptor()
           .set_name(name_)

src/ray/stats/tests/metric_with_open_telemetry_test.cc

@@ -105,44 +105,5 @@ TEST_F(MetricTest, TestSumMetric) {
       "legacy_metric_sum_test"));
 }
 
-TEST_F(MetricTest, TestSumMetric) {
-  ASSERT_TRUE(
-      OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered("metric_sum_test"));
-  // We only test that recording is not crashing. The actual value is not checked
-  // because open telemetry does not provide a way to retrieve the value of a counter.
-  // Checking value is performed via e2e tests instead (e.g., in test_metrics_agent.py).
-  STATS_metric_sum_test.Record(200.0, {{"Tag1", "Value1"}, {"Tag2", "Value2"}});
-}
-
-TEST_F(MetricTest, TestSumMetric) {
-  ASSERT_TRUE(
-      OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered("metric_sum_test"));
-  // We only test that recording is not crashing. The actual value is not checked
-  // because open telemetry does not provide a way to retrieve the value of a counter.
-  // Checking value is performed via e2e tests instead (e.g., in test_metrics_agent.py).
-  STATS_metric_sum_test.Record(200.0, {{"Tag1", "Value1"}, {"Tag2", "Value2"}});
-  LegacyMetricSumTest.Record(200.0, {{"Tag1"sv, "Value1"}, {"Tag2"sv, "Value2"}});
-  ASSERT_TRUE(OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered(
-      "legacy_metric_sum_test"));
-}
-
-TEST_F(MetricTest, TestCounterMetric) {
-  ASSERT_TRUE(OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered(
-      "metric_counter_test"));
-  STATS_metric_counter_test.Record(100.0, {{"Tag1", "Value1"}, {"Tag2", "Value2"}});
-  LegacyMetricCounterTest.Record(100.0, {{"Tag1"sv, "Value1"}, {"Tag2"sv, "Value2"}});
-  ASSERT_TRUE(OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered(
-      "legacy_metric_counter_test"));
-}
-
-TEST_F(MetricTest, TestSumMetric) {
-  ASSERT_TRUE(
-      OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered("metric_sum_test"));
-  STATS_metric_sum_test.Record(200.0, {{"Tag1", "Value1"}, {"Tag2", "Value2"}});
-  LegacyMetricSumTest.Record(200.0, {{"Tag1"sv, "Value1"}, {"Tag2"sv, "Value2"}});
-  ASSERT_TRUE(OpenTelemetryMetricRecorder::GetInstance().IsMetricRegistered(
-      "legacy_metric_sum_test"));
-}
-
 }  // namespace telemetry
 }  // namespace ray

src/ray/telemetry/open_telemetry_metric_recorder.cc

@@ -13,11 +13,17 @@
 // limitations under the License.
 #include "ray/telemetry/open_telemetry_metric_recorder.h"
 
+#include <opentelemetry/context/context.h>
 #include <opentelemetry/exporters/otlp/otlp_grpc_metric_exporter.h>
 #include <opentelemetry/metrics/provider.h>
 #include <opentelemetry/nostd/variant.h>
+#include <opentelemetry/sdk/metrics/aggregation/histogram_aggregation.h>
 #include <opentelemetry/sdk/metrics/export/periodic_exporting_metric_reader.h>
 #include <opentelemetry/sdk/metrics/instruments.h>
+#include <opentelemetry/sdk/metrics/view/instrument_selector.h>
+#include <opentelemetry/sdk/metrics/view/meter_selector.h>
+#include <opentelemetry/sdk/metrics/view/view.h>
+#include <opentelemetry/sdk/metrics/view/view_registry.h>
 
 #include <cassert>
 #include <utility>
@@ -165,6 +171,36 @@ void OpenTelemetryMetricRecorder::RegisterSumMetric(const std::string &name,
   registered_instruments_[name] = std::move(instrument);
 }
 
+void OpenTelemetryMetricRecorder::RegisterHistogramMetric(
+    const std::string &name,
+    const std::string &description,
+    const std::vector<double> &buckets) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  if (registered_instruments_.contains(name)) {
+    return;  // Already registered
+  }
+  // Create a histogram instrument with explicit buckets
+  auto aggregation_config =
+      std::make_shared<opentelemetry::sdk::metrics::HistogramAggregationConfig>();
+  aggregation_config->boundaries_ = buckets;
+  auto view = std::make_unique<opentelemetry::sdk::metrics::View>(
+      name,
+      description,
+      "",
+      opentelemetry::sdk::metrics::AggregationType::kHistogram,
+      aggregation_config);
+
+  auto instrument_selector =
+      std::make_unique<opentelemetry::sdk::metrics::InstrumentSelector>(
+          opentelemetry::sdk::metrics::InstrumentType::kHistogram, name, "");
+  auto meter_selector =
+      std::make_unique<opentelemetry::sdk::metrics::MeterSelector>(meter_name_, "", "");
+  meter_provider_->AddView(
+      std::move(instrument_selector), std::move(meter_selector), std::move(view));
+  auto instrument = GetMeter()->CreateDoubleHistogram(name, description, "");
+  registered_instruments_[name] = std::move(instrument);
+}
+
 void OpenTelemetryMetricRecorder::SetMetricValue(
     const std::string &name,
     absl::flat_hash_map<std::string, std::string> &&tags,
@@ -222,6 +258,9 @@ void OpenTelemetryMetricRecorder::SetSynchronousMetricValue(
   } else if (auto *sum = dynamic_cast<opentelemetry::metrics::UpDownCounter<double> *>(
                  sync_instr_ptr->get())) {
     sum->Add(value, std::move(tags));
+  } else if (auto *histogram = dynamic_cast<opentelemetry::metrics::Histogram<double> *>(
+                 sync_instr_ptr->get())) {
+    histogram->Record(value, std::move(tags), opentelemetry::context::Context());
   } else {
     // Unknown or unsupported instrument type
     RAY_CHECK(false) << "Unsupported synchronous instrument type for metric: " << name;

src/ray/telemetry/open_telemetry_metric_recorder.h

@@ -61,6 +61,11 @@ class OpenTelemetryMetricRecorder {
   // Registers a sum metric with the given name and description
   void RegisterSumMetric(const std::string &name, const std::string &description);
 
+  // Registers a histogram metric with the given name, description, and buckets
+  void RegisterHistogramMetric(const std::string &name,
+                               const std::string &description,
+                               const std::vector<double> &buckets);
+
   // Check if a metric with the given name is registered.
   bool IsMetricRegistered(const std::string &name);
 
@@ -110,6 +115,8 @@ class OpenTelemetryMetricRecorder {
   // Flag to indicate if the recorder is shutting down. This is used to make sure that
   // the recorder will only shutdown once.
   std::atomic<bool> is_shutdown_{false};
+  // The name of the meter used for this recorder.
+  const std::string meter_name_ = "ray";
 
   void SetObservableMetricValue(const std::string &name,
                                 absl::flat_hash_map<std::string, std::string> &&tags,
@@ -123,7 +130,7 @@ class OpenTelemetryMetricRecorder {
       const std::string &name, const absl::flat_hash_map<std::string, std::string> &tags);
 
   opentelemetry::nostd::shared_ptr<opentelemetry::metrics::Meter> GetMeter() {
-    return meter_provider_->GetMeter("ray");
+    return meter_provider_->GetMeter(meter_name_);
   }
 
   // Declare the test class as a friend to allow access to private members for testing.

src/ray/telemetry/tests/open_telemetry_metric_recorder_test.cc

@@ -64,5 +64,12 @@ TEST_F(OpenTelemetryMetricRecorderTest, TestSumMetric) {
   ASSERT_TRUE(recorder_.IsMetricRegistered("test_sum"));
 }
 
+TEST_F(OpenTelemetryMetricRecorderTest, TestHistogramMetric) {
+  recorder_.RegisterHistogramMetric(
+      "test_histogram", "Test histogram description", {0.0, 10.0, 20.0, 30.0});
+  // Check that the histogram metric is registered
+  ASSERT_TRUE(recorder_.IsMetricRegistered("test_histogram"));
+}
+
 }  // namespace telemetry
 }  // namespace ray