From 34e6e52b3e25b93d6d13c49e828f6057db3ad330 Mon Sep 17 00:00:00 2001 From: jean-christophe81 <98889244+jean-christophe81@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:20:06 +0200 Subject: [PATCH] MON-205543-hostgroups-disappear * fix(broker): when a host, service, group member is deleted by a poller we insure that another poller is not the owner * review comments * review comments * review comment * Update broker/unified_sql/src/stream_sql.cc Co-authored-by: Sechkem --------- Co-authored-by: Sechkem --- broker/bam/doc/bam.md | 436 ++++++++++ broker/bam/test/monitoring_stream.cc | 19 +- broker/core/sql/src/query_preparator.cc | 29 +- broker/core/test/mysql/mysql.cc | 75 +- .../com/centreon/broker/unified_sql/stream.hh | 20 +- broker/unified_sql/src/stream.cc | 10 +- broker/unified_sql/src/stream_sql.cc | 795 ++++++++++-------- tests/broker-engine/hostgroups.robot | 61 ++ tests/broker-engine/servicegroups.robot | 79 ++ 9 files changed, 1103 insertions(+), 421 deletions(-) create mode 100644 broker/bam/doc/bam.md diff --git a/broker/bam/doc/bam.md b/broker/bam/doc/bam.md new file mode 100644 index 00000000000..80bf4277cba --- /dev/null +++ b/broker/bam/doc/bam.md @@ -0,0 +1,436 @@ +# BAM — Business Activity Monitoring + +BAM is a Broker module that aggregates monitoring states from raw services and hosts into high-level *Business Activities* (BAs). It translates low-level alerts into business-impact indicators visible to non-technical stakeholders. + +--- + +## Core concepts + +### Business Activity (BA) + +A BA represents a business service (e.g. "Online Payment", "VPN"). It has: + +- a **level** (0–100 for impact mode, or a ratio for other modes) +- a **state**: OK, WARNING, CRITICAL, or UNKNOWN +- **warning** and **critical** thresholds that determine when the state changes +- an optional **downtime behaviour** + +Each BA is surfaced to the monitoring engine as a virtual host/service pair, so it can be acknowledged, scheduled for downtime, and graphed like any other service. + +#### BA computation modes + +There are five computation strategies, chosen at configuration time via `state_source`: + +| Mode | Class | Behaviour | +|------|-------|-----------| +| `impact` | `ba_impact` | Level starts at 100. Each non-OK KPI subtracts its configured impact points. State is OK/WARNING/CRITICAL based on thresholds. | +| `best` | `ba_best` | State equals the best (least severe) state among all KPIs. | +| `worst` | `ba_worst` | State equals the worst (most severe) state among all KPIs. | +| `ratio_percent` | `ba_ratio_percent` | State is WARNING/CRITICAL when the percentage of critical KPIs exceeds the configured threshold. | +| `ratio_number` | `ba_ratio_number` | Same as ratio_percent but based on a raw count instead of a percentage. | + +##### Impact BA details + +`_level_hard` and `_level_soft` start at 100. When a KPI is not OK, its configured *nominal impact* is subtracted. Re-adding impacts every 100 operations (`_recompute_limit`) prevents floating-point drift. + +``` +state_ok if _level_hard > _level_warning +state_warning if _level_hard <= _level_warning +state_critical if _level_hard <= _level_critical +``` + +Perfdata: `BA_Level=;;;0;100` + +##### Ratio percent details + +`_level_hard` counts how many KPIs are in CRITICAL state. State is: + +``` +state_critical if (_level_hard / total_kpis * 100) >= _level_critical +state_warning if (_level_hard / total_kpis * 100) >= _level_warning +``` + +#### Downtime behaviour + +| Value | Behaviour | +|-------|-----------| +| `dt_ignore` (0) | Downtime on KPIs has no effect on the BA computation. | +| `dt_inherit` | If all non-OK KPIs are in downtime, the BA inherits a virtual downtime. | +| `dt_ignore_kpi` | KPIs currently in downtime are excluded from the impact calculation. | + +--- + +### Key Performance Indicator (KPI) + +A KPI is a leaf or intermediate node in the BA computation tree. There are four KPI types (field `kpi_type` in `mod_bam_kpi`): + +| Type | Value | Class | Description | +|------|-------|-------|-------------| +| Service | `0` | `kpi_service` | A host/service check result. | +| Meta-service | `1` | — | Aggregated meta-service value. | +| BA | `2` | `kpi_ba` | Another BA used as a KPI (enables hierarchical BAs). | +| Boolean expression | `3` | `kpi_boolexp` | A boolean rule that evaluates to OK or CRITICAL. | + +Every KPI has separate **hard** and **soft** impact values and an optional **downtime** and **acknowledgement** impact. The `impact_values` struct carries: + +- `nominal`: the points removed from the parent BA level +- `acknowledgement`: additional points removed when the problem is acknowledged +- `downtime`: additional points removed when in downtime +- `state`: the underlying state (for ratio/best/worst modes) + +For service KPIs, separate impact values are configured for WARNING, CRITICAL, and UNKNOWN states (`drop_warning`, `drop_critical`, `drop_unknown`). + +--- + +### Boolean expressions + +A boolean expression (`bool_expression`) is a tree of boolean operations evaluated against service states. It exposes a single OK/CRITICAL state to its parent `kpi_boolexp`. + +#### Expression tree nodes + +All nodes implement `bool_value`: + +| Class | Description | +|-------|-------------| +| `bool_service` | Leaf: evaluates the state of a specific host/service. | +| `bool_constant` | Leaf: always true or always false. | +| `bool_and` | Logical AND of two sub-expressions. | +| `bool_or` | Logical OR of two sub-expressions. | +| `bool_not` | Logical NOT. | +| `bool_xor` | Logical XOR. | +| `bool_equal` | Numeric equality between two sub-expressions. | +| `bool_less_than` | Numeric less-than. | +| `bool_more_than` | Numeric greater-than. | +| `bool_not_equal` | Numeric inequality. | + +The expression text is stored in `mod_bam_boolean.expression` and parsed at startup by `exp_parser` / `exp_tokenizer` / `exp_builder`. + +#### Impact direction + +The field `bool_state` (also called `impact_if`) controls which boolean result triggers the impact: + +- `impact_if = true` → the KPI fires (is CRITICAL) when the expression evaluates to **true** +- `impact_if = false` → the KPI fires when the expression evaluates to **false** + +--- + +### Computable tree + +The entire BA/KPI/boolean graph implements the `computable` interface: + +``` +computable +├── ba (abstract base for all BA types) +│ ├── ba_impact +│ ├── ba_best +│ ├── ba_worst +│ ├── ba_ratio_percent +│ └── ba_ratio_number +├── kpi (abstract base for all KPI types) +│ ├── kpi_service +│ ├── kpi_ba +│ └── kpi_boolexp +└── bool_expression + └── bool_value subtree (bool_and, bool_or, bool_service, …) +``` + +When a leaf value changes (e.g. a service check result arrives), the change propagates upward via: + +1. `update_from(child, visitor)` — recomputes this node given the child that changed. +2. `notify_parents_of_change(visitor)` — if this node's state changed, calls `update_from` on every parent. + +The `visitor` (an `io::stream*`) collects the events produced during the propagation (BA status, KPI status, BA events). + +--- + +## `service_book` + +`service_book` is a dispatch table that routes incoming NEB events to every KPI that depends on a given host/service pair. It is owned by `configuration::applier::state` and shared by all appliers. + +### Internal structure + +The book holds a `std::unordered_map` keyed on `(host_id, service_id)`. Each entry contains: + +- a list of `service_listener*` — the `kpi_service` and `bool_service` objects that care about this service +- a `service_state` snapshot — the last known state (current state, last hard state, state type, last check, acknowledged flag) + +### Registration + +When a `kpi_service` or a `bool_service` is created by its applier, it calls `service_book::listen(host_id, service_id, this)`. The reverse call `unlisten()` removes the registration on teardown. Multiple KPIs can listen to the same service simultaneously. + +### Event dispatch + +`monitoring_stream` forwards every relevant NEB event to the book via one of the overloaded `update()` methods: + +| Event type | State field updated | +|------------|---------------------| +| `service_status` / `pb_service` / `pb_service_status` | `current_state`, `last_hard_state`, `state_type`, `last_check` | +| `pb_adaptive_service_status` | forwarded as-is (scheduled downtime flag, etc.) | +| `acknowledgement` / `pb_acknowledgement` | `acknowledged` | +| `downtime` / `pb_downtime` | forwarded as-is (no local state field) | + +Each `update()` call looks up the `(host_id, service_id)` key, updates the cached `service_state` when applicable, and then calls `service_listener::service_update()` on every registered listener so the KPI can recompute its impact and propagate the change upward through the BA tree. + +### Persistence + +On clean shutdown, `save_to_cache()` serializes all `service_state` snapshots into the `persistent_cache` as a `ServicesBookState` protobuf message. On the next startup, `apply_services_state()` restores those snapshots and immediately re-notifies every listener, so KPIs start with a consistent state before the first live check arrives. + +--- + +## `monitoring_stream` + +`monitoring_stream` is the real-time computation engine. It is an `io::stream` that: + +1. **Receives** NEB events from the broker multiplexer: + - `service_status` / `pb_service_status` / `pb_service` → feeds `kpi_service` + - `acknowledgement` / `pb_acknowledgement` → updates acknowledged state + - `downtime` / `pb_downtime` → propagates downtime to BAs and KPIs + + It also **produces** a `rebuild` event: on `update()`, `_rebuild()` queries `mod_bam` for BAs flagged `must_be_rebuild='1'` and publishes a `rebuild` event listing their IDs. `reporting_stream` is the consumer — its `_process_rebuild()` uses it to trigger an availability rebuild. + +2. **Maintains** the computable tree in memory via `configuration::applier::state`, which is populated from the `centreon` database at startup and on `update()`. + +3. **Writes BA/KPI statuses** to the `centreon_storage` database: + - `mod_bam`: current BA levels and states + - `mod_bam_kpi`: current KPI states and last impact + +4. **Sends external commands** to the monitoring engine via the Engine command pipe (`_ext_cmd_file`): + - **Forced service checks** — because each BA maps to a virtual service, the engine must be told to re-check that service when the BA state changes. Checks are deduplicated and batched over a 5-second window to avoid flooding the engine with redundant checks when a BA tree re-evaluates multiple times in rapid succession. + - **Downtimes** — when a BA inherits a downtime from its KPIs, a corresponding downtime command is sent to the virtual service. + +5. **Persists a cache** of inherited downtimes across restarts via `persistent_cache`. + +### Startup sequence + +``` +monitoring_stream() + └── _prepare() -- prepare SQL statements + └── update() + ├── _applier.apply() -- load configuration from DB, build computable tree + ├── _rebuild() -- publish a rebuild event for BAs flagged must_be_rebuild + ├── initialize() -- publish initial states of all BAs/KPIs + └── _read_cache() -- restore inherited downtimes from persistent cache +``` + +### Initial event restoration (`set_initial_event`) + +On restart, BA and KPI objects must be seeded with the event that was still open when broker last stopped, so the reporting timeline is continuous and no gap appears between sessions. + +The flow has three stages: + +**1. Database read (`reader_v2`)** — executed during `monitoring_stream::update()`. + +- For each BA row in `mod_bam`, `reader_v2` reads `last_state_change` (col 5), `current_status` (col 6), and `in_downtime` (col 7). If `last_state_change` is not `NULL`, it builds a `pb_ba_event` with no `end_time` and stores it in the configuration object via `configuration::ba::set_opened_event()`. +- For each KPI row in `mod_bam_kpi`, `reader_v2` reads `last_state_change` (col 16), `in_downtime` (col 17), and `last_impact` (col 18). If `last_state_change` is not `NULL`, it builds a `KpiEvent` and stores it via `configuration::kpi::set_opened_event()`. + +**2. Object construction (configuration appliers)** — when the computable tree is built from the loaded configuration. + +- `applier::ba::_new_ba()` checks `cfg.get_opened_event().obj().ba_id()`: if non-zero, it calls `ba::set_initial_event()`, which stores the event as `_event` (the currently open period) and appends it to `_initial_events`. +- `applier::kpi::_resolve_kpi()` checks `cfg.get_opened_event().kpi_id()`: if non-zero, it calls `kpi::set_initial_event()`. This variant also reconciles the impact: if the stored `impact_level` differs from the value computed by the current configuration (the KPI's `impact_hard()` output), the old event is closed at `now` and a fresh event is opened — both are pushed to `_initial_events` — so that stale impact values do not carry over into the new session. + +**3. Flushing to the broker bus (`initialize()`)** — the last step of the startup sequence. + +``` +monitoring_stream::initialize() + └── event_cache_visitor ev_cache + └── _applier.visit(&ev_cache) + ├── ba_applier.visit() → ba::visit() → ba::_commit_initial_events(ev_cache) + └── kpi_applier.visit() → kpi::visit() → kpi::commit_initial_events(ev_cache) + └── ev_cache.commit_to(publisher) + └── reporting_stream receives events + ├── writes open BA events → mod_bam_reporting_ba_events + └── writes open KPI events → mod_bam_reporting_kpi_events +``` + +`_commit_initial_events()` / `commit_initial_events()` write every entry in `_initial_events` into the visitor, then clear the vector. `ev_cache.commit_to(publisher)` forwards all collected events to the multiplexing publisher; `reporting_stream` receives them and persists them, closing the gap left by the restart. + +--- + +## `reporting_stream` + +`reporting_stream` is the historical recording engine (stream name: `BAM-BI`). It persists the timeline of BA and KPI state changes for reporting and SLA computation. + +### Events processed + +| Event type | Method | Target table | +|------------|--------|--------------| +| `ba_event` | `_process_ba_event` | `mod_bam_reporting_ba_events` | +| `pb_ba_event` | `_process_pb_ba_event` | `mod_bam_reporting_ba_events` | +| `pb_ba_duration_event` | `_process_pb_ba_duration_event` | `mod_bam_reporting_ba_events_durations` | +| `kpi_event` | `_process_kpi_event` | `mod_bam_reporting_kpi_events` | +| `pb_kpi_event` | `_process_pb_kpi_event` | `mod_bam_reporting_kpi_events` | +| Dimension events (legacy) | `_process_dimension` | `mod_bam_reporting_ba`, `_bv`, `_kpi`, `_timeperiods`, … | +| Dimension events (`pb_dimension_*`) | `_process_pb_dimension` | `mod_bam_reporting_ba`, `_bv`, `_kpi`, `_timeperiods`, … | +| `rebuild` | `_process_rebuild` | triggers availability rebuild | + +Dimension events carry the configuration snapshot (BA names, BV memberships, KPI definitions, time periods). They are received at startup or after a configuration change and used to populate the `mod_bam_reporting_*` dimension tables. + +### BA event lifecycle + +A BA event records a continuous period during which a BA was in a given state: + +```sql +-- mod_bam_reporting_ba_events +ba_event_id -- auto-increment PK +ba_id -- which BA +start_time -- epoch when the state started +end_time -- epoch when the state ended (NULL = still open) +status -- 0=OK, 1=WARNING, 2=CRITICAL, 3=UNKNOWN +in_downtime -- was the BA in downtime during this event? +first_level -- BA level at the start of the event +``` + +On startup, the stream closes any events that have no `end_time` (inconsistent events from a previous crash), then sets `end_time` to now on all remaining open events (`_close_all_events`). New events are opened as statuses arrive. + +### KPI event lifecycle + +Similar to BA events but per KPI: + +```sql +-- mod_bam_reporting_kpi_events +kpi_event_id +kpi_id +start_time +end_time +status +in_downtime +impact_level +first_output +first_perfdata +``` + +### Startup sequence + +``` +reporting_stream() + └── _prepare() -- prepare SQL statements + └── _load_timeperiods() -- load timeperiod definitions from DB + └── _load_kpi_ba_events() -- load open events into memory cache + └── _close_inconsistent_events() -- close events open in only one table + └── _close_all_events() -- close remaining open events + └── availability_thread.start() -- start the availability computation thread +``` + +--- + +## Availability computation + +Availability is computed by `availability_thread`, a background thread owned by `reporting_stream`. + +### Trigger + +The thread wakes up once per day, just after midnight, and also on-demand when a `rebuild` event is received naming specific BAs. + +### Algorithm (`availability_builder`) + +For each BA × time period combination: + +1. Query `mod_bam_reporting_ba_events` for all events that overlap the target day. +2. For each event, compute the intersection of its `[start_time, end_time]` interval with the time period (via `timeperiod::duration_intersect`). +3. Accumulate the intersection duration into the appropriate counter. + +``` +available -- seconds BA was OK, within the time period +degraded -- seconds BA was WARNING +unavailable -- seconds BA was CRITICAL +unknown -- seconds BA was UNKNOWN +downtime -- seconds BA was in downtime (regardless of state) +``` + +Alert counters record how many distinct events *opened* during the day: + +``` +alert_unavailable_opened +alert_degraded_opened +alert_unknown_opened +nb_downtime +``` + +### Output table + +```sql +-- mod_bam_reporting_ba_availabilities +ba_id +time_id -- epoch of day start (midnight) +timeperiod_id +available -- seconds +unavailable -- seconds +degraded -- seconds +unknown -- seconds +downtime -- seconds +alert_unavailable_opened +alert_degraded_opened +alert_unknown_opened +nb_downtime +timeperiod_is_default +``` + +Each BA can have multiple time periods; the default time period is flagged with `timeperiod_is_default = true`. + +--- + +## Database schema overview + +### Configuration tables (read by `monitoring_stream`) + +| Table | Purpose | +|-------|---------| +| `mod_bam` | BA definitions (id, name, type, thresholds, downtime behaviour) | +| `mod_bam_kpi` | KPI definitions (type, host/service/ba/bool references, impact values) | +| `mod_bam_boolean` | Boolean expression definitions (name, expression text, `bool_state`) | +| `mod_bam_impacts` | Named impact values referenced by KPIs | +| `mod_bam_ba_groups` / `mod_bam_bagroup_ba_relation` | Business Views (BV) and BA memberships | +| `mod_bam_relations_ba_timeperiods` | Time period assignments per BA | + +### Reporting tables (written by `reporting_stream`) + +| Table | Purpose | +|-------|---------| +| `mod_bam_reporting_ba` | BA dimension (name, description) | +| `mod_bam_reporting_bv` | BV dimension | +| `mod_bam_reporting_kpi` | KPI dimension | +| `mod_bam_reporting_ba_events` | BA state timeline | +| `mod_bam_reporting_kpi_events` | KPI state timeline | +| `mod_bam_reporting_ba_events_durations` | Derived durations per BA event × time period | +| `mod_bam_reporting_ba_availabilities` | Daily availability per BA × time period | +| `mod_bam_reporting_timeperiods` | Time period definitions | +| `mod_bam_reporting_relations_ba_bv` | BA ↔ BV memberships | +| `mod_bam_reporting_relations_ba_kpi_events` | Links BA events to KPI events | +| `mod_bam_reporting_relations_ba_timeperiods` | BA ↔ time period assignments | + +--- + +## Data flow summary + +``` +Engine checks + │ + ▼ NEB events (service_status, downtime, acknowledgement) +monitoring_stream + │ + ▼ update(event, visitor) +service_book (dispatch table: (host_id, service_id) → [kpi_service, bool_service, …]) + │ + ▼ service_listener::service_update() + ├── updates computable tree (kpi_service → ba) + │ propagates via update_from() / notify_parents_of_change() + │ + ├── writes BA/KPI current statuses → centreon_storage (mod_bam, mod_bam_kpi) + │ + ├── sends forced service check commands → Engine named pipe + │ (batched, deduplicated, 5 s window) + │ + └── publishes BAM events to broker bus + │ + ▼ BAM events (ba_event, kpi_event, dimension events) + reporting_stream + │ + ├── writes event timeline → mod_bam_reporting_ba_events + │ → mod_bam_reporting_kpi_events + │ + ├── writes dimension data → mod_bam_reporting_ba, _bv, _kpi, … + │ + └── availability_thread (wakes at midnight) + └── reads ba_events, intersects with timeperiods + └── writes → mod_bam_reporting_ba_availabilities +``` diff --git a/broker/bam/test/monitoring_stream.cc b/broker/bam/test/monitoring_stream.cc index 129c0619f68..9d302758f7b 100644 --- a/broker/bam/test/monitoring_stream.cc +++ b/broker/bam/test/monitoring_stream.cc @@ -30,6 +30,9 @@ using log_v2 = com::centreon::common::log_v2::log_v2; using namespace com::centreon::broker; using namespace com::centreon::broker::bam; +const std::string db_user = "root"; +const std::string db_password = "centreon"; + class BamMonitoringStream : public testing::Test { void SetUp() override { config::applier::init(com::centreon::common::BROKER, 0, "test_broker", 0); @@ -38,9 +41,9 @@ class BamMonitoringStream : public testing::Test { }; TEST_F(BamMonitoringStream, WriteKpi) { - database_config cfg("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config cfg("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon"); - database_config storage("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config storage("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon_storage"); std::shared_ptr cache; @@ -56,9 +59,9 @@ TEST_F(BamMonitoringStream, WriteKpi) { } TEST_F(BamMonitoringStream, WriteBA) { - database_config cfg("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config cfg("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon"); - database_config storage("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config storage("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon_storage"); ; std::shared_ptr cache; @@ -73,9 +76,9 @@ TEST_F(BamMonitoringStream, WriteBA) { } TEST_F(BamMonitoringStream, WorkWithNoPendigMysqlRequest) { - database_config cfg("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config cfg("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon", 0); - database_config storage("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config storage("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon_storage", 0); ; std::shared_ptr cache; @@ -95,9 +98,9 @@ TEST_F(BamMonitoringStream, WorkWithNoPendigMysqlRequest) { } TEST_F(BamMonitoringStream, WorkWithPendigMysqlRequest) { - database_config cfg("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config cfg("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon", 5); - database_config storage("MySQL", "127.0.0.1", "", 3306, "root", "centreon", + database_config storage("MySQL", "127.0.0.1", "", 3306, db_user, db_password, "centreon_storage", 5); ; std::shared_ptr cache; diff --git a/broker/core/sql/src/query_preparator.cc b/broker/core/sql/src/query_preparator.cc index 55ae1dd3831..368cb9b2488 100644 --- a/broker/core/sql/src/query_preparator.cc +++ b/broker/core/sql/src/query_preparator.cc @@ -548,13 +548,6 @@ mysql_stmt query_preparator::prepare_update_table( query.append("=?,"); query_bind_mapping.insert(std::make_pair(key, query_size++)); } - // Part of ID field. - else { - where.append(e.name); - where.append("=? AND "); - key = fmt::format(":{}", entry_name); - where_bind_mapping.insert(std::make_pair(key, where_size++)); - } } else throw msg_fmt( "could not prepare update query for event of type {}:" @@ -562,6 +555,28 @@ mysql_stmt query_preparator::prepare_update_table( "object", _event_id, e.number, info->get_name()); } + + for (const auto& e : _pb_unique) { + const google::protobuf::FieldDescriptor* f = + desc->FindFieldByNumber(e.number); + if (!f) + throw msg_fmt( + "could not prepare update query for event of type {}:" + "protobuf field with number {} does not exist in '{}' protobuf " + "object", + _event_id, e.number, info->get_name()); + std::string_view entry_name = f->name(); + if (static_cast(f->index()) >= pb_mapping.size()) + pb_mapping.resize(f->index() + 1); + if (std::get<0>(pb_mapping[f->index()]).empty()) + pb_mapping[f->index()] = std::make_tuple(entry_name, e.max_length, e.attribute); + + where.append(e.name); + where.append("=? AND "); + key = fmt::format(":{}", entry_name); + where_bind_mapping.insert(std::make_pair(key, where_size++)); + } + query.resize(query.size() - 1); query.append(where, 0, where.size() - 5); diff --git a/broker/core/test/mysql/mysql.cc b/broker/core/test/mysql/mysql.cc index d013684c750..85dac2fe629 100644 --- a/broker/core/test/mysql/mysql.cc +++ b/broker/core/test/mysql/mysql.cc @@ -50,6 +50,9 @@ using namespace com::centreon::broker; using namespace com::centreon::broker::database; using log_v2 = com::centreon::common::log_v2::log_v2; +const std::string db_user = "root"; +const std::string db_password = "centreon"; + class DatabaseStorageTest : public ::testing::Test { public: void SetUp() override { @@ -65,8 +68,8 @@ class DatabaseStorageTest : public ::testing::Test { // When there is no database // Then the mysql creation throws an exception TEST_F(DatabaseStorageTest, NoDatabase) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 9876, "root", - "centreon", "centreon_storage"); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 9876, db_user, + db_password, "centreon_storage"); std::unique_ptr ms; ASSERT_THROW(ms.reset(new mysql(db_cfg)), msg_fmt); } @@ -75,8 +78,8 @@ TEST_F(DatabaseStorageTest, NoDatabase) { // And when the connection is well done // Then no exception is thrown and the mysql object is well built. TEST_F(DatabaseStorageTest, ConnectionOk) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage"); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage"); std::unique_ptr ms; ASSERT_NO_THROW(ms = std::make_unique(db_cfg)); } @@ -506,8 +509,8 @@ TEST_F(DatabaseStorageTest, ConnectionOk) { TEST_F(DatabaseStorageTest, CustomVarStatement) { config::applier::modules modules(log_v2::instance().get(log_v2::SQL)); modules.load_file("./broker/lib/10-neb.so"); - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); std::unique_ptr ms(new mysql(db_cfg)); query_preparator::event_unique unique; unique.insert("host_id"); @@ -1258,8 +1261,8 @@ TEST_F(DatabaseStorageTest, CustomVarStatement) { ////} // TEST_F(DatabaseStorageTest, ChooseConnectionByName) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms = std::make_unique(db_cfg); int thread_foo(ms->choose_connection_by_name("foo")); int thread_bar(ms->choose_connection_by_name("bar")); @@ -1280,8 +1283,8 @@ TEST_F(DatabaseStorageTest, ChooseConnectionByName) { // Then we can bind values to it and execute the statement. // Then a commit makes data available in the database. TEST_F(DatabaseStorageTest, RepeatStatements) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1359,8 +1362,8 @@ TEST_F(DatabaseStorageTest, RepeatStatements) { } TEST_F(DatabaseStorageTest, CheckBulkStatement) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string version = ms->get_server_version(); std::vector arr = @@ -1428,8 +1431,8 @@ TEST_F(DatabaseStorageTest, CheckBulkStatement) { TEST_F(DatabaseStorageTest, UpdateBulkStatement) { constexpr int TOTAL = 20; - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; if (ms->support_bulk_statement()) { std::string query{ @@ -1492,8 +1495,8 @@ TEST_F(DatabaseStorageTest, UpdateBulkStatement) { // Then we can bind values to it and execute the statement. // Then a commit makes data available in the database. TEST_F(DatabaseStorageTest, LastInsertId) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); time_t now = time(nullptr); std::string query( fmt::format("INSERT INTO metrics" @@ -1533,8 +1536,8 @@ TEST_F(DatabaseStorageTest, LastInsertId) { } TEST_F(DatabaseStorageTest, BulkStatementWithNullStr) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; if (ms->support_bulk_statement()) { std::string query1{"DROP TABLE IF EXISTS ut_test"}; @@ -1596,8 +1599,8 @@ TEST_F(DatabaseStorageTest, BulkStatementWithNullStr) { } TEST_F(DatabaseStorageTest, RepeatStatementsWithNull) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1645,8 +1648,8 @@ TEST_F(DatabaseStorageTest, RepeatStatementsWithNull) { } TEST_F(DatabaseStorageTest, RepeatStatementsWithBigStrings) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1747,8 +1750,8 @@ TEST_F(DatabaseStorageTest, RepeatStatementsWithBigStrings) { } TEST_F(DatabaseStorageTest, RepeatStatementsWithNullValues) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1825,8 +1828,8 @@ TEST_F(DatabaseStorageTest, RepeatStatementsWithNullValues) { } TEST_F(DatabaseStorageTest, BulkStatementsWithNullValues) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1933,8 +1936,8 @@ TEST_F(DatabaseStorageTest, BulkStatementsWithNullValues) { } TEST_F(DatabaseStorageTest, RepeatStatementsWithBooleanValues) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -1973,8 +1976,8 @@ TEST_F(DatabaseStorageTest, RepeatStatementsWithBooleanValues) { } TEST_F(DatabaseStorageTest, BulkStatementsWithBooleanValues) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -2042,8 +2045,8 @@ static std::string row_filler2(const row& data) { } TEST_F(DatabaseStorageTest, MySqlMultiInsert) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -2191,8 +2194,8 @@ struct multi_event_binder { }; TEST_F(DatabaseStorageTest, bulk_or_multi_bbdo_event_bulk) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ @@ -2246,8 +2249,8 @@ TEST_F(DatabaseStorageTest, bulk_or_multi_bbdo_event_bulk) { } TEST_F(DatabaseStorageTest, bulk_or_multi_bbdo_event_multi) { - database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, "root", - "centreon", "centreon_storage", 5, true, 5); + database_config db_cfg("MySQL", "127.0.0.1", MYSQL_SOCKET, 3306, db_user, + db_password, "centreon_storage", 5, true, 5); auto ms{std::make_unique(db_cfg)}; std::string query1{"DROP TABLE IF EXISTS ut_test"}; std::string query2{ diff --git a/broker/unified_sql/inc/com/centreon/broker/unified_sql/stream.hh b/broker/unified_sql/inc/com/centreon/broker/unified_sql/stream.hh index d5ae94bd34d..fc79c2882d7 100644 --- a/broker/unified_sql/inc/com/centreon/broker/unified_sql/stream.hh +++ b/broker/unified_sql/inc/com/centreon/broker/unified_sql/stream.hh @@ -261,8 +261,9 @@ class stream : public io::stream { std::shared_ptr _center; ConflictManagerStats* _stats; - absl::flat_hash_set _cache_deleted_instance_id; - std::unordered_map _cache_host_instance; + absl::flat_hash_set _cache_deleted_instance_id; + std::unordered_map + _cache_host_instance; absl::flat_hash_map _cache_hst_cmd; absl::flat_hash_map, size_t> _cache_svc_cmd; absl::flat_hash_map, index_info> _index_cache; @@ -274,7 +275,11 @@ class stream : public io::stream { absl::flat_hash_map, uint64_t> _severity_cache; absl::flat_hash_map, uint64_t> _tags_cache; - absl::flat_hash_map, uint64_t> _resource_cache; + absl::flat_hash_map, + uint64_t> + _resource_cache; mutable absl::Mutex _timer_m; /* This is a barrier for timers. It must be locked in shared mode in the @@ -329,11 +334,13 @@ class stream : public io::stream { database::mysql_stmt _pb_host_check_update; database::mysql_stmt _host_group_insupdate; database::mysql_stmt _pb_host_group_insupdate; - database::mysql_stmt _host_group_member_delete; + std::unique_ptr _host_group_member_delete; database::mysql_stmt _host_group_member_insert; database::mysql_stmt _pb_host_group_member_insert; database::mysql_stmt _host_insupdate; + database::mysql_stmt _host_update; database::mysql_stmt _pb_host_insupdate; + database::mysql_stmt _pb_host_update; database::mysql_stmt _host_parent_delete; database::mysql_stmt _host_parent_insert; database::mysql_stmt _pb_host_parent_delete; @@ -347,12 +354,12 @@ class stream : public io::stream { database::mysql_stmt _pb_service_check_update; database::mysql_stmt _service_group_insupdate; database::mysql_stmt _pb_service_group_insupdate; - database::mysql_stmt _service_group_member_delete; + std::unique_ptr _service_group_member_delete; database::mysql_stmt _service_group_member_insert; - database::mysql_stmt _pb_service_group_member_delete; database::mysql_stmt _pb_service_group_member_insert; database::mysql_stmt _service_insupdate; database::mysql_stmt _pb_service_insupdate; + database::mysql_stmt _pb_service_update; database::mysql_stmt _service_status_update; std::unique_ptr _hscr_update; @@ -451,6 +458,7 @@ class stream : public io::stream { void _process_service_status(const std::shared_ptr& d); void _process_responsive_instance(const std::shared_ptr& d); + void _prepare_pb_requests(); void _process_pb_host(const std::shared_ptr& d); uint64_t _process_pb_host_in_resources(const Host& h, int32_t conn); void _process_pb_instance_configuration(const std::shared_ptr& d); diff --git a/broker/unified_sql/src/stream.cc b/broker/unified_sql/src/stream.cc index bdbd0adc394..e34b741c2e6 100644 --- a/broker/unified_sql/src/stream.cc +++ b/broker/unified_sql/src/stream.cc @@ -363,7 +363,7 @@ void stream::_load_caches() { /* resources => _resources_cache */ _mysql.run_query_and_get_result( - "SELECT resource_id, id, parent_id FROM resources", + "SELECT resource_id, id, parent_id, poller_id FROM resources", std::move(promise_resource)); /* severities => _severity_cache */ @@ -564,8 +564,8 @@ void stream::_load_caches() { try { mysql_result res{future_resource.get()}; while (_mysql.fetch_row(res)) { - _resource_cache[{res.value_as_u64(1), res.value_as_u64(2)}] = - res.value_as_u64(0); + _resource_cache[{res.value_as_u64(1), res.value_as_u64(2), + res.value_as_u64(3)}] = res.value_as_u64(0); } } catch (const std::exception& e) { throw msg_fmt("unified sql: could not get the list of resources: {}", @@ -1173,11 +1173,11 @@ void stream::_clear_instances_cache(const std::list& ids) { _cache_svc_cmd.erase(itt); // resources - auto res_it = _resource_cache.find({svc_id, host_id}); + auto res_it = _resource_cache.find({svc_id, host_id, it->second}); if (res_it != _resource_cache.end()) _resource_cache.erase(res_it); } - auto res_it = _resource_cache.find({host_id, 0}); + auto res_it = _resource_cache.find({host_id, 0, it->second}); if (res_it != _resource_cache.end()) _resource_cache.erase(res_it); } diff --git a/broker/unified_sql/src/stream_sql.cc b/broker/unified_sql/src/stream_sql.cc index 2ef09a62244..f8591029b01 100644 --- a/broker/unified_sql/src/stream_sql.cc +++ b/broker/unified_sql/src/stream_sql.cc @@ -1430,15 +1430,18 @@ void stream::_process_host_group_member(const std::shared_ptr& d) { "SQL: disabling membership of host {} to host group {} on instance {}", hgm.host_id, hgm.group_id, hgm.poller_id); - if (!_host_group_member_delete.prepared()) { - query_preparator::event_unique unique; - unique.insert("hostgroup_id"); - unique.insert("host_id"); - query_preparator qp(neb::host_group_member::static_type(), unique); - _host_group_member_delete = qp.prepare_delete(_mysql); + if (!_host_group_member_delete) { + _host_group_member_delete = std::make_unique( + "DELETE hosts_hostgroups FROM hosts_hostgroups LEFT JOIN hosts ON " + "hosts_hostgroups.host_id=hosts.host_id " + "WHERE hosts_hostgroups.host_id=? and hostgroup_id = ? and " + "(instance_id = ? OR instance_id is NULL)"); + _mysql.prepare_statement(*_host_group_member_delete); } - _host_group_member_delete << hgm; - _mysql.run_statement(_host_group_member_delete, + _host_group_member_delete->bind_value_as_u64(0, hgm.host_id); + _host_group_member_delete->bind_value_as_u64(1, hgm.group_id); + _host_group_member_delete->bind_value_as_u64(2, hgm.poller_id); + _mysql.run_statement(*_host_group_member_delete, database::mysql_error::delete_host_group_member, conn); _add_action(conn, actions::hostgroups); } @@ -1535,12 +1538,20 @@ void stream::_process_pb_host_group_member(const std::shared_ptr& d) { "SQL: disabling membership of host {} to host group {} on instance {}", hgm.host_id(), hgm.hostgroup_id(), hgm.poller_id()); - std::string query = fmt::format( - "DELETE FROM hosts_hostgroups WHERE host_id={} and hostgroup_id = {}", - hgm.host_id(), hgm.hostgroup_id()); + if (!_host_group_member_delete) { + _host_group_member_delete = std::make_unique( + "DELETE hosts_hostgroups FROM hosts_hostgroups LEFT JOIN hosts ON " + "hosts_hostgroups.host_id=hosts.host_id " + "WHERE hosts_hostgroups.host_id=? and hostgroup_id = ? and " + "(instance_id = ? OR instance_id is NULL)"); + _mysql.prepare_statement(*_host_group_member_delete); + } + _host_group_member_delete->bind_value_as_u64(0, hgm.host_id()); + _host_group_member_delete->bind_value_as_u64(1, hgm.hostgroup_id()); + _host_group_member_delete->bind_value_as_u64(2, hgm.poller_id()); + _mysql.run_statement(*_host_group_member_delete, + database::mysql_error::delete_host_group_member, conn); - _mysql.run_query(query, database::mysql_error::delete_host_group_member, - conn); _add_action(conn, actions::hostgroups); } } @@ -1578,19 +1589,31 @@ void stream::_process_host(const std::shared_ptr& d) { unique.insert("host_id"); query_preparator qp(neb::host::static_type(), unique); _host_insupdate = qp.prepare_insert_or_update(_mysql); - } - // Process object. - _host_insupdate << h; - _mysql.run_statement(_host_insupdate, database::mysql_error::store_host, - conn); - _add_action(conn, actions::hosts); + query_preparator::event_unique update_unique; + update_unique.insert("host_id"); + update_unique.insert("instance_id"); + query_preparator update_qp(neb::host::static_type(), update_unique); + _host_update = update_qp.prepare_update(_mysql); + } - // Fill the cache... - if (h.enabled) + // Process object and fill the cache... + if (h.enabled) { _cache_host_instance[h.host_id] = h.poller_id; - else - _cache_host_instance.erase(h.host_id); + _host_insupdate << h; + _mysql.run_statement(_host_insupdate, database::mysql_error::store_host, + conn); + } else { + auto cache_to_delete = _cache_host_instance.find(h.host_id); + if (cache_to_delete != _cache_host_instance.end() && + cache_to_delete->second == h.poller_id) { + _cache_host_instance.erase(cache_to_delete); + } + _host_update << h; + _mysql.run_statement(_host_update, database::mysql_error::store_host, + conn); + } + _add_action(conn, actions::hosts); } else SPDLOG_LOGGER_TRACE( _logger_sql, @@ -2072,6 +2095,316 @@ void stream::_process_host_status(const std::shared_ptr& d) { now, hs.current_state, hs.state_type); } +void stream::_prepare_pb_requests() { + if (!_pb_host_insupdate.prepared()) { + query_preparator::event_pb_unique unique{ + {1, "host_id", io::protobuf_base::invalid_on_zero, 0}}; + query_preparator qp(neb::pb_host::static_type(), unique); + + std::vector host_entries = { + {1, "host_id", io::protobuf_base::invalid_on_zero, 0}, + {2, "acknowledged", 0, 0}, + {3, "acknowledgement_type", 0, 0}, + {4, "active_checks", 0, 0}, + {5, "enabled", 0, 0}, + {6, "scheduled_downtime_depth", 0, 0}, + {7, "check_command", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_check_command)}, + {8, "check_interval", 0, 0}, + {9, "check_period", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_check_period)}, + {10, "check_type", 0, 0}, + {11, "check_attempt", 0, 0}, + {12, "state", 0, 0}, + {13, "event_handler_enabled", 0, 0}, + {14, "event_handler", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_event_handler)}, + {15, "execution_time", 0, 0}, + {16, "flap_detection", 0, 0}, + {17, "checked", 0, 0}, + {18, "flapping", 0, 0}, + {19, "last_check", io::protobuf_base::invalid_on_zero, 0}, + {20, "last_hard_state", 0, 0}, + {21, "last_hard_state_change", io::protobuf_base::invalid_on_zero, 0}, + {22, "last_notification", io::protobuf_base::invalid_on_zero, 0}, + {23, "notification_number", 0, 0}, + {24, "last_state_change", io::protobuf_base::invalid_on_zero, 0}, + {25, "last_time_down", io::protobuf_base::invalid_on_zero, 0}, + {26, "last_time_unreachable", io::protobuf_base::invalid_on_zero, 0}, + {27, "last_time_up", io::protobuf_base::invalid_on_zero, 0}, + {28, "last_update", io::protobuf_base::invalid_on_zero, 0}, + {29, "latency", 0, 0}, + {30, "max_check_attempts", 0, 0}, + {31, "next_check", io::protobuf_base::invalid_on_zero, 0}, + {32, "next_host_notification", io::protobuf_base::invalid_on_zero, 0}, + {33, "no_more_notifications", 0, 0}, + {34, "notify", 0, 0}, + {35, "output", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_output)}, + {36, "passive_checks", 0, 0}, + {37, "percent_state_change", 0, 0}, + {38, "perfdata", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_perfdata)}, + {39, "retry_interval", 0, 0}, + {40, "should_be_scheduled", 0, 0}, + {41, "obsess_over_host", 0, 0}, + {42, "state_type", 0, 0}, + {43, "action_url", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_action_url)}, + {44, "address", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_address)}, + {45, "alias", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_alias)}, + {46, "check_freshness", 0, 0}, + {47, "default_active_checks", 0, 0}, + {48, "default_event_handler_enabled", 0, 0}, + {49, "default_flap_detection", 0, 0}, + {50, "default_notify", 0, 0}, + {51, "default_passive_checks", 0, 0}, + {52, "display_name", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_display_name)}, + {53, "first_notification_delay", 0, 0}, + {54, "flap_detection_on_down", 0, 0}, + {55, "flap_detection_on_unreachable", 0, 0}, + {56, "flap_detection_on_up", 0, 0}, + {57, "freshness_threshold", 0, 0}, + {58, "high_flap_threshold", 0, 0}, + {59, "name", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_name)}, + {60, "icon_image", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_icon_image)}, + {61, "icon_image_alt", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_icon_image_alt)}, + {62, "instance_id", mapping::entry::invalid_on_zero, 0}, + {63, "low_flap_threshold", 0, 0}, + {64, "notes", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_notes)}, + {65, "notes_url", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_notes_url)}, + {66, "notification_interval", 0, 0}, + {67, "notification_period", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_notification_period)}, + {68, "notify_on_down", 0, 0}, + {69, "notify_on_downtime", 0, 0}, + {70, "notify_on_flapping", 0, 0}, + {71, "notify_on_recovery", 0, 0}, + {72, "notify_on_unreachable", 0, 0}, + {73, "stalk_on_down", 0, 0}, + {74, "stalk_on_unreachable", 0, 0}, + {75, "stalk_on_up", 0, 0}, + {76, "statusmap_image", 0, + get_centreon_storage_hosts_col_size( + centreon_storage_hosts_statusmap_image)}, + {77, "retain_nonstatus_information", 0, 0}, + {78, "retain_status_information", 0, 0}, + {79, "timezone", 0, + get_centreon_storage_hosts_col_size(centreon_storage_hosts_timezone)}}; + + _pb_host_insupdate = + qp.prepare_insert_or_update_table(_mysql, "hosts", host_entries); + + query_preparator::event_pb_unique update_unique{{1, "host_id", 0, 0}, + {62, "instance_id", 0, 0}}; + query_preparator update_qp(neb::pb_host::static_type(), update_unique); + + _pb_host_update = + update_qp.prepare_update_table(_mysql, "hosts", host_entries); + + if (_store_in_resources) { + _resources_host_insert_or_update = _mysql.prepare_query( + "INSERT INTO resources " + "(id,parent_id,type,status,status_ordered,last_" + "status_change," + "in_downtime,acknowledged," + "status_confirmed,check_attempts,max_check_attempts," + "poller_id," + "severity_id,name,address,alias,parent_name,notes_url," + "notes," + "action_url," + "notifications_enabled,passive_checks_enabled," + "active_checks_enabled,enabled,icon_id," + "flapping,percent_state_change)" + "VALUES(?,0,1,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?" + ") ON DUPLICATE KEY UPDATE " + "resource_id=LAST_INSERT_ID(resource_id)," + " type=1" BOOST_PP_SEQ_FOR_EACH( + for_each_to_duplicate_values, + , (status)(status_ordered)(last_status_change)(in_downtime)(acknowledged)(status_confirmed)(check_attempts)(max_check_attempts)(poller_id)(severity_id)(name)(address)(alias)(parent_name)(notes_url)(notes)(action_url)(notifications_enabled)(passive_checks_enabled)(active_checks_enabled)(enabled)(icon_id)(flapping)(percent_state_change))); + } + } + if (!_pb_service_insupdate.prepared()) { + query_preparator::event_pb_unique unique{ + {1, "host_id", io::protobuf_base::invalid_on_zero, 0}, + {2, "service_id", io::protobuf_base::invalid_on_zero, 0}, + }; + query_preparator qp(neb::pb_service::static_type(), unique); + + std::vector service_entries = { + {1, "services.host_id", io::protobuf_base::invalid_on_zero, 0}, + {2, "services.service_id", io::protobuf_base::invalid_on_zero, 0}, + {3, "services.acknowledged", 0, 0}, + {4, "services.acknowledgement_type", 0, 0}, + {5, "services.active_checks", 0, 0}, + {6, "services.enabled", 0, 0}, + {7, "services.scheduled_downtime_depth", 0, 0}, + {8, "services.check_command", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_check_command)}, + {9, "services.check_interval", 0, 0}, + {10, "services.check_period", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_check_period)}, + {11, "services.check_type", 0, 0}, + {12, "services.check_attempt", 0, 0}, + {13, "services.state", 0, 0}, + {14, "services.event_handler_enabled", 0, 0}, + {15, "services.event_handler", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_event_handler)}, + {16, "services.execution_time", 0, 0}, + {17, "services.flap_detection", 0, 0}, + {18, "services.checked", 0, 0}, + {19, "services.flapping", 0, 0}, + {20, "services.last_check", io::protobuf_base::invalid_on_zero, 0}, + {21, "services.last_hard_state", 0, 0}, + {22, "services.last_hard_state_change", + io::protobuf_base::invalid_on_zero, 0}, + {23, "services.last_notification", io::protobuf_base::invalid_on_zero, + 0}, + {24, "services.notification_number", 0, 0}, + {25, "services.last_state_change", io::protobuf_base::invalid_on_zero, + 0}, + {26, "services.last_time_ok", io::protobuf_base::invalid_on_zero, 0}, + {27, "services.last_time_warning", io::protobuf_base::invalid_on_zero, + 0}, + {28, "services.last_time_critical", io::protobuf_base::invalid_on_zero, + 0}, + {29, "services.last_time_unknown", io::protobuf_base::invalid_on_zero, + 0}, + {30, "services.last_update", io::protobuf_base::invalid_on_zero, 0}, + {31, "services.latency", 0, 0}, + {32, "services.max_check_attempts", 0, 0}, + {33, "services.next_check", io::protobuf_base::invalid_on_zero, 0}, + {34, "services.next_notification", io::protobuf_base::invalid_on_zero, + 0}, + {35, "services.no_more_notifications", 0, 0}, + {36, "services.notify", 0, 0}, + {37, "services.output", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_output)}, + + {39, "services.passive_checks", 0, 0}, + {40, "services.percent_state_change", 0, 0}, + {41, "services.perfdata", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_perfdata)}, + {42, "services.retry_interval", 0, 0}, + + {44, "services.description", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_description)}, + {45, "services.should_be_scheduled", 0, 0}, + {46, "services.obsess_over_service", 0, 0}, + {47, "services.state_type", 0, 0}, + {48, "services.action_url", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_action_url)}, + {49, "services.check_freshness", 0, 0}, + {50, "services.default_active_checks", 0, 0}, + {51, "services.default_event_handler_enabled", 0, 0}, + {52, "services.default_flap_detection", 0, 0}, + {53, "services.default_notify", 0, 0}, + {54, "services.default_passive_checks", 0, 0}, + {55, "services.display_name", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_display_name)}, + {56, "services.first_notification_delay", 0, 0}, + {57, "services.flap_detection_on_critical", 0, 0}, + {58, "services.flap_detection_on_ok", 0, 0}, + {59, "services.flap_detection_on_unknown", 0, 0}, + {60, "services.flap_detection_on_warning", 0, 0}, + {61, "services.freshness_threshold", 0, 0}, + {62, "services.high_flap_threshold", 0, 0}, + {63, "services.icon_image", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_icon_image)}, + {64, "services.icon_image_alt", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_icon_image_alt)}, + {65, "services.volatile", 0, 0}, + {66, "services.low_flap_threshold", 0, 0}, + {67, "services.notes", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_notes)}, + {68, "services.notes_url", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_notes_url)}, + {69, "services.notification_interval", 0, 0}, + {70, "services.notification_period", 0, + get_centreon_storage_services_col_size( + centreon_storage_services_notification_period)}, + {71, "services.notify_on_critical", 0, 0}, + {72, "services.notify_on_downtime", 0, 0}, + {73, "services.notify_on_flapping", 0, 0}, + {74, "services.notify_on_recovery", 0, 0}, + {75, "services.notify_on_unknown", 0, 0}, + {76, "services.notify_on_warning", 0, 0}, + {77, "services.stalk_on_critical", 0, 0}, + {78, "services.stalk_on_ok", 0, 0}, + {79, "services.stalk_on_unknown", 0, 0}, + {80, "services.stalk_on_warning", 0, 0}, + {81, "services.retain_nonstatus_information", 0, 0}, + {82, "services.retain_status_information", 0, 0}}; + + _pb_service_insupdate = + qp.prepare_insert_or_update_table(_mysql, "services", service_entries); + + query_preparator::event_pb_unique update_unique{ + {1, "services.host_id", 0, 0}, + {2, "service_id", 0, 0}, + {88, "instance_id", 0, 0}}; + query_preparator update_qp(neb::pb_service::static_type(), update_unique); + _pb_service_update = update_qp.prepare_update_table( + _mysql, "services INNER JOIN hosts ON services.host_id = hosts.host_id", + service_entries); + + if (_store_in_resources) { + _resources_service_insert_or_update = _mysql.prepare_query( + "INSERT INTO resources " + "(id,parent_id,type,internal_id,status,status_" + "ordered,last_" + "status_change,in_downtime,acknowledged," + "status_confirmed,check_attempts,max_check_attempts,poller_" + "id," + "severity_id,name,parent_name,notes_url,notes,action_url," + "notifications_enabled,passive_checks_enabled,active_" + "checks_" + "enabled,enabled,icon_id, flapping, percent_state_change) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)" + "ON DUPLICATE KEY UPDATE " + "resource_id=LAST_INSERT_ID(resource_id)," + " type=1, status = VALUES(status)" BOOST_PP_SEQ_FOR_EACH(for_each_to_duplicate_values, + , (type)(internal_id)(status)(status_ordered)(last_status_change)(in_downtime)(acknowledged)(status_confirmed)(check_attempts)(max_check_attempts)(poller_id)(severity_id)(name)(parent_name)(notes_url)(notes)(action_url)(notifications_enabled)(passive_checks_enabled)(active_checks_enabled)(enabled)(icon_id)(flapping)(percent_state_change))); + } + } + if (!_resources_tags_remove.prepared()) + _resources_tags_remove = + _mysql.prepare_query("DELETE FROM resources_tags WHERE resource_id=?"); + if (!_resources_disable.prepared()) { + _resources_disable = _mysql.prepare_query( + "UPDATE resources SET enabled=0 WHERE resource_id=? AND " + "poller_id=?"); + } +} + /** * Process a host status protobuf event. * @@ -2089,10 +2422,10 @@ void stream::_process_pb_host(const std::shared_ptr& d) { auto& h = hst->obj(); // Log message. - SPDLOG_LOGGER_INFO( - _logger_sql, - "unified_sql: processing pb host event (poller: {}, host: {}, name: {})", - h.instance_id(), h.host_id(), h.name()); + SPDLOG_LOGGER_INFO(_logger_sql, + "unified_sql: processing pb host event (poller: {}, host: " + "{}, name: {}, enabled: {})", + h.instance_id(), h.host_id(), h.name(), h.enabled()); // Processing if (_is_valid_poller(h.instance_id())) { @@ -2103,169 +2436,31 @@ void stream::_process_pb_host(const std::shared_ptr& d) { int32_t conn = _mysql.choose_connection_by_instance(h.instance_id()); // Prepare queries. - if (!_pb_host_insupdate.prepared()) { - query_preparator::event_pb_unique unique{ - {1, "host_id", io::protobuf_base::invalid_on_zero, 0}}; - query_preparator qp(neb::pb_host::static_type(), unique); - _pb_host_insupdate = qp.prepare_insert_or_update_table( - _mysql, "hosts", - {{1, "host_id", io::protobuf_base::invalid_on_zero, 0}, - {2, "acknowledged", 0, 0}, - {3, "acknowledgement_type", 0, 0}, - {4, "active_checks", 0, 0}, - {5, "enabled", 0, 0}, - {6, "scheduled_downtime_depth", 0, 0}, - {7, "check_command", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_check_command)}, - {8, "check_interval", 0, 0}, - {9, "check_period", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_check_period)}, - {10, "check_type", 0, 0}, - {11, "check_attempt", 0, 0}, - {12, "state", 0, 0}, - {13, "event_handler_enabled", 0, 0}, - {14, "event_handler", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_event_handler)}, - {15, "execution_time", 0, 0}, - {16, "flap_detection", 0, 0}, - {17, "checked", 0, 0}, - {18, "flapping", 0, 0}, - {19, "last_check", io::protobuf_base::invalid_on_zero, 0}, - {20, "last_hard_state", 0, 0}, - {21, "last_hard_state_change", io::protobuf_base::invalid_on_zero, - 0}, - {22, "last_notification", io::protobuf_base::invalid_on_zero, 0}, - {23, "notification_number", 0, 0}, - {24, "last_state_change", io::protobuf_base::invalid_on_zero, 0}, - {25, "last_time_down", io::protobuf_base::invalid_on_zero, 0}, - {26, "last_time_unreachable", io::protobuf_base::invalid_on_zero, - 0}, - {27, "last_time_up", io::protobuf_base::invalid_on_zero, 0}, - {28, "last_update", io::protobuf_base::invalid_on_zero, 0}, - {29, "latency", 0, 0}, - {30, "max_check_attempts", 0, 0}, - {31, "next_check", io::protobuf_base::invalid_on_zero, 0}, - {32, "next_host_notification", io::protobuf_base::invalid_on_zero, - 0}, - {33, "no_more_notifications", 0, 0}, - {34, "notify", 0, 0}, - {35, "output", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_output)}, - {36, "passive_checks", 0, 0}, - {37, "percent_state_change", 0, 0}, - {38, "perfdata", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_perfdata)}, - {39, "retry_interval", 0, 0}, - {40, "should_be_scheduled", 0, 0}, - {41, "obsess_over_host", 0, 0}, - {42, "state_type", 0, 0}, - {43, "action_url", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_action_url)}, - {44, "address", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_address)}, - {45, "alias", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_alias)}, - {46, "check_freshness", 0, 0}, - {47, "default_active_checks", 0, 0}, - {48, "default_event_handler_enabled", 0, 0}, - {49, "default_flap_detection", 0, 0}, - {50, "default_notify", 0, 0}, - {51, "default_passive_checks", 0, 0}, - {52, "display_name", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_display_name)}, - {53, "first_notification_delay", 0, 0}, - {54, "flap_detection_on_down", 0, 0}, - {55, "flap_detection_on_unreachable", 0, 0}, - {56, "flap_detection_on_up", 0, 0}, - {57, "freshness_threshold", 0, 0}, - {58, "high_flap_threshold", 0, 0}, - {59, "name", 0, - get_centreon_storage_hosts_col_size(centreon_storage_hosts_name)}, - {60, "icon_image", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_icon_image)}, - {61, "icon_image_alt", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_icon_image_alt)}, - {62, "instance_id", mapping::entry::invalid_on_zero, 0}, - {63, "low_flap_threshold", 0, 0}, - {64, "notes", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_notes)}, - {65, "notes_url", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_notes_url)}, - {66, "notification_interval", 0, 0}, - {67, "notification_period", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_notification_period)}, - {68, "notify_on_down", 0, 0}, - {69, "notify_on_downtime", 0, 0}, - {70, "notify_on_flapping", 0, 0}, - {71, "notify_on_recovery", 0, 0}, - {72, "notify_on_unreachable", 0, 0}, - {73, "stalk_on_down", 0, 0}, - {74, "stalk_on_unreachable", 0, 0}, - {75, "stalk_on_up", 0, 0}, - {76, "statusmap_image", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_statusmap_image)}, - {77, "retain_nonstatus_information", 0, 0}, - {78, "retain_status_information", 0, 0}, - {79, "timezone", 0, - get_centreon_storage_hosts_col_size( - centreon_storage_hosts_timezone)}}); - if (_store_in_resources) { - _resources_host_insert_or_update = _mysql.prepare_query( - "INSERT INTO resources " - "(id,parent_id,type,status,status_ordered,last_" - "status_change," - "in_downtime,acknowledged," - "status_confirmed,check_attempts,max_check_attempts," - "poller_id," - "severity_id,name,address,alias,parent_name,notes_url," - "notes," - "action_url," - "notifications_enabled,passive_checks_enabled," - "active_checks_enabled,enabled,icon_id," - "flapping,percent_state_change)" - "VALUES(?,0,1,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?" - ") ON DUPLICATE KEY UPDATE " - "resource_id=LAST_INSERT_ID(resource_id)," - " type=1" BOOST_PP_SEQ_FOR_EACH( - for_each_to_duplicate_values, , - ( - status)(status_ordered)(last_status_change)(in_downtime)(acknowledged)(status_confirmed)(check_attempts)(max_check_attempts)(poller_id)(severity_id)(name)(address)(alias)(parent_name)(notes_url)(notes)(action_url)(notifications_enabled)(passive_checks_enabled)(active_checks_enabled)(enabled)(icon_id)(flapping)(percent_state_change))); - if (!_resources_tags_remove.prepared()) - _resources_tags_remove = _mysql.prepare_query( - "DELETE FROM resources_tags WHERE resource_id=?"); - if (!_resources_disable.prepared()) { - _resources_disable = _mysql.prepare_query( - "UPDATE resources SET enabled=0 WHERE resource_id=?"); - } - } - } - + _prepare_pb_requests(); // Process object. - _pb_host_insupdate << *hst; - _mysql.run_statement(_pb_host_insupdate, - database::mysql_error::store_host, conn); + if (h.enabled()) { + _pb_host_insupdate << *hst; + _mysql.run_statement(_pb_host_insupdate, + database::mysql_error::store_host, conn); + } else { + // when we disable a host, we care about instance_id in order to not + // modify a host yet moved to another poller and inserted + _pb_host_update << *hst; + _mysql.run_statement(_pb_host_update, database::mysql_error::store_host, + conn); + } _add_action(conn, actions::hosts); // Fill the cache... if (h.enabled()) _cache_host_instance[h.host_id()] = h.instance_id(); - else - _cache_host_instance.erase(h.host_id()); + else { + auto cache_to_delete = _cache_host_instance.find(h.host_id()); + if (cache_to_delete != _cache_host_instance.end() && + cache_to_delete->second == h.instance_id()) { + _cache_host_instance.erase(cache_to_delete); + } + } if (_store_in_resources) { _process_pb_host_in_resources(h, conn); @@ -2280,7 +2475,7 @@ void stream::_process_pb_host(const std::shared_ptr& d) { } uint64_t stream::_process_pb_host_in_resources(const Host& h, int32_t conn) { - auto found = _resource_cache.find({h.host_id(), 0}); + auto found = _resource_cache.find({h.host_id(), 0, h.instance_id()}); uint64_t res_id = 0; if (h.enabled()) { @@ -2361,7 +2556,7 @@ uint64_t stream::_process_pb_host_in_resources(const Host& h, int32_t conn) { _add_action(conn, actions::resources); try { res_id = future.get(); - _resource_cache.insert({{h.host_id(), 0}, res_id}); + _resource_cache.insert({{h.host_id(), 0, h.instance_id()}, res_id}); } catch (const std::exception& e) { SPDLOG_LOGGER_CRITICAL(_logger_sql, "SQL: unable to insert new host resource {}: {}", @@ -2391,15 +2586,17 @@ uint64_t stream::_process_pb_host_in_resources(const Host& h, int32_t conn) { } else { if (found != _resource_cache.end()) { _resources_disable.bind_value_as_u64(0, found->second); + _resources_disable.bind_value_as_u64(1, h.instance_id()); _mysql.run_statement(_resources_disable, database::mysql_error::clean_resources, conn); _resource_cache.erase(found); _add_action(conn, actions::resources); } else { - SPDLOG_LOGGER_INFO( - _logger_sql, "SQL: no need to remove host {}, it is not in database", - h.host_id()); + SPDLOG_LOGGER_INFO(_logger_sql, + "SQL: no need to remove host {}, it is not in " + "database or is not owned by poller {}", + h.host_id(), h.instance_id()); } } return res_id; @@ -3674,16 +3871,23 @@ void stream::_process_service_group_member(const std::shared_ptr& d) { "instance {}", sgm.host_id, sgm.service_id, sgm.group_id, sgm.poller_id); - if (!_service_group_member_delete.prepared()) { - query_preparator::event_unique unique; - unique.insert("servicegroup_id"); - unique.insert("host_id"); - unique.insert("service_id"); - query_preparator qp(neb::service_group_member::static_type(), unique); - _service_group_member_delete = qp.prepare_delete(_mysql); + if (!_service_group_member_delete) { + _service_group_member_delete = std::make_unique( + "DELETE services_servicegroups FROM services_servicegroups " + "LEFT JOIN hosts ON services_servicegroups.host_id=hosts.host_id " + "WHERE " + "services_servicegroups.servicegroup_id=? AND " + "services_servicegroups.host_id=? AND " + "services_servicegroups.service_id=? AND " + "(hosts.instance_id=? OR hosts.instance_id is NULL)"); + _mysql.prepare_statement(*_service_group_member_delete); } - _service_group_member_delete << sgm; - _mysql.run_statement(_service_group_member_delete, + + _service_group_member_delete->bind_value_as_u64(0, sgm.group_id); + _service_group_member_delete->bind_value_as_u64(1, sgm.host_id); + _service_group_member_delete->bind_value_as_u64(2, sgm.service_id); + _service_group_member_delete->bind_value_as_u64(3, sgm.poller_id); + _mysql.run_statement(*_service_group_member_delete, database::mysql_error::delete_service_group_member, conn); _add_action(conn, actions::servicegroups); @@ -3774,21 +3978,26 @@ void stream::_process_pb_service_group_member( "instance {}", sgm.host_id(), sgm.service_id(), sgm.servicegroup_id(), sgm.poller_id()); - - if (!_pb_service_group_member_delete.prepared()) { - query_preparator::event_pb_unique unique{ - {3, "servicegroup_id", io::protobuf_base::invalid_on_zero, 0}, - {5, "host_id", io::protobuf_base::invalid_on_zero, 0}, - {7, "service_id", io::protobuf_base::invalid_on_zero, 0}, - }; - query_preparator qp(neb::pb_service_group_member::static_type(), unique); - _pb_service_group_member_delete = - qp.prepare_delete_table(_mysql, "services_servicegroups "); + if (!_service_group_member_delete) { + _service_group_member_delete = std::make_unique( + "DELETE services_servicegroups FROM services_servicegroups " + "LEFT JOIN hosts ON services_servicegroups.host_id=hosts.host_id " + "WHERE " + "services_servicegroups.servicegroup_id=? AND " + "services_servicegroups.host_id=? AND " + "services_servicegroups.service_id=? AND " + "(hosts.instance_id=? OR hosts.instance_id is NULL)"); + _mysql.prepare_statement(*_service_group_member_delete); } - _pb_service_group_member_delete << sgmp; - _mysql.run_statement(_pb_service_group_member_delete, + + _service_group_member_delete->bind_value_as_u64(0, sgm.servicegroup_id()); + _service_group_member_delete->bind_value_as_u64(1, sgm.host_id()); + _service_group_member_delete->bind_value_as_u64(2, sgm.service_id()); + _service_group_member_delete->bind_value_as_u64(3, sgm.poller_id()); + _mysql.run_statement(*_service_group_member_delete, database::mysql_error::delete_service_group_member, conn); + _add_action(conn, actions::servicegroups); } } @@ -3886,153 +4095,18 @@ void stream::_process_pb_service(const std::shared_ptr& d) { if (s.host_id() && s.service_id()) { // Prepare queries. - if (!_pb_service_insupdate.prepared()) { - query_preparator::event_pb_unique unique{ - {1, "host_id", io::protobuf_base::invalid_on_zero, 0}, - {2, "service_id", io::protobuf_base::invalid_on_zero, 0}, - }; - query_preparator qp(neb::pb_service::static_type(), unique); - - _pb_service_insupdate = qp.prepare_insert_or_update_table( - _mysql, "services", - {{1, "host_id", io::protobuf_base::invalid_on_zero, 0}, - {2, "service_id", io::protobuf_base::invalid_on_zero, 0}, - {3, "acknowledged", 0, 0}, - {4, "acknowledgement_type", 0, 0}, - {5, "active_checks", 0, 0}, - {6, "enabled", 0, 0}, - {7, "scheduled_downtime_depth", 0, 0}, - {8, "check_command", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_check_command)}, - {9, "check_interval", 0, 0}, - {10, "check_period", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_check_period)}, - {11, "check_type", 0, 0}, - {12, "check_attempt", 0, 0}, - {13, "state", 0, 0}, - {14, "event_handler_enabled", 0, 0}, - {15, "event_handler", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_event_handler)}, - {16, "execution_time", 0, 0}, - {17, "flap_detection", 0, 0}, - {18, "checked", 0, 0}, - {19, "flapping", 0, 0}, - {20, "last_check", io::protobuf_base::invalid_on_zero, 0}, - {21, "last_hard_state", 0, 0}, - {22, "last_hard_state_change", io::protobuf_base::invalid_on_zero, - 0}, - {23, "last_notification", io::protobuf_base::invalid_on_zero, 0}, - {24, "notification_number", 0, 0}, - {25, "last_state_change", io::protobuf_base::invalid_on_zero, 0}, - {26, "last_time_ok", io::protobuf_base::invalid_on_zero, 0}, - {27, "last_time_warning", io::protobuf_base::invalid_on_zero, 0}, - {28, "last_time_critical", io::protobuf_base::invalid_on_zero, 0}, - {29, "last_time_unknown", io::protobuf_base::invalid_on_zero, 0}, - {30, "last_update", io::protobuf_base::invalid_on_zero, 0}, - {31, "latency", 0, 0}, - {32, "max_check_attempts", 0, 0}, - {33, "next_check", io::protobuf_base::invalid_on_zero, 0}, - {34, "next_notification", io::protobuf_base::invalid_on_zero, 0}, - {35, "no_more_notifications", 0, 0}, - {36, "notify", 0, 0}, - {37, "output", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_output)}, - - {39, "passive_checks", 0, 0}, - {40, "percent_state_change", 0, 0}, - {41, "perfdata", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_perfdata)}, - {42, "retry_interval", 0, 0}, - - {44, "description", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_description)}, - {45, "should_be_scheduled", 0, 0}, - {46, "obsess_over_service", 0, 0}, - {47, "state_type", 0, 0}, - {48, "action_url", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_action_url)}, - {49, "check_freshness", 0, 0}, - {50, "default_active_checks", 0, 0}, - {51, "default_event_handler_enabled", 0, 0}, - {52, "default_flap_detection", 0, 0}, - {53, "default_notify", 0, 0}, - {54, "default_passive_checks", 0, 0}, - {55, "display_name", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_display_name)}, - {56, "first_notification_delay", 0, 0}, - {57, "flap_detection_on_critical", 0, 0}, - {58, "flap_detection_on_ok", 0, 0}, - {59, "flap_detection_on_unknown", 0, 0}, - {60, "flap_detection_on_warning", 0, 0}, - {61, "freshness_threshold", 0, 0}, - {62, "high_flap_threshold", 0, 0}, - {63, "icon_image", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_icon_image)}, - {64, "icon_image_alt", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_icon_image_alt)}, - {65, "volatile", 0, 0}, - {66, "low_flap_threshold", 0, 0}, - {67, "notes", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_notes)}, - {68, "notes_url", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_notes_url)}, - {69, "notification_interval", 0, 0}, - {70, "notification_period", 0, - get_centreon_storage_services_col_size( - centreon_storage_services_notification_period)}, - {71, "notify_on_critical", 0, 0}, - {72, "notify_on_downtime", 0, 0}, - {73, "notify_on_flapping", 0, 0}, - {74, "notify_on_recovery", 0, 0}, - {75, "notify_on_unknown", 0, 0}, - {76, "notify_on_warning", 0, 0}, - {77, "stalk_on_critical", 0, 0}, - {78, "stalk_on_ok", 0, 0}, - {79, "stalk_on_unknown", 0, 0}, - {80, "stalk_on_warning", 0, 0}, - {81, "retain_nonstatus_information", 0, 0}, - {82, "retain_status_information", 0, 0}}); - if (_store_in_resources) { - _resources_service_insert_or_update = _mysql.prepare_query( - "INSERT INTO resources " - "(id,parent_id,type,internal_id,status,status_" - "ordered,last_" - "status_change,in_downtime,acknowledged," - "status_confirmed,check_attempts,max_check_attempts,poller_" - "id," - "severity_id,name,parent_name,notes_url,notes,action_url," - "notifications_enabled,passive_checks_enabled,active_" - "checks_" - "enabled,enabled,icon_id, flapping, percent_state_change) " - "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)" - "ON DUPLICATE KEY UPDATE " - "resource_id=LAST_INSERT_ID(resource_id)," - " type=1, status = VALUES(status)" BOOST_PP_SEQ_FOR_EACH(for_each_to_duplicate_values, - , - (type)(internal_id)(status)(status_ordered)(last_status_change)(in_downtime)(acknowledged)(status_confirmed)(check_attempts)(max_check_attempts)(poller_id)(severity_id)(name)(parent_name)(notes_url)(notes)(action_url)(notifications_enabled)(passive_checks_enabled)(active_checks_enabled)(enabled)(icon_id)(flapping)(percent_state_change))); - if (!_resources_disable.prepared()) { - _resources_disable = _mysql.prepare_query( - "UPDATE resources SET enabled=0 WHERE resource_id=?"); - } - } - } + _prepare_pb_requests(); // Process object. - _pb_service_insupdate << *svc; - _mysql.run_statement(_pb_service_insupdate, - database::mysql_error::store_service, conn); + if (s.enabled()) { + _pb_service_insupdate << *svc; + _mysql.run_statement(_pb_service_insupdate, + database::mysql_error::store_service, conn); + } else { + _pb_service_update << *svc; + _mysql.run_statement(_pb_service_update, + database::mysql_error::store_service, conn); + } _add_action(conn, actions::services); _check_and_update_index_cache(s); @@ -4052,7 +4126,8 @@ uint64_t stream::_process_pb_service_in_resources(const Service& s, int32_t conn) { uint64_t res_id = 0; - auto found = _resource_cache.find({s.service_id(), s.host_id()}); + auto found = + _resource_cache.find({s.service_id(), s.host_id(), s.instance_id()}); if (s.enabled()) { uint64_t sid = 0; @@ -4133,7 +4208,8 @@ uint64_t stream::_process_pb_service_in_resources(const Service& s, _add_action(conn, actions::resources); try { res_id = future.get(); - _resource_cache.insert({{s.service_id(), s.host_id()}, res_id}); + _resource_cache.insert( + {{s.service_id(), s.host_id(), s.instance_id()}, res_id}); } catch (const std::exception& e) { SPDLOG_LOGGER_CRITICAL( _logger_sql, @@ -4164,6 +4240,7 @@ uint64_t stream::_process_pb_service_in_resources(const Service& s, } else { if (found != _resource_cache.end()) { _resources_disable.bind_value_as_u64(0, found->second); + _resources_disable.bind_value_as_u64(1, s.instance_id()); _mysql.run_statement(_resources_disable, database::mysql_error::clean_resources, conn); @@ -4173,8 +4250,8 @@ uint64_t stream::_process_pb_service_in_resources(const Service& s, SPDLOG_LOGGER_INFO( _logger_sql, "SQL: no need to remove service ({}, {}), it is not in " - "database", - s.host_id(), s.service_id()); + "database or is not owned by poller {}", + s.host_id(), s.service_id(), s.instance_id()); } } return res_id; diff --git a/tests/broker-engine/hostgroups.robot b/tests/broker-engine/hostgroups.robot index a257074c2d2..79a7d6c33c0 100644 --- a/tests/broker-engine/hostgroups.robot +++ b/tests/broker-engine/hostgroups.robot @@ -364,3 +364,64 @@ EBNHG5 ${result} Ctn Check Number Of Relations Between Hostgroup And Hosts 1 6 30 Should Be True ${result} We should have 6 hosts members in the hostgroup 1. + +EBNHG6 + [Documentation] Scenario: Two pollers are connected, hostgroup 1 is defined on poller 0 with host_1 and + ... host_2 as members. Both hosts are then moved from poller 0's configuration to poller 1's + ... configuration, poller 1 is reloaded first, then poller 0. The hostgroup and its host + ... memberships must not disappear from the database. + [Tags] broker engine hostgroup unified_sql MON-169103 + Ctn Config Engine ${2} ${5} ${0} + Ctn Config Broker rrd + Ctn Config Broker central + Ctn Config Broker module ${2} + Ctn Config BBDO3 ${2} + + Ctn Broker Config Log central sql trace + Ctn Add Host Group ${0} ${1} ["host_1", "host_2"] + + ${start} Get Current Date + Ctn Start Broker + Ctn Start Engine + + Ctn Wait For Engine To Be Ready ${start} ${2} + + ${result} Ctn Check Number Of Relations Between Hostgroup And Hosts 1 2 30 + Should Be True ${result} We should have 2 hosts members in the hostgroup 1 before the move. + + Connect To Database pymysql ${DBName} ${DBUser} ${DBPass} ${DBHost} ${DBPort} + ${output} Query SELECT count(*) FROM hostgroups WHERE hostgroup_id = 1 + Should Be Equal As Strings ${output} ((1,),) The hostgroup 1 should exist before the move. + + # Move host_1 and host_2 from poller 0's configuration to poller 1's configuration. + Ctn Engine Config Move Host To Engine 0 1 host_1 + Ctn Engine Config Move Host To Engine 0 1 host_2 + + Ctn Add Host Group ${1} ${1} ["host_1", "host_2"] + + # Reload the second poller first, then the first poller. + Ctn Reload Engine 1 + Sleep 1 + Ctn Reload Engine 0 + + Sleep 5 + + ${result} Ctn Check Number Of Relations Between Hostgroup And Hosts 1 2 30 + Should Be True + ... ${result} + ... host_1 and host_2 should still be members of hostgroup 1 in hosts_hostgroups after the move. + + ${output} Query SELECT count(*) FROM hostgroups WHERE hostgroup_id = 1 + Should Be Equal As Strings ${output} ((1,),) The hostgroup 1 should still exist after the move. + + ${output} Query SELECT instance_id FROM hosts WHERE name='host_1' AND enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... host_1 should be enabled and owned by poller 1 (instance_id=2) after the move. + + ${output} Query SELECT instance_id FROM hosts WHERE name='host_2' AND enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... host_2 should be enabled and owned by poller 1 (instance_id=2) after the move. diff --git a/tests/broker-engine/servicegroups.robot b/tests/broker-engine/servicegroups.robot index 69a98f85d2c..36cd6bb579a 100644 --- a/tests/broker-engine/servicegroups.robot +++ b/tests/broker-engine/servicegroups.robot @@ -263,5 +263,84 @@ EBSG_1 ${result} Ctn Check Number Of Relations Between Servicegroup And Services 1 6 30 Should Be True ${result} We should get 3 relations between the servicegroup 1 and services. +EBNSG2 + [Documentation] Scenario: Two pollers are connected, servicegroup 1 is defined on poller 0 with + ... host_1/service_1 and host_2/service_2 as members. Both hosts and their services are then + ... moved from poller 0's configuration to poller 1's configuration, poller 1 is reloaded first, + ... then poller 0. The servicegroup and its service memberships must not disappear from the + ... database. + [Tags] broker engine servicegroup unified_sql MON-169103 + Ctn Config Engine ${2} ${5} ${1} + Ctn Config Broker rrd + Ctn Config Broker central + Ctn Config Broker module ${2} + Ctn Config BBDO3 ${2} + + Ctn Broker Config Log central sql trace + Ctn Add Service Group ${0} ${1} ["host_1","service_1","host_2","service_2"] + Ctn Config Engine Add Cfg File ${0} servicegroups.cfg + + ${start} Get Current Date + Ctn Start Broker + Ctn Start Engine + + Ctn Wait For Engine To Be Ready ${start} ${2} + + ${result} Ctn Check Number Of Relations Between Servicegroup And Services 1 2 30 + Should Be True ${result} We should have 2 services members in the servicegroup 1 before the move. + + Connect To Database pymysql ${DBName} ${DBUser} ${DBPass} ${DBHost} ${DBPort} + ${output} Query SELECT count(*) FROM servicegroups WHERE servicegroup_id = 1 + Should Be Equal As Strings ${output} ((1,),) The servicegroup 1 should exist before the move. + + # Move host_1 and host_2 with their services from poller 0's configuration to poller 1's configuration. + Ctn Engine Config Move Host To Engine 0 1 host_1 + Ctn Engine Config Move Services To Engine 0 1 host_1 + Ctn Engine Config Move Host To Engine 0 1 host_2 + Ctn Engine Config Move Services To Engine 0 1 host_2 + + Ctn Add Service Group ${1} ${1} ["host_1","service_1","host_2","service_2"] + Ctn Config Engine Add Cfg File ${1} servicegroups.cfg + + # Reload the second poller first, then the first poller. + Ctn Reload Engine 1 + Sleep 1 + Ctn Reload Engine 0 + + Sleep 5 + + ${result} Ctn Check Number Of Relations Between Servicegroup And Services 1 2 30 + Should Be True + ... ${result} + ... host_1/service_1 and host_2/service_2 should still be members of servicegroup 1 in services_servicegroups after the move. + + ${output} Query SELECT count(*) FROM servicegroups WHERE servicegroup_id = 1 + Should Be Equal As Strings ${output} ((1,),) The servicegroup 1 should still exist after the move. + + ${output} Query SELECT instance_id FROM hosts WHERE name='host_1' AND enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... host_1 should be enabled and owned by poller 1 (instance_id=2) after the move. + + ${output} Query SELECT instance_id FROM hosts WHERE name='host_2' AND enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... host_2 should be enabled and owned by poller 1 (instance_id=2) after the move. + + ${output} Query + ... SELECT h.instance_id FROM services s JOIN hosts h ON s.host_id=h.host_id WHERE h.name='host_1' AND s.description='service_1' AND s.enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... service_1 on host_1 should be enabled and owned by poller 1 (instance_id=2) after the move. + + ${output} Query + ... SELECT h.instance_id FROM services s JOIN hosts h ON s.host_id=h.host_id WHERE h.name='host_2' AND s.description='service_2' AND s.enabled=1 + Should Be Equal As Strings + ... ${output} + ... ((2,),) + ... service_2 on host_2 should be enabled and owned by poller 1 (instance_id=2) after the move.