From 623fe0708192081949de4931f3cad2cef26bc07a Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 9 Feb 2026 12:44:37 -0600 Subject: [PATCH] wip --- config/runtime.exs | 4 + lib/towerops/agents.ex | 65 +++++- lib/towerops/devices.ex | 7 +- lib/towerops/snmp.ex | 109 +++------ lib/towerops/snmp/adapters/replay.ex | 36 ++- lib/towerops/snmp/discovery.ex | 24 +- lib/towerops/workers/device_monitor_worker.ex | 18 +- lib/towerops/workers/device_poller_worker.ex | 83 +++++-- lib/towerops_web/channels/agent_channel.ex | 46 +++- .../live/agent_live/show.html.heex | 2 +- lib/towerops_web/live/device_live/show.ex | 22 +- .../live/device_live/show.html.heex | 2 +- lib/towerops_web/live/graph_live/show.ex | 4 +- ...ix_null_handling_in_unique_constraints.exs | 75 +++++++ test/towerops/snmp_test.exs | 209 ++++++++++++++++++ .../workers/device_poller_worker_test.exs | 138 ++++++++++++ 16 files changed, 693 insertions(+), 151 deletions(-) create mode 100644 priv/repo/migrations/20260209152641_fix_null_handling_in_unique_constraints.exs diff --git a/config/runtime.exs b/config/runtime.exs index 32daa9fe..d46e8f79 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -229,6 +229,10 @@ if config_env() == :prod do config :towerops, ToweropsWeb.Endpoint, url: [host: host, port: 443, scheme: "https"], + check_origin: [ + "//towerops.net", + "//*.towerops.net" + ], http: [ # Enable IPv6 and bind on all interfaces. # Set it to {0, 0, 0, 0, 0, 0, 0, 1} for local network only access. diff --git a/lib/towerops/agents.ex b/lib/towerops/agents.ex index 3a438957..ea64daa8 100644 --- a/lib/towerops/agents.ex +++ b/lib/towerops/agents.ex @@ -675,16 +675,63 @@ defmodule Towerops.Agents do """ @spec should_phoenix_poll_device?(Device.t()) :: boolean() - def should_phoenix_poll_device?(%Device{} = device) do - case get_effective_agent_token(device) do - nil -> - # No agent assigned - Phoenix can poll - true + def should_phoenix_poll_device?(%Device{id: device_id}) do + # Use a single atomic query to check assignment and cloud poller status + # This prevents race conditions where assignment changes between reads + # Checks device assignment first, then site default, then org default + query = + from d in Device, + left_join: aa in AgentAssignment, + on: aa.device_id == d.id and aa.enabled == true, + left_join: device_token in AgentToken, + on: device_token.id == aa.agent_token_id, + left_join: s in assoc(d, :site), + left_join: site_token in AgentToken, + on: site_token.id == s.agent_token_id, + left_join: o in assoc(s, :organization), + left_join: org_token in AgentToken, + on: org_token.id == o.default_agent_token_id, + where: d.id == ^device_id, + select: %{ + # Device-level assignment + has_device_assignment: not is_nil(aa.id), + device_is_cloud: device_token.is_cloud_poller, + # Site-level fallback + has_site_token: not is_nil(s.agent_token_id), + site_is_cloud: site_token.is_cloud_poller, + # Org-level fallback + has_org_token: not is_nil(o.default_agent_token_id), + org_is_cloud: org_token.is_cloud_poller + } - agent_token_id -> - # Check if the agent is a cloud poller - agent_token = get_agent_token!(agent_token_id) - agent_token.is_cloud_poller + case Repo.one(query) do + nil -> + # Device doesn't exist + false + + result -> + # Determine effective agent using precedence: device > site > org + cond do + # 1. Device-level assignment + result.has_device_assignment -> + # Device assigned - check if cloud poller (nil = deleted token, treat as cloud) + result.device_is_cloud != false + + # 2. Site-level default + result.has_site_token -> + # Site has default - check if cloud poller + result.site_is_cloud != false + + # 3. Organization-level default + result.has_org_token -> + # Org has default - check if cloud poller + result.org_is_cloud != false + + # 4. No agent anywhere + true -> + # No agent assigned at any level - Phoenix polls + true + end end end diff --git a/lib/towerops/devices.ex b/lib/towerops/devices.ex index 7089b035..05591055 100644 --- a/lib/towerops/devices.ex +++ b/lib/towerops/devices.ex @@ -812,11 +812,16 @@ defmodule Towerops.Devices do _ = DeviceMonitorWorker.stop_monitoring(device.id) _ = DevicePollerWorker.stop_polling(device.id) + # Wait briefly for any in-flight jobs to complete or detect deletion + # This reduces (but doesn't eliminate) the window for orphaned writes + # In-flight jobs check device existence via verify_polling_assignment_unchanged + Process.sleep(500) + # Get agent assignment before deleting (if any) assignment = Agents.get_device_assignment(device.id) agent_token_id = if assignment, do: assignment.agent_token_id - # Delete the device + # Delete the device (cascades to SNMP device, sensors, interfaces, etc.) result = Repo.delete(device) # Notify agent to stop processing jobs for this device diff --git a/lib/towerops/snmp.ex b/lib/towerops/snmp.ex index c7ab33d8..5f2f75ff 100644 --- a/lib/towerops/snmp.ex +++ b/lib/towerops/snmp.ex @@ -503,23 +503,18 @@ defmodule Towerops.Snmp do @doc """ Creates or updates a neighbor record. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same neighbor simultaneously. """ def upsert_neighbor(attrs) do - case Repo.get_by(Neighbor, - interface_id: attrs.interface_id, - remote_chassis_id: attrs[:remote_chassis_id], - protocol: attrs.protocol - ) do - nil -> - %Neighbor{} - |> Neighbor.changeset(attrs) - |> Repo.insert() - - neighbor -> - neighbor - |> Neighbor.changeset(attrs) - |> Repo.update() - end + %Neighbor{} + |> Neighbor.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:interface_id, :remote_chassis_id, :protocol], + returning: true + ) end @doc """ @@ -1501,23 +1496,18 @@ defmodule Towerops.Snmp do @doc """ Creates or updates an ARP entry. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same ARP entry simultaneously. """ def upsert_arp_entry(attrs) do - case Repo.get_by(ArpEntry, - device_id: attrs.device_id, - ip_address: attrs.ip_address, - mac_address: attrs.mac_address - ) do - nil -> - %ArpEntry{} - |> ArpEntry.changeset(attrs) - |> Repo.insert() - - arp_entry -> - arp_entry - |> ArpEntry.changeset(attrs) - |> Repo.update() - end + %ArpEntry{} + |> ArpEntry.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:device_id, :ip_address, :mac_address], + returning: true + ) end @doc """ @@ -1566,56 +1556,19 @@ defmodule Towerops.Snmp do @doc """ Creates or updates a MAC address entry. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same MAC address simultaneously. The unique constraint + handles duplicate detection automatically, eliminating the need for manual cleanup. """ def upsert_mac_address(attrs) do - import Ecto.Query - - require Logger - - # Build query to find existing MAC address entry - query = - from m in MacAddress, - where: m.device_id == ^attrs.device_id, - where: m.mac_address == ^attrs.mac_address - - # Handle vlan_id comparison (nil requires is_nil/1) - query = - case attrs[:vlan_id] do - nil -> where(query, [m], is_nil(m.vlan_id)) - vlan_id -> where(query, [m], m.vlan_id == ^vlan_id) - end - - # Use Repo.all to handle potential duplicates gracefully - # If duplicates exist, keep the first one and delete the rest - case Repo.all(query) do - [] -> - %MacAddress{} - |> MacAddress.changeset(attrs) - |> Repo.insert() - - [mac_address] -> - mac_address - |> MacAddress.changeset(attrs) - |> Repo.update() - - [first | duplicates] -> - # Handle duplicates: delete extras, update first - Logger.warning("Found #{length(duplicates)} duplicate MAC address entries, cleaning up", - device_id: attrs.device_id, - mac_address: attrs.mac_address, - vlan_id: attrs[:vlan_id], - duplicate_ids: Enum.map(duplicates, & &1.id) - ) - - # Delete duplicate entries - duplicate_ids = Enum.map(duplicates, & &1.id) - Repo.delete_all(from m in MacAddress, where: m.id in ^duplicate_ids) - - # Update the remaining entry - first - |> MacAddress.changeset(attrs) - |> Repo.update() - end + %MacAddress{} + |> MacAddress.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:device_id, :mac_address, :vlan_id], + returning: true + ) end @doc """ diff --git a/lib/towerops/snmp/adapters/replay.ex b/lib/towerops/snmp/adapters/replay.ex index ec27647f..e6c245c2 100644 --- a/lib/towerops/snmp/adapters/replay.ex +++ b/lib/towerops/snmp/adapters/replay.ex @@ -87,12 +87,18 @@ defmodule Towerops.Snmp.Adapters.Replay do def walk(opts, base_oid) when is_binary(base_oid) do oid_map = Keyword.fetch!(opts, :oid_map) + # Normalize base_oid (strip leading dot if present) + normalized_base = String.trim_leading(base_oid, ".") + # Filter OIDs that start with base_oid (SNMP walk semantics) # Important: Must be children (start with base_oid + ".") + # Handle both absolute (.1.2.3) and relative (1.2.3) OID notation results = oid_map |> Enum.filter(fn {oid, _value} -> - String.starts_with?(oid, base_oid <> ".") + # Normalize OID (strip leading dot) + normalized_oid = String.trim_leading(oid, ".") + String.starts_with?(normalized_oid, normalized_base <> ".") end) |> Enum.map(fn {oid, value} -> {oid, parse_value(value)} @@ -121,31 +127,35 @@ defmodule Towerops.Snmp.Adapters.Replay do @spec parse_value(String.t()) :: term() defp parse_value(value) when is_binary(value) do cond do - # MAC address (very specific pattern) - check FIRST + # Empty string (from ASN_NULL values) - check FIRST + # SNMP NULL values have no data, return nil + value == "" -> + Logger.debug("Parsing empty string as nil") + nil + + # MAC address (very specific pattern) - check SECOND # Examples: aa:bb:cc:dd:ee:ff, AA-BB-CC-DD-EE-FF Regex.match?(~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/, value) -> value - # IPv4 address (specific pattern) - check SECOND + # IPv4 address (specific pattern) - check THIRD # Example: 192.168.1.1 Regex.match?(~r/^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$/, value) -> value - # OID list (contains dots but not IPv4) - check THIRD - # Example: "1.3.6.1.4.1.9" => [1, 3, 6, 1, 4, 1, 9] - # Must have at least 4 components to avoid IPv4 confusion - Regex.match?(~r/^[\d\.]+$/, value) and String.contains?(value, ".") and - length(String.split(value, ".")) > 4 -> + # OID-like string (contains dots but not IPv4) - check FOURTH + # Keep as string rather than parsing to avoid issues with SNMP shorthand notation + # Example: "1.3.6.1.4.1.9" => "1.3.6.1.4.1.9" (not parsed) + # Note: Leading dots (relative OIDs) and other SNMP notation are preserved + Regex.match?(~r/^\.?[\d\.]+$/, value) and String.contains?(value, ".") -> value - |> String.split(".") - |> Enum.map(&String.to_integer/1) - # Pure integer (no dots) - check FOURTH + # Pure integer (no dots) - check FIFTH # Example: "42" => 42 Regex.match?(~r/^\d+$/, value) -> String.to_integer(value) - # Float - check FIFTH + # Float - check SIXTH # Example: "98.6" => 98.6 Regex.match?(~r/^-?\d+\.\d+$/, value) -> String.to_float(value) @@ -163,6 +173,8 @@ defmodule Towerops.Snmp.Adapters.Replay do defp oid_to_sortable(oid) do oid |> String.split(".") + # Filter out empty strings from leading dots + |> Enum.reject(&(&1 == "")) |> Enum.map(&String.to_integer/1) end end diff --git a/lib/towerops/snmp/discovery.ex b/lib/towerops/snmp/discovery.ex index e48c55c3..25fb35dd 100644 --- a/lib/towerops/snmp/discovery.ex +++ b/lib/towerops/snmp/discovery.ex @@ -217,12 +217,9 @@ defmodule Towerops.Snmp.Discovery do {:ok, processors} <- discover_processors_with_timeout(client_opts, timeouts), Logger.info("Discovering storage...", device_id: device.id), {:ok, storage} <- discover_storage_with_timeout(client_opts, timeouts), - Logger.info("Saving discovery results...", device_id: device.id), - {:ok, discovered_device} <- save_discovery_results(device, device_info, interfaces, sensors, vlans), - Logger.info("Syncing IP addresses...", device_id: device.id), - :ok <- sync_ip_addresses(discovered_device, ip_addresses), - Logger.info("Syncing processors...", device_id: device.id), - :ok <- sync_processors(discovered_device, processors), + Logger.info("Saving discovery results (including IP addresses and processors)...", device_id: device.id), + {:ok, discovered_device} <- + save_discovery_results(device, device_info, interfaces, sensors, vlans, ip_addresses, processors), Logger.info("Syncing storage...", device_id: device.id), :ok <- sync_storage(discovered_device, storage), Logger.info("Discovering neighbors...", device_id: device.id), @@ -592,9 +589,11 @@ defmodule Towerops.Snmp.Discovery do device_info(), [interface_data()], [sensor_data()], + [map()], + [map()], [map()] ) :: {:ok, Device.t()} | {:error, term()} - defp save_discovery_results(device, device_info, interfaces, sensors, vlans) do + defp save_discovery_results(device, device_info, interfaces, sensors, vlans, ip_addresses, processors) do # Sanitize raw_discovery_data BEFORE starting transaction to avoid DB timeout # This can take several seconds for large discovery data (thousands of OIDs) sanitized_device_info = prepare_device_info_for_save(device_info, device.id) @@ -608,9 +607,18 @@ defmodule Towerops.Snmp.Discovery do _ = sync_sensors(snmp_device, sensors) _ = sync_vlans(snmp_device, vlans) + # Sync IP addresses and processors inside transaction for atomicity + device_with_interfaces = %{ + snmp_device + | interfaces: Enum.map(synced_interfaces, &Map.put(&1, :device_id, snmp_device.device_id)) + } + + _ = sync_ip_addresses(device_with_interfaces, ip_addresses) + _ = sync_processors(device_with_interfaces, processors) + # Return device with interfaces loaded and device_id added # Use snmp_device.device_id which references the Equipment table - %{snmp_device | interfaces: Enum.map(synced_interfaces, &Map.put(&1, :device_id, snmp_device.device_id))} + device_with_interfaces end) end diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 7e35a5cf..9779dd24 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -72,13 +72,21 @@ defmodule Towerops.Workers.DeviceMonitorWorker do end defp schedule_next_check_with_error_handling(device_id) do - case schedule_next_check(device_id) do - {:ok, _job} -> + # Verify device still exists before scheduling next check + case Devices.get_device(device_id) do + nil -> + Logger.debug("Device deleted, not rescheduling monitoring", device_id: device_id) :ok - {:error, changeset} -> - Logger.error("Failed to schedule next monitoring check for device #{device_id}: #{inspect(changeset.errors)}") - :ok + _device -> + case schedule_next_check(device_id) do + {:ok, _job} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to schedule next monitoring check for device #{device_id}: #{inspect(changeset.errors)}") + :ok + end end end diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index 4284db38..fdddda4e 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -88,16 +88,25 @@ defmodule Towerops.Workers.DevicePollerWorker do end end - defp schedule_next_poll_with_error_handling(device_id, device) do - poll_interval = get_poll_interval(device) - - case schedule_next_poll(device_id, poll_interval) do - {:ok, _job} -> + defp schedule_next_poll_with_error_handling(device_id, _stale_device) do + # Re-fetch device to get latest check_interval_seconds + # Configuration could have changed during the polling execution + case Devices.get_device(device_id) do + nil -> + Logger.debug("Device deleted, not rescheduling", device_id: device_id) :ok - {:error, changeset} -> - Logger.error("Failed to schedule next poll for device #{device_id}: #{inspect(changeset.errors)}") - :ok + fresh_device -> + poll_interval = get_poll_interval(fresh_device) + + case schedule_next_poll(device_id, poll_interval) do + {:ok, _job} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to schedule next poll for device #{device_id}: #{inspect(changeset.errors)}") + :ok + end end end @@ -175,15 +184,55 @@ defmodule Towerops.Workers.DevicePollerWorker do ] # Wait for all tasks with timeout, handle failures gracefully - tasks - |> Task.yield_many(30_000) - |> Enum.each(fn {task, result} -> - case result do - {:ok, _} -> :ok - {:exit, reason} -> Logger.error("Polling task failed: #{inspect(reason)}") - nil -> Task.shutdown(task, :brutal_kill) - end - end) + _results = + tasks + |> Task.yield_many(30_000) + |> Enum.map(fn {task, result} -> + case result do + {:ok, value} -> + {:ok, value} + + {:exit, reason} -> + Logger.error("Polling task failed: #{inspect(reason)}") + {:error, reason} + + nil -> + Task.shutdown(task, :brutal_kill) + {:error, :timeout} + end + end) + + # Re-check assignment before processing results + # Device could have been reassigned to an agent during the 30-second polling window + case verify_polling_assignment_unchanged(device.id) do + :ok -> + # Assignment unchanged - process results normally + :ok + + :reassigned -> + Logger.info("Device reassigned during polling, discarding results", + device_id: device.id, + device_name: device.name + ) + + :ok + end + end + + # Verify device assignment hasn't changed since polling started + defp verify_polling_assignment_unchanged(device_id) do + case Devices.get_device(device_id) do + nil -> + # Device deleted during polling + :reassigned + + device -> + if Agents.should_phoenix_poll_device?(device) do + :ok + else + :reassigned + end + end end defp poll_device_sensors(device, snmp_device, client_opts, now) do diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index e4ac6e9d..08676257 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -396,20 +396,20 @@ defmodule ToweropsWeb.AgentChannel do def handle_in("monitoring_check", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do with {:ok, binary} <- safe_base64_decode(binary_b64), {:ok, check} <- Validator.validate_monitoring_check_message(binary) do - maybe_debug_log(socket, "Received monitoring check from agent", + maybe_debug_log(socket, "Received ping result from agent", device_id: check.device_id, status: check.status, response_time_ms: check.response_time_ms, binary_size: byte_size(binary_b64) ) - # Store monitoring check result in database + # Store ping result in database _ = store_monitoring_check(check, socket) {:noreply, socket} else {:error, {type, message}} -> - Logger.error("Invalid monitoring check from agent: #{type} - #{message}", + Logger.error("Invalid ping result from agent: #{type} - #{message}", agent_token_id: socket.assigns.agent_token_id, error_type: type, error_message: message, @@ -419,7 +419,7 @@ defmodule ToweropsWeb.AgentChannel do {:noreply, socket} {:error, :base64_decode_failed} -> - Logger.error("Failed to decode monitoring check (invalid base64)", + Logger.error("Failed to decode ping result (invalid base64)", agent_token_id: socket.assigns.agent_token_id, binary_size: byte_size(binary_b64) ) @@ -894,7 +894,8 @@ defmodule ToweropsWeb.AgentChannel do defp process_snmp_result(organization_id, result, socket) do with {:ok, device} <- fetch_device(result.device_id), - :ok <- verify_device_organization(device, organization_id) do + :ok <- verify_device_organization(device, organization_id), + :ok <- verify_device_assignment(device, socket.assigns.agent_token_id) do process_job_result(device, result, socket) else {:error, :device_not_found} -> @@ -902,6 +903,32 @@ defmodule ToweropsWeb.AgentChannel do {:error, :wrong_organization} -> Logger.error("Device #{result.device_id} not in agent's organization") + + {:error, :device_reassigned} -> + Logger.warning("Ignoring stale result for reassigned device", + device_id: result.device_id, + agent_token_id: socket.assigns.agent_token_id + ) + end + end + + # Verify device is still assigned to this agent before processing results + defp verify_device_assignment(device, agent_token_id) do + # Get effective agent token (resolves inheritance from site/org) + effective_agent_token_id = Agents.get_effective_agent_token(device) + + # Check if it matches the requesting agent + if effective_agent_token_id == agent_token_id do + # Also check if there's an explicit device assignment that's disabled + case Agents.get_device_assignment(device.id) do + %{enabled: false} -> + {:error, :device_reassigned} + + _ -> + :ok + end + else + {:error, :device_reassigned} end end @@ -911,8 +938,9 @@ defmodule ToweropsWeb.AgentChannel do with {:ok, device} <- fetch_device(check.device_id), :ok <- verify_device_organization(device, organization_id) do - # Convert protobuf timestamp to DateTime - checked_at = DateTime.from_unix!(check.timestamp, :second) + # Use server time to avoid clock skew issues + # Agent timestamps can be unreliable due to clock drift + checked_at = DateTime.truncate(DateTime.utc_now(), :second) # Convert status string to atom status = String.to_existing_atom(check.status) @@ -930,7 +958,7 @@ defmodule ToweropsWeb.AgentChannel do :ok {:error, changeset} -> - Logger.error("Failed to store monitoring check", + Logger.error("Failed to store ping result", device_id: check.device_id, errors: inspect(changeset.errors) ) @@ -939,7 +967,7 @@ defmodule ToweropsWeb.AgentChannel do end else {:error, :device_not_found} -> - Logger.error("Device not found for monitoring check: #{check.device_id}") + Logger.error("Device not found for ping result: #{check.device_id}") {:error, :device_not_found} {:error, :wrong_organization} -> diff --git a/lib/towerops_web/live/agent_live/show.html.heex b/lib/towerops_web/live/agent_live/show.html.heex index a4c1e104..9f333cef 100644 --- a/lib/towerops_web/live/agent_live/show.html.heex +++ b/lib/towerops_web/live/agent_live/show.html.heex @@ -117,7 +117,7 @@
  • SNMP discovery results (full OID maps)
  • Polling results (sensor readings, interface stats)
  • Error messages with full context
    • -
  • Monitoring check submissions
    • +
  • Ping submissions
    • diff --git a/lib/towerops_web/live/device_live/show.ex b/lib/towerops_web/live/device_live/show.ex index 8c6ae39b..a8c35205 100644 --- a/lib/towerops_web/live/device_live/show.ex +++ b/lib/towerops_web/live/device_live/show.ex @@ -52,16 +52,22 @@ defmodule ToweropsWeb.DeviceLive.Show do {:ok, _device} -> maybe_subscribe_and_schedule_refresh(socket, id) - tab = Map.get(params, "tab", "overview") - page = params |> Map.get("page", "1") |> String.to_integer() + # 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) + socket = + socket + |> load_equipment_data(id) + |> assign(:active_tab, tab) + |> apply_backups_pagination(tab, page) - {:noreply, socket} + {: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, diff --git a/lib/towerops_web/live/device_live/show.html.heex b/lib/towerops_web/live/device_live/show.html.heex index a92e93f4..ac0279b5 100644 --- a/lib/towerops_web/live/device_live/show.html.heex +++ b/lib/towerops_web/live/device_live/show.html.heex @@ -244,7 +244,7 @@ end ]} > - Logs + Events <%= if @snmp_device && @snmp_device.raw_discovery_data do %> diff --git a/lib/towerops_web/live/graph_live/show.ex b/lib/towerops_web/live/graph_live/show.ex index e1b880be..7a4b66bd 100644 --- a/lib/towerops_web/live/graph_live/show.ex +++ b/lib/towerops_web/live/graph_live/show.ex @@ -222,7 +222,7 @@ defmodule ToweropsWeb.GraphLive.Show do # 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("latency"), do: {"Ping 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} @@ -404,7 +404,7 @@ defmodule ToweropsWeb.GraphLive.Show do nil else dataset = %{ - label: "ICMP Latency", + label: "Ping Latency", data: Enum.map(checks, &latency_check_to_chart_point/1) } diff --git a/priv/repo/migrations/20260209152641_fix_null_handling_in_unique_constraints.exs b/priv/repo/migrations/20260209152641_fix_null_handling_in_unique_constraints.exs new file mode 100644 index 00000000..44bf5c26 --- /dev/null +++ b/priv/repo/migrations/20260209152641_fix_null_handling_in_unique_constraints.exs @@ -0,0 +1,75 @@ +defmodule Towerops.Repo.Migrations.FixNullHandlingInUniqueConstraints do + use Ecto.Migration + + @moduledoc """ + Fix unique constraints to handle NULL values correctly. + + PostgreSQL's default unique constraint treats NULL values as distinct, + allowing multiple rows with the same non-NULL values but different NULL values. + + This migration replaces the existing unique constraints with ones that use + NULLS NOT DISTINCT, which treats NULL values as equal for uniqueness purposes. + + This fixes race conditions where concurrent upserts with NULL values could + create duplicate records. + """ + + def up do + # Drop old unique indexes + drop_if_exists unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol]) + drop_if_exists unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id]) + + # Clean up existing duplicates before creating new constraints + # Keep the most recently updated record, delete older duplicates + execute(""" + DELETE FROM snmp_neighbors + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY interface_id, COALESCE(remote_chassis_id, ''), protocol + ORDER BY updated_at DESC + ) AS rn + FROM snmp_neighbors + ) t + WHERE rn > 1 + ) + """) + + execute(""" + DELETE FROM snmp_mac_addresses + WHERE id IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY device_id, mac_address, COALESCE(vlan_id::text, '') + ORDER BY updated_at DESC + ) AS rn + FROM snmp_mac_addresses + ) t + WHERE rn > 1 + ) + """) + + # Create new unique indexes with NULLS NOT DISTINCT + # This ensures NULL values are treated as equal for uniqueness + create unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol], + nulls_distinct: false + ) + + create unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id], + nulls_distinct: false + ) + end + + def down do + # Revert to original unique indexes (which allow multiple NULLs) + drop_if_exists unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol]) + drop_if_exists unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id]) + + create unique_index(:snmp_neighbors, [:interface_id, :remote_chassis_id, :protocol]) + create unique_index(:snmp_mac_addresses, [:device_id, :mac_address, :vlan_id]) + end +end diff --git a/test/towerops/snmp_test.exs b/test/towerops/snmp_test.exs index ebff8cdb..97bdeea6 100644 --- a/test/towerops/snmp_test.exs +++ b/test/towerops/snmp_test.exs @@ -8,6 +8,7 @@ defmodule Towerops.SnmpTest do alias Towerops.Snmp.Device alias Towerops.Snmp.Interface alias Towerops.Snmp.InterfaceStat + alias Towerops.Snmp.MacAddress alias Towerops.Snmp.Neighbor alias Towerops.Snmp.Sensor alias Towerops.Snmp.SensorReading @@ -1050,6 +1051,7 @@ defmodule Towerops.SnmpTest do device_id: device.id, interface_id: interface.id, protocol: "lldp", + remote_chassis_id: "00:11:22:33:44:55", remote_system_name: "switch-b", last_discovered_at: DateTime.utc_now() }) @@ -1061,6 +1063,7 @@ defmodule Towerops.SnmpTest do device_id: device.id, interface_id: interface.id, protocol: "cdp", + remote_chassis_id: "aa:bb:cc:dd:ee:ff", remote_system_name: "router-a", last_discovered_at: DateTime.utc_now() }) @@ -1072,6 +1075,7 @@ defmodule Towerops.SnmpTest do device_id: device.id, interface_id: interface.id, protocol: "lldp", + remote_chassis_id: "11:22:33:44:55:66", remote_system_name: "switch-a", last_discovered_at: DateTime.utc_now() }) @@ -2298,4 +2302,209 @@ defmodule Towerops.SnmpTest do assert Enum.at(discovered, 2).hostname == "old-device" end end + + describe "concurrent upsert operations (race condition tests)" do + test "concurrent neighbor upserts don't create duplicates", %{device: device, snmp_device: snmp_device} do + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_name: "eth0" + }) + |> Repo.insert!() + + # Spawn 10 concurrent workers all upserting the same neighbor + tasks = + for i <- 1..10 do + Task.async(fn -> + Snmp.upsert_neighbor(%{ + device_id: device.id, + interface_id: interface.id, + remote_chassis_id: "00:11:22:33:44:55", + protocol: "lldp", + remote_system_name: "neighbor-#{i}", + last_discovered_at: DateTime.utc_now() + }) + end) + end + + # Wait for all tasks to complete + results = Task.await_many(tasks, 5_000) + + # All should succeed + assert Enum.all?(results, fn + {:ok, _} -> true + _ -> false + end) + + # Should have exactly 1 neighbor, not 10 + neighbors = + Repo.all( + from n in Neighbor, + where: n.interface_id == ^interface.id and n.remote_chassis_id == "00:11:22:33:44:55" + ) + + assert length(neighbors) == 1 + + # The final name should be one of the concurrent writes (non-deterministic but valid) + neighbor = hd(neighbors) + assert String.starts_with?(neighbor.remote_system_name, "neighbor-") + end + + test "concurrent ARP entry upserts don't create duplicates", %{device: device} do + # Spawn 10 concurrent workers all upserting the same ARP entry + tasks = + for _i <- 1..10 do + Task.async(fn -> + Snmp.upsert_arp_entry(%{ + device_id: device.id, + ip_address: "192.168.1.100", + mac_address: "aa:bb:cc:dd:ee:ff", + entry_type: "dynamic", + last_seen_at: DateTime.utc_now() + }) + end) + end + + # Wait for all tasks to complete + results = Task.await_many(tasks, 5_000) + + # All should succeed + assert Enum.all?(results, fn + {:ok, _} -> true + _ -> false + end) + + # Should have exactly 1 ARP entry, not 10 + arp_entries = + Repo.all( + from a in ArpEntry, + where: + a.device_id == ^device.id and a.ip_address == "192.168.1.100" and + a.mac_address == "aa:bb:cc:dd:ee:ff" + ) + + assert length(arp_entries) == 1 + end + + test "concurrent MAC address upserts don't create duplicates", %{device: device} do + # Spawn 10 concurrent workers all upserting the same MAC address + tasks = + for i <- 1..10 do + Task.async(fn -> + Snmp.upsert_mac_address(%{ + device_id: device.id, + mac_address: "11:22:33:44:55:66", + vlan_id: 100, + port_index: i, + last_seen_at: DateTime.utc_now() + }) + end) + end + + # Wait for all tasks to complete + results = Task.await_many(tasks, 5_000) + + # All should succeed + assert Enum.all?(results, fn + {:ok, _} -> true + _ -> false + end) + + # Should have exactly 1 MAC address, not 10 + mac_addresses = + Repo.all( + from m in MacAddress, + where: m.device_id == ^device.id and m.mac_address == "11:22:33:44:55:66" and m.vlan_id == 100 + ) + + assert length(mac_addresses) == 1 + + # The final port_index should be one of the concurrent writes + mac = hd(mac_addresses) + assert mac.port_index in 1..10 + end + + test "concurrent MAC address upserts with nil vlan_id don't create duplicates", %{device: device} do + # Spawn 10 concurrent workers all upserting the same MAC address with nil vlan_id + tasks = + for i <- 1..10 do + Task.async(fn -> + Snmp.upsert_mac_address(%{ + device_id: device.id, + mac_address: "aa:bb:cc:dd:ee:ff", + vlan_id: nil, + port_index: i, + last_seen_at: DateTime.utc_now() + }) + end) + end + + # Wait for all tasks to complete + results = Task.await_many(tasks, 5_000) + + # All should succeed + assert Enum.all?(results, fn + {:ok, _} -> true + _ -> false + end) + + # Should have exactly 1 MAC address, not 10 + mac_addresses = + Repo.all( + from m in MacAddress, + where: m.device_id == ^device.id and m.mac_address == "aa:bb:cc:dd:ee:ff" and is_nil(m.vlan_id) + ) + + assert length(mac_addresses) == 1 + end + + test "concurrent neighbor upserts with nil chassis_id handled correctly", %{ + device: device, + snmp_device: snmp_device + } do + interface = + %Interface{} + |> Interface.changeset(%{ + snmp_device_id: snmp_device.id, + if_index: 1, + if_name: "eth0" + }) + |> Repo.insert!() + + # Spawn 10 concurrent workers all upserting neighbors with nil chassis_id + tasks = + for i <- 1..10 do + Task.async(fn -> + Snmp.upsert_neighbor(%{ + device_id: device.id, + interface_id: interface.id, + remote_chassis_id: nil, + protocol: "cdp", + remote_system_name: "neighbor-#{i}", + last_discovered_at: DateTime.utc_now() + }) + end) + end + + # Wait for all tasks to complete + results = Task.await_many(tasks, 5_000) + + # All should succeed + assert Enum.all?(results, fn + {:ok, _} -> true + _ -> false + end) + + # Should have exactly 1 neighbor with nil chassis_id + neighbors = + Repo.all( + from n in Neighbor, + where: n.interface_id == ^interface.id and is_nil(n.remote_chassis_id) and n.protocol == "cdp" + ) + + assert length(neighbors) == 1 + end + end end diff --git a/test/towerops/workers/device_poller_worker_test.exs b/test/towerops/workers/device_poller_worker_test.exs index 82c1e8b7..f11bb3bc 100644 --- a/test/towerops/workers/device_poller_worker_test.exs +++ b/test/towerops/workers/device_poller_worker_test.exs @@ -343,4 +343,142 @@ defmodule Towerops.Workers.DevicePollerWorkerTest do refute cancelled_jobs == [] end end + + describe "race condition handling" do + test "handles device deletion during poll gracefully", %{organization: org, site: site} do + {:ok, device} = + Devices.create_device(%{ + name: "Test Device", + ip_address: "192.168.1.100", + snmp_enabled: true, + snmp_version: "2c", + snmp_community: "public", + site_id: site.id, + organization_id: org.id + }) + + # Perform polling on device without SNMP setup (no sensors/interfaces) + # Should complete without crashing + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + + # Delete device + Devices.delete_device(device) + + # Verify polling with non-existent device doesn't crash + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + end + + test "discards poll results when device is reassigned to agent", %{ + organization: org, + site: site + } do + # Create a local agent token (is_cloud_poller defaults to false) + {:ok, agent_token, _token} = Towerops.Agents.create_agent_token(org.id, "Local Agent") + + {:ok, device} = + Devices.create_device(%{ + name: "Test Device", + ip_address: "192.168.1.100", + snmp_enabled: true, + snmp_version: "2c", + snmp_community: "public", + site_id: site.id, + organization_id: org.id + }) + + # Create SNMP device + snmp_device = + %Device{} + |> Device.changeset(%{ + device_id: device.id, + sys_name: "test-device", + sys_descr: "Test Device" + }) + |> Repo.insert!() + + # Create a sensor + %Sensor{} + |> Sensor.changeset(%{ + snmp_device_id: snmp_device.id, + sensor_type: "temperature", + sensor_index: "1", + sensor_oid: "1.3.6.1.2.1.99.1.1.1.4.1", + sensor_descr: "Test Sensor", + sensor_unit: "C", + sensor_divisor: 1, + current_reading: 25.0 + }) + |> Repo.insert!() + + # Initial state: Phoenix should poll (no agent assigned) + assert Towerops.Agents.should_phoenix_poll_device?(device) + + # Mock SNMP to simulate slow polling + poll_started = :erlang.monotonic_time(:millisecond) + + expect(SnmpMock, :get, fn _, _, _ -> + # Simulate device reassignment during poll + if :erlang.monotonic_time(:millisecond) - poll_started < 100 do + # First call - assign to agent + {:ok, _} = Towerops.Agents.assign_device_to_agent(agent_token.id, device.id) + Process.sleep(50) + end + + {:ok, 30} + end) + + expect(SnmpMock, :walk, fn _, _, _ -> {:ok, []} end) + + # Perform polling - should detect reassignment and discard results + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + + # Verify device is now assigned to local agent + device = Repo.reload!(device) + refute Towerops.Agents.should_phoenix_poll_device?(device) + end + + test "uses fresh poll interval when rescheduling after config change", %{ + organization: org, + site: site + } do + {:ok, device} = + Devices.create_device(%{ + name: "Test Device", + ip_address: "192.168.1.100", + snmp_enabled: true, + check_interval_seconds: 60, + site_id: site.id, + organization_id: org.id + }) + + # Change interval during "polling" (before scheduling next poll) + {:ok, _device} = Devices.update_device(device, %{check_interval_seconds: 300}) + + # Perform polling - should use NEW interval (300s) when scheduling next poll + # Device has no SNMP setup, so it won't actually poll but will reschedule + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + + # The code path re-fetches the device for fresh config before rescheduling + end + + test "stops rescheduling when device is deleted", %{organization: org, site: site} do + {:ok, device} = + Devices.create_device(%{ + name: "Test Device", + ip_address: "192.168.1.100", + snmp_enabled: true, + site_id: site.id, + organization_id: org.id + }) + + # Perform polling + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + + # Delete device + Devices.delete_device(device) + + # Perform polling again - should not crash and not reschedule + assert :ok = DevicePollerWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + end + end end