fix(collect): develop backport of 22.04 patches

* Mon 16333 add periodic log flush  develop (#610)
* Mon 17390 scheduler test (#609)
* fix(engine/config): tag and severity had several issues
* enh(conan): grpc upgraded
* fix(cmake): we force conan version to avoid issues with version 2
* MON-17390 fix scheduler and connector_ssh tests
* MON-17390 fix perl-connectors test
* MON-17390 fix engine/stat-stop
* MON-17390 engine dont core if cbmod fail start, add sql traces
* MON-17390 add periodical commit
* MON-17390 review remarks
* MON-17390 less lock
* MON-16333 develop backport
* MON-16333 backport periodic flush
* MON-16333 an asio timer flushes logs
* MON-16333 tcp_async is now a sahred_from_this
* MON-16333 add somes modifications, a logger non configured can be activated with ccc
* MON-16333 develop add SetLogLevel
* MON-16333 issues after rebase
* MON-16333-add-SetLogFlushPeriod
* MON-16333-force push
* MON-16333 review remarks
* fix(broker/mysql): new tests added and a little fix in mysql on stats (#616)
* fix(broker/unified_sql): A bad query is fixed here (#617)
* fix(broker): compilation warnings and errors fixed for some compilers
* MON-17457-improve-test-coverage-develop (#619)
* MON-17458-develop-improve-monitoring-stream-ack (#620)
* enh(collect): log_v2 is removed from main static libraries
* fix(collect/log_v2): uniformization of log_v2 between engine and broker
* fix(broker): unit tests on rrd cached + deadlock in tcp_async
* MON-17461 develop update doc (#622)
* fix(broker/mysql): Bind for results was missing
* fix(broker/sql): issue in bulkd_queries fixed and new test
* cleanup(broker/mysql): unused functions removed
* fix(broker): little fixes due cherry-picks
* MON-17465 fix uncached exceptions (#624)
* fix(broker/splitter): splitter better protected during read/write
* MON-16986 splitter open files in binary mode and write bug fix
* wip(broker/splitter)
* fix(broker/splitter): splitter improved with new tests
* cleanup(broker/splitter): comments added.
* cleanup(broker/compression)
* fix(broker/tests): bad comment fixed
* enh(broker/compression): optimization
* fix(broker/mysql): An assert was there by mistake
* fix(broker/splitter): fwrite error management better
* cleanup(broker/splitter): log updated
* fix(broker/splitter): log fixed
* fix(broker/splitter): no exception when limit size reached
* fix(broker/unified_sql): bulk_bind row filling must be protected (#615)
* wip(broker/splitter)
* fix(broker/splitter): splitter improved with new tests
* cleanup(broker/config): logs uniformized
* fix(broker/muxer): reuse muxer in case of reload
* cleanup(broker/muxer): comments fixed
* cleanup(broker/muxer): comments updated
* enh(broker/engine): little optimization

REFS: MON-17453 MON-17454 MON-17455 MON-17456 MON-17457 MON-17458 MON-17459 MON-17460 MON-17461 MON-17462 MON-17463 MON-17464 MON-17465 MON-17466 MON-17467 MON-17468 MON-17897

Co-authored-by: Jean Christophe Roques <jcroques@centreon.com>
Co-authored-by: David Boucher <dboucher@centreon.com>

CMakeLists.txt

@@ -1,5 +1,5 @@
 #
-# Copyright 2009-2021 Centreon
+# Copyright 2009-2023 Centreon
 #
 # Licensed under the Apache License, Version 2.0 (the "License"); you may not
 # use this file except in compliance with the License. You may obtain a copy of
@@ -26,15 +26,21 @@ set(CMAKE_COLOR_DIAGNOSTICS ON)
 cmake_minimum_required(VERSION 3.16)
 project("Centreon Collect" C CXX)
 
+option(WITH_ASAN
+       "Add the libasan to check memory leaks and other memory issues." OFF)
+
+option(WITH_TSAN
+       "Add the libtsan to check threads and other multithreading issues." OFF)
 if(NOT CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND NOT CMAKE_CXX_COMPILER_ID
-  STREQUAL "Clang")
+                                                STREQUAL "Clang")
   message(
     FATAL_ERROR "You can build broker with g++ or clang++. CMake will exit.")
 endif()
 
 # set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++14 -stdlib=libc++")
 # set(CMAKE_CXX_COMPILER "clang++")
 add_definitions("-D_GLIBCXX_USE_CXX11_ABI=1")
+
 option(NG "C++17 build." OFF)
 
 if(NG)
@@ -46,6 +52,28 @@ endif()
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 
+if(WITH_TSAN)
+  set(CMAKE_CXX_FLAGS_DEBUG
+      "${CMAKE_CXX_FLAGS_DEBUG} -fno-omit-frame-pointer -fsanitize=thread")
+  set(CMAKE_LINKER_FLAGS_DEBUG
+      "${CMAKE_LINKER_FLAGS_DEBUG} -fno-omit-frame-pointer -fsanitize=thread")
+endif()
+
+if(WITH_ASAN)
+  set(CMAKE_BUILD_TYPE Debug)
+  if(WITH_CLANG)
+    set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -fsanitize=address")
+    set(CMAKE_LINKER_FLAGS_DEBUG
+        "${CMAKE_LINKER_FLAGS_DEBUG} -fsanitize=address")
+  else()
+    set(CMAKE_CXX_FLAGS_DEBUG
+        "${CMAKE_CXX_FLAGS_DEBUG} -fno-omit-frame-pointer -fsanitize=address")
+    set(CMAKE_LINKER_FLAGS_DEBUG
+        "${CMAKE_LINKER_FLAGS_DEBUG} -fno-omit-frame-pointer -fsanitize=address"
+    )
+  endif()
+endif()
+
 set(ALLOW_DUPLICATE_EXECUTABLE TRUE)
 
 set(BUILD_ARGS "-w" "dupbuild=warn")
@@ -70,11 +98,11 @@ if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
   string(TOLOWER "${id}" id)
 
   if(("${id}" MATCHES "debian")
-    OR("${like}" MATCHES "debian")
-    OR("${id}" MATCHES "ubuntu")
-    OR("${like}" MATCHES "ubuntu"))
+     OR ("${like}" MATCHES "debian")
+     OR ("${id}" MATCHES "ubuntu")
+     OR ("${like}" MATCHES "ubuntu"))
     set(OS_DISTRIBUTOR "Debian")
-  elseif(("${id}" MATCHES "centos") OR("${like}" MATCHES "centos"))
+  elseif(("${id}" MATCHES "centos") OR ("${like}" MATCHES "centos"))
     set(OS_DISTRIBUTOR "CentOS")
   else()
     message(WARNING "lsb_release in not installed")
@@ -116,15 +144,15 @@ include(GNUInstallDirs)
 
 # var directories.
 set(BROKER_VAR_LOG_DIR
-  "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-broker")
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-broker")
 set(BROKER_VAR_LIB_DIR
-  "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-broker")
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-broker")
 set(ENGINE_VAR_LOG_DIR
-  "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine")
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine")
 set(ENGINE_VAR_LOG_ARCHIVE_DIR
-  "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine/archives")
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine/archives")
 set(ENGINE_VAR_LIB_DIR
-  "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-engine")
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-engine")
 
 set(CMAKE_INSTALL_PREFIX "/usr")
 option(WITH_TESTING "Build unit tests." OFF)
@@ -145,7 +173,7 @@ set(protobuf_MODULE_COMPATIBLE True)
 add_definitions(${CONAN_DEFINES_SPDLOG})
 
 include_directories(${CMAKE_SOURCE_DIR} ${CONAN_INCLUDE_DIRS}
-  ${CMAKE_SOURCE_DIR}/clib/inc)
+                    ${CMAKE_SOURCE_DIR}/clib/inc)
 
 add_subdirectory(bbdo)
 add_subdirectory(broker)
@@ -162,4 +190,4 @@ add_custom_target(test-connector COMMAND tests/ut_connector)
 add_custom_target(test DEPENDS test-broker test-engine test-clib test-connector)
 
 add_custom_target(test-coverage DEPENDS broker-test-coverage
-  engine-test-coverage clib-test-coverage)
+                                        engine-test-coverage clib-test-coverage)

README.md

@@ -192,6 +192,33 @@ test/ut
 
 You're done!
 
+### inside a docker, ubuntu for example
+In my case I have the home directory of the centreon project in /data/dev/centreon-collect
+
+First create the ubuntu container and jump into
+```shell
+docker container run --name ubuntu22.04 -ti -v /data/dev/centreon-collect:/root/centreon-collect  ubuntu:22.04 /bin/bash
+```
+Now you are in your ubuntu container
+```shell
+cd /root/centreon-collect/
+apt update
+apt install python3
+apt install python3-pip
+pip3 install conan
+apt install libgnutls28-dev
+apt install liblua5.4-dev
+apt install librrd-dev
+apt install cmake
+
+./cmake.sh 
+cd build
+make
+```
+Then you have all binaries compiled in the ubuntu distribution.
+You can access it outside the container in the /data/dev/centreon-collect/build directory
+
+
 ## Bug reports / Feature requests
 
 The best way to report a bug or to request a feature is to open an issue

broker/CMakeLists.txt

@@ -343,7 +343,6 @@ set(LIBROKER_SOURCES
     ${SRC_DIR}/io/protocols.cc
     ${SRC_DIR}/io/raw.cc
     ${SRC_DIR}/io/stream.cc
-    ${SRC_DIR}/log_v2.cc
     ${SRC_DIR}/mapping/entry.cc
     ${SRC_DIR}/misc/diagnostic.cc
     ${SRC_DIR}/misc/filesystem.cc
@@ -363,6 +362,7 @@ set(LIBROKER_SOURCES
     ${SRC_DIR}/pool.cc
     ${SRC_DIR}/database/mysql_bind_base.cc
     ${SRC_DIR}/database/mysql_bind.cc
+    ${SRC_DIR}/database/mysql_bind_result.cc
     ${SRC_DIR}/database/mysql_bulk_bind.cc
     ${SRC_DIR}/database/mysql_column.cc
     ${SRC_DIR}/mysql_manager.cc
@@ -447,8 +447,6 @@ set(LIBROKER_SOURCES
     ${INC_DIR}/multiplexing/publisher.hh
     ${INC_DIR}/mysql.hh
     ${INC_DIR}/pool.hh
-    ${INC_DIR}/database/mysql_bind_base.hh
-    ${INC_DIR}/database/mysql_bind.hh
     ${INC_DIR}/database/mysql_bulk_bind.hh
     ${INC_DIR}/database/mysql_column.hh
     ${INC_DIR}/database/mysql_error.hh
@@ -480,6 +478,11 @@ set(LIBROKER_SOURCES
     ${INC_DIR}/version.hh)
 
 # Static libraries.
+add_library(rokerlog STATIC ${SRC_DIR}/log_v2.cc)
+set_target_properties(rokerlog PROPERTIES COMPILE_FLAGS "-fPIC")
+target_precompile_headers(rokerlog PRIVATE core/precomp_inc/precomp.hpp)
+target_link_libraries(rokerlog CONAN_PKG::spdlog)
+
 add_library(rokerbase STATIC ${LIBROKER_SOURCES})
 set_target_properties(rokerbase PROPERTIES COMPILE_FLAGS "-fPIC")
 target_precompile_headers(rokerbase PRIVATE core/precomp_inc/precomp.hpp)
@@ -506,7 +509,8 @@ add_executable(cbd ${SRC_DIR}/main.cc)
 # Flags needed to include all symbols in binary.
 target_link_libraries(
   cbd
-  "-export-dynamic"
+  "-rdynamic"
+  rokerlog
   "-Wl,--whole-archive"
   rokerbase
   roker

broker/bam/inc/com/centreon/broker/bam/availability_thread.hh

@@ -39,7 +39,7 @@ namespace bam {
  *  @brief Availability thread
  *
  */
-class availability_thread {
+class availability_thread final {
  public:
   availability_thread(database_config const& db_cfg,
                       timeperiod_map& shared_map);

broker/bam/inc/com/centreon/broker/bam/monitoring_stream.hh

@@ -73,6 +73,7 @@ class monitoring_stream : public io::stream {
   database::mysql_stmt _ba_update;
   database::mysql_stmt _kpi_update;
   int32_t _pending_events;
+  unsigned _pending_request;
   database_config _storage_db_cfg;
   std::shared_ptr<persistent_cache> _cache;
 
@@ -85,7 +86,6 @@ class monitoring_stream : public io::stream {
                      absl::Hash<std::pair<std::string, std::string>>>
       _timer_forced_svc_checks;
 
-  void _check_replication();
   void _prepare();
   void _rebuild();
   void _write_external_command(const std::string& cmd);

broker/bam/src/monitoring_stream.cc

@@ -64,6 +64,7 @@ monitoring_stream::monitoring_stream(const std::string& ext_cmd_file,
       _ext_cmd_file(ext_cmd_file),
       _mysql(db_cfg),
       _pending_events(0),
+      _pending_request(0),
       _storage_db_cfg(storage_db_cfg),
       _cache(std::move(cache)),
       _forced_svc_checks_timer{pool::io_context()} {
@@ -99,6 +100,7 @@ monitoring_stream::~monitoring_stream() {
  */
 int32_t monitoring_stream::flush() {
   _mysql.commit();
+  _pending_request = 0;
   int retval = _pending_events;
   SPDLOG_LOGGER_TRACE(log_v2::bam(), "BAM: monitoring_stream flush: {} events",
                       retval);
@@ -195,8 +197,21 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
   SPDLOG_LOGGER_TRACE(log_v2::bam(), "BAM: monitoring_stream write {}", *data);
   // Take this event into account.
   ++_pending_events;
-  if (!validate(data, get_name()))
-    return 0;
+
+  // this lambda ask mysql to do a commit if we have get_queries_per_transaction
+  // waiting requests
+  auto commit_if_needed = [this]() {
+    if (_mysql.get_config().get_queries_per_transaction() > 0) {
+      if (_pending_request >=
+          _mysql.get_config().get_queries_per_transaction()) {
+        _mysql.commit();
+        _pending_request = 0;
+      }
+    } else {  // auto commit => nothing to do
+      _pending_request = 0;
+    }
+  };
+
   SPDLOG_LOGGER_TRACE(log_v2::bam(), "BAM: {} pending events", _pending_events);
 
   // Process service status events.
@@ -313,6 +328,8 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
       _ba_update.bind_value_as_i32(5, status->state);
 
       _mysql.run_statement(_ba_update, database::mysql_error::update_ba, true);
+      ++_pending_request;
+      commit_if_needed();
 
       if (status->state_changed) {
         std::pair<std::string, std::string> ba_svc_name(
@@ -354,6 +371,8 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
       _ba_update.bind_value_as_i32(5, status.state());
 
       _mysql.run_statement(_ba_update, database::mysql_error::update_ba, true);
+      ++_pending_request;
+      commit_if_needed();
 
       if (status.state_changed()) {
         std::pair<std::string, std::string> ba_svc_name(
@@ -398,6 +417,9 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
 
       _mysql.run_statement(_kpi_update, database::mysql_error::update_kpi,
                            true);
+      ++_pending_request;
+      commit_if_needed();
+
     } break;
     case bam::pb_kpi_status::static_type(): {
       const KpiStatus& status(
@@ -425,6 +447,9 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
 
       _mysql.run_statement(_kpi_update, database::mysql_error::update_kpi,
                            true);
+      ++_pending_request;
+      commit_if_needed();
+
     } break;
     case inherited_downtime::static_type(): {
       std::string cmd;
@@ -480,8 +505,26 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
       break;
   }
 
-  // Event acknowledgement.
-  return 0;
+  // if uncommited request, we can't yet acknowledge
+  if (_pending_request) {
+    if (_pending_events >=
+        10 * _mysql.get_config().get_queries_per_transaction()) {
+      log_v2::bam()->trace(
+          "BAM: monitoring_stream write: too many pending events =>flush and "
+          "acknowledge {} events",
+          _pending_events);
+      return flush();
+    }
+    log_v2::bam()->trace(
+        "BAM: monitoring_stream write: 0 events (request pending) {} to "
+        "acknowledge",
+        _pending_events);
+    return 0;
+  }
+  int retval = _pending_events;
+  _pending_events = 0;
+  log_v2::bam()->trace("BAM: monitoring_stream write: {} events", retval);
+  return retval;
 }
 
 /**

broker/bam/test/ba/kpi_ba.cc

@@ -31,6 +31,8 @@
 
 using namespace com::centreon::broker;
 
+extern std::shared_ptr<asio::io_context> g_io_context;
+
 class KpiBA : public ::testing::Test {
  protected:
   std::unique_ptr<bam::configuration::applier::state> _aply_state;
@@ -40,6 +42,7 @@ class KpiBA : public ::testing::Test {
  public:
   void SetUp() override {
     // Initialization.
+    g_io_context->restart();
     config::applier::init(0, "test_broker", 0);
 
     _aply_state = std::make_unique<bam::configuration::applier::state>();

broker/bam/test/ba/kpi_service.cc

@@ -36,6 +36,8 @@
 
 using namespace com::centreon::broker;
 
+extern std::shared_ptr<asio::io_context> g_io_context;
+
 class BamBA : public ::testing::Test {
  protected:
   std::unique_ptr<bam::configuration::applier::state> _aply_state;
@@ -45,6 +47,7 @@ class BamBA : public ::testing::Test {
  public:
   void SetUp() override {
     // Initialization.
+    g_io_context->restart();
     config::applier::init(0, "test_broker", 0);
 
     _aply_state = std::make_unique<bam::configuration::applier::state>();

broker/bam/test/configuration/applier-boolexp.cc

@@ -24,9 +24,12 @@
 
 using namespace com::centreon::broker;
 
+extern std::shared_ptr<asio::io_context> g_io_context;
+
 class ApplierBoolexp : public ::testing::Test {
  public:
   void SetUp() override {
+    g_io_context->restart();
     // Initialization.
     config::applier::init(0, "test_broker", 0);