Упоминание Аркадии

library/cpp/actors/core/interconnect.cpp

@@ -106,7 +106,7 @@ namespace NActors {
         return res;
     }
 
-    void TNodeLocation::Serialize(NActorsInterconnect::TNodeLocation *pb) const {
+    void TNodeLocation::Serialize(NActorsInterconnect::TNodeLocation *pb, bool compatibleWithOlderVersions) const {
         const NProtoBuf::Descriptor *descriptor = NActorsInterconnect::TNodeLocation::descriptor();
         const NProtoBuf::Reflection *reflection = pb->GetReflection();
         NProtoBuf::UnknownFieldSet *unknown = pb->mutable_unknown_fields();
@@ -117,11 +117,14 @@ namespace NActors {
                 unknown->AddLengthDelimited(key)->assign(value);
             }
         }
+        if (compatibleWithOlderVersions) {
+            GetLegacyValue().Serialize(pb);
+        }
     }
 
     TString TNodeLocation::GetSerializedLocation() const {
         NActorsInterconnect::TNodeLocation pb;
-        Serialize(&pb);
+        Serialize(&pb, false);
         TString s;
         const bool success = pb.SerializeToString(&s);
         Y_VERIFY(success);

library/cpp/actors/core/interconnect.h

@@ -72,7 +72,8 @@ namespace NActors {
         TNodeLocation& operator =(const TNodeLocation&) = default;
         TNodeLocation& operator =(TNodeLocation&&) = default;
 
-        void Serialize(NActorsInterconnect::TNodeLocation *pb) const;
+        // compatibleWithOlderVersions should be set to true when this protobuf is possibly going to be delivered to 21-4
+        void Serialize(NActorsInterconnect::TNodeLocation *pb, bool compatibleWithOlderVersions) const;
         TString GetSerializedLocation() const;
 
         TString GetDataCenterId() const { return ToStringUpTo(TKeys::DataCenter); }

library/cpp/actors/core/monotonic_provider.cpp

@@ -0,0 +1,16 @@
+#include "monotonic_provider.h"
+
+namespace NActors {
+
+class TDefaultMonotonicTimeProvider : public IMonotonicTimeProvider {
+public:
+    TMonotonic Now() override {
+        return TMonotonic::Now();
+    }
+};
+
+TIntrusivePtr<IMonotonicTimeProvider> CreateDefaultMonotonicTimeProvider() {
+    return TIntrusivePtr<IMonotonicTimeProvider>(new TDefaultMonotonicTimeProvider);
+}
+
+} // namespace NActors

library/cpp/actors/core/monotonic_provider.h

@@ -0,0 +1,14 @@
+#pragma once
+
+#include "monotonic.h"
+
+namespace NActors {
+
+class IMonotonicTimeProvider : public TThrRefBase {
+public:
+    virtual TMonotonic Now() = 0;
+};
+
+TIntrusivePtr<IMonotonicTimeProvider> CreateDefaultMonotonicTimeProvider();
+
+} // namespace NActors

library/cpp/actors/core/process_stats.cpp

@@ -232,7 +232,6 @@ namespace {
         NMonitoring::TDynamicCounters::TCounterPtr SystemUptimeSeconds;
     };
 
-
     class TRegistryCollector: public TProcStatCollectingActor<TRegistryCollector> {
         using TBase = TProcStatCollectingActor<TRegistryCollector>;
     public:
@@ -291,6 +290,52 @@ namespace {
         NMonitoring::TIntGauge* NumThreads;
         NMonitoring::TIntGauge* SystemUptimeSeconds;
     };
+
+    class TRegistryCollectorShared: public TProcStatCollectingActor<TRegistryCollectorShared> {
+        using TBase = TProcStatCollectingActor<TRegistryCollectorShared>;
+    public:
+        TRegistryCollectorShared(TDuration interval, std::weak_ptr<NMonitoring::TMetricRegistry> registry)
+            : TBase{interval}
+            , Registry(std::move(registry))
+        {
+        }
+
+        void UpdateCounters(const TProcStat& procStat) {
+            std::shared_ptr<NMonitoring::TMetricRegistry> registry = Registry.lock();
+            if (registry) {
+                registry->IntGauge({{"sensor", "process.VmSize"}})->Set(procStat.Vsize);
+                registry->IntGauge({{"sensor", "process.AnonRssSize"}})->Set(procStat.AnonRss);
+                registry->IntGauge({{"sensor", "process.FileRssSize"}})->Set(procStat.FileRss);
+                registry->IntGauge({{"sensor", "process.CGroupMemLimit"}})->Set(procStat.CGroupMemLim);
+                registry->IntGauge({{"sensor", "process.UptimeSeconds"}})->Set(procStat.Uptime.Seconds());
+                registry->IntGauge({{"sensor", "process.NumThreads"}})->Set(procStat.NumThreads);
+                registry->IntGauge({{"sensor", "system.UptimeSeconds"}})->Set(procStat.SystemUptime.Seconds());
+
+                // it is ok here to reset and add metric value, because mutation
+                // is performed in siglethreaded context
+
+                NMonitoring::TRate* userTime = registry->Rate({{"sensor", "process.UserTime"}});
+                NMonitoring::TRate* sysTime = registry->Rate({{"sensor", "process.SystemTime"}});
+                NMonitoring::TRate* minorPageFaults = registry->Rate({{"sensor", "process.MinorPageFaults"}});
+                NMonitoring::TRate* majorPageFaults = registry->Rate({{"sensor", "process.MajorPageFaults"}});
+
+                userTime->Reset();
+                userTime->Add(procStat.Utime);
+
+                sysTime->Reset();
+                sysTime->Add(procStat.Stime);
+
+                minorPageFaults->Reset();
+                minorPageFaults->Add(procStat.MinFlt);
+
+                majorPageFaults->Reset();
+                majorPageFaults->Add(procStat.MajFlt);
+            }
+        }
+
+    private:
+        std::weak_ptr<NMonitoring::TMetricRegistry> Registry;
+    };
 } // namespace
 
     IActor* CreateProcStatCollector(ui32 intervalSec, NMonitoring::TDynamicCounterPtr counters) {
@@ -300,4 +345,8 @@ namespace {
     IActor* CreateProcStatCollector(TDuration interval, NMonitoring::TMetricRegistry& registry) {
         return new TRegistryCollector(interval, registry);
     }
+
+    IActor* CreateProcStatCollector(TDuration interval, std::weak_ptr<NMonitoring::TMetricRegistry> registry) {
+        return new TRegistryCollectorShared(interval, std::move(registry));
+    }
 }

library/cpp/actors/core/process_stats.h

@@ -63,4 +63,5 @@ namespace NActors {
 
     IActor* CreateProcStatCollector(ui32 intervalSec, NMonitoring::TDynamicCounterPtr counters);
     IActor* CreateProcStatCollector(TDuration interval, NMonitoring::TMetricRegistry& registry);
+    IActor* CreateProcStatCollector(TDuration interval, std::weak_ptr<NMonitoring::TMetricRegistry> registry);
 }

library/cpp/actors/core/ya.make

@@ -83,6 +83,8 @@ SRCS(
     mon_stats.h
     monotonic.cpp
     monotonic.h
+    monotonic_provider.cpp
+    monotonic_provider.h
     worker_context.cpp
     worker_context.h
     probes.cpp

library/cpp/actors/http/http_proxy.cpp

@@ -26,8 +26,8 @@ class THttpProxy : public NActors::TActorBootstrapped<THttpProxy>, public THttpC
         Become(&THttpProxy::StateWork);
     }
 
-    THttpProxy(NMonitoring::TMetricRegistry& sensors)
-        : Sensors(sensors)
+    THttpProxy(std::weak_ptr<NMonitoring::TMetricRegistry> registry)
+        : Registry(std::move(registry))
     {}
 
 protected:
@@ -175,29 +175,35 @@ class THttpProxy : public NActors::TActorBootstrapped<THttpProxy>, public THttpC
         const static TString urlNotFound = "not-found";
         const TString& url = (sensors.Status == "404" ? urlNotFound : sensors.Url);
 
-        Sensors.Rate({
-                         {"sensor", "count"},
-                         {"direction", sensors.Direction},
-                         {"peer", sensors.Host},
-                         {"url", url},
-                         {"status", sensors.Status}
-                     })->Inc();
-        Sensors.HistogramRate({
-                                  {"sensor", "time_us"},
-                                  {"direction", sensors.Direction},
-                                  {"peer", sensors.Host},
-                                  {"url", url},
-                                  {"status", sensors.Status}
-                              },
-                              NMonitoring::ExplicitHistogram({1, 5, 10, 50, 100, 500, 1000, 5000, 10000, 30000, 60000}))->Record(sensors.Time.MicroSeconds());
-        Sensors.HistogramRate({
-                                  {"sensor", "time_ms"},
-                                  {"direction", sensors.Direction},
-                                  {"peer", sensors.Host},
-                                  {"url", url},
-                                  {"status", sensors.Status}
-                              },
-                              NMonitoring::ExplicitHistogram({1, 5, 10, 50, 100, 500, 1000, 5000, 10000, 30000, 60000}))->Record(sensors.Time.MilliSeconds());
+        std::shared_ptr<NMonitoring::TMetricRegistry> registry = Registry.lock();
+        if (registry) {
+            registry->Rate(
+                {
+                    {"sensor", "count"},
+                    {"direction", sensors.Direction},
+                    {"peer", sensors.Host},
+                    {"url", url},
+                    {"status", sensors.Status}
+                })->Inc();
+            registry->HistogramRate(
+                {
+                    {"sensor", "time_us"},
+                    {"direction", sensors.Direction},
+                    {"peer", sensors.Host},
+                    {"url", url},
+                    {"status", sensors.Status}
+                },
+                NMonitoring::ExplicitHistogram({1, 5, 10, 50, 100, 500, 1000, 5000, 10000, 30000, 60000}))->Record(sensors.Time.MicroSeconds());
+            registry->HistogramRate(
+                {
+                    {"sensor", "time_ms"},
+                    {"direction", sensors.Direction},
+                    {"peer", sensors.Host},
+                    {"url", url},
+                    {"status", sensors.Status}
+                },
+                NMonitoring::ExplicitHistogram({1, 5, 10, 50, 100, 500, 1000, 5000, 10000, 30000, 60000}))->Record(sensors.Time.MilliSeconds());
+        }
     }
 
     void Handle(NActors::TEvents::TEvPoison::TPtr, const NActors::TActorContext&) {
@@ -217,7 +223,7 @@ class THttpProxy : public NActors::TActorBootstrapped<THttpProxy>, public THttpC
     THashMap<TString, THostEntry> Hosts;
     THashMap<TString, TActorId> Handlers;
     THashSet<TActorId> Connections; // outgoing
-    NMonitoring::TMetricRegistry& Sensors;
+    std::weak_ptr<NMonitoring::TMetricRegistry> Registry;
 };
 
 TEvHttpProxy::TEvReportSensors* BuildOutgoingRequestSensors(const THttpOutgoingRequestPtr& request, const THttpIncomingResponsePtr& response) {
@@ -240,8 +246,8 @@ TEvHttpProxy::TEvReportSensors* BuildIncomingRequestSensors(const THttpIncomingR
     );
 }
 
-NActors::IActor* CreateHttpProxy(NMonitoring::TMetricRegistry& sensors) {
-    return new THttpProxy(sensors);
+NActors::IActor* CreateHttpProxy(std::weak_ptr<NMonitoring::TMetricRegistry> registry) {
+    return new THttpProxy(std::move(registry));
 }
 
 bool IsIPv6(const TString& host) {

library/cpp/actors/http/http_proxy.h

@@ -225,7 +225,7 @@ struct TEndpointInfo {
     TSslHelpers::TSslHolder<SSL_CTX> SecureContext;
 };
 
-NActors::IActor* CreateHttpProxy(NMonitoring::TMetricRegistry& sensors);
+NActors::IActor* CreateHttpProxy(std::weak_ptr<NMonitoring::TMetricRegistry> registry = NMonitoring::TMetricRegistry::SharedInstance());
 NActors::IActor* CreateHttpAcceptorActor(const TActorId& owner, const TActorId& poller);
 NActors::IActor* CreateOutgoingConnectionActor(const TActorId& owner, const TString& host, bool secure, const TActorId& poller);
 NActors::IActor* CreateIncomingConnectionActor(

library/cpp/actors/http/http_ut.cpp

@@ -180,9 +180,8 @@ Y_UNIT_TEST_SUITE(HttpProxy) {
         TIpPort port = portManager.GetTcpPort();
         TAutoPtr<NActors::IEventHandle> handle;
         actorSystem.Initialize();
-        NMonitoring::TMetricRegistry sensors;
 
-        NActors::IActor* proxy = NHttp::CreateHttpProxy(sensors);
+        NActors::IActor* proxy = NHttp::CreateHttpProxy();
         NActors::TActorId proxyId = actorSystem.Register(proxy);
         actorSystem.Send(new NActors::IEventHandle(proxyId, TActorId(), new NHttp::TEvHttpProxy::TEvAddListeningPort(port)), 0, true);
         actorSystem.DispatchEvents();
@@ -213,7 +212,6 @@ Y_UNIT_TEST_SUITE(HttpProxy) {
         TIpPort port = portManager.GetTcpPort();
         TAutoPtr<NActors::IEventHandle> handle;
         actorSystem.Initialize();
-        NMonitoring::TMetricRegistry sensors;
 
         TString certificateContent = R"___(-----BEGIN PRIVATE KEY-----
 MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCzRZjodO7Aqe1w
@@ -272,7 +270,7 @@ CRA/5XcX13GJwHHj6LCoc3sL7mt8qV9HKY2AOZ88mpObzISZxgPpdKCfjsrdm63V
 
         certificateFile.Write(certificateContent.data(), certificateContent.size());
 
-        NActors::IActor* proxy = NHttp::CreateHttpProxy(sensors);
+        NActors::IActor* proxy = NHttp::CreateHttpProxy();
         NActors::TActorId proxyId = actorSystem.Register(proxy);
 
         THolder<NHttp::TEvHttpProxy::TEvAddListeningPort> add = MakeHolder<NHttp::TEvHttpProxy::TEvAddListeningPort>(port);
@@ -332,9 +330,8 @@ CRA/5XcX13GJwHHj6LCoc3sL7mt8qV9HKY2AOZ88mpObzISZxgPpdKCfjsrdm63V
         TIpPort port = portManager.GetTcpPort();
         TAutoPtr<NActors::IEventHandle> handle;
         actorSystem.Initialize();
-        NMonitoring::TMetricRegistry sensors;
 
-        NActors::IActor* proxy = NHttp::CreateHttpProxy(sensors);
+        NActors::IActor* proxy = NHttp::CreateHttpProxy();
         NActors::TActorId proxyId = actorSystem.Register(proxy);
         actorSystem.Send(new NActors::IEventHandle(proxyId, TActorId(), new NHttp::TEvHttpProxy::TEvAddListeningPort(port)), 0, true);
         actorSystem.DispatchEvents();

library/cpp/actors/interconnect/interconnect_tcp_input_session.cpp

@@ -430,6 +430,11 @@ namespace NActors {
         Socket.Reset();
     }
 
+    void TInputSessionTCP::PassAway() {
+        Metrics->SetClockSkewMicrosec(0);
+        TActorBootstrapped::PassAway();
+    }
+
     void TInputSessionTCP::HandleCheckDeadPeer() {
         const TInstant now = TActivationContext::Now();
         if (now >= LastReceiveTimestamp + DeadPeerTimeout) {

library/cpp/actors/interconnect/interconnect_tcp_session.h

@@ -255,6 +255,7 @@ namespace NActors {
 
         void ReestablishConnection(TDisconnectReason reason);
         void DestroySession(TDisconnectReason reason);
+        void PassAway() override;
 
         TDeque<TIntrusivePtr<TRopeAlignedBuffer>> Buffers;
 

library/cpp/actors/testlib/test_runtime.cpp

@@ -233,6 +233,20 @@ namespace NActors {
         TTestActorRuntimeBase& Runtime;
     };
 
+    class TTestActorRuntimeBase::TMonotonicTimeProvider : public IMonotonicTimeProvider {
+    public:
+        TMonotonicTimeProvider(TTestActorRuntimeBase& runtime)
+            : Runtime(runtime)
+        { }
+
+        TMonotonic Now() override {
+            return Runtime.GetCurrentMonotonicTime();
+        }
+
+    private:
+        TTestActorRuntimeBase& Runtime;
+    };
+
     class TTestActorRuntimeBase::TSchedulerThreadStub : public ISchedulerThread {
     public:
         TSchedulerThreadStub(TTestActorRuntimeBase* runtime, TTestActorRuntimeBase::TNodeDataBase* node)
@@ -470,6 +484,7 @@ namespace NActors {
         , NeedMonitoring(false)
         , RandomProvider(CreateDeterministicRandomProvider(DefaultRandomSeed))
         , TimeProvider(new TTimeProvider(*this))
+        , MonotonicTimeProvider(new TMonotonicTimeProvider(*this))
         , ShouldContinue()
         , CurrentTimestamp(0)
         , DispatchTimeout(DEFAULT_DISPATCH_TIMEOUT)
@@ -797,6 +812,12 @@ namespace NActors {
         return TInstant::MicroSeconds(CurrentTimestamp);
     }
 
+    TMonotonic TTestActorRuntimeBase::GetCurrentMonotonicTime() const {
+        TGuard<TMutex> guard(Mutex);
+        Y_VERIFY(!UseRealThreads);
+        return TMonotonic::MicroSeconds(CurrentTimestamp);
+    }
+
     void TTestActorRuntimeBase::UpdateCurrentTime(TInstant newTime) {
         static int counter = 0;
         ++counter;
@@ -823,6 +844,11 @@ namespace NActors {
         return TimeProvider;
     }
 
+    TIntrusivePtr<IMonotonicTimeProvider> TTestActorRuntimeBase::GetMonotonicTimeProvider() {
+        Y_VERIFY(!UseRealThreads);
+        return MonotonicTimeProvider;
+    }
+
     ui32 TTestActorRuntimeBase::GetNodeId(ui32 index) const {
         Y_VERIFY(index < NodeCount);
         return FirstNodeId + index;

library/cpp/actors/testlib/test_runtime.h

@@ -6,6 +6,7 @@
 #include <library/cpp/actors/core/events.h>
 #include <library/cpp/actors/core/executor_thread.h>
 #include <library/cpp/actors/core/mailbox.h>
+#include <library/cpp/actors/core/monotonic_provider.h>
 #include <library/cpp/actors/util/should_continue.h>
 #include <library/cpp/actors/interconnect/poller_tcp.h>
 #include <library/cpp/actors/interconnect/mock/ic_mock.h>
@@ -188,6 +189,7 @@ namespace NActors {
         class TSchedulerThreadStub;
         class TExecutorPoolStub;
         class TTimeProvider;
+        class TMonotonicTimeProvider;
 
         enum class EEventAction {
             PROCESS,
@@ -229,7 +231,9 @@ namespace NActors {
         void SetLogBackend(const TAutoPtr<TLogBackend> logBackend);
         void SetLogPriority(NActors::NLog::EComponent component, NActors::NLog::EPriority priority);
         TIntrusivePtr<ITimeProvider> GetTimeProvider();
+        TIntrusivePtr<IMonotonicTimeProvider> GetMonotonicTimeProvider();
         TInstant GetCurrentTime() const;
+        TMonotonic GetCurrentMonotonicTime() const;
         void UpdateCurrentTime(TInstant newTime);
         void AdvanceCurrentTime(TDuration duration);
         void AddLocalService(const TActorId& actorId, const TActorSetupCmd& cmd, ui32 nodeIndex = 0);
@@ -534,6 +538,7 @@ namespace NActors {
 
         TIntrusivePtr<IRandomProvider> RandomProvider;
         TIntrusivePtr<ITimeProvider> TimeProvider;
+        TIntrusivePtr<IMonotonicTimeProvider> MonotonicTimeProvider;
 
     protected:
         struct TNodeDataBase: public TThrRefBase {