[METRICS SDK] Enforce a runtime minimum scale for Base2ExponentialHistogramAggregation

CHANGELOG.md

@@ -15,6 +15,10 @@ Increment the:
 
 ## [Unreleased]
 
+* [METRICS SDK] Enforce a runtime minimum scale of `-11` for
+  `Base2ExponentialHistogramAggregation`, so a recording that spans the full
+  double range no longer downscales without end.
+  [#4353](https://github.com/open-telemetry/opentelemetry-cpp/pull/4353)
 * [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

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

@@ -72,6 +72,12 @@ constexpr std::int32_t kMaxScaleMin = -10;
 constexpr std::int32_t kMaxScaleMax = 20;
 constexpr std::size_t kMaxSizeMin   = 2;
 
+// Lower bound for the scale chosen at runtime by automatic downscaling. The specification only
+// requires a "reasonable minimum". At -11 every finite double maps to a bucket index in [-1, 0],
+// so the whole range fits in kMaxSizeMin buckets and no configuration needs more than its
+// max_size. This bounds the runtime scale only; max_scale still starts no lower than kMaxScaleMin.
+constexpr std::int32_t kMinRuntimeScale = -11;
+
 class Base2ExponentialHistogramAggregationConfig : public AggregationConfig
 {
 public:

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

@@ -47,12 +47,20 @@ class Base2ExponentialHistogramAggregation : public Aggregation
 private:
   void AggregateIntoBuckets(std::unique_ptr<AdaptingCircularBufferCounter> &buckets,
                             double value) noexcept;
-  void Downscale(uint32_t by) noexcept;
+
+  /* Reduces the scale by up to `by`, stopping at kMinRuntimeScale. Returns the reduction that was
+   * actually applied, which callers must use to shift bucket indices. */
+  uint32_t Downscale(uint32_t by) noexcept;
 
   mutable opentelemetry::common::SpinLockMutex lock_;
   Base2ExponentialHistogramPointData point_data_;
   Base2ExponentialHistogramIndexer indexer_;
   bool record_min_max_ = true;
+  // Keeps the scale floor warning off the record hot path after the first occurrence.
+  bool floor_warning_emitted_ = false;
+  // Same for the dropped-recording error, which repeats on every call once point data arrives with
+  // buckets that do not match its scale.
+  bool bucket_index_error_emitted_ = false;
 };
 
 }  // namespace metrics

sdk/src/metrics/aggregation/base2_exponential_histogram_aggregation.cc

@@ -35,7 +35,10 @@ namespace
 uint32_t GetScaleReduction(int32_t start_index, int32_t end_index, size_t max_buckets) noexcept
 {
   uint32_t scale_reduction = 0;
-  while (static_cast<int64_t>(end_index) - start_index + 1 > static_cast<int64_t>(max_buckets))
+  // Both indices have collapsed to -1 or 0 after 31 shifts, so further iterations cannot narrow
+  // the span; the bound keeps a degenerate max_buckets from spinning forever.
+  while (scale_reduction < 31 &&
+         static_cast<int64_t>(end_index) - start_index + 1 > static_cast<int64_t>(max_buckets))
   {
     start_index >>= 1;
     end_index >>= 1;
@@ -94,6 +97,102 @@ void DownscaleBuckets(std::unique_ptr<AdaptingCircularBufferCounter> &buckets, u
   buckets->Downscale(by);
 }
 
+// Guards point data that arrives through the public constructors with a smaller budget than the
+// configuration validator would ever produce. A configured max_size is at least kMaxSizeMin, so
+// this never allocates more buckets than the user asked for.
+size_t BucketCapacity(size_t max_buckets) noexcept
+{
+  return (std::max)(max_buckets, kMaxSizeMin);
+}
+
+// Point data handed to the public constructors carries buffers the caller sized, which can be
+// narrower than the capacity this class guarantees; move the counts into a wide enough buffer.
+void EnsureBucketCapacity(std::unique_ptr<AdaptingCircularBufferCounter> &buckets,
+                          size_t capacity) noexcept
+{
+  if (!buckets || buckets->MaxSize() >= capacity)
+  {
+    return;
+  }
+
+  auto widened = std::make_unique<AdaptingCircularBufferCounter>(capacity);
+  if (!buckets->Empty())
+  {
+    for (int32_t index = buckets->StartIndex(); index <= buckets->EndIndex(); ++index)
+    {
+      const uint64_t count = buckets->Get(index);
+      if (count > 0 && !widened->Increment(index, count))
+      {
+        OTEL_INTERNAL_LOG_ERROR(
+            "[Base2ExponentialHistogramAggregation::EnsureBucketCapacity] bucket index "
+            << index << " out of range; count " << count << " dropped. SDK invariant violation");
+        assert(false && "EnsureBucketCapacity: bucket index out of range");
+      }
+    }
+  }
+  buckets = std::move(widened);
+}
+
+// Truncates `requested` to the reduction that can be applied without pushing `current_scale` below
+// kMinRuntimeScale. Returns 0 once the floor is reached.
+uint32_t ClampScaleReduction(int32_t current_scale, uint32_t requested) noexcept
+{
+  const int64_t headroom = static_cast<int64_t>(current_scale) - kMinRuntimeScale;
+  if (headroom <= 0)
+  {
+    return 0;
+  }
+  return static_cast<uint32_t>((std::min)(static_cast<int64_t>(requested), headroom));
+}
+
+// Single entry point for scale reduction: clamps to the runtime floor, folds both bucket arrays by
+// the clamped amount and moves scale_ by exactly that amount. Returns the reduction applied.
+uint32_t ApplyDownscale(Base2ExponentialHistogramPointData &point_data, uint32_t requested) noexcept
+{
+  const uint32_t applied = ClampScaleReduction(point_data.scale_, requested);
+  if (applied == 0)
+  {
+    return 0;
+  }
+
+  if (point_data.positive_buckets_)
+  {
+    DownscaleBuckets(point_data.positive_buckets_, applied);
+  }
+  if (point_data.negative_buckets_)
+  {
+    DownscaleBuckets(point_data.negative_buckets_, applied);
+  }
+  point_data.scale_ -= static_cast<int32_t>(applied);
+  return applied;
+}
+
+// Folds `high_res` onto `target_scale`. The bucket shift has to match the scale delta exactly, so
+// the runtime floor is deliberately not applied here: it bounds the reductions the SDK chooses,
+// not the alignment of an operand that already sits lower.
+void AlignToScale(Base2ExponentialHistogramPointData &high_res, int32_t target_scale) noexcept
+{
+  if (high_res.scale_ <= target_scale)
+  {
+    return;
+  }
+
+  // AdaptingCircularBufferCounter::Downscale() saturates at 31, which is idempotent for int32_t
+  // indices, so a larger delta needs no special handling.
+  const int64_t delta = static_cast<int64_t>(high_res.scale_) - target_scale;
+  const uint32_t by   = delta > 31 ? 31u : static_cast<uint32_t>(delta);
+
+  if (high_res.positive_buckets_)
+  {
+    DownscaleBuckets(high_res.positive_buckets_, by);
+  }
+  if (high_res.negative_buckets_)
+  {
+    DownscaleBuckets(high_res.negative_buckets_, by);
+  }
+  high_res.scale_ = target_scale;
+}
+
 }  // namespace
 
 Base2ExponentialHistogramAggregation::Base2ExponentialHistogramAggregation(
@@ -132,9 +231,9 @@ Base2ExponentialHistogramAggregation::Base2ExponentialHistogramAggregation(
 
   // Initialize buckets
   point_data_.positive_buckets_ =
-      std::make_unique<AdaptingCircularBufferCounter>(point_data_.max_buckets_);
+      std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(point_data_.max_buckets_));
   point_data_.negative_buckets_ =
-      std::make_unique<AdaptingCircularBufferCounter>(point_data_.max_buckets_);
+      std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(point_data_.max_buckets_));
 
   indexer_ = Base2ExponentialHistogramIndexer(point_data_.scale_);
 }
@@ -164,14 +263,20 @@ Base2ExponentialHistogramAggregation::Base2ExponentialHistogramAggregation(
     point_data_.negative_buckets_ =
         std::make_unique<AdaptingCircularBufferCounter>(*point_data.negative_buckets_);
   }
+
+  EnsureBucketCapacity(point_data_.positive_buckets_, BucketCapacity(point_data_.max_buckets_));
+  EnsureBucketCapacity(point_data_.negative_buckets_, BucketCapacity(point_data_.max_buckets_));
 }
 
 Base2ExponentialHistogramAggregation::Base2ExponentialHistogramAggregation(
     Base2ExponentialHistogramPointData &&point_data)
     : point_data_{std::move(point_data)},
       indexer_(point_data_.scale_),
       record_min_max_{point_data_.record_min_max_}
-{}
+{
+  EnsureBucketCapacity(point_data_.positive_buckets_, BucketCapacity(point_data_.max_buckets_));
+  EnsureBucketCapacity(point_data_.negative_buckets_, BucketCapacity(point_data_.max_buckets_));
+}
 
 void Base2ExponentialHistogramAggregation::Aggregate(
     int64_t value,
@@ -221,45 +326,70 @@ void Base2ExponentialHistogramAggregation::AggregateIntoBuckets(
 {
   if (!buckets)
   {
-    buckets = std::make_unique<AdaptingCircularBufferCounter>(point_data_.max_buckets_);
+    buckets =
+        std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(point_data_.max_buckets_));
   }
 
   if (buckets->MaxSize() == 0)
   {
-    buckets = std::make_unique<AdaptingCircularBufferCounter>(point_data_.max_buckets_);
+    buckets =
+        std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(point_data_.max_buckets_));
   }
 
   const int32_t index = indexer_.ComputeIndex(value);
-  if (!buckets->Increment(index, 1))
+  if (buckets->Increment(index, 1))
   {
-    const int32_t start_index = (std::min)(buckets->StartIndex(), index);
-    const int32_t end_index   = (std::max)(buckets->EndIndex(), index);
-    const uint32_t scale_reduction =
-        GetScaleReduction(start_index, end_index, point_data_.max_buckets_);
-    Downscale(scale_reduction);
+    return;
+  }
 
-    buckets->Increment(index >> scale_reduction, 1);
+  // A configured max_size is never below kMaxSizeMin, so the buffer capacity equals max_buckets_
+  // and the failure above already means the span exceeds the budget.
+  const uint32_t scale_reduction =
+      GetScaleReduction((std::min)(buckets->StartIndex(), index),
+                        (std::max)(buckets->EndIndex(), index), point_data_.max_buckets_);
+
+  // Downscale() may stop short of the request at the floor, so shift the index by what was
+  // actually applied.
+  const uint32_t applied = Downscale(scale_reduction);
+  if (!buckets->Increment(index >> applied, 1) && !bucket_index_error_emitted_)
+  {
+    // Unreachable for buckets this class produced: at the floor every finite double maps to -1 or
+    // 0, which fits kMaxSizeMin. It is reachable through the point data constructors, so it is a
+    // caller input error rather than an assertable invariant.
+    bucket_index_error_emitted_ = true;
+    OTEL_INTERNAL_LOG_ERROR(
+        "[Base2ExponentialHistogramAggregation::AggregateIntoBuckets] bucket index "
+        << (index >> applied) << " does not fit the buckets supplied at scale "
+        << point_data_.scale_
+        << "; recording dropped. Further drops on this aggregation are not logged");
   }
 }
 
-void Base2ExponentialHistogramAggregation::Downscale(uint32_t by) noexcept
+uint32_t Base2ExponentialHistogramAggregation::Downscale(uint32_t by) noexcept
 {
   if (by == 0)
   {
-    return;
+    return 0;
   }
 
-  if (point_data_.positive_buckets_)
+  const uint32_t applied = ApplyDownscale(point_data_, by);
+
+  if (applied < by && !floor_warning_emitted_)
   {
-    DownscaleBuckets(point_data_.positive_buckets_, by);
+    floor_warning_emitted_ = true;
+    OTEL_INTERNAL_LOG_WARN("[Base2ExponentialHistogramAggregation] scale "
+                           << point_data_.scale_ << " reached the runtime minimum "
+                           << kMinRuntimeScale
+                           << "; recorded values now share buckets instead of downscaling further");
   }
-  if (point_data_.negative_buckets_)
+
+  if (applied == 0)
   {
-    DownscaleBuckets(point_data_.negative_buckets_, by);
+    return 0;
   }
 
-  point_data_.scale_ -= static_cast<int32_t>(by);
   indexer_ = Base2ExponentialHistogramIndexer(point_data_.scale_);
+  return applied;
 }
 
 // Merge A and B into a new circular buffer C.
@@ -329,7 +459,6 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Merge(
   result_value.count_      = low_res.count_ + high_res.count_;
   result_value.sum_        = low_res.sum_ + high_res.sum_;
   result_value.zero_count_ = low_res.zero_count_ + high_res.zero_count_;
-  result_value.scale_      = (std::min)(low_res.scale_, high_res.scale_);
   result_value.max_buckets_ =
       low_res.max_buckets_ >= high_res.max_buckets_ ? low_res.max_buckets_ : high_res.max_buckets_;
   result_value.record_min_max_ = low_res.record_min_max_ && high_res.record_min_max_;
@@ -340,16 +469,7 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Merge(
     result_value.max_ = (std::max)(low_res.max_, high_res.max_);
   }
 
-  {
-    auto scale_reduction = high_res.scale_ - low_res.scale_;
-
-    if (scale_reduction > 0)
-    {
-      DownscaleBuckets(high_res.positive_buckets_, scale_reduction);
-      DownscaleBuckets(high_res.negative_buckets_, scale_reduction);
-      high_res.scale_ -= scale_reduction;
-    }
-  }
+  AlignToScale(high_res, low_res.scale_);
 
   // positive_buckets_ and negative_buckets_ share a single scale_; apply
   // the maximum required reduction across both bucket types.
@@ -359,21 +479,18 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Merge(
                  GetScaleReductionForUnion(*low_res.negative_buckets_, *high_res.negative_buckets_,
                                            result_value.max_buckets_));
 
-  if (scale_reduction > 0)
-  {
-    DownscaleBuckets(low_res.positive_buckets_, scale_reduction);
-    DownscaleBuckets(high_res.positive_buckets_, scale_reduction);
-    DownscaleBuckets(low_res.negative_buckets_, scale_reduction);
-    DownscaleBuckets(high_res.negative_buckets_, scale_reduction);
-    low_res.scale_ -= static_cast<int32_t>(scale_reduction);
-    high_res.scale_ -= static_cast<int32_t>(scale_reduction);
-    result_value.scale_ -= static_cast<int32_t>(scale_reduction);
-  }
+  // Both operands share a scale after the alignment above, so one clamped amount applies to both.
+  const uint32_t applied = ClampScaleReduction(low_res.scale_, scale_reduction);
+  ApplyDownscale(low_res, applied);
+  ApplyDownscale(high_res, applied);
+  result_value.scale_ = low_res.scale_;
 
-  result_value.positive_buckets_ = std::make_unique<AdaptingCircularBufferCounter>(MergeBuckets(
-      result_value.max_buckets_, *low_res.positive_buckets_, *high_res.positive_buckets_));
-  result_value.negative_buckets_ = std::make_unique<AdaptingCircularBufferCounter>(MergeBuckets(
-      result_value.max_buckets_, *low_res.negative_buckets_, *high_res.negative_buckets_));
+  result_value.positive_buckets_ = std::make_unique<AdaptingCircularBufferCounter>(
+      MergeBuckets(BucketCapacity(result_value.max_buckets_), *low_res.positive_buckets_,
+                   *high_res.positive_buckets_));
+  result_value.negative_buckets_ = std::make_unique<AdaptingCircularBufferCounter>(
+      MergeBuckets(BucketCapacity(result_value.max_buckets_), *low_res.negative_buckets_,
+                   *high_res.negative_buckets_));
 
   return std::unique_ptr<Base2ExponentialHistogramAggregation>{
       new Base2ExponentialHistogramAggregation(std::move(result_value))};
@@ -389,24 +506,7 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Diff(
   auto &low_res  = left.scale_ < right.scale_ ? left : right;
   auto &high_res = left.scale_ < right.scale_ ? right : left;
 
-  {
-    const auto scale_reduction = high_res.scale_ - low_res.scale_;
-
-    if (scale_reduction > 0)
-    {
-      if (high_res.positive_buckets_)
-      {
-        DownscaleBuckets(high_res.positive_buckets_, scale_reduction);
-      }
-
-      if (high_res.negative_buckets_)
-      {
-        DownscaleBuckets(high_res.negative_buckets_, scale_reduction);
-      }
-
-      high_res.scale_ -= scale_reduction;
-    }
-  }
+  AlignToScale(high_res, low_res.scale_);
 
   // positive_buckets_ and negative_buckets_ share a single scale_; apply
   // the maximum required reduction across both bucket types.
@@ -416,15 +516,10 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Diff(
                  GetScaleReductionForUnion(*low_res.negative_buckets_, *high_res.negative_buckets_,
                                            low_res.max_buckets_));
 
-  if (scale_reduction > 0)
-  {
-    DownscaleBuckets(low_res.positive_buckets_, scale_reduction);
-    DownscaleBuckets(high_res.positive_buckets_, scale_reduction);
-    DownscaleBuckets(low_res.negative_buckets_, scale_reduction);
-    DownscaleBuckets(high_res.negative_buckets_, scale_reduction);
-    low_res.scale_ -= static_cast<int32_t>(scale_reduction);
-    high_res.scale_ -= static_cast<int32_t>(scale_reduction);
-  }
+  // Both operands share a scale after the alignment above, so one clamped amount applies to both.
+  const uint32_t applied = ClampScaleReduction(low_res.scale_, scale_reduction);
+  ApplyDownscale(low_res, applied);
+  ApplyDownscale(high_res, applied);
 
   Base2ExponentialHistogramPointData result_value;
   result_value.scale_          = low_res.scale_;
@@ -436,9 +531,9 @@ std::unique_ptr<Aggregation> Base2ExponentialHistogramAggregation::Diff(
       (right.zero_count_ >= left.zero_count_) ? (right.zero_count_ - left.zero_count_) : 0;
 
   result_value.positive_buckets_ =
-      std::make_unique<AdaptingCircularBufferCounter>(right.max_buckets_);
+      std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(right.max_buckets_));
   result_value.negative_buckets_ =
-      std::make_unique<AdaptingCircularBufferCounter>(right.max_buckets_);
+      std::make_unique<AdaptingCircularBufferCounter>(BucketCapacity(right.max_buckets_));
 
   if (!left.positive_buckets_->Empty() || !right.positive_buckets_->Empty())
   {