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
This commit is contained in:
parent
3eb3d4890e
commit
141c775230
2 changed files with 90 additions and 76 deletions
|
|
@ -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))
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue