fix(broker/bam): fix inherited downtimes duplicated

* enh(broker/bam): log on cache more explicit
* fix(bam): engine restart does not duplicate ba downtimes
* fix(bam): downtime are well restored after a restart of cbd
* enh(bam): flat_hash_map is better for host/svc mapping
* cleanup(various): logs and coding style
* enh(bam): better iteration while erasing data
* cleanup(ceof): no more needed
* fix(applier::endpoint): protection of stop() call

REFS: MON-12068

Author: bouda1

CHANGELOG.md

@@ -35,6 +35,8 @@ the kpi were in critical, the state appeared as OK.
 The cache in bam is lighter, metrics are no
 more loaded. And queries to load the cache are parallelized.
 
+Inherited downtimes were duplicated on cbd reload. This is fixed.
+
 *mysql_connection*
 
 A timeout is added on mysql\_ping and this function is less called than before.

CMakeLists.txt

@@ -389,12 +389,6 @@ set(LIBROKER_SOURCES
   ${SRC_DIR}/bbdo/version_response.cc
   ${SRC_DIR}/broker_impl.cc
   ${SRC_DIR}/brokerrpc.cc
-  ${SRC_DIR}/ceof/ceof_deserializer.cc
-  ${SRC_DIR}/ceof/ceof_iterator.cc
-  ${SRC_DIR}/ceof/ceof_parser.cc
-  ${SRC_DIR}/ceof/ceof_serializer.cc
-  ${SRC_DIR}/ceof/ceof_token.cc
-  ${SRC_DIR}/ceof/ceof_writer.cc
   ${SRC_DIR}/compression/factory.cc
   ${SRC_DIR}/compression/opener.cc
   ${SRC_DIR}/compression/stack_array.cc
@@ -479,14 +473,6 @@ set(LIBROKER_SOURCES
   ${INC_DIR}/bbdo/version_response.hh
   ${INC_DIR}/broker_impl.hh
   ${INC_DIR}/brokerrpc.hh
-  ${INC_DIR}/ceof/ceof_deserializer.hh
-  ${INC_DIR}/ceof/ceof_iterator.hh
-  ${INC_DIR}/ceof/ceof_parser.hh
-  ${INC_DIR}/ceof/ceof_serializable.hh
-  ${INC_DIR}/ceof/ceof_serializer.hh
-  ${INC_DIR}/ceof/ceof_token.hh
-  ${INC_DIR}/ceof/ceof_visitor.hh
-  ${INC_DIR}/ceof/ceof_writer.hh
   ${INC_DIR}/compression/factory.hh
   ${INC_DIR}/compression/opener.hh
   ${INC_DIR}/compression/stack_array.hh

bam/CMakeLists.txt

@@ -162,6 +162,7 @@ add_library("${BAM}" SHARED
 )
 set_target_properties("${BAM}" PROPERTIES
     PREFIX "")
+target_link_libraries("${BAM}" ${absl_LIBS})
 
 # Testing.
 if (WITH_SQL_TESTS)

bam/inc/com/centreon/broker/bam/configuration/applier/ba.hh

@@ -67,7 +67,8 @@ class ba {
   std::shared_ptr<neb::host> _ba_host(uint32_t host_id);
   std::shared_ptr<neb::service> _ba_service(uint32_t ba_id,
                                             uint32_t host_id,
-                                            uint32_t service_id);
+                                            uint32_t service_id,
+                                            bool in_downtime = false);
   void _internal_copy(ba const& other);
   std::shared_ptr<bam::ba> _new_ba(configuration::ba const& cfg,
                                    service_book& book);

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

@@ -19,8 +19,7 @@
 #ifndef CCB_BAM_HST_SVC_MAPPING_HH
 #define CCB_BAM_HST_SVC_MAPPING_HH
 
-#include "com/centreon/broker/misc/pair.hh"
-#include <unordered_map>
+#include <absl/container/flat_hash_map.h>
 #include <string>
 #include <utility>
 
@@ -37,10 +36,11 @@ namespace bam {
  *  Allow to find an ID of a host or service by its name.
  */
 class hst_svc_mapping {
-  std::unordered_map<std::pair<std::string, std::string>, std::pair<uint32_t, uint32_t>>
+  absl::flat_hash_map<std::pair<std::string, std::string>,
+                      std::pair<uint32_t, uint32_t>>
       _mapping;
 
-  std::unordered_map<std::pair<uint32_t, uint32_t>, bool> _activated_mapping;
+  absl::flat_hash_map<std::pair<uint32_t, uint32_t>, bool> _activated_mapping;
 
  public:
   hst_svc_mapping() = default;

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

@@ -20,6 +20,7 @@
 #define CCB_BAM_KPI_SERVICE_HH
 
 #include <array>
+#include <absl/container/flat_hash_set.h>
 #include "com/centreon/broker/bam/kpi.hh"
 #include "com/centreon/broker/bam/kpi_event.hh"
 #include "com/centreon/broker/bam/service_listener.hh"
@@ -50,6 +51,7 @@ class kpi_service : public service_listener, public kpi {
 
   bool _acknowledged;
   bool _downtimed;
+  absl::flat_hash_set<uint32_t> _downtime_ids;
   std::array<double, 5> _impacts;
   timestamp _last_check;
   std::string _output;

bam/src/ba.cc

@@ -508,13 +508,17 @@ void ba::visit(io::stream* visitor) {
     short hard_state(get_state_hard());
     bool state_changed(false);
     if (!_event) {
+      log_v2::bam()->trace("BAM: ba::visit no event => creation of one");
       if (_last_kpi_update.is_null())
         _last_kpi_update = time(nullptr);
       _open_new_event(visitor, hard_state);
     }
     // If state changed, close event and open a new one.
     else if (_in_downtime != _event->in_downtime ||
              hard_state != _event->status) {
+      log_v2::bam()->trace(
+          "BAM: ba::visit event needs update downtime: {}, state: {}",
+          _in_downtime != _event->in_downtime, hard_state != _event->status);
       state_changed = true;
       _event->end_time = _last_kpi_update;
       visitor->write(std::static_pointer_cast<io::data>(_event));
@@ -546,7 +550,7 @@ void ba::visit(io::stream* visitor) {
 
     // Generate virtual service status event.
     if (_generate_virtual_status) {
-      std::shared_ptr<neb::service_status> status(new neb::service_status);
+      auto status{std::make_shared<neb::service_status>()};
       status->active_checks_enabled = false;
       status->check_interval = 0.0;
       status->check_type = 1;  // Passive.
@@ -558,6 +562,7 @@ void ba::visit(io::stream* visitor) {
       status->flap_detection_enabled = false;
       status->has_been_checked = true;
       status->host_id = _host_id;
+      status->downtime_depth = _in_downtime;
       // status->host_name = XXX;
       status->is_flapping = false;
       if (_event)
@@ -602,7 +607,7 @@ void ba::visit(io::stream* visitor) {
 void ba::service_update(const std::shared_ptr<neb::downtime>& dt,
                         io::stream* visitor) {
   (void)visitor;
-  if ((dt->host_id == _host_id) && (dt->service_id == _service_id)) {
+  if (dt->host_id == _host_id && dt->service_id == _service_id) {
     // Log message.
     log_v2::bam()->debug(
         "BAM: BA {} '{}' is getting notified of a downtime on its service ({}, "
@@ -645,8 +650,10 @@ void ba::save_inherited_downtime(persistent_cache& cache) const {
  *
  *  @param[in] dwn  The inherited downtime.
  */
-void ba::set_inherited_downtime(inherited_downtime const& dwn) {
+void ba::set_inherited_downtime(const inherited_downtime& dwn) {
   _inherited_downtime.reset(new inherited_downtime(dwn));
+  if (_inherited_downtime->in_downtime)
+    _in_downtime = true;
 }
 
 /**

bam/src/bool_expression.cc

@@ -46,7 +46,8 @@ bool bool_expression::child_has_update(computable* child, io::stream* visitor) {
   // class, as the bool_* classes already cache most of them.
   if (child == _expression.get()) {
     // Logging.
-    log_v2::bam()->debug("BAM: boolean expression {} is getting notified of child update", _id);
+    log_v2::bam()->debug(
+        "BAM: boolean expression {} is getting notified of child update", _id);
   }
   return true;
 }
@@ -58,8 +59,8 @@ bool bool_expression::child_has_update(computable* child, io::stream* visitor) {
  */
 impact_values::state bool_expression::get_state() const {
   return (_expression->value_hard() == _impact_if)
-              ? bool_expression::state::state_critical
-              : bool_expression::state::state_ok;
+             ? bool_expression::state::state_critical
+             : bool_expression::state::state_ok;
 }
 
 /**

bam/src/configuration/applier/ba.cc

@@ -84,19 +84,15 @@ void applier::ba::apply(bam::configuration::state::bas const& my_bas,
   std::list<bam::configuration::ba> to_modify;
 
   // Iterate through configuration.
-  for (bam::configuration::state::bas::iterator it(to_create.begin()),
-       end(to_create.end());
-       it != end;) {
-    std::map<uint32_t, applied>::iterator cfg_it(to_delete.find(it->first));
+  for (auto it = to_create.begin(), end = to_create.end(); it != end;) {
+    auto cfg_it = to_delete.find(it->first);
     // Found = modify (or not).
     if (cfg_it != to_delete.end()) {
       // Configuration mismatch, modify object.
       if (cfg_it->second.cfg != it->second)
         to_modify.push_back(it->second);
       to_delete.erase(cfg_it);
-      bam::configuration::state::bas::iterator tmp = it;
-      ++it;
-      to_create.erase(tmp);
+      it = to_create.erase(it);
     }
     // Not found = create.
     else
@@ -220,13 +216,17 @@ std::shared_ptr<neb::host> applier::ba::_ba_host(uint32_t host_id) {
  */
 std::shared_ptr<neb::service> applier::ba::_ba_service(uint32_t ba_id,
                                                        uint32_t host_id,
-                                                       uint32_t service_id) {
-  std::shared_ptr<neb::service> s(new neb::service);
+                                                       uint32_t service_id,
+                                                       bool in_downtime) {
+  log_v2::bam()->trace("_ba_service ba {}, service {}:{} with downtime {}",
+                       ba_id, host_id, service_id, in_downtime);
+  auto s{std::make_shared<neb::service>()};
   s->host_id = host_id;
   s->service_id = service_id;
   s->service_description = fmt::format("ba_{}", ba_id);
   s->display_name = s->service_description;
   s->last_update = time(nullptr);
+  s->downtime_depth = in_downtime ? 1 : 0;
   return s;
 }
 
@@ -282,6 +282,7 @@ void applier::ba::save_to_cache(persistent_cache& cache) {
  *  @param[in] cache  The cache.
  */
 void applier::ba::load_from_cache(persistent_cache& cache) {
+  log_v2::bam()->trace("BAM: loading inherited downtimes from cache");
   std::shared_ptr<io::data> d;
   cache.get(d);
   while (d) {
@@ -294,6 +295,9 @@ void applier::ba::load_from_cache(persistent_cache& cache) {
       log_v2::bam()->debug("BAM: found an inherited downtime for BA {}",
                            found->first);
       found->second.obj->set_inherited_downtime(dwn);
+      auto s = _ba_service(found->first, found->second.cfg.get_host_id(),
+                           found->second.cfg.get_service_id(), dwn.in_downtime);
+      multiplexing::publisher().write(s);
     }
     cache.get(d);
   }

bam/src/configuration/applier/bool_expression.cc

@@ -24,7 +24,6 @@
 #include "com/centreon/broker/bam/exp_builder.hh"
 #include "com/centreon/broker/bam/exp_parser.hh"
 #include "com/centreon/broker/bam/service_book.hh"
-#include "com/centreon/broker/logging/logging.hh"
 #include "com/centreon/broker/log_v2.hh"
 
 using namespace com::centreon::broker;
@@ -86,8 +85,8 @@ void applier::bool_expression::apply(
   for (std::map<uint32_t, applied>::iterator it(to_delete.begin()),
        end(to_delete.end());
        it != end; ++it) {
-    log_v2::bam()->info(
-        "BAM: removing boolean expression {}", it->second.cfg.get_id());
+    log_v2::bam()->info("BAM: removing boolean expression {}",
+                        it->second.cfg.get_id());
     for (std::list<bool_service::ptr>::const_iterator
              it2(it->second.svc.begin()),
          end2(it->second.svc.end());
@@ -102,8 +101,7 @@ void applier::bool_expression::apply(
   for (bam::configuration::state::bool_exps::iterator it(to_create.begin()),
        end(to_create.end());
        it != end; ++it) {
-    log_v2::bam()->info(
-        "BAM: creating new boolean expression {}", it->first);
+    log_v2::bam()->info("BAM: creating new boolean expression {}", it->first);
     std::shared_ptr<bam::bool_expression> new_bool_exp(
         new bam::bool_expression);
     try {
@@ -131,9 +129,6 @@ void applier::bool_expression::apply(
           "BAM: could not create boolean expression {} so it will be "
           "discarded: {}",
           it->first, e.what());
-      logging::error(logging::high)
-          << "BAM: could not create boolean expression " << it->first
-          << " so it will be discarded: " << e.what();
     }
     new_bool_exp->set_id(it->first);
     new_bool_exp->set_impact_if(it->second.get_impact_if());
@@ -178,11 +173,9 @@ void applier::bool_expression::_resolve_expression_calls() {
           _name_to_ids.find((*call_it)->get_name());
       if (found == _name_to_ids.end()) {
         log_v2::bam()->error(
-            "BAM: could not resolve the external boolean called '{}' for expression '{}'", (*call_it)->get_name(), it->second.cfg.get_name());
-        logging::error(logging::high)
-            << "BAM: could not resolve the external boolean called '"
-            << (*call_it)->get_name() << "' for expression '"
-            << it->second.cfg.get_name() << "'";
+            "BAM: could not resolve the external boolean called '{}' for "
+            "expression '{}'",
+            (*call_it)->get_name(), it->second.cfg.get_name());
         break;
       } else {
         (*call_it)->set_expression(