Add view data expiry. (#452)

Fixes #451.

Author: bianpengyuan

opencensus/stats/internal/view_data_impl.cc

@@ -65,6 +65,7 @@ ViewDataImpl::ViewDataImpl(absl::Time start_time,
       aggregation_window_(descriptor.aggregation_window_),
       type_(TypeForDescriptor(descriptor)),
       start_times_(),
+      expiry_duration_(descriptor.expiry_duration_),
       start_time_(start_time) {
   switch (type_) {
     case Type::kDouble: {
@@ -195,6 +196,8 @@ 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);
+  SetUpdateTime(tag_values, now);
+  PurgeExpired(now);
   switch (type_) {
     case Type::kDouble: {
       if (aggregation_.type() == Aggregation::Type::kSum) {
@@ -302,5 +305,60 @@ ViewDataImpl::ViewDataImpl(ViewDataImpl* source, absl::Time now)
   }
 }
 
+void ViewDataImpl::SetUpdateTime(const std::vector<std::string>& tag_values,
+                                 absl::Time now) {
+  if (expiry_duration_ == absl::ZeroDuration()) {
+    // No need to track last update time if expiry duration is not set.
+    return;
+  }
+
+  auto update_time_map_iter = update_time_entries_.find(tag_values);
+  if (update_time_map_iter == update_time_entries_.end()) {
+    // The timeseries is not tracked, add it to the update time list and map.
+    update_times_.emplace_front(now, tag_values);
+    update_time_entries_[tag_values] = update_times_.begin();
+  } else {
+    // The timeseries is tracked, update its updated time and move it to the
+    // front of the list.
+    auto update_time_list_iter = update_time_map_iter->second;
+    update_time_list_iter->first = now;
+    update_times_.splice(update_times_.begin(), update_times_,
+                         update_time_list_iter);
+  }
+}
+
+void ViewDataImpl::PurgeExpired(absl::Time now) {
+  if (expiry_duration_ == absl::ZeroDuration() || update_times_.empty()) {
+    // No need to remove expired entries since either expiry is not set or there
+    // is no data entry yet.
+    return;
+  }
+  // Remove data that has not been updated for expiry.
+  while (now - update_times_.back().first > expiry_duration_) {
+    const auto& tags = update_times_.back().second;
+    update_time_entries_.erase(tags);
+    start_times_.erase(tags);
+    switch (type_) {
+      case Type::kDouble: {
+        double_data_.erase(tags);
+        break;
+      }
+      case Type::kInt64: {
+        int_data_.erase(tags);
+        break;
+      }
+      case Type::kDistribution: {
+        distribution_data_.erase(tags);
+        break;
+      }
+      case Type::kStatsObject: {
+        interval_data_.erase(tags);
+        break;
+      }
+    }
+    update_times_.pop_back();
+  }
+}
+
 }  // namespace stats
 }  // namespace opencensus

opencensus/stats/internal/view_data_impl.h

@@ -134,6 +134,14 @@ class ViewDataImpl {
     }
   }
 
+  // If expiry duration is set, keep track of the last update time of each view
+  // data.
+  void SetUpdateTime(const std::vector<std::string>& tag_values,
+                     absl::Time now);
+
+  // Purge view data that has not been updated for expiry duration.
+  void PurgeExpired(absl::Time now);
+
   const Aggregation aggregation_;
   const AggregationWindow aggregation_window_;
   const Type type_;
@@ -147,6 +155,16 @@ class ViewDataImpl {
   // A start time for each timeseries/tag map.
   DataMap<absl::Time> start_times_;
 
+  using UpdateTimeList =
+      std::list<std::pair<absl::Time, std::vector<std::string>>>;
+
+  // A list of last update time for each timeseries.
+  UpdateTimeList update_times_;
+  // A map from view data tags to the last update time list iterator.
+  DataMap<UpdateTimeList::iterator> update_time_entries_;
+
+  const absl::Duration expiry_duration_;
+
   // 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

opencensus/stats/internal/view_data_impl_test.cc

@@ -301,6 +301,28 @@ TEST(ViewDataImplTest, StatsObjectToDistribution) {
   EXPECT_THAT(distribution_2_2.bucket_counts(), ::testing::ElementsAre(0, 0));
 }
 
+TEST(ViewDataImplTest, ViewDataExpiry) {
+  const absl::Time start_time = absl::UnixEpoch();
+  const auto descriptor = ViewDescriptor()
+                              .set_aggregation(Aggregation::Sum())
+                              .set_expiry_duration(absl::Seconds(1));
+  ViewDataImpl data(start_time, descriptor);
+  const std::vector<std::string> tags1({"value1", "value2a"});
+  const std::vector<std::string> tags2({"value1", "value2b"});
+
+  AddToViewDataImpl(1, tags1, start_time, {}, &data);
+  AddToViewDataImpl(1, tags2, start_time, {}, &data);
+
+  EXPECT_THAT(data.double_data(),
+              ::testing::UnorderedElementsAre(::testing::Pair(tags1, 1),
+                                              ::testing::Pair(tags2, 1)));
+
+  AddToViewDataImpl(1, tags1, start_time + absl::Seconds(2), {}, &data);
+
+  EXPECT_THAT(data.double_data(),
+              ::testing::UnorderedElementsAre(::testing::Pair(tags1, 2)));
+}
+
 }  // namespace
 }  // namespace stats
 }  // namespace opencensus

opencensus/stats/internal/view_descriptor.cc

@@ -74,6 +74,12 @@ ViewDescriptor& ViewDescriptor::set_description(absl::string_view description) {
   return *this;
 }
 
+ViewDescriptor& ViewDescriptor::set_expiry_duration(
+    absl::Duration expiry_duration) {
+  expiry_duration_ = expiry_duration;
+  return *this;
+}
+
 void ViewDescriptor::RegisterForExport() const {
   if (aggregation_window_.type() == AggregationWindow::Type::kCumulative) {
     StatsExporterImpl::Get()->AddView(*this);
@@ -93,14 +99,16 @@ std::string ViewDescriptor::DebugString() const {
                     [](std::string* out, opencensus::tags::TagKey key) {
                       return out->append(key.name());
                     }),
-      "\n  description: \"", description_, "\"");
+      "\n  description: \"", description_, "\"",
+      "\n  expiry duration: ", absl::FormatDuration(expiry_duration_));
 }
 
 bool ViewDescriptor::operator==(const ViewDescriptor& other) const {
   return name_ == other.name_ && measure_id_ == other.measure_id_ &&
          aggregation_ == other.aggregation_ &&
          aggregation_window_ == other.aggregation_window_ &&
-         columns_ == other.columns_ && description_ == other.description_;
+         columns_ == other.columns_ && description_ == other.description_ &&
+         expiry_duration_ == other.expiry_duration_;
 }
 
 }  // namespace stats

opencensus/stats/view_descriptor.h

@@ -82,6 +82,11 @@ class ViewDescriptor final {
   ViewDescriptor& set_description(absl::string_view description);
   const std::string& description() const { return description_; }
 
+  // Sets the desired lifetime of the view's data. If the view data is not
+  // updated at any point during this duration, it will be removed from the set
+  // of data recorded by the view when periodic aggregation happens.
+  ViewDescriptor& set_expiry_duration(absl::Duration expiry_duration);
+
   //////////////////////////////////////////////////////////////////////////////
   // View registration
 
@@ -113,6 +118,7 @@ class ViewDescriptor final {
   AggregationWindow aggregation_window_;
   std::vector<opencensus::tags::TagKey> columns_;
   std::string description_;
+  absl::Duration expiry_duration_;
 };
 
 }  // namespace stats