Host alerts migration to new api

scripts/lua/modules/alerts_api.lua

@@ -6,9 +6,13 @@ package.path = dirs.installdir .. "/scripts/lua/modules/?.lua;" .. package.path
 
 local json = require("dkjson")
 local alert_endpoints = require("alert_endpoints_utils")
+local alert_consts = require("alert_consts")
 
 local alerts = {}
 
+local MAX_NUM_ENQUEUED_ALERTS_EVENTS = 100
+local ALERTS_EVENTS_QUEUE = "ntopng.cache.alerts_events_queue"
+
 -- Just helpers
 local str_2_periodicity = {
   ["min"]     = 60,
@@ -201,4 +205,186 @@ end
 
 -- ##############################################
 
+function get_alert_triggered_key(type_info)
+  return(string.format("%d_%s", type_info.alert_type.alert_id, type_info.alert_subtype or ""))
+end
+
+-- ##############################################
+
+local function enqueueAlertEvent(alert_event)
+  local event_json = json.encode(alert_event)
+
+  ntop.rpushCache(ALERTS_EVENTS_QUEUE, event_json, MAX_NUM_ENQUEUED_ALERTS_EVENTS)
+
+  return(true)
+end
+
+-- ##############################################
+
+-- Performs the trigger/release asynchronously.
+-- This is necessary both to avoid paying the database io cost inside
+-- the other scripts and as a necessity to avoid a deadlock on the
+-- host hash in the host.lua script
+function alerts.processPendingAlertEvents(deadline)
+  while(true) do
+    local event_json = ntop.lpopCache(ALERTS_EVENTS_QUEUE)
+
+    if(not event_json) then
+      break
+    end
+
+    local event = json.decode(event_json)
+    local to_call
+
+    interface.select(tostring(event.ifid))
+
+    if(event.action == "release") then
+      to_call = interface.releaseAlert
+    else
+      to_call = interface.triggerAlert
+    end
+
+    rv = to_call(
+      event.tstamp, event.granularity,
+      event.type, event.severity,
+      event.entity_type, event.entity_value,
+      event.message, event.subtype)
+
+    if(rv.success) then
+      alert_endpoints.dispatchNotification(event, event_json)
+    end
+
+    if(os.time() > deadline) then
+      break
+    end
+  end
+end
+
+-- ##############################################
+
+-- TODO: remove the "new_" prefix and unify with other alerts
+
+--! @brief Trigger an alert of given type on the entity
+--! @param entity_info data returned by one of the entity_info building functions
+--! @param type_info data returned by one of the type_info building functions
+--! @param when (optional) the time when the release event occurs
+--! @note The actual trigger is performed asynchronously
+--! @return true on success, false otherwise
+function alerts.new_trigger(entity_info, type_info, when)
+  when = when or os.time()
+  local granularity_sec = type_info.alert_granularity and type_info.alert_granularity.granularity_seconds or 0
+  local granularity_id = type_info.alert_granularity and type_info.alert_granularity.granularity_id or nil
+
+  if(granularity_id ~= nil) then
+    local triggered = true
+    local alert_key_name = get_alert_triggered_key(type_info)
+
+    if((not table.empty(host)) and (entity_info.alert_entity.entity_id == alertEntity("host"))) then
+      triggered = host.storeTriggeredAlert(alert_key_name, granularity_id)
+    end
+
+    if(not triggered) then
+      return(false)
+    end
+  end
+
+  local alert_json = json.encode(type_info.alert_type_params)
+  local action = ternary((granularity_id ~= nil), "engaged", "stored")
+
+  local alert_event = {
+    ifid = interface.getId(),
+    granularity = granularity_sec,
+    entity_type = entity_info.alert_entity.entity_id,
+    entity_value = entity_info.alert_entity_val,
+    type = type_info.alert_type.alert_id,
+    severity = type_info.alert_type.severity.severity_id,
+    message = alert_json,
+    subtype = type_info.alert_subtype or "",
+    tstamp = when,
+    action = action,
+  }
+
+  return(enqueueAlertEvent(alert_event))
+end
+
+-- ##############################################
+
+--! @brief Release an alert of given type on the entity
+--! @param entity_info data returned by one of the entity_info building functions
+--! @param type_info data returned by one of the type_info building functions
+--! @param when (optional) the time when the release event occurs
+--! @note The actual release is performed asynchronously
+--! @return true on success, false otherwise
+function alerts.new_release(entity_info, type_info)
+  when = when or os.time()
+  local granularity_sec = type_info.alert_granularity and type_info.alert_granularity.granularity_seconds or 0
+  local granularity_id = type_info.alert_granularity and type_info.alert_granularity.granularity_id or nil
+
+  if(granularity_id ~= nil) then
+    local released = true
+    local alert_key_name = get_alert_triggered_key(type_info)
+
+    if((not table.empty(host)) and (entity_info.alert_entity.entity_id == alertEntity("host"))) then
+      triggered = host.releaseTriggeredAlert(alert_key_name, granularity_id)
+    end
+
+    if(not released) then
+      return(false)
+    end
+  end
+
+  local alert_event = {
+    ifid = interface.getId(),
+    granularity = granularity_sec,
+    entity_type = entity_info.alert_entity.entity_id,
+    entity_value = entity_info.alert_entity_val,
+    type = type_info.alert_type.alert_id,
+    severity = type_info.alert_type.severity.severity_id,
+    message = alert_json,
+    subtype = type_info.alert_subtype or "",
+    tstamp = when,
+    action = "release",
+  }
+
+  return(enqueueAlertEvent(alert_event))
+end
+
+-- ##############################################
+-- entity_info building functions
+-- ##############################################
+
+function alerts.hostAlertEntity(hostip, hostvlan)
+  return {
+    alert_entity = alert_consts.alert_entities.host,
+    alert_entity_val = hostinfo2hostkey({ip = hostip, vlan = hostvlan}, nil, true)
+  }
+end
+
+-- ##############################################
+-- type_info building functions
+-- ##############################################
+
+function alerts.thresholdCrossType(granularity, metric, value, operator, threshold)
+  local res = {
+    alert_type = alert_consts.alert_types.threshold_cross,
+    alert_subtype = string.format("%s_%s", granularity, metric),
+    alert_granularity = alert_consts.alerts_granularities.min,
+    alert_type_params = {
+      metric = metric, value = value,
+      operator = operator, threshold = threshold,
+    }
+  }
+  --~ return {
+    --~ alert_severity = alertSeverity("error"),
+    --~ alert_type = alertType("threshold_cross"),
+    --~ alert_subtype = string.format("%s_%s", granularity, metric),
+    
+  --~ }
+
+  --~ return(alerts_table.tcp_syn_flood)
+  return(res)
+end
+
+-- ##############################################
+
 return(alerts)