111 lines
3.4 KiB
Elixir
111 lines
3.4 KiB
Elixir
defmodule Towerops.Workers.JobHealthCheckWorker do
|
|
@moduledoc """
|
|
Oban worker that ensures all enabled jobs have active workers.
|
|
|
|
Runs every 10 minutes via Oban.Plugins.Cron as a safety net to:
|
|
1. Find devices with monitoring_enabled=true but no active DeviceMonitorWorker job
|
|
2. Find enabled checks with no active CheckExecutorWorker job
|
|
3. Automatically create missing jobs
|
|
4. Log when missing jobs are found and recovered
|
|
|
|
This handles edge cases where jobs might get cancelled unexpectedly or database
|
|
inconsistencies occur during deployments or pod failures.
|
|
"""
|
|
use Oban.Pro.Worker, queue: :maintenance
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Devices
|
|
alias Towerops.Monitoring
|
|
alias Towerops.Monitoring.Check
|
|
alias Towerops.Repo
|
|
alias Towerops.Snmp.Client
|
|
alias Towerops.Workers.DeviceMonitorWorker
|
|
|
|
require Logger
|
|
|
|
@impl Oban.Worker
|
|
@spec perform(Oban.Job.t()) :: :ok
|
|
def perform(%Oban.Job{}) do
|
|
check_recoveries = recover_missing_check_executor_jobs()
|
|
|
|
if check_recoveries > 0 do
|
|
Logger.warning("Job health check recovered #{check_recoveries} missing check executor job(s)")
|
|
end
|
|
|
|
if Client.phoenix_snmp_disabled() do
|
|
Logger.debug("Job health check: skipping monitor job recovery (Phoenix SNMP polling disabled)")
|
|
else
|
|
monitor_recoveries = recover_missing_monitor_jobs()
|
|
|
|
if monitor_recoveries > 0 do
|
|
Logger.warning("Job health check recovered #{monitor_recoveries} missing monitor job(s)")
|
|
end
|
|
end
|
|
|
|
if check_recoveries == 0 do
|
|
Logger.debug("Job health check completed: all jobs healthy")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
defp recover_missing_check_executor_jobs do
|
|
active_check_ids = get_active_check_executor_check_ids()
|
|
|
|
orphaned_checks =
|
|
Check
|
|
|> where([c], c.enabled == true)
|
|
|> Repo.all()
|
|
|> Enum.reject(&MapSet.member?(active_check_ids, &1.id))
|
|
|
|
Enum.each(orphaned_checks, fn check ->
|
|
Logger.debug(
|
|
"Check '#{check.name}' (#{check.check_type}) enabled but has no active executor job, scheduling",
|
|
check_id: check.id,
|
|
check_type: check.check_type
|
|
)
|
|
|
|
Monitoring.schedule_check(check)
|
|
end)
|
|
|
|
length(orphaned_checks)
|
|
end
|
|
|
|
defp get_active_check_executor_check_ids do
|
|
Oban.Job
|
|
|> where([j], j.worker == "Towerops.Workers.CheckExecutorWorker")
|
|
|> where([j], j.state in ["available", "scheduled", "executing", "retryable"])
|
|
|> select([j], fragment("args->>'check_id'"))
|
|
|> Repo.all()
|
|
|> MapSet.new()
|
|
end
|
|
|
|
defp recover_missing_monitor_jobs do
|
|
devices_with_monitoring = Devices.list_monitored_devices()
|
|
active_monitor_job_device_ids = get_active_monitor_job_device_ids()
|
|
|
|
missing_monitor_devices = Enum.reject(devices_with_monitoring, &MapSet.member?(active_monitor_job_device_ids, &1.id))
|
|
|
|
Enum.each(missing_monitor_devices, fn device ->
|
|
Logger.debug(
|
|
"Device '#{device.name}' has monitoring_enabled=true but no active monitor job, creating job",
|
|
device_id: device.id,
|
|
device_name: device.name
|
|
)
|
|
|
|
DeviceMonitorWorker.start_monitoring(device.id)
|
|
end)
|
|
|
|
length(missing_monitor_devices)
|
|
end
|
|
|
|
defp get_active_monitor_job_device_ids do
|
|
Oban.Job
|
|
|> where([j], j.worker == "Towerops.Workers.DeviceMonitorWorker")
|
|
|> where([j], j.state in ["available", "scheduled", "executing", "retryable"])
|
|
|> select([j], fragment("args->>'device_id'"))
|
|
|> Repo.all()
|
|
|> MapSet.new()
|
|
end
|
|
end
|