From cc0a4f4ed6a362222a3132ef8c9a9d0ff3f52ce4 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 5 Mar 2026 09:22:14 -0600 Subject: [PATCH] refactor: replace Task.start with Oban for alert notifications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaced fire-and-forget Task.start calls with reliable Oban workers for all alert notifications. **Why Oban instead of Task.start:** - ✅ Persistent - jobs survive crashes/restarts - ✅ Automatic retries with backoff (3 attempts) - ✅ Monitoring via Oban dashboard - ✅ Rate limiting via queue concurrency - ✅ Distributed coordination across servers **Changes:** - Created AlertNotificationWorker for trigger/acknowledge/resolve - Added notifications queue (concurrency: 10) to Oban config - Replaced 4 Task.start calls in alerts.ex and device_monitor_worker.ex - Updated 2 test assertions for string alert_type **Files changed:** - lib/towerops/workers/alert_notification_worker.ex (new) - lib/towerops/alerts.ex - lib/towerops/workers/device_monitor_worker.ex - config/dev.exs - config/runtime.exs - test/towerops/alerts_test.exs Notifications are now guaranteed to be delivered with automatic retry. --- config/dev.exs | 2 + config/runtime.exs | 2 + lib/towerops/alerts.ex | 6 +- .../workers/alert_notification_worker.ex | 101 ++++++++++++++++++ lib/towerops/workers/device_monitor_worker.ex | 38 +------ test/towerops/alerts_test.exs | 6 +- 6 files changed, 116 insertions(+), 39 deletions(-) create mode 100644 lib/towerops/workers/alert_notification_worker.ex diff --git a/config/dev.exs b/config/dev.exs index 49e3ca53..33059885 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -52,6 +52,8 @@ config :towerops, Oban, pollers: 50, # Device monitoring jobs - health checks monitors: 50, + # Alert notifications (PagerDuty, email, etc.) + notifications: 10, maintenance: 5, weather: 2 ], diff --git a/config/runtime.exs b/config/runtime.exs index 16fdc5d1..84767481 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -165,6 +165,8 @@ if config_env() == :prod do # Service checks - HTTP/TCP/DNS checks: 50, check_executors: 50, + # Alert notifications (PagerDuty, email, etc.) + notifications: 10, maintenance: 5, weather: 2 ], diff --git a/lib/towerops/alerts.ex b/lib/towerops/alerts.ex index 761c5ad0..d4e70fe5 100644 --- a/lib/towerops/alerts.ex +++ b/lib/towerops/alerts.ex @@ -7,8 +7,8 @@ defmodule Towerops.Alerts do alias Towerops.Alerts.Alert alias Towerops.Gaiia.ImpactAnalysis - alias Towerops.PagerDuty.Notifier alias Towerops.Repo + alias Towerops.Workers.AlertNotificationWorker @doc """ Creates an alert for device status change. @@ -248,7 +248,7 @@ defmodule Towerops.Alerts do case result do {:ok, updated_alert} -> - Task.start(fn -> Notifier.notify_acknowledge(updated_alert) end) + AlertNotificationWorker.enqueue_acknowledge(updated_alert.id) {:ok, updated_alert} error -> @@ -267,7 +267,7 @@ defmodule Towerops.Alerts do case result do {:ok, updated_alert} -> - Task.start(fn -> Notifier.notify_resolve(updated_alert) end) + AlertNotificationWorker.enqueue_resolve(updated_alert.id) {:ok, updated_alert} error -> diff --git a/lib/towerops/workers/alert_notification_worker.ex b/lib/towerops/workers/alert_notification_worker.ex new file mode 100644 index 00000000..93a40866 --- /dev/null +++ b/lib/towerops/workers/alert_notification_worker.ex @@ -0,0 +1,101 @@ +defmodule Towerops.Workers.AlertNotificationWorker do + @moduledoc """ + Oban worker for sending alert notifications reliably. + + Handles trigger, acknowledge, and resolve notifications with automatic + retries and error tracking. + """ + use Oban.Worker, + queue: :notifications, + max_attempts: 3, + priority: 1 + + alias Towerops.Alerts + alias Towerops.Devices + alias Towerops.PagerDuty.Notifier + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"action" => "trigger", "alert_id" => alert_id}}) do + with {:ok, alert} <- fetch_alert(alert_id), + {:ok, device} <- fetch_device(alert.device_id) do + Notifier.notify_trigger(alert, device) + else + {:error, :alert_not_found} -> + Logger.warning("Alert #{alert_id} not found for notification") + :ok + + {:error, :device_not_found} -> + Logger.warning("Device not found for alert #{alert_id}") + :ok + + {:error, reason} -> + Logger.error("Failed to send trigger notification for alert #{alert_id}: #{inspect(reason)}") + {:error, reason} + end + end + + def perform(%Oban.Job{args: %{"action" => "acknowledge", "alert_id" => alert_id}}) do + case fetch_alert(alert_id) do + {:ok, alert} -> + Notifier.notify_acknowledge(alert) + + {:error, :alert_not_found} -> + Logger.warning("Alert #{alert_id} not found for acknowledge notification") + :ok + + {:error, reason} -> + Logger.error("Failed to send acknowledge notification for alert #{alert_id}: #{inspect(reason)}") + {:error, reason} + end + end + + def perform(%Oban.Job{args: %{"action" => "resolve", "alert_id" => alert_id}}) do + case fetch_alert(alert_id) do + {:ok, alert} -> + Notifier.notify_resolve(alert) + + {:error, :alert_not_found} -> + Logger.warning("Alert #{alert_id} not found for resolve notification") + :ok + + {:error, reason} -> + Logger.error("Failed to send resolve notification for alert #{alert_id}: #{inspect(reason)}") + {:error, reason} + end + end + + # Helper to enqueue notification jobs + def enqueue_trigger(alert_id) do + %{action: "trigger", alert_id: alert_id} + |> new() + |> Oban.insert() + end + + def enqueue_acknowledge(alert_id) do + %{action: "acknowledge", alert_id: alert_id} + |> new() + |> Oban.insert() + end + + def enqueue_resolve(alert_id) do + %{action: "resolve", alert_id: alert_id} + |> new() + |> Oban.insert() + end + + defp fetch_alert(alert_id) do + case Alerts.get_alert(alert_id) do + nil -> {:error, :alert_not_found} + alert -> {:ok, alert} + end + end + + defp fetch_device(device_id) do + case Devices.get_device(device_id) do + nil -> {:error, :device_not_found} + device -> {:ok, device} + end + end +end diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 3df21203..4e5f169e 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -24,8 +24,8 @@ defmodule Towerops.Workers.DeviceMonitorWorker do alias Towerops.Devices alias Towerops.Maintenance alias Towerops.Monitoring - alias Towerops.PagerDuty.Notifier alias Towerops.Snmp.Client + alias Towerops.Workers.AlertNotificationWorker alias Towerops.Workers.PollingOffset require Logger @@ -223,22 +223,8 @@ defmodule Towerops.Workers.DeviceMonitorWorker do message: alert_message }) do {:ok, alert} -> - # Wrap notification in try-catch to prevent silent failures - Task.start(fn -> - try do - Notifier.notify_trigger(alert, device) - rescue - error -> - require Logger - - Logger.error( - "Failed to send notification for device down alert: #{inspect(error)}", - alert_id: alert.id, - device_id: device.id, - error: error - ) - end - end) + # Enqueue notification job - Oban handles retries and persistence + AlertNotificationWorker.enqueue_trigger(alert.id) _ = Phoenix.PubSub.broadcast( @@ -289,22 +275,8 @@ defmodule Towerops.Workers.DeviceMonitorWorker do message: recovery_message }) do {:ok, alert} -> - # Wrap notification in try-catch to prevent silent failures - Task.start(fn -> - try do - Notifier.notify_trigger(alert, device) - rescue - error -> - require Logger - - Logger.error( - "Failed to send notification for device up alert: #{inspect(error)}", - alert_id: alert.id, - device_id: device.id, - error: error - ) - end - end) + # Enqueue notification job - Oban handles retries and persistence + AlertNotificationWorker.enqueue_trigger(alert.id) resolve_down_alert(device) diff --git a/test/towerops/alerts_test.exs b/test/towerops/alerts_test.exs index 18c3c228..4505a21a 100644 --- a/test/towerops/alerts_test.exs +++ b/test/towerops/alerts_test.exs @@ -38,7 +38,7 @@ defmodule Towerops.AlertsTest do test "create_alert/1 with valid data creates an alert", %{device: device} do attrs = Map.put(@valid_attrs, :device_id, device.id) assert {:ok, %Alert{} = alert} = Alerts.create_alert(attrs) - assert alert.alert_type == :device_down + assert alert.alert_type == "device_down" assert alert.message == "Equipment is not responding" end @@ -622,8 +622,8 @@ defmodule Towerops.AlertsTest do assert length(result) == 2 alert_types = Enum.map(result, & &1.alert_type) - assert :device_down in alert_types - assert :device_up in alert_types + assert "device_down" in alert_types + assert "device_up" in alert_types end end end