graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: integrate lifecycle event broadcasting into DiscoveryWorker

Browse source 
Add Events.broadcast_job_event calls to track job lifecycle:
- :started when job begins executing
- :completed when job finishes (including :discard status)
- :failed when job crashes with error
- Track duration in seconds for all job executions

Wrap main logic in try/rescue to ensure events broadcast even on crash.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Graham McIntire 2026-02-06 18:24:48 -06:00 • committed by Graham McIntire
parent a2778d9b7c
commit be073fa4b7
No known key found for this signature in database
1 changed files with 69 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

109
lib/towerops/workers/discovery_worker.ex
View file

@ -23,6 +23,7 @@ defmodule Towerops.Workers.DiscoveryWorker do
alias Towerops.Agents
alias Towerops.Devices
alias Towerops.JobMonitoring.Events
alias Towerops.Snmp
require Logger
@ -38,50 +39,78 @@ defmodule Towerops.Workers.DiscoveryWorker do
@agent_online_threshold_minutes 10
@impl Oban.Worker
def perform(%Oban.Job{args: %{"device_id" => device_id}}) do
case Devices.get_device_with_details(device_id) do
nil ->
Logger.warning("Device #{device_id} not found, discarding discovery job")
:discard
device ->
# 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)
case Task.yield(task, @job_timeout_ms) || Task.shutdown(task) do
{:ok, :ok} ->
:ok
{:ok, {:error, reason}} ->
Logger.warning(
"Discovery failed for device #{device_id}, discarding job",
device_id: device_id,
error: reason
)
:discard
def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
Events.broadcast_job_event(job, :started)
start_time = System.monotonic_time(:second)
try do
result =
case Devices.get_device_with_details(device_id) do
nil ->
Logger.error(
"Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries",
device_id: device_id,
timeout_ms: @job_timeout_ms
)
Logger.warning("Device #{device_id} not found, discarding discovery job")
:discard
end
end
rescue
error ->
Logger.error(
"Discovery job crashed unexpectedly, discarding to prevent retries",
device_id: device_id,
error: Exception.format(:error, error, __STACKTRACE__)
)
:discard
device ->
# 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)
case Task.yield(task, @job_timeout_ms) || Task.shutdown(task) do
{:ok, :ok} ->
:ok
{:ok, {:error, reason}} ->
Logger.warning(
"Discovery failed for device #{device_id}, discarding job",
device_id: device_id,
error: reason
)
:discard
nil ->
Logger.error(
"Discovery job timeout after #{@job_timeout_ms}ms, discarding to prevent retries",
device_id: device_id,
timeout_ms: @job_timeout_ms
)
:discard
end
end
duration = System.monotonic_time(:second) - start_time
case result do
:ok ->
Events.broadcast_job_event(job, :completed, %{duration: duration})
: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
Logger.error(
"Discovery job crashed unexpectedly, discarding to prevent retries",
device_id: device_id,
error: Exception.format(:error, error, __STACKTRACE__)
)
Events.broadcast_job_event(job, :failed, %{
error: Exception.format(:error, error, __STACKTRACE__),
duration: duration
})
:discard
end
end
defp perform_discovery(device) do