Change the stats exporter interface to batch all views together. (#76)

This allows the Stackdriver exporter to batch TimeSeries from different views into fewer RPCs.

Author: Ian Sturdy

opencensus/exporters/stats/stackdriver/internal/stackdriver_exporter.cc

@@ -40,8 +40,9 @@ class StackdriverExporter::Handler
  public:
   Handler(absl::string_view project_id, absl::string_view opencensus_task);
 
-  void ExportViewData(const opencensus::stats::ViewDescriptor& descriptor,
-                      const opencensus::stats::ViewData& data)
+  void ExportViewData(
+      const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                  opencensus::stats::ViewData>>& data)
       LOCKS_EXCLUDED(mu_) override;
 
  private:
@@ -70,20 +71,26 @@ StackdriverExporter::Handler::Handler(absl::string_view project_id,
                                 ::grpc::GoogleDefaultCredentials()))) {}
 
 void StackdriverExporter::Handler::ExportViewData(
-    const opencensus::stats::ViewDescriptor& descriptor,
-    const opencensus::stats::ViewData& data) {
+    const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                opencensus::stats::ViewData>>& data) {
+  // TODO: refactor to avoid copying the timeseries.
   absl::MutexLock l(&mu_);
-  if (!MaybeRegisterView(descriptor)) {
-    return;
+  std::vector<google::monitoring::v3::TimeSeries> time_series;
+  for (const auto& datum : data) {
+    if (!MaybeRegisterView(datum.first)) {
+      continue;
+    }
+    const auto view_time_series =
+        MakeTimeSeries(datum.first, datum.second, opencensus_task_);
+    time_series.insert(time_series.end(), view_time_series.begin(),
+                       view_time_series.end());
   }
 
   // TODO: use asynchronous RPCs.
-  const auto time_series = MakeTimeSeries(descriptor, data, opencensus_task_);
   int i = 0;
   while (i < time_series.size()) {
     auto request = google::monitoring::v3::CreateTimeSeriesRequest();
     request.set_name(project_id_);
-    // TODO: refactor to avoid copying each timeseries.
     for (int batch_index = 0; batch_index < kTimeSeriesBatchSize;
          ++batch_index) {
       *request.add_time_series() = time_series[i];

opencensus/exporters/stats/stdout/internal/stdout_exporter.cc

@@ -50,8 +50,9 @@ std::string DataToString(const opencensus::stats::Distribution& data) {
 class StdoutExporter::Handler
     : public opencensus::stats::StatsExporter::Handler {
  public:
-  void ExportViewData(const opencensus::stats::ViewDescriptor& descriptor,
-                      const opencensus::stats::ViewData& data) override;
+  void ExportViewData(
+      const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                  opencensus::stats::ViewData>>& data) override;
 
  private:
   // Implements ExportViewData for supported data types.
@@ -69,21 +70,24 @@ void StdoutExporter::Register() {
 }
 
 void StdoutExporter::Handler::ExportViewData(
-    const opencensus::stats::ViewDescriptor& descriptor,
-    const opencensus::stats::ViewData& data) {
-  switch (data.type()) {
-    case opencensus::stats::ViewData::Type::kDouble:
-      ExportViewDataImpl(descriptor, data.start_time(), data.end_time(),
-                         data.double_data());
-      break;
-    case opencensus::stats::ViewData::Type::kInt64:
-      ExportViewDataImpl(descriptor, data.start_time(), data.end_time(),
-                         data.int_data());
-      break;
-    case opencensus::stats::ViewData::Type::kDistribution:
-      ExportViewDataImpl(descriptor, data.start_time(), data.end_time(),
-                         data.distribution_data());
-      break;
+    const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                opencensus::stats::ViewData>>& data) {
+  for (const auto& datum : data) {
+    const auto& view_data = datum.second;
+    switch (view_data.type()) {
+      case opencensus::stats::ViewData::Type::kDouble:
+        ExportViewDataImpl(datum.first, view_data.start_time(),
+                           view_data.end_time(), view_data.double_data());
+        break;
+      case opencensus::stats::ViewData::Type::kInt64:
+        ExportViewDataImpl(datum.first, view_data.start_time(),
+                           view_data.end_time(), view_data.int_data());
+        break;
+      case opencensus::stats::ViewData::Type::kDistribution:
+        ExportViewDataImpl(datum.first, view_data.start_time(),
+                           view_data.end_time(), view_data.distribution_data());
+        break;
+    }
   }
 }
 

opencensus/stats/examples/exporter_example.cc

@@ -42,24 +42,30 @@ class ExampleExporter : public opencensus::stats::StatsExporter::Handler {
         absl::make_unique<ExampleExporter>());
   }
 
-  void ExportViewData(const opencensus::stats::ViewDescriptor& descriptor,
-                      const opencensus::stats::ViewData& data) override {
-    if (data.type() != opencensus::stats::ViewData::Type::kDouble) {
-      // This example only supports double data (i.e. Sum() aggregation).
-      return;
-    }
-    std::string output;
-    absl::StrAppend(&output, "\nData for view \"", descriptor.name(),
-                    "\" from ", absl::FormatTime(data.start_time()), " to ",
-                    absl::FormatTime(data.end_time()), ":\n");
-    for (const auto& row : data.double_data()) {
-      for (int i = 0; i < descriptor.columns().size(); ++i) {
-        absl::StrAppend(&output, descriptor.columns()[i], ":", row.first[i],
-                        ", ");
+  void ExportViewData(
+      const std::vector<std::pair<opencensus::stats::ViewDescriptor,
+                                  opencensus::stats::ViewData>>& data)
+      override {
+    for (const auto& datum : data) {
+      const auto& descriptor = datum.first;
+      const auto& view_data = datum.second;
+      if (view_data.type() != opencensus::stats::ViewData::Type::kDouble) {
+        // This example only supports double data (i.e. Sum() aggregation).
+        return;
+      }
+      std::string output;
+      absl::StrAppend(&output, "\nData for view \"", descriptor.name(),
+                      "\" from ", absl::FormatTime(view_data.start_time()),
+                      " to ", absl::FormatTime(view_data.end_time()), ":\n");
+      for (const auto& row : view_data.double_data()) {
+        for (int i = 0; i < descriptor.columns().size(); ++i) {
+          absl::StrAppend(&output, descriptor.columns()[i], ":", row.first[i],
+                          ", ");
+        }
+        absl::StrAppend(&output, row.second, "\n");
       }
-      absl::StrAppend(&output, row.second, "\n");
+      std::cout << output;
     }
-    std::cout << output;
   }
 };
 

opencensus/stats/internal/stats_exporter.cc

@@ -69,8 +69,13 @@ StatsExporterImpl::GetViewData() {
 
 void StatsExporterImpl::Export() {
   absl::ReaderMutexLock l(&mu_);
+  std::vector<std::pair<ViewDescriptor, ViewData>> data;
+  data.reserve(views_.size());
   for (const auto& view : views_) {
-    SendToHandlers(view.second->descriptor(), view.second->GetData());
+    data.emplace_back(view.second->descriptor(), view.second->GetData());
+  }
+  for (auto& handler : handlers_) {
+    handler->ExportViewData(data);
   }
 }
 
@@ -79,14 +84,6 @@ void StatsExporterImpl::ClearHandlersForTesting() {
   handlers_.clear();
 }
 
-void StatsExporterImpl::SendToHandlers(const ViewDescriptor& descriptor,
-                                       const ViewData& data)
-    SHARED_LOCKS_REQUIRED(mu_) {
-  for (auto& handler : handlers_) {
-    handler->ExportViewData(descriptor, data);
-  }
-}
-
 void StatsExporterImpl::StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(mu_) {
   t_ = std::thread(&StatsExporterImpl::RunWorkerLoop, this);
   thread_started_ = true;

opencensus/stats/internal/stats_exporter_impl.h

@@ -49,9 +49,6 @@ class StatsExporterImpl {
  private:
   StatsExporterImpl() {}
 
-  void SendToHandlers(const ViewDescriptor& descriptor, const ViewData& data)
-      SHARED_LOCKS_REQUIRED(mu_);
-
   void StartExportThread() EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
   // Loops forever, calling Export() every export_interval_.

opencensus/stats/internal/stats_exporter_test.cc

@@ -32,32 +32,29 @@
 namespace opencensus {
 namespace stats {
 
-// A mock exporter that checks whether it is called with the right set of view
-// descriptors.
+// A mock exporter that assigns exported data to the provided pointer.
 class MockExporter : public StatsExporter::Handler {
  public:
-  static void Register(absl::Span<const ViewDescriptor> expected_descriptors) {
+  static void Register(
+      std::vector<std::pair<ViewDescriptor, ViewData>>* output) {
     opencensus::stats::StatsExporter::RegisterPushHandler(
-        absl::make_unique<MockExporter>(expected_descriptors));
+        absl::make_unique<MockExporter>(output));
   }
 
-  MockExporter(absl::Span<const ViewDescriptor> expected_descriptors)
-      : expected_descriptors_(expected_descriptors.begin(),
-                              expected_descriptors.end()) {}
+  MockExporter(std::vector<std::pair<ViewDescriptor, ViewData>>* output)
+      : output_(output) {}
 
-  ~MockExporter() {
-    EXPECT_THAT(actual_descriptors_,
-                ::testing::UnorderedElementsAreArray(expected_descriptors_));
-  }
-
-  void ExportViewData(const ViewDescriptor& descriptor,
-                      const ViewData& data) override {
-    actual_descriptors_.push_back(descriptor);
+  void ExportViewData(
+      const std::vector<std::pair<ViewDescriptor, ViewData>>& data) override {
+    // Looping because ViewData is (intentionally) copy-constructable but not
+    // copy_assignable.
+    for (const auto& datum : data) {
+      output_->emplace_back(datum.first, datum.second);
+    }
   }
 
  private:
-  const std::vector<ViewDescriptor> expected_descriptors_;
-  std::vector<ViewDescriptor> actual_descriptors_;
+  std::vector<std::pair<ViewDescriptor, ViewData>>* output_;
 };
 
 constexpr char kMeasureId[] = "test_measure_id";
@@ -99,17 +96,22 @@ class StatsExporterTest : public ::testing::Test {
 };
 
 TEST_F(StatsExporterTest, AddView) {
-  MockExporter::Register({descriptor1_, descriptor2_});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data;
+  MockExporter::Register(&exported_data);
   descriptor1_.RegisterForExport();
   descriptor2_.RegisterForExport();
   EXPECT_THAT(StatsExporter::GetViewData(),
               ::testing::UnorderedElementsAre(::testing::Key(descriptor1_),
                                               ::testing::Key(descriptor2_)));
   Export();
+  EXPECT_THAT(exported_data,
+              ::testing::UnorderedElementsAre(::testing::Key(descriptor1_),
+                                              ::testing::Key(descriptor2_)));
 }
 
 TEST_F(StatsExporterTest, UpdateView) {
-  MockExporter::Register({descriptor1_edited_, descriptor2_});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data;
+  MockExporter::Register(&exported_data);
   descriptor1_.RegisterForExport();
   descriptor2_.RegisterForExport();
   descriptor1_edited_.RegisterForExport();
@@ -118,39 +120,56 @@ TEST_F(StatsExporterTest, UpdateView) {
       ::testing::UnorderedElementsAre(::testing::Key(descriptor1_edited_),
                                       ::testing::Key(descriptor2_)));
   Export();
+  EXPECT_THAT(exported_data, ::testing::UnorderedElementsAre(
+                                 ::testing::Key(descriptor1_edited_),
+                                 ::testing::Key(descriptor2_)));
 }
 
 TEST_F(StatsExporterTest, RemoveView) {
-  MockExporter::Register({descriptor2_});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data;
+  MockExporter::Register(&exported_data);
   descriptor1_.RegisterForExport();
   descriptor2_.RegisterForExport();
   StatsExporter::RemoveView(descriptor1_.name());
   EXPECT_THAT(StatsExporter::GetViewData(),
               ::testing::UnorderedElementsAre(::testing::Key(descriptor2_)));
   Export();
+  EXPECT_THAT(exported_data,
+              ::testing::UnorderedElementsAre(::testing::Key(descriptor2_)));
 }
 
 TEST_F(StatsExporterTest, MultipleExporters) {
-  MockExporter::Register({descriptor1_});
-  MockExporter::Register({descriptor1_});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data_1;
+  MockExporter::Register(&exported_data_1);
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data_2;
+  MockExporter::Register(&exported_data_2);
   descriptor1_.RegisterForExport();
   Export();
+  EXPECT_THAT(exported_data_1,
+              ::testing::UnorderedElementsAre(::testing::Key(descriptor1_)));
+  EXPECT_THAT(exported_data_2,
+              ::testing::UnorderedElementsAre(::testing::Key(descriptor1_)));
 }
 
 TEST_F(StatsExporterTest, IntervalViewRejected) {
-  MockExporter::Register({});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data;
+  MockExporter::Register(&exported_data);
   ViewDescriptor interval_descriptor = ViewDescriptor().set_name("interval");
   SetAggregationWindow(AggregationWindow::Interval(absl::Hours(1)),
                        &interval_descriptor);
   interval_descriptor.RegisterForExport();
   EXPECT_TRUE(StatsExporter::GetViewData().empty());
   Export();
+  EXPECT_TRUE(exported_data.empty());
 }
 
 TEST_F(StatsExporterTest, TimedExport) {
-  MockExporter::Register({descriptor1_});
+  std::vector<std::pair<ViewDescriptor, ViewData>> exported_data;
+  MockExporter::Register(&exported_data);
   descriptor1_.RegisterForExport();
   absl::SleepFor(absl::Seconds(11));
+  EXPECT_THAT(exported_data,
+              ::testing::UnorderedElementsAre(::testing::Key(descriptor1_)));
 }
 
 }  // namespace stats

opencensus/stats/stats_exporter.h

@@ -45,8 +45,8 @@ class StatsExporter final {
   class Handler {
    public:
     virtual ~Handler() = default;
-    virtual void ExportViewData(const ViewDescriptor& descriptor,
-                                const ViewData& data) = 0;
+    virtual void ExportViewData(
+        const std::vector<std::pair<ViewDescriptor, ViewData>>& data) = 0;
   };
 
   // This should only be called by push exporters' Register() methods.