[Not For Review] Metrics Aggregation SDK

api/include/opentelemetry/common/timestamp.h

@@ -45,6 +45,18 @@ class SystemTimestamp
       : SystemTimestamp{time_point.time_since_epoch()}
   {}
 
+  /**
+   * @brief Reset a system timestamp based on a point in time.
+   *
+   * @param time_point A point in time.
+   */
+  void Reset(const std::chrono::system_clock::time_point &time_point) noexcept
+  {
+    nanos_since_epoch_ = static_cast<int64_t>(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(time_point.time_since_epoch())
+            .count());
+  }
+
   /**
    * @brief Returns a time point for the time stamp.
    *

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

@@ -0,0 +1,180 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/common/timestamp.h"
+#  include "opentelemetry/sdk/metrics/data/point_data.h"
+#  include "opentelemetry/version.h"
+
+#  include <chrono>
+#  include <vector>
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+class Accumulation
+{
+public:
+  virtual void Record(long value) noexcept   = 0;
+  virtual void Record(double value) noexcept = 0;
+};
+
+class LongSumAccumulation : public Accumulation
+{
+public:
+  LongSumAccumulation(long value = 0) : sum_(value) {}
+
+  void Record(long value) noexcept override { sum_ += value; }
+  void Record(double value) noexcept override {}
+
+  LongSingularPointData ToPointData() { return LongSingularPointData(sum_); }
+
+private:
+  long sum_;
+};
+
+class DoubleSumAccumulation : public Accumulation
+{
+public:
+  DoubleSumAccumulation(double value = 0) : sum_(value) {}
+
+  void Record(double value) noexcept override { sum_ += value; }
+  void Record(long value) noexcept override {}
+
+  DoubleSingularPointData ToPointData() { return DoubleSingularPointData(sum_); }
+
+private:
+  double sum_;
+};
+
+class LongLastValueAccumulation : public Accumulation
+{
+public:
+  LongLastValueAccumulation(
+      long value                                = 0,
+      opentelemetry::common::SystemTimestamp ts = std::chrono::system_clock::now())
+      : last_value_(value), ts_{ts}
+  {}
+
+  void Record(long value) noexcept override
+  {
+    last_value_ = value;
+    ts_.Reset(std::chrono::system_clock::now());
+  }
+
+  void Record(double value) noexcept override {}
+
+  const opentelemetry::common::SystemTimestamp &GetLastValueTimeStamp() const { return ts_; }
+
+  LongSingularPointData ToPointData() { return LongSingularPointData(last_value_); }
+
+private:
+  long last_value_;
+  opentelemetry::common::SystemTimestamp ts_;
+};
+
+class DoubleLastValueAccumulation : public Accumulation
+{
+public:
+  DoubleLastValueAccumulation(
+      long value                                = 0,
+      opentelemetry::common::SystemTimestamp ts = std::chrono::system_clock::now())
+      : last_value_(value), ts_{ts}
+  {}
+
+  void Record(double value) noexcept override
+  {
+    last_value_ = value;
+    ts_.Reset(std::chrono::system_clock::now());
+  }
+
+  void Record(long value) noexcept override {}
+
+  const opentelemetry::common::SystemTimestamp &GetLastValueTimeStamp() const { return ts_; }
+
+  DoubleSingularPointData ToPointData() { return DoubleSingularPointData(last_value_); }
+
+private:
+  double last_value_;
+  opentelemetry::common::SystemTimestamp ts_;
+};
+
+class LongHistogramAccumulation : public Accumulation
+{
+public:
+  LongHistogramAccumulation(std::vector<long> &boundaries) : histogram_(boundaries) {}
+
+  void Record(long value) noexcept override
+  {
+    histogram_.count_ += 1;
+    histogram_.sum_ += value;
+    for (auto it = histogram_.boundaries_.begin(); it != histogram_.boundaries_.end(); ++it)
+    {
+      if (value < *it)
+      {
+        histogram_.counts_[std::distance(histogram_.boundaries_.begin(), it)] += 1;
+        return;
+      }
+    }
+  }
+
+  void Record(double value) noexcept override {}
+
+  LongHistogramPointData &ToPointData() { return histogram_; }
+
+private:
+  LongHistogramPointData histogram_;
+};
+
+class DoubleHistogramAccumulation : public Accumulation
+{
+public:
+  DoubleHistogramAccumulation(std::vector<double> &boundaries) : histogram_(boundaries) {}
+
+  void Record(double value) noexcept override
+  {
+    histogram_.count_ += 1;
+    histogram_.sum_ += value;
+    for (auto it = histogram_.boundaries_.begin(); it != histogram_.boundaries_.end(); ++it)
+    {
+      if (value < *it)
+      {
+        histogram_.counts_[std::distance(histogram_.boundaries_.begin(), it)] += 1;
+        return;
+      }
+    }
+  }
+
+  void Record(long value) noexcept override {}
+
+  DoubleHistogramPointData &ToPointData() { return histogram_; }
+
+private:
+  DoubleHistogramPointData histogram_;
+};
+
+class DropAccumulation : public Accumulation
+{
+public:
+  DropAccumulation() {}
+
+  void Record(long value) noexcept override {}
+  void Record(double value) noexcept override {}
+
+  DropPointData ToPointData() { return DropPointData(); }
+};
+
+template <class T>
+struct AccumulationRecord
+{
+  T accumulation;
+  opentelemetry::sdk::common::AttributeMap attributes;
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

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

@@ -3,6 +3,8 @@
 
 #pragma once
 #ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/sdk/metrics/aggregator/accumulation.h"
+#  include "opentelemetry/sdk/metrics/instruments.h"
 #  include "opentelemetry/version.h"
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
@@ -12,14 +14,15 @@ namespace metrics
 
 class Aggregator
 {
-  // TBD
-};
+public:
+  virtual std::unique_ptr<Accumulation> CreateAccumulation() noexcept = 0;
 
-class NoOpAggregator : public Aggregator
-{
-  // TBD
-};
+  virtual std::unique_ptr<Accumulation> Merge(Accumulation &prev,
+                                              Accumulation &current) noexcept = 0;
 
+  virtual std::unique_ptr<Accumulation> diff(Accumulation &prev,
+                                             Accumulation &current) noexcept = 0;
+};
 }  // namespace metrics
 }  // namespace sdk
 OPENTELEMETRY_END_NAMESPACE

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

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/sdk/metrics/aggregator/accumulation.h"
+#  include "opentelemetry/sdk/metrics/aggregator/aggregator.h"
+#  include "opentelemetry/sdk/metrics/data/metric_data.h"
+#  include "opentelemetry/version.h"
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+/** Aggregators to drop the instruments. */
+class DropAggregator : public Aggregator
+{
+public:
+  std::unique_ptr<Accumulation> CreateAccumulation() noexcept override
+  {
+    return std::move(std::unique_ptr<Accumulation>(new DropAccumulation()));
+  }
+
+  /** Returns the result of the merge of the given accumulations.*/
+  std::unique_ptr<Accumulation> Merge(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    return std::move(
+        std::unique_ptr<Accumulation>(std::unique_ptr<Accumulation>(new DropAccumulation())));
+  }
+
+  /** Returns a new delta aggregation by comparing two cumulative measurements.*/
+  std::unique_ptr<Accumulation> diff(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    return std::move(
+        std::unique_ptr<Accumulation>(std::unique_ptr<Accumulation>(new DropAccumulation())));
+  }
+
+  DropMetricData ToMetricData(
+      opentelemetry::sdk::resource::Resource *resource,
+      opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
+      opentelemetry::sdk::metrics::InstrumentDescriptor &instrument_descriptor,
+      std::vector<opentelemetry::sdk::metrics::AccumulationRecord<DropAccumulation>>
+          &accumulation_by_attributes,
+      opentelemetry::common::SystemTimestamp &start_epoch_ns,
+      opentelemetry::common::SystemTimestamp &end_epoch_ns)
+  {
+    DropMetricData metrics_data;
+    return metrics_data;
+  }
+};
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

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

@@ -0,0 +1,108 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/sdk/metrics/aggregator/accumulation.h"
+#  include "opentelemetry/sdk/metrics/aggregator/aggregator.h"
+#  include "opentelemetry/sdk/metrics/data/metric_data.h"
+#  include "opentelemetry/version.h"
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+/**
+ * Basic aggregator which observes events and counts them in pre-defined buckets
+ * and provides the total sum and count of all observations.
+ */
+class LongHistogramAggregator : public Aggregator
+{
+  // TBD - This class is placeholder, and needs to be implemented
+public:
+  LongHistogramAggregator(const std::vector<long> &boundaries) : boundaries_(boundaries) {}
+  std::unique_ptr<Accumulation> CreateAccumulation() noexcept override
+  {
+    // TBD
+    return std::move(std::unique_ptr<Accumulation>(new LongHistogramAccumulation(boundaries_)));
+  }
+
+  /** Returns the result of the merge of the given accumulations.*/
+  std::unique_ptr<Accumulation> Merge(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    // TBD
+    return std::move(std::unique_ptr<Accumulation>(new LongHistogramAccumulation(boundaries_)));
+  }
+
+  /** Returns a new delta aggregation by comparing two cumulative measurements.*/
+  std::unique_ptr<Accumulation> diff(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    return std::move(std::unique_ptr<Accumulation>(new LongHistogramAccumulation(boundaries_)));
+  }
+
+  LongHistogramMetricData ToMetricData(
+      opentelemetry::sdk::resource::Resource *resource,
+      opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
+      opentelemetry::sdk::metrics::InstrumentDescriptor &instrument_descriptor,
+      std::vector<opentelemetry::sdk::metrics::AccumulationRecord<LongSumAccumulation>>
+          &accumulation_by_attributes,
+      opentelemetry::common::SystemTimestamp &start_epoch_ns,
+      opentelemetry::common::SystemTimestamp &end_epoch_ns)
+  {
+    LongHistogramMetricData metrics_data = {resource, instrumentation_library,
+                                            instrument_descriptor};
+    // TBD - Populate me
+    return metrics_data;
+  }
+
+private:
+  std::vector<long> boundaries_;
+};
+
+class DoubleHistogramAggregator : public Aggregator
+{
+  // TBD - This class is placeholder, and needs to be implemented
+public:
+  DoubleHistogramAggregator(const std::vector<double> &boundaries) : boundaries_(boundaries) {}
+  std::unique_ptr<Accumulation> CreateAccumulation() noexcept override
+  {
+    return std::move(std::unique_ptr<Accumulation>(new DoubleHistogramAccumulation(boundaries_)));
+  }
+
+  /** Returns the result of the merge of the given accumulations.*/
+  std::unique_ptr<Accumulation> Merge(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    // TBD
+    return std::move(std::unique_ptr<Accumulation>(new DoubleHistogramAccumulation(boundaries_)));
+  }
+
+  /** Returns a new delta aggregation by comparing two cumulative measurements.*/
+  std::unique_ptr<Accumulation> diff(Accumulation &prev, Accumulation &current) noexcept override
+  {
+    // TBD
+    return std::move(std::unique_ptr<Accumulation>(new DoubleHistogramAccumulation(boundaries_)));
+  }
+
+  DoubleHistogramMetricData ToMetricData(
+      opentelemetry::sdk::resource::Resource *resource,
+      opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
+      opentelemetry::sdk::metrics::InstrumentDescriptor &instrument_descriptor,
+      std::vector<opentelemetry::sdk::metrics::AccumulationRecord<DoubleSumAccumulation>>
+          &accumulation_by_attributes,
+      opentelemetry::common::SystemTimestamp &start_epoch_ns,
+      opentelemetry::common::SystemTimestamp &end_epoch_ns)
+  {
+    DoubleHistogramMetricData metrics_data = {resource, instrumentation_library,
+                                              instrument_descriptor};
+    // TBD - Populate me
+    return metrics_data;
+  }
+
+private:
+  std::vector<double> boundaries_;
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif