Implementing configurable aggregation cardinality limit (open-telemetry#3624)

Author: ThomsonTan

CHANGELOG.md

@@ -33,6 +33,9 @@ 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
+  [#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,21 @@ namespace metrics
 class AggregationConfig
 {
 public:
+  AggregationConfig(size_t cardinality_limit = kAggregationCardinalityLimit)
+      : cardinality_limit_(cardinality_limit)
+  {}
+
+  static const AggregationConfig *GetOrDefault(const AggregationConfig *config)
+  {
+    if (config)
+    {
+      return config;
+    }
+    static const AggregationConfig default_config{};
+    return &default_config;
+  }
+
+  size_t cardinality_limit_;
   virtual ~AggregationConfig() = default;
 };
 

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

@@ -9,6 +9,7 @@
 
 #include "opentelemetry/nostd/shared_ptr.h"
 #include "opentelemetry/sdk/common/attributemap_hash.h"
+#include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
 #include "opentelemetry/sdk/metrics/aggregation/default_aggregation.h"
 
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
@@ -42,8 +43,11 @@ 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_{AggregationConfig::GetOrDefault(aggregation_config)},
+        cumulative_hash_map_(
+            std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
+        delta_hash_map_(
+            std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exempler_filter_type),
         exemplar_reservoir_(exemplar_reservoir),
@@ -124,7 +128,8 @@ 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_ =
+          std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_);
     }
 
     auto status =
@@ -136,6 +141,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

@@ -18,6 +18,7 @@
 #include "opentelemetry/nostd/string_view.h"
 #include "opentelemetry/sdk/common/attributemap_hash.h"
 #include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
+#include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
 #include "opentelemetry/sdk/metrics/aggregation/default_aggregation.h"
 #include "opentelemetry/sdk/metrics/aggregation/histogram_aggregation.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
@@ -63,10 +64,11 @@ 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_(AggregationConfig::GetOrDefault(aggregation_config)),
+        attributes_hashmap_(
+            std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
         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,7 @@ 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_->cardinality_limit_));
   }
 
   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.