towerops/lib/towerops_web/live/device_live/show.ex
Graham McIntire aa1bac0d46
perf: batch inserts, selective LiveView reloading, and database indexes for scale
Replace individual Repo.insert() with Repo.insert_all() for sensor readings,
interface stats, processor readings, and storage readings in agent channel and
device poller worker. Add partial/composite database indexes for common query
patterns. Dashboard now uses GROUP BY aggregation instead of loading all devices.
DeviceLive.Show only reloads data for the active tab on PubSub events.
DeviceLive.Index debounces status changes and skips quota queries on updates.
Agent polling preloads filtered to monitored sensors/interfaces only.
2026-02-12 10:16:10 -06:00

1362 lines
42 KiB
Elixir

defmodule ToweropsWeb.DeviceLive.Show do
@moduledoc false
use ToweropsWeb, :live_view
alias Towerops.Agents
alias Towerops.Devices
alias Towerops.Devices.Firmware
alias Towerops.Devices.MikrotikBackups
alias Towerops.Devices.VersionComparator
alias Towerops.Monitoring
alias Towerops.Repo
alias Towerops.Snmp
alias ToweropsWeb.Live.Helpers.AccessControl
# SNMP interface type to category mappings
@interface_type_categories %{
6 => "Ethernet",
24 => "Loopback",
23 => "PPP",
108 => "PPPoE",
131 => "Tunnel",
135 => "VLAN",
136 => "VLAN",
161 => "IEEE 802.11",
244 => "WWP"
}
# Display order for interface categories
@interface_category_order %{
"Ethernet" => 1,
"VLAN" => 2,
"IEEE 802.11" => 3,
"PPPoE" => 4,
"PPP" => 5,
"Tunnel" => 6,
"Loopback" => 7,
"WWP" => 8,
"Other" => 99
}
@impl true
def mount(_params, _session, socket) do
{:ok, socket}
end
@impl true
def handle_params(%{"id" => id} = params, _, socket) do
organization = socket.assigns.current_scope.organization
# Check if device exists and verify organization access
case AccessControl.verify_device_access(id, organization.id) do
{:ok, _device} ->
maybe_subscribe_and_schedule_refresh(socket, id)
# If no tab parameter, redirect to add it to the URL
if Map.has_key?(params, "tab") do
tab = params["tab"]
page = params |> Map.get("page", "1") |> String.to_integer()
socket =
socket
|> load_equipment_data(id)
|> assign(:active_tab, tab)
|> apply_backups_pagination(tab, page)
{:noreply, socket}
else
# Redirect to add default tab to URL
{:noreply, push_patch(socket, to: ~p"/devices/#{id}?tab=overview")}
end
{:error, :not_found} ->
{:noreply,
socket
|> put_flash(:error, t_equipment("Device not found"))
|> push_navigate(to: ~p"/devices")}
{:error, :unauthorized} ->
{:noreply,
socket
|> put_flash(:error, t_equipment("You don't have access to this device"))
|> push_navigate(to: ~p"/devices")}
end
end
@impl true
def handle_info(:refresh_data, socket) do
# Check if device still exists before scheduling next refresh
case Devices.get_device(socket.assigns.device.id) do
nil ->
# Device was deleted, don't schedule another refresh
{:noreply,
socket
|> put_flash(:error, t_equipment("Device no longer exists"))
|> push_navigate(to: ~p"/devices")}
_device ->
Process.send_after(self(), :refresh_data, 10_000)
{:noreply, reload_active_tab_data(socket)}
end
end
@impl true
def handle_info({:device_status_changed, _device_id, _new_status, _response_time}, socket) do
# Device status shows on all tabs in the header - reload base data only
{:noreply, reload_base_data(socket)}
end
@impl true
def handle_info({:discovery_completed, _device_id}, socket) do
# Structural change - full reload needed
{:noreply,
socket
|> load_equipment_data(socket.assigns.device.id)
|> put_flash(:info, t_equipment("Discovery completed"))}
end
@impl true
def handle_info({:sensors_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview"])}
end
@impl true
def handle_info({:interfaces_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview", "ports"])}
end
@impl true
def handle_info({:neighbors_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["neighbors"])}
end
@impl true
def handle_info({:arp_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["arp"])}
end
@impl true
def handle_info({:mac_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["mac"])}
end
@impl true
def handle_info({:state_sensors_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview"])}
end
@impl true
def handle_info({:processors_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview"])}
end
@impl true
def handle_info({:storage_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview"])}
end
@impl true
def handle_info({:monitoring_check_updated, _device_id}, socket) do
{:noreply, reload_if_tab(socket, ["overview"])}
end
@impl true
def handle_info({:device_event, _event_attrs}, socket) do
# Device event logged (interface changes, sensor thresholds, etc.)
{:noreply, reload_if_tab(socket, ["logs"])}
end
@impl true
def handle_info({:agent_connected, agent_token_id, _organization_id}, socket) do
# Reload agent info if this device's agent just connected
if socket.assigns.agent_info.agent_token_id == agent_token_id do
{:noreply, reload_base_data(socket)}
else
{:noreply, socket}
end
end
@impl true
def handle_info({:agent_disconnected, agent_token_id, _organization_id}, socket) do
# Reload agent info if this device's agent just disconnected
if socket.assigns.agent_info.agent_token_id == agent_token_id do
{:noreply, reload_base_data(socket)}
else
{:noreply, socket}
end
end
@impl true
def handle_info({:agent_heartbeat, agent_token_id, _organization_id}, socket) do
# Reload agent info if this device's agent sent a heartbeat
if socket.assigns.agent_info.agent_token_id == agent_token_id do
{:noreply, reload_base_data(socket)}
else
{:noreply, socket}
end
end
@impl true
def handle_info({:agents_stale, stale_agents}, socket) do
# Reload agent info if this device's agent went stale
device_agent_id = socket.assigns.agent_info.agent_token_id
if device_agent_id && Enum.any?(stale_agents, &(&1.id == device_agent_id)) do
{:noreply, reload_base_data(socket)}
else
{:noreply, socket}
end
end
# Private functions
defp maybe_subscribe_and_schedule_refresh(socket, device_id) do
_ =
if connected?(socket) do
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device_id}")
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health")
Process.send_after(self(), :refresh_data, 10_000)
end
:ok
end
# -- Full data load (initial page load and discovery_completed) --
# Loads all data for every tab. Called from handle_params (initial load)
# and discovery_completed (structural change requiring full refresh).
defp load_equipment_data(socket, device_id) do
socket
|> assign_base_data(device_id)
|> assign_tab_nav_data(device_id)
|> assign_overview_data(device_id)
|> assign_ports_data()
|> assign_logs_data(device_id)
|> assign_backups_data(device_id)
end
# -- Selective reload helpers (used by handle_info handlers) --
# Reload only base data (device, agent_info, snmp_device).
# Used when device status or agent state changes - these show in the header on all tabs.
defp reload_base_data(socket) do
assign_base_data(socket, socket.assigns.device.id)
end
# Reload base data + data for the currently active tab only.
# Used by the periodic refresh timer.
defp reload_active_tab_data(socket) do
device_id = socket.assigns.device.id
socket
|> assign_base_data(device_id)
|> assign_tab_nav_data(device_id)
|> reload_current_tab_data(device_id)
end
# Reload data only if the active tab is one of the relevant tabs.
# Otherwise does nothing - the event is irrelevant to what the user sees.
defp reload_if_tab(socket, relevant_tabs) do
if socket.assigns.active_tab in relevant_tabs do
reload_current_tab_data(socket, socket.assigns.device.id)
else
socket
end
end
# Dispatch to the appropriate tab-specific loader based on active_tab.
defp reload_current_tab_data(socket, device_id) do
case socket.assigns.active_tab do
"overview" -> assign_overview_data(socket, device_id)
"ports" -> assign_ports_data(socket)
"logs" -> assign_logs_data(socket, device_id)
"backups" -> assign_backups_data(socket, device_id)
# neighbors, arp, mac, vlans, ip_addresses, debug use data from tab_nav/base
_ -> socket
end
end
# -- Focused data loaders --
# Base data needed by all tabs: device record, agent info, SNMP device/sensors/interfaces.
defp assign_base_data(socket, device_id) do
device = Devices.get_device!(device_id)
device_with_associations = Repo.preload(device, [:organization, site: [organization: :default_agent_token]])
{agent_token_id, agent_source} = Agents.get_effective_agent_token_with_source(device_with_associations)
agent_info = load_agent_info(agent_token_id, agent_source)
snmp_data = load_snmp_data(device_id)
socket
|> assign(:page_title, device.name)
|> assign(:timezone, socket.assigns.current_scope.timezone)
|> assign(:device, device)
|> assign(:agent_info, agent_info)
|> assign(:snmp_device, snmp_data.device)
|> assign(:snmp_interfaces, snmp_data.interfaces)
|> assign(:snmp_sensors, snmp_data.sensors)
end
# Data needed for the tab navigation bar to decide which tabs to show.
# Neighbors, ARP, MAC, VLANs, and IP addresses determine tab visibility.
defp assign_tab_nav_data(socket, device_id) do
snmp_device = socket.assigns.snmp_device
neighbors = Snmp.list_neighbors_with_equipment(device_id)
arp_entries = Snmp.list_arp_entries(device_id)
mac_addresses = Snmp.list_mac_addresses(device_id)
vlans = load_vlans(snmp_device)
ip_addresses = load_ip_addresses(snmp_device)
socket
|> assign(:neighbors, neighbors)
|> assign(:arp_entries, arp_entries)
|> assign(:mac_addresses, mac_addresses)
|> assign(:vlans, vlans)
|> assign(:ipv4_addresses, Enum.filter(ip_addresses, &(&1.ip_type == "ipv4")))
|> assign(:ipv6_addresses, Enum.filter(ip_addresses, &(&1.ip_type == "ipv6")))
|> assign(:ip_addresses_by_interface, group_ip_addresses_by_interface(ip_addresses))
end
# Overview tab: charts, sensor classifications, metrics, processors, storage, firmware.
# This is the most expensive loader due to time-series chart data queries.
defp assign_overview_data(socket, device_id) do
device = socket.assigns.device
snmp_device = socket.assigns.snmp_device
sensors = socket.assigns.snmp_sensors
recent_checks = Monitoring.list_devices_checks(device_id, 50)
processors = load_processors(snmp_device)
storage = load_storage(snmp_device)
latency_chart_data = load_latency_chart_data(device_id)
processor_chart_data = load_processor_chart_data(processors)
memory_chart_data = load_sensor_chart_data(device_id, ["memory_usage"])
storage_chart_data = load_sensor_chart_data(device_id, ["disk_usage"])
storage_volume_chart_data = load_storage_volume_chart_data(storage)
traffic_chart_data = load_overall_traffic_chart_data(device_id)
wireless_chart_data =
load_sensor_chart_data(device_id, [
"frequency",
"power",
"rssi",
"ccq",
"clients",
"distance",
"noise-floor",
"quality",
"rate",
"utilization"
])
# Classify sensors by type
{transceiver_sensors, non_transceiver_sensors} =
Enum.split_with(sensors, fn sensor ->
sensor.sensor_descr && String.contains?(String.downcase(sensor.sensor_descr), "sfp")
end)
temperature_sensors = Enum.filter(non_transceiver_sensors, &temperature_sensor?/1)
voltage_sensors = Enum.filter(non_transceiver_sensors, &voltage_sensor?/1)
current_sensors = Enum.filter(non_transceiver_sensors, &current_sensor?/1)
power_sensors = Enum.filter(non_transceiver_sensors, &power_sensor?/1)
fan_sensors = Enum.filter(non_transceiver_sensors, &fan_sensor?/1)
load_sensors = Enum.filter(non_transceiver_sensors, &load_sensor?/1)
storage_sensors = Enum.filter(non_transceiver_sensors, &(&1.sensor_type in ["disk_usage"]))
count_sensors = Enum.filter(non_transceiver_sensors, &counter_sensor?/1)
{firewall_counters, general_counters} =
Enum.split_with(count_sensors, &firewall_counter?/1)
wireless_sensors = Enum.filter(non_transceiver_sensors, &wireless_sensor?/1)
signal_sensors = Enum.filter(non_transceiver_sensors, &signal_sensor?/1)
grouped_transceivers = group_transceiver_sensors(transceiver_sensors)
metrics = calculate_metrics(recent_checks, device)
available_firmware = get_available_firmware(snmp_device)
socket
|> assign(:recent_checks, recent_checks)
|> assign(:metrics, metrics)
|> assign(:processors, processors)
|> assign(:storage, storage)
|> assign(:latency_chart_data, latency_chart_data)
|> assign(:processor_chart_data, processor_chart_data)
|> assign(:memory_chart_data, memory_chart_data)
|> assign(:storage_chart_data, storage_chart_data)
|> assign(:storage_volume_chart_data, storage_volume_chart_data)
|> assign(:traffic_chart_data, traffic_chart_data)
|> assign(:wireless_chart_data, wireless_chart_data)
|> assign(:temperature_sensors, temperature_sensors)
|> assign(:voltage_sensors, voltage_sensors)
|> assign(:current_sensors, current_sensors)
|> assign(:power_sensors, power_sensors)
|> assign(:fan_sensors, fan_sensors)
|> assign(:load_sensors, load_sensors)
|> assign(:storage_sensors, storage_sensors)
|> assign(:count_sensors, count_sensors)
|> assign(:general_counters, general_counters)
|> assign(:firewall_counters, firewall_counters)
|> assign(:wireless_sensors, wireless_sensors)
|> assign(:signal_sensors, signal_sensors)
|> assign(:grouped_transceivers, grouped_transceivers)
|> assign(:available_firmware, available_firmware)
end
# Ports tab: interfaces grouped by type.
# Uses snmp_interfaces already loaded in base data.
defp assign_ports_data(socket) do
interfaces = socket.assigns.snmp_interfaces
interfaces_by_type = group_interfaces_by_type(interfaces)
assign(socket, :interfaces_by_type, interfaces_by_type)
end
# Logs/events tab data.
defp assign_logs_data(socket, device_id) do
events = Devices.list_devices_events(device_id, 100)
assign(socket, :events, events)
end
# Backups tab data.
defp assign_backups_data(socket, device_id) do
device = socket.assigns.device
mikrotik_backups =
if device.mikrotik_enabled do
MikrotikBackups.list_device_backups(device_id, limit: 50)
else
[]
end
socket
|> assign(:mikrotik_backups, mikrotik_backups)
|> assign(:selected_backup_ids, MapSet.new())
end
defp get_available_firmware(nil), do: nil
defp get_available_firmware(snmp_device) do
vendor = determine_vendor(snmp_device.manufacturer)
product_line = determine_product_line(snmp_device.manufacturer)
# Only check firmware if both vendor and product_line are known
# Currently only MikroTik/RouterOS is supported
if vendor && product_line do
Firmware.get_latest_firmware_release(vendor, product_line)
end
end
defp determine_vendor(manufacturer) when is_binary(manufacturer) do
manufacturer_lower = String.downcase(manufacturer)
cond do
String.contains?(manufacturer_lower, "mikrotik") -> "mikrotik"
String.contains?(manufacturer_lower, "cisco") -> "cisco"
String.contains?(manufacturer_lower, "ubiquiti") -> "ubiquiti"
true -> nil
end
end
defp determine_vendor(_), do: nil
defp determine_product_line(manufacturer) when is_binary(manufacturer) do
manufacturer_lower = String.downcase(manufacturer)
if String.contains?(manufacturer_lower, "mikrotik") do
"routeros"
end
end
defp determine_product_line(_), do: nil
defp firmware_update_available?(nil, _), do: false
defp firmware_update_available?(_, nil), do: false
defp firmware_update_available?(snmp_device, available_firmware) do
current = snmp_device.firmware_version
available = available_firmware.version
# Extract clean version numbers for comparison
# "RouterOS 7.20.6 (Level 6)" -> "7.20.6"
# "7.21.2" -> "7.21.2"
current_clean = extract_version_number(current)
available_clean = extract_version_number(available)
current_clean && available_clean && VersionComparator.newer?(current_clean, available_clean)
end
# Extract semantic version number from version string
# Examples:
# "RouterOS 7.20.6 (Level 6)" -> "7.20.6"
# "7.21.2" -> "7.21.2"
# "v7.14.1" -> "7.14.1"
defp extract_version_number(nil), do: nil
defp extract_version_number(""), do: nil
defp extract_version_number(version_string) when is_binary(version_string) do
# Match semantic version pattern: X.Y or X.Y.Z
case Regex.run(~r/\d+\.\d+(?:\.\d+)?/, version_string) do
[version] -> version
_ -> nil
end
end
defp format_date(nil), do: ""
defp format_date(date) do
Calendar.strftime(date, "%B %-d, %Y")
end
defp load_agent_info(nil, :none) do
%{
agent_token_id: nil,
type: :cloud,
name: "Cloud Polling",
source: nil,
last_seen_at: nil,
is_offline: false
}
end
defp load_agent_info(agent_token_id, source) do
agent_token = Agents.get_agent_token!(agent_token_id)
is_offline = agent_offline?(agent_token)
%{
agent_token_id: agent_token_id,
type: if(agent_token.is_cloud_poller, do: :cloud, else: :agent),
name: agent_token.name,
source: source,
last_seen_at: agent_token.last_seen_at,
is_offline: is_offline
}
rescue
Ecto.NoResultsError ->
# Agent token no longer exists, fall back to cloud polling display
%{
agent_token_id: nil,
type: :cloud,
name: "Cloud Polling (agent not found)",
source: nil,
last_seen_at: nil,
is_offline: false
}
end
# Agent is considered offline if it hasn't been seen in the last 5 minutes
defp agent_offline?(%{is_cloud_poller: true}), do: false
defp agent_offline?(%{last_seen_at: nil}), do: true
defp agent_offline?(%{last_seen_at: last_seen_at}) do
five_minutes_ago = DateTime.add(DateTime.utc_now(), -5, :minute)
DateTime.before?(last_seen_at, five_minutes_ago)
end
defp calculate_metrics(checks, _equipment) do
total_checks = length(checks)
if total_checks > 0 do
successful_checks = Enum.count(checks, &(&1.status == :success))
uptime_percentage = Float.round(successful_checks / total_checks * 100, 1)
%{
uptime_percentage: uptime_percentage,
total_checks: total_checks,
successful_checks: successful_checks
}
else
%{
uptime_percentage: 0,
total_checks: 0,
successful_checks: 0
}
end
end
defp format_speed(speed_bps) when is_integer(speed_bps) do
cond do
speed_bps >= 1_000_000_000 ->
"#{Float.round(speed_bps / 1_000_000_000, 1)} Gbps"
speed_bps >= 1_000_000 ->
"#{Float.round(speed_bps / 1_000_000, 1)} Mbps"
speed_bps >= 1_000 ->
"#{Float.round(speed_bps / 1_000, 1)} Kbps"
true ->
"#{speed_bps} bps"
end
end
defp format_speed(_), do: "-"
defp format_bytes(nil), do: "-"
defp format_bytes(0), do: "0 B"
defp format_bytes(bytes) when is_integer(bytes) do
cond do
bytes >= 1_099_511_627_776 ->
"#{Float.round(bytes / 1_099_511_627_776, 1)} TB"
bytes >= 1_073_741_824 ->
"#{Float.round(bytes / 1_073_741_824, 1)} GB"
bytes >= 1_048_576 ->
"#{Float.round(bytes / 1_048_576, 1)} MB"
bytes >= 1024 ->
"#{Float.round(bytes / 1024, 1)} KB"
true ->
"#{bytes} B"
end
end
defp format_bytes(_), do: "-"
defp format_sensor_value(value, _divisor) when is_number(value) do
# Value is already divided by divisor when stored in sensor_reading
Float.round(value, 1)
end
defp format_sensor_value(_, _), do: "-"
defp time_ago(nil), do: "Never"
defp time_ago(datetime) do
diff = DateTime.diff(DateTime.utc_now(), datetime, :second)
cond do
diff < 60 -> "#{diff}s ago"
diff < 3600 -> "#{div(diff, 60)}m ago"
diff < 86_400 -> "#{div(diff, 3600)}h ago"
true -> "#{div(diff, 86_400)}d ago"
end
end
defp format_uptime(nil), do: "N/A"
defp format_uptime(timeticks) when is_integer(timeticks) do
# Timeticks are in hundredths of a second
seconds = div(timeticks, 100)
format_duration(seconds)
end
defp format_uptime(_), do: "N/A"
defp format_duration(total_seconds) do
{weeks, remainder} = seconds_to_weeks(total_seconds)
{days, remainder} = seconds_to_days(remainder)
{hours, remainder} = seconds_to_hours(remainder)
{minutes, seconds} = seconds_to_minutes(remainder)
parts =
Enum.reject(
[
format_time_part(weeks, "week"),
format_time_part(days, "day"),
format_time_part(hours, "hour"),
format_time_part(minutes, "minute"),
format_time_part(seconds, "second")
],
&is_nil/1
)
case parts do
[] -> "0 seconds"
parts -> Enum.join(parts, " ")
end
end
@spec seconds_to_weeks(integer()) :: {integer(), integer()}
defp seconds_to_weeks(seconds), do: {div(seconds, 604_800), rem(seconds, 604_800)}
@spec seconds_to_days(integer()) :: {integer(), integer()}
defp seconds_to_days(seconds), do: {div(seconds, 86_400), rem(seconds, 86_400)}
@spec seconds_to_hours(integer()) :: {integer(), integer()}
defp seconds_to_hours(seconds), do: {div(seconds, 3600), rem(seconds, 3600)}
@spec seconds_to_minutes(integer()) :: {integer(), integer()}
defp seconds_to_minutes(seconds), do: {div(seconds, 60), rem(seconds, 60)}
@spec format_time_part(integer(), String.t()) :: String.t() | nil
defp format_time_part(0, _unit), do: nil
defp format_time_part(1, unit), do: "1 #{unit}"
defp format_time_part(count, unit), do: "#{count} #{unit}s"
defp format_device_age(nil), do: "N/A"
defp format_device_age(datetime) do
diff = DateTime.diff(DateTime.utc_now(), datetime, :second)
format_duration(diff) <> " ago"
end
defp format_event_type(event_type) do
event_type
|> String.replace("_", " ")
|> String.capitalize()
end
defp load_snmp_data(device_id) do
case Snmp.get_device_with_associations(device_id) do
nil ->
%{device: nil, interfaces: [], sensors: []}
device ->
# Batch load latest readings for all sensors in a single query
sensor_ids = Enum.map(device.sensors, & &1.id)
latest_readings_map = Snmp.get_latest_sensor_readings_batch(sensor_ids)
sensors_with_readings =
Enum.map(device.sensors, fn sensor ->
latest_reading = Map.get(latest_readings_map, sensor.id)
Map.put(sensor, :latest_reading, latest_reading)
end)
# Batch load latest stats for all interfaces in a single query
interface_ids = Enum.map(device.interfaces, & &1.id)
latest_stats_map = Snmp.get_latest_interface_stats_batch(interface_ids)
interfaces_with_stats =
Enum.map(device.interfaces, fn interface ->
latest_stat = Map.get(latest_stats_map, interface.id)
Map.put(interface, :latest_stat, latest_stat)
end)
%{
device: device,
interfaces: interfaces_with_stats,
sensors: sensors_with_readings
}
end
end
defp load_vlans(nil), do: []
defp load_vlans(snmp_device), do: Snmp.list_vlans(snmp_device.id)
defp load_ip_addresses(nil), do: []
defp load_ip_addresses(snmp_device), do: Snmp.list_ip_addresses(snmp_device.id)
defp load_processors(nil), do: []
defp load_processors(snmp_device), do: Snmp.list_processors(snmp_device.id)
defp load_storage(nil), do: []
defp load_storage(snmp_device), do: Snmp.list_storage(snmp_device.id)
defp load_processor_chart_data([]), do: nil
defp load_processor_chart_data(processors) do
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
datasets =
processors
|> Enum.map(&processor_to_chart_dataset(&1, twenty_four_hours_ago))
|> Enum.reject(&Enum.empty?(&1.data))
if Enum.empty?(datasets) do
nil
else
Jason.encode!(%{datasets: datasets})
end
end
defp processor_to_chart_dataset(processor, since) do
readings =
processor.id
|> Snmp.get_processor_readings(since: since, limit: 1000)
|> Enum.reverse()
label =
if processor.description do
processor.description
else
"CPU #{processor.processor_index}"
end
%{
label: label,
data: Enum.map(readings, &processor_reading_to_data_point/1)
}
end
defp processor_reading_to_data_point(reading) do
%{
x: DateTime.to_unix(reading.checked_at, :millisecond),
y: if(reading.load_percent, do: Float.round(reading.load_percent, 1))
}
end
defp load_storage_volume_chart_data([]), do: nil
defp load_storage_volume_chart_data(storage_volumes) do
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
datasets =
storage_volumes
|> Enum.map(&storage_volume_to_chart_dataset(&1, twenty_four_hours_ago))
|> Enum.reject(&Enum.empty?(&1.data))
if Enum.empty?(datasets) do
nil
else
Jason.encode!(%{datasets: datasets})
end
end
defp storage_volume_to_chart_dataset(storage, since) do
readings =
storage.id
|> Snmp.get_storage_readings(since: since, limit: 1000)
|> Enum.reverse()
label = storage.description || "Storage #{storage.storage_index}"
%{
label: label,
data: Enum.map(readings, &storage_reading_to_data_point/1)
}
end
defp storage_reading_to_data_point(reading) do
%{
x: DateTime.to_unix(reading.checked_at, :millisecond),
y: if(reading.usage_percent, do: Float.round(reading.usage_percent, 1))
}
end
defp load_sensor_chart_data(device_id, sensor_types) when is_list(sensor_types) do
case Snmp.get_device_with_associations(device_id) do
nil -> nil
device -> build_sensor_datasets_json(device, sensor_types)
end
end
defp build_sensor_datasets_json(device, sensor_types) do
sensors =
device.sensors
|> Enum.filter(&(&1.sensor_type in sensor_types))
|> Enum.sort_by(& &1.sensor_index)
if Enum.empty?(sensors) do
nil
else
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
datasets = Enum.map(sensors, &sensor_to_chart_dataset(&1, twenty_four_hours_ago))
Jason.encode!(%{datasets: datasets})
end
end
defp sensor_to_chart_dataset(sensor, since) do
readings =
sensor.id
|> Snmp.get_sensor_readings(since: since, limit: 1000)
|> Enum.reverse()
%{
label: sensor.sensor_descr,
data: Enum.map(readings, &reading_to_data_point/1)
}
end
defp reading_to_data_point(reading) do
%{
x: DateTime.to_unix(reading.checked_at, :millisecond),
y: if(reading.value, do: Float.round(reading.value, 1))
}
end
defp load_latency_chart_data(device_id) do
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
checks =
device_id
|> Monitoring.get_latency_data(since: twenty_four_hours_ago, limit: 1000)
|> Enum.reverse()
if Enum.empty?(checks) do
nil
else
dataset = %{
label: "ICMP Latency",
data: Enum.map(checks, &latency_check_to_data_point/1)
}
Jason.encode!(%{datasets: [dataset]})
end
end
defp latency_check_to_data_point(check) do
%{
x: DateTime.to_unix(check.checked_at, :millisecond),
y: check.response_time_ms
}
end
defp load_overall_traffic_chart_data(device_id) do
case Snmp.get_device_with_associations(device_id) do
nil -> nil
device -> build_overall_traffic_chart_json(device)
end
end
defp build_overall_traffic_chart_json(device) do
if Enum.empty?(device.interfaces) do
nil
else
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
all_stats = aggregate_interface_traffic_stats(device.interfaces, twenty_four_hours_ago)
if Enum.empty?(all_stats) do
nil
else
build_traffic_json_datasets(all_stats)
end
end
end
# Calculate BPS per interface first, then aggregate by timestamp
# This avoids issues with misaligned timestamps between interfaces
defp aggregate_interface_traffic_stats(interfaces, since) do
interfaces
|> Enum.flat_map(fn interface ->
interface.id
|> Snmp.get_interface_stats(since: since, limit: 1000)
|> Enum.reverse()
|> calculate_interface_bps()
end)
|> Enum.group_by(& &1.timestamp_ms)
|> Enum.map(fn {ts, bps_list} ->
total_in = Enum.reduce(bps_list, 0.0, fn x, acc -> acc + x.in_bps end)
total_out = Enum.reduce(bps_list, 0.0, fn x, acc -> acc + x.out_bps end)
%{timestamp_ms: ts, in_bps: total_in, out_bps: total_out}
end)
|> Enum.sort_by(& &1.timestamp_ms)
end
# Calculate BPS for a single interface's stats
defp calculate_interface_bps(stats) do
stats
|> Enum.chunk_every(2, 1, :discard)
|> Enum.map(fn [s1, s2] ->
time_diff = s2.checked_at |> DateTime.diff(s1.checked_at, :second) |> max(1)
in_bps =
if s2.if_in_octets && s1.if_in_octets do
max((s2.if_in_octets - s1.if_in_octets) * 8 / time_diff, 0)
else
0.0
end
out_bps =
if s2.if_out_octets && s1.if_out_octets do
max((s2.if_out_octets - s1.if_out_octets) * 8 / time_diff, 0)
else
0.0
end
# Round timestamp to nearest minute for aggregation (avoids microsecond misalignment)
ts_seconds = DateTime.to_unix(s2.checked_at)
rounded_ts_ms = div(ts_seconds, 60) * 60 * 1000
%{timestamp_ms: rounded_ts_ms, in_bps: in_bps, out_bps: out_bps}
end)
end
defp build_traffic_json_datasets(all_stats) do
in_data = Enum.map(all_stats, fn s -> %{x: s.timestamp_ms, y: Float.round(s.in_bps, 2)} end)
out_data = Enum.map(all_stats, fn s -> %{x: s.timestamp_ms, y: -Float.round(s.out_bps, 2)} end)
datasets = [
%{label: "Outbound", data: out_data},
%{label: "Inbound", data: in_data}
]
Jason.encode!(%{datasets: datasets})
end
# Group interfaces by type for organized display
defp group_interfaces_by_type(interfaces) do
interfaces
|> Enum.group_by(&get_interface_type_category(&1.if_type))
|> Enum.map(fn {category, interfaces} ->
{category, Enum.sort_by(interfaces, & &1.if_index)}
end)
|> Enum.sort_by(fn {category, _} -> interface_type_order(category) end)
end
# Map SNMP interface types to human-readable categories
defp get_interface_type_category(if_type) when is_integer(if_type) do
Map.get(@interface_type_categories, if_type, "Other")
end
defp get_interface_type_category(_), do: "Other"
# Define display order for interface categories
defp interface_type_order(category) do
Map.get(@interface_category_order, category, 99)
end
# Group transceiver sensors by transceiver name
# Returns list of {name, sensors} tuples sorted by name
defp group_transceiver_sensors(transceiver_sensors) do
transceiver_sensors
|> Enum.group_by(&extract_transceiver_name/1)
|> Enum.sort_by(fn {name, _sensors} -> name end)
end
# Extract transceiver name from sensor description
# Examples:
# "sfp-sfpplus1 Rx Power" -> "sfp-sfpplus1"
# "sfp-sfpplus2 Temperature" -> "sfp-sfpplus2"
# "SFP1 Voltage" -> "SFP1"
defp extract_transceiver_name(sensor) do
case sensor.sensor_descr do
nil ->
"unknown"
descr ->
# Extract the first word (transceiver identifier)
descr
|> String.split(" ")
|> List.first()
|> String.downcase()
end
end
# Group IP addresses by interface ID for efficient lookup in templates
defp group_ip_addresses_by_interface(ip_addresses) do
Enum.group_by(ip_addresses, & &1.snmp_interface_id)
end
# Sensor type classification helpers
# Trust sensor_type first (set during SNMP discovery based on OID)
# Only fall back to unit/description matching if sensor_type is missing/unknown
defp temperature_sensor?(sensor) do
# Primary: Check sensor_type (set during discovery based on OID)
type_match = sensor.sensor_type in ["temperature", "cpu_temperature", "celsius"]
# Fallback: Check unit for devices that don't set sensor_type correctly
unit_match = sensor.sensor_unit in ["°C", "C", "celsius"]
# Only use description as last resort if type and unit don't match
# and only for unambiguous temperature keywords
descr_match =
!type_match && !unit_match &&
sensor.sensor_descr &&
String.contains?(String.downcase(sensor.sensor_descr), "temperature") &&
!String.contains?(String.downcase(sensor.sensor_descr), "voltage")
(type_match || unit_match || descr_match) && !voltage_by_description?(sensor)
end
defp voltage_sensor?(sensor) do
type_match = sensor.sensor_type in ["voltage", "volts"]
unit_match = sensor.sensor_unit in ["V", "mV", "volts"]
descr_match = voltage_by_description?(sensor)
type_match || unit_match || descr_match
end
defp voltage_by_description?(sensor) do
sensor.sensor_descr &&
String.contains?(String.downcase(sensor.sensor_descr), "voltage")
end
defp wireless_sensor?(sensor) do
sensor.sensor_type in [
"frequency",
"power",
"rssi",
"ccq",
"clients",
"distance",
"noise-floor",
"quality",
"rate",
"utilization"
]
end
# Counter sensors include various count and gauge types
# (connections, sessions, clients, DHCP leases, etc.)
defp counter_sensor?(sensor) do
sensor.sensor_type in [
"count",
"pppoe_sessions",
"connections",
"clients",
"leases",
"sessions"
]
end
# Firewall counters are connection-related metrics
defp firewall_counter?(sensor) do
descr = String.downcase(sensor.sensor_descr || "")
String.contains?(descr, "connection")
end
# Current sensors (amperes)
defp current_sensor?(sensor) do
type_match = sensor.sensor_type in ["current", "amperes"]
unit_match = sensor.sensor_unit in ["A", "mA", "amperes"]
type_match || unit_match
end
# Power sensors (watts) - separate from wireless power
defp power_sensor?(sensor) do
type_match = sensor.sensor_type in ["power", "watts"]
unit_match = sensor.sensor_unit in ["W", "mW", "watts"]
# Exclude wireless power sensors (those are in wireless category)
(type_match || unit_match) && !wireless_sensor?(sensor)
end
# Fan speed sensors (RPM)
defp fan_sensor?(sensor) do
type_match = sensor.sensor_type in ["fanspeed", "rpm"]
unit_match = sensor.sensor_unit in ["RPM", "rpm"]
descr_match = sensor.sensor_descr && String.contains?(String.downcase(sensor.sensor_descr), "fan")
type_match || unit_match || descr_match
end
# System load sensors (CPU load, etc.)
defp load_sensor?(sensor) do
type_match = sensor.sensor_type in ["load", "cpu_load", "system_load"]
descr_match = sensor.sensor_descr && String.contains?(String.downcase(sensor.sensor_descr), "load")
type_match || descr_match
end
# Signal quality sensors (SNR, RSSI, RSRP, etc.)
defp signal_sensor?(sensor) do
sensor.sensor_type in [
"snr",
"rssi",
"rsrp",
"rsrq",
"sinr",
"ssr",
"mse",
"noise",
"noise-floor",
"dbm"
]
end
@impl true
def handle_event("download_backup", %{"id" => backup_id}, socket) do
backup = MikrotikBackups.get_backup!(backup_id)
organization = socket.assigns.current_scope.organization
# Verify the backup belongs to a device in the user's organization
case AccessControl.verify_device_access(backup.device_id, organization.id) do
{:ok, device} ->
config_text = MikrotikBackups.decompress_config(backup.config_compressed)
filename = "#{device.name}_backup_#{Calendar.strftime(backup.backed_up_at, "%Y%m%d_%H%M%S")}.rsc"
{:noreply, push_event(socket, "download", %{content: config_text, filename: filename, mime_type: "text/plain"})}
{:error, _} ->
{:noreply, put_flash(socket, :error, t_equipment("You don't have permission to access this backup"))}
end
end
@impl true
def handle_event("toggle_backup_selection", %{"id" => backup_id}, socket) do
selected = socket.assigns.selected_backup_ids
updated_selected =
if MapSet.member?(selected, backup_id) do
MapSet.delete(selected, backup_id)
else
# Limit to 2 selections max
if MapSet.size(selected) >= 2 do
selected
else
MapSet.put(selected, backup_id)
end
end
{:noreply, assign(socket, :selected_backup_ids, updated_selected)}
end
@impl true
def handle_event("compare_selected", _params, socket) do
device_id = socket.assigns.device.id
selected = MapSet.to_list(socket.assigns.selected_backup_ids)
case selected do
[backup_a, backup_b] ->
{:noreply,
socket
|> assign(:selected_backup_ids, MapSet.new())
|> push_navigate(to: ~p"/devices/#{device_id}/backups/compare?backup_a=#{backup_a}&backup_b=#{backup_b}")}
_ ->
{:noreply, put_flash(socket, :error, t_equipment("Please select exactly 2 backups to compare"))}
end
end
@impl true
def handle_event("clear_selection", _params, socket) do
{:noreply, assign(socket, :selected_backup_ids, MapSet.new())}
end
@impl true
def handle_event("delete_backup", %{"id" => backup_id}, socket) do
import ToweropsWeb.Permissions
if owner?(socket) do
do_delete_backup(backup_id, socket)
else
{:noreply, put_flash(socket, :error, t_equipment("Only organization owners can delete backups"))}
end
end
@impl true
def handle_event("backup_now", _params, socket) do
device = socket.assigns.device
agent_token_id = socket.assigns.agent_info.agent_token_id
cond do
is_nil(agent_token_id) ->
{:noreply,
put_flash(socket, :error, t_equipment("Device has no agent assigned. Assign an agent to create backups."))}
!device.mikrotik_enabled ->
{:noreply, put_flash(socket, :error, t_equipment("MikroTik API is not enabled for this device."))}
true ->
trigger_manual_backup(socket, device, agent_token_id)
end
end
defp trigger_manual_backup(socket, device, agent_token_id) do
alias Towerops.Agent.AgentJob
alias Towerops.Agent.MikrotikCommand
alias Towerops.Agent.MikrotikDevice
alias Towerops.Devices.BackupRequests
job_id = "backup:#{device.id}:#{DateTime.to_unix(DateTime.utc_now())}"
mikrotik_config = Devices.get_mikrotik_config(device)
# Generate unique filename for this backup (without .rsc extension, RouterOS adds it)
backup_filename = "towerops-backup-#{DateTime.to_unix(DateTime.utc_now())}"
# Build chunked read commands
# Max chunk size is 32KB (32768 bytes), we'll read 10 chunks to cover 320KB
chunk_size = 32_768
num_chunks = 10
read_commands =
for i <- 0..(num_chunks - 1) do
%MikrotikCommand{
command: "/file/read",
args: %{
"file" => "#{backup_filename}.rsc",
"offset" => to_string(i * chunk_size),
"chunk-size" => to_string(chunk_size)
}
}
end
job = %AgentJob{
job_id: job_id,
job_type: :MIKROTIK,
device_id: device.id,
mikrotik_device: %MikrotikDevice{
ip: device.ip_address,
username: mikrotik_config.username,
password: mikrotik_config.password,
port: mikrotik_config.port,
ssh_port: mikrotik_config.ssh_port,
use_ssl: mikrotik_config.use_ssl
},
mikrotik_commands:
[
# Step 1: Export config to file on router (compact format, no defaults)
%MikrotikCommand{
command: "/export",
args: %{"file" => backup_filename, "compact" => ""}
}
] ++
read_commands ++
[
# Final step: Delete the temporary file
%MikrotikCommand{
command: "/file/remove",
args: %{"numbers" => "#{backup_filename}.rsc"}
}
]
}
case BackupRequests.create_request(device.id, job_id, "manual") do
{:ok, _request} ->
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"agent:#{agent_token_id}:backup",
{:backup_requested, job}
)
{:noreply,
socket
|> put_flash(:info, t_equipment("Manual backup requested. Results will appear shortly."))
|> load_equipment_data(device.id)}
{:error, _reason} ->
{:noreply, put_flash(socket, :error, t_equipment("Failed to create backup request. Please try again."))}
end
end
# Private helper for deleting backups
defp do_delete_backup(backup_id, socket) do
backup = MikrotikBackups.get_backup!(backup_id)
organization = socket.assigns.current_scope.organization
# Verify the backup belongs to a device in the user's organization
case AccessControl.verify_device_access(backup.device_id, organization.id) do
{:ok, _device} ->
case MikrotikBackups.delete_backup(backup) do
{:ok, _} ->
{:noreply,
socket
|> put_flash(:info, t_equipment("Backup deleted successfully"))
|> load_equipment_data(socket.assigns.device.id)}
{:error, _changeset} ->
{:noreply, put_flash(socket, :error, t_equipment("Failed to delete backup"))}
end
{:error, _} ->
{:noreply, put_flash(socket, :error, t_equipment("You don't have permission to access this backup"))}
end
end
# Apply pagination to backups list if on the backups tab
defp apply_backups_pagination(socket, "backups", page) when is_map_key(socket.assigns, :mikrotik_backups) do
per_page = 20
all_backups = socket.assigns.mikrotik_backups
total_count = length(all_backups)
total_pages = ceil(total_count / per_page)
# Ensure page is within valid range
page = max(1, min(page, max(1, total_pages)))
# Slice backups for current page
offset = (page - 1) * per_page
backups_page = Enum.slice(all_backups, offset, per_page)
socket
|> assign(:mikrotik_backups, backups_page)
|> assign(:pagination, %{
page: page,
per_page: per_page,
total_count: total_count,
total_pages: total_pages
})
end
defp apply_backups_pagination(socket, _tab, _page), do: socket
end