From 9a366c2665fc7b738e3c73d004ecef18641e8bba Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 23 Mar 2026 13:35:12 -0500 Subject: [PATCH] debug/traffic-graph-logging (#129) Reviewed-on: https://git.mcintire.me/graham/towerops-web/pulls/129 --- lib/towerops_web/live/device_live/show.ex | 9 +- lib/towerops_web/live/graph_live/show.ex | 254 +++++++++++++++++++--- 2 files changed, 225 insertions(+), 38 deletions(-) diff --git a/lib/towerops_web/live/device_live/show.ex b/lib/towerops_web/live/device_live/show.ex index 9c42a5bb..4c5a242f 100644 --- a/lib/towerops_web/live/device_live/show.ex +++ b/lib/towerops_web/live/device_live/show.ex @@ -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) diff --git a/lib/towerops_web/live/graph_live/show.ex b/lib/towerops_web/live/graph_live/show.ex index 3c61f4a1..7717f89b 100644 --- a/lib/towerops_web/live/graph_live/show.ex +++ b/lib/towerops_web/live/graph_live/show.ex @@ -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