diff --git a/BUILD.bazel b/BUILD.bazel index 8ffc27954..f74780747 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -1011,6 +1011,7 @@ cc_test( name = "stats_test", srcs = ["src/ray/stats/stats_test.cc"], copts = COPTS, + tags = ["stats"], deps = [ ":stats_lib", "@com_google_googletest//:gtest_main", @@ -1021,6 +1022,7 @@ cc_test( name = "metric_exporter_client_test", srcs = ["src/ray/stats/metric_exporter_client_test.cc"], copts = COPTS, + tags = ["stats"], deps = [ ":stats_lib", "@com_google_googletest//:gtest_main", diff --git a/src/ray/stats/metric.h b/src/ray/stats/metric.h index b1acc43a8..20520f271 100644 --- a/src/ray/stats/metric.h +++ b/src/ray/stats/metric.h @@ -18,6 +18,7 @@ #include #include "opencensus/stats/stats.h" +#include "opencensus/stats/stats_exporter.h" #include "opencensus/tags/tag_key.h" #include "ray/util/logging.h" @@ -83,7 +84,7 @@ class Metric { unit_(unit), tag_keys_(tag_keys){}; - virtual ~Metric() = default; + virtual ~Metric() { opencensus::stats::StatsExporter::RemoveView(name_); } Metric &operator()() { return *this; } diff --git a/src/ray/stats/metric_exporter.cc b/src/ray/stats/metric_exporter.cc index 4a714e7d7..306a4354e 100644 --- a/src/ray/stats/metric_exporter.cc +++ b/src/ray/stats/metric_exporter.cc @@ -52,18 +52,10 @@ void MetricExporter::ExportToPoints( } } hist_mean /= view_data.size(); - MetricPoint mean_point = {metric_name + ".mean", - current_sys_time_ms(), - hist_mean, + MetricPoint mean_point = {metric_name + ".mean", current_sys_time_ms(), hist_mean, tags}; - MetricPoint max_point = {metric_name + ".max", - current_sys_time_ms(), - hist_max, - tags}; - MetricPoint min_point = {metric_name + ".min", - current_sys_time_ms(), - hist_min, - tags}; + MetricPoint max_point = {metric_name + ".max", current_sys_time_ms(), hist_max, tags}; + MetricPoint min_point = {metric_name + ".min", current_sys_time_ms(), hist_min, tags}; points.push_back(std::move(mean_point)); points.push_back(std::move(max_point)); points.push_back(std::move(min_point)); diff --git a/src/ray/stats/metric_exporter_client_test.cc b/src/ray/stats/metric_exporter_client_test.cc index 30929d675..a1b39c299 100644 --- a/src/ray/stats/metric_exporter_client_test.cc +++ b/src/ray/stats/metric_exporter_client_test.cc @@ -21,6 +21,8 @@ #include #include "absl/memory/memory.h" +#include "opencensus/stats/internal/delta_producer.h" +#include "opencensus/stats/internal/stats_exporter_impl.h" #include "ray/stats/metric_exporter.h" #include "ray/stats/metric_exporter_client.h" #include "ray/stats/stats.h" @@ -33,19 +35,19 @@ 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; - } + : MetricExporterDecorator(exporter), + client1_count(0), + client1_value(0), + lastest_hist_min(0.0), + lastest_hist_mean(0.0), + lastest_hist_max(0.0) {} void ReportMetrics(const std::vector &points) override { if (points.empty()) { return; } MetricExporterDecorator::ReportMetrics(points); - client1_count += points.size(); + client1_count++; client1_value = points.back().value; RAY_LOG(DEBUG) << "Client 1 " << client1_count << " last metric " << points.back().metric_name << ", value " << points.back().value; @@ -53,13 +55,11 @@ class MockExporterClient1 : public MetricExporterDecorator { // 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; } + int GetCount() { return client1_count; } + int GetValue() { return client1_value; } + double GetLastestHistMin() { return lastest_hist_min; } + double GetLastestHistMean() { return lastest_hist_mean; } + double GetLastestHistMax() { return lastest_hist_max; } private: void RecordLastHistData(const std::vector &points) { @@ -77,36 +77,33 @@ class MockExporterClient1 : public MetricExporterDecorator { } private: - static int client1_count; - static int client1_value; - static double lastest_hist_min; - static double lastest_hist_mean; - static double lastest_hist_max; + int client1_count; + int client1_value; + double lastest_hist_min; + double lastest_hist_mean; + double lastest_hist_max; }; class MockExporterClient2 : public MetricExporterDecorator { public: MockExporterClient2(std::shared_ptr exporter) - : MetricExporterDecorator(exporter) { - client2_count = 0; - } + : MetricExporterDecorator(exporter), client2_count(0), client2_value(0) {} void ReportMetrics(const std::vector &points) override { if (points.empty()) { return; } MetricExporterDecorator::ReportMetrics(points); - client2_count += points.size(); + client2_count++; 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; } + int GetCount() { return client2_count; } + int GetValue() { return client2_value; } private: - static int client2_count; - static int client2_value; + int client2_count; + int client2_value; }; /// Default report flush interval is 500ms, so we may wait a while for data @@ -115,7 +112,7 @@ uint32_t kReportFlushInterval = 500; class MetricExporterClientTest : public ::testing::Test { public: - void SetUp() { + virtual void SetUp() override { const stats::TagsType global_tags = {{stats::LanguageKey, "CPP"}, {stats::WorkerPidKey, "1000"}}; absl::Duration report_interval = absl::Milliseconds(kReportFlushInterval); @@ -123,70 +120,68 @@ class MetricExporterClientTest : public ::testing::Test { ray::stats::StatsConfig::instance().SetReportInterval(report_interval); ray::stats::StatsConfig::instance().SetHarvestInterval(harvest_interval); - std::shared_ptr exporter(new stats::StdoutExporterClient()); - std::shared_ptr mock1(new MockExporterClient1(exporter)); - std::shared_ptr mock2(new MockExporterClient2(mock1)); + exporter.reset(new stats::StdoutExporterClient()); + mock1.reset(new MockExporterClient1(exporter)); + mock2.reset(new MockExporterClient2(mock1)); ray::stats::Init(global_tags, 10054, io_service_, mock2, kMockReportBatchSize); } + virtual void TearDown() override { Shutdown(); } + void Shutdown() { - MockExporterClient1::ResetCount(); - MockExporterClient2::ResetCount(); + opencensus::stats::StatsExporterImpl::Get()->ClearHandlersForTesting(); } - private: + protected: boost::asio::io_service io_service_; + std::shared_ptr exporter; + std::shared_ptr mock1; + std::shared_ptr mock2; }; -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; - 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. + // Export client should emit at least once in report flush interval. for (size_t i = 0; i < 100; ++i) { stats::CurrentWorker().Record(i + 1); } - std::this_thread::sleep_for(std::chrono::milliseconds(kReportFlushInterval + 20)); - ASSERT_GE(100, MockExporterClient1::GetValue()); - ASSERT_GE(100, MockExporterClient2::GetValue()); - ASSERT_EQ(1, MockExporterClient1::GetCount()); - ASSERT_EQ(1, MockExporterClient2::GetCount()); + opencensus::stats::DeltaProducer::Get()->Flush(); + opencensus::stats::StatsExporterImpl::Get()->Export(); + ASSERT_GE(100, mock1->GetValue()); + ASSERT_EQ(1, mock1->GetCount()); + ASSERT_GE(100, mock2->GetValue()); + ASSERT_EQ(1, mock2->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}); + static stats::Count random_counter("ray.random.counter", "", "", {tag1, tag2}); + static stats::Gauge random_gauge("ray.random.gauge", "", "", {tag1, tag2}); + static 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}); + static 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 + 20)); - 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)); + opencensus::stats::DeltaProducer::Get()->Flush(); + opencensus::stats::StatsExporterImpl::Get()->Export(); + RAY_LOG(INFO) << "Min " << mock1->GetLastestHistMin() << ", mean " + << mock1->GetLastestHistMean() << ", max " << mock1->GetLastestHistMax(); + ASSERT_TRUE(DoubleEqualTo(mock1->GetLastestHistMin(), 0.0)); + ASSERT_TRUE(DoubleEqualTo(mock1->GetLastestHistMean(), 249.5)); + ASSERT_TRUE(DoubleEqualTo(mock1->GetLastestHistMax(), 499.0)); } } // namespace ray diff --git a/src/ray/stats/stats_test.cc b/src/ray/stats/stats_test.cc index d787d6937..1546c398b 100644 --- a/src/ray/stats/stats_test.cc +++ b/src/ray/stats/stats_test.cc @@ -56,9 +56,17 @@ class MockExporter : public opencensus::stats::StatsExporter::Handler { } }; +/// Default report flush interval is 500ms, so we may wait a while for data +/// exporting. +uint32_t kReportFlushInterval = 500; + class StatsTest : public ::testing::Test { public: void SetUp() { + absl::Duration report_interval = absl::Milliseconds(kReportFlushInterval); + absl::Duration harvest_interval = absl::Milliseconds(kReportFlushInterval / 2); + ray::stats::StatsConfig::instance().SetReportInterval(report_interval); + ray::stats::StatsConfig::instance().SetHarvestInterval(harvest_interval); const stats::TagsType global_tags = {{stats::LanguageKey, "CPP"}, {stats::WorkerPidKey, "1000"}}; std::shared_ptr exporter( @@ -74,7 +82,7 @@ class StatsTest : public ::testing::Test { }; TEST_F(StatsTest, F) { - for (size_t i = 0; i < 500; ++i) { + for (size_t i = 0; i < 20; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); stats::CurrentWorker().Record(2345); }