Generate alerts when a quota is exceeded

scripts/lua/modules/alert_utils.lua

@@ -9,6 +9,12 @@ local verbose = ntop.getCache("ntopng.prefs.alerts.debug") == "1"
 local callback_utils = require "callback_utils"
 local template = require "template_utils"
 local host_pools_utils = require("host_pools_utils")
+local shaper_utils = nil
+
+if(ntop.isPro()) then
+   package.path = dirs.installdir .. "/pro/scripts/lua/modules/?.lua;" .. package.path
+   shaper_utils = require("shaper_utils")
+end
 
 alerts_granularity = {
    { "min", i18n("alerts_thresholds_config.every_minute"), 60 },
@@ -2118,6 +2124,17 @@ end
 
 -- #################################
 
+-- Redis hashe with key=pool and value=list of quota_exceed_items, separated by |
+local function getPoolsQuotaExceededItemsKey(ifid)
+   return "ntopng.cache.quota_exceeded_pools.ifid_" .. ifid
+end
+
+function deletePoolsQuotaExceededItemsKey(ifid)
+   ntop.delCache(getPoolsQuotaExceededItemsKey(ifid))
+end
+
+-- #################################
+
 local function getHostPoolUrl(pool_id)
    return ntop.getHttpPrefix() .. "/lua/hosts_stats.lua?pool=" .. pool_id
 end
@@ -2130,9 +2147,86 @@ function check_host_pools_alerts(ifid, working_status)
    local active_pools_set = getActivePoolsHashKey(ifid)
    local prev_active_pools = swapKeysValues(ntop.getMembersCache(active_pools_set)) or {}
    local alert_pool_connection_enabled = ntop.getPref("ntopng.prefs.alerts.pool_connection_alert") == "1"
+   local alerts_on_quota_exceeded = ntop.isPro() and ntop.getPref("ntopng.prefs.alerts.quota_exceeded_alert") == "1"
+   local pools_stats = interface.getHostPoolsStats()
+   local quota_exceeded_pools_key = getPoolsQuotaExceededItemsKey(ifid)
+   local quota_exceeded_pools_values = ntop.getHashAllCache(quota_exceeded_pools_key) or {}
+   local quota_exceeded_pools = {}
    local now_active_pools = {}
 
+   -- Deserialize quota_exceeded_pools
+   for pool, v in pairs(quota_exceeded_pools_values) do
+      quota_exceeded_pools[pool] = {}
+
+      for _, group in pairs(split(quota_exceeded_pools_values[pool], "|")) do
+         local parts = split(group, "=")
+
+         if #parts == 2 then
+            local proto = parts[1]
+            local quota = parts[2]
+
+            local parts = split(quota, ",")
+            quota_exceeded_pools[pool][proto] = {toboolean(parts[1]), toboolean(parts[2])}
+         end
+      end
+      -- quota_exceeded_pools[pool] is like {Youtube={true, false}}, where true is bytes_exceeded, false is time_exceeded
+   end
+
    for pool, info in pairs(interface.getHostPoolsInfo().num_members_per_pool) do
+      local pool_stats = pools_stats[tonumber(pool)]
+      local pool_exceeded_quotas = quota_exceeded_pools[pool] or {}
+
+      -- Pool quota
+      if pool_stats and ntop.isPro() then
+         local quotas_info = shaper_utils.getQuotasInfo(ifid, pool, pool_stats)
+
+         for proto, info in pairs(quotas_info) do
+            local prev_exceeded = pool_exceeded_quotas[proto] or {false,false}
+
+            if alerts_on_quota_exceeded then
+               if info.bytes_exceeded and not prev_exceeded[1] then
+                  interface.storeHostPoolAlert(tonumber(pool), alertType("quota_exceeded"), alertSeverity("warning"),
+                     i18n("alert_messages.subject_quota_exceeded", {
+                        pool = host_pools_utils.getPoolName(ifid, pool),
+                        url = getHostPoolUrl(pool),
+                        subject = i18n("alert_messages.proto_bytes_quotas", {proto=proto}),
+                        quota = bytesToSize(info.bytes_quota),
+                        value = bytesToSize(info.bytes_value)}))
+               end
+
+               if info.time_exceeded and not prev_exceeded[2] then
+                  interface.storeHostPoolAlert(tonumber(pool), alertType("quota_exceeded"), alertSeverity("warning"),
+                     i18n("alert_messages.subject_quota_exceeded", {
+                        pool = host_pools_utils.getPoolName(ifid, pool),
+                        url = getHostPoolUrl(pool),
+                        subject = i18n("alert_messages.proto_time_quotas", {proto=proto}),
+                        quota = secondsToTime(info.time_quota),
+                        value = secondsToTime(info.time_value)}))
+               end
+            end
+
+            if not info.bytes_exceeded and not info.time_exceeded then
+               -- delete as no quota is left
+               pool_exceeded_quotas[proto] = nil
+            else
+               -- update/add serialized
+               pool_exceeded_quotas[proto] = {info.bytes_exceeded, info.time_exceeded}
+            end
+         end
+
+         if table.empty(pool_exceeded_quotas) then
+            ntop.delHashCache(quota_exceeded_pools_key, pool)
+         else
+            -- Serialize the new quota information for the pool
+            for proto, value in pairs(pool_exceeded_quotas) do
+               pool_exceeded_quotas[proto] = table.concat({tostring(value[1]), tostring(value[2])}, ",")
+            end
+
+            ntop.setHashCache(quota_exceeded_pools_key, pool, table.tconcat(pool_exceeded_quotas, "=", "|"))
+         end
+      end
+
+      -- Pool presence
       if (pool ~= host_pools_utils.DEFAULT_POOL_ID) and (info.num_hosts > 0) then
          now_active_pools[pool] = 1
 
@@ -2148,6 +2242,7 @@ function check_host_pools_alerts(ifid, working_status)
       end
    end
 
+   -- Pool presence
    for pool in pairs(prev_active_pools) do
       if not now_active_pools[pool] then
          -- Pool disconnection