defmodule Towerops.Workers.CheckExecutorWorker do @moduledoc """ Unified worker for executing all check types. Dispatches to appropriate executor based on check_type: - snmp_sensor → SnmpSensorExecutor - snmp_interface → SnmpInterfaceExecutor - snmp_processor → SnmpProcessorExecutor - snmp_storage → SnmpStorageExecutor - http → HttpExecutor - tcp → TcpExecutor - dns → DnsExecutor - ping → PingExecutor (future) Records results in check_results TimescaleDB hypertable and updates check state (OK/WARNING/CRITICAL/UNKNOWN). Self-schedules next execution based on check interval with distributed polling offsets to prevent thundering herd. """ use Oban.Worker, queue: :check_executors, max_attempts: 3 alias Towerops.Agents alias Towerops.Devices.Device alias Towerops.Monitoring alias Towerops.Monitoring.Executors.DnsExecutor alias Towerops.Monitoring.Executors.HttpExecutor alias Towerops.Monitoring.Executors.SnmpInterfaceExecutor alias Towerops.Monitoring.Executors.SnmpProcessorExecutor alias Towerops.Monitoring.Executors.SnmpSensorExecutor alias Towerops.Monitoring.Executors.SnmpStorageExecutor alias Towerops.Monitoring.Executors.TcpExecutor alias Towerops.Snmp.Client alias Towerops.Workers.PollingOffset require Logger @snmp_check_types ~w(snmp_sensor snmp_interface snmp_processor snmp_storage) @impl Oban.Worker def perform(%Oban.Job{args: %{"check_id" => check_id}}) do case Monitoring.get_check(check_id) do nil -> Logger.debug("Check #{check_id} deleted, skipping execution") :ok check -> cond do not check.enabled -> Logger.debug("Check #{check_id} disabled, skipping execution") should_skip_snmp_check?(check) -> Logger.debug("Skipping SNMP check #{check_id} - Phoenix SNMP disabled or device assigned to agent") true -> execute_and_record(check) schedule_next_check(check) end :ok end end defp should_skip_snmp_check?(%{check_type: check_type} = check) when check_type in @snmp_check_types do Client.phoenix_snmp_disabled() or (check.device_id != nil and not Agents.should_phoenix_poll_device?(%Device{id: check.device_id})) end defp should_skip_snmp_check?(_check), do: false defp execute_and_record(check) do Logger.debug("Executing check #{check.id} (#{check.check_type})") check |> dispatch_executor() |> record_result(check) end defp dispatch_executor(%{check_type: "snmp_sensor"} = check), do: SnmpSensorExecutor.execute(check) defp dispatch_executor(%{check_type: "snmp_interface"} = check), do: SnmpInterfaceExecutor.execute(check) defp dispatch_executor(%{check_type: "snmp_processor"} = check), do: SnmpProcessorExecutor.execute(check) defp dispatch_executor(%{check_type: "snmp_storage"} = check), do: SnmpStorageExecutor.execute(check) defp dispatch_executor(%{check_type: "http"} = check), do: execute_http_check(check) defp dispatch_executor(%{check_type: "tcp"} = check), do: execute_tcp_check(check) defp dispatch_executor(%{check_type: "dns"} = check), do: execute_dns_check(check) defp dispatch_executor(%{check_type: "ping"}), do: {:error, "Ping executor not yet implemented"} defp dispatch_executor(%{check_type: type}), do: {:error, "Unknown check type: #{type}"} defp record_result({:ok, %{value: value, status: status, output: output, response_time_ms: time}}, check) do Monitoring.create_check_result(%{ check_id: check.id, organization_id: check.organization_id, value: value, status: status, output: output, response_time_ms: time, checked_at: DateTime.utc_now(), agent_token_id: check.agent_token_id }) Monitoring.update_check_state(check, status, output) Logger.debug("Check #{check.id} completed: status=#{status}, value=#{value}, output=#{output}") end defp record_result({:error, reason}, check) do Logger.warning("Check #{check.id} failed: #{inspect(reason)}") Monitoring.create_check_result(%{ check_id: check.id, organization_id: check.organization_id, status: 3, output: "Error: #{inspect(reason)}", checked_at: DateTime.utc_now(), agent_token_id: check.agent_token_id }) Monitoring.update_check_state(check, 3, "Error: #{inspect(reason)}") end # Adapter for HTTP executor which returns different format defp execute_http_check(check) do case HttpExecutor.execute(check.config, check.timeout_ms) do {:ok, response_time, output} -> {:ok, %{ value: nil, status: 0, output: output, response_time_ms: response_time }} {:error, reason} -> {:error, reason} end end # Adapter for TCP executor which returns different format defp execute_tcp_check(check) do case TcpExecutor.execute(check.config, check.timeout_ms) do {:ok, response_time, output} -> {:ok, %{ value: nil, status: 0, output: output, response_time_ms: response_time }} {:error, reason} -> {:error, reason} end end # Adapter for DNS executor which returns different format defp execute_dns_check(check) do case DnsExecutor.execute(check.config, check.timeout_ms) do {:ok, response_time, output} -> {:ok, %{ value: nil, status: 0, output: output, response_time_ms: response_time }} {:error, reason} -> {:error, reason} end end defp schedule_next_check(check) do # Calculate staggered offset based on check ID offset = PollingOffset.calculate_offset(check.id, check.interval_seconds) %{check_id: check.id} |> new(schedule_in: offset) |> Oban.insert() end end