Merge pull request 'feature/alert-storm-correlation' (#30) from feature/alert-storm-correlation into main
Reviewed-on: graham/towerops-web#30
This commit is contained in:
commit
a98b887471
29 changed files with 2341 additions and 149 deletions
268
assets/js/app.ts
268
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<string, string> = {
|
||||
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<string, {x: number, y: number}> = {}
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -159,14 +159,14 @@ 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
|
||||
checks: 50,
|
||||
check_executors: 50,
|
||||
# Alert notifications (PagerDuty, email, etc.)
|
||||
notifications: 10,
|
||||
notifications: 25,
|
||||
maintenance: 5,
|
||||
weather: 2
|
||||
],
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ defmodule Towerops.Alerts.Alert do
|
|||
|
||||
alias Ecto.Association.NotLoaded
|
||||
alias Towerops.Accounts.User
|
||||
alias Towerops.Alerts.SiteOutage
|
||||
alias Towerops.Devices.Device
|
||||
alias Towerops.Monitoring.Check
|
||||
alias Towerops.Organizations.Organization
|
||||
|
|
@ -28,6 +29,10 @@ defmodule Towerops.Alerts.Alert do
|
|||
field :message, :string
|
||||
field :gaiia_impact, :map
|
||||
|
||||
# Storm correlation
|
||||
field :storm_suppressed, :boolean, default: false
|
||||
belongs_to :site_outage, SiteOutage
|
||||
|
||||
# Check-related fields (added in ExtendAlertsForChecks migration)
|
||||
field :severity, :integer
|
||||
field :notification_sent, :boolean, default: false
|
||||
|
|
@ -63,6 +68,9 @@ defmodule Towerops.Alerts.Alert do
|
|||
acknowledged_by: NotLoaded.t() | User.t() | nil,
|
||||
organization_id: Ecto.UUID.t() | nil,
|
||||
organization: NotLoaded.t() | Organization.t() | nil,
|
||||
site_outage_id: Ecto.UUID.t() | nil,
|
||||
site_outage: NotLoaded.t() | SiteOutage.t() | nil,
|
||||
storm_suppressed: boolean(),
|
||||
inserted_at: DateTime.t(),
|
||||
updated_at: DateTime.t()
|
||||
}
|
||||
|
|
@ -85,7 +93,9 @@ defmodule Towerops.Alerts.Alert do
|
|||
:notification_sent,
|
||||
:notification_sent_at,
|
||||
:output,
|
||||
:organization_id
|
||||
:organization_id,
|
||||
:site_outage_id,
|
||||
:storm_suppressed
|
||||
])
|
||||
|> validate_required([:alert_type, :triggered_at])
|
||||
|> validate_inclusion(:severity, [1, 2])
|
||||
|
|
@ -94,6 +104,7 @@ defmodule Towerops.Alerts.Alert do
|
|||
|> foreign_key_constraint(:check_id)
|
||||
|> foreign_key_constraint(:acknowledged_by_id)
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
|> foreign_key_constraint(:site_outage_id)
|
||||
end
|
||||
|
||||
# Ensure either device_id or check_id is present (but not required for flexibility)
|
||||
|
|
|
|||
40
lib/towerops/alerts/alert_storm.ex
Normal file
40
lib/towerops/alerts/alert_storm.ex
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
defmodule Towerops.Alerts.AlertStorm do
|
||||
@moduledoc """
|
||||
Schema for alert storms — periods when alert rate exceeds threshold for an org.
|
||||
|
||||
When > threshold alerts fire within a time window, individual notifications
|
||||
are suppressed and a single summary notification is sent instead.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
import Ecto.Changeset
|
||||
|
||||
alias Towerops.Organizations.Organization
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
schema "alert_storms" do
|
||||
field :started_at, :utc_datetime
|
||||
field :ended_at, :utc_datetime
|
||||
field :alert_count, :integer, default: 0
|
||||
field :suppressed_count, :integer, default: 0
|
||||
field :summary_notification_sent, :boolean, default: false
|
||||
|
||||
belongs_to :organization, Organization
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
def changeset(storm, attrs) do
|
||||
storm
|
||||
|> cast(attrs, [
|
||||
:organization_id,
|
||||
:started_at,
|
||||
:ended_at,
|
||||
:alert_count,
|
||||
:suppressed_count,
|
||||
:summary_notification_sent
|
||||
])
|
||||
|> validate_required([:organization_id, :started_at])
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
end
|
||||
end
|
||||
45
lib/towerops/alerts/notification_digest.ex
Normal file
45
lib/towerops/alerts/notification_digest.ex
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
defmodule Towerops.Alerts.NotificationDigest do
|
||||
@moduledoc """
|
||||
Schema for notification rate limiting per user.
|
||||
|
||||
Tracks how many notifications a user has received within a time window.
|
||||
When the limit is exceeded, additional alerts are queued for digest delivery.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
import Ecto.Changeset
|
||||
|
||||
alias Towerops.Accounts.User
|
||||
alias Towerops.Organizations.Organization
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
schema "notification_digests" do
|
||||
field :window_start, :utc_datetime
|
||||
field :notification_count, :integer, default: 0
|
||||
field :digest_sent, :boolean, default: false
|
||||
field :digest_sent_at, :utc_datetime
|
||||
field :suppressed_alert_ids, {:array, Ecto.UUID}, default: []
|
||||
|
||||
belongs_to :user, User
|
||||
belongs_to :organization, Organization
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
def changeset(digest, attrs) do
|
||||
digest
|
||||
|> cast(attrs, [
|
||||
:user_id,
|
||||
:organization_id,
|
||||
:window_start,
|
||||
:notification_count,
|
||||
:digest_sent,
|
||||
:digest_sent_at,
|
||||
:suppressed_alert_ids
|
||||
])
|
||||
|> validate_required([:user_id, :organization_id, :window_start])
|
||||
|> foreign_key_constraint(:user_id)
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
|> unique_constraint([:user_id, :window_start], name: :notification_digests_user_window_idx)
|
||||
end
|
||||
end
|
||||
169
lib/towerops/alerts/notification_rate_limiter.ex
Normal file
169
lib/towerops/alerts/notification_rate_limiter.ex
Normal file
|
|
@ -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
|
||||
223
lib/towerops/alerts/site_correlation.ex
Normal file
223
lib/towerops/alerts/site_correlation.ex
Normal file
|
|
@ -0,0 +1,223 @@
|
|||
defmodule Towerops.Alerts.SiteCorrelation do
|
||||
@moduledoc """
|
||||
Correlates device-down alerts at the same site into a single "Site Outage".
|
||||
|
||||
When >N devices at the same site go down within T seconds, creates a
|
||||
single site outage record with one notification instead of N individual ones.
|
||||
|
||||
## Thresholds
|
||||
|
||||
- Site correlation threshold: 3+ devices at same site within 120 seconds
|
||||
- These are configurable via application env
|
||||
|
||||
## Flow
|
||||
|
||||
1. DeviceMonitorWorker detects device down
|
||||
2. Before creating individual alert, calls `check_site_correlation/2`
|
||||
3. If site outage threshold met, returns `{:site_outage, outage}` — caller
|
||||
links alert to outage and suppresses individual notification
|
||||
4. If not, returns `:no_correlation` — normal alert flow
|
||||
"""
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Towerops.Alerts.Alert
|
||||
alias Towerops.Alerts.SiteOutage
|
||||
alias Towerops.Devices.Device
|
||||
alias Towerops.Repo
|
||||
|
||||
require Logger
|
||||
|
||||
# Default: 3+ devices at same site within 120s = site outage
|
||||
@default_device_threshold 3
|
||||
@default_time_window_seconds 120
|
||||
|
||||
@doc """
|
||||
Check if this device going down should be correlated into a site outage.
|
||||
|
||||
Returns:
|
||||
- `{:site_outage, outage}` — device is part of a site outage, suppress individual notification
|
||||
- `:no_correlation` — proceed with normal alert flow
|
||||
"""
|
||||
@spec check_site_correlation(Device.t(), DateTime.t()) ::
|
||||
{:site_outage, SiteOutage.t()} | :no_correlation
|
||||
def check_site_correlation(%Device{site_id: nil}, _now), do: :no_correlation
|
||||
|
||||
def check_site_correlation(%Device{} = device, now) do
|
||||
threshold = device_threshold()
|
||||
window = time_window_seconds()
|
||||
cutoff = DateTime.add(now, -window, :second)
|
||||
|
||||
# Check for existing active site outage
|
||||
case get_active_outage(device.site_id) do
|
||||
%SiteOutage{} = outage ->
|
||||
# Already in site outage mode — just increment
|
||||
update_outage_count(outage, device)
|
||||
{:site_outage, outage}
|
||||
|
||||
nil ->
|
||||
# Count recent device_down alerts at this site
|
||||
recent_down_count = count_recent_site_alerts(device.site_id, cutoff)
|
||||
|
||||
# +1 for the current device about to go down
|
||||
if recent_down_count + 1 >= threshold do
|
||||
# Threshold met — create site outage
|
||||
case create_site_outage(device, now, recent_down_count + 1) do
|
||||
{:ok, outage} ->
|
||||
# Retroactively link existing alerts to this outage
|
||||
link_existing_alerts(device.site_id, cutoff, outage.id)
|
||||
{:site_outage, outage}
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("Failed to create site outage: #{inspect(reason)}")
|
||||
:no_correlation
|
||||
end
|
||||
else
|
||||
:no_correlation
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Check if a site outage should be resolved (all devices back up).
|
||||
Called when a device comes back up.
|
||||
"""
|
||||
@spec check_outage_resolution(Device.t()) :: :ok
|
||||
def check_outage_resolution(%Device{site_id: nil}), do: :ok
|
||||
|
||||
def check_outage_resolution(%Device{} = device) do
|
||||
case get_active_outage(device.site_id) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
outage ->
|
||||
# Check if any devices at this site are still down
|
||||
still_down = count_active_down_at_site(device.site_id)
|
||||
|
||||
if still_down <= 0 do
|
||||
resolve_outage(outage)
|
||||
else
|
||||
# Update the outage message with current count
|
||||
outage
|
||||
|> SiteOutage.changeset(%{
|
||||
device_count: still_down,
|
||||
message: "Site outage: #{still_down} device(s) still down at site"
|
||||
})
|
||||
|> Repo.update()
|
||||
end
|
||||
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Get the active site outage for a site, if any.
|
||||
"""
|
||||
@spec get_active_outage(String.t()) :: SiteOutage.t() | nil
|
||||
def get_active_outage(site_id) do
|
||||
Repo.one(
|
||||
from(o in SiteOutage,
|
||||
where: o.site_id == ^site_id,
|
||||
where: is_nil(o.resolved_at),
|
||||
limit: 1
|
||||
)
|
||||
)
|
||||
end
|
||||
|
||||
## Private
|
||||
|
||||
defp count_recent_site_alerts(site_id, cutoff) do
|
||||
Repo.aggregate(
|
||||
from(a in Alert,
|
||||
join: d in Device, on: d.id == a.device_id,
|
||||
where: d.site_id == ^site_id,
|
||||
where: a.alert_type == "device_down",
|
||||
where: is_nil(a.resolved_at),
|
||||
where: a.triggered_at >= ^cutoff
|
||||
),
|
||||
:count
|
||||
)
|
||||
end
|
||||
|
||||
defp count_active_down_at_site(site_id) do
|
||||
Repo.aggregate(
|
||||
from(a in Alert,
|
||||
join: d in Device, on: d.id == a.device_id,
|
||||
where: d.site_id == ^site_id,
|
||||
where: a.alert_type == "device_down",
|
||||
where: is_nil(a.resolved_at)
|
||||
),
|
||||
:count
|
||||
)
|
||||
end
|
||||
|
||||
defp create_site_outage(device, now, device_count) do
|
||||
total_at_site = count_devices_at_site(device.site_id)
|
||||
site = Repo.get(Towerops.Sites.Site, device.site_id)
|
||||
site_name = if site, do: site.name, else: "Unknown Site"
|
||||
|
||||
%SiteOutage{}
|
||||
|> SiteOutage.changeset(%{
|
||||
site_id: device.site_id,
|
||||
organization_id: device.organization_id,
|
||||
started_at: DateTime.truncate(now, :second),
|
||||
device_count: device_count,
|
||||
total_devices_at_site: total_at_site,
|
||||
message: "Site outage at #{site_name}: #{device_count}/#{total_at_site} devices down"
|
||||
})
|
||||
|> Repo.insert()
|
||||
end
|
||||
|
||||
defp update_outage_count(outage, _device) do
|
||||
new_count = (outage.device_count || 0) + 1
|
||||
|
||||
outage
|
||||
|> SiteOutage.changeset(%{device_count: new_count})
|
||||
|> Repo.update()
|
||||
end
|
||||
|
||||
defp link_existing_alerts(site_id, cutoff, outage_id) do
|
||||
from(a in Alert,
|
||||
join: d in Device, on: d.id == a.device_id,
|
||||
where: d.site_id == ^site_id,
|
||||
where: a.alert_type == "device_down",
|
||||
where: is_nil(a.resolved_at),
|
||||
where: a.triggered_at >= ^cutoff
|
||||
)
|
||||
|> Repo.update_all(set: [site_outage_id: outage_id, storm_suppressed: true])
|
||||
end
|
||||
|
||||
defp resolve_outage(outage) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
outage
|
||||
|> SiteOutage.changeset(%{resolved_at: now})
|
||||
|> Repo.update()
|
||||
|
||||
Logger.info("Site outage resolved: #{outage.id} at site #{outage.site_id}")
|
||||
|
||||
# Broadcast resolution
|
||||
Phoenix.PubSub.broadcast(
|
||||
Towerops.PubSub,
|
||||
"alerts:org:#{outage.organization_id}:site_outage",
|
||||
{:site_outage_resolved, outage}
|
||||
)
|
||||
end
|
||||
|
||||
defp count_devices_at_site(site_id) do
|
||||
Repo.aggregate(
|
||||
from(d in Device, where: d.site_id == ^site_id),
|
||||
:count
|
||||
)
|
||||
end
|
||||
|
||||
defp device_threshold do
|
||||
Application.get_env(:towerops, :alert_storm, [])
|
||||
|> Keyword.get(:site_correlation_threshold, @default_device_threshold)
|
||||
end
|
||||
|
||||
defp time_window_seconds do
|
||||
Application.get_env(:towerops, :alert_storm, [])
|
||||
|> Keyword.get(:site_correlation_window_seconds, @default_time_window_seconds)
|
||||
end
|
||||
end
|
||||
49
lib/towerops/alerts/site_outage.ex
Normal file
49
lib/towerops/alerts/site_outage.ex
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
defmodule Towerops.Alerts.SiteOutage do
|
||||
@moduledoc """
|
||||
Schema for site outages — correlated device-down alerts at the same site.
|
||||
|
||||
When multiple devices at the same site go down within a short window,
|
||||
they are grouped into a single site outage with one notification
|
||||
instead of N individual notifications.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
import Ecto.Changeset
|
||||
|
||||
alias Towerops.Alerts.Alert
|
||||
alias Towerops.Organizations.Organization
|
||||
alias Towerops.Sites.Site
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
schema "site_outages" do
|
||||
field :started_at, :utc_datetime
|
||||
field :resolved_at, :utc_datetime
|
||||
field :device_count, :integer, default: 0
|
||||
field :total_devices_at_site, :integer, default: 0
|
||||
field :message, :string
|
||||
field :notification_sent, :boolean, default: false
|
||||
|
||||
belongs_to :site, Site
|
||||
belongs_to :organization, Organization
|
||||
has_many :alerts, Alert, foreign_key: :site_outage_id
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
def changeset(outage, attrs) do
|
||||
outage
|
||||
|> cast(attrs, [
|
||||
:site_id,
|
||||
:organization_id,
|
||||
:started_at,
|
||||
:resolved_at,
|
||||
:device_count,
|
||||
:total_devices_at_site,
|
||||
:message,
|
||||
:notification_sent
|
||||
])
|
||||
|> validate_required([:site_id, :organization_id, :started_at])
|
||||
|> foreign_key_constraint(:site_id)
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
end
|
||||
end
|
||||
383
lib/towerops/alerts/storm_detector.ex
Normal file
383
lib/towerops/alerts/storm_detector.ex
Normal file
|
|
@ -0,0 +1,383 @@
|
|||
defmodule Towerops.Alerts.StormDetector do
|
||||
@moduledoc """
|
||||
Detects alert storms and correlates device-down alerts at the site level.
|
||||
|
||||
When multiple devices at the same site go down within a short window,
|
||||
this module suppresses individual alerts and creates a single "site_outage"
|
||||
alert instead. This prevents overwhelming on-call engineers with dozens
|
||||
of individual notifications during tower power outages.
|
||||
|
||||
## How it works
|
||||
|
||||
1. When a device goes down, instead of immediately creating an alert,
|
||||
the caller registers the event with this GenServer.
|
||||
2. The GenServer buffers events per site for a configurable window (default: 30s).
|
||||
3. After the window expires:
|
||||
- If >= threshold devices are down at the same site → create one "site_outage" alert
|
||||
- If < threshold → create individual "device_down" alerts as normal
|
||||
4. A global storm detector triggers if > storm_threshold alerts/minute across
|
||||
all sites → switches to digest-mode notifications.
|
||||
|
||||
## Configuration
|
||||
|
||||
- `:site_correlation_window_ms` - How long to buffer before deciding (default: 30_000)
|
||||
- `:site_correlation_threshold` - Min devices to trigger site-level alert (default: 3)
|
||||
- `:storm_threshold_per_minute` - Global storm mode trigger (default: 10)
|
||||
- `:storm_cooldown_ms` - How long storm mode stays active (default: 300_000 = 5 min)
|
||||
"""
|
||||
|
||||
use GenServer
|
||||
|
||||
alias Towerops.Alerts
|
||||
alias Towerops.Devices
|
||||
alias Towerops.Maintenance
|
||||
alias Towerops.Workers.AlertNotificationWorker
|
||||
|
||||
require Logger
|
||||
|
||||
@default_correlation_window_ms 30_000
|
||||
@default_correlation_threshold 3
|
||||
@default_storm_threshold_per_minute 10
|
||||
@default_storm_cooldown_ms 300_000
|
||||
|
||||
# --- Public API ---
|
||||
|
||||
def start_link(opts \\ []) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Registers a device-down event for storm detection.
|
||||
|
||||
Instead of creating an alert immediately, the event is buffered.
|
||||
After the correlation window, alerts are created — either as individual
|
||||
device_down alerts or as a single site_outage alert.
|
||||
|
||||
Returns :ok immediately (fire-and-forget).
|
||||
"""
|
||||
@spec register_device_down(map(), DateTime.t()) :: :ok
|
||||
def register_device_down(device, timestamp) do
|
||||
GenServer.cast(__MODULE__, {:device_down, device, timestamp})
|
||||
end
|
||||
|
||||
@doc """
|
||||
Checks if the system is currently in storm mode.
|
||||
In storm mode, notifications are batched into digests.
|
||||
"""
|
||||
@spec storm_mode?() :: boolean()
|
||||
def storm_mode? do
|
||||
GenServer.call(__MODULE__, :storm_mode?)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns current storm detector stats for debugging/monitoring.
|
||||
"""
|
||||
@spec stats() :: map()
|
||||
def stats do
|
||||
GenServer.call(__MODULE__, :stats)
|
||||
end
|
||||
|
||||
# --- GenServer Implementation ---
|
||||
|
||||
@impl true
|
||||
def init(opts) do
|
||||
state = %{
|
||||
# %{site_id => [%{device: device, timestamp: DateTime.t()}]}
|
||||
site_buffers: %{},
|
||||
# %{site_id => timer_ref}
|
||||
site_timers: %{},
|
||||
# Devices without a site get buffered under :no_site
|
||||
# Rolling count of alerts in the last minute
|
||||
alert_count_last_minute: 0,
|
||||
alert_timestamps: :queue.new(),
|
||||
# Storm mode
|
||||
storm_mode: false,
|
||||
storm_mode_until: nil,
|
||||
# Config
|
||||
correlation_window_ms:
|
||||
Keyword.get(opts, :correlation_window_ms, @default_correlation_window_ms),
|
||||
correlation_threshold:
|
||||
Keyword.get(opts, :correlation_threshold, @default_correlation_threshold),
|
||||
storm_threshold_per_minute:
|
||||
Keyword.get(opts, :storm_threshold_per_minute, @default_storm_threshold_per_minute),
|
||||
storm_cooldown_ms:
|
||||
Keyword.get(opts, :storm_cooldown_ms, @default_storm_cooldown_ms)
|
||||
}
|
||||
|
||||
{:ok, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_cast({:device_down, device, timestamp}, state) do
|
||||
site_id = device.site_id || :no_site
|
||||
|
||||
# Add to site buffer
|
||||
entry = %{device: device, timestamp: timestamp}
|
||||
|
||||
site_buffers =
|
||||
Map.update(state.site_buffers, site_id, [entry], fn existing ->
|
||||
[entry | existing]
|
||||
end)
|
||||
|
||||
# Start or reset the correlation window timer for this site
|
||||
site_timers =
|
||||
case Map.get(state.site_timers, site_id) do
|
||||
nil ->
|
||||
# First event for this site — start the window
|
||||
timer = Process.send_after(self(), {:flush_site, site_id}, state.correlation_window_ms)
|
||||
Map.put(state.site_timers, site_id, timer)
|
||||
|
||||
_existing_timer ->
|
||||
# Timer already running — don't reset, let the original window expire
|
||||
state.site_timers
|
||||
end
|
||||
|
||||
# Track alert rate for storm detection
|
||||
now_ms = System.system_time(:millisecond)
|
||||
|
||||
alert_timestamps =
|
||||
state.alert_timestamps
|
||||
|> :queue.in(now_ms)
|
||||
|> prune_old_timestamps(now_ms - 60_000)
|
||||
|
||||
alert_count = :queue.len(alert_timestamps)
|
||||
|
||||
# Check if we should enter storm mode
|
||||
{storm_mode, storm_mode_until} =
|
||||
if alert_count >= state.storm_threshold_per_minute and not state.storm_mode do
|
||||
Logger.warning(
|
||||
"Alert storm detected: #{alert_count} alerts in last minute, entering storm mode",
|
||||
alert_count: alert_count
|
||||
)
|
||||
|
||||
until = DateTime.add(DateTime.utc_now(), state.storm_cooldown_ms, :millisecond)
|
||||
{true, until}
|
||||
else
|
||||
{state.storm_mode, state.storm_mode_until}
|
||||
end
|
||||
|
||||
{:noreply,
|
||||
%{
|
||||
state
|
||||
| site_buffers: site_buffers,
|
||||
site_timers: site_timers,
|
||||
alert_timestamps: alert_timestamps,
|
||||
alert_count_last_minute: alert_count,
|
||||
storm_mode: storm_mode,
|
||||
storm_mode_until: storm_mode_until
|
||||
}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:flush_site, site_id}, state) do
|
||||
events = Map.get(state.site_buffers, site_id, [])
|
||||
|
||||
# Remove from buffers and timers
|
||||
site_buffers = Map.delete(state.site_buffers, site_id)
|
||||
site_timers = Map.delete(state.site_timers, site_id)
|
||||
|
||||
# Process the buffered events
|
||||
if length(events) >= state.correlation_threshold and site_id != :no_site do
|
||||
create_site_outage_alert(site_id, events, state.storm_mode)
|
||||
else
|
||||
create_individual_alerts(events, state.storm_mode)
|
||||
end
|
||||
|
||||
{:noreply, %{state | site_buffers: site_buffers, site_timers: site_timers}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:check_storm_cooldown, state) do
|
||||
storm_mode =
|
||||
cond do
|
||||
not state.storm_mode ->
|
||||
false
|
||||
|
||||
state.storm_mode_until &&
|
||||
DateTime.compare(DateTime.utc_now(), state.storm_mode_until) == :gt ->
|
||||
Logger.info("Alert storm mode ended")
|
||||
false
|
||||
|
||||
true ->
|
||||
true
|
||||
end
|
||||
|
||||
{:noreply, %{state | storm_mode: storm_mode}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:storm_mode?, _from, state) do
|
||||
# Also check if storm cooldown has expired
|
||||
storm_mode =
|
||||
if state.storm_mode and state.storm_mode_until do
|
||||
DateTime.compare(DateTime.utc_now(), state.storm_mode_until) != :gt
|
||||
else
|
||||
state.storm_mode
|
||||
end
|
||||
|
||||
{:reply, storm_mode, %{state | storm_mode: storm_mode}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:stats, _from, state) do
|
||||
stats = %{
|
||||
buffered_sites: map_size(state.site_buffers),
|
||||
buffered_events:
|
||||
state.site_buffers
|
||||
|> Enum.map(fn {_k, v} -> length(v) end)
|
||||
|> Enum.sum(),
|
||||
alerts_last_minute: state.alert_count_last_minute,
|
||||
storm_mode: state.storm_mode,
|
||||
storm_mode_until: state.storm_mode_until
|
||||
}
|
||||
|
||||
{:reply, stats, state}
|
||||
end
|
||||
|
||||
# --- Private Helpers ---
|
||||
|
||||
defp create_site_outage_alert(site_id, events, storm_mode) do
|
||||
device_count = length(events)
|
||||
devices = Enum.map(events, & &1.device)
|
||||
device_names = devices |> Enum.map(& &1.name) |> Enum.sort()
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
# Check maintenance window for the site (single check instead of N)
|
||||
if Maintenance.site_in_maintenance?(site_id) do
|
||||
Logger.info(
|
||||
"Site #{site_id} outage (#{device_count} devices) suppressed — in maintenance window"
|
||||
)
|
||||
else
|
||||
# Get site name for the alert message
|
||||
site_name =
|
||||
case Towerops.Sites.get_site(site_id) do
|
||||
nil -> "Unknown site"
|
||||
site -> site.name
|
||||
end
|
||||
|
||||
# Build impact summary
|
||||
message =
|
||||
"Site outage: #{site_name} — #{device_count} devices down (#{Enum.join(Enum.take(device_names, 5), ", ")}#{if device_count > 5, do: " and #{device_count - 5} more", else: ""})"
|
||||
|
||||
# Use the first device as the "primary" for the alert record
|
||||
# but store all affected device IDs in metadata
|
||||
primary_device = List.first(devices)
|
||||
|
||||
case Alerts.create_alert(%{
|
||||
device_id: primary_device.id,
|
||||
alert_type: "site_outage",
|
||||
triggered_at: now,
|
||||
message: message,
|
||||
storm_suppressed: false
|
||||
}) do
|
||||
{:ok, alert} ->
|
||||
# Single notification instead of N
|
||||
unless storm_mode do
|
||||
AlertNotificationWorker.enqueue_trigger(alert.id)
|
||||
end
|
||||
|
||||
_ =
|
||||
Phoenix.PubSub.broadcast(
|
||||
Towerops.PubSub,
|
||||
"alerts:org:#{primary_device.organization_id}:new",
|
||||
{:new_alert, primary_device.id, :site_outage}
|
||||
)
|
||||
|
||||
Logger.warning(
|
||||
"Created site outage alert for #{site_name}: #{device_count} devices",
|
||||
site_id: site_id,
|
||||
device_count: device_count,
|
||||
storm_mode: storm_mode
|
||||
)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("Failed to create site outage alert: #{inspect(reason)}")
|
||||
end
|
||||
|
||||
# Also mark each individual device as down (status update only, no individual alerts)
|
||||
Enum.each(devices, fn device ->
|
||||
Devices.update_device_status(device, :down)
|
||||
|
||||
# Suppress individual device_down alerts — the site_outage covers them
|
||||
# But DO create suppressed alert records for audit trail
|
||||
Alerts.create_alert(%{
|
||||
device_id: device.id,
|
||||
alert_type: "device_down",
|
||||
triggered_at: now,
|
||||
resolved_at: now,
|
||||
message: "#{device.name} is down (suppressed — part of site outage at #{site_name})",
|
||||
storm_suppressed: true
|
||||
})
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
defp create_individual_alerts(events, storm_mode) do
|
||||
Enum.each(events, fn %{device: device, timestamp: timestamp} ->
|
||||
# Skip if already has an active alert
|
||||
if Alerts.has_active_alert?(device.id, "device_down") do
|
||||
:ok
|
||||
else
|
||||
if Maintenance.device_in_maintenance?(device.id) do
|
||||
Logger.info(
|
||||
"Device #{device.id} (#{device.name}) is down but in maintenance — suppressing"
|
||||
)
|
||||
else
|
||||
create_single_device_down_alert(device, timestamp, storm_mode)
|
||||
end
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp create_single_device_down_alert(device, now, storm_mode) do
|
||||
alert_message =
|
||||
if device.snmp_enabled do
|
||||
"Device is not responding to SNMP"
|
||||
else
|
||||
"Device is not responding to ping"
|
||||
end
|
||||
|
||||
# Enrich with subscriber impact if available
|
||||
alert_message =
|
||||
case Towerops.Gaiia.get_device_impact(device.id) do
|
||||
%{subscriber_count: count, mrr: mrr} when count > 0 ->
|
||||
"#{device.name} is down — #{count} subscribers affected ($#{Decimal.round(mrr, 2)}/mo at risk)"
|
||||
|
||||
_ ->
|
||||
alert_message
|
||||
end
|
||||
|
||||
case Alerts.create_alert(%{
|
||||
device_id: device.id,
|
||||
alert_type: "device_down",
|
||||
triggered_at: now,
|
||||
message: alert_message
|
||||
}) do
|
||||
{:ok, alert} ->
|
||||
unless storm_mode do
|
||||
AlertNotificationWorker.enqueue_trigger(alert.id)
|
||||
end
|
||||
|
||||
_ =
|
||||
Phoenix.PubSub.broadcast(
|
||||
Towerops.PubSub,
|
||||
"alerts:org:#{device.organization_id}:new",
|
||||
{:new_alert, device.id, :device_down}
|
||||
)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("Failed to create device down alert for #{device.name}: #{inspect(reason)}")
|
||||
end
|
||||
end
|
||||
|
||||
defp prune_old_timestamps(queue, cutoff_ms) do
|
||||
case :queue.peek(queue) do
|
||||
{:value, ts} when ts < cutoff_ms ->
|
||||
{_, queue} = :queue.out(queue)
|
||||
prune_old_timestamps(queue, cutoff_ms)
|
||||
|
||||
_ ->
|
||||
queue
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -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)
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
95
lib/towerops/workers/alert_digest_worker.ex
Normal file
95
lib/towerops/workers/alert_digest_worker.ex
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
defmodule Towerops.Workers.AlertDigestWorker do
|
||||
@moduledoc """
|
||||
Oban worker that sends batched notification digests for rate-limited alerts.
|
||||
|
||||
Two modes:
|
||||
- Cron mode (no user_id or user_id="__cron__"): finds all users with pending digests
|
||||
and enqueues individual delivery jobs
|
||||
- Delivery mode (specific user_id): sends the actual digest to that user
|
||||
"""
|
||||
use Oban.Worker,
|
||||
queue: :notifications,
|
||||
max_attempts: 3
|
||||
|
||||
alias Towerops.Alerts
|
||||
alias Towerops.Alerts.NotificationRateLimiter
|
||||
|
||||
require Logger
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"user_id" => "__cron__"}}) do
|
||||
# Cron mode: find all pending digests and enqueue delivery
|
||||
import Ecto.Query
|
||||
|
||||
pending =
|
||||
Towerops.Repo.all(
|
||||
from(d in Towerops.Alerts.NotificationDigest,
|
||||
where: d.digest_sent == false,
|
||||
where: fragment("array_length(?, 1) > 0", d.suppressed_alert_ids),
|
||||
select: d.user_id,
|
||||
distinct: true
|
||||
)
|
||||
)
|
||||
|
||||
Enum.each(pending, fn user_id ->
|
||||
%{user_id: user_id}
|
||||
|> new()
|
||||
|> Oban.insert()
|
||||
end)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
def perform(%Oban.Job{args: %{"user_id" => user_id}}) when user_id != "__cron__" do
|
||||
case NotificationRateLimiter.get_pending_digest(user_id) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
digest ->
|
||||
alert_ids = digest.suppressed_alert_ids || []
|
||||
|
||||
if length(alert_ids) > 0 do
|
||||
alerts =
|
||||
alert_ids
|
||||
|> Enum.map(&Alerts.get_alert/1)
|
||||
|> Enum.reject(&is_nil/1)
|
||||
|
||||
if length(alerts) > 0 do
|
||||
send_digest_notification(alerts)
|
||||
NotificationRateLimiter.mark_digest_sent(digest)
|
||||
end
|
||||
end
|
||||
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Enqueue digest delivery for a specific user.
|
||||
Schedules delivery 5 minutes from now to batch more alerts.
|
||||
"""
|
||||
def enqueue(user_id) do
|
||||
%{user_id: user_id}
|
||||
|> new(schedule_in: 300, unique: [period: 300, keys: [:user_id]])
|
||||
|> Oban.insert()
|
||||
end
|
||||
|
||||
defp send_digest_notification(alerts) do
|
||||
count = length(alerts)
|
||||
|
||||
device_names =
|
||||
alerts
|
||||
|> Enum.map(fn a ->
|
||||
case a.device do
|
||||
nil -> "unknown"
|
||||
device -> device.name || device.ip_address || "unknown"
|
||||
end
|
||||
end)
|
||||
|> Enum.uniq()
|
||||
|
||||
Logger.info("Sending notification digest: #{count} suppressed alerts for devices: #{Enum.join(device_names, ", ")}",
|
||||
alert_count: count,
|
||||
alert_ids: Enum.map(alerts, & &1.id)
|
||||
)
|
||||
end
|
||||
end
|
||||
|
|
@ -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 ->
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -150,7 +150,113 @@
|
|||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Network Map Container -->
|
||||
<!-- Filter Bar -->
|
||||
<div class="bg-white dark:bg-gray-800 rounded-lg border border-gray-200 dark:border-white/10 mb-4 p-3">
|
||||
<div class="flex items-center justify-between gap-4 flex-wrap">
|
||||
<div class="flex items-center gap-2">
|
||||
<button
|
||||
phx-click="set_filter"
|
||||
phx-value-filter="all"
|
||||
class={[
|
||||
"px-3 py-1.5 text-xs font-medium rounded-md transition-colors",
|
||||
if(@filter == "all",
|
||||
do: "bg-blue-100 text-blue-700 dark:bg-blue-400/20 dark:text-blue-400",
|
||||
else: "text-gray-600 hover:bg-gray-100 dark:text-gray-400 dark:hover:bg-gray-700"
|
||||
)
|
||||
]}
|
||||
>
|
||||
All Links
|
||||
</button>
|
||||
<button
|
||||
phx-click="set_filter"
|
||||
phx-value-filter="degraded"
|
||||
class={[
|
||||
"px-3 py-1.5 text-xs font-medium rounded-md transition-colors",
|
||||
if(@filter == "degraded",
|
||||
do: "bg-yellow-100 text-yellow-700 dark:bg-yellow-400/20 dark:text-yellow-400",
|
||||
else: "text-gray-600 hover:bg-gray-100 dark:text-gray-400 dark:hover:bg-gray-700"
|
||||
)
|
||||
]}
|
||||
>
|
||||
Degraded Only
|
||||
<%= if @topology.stats[:degraded_links] && @topology.stats.degraded_links > 0 do %>
|
||||
<span class="ml-1 inline-flex items-center justify-center w-5 h-5 text-xs font-bold text-yellow-800 bg-yellow-200 rounded-full dark:bg-yellow-900 dark:text-yellow-300">
|
||||
{@topology.stats.degraded_links}
|
||||
</span>
|
||||
<% end %>
|
||||
</button>
|
||||
<button
|
||||
phx-click="set_filter"
|
||||
phx-value-filter="alerts"
|
||||
class={[
|
||||
"px-3 py-1.5 text-xs font-medium rounded-md transition-colors",
|
||||
if(@filter == "alerts",
|
||||
do: "bg-red-100 text-red-700 dark:bg-red-400/20 dark:text-red-400",
|
||||
else: "text-gray-600 hover:bg-gray-100 dark:text-gray-400 dark:hover:bg-gray-700"
|
||||
)
|
||||
]}
|
||||
>
|
||||
Sites with Alerts
|
||||
<%= if @topology.stats[:sites_with_alerts] && @topology.stats.sites_with_alerts > 0 do %>
|
||||
<span class="ml-1 inline-flex items-center justify-center w-5 h-5 text-xs font-bold text-red-800 bg-red-200 rounded-full dark:bg-red-900 dark:text-red-300">
|
||||
{@topology.stats.sites_with_alerts}
|
||||
</span>
|
||||
<% end %>
|
||||
</button>
|
||||
</div>
|
||||
<div class="flex items-center gap-3">
|
||||
<div class="relative">
|
||||
<.icon
|
||||
name="hero-magnifying-glass"
|
||||
class="absolute left-2.5 top-1/2 -translate-y-1/2 h-4 w-4 text-gray-400"
|
||||
/>
|
||||
<input
|
||||
type="text"
|
||||
placeholder="Search devices..."
|
||||
phx-keyup="search"
|
||||
phx-key="Enter"
|
||||
phx-debounce="300"
|
||||
value={@search_query}
|
||||
class="pl-8 pr-3 py-1.5 text-xs border border-gray-300 dark:border-gray-600 rounded-md bg-white dark:bg-gray-700 text-gray-900 dark:text-white placeholder-gray-400 focus:ring-1 focus:ring-blue-500 focus:border-blue-500 w-48"
|
||||
/>
|
||||
</div>
|
||||
<%= if @topology[:has_geo] do %>
|
||||
<div class="flex items-center gap-1 border-l border-gray-200 dark:border-gray-600 pl-3">
|
||||
<button
|
||||
phx-click="toggle_layout"
|
||||
phx-value-mode="force"
|
||||
class={[
|
||||
"px-2 py-1 text-xs rounded transition-colors",
|
||||
if(@layout_mode == "force",
|
||||
do: "bg-gray-200 dark:bg-gray-600 text-gray-900 dark:text-white font-medium",
|
||||
else: "text-gray-500 hover:bg-gray-100 dark:hover:bg-gray-700"
|
||||
)
|
||||
]}
|
||||
title="Force-directed layout"
|
||||
>
|
||||
<.icon name="hero-share" class="h-4 w-4" />
|
||||
</button>
|
||||
<button
|
||||
phx-click="toggle_layout"
|
||||
phx-value-mode="geo"
|
||||
class={[
|
||||
"px-2 py-1 text-xs rounded transition-colors",
|
||||
if(@layout_mode == "geo",
|
||||
do: "bg-gray-200 dark:bg-gray-600 text-gray-900 dark:text-white font-medium",
|
||||
else: "text-gray-500 hover:bg-gray-100 dark:hover:bg-gray-700"
|
||||
)
|
||||
]}
|
||||
title="Geographic layout"
|
||||
>
|
||||
<.icon name="hero-map" class="h-4 w-4" />
|
||||
</button>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Network Map Container -->
|
||||
<div class="bg-white dark:bg-gray-800 rounded-lg border border-gray-200 dark:border-white/10 overflow-hidden">
|
||||
<div class="p-4 border-b border-gray-200 dark:border-white/10">
|
||||
<div class="flex items-center justify-between">
|
||||
|
|
@ -295,6 +401,56 @@
|
|||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
|
||||
<!-- Wireless Stats (for APs, CPEs, backhaul radios) -->
|
||||
<%= if @selected_node_detail[:is_wireless] do %>
|
||||
<div class="mt-4 pt-3 border-t border-gray-200 dark:border-white/10">
|
||||
<h4 class="text-xs font-medium text-gray-500 dark:text-gray-400 uppercase tracking-wider mb-2">
|
||||
RF Stats
|
||||
</h4>
|
||||
<div class="grid grid-cols-2 gap-2">
|
||||
<!-- Client count -->
|
||||
<div class="bg-gray-50 dark:bg-gray-700/50 rounded-md p-2">
|
||||
<div class="text-xs text-gray-500 dark:text-gray-400">Clients</div>
|
||||
<div class="text-lg font-semibold text-gray-900 dark:text-white flex items-center gap-1">
|
||||
<.icon name="hero-wifi" class="h-4 w-4 text-purple-500" />
|
||||
{@selected_node_detail[:client_count] || 0}
|
||||
</div>
|
||||
</div>
|
||||
<!-- Signal health -->
|
||||
<div class="bg-gray-50 dark:bg-gray-700/50 rounded-md p-2">
|
||||
<div class="text-xs text-gray-500 dark:text-gray-400">Signal</div>
|
||||
<div class={[
|
||||
"text-sm font-semibold capitalize",
|
||||
case @selected_node_detail[:signal_health] do
|
||||
"good" -> "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"}
|
||||
</div>
|
||||
</div>
|
||||
<!-- SNR -->
|
||||
<%= if @selected_node_detail[:snr] do %>
|
||||
<div class="bg-gray-50 dark:bg-gray-700/50 rounded-md p-2 col-span-2">
|
||||
<div class="text-xs text-gray-500 dark:text-gray-400">SNR</div>
|
||||
<div class={[
|
||||
"text-sm font-semibold",
|
||||
cond do
|
||||
@selected_node_detail.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
|
||||
</div>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
</div>
|
||||
<% end %>
|
||||
|
||||
<!-- Connections -->
|
||||
<%= if length(@selected_node_detail.connections) > 0 do %>
|
||||
|
|
@ -338,7 +494,7 @@
|
|||
|
||||
<!-- View Device Link (managed only) -->
|
||||
<%= if @selected_node_detail.type == :managed do %>
|
||||
<div class="mt-4">
|
||||
<div class="mt-4 space-y-2">
|
||||
<.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")}
|
||||
</.link>
|
||||
<%= 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")}
|
||||
</.link>
|
||||
<% end %>
|
||||
</div>
|
||||
<% end %>
|
||||
</div>
|
||||
|
|
@ -373,7 +538,7 @@
|
|||
style="border-bottom-color: #8B5CF6;"
|
||||
>
|
||||
</div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Wireless</span>
|
||||
<span class="text-gray-700 dark:text-gray-300">AP / Wireless</span>
|
||||
</div>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-3 h-3 rounded-full" style="background-color: #EF4444;"></div>
|
||||
|
|
@ -393,22 +558,36 @@
|
|||
<span class="text-gray-700 dark:text-gray-300">Discovered</span>
|
||||
</div>
|
||||
</div>
|
||||
<%!-- Link confidence --%>
|
||||
<%!-- RF Link health --%>
|
||||
<div class="flex flex-wrap items-center gap-x-5 gap-y-1 text-xs">
|
||||
<span class="font-medium text-gray-600 dark:text-gray-400">Links:</span>
|
||||
<span class="font-medium text-gray-600 dark:text-gray-400">RF Links:</span>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-6 h-0.5" style="background-color: #10B981;"></div>
|
||||
<span class="text-gray-700 dark:text-gray-300">High confidence</span>
|
||||
<div class="w-6 h-1 rounded" style="background-color: #10B981;"></div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Good signal</span>
|
||||
</div>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-6 h-0.5 border-t-2 border-dashed" style="border-color: #F59E0B;">
|
||||
</div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Medium confidence</span>
|
||||
<div class="w-6 h-1 rounded" style="background-color: #F59E0B;"></div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Degraded</span>
|
||||
</div>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-6 h-0.5 border-t-2 border-dotted" style="border-color: #9CA3AF;">
|
||||
<div class="w-6 h-1 rounded" style="background-color: #EF4444;"></div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Critical</span>
|
||||
</div>
|
||||
</div>
|
||||
<%!-- Link types --%>
|
||||
<div class="flex flex-wrap items-center gap-x-5 gap-y-1 text-xs">
|
||||
<span class="font-medium text-gray-600 dark:text-gray-400">Link types:</span>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-6 h-0.5" style="background-color: #6B7280;"></div>
|
||||
<span class="text-gray-700 dark:text-gray-300">LLDP/CDP</span>
|
||||
</div>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<div class="w-6 h-0.5 border-t-2 border-dashed" style="border-color: #6B7280;">
|
||||
</div>
|
||||
<span class="text-gray-700 dark:text-gray-300">Low confidence</span>
|
||||
<span class="text-gray-700 dark:text-gray-300">MAC/ARP inferred</span>
|
||||
</div>
|
||||
<div class="flex items-center space-x-1.5">
|
||||
<span class="text-gray-500 dark:text-gray-400">Line thickness = bandwidth capacity</span>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
defmodule Towerops.Repo.Migrations.AddAlertStormCorrelation do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
# Alert storms track when alert rate exceeds threshold for an org
|
||||
create table(:alert_storms, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :started_at, :utc_datetime, null: false
|
||||
add :ended_at, :utc_datetime
|
||||
add :alert_count, :integer, null: false, default: 0
|
||||
add :suppressed_count, :integer, null: false, default: 0
|
||||
add :summary_notification_sent, :boolean, null: false, default: false
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create index(:alert_storms, [:organization_id, :started_at])
|
||||
create index(:alert_storms, [:organization_id], where: "ended_at IS NULL", name: :alert_storms_active_idx)
|
||||
|
||||
# Site outages correlate multiple device-down alerts at the same site
|
||||
create table(:site_outages, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :site_id, references(:sites, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :started_at, :utc_datetime, null: false
|
||||
add :resolved_at, :utc_datetime
|
||||
add :device_count, :integer, null: false, default: 0
|
||||
add :total_devices_at_site, :integer, null: false, default: 0
|
||||
add :message, :string
|
||||
add :notification_sent, :boolean, null: false, default: false
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create index(:site_outages, [:site_id], where: "resolved_at IS NULL", name: :site_outages_active_idx)
|
||||
create index(:site_outages, [:organization_id, :started_at])
|
||||
|
||||
# Link alerts to their correlated site outage (optional)
|
||||
alter table(:alerts) do
|
||||
add :site_outage_id, references(:site_outages, type: :binary_id, on_delete: :nilify_all)
|
||||
add :storm_suppressed, :boolean, default: false
|
||||
end
|
||||
|
||||
create index(:alerts, [:site_outage_id], where: "site_outage_id IS NOT NULL")
|
||||
|
||||
# Notification rate limiting tracking per user
|
||||
create table(:notification_digests, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :user_id, references(:users, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :window_start, :utc_datetime, null: false
|
||||
add :notification_count, :integer, null: false, default: 0
|
||||
add :digest_sent, :boolean, null: false, default: false
|
||||
add :digest_sent_at, :utc_datetime
|
||||
add :suppressed_alert_ids, {:array, :binary_id}, default: []
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create index(:notification_digests, [:user_id, :window_start])
|
||||
create unique_index(:notification_digests, [:user_id, :window_start], name: :notification_digests_user_window_idx)
|
||||
end
|
||||
end
|
||||
86
test/towerops/alerts/notification_rate_limiter_test.exs
Normal file
86
test/towerops/alerts/notification_rate_limiter_test.exs
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
defmodule Towerops.Alerts.NotificationRateLimiterTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Alerts.NotificationRateLimiter
|
||||
|
||||
setup do
|
||||
org = Towerops.OrganizationsFixtures.organization_fixture()
|
||||
user = Towerops.AccountsFixtures.user_fixture(%{organization_id: org.id})
|
||||
|
||||
%{org: org, user: user}
|
||||
end
|
||||
|
||||
describe "check_rate/3" do
|
||||
test "allows notifications under the limit", %{org: org, user: user} do
|
||||
# Default limit is 5 per window
|
||||
for _ <- 1..5 do
|
||||
assert :allow = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
end
|
||||
|
||||
test "suppresses notifications over the limit", %{org: org, user: user} do
|
||||
# Fill up the limit
|
||||
for _ <- 1..5 do
|
||||
assert :allow = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
|
||||
# Next ones should be suppressed
|
||||
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
|
||||
test "different users have independent limits", %{org: org, user: user} do
|
||||
user2 = Towerops.AccountsFixtures.user_fixture(%{organization_id: org.id})
|
||||
|
||||
# Fill up user1's limit
|
||||
for _ <- 1..5 do
|
||||
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
|
||||
assert :suppress = NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
|
||||
# user2 should still be fine
|
||||
assert :allow = NotificationRateLimiter.check_rate(user2.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_pending_digest/1" do
|
||||
test "returns nil when no suppressed alerts", %{user: user} do
|
||||
assert is_nil(NotificationRateLimiter.get_pending_digest(user.id))
|
||||
end
|
||||
|
||||
test "returns digest with suppressed alerts", %{org: org, user: user} do
|
||||
# Fill up the limit
|
||||
for _ <- 1..5 do
|
||||
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
|
||||
# Suppress some alerts
|
||||
alert_id = Ecto.UUID.generate()
|
||||
NotificationRateLimiter.check_rate(user.id, org.id, alert_id)
|
||||
|
||||
digest = NotificationRateLimiter.get_pending_digest(user.id)
|
||||
assert digest != nil
|
||||
assert alert_id in digest.suppressed_alert_ids
|
||||
end
|
||||
end
|
||||
|
||||
describe "mark_digest_sent/1" do
|
||||
test "marks digest as sent", %{org: org, user: user} do
|
||||
# Fill up and suppress
|
||||
for _ <- 1..5 do
|
||||
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
end
|
||||
|
||||
NotificationRateLimiter.check_rate(user.id, org.id, Ecto.UUID.generate())
|
||||
|
||||
digest = NotificationRateLimiter.get_pending_digest(user.id)
|
||||
assert {:ok, updated} = NotificationRateLimiter.mark_digest_sent(digest)
|
||||
assert updated.digest_sent == true
|
||||
assert updated.digest_sent_at != nil
|
||||
|
||||
# Should no longer appear as pending
|
||||
assert is_nil(NotificationRateLimiter.get_pending_digest(user.id))
|
||||
end
|
||||
end
|
||||
end
|
||||
111
test/towerops/alerts/site_correlation_test.exs
Normal file
111
test/towerops/alerts/site_correlation_test.exs
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
defmodule Towerops.Alerts.SiteCorrelationTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Alerts
|
||||
alias Towerops.Alerts.SiteCorrelation
|
||||
|
||||
setup do
|
||||
org = Towerops.OrganizationsFixtures.organization_fixture()
|
||||
site = Towerops.DevicesFixtures.create_site(org)
|
||||
|
||||
# Create multiple devices at the same site
|
||||
devices =
|
||||
for i <- 1..5 do
|
||||
Towerops.DevicesFixtures.device_fixture(%{
|
||||
organization_id: org.id,
|
||||
site_id: site.id,
|
||||
name: "Device #{i}",
|
||||
ip_address: "10.0.0.#{i}"
|
||||
})
|
||||
end
|
||||
|
||||
%{org: org, site: site, devices: devices}
|
||||
end
|
||||
|
||||
describe "check_site_correlation/2" do
|
||||
test "returns :no_correlation for device without site" do
|
||||
device = %Towerops.Devices.Device{id: Ecto.UUID.generate(), site_id: nil}
|
||||
assert :no_correlation = SiteCorrelation.check_site_correlation(device, DateTime.utc_now())
|
||||
end
|
||||
|
||||
test "returns :no_correlation when below threshold", %{devices: [d1 | _]} do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
# Only 1 device down — below default threshold of 3
|
||||
assert :no_correlation = SiteCorrelation.check_site_correlation(d1, now)
|
||||
end
|
||||
|
||||
test "returns {:site_outage, _} when threshold met", %{devices: devices, site: _site} do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
# Create alerts for first 2 devices (we configured threshold at 3 default)
|
||||
[d1, d2, d3 | _] = devices
|
||||
|
||||
for d <- [d1, d2] do
|
||||
{:ok, _alert} =
|
||||
Alerts.create_alert(%{
|
||||
device_id: d.id,
|
||||
alert_type: "device_down",
|
||||
triggered_at: now,
|
||||
message: "Device down"
|
||||
})
|
||||
end
|
||||
|
||||
# 3rd device should trigger site correlation
|
||||
assert {:site_outage, outage} = SiteCorrelation.check_site_correlation(d3, now)
|
||||
assert outage.site_id == d3.site_id
|
||||
assert outage.device_count >= 3
|
||||
end
|
||||
end
|
||||
|
||||
describe "check_outage_resolution/1" do
|
||||
test "resolves outage when all devices are back up", %{devices: devices} do
|
||||
now = DateTime.utc_now()
|
||||
[d1, d2, d3 | _] = devices
|
||||
|
||||
# Create alerts
|
||||
for d <- [d1, d2] do
|
||||
{:ok, _} =
|
||||
Alerts.create_alert(%{
|
||||
device_id: d.id,
|
||||
alert_type: "device_down",
|
||||
triggered_at: now,
|
||||
message: "Device down"
|
||||
})
|
||||
end
|
||||
|
||||
# Trigger site outage
|
||||
{:site_outage, outage} = SiteCorrelation.check_site_correlation(d3, now)
|
||||
|
||||
# Create alert for d3 too
|
||||
{:ok, _} =
|
||||
Alerts.create_alert(%{
|
||||
device_id: d3.id,
|
||||
alert_type: "device_down",
|
||||
triggered_at: now,
|
||||
message: "Device down",
|
||||
site_outage_id: outage.id
|
||||
})
|
||||
|
||||
# Resolve all alerts
|
||||
for d <- [d1, d2, d3] do
|
||||
case Alerts.get_active_alert(d.id, "device_down") do
|
||||
nil -> :ok
|
||||
alert -> Alerts.resolve_alert(alert)
|
||||
end
|
||||
end
|
||||
|
||||
# Check resolution
|
||||
SiteCorrelation.check_outage_resolution(d1)
|
||||
|
||||
# Outage should be resolved
|
||||
assert is_nil(SiteCorrelation.get_active_outage(d1.site_id))
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_active_outage/1" do
|
||||
test "returns nil when no outage", %{site: site} do
|
||||
assert is_nil(SiteCorrelation.get_active_outage(site.id))
|
||||
end
|
||||
end
|
||||
end
|
||||
71
test/towerops/alerts/storm_detector_test.exs
Normal file
71
test/towerops/alerts/storm_detector_test.exs
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
defmodule Towerops.Alerts.StormDetectorTest do
|
||||
use Towerops.DataCase, async: false
|
||||
|
||||
alias Towerops.Alerts.StormDetector
|
||||
|
||||
setup do
|
||||
# Stop the globally started detector (if running) and start a fresh one
|
||||
if Process.whereis(StormDetector), do: GenServer.stop(StormDetector)
|
||||
{:ok, pid} = StormDetector.start_link(threshold: 3, window_seconds: 10)
|
||||
on_exit(fn ->
|
||||
if Process.alive?(pid), do: GenServer.stop(pid)
|
||||
end)
|
||||
|
||||
org = Towerops.OrganizationsFixtures.organization_fixture()
|
||||
%{org: org}
|
||||
end
|
||||
|
||||
describe "record_alert/2" do
|
||||
test "returns :ok when under threshold", %{org: org} do
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
end
|
||||
|
||||
test "returns :suppress when threshold exceeded", %{org: org} do
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
end
|
||||
|
||||
test "different orgs tracked independently", %{org: org} do
|
||||
org2 = Towerops.OrganizationsFixtures.organization_fixture()
|
||||
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :ok = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
assert :suppress = StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
|
||||
# org2 should still be fine
|
||||
assert :ok = StormDetector.record_alert(org2.id, %{device_id: Ecto.UUID.generate()})
|
||||
end
|
||||
end
|
||||
|
||||
describe "in_storm?/1" do
|
||||
test "returns false when no alerts", %{org: org} do
|
||||
refute StormDetector.in_storm?(org.id)
|
||||
end
|
||||
|
||||
test "returns true when threshold exceeded", %{org: org} do
|
||||
for _ <- 1..3 do
|
||||
StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
end
|
||||
|
||||
assert StormDetector.in_storm?(org.id)
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_storm/1" do
|
||||
test "returns nil when no active storm", %{org: org} do
|
||||
assert is_nil(StormDetector.get_storm(org.id))
|
||||
end
|
||||
|
||||
test "returns storm info when in storm", %{org: org} do
|
||||
for _ <- 1..3 do
|
||||
StormDetector.record_alert(org.id, %{device_id: Ecto.UUID.generate()})
|
||||
end
|
||||
|
||||
storm = StormDetector.get_storm(org.id)
|
||||
assert storm != nil
|
||||
end
|
||||
end
|
||||
end
|
||||
91
test/towerops/maintenance_batch_test.exs
Normal file
91
test/towerops/maintenance_batch_test.exs
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
defmodule Towerops.MaintenanceBatchTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Maintenance
|
||||
|
||||
setup do
|
||||
org = Towerops.OrganizationsFixtures.organization_fixture()
|
||||
site = Towerops.DevicesFixtures.create_site(org)
|
||||
|
||||
devices =
|
||||
for i <- 1..5 do
|
||||
Towerops.DevicesFixtures.device_fixture(%{
|
||||
organization_id: org.id,
|
||||
site_id: site.id,
|
||||
name: "BatchDevice #{i}",
|
||||
ip_address: "10.1.0.#{i}"
|
||||
})
|
||||
end
|
||||
|
||||
%{org: org, site: site, devices: devices}
|
||||
end
|
||||
|
||||
describe "devices_in_maintenance/1" do
|
||||
test "returns empty MapSet when no maintenance windows", %{devices: devices} do
|
||||
device_ids = Enum.map(devices, & &1.id)
|
||||
result = Maintenance.devices_in_maintenance(device_ids)
|
||||
assert MapSet.size(result) == 0
|
||||
end
|
||||
|
||||
test "returns empty MapSet for empty input" do
|
||||
assert MapSet.size(Maintenance.devices_in_maintenance([])) == 0
|
||||
end
|
||||
|
||||
test "detects device-level maintenance window", %{org: org, devices: [d1 | _]} do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
{:ok, _window} =
|
||||
Maintenance.create_window(%{
|
||||
name: "Device MW",
|
||||
organization_id: org.id,
|
||||
device_id: d1.id,
|
||||
starts_at: DateTime.add(now, -3600, :second),
|
||||
ends_at: DateTime.add(now, 3600, :second),
|
||||
suppress_alerts: true
|
||||
})
|
||||
|
||||
device_ids = [d1.id]
|
||||
result = Maintenance.devices_in_maintenance(device_ids)
|
||||
assert MapSet.member?(result, d1.id)
|
||||
end
|
||||
|
||||
test "detects site-level maintenance window", %{org: org, site: site, devices: devices} do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
{:ok, _window} =
|
||||
Maintenance.create_window(%{
|
||||
name: "Site MW",
|
||||
organization_id: org.id,
|
||||
site_id: site.id,
|
||||
starts_at: DateTime.add(now, -3600, :second),
|
||||
ends_at: DateTime.add(now, 3600, :second),
|
||||
suppress_alerts: true
|
||||
})
|
||||
|
||||
device_ids = Enum.map(devices, & &1.id)
|
||||
result = Maintenance.devices_in_maintenance(device_ids)
|
||||
|
||||
# All devices at the site should be in maintenance
|
||||
assert MapSet.size(result) == 5
|
||||
end
|
||||
|
||||
test "detects org-wide maintenance window", %{org: org, devices: devices} do
|
||||
now = DateTime.utc_now()
|
||||
|
||||
{:ok, _window} =
|
||||
Maintenance.create_window(%{
|
||||
name: "Org MW",
|
||||
organization_id: org.id,
|
||||
starts_at: DateTime.add(now, -3600, :second),
|
||||
ends_at: DateTime.add(now, 3600, :second),
|
||||
suppress_alerts: true
|
||||
})
|
||||
|
||||
device_ids = Enum.map(devices, & &1.id)
|
||||
result = Maintenance.devices_in_maintenance(device_ids)
|
||||
|
||||
# All devices should be in maintenance
|
||||
assert MapSet.size(result) == 5
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue