Merge pull request 'feature/subscriber-impact' (#6) from feature/subscriber-impact into main
Reviewed-on: graham/towerops-web#6
This commit is contained in:
commit
44d9bb4c49
11 changed files with 1083 additions and 13 deletions
|
|
@ -8,6 +8,7 @@ defmodule Towerops.Gaiia do
|
|||
alias Towerops.Devices.Device
|
||||
alias Towerops.Gaiia.Account
|
||||
alias Towerops.Gaiia.BillingSubscription
|
||||
alias Towerops.Gaiia.DeviceSubscriberLink
|
||||
alias Towerops.Gaiia.InventoryItem
|
||||
alias Towerops.Gaiia.NetworkSite
|
||||
alias Towerops.Repo
|
||||
|
|
@ -241,13 +242,12 @@ defmodule Towerops.Gaiia do
|
|||
if device_ids == [] do
|
||||
%{}
|
||||
else
|
||||
# Count inventory items with assigned accounts per device as subscriber proxy
|
||||
InventoryItem
|
||||
|> where([ii], ii.device_id in ^device_ids and not is_nil(ii.assigned_account_gaiia_id))
|
||||
|> group_by([ii], ii.device_id)
|
||||
|> select([ii], %{
|
||||
device_id: ii.device_id,
|
||||
subscriber_count: count(ii.id, :distinct)
|
||||
DeviceSubscriberLink
|
||||
|> where([l], l.device_id in ^device_ids)
|
||||
|> group_by([l], l.device_id)
|
||||
|> select([l], %{
|
||||
device_id: l.device_id,
|
||||
subscriber_count: count(l.gaiia_account_id, :distinct)
|
||||
})
|
||||
|> Repo.all()
|
||||
|> Map.new(fn row ->
|
||||
|
|
@ -264,4 +264,117 @@ defmodule Towerops.Gaiia do
|
|||
|> select([ns], %{account_count: ns.account_count, total_mrr: ns.total_mrr})
|
||||
|> Repo.one()
|
||||
end
|
||||
|
||||
# --- Device Impact Queries ---
|
||||
|
||||
@doc """
|
||||
Get subscriber impact for a single device.
|
||||
Returns `%{subscriber_count: integer, mrr: Decimal.t(), accounts: [%Account{}]}`.
|
||||
"""
|
||||
def get_device_impact(device_id) do
|
||||
account_ids =
|
||||
DeviceSubscriberLink
|
||||
|> where(device_id: ^device_id)
|
||||
|> select([l], l.gaiia_account_id)
|
||||
|> Repo.all()
|
||||
|> Enum.uniq()
|
||||
|
||||
if account_ids == [] do
|
||||
%{subscriber_count: 0, mrr: Decimal.new(0), accounts: []}
|
||||
else
|
||||
accounts =
|
||||
Account
|
||||
|> where([a], a.id in ^account_ids)
|
||||
|> Repo.all()
|
||||
|
||||
mrr =
|
||||
BillingSubscription
|
||||
|> join(:inner, [bs], a in Account, on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id)
|
||||
|> where([bs, a], a.id in ^account_ids)
|
||||
|> where([bs], bs.status == "ACTIVE")
|
||||
|> Repo.aggregate(:sum, :mrr_amount) || Decimal.new(0)
|
||||
|
||||
%{subscriber_count: length(accounts), mrr: mrr, accounts: accounts}
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Get subscriber impact for multiple devices (batch).
|
||||
Returns `%{device_id => %{subscriber_count: integer, mrr: Decimal.t()}}`.
|
||||
"""
|
||||
def get_devices_impact(device_ids) when is_list(device_ids) do
|
||||
if device_ids == [] do
|
||||
%{}
|
||||
else
|
||||
# Get account links per device
|
||||
links =
|
||||
DeviceSubscriberLink
|
||||
|> where([l], l.device_id in ^device_ids)
|
||||
|> join(:inner, [l], a in Account, on: l.gaiia_account_id == a.id)
|
||||
|> join(:left, [l, a], bs in BillingSubscription,
|
||||
on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id and bs.status == "ACTIVE"
|
||||
)
|
||||
|> group_by([l, a, bs], l.device_id)
|
||||
|> select([l, a, bs], %{
|
||||
device_id: l.device_id,
|
||||
subscriber_count: count(a.id, :distinct),
|
||||
mrr: coalesce(sum(bs.mrr_amount), 0)
|
||||
})
|
||||
|> Repo.all()
|
||||
|
||||
Map.new(links, fn row ->
|
||||
{row.device_id, %{subscriber_count: row.subscriber_count, mrr: row.mrr}}
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Get subscriber impact for a site (all devices at site).
|
||||
Returns `%{subscriber_count: integer, mrr: Decimal.t(), devices: [%{device_id, name, count, mrr}]}`.
|
||||
"""
|
||||
def get_site_impact(site_id) do
|
||||
alias Towerops.Devices.Device
|
||||
|
||||
device_ids =
|
||||
Device
|
||||
|> where(site_id: ^site_id)
|
||||
|> select([d], d.id)
|
||||
|> Repo.all()
|
||||
|
||||
impact = get_devices_impact(device_ids)
|
||||
|
||||
devices_with_names =
|
||||
Device
|
||||
|> where([d], d.id in ^device_ids)
|
||||
|> select([d], {d.id, d.name})
|
||||
|> Repo.all()
|
||||
|> Enum.map(fn {id, name} ->
|
||||
di = Map.get(impact, id, %{subscriber_count: 0, mrr: Decimal.new(0)})
|
||||
%{device_id: id, name: name, count: di.subscriber_count, mrr: di.mrr}
|
||||
end)
|
||||
|
||||
total_account_ids =
|
||||
DeviceSubscriberLink
|
||||
|> where([l], l.device_id in ^device_ids)
|
||||
|> select([l], l.gaiia_account_id)
|
||||
|> Repo.all()
|
||||
|> Enum.uniq()
|
||||
|
||||
total_mrr =
|
||||
if total_account_ids == [] do
|
||||
Decimal.new(0)
|
||||
else
|
||||
BillingSubscription
|
||||
|> join(:inner, [bs], a in Account, on: bs.account_gaiia_id == a.gaiia_id and bs.organization_id == a.organization_id)
|
||||
|> where([bs, a], a.id in ^total_account_ids)
|
||||
|> where([bs], bs.status == "ACTIVE")
|
||||
|> Repo.aggregate(:sum, :mrr_amount) || Decimal.new(0)
|
||||
end
|
||||
|
||||
%{
|
||||
subscriber_count: length(total_account_ids),
|
||||
mrr: total_mrr,
|
||||
devices: devices_with_names
|
||||
}
|
||||
end
|
||||
end
|
||||
|
|
|
|||
50
lib/towerops/gaiia/device_subscriber_link.ex
Normal file
50
lib/towerops/gaiia/device_subscriber_link.ex
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
defmodule Towerops.Gaiia.DeviceSubscriberLink do
|
||||
@moduledoc """
|
||||
Schema for materialized subscriber-to-device (AP) mappings.
|
||||
|
||||
Links a Gaiia subscriber account to a TowerOps device based on
|
||||
ARP table matching or site-level fallback.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
schema "device_subscriber_links" do
|
||||
field :match_method, :string
|
||||
field :confidence, :string
|
||||
field :subscriber_ip, :string
|
||||
field :subscriber_mac, :string
|
||||
|
||||
belongs_to :organization, Towerops.Organizations.Organization
|
||||
belongs_to :device, Towerops.Devices.Device
|
||||
belongs_to :gaiia_account, Towerops.Gaiia.Account
|
||||
belongs_to :gaiia_inventory_item, Towerops.Gaiia.InventoryItem
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
def changeset(link, attrs) do
|
||||
link
|
||||
|> cast(attrs, [
|
||||
:organization_id,
|
||||
:device_id,
|
||||
:gaiia_account_id,
|
||||
:gaiia_inventory_item_id,
|
||||
:match_method,
|
||||
:confidence,
|
||||
:subscriber_ip,
|
||||
:subscriber_mac
|
||||
])
|
||||
|> validate_required([:organization_id, :device_id, :gaiia_account_id, :match_method, :confidence])
|
||||
|> validate_inclusion(:match_method, ~w(arp_ip arp_mac subnet site_fallback))
|
||||
|> validate_inclusion(:confidence, ~w(high medium low))
|
||||
|> unique_constraint([:device_id, :gaiia_account_id])
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
|> foreign_key_constraint(:device_id)
|
||||
|> foreign_key_constraint(:gaiia_account_id)
|
||||
|> foreign_key_constraint(:gaiia_inventory_item_id)
|
||||
end
|
||||
end
|
||||
390
lib/towerops/gaiia/subscriber_matching.ex
Normal file
390
lib/towerops/gaiia/subscriber_matching.ex
Normal file
|
|
@ -0,0 +1,390 @@
|
|||
defmodule Towerops.Gaiia.SubscriberMatching do
|
||||
@moduledoc """
|
||||
Matches Gaiia subscriber accounts to TowerOps devices (APs) using
|
||||
ARP table entries and site-level fallback.
|
||||
|
||||
Builds a materialized `device_subscriber_links` table that maps
|
||||
each subscriber to the AP(s) serving them.
|
||||
"""
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Towerops.Devices.Device
|
||||
alias Towerops.Gaiia.Account
|
||||
alias Towerops.Gaiia.DeviceSubscriberLink
|
||||
alias Towerops.Gaiia.InventoryItem
|
||||
alias Towerops.Gaiia.NetworkSite
|
||||
alias Towerops.Integrations
|
||||
alias Towerops.Repo
|
||||
alias Towerops.Snmp.ArpEntry
|
||||
alias Towerops.Snmp.WirelessClient
|
||||
|
||||
require Logger
|
||||
|
||||
@doc """
|
||||
Rebuilds all device-subscriber links for an organization.
|
||||
Deletes existing links and re-computes from ARP + inventory data.
|
||||
"""
|
||||
def refresh_device_links(organization_id) do
|
||||
unless has_gaiia_integration?(organization_id) do
|
||||
Logger.debug("Skipping subscriber matching for org #{organization_id} — no Gaiia integration")
|
||||
:ok
|
||||
else
|
||||
do_refresh(organization_id)
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Lightweight refresh for a single device after ARP poll.
|
||||
Rebuilds links only for the given device.
|
||||
"""
|
||||
def refresh_links_for_device(device_id) do
|
||||
device = Repo.get(Device, device_id)
|
||||
|
||||
if device do
|
||||
unless has_gaiia_integration?(device.organization_id) do
|
||||
:ok
|
||||
else
|
||||
do_refresh_for_device(device)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
# --- Private ---
|
||||
|
||||
defp has_gaiia_integration?(organization_id) do
|
||||
case Integrations.get_integration(organization_id, "gaiia") do
|
||||
{:ok, _integration} -> true
|
||||
{:error, :not_found} -> false
|
||||
end
|
||||
end
|
||||
|
||||
defp do_refresh(organization_id) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
# Load all inventory items with assigned accounts
|
||||
items = load_assigned_items(organization_id)
|
||||
# Load all ARP entries for this org's devices
|
||||
arp_entries = load_arp_entries(organization_id)
|
||||
# Load all wireless client entries for this org's devices
|
||||
wireless_clients = load_wireless_clients(organization_id)
|
||||
|
||||
# Build lookup maps
|
||||
arp_by_ip = Map.new(arp_entries, fn e -> {e.ip_address, e} end)
|
||||
arp_by_mac = Map.new(arp_entries, fn e -> {normalize_mac(e.mac_address), e} end)
|
||||
wc_by_mac = Map.new(wireless_clients, fn w -> {normalize_mac(w.mac_address), w} end)
|
||||
wc_by_ip = wireless_clients |> Enum.filter(& &1.ip_address) |> Map.new(fn w -> {w.ip_address, w} end)
|
||||
|
||||
# Phase 1: Wireless client MAC match (highest confidence)
|
||||
{wc_mac_matched, wc_mac_links} = match_by_wireless_mac(items, wc_by_mac, organization_id, now)
|
||||
|
||||
# Phase 2: Wireless client IP match (high confidence)
|
||||
remaining_after_wc_mac = Enum.reject(items, &MapSet.member?(wc_mac_matched, &1.id))
|
||||
{wc_ip_matched, wc_ip_links} = match_by_wireless_ip(remaining_after_wc_mac, wc_by_ip, organization_id, now)
|
||||
|
||||
# Phase 3: ARP IP match (medium confidence)
|
||||
matched_so_far = MapSet.union(wc_mac_matched, wc_ip_matched)
|
||||
remaining_after_wc = Enum.reject(items, &MapSet.member?(matched_so_far, &1.id))
|
||||
{ip_matched, ip_links} = match_by_arp_ip(remaining_after_wc, arp_by_ip, organization_id, now)
|
||||
|
||||
# Phase 4: ARP MAC match (medium confidence)
|
||||
matched_so_far = MapSet.union(matched_so_far, ip_matched)
|
||||
remaining_after_ip = Enum.reject(items, &MapSet.member?(matched_so_far, &1.id))
|
||||
{mac_matched, mac_links} = match_by_arp_mac(remaining_after_ip, arp_by_mac, organization_id, now)
|
||||
|
||||
# Phase 5: Site fallback (low confidence)
|
||||
matched_ids = MapSet.union(matched_so_far, mac_matched)
|
||||
remaining = Enum.reject(items, &MapSet.member?(matched_ids, &1.id))
|
||||
site_links = match_by_site_fallback(remaining, organization_id, now)
|
||||
|
||||
all_links = wc_mac_links ++ wc_ip_links ++ ip_links ++ mac_links ++ site_links
|
||||
|
||||
# Deduplicate by {device_id, gaiia_account_id} — first match wins (highest priority)
|
||||
unique_links =
|
||||
all_links
|
||||
|> Enum.uniq_by(fn l -> {l.device_id, l.gaiia_account_id} end)
|
||||
|
||||
# Transaction: delete old + insert new
|
||||
Repo.transaction(fn ->
|
||||
from(d in DeviceSubscriberLink, where: d.organization_id == ^organization_id)
|
||||
|> Repo.delete_all()
|
||||
|
||||
if unique_links != [] do
|
||||
Repo.insert_all(DeviceSubscriberLink, unique_links)
|
||||
end
|
||||
end)
|
||||
|
||||
Logger.info(
|
||||
"Subscriber matching for org #{organization_id}: " <>
|
||||
"#{length(unique_links)}/#{length(items)} items matched " <>
|
||||
"(#{length(wc_mac_links)} wireless MAC, #{length(wc_ip_links)} wireless IP, " <>
|
||||
"#{length(ip_links)} ARP IP, #{length(mac_links)} ARP MAC, #{length(site_links)} site fallback)"
|
||||
)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp do_refresh_for_device(device) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
org_id = device.organization_id
|
||||
|
||||
items = load_assigned_items(org_id)
|
||||
arp_entries = load_arp_entries_for_device(device.id)
|
||||
wireless_clients = load_wireless_clients_for_device(device.id)
|
||||
|
||||
arp_by_ip = Map.new(arp_entries, fn e -> {e.ip_address, e} end)
|
||||
arp_by_mac = Map.new(arp_entries, fn e -> {normalize_mac(e.mac_address), e} end)
|
||||
wc_by_mac = Map.new(wireless_clients, fn w -> {normalize_mac(w.mac_address), w} end)
|
||||
wc_by_ip = wireless_clients |> Enum.filter(& &1.ip_address) |> Map.new(fn w -> {w.ip_address, w} end)
|
||||
|
||||
# Resolve account IDs from gaiia_id
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
links =
|
||||
Enum.flat_map(items, fn item ->
|
||||
cond do
|
||||
# Wireless MAC match (highest confidence)
|
||||
item.mac_address && Map.has_key?(wc_by_mac, normalize_mac(item.mac_address)) ->
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil -> []
|
||||
account_id ->
|
||||
[build_link(org_id, device.id, account_id, item, "wireless_mac", "high", nil, normalize_mac(item.mac_address), now)]
|
||||
end
|
||||
|
||||
# Wireless IP match (high confidence)
|
||||
item.ip_address && Map.has_key?(wc_by_ip, item.ip_address) ->
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil -> []
|
||||
account_id ->
|
||||
[build_link(org_id, device.id, account_id, item, "wireless_ip", "high", item.ip_address, nil, now)]
|
||||
end
|
||||
|
||||
# ARP IP match (medium confidence)
|
||||
item.ip_address && Map.has_key?(arp_by_ip, item.ip_address) ->
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil -> []
|
||||
account_id ->
|
||||
[build_link(org_id, device.id, account_id, item, "arp_ip", "medium", item.ip_address, nil, now)]
|
||||
end
|
||||
|
||||
# ARP MAC match (medium confidence)
|
||||
item.mac_address && Map.has_key?(arp_by_mac, normalize_mac(item.mac_address)) ->
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil -> []
|
||||
account_id ->
|
||||
[build_link(org_id, device.id, account_id, item, "arp_mac", "medium", nil, normalize_mac(item.mac_address), now)]
|
||||
end
|
||||
|
||||
true ->
|
||||
[]
|
||||
end
|
||||
end)
|
||||
|> Enum.uniq_by(fn l -> {l.device_id, l.gaiia_account_id} end)
|
||||
|
||||
Repo.transaction(fn ->
|
||||
from(d in DeviceSubscriberLink, where: d.device_id == ^device.id)
|
||||
|> Repo.delete_all()
|
||||
|
||||
if links != [] do
|
||||
Repo.insert_all(DeviceSubscriberLink, links)
|
||||
end
|
||||
end)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp load_assigned_items(organization_id) do
|
||||
InventoryItem
|
||||
|> where(organization_id: ^organization_id)
|
||||
|> where([ii], not is_nil(ii.assigned_account_gaiia_id))
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
defp load_arp_entries(organization_id) do
|
||||
ArpEntry
|
||||
|> join(:inner, [a], d in Device, on: a.device_id == d.id)
|
||||
|> where([a, d], d.organization_id == ^organization_id)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
defp load_wireless_clients(organization_id) do
|
||||
WirelessClient
|
||||
|> where([w], w.organization_id == ^organization_id)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
defp load_wireless_clients_for_device(device_id) do
|
||||
WirelessClient
|
||||
|> where(device_id: ^device_id)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
defp load_arp_entries_for_device(device_id) do
|
||||
ArpEntry
|
||||
|> where(device_id: ^device_id)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
defp load_account_id_lookup(organization_id) do
|
||||
Account
|
||||
|> where(organization_id: ^organization_id)
|
||||
|> select([a], {a.gaiia_id, a.id})
|
||||
|> Repo.all()
|
||||
|> Map.new()
|
||||
end
|
||||
|
||||
defp match_by_wireless_mac(items, wc_by_mac, org_id, now) do
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
{matched_ids, links} =
|
||||
Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} ->
|
||||
normalized = normalize_mac(item.mac_address)
|
||||
if normalized && Map.has_key?(wc_by_mac, normalized) do
|
||||
wc = Map.get(wc_by_mac, normalized)
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, wc.device_id, account_id, item, "wireless_mac", "high", nil, normalized, now)
|
||||
{MapSet.put(matched, item.id), [link | acc]}
|
||||
end
|
||||
else
|
||||
{matched, acc}
|
||||
end
|
||||
end)
|
||||
|
||||
{matched_ids, Enum.reverse(links)}
|
||||
end
|
||||
|
||||
defp match_by_wireless_ip(items, wc_by_ip, org_id, now) do
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
{matched_ids, links} =
|
||||
Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} ->
|
||||
if item.ip_address && Map.has_key?(wc_by_ip, item.ip_address) do
|
||||
wc = Map.get(wc_by_ip, item.ip_address)
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, wc.device_id, account_id, item, "wireless_ip", "high", item.ip_address, nil, now)
|
||||
{MapSet.put(matched, item.id), [link | acc]}
|
||||
end
|
||||
else
|
||||
{matched, acc}
|
||||
end
|
||||
end)
|
||||
|
||||
{matched_ids, Enum.reverse(links)}
|
||||
end
|
||||
|
||||
defp match_by_arp_ip(items, arp_by_ip, org_id, now) do
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
{matched_ids, links} =
|
||||
Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} ->
|
||||
if item.ip_address && Map.has_key?(arp_by_ip, item.ip_address) do
|
||||
arp = Map.get(arp_by_ip, item.ip_address)
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, arp.device_id, account_id, item, "arp_ip", "medium", item.ip_address, nil, now)
|
||||
{MapSet.put(matched, item.id), [link | acc]}
|
||||
end
|
||||
else
|
||||
{matched, acc}
|
||||
end
|
||||
end)
|
||||
|
||||
{matched_ids, Enum.reverse(links)}
|
||||
end
|
||||
|
||||
defp match_by_arp_mac(items, arp_by_mac, org_id, now) do
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
{matched_ids, links} =
|
||||
Enum.reduce(items, {MapSet.new(), []}, fn item, {matched, acc} ->
|
||||
normalized = normalize_mac(item.mac_address)
|
||||
if normalized && Map.has_key?(arp_by_mac, normalized) do
|
||||
arp = Map.get(arp_by_mac, normalized)
|
||||
case Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, arp.device_id, account_id, item, "arp_mac", "medium", nil, normalized, now)
|
||||
{MapSet.put(matched, item.id), [link | acc]}
|
||||
end
|
||||
else
|
||||
{matched, acc}
|
||||
end
|
||||
end)
|
||||
|
||||
{matched_ids, Enum.reverse(links)}
|
||||
end
|
||||
|
||||
defp match_by_site_fallback(items, org_id, now) do
|
||||
account_lookup = load_account_id_lookup(org_id)
|
||||
|
||||
# Load network site -> TowerOps site mappings
|
||||
site_mappings =
|
||||
NetworkSite
|
||||
|> where(organization_id: ^org_id)
|
||||
|> where([ns], not is_nil(ns.site_id))
|
||||
|> select([ns], {ns.gaiia_id, ns.site_id})
|
||||
|> Repo.all()
|
||||
|> Map.new()
|
||||
|
||||
# Load devices per site
|
||||
devices_by_site =
|
||||
Device
|
||||
|> where(organization_id: ^org_id)
|
||||
|> where([d], not is_nil(d.site_id))
|
||||
|> select([d], {d.site_id, d.id})
|
||||
|> Repo.all()
|
||||
|> Enum.group_by(&elem(&1, 0), &elem(&1, 1))
|
||||
|
||||
Enum.flat_map(items, fn item ->
|
||||
with gaiia_site_id when not is_nil(gaiia_site_id) <- item.assigned_network_site_gaiia_id,
|
||||
site_id when not is_nil(site_id) <- Map.get(site_mappings, gaiia_site_id),
|
||||
device_ids when device_ids != nil <- Map.get(devices_by_site, site_id),
|
||||
account_id when not is_nil(account_id) <- Map.get(account_lookup, item.assigned_account_gaiia_id) do
|
||||
Enum.map(device_ids, fn device_id ->
|
||||
build_link(org_id, device_id, account_id, item, "site_fallback", "low", nil, nil, now)
|
||||
end)
|
||||
else
|
||||
_ -> []
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp build_link(org_id, device_id, account_id, item, method, confidence, ip, mac, now) do
|
||||
%{
|
||||
id: Ecto.UUID.generate(),
|
||||
organization_id: org_id,
|
||||
device_id: device_id,
|
||||
gaiia_account_id: account_id,
|
||||
gaiia_inventory_item_id: item.id,
|
||||
match_method: method,
|
||||
confidence: confidence,
|
||||
subscriber_ip: ip,
|
||||
subscriber_mac: mac,
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end
|
||||
|
||||
defp normalize_mac(nil), do: nil
|
||||
|
||||
defp normalize_mac(mac) do
|
||||
mac
|
||||
|> String.downcase()
|
||||
|> String.replace(~r/[^0-9a-f]/, "")
|
||||
|> case do
|
||||
<<a::binary-2, b::binary-2, c::binary-2, d::binary-2, e::binary-2, f::binary-2>> ->
|
||||
"#{a}:#{b}:#{c}:#{d}:#{e}:#{f}"
|
||||
_ ->
|
||||
nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -39,6 +39,7 @@ defmodule Towerops.Gaiia.Sync do
|
|||
Integrations.update_sync_status(integration, "success", message)
|
||||
persist_billing_totals(integration)
|
||||
Gaiia.SiteAggregation.compute_and_store(org_id)
|
||||
Towerops.Gaiia.SubscriberMatching.refresh_device_links(org_id)
|
||||
|
||||
{:ok,
|
||||
%{
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ defmodule Towerops.Snmp do
|
|||
alias Towerops.Snmp.Client
|
||||
alias Towerops.Snmp.Device
|
||||
alias Towerops.Snmp.Discovery
|
||||
alias Towerops.Snmp.WirelessClient
|
||||
alias Towerops.Snmp.Interface
|
||||
alias Towerops.Snmp.InterfaceStat
|
||||
alias Towerops.Snmp.IpAddress
|
||||
|
|
@ -2285,4 +2286,78 @@ defmodule Towerops.Snmp do
|
|||
|> String.replace("%", "\\%")
|
||||
|> String.replace("_", "\\_")
|
||||
end
|
||||
|
||||
# Wireless Client queries
|
||||
|
||||
@doc """
|
||||
Lists all wireless clients for a device.
|
||||
"""
|
||||
def list_wireless_clients(device_id) do
|
||||
WirelessClient
|
||||
|> where([w], w.device_id == ^device_id)
|
||||
|> order_by([w], w.mac_address)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
@doc """
|
||||
Gets wireless clients by MAC addresses within an organization.
|
||||
"""
|
||||
def get_wireless_clients_by_mac(organization_id, mac_addresses) do
|
||||
WirelessClient
|
||||
|> where([w], w.organization_id == ^organization_id)
|
||||
|> where([w], w.mac_address in ^mac_addresses)
|
||||
|> Repo.all()
|
||||
end
|
||||
|
||||
@doc """
|
||||
Creates or updates a wireless client entry.
|
||||
"""
|
||||
def upsert_wireless_client(attrs) do
|
||||
%WirelessClient{}
|
||||
|> WirelessClient.changeset(attrs)
|
||||
|> Repo.insert(
|
||||
on_conflict: {:replace_all_except, [:id, :inserted_at]},
|
||||
conflict_target: [:device_id, :mac_address],
|
||||
returning: true
|
||||
)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Upserts a list of wireless clients for a device.
|
||||
Returns `{success_count, error_count}`.
|
||||
"""
|
||||
def upsert_wireless_clients(device_id, organization_id, clients) do
|
||||
clients
|
||||
|> Enum.reduce({0, 0}, fn client, {s, e} ->
|
||||
attrs =
|
||||
client
|
||||
|> Map.put(:device_id, device_id)
|
||||
|> Map.put(:organization_id, organization_id)
|
||||
|
||||
case upsert_wireless_client(attrs) do
|
||||
{:ok, _} -> {s + 1, e}
|
||||
{:error, _} -> {s, e + 1}
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Deletes stale wireless clients that haven't been seen since the given datetime.
|
||||
"""
|
||||
def delete_stale_wireless_clients(device_id, cutoff_datetime) do
|
||||
WirelessClient
|
||||
|> where([w], w.device_id == ^device_id)
|
||||
|> where([w], w.last_seen_at < ^cutoff_datetime)
|
||||
|> Repo.delete_all()
|
||||
end
|
||||
|
||||
@doc """
|
||||
Atomically deletes stale wireless clients and upserts new ones.
|
||||
"""
|
||||
def delete_stale_and_upsert_wireless_clients(device_id, organization_id, clients, cutoff) do
|
||||
Repo.transaction(fn ->
|
||||
delete_stale_wireless_clients(device_id, cutoff)
|
||||
upsert_wireless_clients(device_id, organization_id, clients)
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
56
lib/towerops/snmp/wireless_client.ex
Normal file
56
lib/towerops/snmp/wireless_client.ex
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
defmodule Towerops.Snmp.WirelessClient do
|
||||
@moduledoc """
|
||||
Schema for wireless client associations polled from APs.
|
||||
|
||||
Stores the connected subscriber radios (CPEs/SMs) discovered via
|
||||
vendor-specific SNMP wireless registration/client tables.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
schema "wireless_clients" do
|
||||
field :mac_address, :string
|
||||
field :ip_address, :string
|
||||
field :hostname, :string
|
||||
field :signal_strength, :integer
|
||||
field :snr, :integer
|
||||
field :distance, :integer
|
||||
field :tx_rate, :integer
|
||||
field :rx_rate, :integer
|
||||
field :uptime_seconds, :integer
|
||||
field :last_seen_at, :utc_datetime
|
||||
field :metadata, :map, default: %{}
|
||||
|
||||
belongs_to :device, Towerops.Devices.Device
|
||||
belongs_to :organization, Towerops.Organizations.Organization
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
@doc false
|
||||
def changeset(wireless_client, attrs) do
|
||||
wireless_client
|
||||
|> cast(attrs, [
|
||||
:device_id,
|
||||
:organization_id,
|
||||
:mac_address,
|
||||
:ip_address,
|
||||
:hostname,
|
||||
:signal_strength,
|
||||
:snr,
|
||||
:distance,
|
||||
:tx_rate,
|
||||
:rx_rate,
|
||||
:uptime_seconds,
|
||||
:last_seen_at,
|
||||
:metadata
|
||||
])
|
||||
|> validate_required([:device_id, :organization_id, :mac_address, :last_seen_at])
|
||||
|> foreign_key_constraint(:device_id)
|
||||
|> foreign_key_constraint(:organization_id)
|
||||
|> unique_constraint([:device_id, :mac_address])
|
||||
end
|
||||
end
|
||||
289
lib/towerops/snmp/wireless_client_discovery.ex
Normal file
289
lib/towerops/snmp/wireless_client_discovery.ex
Normal file
|
|
@ -0,0 +1,289 @@
|
|||
defmodule Towerops.Snmp.WirelessClientDiscovery do
|
||||
@moduledoc """
|
||||
Discovers wireless client/subscriber module associations from APs via SNMP.
|
||||
|
||||
Walks vendor-specific wireless registration/client tables:
|
||||
- Cambium ePMP: cambiumAPConnectedSTATable
|
||||
- Ubiquiti airOS: ubntStaTable
|
||||
- MikroTik RouterOS: mtxrWlRtabTable
|
||||
"""
|
||||
|
||||
alias Towerops.Snmp.Client
|
||||
|
||||
require Logger
|
||||
|
||||
# Cambium ePMP - cambiumAPConnectedSTATable
|
||||
@epmp_base "1.3.6.1.4.1.17713.21.1.2.30.1"
|
||||
|
||||
# Ubiquiti airOS - ubntStaTable
|
||||
@ubnt_base "1.3.6.1.4.1.41112.1.4.7.1"
|
||||
|
||||
# MikroTik RouterOS - mtxrWlRtabTable
|
||||
@mikrotik_base "1.3.6.1.4.1.14988.1.1.1.2.1"
|
||||
|
||||
@doc """
|
||||
Discovers wireless clients connected to an AP.
|
||||
|
||||
Returns `{:ok, [client_map]}` where each client_map has keys matching
|
||||
the wireless_clients schema fields.
|
||||
"""
|
||||
@spec discover_wireless_clients(Client.connection_opts(), String.t()) :: {:ok, [map()]}
|
||||
def discover_wireless_clients(client_opts, vendor_profile) do
|
||||
cond do
|
||||
vendor_profile in ["epmp", "cambium", "pmp"] -> discover_epmp_clients(client_opts)
|
||||
vendor_profile in ["airos", "airfiber", "airos-af", "airos-af60", "airos-af-ltu"] -> discover_ubnt_clients(client_opts)
|
||||
vendor_profile in ["routeros", "mikrotik"] -> discover_mikrotik_clients(client_opts)
|
||||
true -> {:ok, []}
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Determines the vendor profile string from an SNMP device's sys_object_id and manufacturer.
|
||||
"""
|
||||
@spec detect_vendor_profile(map()) :: String.t() | nil
|
||||
def detect_vendor_profile(snmp_device) do
|
||||
sys_oid = snmp_device.sys_object_id || ""
|
||||
manufacturer = String.downcase(snmp_device.manufacturer || "")
|
||||
|
||||
cond do
|
||||
String.starts_with?(sys_oid, "1.3.6.1.4.1.17713") or String.contains?(manufacturer, "cambium") -> "epmp"
|
||||
String.starts_with?(sys_oid, "1.3.6.1.4.1.41112") or String.contains?(manufacturer, "ubiquiti") -> "airos"
|
||||
String.starts_with?(sys_oid, "1.3.6.1.4.1.14988") or String.contains?(manufacturer, "mikrotik") -> "routeros"
|
||||
true -> nil
|
||||
end
|
||||
end
|
||||
|
||||
# --- Cambium ePMP ---
|
||||
|
||||
defp discover_epmp_clients(client_opts) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
with {:ok, macs} <- Client.walk(client_opts, "#{@epmp_base}.1"),
|
||||
{:ok, ips} <- walk_or_empty(client_opts, "#{@epmp_base}.10"),
|
||||
{:ok, ul_rssi} <- walk_or_empty(client_opts, "#{@epmp_base}.4"),
|
||||
{:ok, dl_rssi} <- walk_or_empty(client_opts, "#{@epmp_base}.5"),
|
||||
{:ok, ul_snr} <- walk_or_empty(client_opts, "#{@epmp_base}.6"),
|
||||
{:ok, dl_snr} <- walk_or_empty(client_opts, "#{@epmp_base}.7"),
|
||||
{:ok, distances} <- walk_or_empty(client_opts, "#{@epmp_base}.29"),
|
||||
{:ok, ul_mcs} <- walk_or_empty(client_opts, "#{@epmp_base}.8"),
|
||||
{:ok, dl_mcs} <- walk_or_empty(client_opts, "#{@epmp_base}.9"),
|
||||
{:ok, models} <- walk_or_empty(client_opts, "#{@epmp_base}.38"),
|
||||
{:ok, sessions} <- walk_or_empty(client_opts, "#{@epmp_base}.27") do
|
||||
clients =
|
||||
macs
|
||||
|> group_by_index("#{@epmp_base}.1")
|
||||
|> Enum.map(fn {idx, mac_raw} ->
|
||||
%{
|
||||
mac_address: format_mac(mac_raw),
|
||||
ip_address: get_indexed_value(ips, "#{@epmp_base}.10", idx) |> format_ip(),
|
||||
signal_strength: best_value(
|
||||
get_indexed_value(dl_rssi, "#{@epmp_base}.5", idx),
|
||||
get_indexed_value(ul_rssi, "#{@epmp_base}.4", idx)
|
||||
),
|
||||
snr: best_value(
|
||||
get_indexed_value(dl_snr, "#{@epmp_base}.7", idx),
|
||||
get_indexed_value(ul_snr, "#{@epmp_base}.6", idx)
|
||||
),
|
||||
distance: get_indexed_int(distances, "#{@epmp_base}.29", idx),
|
||||
tx_rate: nil,
|
||||
rx_rate: nil,
|
||||
uptime_seconds: parse_session_time(get_indexed_value(sessions, "#{@epmp_base}.27", idx)),
|
||||
last_seen_at: now,
|
||||
metadata: %{
|
||||
"ul_mcs" => get_indexed_value(ul_mcs, "#{@epmp_base}.8", idx),
|
||||
"dl_mcs" => get_indexed_value(dl_mcs, "#{@epmp_base}.9", idx),
|
||||
"model" => get_indexed_string(models, "#{@epmp_base}.38", idx),
|
||||
"vendor" => "cambium_epmp"
|
||||
}
|
||||
}
|
||||
end)
|
||||
|> Enum.reject(&is_nil(&1.mac_address))
|
||||
|
||||
{:ok, clients}
|
||||
else
|
||||
_ -> {:ok, []}
|
||||
end
|
||||
end
|
||||
|
||||
# --- Ubiquiti airOS ---
|
||||
|
||||
defp discover_ubnt_clients(client_opts) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
with {:ok, macs} <- Client.walk(client_opts, "#{@ubnt_base}.1"),
|
||||
{:ok, ips} <- walk_or_empty(client_opts, "#{@ubnt_base}.10"),
|
||||
{:ok, names} <- walk_or_empty(client_opts, "#{@ubnt_base}.2"),
|
||||
{:ok, signals} <- walk_or_empty(client_opts, "#{@ubnt_base}.3"),
|
||||
{:ok, distances} <- walk_or_empty(client_opts, "#{@ubnt_base}.5"),
|
||||
{:ok, ccqs} <- walk_or_empty(client_opts, "#{@ubnt_base}.6"),
|
||||
{:ok, tx_rates} <- walk_or_empty(client_opts, "#{@ubnt_base}.11"),
|
||||
{:ok, rx_rates} <- walk_or_empty(client_opts, "#{@ubnt_base}.12"),
|
||||
{:ok, conn_times} <- walk_or_empty(client_opts, "#{@ubnt_base}.15") do
|
||||
clients =
|
||||
macs
|
||||
|> group_by_index("#{@ubnt_base}.1")
|
||||
|> Enum.map(fn {idx, mac_raw} ->
|
||||
%{
|
||||
mac_address: format_mac(mac_raw),
|
||||
ip_address: get_indexed_value(ips, "#{@ubnt_base}.10", idx) |> format_ip(),
|
||||
hostname: get_indexed_string(names, "#{@ubnt_base}.2", idx),
|
||||
signal_strength: get_indexed_int(signals, "#{@ubnt_base}.3", idx),
|
||||
snr: nil,
|
||||
distance: get_indexed_int(distances, "#{@ubnt_base}.5", idx),
|
||||
tx_rate: get_indexed_int(tx_rates, "#{@ubnt_base}.11", idx),
|
||||
rx_rate: get_indexed_int(rx_rates, "#{@ubnt_base}.12", idx),
|
||||
uptime_seconds: timeticks_to_seconds(get_indexed_value(conn_times, "#{@ubnt_base}.15", idx)),
|
||||
last_seen_at: now,
|
||||
metadata: %{
|
||||
"ccq" => get_indexed_value(ccqs, "#{@ubnt_base}.6", idx),
|
||||
"vendor" => "ubiquiti_airos"
|
||||
}
|
||||
}
|
||||
end)
|
||||
|> Enum.reject(&is_nil(&1.mac_address))
|
||||
|
||||
{:ok, clients}
|
||||
else
|
||||
_ -> {:ok, []}
|
||||
end
|
||||
end
|
||||
|
||||
# --- MikroTik RouterOS ---
|
||||
|
||||
defp discover_mikrotik_clients(client_opts) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
with {:ok, macs} <- Client.walk(client_opts, "#{@mikrotik_base}.1"),
|
||||
{:ok, signals} <- walk_or_empty(client_opts, "#{@mikrotik_base}.3"),
|
||||
{:ok, tx_rates} <- walk_or_empty(client_opts, "#{@mikrotik_base}.11"),
|
||||
{:ok, rx_rates} <- walk_or_empty(client_opts, "#{@mikrotik_base}.12") do
|
||||
clients =
|
||||
macs
|
||||
|> group_by_index("#{@mikrotik_base}.1")
|
||||
|> Enum.map(fn {idx, mac_raw} ->
|
||||
%{
|
||||
mac_address: format_mac(mac_raw),
|
||||
ip_address: nil,
|
||||
hostname: nil,
|
||||
signal_strength: get_indexed_int(signals, "#{@mikrotik_base}.3", idx),
|
||||
snr: nil,
|
||||
distance: nil,
|
||||
tx_rate: get_indexed_int(tx_rates, "#{@mikrotik_base}.11", idx),
|
||||
rx_rate: get_indexed_int(rx_rates, "#{@mikrotik_base}.12", idx),
|
||||
uptime_seconds: nil,
|
||||
last_seen_at: now,
|
||||
metadata: %{"vendor" => "mikrotik_routeros"}
|
||||
}
|
||||
end)
|
||||
|> Enum.reject(&is_nil(&1.mac_address))
|
||||
|
||||
{:ok, clients}
|
||||
else
|
||||
_ -> {:ok, []}
|
||||
end
|
||||
end
|
||||
|
||||
# --- Helpers ---
|
||||
|
||||
defp walk_or_empty(client_opts, oid) do
|
||||
case Client.walk(client_opts, oid) do
|
||||
{:ok, entries} -> {:ok, entries}
|
||||
{:error, _} -> {:ok, %{}}
|
||||
end
|
||||
end
|
||||
|
||||
# Group walk results by row index, extracting the index suffix from OID keys
|
||||
defp group_by_index(walk_results, base_oid) do
|
||||
base_len = String.length(base_oid) + 1
|
||||
|
||||
walk_results
|
||||
|> Enum.map(fn {oid, value} ->
|
||||
idx = String.slice(oid, base_len..-1//1)
|
||||
{idx, value}
|
||||
end)
|
||||
|> Enum.reject(fn {idx, _} -> idx == "" end)
|
||||
end
|
||||
|
||||
defp get_indexed_value(walk_results, base_oid, idx) do
|
||||
Map.get(walk_results, "#{base_oid}.#{idx}")
|
||||
end
|
||||
|
||||
defp get_indexed_int(walk_results, base_oid, idx) do
|
||||
case get_indexed_value(walk_results, base_oid, idx) do
|
||||
v when is_integer(v) -> v
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp get_indexed_string(walk_results, base_oid, idx) do
|
||||
case get_indexed_value(walk_results, base_oid, idx) do
|
||||
v when is_binary(v) and v != "" -> v
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
# Format MAC from binary (6 bytes) or string to lowercase colon-separated
|
||||
defp format_mac(value) when is_binary(value) and byte_size(value) == 6 do
|
||||
value
|
||||
|> :binary.bin_to_list()
|
||||
|> Enum.map_join(":", &String.pad_leading(Integer.to_string(&1, 16), 2, "0"))
|
||||
|> String.downcase()
|
||||
end
|
||||
|
||||
defp format_mac(value) when is_binary(value) do
|
||||
if String.contains?(value, ":") or String.contains?(value, "-") do
|
||||
value
|
||||
|> String.downcase()
|
||||
|> String.replace("-", ":")
|
||||
else
|
||||
value
|
||||
|> :binary.bin_to_list()
|
||||
|> Enum.map_join(":", &String.pad_leading(Integer.to_string(&1, 16), 2, "0"))
|
||||
|> String.downcase()
|
||||
end
|
||||
end
|
||||
|
||||
defp format_mac(_), do: nil
|
||||
|
||||
# Format IP address from SNMP IpAddress type or string
|
||||
defp format_ip(value) when is_binary(value) and byte_size(value) == 4 do
|
||||
value
|
||||
|> :binary.bin_to_list()
|
||||
|> Enum.join(".")
|
||||
end
|
||||
|
||||
defp format_ip(value) when is_binary(value) do
|
||||
if String.match?(value, ~r/^\d+\.\d+\.\d+\.\d+$/) do
|
||||
value
|
||||
else
|
||||
nil
|
||||
end
|
||||
end
|
||||
|
||||
defp format_ip(_), do: nil
|
||||
|
||||
# Pick the best (non-nil) of two values, preferring the first
|
||||
defp best_value(nil, b), do: if(is_integer(b), do: b, else: nil)
|
||||
defp best_value(a, _) when is_integer(a), do: a
|
||||
defp best_value(_, b) when is_integer(b), do: b
|
||||
defp best_value(_, _), do: nil
|
||||
|
||||
# Parse ePMP session time (DisplayString like "1d 2h 3m 4s" or seconds)
|
||||
defp parse_session_time(nil), do: nil
|
||||
|
||||
defp parse_session_time(value) when is_binary(value) do
|
||||
# Try to parse common formats
|
||||
case Integer.parse(value) do
|
||||
{seconds, ""} -> seconds
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_session_time(value) when is_integer(value), do: value
|
||||
defp parse_session_time(_), do: nil
|
||||
|
||||
# Convert TimeTicks (hundredths of seconds) to seconds
|
||||
defp timeticks_to_seconds(nil), do: nil
|
||||
defp timeticks_to_seconds(value) when is_integer(value), do: div(value, 100)
|
||||
defp timeticks_to_seconds(_), do: nil
|
||||
end
|
||||
|
|
@ -234,10 +234,19 @@ defmodule Towerops.Workers.DeviceMonitorWorker do
|
|||
end
|
||||
|
||||
defp get_down_alert_message(device) do
|
||||
if device.snmp_enabled do
|
||||
"Device is not responding to SNMP"
|
||||
else
|
||||
"Device is not responding to ping"
|
||||
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
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
alias Towerops.Snmp.MacDiscovery
|
||||
alias Towerops.Snmp.NeighborDiscovery
|
||||
alias Towerops.Snmp.SensorChangeDetector
|
||||
alias Towerops.Snmp.WirelessClientDiscovery
|
||||
alias Towerops.Workers.PollingOffset
|
||||
|
||||
require Logger
|
||||
|
|
@ -185,7 +186,8 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
"arp",
|
||||
"mac",
|
||||
"processors",
|
||||
"storage"
|
||||
"storage",
|
||||
"wireless_clients"
|
||||
]
|
||||
|
||||
tasks = [
|
||||
|
|
@ -197,7 +199,8 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
Task.async(fn -> poll_device_arp(device, snmp_device, client_opts) end),
|
||||
Task.async(fn -> poll_device_mac(device, snmp_device, client_opts) end),
|
||||
Task.async(fn -> poll_device_processors(device, snmp_device, client_opts, now) end),
|
||||
Task.async(fn -> poll_device_storage(device, snmp_device, client_opts, now) end)
|
||||
Task.async(fn -> poll_device_storage(device, snmp_device, client_opts, now) end),
|
||||
Task.async(fn -> poll_wireless_clients(device, snmp_device, client_opts) end)
|
||||
]
|
||||
|
||||
# Wait for all tasks with timeout, log failures
|
||||
|
|
@ -356,6 +359,9 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
"device:#{device.id}",
|
||||
{:arp_updated, device.id}
|
||||
)
|
||||
|
||||
# Refresh subscriber-to-device links based on new ARP data
|
||||
Towerops.Gaiia.SubscriberMatching.refresh_links_for_device(device.id)
|
||||
rescue
|
||||
error ->
|
||||
Logger.error("Error polling ARP table for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}")
|
||||
|
|
@ -380,6 +386,34 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
Logger.error("Error polling MAC FDB table for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}")
|
||||
end
|
||||
|
||||
defp poll_wireless_clients(device, snmp_device, client_opts) do
|
||||
vendor_profile = WirelessClientDiscovery.detect_vendor_profile(snmp_device)
|
||||
|
||||
if vendor_profile do
|
||||
{:ok, clients} = WirelessClientDiscovery.discover_wireless_clients(client_opts, vendor_profile)
|
||||
|
||||
if clients != [] do
|
||||
cutoff = DateTime.add(DateTime.utc_now(), -5, :minute)
|
||||
Snmp.delete_stale_and_upsert_wireless_clients(device.id, device.organization_id, clients, cutoff)
|
||||
|
||||
Logger.debug("Polled and saved #{length(clients)} wireless clients for #{device.name}")
|
||||
|
||||
_ =
|
||||
Phoenix.PubSub.broadcast(
|
||||
Towerops.PubSub,
|
||||
"device:#{device.id}",
|
||||
{:wireless_clients_updated, device.id}
|
||||
)
|
||||
|
||||
# Refresh subscriber-to-device links with new wireless client data
|
||||
Towerops.Gaiia.SubscriberMatching.refresh_links_for_device(device.id)
|
||||
end
|
||||
end
|
||||
rescue
|
||||
error ->
|
||||
Logger.error("Error polling wireless clients for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}")
|
||||
end
|
||||
|
||||
defp poll_device_processors(device, snmp_device, client_opts, now) do
|
||||
processors = snmp_device.processors || []
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
defmodule Towerops.Repo.Migrations.CreateDeviceSubscriberLinks do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
create table(:device_subscriber_links, 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 :device_id, references(:devices, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :gaiia_account_id, references(:gaiia_accounts, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :gaiia_inventory_item_id, references(:gaiia_inventory_items, type: :binary_id, on_delete: :nilify_all)
|
||||
add :match_method, :string, null: false
|
||||
add :confidence, :string, null: false
|
||||
add :subscriber_ip, :string
|
||||
add :subscriber_mac, :string
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create unique_index(:device_subscriber_links, [:device_id, :gaiia_account_id])
|
||||
create index(:device_subscriber_links, [:device_id])
|
||||
create index(:device_subscriber_links, [:gaiia_account_id])
|
||||
create index(:device_subscriber_links, [:organization_id])
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
defmodule Towerops.Repo.Migrations.CreateWirelessClients do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
create table(:wireless_clients, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :device_id, references(:devices, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), null: false
|
||||
add :mac_address, :string, null: false
|
||||
add :ip_address, :string
|
||||
add :hostname, :string
|
||||
add :signal_strength, :integer
|
||||
add :snr, :integer
|
||||
add :distance, :integer
|
||||
add :tx_rate, :integer
|
||||
add :rx_rate, :integer
|
||||
add :uptime_seconds, :integer
|
||||
add :last_seen_at, :utc_datetime, null: false
|
||||
add :metadata, :map, default: %{}
|
||||
|
||||
timestamps(type: :utc_datetime)
|
||||
end
|
||||
|
||||
create unique_index(:wireless_clients, [:device_id, :mac_address])
|
||||
create index(:wireless_clients, [:device_id])
|
||||
create index(:wireless_clients, [:mac_address])
|
||||
create index(:wireless_clients, [:organization_id])
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue