Update ViewData to store a separate start_timestamp for each tag map. (#444)

 Update the stackdriver_exporter to use a separate timestamp for each timeseries/tag map.

Author: Alex Amato

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

@@ -132,6 +132,7 @@ template <typename DataValueT>
 std::vector<google::monitoring::v3::TimeSeries> DataToTimeSeries(
     const opencensus::stats::ViewDescriptor& view_descriptor,
     const opencensus::stats::ViewData::DataMap<DataValueT>& data,
+    const opencensus::stats::ViewData::DataMap<absl::Time>& start_times,
     const google::monitoring::v3::TimeSeries& base_time_series) {
   const google::api::MetricDescriptor::ValueType type =
       GetValueType(view_descriptor);
@@ -147,6 +148,17 @@ std::vector<google::monitoring::v3::TimeSeries> DataToTimeSeries(
     // The point is already created in the base_time_series to set the times.
     SetTypedValue(row.second, type,
                   time_series.mutable_points(0)->mutable_value());
+
+    // Stackdriver doesn't like start_time and end_time being different for
+    // GAUGE metrics. Don't set the start time for GAUGE.
+    if (view_descriptor.aggregation().type() !=
+        opencensus::stats::Aggregation::Type::kLastValue) {
+      // Use the start time stored for the specific row's tags.
+      auto* interval = time_series.mutable_points(0)->mutable_interval();
+      absl::Time start_time = start_times.at(row.first);
+      opencensus::common::SetTimestamp(start_time,
+                                       interval->mutable_start_time());
+    }
   }
   return vector;
 }
@@ -223,13 +235,6 @@ std::vector<google::monitoring::v3::TimeSeries> MakeTimeSeries(
     *base_time_series.mutable_resource() = *monitored_resource_for_view;
   }
   auto* interval = base_time_series.add_points()->mutable_interval();
-  // Stackdriver doesn't like start_time and end_time being different for GAUGE
-  // metrics.
-  if (view_descriptor.aggregation().type() !=
-      opencensus::stats::Aggregation::Type::kLastValue) {
-    opencensus::common::SetTimestamp(data.start_time(),
-                                     interval->mutable_start_time());
-  }
   opencensus::common::SetTimestamp(data.end_time(),
                                    interval->mutable_end_time());
   if (add_task_label) {
@@ -239,13 +244,13 @@ std::vector<google::monitoring::v3::TimeSeries> MakeTimeSeries(
   switch (data.type()) {
     case opencensus::stats::ViewData::Type::kDouble:
       return DataToTimeSeries(view_descriptor, data.double_data(),
-                              base_time_series);
+                              data.start_times(), base_time_series);
     case opencensus::stats::ViewData::Type::kInt64:
       return DataToTimeSeries(view_descriptor, data.int_data(),
-                              base_time_series);
+                              data.start_times(), base_time_series);
     case opencensus::stats::ViewData::Type::kDistribution:
       return DataToTimeSeries(view_descriptor, data.distribution_data(),
-                              base_time_series);
+                              data.start_times(), base_time_series);
   }
   ABSL_ASSERT(false && "Bad ViewData.type().");
   return {};

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

@@ -31,6 +31,7 @@
 #include "opencensus/stats/testing/test_utils.h"
 
 using opencensus::stats::testing::TestUtils;
+using opencensus::stats::testing::TestViewValue;
 
 namespace opencensus {
 namespace exporters {
@@ -67,6 +68,19 @@ DefaultPerMetricResource() {
   return std::unordered_map<std::string, google::api::MonitoredResource>();
 }
 
+std::vector<std::string> ToDataMapKeys(
+    const std::initializer_list<std::string> label_names,
+    const google::monitoring::v3::TimeSeries& time_series) {
+  // label_names defines the key order and is needed because
+  // time_series.metric().labels() does not maintain the original order.
+  std::vector<std::string> data_map_keys;
+  const auto& label_map = time_series.metric().labels();
+  for (const auto& label_name : label_names) {
+    data_map_keys.push_back(label_map.at(label_name));
+  }
+  return data_map_keys;
+}
+
 // TODO: MonitoredResourceForView
 TEST(StackdriverUtilsTest, MonitoredResourceForViewIsDefault) {
   const auto view_descriptor =
@@ -368,6 +382,9 @@ TEST(StackdriverUtilsTest, MakeTimeSeriesSumDoubleAndTypes) {
   for (const auto& ts : time_series) {
     EXPECT_EQ("custom.googleapis.com/test/test_view", ts.metric().type());
     ASSERT_EQ(1, ts.points_size());
+    const auto& tags = ToDataMapKeys({"foo", "bar"}, ts);
+    EXPECT_EQ(absl::ToUnixSeconds(data.start_times().at(tags)),
+              ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.start_time()),
               ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.end_time()),
@@ -400,19 +417,42 @@ TEST(StackdriverUtilsTest, MakeTimeSeriesSumInt) {
           .set_aggregation(opencensus::stats::Aggregation::Sum())
           .add_column(tag_key_1)
           .add_column(tag_key_2);
-  const opencensus::stats::ViewData data = TestUtils::MakeViewData(
-      view_descriptor, {{{"v1", "v1"}, 1.0}, {{"v1", "v2"}, 2.0}});
+  absl::Time t1 = absl::UnixEpoch();
+  absl::Time t2 = t1 + absl::Seconds(15);
+  std::map<std::vector<std::string>, absl::Time> start_times{
+      {{"v1", "v1"}, t1},
+      {{"v1", "v2"}, t2},
+  };
+
+  std::vector<TestViewValue> view_values;
+  TestViewValue view_value1;
+  view_value1.tag_values = {"v1", "v1"};
+  view_value1.value = 1.0;
+  view_value1.start_time = t1;
+  view_values.push_back(view_value1);
+
+  TestViewValue view_value2;
+  view_value2.tag_values = {"v1", "v2"};
+  view_value2.value = 2.0;
+  view_value2.start_time = t2;
+  view_values.push_back(view_value2);
+
+  const opencensus::stats::ViewData data =
+      TestUtils::MakeViewDataWithStartTimes(view_descriptor, view_values);
   const std::vector<google::monitoring::v3::TimeSeries> time_series =
       MakeTimeSeries(kMetricNamePrefix, kDefaultResource, view_descriptor, data,
                      kAddTaskLabel, task);
 
   for (const auto& ts : time_series) {
     ASSERT_EQ(1, ts.points_size());
-    EXPECT_EQ(absl::ToUnixSeconds(data.start_time()),
+    const auto& tags = ToDataMapKeys({"foo", "bar"}, ts);
+    EXPECT_EQ(absl::ToUnixSeconds(start_times[tags]),
               ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.end_time()),
               ts.points(0).interval().end_time().seconds());
   }
+  EXPECT_EQ(absl::ToUnixSeconds(data.start_time()),
+            absl::ToUnixSeconds(absl::UnixEpoch()));
 
   EXPECT_THAT(time_series,
               ::testing::UnorderedElementsAre(
@@ -449,6 +489,9 @@ TEST(StackdriverUtilsTest, MakeTimeSeriesCountDouble) {
 
   for (const auto& ts : time_series) {
     ASSERT_EQ(1, ts.points_size());
+    const auto& tags = ToDataMapKeys({"foo", "bar"}, ts);
+    EXPECT_EQ(absl::ToUnixSeconds(data.start_times().at(tags)),
+              ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.start_time()),
               ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.end_time()),
@@ -493,6 +536,9 @@ TEST(StackdriverUtilsTest, MakeTimeSeriesDistributionDouble) {
 
   for (const auto& ts : time_series) {
     ASSERT_EQ(1, ts.points_size());
+    const auto& tags = ToDataMapKeys({"foo", "bar"}, ts);
+    EXPECT_EQ(absl::ToUnixSeconds(data.start_times().at(tags)),
+              ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.start_time()),
               ts.points(0).interval().start_time().seconds());
     EXPECT_EQ(absl::ToUnixSeconds(data.end_time()),
@@ -540,6 +586,9 @@ TEST(StackdriverUtilsTest, MakeTimeSeriesLastValueInt) {
 
   for (const auto& ts : time_series) {
     ASSERT_EQ(1, ts.points_size());
+    const auto& tags = ToDataMapKeys({"foo", "bar"}, ts);
+    EXPECT_EQ(absl::ToUnixSeconds(data.start_times().at(tags)),
+              ts.points(0).interval().start_time().seconds());
     EXPECT_FALSE(ts.points(0).interval().has_start_time());
     EXPECT_EQ(absl::ToUnixSeconds(data.end_time()),
               ts.points(0).interval().end_time().seconds());

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

@@ -61,7 +61,8 @@ class Handler : public opencensus::stats::StatsExporter::Handler {
   template <typename DataValueT>
   void ExportViewDataImpl(
       const opencensus::stats::ViewDescriptor& descriptor,
-      absl::Time start_time, absl::Time end_time,
+      const ::opencensus::stats::ViewData::DataMap<absl::Time>& start_times,
+      absl::Time end_time,
       const opencensus::stats::ViewData::DataMap<DataValueT>& data);
 
   std::ostream* stream_;
@@ -74,15 +75,15 @@ void Handler::ExportViewData(
     const auto& view_data = datum.second;
     switch (view_data.type()) {
       case opencensus::stats::ViewData::Type::kDouble:
-        ExportViewDataImpl(datum.first, view_data.start_time(),
+        ExportViewDataImpl(datum.first, view_data.start_times(),
                            view_data.end_time(), view_data.double_data());
         break;
       case opencensus::stats::ViewData::Type::kInt64:
-        ExportViewDataImpl(datum.first, view_data.start_time(),
+        ExportViewDataImpl(datum.first, view_data.start_times(),
                            view_data.end_time(), view_data.int_data());
         break;
       case opencensus::stats::ViewData::Type::kDistribution:
-        ExportViewDataImpl(datum.first, view_data.start_time(),
+        ExportViewDataImpl(datum.first, view_data.start_times(),
                            view_data.end_time(), view_data.distribution_data());
         break;
     }
@@ -92,20 +93,23 @@ void Handler::ExportViewData(
 
 template <typename DataValueT>
 void Handler::ExportViewDataImpl(
-    const opencensus::stats::ViewDescriptor& descriptor, absl::Time start_time,
+    const opencensus::stats::ViewDescriptor& descriptor,
+    const ::opencensus::stats::ViewData::DataMap<absl::Time>& start_times,
     absl::Time end_time,
     const opencensus::stats::ViewData::DataMap<DataValueT>& data) {
   if (data.empty()) {
-    *stream_ << absl::StrCat("No data for view \"", descriptor.name(),
-                             "\" from ", absl::FormatTime(start_time), ".\n\n");
+    *stream_ << absl::StrCat("No data for view \"", descriptor.name(), "\".\n");
     return;
   }
   // Build a string so we can write it in one shot to minimize crosstalk if
   // multiple threads write to stream_ simultaneously.
-  std::string output = absl::StrCat("Data for view \"", descriptor.name(),
-                                    "\" from ", absl::FormatTime(start_time),
-                                    " to ", absl::FormatTime(end_time), ":\n");
+  std::string output =
+      absl::StrCat("Data for view \"", descriptor.name(), "\":\n");
+
   for (const auto& row : data) {
+    auto start_time = start_times.at(row.first);
+    absl::StrAppend(&output, "Row data from ", absl::FormatTime(start_time),
+                    " to ", absl::FormatTime(end_time), ":\n");
     absl::StrAppend(&output, "  ");
     for (int i = 0; i < descriptor.columns().size(); ++i) {
       absl::StrAppend(&output, descriptor.columns()[i].name(), "=",

opencensus/stats/examples/exporter_example.cc

@@ -60,10 +60,12 @@ class ExampleExporter : public opencensus::stats::StatsExporter::Handler {
         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");
+      absl::StrAppend(&output, "\nData for view \"", descriptor.name(), "\n");
       for (const auto& row : view_data.double_data()) {
+        auto start_time = view_data.start_times().at(row.first);
+        absl::StrAppend(&output, "\nRow data from ",
+                        absl::FormatTime(start_time), " to ",
+                        absl::FormatTime(view_data.end_time()), ":\n");
         for (int i = 0; i < descriptor.columns().size(); ++i) {
           absl::StrAppend(&output, descriptor.columns()[i].name(), ":",
                           row.first[i], ", ");

opencensus/stats/internal/view_data.cc

@@ -81,6 +81,11 @@ const ViewData::DataMap<Distribution>& ViewData::distribution_data() const {
 }
 
 absl::Time ViewData::start_time() const { return impl_->start_time(); }
+
+const ViewData::DataMap<absl::Time>& ViewData::start_times() const {
+  return impl_->start_times();
+}
+
 absl::Time ViewData::end_time() const { return end_time_; }
 
 ViewData::ViewData(const ViewData& other)

opencensus/stats/internal/view_data_impl.cc

@@ -64,6 +64,7 @@ ViewDataImpl::ViewDataImpl(absl::Time start_time,
     : aggregation_(descriptor.aggregation()),
       aggregation_window_(descriptor.aggregation_window_),
       type_(TypeForDescriptor(descriptor)),
+      start_times_(),
       start_time_(start_time) {
   switch (type_) {
     case Type::kDouble: {
@@ -91,9 +92,18 @@ ViewDataImpl::ViewDataImpl(const ViewDataImpl& other, absl::Time now)
       type_(other.aggregation().type() == Aggregation::Type::kDistribution
                 ? Type::kDistribution
                 : Type::kDouble),
+      start_times_(),
       start_time_(std::max(other.start_time(),
                            now - other.aggregation_window().duration())) {
+  // Intentionally reset the source with a new start time.
+  for (const auto& it : other.start_times_) {
+    auto new_start_time =
+        std::max(it.second, now - other.aggregation_window().duration());
+    start_times_[it.first] = new_start_time;
+  }
+
   ABSL_ASSERT(aggregation_window_.type() == AggregationWindow::Type::kInterval);
+
   switch (aggregation_.type()) {
     case Aggregation::Type::kSum:
     case Aggregation::Type::kCount: {
@@ -156,6 +166,7 @@ ViewDataImpl::ViewDataImpl(const ViewDataImpl& other)
     : aggregation_(other.aggregation_),
       aggregation_window_(other.aggregation_window_),
       type_(other.type()),
+      start_times_(other.start_times_),
       start_time_(other.start_time_) {
   switch (type_) {
     case Type::kDouble: {
@@ -182,6 +193,8 @@ ViewDataImpl::ViewDataImpl(const ViewDataImpl& other)
 
 void ViewDataImpl::Merge(const std::vector<std::string>& tag_values,
                          const MeasureData& data, absl::Time now) {
+  // A value is set here. Set a start time if it is unset.
+  SetStartTimeIfUnset(tag_values, now);
   switch (type_) {
     case Type::kDouble: {
       if (aggregation_.type() == Aggregation::Type::kSum) {
@@ -256,6 +269,7 @@ ViewDataImpl::ViewDataImpl(ViewDataImpl* source, absl::Time now)
     : aggregation_(source->aggregation_),
       aggregation_window_(source->aggregation_window_),
       type_(source->type_),
+      start_times_(source->start_times_),
       start_time_(source->start_time_) {
   switch (type_) {
     case Type::kDouble: {
@@ -280,7 +294,12 @@ ViewDataImpl::ViewDataImpl(ViewDataImpl* source, absl::Time now)
       break;
     }
   }
+  // Intentionally reset the source with new start times.
   source->start_time_ = now;
+
+  for (const auto& it : source->start_times_) {
+    source->start_times_[it.first] = now;
+  }
 }
 
 }  // namespace stats

opencensus/stats/internal/view_data_impl.h

@@ -103,6 +103,12 @@ class ViewDataImpl {
     return interval_data_;
   }
 
+  // Returns a start time for each timeseries/tag map.
+  const DataMap<absl::Time>& start_times() const { return start_times_; }
+
+  // DEPRECATED: Legacy start_time_ for the entire view.
+  // This should be deleted if custom exporters are updated to
+  // use start_times_ and stop depending on this field.
   absl::Time start_time() const { return start_time_; }
 
   // Merges bulk data for the given tag values at 'now'. tag_values must be
@@ -120,6 +126,14 @@ class ViewDataImpl {
 
   Type TypeForDescriptor(const ViewDescriptor& descriptor);
 
+  void SetStartTimeIfUnset(const std::vector<std::string>& tag_values,
+                           absl::Time now) {
+    // If the time is not set.
+    if (start_times_.find(tag_values) == start_times_.end()) {
+      start_times_[tag_values] = now;
+    }
+  }
+
   const Aggregation aggregation_;
   const AggregationWindow aggregation_window_;
   const Type type_;
@@ -129,6 +143,13 @@ class ViewDataImpl {
     DataMap<Distribution> distribution_data_;
     DataMap<IntervalStatsObject> interval_data_;
   };
+
+  // A start time for each timeseries/tag map.
+  DataMap<absl::Time> start_times_;
+
+  // DEPRECATED: Legacy start_time_ for the entire view.
+  // This should be deleted if custom exporters are updated to
+  // use start_times_ and stop depending on this field
   absl::Time start_time_;
 };