diff --git a/lib/towerops/application.ex b/lib/towerops/application.ex index 64e95638..b2df27dd 100644 --- a/lib/towerops/application.ex +++ b/lib/towerops/application.ex @@ -101,7 +101,9 @@ defmodule Towerops.Application do Towerops.Profiles.YamlProfiles, # SnmpKit services for SNMP operations (used for SNMP protocol, not MIB resolution) SnmpKit.SnmpMgr.Config, - SnmpKit.SnmpMgr.MIB + SnmpKit.SnmpMgr.MIB, + # Task supervisor for fire-and-forget background tasks (e.g. polling data processing) + {Task.Supervisor, name: Towerops.TaskSupervisor} ] ++ dev_only_workers() ++ background_workers() ++ diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index fdddda4e..7f837560 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -171,6 +171,18 @@ defmodule Towerops.Workers.DevicePollerWorker do now = DateTime.truncate(DateTime.utc_now(), :second) # Run all polling operations in parallel using async tasks + task_names = [ + "sensors", + "state_sensors", + "interfaces", + "interface_changes", + "neighbors", + "arp", + "mac", + "processors", + "storage" + ] + tasks = [ Task.async(fn -> poll_device_sensors(device, snmp_device, client_opts, now) end), Task.async(fn -> poll_device_state_sensors(device, snmp_device, client_opts, now) end), @@ -183,24 +195,30 @@ defmodule Towerops.Workers.DevicePollerWorker do Task.async(fn -> poll_device_storage(device, snmp_device, client_opts, now) end) ] - # Wait for all tasks with timeout, handle failures gracefully - _results = - tasks - |> Task.yield_many(30_000) - |> Enum.map(fn {task, result} -> - case result do - {:ok, value} -> - {:ok, value} + # Wait for all tasks with timeout, log failures + tasks + |> Task.yield_many(30_000) + |> Enum.zip(task_names) + |> Enum.each(fn {{task, result}, name} -> + case result do + {:ok, _value} -> + :ok - {:exit, reason} -> - Logger.error("Polling task failed: #{inspect(reason)}") - {:error, reason} + {:exit, reason} -> + Logger.warning("Polling task '#{name}' crashed for #{device.name}: #{inspect(reason)}", + device_id: device.id, + task: name + ) - nil -> - Task.shutdown(task, :brutal_kill) - {:error, :timeout} - end - end) + nil -> + Task.shutdown(task, :brutal_kill) + + Logger.warning("Polling task '#{name}' timed out for #{device.name}", + device_id: device.id, + task: name + ) + end + end) # Re-check assignment before processing results # Device could have been reassigned to an agent during the 30-second polling window @@ -549,12 +567,16 @@ defmodule Towerops.Workers.DevicePollerWorker do fn sensor -> result = poll_sensor_value(sensor, client_opts) handle_sensor_poll_result(sensor, result, timestamp) + sensor.sensor_descr end, max_concurrency: 2, timeout: 40_000, on_timeout: :kill_task ) - |> Stream.run() + |> Enum.each(fn + {:ok, _} -> :ok + {:exit, reason} -> Logger.warning("Sensor poll task failed: #{inspect(reason)}") + end) end defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id) do @@ -563,12 +585,16 @@ defmodule Towerops.Workers.DevicePollerWorker do fn state_sensor -> result = poll_state_sensor_value(state_sensor, client_opts) handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id) + state_sensor.sensor_descr end, max_concurrency: 2, timeout: 40_000, on_timeout: :kill_task ) - |> Stream.run() + |> Enum.each(fn + {:ok, _} -> :ok + {:exit, reason} -> Logger.warning("State sensor poll task failed: #{inspect(reason)}") + end) end defp poll_state_sensor_value(state_sensor, client_opts) do diff --git a/lib/towerops/workers/discovery_worker.ex b/lib/towerops/workers/discovery_worker.ex index 37d6d650..86f0c3d5 100644 --- a/lib/towerops/workers/discovery_worker.ex +++ b/lib/towerops/workers/discovery_worker.ex @@ -19,7 +19,7 @@ defmodule Towerops.Workers.DiscoveryWorker do use Oban.Worker, queue: :discovery, unique: [period: 60, states: [:available, :scheduled, :executing]], - max_attempts: 1 + max_attempts: 3 alias Towerops.Agents alias Towerops.Devices @@ -39,7 +39,7 @@ defmodule Towerops.Workers.DiscoveryWorker do @agent_online_threshold_minutes 10 @impl Oban.Worker - @spec perform(Oban.Job.t()) :: :ok | :discard + @spec perform(Oban.Job.t()) :: :ok | :discard | {:error, term()} def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do Events.broadcast_job_event(job, :started) start_time = System.monotonic_time(:second) @@ -63,31 +63,37 @@ defmodule Towerops.Workers.DiscoveryWorker do :discard -> Events.broadcast_job_event(job, :completed, %{duration: duration}) + + {:error, reason} -> + Events.broadcast_job_event(job, :failed, %{ + error: inspect(reason), + duration: duration + }) end result rescue error -> duration = System.monotonic_time(:second) - start_time + error_string = Exception.format(:error, error, __STACKTRACE__) Logger.error( - "Discovery job crashed unexpectedly, discarding to prevent retries", + "Discovery job crashed unexpectedly", device_id: device_id, - error: Exception.format(:error, error, __STACKTRACE__) + error: error_string ) Events.broadcast_job_event(job, :failed, %{ - error: Exception.format(:error, error, __STACKTRACE__), + error: error_string, duration: duration }) + # Crashes are not retryable - they indicate a code bug, not a transient issue :discard end end defp perform_device_discovery(device, device_id) do - # Discovery is a one-shot operation - always discard on completion (success or failure) - # This prevents stuck jobs from retrying indefinitely # Use Task.async/await with timeout to prevent jobs from hanging indefinitely task = Task.async(fn -> perform_discovery(device) end) @@ -96,22 +102,32 @@ defmodule Towerops.Workers.DiscoveryWorker do :ok {:ok, {:error, reason}} -> - Logger.warning( - "Discovery failed for device #{device_id}, discarding job", - device_id: device_id, - error: reason - ) + if retryable_error?(reason) do + Logger.warning( + "Discovery failed for device #{device_id} with transient error, will retry", + device_id: device_id, + error: inspect(reason) + ) - :discard + {:error, reason} + else + Logger.warning( + "Discovery failed for device #{device_id} with permanent error, discarding", + device_id: device_id, + error: inspect(reason) + ) + + :discard + end nil -> Logger.error( - "Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries", + "Discovery job timeout after #{@job_timeout_ms}ms, will retry", device_id: device_id, timeout_ms: @job_timeout_ms ) - :discard + {:error, :job_timeout} end end @@ -254,6 +270,20 @@ defmodule Towerops.Workers.DiscoveryWorker do end end + @doc """ + Classifies whether a discovery error is transient (retryable) or permanent. + + Transient errors (network issues, timeouts) will be retried by Oban. + Permanent errors (device not found, invalid config) are discarded immediately. + Unknown errors default to retryable to avoid silent data loss. + """ + @spec retryable_error?(atom() | term()) :: boolean() + def retryable_error?(:device_not_found), do: false + def retryable_error?(:invalid_config), do: false + def retryable_error?(:no_snmp_credentials), do: false + def retryable_error?(:device_deleted), do: false + def retryable_error?(_reason), do: true + # Private helpers defp get_assigned_agent(device) do diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index 40cd3bd9..fe90e240 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -1138,7 +1138,7 @@ defmodule ToweropsWeb.AgentChannel do device_name = device.name snmp_device_id = device.snmp_device.id - spawn(fn -> + Task.Supervisor.start_child(Towerops.TaskSupervisor, fn -> try do alias Towerops.Snmp.ArpDiscovery alias Towerops.Snmp.MacDiscovery diff --git a/test/towerops/application_test.exs b/test/towerops/application_test.exs new file mode 100644 index 00000000..017d6b7d --- /dev/null +++ b/test/towerops/application_test.exs @@ -0,0 +1,9 @@ +defmodule Towerops.ApplicationTest do + use ExUnit.Case, async: true + + describe "supervision tree" do + test "Towerops.TaskSupervisor is running" do + assert Process.whereis(Towerops.TaskSupervisor) + end + end +end diff --git a/test/towerops/workers/discovery_worker_test.exs b/test/towerops/workers/discovery_worker_test.exs index e8b26bf7..fb477916 100644 --- a/test/towerops/workers/discovery_worker_test.exs +++ b/test/towerops/workers/discovery_worker_test.exs @@ -78,14 +78,14 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => non_existent_id}}) end - test "returns discard when discovery fails", %{device: device} do - # Mock SNMP failure (test_connection fails immediately) - expect(SnmpMock, :get, fn _target, _oid, _opts -> + test "returns error for transient failures to allow retry", %{device: device} do + # Mock SNMP failure with a transient error (timeout) + stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end) - # Discovery worker always returns :discard to prevent infinite retries - assert :discard = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) + # Transient failures should return {:error, reason} so Oban can retry + assert {:error, _reason} = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}}) end test "successfully completes discovery with proper mocks", %{device: device} do @@ -119,7 +119,7 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do test "logs error when discovery fails", %{device: device} do import ExUnit.CaptureLog - expect(SnmpMock, :get, fn _target, _oid, _opts -> + stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :network_unreachable} end) @@ -131,4 +131,34 @@ defmodule Towerops.Workers.DiscoveryWorkerTest do assert log =~ "Direct SNMP discovery failed:" end end + + describe "retryable_error?/1" do + test "timeout is retryable" do + assert DiscoveryWorker.retryable_error?(:timeout) + end + + test "network_unreachable is retryable" do + assert DiscoveryWorker.retryable_error?(:network_unreachable) + end + + test "connection_refused is retryable" do + assert DiscoveryWorker.retryable_error?(:connection_refused) + end + + test "econnrefused is retryable" do + assert DiscoveryWorker.retryable_error?(:econnrefused) + end + + test "device_not_found is not retryable" do + refute DiscoveryWorker.retryable_error?(:device_not_found) + end + + test "invalid_config is not retryable" do + refute DiscoveryWorker.retryable_error?(:invalid_config) + end + + test "unknown errors are retryable by default" do + assert DiscoveryWorker.retryable_error?(:some_unknown_error) + end + end end