add-log-buffer 2

add log buffer

changed names

added move

changed details

fixed released buffer procedure

returned condition IgnoredCount > 0

default bufferSizeLimitBytes = 0

returned passedCount declaration

returned Y_VERIFY

removed passedcount

changed buffer reducing

no new line at the end of file

added srcs in ya.make

add log buffer

1 files changed, 56 insertions, 145 deletions

diff --git a/library/cpp/actors/core/log.cpp b/library/cpp/actors/core/log.cpp
index d005be6535..eb8b19193c 100644
--- a/library/cpp/actors/core/log.cpp
+++ b/library/cpp/actors/core/log.cpp
@@ -1,7 +1,4 @@
 #include "log.h"
-#include "log_settings.h"
-
-#include <library/cpp/monlib/service/pages/templates.h>
 
 static_assert(int(NActors::NLog::PRI_EMERG) == int(::TLOG_EMERG), "expect int(NActors::NLog::PRI_EMERG) == int(::TLOG_EMERG)");
 static_assert(int(NActors::NLog::PRI_ALERT) == int(::TLOG_ALERT), "expect int(NActors::NLog::PRI_ALERT) == int(::TLOG_ALERT)");
@@ -33,137 +30,6 @@ namespace {
 }
 
 namespace NActors {
-
-    class TLoggerCounters : public ILoggerMetrics {
-    public:
-        TLoggerCounters(TIntrusivePtr<NMonitoring::TDynamicCounters> counters)
-            : DynamicCounters(counters)
-        {
-            ActorMsgs_ = DynamicCounters->GetCounter("ActorMsgs", true);
-            DirectMsgs_ = DynamicCounters->GetCounter("DirectMsgs", true);
-            LevelRequests_ = DynamicCounters->GetCounter("LevelRequests", true);
-            IgnoredMsgs_ = DynamicCounters->GetCounter("IgnoredMsgs", true);
-            DroppedMsgs_ = DynamicCounters->GetCounter("DroppedMsgs", true);
-
-            AlertMsgs_ = DynamicCounters->GetCounter("AlertMsgs", true);
-            EmergMsgs_ = DynamicCounters->GetCounter("EmergMsgs", true);
-        }
-
-        ~TLoggerCounters() = default;
-
-        void IncActorMsgs() override {
-            ++*ActorMsgs_;
-        }
-        void IncDirectMsgs() override {
-            ++*DirectMsgs_;
-        }
-        void IncLevelRequests() override {
-            ++*LevelRequests_;
-        }
-        void IncIgnoredMsgs() override {
-            ++*IgnoredMsgs_;
-        }
-        void IncAlertMsgs() override {
-            ++*AlertMsgs_;
-        }
-        void IncEmergMsgs() override {
-            ++*EmergMsgs_;
-        }
-        void IncDroppedMsgs() override {
-            DroppedMsgs_->Inc();
-        };
-
-        void GetOutputHtml(IOutputStream& str) override {
-            HTML(str) {
-                DIV_CLASS("row") {
-                    DIV_CLASS("col-md-12") {
-                        TAG(TH4) {
-                            str << "Counters" << Endl;
-                        }
-                        DynamicCounters->OutputHtml(str);
-                    }
-                }
-            }
-        }
-
-    private:
-        NMonitoring::TDynamicCounters::TCounterPtr ActorMsgs_;
-        NMonitoring::TDynamicCounters::TCounterPtr DirectMsgs_;
-        NMonitoring::TDynamicCounters::TCounterPtr LevelRequests_;
-        NMonitoring::TDynamicCounters::TCounterPtr IgnoredMsgs_;
-        NMonitoring::TDynamicCounters::TCounterPtr AlertMsgs_;
-        NMonitoring::TDynamicCounters::TCounterPtr EmergMsgs_;
-        // Dropped while the logger backend was unavailable
-        NMonitoring::TDynamicCounters::TCounterPtr DroppedMsgs_;
-
-        TIntrusivePtr<NMonitoring::TDynamicCounters> DynamicCounters;
-    };
-
-    class TLoggerMetrics : public ILoggerMetrics {
-    public:
-        TLoggerMetrics(std::shared_ptr<NMonitoring::TMetricRegistry> metrics)
-            : Metrics(metrics)
-        {
-            ActorMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.actor_msgs"}});
-            DirectMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.direct_msgs"}});
-            LevelRequests_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.level_requests"}});
-            IgnoredMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.ignored_msgs"}});
-            DroppedMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.dropped_msgs"}});
-
-            AlertMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.alert_msgs"}});
-            EmergMsgs_ = Metrics->Rate(NMonitoring::TLabels{{"sensor", "logger.emerg_msgs"}});
-        }
-
-        ~TLoggerMetrics() = default;
-
-        void IncActorMsgs() override {
-            ActorMsgs_->Inc();
-        }
-        void IncDirectMsgs() override {
-            DirectMsgs_->Inc();
-        }
-        void IncLevelRequests() override {
-            LevelRequests_->Inc();
-        }
-        void IncIgnoredMsgs() override {
-            IgnoredMsgs_->Inc();
-        }
-        void IncAlertMsgs() override {
-            AlertMsgs_->Inc();
-        }
-        void IncEmergMsgs() override {
-            EmergMsgs_->Inc();
-        }
-        void IncDroppedMsgs() override {
-            DroppedMsgs_->Inc();
-        };
-
-        void GetOutputHtml(IOutputStream& str) override {
-            HTML(str) {
-                DIV_CLASS("row") {
-                    DIV_CLASS("col-md-12") {
-                        TAG(TH4) {
-                            str << "Metrics" << Endl;
-                        }
-                        // TODO: Now, TMetricRegistry does not have the GetOutputHtml function
-                    }
-                }
-            }
-        }
-
-    private:
-        NMonitoring::TRate* ActorMsgs_;
-        NMonitoring::TRate* DirectMsgs_;
-        NMonitoring::TRate* LevelRequests_;
-        NMonitoring::TRate* IgnoredMsgs_;
-        NMonitoring::TRate* AlertMsgs_;
-        NMonitoring::TRate* EmergMsgs_;
-        // Dropped while the logger backend was unavailable
-        NMonitoring::TRate* DroppedMsgs_;
-
-        std::shared_ptr<NMonitoring::TMetricRegistry> Metrics;
-    };
-
     TAtomic TLoggerActor::IsOverflow = 0;
 
     TLoggerActor::TLoggerActor(TIntrusivePtr<NLog::TSettings> settings,
@@ -173,6 +39,7 @@ namespace NActors {
         , Settings(settings)
         , LogBackend(logBackend.Release())
         , Metrics(std::make_unique<TLoggerCounters>(counters))
+        , LogBuffer(Metrics.get())
     {
     }
 
@@ -183,6 +50,7 @@ namespace NActors {
         , Settings(settings)
         , LogBackend(logBackend)
         , Metrics(std::make_unique<TLoggerCounters>(counters))
+        , LogBuffer(Metrics.get())
     {
     }
 
@@ -193,6 +61,7 @@ namespace NActors {
         , Settings(settings)
         , LogBackend(logBackend.Release())
         , Metrics(std::make_unique<TLoggerMetrics>(metrics))
+        , LogBuffer(Metrics.get())
     {
     }
 
@@ -203,6 +72,7 @@ namespace NActors {
         , Settings(settings)
         , LogBackend(logBackend)
         , Metrics(std::make_unique<TLoggerMetrics>(metrics))
+        , LogBuffer(Metrics.get())
     {
     }
 
@@ -242,6 +112,23 @@ namespace NActors {
         PassedCount = 0;
     }
 
+    void TLoggerActor::ReleaseLogBufferMessage() {
+        if (LogBuffer.GetLogsNumber() > 0) {
+            auto message = LogBuffer.GetMessage();  
+            if (!OutputRecord(message.Time, message.Priority, message.Component, message.Formatted)) {
+                BecomeDefunct();
+            }
+        }
+    }
+
+    void TLoggerActor::ReleaseLogBufferMessageEvent(TReleaseLogBuffer::TPtr& ev, const NActors::TActorContext& ctx) {
+        Y_UNUSED(ev);
+        ReleaseLogBufferMessage();
+        if (LogBuffer.GetLogsNumber() > 0) {
+            ctx.Send(ctx.SelfID, new TReleaseLogBuffer());
+        }
+    }
+
     void TLoggerActor::HandleIgnoredEventDrop() {
         // logger backend is unavailable, just ignore
     }
@@ -264,6 +151,30 @@ namespace NActors {
 
     }
 
+    bool TLoggerActor::TryAddLogBuffer(TLogBufferMessage& lbm) {
+        ui64 messageSize = sizeof(lbm);
+        if (LogBuffer.GetSizeBytes() + messageSize < Settings->BufferSizeLimitBytes)
+            return LogBuffer.TryAddMessage(lbm);
+
+        return false;
+    }
+
+    bool TLoggerActor::TryKeepLog(TLogBufferMessage& lbm, const NActors::TActorContext& ctx) {
+        bool wasEmtpyBuffer = LogBuffer.IsEmpty();
+        
+        if (TryAddLogBuffer(lbm)) {
+        } else if (!LogBuffer.TryReduceLogsNumber()) {
+            return false;
+        } else if (!TryAddLogBuffer(lbm)) {
+            return false;
+        }
+
+        if (wasEmtpyBuffer) {
+            ctx.Send(ctx.SelfID, new TReleaseLogBuffer());
+        }
+        return true;
+    }
+
     void TLoggerActor::HandleLogEvent(NLog::TEvLog::TPtr& ev, const NActors::TActorContext& ctx) {
         i64 delayMillisec = (ctx.Now() - ev->Get()->Stamp).MilliSeconds();
         WriteMessageStat(*ev->Get());
@@ -271,16 +182,17 @@ namespace NActors {
             // Disable throttling if it was enabled previously
             if (AtomicGet(IsOverflow))
                 AtomicSet(IsOverflow, 0);
-
-            // Check if some records have to be dropped
-            if ((PassedCount > 10 && delayMillisec > (i64)Settings->TimeThresholdMs) || IgnoredCount > 0) {
-                Metrics->IncIgnoredMsgs();
-                if (IgnoredCount == 0) {
-                    ctx.Send(ctx.SelfID, new TLogIgnored());
+            if (PassedCount > 10 && delayMillisec > (i64)Settings->TimeThresholdMs || IgnoredCount > 0 || !LogBuffer.IsEmpty()) {
+                TLogBufferMessage lbm(ev);
+                if (!TryKeepLog(lbm, ctx)) {
+                    Metrics->IncIgnoredMsgs();
+                    if (IgnoredCount == 0) {
+                        ctx.Send(ctx.SelfID, new TLogIgnored());
+                    }
+                    ++IgnoredCount;
+                    PassedCount = 0;
                 }
-                ++IgnoredCount;
-                PassedCount = 0;
-                return;
+                return; 
             }
             PassedCount++;
         } else {
@@ -291,8 +203,7 @@ namespace NActors {
                 AtomicSet(IsOverflow, 0);
         }
 
-        const auto prio = ev->Get()->Level.ToPrio();
-        if (!OutputRecord(ev->Get()->Stamp, prio, ev->Get()->Component, ev->Get()->Line)) {
+        if (!OutputRecord(ev->Get()->Stamp, ev->Get()->Level.ToPrio(), ev->Get()->Component, ev->Get()->Line)) {
             BecomeDefunct();
         }
     }