From 141c775230c2d0901962a5c554822168d41d36c8 Mon Sep 17 00:00:00 2001 From: mayor Date: Sat, 7 Feb 2026 09:17:09 -0600 Subject: [PATCH] fix: discovery retry, disabled assignment cascade, and silent spawn crashes - Move last_discovery_at update to after successful processing so failed discoveries are retried on the next poll cycle instead of being blocked for 24 hours - Add disabled assignment check to agent polling target query so devices with enabled=false assignments don't leak through site/org cascade - Wrap spawned polling data processing in try/rescue so crashes are logged instead of silently disappearing --- lib/towerops/agents.ex | 20 ++- lib/towerops_web/channels/agent_channel.ex | 146 +++++++++++---------- 2 files changed, 90 insertions(+), 76 deletions(-) diff --git a/lib/towerops/agents.ex b/lib/towerops/agents.ex index 0ad08aeb..3a438957 100644 --- a/lib/towerops/agents.ex +++ b/lib/towerops/agents.ex @@ -429,6 +429,12 @@ defmodule Towerops.Agents do join: o in assoc(e, :organization), left_join: aa in AgentAssignment, on: aa.device_id == e.id and aa.enabled == true, + # Detect disabled assignments for this specific agent so cascade + # doesn't route devices that were explicitly disabled + left_join: disabled in AgentAssignment, + on: + disabled.device_id == e.id and disabled.agent_token_id == ^agent_token_id and + disabled.enabled == false, where: e.snmp_enabled == true, preload: [ :agent_assignments, @@ -443,19 +449,21 @@ defmodule Towerops.Agents do |> Repo.all() end - # Adds where clause to match devices assigned to this agent - # Supports devices assigned directly to org (no site) or to a site + # Adds where clause to match devices assigned to this agent. + # Supports direct assignment, site cascade, org cascade, and global default. + # A disabled assignment for this agent blocks cascade to prevent polling + # devices that were explicitly disabled. # 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, + from([e, s, o, aa, disabled] in query, where: aa.agent_token_id == ^agent_token_id or - (is_nil(aa.agent_token_id) and not is_nil(s.id) and + (is_nil(aa.agent_token_id) and is_nil(disabled.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(aa.agent_token_id) and is_nil(disabled.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_global_default and is_nil(aa.agent_token_id) and is_nil(disabled.id) and (is_nil(s.id) or is_nil(s.agent_token_id)) and is_nil(o.default_agent_token_id)) ) diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index b483f429..e210d67a 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -863,22 +863,12 @@ defmodule ToweropsWeb.AgentChannel do end defp process_discovery_result(device, result, socket) do - # Agent has completed discovery and sent back SNMP data - # Update last_discovery_at to signal completion to DiscoveryWorker Logger.info("Discovery results received from agent for #{device.name}, processing data") - # Update device timestamp to mark discovery as complete - case Devices.update_device(device, %{last_discovery_at: DateTime.utc_now()}) do - {:ok, updated_device} -> - Logger.info("Updated last_discovery_at for device #{device.name}") - - # Process the SNMP data from agent's discovery queries - # The agent sends back oid_values map with all the collected data - process_discovery_data(updated_device, result, socket) - - {:error, changeset} -> - Logger.error("Failed to update last_discovery_at for device #{device.name}: #{inspect(changeset)}") - end + # Process discovery data first, only update last_discovery_at on success. + # This ensures failed discoveries are retried on the next poll cycle + # instead of being blocked for 24 hours. + process_discovery_data(device, result, socket) end defp process_discovery_data(device, result, socket) do @@ -901,6 +891,10 @@ defmodule ToweropsWeb.AgentChannel do # Process full discovery using agent data case AgentDiscovery.process_agent_discovery(device, oid_values) do {:ok, _discovered_device} -> + # Only mark discovery timestamp on success so failures are retried + # on the next poll cycle instead of waiting 24 hours + _ = Devices.update_device(device, %{last_discovery_at: DateTime.utc_now()}) + Logger.info("Agent discovery completed", device_id: device.id, device_name: device.name @@ -933,8 +927,6 @@ defmodule ToweropsWeb.AgentChannel do error_reason: inspect(reason) ) - # Don't update last_discovery_at - DiscoveryWorker will retry - # with direct discovery as fallback {:error, reason} end end @@ -971,76 +963,90 @@ defmodule ToweropsWeb.AgentChannel do oid_map: oid_values ] - # Reuse existing polling functions with Replay adapter + # Reuse existing polling functions with Replay adapter. + # Runs in a spawned process to avoid blocking the channel. + device_id = device.id + device_name = device.name + snmp_device_id = device.snmp_device.id + spawn(fn -> - alias Towerops.Snmp.Profiles.Base - alias Towerops.Snmp.NeighborDiscovery + try do + alias Towerops.Snmp.Profiles.Base + alias Towerops.Snmp.NeighborDiscovery - # Fetch interfaces once for reuse across multiple operations - interfaces = Towerops.Snmp.list_interfaces(device.snmp_device.id) + # Fetch interfaces once for reuse across multiple operations + interfaces = Towerops.Snmp.list_interfaces(snmp_device_id) - # Add device_id to interfaces for neighbor discovery (required by build_neighbor_record) - interfaces_with_device_id = Enum.map(interfaces, &Map.put(&1, :device_id, device.id)) + # Add device_id to interfaces for neighbor discovery (required by build_neighbor_record) + interfaces_with_device_id = Enum.map(interfaces, &Map.put(&1, :device_id, device_id)) - # Process neighbors (LLDP/CDP) - case NeighborDiscovery.discover_neighbors(client_opts, interfaces_with_device_id) do - {:ok, neighbors} when neighbors != [] -> - Towerops.Snmp.Discovery.save_neighbors(device.id, neighbors) + # Process neighbors (LLDP/CDP) + case NeighborDiscovery.discover_neighbors(client_opts, interfaces_with_device_id) do + {:ok, neighbors} when neighbors != [] -> + Towerops.Snmp.Discovery.save_neighbors(device_id, neighbors) - _ -> - :ok - end + _ -> + :ok + end - # Process ARP entries - case Towerops.Snmp.ArpDiscovery.discover_arp_table(client_opts) do - {:ok, arp_entries} when arp_entries != [] -> - Towerops.Snmp.Discovery.save_arp_entries(device.id, arp_entries, interfaces) + # Process ARP entries + case Towerops.Snmp.ArpDiscovery.discover_arp_table(client_opts) do + {:ok, arp_entries} when arp_entries != [] -> + Towerops.Snmp.Discovery.save_arp_entries(device_id, arp_entries, interfaces) - _ -> - :ok - end + _ -> + :ok + end - # Process MAC addresses - case Towerops.Snmp.MacDiscovery.discover_mac_table(client_opts) do - {:ok, mac_addresses} when mac_addresses != [] -> - Towerops.Snmp.upsert_mac_addresses(device.id, mac_addresses, interfaces) + # Process MAC addresses + case Towerops.Snmp.MacDiscovery.discover_mac_table(client_opts) do + {:ok, mac_addresses} when mac_addresses != [] -> + Towerops.Snmp.upsert_mac_addresses(device_id, mac_addresses, interfaces) - _ -> - :ok - end + _ -> + :ok + end - # Process IP addresses - case Base.discover_all_ip_addresses(client_opts) do - {:ok, ip_addresses} when ip_addresses != [] -> - discovered_device = %{ - device_id: device.id, - interfaces: interfaces - } + # Process IP addresses + case Base.discover_all_ip_addresses(client_opts) do + {:ok, ip_addresses} when ip_addresses != [] -> + discovered_device = %{ + device_id: device_id, + interfaces: interfaces + } - Towerops.Snmp.Discovery.sync_ip_addresses(discovered_device, ip_addresses) + Towerops.Snmp.Discovery.sync_ip_addresses(discovered_device, ip_addresses) - _ -> - :ok - end + _ -> + :ok + end - # Process processors - case Base.discover_processors(client_opts) do - {:ok, processors} when processors != [] -> - discovered_device = %{id: device.snmp_device.id} - Towerops.Snmp.Discovery.sync_processors(discovered_device, processors) + # Process processors + case Base.discover_processors(client_opts) do + {:ok, processors} when processors != [] -> + discovered_device = %{id: snmp_device_id} + Towerops.Snmp.Discovery.sync_processors(discovered_device, processors) - _ -> - :ok - end + _ -> + :ok + end - # Process storage - case Base.discover_storage(client_opts) do - {:ok, storage} when storage != [] -> - discovered_device = %{id: device.snmp_device.id} - Towerops.Snmp.Discovery.sync_storage(discovered_device, storage) + # Process storage + case Base.discover_storage(client_opts) do + {:ok, storage} when storage != [] -> + discovered_device = %{id: snmp_device_id} + Towerops.Snmp.Discovery.sync_storage(discovered_device, storage) - _ -> - :ok + _ -> + :ok + end + rescue + e -> + Logger.error( + "Polling data processing crashed for device #{device_name}: #{Exception.message(e)}", + device_id: device_id, + error: Exception.format(:error, e, __STACKTRACE__) + ) end end) end