Add delta aggregation as an internal interface. (#117)

Author: Ian Sturdy

opencensus/stats/internal/aggregation_window.cc

@@ -23,6 +23,8 @@ std::string AggregationWindow::DebugString() const {
   switch (type_) {
     case Type::kCumulative:
       return "Cumulative";
+    case Type::kDelta:
+      return "Delta";
     case Type::kInterval:
       return absl::StrCat("Interval (", absl::ToDoubleSeconds(duration_),
                           "s window)");

opencensus/stats/internal/aggregation_window.h

@@ -32,6 +32,12 @@ class AggregationWindow final {
     return AggregationWindow(Type::kCumulative, absl::InfiniteDuration());
   }
 
+  // Delta aggregation accumulates data until it is requested and then resets
+  // it, so that each recorded value appears in exactly one delta.
+  static AggregationWindow Delta() {
+    return AggregationWindow(Type::kDelta, absl::InfiniteDuration());
+  }
+
   // Interval aggregation keeps a rolling total of usage over the previous
   // 'interval' of time.
   static AggregationWindow Interval(absl::Duration interval) {
@@ -40,6 +46,7 @@ class AggregationWindow final {
 
   enum class Type {
     kCumulative,
+    kDelta,
     kInterval,
   };
 

opencensus/stats/internal/debug_string_test.cc

@@ -37,6 +37,7 @@ TEST(DebugStringTest, Aggregation) {
 
 TEST(DebugStringTest, AggregationWindow) {
   EXPECT_NE("", AggregationWindow::Cumulative().DebugString());
+  EXPECT_NE("", AggregationWindow::Delta().DebugString());
   EXPECT_NE("", AggregationWindow::Interval(absl::Minutes(1)).DebugString());
 }
 

opencensus/stats/internal/stats_manager.cc

@@ -15,8 +15,10 @@
 #include "opencensus/stats/internal/stats_manager.h"
 
 #include <iostream>
+#include <memory>
 
 #include "absl/base/macros.h"
+#include "absl/memory/memory.h"
 #include "absl/time/time.h"
 
 namespace opencensus {
@@ -73,12 +75,15 @@ void StatsManager::ViewInformation::Record(
   data_.Add(value, tag_values, now);
 }
 
-ViewDataImpl StatsManager::ViewInformation::GetData() const {
+std::unique_ptr<ViewDataImpl> StatsManager::ViewInformation::GetData() {
   absl::ReaderMutexLock l(mu_);
   if (data_.type() == ViewDataImpl::Type::kStatsObject) {
-    return ViewDataImpl(data_, absl::Now());
+    return absl::make_unique<ViewDataImpl>(data_, absl::Now());
+  } else if (descriptor_.aggregation_window_.type() ==
+             AggregationWindow::Type::kDelta) {
+    return data_.GetDeltaAndReset(absl::Now());
   } else {
-    return data_;
+    return absl::make_unique<ViewDataImpl>(data_);
   }
 }
 

opencensus/stats/internal/stats_manager.h

@@ -61,7 +61,7 @@ class StatsManager final {
         absl::Time now);
 
     // Retrieves a copy of the data.
-    ViewDataImpl GetData() const LOCKS_EXCLUDED(*mu_);
+    std::unique_ptr<ViewDataImpl> GetData() LOCKS_EXCLUDED(*mu_);
 
     const ViewDescriptor& view_descriptor() const { return descriptor_; }
 

opencensus/stats/internal/stats_manager_test.cc

@@ -136,6 +136,37 @@ TEST_F(StatsManagerTest, Distribution) {
               ::testing::ElementsAre(1, 0));
 }
 
+TEST_F(StatsManagerTest, Delta) {
+  ViewDescriptor view_descriptor = ViewDescriptor()
+                                       .set_measure(kFirstMeasureId)
+                                       .set_name("delta")
+                                       .set_aggregation(Aggregation::Count())
+                                       .add_column(key1_)
+                                       .add_column(key2_);
+  SetAggregationWindow(AggregationWindow::Delta(), &view_descriptor);
+  View view(view_descriptor);
+  ASSERT_EQ(ViewData::Type::kInt64, view.GetData().type());
+  EXPECT_TRUE(view.GetData().int_data().empty());
+  // Stats under a different measure should be ignored.
+  Record({{SecondMeasure(), 1}}, {});
+  EXPECT_TRUE(view.GetData().int_data().empty());
+
+  Record({{FirstMeasure(), 2.0}}, {});
+  Record({{FirstMeasure(), 3.0}}, {});
+  Record({{FirstMeasure(), 4.0}},
+         {{key1_, "value1"}, {key2_, "value2"}, {key3_, "value3"}});
+  EXPECT_THAT(
+      view.GetData().int_data(),
+      ::testing::UnorderedElementsAre(
+          ::testing::Pair(::testing::ElementsAre("", ""), 2),
+          ::testing::Pair(::testing::ElementsAre("value1", "value2"), 1)));
+
+  Record({{FirstMeasure(), 4.0}}, {{key1_, "new_value"}});
+  EXPECT_THAT(view.GetData().int_data(),
+              ::testing::UnorderedElementsAre(
+                  ::testing::Pair(::testing::ElementsAre("new_value", ""), 1)));
+}
+
 // TODO: Test window expiration if we add a simulated clock.
 TEST_F(StatsManagerTest, IntervalCount) {
   ViewDescriptor view_descriptor = ViewDescriptor()

opencensus/stats/internal/view.cc

@@ -15,6 +15,7 @@
 #include "opencensus/stats/view.h"
 
 #include <iostream>
+#include <utility>
 
 #include "absl/base/macros.h"
 #include "absl/memory/memory.h"
@@ -43,7 +44,7 @@ const ViewData View::GetData() {
     ABSL_ASSERT(0);
     return ViewData(absl::make_unique<ViewDataImpl>(absl::Now(), descriptor_));
   }
-  return ViewData((absl::make_unique<ViewDataImpl>(handle_->GetData())));
+  return ViewData(handle_->GetData());
 }
 
 }  // namespace stats

opencensus/stats/internal/view_data_impl.cc

@@ -16,7 +16,9 @@
 
 #include <cstdint>
 #include <iostream>
+#include <memory>
 
+#include "absl/memory/memory.h"
 #include "opencensus/stats/distribution.h"
 
 namespace opencensus {
@@ -26,6 +28,7 @@ ViewDataImpl::Type ViewDataImpl::TypeForDescriptor(
     const ViewDescriptor& descriptor) {
   switch (descriptor.aggregation_window_.type()) {
     case AggregationWindow::Type::kCumulative:
+    case AggregationWindow::Type::kDelta:
       switch (descriptor.aggregation().type()) {
         case Aggregation::Type::kSum:
           return ViewDataImpl::Type::kDouble;
@@ -123,6 +126,11 @@ ViewDataImpl::~ViewDataImpl() {
   }
 }
 
+std::unique_ptr<ViewDataImpl> ViewDataImpl::GetDeltaAndReset(absl::Time now) {
+  // Need to use wrap_unique because this is a private constructor.
+  return absl::WrapUnique(new ViewDataImpl(this, now));
+}
+
 ViewDataImpl::ViewDataImpl(const ViewDataImpl& other)
     : aggregation_(other.aggregation_),
       aggregation_window_(other.aggregation_window_),
@@ -201,5 +209,38 @@ void ViewDataImpl::Add(double value, const std::vector<std::string>& tag_values,
   }
 }
 
+ViewDataImpl::ViewDataImpl(ViewDataImpl* source, absl::Time now)
+    : aggregation_(source->aggregation_),
+      aggregation_window_(source->aggregation_window_),
+      type_(source->type_),
+      start_time_(source->start_time_),
+      end_time_(now) {
+  switch (type_) {
+    case Type::kDouble: {
+      new (&double_data_) DataMap<double>();
+      double_data_.swap(source->double_data_);
+      break;
+    }
+    case Type::kInt64: {
+      new (&int_data_) DataMap<int64_t>();
+      int_data_.swap(source->int_data_);
+      break;
+    }
+    case Type::kDistribution: {
+      new (&distribution_data_) DataMap<Distribution>();
+      distribution_data_.swap(source->distribution_data_);
+      break;
+    }
+    case Type::kStatsObject: {
+      std::cerr << "GetDeltaAndReset should not be called on ViewDataImpl for "
+                   "interval stats.";
+      ABSL_ASSERT(0);
+      break;
+    }
+  }
+  source->start_time_ = now;
+  source->end_time_ = now;
+}
+
 }  // namespace stats
 }  // namespace opencensus

opencensus/stats/internal/view_data_impl.h

@@ -15,6 +15,7 @@
 #ifndef OPENCENSUS_STATS_INTERNAL_VIEW_DATA_IMPL_H_
 #define OPENCENSUS_STATS_INTERNAL_VIEW_DATA_IMPL_H_
 
+#include <memory>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -62,6 +63,10 @@ class ViewDataImpl {
   ViewDataImpl(const ViewDataImpl& other);
   ~ViewDataImpl();
 
+  // Returns a copy of the present state of the object and resets data() and
+  // start_time().
+  std::unique_ptr<ViewDataImpl> GetDeltaAndReset(absl::Time now);
+
   const Aggregation& aggregation() const { return aggregation_; }
   const AggregationWindow& aggregation_window() const {
     return aggregation_window_;
@@ -108,6 +113,11 @@ class ViewDataImpl {
            absl::Time now);
 
  private:
+  // Implements GetDeltaAndReset(), copying aggregation_ and swapping data_ and
+  // start/end times. This is private so that it can be given a more descriptive
+  // name in the public API.
+  ViewDataImpl(ViewDataImpl* source, absl::Time now);
+
   Type TypeForDescriptor(const ViewDescriptor& descriptor);
 
   const Aggregation aggregation_;