From 7a25e1f7f625dc2371e40d41fe97ed1a3b572233 Mon Sep 17 00:00:00 2001 From: Debjit Date: Thu, 26 Dec 2024 14:32:52 +0530 Subject: [PATCH 1/3] Perf improvement - Summary metric --- core/src/detail/ckms_quantiles.cc | 75 ++++++++++++++++++------------- 1 file changed, 44 insertions(+), 31 deletions(-) diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 395b6ff3..94ce4073 100644 --- a/core/src/detail/ckms_quantiles.cc +++ b/core/src/detail/ckms_quantiles.cc @@ -85,42 +85,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(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; + float value = buffer_[i]; + + auto iterator = std::lower_bound( + sample_.begin(), sample_.end(), value, + [](const Item& item, float 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))) + + 1; } - int delta; - if (idx - 1 == 0 || idx + 1 == sample_.size()) { - delta = 0; - } else { - delta = static_cast(std::floor(allowableError(idx + 1))) + 1; - } - - sample_.emplace(sample_.begin() + idx, v, 1, delta); - count_++; - item = idx++; + sample_.emplace(iterator, value, 1, delta); + ++count_; } buffer_count_ = 0; @@ -128,24 +128,37 @@ 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++; + 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]); - while (idx < sample_.size()) { - prev = next; - next = idx++; + for (std::size_t idx = 1; idx < sample_.size(); ++idx) { + const Item& current_sample = sample_[idx]; + Item& last_compressed_sample = compressed_samples.back(); - if (sample_[prev].g + sample_[next].g + sample_[next].delta <= - allowableError(idx - 1)) { - sample_[next].g += sample_[prev].g; - sample_.erase(sample_.begin() + prev); + // 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(static_cast(compressed_samples.size()) - 1)) { + // Merge current sample into last compressed sample + last_compressed_sample.g += current_sample.g; // Update weight + } 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 From ace740a4fe2604a7b76e56d80dfdec1917392076 Mon Sep 17 00:00:00 2001 From: Gregor Jasny Date: Sat, 29 Aug 2026 21:34:49 +0200 Subject: [PATCH 2/3] chore: add test cases and fix code --- core/src/detail/ckms_quantiles.cc | 11 +++++++---- core/tests/summary_test.cc | 25 +++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/core/src/detail/ckms_quantiles.cc b/core/src/detail/ckms_quantiles.cc index 94ce4073..8e17e811 100644 --- a/core/src/detail/ckms_quantiles.cc +++ b/core/src/detail/ckms_quantiles.cc @@ -105,11 +105,11 @@ bool CKMSQuantiles::insertBatch() { // Loop through the buffer and insert the items into the sample set for (std::size_t i = start; i < buffer_count_; ++i) { - float value = buffer_[i]; + double value = buffer_[i]; auto iterator = std::lower_bound( sample_.begin(), sample_.end(), value, - [](const Item& item, float val) { return item.value < val; }); + [](const Item& item, double val) { return item.value < val; }); std::size_t idx = std::distance(sample_.begin(), iterator); int delta = 0; @@ -149,8 +149,11 @@ void CKMSQuantiles::compress() { // sample if (last_compressed_sample.g + current_sample.g + current_sample.delta <= allowableError(static_cast(compressed_samples.size()) - 1)) { - // Merge current sample into last compressed sample - last_compressed_sample.g += current_sample.g; // Update weight + // 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; } else { // If not compressible, add current sample to compressed samples compressed_samples.push_back(current_sample); diff --git a/core/tests/summary_test.cc b/core/tests/summary_test.cc index f3549fb5..e5bf0ee3 100644 --- a/core/tests/summary_test.cc +++ b/core/tests/summary_test.cc @@ -102,6 +102,31 @@ 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). From e1e26448ddd7a366486cf4458a18ffbcc536c877 Mon Sep 17 00:00:00 2001 From: Debjit Mondal Date: Tue, 15 Sep 2026 07:10:13 -0700 Subject: [PATCH 3/3] Fix CKMS compression accounting and strengthen quantile coverage --- .../prometheus/detail/ckms_quantiles.h | 2 +- core/src/detail/ckms_quantiles.cc | 18 +++++++---- core/tests/summary_test.cc | 32 +++++++++++++++---- 3 files changed, 37 insertions(+), 15 deletions(-) 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 8e17e811..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()) { @@ -95,7 +94,7 @@ bool CKMSQuantiles::insertBatch() { std::size_t start = 0; - sample_.reserve(buffer_count_); + 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); @@ -114,8 +113,8 @@ bool CKMSQuantiles::insertBatch() { int delta = 0; if (idx > 0 && idx < sample_.size()) { - delta = static_cast( - std::floor(allowableError(static_cast(idx) + 1))) + + delta = static_cast(std::floor( + allowableError(static_cast(idx) + 1, sample_.size()))) + 1; } @@ -141,19 +140,24 @@ void CKMSQuantiles::compress() { // 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(static_cast(compressed_samples.size()) - 1)) { + 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); diff --git a/core/tests/summary_test.cc b/core/tests/summary_test.cc index e5bf0ee3..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}; @@ -128,24 +146,24 @@ TEST(SummaryTest, insert_preserves_double_precision) { } 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