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 @@