Add Aggregation Storage

sdk/include/opentelemetry/sdk/metrics/data/point_data.h

@@ -5,6 +5,7 @@
 #ifndef ENABLE_METRICS_PREVIEW
 #  include "opentelemetry/common/timestamp.h"
 #  include "opentelemetry/nostd/variant.h"
+#  include "opentelemetry/sdk/metrics/instruments.h"
 #  include "opentelemetry/version.h"
 
 #  include <vector>
@@ -18,13 +19,6 @@ namespace metrics
 using ValueType = nostd::variant<long, double>;
 using ListType  = nostd::variant<std::vector<long>, std::vector<double>>;
 
-enum class AggregationTemporarily
-{
-  kUnspecified,
-  kDelta,
-  kCummulative
-};
-
 class SumPointData
 {
 public:

sdk/include/opentelemetry/sdk/metrics/instruments.h

@@ -3,7 +3,7 @@
 
 #pragma once
 #ifndef ENABLE_METRICS_PREVIEW
-#  include "opentelemetry/nostd/string_view.h"
+#  include "opentelemetry/sdk/common/attribute_utils.h"
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
 {
@@ -27,6 +27,22 @@ enum class InstrumentValueType
   kDouble
 };
 
+enum class AggregationType
+{
+  kDrop,
+  kHistogram,
+  kLastValue,
+  kSum,
+  kDefault
+};
+
+enum class AggregationTemporarily
+{
+  kUnspecified,
+  kDelta,
+  kCummulative
+};
+
 struct InstrumentDescriptor
 {
   std::string name_;
@@ -36,6 +52,15 @@ struct InstrumentDescriptor
   InstrumentValueType value_type_;
 };
 
+using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
+
+// TBD -> Remove once MetricCollector is imoplemeted
+class MetricCollector
+{
+public:
+  AggregationTemporarily aggregation_temporarily_;
+};
+
 /*class InstrumentSelector {
 public:
 InstrumentSelector(opentelemetry::nostd::string_view name,

sdk/include/opentelemetry/sdk/metrics/measurement_processor.h

@@ -21,6 +21,7 @@ static std::size_t MakeKey(const MetricReader &metric_reader)
 {
   return reinterpret_cast<std::size_t>(&metric_reader);
 }
+
 class MeasurementProcessor
 {
 public:
@@ -41,12 +42,15 @@ class MeasurementProcessor
 
 class DefaultMeasurementProcessor : public MeasurementProcessor
 {
-
 public:
   bool AddMetricStorage(const MetricReader &reader)
   {
-    // TBD = check if already present.
-    metric_storages_[MakeKey(reader)] = std::unique_ptr<SyncMetricStorage>(new SyncMetricStorage());
+    // TBD Check if already present
+    // pass intrumentation type, and aggregation type instead of hardcodig below.
+    InstrumentDescriptor instr_desc   = {"name", "desc", "1unit", InstrumentType::kCounter,
+                                       InstrumentValueType::kLong};
+    metric_storages_[MakeKey(reader)] = std::unique_ptr<SyncMetricStorage>(
+        new SyncMetricStorage(instr_desc, AggregationType::kSum));
     return true;
   }
 
@@ -91,9 +95,13 @@ class DefaultMeasurementProcessor : public MeasurementProcessor
                nostd::function_ref<bool(MetricData)> callback) noexcept override
   {
     auto i = metric_storages_.find(MakeKey(reader));
+
+    // TBD - Remove hardcodings below
+    std::vector<MetricCollector *> collectors;
     if (i != metric_storages_.end())
     {
-      return i->second->Collect(aggregation_temporarily, callback);
+
+      return i->second->Collect(nullptr, collectors, nullptr, nullptr, callback);
     }
     return false;
   }

sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h

@@ -0,0 +1,127 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/common/spin_lock_mutex.h"
+#  include "opentelemetry/nostd/function_ref.h"
+#  include "opentelemetry/sdk/common/attribute_utils.h"
+#  include "opentelemetry/sdk/common/attributemap_hash.h"
+#  include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
+#  include "opentelemetry/sdk/metrics/instruments.h"
+#  include "opentelemetry/version.h"
+
+#  include <functional>
+#  include <memory>
+#  include <mutex>
+#  include <unordered_map>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+using opentelemetry::sdk::common::OrderedAttributeMap;
+
+class AttributeHashGenerator
+{
+public:
+  size_t operator()(const MetricAttributes &attributes) const
+  {
+    return opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
+  }
+};
+
+class AttributesHashMap
+{
+public:
+  Aggregation *Get(const MetricAttributes &attributes) const
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    auto it = hash_map_.find(attributes);
+    if (it != hash_map_.end())
+    {
+      return it->second.get();
+    }
+    return nullptr;
+  }
+
+  /**
+   * @return check if key is present in hash
+   *
+   */
+  bool Has(const MetricAttributes &attributes) const
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    return (hash_map_.find(attributes) == hash_map_.end()) ? false : true;
+  }
+
+  /**
+   * @return the pointer to value for given key if present.
+   * If not present, it uses the provided callback to generate
+   * value and store in the hash
+   */
+  Aggregation *GetOrSetDefault(const MetricAttributes &attributes,
+                               std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    auto it = hash_map_.find(attributes);
+    if (it != hash_map_.end())
+    {
+      return it->second.get();
+    }
+
+    hash_map_[attributes] = std::move(aggregation_callback());
+    return hash_map_[attributes].get();
+  }
+
+  /**
+   * Set the value for given key, overwriting the value if already present
+   */
+  void Set(const MetricAttributes &attributes, std::unique_ptr<Aggregation> value)
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    hash_map_[attributes] = std::move(value);
+  }
+
+  /**
+   * Iterate the hash to yield key and value stored in hash.
+   */
+  bool GetAllEnteries(
+      nostd::function_ref<bool(const MetricAttributes &, Aggregation &)> callback) const
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    for (auto &kv : hash_map_)
+    {
+      if (!callback(kv.first, *(kv.second.get())))
+      {
+        return false;  // callback is not prepared to consume data
+      }
+    }
+    return true;
+  }
+
+  /**
+   * Return the size of hash.
+   */
+  size_t Size()
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
+    return hash_map_.size();
+  }
+
+private:
+  std::unordered_map<MetricAttributes, std::unique_ptr<Aggregation>, AttributeHashGenerator>
+      hash_map_;
+
+  static opentelemetry::common::SpinLockMutex &GetLock() noexcept
+  {
+    static opentelemetry::common::SpinLockMutex lock;
+    return lock;
+  }
+};
+}  // namespace metrics
+
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h

@@ -18,8 +18,12 @@ class MetricStorage
 {
 public:
   /* collect the metrics from this storage */
-  virtual bool Collect(AggregationTemporarily aggregation_temporarily,
-                       nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
+  virtual bool Collect(
+      MetricCollector *collector,
+      nostd::span<MetricCollector *> collectors,
+      opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
+      opentelemetry::sdk::resource::Resource *resource,
+      nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
 };
 
 class WritableMetricStorage
@@ -39,8 +43,12 @@ class WritableMetricStorage
 class NoopMetricStorage : public MetricStorage
 {
 public:
-  bool Collect(AggregationTemporarily aggregation_temporarily,
-               nostd::function_ref<bool(MetricData)> callback) noexcept override
+  bool Collect(
+      MetricCollector *collector,
+      nostd::span<MetricCollector *> collectors,
+      opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
+      opentelemetry::sdk::resource::Resource *resource,
+      nostd::function_ref<bool(MetricData)> callback) noexcept override
   {
     MetricData metric_data;
     if (callback(metric_data))