diff --git a/lib/towerops/workers/discovery_worker.ex b/lib/towerops/workers/discovery_worker.ex index a2dc77ca..f9eb3ee4 100644 --- a/lib/towerops/workers/discovery_worker.ex +++ b/lib/towerops/workers/discovery_worker.ex @@ -79,18 +79,44 @@ defmodule Towerops.Workers.DiscoveryWorker do end defp attempt_agent_discovery(device, agent_token_id, tried_agents) do - case trigger_agent_discovery(device, agent_token_id) do - :ok -> - :ok + # 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 + ) - {:error, :timeout} -> - Logger.warning( - "Agent discovery timeout, trying next cloud poller", - device_id: device.id, - failed_agent_id: agent_token_id - ) + case trigger_agent_discovery(device, agent_token_id) do + :ok -> + :ok - attempt_cloud_poller_discovery(device, tried_agents) + {: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 @@ -116,19 +142,23 @@ defmodule Towerops.Workers.DiscoveryWorker do end defp get_next_cloud_poller(tried_agents) do - cutoff = DateTime.add(DateTime.utc_now(), -@agent_online_threshold_minutes, :minute) - Agents.list_cloud_pollers() |> Enum.reject(fn agent -> agent.id in tried_agents end) - |> Enum.find(fn agent -> - agent.enabled && agent.last_seen_at && DateTime.after?(agent.last_seen_at, cutoff) - 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 @@ -140,44 +170,60 @@ defmodule Towerops.Workers.DiscoveryWorker do # Wait for agent to complete discovery (with timeout) # The agent will update last_discovery_at when done - wait_for_agent_discovery(device, @agent_discovery_timeout_ms) + wait_for_agent_discovery(device, agent_token_id, @agent_discovery_timeout_ms) end - defp wait_for_agent_discovery(device, timeout_ms) do + 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, initial_discovery_at, start_time, timeout_ms) + wait_loop(device.id, agent_token_id, initial_discovery_at, start_time, timeout_ms) end - defp wait_loop(device_id, initial_discovery_at, start_time, timeout_ms) do + defp wait_loop(device_id, agent_token_id, initial_discovery_at, start_time, timeout_ms) do elapsed = System.monotonic_time(:millisecond) - start_time - if elapsed >= timeout_ms do - Logger.warning( - "Agent discovery timeout", - device_id: device_id, - elapsed_ms: elapsed - ) - - {:error, :timeout} - else - # Check if discovery completed (last_discovery_at changed) - 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, initial_discovery_at, start_time, timeout_ms) - else - Logger.info( - "Agent discovery completed successfully", + 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 ) - :ok - end + {: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