145 lines
5 KiB
Elixir
145 lines
5 KiB
Elixir
defmodule Towerops.Workers.CapacityInsightWorker do
|
|
@moduledoc """
|
|
Evaluates backhaul capacity utilization and generates insights when
|
|
interfaces approach or exceed their configured capacity.
|
|
|
|
Runs every 15 minutes. Generates:
|
|
- `backhaul_over_capacity` (critical) when utilization >= 90%
|
|
- `backhaul_near_capacity` (warning) when utilization >= 75%
|
|
|
|
Auto-resolves insights when utilization drops below 70%.
|
|
"""
|
|
|
|
use Oban.Pro.Worker, queue: :maintenance
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Capacity
|
|
alias Towerops.Devices.Device
|
|
alias Towerops.Organizations.Organization
|
|
alias Towerops.Preseem.Insight
|
|
alias Towerops.Preseem.Insights
|
|
alias Towerops.Repo
|
|
alias Towerops.Snmp.Interface
|
|
|
|
@critical_threshold 90
|
|
@warning_threshold 75
|
|
@resolve_threshold 70
|
|
|
|
@impl Oban.Pro.Worker
|
|
def process(%Oban.Job{}) do
|
|
# Use streaming to avoid loading all org IDs into memory
|
|
Repo.transaction(fn ->
|
|
Organization
|
|
|> select([o], o.id)
|
|
|> Repo.stream()
|
|
|> Enum.each(&evaluate_organization/1)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
defp evaluate_organization(organization_id) do
|
|
interfaces = list_capacity_interfaces(organization_id)
|
|
|
|
evaluated =
|
|
Enum.map(interfaces, fn {interface, device} ->
|
|
utilization = Capacity.get_utilization(interface)
|
|
{interface, device, utilization}
|
|
end)
|
|
|
|
# Generate insights for high utilization
|
|
Enum.each(evaluated, fn {interface, device, utilization} ->
|
|
generate_capacity_insight(organization_id, interface, device, utilization)
|
|
end)
|
|
|
|
# Auto-resolve insights for recovered interfaces
|
|
evaluated_ids = MapSet.new(evaluated, fn {iface, _, _} -> iface.id end)
|
|
auto_resolve_recovered(organization_id, evaluated, evaluated_ids)
|
|
end
|
|
|
|
defp generate_capacity_insight(_org_id, _interface, _device, nil), do: :ok
|
|
|
|
defp generate_capacity_insight(organization_id, interface, device, utilization) do
|
|
pct = utilization.utilization_pct
|
|
|
|
cond do
|
|
pct >= @critical_threshold ->
|
|
insert_insight(organization_id, interface, device, utilization, "backhaul_over_capacity", "critical")
|
|
|
|
pct >= @warning_threshold ->
|
|
insert_insight(organization_id, interface, device, utilization, "backhaul_near_capacity", "warning")
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp insert_insight(organization_id, interface, device, utilization, type, urgency) do
|
|
Insights.insert_insight_if_new(%{
|
|
organization_id: organization_id,
|
|
device_id: device.id,
|
|
type: type,
|
|
urgency: urgency,
|
|
channel: "proactive",
|
|
source: "snmp",
|
|
title: "#{interface.if_name} on #{device.name} #{humanize_type(type)}",
|
|
description:
|
|
"Interface #{interface.if_name} is at #{Float.round(utilization.utilization_pct, 1)}% of its #{format_capacity(interface.configured_capacity_bps)} capacity.",
|
|
dedup_key: interface.id,
|
|
metadata: %{
|
|
"dedup_key" => interface.id,
|
|
"interface_id" => interface.id,
|
|
"interface_name" => interface.if_name,
|
|
"capacity_bps" => interface.configured_capacity_bps,
|
|
"throughput_bps" => utilization.throughput.max_bps,
|
|
"utilization_pct" => Float.round(utilization.utilization_pct, 1),
|
|
"device_name" => device.name
|
|
}
|
|
})
|
|
end
|
|
|
|
defp auto_resolve_recovered(organization_id, evaluated, _evaluated_ids) do
|
|
active_insights =
|
|
Insight
|
|
|> where([i], i.organization_id == ^organization_id)
|
|
|> where([i], i.type in ["backhaul_over_capacity", "backhaul_near_capacity"])
|
|
|> where([i], i.status == "active")
|
|
|> Repo.all()
|
|
|
|
Enum.each(active_insights, fn insight ->
|
|
interface_id = insight.metadata["interface_id"]
|
|
match = Enum.find(evaluated, fn {iface, _, _} -> iface.id == interface_id end)
|
|
maybe_resolve_insight(insight, match)
|
|
end)
|
|
end
|
|
|
|
defp maybe_resolve_insight(_insight, nil), do: :ok
|
|
defp maybe_resolve_insight(_insight, {_iface, _device, nil}), do: :ok
|
|
|
|
defp maybe_resolve_insight(insight, {_iface, _device, utilization}) do
|
|
if utilization.utilization_pct < @resolve_threshold do
|
|
insight
|
|
|> Insight.changeset(%{status: "resolved"})
|
|
|> Repo.update()
|
|
end
|
|
end
|
|
|
|
defp list_capacity_interfaces(organization_id) do
|
|
Interface
|
|
|> join(:inner, [i], sd in Towerops.Snmp.Device, on: i.snmp_device_id == sd.id)
|
|
|> join(:inner, [_i, sd], d in Device, on: sd.device_id == d.id)
|
|
|> where([_i, _sd, d], d.organization_id == ^organization_id)
|
|
|> where([i], not is_nil(i.configured_capacity_bps))
|
|
|> select([i, _sd, d], {i, d})
|
|
|> Repo.all()
|
|
end
|
|
|
|
defp humanize_type("backhaul_over_capacity"), do: "over capacity"
|
|
defp humanize_type("backhaul_near_capacity"), do: "approaching capacity"
|
|
|
|
defp format_capacity(bps) when bps >= 1_000_000_000, do: "#{Float.round(bps / 1_000_000_000, 1)} Gbps"
|
|
defp format_capacity(bps) when bps >= 1_000_000, do: "#{Float.round(bps / 1_000_000, 1)} Mbps"
|
|
defp format_capacity(bps) when bps >= 1_000, do: "#{Float.round(bps / 1_000, 1)} Kbps"
|
|
defp format_capacity(bps), do: "#{bps} bps"
|
|
end
|