Display latency values with single decimal precision (e.g., 114.1ms) for improved readability in graphs. This ensures consistency between the raw chart data and tooltip formatting.
911 lines
27 KiB
Elixir
911 lines
27 KiB
Elixir
defmodule ToweropsWeb.GraphLive.Show do
|
|
@moduledoc false
|
|
use ToweropsWeb, :live_view
|
|
|
|
alias Towerops.Agents
|
|
alias Towerops.Devices
|
|
alias Towerops.Monitoring
|
|
alias Towerops.Snmp
|
|
|
|
require Logger
|
|
|
|
# Maximum live data points (5 minutes at 1 second intervals)
|
|
@max_live_points 300
|
|
|
|
@impl true
|
|
def mount(_params, _session, socket) do
|
|
{:ok, socket}
|
|
end
|
|
|
|
@impl true
|
|
def handle_params(%{"id" => device_id, "sensor_type" => sensor_type} = params, _, socket) do
|
|
organization = socket.assigns.current_scope.organization
|
|
|
|
with {:ok, device} <- find_device(device_id),
|
|
:ok <- verify_device_access(device, organization) do
|
|
socket = initialize_graph_view(socket, device_id, sensor_type, params)
|
|
{:noreply, socket}
|
|
else
|
|
{:error, :not_found} ->
|
|
{:noreply,
|
|
socket
|
|
|> put_flash(:error, "Device not found")
|
|
|> push_navigate(to: ~p"/devices")}
|
|
|
|
{:error, :access_denied} ->
|
|
{:noreply,
|
|
socket
|
|
|> put_flash(:error, "You don't have access to this device")
|
|
|> push_navigate(to: ~p"/devices")}
|
|
end
|
|
end
|
|
|
|
defp find_device(device_id) do
|
|
case Devices.get_device(device_id) do
|
|
nil -> {:error, :not_found}
|
|
device -> {:ok, device}
|
|
end
|
|
end
|
|
|
|
defp verify_device_access(device, organization) do
|
|
if device.organization_id == organization.id do
|
|
:ok
|
|
else
|
|
{:error, :access_denied}
|
|
end
|
|
end
|
|
|
|
defp initialize_graph_view(socket, device_id, sensor_type, params) do
|
|
maybe_subscribe_to_device(socket, device_id)
|
|
|
|
range = Map.get(params, "range", "24h")
|
|
interface_id = Map.get(params, "interface_id")
|
|
sensor_id = Map.get(params, "sensor_id")
|
|
storage_id = Map.get(params, "storage_id")
|
|
|
|
# Check if device has agent assignment
|
|
has_agent = Agents.get_device_assignment(device_id) != nil
|
|
|
|
socket
|
|
|> assign(:device_id, device_id)
|
|
|> assign(:sensor_type, sensor_type)
|
|
|> assign(:range, range)
|
|
|> assign(:interface_id, interface_id)
|
|
|> assign(:sensor_id, sensor_id)
|
|
|> assign(:storage_id, storage_id)
|
|
|> assign(:is_live_mode, range == "live")
|
|
|> assign(:has_agent_assignment, has_agent)
|
|
|> maybe_start_live_polling()
|
|
|> load_graph_data()
|
|
end
|
|
|
|
defp maybe_subscribe_to_device(socket, device_id) do
|
|
_ =
|
|
if connected?(socket) do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device_id}")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
@impl true
|
|
def handle_event("change_range", %{"range" => range}, socket) do
|
|
params = build_graph_params(socket.assigns, range)
|
|
|
|
{:noreply,
|
|
push_patch(socket,
|
|
to: ~p"/devices/#{socket.assigns.device_id}/graph/#{socket.assigns.sensor_type}?#{params}"
|
|
)}
|
|
end
|
|
|
|
defp build_graph_params(assigns, range) do
|
|
base_params = %{"range" => range}
|
|
|
|
base_params =
|
|
if assigns.interface_id do
|
|
Map.put(base_params, "interface_id", assigns.interface_id)
|
|
else
|
|
base_params
|
|
end
|
|
|
|
base_params =
|
|
if assigns.sensor_id do
|
|
Map.put(base_params, "sensor_id", assigns.sensor_id)
|
|
else
|
|
base_params
|
|
end
|
|
|
|
if assigns.storage_id do
|
|
Map.put(base_params, "storage_id", assigns.storage_id)
|
|
else
|
|
base_params
|
|
end
|
|
end
|
|
|
|
# Private functions
|
|
|
|
defp load_graph_data(socket) do
|
|
if socket.assigns.is_live_mode do
|
|
load_live_mode_graph_data(socket)
|
|
else
|
|
# Historical mode - existing logic
|
|
device_id = socket.assigns.device_id
|
|
sensor_type = socket.assigns.sensor_type
|
|
range = socket.assigns.range
|
|
|
|
device = Devices.get_device!(device_id)
|
|
{chart_data, title_suffix} = load_chart_data_for_type(socket.assigns, range)
|
|
|
|
{title, unit, auto_scale} = get_chart_config(sensor_type)
|
|
chart_title = if title_suffix, do: "#{title} - #{title_suffix}", else: title
|
|
show_zero_line = sensor_type == "traffic"
|
|
|
|
# Calculate max/min values from chart data
|
|
{max_value, min_value} = calculate_chart_stats(chart_data)
|
|
|
|
socket
|
|
|> assign(:device, device)
|
|
|> assign(:page_title, "#{device.name} - #{chart_title}")
|
|
|> assign(:chart_title, chart_title)
|
|
|> assign(:chart_data, chart_data)
|
|
|> assign(:unit, unit)
|
|
|> assign(:auto_scale, auto_scale)
|
|
|> assign(:show_zero_line, show_zero_line)
|
|
|> assign(:max_value, max_value)
|
|
|> assign(:min_value, min_value)
|
|
end
|
|
end
|
|
|
|
defp load_live_mode_graph_data(socket) do
|
|
device_id = socket.assigns.device_id
|
|
sensor_type = socket.assigns.sensor_type
|
|
device = Devices.get_device!(device_id)
|
|
{title, unit, auto_scale} = get_chart_config(sensor_type)
|
|
|
|
title_suffix = get_title_suffix(socket.assigns)
|
|
chart_title = if title_suffix, do: "#{title} - #{title_suffix}", else: title
|
|
|
|
socket
|
|
|> assign(:device, device)
|
|
|> assign(:page_title, "#{device.name} - #{chart_title} (Live)")
|
|
|> assign(:chart_title, chart_title)
|
|
|> assign(:chart_data, Jason.encode!(%{datasets: []}))
|
|
|> assign(:unit, unit)
|
|
|> assign(:auto_scale, auto_scale)
|
|
|> assign(:show_zero_line, sensor_type == "traffic")
|
|
|> assign(:max_value, nil)
|
|
|> assign(:min_value, nil)
|
|
end
|
|
|
|
defp get_title_suffix(assigns) do
|
|
cond do
|
|
assigns[:sensor_id] -> get_sensor_name(assigns[:sensor_id])
|
|
assigns[:interface_id] && assigns[:sensor_type] == "traffic" -> get_interface_name(assigns[:interface_id])
|
|
assigns[:storage_id] && assigns[:sensor_type] == "storage_volume" -> get_storage_name(assigns[:storage_id])
|
|
true -> nil
|
|
end
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_type: "latency", device_id: device_id}, range) do
|
|
{load_latency_chart_data(device_id, range), nil}
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_type: "traffic", interface_id: interface_id}, range)
|
|
when not is_nil(interface_id) do
|
|
{load_interface_traffic_chart_data(interface_id, range), get_interface_name(interface_id)}
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_type: "traffic", device_id: device_id}, range) do
|
|
{load_traffic_chart_data(device_id, range), nil}
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_type: "storage_volume", storage_id: storage_id}, range)
|
|
when not is_nil(storage_id) do
|
|
{load_storage_chart_data(storage_id, range), get_storage_name(storage_id)}
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_id: sensor_id}, range) when not is_nil(sensor_id) do
|
|
{load_single_sensor_chart_data(sensor_id, range), get_sensor_name(sensor_id)}
|
|
end
|
|
|
|
defp load_chart_data_for_type(%{sensor_type: sensor_type, device_id: device_id}, range) do
|
|
sensor_types = get_sensor_types_for_chart(sensor_type)
|
|
{load_sensor_chart_data(device_id, sensor_types, range), nil}
|
|
end
|
|
|
|
defp get_sensor_types_for_chart("processors"), do: ["cpu_load"]
|
|
defp get_sensor_types_for_chart("memory"), do: ["memory_usage"]
|
|
defp get_sensor_types_for_chart("storage"), do: ["disk_usage"]
|
|
defp get_sensor_types_for_chart("temperature"), do: ["temperature", "cpu_temperature", "celsius"]
|
|
defp get_sensor_types_for_chart("voltage"), do: ["voltage", "volts"]
|
|
defp get_sensor_types_for_chart("traffic"), do: nil
|
|
# For unknown sensor types, use the type directly (supports count, pppoe_sessions, etc.)
|
|
defp get_sensor_types_for_chart(sensor_type), do: [sensor_type]
|
|
|
|
defp get_chart_config("latency"), do: {"ICMP Latency", "ms", true}
|
|
defp get_chart_config("processors"), do: {"Processor Usage", "%", false}
|
|
defp get_chart_config("memory"), do: {"Memory Usage", "%", false}
|
|
defp get_chart_config("storage"), do: {"Storage Usage", "%", false}
|
|
defp get_chart_config("storage_volume"), do: {"Storage Usage", "%", false}
|
|
defp get_chart_config("temperature"), do: {"Temperature", "°C", true}
|
|
defp get_chart_config("voltage"), do: {"Voltage", "V", true}
|
|
defp get_chart_config("traffic"), do: {"Overall Traffic", "bps", true}
|
|
defp get_chart_config("count"), do: {"Count", "", true}
|
|
defp get_chart_config("pppoe_sessions"), do: {"PPPoE Sessions", "", true}
|
|
defp get_chart_config("connections"), do: {"Connections", "", true}
|
|
defp get_chart_config("wireless"), do: {"Wireless Metrics", "", true}
|
|
# Humanize unknown sensor types for title
|
|
defp get_chart_config(sensor_type), do: {humanize_sensor_type(sensor_type), "", true}
|
|
|
|
defp humanize_sensor_type(sensor_type) do
|
|
sensor_type
|
|
|> String.replace("_", " ")
|
|
|> String.split()
|
|
|> Enum.map_join(" ", &String.capitalize/1)
|
|
end
|
|
|
|
defp load_sensor_chart_data(device_id, sensor_types, range) do
|
|
case Snmp.get_device_with_associations(device_id) do
|
|
nil ->
|
|
nil
|
|
|
|
device ->
|
|
build_sensor_chart_json(device, sensor_types, range)
|
|
end
|
|
end
|
|
|
|
defp load_single_sensor_chart_data(sensor_id, range) do
|
|
case Snmp.get_sensor(sensor_id) do
|
|
nil ->
|
|
nil
|
|
|
|
sensor ->
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
dataset = sensor_to_dataset(sensor, since, limit)
|
|
|
|
# Return nil if no data points (empty readings)
|
|
if Enum.empty?(dataset.data) do
|
|
nil
|
|
else
|
|
Jason.encode!(%{datasets: [dataset]})
|
|
end
|
|
end
|
|
end
|
|
|
|
defp get_sensor_name(sensor_id) do
|
|
case Snmp.get_sensor(sensor_id) do
|
|
nil -> nil
|
|
sensor -> sensor.sensor_descr
|
|
end
|
|
end
|
|
|
|
defp load_storage_chart_data(storage_id, range) do
|
|
case Snmp.get_storage(storage_id) do
|
|
nil ->
|
|
nil
|
|
|
|
storage ->
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
dataset = storage_to_dataset(storage, since, limit)
|
|
Jason.encode!(%{datasets: [dataset]})
|
|
end
|
|
end
|
|
|
|
defp get_storage_name(storage_id) do
|
|
case Snmp.get_storage(storage_id) do
|
|
nil -> nil
|
|
storage -> storage.description || "Storage #{storage.storage_index}"
|
|
end
|
|
end
|
|
|
|
defp storage_to_dataset(storage, since, limit) do
|
|
readings =
|
|
storage.id
|
|
|> Snmp.get_storage_readings(since: since, limit: limit)
|
|
|> Enum.reverse()
|
|
|
|
%{
|
|
label: storage.description || "Storage #{storage.storage_index}",
|
|
data: Enum.map(readings, &storage_reading_to_chart_point/1)
|
|
}
|
|
end
|
|
|
|
defp storage_reading_to_chart_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 build_sensor_chart_json(device, sensor_types, range) 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
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
datasets = Enum.map(sensors, &sensor_to_dataset(&1, since, limit))
|
|
Jason.encode!(%{datasets: datasets})
|
|
end
|
|
end
|
|
|
|
defp sensor_to_dataset(sensor, since, limit) do
|
|
readings =
|
|
sensor.id
|
|
|> Snmp.get_sensor_readings(since: since, limit: limit)
|
|
|> Enum.reverse()
|
|
|
|
%{
|
|
label: sensor.sensor_descr,
|
|
data: Enum.map(readings, &reading_to_chart_point/1)
|
|
}
|
|
end
|
|
|
|
defp reading_to_chart_point(reading) do
|
|
%{
|
|
x: DateTime.to_unix(reading.checked_at, :millisecond),
|
|
y: if(reading.value, do: Float.round(reading.value, 1))
|
|
}
|
|
end
|
|
|
|
defp get_datetime_from_range("1h"), do: DateTime.add(DateTime.utc_now(), -1, :hour)
|
|
defp get_datetime_from_range("6h"), do: DateTime.add(DateTime.utc_now(), -6, :hour)
|
|
defp get_datetime_from_range("12h"), do: DateTime.add(DateTime.utc_now(), -12, :hour)
|
|
defp get_datetime_from_range("24h"), do: DateTime.add(DateTime.utc_now(), -24, :hour)
|
|
defp get_datetime_from_range("7d"), do: DateTime.add(DateTime.utc_now(), -7, :day)
|
|
defp get_datetime_from_range("30d"), do: DateTime.add(DateTime.utc_now(), -30, :day)
|
|
defp get_datetime_from_range(_), do: DateTime.add(DateTime.utc_now(), -24, :hour)
|
|
|
|
defp get_limit_for_range("1h"), do: 360
|
|
defp get_limit_for_range("6h"), do: 720
|
|
defp get_limit_for_range("12h"), do: 1440
|
|
defp get_limit_for_range("24h"), do: 2880
|
|
defp get_limit_for_range("7d"), do: 10_080
|
|
defp get_limit_for_range("30d"), do: 43_200
|
|
defp get_limit_for_range(_), do: 2880
|
|
|
|
defp load_traffic_chart_data(device_id, range) do
|
|
case Snmp.get_device_with_associations(device_id) do
|
|
nil -> nil
|
|
device -> build_traffic_chart_json(device, range)
|
|
end
|
|
end
|
|
|
|
defp load_interface_traffic_chart_data(interface_id, range) do
|
|
case Snmp.get_interface(interface_id) do
|
|
nil -> nil
|
|
interface -> build_interface_traffic_chart_json(interface, range)
|
|
end
|
|
end
|
|
|
|
defp get_interface_name(interface_id) do
|
|
case Snmp.get_interface(interface_id) do
|
|
nil -> "Unknown Interface"
|
|
interface -> interface.if_name || interface.if_descr || "Interface #{interface.if_index}"
|
|
end
|
|
end
|
|
|
|
defp load_latency_chart_data(device_id, range) do
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
|
|
checks =
|
|
device_id
|
|
|> Monitoring.get_latency_data(since: since, limit: limit)
|
|
|> Enum.reverse()
|
|
|
|
if Enum.empty?(checks) do
|
|
nil
|
|
else
|
|
dataset = %{
|
|
label: "ICMP Latency",
|
|
data: Enum.map(checks, &latency_check_to_chart_point/1)
|
|
}
|
|
|
|
Jason.encode!(%{datasets: [dataset]})
|
|
end
|
|
end
|
|
|
|
defp latency_check_to_chart_point(check) do
|
|
%{
|
|
x: DateTime.to_unix(check.checked_at, :millisecond),
|
|
y: Float.round(check.response_time_ms, 1)
|
|
}
|
|
end
|
|
|
|
# Calculate max and min values from chart data
|
|
defp calculate_chart_stats(nil), do: {nil, nil}
|
|
|
|
defp calculate_chart_stats(chart_data) when is_binary(chart_data) do
|
|
case Jason.decode(chart_data) do
|
|
{:ok, %{"datasets" => datasets}} ->
|
|
extract_min_max_from_datasets(datasets)
|
|
|
|
_ ->
|
|
{nil, nil}
|
|
end
|
|
end
|
|
|
|
defp extract_min_max_from_datasets(datasets) do
|
|
values = Enum.flat_map(datasets, &extract_values_from_dataset/1)
|
|
|
|
case values do
|
|
[] -> {nil, nil}
|
|
_ -> {Enum.max(values), Enum.min(values)}
|
|
end
|
|
end
|
|
|
|
defp extract_values_from_dataset(%{"data" => data}) when is_list(data) do
|
|
Enum.flat_map(data, &extract_y_value/1)
|
|
end
|
|
|
|
defp extract_values_from_dataset(_), do: []
|
|
|
|
defp extract_y_value(%{"y" => y}) when is_number(y), do: [y]
|
|
defp extract_y_value(_), do: []
|
|
|
|
defp build_traffic_chart_json(device, range) do
|
|
if Enum.empty?(device.interfaces), do: nil, else: build_traffic_datasets(device, range)
|
|
end
|
|
|
|
defp build_traffic_datasets(device, range) do
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
active_interfaces = Enum.filter(device.interfaces, &(&1.if_oper_status == "up"))
|
|
|
|
if Enum.empty?(active_interfaces), do: nil, else: encode_interface_datasets(active_interfaces, since, limit)
|
|
end
|
|
|
|
defp encode_interface_datasets(interfaces, since, limit) do
|
|
datasets = build_per_interface_datasets(interfaces, since, limit)
|
|
if Enum.empty?(datasets), do: nil, else: Jason.encode!(%{datasets: datasets})
|
|
end
|
|
|
|
defp build_per_interface_datasets(interfaces, since, limit) do
|
|
Enum.flat_map(interfaces, fn interface ->
|
|
stats =
|
|
interface.id
|
|
|> Snmp.get_interface_stats(since: since, limit: limit)
|
|
|> Enum.reverse()
|
|
|
|
if Enum.empty?(stats) do
|
|
[]
|
|
else
|
|
interface_name = interface.if_name || interface.if_descr
|
|
in_data = calculate_interface_traffic_data(stats, :inbound)
|
|
out_data = calculate_interface_traffic_data(stats, :outbound)
|
|
|
|
[
|
|
%{label: "#{interface_name} Out", data: out_data},
|
|
%{label: "#{interface_name} In", data: in_data}
|
|
]
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp build_interface_traffic_chart_json(interface, range) do
|
|
since = get_datetime_from_range(range)
|
|
limit = get_limit_for_range(range)
|
|
|
|
stats =
|
|
interface.id
|
|
|> Snmp.get_interface_stats(since: since, limit: limit)
|
|
|> Enum.reverse()
|
|
|
|
if Enum.empty?(stats) do
|
|
nil
|
|
else
|
|
in_data = calculate_interface_traffic_data(stats, :inbound)
|
|
out_data = calculate_interface_traffic_data(stats, :outbound)
|
|
|
|
datasets = [
|
|
%{label: "Outbound", data: out_data},
|
|
%{label: "Inbound", data: in_data}
|
|
]
|
|
|
|
Jason.encode!(%{datasets: datasets})
|
|
end
|
|
end
|
|
|
|
defp calculate_interface_traffic_data(stats, :inbound) do
|
|
stats
|
|
|> Enum.chunk_every(2, 1, :discard)
|
|
|> Enum.map(fn [stat1, stat2] ->
|
|
# Skip if octets are nil
|
|
if is_nil(stat1.if_in_octets) or is_nil(stat2.if_in_octets) do
|
|
nil
|
|
else
|
|
time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1)
|
|
|
|
bps =
|
|
((stat2.if_in_octets - stat1.if_in_octets) * 8 / time_diff)
|
|
|> max(0)
|
|
|> :erlang.float()
|
|
|> Float.round(2)
|
|
|
|
%{
|
|
x: DateTime.to_unix(stat2.checked_at, :millisecond),
|
|
y: -bps
|
|
}
|
|
end
|
|
end)
|
|
|> Enum.reject(&is_nil/1)
|
|
end
|
|
|
|
defp calculate_interface_traffic_data(stats, :outbound) do
|
|
stats
|
|
|> Enum.chunk_every(2, 1, :discard)
|
|
|> Enum.map(fn [stat1, stat2] ->
|
|
# Skip if octets are nil
|
|
if is_nil(stat1.if_out_octets) or is_nil(stat2.if_out_octets) do
|
|
nil
|
|
else
|
|
time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1)
|
|
|
|
bps =
|
|
((stat2.if_out_octets - stat1.if_out_octets) * 8 / time_diff)
|
|
|> max(0)
|
|
|> :erlang.float()
|
|
|> Float.round(2)
|
|
|
|
%{
|
|
x: DateTime.to_unix(stat2.checked_at, :millisecond),
|
|
y: bps
|
|
}
|
|
end
|
|
end)
|
|
|> Enum.reject(&is_nil/1)
|
|
end
|
|
|
|
# Format value with unit for display
|
|
defp format_value(value, "bps") do
|
|
abs_value = abs(value)
|
|
|
|
cond do
|
|
abs_value >= 1_000_000_000 ->
|
|
"#{Float.round(abs_value / 1_000_000_000, 2)} Gbps"
|
|
|
|
abs_value >= 1_000_000 ->
|
|
"#{Float.round(abs_value / 1_000_000, 2)} Mbps"
|
|
|
|
abs_value >= 1_000 ->
|
|
"#{Float.round(abs_value / 1_000, 2)} Kbps"
|
|
|
|
true ->
|
|
"#{Float.round(abs_value, 2)} bps"
|
|
end
|
|
end
|
|
|
|
# For counts (empty unit), format as integer
|
|
defp format_value(value, "") do
|
|
"#{round(value)}"
|
|
end
|
|
|
|
# For other units, format with one decimal place
|
|
defp format_value(value, unit) do
|
|
"#{Float.round(value, 1)}#{unit}"
|
|
end
|
|
|
|
# Live polling functions
|
|
|
|
# Start/stop live polling based on mode
|
|
defp maybe_start_live_polling(socket) do
|
|
if socket.assigns[:is_live_mode] && connected?(socket) do
|
|
cancel_live_polling_timer(socket)
|
|
|
|
socket
|
|
|> assign(:live_data_buffer, %{})
|
|
|> assign(:live_data_points, 0)
|
|
|> schedule_live_poll()
|
|
else
|
|
cancel_live_polling_timer(socket)
|
|
end
|
|
end
|
|
|
|
# Schedule next poll in 1 second
|
|
defp schedule_live_poll(socket) do
|
|
timer_ref = Process.send_after(self(), :live_poll, 1_000)
|
|
assign(socket, :live_poll_timer, timer_ref)
|
|
end
|
|
|
|
# Cancel existing timer
|
|
defp cancel_live_polling_timer(socket) do
|
|
case Map.get(socket.assigns, :live_poll_timer) do
|
|
nil ->
|
|
socket
|
|
|
|
timer_ref ->
|
|
Process.cancel_timer(timer_ref)
|
|
assign(socket, :live_poll_timer, nil)
|
|
end
|
|
end
|
|
|
|
# Handle device status changes from PubSub
|
|
@impl true
|
|
def handle_info({:device_status_changed, _device_id, _new_status, _response_time}, socket) do
|
|
# Reload device to get updated status
|
|
device = Devices.get_device(socket.assigns.device_id)
|
|
{:noreply, assign(socket, :device, device)}
|
|
end
|
|
|
|
# Handle sensor updates from PubSub
|
|
@impl true
|
|
def handle_info({:state_sensors_updated, _device_id}, socket) do
|
|
{:noreply, load_graph_data(socket)}
|
|
end
|
|
|
|
# Handle neighbor updates from PubSub (ignored - not relevant to graphs)
|
|
@impl true
|
|
def handle_info({:neighbors_updated, _device_id}, socket) do
|
|
{:noreply, socket}
|
|
end
|
|
|
|
# Handle live poll event
|
|
@impl true
|
|
def handle_info(:live_poll, socket) do
|
|
if socket.assigns[:is_live_mode] do
|
|
socket = perform_live_poll(socket)
|
|
{:noreply, schedule_live_poll(socket)}
|
|
else
|
|
{:noreply, socket}
|
|
end
|
|
end
|
|
|
|
# Perform SNMP polling and update buffer
|
|
defp perform_live_poll(socket) do
|
|
device_id = socket.assigns.device_id
|
|
|
|
# Verify device still exists and access is valid
|
|
case Devices.get_device(device_id) do
|
|
nil ->
|
|
socket
|
|
|> put_flash(:error, "Device no longer exists")
|
|
|> push_navigate(to: ~p"/devices")
|
|
|
|
device ->
|
|
organization = socket.assigns.current_scope.organization
|
|
|
|
if device.organization_id == organization.id do
|
|
poll_and_update_buffer(socket, device)
|
|
else
|
|
socket
|
|
|> put_flash(:error, "Access to device revoked")
|
|
|> push_navigate(to: ~p"/devices")
|
|
end
|
|
end
|
|
end
|
|
|
|
# Poll sensors and update buffer
|
|
defp poll_and_update_buffer(socket, device) do
|
|
now = DateTime.utc_now()
|
|
timestamp_ms = DateTime.to_unix(now, :millisecond)
|
|
|
|
# Check if device has agent assignment (fresh check each poll to avoid stale data)
|
|
has_agent = Agents.get_device_assignment(device.id) != nil
|
|
|
|
Logger.info("Live poll for device #{device.name}: has_agent=#{has_agent}")
|
|
|
|
new_points =
|
|
if has_agent do
|
|
# Request agent to poll NOW and wait for result
|
|
Logger.info("Requesting live poll from agent")
|
|
request_agent_live_poll_and_wait(device.id, socket.assigns)
|
|
else
|
|
# Do direct SNMP polling for specific sensors only
|
|
Logger.info("Using direct SNMP for live polling")
|
|
client_opts = build_snmp_client_opts(device)
|
|
poll_sensors_for_live_mode(socket.assigns, client_opts, timestamp_ms)
|
|
end
|
|
|
|
Logger.info("Live poll fetched #{length(new_points)} data points: #{inspect(new_points)}")
|
|
|
|
# Update buffer with new points (rolling window)
|
|
{updated_buffer, point_count} =
|
|
update_live_data_buffer(
|
|
socket.assigns.live_data_buffer,
|
|
socket.assigns.live_data_points,
|
|
new_points
|
|
)
|
|
|
|
# Push update to client
|
|
socket
|
|
|> assign(:live_data_buffer, updated_buffer)
|
|
|> assign(:live_data_points, point_count)
|
|
|> push_event("live_data_update", %{
|
|
timestamp: timestamp_ms,
|
|
points: new_points
|
|
})
|
|
end
|
|
|
|
# Request live poll from agent and wait for response
|
|
defp request_agent_live_poll_and_wait(device_id, assigns) do
|
|
case Agents.get_device_assignment(device_id) do
|
|
nil ->
|
|
[]
|
|
|
|
assignment ->
|
|
# Generate unique reply topic for this request
|
|
reply_topic = "live_poll_reply:#{:erlang.unique_integer([:positive])}"
|
|
|
|
# Subscribe to reply topic
|
|
:ok = Phoenix.PubSub.subscribe(Towerops.PubSub, reply_topic)
|
|
|
|
# Get sensors we're interested in
|
|
sensors = get_sensors_for_live_mode(assigns)
|
|
sensor_oids = Enum.map(sensors, & &1.sensor_oid)
|
|
|
|
# Broadcast live poll request with reply topic and sensor OIDs
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{assignment.agent_token_id}:live_poll",
|
|
{:live_poll_requested, device_id, sensor_oids, reply_topic}
|
|
)
|
|
|
|
# Wait for response with timeout
|
|
receive do
|
|
{:live_poll_result, oid_values} ->
|
|
# Process the OID values into data points
|
|
process_live_poll_oids(sensors, oid_values)
|
|
after
|
|
3000 ->
|
|
Logger.warning("Live poll timeout waiting for agent response")
|
|
[]
|
|
end
|
|
end
|
|
end
|
|
|
|
# Convert OID values from agent into data points
|
|
defp process_live_poll_oids(sensors, oid_values) do
|
|
timestamp_ms = DateTime.to_unix(DateTime.utc_now(), :millisecond)
|
|
|
|
sensors
|
|
|> Enum.map(&build_live_poll_data_point(&1, oid_values, timestamp_ms))
|
|
|> Enum.reject(&is_nil/1)
|
|
end
|
|
|
|
defp build_live_poll_data_point(sensor, oid_values, timestamp_ms) do
|
|
with value when not is_nil(value) <- Map.get(oid_values, sensor.sensor_oid),
|
|
parsed_value when not is_nil(parsed_value) <- parse_sensor_value(value) do
|
|
final_value = parsed_value / sensor.sensor_divisor
|
|
|
|
%{
|
|
sensor_id: sensor.id,
|
|
label: sensor.sensor_descr,
|
|
value: Float.round(final_value, 1),
|
|
timestamp: timestamp_ms
|
|
}
|
|
else
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
defp parse_sensor_value(value) when is_float(value), do: value
|
|
defp parse_sensor_value(value) when is_integer(value), do: value / 1.0
|
|
|
|
defp parse_sensor_value(value) when is_binary(value) do
|
|
case Float.parse(value) do
|
|
{float, _} -> float
|
|
:error -> nil
|
|
end
|
|
end
|
|
|
|
defp parse_sensor_value(_), do: nil
|
|
|
|
# Build SNMP client options
|
|
defp build_snmp_client_opts(device) do
|
|
snmp_config = Devices.get_snmp_config(device)
|
|
|
|
[
|
|
ip: device.ip_address,
|
|
community: snmp_config.community,
|
|
version: snmp_config.version,
|
|
port: device.snmp_port || 161,
|
|
# Increased for live polling - devices may be slow to respond
|
|
timeout: 3000
|
|
]
|
|
end
|
|
|
|
# Poll sensors in parallel
|
|
defp poll_sensors_for_live_mode(assigns, client_opts, timestamp_ms) do
|
|
sensors = get_sensors_for_live_mode(assigns)
|
|
|
|
sensors
|
|
|> Task.async_stream(
|
|
fn sensor ->
|
|
case poll_single_sensor(sensor, client_opts) do
|
|
{:ok, value} ->
|
|
%{
|
|
sensor_id: sensor.id,
|
|
label: sensor.sensor_descr,
|
|
value: Float.round(value, 1),
|
|
timestamp: timestamp_ms
|
|
}
|
|
|
|
{:error, _reason} ->
|
|
# Skip failed polls - chart will show gap
|
|
nil
|
|
end
|
|
end,
|
|
max_concurrency: 10,
|
|
# Give SNMP client (3000ms timeout) enough time plus overhead
|
|
timeout: 4000,
|
|
on_timeout: :kill_task
|
|
)
|
|
|> Enum.reduce([], fn
|
|
{:ok, nil}, acc -> acc
|
|
{:ok, point}, acc -> [point | acc]
|
|
{:exit, _reason}, acc -> acc
|
|
end)
|
|
end
|
|
|
|
# Get sensors based on graph type
|
|
defp get_sensors_for_live_mode(%{sensor_id: sensor_id}) when not is_nil(sensor_id) do
|
|
case Snmp.get_sensor(sensor_id) do
|
|
nil -> []
|
|
sensor -> [sensor]
|
|
end
|
|
end
|
|
|
|
defp get_sensors_for_live_mode(%{sensor_type: sensor_type, device_id: device_id}) do
|
|
sensor_types = get_sensor_types_for_chart(sensor_type)
|
|
device = Snmp.get_device_with_associations(device_id)
|
|
|
|
device.sensors
|
|
|> Enum.filter(&(&1.sensor_type in sensor_types))
|
|
|> Enum.sort_by(& &1.sensor_index)
|
|
end
|
|
|
|
# Poll single sensor via SNMP
|
|
defp poll_single_sensor(sensor, client_opts) do
|
|
case Snmp.Client.get(client_opts, sensor.sensor_oid) do
|
|
{:ok, raw_value} when is_number(raw_value) ->
|
|
value = raw_value / sensor.sensor_divisor
|
|
{:ok, value}
|
|
|
|
{:ok, _non_numeric} ->
|
|
{:error, :non_numeric}
|
|
|
|
{:error, reason} ->
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
# Update live data buffer with rolling window
|
|
defp update_live_data_buffer(buffer, current_count, new_points) do
|
|
updated_buffer =
|
|
Enum.reduce(new_points, buffer, fn point, acc ->
|
|
sensor_buffer = Map.get(acc, point.sensor_id, [])
|
|
|
|
# Add new point at the front
|
|
new_buffer = [
|
|
%{x: point.timestamp, y: point.value} | sensor_buffer
|
|
]
|
|
|
|
# Trim to max size
|
|
trimmed_buffer =
|
|
if current_count >= @max_live_points do
|
|
Enum.take(new_buffer, @max_live_points)
|
|
else
|
|
new_buffer
|
|
end
|
|
|
|
Map.put(acc, point.sensor_id, trimmed_buffer)
|
|
end)
|
|
|
|
new_count = min(current_count + 1, @max_live_points)
|
|
{updated_buffer, new_count}
|
|
end
|
|
|
|
# Cleanup on LiveView termination
|
|
@impl true
|
|
def terminate(_reason, socket) do
|
|
cancel_live_polling_timer(socket)
|
|
:ok
|
|
end
|
|
end
|