[ML] Warn and error level log throttling (#1615)

This introduces a simple mechanism to throttle logging at WARN and ERROR level per process. In particular, it
  1. Always outputs the first WARN or ERROR for a particular log line
  2. Thereafter limits to outputting a duplicate line once per hour (wall clock)
This should be enough to allow us to check for presence of errors (useful for QA). Also, nearly always, only the first
instance of an error is useful.

Author: tveasey

docs/CHANGELOG.asciidoc

@@ -55,6 +55,7 @@
   unnecessary computation. (See {ml-pull}1537[#1537].)
 * Improvements to time series modeling particularly in relation to adaption to change.
   (See {ml-pull})1614[#1614].)
+* Warn and error log throttling. (See {ml-pull}1615[#1615].)
 
 === Bug Fixes
 

include/core/CLogger.h

@@ -6,6 +6,7 @@
 #ifndef INCLUDED_ml_core_CLogger_h
 #define INCLUDED_ml_core_CLogger_h
 
+#include <core/CLoggerThrottler.h>
 #include <core/CNamedPipeFactory.h>
 #include <core/CNonCopyable.h>
 #include <core/ImportExport.h>
@@ -160,6 +161,9 @@ class CORE_EXPORT CLogger : private CNonCopyable {
     //! CLogger is a singleton, so we can not just create new instances
     void reset();
 
+    //! Get the object which performs log throttling.
+    CLoggerThrottler& throttler();
+
 private:
     //! Constructor for a singleton is private.
     CLogger();
@@ -193,6 +197,9 @@ class CORE_EXPORT CLogger : private CNonCopyable {
 
     //! The default handler for fatal errors.
     TFatalErrorHandler m_FatalErrorHandler;
+
+    //! The log throttler.
+    CLoggerThrottler m_Throttler;
 };
 
 CORE_EXPORT std::ostream& operator<<(std::ostream& strm, CLogger::ELevel level);

include/core/CLoggerThrottler.h

@@ -0,0 +1,82 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+#ifndef INCLUDED_ml_core_CLoggerThrottler_h
+#define INCLUDED_ml_core_CLoggerThrottler_h
+
+#include <core/ImportExport.h>
+
+#include <boost/unordered_map.hpp>
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+
+namespace ml {
+namespace core {
+
+//! \brief Implements per log line throttling.
+//!
+//! DESCRIPTION:\n
+//! This implements global throttling for a program per line log line. By default
+//! the same log line will only be emitted once per hour.
+//!
+//! IMPLEMENTATION:\n
+//! This is thread safe but uses a very simple strategy: all accesses to a single
+//! hash map are sychronised. We assume that log throttling is only applied to
+//! messages which normally occur infrequently; for example, this is only currently
+//! applied to WARN and ERROR level logging (see LogMacros.h). So there will be
+//! little contention. Furthermore, the overhead of locking and unlocking the mutex
+//! should be neglible compared to the work done if the log line were actually
+//! emitted. So this should actually give a significant performance improvement
+//! if a log line is spamming.
+class CORE_EXPORT CLoggerThrottler {
+public:
+    CLoggerThrottler();
+    CLoggerThrottler(const CLoggerThrottler&) = delete;
+    CLoggerThrottler& operator=(const CLoggerThrottler&) = delete;
+
+    //! Set the minimum interval between repeated log messages.
+    //!
+    //! \note This expects the interval in milliseconds.
+    void minimumLogIntervalMs(std::int64_t minimumLogIntervalMs);
+
+    //! Should we skip logging of \p line in \p file?
+    //!
+    //! \param[in] file The file containing the log line.
+    //! \param[in] line The log line.
+    //! \return A pair comprising the count of log lines since the log line was
+    //! last output (skip returned false) and whether to output the line or not.
+    //! \note This is expected to be used in conjuction with __FILE__ and __LINE__
+    //! macros although any unique identifiers which are a string literal and an
+    //! integer are permitted.
+    //! \warning We make no attempt to deduplicate logging from the same line for
+    //! which __FILE__ resolves to a different string literal; this might conceivably
+    //! happen for logging in a header. This is not expected to be important (it
+    //! might result in slightly more logging) and it is possible to work around
+    //! this by explicitly wrapping the logging, for example:
+    //! \code
+    //! if (core::CLogger::instance().throttler().skip("my unique string", 0) == false) {
+    //!     LOG_ERROR(<< ...)
+    //! }
+    //! \endcode
+    std::pair<std::size_t, bool> skip(const char* file, int line);
+
+private:
+    using TConstCharPtrIntPr = std::pair<const char*, int>;
+    using TInt64SizePr = std::pair<std::int64_t, std::size_t>;
+    using TConstCharPtrIntPrInt64SizePrUMap =
+        boost::unordered_map<TConstCharPtrIntPr, TInt64SizePr>;
+
+private:
+    std::int64_t m_MinimumLogIntervalMs;
+    std::mutex m_Mutex;
+    TConstCharPtrIntPrInt64SizePrUMap m_LastLogTimesAndCounts;
+};
+}
+}
+
+#endif // INCLUDED_ml_core_CLoggerThrottler_h

include/core/LogMacros.h

@@ -12,7 +12,10 @@
 #include <boost/log/sources/severity_logger.hpp>
 #include <boost/log/utility/manipulators/add_value.hpp>
 
+#include <cstddef>
 #include <sstream>
+#include <string>
+#include <tuple>
 
 // Location info
 #ifdef LOG_LOCATION_INFO
@@ -73,25 +76,46 @@
 #ifdef LOG_WARN
 #undef LOG_WARN
 #endif
-#define LOG_WARN(message)                                                                   \
-    BOOST_LOG_STREAM_SEV(ml::core::CLogger::instance().logger(), ml::core::CLogger::E_Warn) \
-    LOG_LOCATION_INFO                                                                       \
-    message
+#define LOG_WARN(message)                                                                 \
+    {                                                                                     \
+        std::size_t countOfWarnMessages;                                                  \
+        bool skipWarnMessage;                                                             \
+        std::tie(countOfWarnMessages, skipWarnMessage) =                                  \
+            ml::core::CLogger::instance().throttler().skip(__FILE__, __LINE__);           \
+        if (skipWarnMessage == false) {                                                   \
+            BOOST_LOG_STREAM_SEV(ml::core::CLogger::instance().logger(),                  \
+                                 ml::core::CLogger::E_Warn)                               \
+            LOG_LOCATION_INFO                                                             \
+            message << (countOfWarnMessages > 1                                           \
+                            ? " | repeated [" + std::to_string(countOfWarnMessages) + "]" \
+                            : "");                                                        \
+        }                                                                                 \
+    }
 #ifdef LOG_ERROR
 #undef LOG_ERROR
 #endif
-#define LOG_ERROR(message)                                                                   \
-    BOOST_LOG_STREAM_SEV(ml::core::CLogger::instance().logger(), ml::core::CLogger::E_Error) \
-    LOG_LOCATION_INFO                                                                        \
-    message
+#define LOG_ERROR(message)                                                                 \
+    {                                                                                      \
+        std::size_t countOfErrorMessages;                                                  \
+        bool skipErrorMessage;                                                             \
+        std::tie(countOfErrorMessages, skipErrorMessage) =                                 \
+            ml::core::CLogger::instance().throttler().skip(__FILE__, __LINE__);            \
+        if (skipErrorMessage == false) {                                                   \
+            BOOST_LOG_STREAM_SEV(ml::core::CLogger::instance().logger(),                   \
+                                 ml::core::CLogger::E_Error)                               \
+            LOG_LOCATION_INFO                                                              \
+            message << (countOfErrorMessages > 1                                           \
+                            ? " | repeated [" + std::to_string(countOfErrorMessages) + "]" \
+                            : "");                                                         \
+        }                                                                                  \
+    }
 #ifdef LOG_FATAL
 #undef LOG_FATAL
 #endif
 #define LOG_FATAL(message)                                                                   \
     BOOST_LOG_STREAM_SEV(ml::core::CLogger::instance().logger(), ml::core::CLogger::E_Fatal) \
     LOG_LOCATION_INFO                                                                        \
     message
-
 #ifdef HANDLE_FATAL
 #undef HANDLE_FATAL
 #endif

lib/core/CLogger.cc

@@ -177,6 +177,10 @@ void CLogger::reset() {
     m_Reconfigured.store(false);
 }
 
+CLoggerThrottler& CLogger::throttler() {
+    return m_Throttler;
+}
+
 CLogger& CLogger::instance() {
     static CLogger instance;
     return instance;

lib/core/CLoggerThrottler.cc

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+#include <core/CLoggerThrottler.h>
+
+#include <core/CTimeUtils.h>
+
+#include <limits>
+
+namespace ml {
+namespace core {
+
+CLoggerThrottler::CLoggerThrottler() : m_MinimumLogIntervalMs{3600 * 1000} {
+}
+
+void CLoggerThrottler::minimumLogIntervalMs(std::int64_t minimumLogIntervalMs) {
+    m_MinimumLogIntervalMs = minimumLogIntervalMs;
+}
+
+std::pair<std::size_t, bool> CLoggerThrottler::skip(const char* file, int line) {
+
+    auto key = std::make_pair(file, line);
+    auto now = CTimeUtils::nowMs();
+    auto value = std::make_pair(std::numeric_limits<std::int64_t>::min(), 0);
+
+    // We have to hold the lock while updating the map entry because the same log
+    // line may be triggered from different threads and we don't try and update
+    // the last time and count atomically.
+
+    std::unique_lock<std::mutex> lock{m_Mutex};
+    auto& valueRef = m_LastLogTimesAndCounts.emplace(key, value).first->second;
+    std::size_t count{valueRef.second + 1};
+
+    if (now >= valueRef.first + m_MinimumLogIntervalMs) {
+        valueRef = std::make_pair(now, 0);
+        return std::make_pair(count, false);
+    }
+    ++valueRef.second;
+    return std::make_pair(count, true);
+}
+}
+}

lib/core/Makefile

@@ -87,6 +87,7 @@ CJsonOutputStreamWrapper.cc \
 CJsonStatePersistInserter.cc \
 CJsonStateRestoreTraverser.cc \
 CLogger.cc \
+CLoggerThrottler.cc \
 CLoopProgress.cc \
 CMemory.cc \
 CMemoryUsage.cc \