From 9e35e1eef2e349611e5e5e6983fa95c69db87917 Mon Sep 17 00:00:00 2001 From: Graham McIntie Date: Sun, 15 Mar 2026 14:06:06 -0500 Subject: [PATCH 1/3] feat: WISP-tuned network map enhancements (TOW-16) - Tower/device icons: role-based shapes (triangle for APs/wireless, diamond for switches, round-rect for routers, hexagon for firewalls) - RF link overlays: edges colored by signal health (green/yellow/red) with bandwidth-proportional line thickness - Enhanced hover tooltips: device-to-device name, SNR dB, bandwidth, signal health indicator - Wireless client count badge on AP nodes - Filter bar: All Links | Degraded Only | Sites with Alerts | Search - Search with zoom-to-match and highlight - Geographic layout toggle (when sites have lat/lng) - Enhanced detail panel: RF stats (client count, signal health, SNR), View RF Links button for wireless devices - Backend: compute_wireless_stats/compute_rf_link_stats enrich topology - Site nodes include lat/lng for geographic layout - Edge enrichment with RF signal health from SNR sensors - Fix pre-existing compile error in notification_rate_limiter.ex --- assets/js/app.ts | 268 ++++++++++++++++-- .../alerts/notification_rate_limiter.ex | 169 +++++++++++ lib/towerops/topology.ex | 156 +++++++++- lib/towerops_web/live/network_map_live.ex | 29 +- .../live/network_map_live.html.heex | 203 ++++++++++++- 5 files changed, 775 insertions(+), 50 deletions(-) create mode 100644 lib/towerops/alerts/notification_rate_limiter.ex diff --git a/assets/js/app.ts b/assets/js/app.ts index 8af7b809..7fa5d476 100644 --- a/assets/js/app.ts +++ b/assets/js/app.ts @@ -584,16 +584,37 @@ const BetaBannerDismiss = { // Network Map hook for Cytoscape.js topology visualization const NetworkMap = { cy: null as any, + _topologyData: null as any, mounted(this: any) { const topologyData = JSON.parse(this.el.dataset.topology || '{}') + this._topologyData = topologyData this.initializeCytoscape(topologyData) // Listen for topology updates from LiveView this.handleEvent("update_topology", (payload: any) => { if (this.cy) { + this._topologyData = payload.topology this.updateTopology(payload.topology) } }) + + // Filter handler + this.handleEvent("apply_filter", (payload: any) => { + if (!this.cy) return + this._applyFilter(payload.filter) + }) + + // Search handler + this.handleEvent("search_nodes", (payload: any) => { + if (!this.cy) return + this._searchNodes(payload.query) + }) + + // Layout toggle handler + this.handleEvent("change_layout", (payload: any) => { + if (!this.cy || !this._topologyData) return + this._changeLayout(payload.mode, this._topologyData) + }) }, updated(this: any) { @@ -671,14 +692,24 @@ const NetworkMap = { unknown: 'ellipse' } - // Confidence-based edge styling helper - const getEdgeStyle = (confidence: number) => { + // Signal health color map for RF links + const signalHealthColors: Record = { + good: '#10B981', // green + degraded: '#F59E0B', // yellow + critical: '#EF4444' // red + } + + // Edge styling: prefer signal_health coloring for RF links, fall back to confidence + const getEdgeStyle = (confidence: number, signalHealth?: string) => { + if (signalHealth && signalHealthColors[signalHealth]) { + return { color: signalHealthColors[signalHealth], style: 'solid' as const, dashPattern: [] } + } if (confidence > 0.9) { - return { color: '#10B981', style: 'solid', dashPattern: [] } + return { color: '#10B981', style: 'solid' as const, dashPattern: [] } } else if (confidence >= 0.5) { - return { color: '#F59E0B', style: 'dashed', dashPattern: [6, 3] } + return { color: '#F59E0B', style: 'dashed' as const, dashPattern: [6, 3] } } else { - return { color: '#9CA3AF', style: 'dashed', dashPattern: [2, 3] } + return { color: '#9CA3AF', style: 'dashed' as const, dashPattern: [2, 3] } } } @@ -766,7 +797,13 @@ const NetworkMap = { discovered: node.discovered, ip_address: node.ip_address, site_name: node.site_name, - manufacturer: node.manufacturer + manufacturer: node.manufacturer, + client_count: node.client_count || 0, + signal_health: node.signal_health, + has_alerts: node.has_alerts || false, + latitude: node.latitude, + longitude: node.longitude, + device_role: node.device_role } if (node.parent) data.parent = node.parent elements.push({ @@ -792,17 +829,20 @@ const NetworkMap = { link_type: edge.link_type, confidence: edge.confidence ?? 0.5, if_speed: edge.if_speed, + signal_health: edge.signal_health, + snr: edge.snr, label: label } }) }) - // Map interface speed to edge width + // Map interface speed to edge width ā€” proportional to bandwidth capacity const getEdgeWidth = (speed: number | null | undefined): number => { - if (!speed) return 1 - if (speed >= 10_000_000_000) return 4 // 10Gbps+ - if (speed >= 1_000_000_000) return 2 // 1Gbps - return 1 // < 1Gbps or unknown + if (!speed) return 1.5 + if (speed >= 10_000_000_000) return 5 // 10Gbps+ + if (speed >= 1_000_000_000) return 3.5 // 1Gbps + if (speed >= 100_000_000) return 2.5 // 100Mbps + return 1.5 // < 100Mbps or unknown } // Initialize Cytoscape @@ -819,7 +859,14 @@ const NetworkMap = { 'shape': function(ele: any) { return roleShapes[ele.data('type')] || roleShapes.unknown }, - 'label': 'data(label)', + 'label': function(ele: any) { + const label = ele.data('label') || '' + const clients = ele.data('client_count') + if (clients > 0) { + return `${label}\nšŸ“” ${clients} clients` + } + return label + }, 'color': isDark ? '#fff' : '#000', 'text-valign': 'bottom', 'text-halign': 'center', @@ -827,7 +874,7 @@ const NetworkMap = { 'font-size': '13px', 'font-weight': '500', 'text-wrap': 'wrap', - 'text-max-width': '120px', + 'text-max-width': '140px', 'width': 44, 'height': 44, 'border-width': 2, @@ -895,6 +942,16 @@ const NetworkMap = { 'overlay-opacity': 0.1 } }, + { + selector: 'node.search-match', + style: { + 'border-width': 4, + 'border-color': '#f59e0b', + 'overlay-color': '#f59e0b', + 'overlay-opacity': 0.15, + 'z-index': 999 + } + }, { selector: '$node > node', // compound parent selector style: { @@ -921,17 +978,17 @@ const NetworkMap = { return getEdgeWidth(ele.data('if_speed')) }, 'line-color': function(ele: any) { - return getEdgeStyle(ele.data('confidence')).color + return getEdgeStyle(ele.data('confidence'), ele.data('signal_health')).color }, 'line-style': function(ele: any) { - return getEdgeStyle(ele.data('confidence')).style + return getEdgeStyle(ele.data('confidence'), ele.data('signal_health')).style }, 'line-dash-pattern': function(ele: any) { - const pattern = getEdgeStyle(ele.data('confidence')).dashPattern + const pattern = getEdgeStyle(ele.data('confidence'), ele.data('signal_health')).dashPattern return pattern.length > 0 ? pattern : [1, 0] }, 'target-arrow-color': function(ele: any) { - return getEdgeStyle(ele.data('confidence')).color + return getEdgeStyle(ele.data('confidence'), ele.data('signal_health')).color }, 'curve-style': 'bezier', 'label': 'data(label)', @@ -998,40 +1055,60 @@ const NetworkMap = { this.cy.nodes().removeClass('highlighted') }) - // Add hover tooltips + // Add hover tooltips with WISP RF info this.cy.on('mouseover', 'node', (evt: any) => { const node = evt.target - const nodeData = node.data() + const d = node.data() - let tooltip = nodeData.label - if (nodeData.ip_address) tooltip += `\n${nodeData.ip_address}` - if (nodeData.site_name) tooltip += `\n${nodeData.site_name}` - if (nodeData.manufacturer) tooltip += `\n${nodeData.manufacturer}` + let tooltip = d.label || '' + if (d.ip_address) tooltip += `\n${d.ip_address}` + if (d.site_name) tooltip += `\nšŸ“ ${d.site_name}` + if (d.manufacturer) tooltip += `\n${d.manufacturer}` + if (d.client_count > 0) tooltip += `\nšŸ“” ${d.client_count} clients` + if (d.signal_health) { + const healthIcon = d.signal_health === 'good' ? 'šŸŸ¢' : d.signal_health === 'degraded' ? 'šŸŸ”' : 'šŸ”“' + tooltip += `\n${healthIcon} Signal: ${d.signal_health}` + } node.style('label', tooltip) }) this.cy.on('mouseout', 'node', (evt: any) => { const node = evt.target - node.style('label', node.data('label')) + const d = node.data() + const clients = d.client_count + const label = d.label || '' + node.style('label', clients > 0 ? `${label}\nšŸ“” ${clients} clients` : label) }) - // Add hover for edges + // Enhanced edge hover with PTP/backhaul RF stats this.cy.on('mouseover', 'edge', (evt: any) => { const edge = evt.target const data = edge.data() - // Build detailed tooltip with interfaces and link type - let tooltip = '' - if (data.source_interface) tooltip += `${data.source_interface}` - if (data.target_interface) tooltip += ` \u2194 ${data.target_interface}` + // Get source and target node names for the tooltip + const srcNode = this.cy.getElementById(data.source) + const tgtNode = this.cy.getElementById(data.target) + const srcName = srcNode.data('label') || data.source + const tgtName = tgtNode.data('label') || data.target + + let tooltip = `${srcName} ā†” ${tgtName}` + if (data.snr != null) tooltip += `\nSNR ${data.snr} dB` + if (data.signal_health) { + const healthIcon = data.signal_health === 'good' ? 'šŸŸ¢' : data.signal_health === 'degraded' ? 'šŸŸ”' : 'šŸ”“' + tooltip += ` ${healthIcon}` + } + if (data.if_speed) { + const mbps = data.if_speed / 1_000_000 + tooltip += `\n${mbps >= 1000 ? (mbps/1000).toFixed(1) + ' Gbps' : mbps + ' Mbps'}` + } if (data.link_type) tooltip += `\n${data.link_type.toUpperCase()}` const confidence = data.confidence if (confidence != null) tooltip += ` (${Math.round(confidence * 100)}%)` edge.style({ 'line-color': '#fbbf24', - 'width': 3, + 'width': Math.max(getEdgeWidth(data.if_speed), 3), 'label': tooltip }) }) @@ -1039,7 +1116,7 @@ const NetworkMap = { this.cy.on('mouseout', 'edge', (evt: any) => { const edge = evt.target const data = edge.data() - const edgeStyle = getEdgeStyle(data.confidence ?? 0.5) + const edgeStyle = getEdgeStyle(data.confidence ?? 0.5, data.signal_health) edge.style({ 'line-color': edgeStyle.color, 'width': getEdgeWidth(data.if_speed), @@ -1230,6 +1307,135 @@ const NetworkMap = { }) return positions + }, + + // --- Filter, Search, Layout methods --- + + _applyFilter(this: any, filter: string) { + if (!this.cy) return + + // Show all first + this.cy.elements().style('display', 'element') + + if (filter === 'degraded') { + // Show only edges with degraded/critical signal health and their connected nodes + const matchEdges = this.cy.edges().filter((e: any) => { + const health = e.data('signal_health') + return health === 'degraded' || health === 'critical' + }) + const connectedNodes = matchEdges.connectedNodes() + const parents = connectedNodes.parents() + const visible = matchEdges.union(connectedNodes).union(parents) + this.cy.elements().not(visible).style('display', 'none') + } else if (filter === 'alerts') { + // Show only nodes with alerts and their connected edges + const alertNodes = this.cy.nodes().filter((n: any) => { + return n.data('has_alerts') === true || n.data('status') === 'down' + }) + const connectedEdges = alertNodes.connectedEdges() + const parents = alertNodes.parents() + const visible = alertNodes.union(connectedEdges).union(parents) + this.cy.elements().not(visible).style('display', 'none') + } + // 'all' filter shows everything (already reset above) + }, + + _searchNodes(this: any, query: string) { + if (!this.cy) return + + // Remove previous search highlights + this.cy.nodes().removeClass('search-match') + + if (!query || query.trim() === '') return + + const q = query.toLowerCase() + const matches = this.cy.nodes().filter((n: any) => { + const d = n.data() + return (d.label && d.label.toLowerCase().includes(q)) || + (d.ip_address && d.ip_address.toLowerCase().includes(q)) || + (d.site_name && d.site_name.toLowerCase().includes(q)) || + (d.manufacturer && d.manufacturer.toLowerCase().includes(q)) + }) + + matches.addClass('search-match') + + // Zoom to fit matches if any found + if (matches.length > 0) { + this.cy.fit(matches, 80) + } + }, + + _changeLayout(this: any, mode: string, topology: any) { + if (!this.cy) return + + if (mode === 'geo') { + // Geographic layout using lat/lng positions + const geoNodes = (topology.nodes || []).filter((n: any) => n.latitude != null && n.longitude != null) + + if (geoNodes.length === 0) return + + // Find bounds + const lats = geoNodes.map((n: any) => n.latitude) + const lngs = geoNodes.map((n: any) => n.longitude) + const minLat = Math.min(...lats), maxLat = Math.max(...lats) + const minLng = Math.min(...lngs), maxLng = Math.max(...lngs) + + const width = this.cy.width() - 100 + const height = this.cy.height() - 100 + + // Map geo coordinates to canvas positions + const positions: Record = {} + const latRange = maxLat - minLat || 1 + const lngRange = maxLng - minLng || 1 + + ;(topology.nodes || []).forEach((n: any) => { + if (n.latitude != null && n.longitude != null) { + positions[n.id] = { + x: 50 + ((n.longitude - minLng) / lngRange) * width, + y: 50 + ((maxLat - n.latitude) / latRange) * height // invert Y for map orientation + } + } + }) + + // Nodes without geo get placed at the bottom + let offsetX = 0 + ;(topology.nodes || []).forEach((n: any) => { + if (n.type === 'site') return + if (!positions[n.id]) { + // If node has a parent site with geo, place near it + if (n.parent && positions[n.parent]) { + positions[n.id] = { + x: positions[n.parent].x + (Math.random() - 0.5) * 60, + y: positions[n.parent].y + (Math.random() - 0.5) * 60 + } + } else { + positions[n.id] = { x: offsetX * 80, y: height + 100 } + offsetX++ + } + } + }) + + this.cy.layout({ + name: 'preset', + positions: (node: any) => positions[node.id()] || { x: 0, y: 0 }, + fit: true, + padding: 50, + animate: true, + animationDuration: 500 + }).run() + } else { + // Force-directed (default preset layout) + const allNodes = topology.nodes || [] + const newPositions = this._computePresetPositions(allNodes) + this.cy.layout({ + name: 'preset', + positions: (node: any) => newPositions[node.id()] || { x: 0, y: 0 }, + fit: true, + padding: 50, + animate: true, + animationDuration: 500 + }).run() + } } } diff --git a/lib/towerops/alerts/notification_rate_limiter.ex b/lib/towerops/alerts/notification_rate_limiter.ex new file mode 100644 index 00000000..6c3164d3 --- /dev/null +++ b/lib/towerops/alerts/notification_rate_limiter.ex @@ -0,0 +1,169 @@ +defmodule Towerops.Alerts.NotificationRateLimiter do + @moduledoc """ + Per-user notification rate limiting for alert notifications. + + Limits max N notifications per user per T minutes. When exceeded, + additional alerts are queued for batch digest delivery. + + ## Defaults + - Max 5 notifications per 15-minute window + - Digest sent at end of window with suppressed alerts + + ## Usage + + case NotificationRateLimiter.check_rate(user_id, org_id, alert_id) do + :allow -> send notification normally + :suppress -> alert queued for digest, skip individual notification + end + """ + + import Ecto.Query + + alias Towerops.Alerts.NotificationDigest + alias Towerops.Repo + + require Logger + + @default_max_per_window 5 + @default_window_minutes 15 + + @doc """ + Check if a notification should be sent or suppressed for rate limiting. + + Returns :allow or :suppress. + """ + @spec check_rate(String.t(), String.t(), String.t()) :: :allow | :suppress + def check_rate(user_id, organization_id, alert_id) do + now = DateTime.utc_now() + window_start = calculate_window_start(now) + + case get_or_create_digest(user_id, organization_id, window_start) do + {:ok, digest} -> + max = max_per_window() + + if digest.notification_count < max do + # Under limit ā€” increment and allow + increment_count(digest) + :allow + else + # Over limit ā€” add to suppressed list + add_suppressed_alert(digest, alert_id) + Logger.info("Notification rate limited for user #{user_id}: #{digest.notification_count}/#{max} in window", + user_id: user_id, + alert_id: alert_id + ) + :suppress + end + + {:error, reason} -> + Logger.error("Failed to check notification rate: #{inspect(reason)}") + # Fail open ā€” allow notification + :allow + end + end + + @doc """ + Get suppressed alert IDs for a user's current or recent window. + Used by the digest worker to build the digest notification. + """ + @spec get_pending_digest(String.t()) :: NotificationDigest.t() | nil + def get_pending_digest(user_id) do + Repo.one( + from(d in NotificationDigest, + where: d.user_id == ^user_id, + where: d.digest_sent == false, + where: fragment("array_length(?, 1) > 0", d.suppressed_alert_ids), + order_by: [desc: d.window_start], + limit: 1 + ) + ) + end + + @doc """ + Mark a digest as sent. + """ + @spec mark_digest_sent(NotificationDigest.t()) :: {:ok, NotificationDigest.t()} | {:error, any()} + def mark_digest_sent(digest) do + digest + |> NotificationDigest.changeset(%{ + digest_sent: true, + digest_sent_at: DateTime.truncate(DateTime.utc_now(), :second) + }) + |> Repo.update() + end + + @doc """ + List users with pending digests for an organization. + """ + @spec users_with_pending_digests(String.t()) :: [String.t()] + def users_with_pending_digests(organization_id) do + Repo.all( + from(d in NotificationDigest, + where: d.organization_id == ^organization_id, + where: d.digest_sent == false, + where: fragment("array_length(?, 1) > 0", d.suppressed_alert_ids), + select: d.user_id, + distinct: true + ) + ) + end + + ## Private + + defp get_or_create_digest(user_id, organization_id, window_start) do + case Repo.one( + from(d in NotificationDigest, + where: d.user_id == ^user_id, + where: d.window_start == ^window_start, + limit: 1 + ) + ) do + %NotificationDigest{} = digest -> + {:ok, digest} + + nil -> + %NotificationDigest{} + |> NotificationDigest.changeset(%{ + user_id: user_id, + organization_id: organization_id, + window_start: window_start, + notification_count: 0 + }) + |> Repo.insert() + end + end + + defp increment_count(digest) do + from(d in NotificationDigest, where: d.id == ^digest.id) + |> Repo.update_all(inc: [notification_count: 1]) + end + + defp add_suppressed_alert(digest, alert_id) do + new_ids = (digest.suppressed_alert_ids || []) ++ [alert_id] + + digest + |> NotificationDigest.changeset(%{ + suppressed_alert_ids: new_ids, + notification_count: (digest.notification_count || 0) + 1 + }) + |> Repo.update() + end + + defp calculate_window_start(now) do + # Round down to nearest window boundary + minutes = max_per_window_minutes() + unix = DateTime.to_unix(now) + window_unix = div(unix, minutes * 60) * (minutes * 60) + DateTime.from_unix!(window_unix) |> DateTime.truncate(:second) + end + + defp max_per_window do + Application.get_env(:towerops, :alert_storm, []) + |> Keyword.get(:notification_max_per_window, @default_max_per_window) + end + + defp max_per_window_minutes do + Application.get_env(:towerops, :alert_storm, []) + |> Keyword.get(:notification_window_minutes, @default_window_minutes) + end +end diff --git a/lib/towerops/topology.ex b/lib/towerops/topology.ex index 864021b2..4a8421f4 100644 --- a/lib/towerops/topology.ex +++ b/lib/towerops/topology.ex @@ -13,6 +13,8 @@ defmodule Towerops.Topology do alias Towerops.Snmp.Device, as: SnmpDevice alias Towerops.Snmp.Interface alias Towerops.Snmp.IpAddress + alias Towerops.Snmp.Sensor + alias Towerops.Snmp.WirelessClient alias Towerops.Snmp.MacAddress alias Towerops.Snmp.Neighbor alias Towerops.Topology.DeviceLink @@ -424,7 +426,13 @@ defmodule Towerops.Topology do device_ids = Enum.map(devices, & &1.id) links = list_links_for_devices(device_ids) - managed_nodes = Enum.map(devices, &device_to_node/1) + # Compute wireless stats (client counts + signal health) per device + wireless_stats = compute_wireless_stats(device_ids) + + # Compute RF signal health per link for edge enrichment + rf_link_stats = compute_rf_link_stats(device_ids) + + managed_nodes = Enum.map(devices, &device_to_node(&1, wireless_stats)) discovered_nodes = build_discovered_nodes(links, tab) # Build site compound nodes from devices that have sites @@ -444,7 +452,9 @@ defmodule Towerops.Topology do ip_address: nil, discovered: false, manufacturer: nil, - parent: nil + parent: nil, + latitude: device.site.latitude, + longitude: device.site.longitude } end) @@ -455,6 +465,13 @@ defmodule Towerops.Topology do links |> build_edges_from_links(node_ids) |> merge_bidirectional_edges() + |> enrich_edges_with_rf(rf_link_stats) + + # Compute whether geographic layout is available + has_geo = + Enum.any?(all_nodes, fn n -> + n[:latitude] != nil and n[:longitude] != nil + end) %{ nodes: all_nodes, @@ -463,8 +480,11 @@ defmodule Towerops.Topology do total_devices: length(managed_nodes) + length(discovered_nodes), added_devices: length(managed_nodes), discovered_devices: length(discovered_nodes), - total_links: length(edges) + total_links: length(edges), + degraded_links: Enum.count(edges, fn e -> e[:signal_health] == "degraded" or e[:signal_health] == "critical" end), + sites_with_alerts: devices |> Enum.filter(&(&1.status == :down)) |> Enum.map(& &1.site_id) |> Enum.reject(&is_nil/1) |> Enum.uniq() |> length() }, + has_geo: has_geo, last_updated: DateTime.utc_now() } end @@ -553,16 +573,28 @@ defmodule Towerops.Topology do |> list_connected_devices() |> Enum.map(&link_to_connection(&1, device_id)) + # Wireless stats for detail panel + ws = compute_wireless_stats([device_id]) + device_ws = Map.get(ws, device_id, %{client_count: 0, signal_health: nil}) + + rf = compute_rf_link_stats([device_id]) + device_rf = Map.get(rf, device_id) + %{ id: device.id, name: device.name, ip_address: device.ip_address, type: :managed, role: role_to_type_atom(device.device_role), + device_role: device.device_role, status: device.status, site_name: if(device.site, do: device.site.name), manufacturer: if(device.snmp_device, do: device.snmp_device.manufacturer), - connections: connections + connections: connections, + client_count: device_ws.client_count, + signal_health: device_ws.signal_health, + snr: if(device_rf, do: device_rf[:snr]), + is_wireless: device.device_role in ~w(access_point cpe backhaul_radio) } end end @@ -636,7 +668,9 @@ defmodule Towerops.Topology do ) end - defp device_to_node(device) do + defp device_to_node(device, wireless_stats) do + stats = Map.get(wireless_stats, device.id, %{client_count: 0, signal_health: nil}) + %{ id: device.id, label: device.name || device.ip_address || "Unknown", @@ -648,7 +682,12 @@ defmodule Towerops.Topology do ip_address: device.ip_address, discovered: false, manufacturer: if(device.snmp_device, do: device.snmp_device.manufacturer), - parent: if(device.site_id, do: "site_#{device.site_id}") + parent: if(device.site_id, do: "site_#{device.site_id}"), + client_count: stats.client_count, + signal_health: stats.signal_health, + latitude: if(device.site, do: device.site.latitude), + longitude: if(device.site, do: device.site.longitude), + has_alerts: device.status == :down } end @@ -1154,4 +1193,109 @@ defmodule Towerops.Topology do end end end + + # --- WISP wireless stats helpers --- + + @doc """ + Compute wireless client counts and average signal health per device. + Returns a map of device_id => %{client_count: int, signal_health: "good"|"degraded"|"critical"|nil} + """ + def compute_wireless_stats(device_ids) when is_list(device_ids) do + if Enum.empty?(device_ids) do + %{} + else + # Count wireless clients and average signal per device + results = + Repo.all( + from(wc in WirelessClient, + where: wc.device_id in ^device_ids, + group_by: wc.device_id, + select: {wc.device_id, count(wc.id), avg(wc.signal_strength)} + ) + ) + + Map.new(results, fn {device_id, count, avg_signal} -> + health = classify_signal_health(avg_signal) + {device_id, %{client_count: count, signal_health: health}} + end) + end + end + + @doc """ + Compute RF link stats for edges. Uses SNR sensors on devices to + determine signal health for PTP/backhaul links. + Returns a map of device_id => %{signal_dbm: float, snr: float, signal_health: string} + """ + def compute_rf_link_stats(device_ids) when is_list(device_ids) do + if Enum.empty?(device_ids) do + %{} + else + # Get latest SNR sensor readings per device + results = + Repo.all( + from(s in Sensor, + join: sd in SnmpDevice, + on: s.snmp_device_id == sd.id, + where: sd.device_id in ^device_ids and s.sensor_type == "snr" and not is_nil(s.last_value), + select: {sd.device_id, s.last_value, s.sensor_descr} + ) + ) + + results + |> Enum.group_by(fn {device_id, _val, _descr} -> device_id end) + |> Map.new(fn {device_id, entries} -> + avg_snr = + entries + |> Enum.map(fn {_, val, _} -> val end) + |> then(fn vals -> Enum.sum(vals) / max(length(vals), 1) end) + + {device_id, %{snr: Float.round(avg_snr, 1), signal_health: classify_snr_health(avg_snr)}} + end) + end + end + + defp classify_signal_health(nil), do: nil + defp classify_signal_health(avg_signal) do + avg = if is_float(avg_signal), do: avg_signal, else: Decimal.to_float(avg_signal) + cond do + avg >= -65 -> "good" + avg >= -75 -> "degraded" + true -> "critical" + end + end + + defp classify_snr_health(snr) when snr >= 25, do: "good" + defp classify_snr_health(snr) when snr >= 15, do: "degraded" + defp classify_snr_health(_snr), do: "critical" + + defp enrich_edges_with_rf(edges, rf_stats) do + Enum.map(edges, fn edge -> + source_rf = Map.get(rf_stats, edge.source) + target_rf = Map.get(rf_stats, edge.target) + + # Use the worse health of the two endpoints + signal_health = + case {source_rf, target_rf} do + {nil, nil} -> nil + {s, nil} -> s.signal_health + {nil, t} -> t.signal_health + {s, t} -> worse_health(s.signal_health, t.signal_health) + end + + snr = + case {source_rf, target_rf} do + {nil, nil} -> nil + {s, nil} -> s[:snr] + {nil, t} -> t[:snr] + {s, t} -> min(s[:snr] || 0, t[:snr] || 0) + end + + Map.merge(edge, %{signal_health: signal_health, snr: snr}) + end) + end + + defp worse_health(a, b) do + priority = %{"good" => 0, "degraded" => 1, "critical" => 2} + if (priority[a] || 0) >= (priority[b] || 0), do: a, else: b + end end diff --git a/lib/towerops_web/live/network_map_live.ex b/lib/towerops_web/live/network_map_live.ex index c8a88baa..474ec948 100644 --- a/lib/towerops_web/live/network_map_live.ex +++ b/lib/towerops_web/live/network_map_live.ex @@ -34,7 +34,10 @@ defmodule ToweropsWeb.NetworkMapLive do |> assign(:loading, true) |> assign(:topology, default_topology) |> assign(:active_tab, "added") - |> assign(:selected_node_detail, nil)} + |> assign(:selected_node_detail, nil) + |> assign(:filter, "all") + |> assign(:search_query, "") + |> assign(:layout_mode, "force")} end @impl true @@ -87,6 +90,30 @@ defmodule ToweropsWeb.NetworkMapLive do |> push_event("clear_highlight", %{})} end + @impl true + def handle_event("set_filter", %{"filter" => filter}, socket) do + {:noreply, + socket + |> assign(:filter, filter) + |> push_event("apply_filter", %{filter: filter})} + end + + @impl true + def handle_event("search", %{"query" => query}, socket) do + {:noreply, + socket + |> assign(:search_query, query) + |> push_event("search_nodes", %{query: query})} + end + + @impl true + def handle_event("toggle_layout", %{"mode" => mode}, socket) do + {:noreply, + socket + |> assign(:layout_mode, mode) + |> push_event("change_layout", %{mode: mode})} + end + @impl true def handle_info({:topology_updated, _organization_id}, socket) do # Real-time topology update diff --git a/lib/towerops_web/live/network_map_live.html.heex b/lib/towerops_web/live/network_map_live.html.heex index be3b3077..8a6fe435 100644 --- a/lib/towerops_web/live/network_map_live.html.heex +++ b/lib/towerops_web/live/network_map_live.html.heex @@ -150,7 +150,113 @@ - + +
+
+
+ + + +
+
+
+ <.icon + name="hero-magnifying-glass" + class="absolute left-2.5 top-1/2 -translate-y-1/2 h-4 w-4 text-gray-400" + /> + +
+ <%= if @topology[:has_geo] do %> +
+ + +
+ <% end %> +
+
+
+ +
@@ -295,6 +401,56 @@
<% end %>
+ + + <%= if @selected_node_detail[:is_wireless] do %> +
+

+ RF Stats +

+
+ +
+
Clients
+
+ <.icon name="hero-wifi" class="h-4 w-4 text-purple-500" /> + {@selected_node_detail[:client_count] || 0} +
+
+ +
+
Signal
+
"text-green-600 dark:text-green-400" + "degraded" -> "text-yellow-600 dark:text-yellow-400" + "critical" -> "text-red-600 dark:text-red-400" + _ -> "text-gray-500 dark:text-gray-400" + end + ]}> + {@selected_node_detail[:signal_health] || "N/A"} +
+
+ + <%= if @selected_node_detail[:snr] do %> +
+
SNR
+
= 25 -> "text-green-600 dark:text-green-400" + @selected_node_detail.snr >= 15 -> "text-yellow-600 dark:text-yellow-400" + true -> "text-red-600 dark:text-red-400" + end + ]}> + {@selected_node_detail.snr} dB +
+
+ <% end %> +
+
+ <% end %> <%= if length(@selected_node_detail.connections) > 0 do %> @@ -338,7 +494,7 @@ <%= if @selected_node_detail.type == :managed do %> -
+
<.link navigate={~p"/devices/#{@selected_node_detail.id}"} class="flex items-center justify-center w-full px-3 py-2 text-sm font-medium text-blue-600 dark:text-blue-400 bg-blue-50 dark:bg-blue-400/10 rounded-md hover:bg-blue-100 dark:hover:bg-blue-400/20" @@ -346,6 +502,15 @@ <.icon name="hero-arrow-top-right-on-square" class="h-4 w-4 mr-1.5" /> {t("View Device")} + <%= if @selected_node_detail[:is_wireless] do %> + <.link + navigate={~p"/devices/#{@selected_node_detail.id}?tab=rf"} + class="flex items-center justify-center w-full px-3 py-2 text-sm font-medium text-purple-600 dark:text-purple-400 bg-purple-50 dark:bg-purple-400/10 rounded-md hover:bg-purple-100 dark:hover:bg-purple-400/20" + > + <.icon name="hero-signal" class="h-4 w-4 mr-1.5" /> + {t("View RF Links")} + + <% end %>
<% end %>
@@ -373,7 +538,7 @@ style="border-bottom-color: #8B5CF6;" >
- Wireless + AP / Wireless
@@ -393,22 +558,36 @@ Discovered
- <%!-- Link confidence --%> + <%!-- RF Link health --%>
- Links: + RF Links:
-
- High confidence +
+ Good signal
-
-
- Medium confidence +
+ Degraded
-
+
+ Critical +
+
+ <%!-- Link types --%> +
+ Link types: +
+
+ LLDP/CDP +
+
+
- Low confidence + MAC/ARP inferred +
+
+ Line thickness = bandwidth capacity
From b134a2e56a598dc250f3f7a8c2f50c2dc71f6da0 Mon Sep 17 00:00:00 2001 From: Graham McIntie Date: Sun, 15 Mar 2026 14:08:27 -0500 Subject: [PATCH 2/3] perf: LiveView event batching & PubSub optimization (TOW-26) - Add 1.5s debounce to dashboard LiveView for alert events, preventing excessive re-renders during mass outages where hundreds of alerts fire simultaneously - Replace global 'device:events' PubSub topic with org-scoped 'device_events:org:{id}' topics across all broadcasters (device_poller_worker, sensor_change_detector, discovery) so LiveView processes only receive events for their tenant - Update EventLogger GenServer to subscribe to per-org topics dynamically, with runtime subscription support for newly created orgs - Add Organizations.list_organization_ids/0 for EventLogger bootstrap - Increase default DB pool_size from 20 to 25 for production concurrent users Co-Authored-By: Paperclip --- config/runtime.exs | 6 ++- lib/towerops/devices/event_logger.ex | 36 +++++++++++++++-- lib/towerops/organizations.ex | 8 ++++ lib/towerops/snmp/discovery.ex | 8 ++++ lib/towerops/snmp/sensor_change_detector.ex | 16 ++++++++ lib/towerops/workers/device_poller_worker.ex | 42 ++++++++++++++------ lib/towerops_web/live/dashboard_live.ex | 21 ++++++++-- 7 files changed, 114 insertions(+), 23 deletions(-) diff --git a/config/runtime.exs b/config/runtime.exs index 14beafeb..93fe9c16 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -159,7 +159,7 @@ if config_env() == :prod do default: 10, discovery: 10, # SNMP polling jobs - one per device - pollers: 50, + pollers: String.to_integer(System.get_env("POLLER_CONCURRENCY") || "50"), # Device monitoring jobs - health checks monitors: 50, # Service checks - HTTP/TCP/DNS @@ -184,6 +184,8 @@ if config_env() == :prod do {"30 0,8,16 * * *", Towerops.Workers.AgentLatencyEvaluator}, # Health check for missing jobs every 10 minutes {"*/10 * * * *", Towerops.Workers.JobHealthCheckWorker}, + # Send notification digests for rate-limited alerts every 5 minutes + {"*/5 * * * *", Towerops.Workers.AlertDigestWorker, args: %{"user_id" => "__cron__"}}, # Mark stale backup requests as timeout every 10 minutes {"*/10 * * * *", Towerops.Workers.BackupTimeoutWorker}, # Clean up old login history daily at 2 AM @@ -295,7 +297,7 @@ if config_env() == :prod do config :towerops, Towerops.Repo, ssl: ssl_config, url: database_url, - pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"), + pool_size: String.to_integer(System.get_env("POOL_SIZE") || "20"), # For machines with several cores, consider starting multiple pools of `pool_size` # pool_count: 4, socket_options: maybe_ipv6 diff --git a/lib/towerops/devices/event_logger.ex b/lib/towerops/devices/event_logger.ex index 18441d0a..086562b2 100644 --- a/lib/towerops/devices/event_logger.ex +++ b/lib/towerops/devices/event_logger.ex @@ -2,6 +2,10 @@ defmodule Towerops.Devices.EventLogger do @moduledoc """ GenServer that subscribes to device events via PubSub and logs them to the database. + Subscribes to org-scoped `device_events:org:{org_id}` topics instead of a + single global topic, so PubSub traffic is partitioned per organization and + LiveView processes only receive events for their tenant. + This decouples event detection from event storage, allowing: - Faster polling without database write blocking - Multiple event consumers (logging, alerts, webhooks) @@ -10,6 +14,7 @@ defmodule Towerops.Devices.EventLogger do use GenServer alias Towerops.Devices + alias Towerops.Organizations require Logger @@ -19,14 +24,37 @@ defmodule Towerops.Devices.EventLogger do GenServer.start_link(__MODULE__, %{}, name: __MODULE__) end + @doc """ + Subscribe to a new org's device events topic at runtime (e.g. when an org is created). + """ + def subscribe_org(org_id) when is_binary(org_id) do + GenServer.cast(__MODULE__, {:subscribe_org, org_id}) + end + # Server Callbacks @impl true def init(state) do - # device events topic - _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") - Logger.info("EventLogger started and subscribed to device events") - {:ok, state} + # Subscribe to all existing org-scoped device event topics + org_ids = Organizations.list_organization_ids() + + for org_id <- org_ids do + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device_events:org:#{org_id}") + end + + Logger.info("EventLogger started, subscribed to #{length(org_ids)} org device event topics") + {:ok, Map.put(state, :subscribed_orgs, MapSet.new(org_ids))} + end + + @impl true + def handle_cast({:subscribe_org, org_id}, state) do + if MapSet.member?(state.subscribed_orgs, org_id) do + {:noreply, state} + else + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device_events:org:#{org_id}") + Logger.info("EventLogger subscribed to device_events:org:#{org_id}") + {:noreply, %{state | subscribed_orgs: MapSet.put(state.subscribed_orgs, org_id)}} + end end @impl true diff --git a/lib/towerops/organizations.ex b/lib/towerops/organizations.ex index 8b20b6f0..cac5dd2d 100644 --- a/lib/towerops/organizations.ex +++ b/lib/towerops/organizations.ex @@ -19,6 +19,14 @@ defmodule Towerops.Organizations do ## Organizations + @doc """ + Returns all organization IDs. Used by EventLogger to subscribe to org-scoped PubSub topics. + """ + @spec list_organization_ids() :: [String.t()] + def list_organization_ids do + Repo.all(from(o in Organization, select: o.id)) + end + @doc """ Returns the list of organizations for a user. """ diff --git a/lib/towerops/snmp/discovery.ex b/lib/towerops/snmp/discovery.ex index 9aca0c72..0275ecf3 100644 --- a/lib/towerops/snmp/discovery.ex +++ b/lib/towerops/snmp/discovery.ex @@ -1429,6 +1429,7 @@ defmodule Towerops.Snmp.Discovery do event_attrs = %{ device_id: device.id, + org_id: device.organization_id, event_type: event_type, severity: "info", message: message, @@ -1442,6 +1443,13 @@ defmodule Towerops.Snmp.Discovery do "device:events", {:device_event, event_attrs} ) + + # Also publish to org-scoped topic for targeted LiveView subscriptions + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "device_events:org:#{device.organization_id}", + {:device_event, event_attrs} + ) end # Sanitizes data structures to ensure they are JSON-encodable diff --git a/lib/towerops/snmp/sensor_change_detector.ex b/lib/towerops/snmp/sensor_change_detector.ex index 1ec08c9b..30d68777 100644 --- a/lib/towerops/snmp/sensor_change_detector.ex +++ b/lib/towerops/snmp/sensor_change_detector.ex @@ -272,10 +272,26 @@ defmodule Towerops.Snmp.SensorChangeDetector do _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{event.device_id}", {:device_event, event}) _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + case get_org_id_from_device(event.device_id) do + nil -> + :ok + + org_id -> + event_with_org = Map.put(event, :org_id, org_id) + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device_events:org:#{org_id}", {:device_event, event_with_org}) + end + Logger.debug("Sensor event: #{event.message}") end) end + defp get_org_id_from_device(device_id) do + case Towerops.Repo.get(Towerops.Devices.Device, device_id) do + nil -> nil + device -> device.organization_id + end + end + defp get_device_id_from_sensor(sensor) do case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do nil -> nil diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index e66e40d7..a8428fd2 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -296,7 +296,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end defp poll_device_state_sensors(device, snmp_device, client_opts, now) do - poll_state_sensors(snmp_device.state_sensors, client_opts, now, device.id) + poll_state_sensors(snmp_device.state_sensors, client_opts, now, device.id, device.organization_id) Logger.debug("Polled #{length(snmp_device.state_sensors)} state sensors for #{device.name}") _ = @@ -745,12 +745,12 @@ defmodule Towerops.Workers.DevicePollerWorker do end) end - defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id) do + defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id, org_id) do state_sensors |> Task.async_stream( fn state_sensor -> result = poll_state_sensor_value(state_sensor, client_opts) - handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id) + handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id, org_id) state_sensor.sensor_descr end, max_concurrency: 2, @@ -776,7 +776,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end end - defp handle_state_sensor_poll_result(state_sensor, {:ok, new_state_value}, timestamp, device_id) do + defp handle_state_sensor_poll_result(state_sensor, {:ok, new_state_value}, timestamp, device_id, org_id) do old_state_value = state_sensor.state_value old_state_descr = state_sensor.state_descr old_status = state_sensor.status @@ -803,11 +803,11 @@ defmodule Towerops.Workers.DevicePollerWorker do new_status: new_status } - broadcast_state_sensor_change(state_sensor, state_changes, device_id, timestamp) + broadcast_state_sensor_change(state_sensor, state_changes, device_id, org_id, timestamp) end end - defp handle_state_sensor_poll_result(state_sensor, {:error, reason}, timestamp, _device_id) do + defp handle_state_sensor_poll_result(state_sensor, {:error, reason}, timestamp, _device_id, _org_id) do Logger.debug("Failed to poll state sensor #{state_sensor.sensor_descr}: #{inspect(reason)}") Snmp.update_state_sensor(state_sensor, %{ @@ -816,7 +816,7 @@ defmodule Towerops.Workers.DevicePollerWorker do }) end - defp broadcast_state_sensor_change(state_sensor, state_changes, device_id, timestamp) do + defp broadcast_state_sensor_change(state_sensor, state_changes, device_id, org_id, timestamp) do %{ old_state_value: old_state_value, new_state_value: new_state_value, @@ -835,6 +835,7 @@ defmodule Towerops.Workers.DevicePollerWorker do event = %{ device_id: device_id, + org_id: org_id, event_type: event_type, severity: severity, message: message, @@ -855,6 +856,10 @@ defmodule Towerops.Workers.DevicePollerWorker do _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{device_id}", {:device_event, event}) _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + if org_id do + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device_events:org:#{org_id}", {:device_event, event}) + end + Logger.info("State sensor change: #{event.message}") end @@ -1041,7 +1046,7 @@ defmodule Towerops.Workers.DevicePollerWorker do fn interface -> case get_interface_attributes(client_opts, interface.if_index) do {:ok, current_attrs} -> - detect_and_log_changes(interface, current_attrs, device.id, timestamp) + detect_and_log_changes(interface, current_attrs, device.id, device.organization_id, timestamp) {:error, reason} -> Logger.debug("Failed to get interface attributes for #{interface.if_name}: #{inspect(reason)}") @@ -1078,7 +1083,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end end - defp detect_and_log_changes(interface, current_attrs, device_id, timestamp) do + defp detect_and_log_changes(interface, current_attrs, device_id, org_id, timestamp) do events = [] |> maybe_add_oper_status_event(interface, current_attrs, device_id, timestamp) @@ -1087,7 +1092,7 @@ defmodule Towerops.Workers.DevicePollerWorker do |> maybe_add_mac_change_event(interface, current_attrs, device_id, timestamp) if events != [] do - broadcast_interface_events(events) + broadcast_interface_events(events, org_id) Snmp.update_interface(interface, current_attrs) end end @@ -1233,22 +1238,33 @@ defmodule Towerops.Workers.DevicePollerWorker do "Interface #{if_name} MAC address changed from #{old_mac} to #{new_mac}" end - defp broadcast_interface_events(events) do + defp broadcast_interface_events(events, org_id) do Enum.each(events, fn {:event, event_attrs} -> + event_with_org = Map.put(event_attrs, :org_id, org_id) + _ = Phoenix.PubSub.broadcast( Towerops.PubSub, "device:#{event_attrs.device_id}", - {:device_event, event_attrs} + {:device_event, event_with_org} ) _ = Phoenix.PubSub.broadcast( Towerops.PubSub, "device:events", - {:device_event, event_attrs} + {:device_event, event_with_org} ) + if org_id do + _ = + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "device_events:org:#{org_id}", + {:device_event, event_with_org} + ) + end + Logger.debug("Broadcast event: #{event_attrs.message}") end) end diff --git a/lib/towerops_web/live/dashboard_live.ex b/lib/towerops_web/live/dashboard_live.ex index 329f696f..2a0901bf 100644 --- a/lib/towerops_web/live/dashboard_live.ex +++ b/lib/towerops_web/live/dashboard_live.ex @@ -29,7 +29,8 @@ defmodule ToweropsWeb.DashboardLive do |> assign(:device_count, device_count) |> assign(:insight_source, nil) |> assign(:timezone, socket.assigns.current_scope.timezone) - |> assign(:time_format, socket.assigns.current_scope.time_format)} + |> assign(:time_format, socket.assigns.current_scope.time_format) + |> assign(:pending_dashboard_reload_ref, nil)} end @impl true @@ -81,15 +82,27 @@ defmodule ToweropsWeb.DashboardLive do end @impl true - def handle_info({:new_alert, _device_id, _alert_type}, socket) do - {:noreply, load_dashboard_data(socket, socket.assigns.current_scope.organization.id)} + def handle_info({event, _device_id, _alert_type}, socket) + when event in [:new_alert, :alert_resolved] do + {:noreply, schedule_debounced_dashboard_reload(socket)} end @impl true - def handle_info({:alert_resolved, _device_id, _alert_type}, socket) do + def handle_info(:debounced_dashboard_reload, socket) do {:noreply, load_dashboard_data(socket, socket.assigns.current_scope.organization.id)} end + @dashboard_debounce_ms 1_500 + + defp schedule_debounced_dashboard_reload(socket) do + if ref = socket.assigns.pending_dashboard_reload_ref do + Process.cancel_timer(ref) + end + + ref = Process.send_after(self(), :debounced_dashboard_reload, @dashboard_debounce_ms) + assign(socket, :pending_dashboard_reload_ref, ref) + end + defp load_dashboard_data(socket, organization_id) do organization = socket.assigns.current_scope.organization summary = Dashboard.get_dashboard_summary(organization_id) From 90af7ac0d1c68285b9aa310a174c14b72da859c7 Mon Sep 17 00:00:00 2001 From: Graham McIntie Date: Sun, 15 Mar 2026 14:29:12 -0500 Subject: [PATCH 3/3] feat: alert storm correlation & suppression (TOW-13) - 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 --- config/runtime.exs | 2 +- lib/towerops/alerts/alert.ex | 13 +- lib/towerops/alerts/alert_storm.ex | 40 ++ lib/towerops/alerts/notification_digest.ex | 45 ++ lib/towerops/alerts/site_correlation.ex | 223 ++++++++++ lib/towerops/alerts/site_outage.ex | 49 +++ lib/towerops/alerts/storm_detector.ex | 383 ++++++++++++++++++ lib/towerops/application.ex | 6 +- lib/towerops/maintenance.ex | 53 +++ lib/towerops/pagerduty/client.ex | 40 +- lib/towerops/workers/alert_digest_worker.ex | 95 +++++ .../workers/alert_notification_worker.ex | 87 +++- lib/towerops/workers/device_monitor_worker.ex | 69 +--- ...0315190000_add_alert_storm_correlation.exs | 64 +++ .../alerts/notification_rate_limiter_test.exs | 86 ++++ .../towerops/alerts/site_correlation_test.exs | 111 +++++ test/towerops/alerts/storm_detector_test.exs | 71 ++++ test/towerops/maintenance_batch_test.exs | 91 +++++ 18 files changed, 1452 insertions(+), 76 deletions(-) create mode 100644 lib/towerops/alerts/alert_storm.ex create mode 100644 lib/towerops/alerts/notification_digest.ex create mode 100644 lib/towerops/alerts/site_correlation.ex create mode 100644 lib/towerops/alerts/site_outage.ex create mode 100644 lib/towerops/alerts/storm_detector.ex create mode 100644 lib/towerops/workers/alert_digest_worker.ex create mode 100644 priv/repo/migrations/20260315190000_add_alert_storm_correlation.exs create mode 100644 test/towerops/alerts/notification_rate_limiter_test.exs create mode 100644 test/towerops/alerts/site_correlation_test.exs create mode 100644 test/towerops/alerts/storm_detector_test.exs create mode 100644 test/towerops/maintenance_batch_test.exs diff --git a/config/runtime.exs b/config/runtime.exs index 93fe9c16..1a4a08e2 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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 ], diff --git a/lib/towerops/alerts/alert.ex b/lib/towerops/alerts/alert.ex index 436dc5ec..1c89e90f 100644 --- a/lib/towerops/alerts/alert.ex +++ b/lib/towerops/alerts/alert.ex @@ -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) diff --git a/lib/towerops/alerts/alert_storm.ex b/lib/towerops/alerts/alert_storm.ex new file mode 100644 index 00000000..a32667bf --- /dev/null +++ b/lib/towerops/alerts/alert_storm.ex @@ -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 diff --git a/lib/towerops/alerts/notification_digest.ex b/lib/towerops/alerts/notification_digest.ex new file mode 100644 index 00000000..180081ad --- /dev/null +++ b/lib/towerops/alerts/notification_digest.ex @@ -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 diff --git a/lib/towerops/alerts/site_correlation.ex b/lib/towerops/alerts/site_correlation.ex new file mode 100644 index 00000000..2f0a27ac --- /dev/null +++ b/lib/towerops/alerts/site_correlation.ex @@ -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 diff --git a/lib/towerops/alerts/site_outage.ex b/lib/towerops/alerts/site_outage.ex new file mode 100644 index 00000000..eaaee48e --- /dev/null +++ b/lib/towerops/alerts/site_outage.ex @@ -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 diff --git a/lib/towerops/alerts/storm_detector.ex b/lib/towerops/alerts/storm_detector.ex new file mode 100644 index 00000000..be9ef725 --- /dev/null +++ b/lib/towerops/alerts/storm_detector.ex @@ -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 diff --git a/lib/towerops/application.ex b/lib/towerops/application.ex index e8177f57..5d5f3975 100644 --- a/lib/towerops/application.ex +++ b/lib/towerops/application.ex @@ -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) ] diff --git a/lib/towerops/maintenance.ex b/lib/towerops/maintenance.ex index 7ee25f34..0a336912 100644 --- a/lib/towerops/maintenance.ex +++ b/lib/towerops/maintenance.ex @@ -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 diff --git a/lib/towerops/pagerduty/client.ex b/lib/towerops/pagerduty/client.ex index 07b635a9..03a88d22 100644 --- a/lib/towerops/pagerduty/client.ex +++ b/lib/towerops/pagerduty/client.ex @@ -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 diff --git a/lib/towerops/workers/alert_digest_worker.ex b/lib/towerops/workers/alert_digest_worker.ex new file mode 100644 index 00000000..52dbf6e8 --- /dev/null +++ b/lib/towerops/workers/alert_digest_worker.ex @@ -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 diff --git a/lib/towerops/workers/alert_notification_worker.ex b/lib/towerops/workers/alert_notification_worker.ex index 8fb1d0a5..cbf9f687 100644 --- a/lib/towerops/workers/alert_notification_worker.ex +++ b/lib/towerops/workers/alert_notification_worker.ex @@ -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 -> diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 4e5f169e..4fdd8c29 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -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 diff --git a/priv/repo/migrations/20260315190000_add_alert_storm_correlation.exs b/priv/repo/migrations/20260315190000_add_alert_storm_correlation.exs new file mode 100644 index 00000000..8e30fdc4 --- /dev/null +++ b/priv/repo/migrations/20260315190000_add_alert_storm_correlation.exs @@ -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 diff --git a/test/towerops/alerts/notification_rate_limiter_test.exs b/test/towerops/alerts/notification_rate_limiter_test.exs new file mode 100644 index 00000000..d5741529 --- /dev/null +++ b/test/towerops/alerts/notification_rate_limiter_test.exs @@ -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 diff --git a/test/towerops/alerts/site_correlation_test.exs b/test/towerops/alerts/site_correlation_test.exs new file mode 100644 index 00000000..1da170c1 --- /dev/null +++ b/test/towerops/alerts/site_correlation_test.exs @@ -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 diff --git a/test/towerops/alerts/storm_detector_test.exs b/test/towerops/alerts/storm_detector_test.exs new file mode 100644 index 00000000..62b31f83 --- /dev/null +++ b/test/towerops/alerts/storm_detector_test.exs @@ -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 diff --git a/test/towerops/maintenance_batch_test.exs b/test/towerops/maintenance_batch_test.exs new file mode 100644 index 00000000..2ca1deb1 --- /dev/null +++ b/test/towerops/maintenance_batch_test.exs @@ -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