intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 556 insertions, 0 deletions

diff --git a/library/cpp/monlib/encode/json/json_encoder.cpp b/library/cpp/monlib/encode/json/json_encoder.cpp
new file mode 100644
index 0000000000..20d2bb6283
--- /dev/null
+++ b/library/cpp/monlib/encode/json/json_encoder.cpp
@@ -0,0 +1,556 @@
+#include "json.h"
+#include "typed_point.h"
+
+#include <library/cpp/monlib/encode/buffered/buffered_encoder_base.h>
+#include <library/cpp/monlib/encode/encoder_state.h>
+#include <library/cpp/monlib/metrics/metric.h>
+#include <library/cpp/monlib/metrics/metric_value.h>
+#include <library/cpp/monlib/metrics/labels.h>
+
+#include <library/cpp/json/writer/json.h>
+
+#include <util/charset/utf8.h>
+#include <util/generic/algorithm.h>
+
+namespace NMonitoring {
+    namespace {
+        enum class EJsonStyle {
+            Solomon,
+            Cloud
+        };
+
+        ///////////////////////////////////////////////////////////////////////
+        // TJsonWriter
+        ///////////////////////////////////////////////////////////////////////
+        class TJsonWriter {
+        public:
+            TJsonWriter(IOutputStream* out, int indentation, EJsonStyle style, TStringBuf metricNameLabel)
+                : Buf_(NJsonWriter::HEM_UNSAFE, out)
+                , Style_(style)
+                , MetricNameLabel_(metricNameLabel)
+                , CurrentMetricName_()
+            {
+                Buf_.SetIndentSpaces(indentation);
+                Buf_.SetWriteNanAsString();
+            }
+
+            void WriteTime(TInstant time) {
+                if (time != TInstant::Zero()) {
+                    Buf_.WriteKey(TStringBuf("ts"));
+                    if (Style_ == EJsonStyle::Solomon) {
+                        Buf_.WriteULongLong(time.Seconds());
+                    } else {
+                        Buf_.WriteString(time.ToString());
+                    }
+                }
+            }
+
+            void WriteValue(double value) {
+                Buf_.WriteKey(TStringBuf("value"));
+                Buf_.WriteDouble(value);
+            }
+
+            void WriteValue(i64 value) {
+                Buf_.WriteKey(TStringBuf("value"));
+                Buf_.WriteLongLong(value);
+            }
+
+            void WriteValue(ui64 value) {
+                Buf_.WriteKey(TStringBuf("value"));
+                Buf_.WriteULongLong(value);
+            }
+
+            void WriteValue(IHistogramSnapshot* s) {
+                Y_ENSURE(Style_ == EJsonStyle::Solomon);
+
+                Buf_.WriteKey(TStringBuf("hist"));
+                Buf_.BeginObject();
+                if (ui32 count = s->Count()) {
+                    bool hasInf = (s->UpperBound(count - 1) == Max<double>());
+                    if (hasInf) {
+                        count--;
+                    }
+
+                    Buf_.WriteKey(TStringBuf("bounds"));
+                    Buf_.BeginList();
+                    for (ui32 i = 0; i < count; i++) {
+                        Buf_.WriteDouble(s->UpperBound(i));
+                    }
+                    Buf_.EndList();
+
+                    Buf_.WriteKey(TStringBuf("buckets"));
+                    Buf_.BeginList();
+                    for (ui32 i = 0; i < count; i++) {
+                        Buf_.WriteULongLong(s->Value(i));
+                    }
+                    Buf_.EndList();
+
+                    if (hasInf) {
+                        Buf_.WriteKey(TStringBuf("inf"));
+                        Buf_.WriteULongLong(s->Value(count));
+                    }
+                }
+                Buf_.EndObject();
+            }
+
+            void WriteValue(ISummaryDoubleSnapshot* s) {
+                Y_ENSURE(Style_ == EJsonStyle::Solomon);
+
+                Buf_.WriteKey(TStringBuf("summary"));
+                Buf_.BeginObject();
+
+                Buf_.WriteKey(TStringBuf("sum"));
+                Buf_.WriteDouble(s->GetSum());
+
+                Buf_.WriteKey(TStringBuf("min"));
+                Buf_.WriteDouble(s->GetMin());
+
+                Buf_.WriteKey(TStringBuf("max"));
+                Buf_.WriteDouble(s->GetMax());
+
+                Buf_.WriteKey(TStringBuf("last"));
+                Buf_.WriteDouble(s->GetLast());
+
+                Buf_.WriteKey(TStringBuf("count"));
+                Buf_.WriteULongLong(s->GetCount());
+
+                Buf_.EndObject();
+            }
+
+            void WriteValue(TLogHistogramSnapshot* s) {
+                Y_ENSURE(Style_ == EJsonStyle::Solomon);
+
+                Buf_.WriteKey(TStringBuf("log_hist"));
+                Buf_.BeginObject();
+
+                Buf_.WriteKey(TStringBuf("base"));
+                Buf_.WriteDouble(s->Base());
+
+                Buf_.WriteKey(TStringBuf("zeros_count"));
+                Buf_.WriteULongLong(s->ZerosCount());
+
+                Buf_.WriteKey(TStringBuf("start_power"));
+                Buf_.WriteInt(s->StartPower());
+
+                Buf_.WriteKey(TStringBuf("buckets"));
+                Buf_.BeginList();
+                for (size_t i = 0; i < s->Count(); ++i) {
+                    Buf_.WriteDouble(s->Bucket(i));
+                }
+                Buf_.EndList();
+
+                Buf_.EndObject();
+            }
+
+            void WriteValue(EMetricValueType type, TMetricValue value) {
+                switch (type) {
+                    case EMetricValueType::DOUBLE:
+                        WriteValue(value.AsDouble());
+                        break;
+
+                    case EMetricValueType::INT64:
+                        WriteValue(value.AsInt64());
+                        break;
+
+                    case EMetricValueType::UINT64:
+                        WriteValue(value.AsUint64());
+                        break;
+
+                    case EMetricValueType::HISTOGRAM:
+                        WriteValue(value.AsHistogram());
+                        break;
+
+                    case EMetricValueType::SUMMARY:
+                        WriteValue(value.AsSummaryDouble());
+                        break;
+
+                    case EMetricValueType::LOGHISTOGRAM:
+                        WriteValue(value.AsLogHistogram());
+                        break;
+
+                    case EMetricValueType::UNKNOWN:
+                        ythrow yexception() << "unknown metric value type";
+                }
+            }
+
+            void WriteLabel(TStringBuf name, TStringBuf value) {
+                Y_ENSURE(IsUtf(name), "label name is not valid UTF-8 string");
+                Y_ENSURE(IsUtf(value), "label value is not valid UTF-8 string");
+                if (Style_ == EJsonStyle::Cloud && name == MetricNameLabel_) {
+                    CurrentMetricName_ = value;
+                } else {
+                    Buf_.WriteKey(name);
+                    Buf_.WriteString(value);
+                }
+            }
+
+            void WriteMetricType(EMetricType type) {
+                if (Style_ == EJsonStyle::Cloud) {
+                    Buf_.WriteKey("type");
+                    Buf_.WriteString(MetricTypeToCloudStr(type));
+                } else {
+                    Buf_.WriteKey("kind");
+                    Buf_.WriteString(MetricTypeToStr(type));
+                }
+            }
+
+            void WriteName() {
+                if (Style_ != EJsonStyle::Cloud) {
+                    return;
+                }
+                if (CurrentMetricName_.Empty()) {
+                    ythrow yexception() << "label '" << MetricNameLabel_ << "' is not defined";
+                }
+                Buf_.WriteKey("name");
+                Buf_.WriteString(CurrentMetricName_);
+                CurrentMetricName_.clear();
+            }
+
+        private:
+            static TStringBuf MetricTypeToCloudStr(EMetricType type) {
+                switch (type) {
+                    case EMetricType::GAUGE:
+                        return TStringBuf("DGAUGE");
+                    case EMetricType::COUNTER:
+                        return TStringBuf("COUNTER");
+                    case EMetricType::RATE:
+                        return TStringBuf("RATE");
+                    case EMetricType::IGAUGE:
+                        return TStringBuf("IGAUGE");
+                    default:
+                        ythrow yexception() << "metric type '" << type << "' is not supported by cloud json format";
+                }
+            }
+
+        protected:
+            NJsonWriter::TBuf Buf_;
+            EJsonStyle Style_;
+            TString MetricNameLabel_;
+            TString CurrentMetricName_;
+        };
+
+        ///////////////////////////////////////////////////////////////////////
+        // TEncoderJson
+        ///////////////////////////////////////////////////////////////////////
+        class TEncoderJson final: public IMetricEncoder, public TJsonWriter {
+        public:
+            TEncoderJson(IOutputStream* out, int indentation, EJsonStyle style, TStringBuf metricNameLabel)
+                : TJsonWriter{out, indentation, style, metricNameLabel}
+            {
+            }
+
+            ~TEncoderJson() override {
+                Close();
+            }
+
+        private:
+            void OnStreamBegin() override {
+                State_.Expect(TEncoderState::EState::ROOT);
+                Buf_.BeginObject();
+            }
+
+            void OnStreamEnd() override {
+                State_.Expect(TEncoderState::EState::ROOT);
+                if (!Buf_.KeyExpected()) {
+                    // not closed metrics array
+                    Buf_.EndList();
+                }
+                Buf_.EndObject();
+            }
+
+            void OnCommonTime(TInstant time) override {
+                State_.Expect(TEncoderState::EState::ROOT);
+                WriteTime(time);
+            }
+
+            void OnMetricBegin(EMetricType type) override {
+                State_.Switch(TEncoderState::EState::ROOT, TEncoderState::EState::METRIC);
+                if (Buf_.KeyExpected()) {
+                    // first metric, so open metrics array
+                    Buf_.WriteKey(TStringBuf(Style_ == EJsonStyle::Solomon ? "sensors" : "metrics"));
+                    Buf_.BeginList();
+                }
+                Buf_.BeginObject();
+                WriteMetricType(type);
+            }
+
+            void OnMetricEnd() override {
+                State_.Switch(TEncoderState::EState::METRIC, TEncoderState::EState::ROOT);
+                if (!Buf_.KeyExpected()) {
+                    // not closed timeseries array
+                    Buf_.EndList();
+                }
+
+                if (!TimeSeries_ && LastPoint_.HasValue()) {
+                    // we have seen only one point between OnMetricBegin() and
+                    // OnMetricEnd() calls
+                    WriteTime(LastPoint_.GetTime());
+                    WriteValue(LastPoint_.GetValueType(), LastPoint_.GetValue());
+                }
+                Buf_.EndObject();
+
+                LastPoint_ = {};
+                TimeSeries_ = false;
+            }
+
+            void OnLabelsBegin() override {
+                if (!Buf_.KeyExpected()) {
+                    // not closed metrics or timeseries array if labels go after values
+                    Buf_.EndList();
+                }
+                if (State_ == TEncoderState::EState::ROOT) {
+                    State_ = TEncoderState::EState::COMMON_LABELS;
+                    Buf_.WriteKey(TStringBuf(Style_ == EJsonStyle::Solomon ? "commonLabels" : "labels"));
+                } else if (State_ == TEncoderState::EState::METRIC) {
+                    State_ = TEncoderState::EState::METRIC_LABELS;
+                    Buf_.WriteKey(TStringBuf("labels"));
+                } else {
+                    State_.ThrowInvalid("expected METRIC or ROOT");
+                }
+                Buf_.BeginObject();
+
+                EmptyLabels_ = true;
+            }
+
+            void OnLabelsEnd() override {
+                if (State_ == TEncoderState::EState::METRIC_LABELS) {
+                    State_ = TEncoderState::EState::METRIC;
+                } else if (State_ == TEncoderState::EState::COMMON_LABELS) {
+                    State_ = TEncoderState::EState::ROOT;
+                } else {
+                    State_.ThrowInvalid("expected LABELS or COMMON_LABELS");
+                }
+
+                Y_ENSURE(!EmptyLabels_, "Labels cannot be empty");
+                Buf_.EndObject();
+                if (State_ == TEncoderState::EState::METRIC) {
+                    WriteName();
+                }
+            }
+
+            void OnLabel(TStringBuf name, TStringBuf value) override {
+                if (State_ == TEncoderState::EState::METRIC_LABELS || State_ == TEncoderState::EState::COMMON_LABELS) {
+                    WriteLabel(name, value);
+                } else {
+                    State_.ThrowInvalid("expected LABELS or COMMON_LABELS");
+                }
+
+                EmptyLabels_ = false;
+            }
+
+            void OnDouble(TInstant time, double value) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<double>(time, value);
+            }
+
+            void OnInt64(TInstant time, i64 value) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<i64>(time, value);
+            }
+
+            void OnUint64(TInstant time, ui64 value) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<ui64>(time, value);
+            }
+
+            void OnHistogram(TInstant time, IHistogramSnapshotPtr snapshot) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<IHistogramSnapshot*>(time, snapshot.Get());
+            }
+
+            void OnSummaryDouble(TInstant time, ISummaryDoubleSnapshotPtr snapshot) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<ISummaryDoubleSnapshot*>(time, snapshot.Get());
+            }
+
+            void OnLogHistogram(TInstant time, TLogHistogramSnapshotPtr snapshot) override {
+                State_.Expect(TEncoderState::EState::METRIC);
+                Write<TLogHistogramSnapshot*>(time, snapshot.Get());
+            }
+
+            template <typename T>
+            void Write(TInstant time, T value) {
+                State_.Expect(TEncoderState::EState::METRIC);
+
+                if (!LastPoint_.HasValue()) {
+                    LastPoint_ = {time, value};
+                } else {
+                    // second point
+                    // TODO: output types
+                    Y_ENSURE(LastPoint_.GetValueType() == TValueType<T>::Type,
+                             "mixed metric value types in one metric");
+
+                    if (!TimeSeries_) {
+                        Buf_.WriteKey(TStringBuf("timeseries"));
+                        Buf_.BeginList();
+                        Buf_.BeginObject();
+                        Y_ENSURE(LastPoint_.GetTime() != TInstant::Zero(),
+                            "time cannot be empty or zero in a timeseries point");
+                        WriteTime(LastPoint_.GetTime());
+                        WriteValue(LastPoint_.GetValueType(), LastPoint_.GetValue());
+                        Buf_.EndObject();
+                        TimeSeries_ = true;
+                    }
+
+                    if (TimeSeries_) {
+                        Buf_.BeginObject();
+                        Y_ENSURE(time != TInstant::Zero(),
+                            "time cannot be empty or zero in a timeseries point");
+
+                        WriteTime(time);
+                        WriteValue(value);
+                        Buf_.EndObject();
+                    }
+                }
+            }
+
+            void Close() override {
+                LastPoint_ = {};
+            }
+
+        private:
+            TEncoderState State_;
+            TTypedPoint LastPoint_;
+            bool TimeSeries_ = false;
+            bool EmptyLabels_ = false;
+        };
+
+        ///////////////////////////////////////////////////////////////////////
+        // TBufferedJsonEncoder
+        ///////////////////////////////////////////////////////////////////////
+        class TBufferedJsonEncoder : public TBufferedEncoderBase, public TJsonWriter {
+        public:
+            TBufferedJsonEncoder(IOutputStream* out, int indentation, EJsonStyle style, TStringBuf metricNameLabel)
+                : TJsonWriter{out, indentation, style, metricNameLabel}
+            {
+                MetricsMergingMode_ = EMetricsMergingMode::MERGE_METRICS;
+            }
+
+            ~TBufferedJsonEncoder() override {
+                Close();
+            }
+
+            void OnLabelsBegin() override {
+                TBufferedEncoderBase::OnLabelsBegin();
+                EmptyLabels_ = true;
+            }
+
+            void OnLabel(TStringBuf name, TStringBuf value) override {
+                TBufferedEncoderBase::OnLabel(name, value);
+                EmptyLabels_ = false;
+            }
+
+            void OnLabel(ui32 name, ui32 value) override {
+                TBufferedEncoderBase::OnLabel(name, value);
+                EmptyLabels_ = false;
+            }
+
+            void OnLabelsEnd() override {
+                TBufferedEncoderBase::OnLabelsEnd();
+                Y_ENSURE(!EmptyLabels_, "Labels cannot be empty");
+            }
+
+            void Close() final {
+                if (Closed_) {
+                    return;
+                }
+
+                Closed_ = true;
+
+                LabelValuesPool_.Build();
+                LabelNamesPool_.Build();
+
+                Buf_.BeginObject();
+
+                WriteTime(CommonTime_);
+                if (CommonLabels_.size() > 0) {
+                    Buf_.WriteKey(TStringBuf(Style_ == EJsonStyle::Solomon ? "commonLabels": "labels"));
+                    WriteLabels(CommonLabels_, true);
+                }
+
+                if (Metrics_.size() > 0) {
+                    Buf_.WriteKey(TStringBuf(Style_ == EJsonStyle::Solomon ? "sensors" : "metrics"));
+                    WriteMetrics();
+                }
+
+                Buf_.EndObject();
+            }
+
+        private:
+            void WriteMetrics() {
+                Buf_.BeginList();
+                for (auto&& metric : Metrics_) {
+                    WriteMetric(metric);
+                }
+                Buf_.EndList();
+            }
+
+            void WriteMetric(TMetric& metric) {
+                Buf_.BeginObject();
+
+                WriteMetricType(metric.MetricType);
+
+                Buf_.WriteKey(TStringBuf("labels"));
+                WriteLabels(metric.Labels, false);
+
+                metric.TimeSeries.SortByTs();
+                if (metric.TimeSeries.Size() == 1) {
+                    const auto& point = metric.TimeSeries[0];
+                    WriteTime(point.GetTime());
+                    WriteValue(metric.TimeSeries.GetValueType(), point.GetValue());
+                } else if (metric.TimeSeries.Size() > 1) {
+                    Buf_.WriteKey(TStringBuf("timeseries"));
+                    Buf_.BeginList();
+                    metric.TimeSeries.ForEach([this](TInstant time, EMetricValueType type, TMetricValue value) {
+                        Buf_.BeginObject();
+                        // make gcc 6.1 happy https://gcc.gnu.org/bugzilla/show_bug.cgi?id=61636
+                        this->WriteTime(time);
+                        this->WriteValue(type, value);
+                        Buf_.EndObject();
+                    });
+
+                    Buf_.EndList();
+                }
+
+                Buf_.EndObject();
+            }
+
+            void WriteLabels(const TPooledLabels& labels, bool isCommon) {
+                Buf_.BeginObject();
+
+                for (auto i = 0u; i < labels.size(); ++i) {
+                    TStringBuf name = LabelNamesPool_.Get(labels[i].Key->Index);
+                    TStringBuf value = LabelValuesPool_.Get(labels[i].Value->Index);
+
+                    WriteLabel(name, value);
+                }
+
+                Buf_.EndObject();
+
+                if (!isCommon) {
+                    WriteName();
+                }
+            }
+
+        private:
+            bool Closed_{false};
+            bool EmptyLabels_ = false;
+        };
+    }
+
+    IMetricEncoderPtr EncoderJson(IOutputStream* out, int indentation) {
+        return MakeHolder<TEncoderJson>(out, indentation, EJsonStyle::Solomon, "");
+    }
+
+    IMetricEncoderPtr BufferedEncoderJson(IOutputStream* out, int indentation) {
+        return MakeHolder<TBufferedJsonEncoder>(out, indentation, EJsonStyle::Solomon, "");
+    }
+
+    IMetricEncoderPtr EncoderCloudJson(IOutputStream* out, int indentation, TStringBuf metricNameLabel) {
+        return MakeHolder<TEncoderJson>(out, indentation, EJsonStyle::Cloud, metricNameLabel);
+    }
+
+    IMetricEncoderPtr BufferedEncoderCloudJson(IOutputStream* out, int indentation, TStringBuf metricNameLabel) {
+        return MakeHolder<TBufferedJsonEncoder>(out, indentation, EJsonStyle::Cloud, metricNameLabel);
+    }
+}