409 lines
12 KiB
Elixir
409 lines
12 KiB
Elixir
defmodule Towerops.Workers.DiscoveryWorker do
|
|
@moduledoc """
|
|
Oban worker for SNMP device discovery.
|
|
|
|
Enqueued when:
|
|
- A new device is created with SNMP enabled
|
|
- An existing device's SNMP configuration is changed
|
|
- User manually triggers discovery via "Rediscover Device" button
|
|
|
|
## Agent-based Discovery
|
|
|
|
If the device has an assigned agent, discovery is delegated to that agent.
|
|
The agent receives a discovery job via PubSub broadcast, performs SNMP operations,
|
|
and sends results back to the server.
|
|
|
|
If no agent is assigned or the agent is offline, discovery falls back to direct
|
|
SNMP operations from the Phoenix cluster.
|
|
"""
|
|
use Oban.Worker,
|
|
queue: :discovery,
|
|
unique: [period: 60, states: [:available, :scheduled, :executing]],
|
|
max_attempts: 1
|
|
|
|
alias Towerops.Agents
|
|
alias Towerops.Devices
|
|
alias Towerops.Snmp
|
|
|
|
require Logger
|
|
|
|
# Timeout for waiting for agent-based discovery (60 seconds)
|
|
@agent_discovery_timeout_ms 60_000
|
|
|
|
# Consider agents online if they checked in within last 10 minutes
|
|
@agent_online_threshold_minutes 10
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"device_id" => device_id}}) do
|
|
case Devices.get_device_with_details(device_id) do
|
|
nil ->
|
|
Logger.warning("Device #{device_id} not found, discarding discovery job")
|
|
:discard
|
|
|
|
device ->
|
|
perform_discovery(device)
|
|
end
|
|
end
|
|
|
|
defp perform_discovery(device) do
|
|
start_time = System.monotonic_time(:millisecond)
|
|
|
|
Logger.info(
|
|
"DISCOVERY STARTED: #{device.name} (#{device.ip_address})",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
snmp_enabled: device.snmp_enabled,
|
|
snmp_port: device.snmp_port || 161
|
|
)
|
|
|
|
result = execute_discovery_with_agent_routing(device)
|
|
duration_ms = System.monotonic_time(:millisecond) - start_time
|
|
|
|
log_discovery_result(device, result, duration_ms)
|
|
result
|
|
end
|
|
|
|
defp execute_discovery_with_agent_routing(device) do
|
|
case get_assigned_agent(device) do
|
|
{agent_token_id, source} when not is_nil(agent_token_id) ->
|
|
handle_assigned_agent_discovery(device, agent_token_id, source)
|
|
|
|
{nil, :none} ->
|
|
Logger.info(
|
|
"Starting SNMP discovery for device #{device.id} with no assigned agent, will try cloud pollers then cluster fallback",
|
|
device_id: device.id
|
|
)
|
|
|
|
attempt_cloud_poller_discovery(device, [])
|
|
end
|
|
end
|
|
|
|
defp handle_assigned_agent_discovery(device, agent_token_id, source) do
|
|
agent = Agents.get_agent_token!(agent_token_id)
|
|
log_agent_assignment(device, agent, agent_token_id, source)
|
|
attempt_agent_discovery(device, agent_token_id, [agent_token_id])
|
|
rescue
|
|
Ecto.NoResultsError ->
|
|
Logger.warning(
|
|
"Agent token #{agent_token_id} no longer exists (#{source} assignment), falling back to cloud pollers",
|
|
device_id: device.id,
|
|
agent_token_id: agent_token_id,
|
|
source: source
|
|
)
|
|
|
|
attempt_cloud_poller_discovery(device, [])
|
|
end
|
|
|
|
defp log_agent_assignment(device, agent, agent_token_id, source) do
|
|
poller_type = if agent.cloud_poller, do: "cloud poller", else: "user's agent"
|
|
|
|
Logger.info(
|
|
"Agent assignment found: using #{poller_type} '#{agent.name}' (assigned via #{source})",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
agent_token_id: agent_token_id,
|
|
agent_name: agent.name,
|
|
cloud_poller: agent.cloud_poller,
|
|
assignment_source: source,
|
|
agent_enabled: agent.enabled,
|
|
agent_last_seen: agent.last_seen_at
|
|
)
|
|
end
|
|
|
|
defp log_discovery_result(device, :ok, duration_ms) do
|
|
Logger.info(
|
|
"DISCOVERY COMPLETED: #{device.name} (#{device.ip_address}) in #{duration_ms}ms",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
duration_ms: duration_ms
|
|
)
|
|
end
|
|
|
|
defp log_discovery_result(device, {:error, reason}, duration_ms) do
|
|
Logger.error(
|
|
"DISCOVERY FAILED: #{device.name} (#{device.ip_address}) after #{duration_ms}ms - #{inspect(reason)}",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
duration_ms: duration_ms,
|
|
error: reason
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Enqueues a discovery job for a device.
|
|
|
|
Uses Oban unique constraint to prevent duplicate jobs within 60 seconds.
|
|
"""
|
|
def enqueue(device_id) do
|
|
# Log who's enqueueing this to track down runaway triggers
|
|
caller_module =
|
|
self()
|
|
|> Process.info(:current_stacktrace)
|
|
|> elem(1)
|
|
|> Enum.at(1)
|
|
|> elem(0)
|
|
|
|
result =
|
|
%{device_id: device_id}
|
|
|> new()
|
|
|> Oban.insert()
|
|
|
|
case result do
|
|
{:ok, %Oban.Job{conflict?: true}} ->
|
|
_ =
|
|
Logger.debug(
|
|
"Discovery job already enqueued for device, skipping duplicate",
|
|
device_id: device_id,
|
|
caller: caller_module
|
|
)
|
|
|
|
:ok
|
|
|
|
{:ok, _job} ->
|
|
_ =
|
|
Logger.debug(
|
|
"Discovery job enqueued for device",
|
|
device_id: device_id,
|
|
caller: caller_module
|
|
)
|
|
|
|
:ok
|
|
|
|
{:error, _} = error ->
|
|
_ =
|
|
Logger.warning(
|
|
"Failed to enqueue discovery job",
|
|
device_id: device_id,
|
|
caller: caller_module
|
|
)
|
|
|
|
error
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
# Private helpers
|
|
|
|
defp get_assigned_agent(device) do
|
|
Agents.get_effective_agent_token_with_source(device)
|
|
end
|
|
|
|
defp attempt_agent_discovery(device, agent_token_id, tried_agents) do
|
|
# First check if the agent is online based on recent heartbeat
|
|
if agent_online?(agent_token_id) do
|
|
Logger.info(
|
|
"Agent is online, delegating discovery",
|
|
device_id: device.id,
|
|
agent_token_id: agent_token_id
|
|
)
|
|
|
|
case trigger_agent_discovery(device, agent_token_id) do
|
|
:ok ->
|
|
:ok
|
|
|
|
{:error, :timeout} ->
|
|
Logger.warning(
|
|
"Agent discovery timeout, trying next cloud poller",
|
|
device_id: device.id,
|
|
failed_agent_id: agent_token_id
|
|
)
|
|
|
|
attempt_cloud_poller_discovery(device, tried_agents)
|
|
|
|
{:error, :agent_offline} ->
|
|
Logger.warning(
|
|
"Agent went offline during discovery, trying next cloud poller",
|
|
device_id: device.id,
|
|
failed_agent_id: agent_token_id
|
|
)
|
|
|
|
attempt_cloud_poller_discovery(device, tried_agents)
|
|
end
|
|
else
|
|
Logger.warning(
|
|
"Agent is offline, skipping to next cloud poller",
|
|
device_id: device.id,
|
|
offline_agent_id: agent_token_id
|
|
)
|
|
|
|
attempt_cloud_poller_discovery(device, tried_agents)
|
|
end
|
|
end
|
|
|
|
defp attempt_cloud_poller_discovery(device, tried_agents) do
|
|
Logger.info(
|
|
"Checking for cloud pollers...",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
tried_agents: tried_agents
|
|
)
|
|
|
|
case get_next_cloud_poller(tried_agents) do
|
|
nil ->
|
|
Logger.info(
|
|
"No cloud pollers available, using direct SNMP from Phoenix cluster",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
tried_agents: tried_agents
|
|
)
|
|
|
|
perform_direct_discovery(device)
|
|
|
|
next_agent_id ->
|
|
agent = Agents.get_agent_token!(next_agent_id)
|
|
|
|
Logger.info(
|
|
"Trying global cloud poller '#{agent.name}'",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
cloud_poller_id: next_agent_id,
|
|
cloud_poller_name: agent.name,
|
|
agent_enabled: agent.enabled,
|
|
agent_last_seen: agent.last_seen_at
|
|
)
|
|
|
|
attempt_agent_discovery(device, next_agent_id, [next_agent_id | tried_agents])
|
|
end
|
|
end
|
|
|
|
defp get_next_cloud_poller(tried_agents) do
|
|
Agents.list_cloud_pollers()
|
|
|> Enum.reject(fn agent -> agent.id in tried_agents end)
|
|
|> Enum.find(fn agent -> agent_online?(agent.id) end)
|
|
|> case do
|
|
nil -> nil
|
|
agent -> agent.id
|
|
end
|
|
end
|
|
|
|
defp agent_online?(agent_token_id) do
|
|
agent = Agents.get_agent_token!(agent_token_id)
|
|
cutoff = DateTime.add(DateTime.utc_now(), -@agent_online_threshold_minutes, :minute)
|
|
agent.enabled && agent.last_seen_at && DateTime.after?(agent.last_seen_at, cutoff)
|
|
rescue
|
|
Ecto.NoResultsError -> false
|
|
end
|
|
|
|
defp trigger_agent_discovery(device, agent_token_id) do
|
|
# Broadcast discovery request to agent's channel
|
|
# The agent will receive this via PubSub and execute discovery
|
|
_ =
|
|
Logger.info(
|
|
"Sending discovery request to remote agent via PubSub",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
agent_token_id: agent_token_id,
|
|
pubsub_channel: "agent:#{agent_token_id}:discovery"
|
|
)
|
|
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token_id}:discovery",
|
|
{:discovery_requested, device.id}
|
|
)
|
|
|
|
Logger.info(
|
|
"Discovery request sent, waiting up to #{@agent_discovery_timeout_ms}ms for agent to respond",
|
|
device_id: device.id,
|
|
agent_token_id: agent_token_id,
|
|
timeout_ms: @agent_discovery_timeout_ms
|
|
)
|
|
|
|
# Wait for agent to complete discovery (with timeout)
|
|
# The agent will update last_discovery_at when done
|
|
wait_for_agent_discovery(device, agent_token_id, @agent_discovery_timeout_ms)
|
|
end
|
|
|
|
defp wait_for_agent_discovery(device, agent_token_id, timeout_ms) do
|
|
start_time = System.monotonic_time(:millisecond)
|
|
initial_discovery_at = device.last_discovery_at
|
|
|
|
wait_loop(device.id, agent_token_id, initial_discovery_at, start_time, timeout_ms)
|
|
end
|
|
|
|
defp wait_loop(device_id, agent_token_id, initial_discovery_at, start_time, timeout_ms) do
|
|
elapsed = System.monotonic_time(:millisecond) - start_time
|
|
|
|
cond do
|
|
# Check if agent went offline during discovery
|
|
!agent_online?(agent_token_id) ->
|
|
Logger.warning(
|
|
"Agent went offline during discovery",
|
|
device_id: device_id,
|
|
agent_token_id: agent_token_id,
|
|
elapsed_ms: elapsed
|
|
)
|
|
|
|
{:error, :agent_offline}
|
|
|
|
# Check if we've exceeded timeout
|
|
elapsed >= timeout_ms ->
|
|
Logger.warning(
|
|
"Agent discovery timeout - no response within #{timeout_ms}ms",
|
|
device_id: device_id,
|
|
agent_token_id: agent_token_id,
|
|
elapsed_ms: elapsed
|
|
)
|
|
|
|
{:error, :timeout}
|
|
|
|
# Check if discovery completed (last_discovery_at changed)
|
|
true ->
|
|
device = Devices.get_device!(device_id)
|
|
|
|
if device.last_discovery_at == initial_discovery_at do
|
|
# Wait 500ms before checking again
|
|
Process.sleep(500)
|
|
wait_loop(device_id, agent_token_id, initial_discovery_at, start_time, timeout_ms)
|
|
else
|
|
Logger.info(
|
|
"Agent discovery completed successfully",
|
|
device_id: device_id,
|
|
agent_token_id: agent_token_id,
|
|
elapsed_ms: elapsed
|
|
)
|
|
|
|
:ok
|
|
end
|
|
end
|
|
end
|
|
|
|
defp perform_direct_discovery(device) do
|
|
Logger.info(
|
|
"Starting direct SNMP discovery from Phoenix cluster (no remote agents available)",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address
|
|
)
|
|
|
|
case Snmp.discover_device(device) do
|
|
{:ok, _snmp_device} ->
|
|
Logger.info(
|
|
"Direct SNMP discovery completed successfully",
|
|
device_id: device.id,
|
|
device_name: device.name
|
|
)
|
|
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error(
|
|
"Direct SNMP discovery failed: #{inspect(reason)}",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
device_ip: device.ip_address,
|
|
error: reason
|
|
)
|
|
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|