Implementing configurable aggregation cardinality limit

CHANGELOG.md

@@ -33,6 +33,10 @@ Increment the:
 * [BUILD] Use -dev versions in main branch
   [#3609](https://github.com/open-telemetry/opentelemetry-cpp/pull/3609)
 
+* [SDK] Implementing configurable aggregation cardinality limit and collector
+  emitting all views instead of last view
+  [#3624](https://github.com/open-telemetry/opentelemetry-cpp/pull/3624)
+
 Important changes:
 
 * [CMAKE] Upgrade CMake minimum version to 3.16

ci/do_ci.sh

@@ -558,7 +558,7 @@ elif [[ "$1" == "bazel.tsan" ]]; then
   exit 0
 elif [[ "$1" == "bazel.valgrind" ]]; then
   bazel $BAZEL_STARTUP_OPTIONS build $BAZEL_OPTIONS_ASYNC //...
-  bazel $BAZEL_STARTUP_OPTIONS test --run_under="/usr/bin/valgrind --leak-check=full --error-exitcode=1 --errors-for-leak-kinds=definite --suppressions=\"${SRC_DIR}/ci/valgrind-suppressions\"" $BAZEL_TEST_OPTIONS_ASYNC //...
+  bazel $BAZEL_STARTUP_OPTIONS test --test_timeout=600 --run_under="/usr/bin/valgrind --leak-check=full --error-exitcode=1 --errors-for-leak-kinds=definite --suppressions=\"${SRC_DIR}/ci/valgrind-suppressions\"" $BAZEL_TEST_OPTIONS_ASYNC //...
   exit 0
 elif [[ "$1" == "bazel.e2e" ]]; then
   cd examples/e2e

sdk/include/opentelemetry/sdk/metrics/aggregation/aggregation_config.h

@@ -5,6 +5,7 @@
 
 #include <vector>
 
+#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -15,6 +16,10 @@ namespace metrics
 class AggregationConfig
 {
 public:
+  AggregationConfig(size_t cardinality_limit = kAggregationCardinalityLimit)
+      : cardinality_limit_(cardinality_limit)
+  {}
+  size_t cardinality_limit_;
   virtual ~AggregationConfig() = default;
 };
 

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

@@ -42,8 +42,13 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
                      const AggregationConfig *aggregation_config)
       : instrument_descriptor_(instrument_descriptor),
         aggregation_type_{aggregation_type},
-        cumulative_hash_map_(new AttributesHashMap()),
-        delta_hash_map_(new AttributesHashMap()),
+        aggregation_config_{aggregation_config},
+        cumulative_hash_map_(new AttributesHashMap(aggregation_config
+                                                       ? aggregation_config->cardinality_limit_
+                                                       : kAggregationCardinalityLimit)),
+        delta_hash_map_(new AttributesHashMap(aggregation_config
+                                                  ? aggregation_config->cardinality_limit_
+                                                  : kAggregationCardinalityLimit)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exempler_filter_type),
         exemplar_reservoir_(exemplar_reservoir),
@@ -124,7 +129,9 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     {
       std::lock_guard<opentelemetry::common::SpinLockMutex> guard(hashmap_lock_);
       delta_metrics = std::move(delta_hash_map_);
-      delta_hash_map_.reset(new AttributesHashMap);
+      delta_hash_map_.reset(new AttributesHashMap(aggregation_config_
+                                                      ? aggregation_config_->cardinality_limit_
+                                                      : kAggregationCardinalityLimit));
     }
 
     auto status =
@@ -136,6 +143,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
 private:
   InstrumentDescriptor instrument_descriptor_;
   AggregationType aggregation_type_;
+  const AggregationConfig *aggregation_config_;
   std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
   opentelemetry::common::SpinLockMutex hashmap_lock_;

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

@@ -49,7 +49,13 @@ class AttributesHashMapWithCustomHash
 public:
   AttributesHashMapWithCustomHash(size_t attributes_limit = kAggregationCardinalityLimit)
       : attributes_limit_(attributes_limit)
-  {}
+  {
+    if (attributes_limit_ > kAggregationCardinalityLimit)
+    {
+      hash_map_.reserve(attributes_limit_);
+    }
+  }
+
   Aggregation *Get(const MetricAttributes &attributes) const
   {
     auto it = hash_map_.find(attributes);

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

@@ -63,10 +63,12 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
                     ExemplarFilterType exempler_filter_type,
                     nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
 #endif
-                    const AggregationConfig *aggregation_config,
-                    size_t attributes_limit = kAggregationCardinalityLimit)
+                    const AggregationConfig *aggregation_config)
       : instrument_descriptor_(instrument_descriptor),
-        attributes_hashmap_(new AttributesHashMap(attributes_limit)),
+        aggregation_config_(aggregation_config),
+        attributes_hashmap_(new AttributesHashMap(aggregation_config
+                                                      ? aggregation_config->cardinality_limit_
+                                                      : kAggregationCardinalityLimit)),
         attributes_processor_(std::move(attributes_processor)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exempler_filter_type),
@@ -173,6 +175,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
 private:
   InstrumentDescriptor instrument_descriptor_;
   // hashmap to maintain the metrics for delta collection (i.e, collection since last Collect call)
+  const AggregationConfig *aggregation_config_;
   std::unique_ptr<AttributesHashMap> attributes_hashmap_;
   std::function<std::unique_ptr<Aggregation>()> create_default_aggregation_;
   std::shared_ptr<const AttributesProcessor> attributes_processor_;

sdk/src/metrics/state/sync_metric_storage.cc

@@ -9,6 +9,7 @@
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/span.h"
+#include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
@@ -35,7 +36,9 @@ bool SyncMetricStorage::Collect(CollectorHandle *collector,
   {
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
     delta_metrics = std::move(attributes_hashmap_);
-    attributes_hashmap_.reset(new AttributesHashMap);
+    attributes_hashmap_.reset(new AttributesHashMap(aggregation_config_
+                                                        ? aggregation_config_->cardinality_limit_
+                                                        : kAggregationCardinalityLimit));
   }
 
   return temporal_metric_storage_.buildMetrics(collector, collectors, sdk_start_ts, collection_ts,

sdk/src/metrics/state/temporal_metric_storage.cc

@@ -97,7 +97,9 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,
   }
   auto unreported_list = std::move(present->second);
   // Iterate over the unreporter metrics for `collector` and store result in `merged_metrics`
-  std::unique_ptr<AttributesHashMap> merged_metrics(new AttributesHashMap);
+  std::unique_ptr<AttributesHashMap> merged_metrics(
+      new AttributesHashMap(aggregation_config_ ? aggregation_config_->cardinality_limit_
+                                                : kAggregationCardinalityLimit));
   for (auto &agg_hashmap : unreported_list)
   {
     agg_hashmap->GetAllEnteries(

sdk/test/metrics/async_metric_storage_test.cc

@@ -25,11 +25,13 @@
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
-#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+#  include "opentelemetry/sdk/metrics/data/exemplar_data.h"
 #  include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
 #  include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
+#else
+#  include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 #endif
 
 using namespace opentelemetry::sdk::metrics;
@@ -192,8 +194,10 @@ TEST_P(WritableMetricStorageTestUpDownFixture, TestAggregation)
         }
         return true;
       });
-  // subsequent recording after collection shouldn't fail
-  // monotonic increasing values;
+  // Note: When the cardinality limit is set to n, the attributes hashmap emits n-1 distinct
+  // attribute sets, plus an overflow bucket for additional attributes. The test logic below is made
+  // generic to succeed for either n or n-1 total cardinality. If this behavior is unexpected,
+  // please investigate and file an issue.
   int64_t get_count2 = -50;
   int64_t put_count2 = -70;
 

sdk/test/metrics/cardinality_limit_test.cc

@@ -20,6 +20,7 @@
 #include "opentelemetry/nostd/span.h"
 #include "opentelemetry/nostd/variant.h"
 #include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
+#include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
 #include "opentelemetry/sdk/metrics/aggregation/sum_aggregation.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
 #include "opentelemetry/sdk/metrics/data/point_data.h"
@@ -110,14 +111,15 @@ TEST_P(WritableMetricStorageCardinalityLimitTestFixture, LongCounterSumAggregati
   const size_t attributes_limit   = 10;
   InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kCounter,
                                      InstrumentValueType::kLong};
+  AggregationConfig aggConfig(attributes_limit);
   std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   SyncMetricStorage storage(instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                             ExemplarFilterType::kAlwaysOff,
                             ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
-                            nullptr, attributes_limit);
+                            &aggConfig);
 
   int64_t record_value = 100;
   // add 9 unique metric points, and 6 more above limit.