graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

debug/traffic-graph-logging (#129)

Browse source 
Reviewed-on: graham/towerops-web#129
This commit is contained in:
Graham McIntire 2026-03-23 13:35:12 -05:00 • committed by graham
parent 7655f9d45a
commit 9a366c2665
2 changed files with 225 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
lib/towerops_web/live/device_live/show.ex
View file

@ -1211,10 +1211,9 @@ defmodule ToweropsWeb.DeviceLive.Show do
# Calculate BPS per interface first, then aggregate by timestamp
# This avoids issues with misaligned timestamps between interfaces
# Only processes monitored interfaces (nil treated as true for backwards compatibility)
# Processes all interfaces (regardless of monitored status) to show total device traffic
defp aggregate_interface_traffic_stats(interfaces, since) do
interfaces
|> Enum.filter(&(&1.monitored != false))
|> Enum.flat_map(fn interface ->
interface.id
|> Snmp.get_interface_stats(since: since, limit: 1000)
@ -1414,11 +1413,9 @@ defmodule ToweropsWeb.DeviceLive.Show do
# Sums capacity of active interfaces at each timestamp
defp calculate_interface_capacity_data(all_stats, device) do
# Build a map of interface_id -> {rx_capacity, tx_capacity} for quick lookup
# Only include monitored interfaces (nil treated as true for backwards compatibility)
# Includes all interfaces to match the total device traffic
interface_capacity_map =
device.interfaces
|> Enum.filter(&(&1.monitored != false))
|> Map.new(fn interface ->
Map.new(device.interfaces, fn interface ->
cap = interface.configured_capacity_bps || interface.if_speed || 0
{interface.id, {cap, cap}}
end)

254
lib/towerops_web/live/graph_live/show.ex
View file

@ -347,6 +347,7 @@ defmodule ToweropsWeb.GraphLive.Show do
|> assign(:show_zero_line, sensor_type == "traffic")
|> assign(:max_value, nil)
|> assign(:min_value, nil)
|> assign(:previous_interface_stats, nil)
end
defp get_title_suffix(assigns) do
@ -629,9 +630,9 @@ defmodule ToweropsWeb.GraphLive.Show do
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, device)
# Process all interfaces (regardless of operational status) to show total device traffic
if Enum.empty?(device.interfaces), do: nil, else: encode_interface_datasets(device.interfaces, since, limit, device)
end
# Get device for adding capacity lines to full-page traffic view
@ -675,10 +676,9 @@ defmodule ToweropsWeb.GraphLive.Show do
# Calculates capacity per timestamp based on which interfaces have data at that point
defp add_capacity_to_full_graph(datasets, interfaces, since, limit) do
# Build a map of interface_id -> capacity for quick lookup
# Includes all interfaces to match the total device traffic
interface_capacity_map =
interfaces
|> Enum.filter(& &1.monitored)
|> Map.new(fn interface ->
Map.new(interfaces, fn interface ->
capacity = interface.configured_capacity_bps || interface.if_speed || 0
{interface.id, capacity}
end)
@ -977,16 +977,24 @@ defmodule ToweropsWeb.GraphLive.Show do
Logger.info("Live poll for device #{device.name}: agent_token_id=#{inspect(agent_token_id)}")
new_points =
{new_points, updated_socket} =
if agent_token_id 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, agent_token_id, socket.assigns)
points = request_agent_live_poll_and_wait(device.id, agent_token_id, socket.assigns)
{points, socket}
else
# Do direct SNMP polling for specific sensors only
# Do direct SNMP polling
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)
# For traffic graphs, we need to store interface stats for next poll
if socket.assigns.sensor_type == "traffic" do
poll_traffic_with_state_update(socket, client_opts, timestamp_ms)
else
points = poll_sensors_for_live_mode(socket.assigns, client_opts, timestamp_ms)
{points, socket}
end
end
Logger.info("Live poll fetched #{length(new_points)} data points: #{inspect(new_points)}")
@ -994,13 +1002,13 @@ defmodule ToweropsWeb.GraphLive.Show do
# 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,
updated_socket.assigns.live_data_buffer,
updated_socket.assigns.live_data_points,
new_points
)
# Push update to client
socket
updated_socket
|> assign(:live_data_buffer, updated_buffer)
|> assign(:live_data_points, point_count)
|> push_event("live_data_update", %{
@ -1088,27 +1096,21 @@ defmodule ToweropsWeb.GraphLive.Show do
# Poll sensors in parallel
defp poll_sensors_for_live_mode(assigns, client_opts, timestamp_ms) do
# Special handling for traffic graphs
if assigns.sensor_type == "traffic" do
poll_traffic_for_live_mode(assigns, client_opts, timestamp_ms)
else
poll_regular_sensors_for_live_mode(assigns, client_opts, timestamp_ms)
end
end
# Poll regular sensors (non-traffic) in parallel
defp poll_regular_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,
|> Task.async_stream(&poll_sensor_to_data_point(&1, client_opts, timestamp_ms),
max_concurrency: 10,
# Give SNMP client (3000ms timeout) enough time plus overhead
timeout: 4000,
on_timeout: :kill_task
)
@ -1119,6 +1121,189 @@ defmodule ToweropsWeb.GraphLive.Show do
end)
end
# Poll a single sensor and convert to data point
defp poll_sensor_to_data_point(sensor, client_opts, timestamp_ms) do
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} ->
nil
end
end
# Poll traffic stats and update socket with current stats for next poll
# Returns {data_points, updated_socket}
defp poll_traffic_with_state_update(socket, client_opts, timestamp_ms) do
device = Snmp.get_device_with_associations(socket.assigns.device_id)
if is_nil(device) || Enum.empty?(device.interfaces) do
{[], socket}
else
# Poll all interface stats
current_stats = poll_interface_stats_parallel(device.interfaces, client_opts)
# If we have previous stats, calculate BPS
data_points =
if socket.assigns.previous_interface_stats do
calculate_traffic_bps_from_stats(
socket.assigns.previous_interface_stats,
current_stats,
timestamp_ms
)
else
# First poll - no previous stats, so no BPS to calculate yet
[]
end
# Update socket with current stats for next poll
updated_socket = assign(socket, :previous_interface_stats, current_stats)
{data_points, updated_socket}
end
end
# Poll traffic stats for live mode (legacy - used for non-state-updating path)
# Requires two readings to calculate BPS, so uses previous_interface_stats from socket assigns
defp poll_traffic_for_live_mode(assigns, client_opts, timestamp_ms) do
device = Snmp.get_device_with_associations(assigns.device_id)
if is_nil(device) || Enum.empty?(device.interfaces) do
[]
else
# Poll all interface stats
current_stats = poll_interface_stats_parallel(device.interfaces, client_opts)
# If we have previous stats, calculate BPS
if assigns.previous_interface_stats do
calculate_traffic_bps_from_stats(
assigns.previous_interface_stats,
current_stats,
timestamp_ms
)
else
# First poll - no previous stats, so no BPS to calculate yet
[]
end
end
end
# Poll interface octets for all interfaces in parallel
# Uses 64-bit counters (ifHCInOctets/ifHCOutOctets) for better accuracy on high-speed interfaces
defp poll_interface_stats_parallel(interfaces, client_opts) do
interfaces
|> Task.async_stream(&poll_single_interface_octets(&1, client_opts),
max_concurrency: 10,
timeout: 4000,
on_timeout: :kill_task
)
|> Enum.reduce([], fn
{:ok, nil}, acc -> acc
{:ok, stat}, acc -> [stat | acc]
{:exit, _}, acc -> acc
end)
end
# Poll octets for a single interface (try 64-bit, fall back to 32-bit)
defp poll_single_interface_octets(interface, client_opts) do
# Try 64-bit counters first
in_octets_oid = "1.3.6.1.2.1.31.1.1.1.6.#{interface.if_index}"
out_octets_oid = "1.3.6.1.2.1.31.1.1.1.10.#{interface.if_index}"
with {:ok, in_octets} <- Snmp.Client.get(client_opts, in_octets_oid),
{:ok, out_octets} <- Snmp.Client.get(client_opts, out_octets_oid),
true <- is_number(in_octets) and is_number(out_octets) do
build_interface_stat(interface, in_octets, out_octets)
else
_ -> poll_single_interface_octets_32bit(interface, client_opts)
end
end
# Fall back to 32-bit counters
defp poll_single_interface_octets_32bit(interface, client_opts) do
in_octets_oid = "1.3.6.1.2.1.2.2.1.10.#{interface.if_index}"
out_octets_oid = "1.3.6.1.2.1.2.2.1.16.#{interface.if_index}"
with {:ok, in_octets} <- Snmp.Client.get(client_opts, in_octets_oid),
{:ok, out_octets} <- Snmp.Client.get(client_opts, out_octets_oid),
true <- is_number(in_octets) and is_number(out_octets) do
build_interface_stat(interface, in_octets, out_octets)
else
_ -> nil
end
end
# Build interface stat map
defp build_interface_stat(interface, in_octets, out_octets) do
%{
interface_id: interface.id,
interface_name: interface.if_name || interface.if_descr,
if_in_octets: in_octets,
if_out_octets: out_octets,
polled_at: DateTime.utc_now()
}
end
# Calculate BPS from two sets of interface stats
defp calculate_traffic_bps_from_stats(previous_stats, current_stats, timestamp_ms) do
# Build map of previous stats for quick lookup
previous_map = Map.new(previous_stats, &{&1.interface_id, &1})
# Calculate BPS for each interface that has both previous and current stats
bps_per_interface =
Enum.flat_map(current_stats, fn current ->
calculate_interface_bps_if_previous_exists(current, previous_map)
end)
# Aggregate total traffic across all interfaces
total_in = Enum.reduce(bps_per_interface, 0.0, fn x, acc -> acc + x.in_bps end)
total_out = Enum.reduce(bps_per_interface, 0.0, fn x, acc -> acc + x.out_bps end)
# Return two data points: inbound and outbound
[
%{
sensor_id: "traffic_in",
label: "Inbound",
value: Float.round(total_in, 2),
timestamp: timestamp_ms
},
%{
sensor_id: "traffic_out",
label: "Outbound",
value: Float.round(total_out, 2),
timestamp: timestamp_ms
}
]
end
# Calculate BPS for an interface if previous stat exists
defp calculate_interface_bps_if_previous_exists(current, previous_map) do
case Map.get(previous_map, current.interface_id) do
nil ->
[]
previous ->
time_diff = current.polled_at |> DateTime.diff(previous.polled_at, :second) |> max(1)
in_bps = calculate_bps(current.if_in_octets, previous.if_in_octets, time_diff)
out_bps = calculate_bps(current.if_out_octets, previous.if_out_octets, time_diff)
[%{interface_id: current.interface_id, in_bps: in_bps, out_bps: out_bps}]
end
end
# Calculate BPS from octets difference
defp calculate_bps(current_octets, previous_octets, time_diff) do
if current_octets && previous_octets do
max((current_octets - previous_octets) * 8 / time_diff, 0)
else
0.0
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
@ -1131,9 +1316,14 @@ defmodule ToweropsWeb.GraphLive.Show 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)
# Traffic graphs don't have sensors (return empty list)
if is_nil(sensor_types) || is_nil(device) do
[]
else
device.sensors
|> Enum.filter(&(&1.sensor_type in sensor_types))
|> Enum.sort_by(& &1.sensor_index)
end
end
# Poll single sensor via SNMP