switch metrics registry to shared_ptr KIKIMR-14218

ref:5ef16c21e373b05982980dcd3016950f15a884e8

7 files changed, 96 insertions, 36 deletions

diff --git a/library/cpp/actors/core/process_stats.cpp b/library/cpp/actors/core/process_stats.cpp
index ef793760b54..0fdad83da59 100644
--- a/library/cpp/actors/core/process_stats.cpp
+++ b/library/cpp/actors/core/process_stats.cpp
@@ -231,7 +231,6 @@ namespace {
         NMonitoring::TDynamicCounters::TCounterPtr SystemUptimeSeconds;
     };
 
-
     class TRegistryCollector: public TProcStatCollectingActor<TRegistryCollector> {
         using TBase = TProcStatCollectingActor<TRegistryCollector>;
     public:
@@ -290,6 +289,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) {
@@ -299,4 +344,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));
+    }
 }
diff --git a/library/cpp/actors/core/process_stats.h b/library/cpp/actors/core/process_stats.h
index 66346d0b5aa..5681f0eb1a8 100644
--- a/library/cpp/actors/core/process_stats.h
+++ b/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);
 }
diff --git a/library/cpp/actors/http/http_proxy.cpp b/library/cpp/actors/http/http_proxy.cpp
index 36c6855d93f..e0204f6ed0d 100644
--- a/library/cpp/actors/http/http_proxy.cpp
+++ b/library/cpp/actors/http/http_proxy.cpp
@@ -26,8 +26,8 @@ public:
         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 @@ protected:
         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 @@ protected:
     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) {
diff --git a/library/cpp/actors/http/http_proxy.h b/library/cpp/actors/http/http_proxy.h
index afd0170997e..92002b047d3 100644
--- a/library/cpp/actors/http/http_proxy.h
+++ b/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(
diff --git a/library/cpp/actors/http/http_ut.cpp b/library/cpp/actors/http/http_ut.cpp
index 4c922f8d0fc..caa5b3e1831 100644
--- a/library/cpp/actors/http/http_ut.cpp
+++ b/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();
diff --git a/library/cpp/monlib/metrics/metric_registry.cpp b/library/cpp/monlib/metrics/metric_registry.cpp
index 74260e30693..3d7ffd40cf0 100644
--- a/library/cpp/monlib/metrics/metric_registry.cpp
+++ b/library/cpp/monlib/metrics/metric_registry.cpp
@@ -43,9 +43,15 @@ namespace NMonitoring {
     TMetricRegistry& TMetricRegistry::operator=(TMetricRegistry&& other) = default;
 
     TMetricRegistry* TMetricRegistry::Instance() {
+        //return SharedInstance().get();
         return Singleton<TMetricRegistry>();
     }
 
+    std::shared_ptr<TMetricRegistry> TMetricRegistry::SharedInstance() {
+        static auto instance(std::make_shared<TMetricRegistry>());
+        return instance;
+    }
+
     TGauge* TMetricRegistry::Gauge(TLabels labels) {
         return Metric<TGauge, EMetricType::GAUGE>(std::move(labels));
     }
diff --git a/library/cpp/monlib/metrics/metric_registry.h b/library/cpp/monlib/metrics/metric_registry.h
index 06e61d9991a..faba19e8454 100644
--- a/library/cpp/monlib/metrics/metric_registry.h
+++ b/library/cpp/monlib/metrics/metric_registry.h
@@ -67,6 +67,7 @@ namespace NMonitoring {
          * Get a global metrics registry instance.
          */
         static TMetricRegistry* Instance();
+        static std::shared_ptr<TMetricRegistry> SharedInstance();
 
         TGauge* Gauge(TLabels labels);
         TLazyGauge* LazyGauge(TLabels labels, std::function<double()> supplier);