MON-193976 failover can read endpoint config and increase failed write retries delay (#3160) (#3194)

* splitter read does not loop and only splitter write error per second

Author: jean-christophe81

broker/core/inc/com/centreon/broker/file/splitter.hh

@@ -73,11 +73,19 @@ class splitter : public fs_file {
   std::atomic_int _wid;
   long _woffset;
 
+  /**
+   * @brief when disk is full, we will log all events write failures so we will
+   * log only once by second
+   *
+   */
+  time_t _last_write_error_log;
+
   void _open_read_file();
   bool _open_write_file();
 
  public:
-  splitter(const std::string& path, uint32_t max_file_size = 100000000u,
+  splitter(const std::string& path,
+           uint32_t max_file_size = 100000000u,
            bool auto_delete = false);
   ~splitter();
   splitter(const splitter&) = delete;

broker/core/inc/com/centreon/broker/processing/failover.hh

@@ -19,7 +19,7 @@
 #ifndef CCB_PROCESSING_FAILOVER_HH
 #define CCB_PROCESSING_FAILOVER_HH
 
-#include <climits>
+#include "com/centreon/broker/config/endpoint.hh"
 #include "com/centreon/broker/io/endpoint.hh"
 #include "com/centreon/broker/multiplexing/muxer.hh"
 
@@ -53,13 +53,14 @@ class failover : public endpoint {
   _running_state _state ABSL_GUARDED_BY(_state_m);
   mutable absl::Mutex _state_m;
   std::shared_ptr<spdlog::logger> _logger;
+  unsigned _max_retry_delay;
 
   void _run();
 
  public:
   failover(std::shared_ptr<io::endpoint> endp,
            std::shared_ptr<multiplexing::muxer> mux,
-           std::string const& name);
+           const config::endpoint& cfg);
   failover(failover const& other) = delete;
   failover& operator=(failover const& other) = delete;
   ~failover();

broker/core/src/config/applier/endpoint.cc

@@ -455,7 +455,7 @@ processing::failover* endpoint::_create_failover(
   }
 
   // Return failover thread.
-  auto fo{std::make_unique<processing::failover>(endp, mux, cfg.name)};
+  auto fo{std::make_unique<processing::failover>(endp, mux, cfg)};
   fo->set_buffering_timeout(cfg.buffering_timeout);
   fo->set_retry_interval(cfg.retry_interval);
   fo->set_failover(failovr);

broker/core/src/file/splitter.cc

@@ -57,7 +57,8 @@ splitter::splitter(std::string const& path,
       _roffset{0},
       _write_m{},
       _wfile{},
-      _woffset{0} {
+      _woffset{0},
+      _last_write_error_log(0) {
   // Get IDs of already existing file parts. File parts are suffixed
   // with their order number. A file named /var/lib/foo would have
   // parts named /var/lib/foo, /var/lib/foo1, /var/lib/foo2, ...
@@ -143,8 +144,23 @@ long splitter::read(void* buffer, long max_size) {
   /* No lock here, there is only one consumer. */
   if (!_rfile) {
     _open_read_file();
-    if (!_rfile)
+    if (!_rfile) {
+      // open read failure => next file
+      bool have_to_read_next_file = false;
+      {
+        std::lock_guard l(_write_m);
+        if (_rid < _wid) {
+          _rid++;
+          have_to_read_next_file = true;
+        } else {
+          throw exceptions::shutdown("No more data to read");
+        }
+      }
+      if (have_to_read_next_file) {
+        return read(buffer, max_size);
+      }
       return 0;
+    }
   }
 
   /* We introduce the locker, but don't lock if not necessary */
@@ -448,10 +464,14 @@ bool splitter::_open_write_file() {
     size_t size = disk_accessor::instance().fwrite(
         header.bytes, 1, sizeof(header), _wfile.get());
     if (size != sizeof(header)) {
-      std::string wfile(get_file_path(_wid));
-      char msg[1024];
-      logger->critical("splitter: cannot write to file '{}': {}", wfile,
-                       strerror_r(errno, msg, sizeof(msg)));
+      time_t now = time(nullptr);
+      if (now > _last_write_error_log) {
+        _last_write_error_log = now;
+        std::string wfile(get_file_path(_wid));
+        char msg[1024];
+        logger->critical("splitter: cannot write to file '{}': {}", wfile,
+                         strerror_r(errno, msg, sizeof(msg)));
+      }
       _wfile.reset();
       return false;
     }

broker/core/src/processing/failover.cc

@@ -39,11 +39,12 @@ using log_v2 = com::centreon::common::log_v2::log_v2;
  */
 failover::failover(std::shared_ptr<io::endpoint> endp,
                    std::shared_ptr<multiplexing::muxer> mux,
-                   const std::string& name)
-    : endpoint(false, name),
+                   const config::endpoint& cfg)
+    : endpoint(false, cfg.name),
       _should_exit(false),
       _state(not_started),
       _logger{log_v2::instance().get(log_v2::PROCESSING)},
+      _max_retry_delay(30),
       _buffering_timeout(0),
       _endpoint(endp),
       _failover_launched(false),
@@ -52,6 +53,14 @@ failover::failover(std::shared_ptr<io::endpoint> endp,
       _muxer(mux),
       _update(false) {
   SPDLOG_LOGGER_TRACE(_logger, "failover '{}' construction.", _name);
+
+  auto search = cfg.params.find("max_retry_delay");
+  if (search != cfg.params.end()) {
+    if (!absl::SimpleAtoi(search->second, &_max_retry_delay)) {
+      throw msg_fmt("max_retry_delay needs a numerical value and not {}",
+                    search->second);
+    }
+  }
 }
 
 /**
@@ -158,6 +167,27 @@ void failover::_run() {
     return;
   }
 
+  unsigned error_retry_delay = 0;
+
+  auto increase_retry_delay_and_wait = [&]() {
+    // in order to avoid a write infinite loop, we increase delay between two
+    // failures
+    if (!error_retry_delay) {
+      error_retry_delay = 1;
+    } else {
+      error_retry_delay *= 2;
+    }
+    if (error_retry_delay > _max_retry_delay) {
+      error_retry_delay = _max_retry_delay;
+    }
+    SPDLOG_LOGGER_ERROR(
+        _logger, " {} Failed to send event to stream, we wait {s} before retry",
+        _name, error_retry_delay);
+    for (ssize_t i = 0; !should_exit() && i < error_retry_delay * 10; i++) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+  };
+
   auto on_exception_handler = [&]() {
     if (_stream) {
       int32_t ack_events;
@@ -173,6 +203,7 @@ void failover::_run() {
     }
     set_state("connecting");
     if (!should_exit()) {
+      increase_retry_delay_and_wait();
       _launch_failover();
       _initialized = true;
     }
@@ -355,8 +386,16 @@ void failover::_run() {
             int we(0);
 
             try {
-              std::lock_guard<std::timed_mutex> stream_lock(_stream_m);
-              we = _stream->write(d);
+              {
+                std::lock_guard<std::timed_mutex> stream_lock(_stream_m);
+                we = _stream->write(d);
+              }
+              if (we < 0) {  // stream write failure
+                increase_retry_delay_and_wait();
+              } else {
+                // no exception, no error => reset error_retry_delay
+                error_retry_delay = 0;
+              }
             } catch (exceptions::shutdown const& e) {
               SPDLOG_LOGGER_DEBUG(
                   _logger,
@@ -625,7 +664,8 @@ bool failover::should_exit() const {
 }
 
 bool failover::wait_for_all_events_written(unsigned ms_timeout) {
-  _logger->info("processing::failover::wait_for_all_events_written");
+  SPDLOG_LOGGER_INFO(
+      _logger, "{} processing::failover::wait_for_all_events_written", _name);
   std::lock_guard<std::timed_mutex> stream_lock(_stream_m);
   if (_stream) {
     return _stream->wait_for_all_events_written(ms_timeout);

broker/lua/src/luabinding.cc

@@ -362,12 +362,12 @@ int luabinding::write(std::shared_ptr<io::data> const& data) noexcept {
       else
         SPDLOG_LOGGER_ERROR(
             _logger, "lua: unknown error while running function `filter()'");
-      RETURN_AND_POP(0);
+      RETURN_AND_POP(-1);
     }
 
     if (!lua_isboolean(_L, -1)) {
       SPDLOG_LOGGER_ERROR(_logger, "lua: `filter' must return a boolean");
-      RETURN_AND_POP(0);
+      RETURN_AND_POP(-1);
     }
 
     execute_write = lua_toboolean(_L, -1);
@@ -403,12 +403,14 @@ int luabinding::write(std::shared_ptr<io::data> const& data) noexcept {
     else
       SPDLOG_LOGGER_ERROR(_logger,
                           "lua: unknown error running function `write'");
-    RETURN_AND_POP(0);
+    RETURN_AND_POP(-1);
   }
 
   if (!lua_isboolean(_L, -1)) {
-    SPDLOG_LOGGER_ERROR(_logger, "lua: `write' must return a boolean");
-    RETURN_AND_POP(0);
+    // despite the fact that it's an error, we log it as an info because some SC
+    // use it to delay next event write
+    SPDLOG_LOGGER_INFO(_logger, "lua: `write' must return a boolean");
+    RETURN_AND_POP(-1);
   }
   int acknowledge = lua_toboolean(_L, -1);
 

broker/lua/test/lua.cc

@@ -4871,7 +4871,7 @@ TEST_F(LuaTest, BadLua) {
   s->service_id = 18;
   s->output = "Bonjour";
   std::shared_ptr<io::data> svc(s.release());
-  ASSERT_EQ(binding->write(svc), 0);
+  ASSERT_EQ(binding->write(svc), -1);
   RemoveFile(filename);
 }
 

tests/broker-engine/lua-cache.robot tests/broker-engine/lua.robottests/broker-engine/lua-cache.robot renamed to tests/broker-engine/lua.robot

@@ -840,3 +840,82 @@ LUA_CACHE_SAVE_SERVICE_GROUP
 
 
 
+LUA_FAILURE
+    [Documentation]    Given a engine broker configured with a SC, we check write intervals in case of lua write returns nothing
+    [Tags]    broker    engine    services    lua    MON-193976
+    Ctn Clear Commands Status
+    Ctn Clear Retention
+
+    Remove File    /tmp/test-LUA.log
+    Ctn Config Engine    ${1}    ${5}    ${20}
+    Ctn Config Broker    central
+    Ctn Config Broker    module
+    Ctn Config Broker    rrd
+    Ctn Broker Config Log    central    neb    trace
+    Ctn Broker Config Log    central    sql    error
+    Ctn Broker Config Log    central    lua    debug
+    Ctn Broker Config Log    central    processing    debug
+    Ctn Broker Config Source Log    central    1
+    Ctn Config Broker Sql Output    central    unified_sql
+
+    Remove File    /tmp/test-LUA.log
+
+    ${new_content}    Catenate    SEPARATOR=\n
+    ...    function init(params)
+    ...        broker_log:set_parameters(2, '/tmp/test-LUA.log')
+    ...        broker_log:info(0, 'lua start test')
+    ...    end
+    ...
+    ...    function write(e)
+    ...        local file = io.open("/tmp/test_lua_failure.txt", "r")
+    ...        local now = os.time()
+    ...        if file ~=  nill then
+    ...            broker_log:info(0, now .. ' lua write success')
+    ...            file:close()
+    ...            return true
+    ...        else
+    ...            broker_log:info(0, now .. ' lua write error')
+    ...            return
+    ...        end
+    ...    end
+
+    # Create the initial LUA script file
+    Create File    /tmp/test-LUA.lua    ${new_content}
+    Create File    /tmp/test_lua_failure.txt    aa
+
+    Ctn Broker Config Add Lua Output    central    test-LUA    /tmp/test-LUA.lua
+    Ctn Broker Config Output Set    central    test-LUA    max_retry_delay    16
+
+    ${start}    Get Current Date
+    Ctn Start Broker
+    Ctn Start Engine
+    Ctn Wait For Engine To Be Ready    ${start}    ${1}
+
+    FOR    ${index}    IN RANGE    6
+        ${result}    Grep File    /tmp/test-LUA.log    lua write success
+        IF    len("""${result}""") > 0    BREAK
+        Sleep    5s
+    END
+
+    Should Not Be Empty    ${result}    no write succes found in logs
+
+    Append To File    /tmp/test-LUA.log    now failure\n
+    #now, write will fail
+    Remove File    /tmp/test_lua_failure.txt
+
+    #generate many events
+    Ctn Restart Engine
+
+    FOR    ${index}    IN RANGE    90
+        ${lua_output}    Get File    /tmp/test-LUA.log
+        ${fail_lines}    Get Regexp Matches    ${lua_output}   ([0-9]+) lua write error
+        ${fail_count}    Get Length    ${fail_lines}
+
+        IF    ${fail_count} >= 7    BREAK
+        Sleep    1s
+    END
+
+    Should Be True    ${fail_count} >= 7    no enougth write error found in logs
+
+    ${res}    Ctn Broker Check Failover Lua Retry    ${fail_lines}    16
+    Should Be True    ${res}    delays beetwen lua retry records are not as expected

tests/resources/Broker.py

@@ -3287,3 +3287,34 @@ def ctn_check_acknowledgement_in_logs_table(date: int, timeout: int = TIMEOUT):
                     return True
         time.sleep(2)
     return False
+
+
+def ctn_broker_check_failover_lua_retry(lua_log_lines, max_retry_delay: int):
+    """
+    Check lines as 1770991869 lua write error
+    We check that first field (timestamp) obey to the failover increase interval law
+
+    :param lua_log_lines: lines given by grep
+    :param max_retry_delay: max_retry_delay configured in broker output
+    """
+    last_timestamp = 0
+    last_interval = 1
+
+    timestamp_search = re.compile(r"(\d+).*")
+    for line in lua_log_lines:
+        extract = timestamp_search.search(line)
+        if extract is not None:
+            new_ts = int(extract.group(1))
+        if last_timestamp == 0:
+            last_timestamp = new_ts
+        else:
+            if new_ts != last_timestamp + last_interval:
+                logger.console.log(
+                    f"expected interval: {last_interval}, but interval found: {new_ts} - {last_timestamp} = {new_ts - last_timestamp}")
+                return False
+            else:
+                last_timestamp = new_ts
+                last_interval = last_interval * 2
+                if last_interval > max_retry_delay:
+                    last_interval = max_retry_delay
+    return True