fix(sc-event): replay status-change events after a downtime ends

.github/actions/test-packages/action.yml

@@ -166,8 +166,11 @@ runs:
         # Select the correct repo and version depending on distrib
         # Format <= 24.10: https://packages.centreon.com/{repo}-{stability}/ {distrib} main
         # Format >= 25.10: https://packages.centreon.com/{repo}/ {distrib}-{version}-{stability} main
+        use_standard_stable=true
+        use_standard_testing=true
+        use_standard_unstable=true
         if [[ "$DISTRIB" == "jammy" ]]; then
-          standard_base="ubuntu-standard-24.04"
+          standard_base="ubuntu-standard-24.10"
           standard_distrib="$DISTRIB"
           plugins_prefix="ubuntu-plugins"
           new_repo_format=false
@@ -181,6 +184,8 @@ runs:
           standard_distrib="$DISTRIB"
           plugins_prefix="apt-plugins"
           new_repo_format=false
+          use_standard_testing=false
+          use_standard_unstable=false
         elif [[ "$DISTRIB" == "bookworm" ]]; then
           standard_base="apt-standard"
           standard_distrib="${DISTRIB}-25.10"
@@ -195,13 +200,25 @@ runs:
         # Add Centreon standard repository (provides centreon-broker-core and other core packages)
         if [[ "$TEST_TYPE" != "dependency" ]]; then
           if [[ "$new_repo_format" == "true" ]]; then
-            echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-stable main" | tee /etc/apt/sources.list.d/centreon-standard.list
-            echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-testing main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
-            echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-unstable main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            if [[ "$use_standard_stable" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-stable main" | tee /etc/apt/sources.list.d/centreon-standard.list
+            fi
+            if [[ "$use_standard_testing" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-testing main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            fi
+            if [[ "$use_standard_unstable" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}/ ${standard_distrib}-unstable main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            fi
           else
-            echo "deb https://packages.centreon.com/${standard_base}-stable/ ${standard_distrib} main" | tee /etc/apt/sources.list.d/centreon-standard.list
-            echo "deb https://packages.centreon.com/${standard_base}-testing/ ${standard_distrib} main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
-            echo "deb https://packages.centreon.com/${standard_base}-unstable/ ${standard_distrib} main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            if [[ "$use_standard_stable" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}-stable/ ${standard_distrib} main" | tee /etc/apt/sources.list.d/centreon-standard.list
+            fi
+            if [[ "$use_standard_testing" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}-testing/ ${standard_distrib} main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            fi
+            if [[ "$use_standard_unstable" == "true" ]]; then
+              echo "deb https://packages.centreon.com/${standard_base}-unstable/ ${standard_distrib} main" | tee -a /etc/apt/sources.list.d/centreon-standard.list
+            fi
           fi
         fi
         # Add Centreon plugins repositories (stable, testing, unstable)

centreon-certified/canopsis/canopsis2x-events-apiv2.lua

@@ -631,6 +631,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/datadog/datadog-events-apiv2.lua

@@ -297,6 +297,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/elasticsearch/elastic-events-apiv2.lua

@@ -286,6 +286,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/google/bigquery-events-apiv2.lua

@@ -384,6 +384,9 @@ local queue
 
 function init(params)
   queue = EventQueue.new(params)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 function write(event)

centreon-certified/kafka/kafka-events-apiv2.lua

@@ -243,6 +243,9 @@ local queue
 
 function init(params)
   queue = EventQueue.new(params)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/keep/keep-events-apiv2.lua

@@ -471,6 +471,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/logstash/logstash-events-apiv2.lua

@@ -283,6 +283,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 --------------------------------------------------------------------------------

centreon-certified/omi/omi_events-apiv2.lua

@@ -312,6 +312,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- Fonction write()

centreon-certified/opsgenie/opsgenie-events-apiv2.lua

@@ -424,6 +424,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/pagerduty/pagerduty-events-apiv2.lua

@@ -419,6 +419,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/servicenow/servicenow-em-events-apiv2.lua

@@ -375,6 +375,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 --------------------------------------------------------------------------------

centreon-certified/servicenow/servicenow-incident-events-apiv2.lua

@@ -384,6 +384,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 --------------------------------------------------------------------------------

centreon-certified/signl4/signl4-events-apiv2.lua

@@ -282,6 +282,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 -- --------------------------------------------------------------------------------

centreon-certified/splunk/splunk-events-apiv2.lua

@@ -280,6 +280,9 @@ local queue
 -- Fonction init()
 function init(conf)
   queue = EventQueue.new(conf)
+  sc_event.set_pending_event_handler(function(pending_broker_event)
+    write(pending_broker_event)
+  end)
 end
 
 --------------------------------------------------------------------------------

modules/centreon-stream-connectors-lib/sc_event.lua

@@ -14,6 +14,15 @@ local sc_broker = require("centreon-stream-connectors-lib.sc_broker")
 local sc_storage = require("centreon-stream-connectors-lib.sc_storage")
 
 local ScEvent = {}
+local pending_event_handler = nil
+
+--- sc_event.set_pending_event_handler: register a callback called by the library when a stored event
+-- (saved during a downtime) is ready to be sent because the downtime ended with a status change.
+-- The callback receives an already-validated sc_event object.
+-- @param handler (function) function(sc_event_obj) called for each valid pending event
+function sc_event.set_pending_event_handler(handler)
+  pending_event_handler = handler
+end
 
 function sc_event.new(broker_event, params, common, logger, broker, storage)
   local self = {}
@@ -42,10 +51,10 @@ function sc_event.new(broker_event, params, common, logger, broker, storage)
   self.validation_steps = {}
 
   for accepted_element, info in pairs(self.params.accepted_elements_info) do
-    if not self.validation_steps[info.category_id] then 
+    if not self.validation_steps[info.category_id] then
       self.validation_steps[info.category_id] = {}
     end
-    
+
     self.validation_steps[info.category_id][info.element_id] = {}
   end
 
@@ -55,15 +64,64 @@ end
 
 
 --- is_valid_category: check if the event is in an accepted category
--- @retun true|false (boolean)
+-- @return true|false (boolean)
 function ScEvent:is_valid_category()
   return self:find_in_mapping(self.params.category_mapping, self.params.accepted_categories, self.event.category)
 end
 
 --- is_valid_element: check if the event is an accepted element
 -- @return true|false (boolean)
 function ScEvent:is_valid_element()
-  return self:find_in_mapping(self.params.element_mapping[self.event.category], self.params.accepted_elements, self.event.element)
+  local is_valid_element = false
+  is_valid_element = self:find_in_mapping(self.params.element_mapping[self.event.category], self.params.accepted_elements, self.event.element)
+  if self.event.element == self.params.bbdo.elements.downtime.id and self.params.in_downtime == 0 then
+    local object_id
+    if self.event.type == 1 then
+      object_id = 'downtime_service_' .. self.event.host_id .. '_' .. self.event.service_id
+    elseif self.event.type == 2 then
+      object_id = 'downtime_host_' .. self.event.host_id
+    else
+      self.sc_logger:error("[sc_event:is_valid_element]: unknown downtime type: " .. tostring(self.event.type))
+      return is_valid_element
+    end
+    if self:is_valid_downtime_event_start() then
+      local status = -1
+      if self.event.type == 1 then
+        status = broker_cache:get_service(self.event.host_id, self.event.service_id).state
+      elseif self.event.type == 2 then
+        status = broker_cache:get_host(self.event.host_id).state
+      end
+      local storage_data = {
+        object_type = self.event.type,
+        status = status,
+        downtime_start = self.event.actual_start_time,
+        downtime_end = self.event.actual_end_time
+      }
+      if not self.sc_storage:set_multiple(object_id, storage_data) then
+        self.sc_logger:error("[sc_event:is_valid_element]: Cannot register downtime datas in storage.")
+      end
+    elseif self:is_valid_downtime_event_end() then
+      local ok, stored = self.sc_storage:get_multiple(object_id, {"object_type", "status", "broker_event"})
+      if ok then
+        if stored.broker_event and pending_event_handler then
+          -- delete broker_event from storage before sending to prevent duplicate dispatch
+          -- on the second downtime end event (cancellation + deletion both trigger this path)
+          self.sc_storage:delete(object_id, "broker_event")
+          -- broker_event is stored as a JSON string: decode it explicitly
+          local broker_event = broker.json_decode(stored.broker_event)
+          if broker_event then
+            broker_event.scheduled_downtime_depth = 0
+            pending_event_handler(broker_event)
+          else
+            self.sc_logger:error("[sc_event:is_valid_element]: failed to decode stored broker_event")
+          end
+        end
+      else
+        self.sc_logger:error("[sc_event:is_valid_element]: Cannot get downtime datas from storage.")
+      end
+    end
+  end
+  return is_valid_element
 end
 
 --- find_in_mapping: check if item type is in the mapping and is accepted
@@ -79,7 +137,6 @@ function ScEvent:find_in_mapping(mapping, reference, item)
       end
     end
   end
-
   return false
 end
 
@@ -741,9 +798,26 @@ function ScEvent:is_valid_event_downtime_state()
     self.event.scheduled_downtime_depth = self.event.downtime_depth
   end
 
-  if not self.sc_common:compare_numbers(self.params.in_downtime, self.event.scheduled_downtime_depth, ">=") then
-    self.sc_logger:warning("[sc_event:is_valid_event_downtime_state]: event is not in an valid downtime state. Event downtime state must be below or equal to " .. tostring(self.params.in_downtime) 
-      .. ". Current downtime state: " .. tostring(self.sc_common:boolean_to_number(self.event.scheduled_downtime_depth)))
+  if self.params.in_downtime == 0 and self.event.scheduled_downtime_depth > 0 then
+    local object_id
+    if self.event.service_id and self.event.service_id ~= 0 then
+      object_id = 'downtime_service_' .. self.event.host_id .. '_' .. self.event.service_id
+    else
+      object_id = 'downtime_host_' .. self.event.host_id
+    end
+    local ok, stored = self.sc_storage:get_multiple(object_id, {"object_type", "status"})
+    if ok then
+      if stored.status ~= self.event.state then
+        -- store broker_event as explicit JSON string to avoid relying on storage backend type conversion
+        if not self.sc_storage:set(object_id, "broker_event", broker.json_encode(self.broker_event)) then
+          self.sc_logger:error("[sc_event:is_valid_event_downtime_state]: event hasn't been stored in storage")
+        end
+      else
+        self.sc_storage:set(object_id, "broker_event", nil)
+      end
+    end
+    self.sc_logger:warning("[sc_event:is_valid_event_downtime_state]: event is not in a valid downtime state. Event downtime depth must be equal to " .. tostring(self.params.in_downtime)
+      .. ". Current downtime depth value: " .. tostring(self.sc_common:boolean_to_number(self.event.scheduled_downtime_depth)))
     return false
   end
 
@@ -1244,8 +1318,6 @@ function ScEvent:is_valid_service_severity()
     return true
   end
 
-
-
   -- return false if service severity doesn't match 
   if not self.sc_common:compare_numbers(self.params.service_severity_threshold, self.event.cache.severity.service, self.params.service_severity_operator) then
     self.sc_logger:debug("[sc_event:is_valid_service_severity]: dropping event because service with id: " .. tostring(self.event.service_id) .. " has an invalid severity. Severity is: "
@@ -1341,7 +1413,7 @@ function ScEvent:is_valid_acknowledgement_event()
   return true
 end
 
---- is_vaid_downtime_event: check if the event is a valid downtime event
+--- is_valid_downtime_event: check if the event is a valid downtime event
 -- return true|false (boolean)
 function ScEvent:is_valid_downtime_event()
   -- return false if the event is one of all the "fake" start or end downtime event received from broker
@@ -1613,12 +1685,10 @@ function ScEvent:is_downtime_event_useless()
   if self:is_valid_downtime_event_start() then
     return true
   end
-
   -- return false if downtime event is not a valid end of downtime event
   if self:is_valid_downtime_event_end() then
     return true
   end
-
   return false
 end
 

modules/centreon-stream-connectors-lib/sc_storage.lua

@@ -24,18 +24,20 @@ function sc_storage.new(common, logger, params)
     "host_%d+",
     "service_%d+_%d+",
     "ba_%d+",
-    "metric_.*"
+    "metric_.*",
+    "downtime_host_%d+",
+    "downtime_service_%d+_%d+"
   }
 
-  -- make sure we are able to load the desired storage backend. If not, fall back to the one provided by broker
+  -- make sure we are able to load the desired storage backend. If not, fall back to the one provided by sqlite
   if pcall(require, "centreon-stream-connectors-lib.storage_backends.sc_storage_" .. params.storage_backend) then
     local storage_backend = require("centreon-stream-connectors-lib.storage_backends.sc_storage_" .. params.storage_backend)
     self.storage_backend = storage_backend.new(self.sc_common, logger, params)
   else
     self.sc_logger:error("[sc_storage:new]: Couldn't load storage backend: " .. tostring(params.storage_backend)
       .. ". Make sure that the file sc_storage_" .. tostring(params.storage_backend) .. ".lua exists on your server."
-      .. " The stream connector is going to use the broker storage backend.")
-    self.storage_backend = require("centreon-stream-connectors-lib.storage_backends.sc_storage_broker")
+      .. " The stream connector is going to use the sqlite storage backend.")
+    self.storage_backend = require("centreon-stream-connectors-lib.storage_backends.sc_storage_sqlite")
   end
 
   setmetatable(self, { __index = ScStorage})