[Metrics SDK] Make cardinality limit configurable through View class

sdk/include/opentelemetry/sdk/metrics/metric_reader.h

@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <stddef.h>
 #include <atomic>
 #include <chrono>
 
@@ -41,6 +42,19 @@ class MetricReader
   virtual AggregationTemporality GetAggregationTemporality(
       InstrumentType instrument_type) const noexcept = 0;
 
+  /**
+   * Get the default cardinality limit for given Instrument Type for this reader.
+   * 
+   * @param instrument_type The instrument type to get the cardinality limit for
+   * @return The cardinality limit, or 0 if no limit is set
+   */
+  virtual size_t GetDefaultCardinalityLimit(InstrumentType instrument_type) const noexcept
+  {
+    // Default implementation returns no limit
+    (void)instrument_type;
+    return 0;
+  }
+
   /**
    * Shutdown the metric reader.
    */

sdk/include/opentelemetry/sdk/metrics/view/view.h

@@ -31,13 +31,15 @@ class View
        std::shared_ptr<AggregationConfig> aggregation_config = nullptr,
        std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attributes_processor =
            std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor>(
-               new opentelemetry::sdk::metrics::DefaultAttributesProcessor()))
+               new opentelemetry::sdk::metrics::DefaultAttributesProcessor()),
+       size_t aggregation_cardinality_limit = 0)
       : name_(name),
         description_(description),
         unit_(unit),
         aggregation_type_{aggregation_type},
         aggregation_config_{aggregation_config},
-        attributes_processor_{std::move(attributes_processor)}
+        attributes_processor_{std::move(attributes_processor)},
+        aggregation_cardinality_limit_{aggregation_cardinality_limit}
   {}
 
   virtual ~View() = default;
@@ -59,13 +61,24 @@ class View
     return attributes_processor_;
   }
 
+  virtual size_t GetAggregationCardinalityLimit() const noexcept
+  {
+    return aggregation_cardinality_limit_;
+  }
+
+  virtual bool HasAggregationCardinalityLimit() const noexcept
+  {
+    return aggregation_cardinality_limit_ > 0;
+  }
+
 private:
   std::string name_;
   std::string description_;
   std::string unit_;
   AggregationType aggregation_type_;
   std::shared_ptr<AggregationConfig> aggregation_config_;
   std::shared_ptr<AttributesProcessor> attributes_processor_;
+  size_t aggregation_cardinality_limit_;
 };
 }  // namespace metrics
 }  // namespace sdk

sdk/src/metrics/meter.cc

@@ -1,6 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
+#include <stddef.h>
 #include <cstdint>
 #include <mutex>
 #include <ostream>
@@ -29,6 +30,7 @@
 #include "opentelemetry/sdk/metrics/meter_config.h"
 #include "opentelemetry/sdk/metrics/meter_context.h"
 #include "opentelemetry/sdk/metrics/state/async_metric_storage.h"
+#include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #include "opentelemetry/sdk/metrics/state/metric_storage.h"
 #include "opentelemetry/sdk/metrics/state/multi_metric_storage.h"
@@ -538,14 +540,20 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
         else
         {
           WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc);
+          // Calculate cardinality limit based on specification priority:
+          // 1. View-specific cardinality limit (if set)
+          // 2. Default value of 2000
+          size_t cardinality_limit = view.HasAggregationCardinalityLimit()
+                                         ? view.GetAggregationCardinalityLimit()
+                                         : kAggregationCardinalityLimit;
           sync_storage = std::shared_ptr<SyncMetricStorage>(new SyncMetricStorage(
               view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
               exemplar_filter_type,
               GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),
                                    view_instr_desc),
 #endif
-              view.GetAggregationConfig()));
+              view.GetAggregationConfig(), cardinality_limit));
           storage_registry_.insert({view_instr_desc, sync_storage});
         }
         auto sync_multi_storage = static_cast<SyncMultiMetricStorage *>(storages.get());

sdk/test/metrics/cardinality_limit_test.cc

@@ -29,6 +29,7 @@
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #include "opentelemetry/sdk/metrics/state/sync_metric_storage.h"
 #include "opentelemetry/sdk/metrics/view/attributes_processor.h"
+#include "opentelemetry/sdk/metrics/view/view.h"
 
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
 #  include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
@@ -39,6 +40,22 @@ using namespace opentelemetry::sdk::metrics;
 using namespace opentelemetry::common;
 namespace nostd = opentelemetry::nostd;
 
+TEST(CardinalityLimit, ViewCardinalityLimitConfiguration)
+{
+  // Test View without cardinality limit
+  View view_no_limit("test_view_no_limit");
+  EXPECT_FALSE(view_no_limit.HasAggregationCardinalityLimit());
+  EXPECT_EQ(view_no_limit.GetAggregationCardinalityLimit(), 0);
+
+  // Test View with cardinality limit
+  View view_with_limit("test_view_with_limit", "", "", AggregationType::kDefault, nullptr,
+                       std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor>(
+                           new opentelemetry::sdk::metrics::DefaultAttributesProcessor()),
+                       500);
+  EXPECT_TRUE(view_with_limit.HasAggregationCardinalityLimit());
+  EXPECT_EQ(view_with_limit.GetAggregationCardinalityLimit(), 500);
+}
+
 TEST(CardinalityLimit, AttributesHashMapBasicTests)
 {
   AttributesHashMap hash_map(10);
@@ -156,3 +173,84 @@ TEST_P(WritableMetricStorageCardinalityLimitTestFixture, LongCounterSumAggregati
 INSTANTIATE_TEST_SUITE_P(All,
                          WritableMetricStorageCardinalityLimitTestFixture,
                          ::testing::Values(AggregationTemporality::kDelta));
+
+TEST(CardinalityLimit, SyncMetricStorageWithViewCardinalityLimit)
+{
+  auto sdk_start_ts               = std::chrono::system_clock::now();
+  InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kCounter,
+                                     InstrumentValueType::kLong};
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
+      new DefaultAttributesProcessor{}};
+
+  // Create a view with a cardinality limit of 5
+  View view_with_limit("test_view", "", "", AggregationType::kSum, nullptr,
+                       std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor>(
+                           new opentelemetry::sdk::metrics::DefaultAttributesProcessor()),
+                       5);
+
+  // Test that the view has the cardinality limit
+  EXPECT_TRUE(view_with_limit.HasAggregationCardinalityLimit());
+  EXPECT_EQ(view_with_limit.GetAggregationCardinalityLimit(), 5);
+
+  // Create SyncMetricStorage using the cardinality limit from the view
+  size_t cardinality_limit = view_with_limit.HasAggregationCardinalityLimit()
+                                 ? view_with_limit.GetAggregationCardinalityLimit()
+                                 : kAggregationCardinalityLimit;
+  SyncMetricStorage storage(instr_desc, AggregationType::kSum, default_attributes_processor,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+                            ExemplarFilterType::kAlwaysOff,
+                            ExemplarReservoir::GetNoExemplarReservoir(),
+#endif
+                            nullptr, cardinality_limit);
+
+  int64_t record_value = 100;
+  // Add 5 unique metric points (should all fit within limit)
+  // With cardinality limit 5: first 4 get individual metric points
+  for (auto i = 0; i < 5; i++)
+  {
+    std::map<std::string, std::string> attributes = {{"key", std::to_string(i)}};
+    storage.RecordLong(record_value,
+                       KeyValueIterableView<std::map<std::string, std::string>>(attributes),
+                       opentelemetry::context::Context{});
+  }
+
+  // Add 3 more unique metric points (should trigger overflow behavior)
+  // These will be aggregated with the 5th metric point into overflow bucket
+  for (auto i = 5; i < 8; i++)
+  {
+    std::map<std::string, std::string> attributes = {{"key", std::to_string(i)}};
+    storage.RecordLong(record_value,
+                       KeyValueIterableView<std::map<std::string, std::string>>(attributes),
+                       opentelemetry::context::Context{});
+  }
+
+  AggregationTemporality temporality = AggregationTemporality::kDelta;
+  std::shared_ptr<CollectorHandle> collector(new MockCollectorHandle(temporality));
+  std::vector<std::shared_ptr<CollectorHandle>> collectors;
+  collectors.push_back(collector);
+  auto collection_ts      = std::chrono::system_clock::now();
+  size_t count_attributes = 0;
+  bool overflow_present   = false;
+
+  storage.Collect(
+      collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData &metric_data) {
+        for (const auto &data_attr : metric_data.point_data_attr_)
+        {
+          count_attributes++;
+          if (data_attr.attributes.begin()->first == kAttributesLimitOverflowKey)
+          {
+            // The overflow attribute should contain the aggregated values from the 4 excess metrics
+            // With cardinality limit 5: first 4 get individual points, remaining 4 go to overflow
+            const auto &data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
+            EXPECT_EQ(nostd::get<int64_t>(data.value_), record_value * 4);
+            overflow_present = true;
+          }
+        }
+        return true;
+      });
+
+  // We should have exactly 5 attributes (the cardinality limit)
+  EXPECT_EQ(count_attributes, 5);
+  // And there should be an overflow attribute
+  EXPECT_TRUE(overflow_present);
+}