Graham McIntire
a61abd836c
- create MonitoringCheck schema for the monitoring_checks table, fixing silent data loss where ping results were dropped by the wrong schema - update agent channel and device monitor worker to use new schema - fix Stats queries to use MonitoringCheck instead of Check schema - add list_online_cloud_pollers and list_cloud_polled_devices queries - add CloudLatencyProbeWorker to dispatch cross-agent latency probes every 8 hours via PubSub to all online cloud pollers - scope AgentLatencyEvaluator to cloud pollers only with cloud_only filter, lower min_checks (5), wider time window (48h), and device-level assignments to avoid changing site/org defaults - update cron schedule: probes at 0/8/16h, evaluation at 0:30/8:30/16:30 - refactor monitoring.ex and http_executor.ex to fix credo complexity
106 lines
3.1 KiB
Elixir
106 lines
3.1 KiB
Elixir
defmodule Towerops.Workers.AgentLatencyEvaluator do
|
|
@moduledoc """
|
|
Oban cron worker that evaluates cloud agent latency and reassigns devices
|
|
to the lowest-latency cloud poller.
|
|
|
|
Runs 30 minutes after each CloudLatencyProbeWorker batch (00:30, 08:30, 16:30 UTC) to:
|
|
1. Find cloud-polled devices where an alternative cloud agent has 20%+ better latency
|
|
2. Create device-level assignments to the best cloud poller
|
|
3. Broadcast PubSub events for UI updates
|
|
|
|
Only considers cloud pollers (is_cloud_poller == true). Local agent assignments
|
|
are never affected. All reassignments use device-level assignments to avoid
|
|
changing site/org defaults.
|
|
"""
|
|
use Oban.Worker, queue: :maintenance
|
|
|
|
alias Towerops.Agents
|
|
alias Towerops.Agents.Stats
|
|
|
|
require Logger
|
|
|
|
# Minimum improvement required for reassignment (20%)
|
|
@min_improvement_percent 20
|
|
|
|
# Minimum checks per agent for statistical validity
|
|
# Lower than default (10) because cloud probes run 3x/day with 5 pings each
|
|
@min_checks_per_agent 5
|
|
|
|
# Look back 48 hours to cover multiple 8-hour probe cycles
|
|
@hours_ago 48
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{}) do
|
|
start_time = System.monotonic_time(:millisecond)
|
|
|
|
candidates =
|
|
Stats.find_reassignment_candidates(
|
|
min_improvement_percent: @min_improvement_percent,
|
|
min_checks_per_agent: @min_checks_per_agent,
|
|
hours_ago: @hours_ago,
|
|
cloud_only: true
|
|
)
|
|
|
|
reassignment_count =
|
|
candidates
|
|
|> Enum.map(&reassign_device/1)
|
|
|> Enum.count(&(&1 == :ok))
|
|
|
|
duration_ms = System.monotonic_time(:millisecond) - start_time
|
|
|
|
Logger.info(
|
|
"Latency evaluation completed: #{length(candidates)} candidate(s) found, " <>
|
|
"#{reassignment_count} device(s) reassigned in #{duration_ms}ms"
|
|
)
|
|
|
|
:ok
|
|
end
|
|
|
|
defp reassign_device(candidate) do
|
|
Logger.info(
|
|
"Reassigning device '#{candidate.device_name}' " <>
|
|
"(#{candidate.improvement_percent}% improvement: " <>
|
|
"#{candidate.current_latency_ms}ms → #{candidate.best_latency_ms}ms)",
|
|
device_id: candidate.device_id,
|
|
device_name: candidate.device_name,
|
|
old_agent: candidate.current_agent_token_id,
|
|
new_agent: candidate.best_agent_token_id,
|
|
improvement_percent: candidate.improvement_percent,
|
|
assignment_source: candidate.assignment_source
|
|
)
|
|
|
|
# Cloud-only mode: always use device-level assignments to avoid
|
|
# changing site/org defaults that affect other devices
|
|
result = reassign_at_device_level(candidate)
|
|
|
|
case result do
|
|
{:ok, _} ->
|
|
_ = broadcast_reassignment(candidate)
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error(
|
|
"Failed to reassign device '#{candidate.device_name}': #{inspect(reason)}",
|
|
device_id: candidate.device_id,
|
|
error: reason
|
|
)
|
|
|
|
:error
|
|
end
|
|
end
|
|
|
|
defp reassign_at_device_level(candidate) do
|
|
Agents.update_device_assignment(
|
|
candidate.device_id,
|
|
candidate.best_agent_token_id
|
|
)
|
|
end
|
|
|
|
defp broadcast_reassignment(candidate) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"devices:latency",
|
|
{:device_reassigned, candidate}
|
|
)
|
|
end
|
|
end
|