intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 150 insertions, 0 deletions

diff --git a/library/cpp/monlib/metrics/ewma.cpp b/library/cpp/monlib/metrics/ewma.cpp
new file mode 100644
index 00000000000..8a296c32258
--- /dev/null
+++ b/library/cpp/monlib/metrics/ewma.cpp
@@ -0,0 +1,150 @@
+#include "ewma.h"
+#include "metric.h"
+
+#include <atomic>
+#include <cmath>
+
+namespace NMonitoring {
+namespace {
+    constexpr auto DEFAULT_INTERVAL = TDuration::Seconds(5);
+
+    const double ALPHA1 = 1. - std::exp(-double(DEFAULT_INTERVAL.Seconds())/60./1.);
+    const double ALPHA5 = 1. - std::exp(-double(DEFAULT_INTERVAL.Seconds())/60./5.);
+    const double ALPHA15 = 1. - std::exp(-double(DEFAULT_INTERVAL.Seconds())/60./15.);
+
+    class TExpMovingAverage final: public IExpMovingAverage {
+    public:
+        explicit TExpMovingAverage(IGauge* metric, double alpha, TDuration interval)
+            : Metric_{metric}
+            , Alpha_{alpha}
+            , Interval_{interval.Seconds()}
+        {
+            Y_VERIFY(metric != nullptr, "Passing nullptr metric is not allowed");
+        }
+
+        ~TExpMovingAverage() override = default;
+
+        // This method NOT thread safe
+        void Tick() override {
+            const auto current = Uncounted_.fetch_and(0);
+            const double instantRate = double(current) / Interval_;
+
+            if (Y_UNLIKELY(!IsInitialized())) {
+                Metric_->Set(instantRate);
+                Init_ = true;
+            } else {
+                const double currentRate = Metric_->Get();
+                Metric_->Set(Alpha_ * (instantRate - currentRate) + currentRate);
+            }
+
+        }
+
+        void Update(i64 value) override {
+            Uncounted_ += value;
+        }
+
+        double Rate() const override {
+            return Metric_->Get();
+        }
+
+        void Reset() override {
+            Init_ = false;
+            Uncounted_ = 0;
+        }
+
+    private:
+        bool IsInitialized() const {
+            return Init_;
+        }
+
+    private:
+        std::atomic<i64> Uncounted_{0};
+        std::atomic<bool> Init_{false};
+
+        IGauge* Metric_{nullptr};
+        double Alpha_;
+        ui64 Interval_;
+    };
+
+    struct TFakeEwma: IExpMovingAverage {
+        void Tick() override {}
+        void Update(i64) override {}
+        double Rate() const override { return 0; }
+        void Reset() override {}
+    };
+
+} // namespace
+
+    TEwmaMeter::TEwmaMeter()
+        : Ewma_{MakeHolder<TFakeEwma>()}
+    {
+    }
+
+    TEwmaMeter::TEwmaMeter(IExpMovingAveragePtr&& ewma)
+        : Ewma_{std::move(ewma)}
+    {
+    }
+
+    TEwmaMeter::TEwmaMeter(TEwmaMeter&& other) {
+        if (&other == this) {
+            return;
+        }
+
+        *this = std::move(other);
+    }
+
+    TEwmaMeter& TEwmaMeter::operator=(TEwmaMeter&& other) {
+        Ewma_ = std::move(other.Ewma_);
+        LastTick_.store(other.LastTick_);
+        return *this;
+    }
+
+    void TEwmaMeter::TickIfNeeded() {
+        constexpr ui64 INTERVAL_SECONDS = DEFAULT_INTERVAL.Seconds();
+
+        const auto now = TInstant::Now().Seconds();
+        ui64 old = LastTick_.load();
+        const auto secondsSinceLastTick = now - old;
+
+        if (secondsSinceLastTick > INTERVAL_SECONDS) {
+            // round to the interval grid
+            const ui64 newLast = now - (secondsSinceLastTick % INTERVAL_SECONDS);
+            if (LastTick_.compare_exchange_strong(old, newLast)) {
+                for (size_t i = 0; i < secondsSinceLastTick / INTERVAL_SECONDS; ++i) {
+                    Ewma_->Tick();
+                }
+            }
+        }
+    }
+
+    void TEwmaMeter::Mark() {
+        TickIfNeeded();
+        Ewma_->Update(1);
+    }
+
+    void TEwmaMeter::Mark(i64 value) {
+        TickIfNeeded();
+        Ewma_->Update(value);
+    }
+
+    double TEwmaMeter::Get() {
+        TickIfNeeded();
+        return Ewma_->Rate();
+    }
+
+    IExpMovingAveragePtr OneMinuteEwma(IGauge* metric) {
+        return MakeHolder<TExpMovingAverage>(metric, ALPHA1, DEFAULT_INTERVAL);
+    }
+
+    IExpMovingAveragePtr FiveMinuteEwma(IGauge* metric) {
+        return MakeHolder<TExpMovingAverage>(metric, ALPHA5, DEFAULT_INTERVAL);
+    }
+
+    IExpMovingAveragePtr FiveteenMinuteEwma(IGauge* metric) {
+        return MakeHolder<TExpMovingAverage>(metric, ALPHA15, DEFAULT_INTERVAL);
+    }
+
+    IExpMovingAveragePtr CreateEwma(IGauge* metric, double alpha, TDuration interval) {
+        return MakeHolder<TExpMovingAverage>(metric, alpha, interval);
+    }
+} // namespace NMonitoring