Add MinMaxSumCount and Gauge Aggregators

sdk/include/opentelemetry/sdk/metrics/aggregator/gauge_aggregator.h

@@ -0,0 +1,138 @@
+#pragma once
+
+#include "opentelemetry/core/timestamp.h"
+#include "opentelemetry/metrics/instrument.h"
+#include "opentelemetry/sdk/metrics/aggregator/aggregator.h"
+#include "opentelemetry/version.h"
+
+#include <memory>
+#include <mutex>
+#include <vector>
+
+namespace metrics_api = opentelemetry::metrics;
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+/**
+ * This aggregator stores and maintains a vector of
+ * type T where the contents of the vector simply
+ * include the last value recorded to the aggregator.
+ * The aggregator also maintains a timestamp of when
+ * the last value was recorded.
+ *
+ * @tparam T the type of values stored in this aggregator.
+ */
+template <class T>
+class GaugeAggregator : public Aggregator<T>
+{
+public:
+  explicit GaugeAggregator<T>(metrics_api::InstrumentKind kind)
+  {
+    static_assert(std::is_arithmetic<T>::value, "Not an arithmetic type");
+    this->kind_        = kind;
+    this->values_      = std::vector<T>(1, 0);
+    this->checkpoint_  = this->values_;
+    this->agg_kind_    = AggregatorKind::Gauge;
+    current_timestamp_ = core::SystemTimestamp(std::chrono::system_clock::now());
+  }
+
+  ~GaugeAggregator() = default;
+
+  GaugeAggregator(const GaugeAggregator &cp)
+  {
+    this->values_      = cp.values_;
+    this->checkpoint_  = cp.checkpoint_;
+    this->kind_        = cp.kind_;
+    this->agg_kind_    = cp.agg_kind_;
+    current_timestamp_ = cp.current_timestamp_;
+    // use default initialized mutex as they cannot be copied
+  }
+
+  /**
+   * Receives a captured value from the instrument and applies it to the current aggregator value.
+   *
+   * @param val, the raw value used in aggregation
+   */
+  void update(T val) override
+  {
+    this->mu_.lock();
+    this->values_[0]   = val;
+    current_timestamp_ = core::SystemTimestamp(std::chrono::system_clock::now());
+    this->mu_.unlock();
+  }
+
+  /**
+   * Checkpoints the current value.  This function will overwrite the current checkpoint with the
+   * current value.
+   *
+   * @return none
+   */
+
+  void checkpoint() override
+  {
+    this->mu_.lock();
+
+    this->checkpoint_ = this->values_;
+
+    // Reset the values to default
+    this->values_[0]      = 0;
+    checkpoint_timestamp_ = current_timestamp_;
+    current_timestamp_    = core::SystemTimestamp(std::chrono::system_clock::now());
+
+    this->mu_.unlock();
+  }
+
+  /**
+   * Merges two Gauge aggregators together
+   *
+   * @param other the aggregator to merge with this aggregator
+   */
+  void merge(GaugeAggregator<T> other)
+  {
+    if (this->kind_ == other.kind_)
+    {
+      this->mu_.lock();
+      // First merge values
+      this->values_[0] = other.values_[0];
+      // Now merge checkpoints
+      this->checkpoint_[0] = other.checkpoint_[0];
+      current_timestamp_   = core::SystemTimestamp(std::chrono::system_clock::now());
+      this->mu_.unlock();
+    }
+    else
+    {
+      // Log error
+      return;
+    }
+  }
+
+  /**
+   * @return the value of the latest checkpoint
+   */
+  std::vector<T> get_checkpoint() override { return this->checkpoint_; }
+
+  /**
+   * @return the latest checkpointed timestamp
+   */
+  core::SystemTimestamp get_checkpoint_timestamp() override { return checkpoint_timestamp_; }
+
+  /**
+   * @return the values_ vector stored in this aggregator
+   */
+  std::vector<T> get_values() override { return this->values_; }
+
+  /**
+   * @return the timestamp of when the last value recorded
+   */
+  core::SystemTimestamp get_timestamp() { return current_timestamp_; }
+
+private:
+  core::SystemTimestamp current_timestamp_;
+  core::SystemTimestamp checkpoint_timestamp_;
+};
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE

sdk/include/opentelemetry/sdk/metrics/aggregator/min_max_sum_count_aggregator.h

@@ -0,0 +1,154 @@
+#pragma once
+
+#include "opentelemetry/metrics/instrument.h"
+#include "opentelemetry/sdk/metrics/aggregator/aggregator.h"
+#include "opentelemetry/version.h"
+
+#include <memory>
+#include <mutex>
+#include <vector>
+
+namespace metrics_api = opentelemetry::metrics;
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+const int MinValueIndex   = 0;
+const int MaxValueIndex   = 1;
+const int SumValueIndex   = 2;
+const int CountValueIndex = 3;
+/**
+ * This aggregator stores and maintains a vector of
+ * type T where the contents in the vector are made
+ * up of the minimum value recorded to this instrument,
+ * the maximum value, the sum of all values, and the
+ * count of all values.
+ *
+ * @tparam T the type of values stored in this aggregator.
+ */
+template <class T>
+class MinMaxSumCountAggregator : public Aggregator<T>
+{
+public:
+  explicit MinMaxSumCountAggregator(metrics_api::InstrumentKind kind)
+  {
+    static_assert(std::is_arithmetic<T>::value, "Not an arithmetic type");
+    this->kind_       = kind;
+    this->values_     = std::vector<T>(4, 0);  // {min, max, sum, count}
+    this->checkpoint_ = this->values_;
+    this->agg_kind_   = AggregatorKind::MinMaxSumCount;
+  }
+
+  ~MinMaxSumCountAggregator() = default;
+
+  MinMaxSumCountAggregator(const MinMaxSumCountAggregator &cp)
+  {
+    this->values_     = cp.values_;
+    this->checkpoint_ = cp.checkpoint_;
+    this->kind_       = cp.kind_;
+    this->agg_kind_   = cp.agg_kind_;
+    // use default initialized mutex as they cannot be copied
+  }
+
+  /**
+   * Receives a captured value from the instrument and applies it to the current aggregator value.
+   *
+   * @param val, the raw value used in aggregation
+   */
+  void update(T val) override
+  {
+    this->mu_.lock();
+
+    if (this->values_[CountValueIndex] == 0 || val < this->values_[MinValueIndex])  // set min
+      this->values_[MinValueIndex] = val;
+    if (this->values_[CountValueIndex] == 0 || val > this->values_[MaxValueIndex])  // set max
+      this->values_[MaxValueIndex] = val;
+
+    this->values_[SumValueIndex] += val;  // compute sum
+    this->values_[CountValueIndex]++;     // increment count
+
+    this->mu_.unlock();
+  }
+
+  /**
+   * Checkpoints the current value.  This function will overwrite the current checkpoint with the
+   * current value.
+   *
+   */
+  void checkpoint() override
+  {
+    this->mu_.lock();
+    this->checkpoint_ = this->values_;
+    // Reset the values
+    this->values_[MinValueIndex]   = 0;
+    this->values_[MaxValueIndex]   = 0;
+    this->values_[SumValueIndex]   = 0;
+    this->values_[CountValueIndex] = 0;
+    this->mu_.unlock();
+  }
+
+  /**
+   * Merges two MinMaxSumCount aggregators together
+   *
+   * @param other the aggregator to merge with this aggregator
+   */
+  void merge(const MinMaxSumCountAggregator &other)
+  {
+    if (this->kind_ == other.kind_)
+    {
+      this->mu_.lock();
+      // First merge values
+      // set min
+      if (this->values_[CountValueIndex] == 0 ||
+          other.values_[MinValueIndex] < this->values_[MinValueIndex])
+        this->values_[MinValueIndex] = other.values_[MinValueIndex];
+      // set max
+      if (this->values_[CountValueIndex] == 0 ||
+          other.values_[MaxValueIndex] > this->values_[MaxValueIndex])
+        this->values_[MaxValueIndex] = other.values_[MaxValueIndex];
+      // set sum
+      this->values_[SumValueIndex] += other.values_[SumValueIndex];
+      // set count
+      this->values_[CountValueIndex] += other.values_[CountValueIndex];
+
+      // Now merge checkpoints
+      if (this->checkpoint_[CountValueIndex] == 0 ||
+          other.checkpoint_[MinValueIndex] < this->checkpoint_[MinValueIndex])
+        this->checkpoint_[MinValueIndex] = other.checkpoint_[MinValueIndex];
+      // set max
+      if (this->checkpoint_[CountValueIndex] == 0 ||
+          other.checkpoint_[MaxValueIndex] > this->checkpoint_[MaxValueIndex])
+        this->checkpoint_[MaxValueIndex] = other.checkpoint_[MaxValueIndex];
+      // set sum
+      this->checkpoint_[SumValueIndex] += other.checkpoint_[SumValueIndex];
+      // set count
+      this->checkpoint_[CountValueIndex] += other.checkpoint_[CountValueIndex];
+
+      this->mu_.unlock();
+    }
+    else
+    {
+      // Log error
+      return;
+    }
+  }
+
+  /**
+   * Returns the checkpointed value
+   *
+   * @return the value of the checkpoint
+   */
+  std::vector<T> get_checkpoint() override { return this->checkpoint_; }
+
+  /**
+   * Returns the values currently held by the aggregator
+   *
+   * @return the values held by the aggregator
+   */
+  std::vector<T> get_values() override { return this->values_; }
+};
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE

sdk/test/metrics/BUILD

@@ -1,3 +1,25 @@
+cc_test(
+    name = "gauge_aggregator_test",
+    srcs = [
+        "gauge_aggregator_test.cc",
+    ],
+    deps = [
+        "//sdk/src/metrics",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
+cc_test(
+    name = "min_max_sum_count_aggregator_test",
+    srcs = [
+        "min_max_sum_count_aggregator_test.cc",
+    ],
+    deps = [
+        "//sdk/src/metrics",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "meter_provider_sdk_test",
     srcs = [

sdk/test/metrics/CMakeLists.txt

@@ -1,7 +1,7 @@
-foreach(testname meter_provider_sdk_test)
+foreach(testname meter_provider_sdk_test gauge_aggregator_test
+                 min_max_sum_count_aggregator_test)
   add_executable(${testname} "${testname}.cc")
-  target_link_libraries(
-    ${testname} ${GTEST_BOTH_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT}
-    opentelemetry_common opentelemetry_metrics)
+  target_link_libraries(${testname} ${GTEST_BOTH_LIBRARIES}
+                        ${CMAKE_THREAD_LIBS_INIT} opentelemetry_metrics)
   gtest_add_tests(TARGET ${testname} TEST_PREFIX metrics. TEST_LIST ${testname})
 endforeach()