123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161 |
- #include "json.h"
- #include "typed_point.h"
- #include <library/cpp/monlib/exception/exception.h>
- #include <library/cpp/monlib/metrics/labels.h>
- #include <library/cpp/monlib/metrics/metric_value.h>
- #include <library/cpp/json/json_reader.h>
- #include <util/datetime/base.h>
- #include <util/string/cast.h>
- #include <limits>
- namespace NMonitoring {
- #define DECODE_ENSURE(COND, ...) MONLIB_ENSURE_EX(COND, TJsonDecodeError() << __VA_ARGS__)
- namespace {
- ///////////////////////////////////////////////////////////////////////
- // THistogramBuilder
- ///////////////////////////////////////////////////////////////////////
- class THistogramBuilder {
- public:
- void AddBound(TBucketBound bound) {
- if (!Bounds_.empty()) {
- DECODE_ENSURE(Bounds_.back() < bound,
- "non sorted bounds, " << Bounds_.back() <<
- " >= " << bound);
- }
- Bounds_.push_back(bound);
- }
- void AddValue(TBucketValue value) {
- Values_.push_back(value);
- }
- void AddInf(TBucketValue value) {
- InfPresented_ = true;
- InfValue_ = value;
- }
- IHistogramSnapshotPtr Build() {
- if (InfPresented_) {
- Bounds_.push_back(Max<TBucketBound>());
- Values_.push_back(InfValue_);
- }
- auto snapshot = ExplicitHistogramSnapshot(Bounds_, Values_, true);
- Bounds_.clear();
- Values_.clear();
- InfPresented_ = false;
- return snapshot;
- }
- bool Empty() const noexcept {
- return Bounds_.empty() && Values_.empty();
- }
- void Clear() {
- Bounds_.clear();
- Values_.clear();
- }
- private:
- TBucketBounds Bounds_;
- TBucketValues Values_;
- bool InfPresented_ = false;
- TBucketValue InfValue_;
- };
- class TSummaryDoubleBuilder {
- public:
- ISummaryDoubleSnapshotPtr Build() const {
- return MakeIntrusive<TSummaryDoubleSnapshot>(Sum_, Min_, Max_, Last_, Count_);
- }
- void SetSum(double sum) {
- Empty_ = false;
- Sum_ = sum;
- }
- void SetMin(double min) {
- Empty_ = false;
- Min_ = min;
- }
- void SetMax(double max) {
- Empty_ = false;
- Max_ = max;
- }
- void SetLast(double last) {
- Empty_ = false;
- Last_ = last;
- }
- void SetCount(ui64 count) {
- Empty_ = false;
- Count_ = count;
- }
- void Clear() {
- Empty_ = true;
- Sum_ = 0;
- Min_ = 0;
- Max_ = 0;
- Last_ = 0;
- Count_ = 0;
- }
- bool Empty() const {
- return Empty_;
- }
- private:
- double Sum_ = 0;
- double Min_ = 0;
- double Max_ = 0;
- double Last_ = 0;
- ui64 Count_ = 0;
- bool Empty_ = true;
- };
- class TLogHistogramBuilder {
- public:
- void SetBase(double base) {
- DECODE_ENSURE(base > 0, "base must be positive");
- Base_ = base;
- }
- void SetZerosCount(ui64 zerosCount) {
- ZerosCount_ = zerosCount;
- }
- void SetStartPower(int startPower) {
- StartPower_ = startPower;
- }
- void AddBucketValue(double value) {
- DECODE_ENSURE(value > 0.0, "bucket values must be positive");
- DECODE_ENSURE(value < std::numeric_limits<double>::max(), "bucket values must be finite");
- Buckets_.push_back(value);
- }
- void Clear() {
- Buckets_.clear();
- Base_ = 1.5;
- ZerosCount_ = 0;
- StartPower_ = 0;
- }
- bool Empty() const {
- return Buckets_.empty() && ZerosCount_ == 0;
- }
- TLogHistogramSnapshotPtr Build() {
- return MakeIntrusive<TLogHistogramSnapshot>(Base_, ZerosCount_, StartPower_, std::move(Buckets_));
- }
- private:
- double Base_ = 1.5;
- ui64 ZerosCount_ = 0;
- int StartPower_ = 0;
- TVector<double> Buckets_;
- };
- std::pair<double, bool> ParseSpecDouble(TStringBuf string) {
- if (string == TStringBuf("nan") || string == TStringBuf("NaN")) {
- return {std::numeric_limits<double>::quiet_NaN(), true};
- } else if (string == TStringBuf("inf") || string == TStringBuf("Infinity")) {
- return {std::numeric_limits<double>::infinity(), true};
- } else if (string == TStringBuf("-inf") || string == TStringBuf("-Infinity")) {
- return {-std::numeric_limits<double>::infinity(), true};
- } else {
- return {0, false};
- }
- }
- ///////////////////////////////////////////////////////////////////////
- // TMetricCollector
- ///////////////////////////////////////////////////////////////////////
- struct TMetricCollector {
- EMetricType Type = EMetricType::UNKNOWN;
- TLabels Labels;
- THistogramBuilder HistogramBuilder;
- TSummaryDoubleBuilder SummaryBuilder;
- TLogHistogramBuilder LogHistBuilder;
- TTypedPoint LastPoint;
- TVector<TTypedPoint> TimeSeries;
- bool SeenTsOrValue = false;
- bool SeenTimeseries = false;
- void Clear() {
- Type = EMetricType::UNKNOWN;
- Labels.Clear();
- SeenTsOrValue = false;
- SeenTimeseries = false;
- TimeSeries.clear();
- LastPoint = {};
- HistogramBuilder.Clear();
- SummaryBuilder.Clear();
- LogHistBuilder.Clear();
- }
- void AddLabel(const TLabel& label) {
- Labels.Add(label.Name(), label.Value());
- }
- void SetLastTime(TInstant time) {
- LastPoint.SetTime(time);
- }
- template <typename T>
- void SetLastValue(T value) {
- LastPoint.SetValue(value);
- }
- void SaveLastPoint() {
- DECODE_ENSURE(LastPoint.GetTime() != TInstant::Zero(),
- "cannot add point without or zero timestamp");
- if (!HistogramBuilder.Empty()) {
- auto histogram = HistogramBuilder.Build();
- TimeSeries.emplace_back(LastPoint.GetTime(), histogram.Get());
- } else if (!SummaryBuilder.Empty()) {
- auto summary = SummaryBuilder.Build();
- TimeSeries.emplace_back(LastPoint.GetTime(), summary.Get());
- } else if (!LogHistBuilder.Empty()) {
- auto logHist = LogHistBuilder.Build();
- TimeSeries.emplace_back(LastPoint.GetTime(), logHist.Get());
- } else {
- TimeSeries.push_back(std::move(LastPoint));
- }
- }
- template <typename TConsumer>
- void Consume(TConsumer&& consumer) {
- if (TimeSeries.empty()) {
- const auto& p = LastPoint;
- consumer(p.GetTime(), p.GetValueType(), p.GetValue());
- } else {
- for (const auto& p: TimeSeries) {
- consumer(p.GetTime(), p.GetValueType(), p.GetValue());
- }
- }
- }
- };
- struct TCommonParts {
- TInstant CommonTime;
- TLabels CommonLabels;
- };
- class IHaltableMetricConsumer: public IMetricConsumer {
- public:
- virtual bool NeedToStop() const = 0;
- };
- // TODO(ivanzhukov@): check all states for cases when a json document is invalid
- // e.g. "metrics" or "commonLabels" keys are specified multiple times
- class TCommonPartsCollector: public IHaltableMetricConsumer {
- public:
- TCommonParts&& CommonParts() {
- return std::move(CommonParts_);
- }
- private:
- bool NeedToStop() const override {
- return TInstant::Zero() != CommonParts_.CommonTime && !CommonParts_.CommonLabels.Empty();
- }
- void OnStreamBegin() override {
- }
- void OnStreamEnd() override {
- }
- void OnCommonTime(TInstant time) override {
- CommonParts_.CommonTime = time;
- }
- void OnMetricBegin(EMetricType) override {
- IsMetric_ = true;
- }
- void OnMetricEnd() override {
- IsMetric_ = false;
- }
- void OnLabelsBegin() override {
- }
- void OnLabelsEnd() override {
- }
- void OnLabel(TStringBuf name, TStringBuf value) override {
- if (!IsMetric_) {
- CommonParts_.CommonLabels.Add(std::move(name), std::move(value));
- }
- }
- void OnDouble(TInstant, double) override {
- }
- void OnInt64(TInstant, i64) override {
- }
- void OnUint64(TInstant, ui64) override {
- }
- void OnHistogram(TInstant, IHistogramSnapshotPtr) override {
- }
- void OnLogHistogram(TInstant, TLogHistogramSnapshotPtr) override {
- }
- void OnSummaryDouble(TInstant, ISummaryDoubleSnapshotPtr) override {
- }
- private:
- TCommonParts CommonParts_;
- bool IsMetric_{false};
- };
- class TCommonPartsProxy: public IHaltableMetricConsumer {
- public:
- TCommonPartsProxy(TCommonParts&& commonParts, IMetricConsumer* c)
- : CommonParts_{std::move(commonParts)}
- , Consumer_{c}
- {}
- private:
- bool NeedToStop() const override {
- return false;
- }
- void OnStreamBegin() override {
- Consumer_->OnStreamBegin();
- if (!CommonParts_.CommonLabels.Empty()) {
- Consumer_->OnLabelsBegin();
- for (auto&& label : CommonParts_.CommonLabels) {
- Consumer_->OnLabel(label.Name(), label.Value());
- }
- Consumer_->OnLabelsEnd();
- }
- if (TInstant::Zero() != CommonParts_.CommonTime) {
- Consumer_->OnCommonTime(CommonParts_.CommonTime);
- }
- }
- void OnStreamEnd() override {
- Consumer_->OnStreamEnd();
- }
- void OnCommonTime(TInstant) override {
- }
- void OnMetricBegin(EMetricType type) override {
- IsMetric_ = true;
- Consumer_->OnMetricBegin(type);
- }
- void OnMetricEnd() override {
- IsMetric_ = false;
- Consumer_->OnMetricEnd();
- }
- void OnLabelsBegin() override {
- if (IsMetric_) {
- Consumer_->OnLabelsBegin();
- }
- }
- void OnLabelsEnd() override {
- if (IsMetric_) {
- Consumer_->OnLabelsEnd();
- }
- }
- void OnLabel(TStringBuf name, TStringBuf value) override {
- if (IsMetric_) {
- Consumer_->OnLabel(std::move(name), std::move(value));
- }
- }
- void OnDouble(TInstant time, double value) override {
- Consumer_->OnDouble(time, value);
- }
- void OnInt64(TInstant time, i64 value) override {
- Consumer_->OnInt64(time, value);
- }
- void OnUint64(TInstant time, ui64 value) override {
- Consumer_->OnUint64(time, value);
- }
- void OnHistogram(TInstant time, IHistogramSnapshotPtr snapshot) override {
- Consumer_->OnHistogram(time, std::move(snapshot));
- }
- void OnLogHistogram(TInstant time, TLogHistogramSnapshotPtr snapshot) override {
- Consumer_->OnLogHistogram(time, std::move(snapshot));
- }
- void OnSummaryDouble(TInstant time, ISummaryDoubleSnapshotPtr snapshot) override {
- Consumer_->OnSummaryDouble(time, std::move(snapshot));
- }
- private:
- const TCommonParts CommonParts_;
- IMetricConsumer* Consumer_;
- bool IsMetric_{false};
- };
- ///////////////////////////////////////////////////////////////////////
- // TDecoderJson
- ///////////////////////////////////////////////////////////////////////
- class TDecoderJson final: public NJson::TJsonCallbacks {
- struct TState {
- enum EState {
- ROOT_OBJECT = 0x01,
- COMMON_LABELS,
- COMMON_TS,
- METRICS_ARRAY,
- METRIC_OBJECT,
- METRIC_NAME,
- METRIC_LABELS,
- METRIC_TYPE,
- METRIC_MODE, // TODO: must be deleted
- METRIC_TIMESERIES,
- METRIC_TS,
- METRIC_VALUE,
- METRIC_HIST,
- METRIC_HIST_BOUNDS,
- METRIC_HIST_BUCKETS,
- METRIC_HIST_INF,
- METRIC_DSUMMARY,
- METRIC_DSUMMARY_SUM,
- METRIC_DSUMMARY_MIN,
- METRIC_DSUMMARY_MAX,
- METRIC_DSUMMARY_LAST,
- METRIC_DSUMMARY_COUNT,
- METRIC_LOG_HIST,
- METRIC_LOG_HIST_BASE,
- METRIC_LOG_HIST_ZEROS,
- METRIC_LOG_HIST_START_POWER,
- METRIC_LOG_HIST_BUCKETS,
- };
- constexpr EState Current() const noexcept {
- return static_cast<EState>(State_ & 0xFF);
- }
- void ToNext(EState state) noexcept {
- constexpr auto bitSize = 8 * sizeof(ui8);
- State_ = (State_ << bitSize) | static_cast<ui8>(state);
- }
- void ToPrev() noexcept {
- constexpr auto bitSize = 8 * sizeof(ui8);
- State_ = State_ >> bitSize;
- }
- private:
- ui64 State_ = static_cast<ui64>(ROOT_OBJECT);
- };
- public:
- TDecoderJson(TStringBuf data, IHaltableMetricConsumer* metricConsumer, TStringBuf metricNameLabel)
- : Data_(data)
- , MetricConsumer_(metricConsumer)
- , MetricNameLabel_(metricNameLabel)
- {
- }
- private:
- #define PARSE_ENSURE(CONDITION, ...) \
- do { \
- if (Y_UNLIKELY(!(CONDITION))) { \
- ErrorMsg_ = TStringBuilder() << __VA_ARGS__; \
- return false; \
- } \
- } while (false)
- bool OnInteger(long long value) override {
- switch (State_.Current()) {
- case TState::COMMON_TS:
- PARSE_ENSURE(value >= 0, "unexpected negative number in a common timestamp: " << value);
- MetricConsumer_->OnCommonTime(TInstant::Seconds(value));
- State_.ToPrev();
- if (MetricConsumer_->NeedToStop()) {
- IsIntentionallyHalted_ = true;
- return false;
- }
- break;
- case TState::METRIC_TS:
- PARSE_ENSURE(value >= 0, "unexpected negative number in a metric timestamp: " << value);
- LastMetric_.SetLastTime(TInstant::Seconds(value));
- State_.ToPrev();
- break;
- case TState::METRIC_VALUE:
- LastMetric_.SetLastValue(static_cast<i64>(value));
- State_.ToPrev();
- break;
- case TState::METRIC_HIST_BOUNDS:
- LastMetric_.HistogramBuilder.AddBound(static_cast<double>(value));
- break;
- case TState::METRIC_HIST_BUCKETS:
- PARSE_ENSURE(value >= 0 && static_cast<ui64>(value) <= Max<TBucketValues::value_type>(), "value is out of bounds " << value);
- LastMetric_.HistogramBuilder.AddValue(value);
- break;
- case TState::METRIC_HIST_INF:
- PARSE_ENSURE(value >= 0, "unexpected negative number in histogram inf: " << value);
- LastMetric_.HistogramBuilder.AddInf(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_COUNT:
- LastMetric_.SummaryBuilder.SetCount(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_SUM:
- LastMetric_.SummaryBuilder.SetSum(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MIN:
- LastMetric_.SummaryBuilder.SetMin(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MAX:
- LastMetric_.SummaryBuilder.SetMax(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_LAST:
- LastMetric_.SummaryBuilder.SetLast(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BASE:
- LastMetric_.LogHistBuilder.SetBase(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_ZEROS:
- LastMetric_.LogHistBuilder.SetZerosCount(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_START_POWER:
- LastMetric_.LogHistBuilder.SetStartPower(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BUCKETS:
- LastMetric_.LogHistBuilder.AddBucketValue(value);
- break;
- default:
- return false;
- }
- return true;
- }
- bool OnUInteger(unsigned long long value) override {
- switch (State_.Current()) {
- case TState::COMMON_TS:
- MetricConsumer_->OnCommonTime(TInstant::Seconds(value));
- State_.ToPrev();
- if (MetricConsumer_->NeedToStop()) {
- IsIntentionallyHalted_ = true;
- return false;
- }
- break;
- case TState::METRIC_TS:
- LastMetric_.SetLastTime(TInstant::Seconds(value));
- State_.ToPrev();
- break;
- case TState::METRIC_VALUE:
- PARSE_ENSURE(value <= Max<ui64>(), "Metric value is out of bounds: " << value);
- LastMetric_.SetLastValue(static_cast<ui64>(value));
- State_.ToPrev();
- break;
- case TState::METRIC_HIST_BOUNDS:
- LastMetric_.HistogramBuilder.AddBound(static_cast<double>(value));
- break;
- case TState::METRIC_HIST_BUCKETS:
- PARSE_ENSURE(value <= Max<TBucketValues::value_type>(), "Histogram bucket value is out of bounds: " << value);
- LastMetric_.HistogramBuilder.AddValue(value);
- break;
- case TState::METRIC_HIST_INF:
- LastMetric_.HistogramBuilder.AddInf(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_COUNT:
- LastMetric_.SummaryBuilder.SetCount(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_SUM:
- LastMetric_.SummaryBuilder.SetSum(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MIN:
- LastMetric_.SummaryBuilder.SetMin(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MAX:
- LastMetric_.SummaryBuilder.SetMax(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_LAST:
- LastMetric_.SummaryBuilder.SetLast(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BASE:
- LastMetric_.LogHistBuilder.SetBase(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_ZEROS:
- LastMetric_.LogHistBuilder.SetZerosCount(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_START_POWER:
- LastMetric_.LogHistBuilder.SetStartPower(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BUCKETS:
- LastMetric_.LogHistBuilder.AddBucketValue(value);
- break;
- default:
- return false;
- }
- return true;
- }
- bool OnDouble(double value) override {
- switch (State_.Current()) {
- case TState::METRIC_VALUE:
- LastMetric_.SetLastValue(value);
- State_.ToPrev();
- break;
- case TState::METRIC_HIST_BOUNDS:
- LastMetric_.HistogramBuilder.AddBound(value);
- break;
- case TState::METRIC_DSUMMARY_SUM:
- LastMetric_.SummaryBuilder.SetSum(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MIN:
- LastMetric_.SummaryBuilder.SetMin(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MAX:
- LastMetric_.SummaryBuilder.SetMax(value);
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_LAST:
- LastMetric_.SummaryBuilder.SetLast(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BASE:
- LastMetric_.LogHistBuilder.SetBase(value);
- State_.ToPrev();
- break;
- case TState::METRIC_LOG_HIST_BUCKETS:
- LastMetric_.LogHistBuilder.AddBucketValue(value);
- break;
- default:
- return false;
- }
- return true;
- }
- bool OnString(const TStringBuf& value) override {
- switch (State_.Current()) {
- case TState::COMMON_LABELS:
- PARSE_ENSURE(!LastLabelName_.empty(), "empty label name in common labels");
- MetricConsumer_->OnLabel(LastLabelName_, TString{value});
- break;
- case TState::METRIC_LABELS:
- PARSE_ENSURE(!LastLabelName_.empty(), "empty label name in metric labels");
- LastMetric_.Labels.Add(LastLabelName_, TString{value});
- break;
- case TState::METRIC_NAME:
- PARSE_ENSURE(!value.empty(), "empty metric name");
- LastMetric_.Labels.Add(MetricNameLabel_, TString{value});
- State_.ToPrev();
- break;
- case TState::COMMON_TS:
- MetricConsumer_->OnCommonTime(TInstant::ParseIso8601(value));
- State_.ToPrev();
- if (MetricConsumer_->NeedToStop()) {
- IsIntentionallyHalted_ = true;
- return false;
- }
- break;
- case TState::METRIC_TS:
- LastMetric_.SetLastTime(TInstant::ParseIso8601(value));
- State_.ToPrev();
- break;
- case TState::METRIC_VALUE:
- if (auto [doubleValue, ok] = ParseSpecDouble(value); ok) {
- LastMetric_.SetLastValue(doubleValue);
- } else {
- return false;
- }
- State_.ToPrev();
- break;
- case TState::METRIC_TYPE:
- LastMetric_.Type = MetricTypeFromStr(value);
- State_.ToPrev();
- break;
- case TState::METRIC_MODE:
- if (value == TStringBuf("deriv")) {
- LastMetric_.Type = EMetricType::RATE;
- }
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_SUM:
- if (auto [doubleValue, ok] = ParseSpecDouble(value); ok) {
- LastMetric_.SummaryBuilder.SetSum(doubleValue);
- } else {
- return false;
- }
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MIN:
- if (auto [doubleValue, ok] = ParseSpecDouble(value); ok) {
- LastMetric_.SummaryBuilder.SetMin(doubleValue);
- } else {
- return false;
- }
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_MAX:
- if (auto [doubleValue, ok] = ParseSpecDouble(value); ok) {
- LastMetric_.SummaryBuilder.SetMax(doubleValue);
- } else {
- return false;
- }
- State_.ToPrev();
- break;
- case TState::METRIC_DSUMMARY_LAST:
- if (auto [doubleValue, ok] = ParseSpecDouble(value); ok) {
- LastMetric_.SummaryBuilder.SetLast(doubleValue);
- } else {
- return false;
- }
- State_.ToPrev();
- break;
- default:
- return false;
- }
- return true;
- }
- bool OnMapKey(const TStringBuf& key) override {
- switch (State_.Current()) {
- case TState::ROOT_OBJECT:
- if (key == TStringBuf("commonLabels") || key == TStringBuf("labels")) {
- State_.ToNext(TState::COMMON_LABELS);
- } else if (key == TStringBuf("ts")) {
- State_.ToNext(TState::COMMON_TS);
- } else if (key == TStringBuf("sensors") || key == TStringBuf("metrics")) {
- State_.ToNext(TState::METRICS_ARRAY);
- }
- break;
- case TState::COMMON_LABELS:
- case TState::METRIC_LABELS:
- LastLabelName_ = key;
- break;
- case TState::METRIC_OBJECT:
- if (key == TStringBuf("labels")) {
- State_.ToNext(TState::METRIC_LABELS);
- } else if (key == TStringBuf("name")) {
- State_.ToNext(TState::METRIC_NAME);
- } else if (key == TStringBuf("ts")) {
- PARSE_ENSURE(!LastMetric_.SeenTimeseries,
- "mixed timeseries and ts attributes");
- LastMetric_.SeenTsOrValue = true;
- State_.ToNext(TState::METRIC_TS);
- } else if (key == TStringBuf("value")) {
- PARSE_ENSURE(!LastMetric_.SeenTimeseries,
- "mixed timeseries and value attributes");
- LastMetric_.SeenTsOrValue = true;
- State_.ToNext(TState::METRIC_VALUE);
- } else if (key == TStringBuf("timeseries")) {
- PARSE_ENSURE(!LastMetric_.SeenTsOrValue,
- "mixed timeseries and ts/value attributes");
- LastMetric_.SeenTimeseries = true;
- State_.ToNext(TState::METRIC_TIMESERIES);
- } else if (key == TStringBuf("mode")) {
- State_.ToNext(TState::METRIC_MODE);
- } else if (key == TStringBuf("kind") || key == TStringBuf("type")) {
- State_.ToNext(TState::METRIC_TYPE);
- } else if (key == TStringBuf("hist")) {
- State_.ToNext(TState::METRIC_HIST);
- } else if (key == TStringBuf("summary")) {
- State_.ToNext(TState::METRIC_DSUMMARY);
- } else if (key == TStringBuf("log_hist")) {
- State_.ToNext(TState::METRIC_LOG_HIST);
- } else if (key == TStringBuf("memOnly")) {
- // deprecated. Skip it without errors for backward compatibility
- } else {
- ErrorMsg_ = TStringBuilder() << "unexpected key \"" << key << "\" in a metric schema";
- return false;
- }
- break;
- case TState::METRIC_TIMESERIES:
- if (key == TStringBuf("ts")) {
- State_.ToNext(TState::METRIC_TS);
- } else if (key == TStringBuf("value")) {
- State_.ToNext(TState::METRIC_VALUE);
- } else if (key == TStringBuf("hist")) {
- State_.ToNext(TState::METRIC_HIST);
- } else if (key == TStringBuf("summary")) {
- State_.ToNext(TState::METRIC_DSUMMARY);
- } else if (key == TStringBuf("log_hist")) {
- State_.ToNext(TState::METRIC_LOG_HIST);
- }
- break;
- case TState::METRIC_HIST:
- if (key == TStringBuf("bounds")) {
- State_.ToNext(TState::METRIC_HIST_BOUNDS);
- } else if (key == TStringBuf("buckets")) {
- State_.ToNext(TState::METRIC_HIST_BUCKETS);
- } else if (key == TStringBuf("inf")) {
- State_.ToNext(TState::METRIC_HIST_INF);
- }
- break;
- case TState::METRIC_LOG_HIST:
- if (key == TStringBuf("base")) {
- State_.ToNext(TState::METRIC_LOG_HIST_BASE);
- } else if (key == TStringBuf("zeros_count")) {
- State_.ToNext(TState::METRIC_LOG_HIST_ZEROS);
- } else if (key == TStringBuf("start_power")) {
- State_.ToNext(TState::METRIC_LOG_HIST_START_POWER);
- } else if (key == TStringBuf("buckets")) {
- State_.ToNext(TState::METRIC_LOG_HIST_BUCKETS);
- }
- break;
- case TState::METRIC_DSUMMARY:
- if (key == TStringBuf("sum")) {
- State_.ToNext(TState::METRIC_DSUMMARY_SUM);
- } else if (key == TStringBuf("min")) {
- State_.ToNext(TState::METRIC_DSUMMARY_MIN);
- } else if (key == TStringBuf("max")) {
- State_.ToNext(TState::METRIC_DSUMMARY_MAX);
- } else if (key == TStringBuf("last")) {
- State_.ToNext(TState::METRIC_DSUMMARY_LAST);
- } else if (key == TStringBuf("count")) {
- State_.ToNext(TState::METRIC_DSUMMARY_COUNT);
- }
- break;
- default:
- return false;
- }
- return true;
- }
- bool OnOpenMap() override {
- switch (State_.Current()) {
- case TState::ROOT_OBJECT:
- MetricConsumer_->OnStreamBegin();
- break;
- case TState::COMMON_LABELS:
- MetricConsumer_->OnLabelsBegin();
- break;
- case TState::METRICS_ARRAY:
- State_.ToNext(TState::METRIC_OBJECT);
- LastMetric_.Clear();
- break;
- default:
- break;
- }
- return true;
- }
- bool OnCloseMap() override {
- switch (State_.Current()) {
- case TState::ROOT_OBJECT:
- MetricConsumer_->OnStreamEnd();
- break;
- case TState::METRIC_LABELS:
- State_.ToPrev();
- break;
- case TState::COMMON_LABELS:
- MetricConsumer_->OnLabelsEnd();
- State_.ToPrev();
- if (MetricConsumer_->NeedToStop()) {
- IsIntentionallyHalted_ = true;
- return false;
- }
- break;
- case TState::METRIC_OBJECT:
- ConsumeMetric();
- State_.ToPrev();
- break;
- case TState::METRIC_TIMESERIES:
- LastMetric_.SaveLastPoint();
- break;
- case TState::METRIC_HIST:
- case TState::METRIC_DSUMMARY:
- case TState::METRIC_LOG_HIST:
- State_.ToPrev();
- break;
- default:
- break;
- }
- return true;
- }
- bool OnOpenArray() override {
- auto currentState = State_.Current();
- PARSE_ENSURE(
- currentState == TState::METRICS_ARRAY ||
- currentState == TState::METRIC_TIMESERIES ||
- currentState == TState::METRIC_HIST_BOUNDS ||
- currentState == TState::METRIC_HIST_BUCKETS ||
- currentState == TState::METRIC_LOG_HIST_BUCKETS,
- "unexpected array begin");
- return true;
- }
- bool OnCloseArray() override {
- switch (State_.Current()) {
- case TState::METRICS_ARRAY:
- case TState::METRIC_TIMESERIES:
- case TState::METRIC_HIST_BOUNDS:
- case TState::METRIC_HIST_BUCKETS:
- case TState::METRIC_LOG_HIST_BUCKETS:
- State_.ToPrev();
- break;
- default:
- return false;
- }
- return true;
- }
- void OnError(size_t off, TStringBuf reason) override {
- if (IsIntentionallyHalted_) {
- return;
- }
- size_t snippetBeg = (off < 20) ? 0 : (off - 20);
- TStringBuf snippet = Data_.SubStr(snippetBeg, 40);
- throw TJsonDecodeError()
- << "cannot parse JSON, error at: " << off
- << ", reason: " << (ErrorMsg_.empty() ? reason : TStringBuf{ErrorMsg_})
- << "\nsnippet: ..." << snippet << "...";
- }
- bool OnEnd() override {
- return true;
- }
- void ConsumeMetric() {
- // for backwad compatibility all unknown metrics treated as gauges
- if (LastMetric_.Type == EMetricType::UNKNOWN) {
- if (LastMetric_.HistogramBuilder.Empty()) {
- LastMetric_.Type = EMetricType::GAUGE;
- } else {
- LastMetric_.Type = EMetricType::HIST;
- }
- }
- // (1) begin metric
- MetricConsumer_->OnMetricBegin(LastMetric_.Type);
- // (2) labels
- if (!LastMetric_.Labels.empty()) {
- MetricConsumer_->OnLabelsBegin();
- for (auto&& label : LastMetric_.Labels) {
- MetricConsumer_->OnLabel(label.Name(), label.Value());
- }
- MetricConsumer_->OnLabelsEnd();
- }
- // (3) values
- switch (LastMetric_.Type) {
- case EMetricType::GAUGE:
- LastMetric_.Consume([this](TInstant time, EMetricValueType valueType, TMetricValue value) {
- MetricConsumer_->OnDouble(time, value.AsDouble(valueType));
- });
- break;
- case EMetricType::IGAUGE:
- LastMetric_.Consume([this](TInstant time, EMetricValueType valueType, TMetricValue value) {
- MetricConsumer_->OnInt64(time, value.AsInt64(valueType));
- });
- break;
- case EMetricType::COUNTER:
- case EMetricType::RATE:
- LastMetric_.Consume([this](TInstant time, EMetricValueType valueType, TMetricValue value) {
- MetricConsumer_->OnUint64(time, value.AsUint64(valueType));
- });
- break;
- case EMetricType::HIST:
- case EMetricType::HIST_RATE:
- if (LastMetric_.TimeSeries.empty()) {
- auto time = LastMetric_.LastPoint.GetTime();
- auto histogram = LastMetric_.HistogramBuilder.Build();
- MetricConsumer_->OnHistogram(time, histogram);
- } else {
- for (const auto& p : LastMetric_.TimeSeries) {
- DECODE_ENSURE(p.GetValueType() == EMetricValueType::HISTOGRAM, "Value is not a histogram");
- MetricConsumer_->OnHistogram(p.GetTime(), p.GetValue().AsHistogram());
- }
- }
- break;
- case EMetricType::DSUMMARY:
- if (LastMetric_.TimeSeries.empty()) {
- auto time = LastMetric_.LastPoint.GetTime();
- auto summary = LastMetric_.SummaryBuilder.Build();
- MetricConsumer_->OnSummaryDouble(time, summary);
- } else {
- for (const auto& p : LastMetric_.TimeSeries) {
- DECODE_ENSURE(p.GetValueType() == EMetricValueType::SUMMARY, "Value is not a summary");
- MetricConsumer_->OnSummaryDouble(p.GetTime(), p.GetValue().AsSummaryDouble());
- }
- }
- break;
- case EMetricType::LOGHIST:
- if (LastMetric_.TimeSeries.empty()) {
- auto time = LastMetric_.LastPoint.GetTime();
- auto logHist = LastMetric_.LogHistBuilder.Build();
- MetricConsumer_->OnLogHistogram(time, logHist);
- } else {
- for (const auto& p : LastMetric_.TimeSeries) {
- DECODE_ENSURE(p.GetValueType() == EMetricValueType::LOGHISTOGRAM, "Value is not a log_histogram");
- MetricConsumer_->OnLogHistogram(p.GetTime(), p.GetValue().AsLogHistogram());
- }
- }
- break;
- case EMetricType::UNKNOWN:
- // TODO: output metric labels
- ythrow yexception() << "unknown metric type";
- }
- // (4) end metric
- MetricConsumer_->OnMetricEnd();
- }
- private:
- TStringBuf Data_;
- IHaltableMetricConsumer* MetricConsumer_;
- TString MetricNameLabel_;
- TState State_;
- TString LastLabelName_;
- TMetricCollector LastMetric_;
- TString ErrorMsg_;
- bool IsIntentionallyHalted_{false};
- };
- } // namespace
- void DecodeJson(TStringBuf data, IMetricConsumer* c, TStringBuf metricNameLabel) {
- TCommonPartsCollector commonPartsCollector;
- {
- TMemoryInput memIn(data);
- TDecoderJson decoder(data, &commonPartsCollector, metricNameLabel);
- // no need to check a return value. If there is an error, a TJsonDecodeError is thrown
- NJson::ReadJson(&memIn, &decoder);
- }
- TCommonPartsProxy commonPartsProxy(std::move(commonPartsCollector.CommonParts()), c);
- {
- TMemoryInput memIn(data);
- TDecoderJson decoder(data, &commonPartsProxy, metricNameLabel);
- // no need to check a return value. If there is an error, a TJsonDecodeError is thrown
- NJson::ReadJson(&memIn, &decoder);
- }
- }
- #undef DECODE_ENSURE
- }
|