open-telemetry · ThomsonTan · Sep 11, 2025 · Mar 12, 2025 · Mar 12, 2025 · Mar 12, 2025
@@ -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

@@ -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

@@ -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;
 };
 

@@ -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_;

@@ -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);

@@ -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_;

@@ -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,

@@ -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(

@@ -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;

@@ -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.

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include <gtest/gtest.h>
+#include <stddef.h>
 #include <initializer_list>
+#include <map>
 #include <string>
 #include <unordered_map>
 #include <utility>
@@ -26,6 +28,7 @@
 #include "opentelemetry/sdk/metrics/meter_provider.h"
 #include "opentelemetry/sdk/metrics/metric_reader.h"
 #include "opentelemetry/sdk/metrics/push_metric_exporter.h"
+#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 #include "opentelemetry/sdk/metrics/view/instrument_selector.h"
 #include "opentelemetry/sdk/metrics/view/meter_selector.h"
@@ -142,6 +145,85 @@ TEST(HistogramToSumFilterAttributes, Double)
   });
 }
 
+TEST(HistogramToSumFilterAttributesWithCardinaityLimit, Double)
+{
+  MeterProvider mp;
+  auto m                      = mp.GetMeter("meter1", "version1", "schema1");
+  std::string instrument_unit = "ms";
+  std::string instrument_name = "historgram1";
+  std::string instrument_desc = "histogram metrics";
+  size_t cardinality_limit    = 10000;
+
+  std::unordered_map<std::string, bool> allowedattr;
+  allowedattr["attr1"] = true;
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::FilteringAttributesProcessor(allowedattr)};
+
+  std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig> dummy_aggregation_config{
+      new opentelemetry::sdk::metrics::AggregationConfig(cardinality_limit)};
+  std::unique_ptr<MockMetricExporter> exporter(new MockMetricExporter());
+  std::shared_ptr<MetricReader> reader{new MockMetricReader(std::move(exporter))};
+  mp.AddMetricReader(reader);
+
+  std::unique_ptr<View> view{new View("view1", "view1_description", AggregationType::kSum,
+                                      dummy_aggregation_config, std::move(attrproc))};
+  std::unique_ptr<InstrumentSelector> instrument_selector{
+      new InstrumentSelector(InstrumentType::kHistogram, instrument_name, instrument_unit)};
+  std::unique_ptr<MeterSelector> meter_selector{new MeterSelector("meter1", "version1", "schema1")};
+  mp.AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view));
+
+  auto h = m->CreateDoubleHistogram(instrument_name, instrument_desc, instrument_unit);
+  size_t total_metrics_times = 5;
+
+  size_t agg_repeat_count = 5;
+  for (size_t repeat = 0; repeat < agg_repeat_count; repeat++)
+  {
+
+    for (size_t times = 0; times < total_metrics_times; times++)
+    {
+      for (size_t i = 0; i < 2 * cardinality_limit; i++)
+      {
+        std::unordered_map<std::string, std::string> attr = {{"attr1", std::to_string(i)},
+                                                             {"attr2", "val2"}};
+        h->Record(1, attr, opentelemetry::context::Context{});
+      }
+    }
+
+    reader->Collect([&](ResourceMetrics &rm) {
+      for (const ScopeMetrics &smd : rm.scope_metric_data_)
+      {
+        for (const MetricData &md : smd.metric_data_)
+        {
+          // Something weird about attributes hashmap. If cardinality is setup to n, it emits n-1
+          // including overflow. Just making the logic generic here to succeed for n or n-1 total
+          // cardinality.
+          EXPECT_GE(cardinality_limit, md.point_data_attr_.size());
+          EXPECT_LT(cardinality_limit / 2, md.point_data_attr_.size());
+          for (size_t i = 0; i < md.point_data_attr_.size(); i++)
+          {
+            EXPECT_EQ(1, md.point_data_attr_[i].attributes.size());
+            if (md.point_data_attr_[i].attributes.end() !=
+                md.point_data_attr_[i].attributes.find("attr1"))
+            {
+              EXPECT_EQ(total_metrics_times * (repeat + 1),
+                        opentelemetry::nostd::get<double>(opentelemetry::nostd::get<SumPointData>(
+                                                              md.point_data_attr_[i].point_data)
+                                                              .value_));
+            }
+            else
+            {
+              EXPECT_NE(md.point_data_attr_[i].attributes.end(),
+                        md.point_data_attr_[i].attributes.find(
+                            sdk::metrics::kAttributesLimitOverflowKey));
+            }
+          }
+        }
+      }
+      return true;
+    });
+  }
+}
+
 TEST(CounterToSum, Double)
 {
   MeterProvider mp;
@@ -244,6 +326,85 @@ TEST(CounterToSumFilterAttributes, Double)
   });
 }
 
+TEST(CounterToSumFilterAttributesWithCardinalityLimit, Double)
+{
+  MeterProvider mp;
+  auto m                      = mp.GetMeter("meter1", "version1", "schema1");
+  std::string instrument_unit = "ms";
+  std::string instrument_name = "counter1";
+  std::string instrument_desc = "counter metrics";
+  size_t cardinality_limit    = 10000;
+
+  std::unordered_map<std::string, bool> allowedattr;
+  allowedattr["attr1"] = true;
+  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attrproc{
+      new opentelemetry::sdk::metrics::FilteringAttributesProcessor(allowedattr)};
+
+  std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig> dummy_aggregation_config{
+      new opentelemetry::sdk::metrics::AggregationConfig(cardinality_limit)};
+  std::unique_ptr<MockMetricExporter> exporter(new MockMetricExporter());
+  std::shared_ptr<MetricReader> reader{new MockMetricReader(std::move(exporter))};
+  mp.AddMetricReader(reader);
+
+  std::unique_ptr<View> view{new View("view1", "view1_description", AggregationType::kSum,
+                                      dummy_aggregation_config, std::move(attrproc))};
+  std::unique_ptr<InstrumentSelector> instrument_selector{
+      new InstrumentSelector(InstrumentType::kCounter, instrument_name, instrument_unit)};
+  std::unique_ptr<MeterSelector> meter_selector{new MeterSelector("meter1", "version1", "schema1")};
+  mp.AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view));
+
+  auto c = m->CreateDoubleCounter(instrument_name, instrument_desc, instrument_unit);
+
+  size_t agg_repeat_count = 5;
+  for (size_t repeat = 0; repeat < agg_repeat_count; repeat++)
+  {
+    size_t total_metrics_times = 5;
+
+    for (size_t times = 0; times < total_metrics_times; times++)
+    {
+      for (size_t i = 0; i < 2 * cardinality_limit; i++)
+      {
+        std::unordered_map<std::string, std::string> attr = {{"attr1", std::to_string(i)},
+                                                             {"attr2", "val2"}};
+        c->Add(1, attr, opentelemetry::context::Context{});
+      }
+    }
+
+    reader->Collect([&](ResourceMetrics &rm) {
+      for (const ScopeMetrics &smd : rm.scope_metric_data_)
+      {
+        for (const MetricData &md : smd.metric_data_)
+        {
+          // Something weird about attributes hashmap. If cardinality is setup to n, it emits n-1
+          // including overflow. Just making the logic generic here to succeed for n or n-1 total
+          // cardinality.
+          EXPECT_GE(cardinality_limit, md.point_data_attr_.size());
+          EXPECT_LT(cardinality_limit / 2, md.point_data_attr_.size());
+          for (size_t i = 0; i < md.point_data_attr_.size(); i++)
+          {
+            EXPECT_EQ(1, md.point_data_attr_[i].attributes.size());
+            if (md.point_data_attr_[i].attributes.find("attr1") !=
+                md.point_data_attr_[i].attributes.end())
+            {
+              EXPECT_EQ(total_metrics_times * (repeat + 1),
+                        opentelemetry::nostd::get<double>(opentelemetry::nostd::get<SumPointData>(
+                                                              md.point_data_attr_[i].point_data)
+                                                              .value_));
+            }
+            else
+            {
+              EXPECT_NE(md.point_data_attr_[i].attributes.end(),
+                        md.point_data_attr_[i].attributes.find(
+                            sdk::metrics::kAttributesLimitOverflowKey));
+            }
+          }
+        }
+      }
+      return true;
+    });
+  }
+}
+
 class UpDownCounterToSumFixture : public ::testing::TestWithParam<bool>
 {};