open-telemetry · om7057 · Aug 10, 2026 · Aug 10, 2026 · Aug 11, 2026 · Aug 12, 2026
@@ -15,6 +15,25 @@ Increment the:
 
 ## [Unreleased]
 
+* [METRICS SDK] Enforce `MetricReader`-level cardinality limits as a fallback
+  during collection, when the matching view has no explicit
+  `aggregation_cardinality_limit` of its own (View > Reader > SDK default).
+  Shared recording storage for a view with no explicit limit is now sized at
+  the max limit across all attached readers, so a reader with a higher limit
+  does not lose data; each reader's own (possibly stricter) limit is then
+  re-applied to just its own collected output. A programmatically-constructed
+  `AggregationConfig`/`HistogramAggregationConfig`/
+  `Base2ExponentialHistogramAggregationConfig` (e.g. one built only to carry
+  histogram boundaries) is now also correctly treated as not having an
+  explicit cardinality limit, matching the declarative-configuration path.
+  `AggregationConfig::cardinality_limit_`/`cardinality_limit_explicit_` are
+  now private, set together via `SetCardinalityLimit()`, so they cannot be
+  desynced by direct field assignment. The single-collector delta fast path
+  in `TemporalMetricStorage::buildMetrics()` now re-caps to the collector's
+  own limit instead of emitting the raw recording storage unchecked, and the
+  bound-instrument admission path (preview) now uses the same resolved
+  recording limit as the unbound path.
+  [#4387](https://github.com/open-telemetry/opentelemetry-cpp/issues/4387)
 * [CONFIGURATION] Add a configuration builder for the host resource detector
   [#4451](https://github.com/open-telemetry/opentelemetry-cpp/issues/4451)
 * [CONFIGURATION] Build the configured resource detectors in SdkBuilder, apply

@@ -3,6 +3,8 @@
 
 #pragma once
 
+#include <cstddef>
+#include <limits>
 #include <vector>
 
 #include "opentelemetry/sdk/metrics/instruments.h"
@@ -15,11 +17,21 @@ namespace sdk
 namespace metrics
 {
 
+// Sentinel passed to AggregationConfig's cardinality_limit constructor argument to mean "the
+// caller did not specify a limit", distinct from any real limit value (which is always a small
+// positive count). Lets the constructor tell apart e.g. HistogramAggregationConfig() built only
+// to carry boundaries_ from HistogramAggregationConfig(500) built to set a real limit of 500,
+// without needing <optional> (avoided elsewhere in the SDK for ABI reasons).
+constexpr size_t kCardinalityLimitUnspecified = (std::numeric_limits<size_t>::max)();
+
 class AggregationConfig
 {
 public:
-  AggregationConfig(size_t cardinality_limit = kAggregationCardinalityLimit)
-      : cardinality_limit_(cardinality_limit)
+  AggregationConfig(size_t cardinality_limit = kCardinalityLimitUnspecified)
+      : cardinality_limit_(cardinality_limit == kCardinalityLimitUnspecified
+                               ? kAggregationCardinalityLimit
+                               : cardinality_limit),
+        cardinality_limit_explicit_(cardinality_limit != kCardinalityLimitUnspecified)
   {}
 
   AggregationConfig(const AggregationConfig &)            = default;
@@ -39,14 +51,33 @@ class AggregationConfig
     return &default_config;
   }
 
-  size_t cardinality_limit_;
+  std::size_t GetCardinalityLimit() const noexcept { return cardinality_limit_; }
+
+  // Whether the cardinality limit reflects an intentionally-configured value, as opposed to
+  // just the compiled-in default it was left at because this config was constructed for some
+  // other reason (e.g. histogram boundaries) without a cardinality_limit argument. A
+  // MetricReader-level fallback applies whenever this is false.
+  bool IsCardinalityLimitExplicit() const noexcept { return cardinality_limit_explicit_; }
+
+  // Sets the cardinality limit and marks it explicit, atomically. Use this instead of assigning
+  // a limit some other way, so the limit and its explicit-ness can never go out of sync.
+  void SetCardinalityLimit(std::size_t cardinality_limit) noexcept
+  {
+    cardinality_limit_          = cardinality_limit;
+    cardinality_limit_explicit_ = true;
+  }
+
   virtual ~AggregationConfig() = default;
+
+private:
+  size_t cardinality_limit_;
+  bool cardinality_limit_explicit_;
 };
 
 class HistogramAggregationConfig : public AggregationConfig
 {
 public:
-  HistogramAggregationConfig(size_t cardinality_limit = kAggregationCardinalityLimit)
+  HistogramAggregationConfig(size_t cardinality_limit = kCardinalityLimitUnspecified)
       : AggregationConfig(cardinality_limit)
   {}
 
@@ -76,7 +107,7 @@ class Base2ExponentialHistogramAggregationConfig : public AggregationConfig
 {
 public:
   Base2ExponentialHistogramAggregationConfig(
-      size_t cardinality_limit = kAggregationCardinalityLimit)
+      size_t cardinality_limit = kCardinalityLimitUnspecified)
       : AggregationConfig(cardinality_limit)
   {}
 

@@ -142,6 +142,13 @@ class MeterContext : public std::enable_shared_from_this<MeterContext>
    * Note: This reader may not receive any in-flight meter data, but will get newly created meter
    * data.
    * Note: This method is not thread safe, and should ideally be called from main thread.
+   * Note: For an instrument that already exists when this reader is added, this reader's
+   * cardinality limit for that instrument only takes effect up to whatever recording capacity
+   * was already resolved (as the highest limit across readers attached at instrument-creation
+   * time) when the instrument was first created; that capacity is fixed for the lifetime of the
+   * instrument and is not grown retroactively. A reader added with a higher limit than any
+   * reader present at instrument-creation time will not recover attribute sets that had already
+   * collapsed into the instrument's overflow point before this reader was attached.
    */
   void AddMetricReader(std::shared_ptr<MetricReader> reader,
                        std::unique_ptr<MetricFilter> metric_filter = nullptr) noexcept;

@@ -104,6 +104,13 @@ class OPENTELEMETRY_EXPORT MeterProvider final : public opentelemetry::metrics::
    * Note: This reader may not receive any in-flight meter data, but will get newly created meter
    * data.
    * Note: This method is not thread safe, and should ideally be called from main thread.
+   * Note: For an instrument that already exists when this reader is added, this reader's
+   * cardinality limit for that instrument only takes effect up to whatever recording capacity
+   * was already resolved (as the highest limit across readers attached at instrument-creation
+   * time) when the instrument was first created; that capacity is fixed for the lifetime of the
+   * instrument and is not grown retroactively. A reader added with a higher limit than any
+   * reader present at instrument-creation time will not recover attribute sets that had already
+   * collapsed into the instrument's overflow point before this reader was attached.
    */
   void AddMetricReader(std::shared_ptr<MetricReader> reader,
                        std::unique_ptr<MetricFilter> metric_filter = nullptr) noexcept;

@@ -61,9 +61,8 @@ class MetricReader
   /**
    * Set per-instrument-type cardinality limits for this reader.
    *
-   * TODO: Reader-level limits are stored but not yet enforced as a per-collector
-   * fallback during the collection path. Enforcement will be added in a follow-up.
-   * View-level limits (via AggregationConfig) are enforced today.
+   * Enforced as a fallback during the collection path whenever the matching view has no
+   * explicit AggregationConfig cardinality limit of its own (View > Reader > SDK default).
    *
    * @param limits The cardinality limits to apply
    */

@@ -35,20 +35,40 @@ namespace metrics
 class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStorage
 {
 public:
+  // Back-compat overload preserving the original constructor signature for any external caller.
+  // See SyncMetricStorage's constructor comment for what `recording_cardinality_limit` is for.
   AsyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
                      const AggregationType aggregation_type,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                      ExemplarFilterType exemplar_filter_type,
                      nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
 #endif
                      const AggregationConfig *aggregation_config)
+      : AsyncMetricStorage(
+            instrument_descriptor,
+            aggregation_type,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+            exemplar_filter_type,
+            std::move(exemplar_reservoir),
+#endif
+            aggregation_config,
+            AggregationConfig::GetOrDefault(aggregation_config)->GetCardinalityLimit())
+  {}
+
+  AsyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
+                     const AggregationType aggregation_type,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+                     ExemplarFilterType exemplar_filter_type,
+                     nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
+#endif
+                     const AggregationConfig *aggregation_config,
+                     std::size_t recording_cardinality_limit)
       : instrument_descriptor_(instrument_descriptor),
         aggregation_type_{aggregation_type},
         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_)),
+        recording_cardinality_limit_(recording_cardinality_limit),
+        cumulative_hash_map_(std::make_unique<AttributesHashMap>(recording_cardinality_limit_)),
+        delta_hash_map_(std::make_unique<AttributesHashMap>(recording_cardinality_limit_)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exemplar_filter_type),
         exemplar_reservoir_(std::move(exemplar_reservoir)),
@@ -131,9 +151,8 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     std::shared_ptr<AttributesHashMap> delta_metrics = nullptr;
     {
       std::lock_guard<opentelemetry::common::SpinLockMutex> guard(hashmap_lock_);
-      delta_metrics = std::move(delta_hash_map_);
-      delta_hash_map_ =
-          std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_);
+      delta_metrics   = std::move(delta_hash_map_);
+      delta_hash_map_ = std::make_unique<AttributesHashMap>(recording_cardinality_limit_);
     }
 
     auto status =
@@ -146,6 +165,8 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
   InstrumentDescriptor instrument_descriptor_;
   AggregationType aggregation_type_;
   const AggregationConfig *aggregation_config_;
+  // Capacity used to (re)size cumulative_hash_map_/delta_hash_map_. See the constructor comment.
+  const std::size_t recording_cardinality_limit_;
   std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
   opentelemetry::common::SpinLockMutex hashmap_lock_;

@@ -122,7 +122,12 @@ class AttributesHashMapWithCustomHash
     return result.first->second.get();
   }
   /**
-   * Set the value for given key, overwriting the value if already present
+   * Set the value for given key, overwriting the value if already present.
+   *
+   * When `attributes` is a new key that has to be routed to the overflow entry (capacity
+   * reached), this merges into any existing overflow aggregation rather than replacing it, so
+   * that a second and later over-capacity attribute set doesn't discard values already
+   * accumulated there from an earlier one.
    */
   void Set(const MetricAttributes &attributes, std::unique_ptr<Aggregation> aggr)
   {
@@ -133,7 +138,7 @@ class AttributesHashMapWithCustomHash
     }
     else if (IsOverflowAttributes(attributes))
     {
-      hash_map_[GetOverflowAttributes()] = std::move(aggr);
+      SetOverflowMerged(std::move(aggr));
     }
     else
     {
@@ -150,7 +155,7 @@ class AttributesHashMapWithCustomHash
     }
     else if (IsOverflowAttributes(attributes))
     {
-      hash_map_[GetOverflowAttributes()] = std::move(aggr);
+      SetOverflowMerged(std::move(aggr));
     }
     else
     {
@@ -207,6 +212,23 @@ class AttributesHashMapWithCustomHash
     return result.first->second.get();
   }
 
+  // Merges `agg` into the existing overflow aggregation if one is already present, otherwise
+  // installs it as the overflow entry. Used by Set() when routing a new over-capacity key to
+  // overflow, so multiple distinct attribute sets funneled into the single overflow bucket
+  // accumulate rather than each replacing the last.
+  void SetOverflowMerged(std::unique_ptr<Aggregation> agg)
+  {
+    auto it = hash_map_.find(GetOverflowAttributes());
+    if (it != hash_map_.end())
+    {
+      hash_map_[GetOverflowAttributes()] = it->second->Merge(*agg);
+    }
+    else
+    {
+      hash_map_[GetOverflowAttributes()] = std::move(agg);
+    }
+  }
+
   bool IsOverflowAttributes(const MetricAttributes &attributes) const
   {
     // If the incoming attributes are exactly the overflow sentinel, route

@@ -50,6 +50,9 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
 {
 
 public:
+  // Back-compat overload preserving the original constructor signature for any external caller.
+  // Sizes recording storage using the view's own cardinality limit (or the SDK default when no
+  // AggregationConfig is supplied), matching this class's original behavior exactly.
   SyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
                     const AggregationType aggregation_type,
                     std::shared_ptr<const AttributesProcessor> attributes_processor,
@@ -58,10 +61,38 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
                     nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
 #endif
                     const AggregationConfig *aggregation_config)
+      : SyncMetricStorage(
+            instrument_descriptor,
+            aggregation_type,
+            std::move(attributes_processor),
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+            exemplar_filter_type,
+            std::move(exemplar_reservoir),
+#endif
+            aggregation_config,
+            AggregationConfig::GetOrDefault(aggregation_config)->GetCardinalityLimit())
+  {}
+
+  // `recording_cardinality_limit` sizes the storage that raw measurements are recorded into,
+  // separately from `aggregation_config`. When a view sets an explicit cardinality limit,
+  // callers should pass that same limit (the overload above does this). When a view has no
+  // explicit limit, callers may pass the max cardinality limit configured across all attached
+  // MetricReaders, so no reader loses data purely because the shared recording storage was
+  // capped too low for it; each reader's own (possibly stricter) limit is then re-applied to
+  // its own output during collection. See TemporalMetricStorage::buildMetrics().
+  SyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
+                    const AggregationType aggregation_type,
+                    std::shared_ptr<const AttributesProcessor> attributes_processor,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+                    ExemplarFilterType exemplar_filter_type,
+                    nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
+#endif
+                    const AggregationConfig *aggregation_config,
+                    std::size_t recording_cardinality_limit)
       : instrument_descriptor_(instrument_descriptor),
         aggregation_config_(AggregationConfig::GetOrDefault(aggregation_config)),
-        attributes_hashmap_(
-            std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
+        recording_cardinality_limit_(recording_cardinality_limit),
+        attributes_hashmap_(std::make_unique<AttributesHashMap>(recording_cardinality_limit_)),
         attributes_processor_(std::move(attributes_processor)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exemplar_filter_type),
@@ -259,7 +290,12 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
     {
       return filtered;
     }
-    const size_t limit      = aggregation_config_->cardinality_limit_;
+    // Use the resolved recording capacity, not aggregation_config_->GetCardinalityLimit():
+    // when the limit comes from a MetricReader fallback rather than an explicit view limit,
+    // these can differ, and attributes_hashmap_ (the unbound path) is sized to
+    // recording_cardinality_limit_. Using the same value here keeps bound and unbound
+    // admission consistent.
+    const size_t limit      = recording_cardinality_limit_;
     const bool has_overflow = active_keys_.find(GetOverflowAttributes()) != active_keys_.end();
     // Mirror AttributesHashMap::IsOverflowAttributes() exactly. The configured
     // limit applies to non-overflow attribute sets, while overflow is reserved.
@@ -278,6 +314,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
   InstrumentDescriptor instrument_descriptor_;
   // hashmap to maintain the metrics for delta collection (i.e, collection since last Collect call)
   const AggregationConfig *aggregation_config_;
+  // Capacity used to (re)size attributes_hashmap_. See the constructor comment above.
+  const std::size_t recording_cardinality_limit_;
   std::unique_ptr<AttributesHashMap> attributes_hashmap_;
   std::function<std::unique_ptr<Aggregation>()> create_default_aggregation_;
   std::shared_ptr<const AttributesProcessor> attributes_processor_;

@@ -1989,6 +1989,11 @@ void SdkBuilder::AddView(
 
     if (stream->aggregation)
     {
+      // CreateAggregationConfig() may build a config purely for non-cardinality reasons (e.g.
+      // histogram boundaries) via a config's default constructor, which already leaves the
+      // cardinality limit not-explicit (see AggregationConfig's constructor) so it doesn't
+      // shadow a MetricReader-level fallback; the block below marks it explicit if
+      // aggregation_cardinality_limit is also configured.
       sdk_aggregation_config = CreateAggregationConfig(stream->aggregation, sdk_aggregation_type);
     }
 
@@ -1997,7 +2002,7 @@ void SdkBuilder::AddView(
     {
       if (sdk_aggregation_config)
       {
-        sdk_aggregation_config->cardinality_limit_ = stream->aggregation_cardinality_limit;
+        sdk_aggregation_config->SetCardinalityLimit(stream->aggregation_cardinality_limit);
       }
       else
       {