diff --git a/core/include/prometheus/detail/ckms_quantiles.h b/core/include/prometheus/detail/ckms_quantiles.h index a3b6617f..39ba4f21 100644 --- a/core/include/prometheus/detail/ckms_quantiles.h +++ b/core/include/prometheus/detail/ckms_quantiles.h @@ -39,7 +39,7 @@ class PROMETHEUS_CPP_CORE_EXPORT CKMSQuantiles { void reset(); private: - double allowableError(int rank); + double allowableError(int rank, std::size_t size); bool insertBatch(); void compress(); diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 395b6ff3..66ed8e92 100644 --- a/core/src/detail/ckms_quantiles.cc +++ b/core/src/detail/ckms_quantiles.cc @@ -39,7 +39,7 @@ double CKMSQuantiles::get(double q) { int rankMin = 0; const auto desired = static_cast(q * count_); - const auto bound = desired + (allowableError(desired) / 2); + const auto bound = desired + (allowableError(desired, sample_.size()) / 2); auto it = sample_.begin(); decltype(it) prev; @@ -65,8 +65,7 @@ void CKMSQuantiles::reset() { buffer_count_ = 0; } -double CKMSQuantiles::allowableError(int rank) { - auto size = sample_.size(); +double CKMSQuantiles::allowableError(int rank, std::size_t size) { double minError = size + 1; for (const auto& q : quantiles_.get()) { @@ -85,42 +84,42 @@ double CKMSQuantiles::allowableError(int rank) { } bool CKMSQuantiles::insertBatch() { + // If there is no data to insert return false if (buffer_count_ == 0) { return false; } + // Sort the buffer upto buffer_count_ to prepare for inserting items std::sort(buffer_.begin(), buffer_.begin() + buffer_count_); std::size_t start = 0; + + sample_.reserve(sample_.size() + buffer_count_); + // If the sample set is empty, add the first item if (sample_.empty()) { sample_.emplace_back(buffer_[0], 1, 0); - ++start; + ++start; // Skip the first item since it's already added to the sample ++count_; } - std::size_t idx = 0; - std::size_t item = idx++; - + // Loop through the buffer and insert the items into the sample set for (std::size_t i = start; i < buffer_count_; ++i) { - double v = buffer_[i]; - while (idx < sample_.size() && sample_[item].value < v) { - item = idx++; - } - - if (sample_[item].value > v) { - --idx; - } - - int delta; - if (idx - 1 == 0 || idx + 1 == sample_.size()) { - delta = 0; - } else { - delta = static_cast(std::floor(allowableError(idx + 1))) + 1; + double value = buffer_[i]; + + auto iterator = std::lower_bound( + sample_.begin(), sample_.end(), value, + [](const Item& item, double val) { return item.value < val; }); + std::size_t idx = std::distance(sample_.begin(), iterator); + + int delta = 0; + if (idx > 0 && idx < sample_.size()) { + delta = static_cast(std::floor( + allowableError(static_cast(idx) + 1, sample_.size()))) + + 1; } - sample_.emplace(sample_.begin() + idx, v, 1, delta); - count_++; - item = idx++; + sample_.emplace(iterator, value, 1, delta); + ++count_; } buffer_count_ = 0; @@ -128,24 +127,45 @@ bool CKMSQuantiles::insertBatch() { } void CKMSQuantiles::compress() { + // If there are less than 2 items in the sample set, there's nothing to + // compress if (sample_.size() < 2) { return; } - std::size_t idx = 0; - std::size_t prev; - std::size_t next = idx++; - - while (idx < sample_.size()) { - prev = next; - next = idx++; - - if (sample_[prev].g + sample_[next].g + sample_[next].delta <= - allowableError(idx - 1)) { - sample_[next].g += sample_[prev].g; - sample_.erase(sample_.begin() + prev); + std::vector compressed_samples; // Vector to hold compressed samples + compressed_samples.reserve( + sample_.size()); // Reserve space to avoid multiple allocations + + // Start with the first sample + compressed_samples.push_back(sample_[0]); + + // Merges logically remove items even though sample_ stays unchanged. + auto logical_size = sample_.size(); + + for (std::size_t idx = 1; idx < sample_.size(); ++idx) { + const Item& current_sample = sample_[idx]; + Item& last_compressed_sample = compressed_samples.back(); + const auto logical_index = static_cast(compressed_samples.size()); + + // Check if we can compress the current sample into the last compressed + // sample + if (last_compressed_sample.g + current_sample.g + current_sample.delta <= + allowableError(logical_index, logical_size)) { + // Keep current_sample's value/delta (matches original semantics of + // dropping the earlier, smaller-value sample) but combine weights. + int merged_g = last_compressed_sample.g + current_sample.g; + last_compressed_sample = current_sample; + last_compressed_sample.g = merged_g; + --logical_size; + } else { + // If not compressible, add current sample to compressed samples + compressed_samples.push_back(current_sample); } } + + // Replace old samples with new compressed samples + sample_ = std::move(compressed_samples); } } // namespace prometheus::detail diff --git a/core/tests/summary_test.cc b/core/tests/summary_test.cc index f3549fb5..0e1c214f 100644 --- a/core/tests/summary_test.cc +++ b/core/tests/summary_test.cc @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -71,6 +72,23 @@ TEST(SummaryTest, quantile_values) { EXPECT_NEAR(s.quantile.at(2).value, 0.99 * SAMPLES, 0.001 * SAMPLES); } +TEST(SummaryTest, single_quantile_with_ascending_observations) { + constexpr double quantile = 0.9; + constexpr double error = 0.01; + + for (const int samples : {100, 500}) { + SCOPED_TRACE(samples); + Summary summary{Summary::Quantiles{{quantile, error}}, + std::chrono::hours{1}}; + for (int i = 1; i <= samples; ++i) summary.Observe(i); + + auto metric = summary.Collect(); + const auto& s = metric.summary; + ASSERT_EQ(s.quantile.size(), 1U); + EXPECT_NEAR(s.quantile.at(0).value, quantile * samples, error * samples); + } +} + TEST(SummaryTest, max_age) { Summary summary{Summary::Quantiles{{0.99, 0.001}}, std::chrono::seconds(1), 2}; @@ -102,25 +120,50 @@ TEST(SummaryTest, construction_with_dynamic_quantile_vector) { summary.Observe(8.0); } +TEST(SummaryTest, compress_keeps_larger_value_on_merge) { + // With q=1.0 the two samples are eligible to be merged in compress(). + // The surviving sample must retain the larger value, not the smaller one. + Summary summary{Summary::Quantiles{{1.0, 0.001}}, std::chrono::hours{1}}; + summary.Observe(1.0); + summary.Observe(100.0); + auto metric = summary.Collect(); + auto s = metric.summary; + ASSERT_EQ(s.quantile.size(), 1U); + EXPECT_DOUBLE_EQ(s.quantile.at(0).value, 100.0); +} + +TEST(SummaryTest, insert_preserves_double_precision) { + // 2^24+1 is not exactly representable as float, so a truncating + // double->float->double round trip would corrupt the stored value. + Summary summary{Summary::Quantiles{{1.0, 0.001}}, std::chrono::hours{1}}; + const double v = 16777217.0; + summary.Observe(1.0); + summary.Observe(v); + auto metric = summary.Collect(); + auto s = metric.summary; + ASSERT_EQ(s.quantile.size(), 1U); + EXPECT_DOUBLE_EQ(s.quantile.at(0).value, v); +} + TEST(SummaryTest, quantile_with_out_of_order_batches) { - // Flush large values into the sample first, then insert smaller values so - // that insertBatch() hits the --idx path (value < sample_[item].value). + // Insert small values into a sample that already contains larger values. Summary summary{Summary::Quantiles{{0.5, 0.05}}, std::chrono::hours{1}}; for (int i = 0; i < 10; ++i) summary.Observe(100.0 + i); - summary.Collect(); // flushes buffer → sample_ now contains large values + summary.Collect(); // Flush the first batch before inserting smaller values. for (int i = 0; i < 10; ++i) summary.Observe(1.0 + i); - auto metric = summary.Collect(); // flushes buffer with small values < existing sample → --idx + auto metric = summary.Collect(); auto s = metric.summary; // 20 observations total, sum = (1..10) + (100..109) = 55 + 1045 = 1100 EXPECT_EQ(s.sample_count, 20U); EXPECT_DOUBLE_EQ(s.sample_sum, 1100.0); - // p50 with error 0.05 on 20 samples: rank error ≤ 1, true median is 10 or 100 + // p50 with error 0.05 permits ranks 9..11: values 9, 10, or 100. ASSERT_EQ(s.quantile.size(), 1U); - EXPECT_GE(s.quantile.at(0).value, 1.0); - EXPECT_LE(s.quantile.at(0).value, 109.0); + const auto value = s.quantile.at(0).value; + EXPECT_TRUE(value == 9.0 || value == 10.0 || value == 100.0) + << "Unexpected p50 value: " << value; } } // namespace