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

sdepassio · sdepassio · commit 80edc3351ff3 · 2026-06-25T11:39:36.000+02:00
diff --git a/centreon-certified/canopsis/canopsis2x-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/datadog/datadog-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/elasticsearch/elastic-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/google/bigquery-events-apiv2.lua b/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)
diff --git a/centreon-certified/kafka/kafka-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/keep/keep-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/logstash/logstash-events-apiv2.lua b/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
 
 --------------------------------------------------------------------------------
diff --git a/centreon-certified/omi/omi_events-apiv2.lua b/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()
diff --git a/centreon-certified/opsgenie/opsgenie-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/pagerduty/pagerduty-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/servicenow/servicenow-em-events-apiv2.lua b/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
 
 --------------------------------------------------------------------------------
diff --git a/centreon-certified/servicenow/servicenow-incident-events-apiv2.lua b/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
 
 --------------------------------------------------------------------------------
diff --git a/centreon-certified/signl4/signl4-events-apiv2.lua b/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
 
 -- --------------------------------------------------------------------------------
diff --git a/centreon-certified/splunk/splunk-events-apiv2.lua b/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
 
 --------------------------------------------------------------------------------
diff --git a/modules/centreon-stream-connectors-lib/sc_event.lua b/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,67 @@ 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
+    if not self.downtimes_storage then
+      self.downtimes_storage = sc_storage.new(self.sc_common, self.sc_logger, self.params)
+    end
+    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.downtimes_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.downtimes_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.downtimes_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 +140,6 @@ function ScEvent:find_in_mapping(mapping, reference, item)
       end
     end
   end
-
   return false
 end
 
@@ -741,9 +801,29 @@ 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
+    if not self.downtimes_storage then
+      self.downtimes_storage = sc_storage.new(self.sc_common, self.sc_logger, self.params)
+    end
+    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.downtimes_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.downtimes_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.downtimes_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 +1324,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 +1419,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 +1691,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
 
diff --git a/modules/centreon-stream-connectors-lib/sc_storage.lua b/modules/centreon-stream-connectors-lib/sc_storage.lua
@@ -24,7 +24,9 @@ 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
