Add wireless client table discovery and improve subscriber matching
Poll connected client/SM tables from APs via vendor-specific SNMP: - Cambium ePMP: cambiumAPConnectedSTATable - Ubiquiti airOS: ubntStaTable - MikroTik RouterOS: mtxrWlRtabTable Wireless client MAC matches now take highest priority for subscriber-to-AP mapping, giving accurate per-radio subscriber counts.
This commit is contained in:
parent
520f7b0c54
commit
b12d46c716
6 changed files with 589 additions and 15 deletions
|
|
@ -17,6 +17,7 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
alias Towerops.Integrations
|
||||
alias Towerops.Repo
|
||||
alias Towerops.Snmp.ArpEntry
|
||||
alias Towerops.Snmp.WirelessClient
|
||||
|
||||
require Logger
|
||||
|
||||
|
|
@ -65,24 +66,38 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
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: ARP IP match
|
||||
{ip_matched, ip_links} = match_by_arp_ip(items, arp_by_ip, organization_id, now)
|
||||
# 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: ARP MAC match (remaining items)
|
||||
remaining_after_ip = Enum.reject(items, &MapSet.member?(ip_matched, &1.id))
|
||||
# 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 3: Site fallback (remaining items)
|
||||
matched_ids = MapSet.union(ip_matched, mac_matched)
|
||||
# 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 = ip_links ++ mac_links ++ site_links
|
||||
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 =
|
||||
|
|
@ -102,7 +117,8 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
Logger.info(
|
||||
"Subscriber matching for org #{organization_id}: " <>
|
||||
"#{length(unique_links)}/#{length(items)} items matched " <>
|
||||
"(#{length(ip_links)} ARP IP, #{length(mac_links)} ARP MAC, #{length(site_links)} site fallback)"
|
||||
"(#{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
|
||||
|
|
@ -114,9 +130,12 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
|
||||
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)
|
||||
|
|
@ -124,18 +143,36 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
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", "high", item.ip_address, nil, now)]
|
||||
[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", "high", nil, normalize_mac(item.mac_address), now)]
|
||||
[build_link(org_id, device.id, account_id, item, "arp_mac", "medium", nil, normalize_mac(item.mac_address), now)]
|
||||
end
|
||||
|
||||
true ->
|
||||
|
|
@ -170,6 +207,18 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
|> 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)
|
||||
|
|
@ -184,6 +233,51 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
|> 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)
|
||||
|
||||
|
|
@ -195,7 +289,7 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, arp.device_id, account_id, item, "arp_ip", "high", item.ip_address, nil, now)
|
||||
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
|
||||
|
|
@ -218,7 +312,7 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
nil ->
|
||||
{matched, acc}
|
||||
account_id ->
|
||||
link = build_link(org_id, arp.device_id, account_id, item, "arp_mac", "high", nil, normalized, now)
|
||||
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
|
||||
|
|
@ -256,7 +350,7 @@ defmodule Towerops.Gaiia.SubscriberMatching do
|
|||
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", "medium", nil, nil, now)
|
||||
build_link(org_id, device_id, account_id, item, "site_fallback", "low", nil, nil, now)
|
||||
end)
|
||||
else
|
||||
_ -> []
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -383,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,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