fix(broker/tests): robot tests should be ok now

* fix(tests): fixes in various tests
* fix(tests): scheduler test is split into 2 tests, retry check and then normal check
* fix(tests): external-commands2 contained an issue in its last check
* fix(tests): Tests on forced checks improved
* fix(tests): coding style better followed
* wip(ci): attempt to get information about arm cpu
* wip(ci): attempt to reduce the artifact
* fix(broker/feeder): feeder can be not stopped when switched off
* enh(tests): unstable feature added
* fix(cmake): issue with python 3.x, x < 11 fixed
* fix(tests): no more develop branch in benches
* fix(tests/reverse_bam): the process should be better killed
* fix(broker): engine is directly accessible from muxers
* fix(broker): mysql_connection finish() may finish too late
* fix(tests): reverse-connection, db cleanup added
* fix(ci): DEBUG_ROBOT set to OFF
* enh(broker): shared_from_this no more used
* enh(tests): new keyword applied everywhere
* fix(tests): an issue when the tuple is empty fixed
* fix(broker): bad initialization

REFS: MON-22554

.github/scripts/collect-test-robot.sh

@@ -11,6 +11,11 @@ database_type=$2
 distrib=${ID}
 distrib=$(echo $distrib | tr '[:lower:]' '[:upper:]')
 
+#cpu=$(lscpu | awk '$1 ~ "Architecture" { print $2 }')
+if [[ "$test_file" =~ "unstable" ]] ; then
+  exit 0
+fi
+
 if [ ${database_type} == 'mysql' ] && [ ! -f tests/${test_file}.mysql ]; then
     echo > tests/log.html
     echo '<?xml version="1.0" encoding="UTF-8"?>' > tests/output.xml
@@ -106,4 +111,4 @@ cd tests
 ./init-proto.sh
 
 echo "####################### Run Centreon Collect Robot Tests #######################"
-robot $test_file
+robot -e unstable $test_file

.github/workflows/package-collect.yml

@@ -88,6 +88,7 @@ jobs:
 
           sudo $CMAKE \
                   -G "Ninja" \
+                  -DDEBUG_ROBOT=OFF \
                   -DWITH_TESTING=OFF \
                   -DWITH_BENCH=ON \
                   -DWITH_MODULE_SIMU=OFF \

.github/workflows/robot-nightly.yml

@@ -19,10 +19,10 @@ jobs:
         uses: actions/checkout@v4
 
       - run: |
+          gh workflow run robot-nightly.yml -r "dev-23.10.x"
           gh workflow run robot-nightly.yml -r "dev-23.04.x"
           gh workflow run robot-nightly.yml -r "dev-22.10.x"
           gh workflow run robot-nightly.yml -r "dev-22.04.x"
-          gh workflow run robot-nightly.yml -r "dev-21.10.x"
         shell: bash
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/robot-test.yml

@@ -88,19 +88,18 @@ jobs:
           AWS_BUCKET: centreon-collect-robot-report
 
       - name: Move reports
-        if: ${{ !cancelled() }}
+        if: ${{ failure() }}
         run: |
           mkdir reports
           FILE_PREFIX=`echo "${{ matrix.feature }}" | sed -e "s#/#__#g"`-${{inputs.database_type}}
-          mv tests/log.html reports/$FILE_PREFIX-log.html
-          mv tests/output.xml reports/$FILE_PREFIX-output.xml
-          mv tests/report.html reports/$FILE_PREFIX-report.html
           if [ -d tests/failed ] ; then
+            mv tests/log.html reports/$FILE_PREFIX-log.html
+            mv tests/output.xml reports/$FILE_PREFIX-output.xml
             mv tests/failed reports/$FILE_PREFIX-failed
           fi
 
       - name: Upload Test Results
-        if: ${{ !cancelled() }}
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v3
         with:
           name: robot-test-reports-${{inputs.database_type}}-${{ inputs.distrib }}-${{ inputs.arch }}
@@ -109,7 +108,7 @@ jobs:
 
   robot-test-report:
     needs: [robot-test]
-    if: ${{ !cancelled() }}
+    if: ${{ failure() }}
     runs-on: ubuntu-22.04
 
     steps:

CMakeLists.txt

@@ -42,6 +42,7 @@ endif()
 add_definitions("-D_GLIBCXX_USE_CXX11_ABI=1")
 
 option(NG "C++17 build." OFF)
+option(DEBUG_ROBOT OFF)
 
 if(NG)
   set(CMAKE_CXX_STANDARD 17)
@@ -128,6 +129,10 @@ set(COLLECT_PATCH 0)
 set(COLLECT_VERSION "${COLLECT_MAJOR}.${COLLECT_MINOR}.${COLLECT_PATCH}")
 add_definitions(-DCENTREON_CONNECTOR_VERSION=\"${COLLECT_VERSION}\")
 
+if (DEBUG_ROBOT)
+  add_definitions(-DDEBUG_ROBOT)
+endif()
+
 # ########### CONSTANTS ###########
 set(USER_BROKER centreon-broker)
 set(USER_ENGINE centreon-engine)

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

@@ -59,7 +59,7 @@ class computable {
    * @brief This method is used by the dump() method. It gives a summary of this
    * computable main informations.
    *
-   * @return A multiline strings with various informations.
+   * @return A multiline string with various informations.
    */
   virtual std::string object_info() const = 0;
   /**

broker/bam/src/bool_constant.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016, 2023 Centreon
+ * Copyright 2014, 2023 Centreon
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

broker/bam/src/bool_operation.cc

@@ -110,14 +110,19 @@ std::string bool_operation::object_info() const {
   switch (_type) {
     case addition:
       op = "PLUS";
+      break;
     case substraction:
       op = "MINUS";
+      break;
     case multiplication:
       op = "MUL";
+      break;
     case division:
       op = "DIV";
+      break;
     case modulo:
       op = "MODULO";
+      break;
     default:
       return "unknown operation";
   }

broker/bam/src/connector.cc

@@ -47,7 +47,8 @@ static constexpr multiplexing::muxer_filter _monitoring_stream_filter = {
     neb::downtime::static_type(),        neb::pb_downtime::static_type(),
     bam::ba_status::static_type(),       bam::pb_ba_status::static_type(),
     bam::kpi_status::static_type(),      bam::pb_kpi_status::static_type(),
-    inherited_downtime::static_type(),   pb_inherited_downtime::static_type()};
+    inherited_downtime::static_type(),   pb_inherited_downtime::static_type(),
+    extcmd::pb_ba_info::static_type()};
 
 static constexpr multiplexing::muxer_filter _reporting_stream_filter = {
     bam::kpi_event::static_type(),

broker/bam/src/monitoring_stream.cc

@@ -581,6 +581,7 @@ int monitoring_stream::write(std::shared_ptr<io::data> const& data) {
       _write_external_command(cmd);
     } break;
     case extcmd::pb_ba_info::static_type(): {
+      log_v2::bam()->info("BAM: dump BA");
       extcmd::pb_ba_info const& e =
           *std::static_pointer_cast<const extcmd::pb_ba_info>(data);
       auto& obj = e.obj();

broker/core/inc/com/centreon/broker/cache/global_cache.hh

@@ -150,7 +150,7 @@ class global_cache : public std::enable_shared_from_this<global_cache> {
 
   void allocation_exception_handler();
 
-  virtual void managed_map(bool create){};
+  virtual void managed_map(bool create) {}
 
  public:
   using pointer = std::shared_ptr<global_cache>;

broker/core/inc/com/centreon/broker/misc/misc.hh

@@ -36,8 +36,16 @@ std::string dump_filters(const multiplexing::muxer_filter& filters);
 std::list<perfdata> parse_perfdata(uint32_t host_id,
                                    uint32_t service_id,
                                    const char* str);
+#if DEBUG_ROBOT
+void debug(const std::string& content);
+#endif
 }  // namespace misc
 
+#if DEBUG_ROBOT
+#define DEBUG(content) misc::debug(content)
+#else
+#define DEBUG(content)
+#endif
 CCB_END()
 
 #endif  // !CCB_MISC_MISC_HH

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

@@ -58,6 +58,7 @@ class feeder : public stat_visitable,
 
  protected:
   feeder(const std::string& name,
+         const std::shared_ptr<multiplexing::engine>& parent,
          std::shared_ptr<io::stream>& client,
          const multiplexing::muxer_filter& read_filters,
          const multiplexing::muxer_filter& write_filters);
@@ -84,6 +85,7 @@ class feeder : public stat_visitable,
  public:
   static std::shared_ptr<feeder> create(
       const std::string& name,
+      const std::shared_ptr<multiplexing::engine>& parent,
       std::shared_ptr<io::stream>& client,
       const multiplexing::muxer_filter& read_filters,
       const multiplexing::muxer_filter& write_filters);

broker/core/multiplexing/inc/com/centreon/broker/multiplexing/engine.hh

@@ -1,20 +1,20 @@
 /*
-** Copyright 2009-2012,2015,2019-2021 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 the License at
-**
-**     http://www.apache.org/licenses/LICENSE-2.0
-**
-** Unless required by applicable law or agreed to in writing, software
-** distributed under the License is distributed on an "AS IS" BASIS,
-** WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-** See the License for the specific language governing permissions and
-** limitations under the License.
-**
-** For more information : contact@centreon.com
-*/
+ * Copyright 2009-2012,2015,2019-2021 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 the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * For more information : contact@centreon.com
+ */
 
 #ifndef CCB_MULTIPLEXING_ENGINE_HH
 #define CCB_MULTIPLEXING_ENGINE_HH
@@ -63,7 +63,7 @@ class callback_caller;
  *
  *  @see muxer
  */
-class engine : public std::enable_shared_from_this<engine> {
+class engine {
   static std::mutex _load_m;
   static std::shared_ptr<engine> _instance;
 
@@ -82,8 +82,7 @@ class engine : public std::enable_shared_from_this<engine> {
   std::deque<std::shared_ptr<io::data>> _kiew;
 
   // Subscriber.
-  std::vector<std::shared_ptr<muxer>> _muxers;
-  std::mutex _muxers_m;
+  std::vector<std::weak_ptr<muxer>> _muxers;
 
   // Statistics.
   EngineStats* _stats;
@@ -112,7 +111,7 @@ class engine : public std::enable_shared_from_this<engine> {
   void start();
   void stop();
   void subscribe(const std::shared_ptr<muxer>& subscriber);
-  void unsubscribe(const muxer* subscriber);
+  void unsubscribe_muxer(const muxer* subscriber);
 };
 }  // namespace multiplexing
 

broker/core/multiplexing/inc/com/centreon/broker/multiplexing/muxer.hh

@@ -57,6 +57,7 @@ class muxer : public io::stream {
   static uint32_t _event_queue_max_size;
 
   const std::string _name;
+  std::shared_ptr<engine> _engine;
   const std::string _queue_file_name;
   multiplexing::muxer_filter _read_filter;
   multiplexing::muxer_filter _write_filter;
@@ -83,6 +84,7 @@ class muxer : public io::stream {
   void _update_stats(void) noexcept;
 
   muxer(std::string name,
+        const std::shared_ptr<engine>& parent,
         const muxer_filter& r_filter,
         const muxer_filter& w_filter,
         bool persistent = false);
@@ -94,6 +96,7 @@ class muxer : public io::stream {
   static uint32_t event_queue_max_size() noexcept;
 
   static std::shared_ptr<muxer> create(std::string name,
+                                       const std::shared_ptr<engine>& parent,
                                        const muxer_filter& r_filter,
                                        const muxer_filter& w_filter,
                                        bool persistent = false);
@@ -121,6 +124,7 @@ class muxer : public io::stream {
   void set_read_filter(const muxer_filter& w_filter);
   void set_write_filter(const muxer_filter& w_filter);
   void clear_read_handler();
+  void unsubscribe();
 };
 
 /**

broker/core/multiplexing/src/engine.cc

@@ -25,6 +25,7 @@
 #include "com/centreon/broker/config/applier/state.hh"
 #include "com/centreon/broker/io/events.hh"
 #include "com/centreon/broker/log_v2.hh"
+#include "com/centreon/broker/misc/misc.hh"
 #include "com/centreon/broker/multiplexing/muxer.hh"
 #include "com/centreon/broker/pool.hh"
 
@@ -236,6 +237,7 @@ void engine::stop() {
                                      EngineStats::STOPPED);
   }
   log_v2::core()->debug("multiplexing: engine stopped");
+  DEBUG(fmt::format("STOP engine {:p}", static_cast<void*>(this)));
 }
 
 /**
@@ -248,7 +250,7 @@ void engine::subscribe(const std::shared_ptr<muxer>& subscriber) {
                           subscriber->name());
   std::lock_guard<std::mutex> l(_engine_m);
   for (auto& m : _muxers)
-    if (m == subscriber) {
+    if (m.lock() == subscriber) {
       log_v2::config()->debug("engine: muxer {} already subscribed",
                               subscriber->name());
       return;
@@ -261,10 +263,11 @@ void engine::subscribe(const std::shared_ptr<muxer>& subscriber) {
  *
  *  @param[in] subscriber  Subscriber.
  */
-void engine::unsubscribe(const muxer* subscriber) {
+void engine::unsubscribe_muxer(const muxer* subscriber) {
   std::lock_guard<std::mutex> l(_engine_m);
   for (auto it = _muxers.begin(); it != _muxers.end(); ++it) {
-    if (it->get() == subscriber) {
+    auto w = it->lock();
+    if (!w || w.get() == subscriber) {
       log_v2::config()->debug("engine: muxer {} unsubscribes to engine",
                               subscriber->name());
       _muxers.erase(it);
@@ -281,12 +284,16 @@ engine::engine()
       _stats{stats::center::instance().register_engine()},
       _unprocessed_events{0u},
       _sending_to_subscribers{false} {
+  DEBUG(fmt::format("CONSTRUCTOR engine {:p}", static_cast<void*>(this)));
   stats::center::instance().update(&EngineStats::set_mode, _stats,
                                    EngineStats::NOT_STARTED);
 }
 
 engine::~engine() noexcept {
+  /* Muxers should be unsubscribed before arriving here. */
+  assert(_muxers.empty());
   log_v2::core()->debug("core: cbd engine destroyed.");
+  DEBUG(fmt::format("DESTRUCTOR engine {:p}", static_cast<void*>(this)));
 }
 
 /**
@@ -380,8 +387,8 @@ bool engine::_send_to_subscribers(send_to_mux_callback_type&& callback) {
     // it will be destroyed at the end of the scope of this function and at the
     // end of lambdas posted
     cb = std::make_shared<detail::callback_caller>(std::move(callback),
-                                                   shared_from_this());
-    last_muxer = *_muxers.rbegin();
+                                                   _instance);
+    last_muxer = _muxers.rbegin()->lock();
     if (_muxers.size() > 1) {
       /* Since the sending is parallelized, we use the thread pool for this
        * purpose except for the last muxer where we use this thread. */
@@ -392,7 +399,7 @@ bool engine::_send_to_subscribers(send_to_mux_callback_type&& callback) {
       /* We use the thread pool for the muxers from the first one to the
        * second to last */
       for (auto it = _muxers.begin(); it != it_last; ++it) {
-        pool::io_context().post([kiew, m = *it, cb]() {
+        pool::io_context().post([kiew, m = it->lock(), cb]() {
           try {
             m->publish(*kiew);
           }  // pool threads protection