diff --git a/.gitignore b/.gitignore index 4ae3db95..aea768b2 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,9 @@ # The directory Mix downloads your dependencies sources to. /deps/ +# Development log files +/log/dev.log* + # Vendored package dependencies and build artifacts vendor/*/deps/ vendor/*/_build/ diff --git a/config/dev.exs b/config/dev.exs index 51f85790..80f17e15 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -7,12 +7,22 @@ config :honeybadger, # Don't use custom filter in dev since Honeybadger is excluded filter: Honeybadger.Filter.Default -# Do not include metadata nor timestamps in development logs -config :logger, :default_formatter, format: "[$level] $message\n" +# Console backend configuration +config :logger, :console, + format: "[$level] $message\n", + metadata: [] -# Set minimum log level to info to prevent debug logs from exposing sensitive data -# SNMPKit logs complete SNMP messages at debug level which includes community strings -config :logger, level: :info +# File backend configuration (writes to log/dev.log) +config :logger, :file_log, + path: "log/dev.log", + level: :info, + format: "[$level] $message\n", + metadata: [] + +# Logger configuration with console and file backends +config :logger, + backends: [:console, {LoggerFileBackend, :file_log}], + level: :info # Initialize plugs at runtime for faster development compilation config :phoenix, :plug_init_mode, :runtime diff --git a/lib/towerops/agents.ex b/lib/towerops/agents.ex index 0f8d57b4..59c2bb44 100644 --- a/lib/towerops/agents.ex +++ b/lib/towerops/agents.ex @@ -410,13 +410,14 @@ defmodule Towerops.Agents do query = from(e in Device, - join: s in assoc(e, :site), - join: o in assoc(s, :organization), + left_join: s in assoc(e, :site), + join: o in assoc(e, :organization), left_join: aa in AgentAssignment, on: aa.device_id == e.id and aa.enabled == true, where: e.snmp_enabled == true, preload: [ :agent_assignments, + :organization, site: :organization, snmp_device: [:sensors, :interfaces] ] @@ -428,15 +429,19 @@ defmodule Towerops.Agents do end # Adds where clause to match devices assigned to this agent + # Supports devices assigned directly to org (no site) or to a site # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity defp where_agent_matches(query, agent_token_id, is_global_default) do from([e, s, o, aa] in query, where: aa.agent_token_id == ^agent_token_id or - (is_nil(aa.agent_token_id) and s.agent_token_id == ^agent_token_id) or - (is_nil(aa.agent_token_id) and is_nil(s.agent_token_id) and + (is_nil(aa.agent_token_id) and not is_nil(s.id) and + s.agent_token_id == ^agent_token_id) or + (is_nil(aa.agent_token_id) and + (is_nil(s.id) or is_nil(s.agent_token_id)) and o.default_agent_token_id == ^agent_token_id) or - (^is_global_default and is_nil(aa.agent_token_id) and is_nil(s.agent_token_id) and + (^is_global_default and is_nil(aa.agent_token_id) and + (is_nil(s.id) or is_nil(s.agent_token_id)) and is_nil(o.default_agent_token_id)) ) end diff --git a/lib/towerops/application.ex b/lib/towerops/application.ex index b1d64a08..e98957be 100644 --- a/lib/towerops/application.ex +++ b/lib/towerops/application.ex @@ -7,6 +7,8 @@ defmodule Towerops.Application do use Application + alias Towerops.Workers.JobCleanupTask + require Logger # Capture the build timestamp at compile time (fallback for development) @@ -113,7 +115,7 @@ defmodule Towerops.Application do Task.start(fn -> # Wait for supervisor to fully initialize Process.sleep(2000) - Towerops.Workers.JobCleanupTask.run() + JobCleanupTask.run() end) result diff --git a/lib/towerops/proto/agent.pb.ex b/lib/towerops/proto/agent.pb.ex index cf26e436..13875219 100644 --- a/lib/towerops/proto/agent.pb.ex +++ b/lib/towerops/proto/agent.pb.ex @@ -334,6 +334,7 @@ defmodule Towerops.Agent.SnmpResult do field :job_type, 2, type: Towerops.Agent.JobType, json_name: "jobType", enum: true field :oid_values, 3, repeated: true, type: Towerops.Agent.SnmpResult.OidValuesEntry, json_name: "oidValues", map: true field :timestamp, 4, type: :int64 + field :job_id, 5, type: :string, json_name: "jobId" end defmodule Towerops.Agent.AgentHeartbeat do diff --git a/lib/towerops/workers/discovery_worker.ex b/lib/towerops/workers/discovery_worker.ex index 8c9325bc..2956b300 100644 --- a/lib/towerops/workers/discovery_worker.ex +++ b/lib/towerops/workers/discovery_worker.ex @@ -27,8 +27,9 @@ defmodule Towerops.Workers.DiscoveryWorker do require Logger - # Timeout for waiting for agent-based discovery (60 seconds) - @agent_discovery_timeout_ms 60_000 + # Timeout for waiting for agent-based discovery (120 seconds) + # Increased to match agent SNMP timeout - SNMPv3 operations can take 50+ seconds + @agent_discovery_timeout_ms 120_000 # Overall job timeout (5 minutes) - job will be cancelled if it exceeds this @job_timeout_ms 300_000 diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index 7cea1859..aaeddf50 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -79,6 +79,9 @@ defmodule ToweropsWeb.AgentChannel do # Subscribe to credential test requests for this agent _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:credential_test") + # Subscribe to live poll requests for this agent + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:live_poll") + # Update last_seen_at and IP on join remote_ip = get_remote_ip(socket) _ = Agents.update_agent_token_heartbeat(agent_token.id, remote_ip, %{}) @@ -104,11 +107,7 @@ defmodule ToweropsWeb.AgentChannel do job_list = %AgentJobList{jobs: jobs} binary = AgentJobList.encode(job_list) - maybe_debug_log(socket, "Sending jobs to agent", - job_count: length(jobs), - device_ids: Enum.map(jobs, & &1.device_id), - job_types: Enum.map(jobs, & &1.job_type) - ) + Logger.info("Sending #{length(jobs)} jobs to agent: #{inspect(Enum.map(jobs, & &1.job_id))}") push(socket, "jobs", %{binary: Base.encode64(binary)}) {:noreply, socket} @@ -201,6 +200,36 @@ defmodule ToweropsWeb.AgentChannel do {:noreply, socket} end + # Handle PubSub broadcast when live polling is requested for a device + def handle_info({:live_poll_requested, device_id, sensor_oids, reply_topic}, socket) do + maybe_debug_log(socket, "Live poll requested for device", + device_id: device_id, + sensor_count: length(sensor_oids), + reply_topic: reply_topic + ) + + case Devices.get_device_with_details(device_id) do + nil -> + Logger.error("Device not found for live poll request", device_id: device_id) + {:noreply, socket} + + device -> + # Build live polling job with only requested sensor OIDs + job = build_live_polling_job(device, sensor_oids, reply_topic) + job_list = %AgentJobList{jobs: [job]} + binary = AgentJobList.encode(job_list) + + maybe_debug_log(socket, "Sending live poll job to agent", + device_id: device_id, + device_name: device.name, + oid_count: length(sensor_oids) + ) + + push(socket, "jobs", %{binary: Base.encode64(binary)}) + {:noreply, socket} + end + end + @impl true @spec handle_in(String.t(), map(), Phoenix.Socket.t()) :: {:noreply, Phoenix.Socket.t()} def handle_in("result", %{"binary" => binary_b64}, socket) do @@ -210,11 +239,18 @@ defmodule ToweropsWeb.AgentChannel do maybe_debug_log(socket, "Received SNMP result from agent", device_id: result.device_id, job_type: result.job_type, + job_id: result.job_id, binary_size: byte_size(binary_b64), oid_count: map_size(result.oid_values) ) - _ = process_snmp_result(socket.assigns.organization_id, result, socket) + # Check if this is a live poll result + if String.starts_with?(result.job_id, "live_poll:") do + handle_live_poll_result(result, socket) + else + _ = process_snmp_result(socket.assigns.organization_id, result, socket) + end + {:noreply, socket} end @@ -459,6 +495,45 @@ defmodule ToweropsWeb.AgentChannel do } end + defp build_live_polling_job(device, sensor_oids, reply_topic) do + snmp_credentials = resolve_snmp_credentials(device) + + %AgentJob{ + job_id: "live_poll:#{device.id}:#{reply_topic}", + job_type: :POLL, + device_id: device.id, + snmp_device: build_snmp_device_message(device, snmp_credentials), + queries: [ + %SnmpQuery{ + query_type: :GET, + oids: sensor_oids + } + ] + } + end + + # Handle live poll result - broadcast to reply topic instead of saving + defp handle_live_poll_result(result, socket) do + # Extract reply topic from job_id: "live_poll:device_id:reply_topic" + case String.split(result.job_id, ":", parts: 3) do + ["live_poll", _device_id, reply_topic] -> + maybe_debug_log(socket, "Broadcasting live poll result", + reply_topic: reply_topic, + oid_count: map_size(result.oid_values) + ) + + # Broadcast OID values directly to the waiting LiveView + Phoenix.PubSub.broadcast( + Towerops.PubSub, + reply_topic, + {:live_poll_result, result.oid_values} + ) + + _ -> + Logger.warning("Invalid live poll job_id format: #{result.job_id}") + end + end + defp build_discovery_queries do [ # System info (GET) @@ -728,38 +803,58 @@ defmodule ToweropsWeb.AgentChannel do oid_values = Map.new(result.oid_values) timestamp = DateTime.from_unix!(result.timestamp, :second) + Logger.info( + "Processing polling result for #{device.name}: #{map_size(oid_values)} OIDs, #{length(snmp_device.sensors)} sensors" + ) + # Process sensor readings - Enum.each(snmp_device.sensors, fn sensor -> - if value = Map.get(oid_values, sensor.sensor_oid) do - Snmp.create_sensor_reading(%{ - sensor_id: sensor.id, - value: parse_float(value) / sensor.sensor_divisor, - status: "ok", - checked_at: timestamp - }) - end - end) + Enum.each(snmp_device.sensors, &process_sensor_reading(&1, oid_values, timestamp)) # Process interface stats - Enum.each(device.snmp_device.interfaces, fn iface -> - idx = iface.if_index - - Snmp.create_interface_stat(%{ - interface_id: iface.id, - if_in_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.10.#{idx}")), - if_out_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.16.#{idx}")), - if_in_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.14.#{idx}")), - if_out_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.20.#{idx}")), - if_in_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.13.#{idx}")), - if_out_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.19.#{idx}")), - checked_at: timestamp - }) - end) + Enum.each(device.snmp_device.interfaces, &process_interface_stats(&1, oid_values, timestamp)) # Note: Neighbor discovery is handled by separate polling worker # since it requires complex LLDP/CDP parsing end + defp process_sensor_reading(sensor, oid_values, timestamp) do + case Map.get(oid_values, sensor.sensor_oid) do + nil -> + Logger.info("Sensor #{sensor.sensor_descr} (#{sensor.sensor_oid}): OID not in result") + + value -> + with parsed_value when not is_nil(parsed_value) <- parse_float(value) do + final_value = parsed_value / sensor.sensor_divisor + + Logger.info( + "Sensor #{sensor.sensor_descr}: raw=#{inspect(value)}, parsed=#{inspect(parsed_value)}, divisor=#{sensor.sensor_divisor}, final=#{inspect(final_value)}" + ) + + Snmp.create_sensor_reading(%{ + sensor_id: sensor.id, + value: final_value, + status: "ok", + checked_at: timestamp + }) + end + end + end + + defp process_interface_stats(iface, oid_values, timestamp) do + idx = iface.if_index + + Snmp.create_interface_stat(%{ + interface_id: iface.id, + if_in_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.10.#{idx}")), + if_out_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.16.#{idx}")), + if_in_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.14.#{idx}")), + if_out_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.20.#{idx}")), + if_in_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.13.#{idx}")), + if_out_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.19.#{idx}")), + checked_at: timestamp + }) + end + defp parse_integer(nil), do: nil defp parse_integer(value) when is_integer(value), do: value diff --git a/lib/towerops_web/endpoint.ex b/lib/towerops_web/endpoint.ex index 3ac163cc..c276a822 100644 --- a/lib/towerops_web/endpoint.ex +++ b/lib/towerops_web/endpoint.ex @@ -16,7 +16,7 @@ defmodule ToweropsWeb.Endpoint do longpoll: [connect_info: [session: @session_options]] socket "/socket/agent", ToweropsWeb.AgentSocket, - websocket: true, + websocket: [connect_info: [:peer_data]], longpoll: false # Serve at "/" the static files from "priv/static" directory. diff --git a/lib/towerops_web/live/device_live/form.ex b/lib/towerops_web/live/device_live/form.ex index 9ee405b1..47bcf9bc 100644 --- a/lib/towerops_web/live/device_live/form.ex +++ b/lib/towerops_web/live/device_live/form.ex @@ -271,9 +271,11 @@ defmodule ToweropsWeb.DeviceLive.Form do def handle_event("test_snmp", _params, socket) do changeset = socket.assigns.form.source form_data = Ecto.Changeset.apply_changes(changeset) - {auth_password, priv_password} = resolve_snmpv3_passwords(changeset, form_data) - device_map = build_device_map(form_data, auth_password, priv_password) + # Resolve credentials with inheritance from site/org + resolved_credentials = resolve_snmp_credentials_for_test(form_data, changeset, socket.assigns) + + device_map = build_device_map_with_credentials(form_data, resolved_credentials) effective_agent_id = determine_effective_agent_id(device_map, socket.assigns) handle_snmp_test(socket, device_map, effective_agent_id) @@ -295,44 +297,120 @@ defmodule ToweropsWeb.DeviceLive.Form do end end - defp resolve_snmpv3_passwords(changeset, form_data) do + # Resolve all SNMP credentials with inheritance from site/org + defp resolve_snmp_credentials_for_test(form_data, changeset, assigns) do + snmp_version = form_data.snmp_version || "2c" + + if snmp_version == "3" do + resolve_snmpv3_credentials_for_test(form_data, changeset, assigns) + else + resolve_snmp_community_for_test(form_data, changeset, assigns) + end + end + + # Resolve SNMPv3 credentials with site/org inheritance + defp resolve_snmpv3_credentials_for_test(form_data, changeset, assigns) do changes = changeset.changes - passwords_changed? = Map.has_key?(changes, :snmpv3_auth_password) or Map.has_key?(changes, :snmpv3_priv_password) - cond do - passwords_changed? -> - {Map.get(changes, :snmpv3_auth_password), Map.get(changes, :snmpv3_priv_password)} + # Check if credentials are explicitly set in form + if Map.has_key?(changes, :snmpv3_username) && form_data.snmpv3_username do + # Use form values + %{ + version: "3", + security_level: form_data.snmpv3_security_level, + username: form_data.snmpv3_username, + auth_protocol: form_data.snmpv3_auth_protocol, + auth_password: Map.get(changes, :snmpv3_auth_password), + priv_protocol: form_data.snmpv3_priv_protocol, + priv_password: Map.get(changes, :snmpv3_priv_password) + } + else + # Inherit from site or org + site_id = form_data.site_id + site = site_id && Enum.find(assigns.available_sites, &(&1.id == site_id)) - form_data.id -> - get_existing_device_passwords(form_data.id) + cond do + # Try site-level credentials + site && site.snmpv3_username -> + %{ + version: "3", + security_level: site.snmpv3_security_level, + username: site.snmpv3_username, + auth_protocol: site.snmpv3_auth_protocol, + auth_password: site.snmpv3_auth_password, + priv_protocol: site.snmpv3_priv_protocol, + priv_password: site.snmpv3_priv_password + } - true -> - {Map.get(changes, :snmpv3_auth_password), Map.get(changes, :snmpv3_priv_password)} + # Fall back to org-level credentials + assigns.organization.snmpv3_username -> + %{ + version: "3", + security_level: assigns.organization.snmpv3_security_level, + username: assigns.organization.snmpv3_username, + auth_protocol: assigns.organization.snmpv3_auth_protocol, + auth_password: assigns.organization.snmpv3_auth_password, + priv_protocol: assigns.organization.snmpv3_priv_protocol, + priv_password: assigns.organization.snmpv3_priv_password + } + + true -> + %{version: "3"} + end end end - defp get_existing_device_passwords(device_id) do - case Devices.get_snmpv3_config(device_id) do - nil -> {nil, nil} - config -> {config.auth_password, config.priv_password} + # Resolve SNMP community with site/org inheritance + defp resolve_snmp_community_for_test(form_data, changeset, assigns) do + version = form_data.snmp_version || "2c" + community = get_inherited_community(form_data, changeset, assigns) + + if community do + %{version: version, community: community} + else + %{version: version} end end - defp build_device_map(form_data, auth_password, priv_password) do - %{ + defp get_inherited_community(form_data, changeset, assigns) do + changes = changeset.changes + + # Check if community is explicitly set in form + if Map.has_key?(changes, :snmp_community) && form_data.snmp_community do + form_data.snmp_community + else + # Try site-level, then org-level community + site = form_data.site_id && Enum.find(assigns.available_sites, &(&1.id == form_data.site_id)) + + if site && site.snmp_community do + site.snmp_community + else + assigns.organization.snmp_community + end + end + end + + defp build_device_map_with_credentials(form_data, credentials) do + base_map = %{ "agent_token_id" => Map.get(form_data, :agent_token_id), "site_id" => form_data.site_id, "ip_address" => form_data.ip_address, "snmp_port" => form_data.snmp_port || 161, - "snmp_version" => form_data.snmp_version || "2c", - "snmp_community" => form_data.snmp_community, - "snmpv3_security_level" => form_data.snmpv3_security_level, - "snmpv3_username" => form_data.snmpv3_username, - "snmpv3_auth_protocol" => form_data.snmpv3_auth_protocol, - "snmpv3_auth_password" => auth_password, - "snmpv3_priv_protocol" => form_data.snmpv3_priv_protocol, - "snmpv3_priv_password" => priv_password + "snmp_version" => credentials.version } + + if credentials.version == "3" do + Map.merge(base_map, %{ + "snmpv3_security_level" => credentials[:security_level], + "snmpv3_username" => credentials[:username], + "snmpv3_auth_protocol" => credentials[:auth_protocol], + "snmpv3_auth_password" => credentials[:auth_password], + "snmpv3_priv_protocol" => credentials[:priv_protocol], + "snmpv3_priv_password" => credentials[:priv_password] + }) + else + Map.put(base_map, "snmp_community", credentials[:community]) + end end defp handle_snmp_test(socket, _device_map, nil) do diff --git a/lib/towerops_web/live/graph_live/show.ex b/lib/towerops_web/live/graph_live/show.ex index ef8c25a5..719f6f49 100644 --- a/lib/towerops_web/live/graph_live/show.ex +++ b/lib/towerops_web/live/graph_live/show.ex @@ -517,38 +517,50 @@ defmodule ToweropsWeb.GraphLive.Show do stats |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [stat1, stat2] -> - time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1) + # 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) + 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 - } + %{ + 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] -> - time_diff = stat2.checked_at |> DateTime.diff(stat1.checked_at, :second) |> max(1) + # 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) + 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 - } + %{ + x: DateTime.to_unix(stat2.checked_at, :millisecond), + y: bps + } + end end) + |> Enum.reject(&is_nil/1) end # Format value with unit for display @@ -628,6 +640,12 @@ defmodule ToweropsWeb.GraphLive.Show 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 @@ -671,20 +689,22 @@ defmodule ToweropsWeb.GraphLive.Show do # Check if device has agent assignment has_agent = socket.assigns[:has_agent_assignment] || false - Logger.debug("Live poll for device #{device.name}: has_agent=#{has_agent}") + Logger.info("Live poll for device #{device.name}: has_agent=#{has_agent}") new_points = if has_agent do - # Use agent-collected data from database - Logger.debug("Using agent-collected data for live polling") - fetch_latest_agent_data(socket.assigns, timestamp_ms) + # 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 - Logger.debug("Using direct SNMP for live polling") + # 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( @@ -703,48 +723,80 @@ defmodule ToweropsWeb.GraphLive.Show do }) end - # Fetch latest sensor data collected by agent - defp fetch_latest_agent_data(assigns, _timestamp_ms) do - sensors = get_sensors_for_live_mode(assigns) - - sensors - |> Enum.map(&fetch_sensor_reading/1) - |> Enum.reject(&is_nil/1) - end - - defp fetch_sensor_reading(sensor) do - case Snmp.get_latest_sensor_reading(sensor.id) do + # 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 -> - Logger.debug("No reading found for sensor #{sensor.id} (#{sensor.sensor_descr})") - nil + [] - %{value: nil} = reading -> - Logger.debug("Nil value for sensor #{sensor.id} (#{sensor.sensor_descr}) at #{reading.checked_at}") - nil + assignment -> + # Generate unique reply topic for this request + reply_topic = "live_poll_reply:#{:erlang.unique_integer([:positive])}" - reading -> - validate_and_format_reading(sensor, reading) + # 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 - defp validate_and_format_reading(sensor, reading) do - age_seconds = DateTime.diff(DateTime.utc_now(), reading.checked_at, :second) + # 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) - if age_seconds > 300 do - Logger.debug("Stale reading for sensor #{sensor.id} (#{sensor.sensor_descr}), age: #{age_seconds}s") - nil - else - reading_timestamp_ms = DateTime.to_unix(reading.checked_at, :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(reading.value, 1), - timestamp: reading_timestamp_ms + 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) diff --git a/mix.exs b/mix.exs index 1616c059..70abad84 100644 --- a/mix.exs +++ b/mix.exs @@ -87,7 +87,8 @@ defmodule Towerops.MixProject do {:sobelow, "~> 0.14.1", only: [:dev, :test], runtime: false}, {:mix_audit, "~> 2.1", only: [:dev, :test], runtime: false}, {:hammer, "~> 6.2"}, - {:cloak_ecto, "~> 1.3"} + {:cloak_ecto, "~> 1.3"}, + {:logger_file_backend, "~> 0.0.13", only: :dev} ] end diff --git a/mix.lock b/mix.lock index a533db9b..18fcad60 100644 --- a/mix.lock +++ b/mix.lock @@ -35,6 +35,7 @@ "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"}, "lazy_html": {:hex, :lazy_html, "0.1.8", "677a8642e644eef8de98f3040e2520d42d0f0f8bd6c5cd49db36504e34dffe91", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.9.0", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:fine, "~> 0.1.0", [hex: :fine, repo: "hexpm", optional: false]}], "hexpm", "0d8167d930b704feb94b41414ca7f5779dff9bca7fcf619fcef18de138f08736"}, "libcluster": {:hex, :libcluster, "3.5.0", "5ee4cfde4bdf32b2fef271e33ce3241e89509f4344f6c6a8d4069937484866ba", [:mix], [{:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ebf6561fcedd765a4cd43b4b8c04b1c87f4177b5fb3cbdfe40a780499d72f743"}, + "logger_file_backend": {:hex, :logger_file_backend, "0.0.14", "774bb661f1c3fed51b624d2859180c01e386eb1273dc22de4f4a155ef749a602", [:mix], [], "hexpm", "071354a18196468f3904ef09413af20971d55164267427f6257b52cfba03f9e6"}, "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"}, "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, "mimerl": {:hex, :mimerl, "1.4.0", "3882a5ca67fbbe7117ba8947f27643557adec38fa2307490c4c4207624cb213b", [:rebar3], [], "hexpm", "13af15f9f68c65884ecca3a3891d50a7b57d82152792f3e19d88650aa126b144"}, diff --git a/test/towerops_web/live/graph_live/show_test.exs b/test/towerops_web/live/graph_live/show_test.exs index f79a199b..4e6711b1 100644 --- a/test/towerops_web/live/graph_live/show_test.exs +++ b/test/towerops_web/live/graph_live/show_test.exs @@ -228,6 +228,145 @@ defmodule ToweropsWeb.GraphLive.ShowTest do assert html =~ "Overall Traffic" assert html =~ device.name end + + test "handles nil if_in_octets gracefully", %{ + conn: conn, + device: device, + organization: _org, + snmp_device: snmp_device + } do + # Create an interface + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_descr: "eth0", + if_name: "eth0", + if_oper_status: "up", + if_speed: 1_000_000_000 + }) + |> Repo.insert!() + + # Create interface stats with nil if_in_octets + base_time = DateTime.utc_now() + + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: nil, + if_out_octets: 1000, + checked_at: DateTime.add(base_time, -60, :second) + }) + + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: nil, + if_out_octets: 2000, + checked_at: base_time + }) + + # Should not crash + {:ok, _view, html} = live(conn, ~p"/devices/#{device.id}/graph/traffic") + + assert html =~ "Overall Traffic" + assert html =~ device.name + end + + test "handles nil if_out_octets gracefully", %{ + conn: conn, + device: device, + organization: _org, + snmp_device: snmp_device + } do + # Create an interface + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_descr: "eth0", + if_name: "eth0", + if_oper_status: "up", + if_speed: 1_000_000_000 + }) + |> Repo.insert!() + + # Create interface stats with nil if_out_octets + base_time = DateTime.utc_now() + + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 1000, + if_out_octets: nil, + checked_at: DateTime.add(base_time, -60, :second) + }) + + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 2000, + if_out_octets: nil, + checked_at: base_time + }) + + # Should not crash + {:ok, _view, html} = live(conn, ~p"/devices/#{device.id}/graph/traffic") + + assert html =~ "Overall Traffic" + assert html =~ device.name + end + + test "handles mixed nil and valid octets values", %{ + conn: conn, + device: device, + organization: _org, + snmp_device: snmp_device + } do + # Create an interface + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_descr: "eth0", + if_name: "eth0", + if_oper_status: "up", + if_speed: 1_000_000_000 + }) + |> Repo.insert!() + + # Create interface stats with mixed nil and valid values + base_time = DateTime.utc_now() + + # First stat: nil values + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: nil, + if_out_octets: nil, + checked_at: DateTime.add(base_time, -120, :second) + }) + + # Second stat: valid values + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 1000, + if_out_octets: 2000, + checked_at: DateTime.add(base_time, -60, :second) + }) + + # Third stat: valid values + Snmp.create_interface_stat(%{ + interface_id: interface.id, + if_in_octets: 2000, + if_out_octets: 4000, + checked_at: base_time + }) + + # Should not crash and should render valid data points + {:ok, _view, html} = live(conn, ~p"/devices/#{device.id}/graph/traffic") + + assert html =~ "Overall Traffic" + assert html =~ device.name + end end describe "Time range selection" do