Graham McIntire
52cde2cbc5
Sonar, Splynx, VISP, NetBox, and Gaiia sync workers all followed the same
"cron runs, then Enum.map syncs every integration in the same job" shape.
A single slow tenant — anything from an upstream rate limit to a stuck
GraphQL call — would hold the maintenance queue slot for ten minutes
while every other integration waited in line.
Each worker now uses the dispatcher/per-integration pattern that
UispSyncWorker already had:
- the cron `perform(%Oban.Job{args: %{}})` fans out one job per eligible
integration via `PollingOffset.calculate_offset/2`, staggered across a
5- to 10-minute window with `unique: [period: window_seconds]`
- `perform(%Oban.Job{args: %{"integration_id" => id}})` does the actual
sync, classifying failures via `Towerops.Workers.SyncErrors.transient?/1`
so permanent errors (auth, 4xx) don't get retried forever
Worker-specific defaults preserved:
- gaiia / netbox: 30/600s window, longer interval
- sonar / splynx / visp: 10/300s window
Tests updated to match the two-stage dispatcher pattern — `perform_job/2`
with empty args now asserts the per-integration jobs got enqueued and
the actual sync is invoked via a separate `perform_job/2` call with the
integration_id arg.
82 lines
2.7 KiB
Elixir
82 lines
2.7 KiB
Elixir
defmodule Towerops.Workers.NetBoxSyncWorker do
|
|
@moduledoc """
|
|
Oban worker that syncs NetBox data for enabled integrations.
|
|
|
|
The cron job (no args) runs every 30 minutes and dispatches individual
|
|
per-integration sync jobs staggered across the 600-second window using
|
|
`PollingOffset.calculate_offset/2`. This prevents a thundering herd
|
|
against NetBox APIs and stops one slow tenant from blocking the
|
|
maintenance queue while every other integration waits in line.
|
|
|
|
Individual jobs receive `%{"integration_id" => id}` and perform the
|
|
actual sync via `NetBox.Sync.sync_organization/1`.
|
|
"""
|
|
use Oban.Worker, queue: :maintenance
|
|
|
|
alias Towerops.Integrations
|
|
alias Towerops.NetBox.Sync
|
|
alias Towerops.Workers.PollingOffset
|
|
alias Towerops.Workers.SyncErrors
|
|
|
|
require Logger
|
|
|
|
@provider "netbox"
|
|
@default_interval_minutes 30
|
|
@window_seconds 600
|
|
|
|
# Cron dispatcher: enqueues staggered individual sync jobs.
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: args}) when args == %{} do
|
|
integrations = Integrations.list_enabled_integrations(@provider)
|
|
|
|
eligible = Enum.filter(integrations, &due_for_sync?/1)
|
|
|
|
now = DateTime.utc_now()
|
|
|
|
enqueued =
|
|
Enum.count(eligible, fn integration ->
|
|
offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
|
|
scheduled_at = DateTime.add(now, offset, :second)
|
|
|
|
case %{"integration_id" => integration.id}
|
|
|> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
|
|
|> Oban.insert() do
|
|
{:ok, _job} -> true
|
|
{:error, _reason} -> false
|
|
end
|
|
end)
|
|
|
|
if enqueued > 0 do
|
|
Logger.info("NetBox sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
# Individual sync: loads integration by ID and syncs.
|
|
def perform(%Oban.Job{args: %{"integration_id" => id}}) do
|
|
case Integrations.get_integration_by_id(id) do
|
|
{:ok, integration} ->
|
|
sync_integration(integration)
|
|
|
|
{:error, :not_found} ->
|
|
Logger.warning("NetBox sync: integration #{id} not found, skipping")
|
|
{:error, :not_found}
|
|
end
|
|
end
|
|
|
|
defp sync_integration(integration) do
|
|
case Sync.sync_organization(integration) do
|
|
{:ok, result} ->
|
|
Logger.info("NetBox sync completed for org #{integration.organization_id}: #{inspect(result)}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("NetBox sync failed for org #{integration.organization_id}: #{inspect(reason)}")
|
|
if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
|
|
end
|
|
end
|
|
|
|
@doc false
|
|
def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
|
|
end
|