105 lines
2.8 KiB
Elixir
105 lines
2.8 KiB
Elixir
defmodule Towerops.JobMonitoring do
|
|
@moduledoc """
|
|
Context for monitoring Oban jobs, specifically polling and discovery operations.
|
|
|
|
Provides queries for active, stuck, failed, and completed jobs with device
|
|
and agent context for operational monitoring.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Oban.Job
|
|
alias Towerops.Repo
|
|
|
|
@worker_names [
|
|
"Towerops.Workers.DevicePollerWorker",
|
|
"Towerops.Workers.DiscoveryWorker"
|
|
]
|
|
|
|
# 2 minutes
|
|
@polling_threshold_seconds 120
|
|
# 5 minutes
|
|
@discovery_threshold_seconds 300
|
|
|
|
@doc """
|
|
Lists all currently executing polling and discovery jobs.
|
|
"""
|
|
@spec list_active_jobs() :: [Job.t()]
|
|
def list_active_jobs do
|
|
from(j in Job,
|
|
where: j.state == "executing",
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [asc: j.attempted_at]
|
|
)
|
|
|> Repo.all()
|
|
|> preload_device_context()
|
|
end
|
|
|
|
@doc """
|
|
Lists jobs that are executing longer than expected thresholds.
|
|
|
|
Thresholds:
|
|
- Polling: 2 minutes
|
|
- Discovery: 5 minutes
|
|
"""
|
|
@spec list_stuck_jobs() :: [Job.t()]
|
|
def list_stuck_jobs do
|
|
now = DateTime.utc_now()
|
|
polling_threshold = DateTime.add(now, -@polling_threshold_seconds, :second)
|
|
discovery_threshold = DateTime.add(now, -@discovery_threshold_seconds, :second)
|
|
|
|
from(j in Job,
|
|
where: j.state == "executing",
|
|
where:
|
|
(j.worker == "Towerops.Workers.DevicePollerWorker" and j.attempted_at < ^polling_threshold) or
|
|
(j.worker == "Towerops.Workers.DiscoveryWorker" and j.attempted_at < ^discovery_threshold),
|
|
order_by: [asc: j.attempted_at]
|
|
)
|
|
|> Repo.all()
|
|
|> preload_device_context()
|
|
end
|
|
|
|
@doc """
|
|
Lists jobs that have failed and are retrying or have been cancelled.
|
|
|
|
Limited to last 100 jobs.
|
|
"""
|
|
@spec list_failed_jobs() :: [Job.t()]
|
|
def list_failed_jobs do
|
|
from(j in Job,
|
|
where: j.state in ["retryable", "cancelled"],
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [desc: j.inserted_at],
|
|
limit: 100
|
|
)
|
|
|> Repo.all()
|
|
|> preload_device_context()
|
|
end
|
|
|
|
@doc """
|
|
Lists recently completed jobs (completed, cancelled, or discarded).
|
|
|
|
## Options
|
|
- `:limit` - Maximum number of jobs to return (default: 100)
|
|
"""
|
|
@spec list_recent_completions(integer()) :: [Job.t()]
|
|
def list_recent_completions(limit \\ 100) do
|
|
from(j in Job,
|
|
where: j.state in ["completed", "cancelled", "discarded"],
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [desc: j.completed_at],
|
|
limit: ^limit
|
|
)
|
|
|> Repo.all()
|
|
|> preload_device_context()
|
|
end
|
|
|
|
# Add helper to fetch device context
|
|
defp preload_device_context(jobs) do
|
|
Enum.map(jobs, fn job ->
|
|
device_id = get_in(job.args, ["device_id"])
|
|
device = device_id && Repo.get(Towerops.Devices.Device, device_id)
|
|
Map.put(job, :device, device)
|
|
end)
|
|
end
|
|
end
|