[Metrics] Stats supports metric exporters (#8941)

BUILD.bazel

@@ -935,6 +935,16 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "metric_exporter_client_test",
+    srcs = ["src/ray/stats/metric_exporter_client_test.cc"],
+    copts = COPTS,
+    deps = [
+        ":stats_lib",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_library(
     name = "gcs_test_util_lib",
     hdrs = [

src/ray/stats/metric.h

@@ -141,6 +141,13 @@ class Sum : public Metric {
 
 };  // class Sum
 
+/// Raw metric view point for exporter.
+struct MetricPoint {
+  std::string metric_name;
+  int64_t timestamp;
+  double value;
+  std::unordered_map<std::string, std::string> tags;
+};
 }  // namespace stats
 
 }  // namespace ray

src/ray/stats/metric_exporter.cc

@@ -0,0 +1,116 @@
+// Copyright 2017 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <future>
+
+#include "ray/stats/metric_exporter.h"
+
+namespace ray {
+namespace stats {
+
+template <>
+void MetricExporter::ExportToPoints(
+    const opencensus::stats::ViewData::DataMap<opencensus::stats::Distribution>
+        &view_data,
+    const std::string &metric_name, std::vector<std::string> &keys,
+    std::vector<MetricPoint> &points) {
+  // Return if no raw data found in view map.
+  if (view_data.size() == 0) {
+    return;
+  }
+  // NOTE(lingxuan.zlx): No sampling in histogram data, so all points all be filled in.
+  std::unordered_map<std::string, std::string> tags;
+  for (size_t i = 0; i < view_data.begin()->first.size(); ++i) {
+    tags[keys[i]] = view_data.begin()->first[i];
+  }
+  // Histogram metric will be append suffix with mean/max/min.
+  double hist_mean = 0.0;
+  double hist_max = 0.0;
+  double hist_min = 0.0;
+  bool in_first_hist_data = true;
+  for (const auto &row : view_data) {
+    if (in_first_hist_data) {
+      hist_mean = static_cast<double>(row.second.mean());
+      hist_max = static_cast<double>(row.second.max());
+      hist_min = static_cast<double>(row.second.min());
+      in_first_hist_data = false;
+    } else {
+      hist_mean += static_cast<double>(row.second.mean());
+      hist_max = std::max(hist_max, static_cast<double>(row.second.max()));
+      hist_min = std::min(hist_min, static_cast<double>(row.second.min()));
+    }
+  }
+  hist_mean /= view_data.size();
+  MetricPoint mean_point{.metric_name = metric_name + ".mean",
+                         .timestamp = current_sys_time_ms(),
+                         .value = hist_mean,
+                         .tags = tags};
+  MetricPoint max_point{.metric_name = metric_name + ".max",
+                        .timestamp = current_sys_time_ms(),
+                        .value = hist_max,
+                        .tags = tags};
+  MetricPoint min_point{.metric_name = metric_name + ".min",
+                        .timestamp = current_sys_time_ms(),
+                        .value = hist_min,
+                        .tags = tags};
+  points.push_back(std::move(mean_point));
+  points.push_back(std::move(max_point));
+  points.push_back(std::move(min_point));
+  RAY_LOG(DEBUG) << "Metric name " << metric_name << ", mean value : " << mean_point.value
+                 << " max value : " << max_point.value
+                 << "  min value : " << min_point.value;
+
+  if (points.size() >= report_batch_size_) {
+    RAY_LOG(DEBUG) << "Point size : " << points.size();
+    metric_exporter_client_->ReportMetrics(points);
+    points.clear();
+  }
+}
+
+void MetricExporter::ExportViewData(
+    const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                opencensus::stats::ViewData>> &data) {
+  std::vector<MetricPoint> points;
+  // NOTE(lingxuan.zlx): There is no sampling in view data, so all raw metric
+  // data will be processed.
+  for (const auto &datum : data) {
+    auto &descriptor = datum.first;
+    auto &view_data = datum.second;
+
+    std::vector<std::string> keys;
+    for (size_t i = 0; i < descriptor.columns().size(); ++i) {
+      keys.push_back(descriptor.columns()[i].name());
+    }
+    auto &metric_name = descriptor.name();
+    switch (view_data.type()) {
+    case opencensus::stats::ViewData::Type::kDouble:
+      ExportToPoints<double>(view_data.double_data(), metric_name, keys, points);
+      break;
+    case opencensus::stats::ViewData::Type::kInt64:
+      ExportToPoints<int64_t>(view_data.int_data(), metric_name, keys, points);
+      break;
+    case opencensus::stats::ViewData::Type::kDistribution:
+      ExportToPoints<opencensus::stats::Distribution>(view_data.distribution_data(),
+                                                      metric_name, keys, points);
+      break;
+    default:
+      RAY_LOG(FATAL) << "Unknown view data type.";
+      break;
+    }
+  }
+  RAY_LOG(DEBUG) << "Point size : " << points.size();
+  metric_exporter_client_->ReportMetrics(points);
+}
+}  // namespace stats
+}  // namespace ray

src/ray/stats/metric_exporter.h

@@ -0,0 +1,88 @@
+// Copyright 2017 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+#include "absl/memory/memory.h"
+#include "opencensus/stats/stats.h"
+#include "opencensus/tags/tag_key.h"
+
+#include "ray/stats/metric.h"
+#include "ray/stats/metric_exporter_client.h"
+#include "ray/util/logging.h"
+#include "ray/util/util.h"
+
+namespace ray {
+namespace stats {
+
+/// Main function of metric exporter is collecting indicator information from
+/// opencensus data view, and sends it to the remote (for example
+/// sends metrics to dashboard agents through RPC). How to use it? Register metrics
+/// exporter after a main thread launched.
+class MetricExporter final : public opencensus::stats::StatsExporter::Handler {
+ public:
+  explicit MetricExporter(std::shared_ptr<MetricExporterClient> metric_exporter_client,
+                          size_t report_batch_size = kDefaultBatchSize)
+      : metric_exporter_client_(metric_exporter_client),
+        report_batch_size_(report_batch_size) {}
+  ~MetricExporter() = default;
+  static void Register(std::shared_ptr<MetricExporterClient> metric_exporter_client,
+                       size_t report_batch_size) {
+    opencensus::stats::StatsExporter::RegisterPushHandler(
+        absl::make_unique<MetricExporter>(metric_exporter_client, report_batch_size));
+  }
+
+  void ExportViewData(
+      const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                  opencensus::stats::ViewData>> &data) override;
+
+ private:
+  template <class DTYPE>
+  /// Extract raw data from view data, then metric exporter clients can use them
+  /// in points schema.
+  /// \param view_data, raw data in map
+  /// \param metric_name, metric name of view data
+  /// \param keys, metric tags map
+  /// \param points, memory metric vector instance
+  void ExportToPoints(const opencensus::stats::ViewData::DataMap<DTYPE> &view_data,
+                      const std::string &metric_name, std::vector<std::string> &keys,
+                      std::vector<MetricPoint> &points) {
+    for (const auto &row : view_data) {
+      std::unordered_map<std::string, std::string> tags;
+      for (size_t i = 0; i < keys.size(); ++i) {
+        tags[keys[i]] = row.first[i];
+      }
+      // Current timestamp is used for point not view data time.
+      MetricPoint point{.metric_name = metric_name,
+                        .timestamp = current_sys_time_ms(),
+                        .value = static_cast<double>(row.second),
+                        .tags = tags};
+      RAY_LOG(DEBUG) << "Metric name " << metric_name << ", value " << point.value;
+      points.push_back(std::move(point));
+      if (points.size() >= report_batch_size_) {
+        RAY_LOG(DEBUG) << "Point size : " << points.size();
+        metric_exporter_client_->ReportMetrics(points);
+        points.clear();
+      }
+    }
+  }
+
+ private:
+  std::shared_ptr<MetricExporterClient> metric_exporter_client_;
+  /// Auto max minbatch size for reporting metrics to external components.
+  static constexpr size_t kDefaultBatchSize = 100;
+  size_t report_batch_size_;
+};
+
+}  // namespace stats
+}  // namespace ray

src/ray/stats/metric_exporter_client.cc

@@ -0,0 +1,36 @@
+// Copyright 2017 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <algorithm>
+
+#include "ray/stats/metric_exporter_client.h"
+
+namespace ray {
+namespace stats {
+
+void StdoutExporterClient::ReportMetrics(const std::vector<MetricPoint> &points) {
+  RAY_LOG(DEBUG) << "Metric point size : " << points.size();
+}
+
+MetricExporterDecorator::MetricExporterDecorator(
+    std::shared_ptr<MetricExporterClient> exporter)
+    : exporter_(exporter) {}
+
+void MetricExporterDecorator::ReportMetrics(const std::vector<MetricPoint> &points) {
+  if (exporter_) {
+    exporter_->ReportMetrics(points);
+  }
+}
+}  // namespace stats
+}  // namespace ray

src/ray/stats/metric_exporter_client.h

@@ -0,0 +1,66 @@
+// Copyright 2017 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "ray/stats/metric.h"
+
+namespace ray {
+namespace stats {
+/// Interface class for abstract metrics exporter client.
+class MetricExporterClient {
+ public:
+  virtual void ReportMetrics(const std::vector<MetricPoint> &points) = 0;
+  virtual ~MetricExporterClient() = default;
+};
+
+/// Default stdout exporter client can log metrics info for debug.
+/// In decorator pattern, a basic concrete class is needed, so we
+/// use stdout as the default concrete class.
+class StdoutExporterClient : public MetricExporterClient {
+ public:
+  void ReportMetrics(const std::vector<MetricPoint> &points) override;
+};
+
+/// The decoration mode is that the user can apply it by configuring different
+/// combinations.
+/// Usage:
+/// std::shared_ptr<MetricExporterClient> exporter(new StdoutExporterClient());
+/// std::shared_ptr<MetricExporterClient> dashboard_exporter_client(
+///         new DashboardExporterCLient(exporter, gcs_rpc_client));
+///  Both dahsboard client and std logging will emit when
+//  dahsboard_exporter_client->ReportMetrics(points) is called.
+/// Actually, opentsdb exporter can be added like above mentioned style.
+class MetricExporterDecorator : public MetricExporterClient {
+ public:
+  MetricExporterDecorator(std::shared_ptr<MetricExporterClient> exporter);
+  virtual void ReportMetrics(const std::vector<MetricPoint> &points);
+
+ private:
+  std::shared_ptr<MetricExporterClient> exporter_;
+};
+
+class OpentsdbExporterClient : public MetricExporterDecorator {
+ public:
+  OpentsdbExporterClient(std::shared_ptr<MetricExporterClient> exporter)
+      : MetricExporterDecorator(exporter) {}
+  void ReportMetrics(const std::vector<MetricPoint> &points) override {
+    MetricExporterDecorator::ReportMetrics(points);
+    // TODO(lingxuan.zlx): opentsdb client is used for report to backend
+    // storage.
+  }
+};
+
+}  // namespace stats
+}  // namespace ray

src/ray/stats/metric_exporter_client_test.cc

@@ -0,0 +1,184 @@
+// Copyright 2017 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include <chrono>
+#include <iostream>
+#include <thread>
+#include <vector>
+
+#include "absl/memory/memory.h"
+#include "ray/stats/metric_exporter.h"
+#include "ray/stats/metric_exporter_client.h"
+#include "ray/stats/stats.h"
+
+namespace ray {
+using namespace stats;
+
+const size_t kMockReportBatchSize = 10;
+
+class MockExporterClient1 : public MetricExporterDecorator {
+ public:
+  MockExporterClient1(std::shared_ptr<MetricExporterClient> exporter)
+      : MetricExporterDecorator(exporter) {
+    client1_count = 0;
+    lastest_hist_min = 0.0;
+    lastest_hist_mean = 0.0;
+    lastest_hist_max = 0.0;
+  }
+
+  void ReportMetrics(const std::vector<MetricPoint> &points) override {
+    MetricExporterDecorator::ReportMetrics(points);
+    client1_count += points.size();
+    client1_value = points.back().value;
+    RAY_LOG(DEBUG) << "Client 1 " << client1_count << " last metric "
+                   << points.back().metric_name << ", value " << points.back().value;
+    RecordLastHistData(points);
+    // Point size must be less than or equal to report batch size.
+    ASSERT_GE(kMockReportBatchSize, points.size());
+  }
+
+  static int GetCount() { return client1_count; }
+  static void ResetCount() { client1_count = 0; }
+  static int GetValue() { return client1_value; }
+  static double GetLastestHistMin() { return lastest_hist_min; }
+  static double GetLastestHistMean() { return lastest_hist_mean; }
+  static double GetLastestHistMax() { return lastest_hist_max; }
+
+ private:
+  void RecordLastHistData(const std::vector<MetricPoint> &points) {
+    for (auto &point : points) {
+      if (point.metric_name.find(".min") != std::string::npos) {
+        lastest_hist_min = point.value;
+      }
+      if (point.metric_name.find(".mean") != std::string::npos) {
+        lastest_hist_mean = point.value;
+      }
+      if (point.metric_name.find(".max") != std::string::npos) {
+        lastest_hist_max = point.value;
+      }
+    }
+  }
+
+ private:
+  static int client1_count;
+  static int client1_value;
+  static double lastest_hist_min;
+  static double lastest_hist_mean;
+  static double lastest_hist_max;
+};
+
+class MockExporterClient2 : public MetricExporterDecorator {
+ public:
+  MockExporterClient2(std::shared_ptr<MetricExporterClient> exporter)
+      : MetricExporterDecorator(exporter) {
+    client2_count = 0;
+  }
+  void ReportMetrics(const std::vector<MetricPoint> &points) override {
+    MetricExporterDecorator::ReportMetrics(points);
+    client2_count += points.size();
+    RAY_LOG(DEBUG) << "Client 2 " << client2_count << " last metric "
+                   << points.back().metric_name << ", value " << points.back().value;
+    client2_value = points.back().value;
+  }
+  static int GetCount() { return client2_count; }
+  static void ResetCount() { client2_count = 0; }
+  static int GetValue() { return client2_value; }
+
+ private:
+  static int client2_count;
+  static int client2_value;
+};
+
+class MetricExporterClientTest : public ::testing::Test {
+ public:
+  void SetUp() {
+    const stats::TagsType global_tags = {{stats::LanguageKey, "CPP"},
+                                         {stats::WorkerPidKey, "1000"}};
+    ray::stats::Init("127.0.0.1:8888", global_tags, false);
+    std::shared_ptr<MetricExporterClient> exporter(new stats::StdoutExporterClient());
+    std::shared_ptr<MetricExporterClient> mock1(new MockExporterClient1(exporter));
+    std::shared_ptr<MetricExporterClient> mock2(new MockExporterClient2(mock1));
+    MetricExporter::Register(mock2, kMockReportBatchSize);
+  }
+
+  void Shutdown() {
+    MockExporterClient1::ResetCount();
+    MockExporterClient2::ResetCount();
+  }
+};
+
+int MockExporterClient1::client1_count;
+double MockExporterClient1::lastest_hist_min;
+double MockExporterClient1::lastest_hist_mean;
+double MockExporterClient1::lastest_hist_max;
+int MockExporterClient2::client2_count;
+int MockExporterClient1::client1_value;
+int MockExporterClient2::client2_value;
+
+/// Default report flush interval is 10s, so we may wait a while for data
+/// exporting.
+uint32_t kReportFlushInterval = 10000;
+
+bool DoubleEqualTo(double value, double compared_value) {
+  return value >= compared_value - 1e-5 && value <= compared_value + 1e-5;
+}
+
+TEST_F(MetricExporterClientTest, decorator_test) {
+  // Export client should emit at least once in 10 seconds.
+  for (size_t i = 0; i < 100; ++i) {
+    stats::CurrentWorker().Record(i + 1);
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(kReportFlushInterval + 200));
+  ASSERT_GE(100, MockExporterClient1::GetValue());
+  ASSERT_GE(100, MockExporterClient2::GetValue());
+  ASSERT_EQ(1, MockExporterClient1::GetCount());
+  ASSERT_EQ(1, MockExporterClient2::GetCount());
+}
+
+TEST_F(MetricExporterClientTest, exporter_client_caculation_test) {
+  const stats::TagKeyType tag1 = stats::TagKeyType::Register("k1");
+  const stats::TagKeyType tag2 = stats::TagKeyType::Register("k2");
+  stats::Count random_counter("ray.random.counter", "", "", {tag1, tag2});
+  stats::Gauge random_gauge("ray.random.gauge", "", "", {tag1, tag2});
+  stats::Sum random_sum("ray.random.sum", "", "", {tag1, tag2});
+
+  std::vector<double> hist_vector;
+  for (int i = 0; i < 50; i++) {
+    hist_vector.push_back((double)(i * 10.0));
+  }
+  stats::Histogram random_hist("ray.random.hist", "", "", hist_vector, {tag1, tag2});
+  for (size_t i = 0; i < 500; ++i) {
+    random_counter.Record(i, {{tag1, std::to_string(i)}, {tag2, std::to_string(i * 2)}});
+    random_gauge.Record(i, {{tag1, std::to_string(i)}, {tag2, std::to_string(i * 2)}});
+    random_sum.Record(i, {{tag1, std::to_string(i)}, {tag2, std::to_string(i * 2)}});
+    random_hist.Record(i, {{tag1, std::to_string(i)}, {tag2, std::to_string(i * 2)}});
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(kReportFlushInterval + 200));
+  RAY_LOG(INFO) << "Min " << MockExporterClient1::GetLastestHistMin() << ", mean "
+                << MockExporterClient1::GetLastestHistMean() << ", max "
+                << MockExporterClient1::GetLastestHistMax();
+  ASSERT_TRUE(DoubleEqualTo(MockExporterClient1::GetLastestHistMin(), 0.0));
+  ASSERT_TRUE(DoubleEqualTo(MockExporterClient1::GetLastestHistMean(), 249.5));
+  ASSERT_TRUE(DoubleEqualTo(MockExporterClient1::GetLastestHistMax(), 499.0));
+}
+
+}  // namespace ray
+
+int main(int argc, char **argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}