diff options
| author | akchernikov <[email protected]> | 2022-02-10 16:50:44 +0300 |
|---|---|---|
| committer | Daniil Cherednik <[email protected]> | 2022-02-10 16:50:44 +0300 |
| commit | ea46c401e7900b229add3e6074dbf89adc84ebfc (patch) | |
| tree | 5d5cb817648f650d76cf1076100726fd9b8448e8 /library/cpp/actors/interconnect/interconnect_counters.cpp | |
| parent | 87cccadbd489f00bc6d81b27ad182277cbb25826 (diff) | |
Restoring authorship annotation for <[email protected]>. Commit 2 of 2.
Diffstat (limited to 'library/cpp/actors/interconnect/interconnect_counters.cpp')
| -rw-r--r-- | library/cpp/actors/interconnect/interconnect_counters.cpp | 920 |
1 files changed, 460 insertions, 460 deletions
diff --git a/library/cpp/actors/interconnect/interconnect_counters.cpp b/library/cpp/actors/interconnect/interconnect_counters.cpp index a206818160a..224160d4b4f 100644 --- a/library/cpp/actors/interconnect/interconnect_counters.cpp +++ b/library/cpp/actors/interconnect/interconnect_counters.cpp @@ -1,15 +1,15 @@ -#include "interconnect_counters.h" +#include "interconnect_counters.h" -#include <library/cpp/monlib/metrics/metric_registry.h> -#include <library/cpp/monlib/metrics/metric_sub_registry.h> +#include <library/cpp/monlib/metrics/metric_registry.h> +#include <library/cpp/monlib/metrics/metric_sub_registry.h> -#include <unordered_map> +#include <unordered_map> namespace NActors { -namespace { - - class TInterconnectCounters: public IInterconnectMetrics { +namespace { + + class TInterconnectCounters: public IInterconnectMetrics { public: struct TOutputChannel { NMonitoring::TDynamicCounters::TCounterPtr Traffic; @@ -87,7 +87,7 @@ namespace { NMonitoring::TDynamicCounterPtr PerSessionCounters; NMonitoring::TDynamicCounterPtr PerDataCenterCounters; NMonitoring::TDynamicCounterPtr& AdaptiveCounters; - + bool Initialized = false; NMonitoring::TDynamicCounters::TCounterPtr Traffic; @@ -100,135 +100,135 @@ namespace { , MergePerDataCenterCounters(common->Settings.MergePerDataCenterCounters) , MergePerPeerCounters(common->Settings.MergePerPeerCounters) , Counters(common->MonCounters) - , AdaptiveCounters(MergePerDataCenterCounters - ? PerDataCenterCounters : - MergePerPeerCounters ? Counters : PerSessionCounters) + , AdaptiveCounters(MergePerDataCenterCounters + ? PerDataCenterCounters : + MergePerPeerCounters ? Counters : PerSessionCounters) {} void AddInflightDataAmount(ui64 value) override { - *InflightDataAmount += value; - } - + *InflightDataAmount += value; + } + void SubInflightDataAmount(ui64 value) override { - *InflightDataAmount -= value; - } - + *InflightDataAmount -= value; + } + void AddTotalBytesWritten(ui64 value) override { - *TotalBytesWritten += value; - } - + *TotalBytesWritten += value; + } + void SetClockSkewMicrosec(i64 value) override { - *ClockSkewMicrosec = value; - } - - void IncSessionDeaths() override { - ++*SessionDeaths; - } - - void IncHandshakeFails() override { - ++*HandshakeFails; - } - - void SetConnected(ui32 value) override { - *Connected = value; - } - - void IncSubscribersCount() override { - ++*SubscribersCount; - } - - void SubSubscribersCount(ui32 value) override { - *SubscribersCount -= value; - } - + *ClockSkewMicrosec = value; + } + + void IncSessionDeaths() override { + ++*SessionDeaths; + } + + void IncHandshakeFails() override { + ++*HandshakeFails; + } + + void SetConnected(ui32 value) override { + *Connected = value; + } + + void IncSubscribersCount() override { + ++*SubscribersCount; + } + + void SubSubscribersCount(ui32 value) override { + *SubscribersCount -= value; + } + void SubOutputBuffersTotalSize(ui64 value) override { - *OutputBuffersTotalSize -= value; - } - + *OutputBuffersTotalSize -= value; + } + void AddOutputBuffersTotalSize(ui64 value) override { - *OutputBuffersTotalSize += value; - } - + *OutputBuffersTotalSize += value; + } + ui64 GetOutputBuffersTotalSize() const override { - return *OutputBuffersTotalSize; - } - - void IncDisconnections() override { - ++*Disconnections; - } - - void IncUsefulWriteWakeups() override { - ++*UsefulWriteWakeups; - } - - void IncSpuriousWriteWakeups() override { - ++*SpuriousWriteWakeups; - } - - void IncSendSyscalls() override { - ++*SendSyscalls; - } - - void IncInflyLimitReach() override { - ++*InflyLimitReach; - } - - void IncUsefulReadWakeups() override { - ++*UsefulReadWakeups; - } - - void IncSpuriousReadWakeups() override { - ++*SpuriousReadWakeups; - } - - void IncDisconnectByReason(const TString& s) override { - if (auto it = DisconnectByReason.find(s); it != DisconnectByReason.end()) { - it->second->Inc(); - } - } - + return *OutputBuffersTotalSize; + } + + void IncDisconnections() override { + ++*Disconnections; + } + + void IncUsefulWriteWakeups() override { + ++*UsefulWriteWakeups; + } + + void IncSpuriousWriteWakeups() override { + ++*SpuriousWriteWakeups; + } + + void IncSendSyscalls() override { + ++*SendSyscalls; + } + + void IncInflyLimitReach() override { + ++*InflyLimitReach; + } + + void IncUsefulReadWakeups() override { + ++*UsefulReadWakeups; + } + + void IncSpuriousReadWakeups() override { + ++*SpuriousReadWakeups; + } + + void IncDisconnectByReason(const TString& s) override { + if (auto it = DisconnectByReason.find(s); it != DisconnectByReason.end()) { + it->second->Inc(); + } + } + void AddInputChannelsIncomingTraffic(ui16 channel, ui64 incomingTraffic) override { - auto& ch = InputChannels.Get(channel); - *ch.IncomingTraffic += incomingTraffic; - } - - void IncInputChannelsIncomingEvents(ui16 channel) override { - auto& ch = InputChannels.Get(channel); - ++*ch.IncomingEvents; - } - - void IncRecvSyscalls() override { - ++*RecvSyscalls; - } - + auto& ch = InputChannels.Get(channel); + *ch.IncomingTraffic += incomingTraffic; + } + + void IncInputChannelsIncomingEvents(ui16 channel) override { + auto& ch = InputChannels.Get(channel); + ++*ch.IncomingEvents; + } + + void IncRecvSyscalls() override { + ++*RecvSyscalls; + } + void AddTotalBytesRead(ui64 value) override { - *TotalBytesRead += value; - } - - void UpdateLegacyPingTimeHist(ui64 value) override { - LegacyPingTimeHist.Add(value); - PingTimeHistogram->Collect(value); - } - + *TotalBytesRead += value; + } + + void UpdateLegacyPingTimeHist(ui64 value) override { + LegacyPingTimeHist.Add(value); + PingTimeHistogram->Collect(value); + } + void UpdateOutputChannelTraffic(ui16 channel, ui64 value) override { - if (GetOutputChannel(channel).OutgoingTraffic) { - *(GetOutputChannel(channel).OutgoingTraffic) += value; - } - if (GetOutputChannel(channel).Traffic) { - *(GetOutputChannel(channel).Traffic) += value; - } - } - - void UpdateOutputChannelEvents(ui16 channel) override { - if (GetOutputChannel(channel).OutgoingEvents) { - ++*(GetOutputChannel(channel).OutgoingEvents); - } - if (GetOutputChannel(channel).Events) { - ++*(GetOutputChannel(channel).Events); - } - } - - void SetPeerInfo(const TString& name, const TString& dataCenterId) override { + if (GetOutputChannel(channel).OutgoingTraffic) { + *(GetOutputChannel(channel).OutgoingTraffic) += value; + } + if (GetOutputChannel(channel).Traffic) { + *(GetOutputChannel(channel).Traffic) += value; + } + } + + void UpdateOutputChannelEvents(ui16 channel) override { + if (GetOutputChannel(channel).OutgoingEvents) { + ++*(GetOutputChannel(channel).OutgoingEvents); + } + if (GetOutputChannel(channel).Events) { + ++*(GetOutputChannel(channel).Events); + } + } + + void SetPeerInfo(const TString& name, const TString& dataCenterId) override { if (name != std::exchange(HumanFriendlyPeerHostName, name)) { PerSessionCounters.Reset(); } @@ -312,7 +312,7 @@ namespace { return it != OutputChannels.end() ? it->second : OtherOutputChannel; } - private: + private: NMonitoring::TDynamicCounters::TCounterPtr SessionDeaths; NMonitoring::TDynamicCounters::TCounterPtr HandshakeFails; NMonitoring::TDynamicCounters::TCounterPtr Connected; @@ -340,353 +340,353 @@ namespace { NMonitoring::TDynamicCounters::TCounterPtr TotalBytesWritten, TotalBytesRead; }; - class TInterconnectMetrics: public IInterconnectMetrics { - public: - struct TOutputChannel { - NMonitoring::IRate* Traffic; - NMonitoring::IRate* Events; - NMonitoring::IRate* OutgoingTraffic; - NMonitoring::IRate* OutgoingEvents; - - TOutputChannel() = default; - - TOutputChannel(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, - NMonitoring::IRate* traffic, - NMonitoring::IRate* events) - : Traffic(traffic) - , Events(events) - , OutgoingTraffic(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.outgoing_traffic"}}))) - , OutgoingEvents(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.outgoing_events"}}))) - {} - - TOutputChannel(const TOutputChannel&) = default; - }; - - struct TInputChannel { - NMonitoring::IRate* Traffic; - NMonitoring::IRate* Events; - NMonitoring::IRate* ScopeErrors; - NMonitoring::IRate* IncomingTraffic; - NMonitoring::IRate* IncomingEvents; - - TInputChannel() = default; - - TInputChannel(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, - NMonitoring::IRate* traffic, NMonitoring::IRate* events, - NMonitoring::IRate* scopeErrors) - : Traffic(traffic) - , Events(events) - , ScopeErrors(scopeErrors) - , IncomingTraffic(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.incoming_traffic"}}))) - , IncomingEvents(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.incoming_events"}}))) - {} - - TInputChannel(const TInputChannel&) = default; - }; - - struct TInputChannels : std::unordered_map<ui16, TInputChannel> { - TInputChannel OtherInputChannel; - - TInputChannels() = default; - - TInputChannels(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, - const std::unordered_map<ui16, TString>& names, - NMonitoring::IRate* traffic, NMonitoring::IRate* events, - NMonitoring::IRate* scopeErrors) - : OtherInputChannel(std::make_shared<NMonitoring::TMetricSubRegistry>( - NMonitoring::TLabels{{"channel", "other"}}, metrics), traffic, events, scopeErrors) - { - for (const auto& [id, name] : names) { - try_emplace(id, std::make_shared<NMonitoring::TMetricSubRegistry>(NMonitoring::TLabels{{"channel", name}}, metrics), - traffic, events, scopeErrors); - } - } - - TInputChannels(const TInputChannels&) = default; - - const TInputChannel& Get(ui16 id) const { - const auto it = find(id); - return it != end() ? it->second : OtherInputChannel; - } - }; - - TInterconnectMetrics(const TInterconnectProxyCommon::TPtr& common) - : Common(common) - , MergePerDataCenterMetrics_(common->Settings.MergePerDataCenterCounters) - , MergePerPeerMetrics_(common->Settings.MergePerPeerCounters) - , Metrics_(common->Metrics) - , AdaptiveMetrics_(MergePerDataCenterMetrics_ - ? PerDataCenterMetrics_ : - MergePerPeerMetrics_ ? Metrics_ : PerSessionMetrics_) - {} - + class TInterconnectMetrics: public IInterconnectMetrics { + public: + struct TOutputChannel { + NMonitoring::IRate* Traffic; + NMonitoring::IRate* Events; + NMonitoring::IRate* OutgoingTraffic; + NMonitoring::IRate* OutgoingEvents; + + TOutputChannel() = default; + + TOutputChannel(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, + NMonitoring::IRate* traffic, + NMonitoring::IRate* events) + : Traffic(traffic) + , Events(events) + , OutgoingTraffic(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.outgoing_traffic"}}))) + , OutgoingEvents(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.outgoing_events"}}))) + {} + + TOutputChannel(const TOutputChannel&) = default; + }; + + struct TInputChannel { + NMonitoring::IRate* Traffic; + NMonitoring::IRate* Events; + NMonitoring::IRate* ScopeErrors; + NMonitoring::IRate* IncomingTraffic; + NMonitoring::IRate* IncomingEvents; + + TInputChannel() = default; + + TInputChannel(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, + NMonitoring::IRate* traffic, NMonitoring::IRate* events, + NMonitoring::IRate* scopeErrors) + : Traffic(traffic) + , Events(events) + , ScopeErrors(scopeErrors) + , IncomingTraffic(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.incoming_traffic"}}))) + , IncomingEvents(metrics->Rate(NMonitoring::MakeLabels({{"sensor", "interconnect.incoming_events"}}))) + {} + + TInputChannel(const TInputChannel&) = default; + }; + + struct TInputChannels : std::unordered_map<ui16, TInputChannel> { + TInputChannel OtherInputChannel; + + TInputChannels() = default; + + TInputChannels(const std::shared_ptr<NMonitoring::IMetricRegistry>& metrics, + const std::unordered_map<ui16, TString>& names, + NMonitoring::IRate* traffic, NMonitoring::IRate* events, + NMonitoring::IRate* scopeErrors) + : OtherInputChannel(std::make_shared<NMonitoring::TMetricSubRegistry>( + NMonitoring::TLabels{{"channel", "other"}}, metrics), traffic, events, scopeErrors) + { + for (const auto& [id, name] : names) { + try_emplace(id, std::make_shared<NMonitoring::TMetricSubRegistry>(NMonitoring::TLabels{{"channel", name}}, metrics), + traffic, events, scopeErrors); + } + } + + TInputChannels(const TInputChannels&) = default; + + const TInputChannel& Get(ui16 id) const { + const auto it = find(id); + return it != end() ? it->second : OtherInputChannel; + } + }; + + TInterconnectMetrics(const TInterconnectProxyCommon::TPtr& common) + : Common(common) + , MergePerDataCenterMetrics_(common->Settings.MergePerDataCenterCounters) + , MergePerPeerMetrics_(common->Settings.MergePerPeerCounters) + , Metrics_(common->Metrics) + , AdaptiveMetrics_(MergePerDataCenterMetrics_ + ? PerDataCenterMetrics_ : + MergePerPeerMetrics_ ? Metrics_ : PerSessionMetrics_) + {} + void AddInflightDataAmount(ui64 value) override { - InflightDataAmount_->Add(value); - } - + InflightDataAmount_->Add(value); + } + void SubInflightDataAmount(ui64 value) override { - InflightDataAmount_->Add(-value); - } - + InflightDataAmount_->Add(-value); + } + void AddTotalBytesWritten(ui64 value) override { - TotalBytesWritten_->Add(value); - } - + TotalBytesWritten_->Add(value); + } + void SetClockSkewMicrosec(i64 value) override { - ClockSkewMicrosec_->Set(value); - } - - void IncSessionDeaths() override { - SessionDeaths_->Inc(); - } - - void IncHandshakeFails() override { - HandshakeFails_->Inc(); - } - - void SetConnected(ui32 value) override { - Connected_->Set(value); - } - - void IncSubscribersCount() override { - SubscribersCount_->Inc(); - } - - void SubSubscribersCount(ui32 value) override { - SubscribersCount_->Add(-value); - } - + ClockSkewMicrosec_->Set(value); + } + + void IncSessionDeaths() override { + SessionDeaths_->Inc(); + } + + void IncHandshakeFails() override { + HandshakeFails_->Inc(); + } + + void SetConnected(ui32 value) override { + Connected_->Set(value); + } + + void IncSubscribersCount() override { + SubscribersCount_->Inc(); + } + + void SubSubscribersCount(ui32 value) override { + SubscribersCount_->Add(-value); + } + void SubOutputBuffersTotalSize(ui64 value) override { - OutputBuffersTotalSize_->Add(-value); - } - + OutputBuffersTotalSize_->Add(-value); + } + void AddOutputBuffersTotalSize(ui64 value) override { - OutputBuffersTotalSize_->Add(value); - } - + OutputBuffersTotalSize_->Add(value); + } + ui64 GetOutputBuffersTotalSize() const override { - return OutputBuffersTotalSize_->Get(); - } - - void IncDisconnections() override { - Disconnections_->Inc(); - } - - void IncUsefulWriteWakeups() override { - UsefulWriteWakeups_->Inc(); - } - - void IncSpuriousWriteWakeups() override { - SpuriousWriteWakeups_->Inc(); - } - - void IncSendSyscalls() override { - SendSyscalls_->Inc(); - } - - void IncInflyLimitReach() override { - InflyLimitReach_->Inc(); - } - - void IncUsefulReadWakeups() override { - UsefulReadWakeups_->Inc(); - } - - void IncSpuriousReadWakeups() override { - SpuriousReadWakeups_->Inc(); - } - - void IncDisconnectByReason(const TString& s) override { - if (auto it = DisconnectByReason_.find(s); it != DisconnectByReason_.end()) { - it->second->Inc(); - } - } - + return OutputBuffersTotalSize_->Get(); + } + + void IncDisconnections() override { + Disconnections_->Inc(); + } + + void IncUsefulWriteWakeups() override { + UsefulWriteWakeups_->Inc(); + } + + void IncSpuriousWriteWakeups() override { + SpuriousWriteWakeups_->Inc(); + } + + void IncSendSyscalls() override { + SendSyscalls_->Inc(); + } + + void IncInflyLimitReach() override { + InflyLimitReach_->Inc(); + } + + void IncUsefulReadWakeups() override { + UsefulReadWakeups_->Inc(); + } + + void IncSpuriousReadWakeups() override { + SpuriousReadWakeups_->Inc(); + } + + void IncDisconnectByReason(const TString& s) override { + if (auto it = DisconnectByReason_.find(s); it != DisconnectByReason_.end()) { + it->second->Inc(); + } + } + void AddInputChannelsIncomingTraffic(ui16 channel, ui64 incomingTraffic) override { - auto& ch = InputChannels_.Get(channel); - ch.IncomingTraffic->Add(incomingTraffic); - } - - void IncInputChannelsIncomingEvents(ui16 channel) override { - auto& ch = InputChannels_.Get(channel); - ch.IncomingEvents->Inc(); - } - - void IncRecvSyscalls() override { - RecvSyscalls_->Inc(); - } - + auto& ch = InputChannels_.Get(channel); + ch.IncomingTraffic->Add(incomingTraffic); + } + + void IncInputChannelsIncomingEvents(ui16 channel) override { + auto& ch = InputChannels_.Get(channel); + ch.IncomingEvents->Inc(); + } + + void IncRecvSyscalls() override { + RecvSyscalls_->Inc(); + } + void AddTotalBytesRead(ui64 value) override { - TotalBytesRead_->Add(value); - } - - void UpdateLegacyPingTimeHist(ui64 value) override { - PingTimeHistogram_->Record(value); - } - + TotalBytesRead_->Add(value); + } + + void UpdateLegacyPingTimeHist(ui64 value) override { + PingTimeHistogram_->Record(value); + } + void UpdateOutputChannelTraffic(ui16 channel, ui64 value) override { - if (GetOutputChannel(channel).OutgoingTraffic) { - GetOutputChannel(channel).OutgoingTraffic->Add(value); - } - if (GetOutputChannel(channel).Traffic) { - GetOutputChannel(channel).Traffic->Add(value); - } - } - - void UpdateOutputChannelEvents(ui16 channel) override { - if (GetOutputChannel(channel).OutgoingEvents) { - GetOutputChannel(channel).OutgoingEvents->Inc(); - } - if (GetOutputChannel(channel).Events) { - GetOutputChannel(channel).Events->Inc(); - } - } - - void SetPeerInfo(const TString& name, const TString& dataCenterId) override { - if (name != std::exchange(HumanFriendlyPeerHostName, name)) { - PerSessionMetrics_.reset(); - } - VALGRIND_MAKE_READABLE(&DataCenterId, sizeof(DataCenterId)); - if (dataCenterId != std::exchange(DataCenterId, dataCenterId)) { - PerDataCenterMetrics_.reset(); - } - - const bool updatePerDataCenter = !PerDataCenterMetrics_ && MergePerDataCenterMetrics_; - if (updatePerDataCenter) { - PerDataCenterMetrics_ = std::make_shared<NMonitoring::TMetricSubRegistry>( - NMonitoring::TLabels{{"datacenter_id", *DataCenterId}}, Metrics_); - } - - const bool updatePerSession = !PerSessionMetrics_ || updatePerDataCenter; - if (updatePerSession) { - auto base = MergePerDataCenterMetrics_ ? PerDataCenterMetrics_ : Metrics_; - PerSessionMetrics_ = std::make_shared<NMonitoring::TMetricSubRegistry>( - NMonitoring::TLabels{{"peer", *HumanFriendlyPeerHostName}}, base); - } - - const bool updateGlobal = !Initialized_; - - const bool updateAdaptive = - &AdaptiveMetrics_ == &Metrics_ ? updateGlobal : - &AdaptiveMetrics_ == &PerSessionMetrics_ ? updatePerSession : - &AdaptiveMetrics_ == &PerDataCenterMetrics_ ? updatePerDataCenter : - false; - - auto createRate = [](std::shared_ptr<NMonitoring::IMetricRegistry> metrics, TStringBuf name) mutable { - return metrics->Rate(NMonitoring::MakeLabels(NMonitoring::TLabels{{"sensor", name}})); - }; - auto createIntGauge = [](std::shared_ptr<NMonitoring::IMetricRegistry> metrics, TStringBuf name) mutable { - return metrics->IntGauge(NMonitoring::MakeLabels(NMonitoring::TLabels{{"sensor", name}})); - }; - - if (updatePerSession) { - Connected_ = createIntGauge(PerSessionMetrics_, "interconnect.connected"); - Disconnections_ = createRate(PerSessionMetrics_, "interconnect.disconnections"); - ClockSkewMicrosec_ = createIntGauge(PerSessionMetrics_, "interconnect.clock_skew_microsec"); - Traffic_ = createRate(PerSessionMetrics_, "interconnect.traffic"); - Events_ = createRate(PerSessionMetrics_, "interconnect.events"); - ScopeErrors_ = createRate(PerSessionMetrics_, "interconnect.scope_errors"); - - for (const auto& [id, name] : Common->ChannelName) { - OutputChannels_.try_emplace(id, std::make_shared<NMonitoring::TMetricSubRegistry>( - NMonitoring::TLabels{{"channel", name}}, Metrics_), Traffic_, Events_); - } - OtherOutputChannel_ = TOutputChannel(std::make_shared<NMonitoring::TMetricSubRegistry>( - NMonitoring::TLabels{{"channel", "other"}}, Metrics_), Traffic_, Events_); - - InputChannels_ = TInputChannels(Metrics_, Common->ChannelName, Traffic_, Events_, ScopeErrors_); - } - - if (updateAdaptive) { - SessionDeaths_ = createRate(AdaptiveMetrics_, "interconnect.session_deaths"); - HandshakeFails_ = createRate(AdaptiveMetrics_, "interconnect.handshake_fails"); - InflyLimitReach_ = createRate(AdaptiveMetrics_, "interconnect.infly_limit_reach"); - InflightDataAmount_ = createRate(AdaptiveMetrics_, "interconnect.inflight_data"); - PingTimeHistogram_ = AdaptiveMetrics_->HistogramRate( - NMonitoring::MakeLabels({{"sensor", "interconnect.ping_time_us"}}), NMonitoring::ExponentialHistogram(18, 2, 125)); - } - - if (updateGlobal) { - OutputBuffersTotalSize_ = createRate(Metrics_, "interconnect.output_buffers_total_size"); - SendSyscalls_ = createRate(Metrics_, "interconnect.send_syscalls"); - RecvSyscalls_ = createRate(Metrics_, "interconnect.recv_syscalls"); - SpuriousReadWakeups_ = createRate(Metrics_, "interconnect.spurious_read_wakeups"); - UsefulReadWakeups_ = createRate(Metrics_, "interconnect.useful_read_wakeups"); - SpuriousWriteWakeups_ = createRate(Metrics_, "interconnect.spurious_write_wakeups"); - UsefulWriteWakeups_ = createRate(Metrics_, "interconnect.useful_write_wakeups"); - SubscribersCount_ = createIntGauge(AdaptiveMetrics_, "interconnect.subscribers_count"); - TotalBytesWritten_ = createRate(Metrics_, "interconnect.total_bytes_written"); - TotalBytesRead_ = createRate(Metrics_, "interconnect.total_bytes_read"); - - for (const char *reason : TDisconnectReason::Reasons) { + if (GetOutputChannel(channel).OutgoingTraffic) { + GetOutputChannel(channel).OutgoingTraffic->Add(value); + } + if (GetOutputChannel(channel).Traffic) { + GetOutputChannel(channel).Traffic->Add(value); + } + } + + void UpdateOutputChannelEvents(ui16 channel) override { + if (GetOutputChannel(channel).OutgoingEvents) { + GetOutputChannel(channel).OutgoingEvents->Inc(); + } + if (GetOutputChannel(channel).Events) { + GetOutputChannel(channel).Events->Inc(); + } + } + + void SetPeerInfo(const TString& name, const TString& dataCenterId) override { + if (name != std::exchange(HumanFriendlyPeerHostName, name)) { + PerSessionMetrics_.reset(); + } + VALGRIND_MAKE_READABLE(&DataCenterId, sizeof(DataCenterId)); + if (dataCenterId != std::exchange(DataCenterId, dataCenterId)) { + PerDataCenterMetrics_.reset(); + } + + const bool updatePerDataCenter = !PerDataCenterMetrics_ && MergePerDataCenterMetrics_; + if (updatePerDataCenter) { + PerDataCenterMetrics_ = std::make_shared<NMonitoring::TMetricSubRegistry>( + NMonitoring::TLabels{{"datacenter_id", *DataCenterId}}, Metrics_); + } + + const bool updatePerSession = !PerSessionMetrics_ || updatePerDataCenter; + if (updatePerSession) { + auto base = MergePerDataCenterMetrics_ ? PerDataCenterMetrics_ : Metrics_; + PerSessionMetrics_ = std::make_shared<NMonitoring::TMetricSubRegistry>( + NMonitoring::TLabels{{"peer", *HumanFriendlyPeerHostName}}, base); + } + + const bool updateGlobal = !Initialized_; + + const bool updateAdaptive = + &AdaptiveMetrics_ == &Metrics_ ? updateGlobal : + &AdaptiveMetrics_ == &PerSessionMetrics_ ? updatePerSession : + &AdaptiveMetrics_ == &PerDataCenterMetrics_ ? updatePerDataCenter : + false; + + auto createRate = [](std::shared_ptr<NMonitoring::IMetricRegistry> metrics, TStringBuf name) mutable { + return metrics->Rate(NMonitoring::MakeLabels(NMonitoring::TLabels{{"sensor", name}})); + }; + auto createIntGauge = [](std::shared_ptr<NMonitoring::IMetricRegistry> metrics, TStringBuf name) mutable { + return metrics->IntGauge(NMonitoring::MakeLabels(NMonitoring::TLabels{{"sensor", name}})); + }; + + if (updatePerSession) { + Connected_ = createIntGauge(PerSessionMetrics_, "interconnect.connected"); + Disconnections_ = createRate(PerSessionMetrics_, "interconnect.disconnections"); + ClockSkewMicrosec_ = createIntGauge(PerSessionMetrics_, "interconnect.clock_skew_microsec"); + Traffic_ = createRate(PerSessionMetrics_, "interconnect.traffic"); + Events_ = createRate(PerSessionMetrics_, "interconnect.events"); + ScopeErrors_ = createRate(PerSessionMetrics_, "interconnect.scope_errors"); + + for (const auto& [id, name] : Common->ChannelName) { + OutputChannels_.try_emplace(id, std::make_shared<NMonitoring::TMetricSubRegistry>( + NMonitoring::TLabels{{"channel", name}}, Metrics_), Traffic_, Events_); + } + OtherOutputChannel_ = TOutputChannel(std::make_shared<NMonitoring::TMetricSubRegistry>( + NMonitoring::TLabels{{"channel", "other"}}, Metrics_), Traffic_, Events_); + + InputChannels_ = TInputChannels(Metrics_, Common->ChannelName, Traffic_, Events_, ScopeErrors_); + } + + if (updateAdaptive) { + SessionDeaths_ = createRate(AdaptiveMetrics_, "interconnect.session_deaths"); + HandshakeFails_ = createRate(AdaptiveMetrics_, "interconnect.handshake_fails"); + InflyLimitReach_ = createRate(AdaptiveMetrics_, "interconnect.infly_limit_reach"); + InflightDataAmount_ = createRate(AdaptiveMetrics_, "interconnect.inflight_data"); + PingTimeHistogram_ = AdaptiveMetrics_->HistogramRate( + NMonitoring::MakeLabels({{"sensor", "interconnect.ping_time_us"}}), NMonitoring::ExponentialHistogram(18, 2, 125)); + } + + if (updateGlobal) { + OutputBuffersTotalSize_ = createRate(Metrics_, "interconnect.output_buffers_total_size"); + SendSyscalls_ = createRate(Metrics_, "interconnect.send_syscalls"); + RecvSyscalls_ = createRate(Metrics_, "interconnect.recv_syscalls"); + SpuriousReadWakeups_ = createRate(Metrics_, "interconnect.spurious_read_wakeups"); + UsefulReadWakeups_ = createRate(Metrics_, "interconnect.useful_read_wakeups"); + SpuriousWriteWakeups_ = createRate(Metrics_, "interconnect.spurious_write_wakeups"); + UsefulWriteWakeups_ = createRate(Metrics_, "interconnect.useful_write_wakeups"); + SubscribersCount_ = createIntGauge(AdaptiveMetrics_, "interconnect.subscribers_count"); + TotalBytesWritten_ = createRate(Metrics_, "interconnect.total_bytes_written"); + TotalBytesRead_ = createRate(Metrics_, "interconnect.total_bytes_read"); + + for (const char *reason : TDisconnectReason::Reasons) { DisconnectByReason_[reason] = Metrics_->Rate( NMonitoring::MakeLabels({ {"sensor", "interconnect.disconnect_reason"}, {"reason", reason}, })); - } - } - - Initialized_ = true; - } - - TOutputChannel GetOutputChannel(ui16 index) const { - Y_VERIFY(Initialized_); - const auto it = OutputChannels_.find(index); - return it != OutputChannels_.end() ? it->second : OtherOutputChannel_; - } - - private: - const TInterconnectProxyCommon::TPtr Common; - const bool MergePerDataCenterMetrics_; - const bool MergePerPeerMetrics_; - std::shared_ptr<NMonitoring::IMetricRegistry> Metrics_; - std::shared_ptr<NMonitoring::IMetricRegistry> PerSessionMetrics_; - std::shared_ptr<NMonitoring::IMetricRegistry> PerDataCenterMetrics_; - std::shared_ptr<NMonitoring::IMetricRegistry>& AdaptiveMetrics_; - bool Initialized_ = false; - - NMonitoring::IRate* Traffic_; - - NMonitoring::IRate* Events_; - NMonitoring::IRate* ScopeErrors_; - NMonitoring::IRate* Disconnections_; - NMonitoring::IIntGauge* Connected_; - - NMonitoring::IRate* SessionDeaths_; - NMonitoring::IRate* HandshakeFails_; - NMonitoring::IRate* InflyLimitReach_; - NMonitoring::IRate* InflightDataAmount_; - NMonitoring::IRate* OutputBuffersTotalSize_; - NMonitoring::IIntGauge* SubscribersCount_; - NMonitoring::IRate* SendSyscalls_; - NMonitoring::IRate* RecvSyscalls_; - NMonitoring::IRate* SpuriousWriteWakeups_; - NMonitoring::IRate* UsefulWriteWakeups_; - NMonitoring::IRate* SpuriousReadWakeups_; - NMonitoring::IRate* UsefulReadWakeups_; - NMonitoring::IIntGauge* ClockSkewMicrosec_; - - NMonitoring::IHistogram* PingTimeHistogram_; - - std::unordered_map<ui16, TOutputChannel> OutputChannels_; - TOutputChannel OtherOutputChannel_; - TInputChannels InputChannels_; - - THashMap<TString, NMonitoring::IRate*> DisconnectByReason_; - - NMonitoring::IRate* TotalBytesWritten_; - NMonitoring::IRate* TotalBytesRead_; - }; - -} // namespace - -std::unique_ptr<IInterconnectMetrics> CreateInterconnectCounters(const TInterconnectProxyCommon::TPtr& common) { - return std::make_unique<TInterconnectCounters>(common); -} - -std::unique_ptr<IInterconnectMetrics> CreateInterconnectMetrics(const TInterconnectProxyCommon::TPtr& common) { - return std::make_unique<TInterconnectMetrics>(common); -} - + } + } + + Initialized_ = true; + } + + TOutputChannel GetOutputChannel(ui16 index) const { + Y_VERIFY(Initialized_); + const auto it = OutputChannels_.find(index); + return it != OutputChannels_.end() ? it->second : OtherOutputChannel_; + } + + private: + const TInterconnectProxyCommon::TPtr Common; + const bool MergePerDataCenterMetrics_; + const bool MergePerPeerMetrics_; + std::shared_ptr<NMonitoring::IMetricRegistry> Metrics_; + std::shared_ptr<NMonitoring::IMetricRegistry> PerSessionMetrics_; + std::shared_ptr<NMonitoring::IMetricRegistry> PerDataCenterMetrics_; + std::shared_ptr<NMonitoring::IMetricRegistry>& AdaptiveMetrics_; + bool Initialized_ = false; + + NMonitoring::IRate* Traffic_; + + NMonitoring::IRate* Events_; + NMonitoring::IRate* ScopeErrors_; + NMonitoring::IRate* Disconnections_; + NMonitoring::IIntGauge* Connected_; + + NMonitoring::IRate* SessionDeaths_; + NMonitoring::IRate* HandshakeFails_; + NMonitoring::IRate* InflyLimitReach_; + NMonitoring::IRate* InflightDataAmount_; + NMonitoring::IRate* OutputBuffersTotalSize_; + NMonitoring::IIntGauge* SubscribersCount_; + NMonitoring::IRate* SendSyscalls_; + NMonitoring::IRate* RecvSyscalls_; + NMonitoring::IRate* SpuriousWriteWakeups_; + NMonitoring::IRate* UsefulWriteWakeups_; + NMonitoring::IRate* SpuriousReadWakeups_; + NMonitoring::IRate* UsefulReadWakeups_; + NMonitoring::IIntGauge* ClockSkewMicrosec_; + + NMonitoring::IHistogram* PingTimeHistogram_; + + std::unordered_map<ui16, TOutputChannel> OutputChannels_; + TOutputChannel OtherOutputChannel_; + TInputChannels InputChannels_; + + THashMap<TString, NMonitoring::IRate*> DisconnectByReason_; + + NMonitoring::IRate* TotalBytesWritten_; + NMonitoring::IRate* TotalBytesRead_; + }; + +} // namespace + +std::unique_ptr<IInterconnectMetrics> CreateInterconnectCounters(const TInterconnectProxyCommon::TPtr& common) { + return std::make_unique<TInterconnectCounters>(common); +} + +std::unique_ptr<IInterconnectMetrics> CreateInterconnectMetrics(const TInterconnectProxyCommon::TPtr& common) { + return std::make_unique<TInterconnectMetrics>(common); +} + } // NActors |
