diff --git a/config/dev.exs b/config/dev.exs index 20123b1a..a08035ed 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -25,6 +25,25 @@ config :phoenix_live_view, # Disable swoosh api client as it is only required for production adapters. config :swoosh, :api_client, false +# Configure Oban for background jobs +config :towerops, Oban, + repo: Towerops.Repo, + queues: [ + default: 10, + discovery: 10, + # SNMP polling jobs - one per device + pollers: 50, + # Device monitoring jobs - health checks + monitors: 50, + maintenance: 5 + ], + plugins: [ + # Automatically delete completed jobs after 60 seconds + {Oban.Plugins.Pruner, max_age: 60}, + # Rescue orphaned jobs (when node crashes) + {Oban.Plugins.Reindexer, schedule: "0 2 * * *"} + ] + # Configure your database config :towerops, Towerops.Repo, username: "postgres", diff --git a/lib/towerops_web/telemetry.ex b/lib/towerops_web/telemetry.ex index e7ddcc09..cd874ee9 100644 --- a/lib/towerops_web/telemetry.ex +++ b/lib/towerops_web/telemetry.ex @@ -99,16 +99,16 @@ defmodule ToweropsWeb.Telemetry do summary("vm.total_run_queue_lengths.cpu"), summary("vm.total_run_queue_lengths.io"), - # Exq/Redis Metrics - last_value("towerops.exq.queue.size", + # Oban/Redis Metrics + last_value("towerops.oban.queue.size", tags: [:queue], - description: "Number of jobs in each Exq queue" + description: "Number of jobs in each Oban queue" ), - last_value("towerops.exq.processes.busy", - description: "Number of busy Exq worker processes" + last_value("towerops.oban.jobs.executing", + description: "Number of currently executing Oban jobs" ), - last_value("towerops.exq.processes.total", - description: "Total number of Exq worker processes" + last_value("towerops.oban.jobs.available", + description: "Number of available Oban jobs" ), last_value("towerops.redis.connected_clients", description: "Number of Redis/Valkey connected clients" @@ -125,72 +125,84 @@ defmodule ToweropsWeb.Telemetry do defp periodic_measurements do [ - # Measure Exq and Redis stats every 10 seconds - {__MODULE__, :publish_exq_stats, []}, + # Measure Oban and Redis stats every 10 seconds + {__MODULE__, :publish_oban_stats, []}, {__MODULE__, :publish_redis_stats, []} ] end @doc """ - Publishes Exq queue and process statistics. + Publishes Oban queue and job statistics. """ - def publish_exq_stats do - # Only run if Exq is available (not in test env) and the API process is running - # The Exq.Api.Server is registered as :"Exq.Api" when started with name: Exq - api_process = :"Exq.Api" - - if Application.get_env(:towerops, :env) == :test or is_nil(Process.whereis(api_process)) do + def publish_oban_stats do + if Application.get_env(:towerops, :env) == :test do :ok else try do - queues = ["default", "discovery", "polling", "monitoring", "maintenance"] + import Ecto.Query + queues = ["default", "discovery", "pollers", "monitors", "maintenance"] + + # Query queue sizes for queue <- queues do try do - case Exq.Api.queue_size(api_process, queue) do - {:ok, size} -> - :telemetry.execute( - [:towerops, :exq, :queue, :size], - %{value: size}, - %{queue: queue} - ) + size = + Towerops.Repo.one( + from j in Oban.Job, + where: j.queue == ^queue and j.state in ["available", "scheduled"], + select: count(j.id) + ) - _error -> - :ok - end + :telemetry.execute( + [:towerops, :oban, :queue, :size], + %{value: size}, + %{queue: queue} + ) rescue _ -> :ok - catch - :exit, _ -> :ok end end + # Query executing jobs count try do - with {:ok, processes} <- Exq.Api.processes(api_process), - {:ok, busy} <- Exq.Api.busy(api_process) do - :telemetry.execute( - [:towerops, :exq, :processes, :busy], - %{value: length(busy)}, - %{} + executing_count = + Towerops.Repo.one( + from j in Oban.Job, + where: j.state == "executing", + select: count(j.id) ) - :telemetry.execute( - [:towerops, :exq, :processes, :total], - %{value: length(processes)}, - %{} - ) - end + :telemetry.execute( + [:towerops, :oban, :jobs, :executing], + %{value: executing_count}, + %{} + ) + rescue + _ -> :ok + end + + # Query available jobs count + try do + available_count = + Towerops.Repo.one( + from j in Oban.Job, + where: j.state == "available", + select: count(j.id) + ) + + :telemetry.execute( + [:towerops, :oban, :jobs, :available], + %{value: available_count}, + %{} + ) rescue _ -> :ok - catch - :exit, _ -> :ok end :ok rescue _ -> :ok catch - # Catch exits (like :noproc) and exceptions _, _ -> :ok end end