diff --git a/BUILD.bazel b/BUILD.bazel index 69cb2ffbf..a7e139279 100644 --- a/BUILD.bazel +++ b/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 = [ diff --git a/src/ray/stats/metric.h b/src/ray/stats/metric.h index 964383daa..ede4a9500 100644 --- a/src/ray/stats/metric.h +++ b/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 tags; +}; } // namespace stats } // namespace ray diff --git a/src/ray/stats/metric_exporter.cc b/src/ray/stats/metric_exporter.cc new file mode 100644 index 000000000..9f42edf1e --- /dev/null +++ b/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 + +#include "ray/stats/metric_exporter.h" + +namespace ray { +namespace stats { + +template <> +void MetricExporter::ExportToPoints( + const opencensus::stats::ViewData::DataMap + &view_data, + const std::string &metric_name, std::vector &keys, + std::vector &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 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(row.second.mean()); + hist_max = static_cast(row.second.max()); + hist_min = static_cast(row.second.min()); + in_first_hist_data = false; + } else { + hist_mean += static_cast(row.second.mean()); + hist_max = std::max(hist_max, static_cast(row.second.max())); + hist_min = std::min(hist_min, static_cast(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> &data) { + std::vector 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 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(view_data.double_data(), metric_name, keys, points); + break; + case opencensus::stats::ViewData::Type::kInt64: + ExportToPoints(view_data.int_data(), metric_name, keys, points); + break; + case opencensus::stats::ViewData::Type::kDistribution: + ExportToPoints(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 diff --git a/src/ray/stats/metric_exporter.h b/src/ray/stats/metric_exporter.h new file mode 100644 index 000000000..25b462139 --- /dev/null +++ b/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 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 metric_exporter_client, + size_t report_batch_size) { + opencensus::stats::StatsExporter::RegisterPushHandler( + absl::make_unique(metric_exporter_client, report_batch_size)); + } + + void ExportViewData( + const std::vector> &data) override; + + private: + template + /// 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 &view_data, + const std::string &metric_name, std::vector &keys, + std::vector &points) { + for (const auto &row : view_data) { + std::unordered_map 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(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 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 diff --git a/src/ray/stats/metric_exporter_client.cc b/src/ray/stats/metric_exporter_client.cc new file mode 100644 index 000000000..f5df17738 --- /dev/null +++ b/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 + +#include "ray/stats/metric_exporter_client.h" + +namespace ray { +namespace stats { + +void StdoutExporterClient::ReportMetrics(const std::vector &points) { + RAY_LOG(DEBUG) << "Metric point size : " << points.size(); +} + +MetricExporterDecorator::MetricExporterDecorator( + std::shared_ptr exporter) + : exporter_(exporter) {} + +void MetricExporterDecorator::ReportMetrics(const std::vector &points) { + if (exporter_) { + exporter_->ReportMetrics(points); + } +} +} // namespace stats +} // namespace ray diff --git a/src/ray/stats/metric_exporter_client.h b/src/ray/stats/metric_exporter_client.h new file mode 100644 index 000000000..507f4051c --- /dev/null +++ b/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 &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 &points) override; +}; + +/// The decoration mode is that the user can apply it by configuring different +/// combinations. +/// Usage: +/// std::shared_ptr exporter(new StdoutExporterClient()); +/// std::shared_ptr 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 exporter); + virtual void ReportMetrics(const std::vector &points); + + private: + std::shared_ptr exporter_; +}; + +class OpentsdbExporterClient : public MetricExporterDecorator { + public: + OpentsdbExporterClient(std::shared_ptr exporter) + : MetricExporterDecorator(exporter) {} + void ReportMetrics(const std::vector &points) override { + MetricExporterDecorator::ReportMetrics(points); + // TODO(lingxuan.zlx): opentsdb client is used for report to backend + // storage. + } +}; + +} // namespace stats +} // namespace ray diff --git a/src/ray/stats/metric_exporter_client_test.cc b/src/ray/stats/metric_exporter_client_test.cc new file mode 100644 index 000000000..05eaea73f --- /dev/null +++ b/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 +#include +#include +#include + +#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 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 &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 &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 exporter) + : MetricExporterDecorator(exporter) { + client2_count = 0; + } + void ReportMetrics(const std::vector &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 exporter(new stats::StdoutExporterClient()); + std::shared_ptr mock1(new MockExporterClient1(exporter)); + std::shared_ptr 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 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(); +}