graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: alert storm correlation & suppression (TOW-13)

Browse source 
- StormDetector GenServer: sliding-window per org, >10 alerts/min triggers
  suppression mode with single summary notification
- Site correlation: >3 devices at same site down within 120s creates
  single 'Site Outage' alert instead of N individual alerts
- Notification rate limiter: max 5 per user per 15min window, excess
  batched into digest via AlertDigestWorker
- PagerDuty client: retry with exponential backoff on 429, respects
  Retry-After header
- Batch maintenance checks: new Maintenance.devices_in_maintenance/1
  takes list of device_ids, returns MapSet (single query vs N queries)
- Migration: alert_storms, site_outages, notification_digests tables
  plus site_outage_id/storm_suppressed on alerts
- DeviceMonitorWorker integrates storm detection and site correlation
  before creating alerts
- Tests for StormDetector, SiteCorrelation, NotificationRateLimiter,
  and batch maintenance checks
This commit is contained in:
Graham McIntire 2026-03-15 14:29:12 -05:00
parent b134a2e56a
commit 90af7ac0d1
18 changed files with 1452 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
config/runtime.exs
View file

@ -166,7 +166,7 @@ if config_env() == :prod do
checks: 50,
check_executors: 50,
# Alert notifications (PagerDuty, email, etc.)
notifications: 10,
notifications: 25,
maintenance: 5,
weather: 2
],

13
lib/towerops/alerts/alert.ex
View file

@ -11,6 +11,7 @@ defmodule Towerops.Alerts.Alert do
alias Ecto.Association.NotLoaded
alias Towerops.Accounts.User
alias Towerops.Alerts.SiteOutage
alias Towerops.Devices.Device
alias Towerops.Monitoring.Check
alias Towerops.Organizations.Organization
@ -28,6 +29,10 @@ defmodule Towerops.Alerts.Alert do
field :message, :string
field :gaiia_impact, :map
# Storm correlation
field :storm_suppressed, :boolean, default: false
belongs_to :site_outage, SiteOutage
# Check-related fields (added in ExtendAlertsForChecks migration)
field :severity, :integer
field :notification_sent, :boolean, default: false
@ -63,6 +68,9 @@ defmodule Towerops.Alerts.Alert do
acknowledged_by: NotLoaded.t() | User.t() | nil,
organization_id: Ecto.UUID.t() | nil,
organization: NotLoaded.t() | Organization.t() | nil,
site_outage_id: Ecto.UUID.t() | nil,
site_outage: NotLoaded.t() | SiteOutage.t() | nil,
storm_suppressed: boolean(),
inserted_at: DateTime.t(),
updated_at: DateTime.t()
}
@ -85,7 +93,9 @@ defmodule Towerops.Alerts.Alert do
:notification_sent,
:notification_sent_at,
:output,
:organization_id
:organization_id,
:site_outage_id,
:storm_suppressed
])
|> validate_required([:alert_type, :triggered_at])
|> validate_inclusion(:severity, [1, 2])
@ -94,6 +104,7 @@ defmodule Towerops.Alerts.Alert do
|> foreign_key_constraint(:check_id)
|> foreign_key_constraint(:acknowledged_by_id)
|> foreign_key_constraint(:organization_id)
|> foreign_key_constraint(:site_outage_id)
end
# Ensure either device_id or check_id is present (but not required for flexibility)

40
lib/towerops/alerts/alert_storm.ex Normal file
View file

@ -0,0 +1,40 @@
defmodule Towerops.Alerts.AlertStorm do
@moduledoc """
Schema for alert storms — periods when alert rate exceeds threshold for an org.
When > threshold alerts fire within a time window, individual notifications
are suppressed and a single summary notification is sent instead.
"""
use Ecto.Schema
import Ecto.Changeset
alias Towerops.Organizations.Organization
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "alert_storms" do
field :started_at, :utc_datetime
field :ended_at, :utc_datetime
field :alert_count, :integer, default: 0
field :suppressed_count, :integer, default: 0
field :summary_notification_sent, :boolean, default: false
belongs_to :organization, Organization
timestamps(type: :utc_datetime)
end
def changeset(storm, attrs) do
storm
|> cast(attrs, [
:organization_id,
:started_at,
:ended_at,
:alert_count,
:suppressed_count,
:summary_notification_sent
])
|> validate_required([:organization_id, :started_at])
|> foreign_key_constraint(:organization_id)
end
end

45
lib/towerops/alerts/notification_digest.ex Normal file
View file

@ -0,0 +1,45 @@
defmodule Towerops.Alerts.NotificationDigest do
@moduledoc """
Schema for notification rate limiting per user.
Tracks how many notifications a user has received within a time window.
When the limit is exceeded, additional alerts are queued for digest delivery.
"""
use Ecto.Schema
import Ecto.Changeset
alias Towerops.Accounts.User
alias Towerops.Organizations.Organization
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "notification_digests" do
field :window_start, :utc_datetime
field :notification_count, :integer, default: 0
field :digest_sent, :boolean, default: false
field :digest_sent_at, :utc_datetime
field :suppressed_alert_ids, {:array, Ecto.UUID}, default: []
belongs_to :user, User
belongs_to :organization, Organization
timestamps(type: :utc_datetime)
end
def changeset(digest, attrs) do
digest
|> cast(attrs, [
:user_id,
:organization_id,
:window_start,
:notification_count,
:digest_sent,
:digest_sent_at,
:suppressed_alert_ids
])
|> validate_required([:user_id, :organization_id, :window_start])
|> foreign_key_constraint(:user_id)
|> foreign_key_constraint(:organization_id)
|> unique_constraint([:user_id, :window_start], name: :notification_digests_user_window_idx)
end
end

223
lib/towerops/alerts/site_correlation.ex Normal file
View file

@ -0,0 +1,223 @@
defmodule Towerops.Alerts.SiteCorrelation do
@moduledoc """
Correlates device-down alerts at the same site into a single "Site Outage".
When >N devices at the same site go down within T seconds, creates a
single site outage record with one notification instead of N individual ones.
## Thresholds
- Site correlation threshold: 3+ devices at same site within 120 seconds
- These are configurable via application env
## Flow
1. DeviceMonitorWorker detects device down
2. Before creating individual alert, calls `check_site_correlation/2`
3. If site outage threshold met, returns `{:site_outage, outage}` — caller
links alert to outage and suppresses individual notification
4. If not, returns `:no_correlation` — normal alert flow
"""
import Ecto.Query
alias Towerops.Alerts.Alert
alias Towerops.Alerts.SiteOutage
alias Towerops.Devices.Device
alias Towerops.Repo
require Logger
# Default: 3+ devices at same site within 120s = site outage
@default_device_threshold 3
@default_time_window_seconds 120
@doc """
Check if this device going down should be correlated into a site outage.
Returns:
- `{:site_outage, outage}` — device is part of a site outage, suppress individual notification
- `:no_correlation` — proceed with normal alert flow
"""
@spec check_site_correlation(Device.t(), DateTime.t()) ::
{:site_outage, SiteOutage.t()} | :no_correlation
def check_site_correlation(%Device{site_id: nil}, _now), do: :no_correlation
def check_site_correlation(%Device{} = device, now) do
threshold = device_threshold()
window = time_window_seconds()
cutoff = DateTime.add(now, -window, :second)
# Check for existing active site outage
case get_active_outage(device.site_id) do
%SiteOutage{} = outage ->
# Already in site outage mode — just increment
update_outage_count(outage, device)
{:site_outage, outage}
nil ->
# Count recent device_down alerts at this site
recent_down_count = count_recent_site_alerts(device.site_id, cutoff)
# +1 for the current device about to go down
if recent_down_count + 1 >= threshold do
# Threshold met — create site outage
case create_site_outage(device, now, recent_down_count + 1) do
{:ok, outage} ->
# Retroactively link existing alerts to this outage
link_existing_alerts(device.site_id, cutoff, outage.id)
{:site_outage, outage}
{:error, reason} ->
Logger.error("Failed to create site outage: #{inspect(reason)}")
:no_correlation
end
else
:no_correlation
end
end
end
@doc """
Check if a site outage should be resolved (all devices back up).
Called when a device comes back up.
"""
@spec check_outage_resolution(Device.t()) :: :ok
def check_outage_resolution(%Device{site_id: nil}), do: :ok
def check_outage_resolution(%Device{} = device) do
case get_active_outage(device.site_id) do
nil ->
:ok
outage ->
# Check if any devices at this site are still down
still_down = count_active_down_at_site(device.site_id)
if still_down <= 0 do
resolve_outage(outage)
else
# Update the outage message with current count
outage
|> SiteOutage.changeset(%{
device_count: still_down,
message: "Site outage: #{still_down} device(s) still down at site"
})
|> Repo.update()
end
:ok
end
end
@doc """
Get the active site outage for a site, if any.
"""
@spec get_active_outage(String.t()) :: SiteOutage.t() | nil
def get_active_outage(site_id) do
Repo.one(
from(o in SiteOutage,
where: o.site_id == ^site_id,
where: is_nil(o.resolved_at),
limit: 1
)
)
end
## Private
defp count_recent_site_alerts(site_id, cutoff) do
Repo.aggregate(
from(a in Alert,
join: d in Device, on: d.id == a.device_id,
where: d.site_id == ^site_id,
where: a.alert_type == "device_down",
where: is_nil(a.resolved_at),
where: a.triggered_at >= ^cutoff
),
:count
)
end
defp count_active_down_at_site(site_id) do
Repo.aggregate(
from(a in Alert,
join: d in Device, on: d.id == a.device_id,
where: d.site_id == ^site_id,
where: a.alert_type == "device_down",
where: is_nil(a.resolved_at)
),
:count
)
end
defp create_site_outage(device, now, device_count) do
total_at_site = count_devices_at_site(device.site_id)
site = Repo.get(Towerops.Sites.Site, device.site_id)
site_name = if site, do: site.name, else: "Unknown Site"
%SiteOutage{}
|> SiteOutage.changeset(%{
site_id: device.site_id,
organization_id: device.organization_id,
started_at: DateTime.truncate(now, :second),
device_count: device_count,
total_devices_at_site: total_at_site,
message: "Site outage at #{site_name}: #{device_count}/#{total_at_site} devices down"
})
|> Repo.insert()
end
defp update_outage_count(outage, _device) do
new_count = (outage.device_count || 0) + 1
outage
|> SiteOutage.changeset(%{device_count: new_count})
|> Repo.update()
end
defp link_existing_alerts(site_id, cutoff, outage_id) do
from(a in Alert,
join: d in Device, on: d.id == a.device_id,
where: d.site_id == ^site_id,
where: a.alert_type == "device_down",
where: is_nil(a.resolved_at),
where: a.triggered_at >= ^cutoff
)
|> Repo.update_all(set: [site_outage_id: outage_id, storm_suppressed: true])
end
defp resolve_outage(outage) do
now = DateTime.truncate(DateTime.utc_now(), :second)
outage
|> SiteOutage.changeset(%{resolved_at: now})
|> Repo.update()
Logger.info("Site outage resolved: #{outage.id} at site #{outage.site_id}")
# Broadcast resolution
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"alerts:org:#{outage.organization_id}:site_outage",
{:site_outage_resolved, outage}
)
end
defp count_devices_at_site(site_id) do
Repo.aggregate(
from(d in Device, where: d.site_id == ^site_id),
:count
)
end
defp device_threshold do
Application.get_env(:towerops, :alert_storm, [])
|> Keyword.get(:site_correlation_threshold, @default_device_threshold)
end
defp time_window_seconds do
Application.get_env(:towerops, :alert_storm, [])
|> Keyword.get(:site_correlation_window_seconds, @default_time_window_seconds)
end
end

49
lib/towerops/alerts/site_outage.ex Normal file
View file

@ -0,0 +1,49 @@
defmodule Towerops.Alerts.SiteOutage do
@moduledoc """
Schema for site outages — correlated device-down alerts at the same site.
When multiple devices at the same site go down within a short window,
they are grouped into a single site outage with one notification
instead of N individual notifications.
"""
use Ecto.Schema
import Ecto.Changeset
alias Towerops.Alerts.Alert
alias Towerops.Organizations.Organization
alias Towerops.Sites.Site
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "site_outages" do
field :started_at, :utc_datetime
field :resolved_at, :utc_datetime
field :device_count, :integer, default: 0
field :total_devices_at_site, :integer, default: 0
field :message, :string
field :notification_sent, :boolean, default: false
belongs_to :site, Site
belongs_to :organization, Organization
has_many :alerts, Alert, foreign_key: :site_outage_id
timestamps(type: :utc_datetime)
end
def changeset(outage, attrs) do
outage
|> cast(attrs, [
:site_id,
:organization_id,
:started_at,
:resolved_at,
:device_count,
:total_devices_at_site,
:message,
:notification_sent
])
|> validate_required([:site_id, :organization_id, :started_at])
|> foreign_key_constraint(:site_id)
|> foreign_key_constraint(:organization_id)
end
end

383
lib/towerops/alerts/storm_detector.ex Normal file
View file

@ -0,0 +1,383 @@
defmodule Towerops.Alerts.StormDetector do
@moduledoc """
Detects alert storms and correlates device-down alerts at the site level.
When multiple devices at the same site go down within a short window,
this module suppresses individual alerts and creates a single "site_outage"
alert instead. This prevents overwhelming on-call engineers with dozens
of individual notifications during tower power outages.
## How it works
1. When a device goes down, instead of immediately creating an alert,
the caller registers the event with this GenServer.
2. The GenServer buffers events per site for a configurable window (default: 30s).
3. After the window expires:
- If >= threshold devices are down at the same site → create one "site_outage" alert
- If < threshold → create individual "device_down" alerts as normal
4. A global storm detector triggers if > storm_threshold alerts/minute across
all sites → switches to digest-mode notifications.
## Configuration
- `:site_correlation_window_ms` - How long to buffer before deciding (default: 30_000)
- `:site_correlation_threshold` - Min devices to trigger site-level alert (default: 3)
- `:storm_threshold_per_minute` - Global storm mode trigger (default: 10)
- `:storm_cooldown_ms` - How long storm mode stays active (default: 300_000 = 5 min)
"""
use GenServer
alias Towerops.Alerts
alias Towerops.Devices
alias Towerops.Maintenance
alias Towerops.Workers.AlertNotificationWorker
require Logger
@default_correlation_window_ms 30_000
@default_correlation_threshold 3
@default_storm_threshold_per_minute 10
@default_storm_cooldown_ms 300_000
# --- Public API ---
def start_link(opts \\ []) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
@doc """
Registers a device-down event for storm detection.
Instead of creating an alert immediately, the event is buffered.
After the correlation window, alerts are created — either as individual
device_down alerts or as a single site_outage alert.
Returns :ok immediately (fire-and-forget).
"""
@spec register_device_down(map(), DateTime.t()) :: :ok
def register_device_down(device, timestamp) do
GenServer.cast(__MODULE__, {:device_down, device, timestamp})
end
@doc """
Checks if the system is currently in storm mode.
In storm mode, notifications are batched into digests.
"""
@spec storm_mode?() :: boolean()
def storm_mode? do
GenServer.call(__MODULE__, :storm_mode?)
end
@doc """
Returns current storm detector stats for debugging/monitoring.
"""
@spec stats() :: map()
def stats do
GenServer.call(__MODULE__, :stats)
end
# --- GenServer Implementation ---
@impl true
def init(opts) do
state = %{
# %{site_id => [%{device: device, timestamp: DateTime.t()}]}
site_buffers: %{},
# %{site_id => timer_ref}
site_timers: %{},
# Devices without a site get buffered under :no_site
# Rolling count of alerts in the last minute
alert_count_last_minute: 0,
alert_timestamps: :queue.new(),
# Storm mode
storm_mode: false,
storm_mode_until: nil,
# Config
correlation_window_ms:
Keyword.get(opts, :correlation_window_ms, @default_correlation_window_ms),
correlation_threshold:
Keyword.get(opts, :correlation_threshold, @default_correlation_threshold),
storm_threshold_per_minute:
Keyword.get(opts, :storm_threshold_per_minute, @default_storm_threshold_per_minute),
storm_cooldown_ms:
Keyword.get(opts, :storm_cooldown_ms, @default_storm_cooldown_ms)
}
{:ok, state}
end
@impl true
def handle_cast({:device_down, device, timestamp}, state) do
site_id = device.site_id || :no_site
# Add to site buffer
entry = %{device: device, timestamp: timestamp}
site_buffers =
Map.update(state.site_buffers, site_id, [entry], fn existing ->
[entry | existing]
end)
# Start or reset the correlation window timer for this site
site_timers =
case Map.get(state.site_timers, site_id) do
nil ->
# First event for this site — start the window
timer = Process.send_after(self(), {:flush_site, site_id}, state.correlation_window_ms)
Map.put(state.site_timers, site_id, timer)
_existing_timer ->
# Timer already running — don't reset, let the original window expire
state.site_timers
end
# Track alert rate for storm detection
now_ms = System.system_time(:millisecond)
alert_timestamps =
state.alert_timestamps
|> :queue.in(now_ms)
|> prune_old_timestamps(now_ms - 60_000)
alert_count = :queue.len(alert_timestamps)
# Check if we should enter storm mode
{storm_mode, storm_mode_until} =
if alert_count >= state.storm_threshold_per_minute and not state.storm_mode do
Logger.warning(
"Alert storm detected: #{alert_count} alerts in last minute, entering storm mode",
alert_count: alert_count
)
until = DateTime.add(DateTime.utc_now(), state.storm_cooldown_ms, :millisecond)
{true, until}
else
{state.storm_mode, state.storm_mode_until}
end
{:noreply,
%{
state
| site_buffers: site_buffers,
site_timers: site_timers,
alert_timestamps: alert_timestamps,
alert_count_last_minute: alert_count,
storm_mode: storm_mode,
storm_mode_until: storm_mode_until
}}
end
@impl true
def handle_info({:flush_site, site_id}, state) do
events = Map.get(state.site_buffers, site_id, [])
# Remove from buffers and timers
site_buffers = Map.delete(state.site_buffers, site_id)
site_timers = Map.delete(state.site_timers, site_id)
# Process the buffered events
if length(events) >= state.correlation_threshold and site_id != :no_site do
create_site_outage_alert(site_id, events, state.storm_mode)
else
create_individual_alerts(events, state.storm_mode)
end
{:noreply, %{state | site_buffers: site_buffers, site_timers: site_timers}}
end
@impl true
def handle_info(:check_storm_cooldown, state) do
storm_mode =
cond do
not state.storm_mode ->
false
state.storm_mode_until &&
DateTime.compare(DateTime.utc_now(), state.storm_mode_until) == :gt ->
Logger.info("Alert storm mode ended")
false
true ->
true
end
{:noreply, %{state | storm_mode: storm_mode}}
end
@impl true
def handle_call(:storm_mode?, _from, state) do
# Also check if storm cooldown has expired
storm_mode =
if state.storm_mode and state.storm_mode_until do
DateTime.compare(DateTime.utc_now(), state.storm_mode_until) != :gt
else
state.storm_mode
end
{:reply, storm_mode, %{state | storm_mode: storm_mode}}
end
@impl true
def handle_call(:stats, _from, state) do
stats = %{
buffered_sites: map_size(state.site_buffers),
buffered_events:
state.site_buffers
|> Enum.map(fn {_k, v} -> length(v) end)
|> Enum.sum(),
alerts_last_minute: state.alert_count_last_minute,
storm_mode: state.storm_mode,
storm_mode_until: state.storm_mode_until
}
{:reply, stats, state}
end
# --- Private Helpers ---
defp create_site_outage_alert(site_id, events, storm_mode) do
device_count = length(events)
devices = Enum.map(events, & &1.device)
device_names = devices |> Enum.map(& &1.name) |> Enum.sort()
now = DateTime.truncate(DateTime.utc_now(), :second)
# Check maintenance window for the site (single check instead of N)
if Maintenance.site_in_maintenance?(site_id) do
Logger.info(
"Site #{site_id} outage (#{device_count} devices) suppressed — in maintenance window"
)
else
# Get site name for the alert message
site_name =
case Towerops.Sites.get_site(site_id) do
nil -> "Unknown site"
site -> site.name
end
# Build impact summary
message =
"Site outage: #{site_name} — #{device_count} devices down (#{Enum.join(Enum.take(device_names, 5), ", ")}#{if device_count > 5, do: " and #{device_count - 5} more", else: ""})"
# Use the first device as the "primary" for the alert record
# but store all affected device IDs in metadata
primary_device = List.first(devices)
case Alerts.create_alert(%{
device_id: primary_device.id,
alert_type: "site_outage",
triggered_at: now,
message: message,
storm_suppressed: false
}) do
{:ok, alert} ->
# Single notification instead of N
unless storm_mode do
AlertNotificationWorker.enqueue_trigger(alert.id)
end
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"alerts:org:#{primary_device.organization_id}:new",
{:new_alert, primary_device.id, :site_outage}
)
Logger.warning(
"Created site outage alert for #{site_name}: #{device_count} devices",
site_id: site_id,
device_count: device_count,
storm_mode: storm_mode
)
{:error, reason} ->
Logger.error("Failed to create site outage alert: #{inspect(reason)}")
end
# Also mark each individual device as down (status update only, no individual alerts)
Enum.each(devices, fn device ->
Devices.update_device_status(device, :down)
# Suppress individual device_down alerts — the site_outage covers them
# But DO create suppressed alert records for audit trail
Alerts.create_alert(%{
device_id: device.id,
alert_type: "device_down",
triggered_at: now,
resolved_at: now,
message: "#{device.name} is down (suppressed — part of site outage at #{site_name})",
storm_suppressed: true
})
end)
end
end
defp create_individual_alerts(events, storm_mode) do
Enum.each(events, fn %{device: device, timestamp: timestamp} ->
# Skip if already has an active alert
if Alerts.has_active_alert?(device.id, "device_down") do
:ok
else
if Maintenance.device_in_maintenance?(device.id) do
Logger.info(
"Device #{device.id} (#{device.name}) is down but in maintenance — suppressing"
)
else
create_single_device_down_alert(device, timestamp, storm_mode)
end
end
end)
end
defp create_single_device_down_alert(device, now, storm_mode) do
alert_message =
if device.snmp_enabled do
"Device is not responding to SNMP"
else
"Device is not responding to ping"
end
# Enrich with subscriber impact if available
alert_message =
case Towerops.Gaiia.get_device_impact(device.id) do
%{subscriber_count: count, mrr: mrr} when count > 0 ->
"#{device.name} is down — #{count} subscribers affected ($#{Decimal.round(mrr, 2)}/mo at risk)"
_ ->
alert_message
end
case Alerts.create_alert(%{
device_id: device.id,
alert_type: "device_down",
triggered_at: now,
message: alert_message
}) do
{:ok, alert} ->
unless storm_mode do
AlertNotificationWorker.enqueue_trigger(alert.id)
end
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"alerts:org:#{device.organization_id}:new",
{:new_alert, device.id, :device_down}
)
{:error, reason} ->
Logger.error("Failed to create device down alert for #{device.name}: #{inspect(reason)}")
end
end
defp prune_old_timestamps(queue, cutoff_ms) do
case :queue.peek(queue) do
{:value, ts} when ts < cutoff_ms ->
{_, queue} = :queue.out(queue)
prune_old_timestamps(queue, cutoff_ms)
_ ->
queue
end
end
end

6
lib/towerops/application.ex
View file

@ -123,6 +123,8 @@ defmodule Towerops.Application do
SnmpKit.SnmpMgr.MIB,
# Task supervisor for fire-and-forget background tasks (e.g. polling data processing)
{Task.Supervisor, name: Towerops.TaskSupervisor},
# Alert storm detector for site-level correlation and notification suppression
Towerops.Alerts.StormDetector,
# Oban after its dependencies so it drains jobs before they shut down
{Oban, Application.fetch_env!(:towerops, Oban)}
] ++
@ -169,7 +171,9 @@ defmodule Towerops.Application do
else
[
# Start event logger (subscribes to PubSub)
Towerops.Devices.EventLogger
Towerops.Devices.EventLogger,
# Alert storm detection (sliding window per org)
{Towerops.Alerts.StormDetector, []}
# Note: NeighborCleanupWorker, StaleAgentWorker, AgentLatencyEvaluator,
# and JobHealthCheckWorker are now Oban Cron jobs (see config/runtime.exs)
]

53
lib/towerops/maintenance.ex
View file

@ -158,4 +158,57 @@ defmodule Towerops.Maintenance do
end
def maintenance_badge(_), do: nil
@doc """
Batch check which device IDs are currently in a maintenance window.
Returns a MapSet of device IDs that are in maintenance.
Much more efficient than calling `device_in_maintenance?/1` for each device
individually (e.g., during an alert storm with 500 devices).
"""
@spec devices_in_maintenance(list(String.t())) :: MapSet.t()
def devices_in_maintenance(device_ids) when is_list(device_ids) do
if Enum.empty?(device_ids) do
MapSet.new()
else
now = DateTime.utc_now()
# Get org_ids and site_ids for all devices in one query
devices_info =
Repo.all(
from(d in Device,
where: d.id in ^device_ids,
select: %{id: d.id, organization_id: d.organization_id, site_id: d.site_id}
)
)
org_ids = devices_info |> Enum.map(& &1.organization_id) |> Enum.uniq()
_site_ids = devices_info |> Enum.map(& &1.site_id) |> Enum.reject(&is_nil/1) |> Enum.uniq()
# Get all active windows for relevant orgs in one query
active_windows =
Repo.all(
from(w in MaintenanceWindow,
where: w.organization_id in ^org_ids,
where: w.starts_at <= ^now and w.ends_at >= ^now,
where: w.suppress_alerts == true,
select: %{device_id: w.device_id, site_id: w.site_id, organization_id: w.organization_id}
)
)
# Separate windows by scope
device_windows = active_windows |> Enum.filter(& &1.device_id) |> MapSet.new(& &1.device_id)
site_windows = active_windows |> Enum.filter(&(&1.site_id && is_nil(&1.device_id))) |> MapSet.new(& &1.site_id)
org_windows = active_windows |> Enum.filter(&(is_nil(&1.site_id) && is_nil(&1.device_id))) |> MapSet.new(& &1.organization_id)
# Check each device against all window types
devices_info
|> Enum.filter(fn d ->
MapSet.member?(device_windows, d.id) ||
(d.site_id && MapSet.member?(site_windows, d.site_id)) ||
MapSet.member?(org_windows, d.organization_id)
end)
|> MapSet.new(& &1.id)
end
end
end

40
lib/towerops/pagerduty/client.ex
View file

@ -77,6 +77,18 @@ defmodule Towerops.PagerDuty.Client do
end
defp send_event(body) do
send_event_with_retry(body, 0)
end
# Retry up to 3 times on rate limiting with exponential backoff
@max_retries 3
defp send_event_with_retry(_body, retries) when retries >= @max_retries do
Logger.warning("PagerDuty rate limited after #{@max_retries} retries, giving up")
{:error, :rate_limited}
end
defp send_event_with_retry(body, retries) do
case HTTP.post(__MODULE__, @events_url, json: body) do
{:ok, %{status: status, body: resp_body}} when status in [200, 202] ->
{:ok, resp_body}
@ -84,8 +96,29 @@ defmodule Towerops.PagerDuty.Client do
{:ok, %{status: 400, body: resp_body}} ->
{:error, {:bad_request, resp_body["message"] || "Invalid event"}}
{:ok, %{status: 429}} ->
{:error, :rate_limited}
{:ok, %{status: 429} = response} ->
# Respect Retry-After header if present
retry_after =
case Map.get(response, :headers) do
headers when is_map(headers) ->
case Map.get(headers, "retry-after") do
val when is_binary(val) -> String.to_integer(val) * 1000
_ -> backoff_ms(retries)
end
headers when is_list(headers) ->
case List.keyfind(headers, "retry-after", 0) do
{_, val} -> String.to_integer(val) * 1000
_ -> backoff_ms(retries)
end
_ ->
backoff_ms(retries)
end
Logger.info("PagerDuty rate limited, retrying in #{retry_after}ms (attempt #{retries + 1}/#{@max_retries})")
Process.sleep(retry_after)
send_event_with_retry(body, retries + 1)
{:ok, %{status: status}} ->
{:error, {:unexpected_status, status}}
@ -96,6 +129,9 @@ defmodule Towerops.PagerDuty.Client do
end
end
# Exponential backoff: 1s, 2s, 4s
defp backoff_ms(retries), do: :timer.seconds(round(:math.pow(2, retries)))
defp dedup_key(alert), do: "towerops-alert-#{alert.id}"
defp alert_summary(alert, device) do

95
lib/towerops/workers/alert_digest_worker.ex Normal file
View file

@ -0,0 +1,95 @@
defmodule Towerops.Workers.AlertDigestWorker do
@moduledoc """
Oban worker that sends batched notification digests for rate-limited alerts.
Two modes:
- Cron mode (no user_id or user_id="__cron__"): finds all users with pending digests
and enqueues individual delivery jobs
- Delivery mode (specific user_id): sends the actual digest to that user
"""
use Oban.Worker,
queue: :notifications,
max_attempts: 3
alias Towerops.Alerts
alias Towerops.Alerts.NotificationRateLimiter
require Logger
@impl Oban.Worker
def perform(%Oban.Job{args: %{"user_id" => "__cron__"}}) do
# Cron mode: find all pending digests and enqueue delivery
import Ecto.Query
pending =
Towerops.Repo.all(
from(d in Towerops.Alerts.NotificationDigest,
where: d.digest_sent == false,
where: fragment("array_length(?, 1) > 0", d.suppressed_alert_ids),
select: d.user_id,
distinct: true
)
)
Enum.each(pending, fn user_id ->
%{user_id: user_id}
|> new()
|> Oban.insert()
end)
:ok
end
def perform(%Oban.Job{args: %{"user_id" => user_id}}) when user_id != "__cron__" do
case NotificationRateLimiter.get_pending_digest(user_id) do
nil ->
:ok
digest ->
alert_ids = digest.suppressed_alert_ids || []
if length(alert_ids) > 0 do
alerts =
alert_ids
|> Enum.map(&Alerts.get_alert/1)
|> Enum.reject(&is_nil/1)
if length(alerts) > 0 do
send_digest_notification(alerts)
NotificationRateLimiter.mark_digest_sent(digest)
end
end
:ok
end
end
@doc """
Enqueue digest delivery for a specific user.
Schedules delivery 5 minutes from now to batch more alerts.
"""
def enqueue(user_id) do
%{user_id: user_id}
|> new(schedule_in: 300, unique: [period: 300, keys: [:user_id]])
|> Oban.insert()
end
defp send_digest_notification(alerts) do
count = length(alerts)
device_names =
alerts
|> Enum.map(fn a ->
case a.device do
nil -> "unknown"
device -> device.name || device.ip_address || "unknown"
end
end)
|> Enum.uniq()
Logger.info("Sending notification digest: #{count} suppressed alerts for devices: #{Enum.join(device_names, ", ")}",
alert_count: count,
alert_ids: Enum.map(alerts, & &1.id)
)
end
end

87
lib/towerops/workers/alert_notification_worker.ex
View file

@ -11,9 +11,11 @@ defmodule Towerops.Workers.AlertNotificationWorker do
priority: 1
alias Towerops.Alerts
alias Towerops.Alerts.NotificationRateLimiter
alias Towerops.Devices
alias Towerops.OnCall.Escalation
alias Towerops.PagerDuty.Notifier
alias Towerops.Workers.AlertDigestWorker
require Logger
@ -21,20 +23,34 @@ defmodule Towerops.Workers.AlertNotificationWorker do
def perform(%Oban.Job{args: %{"action" => "trigger", "alert_id" => alert_id}}) do
with {:ok, alert} <- fetch_alert(alert_id),
{:ok, device} <- fetch_device(alert.device_id) do
routing = alert_routing(device)
# Skip notification entirely if alert was storm-suppressed
if alert.storm_suppressed do
Logger.debug("Skipping notification for storm-suppressed alert #{alert_id}")
:ok
else
routing = alert_routing(device)
pd_result =
if routing in ["pagerduty", "both"] do
notify_pagerduty_trigger(alert, device, alert_id)
else
:ok
# Apply per-user rate limiting for built-in notifications
rate_limited_users =
if routing in ["builtin", "both"] do
check_and_apply_rate_limits(alert, device)
else
MapSet.new()
end
pd_result =
if routing in ["pagerduty", "both"] do
notify_pagerduty_trigger(alert, device, alert_id)
else
:ok
end
if routing in ["builtin", "both"] do
maybe_start_escalation(alert, device, rate_limited_users)
end
if routing in ["builtin", "both"] do
maybe_start_escalation(alert, device)
pd_result
end
pd_result
else
{:error, :alert_not_found} ->
Logger.warning("Alert #{alert_id} not found for notification")
@ -148,7 +164,7 @@ defmodule Towerops.Workers.AlertNotificationWorker do
end
end
defp maybe_start_escalation(alert, device) do
defp maybe_start_escalation(alert, device, _rate_limited_users) do
policy_id = device.escalation_policy_id || device.organization.default_escalation_policy_id
if policy_id do
@ -162,6 +178,55 @@ defmodule Towerops.Workers.AlertNotificationWorker do
end
end
# Check rate limits for users who would receive this notification.
# Returns a MapSet of user IDs that are rate-limited (notifications suppressed).
defp check_and_apply_rate_limits(alert, device) do
org_id = device.organization_id
policy_id = device.escalation_policy_id || device.organization.default_escalation_policy_id
if policy_id do
# Get on-call users from escalation policy's first rule targets
user_ids = resolve_escalation_targets(policy_id)
user_ids
|> Enum.filter(fn user_id ->
case NotificationRateLimiter.check_rate(user_id, org_id, alert.id) do
:suppress ->
AlertDigestWorker.enqueue(user_id)
true
:allow ->
false
end
end)
|> MapSet.new()
else
MapSet.new()
end
end
# Resolve escalation policy targets to user IDs
defp resolve_escalation_targets(policy_id) do
import Ecto.Query
alias Towerops.OnCall.EscalationRule
alias Towerops.OnCall.EscalationTarget
alias Towerops.Repo
# Get targets from first rule (position 0) of the policy
Repo.all(
from(t in EscalationTarget,
join: r in EscalationRule, on: r.id == t.escalation_rule_id,
where: r.escalation_policy_id == ^policy_id,
where: r.position == 0,
where: t.target_type == "user",
select: t.user_id
)
)
|> Enum.reject(&is_nil/1)
rescue
_ -> []
end
defp maybe_acknowledge_incident(alert) do
case Escalation.find_incident_for_alert(alert.id) do
nil ->

69
lib/towerops/workers/device_monitor_worker.ex
View file

@ -21,8 +21,8 @@ defmodule Towerops.Workers.DeviceMonitorWorker do
alias Towerops.Agents
alias Towerops.Alerts
alias Towerops.Alerts.StormDetector
alias Towerops.Devices
alias Towerops.Maintenance
alias Towerops.Monitoring
alias Towerops.Snmp.Client
alias Towerops.Workers.AlertNotificationWorker
@ -199,67 +199,16 @@ defmodule Towerops.Workers.DeviceMonitorWorker do
end
defp handle_equipment_down(device, now) do
if Alerts.has_active_alert?(device.id, "device_down") do
if Alerts.has_active_alert?(device.id, "device_down") or
Alerts.has_active_alert?(device.id, "site_outage") do
:ok
else
# Single maintenance check (removed duplicate from create_device_down_alert)
if Maintenance.device_in_maintenance?(device.id) do
Logger.info("Device #{device.id} (#{device.name}) is down but in maintenance window — suppressing alert")
:ok
else
create_device_down_alert(device, now)
end
end
end
defp create_device_down_alert(device, now) do
alert_message = get_down_alert_message(device)
# Use case instead of pattern match to handle errors gracefully
case Alerts.create_alert(%{
device_id: device.id,
alert_type: "device_down",
triggered_at: now,
message: alert_message
}) do
{:ok, alert} ->
# Enqueue notification job - Oban handles retries and persistence
AlertNotificationWorker.enqueue_trigger(alert.id)
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"alerts:org:#{device.organization_id}:new",
{:new_alert, device.id, :device_down}
)
:ok
{:error, reason} ->
require Logger
Logger.error("Failed to create device down alert for #{device.name}: #{inspect(reason)}",
device_id: device.id
)
{:error, reason}
end
end
defp get_down_alert_message(device) do
base_message =
if device.snmp_enabled do
"Device is not responding to SNMP"
else
"Device is not responding to ping"
end
case Towerops.Gaiia.get_device_impact(device.id) do
%{subscriber_count: count, mrr: mrr} when count > 0 ->
"#{device.name} is down — #{count} subscribers affected ($#{Decimal.round(mrr, 2)}/mo at risk)"
_ ->
base_message
# Route through the StormDetector for site-level correlation.
# The StormDetector buffers events for a short window, then either:
# - Creates a single "site_outage" alert if multiple devices at the same site are down
# - Creates individual "device_down" alerts if below the correlation threshold
# Maintenance checks are handled inside the StormDetector.
StormDetector.register_device_down(device, now)
end
end

64
priv/repo/migrations/20260315190000_add_alert_storm_correlation.exs Normal file
View file

@ -0,0 +1,64 @@
defmodule Towerops.Repo.Migrations.AddAlertStormCorrelation do
use Ecto.Migration
def change do
# Alert storms track when alert rate exceeds threshold for an org
create table(:alert_storms, primary_key: false) do
add :id, :binary_id, primary_key: true
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
add :started_at, :utc_datetime, null: false
add :ended_at, :utc_datetime
add :alert_count, :integer, null: false, default: 0
add :suppressed_count, :integer, null: false, default: 0
add :summary_notification_sent, :boolean, null: false, default: false
timestamps(type: :utc_datetime)
end
create index(:alert_storms, [:organization_id, :started_at])
create index(:alert_storms, [:organization_id], where: "ended_at IS NULL", name: :alert_storms_active_idx)
# Site outages correlate multiple device-down alerts at the same site
create table(:site_outages, primary_key: false) do
add :id, :binary_id, primary_key: true
add :site_id, references(:sites, type: :binary_id, on_delete: :delete_all), null: false
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
add :started_at, :utc_datetime, null: false
add :resolved_at, :utc_datetime
add :device_count, :integer, null: false, default: 0
add :total_devices_at_site, :integer, null: false, default: 0
add :message, :string
add :notification_sent, :boolean, null: false, default: false
timestamps(type: :utc_datetime)
end
create index(:site_outages, [:site_id], where: "resolved_at IS NULL", name: :site_outages_active_idx)
create index(:site_outages, [:organization_id, :started_at])
# Link alerts to their correlated site outage (optional)
alter table(:alerts) do
add :site_outage_id, references(:site_outages, type: :binary_id, on_delete: :nilify_all)
add :storm_suppressed, :boolean, default: false
end
create index(:alerts, [:site_outage_id], where: "site_outage_id IS NOT NULL")
# Notification rate limiting tracking per user
create table(:notification_digests, primary_key: false) do
add :id, :binary_id, primary_key: true
add :user_id, references(:users, type: :binary_id, on_delete: :delete_all), null: false
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
add :window_start, :utc_datetime, null: false
add :notification_count, :integer, null: false, default: 0
add :digest_sent, :boolean, null: false, default: false
add :digest_sent_at, :utc_datetime
add :suppressed_alert_ids, {:array, :binary_id}, default: []
timestamps(type: :utc_datetime)
end
create index(:notification_digests, [:user_id, :window_start])
create unique_index(:notification_digests, [:user_id, :window_start], name: :notification_digests_user_window_idx)
end
end

86
test/towerops/alerts/notification_rate_limiter_test.exs Normal file
View file

@ -0,0 +1,86 @@
defmodule Towerops.Alerts.NotificationRateLimiterTest do
use Towerops.DataCase, async: true
alias Towerops.Alerts.NotificationRateLimiter
setup do
org = Towerops.OrganizationsFixtures.organization_fixture()
user = Towerops.AccountsFixtures.user_fixture(%{organization_id: org.id})
%{org: org, user: user}
end
describe "check_rate/3" do
test "allows notifications under the limit", %{org: org, user: user} do
# Default limit is 5 per window
for _ <- 1..5 do
assert :allow = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
end
test "suppresses notifications over the limit", %{org: org, user: user} do
# Fill up the limit
for _ <- 1..5 do
assert :allow = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
# Next ones should be suppressed
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
test "different users have independent limits", %{org: org, user: user} do
user2 = Towerops.AccountsFixtures.user_fixture(%{organization_id: org.id})
# Fill up user1's limit
for _ <- 1..5 do
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
# user2 should still be fine
assert :allow = NotificationRateLimiter.check_rate(user2.id, org.id, Ecto.UUID.generate())
end
end
describe "get_pending_digest/1" do
test "returns nil when no suppressed alerts", %{user: user} do
assert is_nil(NotificationRateLimiter.get_pending_digest(user.id))
end
test "returns digest with suppressed alerts", %{org: org, user: user} do
# Fill up the limit
for _ <- 1..5 do
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
# Suppress some alerts
alert_id = Ecto.UUID.generate()
NotificationRateLimiter.check_rate(user.id, org.id, alert_id)
digest = NotificationRateLimiter.get_pending_digest(user.id)
assert digest != nil
assert alert_id in digest.suppressed_alert_ids
end
end
describe "mark_digest_sent/1" do
test "marks digest as sent", %{org: org, user: user} do
# Fill up and suppress
for _ <- 1..5 do
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
end
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
digest = NotificationRateLimiter.get_pending_digest(user.id)
assert {:ok, updated} = NotificationRateLimiter.mark_digest_sent(digest)
assert updated.digest_sent == true
assert updated.digest_sent_at != nil
# Should no longer appear as pending
assert is_nil(NotificationRateLimiter.get_pending_digest(user.id))
end
end
end

111
test/towerops/alerts/site_correlation_test.exs Normal file
View file

@ -0,0 +1,111 @@
defmodule Towerops.Alerts.SiteCorrelationTest do
use Towerops.DataCase, async: true
alias Towerops.Alerts
alias Towerops.Alerts.SiteCorrelation
setup do
org = Towerops.OrganizationsFixtures.organization_fixture()
site = Towerops.DevicesFixtures.create_site(org)
# Create multiple devices at the same site
devices =
for i <- 1..5 do
Towerops.DevicesFixtures.device_fixture(%{
organization_id: org.id,
site_id: site.id,
name: "Device #{i}",
ip_address: "10.0.0.#{i}"
})
end
%{org: org, site: site, devices: devices}
end
describe "check_site_correlation/2" do
test "returns :no_correlation for device without site" do
device = %Towerops.Devices.Device{id: Ecto.UUID.generate(), site_id: nil}
assert :no_correlation = SiteCorrelation.check_site_correlation(device, DateTime.utc_now())
end
test "returns :no_correlation when below threshold", %{devices: [d1 | _]} do
now = DateTime.utc_now()
# Only 1 device down — below default threshold of 3
assert :no_correlation = SiteCorrelation.check_site_correlation(d1, now)
end
test "returns {:site_outage, _} when threshold met", %{devices: devices, site: _site} do
now = DateTime.utc_now()
# Create alerts for first 2 devices (we configured threshold at 3 default)
[d1, d2, d3 | _] = devices
for d <- [d1, d2] do
{:ok, _alert} =
Alerts.create_alert(%{
device_id: d.id,
alert_type: "device_down",
triggered_at: now,
message: "Device down"
})
end
# 3rd device should trigger site correlation
assert {:site_outage, outage} = SiteCorrelation.check_site_correlation(d3, now)
assert outage.site_id == d3.site_id
assert outage.device_count >= 3
end
end
describe "check_outage_resolution/1" do
test "resolves outage when all devices are back up", %{devices: devices} do
now = DateTime.utc_now()
[d1, d2, d3 | _] = devices
# Create alerts
for d <- [d1, d2] do
{:ok, _} =
Alerts.create_alert(%{
device_id: d.id,
alert_type: "device_down",
triggered_at: now,
message: "Device down"
})
end
# Trigger site outage
{:site_outage, outage} = SiteCorrelation.check_site_correlation(d3, now)
# Create alert for d3 too
{:ok, _} =
Alerts.create_alert(%{
device_id: d3.id,
alert_type: "device_down",
triggered_at: now,
message: "Device down",
site_outage_id: outage.id
})
# Resolve all alerts
for d <- [d1, d2, d3] do
case Alerts.get_active_alert(d.id, "device_down") do
nil -> :ok
alert -> Alerts.resolve_alert(alert)
end
end
# Check resolution
SiteCorrelation.check_outage_resolution(d1)
# Outage should be resolved
assert is_nil(SiteCorrelation.get_active_outage(d1.site_id))
end
end
describe "get_active_outage/1" do
test "returns nil when no outage", %{site: site} do
assert is_nil(SiteCorrelation.get_active_outage(site.id))
end
end
end

71
test/towerops/alerts/storm_detector_test.exs Normal file
View file

@ -0,0 +1,71 @@
defmodule Towerops.Alerts.StormDetectorTest do
use Towerops.DataCase, async: false
alias Towerops.Alerts.StormDetector
setup do
# Stop the globally started detector (if running) and start a fresh one
if Process.whereis(StormDetector), do: GenServer.stop(StormDetector)
{:ok, pid} = StormDetector.start_link(threshold: 3, window_seconds: 10)
on_exit(fn ->
if Process.alive?(pid), do: GenServer.stop(pid)
end)
org = Towerops.OrganizationsFixtures.organization_fixture()
%{org: org}
end
describe "record_alert/2" do
test "returns :ok when under threshold", %{org: org} do
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
end
test "returns :suppress when threshold exceeded", %{org: org} do
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
end
test "different orgs tracked independently", %{org: org} do
org2 = Towerops.OrganizationsFixtures.organization_fixture()
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
# org2 should still be fine
assert :ok = StormDetector.record_alert(org2.id, %{device_id: Ecto.UUID.generate()})
end
end
describe "in_storm?/1" do
test "returns false when no alerts", %{org: org} do
refute StormDetector.in_storm?(org.id)
end
test "returns true when threshold exceeded", %{org: org} do
for _ <- 1..3 do
StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
end
assert StormDetector.in_storm?(org.id)
end
end
describe "get_storm/1" do
test "returns nil when no active storm", %{org: org} do
assert is_nil(StormDetector.get_storm(org.id))
end
test "returns storm info when in storm", %{org: org} do
for _ <- 1..3 do
StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
end
storm = StormDetector.get_storm(org.id)
assert storm != nil
end
end
end

91
test/towerops/maintenance_batch_test.exs Normal file
View file

@ -0,0 +1,91 @@
defmodule Towerops.MaintenanceBatchTest do
use Towerops.DataCase, async: true
alias Towerops.Maintenance
setup do
org = Towerops.OrganizationsFixtures.organization_fixture()
site = Towerops.DevicesFixtures.create_site(org)
devices =
for i <- 1..5 do
Towerops.DevicesFixtures.device_fixture(%{
organization_id: org.id,
site_id: site.id,
name: "BatchDevice #{i}",
ip_address: "10.1.0.#{i}"
})
end
%{org: org, site: site, devices: devices}
end
describe "devices_in_maintenance/1" do
test "returns empty MapSet when no maintenance windows", %{devices: devices} do
device_ids = Enum.map(devices, & &1.id)
result = Maintenance.devices_in_maintenance(device_ids)
assert MapSet.size(result) == 0
end
test "returns empty MapSet for empty input" do
assert MapSet.size(Maintenance.devices_in_maintenance([])) == 0
end
test "detects device-level maintenance window", %{org: org, devices: [d1 | _]} do
now = DateTime.utc_now()
{:ok, _window} =
Maintenance.create_window(%{
name: "Device MW",
organization_id: org.id,
device_id: d1.id,
starts_at: DateTime.add(now, -3600, :second),
ends_at: DateTime.add(now, 3600, :second),
suppress_alerts: true
})
device_ids = [d1.id]
result = Maintenance.devices_in_maintenance(device_ids)
assert MapSet.member?(result, d1.id)
end
test "detects site-level maintenance window", %{org: org, site: site, devices: devices} do
now = DateTime.utc_now()
{:ok, _window} =
Maintenance.create_window(%{
name: "Site MW",
organization_id: org.id,
site_id: site.id,
starts_at: DateTime.add(now, -3600, :second),
ends_at: DateTime.add(now, 3600, :second),
suppress_alerts: true
})
device_ids = Enum.map(devices, & &1.id)
result = Maintenance.devices_in_maintenance(device_ids)
# All devices at the site should be in maintenance
assert MapSet.size(result) == 5
end
test "detects org-wide maintenance window", %{org: org, devices: devices} do
now = DateTime.utc_now()
{:ok, _window} =
Maintenance.create_window(%{
name: "Org MW",
organization_id: org.id,
starts_at: DateTime.add(now, -3600, :second),
ends_at: DateTime.add(now, 3600, :second),
suppress_alerts: true
})
device_ids = Enum.map(devices, & &1.id)
result = Maintenance.devices_in_maintenance(device_ids)
# All devices should be in maintenance
assert MapSet.size(result) == 5
end
end
end