utils: add a powers of two histogram implementation

This histogram type avoids using `hdr_hist` to generate seastar
histograms in metrics. It instead implements a simple powers of
two histogram that reduces memory utilization from kilobytes to
`number_of_buckets` * sizeof(uint64_t).

src/v/utils/log_hist.h

@@ -0,0 +1,188 @@
+/*
+ * Copyright 2023 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include <seastar/core/metrics_types.hh>
+#include <seastar/core/shared_ptr.hh>
+
+#include <boost/intrusive/list.hpp>
+
+#include <bit>
+#include <chrono>
+#include <cstdint>
+#include <vector>
+
+/*
+ * A histogram implementation
+ * The buckets upper bounds are powers of 2 minus 1.
+ * `first_bucket_upper_bound` therefore must be a power of 2.
+ * The number of values represented by each bucket increases by powers 2 as
+ * well.
+ *
+ * Assume `number_of_buckets` is 4 and `first_bucket_upper_bound` is 16 the
+ * bucket value ranges are;
+ *
+ * [1, 16), [16, 32), [32, 64), [64, 128)
+ *
+ * And if 1, 16, 32, and 33 are recorded the buckets will have the following
+ * counts;
+ *
+ * [1, 16)   = 1
+ * [16, 32)  = 1
+ * [32, 64)  = 2
+ * [64, 128) = 0
+ */
+template<
+  typename duration_t,
+  int number_of_buckets,
+  uint64_t first_bucket_upper_bound>
+class log_hist {
+    static_assert(
+      first_bucket_upper_bound >= 1
+        && (first_bucket_upper_bound & (first_bucket_upper_bound - 1)) == 0,
+      "first bucket bound must be power of 2");
+
+    using measurement_canary_t = seastar::lw_shared_ptr<bool>;
+
+public:
+    static constexpr int first_bucket_clz = std::countl_zero(
+      first_bucket_upper_bound - 1);
+    static constexpr int first_bucket_exp = 64 - first_bucket_clz;
+
+    using clock_type = std::chrono::high_resolution_clock;
+
+    /// \brief move-only type to tracking durations
+    /// if the log_hist ptr goes out of scope, it will detach itself
+    /// and the recording will simply be ignored.
+    class measurement {
+    public:
+        explicit measurement(log_hist& h)
+          : _canary(h._canary)
+          , _h(std::ref(h))
+          , _begin_t(log_hist::clock_type::now()) {}
+        measurement(const measurement&) = delete;
+        measurement& operator=(const measurement&) = delete;
+        measurement(measurement&& o) noexcept
+          : _canary(o._canary)
+          , _h(o._h)
+          , _begin_t(o._begin_t) {
+            o.cancel();
+        }
+        measurement& operator=(measurement&& o) noexcept {
+            if (this != &o) {
+                this->~measurement();
+                new (this) measurement(std::move(o));
+            }
+            return *this;
+        }
+        ~measurement() noexcept {
+            if (_canary && *_canary) {
+                _h.get().record(compute_duration());
+            }
+        }
+
+        // Cancels this measurements and prevents any values from
+        // being recorded to the underlying histogram.
+        void cancel() { _canary = nullptr; }
+
+    private:
+        int64_t compute_duration() const {
+            return std::chrono::duration_cast<duration_t>(
+                     log_hist::clock_type::now() - _begin_t)
+              .count();
+        }
+
+        measurement_canary_t _canary;
+        std::reference_wrapper<log_hist> _h;
+        log_hist::clock_type::time_point _begin_t;
+    };
+
+    std::unique_ptr<measurement> auto_measure() {
+        return std::make_unique<measurement>(*this);
+    }
+
+    log_hist()
+      : _canary(seastar::make_lw_shared(true))
+      , _counts(number_of_buckets) {}
+
+    ~log_hist() {
+        // Notify any active measurements that this object no longer exists.
+        *_canary = false;
+    }
+
+    /*
+     * record expects values of that are equivalent to `duration_t::count()`
+     * so make sure the input is scaled correctly.
+     */
+    void record(uint64_t val) {
+        _sample_sum += val;
+        const int i = std::clamp(
+          first_bucket_clz - std::countl_zero(val),
+          0,
+          static_cast<int>(_counts.size() - 1));
+        _counts[i]++;
+    }
+
+    void record(std::unique_ptr<measurement> m) {
+        record(m->compute_duration());
+    }
+
+    seastar::metrics::histogram seastar_histogram_logform(int64_t scale) const {
+        seastar::metrics::histogram hist;
+        hist.buckets.resize(_counts.size());
+        hist.sample_sum = static_cast<double>(_sample_sum)
+                          / static_cast<double>(scale);
+
+        uint64_t cumulative_count = 0;
+        for (uint64_t i = 0; i < _counts.size(); i++) {
+            auto& bucket = hist.buckets[i];
+
+            cumulative_count += _counts[i];
+            bucket.count = cumulative_count;
+            uint64_t unscaled_upper_bound = ((uint64_t)1
+                                             << (first_bucket_exp + i))
+                                            - 1;
+            bucket.upper_bound = static_cast<double>(unscaled_upper_bound)
+                                 / static_cast<double>(scale);
+        }
+
+        hist.sample_count = cumulative_count;
+        return hist;
+    }
+
+private:
+    friend measurement;
+
+    // Used to inform measurements whether `log_hist` has been destroyed
+    measurement_canary_t _canary;
+
+    std::vector<uint64_t> _counts;
+    uint64_t _sample_sum{0};
+};
+
+/*
+ * This histogram produces indentical results as the public metric's `hdr_hist`.
+ * So if this histogram and `hdr_hist` are create and have the same values
+ * recorded to them then `log_hist_public::seastar_histogram_logform(1000000)`
+ * will produce the same seastar histogram as
+ * `ssx::metrics::report_default_histogram(hdr_hist)`.
+ */
+using log_hist_public = log_hist<std::chrono::microseconds, 18, 256>;
+static constexpr int64_t log_hist_public_scale = 1'000'000;
+
+/*
+ * This histogram produces results that are similar, but not indentical to the
+ * internal metric's `hdr_hist`. Some of the first buckets will have the
+ * following bounds; [log_hist_internal upper bounds, internal hdr_hist upper
+ * bounds] [8, 10], [16, 20], [32, 41], [64, 83], [128, 167], [256, 335]
+ */
+using log_hist_internal = log_hist<std::chrono::microseconds, 26, 8>;

src/v/utils/tests/seastar_histogram_test.cc

@@ -1,7 +1,14 @@
 #include "utils/hdr_hist.h"
+#include "utils/log_hist.h"
 
+#include <seastar/core/sleep.hh>
 #include <seastar/testing/thread_test_case.hh>
 
+#include <boost/test/tools/old/interface.hpp>
+
+#include <cstdint>
+#include <random>
+
 SEASTAR_THREAD_TEST_CASE(test_seastar_histograms_match) {
     using namespace std::chrono_literals;
 
@@ -20,3 +27,93 @@ SEASTAR_THREAD_TEST_CASE(test_seastar_histograms_match) {
           logform_b.buckets[idx].upper_bound);
     }
 }
+
+namespace {
+bool approximately_equal(double a, double b) {
+    constexpr double precision_error = 0.0001;
+    return std::abs(a - b) <= precision_error;
+}
+
+struct hist_config {
+    int64_t scale;
+    bool use_approximately_equal;
+};
+
+constexpr std::array hist_configs = {
+  hist_config{log_hist_public_scale, true}, hist_config{1, false}};
+
+template<typename l_hist>
+void validate_histograms_equal(const hdr_hist& a, const l_hist& b) {
+    for (auto cfg : hist_configs) {
+        const auto logform_a = a.seastar_histogram_logform(
+          18, 250, 2.0, cfg.scale);
+        const auto logform_b = b.seastar_histogram_logform(cfg.scale);
+
+        BOOST_CHECK_EQUAL(logform_a.sample_count, logform_b.sample_count);
+        if (cfg.use_approximately_equal) {
+            BOOST_CHECK(
+              approximately_equal(logform_a.sample_sum, logform_b.sample_sum));
+        } else {
+            BOOST_CHECK_EQUAL(logform_a.sample_sum, logform_b.sample_sum);
+        }
+
+        for (size_t idx = 0; idx < logform_a.buckets.size(); ++idx) {
+            if (cfg.use_approximately_equal) {
+                BOOST_CHECK(approximately_equal(
+                  logform_a.buckets[idx].upper_bound,
+                  logform_b.buckets[idx].upper_bound));
+            } else {
+                BOOST_CHECK_EQUAL(
+                  logform_a.buckets[idx].upper_bound,
+                  logform_b.buckets[idx].upper_bound);
+            }
+            BOOST_CHECK_EQUAL(
+              logform_a.buckets[idx].count, logform_b.buckets[idx].count);
+        }
+    }
+}
+} // namespace
+
+// ensures both the log_hist_public and the public hdr_hist return identical
+// seastar histograms for values recorded around bucket bounds.
+SEASTAR_THREAD_TEST_CASE(test_public_log_hist_and_hdr_hist_equal_bounds) {
+    using namespace std::chrono_literals;
+
+    hdr_hist a;
+    log_hist_public b;
+
+    a.record(1);
+    b.record(1);
+
+    for (unsigned i = 0; i < 17; i++) {
+        auto upper_bound
+          = (((unsigned)1 << (log_hist_public::first_bucket_exp + i)) - 1);
+        a.record(upper_bound);
+        a.record(upper_bound + 1);
+        b.record(upper_bound);
+        b.record(upper_bound + 1);
+    }
+
+    validate_histograms_equal(a, b);
+}
+
+// ensures both the log_hist_public and the public hdr_hist return identical
+// seastar histograms for randomly selected values.
+SEASTAR_THREAD_TEST_CASE(test_public_log_hist_and_hdr_hist_equal_rand) {
+    using namespace std::chrono_literals;
+
+    hdr_hist a;
+    log_hist_public b;
+
+    std::random_device rd;
+    std::mt19937 gen(rd());
+    std::uniform_int_distribution<uint64_t> d(1, (1 << (8 + 17)) - 1);
+
+    for (unsigned i = 0; i < 1'000'000; i++) {
+        auto sample = d(gen);
+        a.record(sample);
+        b.record(sample);
+    }
+
+    validate_histograms_equal(a, b);
+}