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

fix: cloud poller automatic rebalancing

Browse source 
- 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
This commit is contained in:
Graham McIntire 2026-02-12 15:32:33 -06:00
parent 07a6128877
commit a61abd836c
No known key found for this signature in database
18 changed files with 1140 additions and 337 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
config/dev.exs
View file

@ -62,8 +62,10 @@ config :towerops, Oban,
{"0 * * * *", Towerops.Snmp.NeighborCleanupWorker},
# Check for stale agents every minute
{"* * * * *", Towerops.Workers.StaleAgentWorker},
# Evaluate latency-based agent reassignment every 5 minutes
{"*/5 * * * *", Towerops.Workers.AgentLatencyEvaluator},
# Cloud latency probes every 8 hours (00:00, 08:00, 16:00)
{"0 */8 * * *", Towerops.Workers.CloudLatencyProbeWorker},
# Evaluate latency-based agent reassignment 30 min after each probe batch
{"30 0,8,16 * * *", Towerops.Workers.AgentLatencyEvaluator},
# Health check for missing jobs every 10 minutes
{"*/10 * * * *", Towerops.Workers.JobHealthCheckWorker},
# Fetch latest firmware versions daily at 2 AM

6
config/runtime.exs
View file

@ -171,8 +171,10 @@ if config_env() == :prod do
{"0 * * * *", Towerops.Snmp.NeighborCleanupWorker},
# Check for stale agents every minute
{"* * * * *", Towerops.Workers.StaleAgentWorker},
# Evaluate latency-based agent reassignment every 5 minutes
{"*/5 * * * *", Towerops.Workers.AgentLatencyEvaluator},
# Cloud latency probes every 8 hours (00:00, 08:00, 16:00)
{"0 */8 * * *", Towerops.Workers.CloudLatencyProbeWorker},
# Evaluate latency-based agent reassignment 30 min after each probe batch
{"30 0,8,16 * * *", Towerops.Workers.AgentLatencyEvaluator},
# Health check for missing jobs every 10 minutes
{"*/10 * * * *", Towerops.Workers.JobHealthCheckWorker},
# Mark stale backup requests as timeout every 10 minutes

86
lib/towerops/agents.ex
View file

@ -464,6 +464,92 @@ defmodule Towerops.Agents do
end
end
## Cloud poller queries
@doc """
Lists all online cloud poller agent tokens.
Returns cloud pollers that are enabled and have been seen within the last 5 minutes.
"""
@spec list_online_cloud_pollers() :: [AgentToken.t()]
def list_online_cloud_pollers do
five_minutes_ago = DateTime.add(DateTime.utc_now(), -5, :minute)
AgentToken
|> where([t], t.is_cloud_poller == true and t.enabled == true)
|> where([t], not is_nil(t.last_seen_at) and t.last_seen_at > ^five_minutes_ago)
|> Repo.all()
end
@doc """
Lists all devices that should be polled by cloud pollers.
Returns devices whose effective agent is either a cloud poller or no agent at all,
filtered to only monitoring-enabled devices.
"""
@spec list_cloud_polled_devices() :: [Device.t()]
def list_cloud_polled_devices do
Repo.all(cloud_polled_devices_query())
end
defp cloud_polled_devices_query do
from(d in Device,
left_join: aa in AgentAssignment,
on: aa.device_id == d.id and aa.enabled == true,
left_join: device_token in AgentToken,
on: device_token.id == aa.agent_token_id,
left_join: s in assoc(d, :site),
left_join: site_token in AgentToken,
on: site_token.id == s.agent_token_id,
left_join: o in assoc(d, :organization),
left_join: org_token in AgentToken,
on: org_token.id == o.default_agent_token_id,
where: d.monitoring_enabled == true,
where: ^cloud_polled_condition()
)
end
# Device is cloud-polled if its effective agent is a cloud poller or no agent exists.
# Split into separate dynamic conditions to satisfy credo complexity limits.
defp cloud_polled_condition do
device_cloud = device_assigned_to_cloud()
site_cloud = site_default_is_cloud()
org_cloud = org_default_is_cloud()
no_agent = no_agent_at_any_level()
dynamic([], ^device_cloud or ^site_cloud or ^org_cloud or ^no_agent)
end
defp device_assigned_to_cloud do
dynamic(
[d, aa, device_token, s, site_token, o, org_token],
not is_nil(aa.id) and device_token.is_cloud_poller == true
)
end
defp site_default_is_cloud do
dynamic(
[d, aa, device_token, s, site_token, o, org_token],
is_nil(aa.id) and not is_nil(s.agent_token_id) and site_token.is_cloud_poller == true
)
end
defp org_default_is_cloud do
dynamic(
[d, aa, device_token, s, site_token, o, org_token],
is_nil(aa.id) and (is_nil(s.id) or is_nil(s.agent_token_id)) and
not is_nil(o.default_agent_token_id) and org_token.is_cloud_poller == true
)
end
defp no_agent_at_any_level do
dynamic(
[d, aa, device_token, s, site_token, o, org_token],
is_nil(aa.id) and (is_nil(s.id) or is_nil(s.agent_token_id)) and
is_nil(o.default_agent_token_id)
)
end
## Assignment management
@doc """

38
lib/towerops/agents/stats.ex
View file

@ -10,7 +10,7 @@ defmodule Towerops.Agents.Stats do
alias Towerops.Agents.AgentAssignment
alias Towerops.Agents.AgentToken
alias Towerops.Devices.Device
alias Towerops.Monitoring.Check
alias Towerops.Monitoring.MonitoringCheck
alias Towerops.Repo
alias Towerops.Snmp.SensorReading
@ -337,10 +337,10 @@ defmodule Towerops.Agents.Stats do
time_threshold = DateTime.add(DateTime.utc_now(), -hours_ago, :hour)
Repo.all(
from(c in Check,
from(c in MonitoringCheck,
where:
c.device_id == ^device_id and
c.status == :success and
c.status == "success" and
not is_nil(c.agent_token_id) and
c.checked_at > ^time_threshold,
group_by: c.agent_token_id,
@ -419,13 +419,19 @@ defmodule Towerops.Agents.Stats do
min_improvement_percent = Keyword.get(opts, :min_improvement_percent, 20)
min_checks = Keyword.get(opts, :min_checks_per_agent, 10)
hours_ago = Keyword.get(opts, :hours_ago, 24)
cloud_only = Keyword.get(opts, :cloud_only, false)
time_threshold = DateTime.add(DateTime.utc_now(), -hours_ago, :hour)
# Get all devices with multiple agent latency measurements
devices_with_agents =
Repo.all(
from(c in Check,
where: c.status == :success and not is_nil(c.agent_token_id) and c.checked_at > ^time_threshold,
# When cloud_only is true, join agent_tokens to filter by is_cloud_poller
query =
if cloud_only do
from(c in MonitoringCheck,
join: a in AgentToken,
on: a.id == c.agent_token_id,
where:
c.status == "success" and not is_nil(c.agent_token_id) and
c.checked_at > ^time_threshold and a.is_cloud_poller == true,
group_by: [c.device_id, c.agent_token_id],
having: count(c.id) >= ^min_checks,
select: %{
@ -435,7 +441,23 @@ defmodule Towerops.Agents.Stats do
check_count: count(c.id)
}
)
)
else
from(c in MonitoringCheck,
where:
c.status == "success" and not is_nil(c.agent_token_id) and
c.checked_at > ^time_threshold,
group_by: [c.device_id, c.agent_token_id],
having: count(c.id) >= ^min_checks,
select: %{
device_id: c.device_id,
agent_token_id: c.agent_token_id,
avg_latency_ms: avg(c.response_time_ms),
check_count: count(c.id)
}
)
end
devices_with_agents = Repo.all(query)
# Group by device and evaluate candidates
devices_with_agents

63
lib/towerops/monitoring.ex
View file

@ -9,6 +9,7 @@ defmodule Towerops.Monitoring do
alias Towerops.Monitoring.Check
alias Towerops.Monitoring.CheckResult
alias Towerops.Monitoring.MonitoringCheck
alias Towerops.Repo
## Checks
@ -168,43 +169,35 @@ defmodule Towerops.Monitoring do
update_check(check, attrs)
end
defp calculate_state_transition(check, _new_status, state_changed) do
if state_changed do
# State changed - increment attempt, stay soft
new_attempt = check.current_check_attempt + 1
if new_attempt >= check.max_check_attempts do
# Reached max attempts - transition to hard
{"hard", new_attempt}
else
# Still in soft state
{"soft", new_attempt}
end
else
# State unchanged
if check.current_state_type == "hard" do
# Already hard, stay hard
{"hard", check.current_check_attempt}
else
# Soft state, increment attempt
new_attempt = min(check.current_check_attempt + 1, check.max_check_attempts)
if new_attempt >= check.max_check_attempts do
{"hard", new_attempt}
else
{"soft", new_attempt}
end
end
end
defp calculate_state_transition(check, _new_status, true = _state_changed) do
new_attempt = check.current_check_attempt + 1
state_type = if new_attempt >= check.max_check_attempts, do: "hard", else: "soft"
{state_type, new_attempt}
end
## Legacy ping-based monitoring compatibility
## TODO: Remove once old monitoring is fully migrated to new checks system
defp calculate_state_transition(%{current_state_type: "hard"} = check, _new_status, false) do
{"hard", check.current_check_attempt}
end
defp calculate_state_transition(check, _new_status, false) do
new_attempt = min(check.current_check_attempt + 1, check.max_check_attempts)
state_type = if new_attempt >= check.max_check_attempts, do: "hard", else: "soft"
{state_type, new_attempt}
end
## Monitoring Checks (ping results from agents)
@doc """
Creates a monitoring check record in the monitoring_checks table.
"""
def create_monitoring_check(attrs \\ %{}) do
%MonitoringCheck{}
|> MonitoringCheck.changeset(attrs)
|> Repo.insert()
end
## Legacy ping-based monitoring compatibility (stub)
@doc false
def get_latency_data(_device_or_site_id, _opts) do
# Stub for backward compatibility - returns empty list
# Old monitoring checks are being replaced by new checks system
[]
end
def get_latency_data(_device_or_site_id, _opts), do: []
end

108
lib/towerops/monitoring/executors/http_executor.ex
View file

@ -27,68 +27,76 @@ defmodule Towerops.Monitoring.Executors.HttpExecutor do
- {:error, reason} on failure
"""
def execute(config, timeout_ms \\ @default_timeout) do
url = Map.fetch!(config, "url")
method = String.downcase(Map.get(config, "method", "GET"))
expected_status = Map.get(config, "expected_status", 200)
verify_ssl = Map.get(config, "verify_ssl", true)
headers = Map.get(config, "headers", %{})
body = Map.get(config, "body")
regex = Map.get(config, "regex")
follow_redirects = Map.get(config, "follow_redirects", true)
start_time = System.monotonic_time(:millisecond)
req_opts = build_req_opts(config, timeout_ms)
req_opts = [
case Req.request(req_opts) do
{:ok, response} ->
response_time = System.monotonic_time(:millisecond) - start_time
validate_response(response, config, response_time)
{:error, error} ->
format_error(error, timeout_ms)
end
rescue
e ->
Logger.error("HTTP check exception: #{inspect(e)}")
{:error, "Exception: #{Exception.message(e)}"}
end
defp build_req_opts(config, timeout_ms) do
method = String.downcase(Map.get(config, "method", "GET"))
verify_ssl = Map.get(config, "verify_ssl", true)
opts = [
method: String.to_atom(method),
url: url,
headers: headers,
url: Map.fetch!(config, "url"),
headers: Map.get(config, "headers", %{}),
receive_timeout: timeout_ms,
connect_options: [
transport_opts: [
verify: if(verify_ssl, do: :verify_peer, else: :verify_none)
]
],
redirect: follow_redirects
redirect: Map.get(config, "follow_redirects", true)
]
req_opts =
if body do
Keyword.put(req_opts, :body, body)
else
req_opts
end
case Req.request(req_opts) do
{:ok, %Req.Response{status: status, body: response_body}} ->
response_time = System.monotonic_time(:millisecond) - start_time
cond do
status != expected_status ->
{:error, "HTTP #{status}, expected #{expected_status}"}
regex && !content_matches?(response_body, regex) ->
{:error, "Content does not match pattern: #{regex}"}
true ->
{:ok, response_time, "HTTP #{status} OK"}
end
{:error, %Req.TransportError{reason: :timeout}} ->
{:error, "Connection timeout after #{timeout_ms}ms"}
{:error, %Req.TransportError{reason: :econnrefused}} ->
{:error, "Connection refused"}
{:error, %Req.TransportError{reason: reason}} ->
{:error, "Transport error: #{inspect(reason)}"}
{:error, reason} ->
{:error, "Request failed: #{inspect(reason)}"}
case Map.get(config, "body") do
nil -> opts
body -> Keyword.put(opts, :body, body)
end
rescue
e ->
Logger.error("HTTP check exception: #{inspect(e)}")
{:error, "Exception: #{Exception.message(e)}"}
end
defp validate_response(%Req.Response{status: status, body: body}, config, response_time) do
expected_status = Map.get(config, "expected_status", 200)
regex = Map.get(config, "regex")
cond do
status != expected_status ->
{:error, "HTTP #{status}, expected #{expected_status}"}
regex && !content_matches?(body, regex) ->
{:error, "Content does not match pattern: #{regex}"}
true ->
{:ok, response_time, "HTTP #{status} OK"}
end
end
defp format_error(%Req.TransportError{reason: :timeout}, timeout_ms) do
{:error, "Connection timeout after #{timeout_ms}ms"}
end
defp format_error(%Req.TransportError{reason: :econnrefused}, _timeout_ms) do
{:error, "Connection refused"}
end
defp format_error(%Req.TransportError{reason: reason}, _timeout_ms) do
{:error, "Transport error: #{inspect(reason)}"}
end
defp format_error(reason, _timeout_ms) do
{:error, "Request failed: #{inspect(reason)}"}
end
defp content_matches?(body, regex_str) when is_binary(body) do

37
lib/towerops/monitoring/monitoring_check.ex Normal file
View file

@ -0,0 +1,37 @@
defmodule Towerops.Monitoring.MonitoringCheck do
@moduledoc """
Schema for the `monitoring_checks` table.
Stores ping/monitoring check results from agents and the Phoenix server.
Each record captures the device status, response time, and which agent
performed the check.
"""
use Ecto.Schema
import Ecto.Changeset
alias Towerops.Agents.AgentToken
alias Towerops.Devices.Device
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "monitoring_checks" do
field :status, :string
field :response_time_ms, :float
field :checked_at, :utc_datetime
belongs_to :device, Device
belongs_to :agent_token, AgentToken
timestamps(type: :utc_datetime, updated_at: false)
end
def changeset(monitoring_check, attrs) do
monitoring_check
|> cast(attrs, [:device_id, :agent_token_id, :status, :response_time_ms, :checked_at])
|> validate_required([:device_id, :status, :checked_at])
|> foreign_key_constraint(:device_id)
|> foreign_key_constraint(:agent_token_id)
end
end

62
lib/towerops/workers/agent_latency_evaluator.ex
View file

@ -1,24 +1,21 @@
defmodule Towerops.Workers.AgentLatencyEvaluator do
@moduledoc """
Oban worker that periodically evaluates device-to-agent latency and performs automatic reassignments.
Oban cron worker that evaluates cloud agent latency and reassigns devices
to the lowest-latency cloud poller.
Runs every 5 minutes via Oban.Plugins.Cron to:
1. Find devices where an alternative agent has significantly better latency (20%+ improvement)
2. Reassign devices based on their assignment source (site, organization, global, or device-level)
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
4. Log all reassignment decisions with improvement metrics
Only considers devices with automatic assignments (site, organization, global, or none).
Manual device-level assignments are never automatically changed.
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
alias Towerops.Devices.Device
alias Towerops.Organizations
alias Towerops.Repo
alias Towerops.Sites
require Logger
@ -26,7 +23,11 @@ defmodule Towerops.Workers.AgentLatencyEvaluator do
@min_improvement_percent 20
# Minimum checks per agent for statistical validity
@min_checks_per_agent 10
# 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
@ -35,7 +36,9 @@ defmodule Towerops.Workers.AgentLatencyEvaluator do
candidates =
Stats.find_reassignment_candidates(
min_improvement_percent: @min_improvement_percent,
min_checks_per_agent: @min_checks_per_agent
min_checks_per_agent: @min_checks_per_agent,
hours_ago: @hours_ago,
cloud_only: true
)
reassignment_count =
@ -66,17 +69,9 @@ defmodule Towerops.Workers.AgentLatencyEvaluator do
assignment_source: candidate.assignment_source
)
result =
case candidate.assignment_source do
:site ->
reassign_at_site_level(candidate)
:organization ->
reassign_at_organization_level(candidate)
source when source in [:global, :none] ->
reassign_at_device_level(candidate)
end
# 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, _} ->
@ -94,25 +89,6 @@ defmodule Towerops.Workers.AgentLatencyEvaluator do
end
end
defp reassign_at_site_level(candidate) do
device = Device |> Repo.get!(candidate.device_id) |> Repo.preload(:site)
Sites.update_site(device.site, %{agent_token_id: candidate.best_agent_token_id})
end
defp reassign_at_organization_level(candidate) do
device =
Device
|> Repo.get!(candidate.device_id)
|> Repo.preload(site: :organization)
organization = device.site.organization
Organizations.update_organization(organization, %{
default_agent_token_id: candidate.best_agent_token_id
})
end
defp reassign_at_device_level(candidate) do
Agents.update_device_assignment(
candidate.device_id,

47
lib/towerops/workers/cloud_latency_probe_worker.ex Normal file
View file

@ -0,0 +1,47 @@
defmodule Towerops.Workers.CloudLatencyProbeWorker do
@moduledoc """
Oban cron worker that dispatches latency probes from all online cloud agents
to all cloud-polled devices.
Runs every 8 hours to collect cross-agent latency data used by
AgentLatencyEvaluator for automatic device reassignment.
Each cloud agent receives a list of devices to ping (5 pings per device).
Results flow back through the existing MonitoringCheck pipeline.
"""
use Oban.Worker, queue: :maintenance
alias Towerops.Agents
require Logger
@impl Oban.Worker
def perform(%Oban.Job{}) do
cloud_pollers = Agents.list_online_cloud_pollers()
devices = Agents.list_cloud_polled_devices()
if cloud_pollers == [] or devices == [] do
Logger.info("Cloud latency probe: no work (#{length(cloud_pollers)} pollers, #{length(devices)} devices)")
:ok
else
dispatch_probes(cloud_pollers, devices)
end
end
defp dispatch_probes(cloud_pollers, devices) do
Logger.info(
"Cloud latency probe: dispatching to #{length(cloud_pollers)} poller(s) " <>
"for #{length(devices)} device(s)"
)
for poller <- cloud_pollers do
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"agent:#{poller.id}:latency_probe",
{:latency_probe_jobs, devices}
)
end
:ok
end
end

4
lib/towerops/workers/device_monitor_worker.ex
View file

@ -143,9 +143,9 @@ defmodule Towerops.Workers.DeviceMonitorWorker do
{:error, _reason} -> {:failure, nil}
end
case Monitoring.create_check(%{
case Monitoring.create_monitoring_check(%{
device_id: device.id,
status: status,
status: to_string(status),
response_time_ms: response_time_ms,
checked_at: now
}) do

52
lib/towerops_web/channels/agent_channel.ex
View file

@ -95,6 +95,11 @@ defmodule ToweropsWeb.AgentChannel do
# Subscribe to token lifecycle events (disable/delete triggers disconnect)
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:lifecycle")
# Subscribe to latency probe requests (cloud pollers only)
if agent_token.is_cloud_poller do
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:latency_probe")
end
# Update last_seen_at and IP on join
remote_ip = get_remote_ip(socket)
_ = Agents.update_agent_token_heartbeat(agent_token.id, remote_ip, %{})
@ -367,6 +372,42 @@ defmodule ToweropsWeb.AgentChannel do
{:noreply, socket}
end
# Handle latency probe requests — send PING jobs for cloud-polled devices
def handle_info({:latency_probe_jobs, devices}, socket) do
jobs =
Enum.flat_map(devices, fn device ->
# 5 pings per device for statistical validity
for i <- 1..5 do
%AgentJob{
job_id: "probe:#{device.id}:#{i}",
job_type: :PING,
device_id: device.id,
snmp_device: %SnmpDevice{
ip: to_string(device.ip_address),
port: 0,
version: "",
community: ""
},
queries: []
}
end
end)
if jobs != [] do
job_list = %AgentJobList{jobs: jobs}
binary = AgentJobList.encode(job_list)
Logger.info("Sending #{length(jobs)} latency probe pings to cloud poller",
agent_token_id: socket.assigns.agent_token_id,
device_count: length(devices)
)
push(socket, "jobs", %{binary: Base.encode64(binary)})
end
{:noreply, socket}
end
@impl true
@spec handle_in(String.t(), map(), socket()) :: {:noreply, socket()}
def handle_in("result", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
@ -1206,20 +1247,21 @@ defmodule ToweropsWeb.AgentChannel do
# Agent timestamps can be unreliable due to clock drift
checked_at = DateTime.truncate(DateTime.utc_now(), :second)
# Convert status string to atom
status = String.to_existing_atom(check.status)
# Keep status as string for storage, convert to atom for device status update
status_string = check.status
status_atom = String.to_existing_atom(status_string)
attrs = %{
device_id: check.device_id,
agent_token_id: agent_token_id,
status: status,
status: status_string,
response_time_ms: check.response_time_ms,
checked_at: checked_at
}
case Monitoring.create_check(attrs) do
case Monitoring.create_monitoring_check(attrs) do
{:ok, _check} ->
update_device_status_from_check(device, status)
update_device_status_from_check(device, status_atom)
:ok
{:error, changeset} ->

1
lib/towerops_web/live/device_live/show.ex
View file

@ -367,7 +367,6 @@ defmodule ToweropsWeb.DeviceLive.Show do
wireless_sensors = Enum.filter(non_transceiver_sensors, &wireless_sensor?/1)
signal_sensors = Enum.filter(non_transceiver_sensors, &signal_sensor?/1)
grouped_transceivers = group_transceiver_sensors(transceiver_sensors)
# TODO: Replace with new checks system metrics
metrics = calculate_metrics([], device)
available_firmware = get_available_firmware(snmp_device)

1
lib/towerops_web/live/device_live/show.html.heex
View file

@ -2184,7 +2184,6 @@
</div>
</div>
</div>
<% _ -> %>
<div class="text-center py-12">
<p class="text-gray-500 dark:text-gray-400">Tab not found</p>

213
test/towerops/agents/cloud_poller_test.exs Normal file
View file

@ -0,0 +1,213 @@
defmodule Towerops.Agents.CloudPollerTest do
use Towerops.DataCase
import Towerops.AccountsFixtures
import Towerops.OrganizationsFixtures
alias Towerops.Agents
alias Towerops.Repo
alias Towerops.Sites
describe "list_online_cloud_pollers/0" do
test "returns cloud pollers seen within 5 minutes" do
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
cloud_poller
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
pollers = Agents.list_online_cloud_pollers()
assert length(pollers) == 1
assert hd(pollers).id == cloud_poller.id
end
test "excludes cloud pollers not seen recently" do
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Stale Cloud")
ten_min_ago = DateTime.utc_now() |> DateTime.add(-10, :minute) |> DateTime.truncate(:second)
cloud_poller
|> Ecto.Changeset.change(%{last_seen_at: ten_min_ago})
|> Repo.update!()
assert Agents.list_online_cloud_pollers() == []
end
test "excludes disabled cloud pollers" do
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Disabled Cloud")
cloud_poller
|> Ecto.Changeset.change(%{
last_seen_at: DateTime.truncate(DateTime.utc_now(), :second),
enabled: false
})
|> Repo.update!()
assert Agents.list_online_cloud_pollers() == []
end
test "excludes regular (non-cloud) agents" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent, _token} = Agents.create_agent_token(org.id, "Regular Agent")
agent
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
assert Agents.list_online_cloud_pollers() == []
end
end
describe "list_cloud_polled_devices/0" do
test "returns devices with no agent assignment" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Unassigned Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
devices = Agents.list_cloud_polled_devices()
assert length(devices) == 1
assert hd(devices).id == device.id
end
test "returns devices assigned to a cloud poller" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Cloud Polled Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
{:ok, _} = Agents.assign_device_to_agent(cloud_poller.id, device.id)
devices = Agents.list_cloud_polled_devices()
assert length(devices) == 1
assert hd(devices).id == device.id
end
test "excludes devices assigned to a local agent" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, local_agent, _token} = Agents.create_agent_token(org.id, "Local Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Local Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
{:ok, _} = Agents.assign_device_to_agent(local_agent.id, device.id)
assert Agents.list_cloud_polled_devices() == []
end
test "excludes devices with monitoring disabled" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, _device} =
Towerops.Devices.create_device(%{
name: "No Monitoring Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: false
})
assert Agents.list_cloud_polled_devices() == []
end
test "includes devices whose site default is a cloud poller" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
{:ok, site} =
Sites.create_site(%{
name: "Cloud Site",
organization_id: org.id,
agent_token_id: cloud_poller.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Site Cloud Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
devices = Agents.list_cloud_polled_devices()
assert length(devices) == 1
assert hd(devices).id == device.id
end
test "excludes devices whose site default is a local agent" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, local_agent, _token} = Agents.create_agent_token(org.id, "Local Agent")
{:ok, site} =
Sites.create_site(%{
name: "Local Site",
organization_id: org.id,
agent_token_id: local_agent.id
})
{:ok, _device} =
Towerops.Devices.create_device(%{
name: "Site Local Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
assert Agents.list_cloud_polled_devices() == []
end
end
end

82
test/towerops/agents/stats_test.exs
View file

@ -8,7 +8,7 @@ defmodule Towerops.Agents.StatsTest do
alias Towerops.Agents.Stats
alias Towerops.Devices.Device
alias Towerops.Monitoring
alias Towerops.Monitoring.Check
alias Towerops.Monitoring.MonitoringCheck
alias Towerops.Repo
alias Towerops.Sites
alias Towerops.Snmp.Device
@ -514,14 +514,14 @@ defmodule Towerops.Agents.StatsTest do
id: Ecto.UUID.generate(),
device_id: device.id,
agent_token_id: agent1.id,
status: :success,
status: "success",
response_time_ms: 50.0,
checked_at: now,
inserted_at: now
}
end
Repo.insert_all(Check, checks_agent1)
Repo.insert_all(MonitoringCheck, checks_agent1)
# Create checks from agent2 with avg 100ms (bulk insert for speed)
checks_agent2 =
@ -530,14 +530,14 @@ defmodule Towerops.Agents.StatsTest do
id: Ecto.UUID.generate(),
device_id: device.id,
agent_token_id: agent2.id,
status: :success,
status: "success",
response_time_ms: 100.0,
checked_at: now,
inserted_at: now
}
end
Repo.insert_all(Check, checks_agent2)
Repo.insert_all(MonitoringCheck, checks_agent2)
latency_by_agent = Stats.get_device_latency_by_agent(device.id)
@ -571,10 +571,10 @@ defmodule Towerops.Agents.StatsTest do
# Create successful checks with 50ms
for _ <- 1..10 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -582,10 +582,10 @@ defmodule Towerops.Agents.StatsTest do
# Create failed checks (should be ignored)
for _ <- 1..5 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :error,
status: "error",
response_time_ms: nil,
checked_at: DateTime.utc_now()
})
@ -619,10 +619,10 @@ defmodule Towerops.Agents.StatsTest do
# Agent1 has 15 checks (above min)
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent1.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -630,10 +630,10 @@ defmodule Towerops.Agents.StatsTest do
# Agent2 has only 5 checks (below min)
for _ <- 1..5 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent2.id,
status: :success,
status: "success",
response_time_ms: 30,
checked_at: DateTime.utc_now()
})
@ -664,10 +664,10 @@ defmodule Towerops.Agents.StatsTest do
# Create recent checks
for _ <- 1..10 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -677,10 +677,10 @@ defmodule Towerops.Agents.StatsTest do
old_time = DateTime.add(DateTime.utc_now(), -25, :hour)
for _ <- 1..5 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: old_time
})
@ -727,18 +727,18 @@ defmodule Towerops.Agents.StatsTest do
# Create checks from both agents
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent1.id,
status: :success,
status: "success",
response_time_ms: 30,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent2.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
@ -804,10 +804,10 @@ defmodule Towerops.Agents.StatsTest do
# Create checks showing this is the only/best agent
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -844,18 +844,18 @@ defmodule Towerops.Agents.StatsTest do
# Create checks: slow agent = 100ms, fast agent = 50ms (50% improvement)
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -897,18 +897,18 @@ defmodule Towerops.Agents.StatsTest do
# Create checks showing fast agent is better
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -946,18 +946,18 @@ defmodule Towerops.Agents.StatsTest do
# Create checks showing fast agent is better
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -997,18 +997,18 @@ defmodule Towerops.Agents.StatsTest do
# Create only 5 checks (below min of 10)
for _ <- 1..5 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
@ -1073,10 +1073,10 @@ defmodule Towerops.Agents.StatsTest do
# Create checks from agent even though device isn't assigned
for _ <- 1..15 do
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: :success,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})

114
test/towerops/monitoring/monitoring_check_test.exs Normal file
View file

@ -0,0 +1,114 @@
defmodule Towerops.Monitoring.MonitoringCheckTest do
use Towerops.DataCase
import Towerops.AccountsFixtures
import Towerops.OrganizationsFixtures
alias Towerops.Agents
alias Towerops.Monitoring
alias Towerops.Monitoring.MonitoringCheck
alias Towerops.Sites
describe "MonitoringCheck schema" do
test "valid changeset with required fields" do
{device, _agent} = create_device_with_agent()
changeset =
MonitoringCheck.changeset(%MonitoringCheck{}, %{
device_id: device.id,
status: "success",
checked_at: DateTime.utc_now()
})
assert changeset.valid?
end
test "invalid changeset missing required fields" do
changeset = MonitoringCheck.changeset(%MonitoringCheck{}, %{})
refute changeset.valid?
assert "can't be blank" in errors_on(changeset).device_id
assert "can't be blank" in errors_on(changeset).status
assert "can't be blank" in errors_on(changeset).checked_at
end
test "valid changeset with all fields" do
{device, agent} = create_device_with_agent()
changeset =
MonitoringCheck.changeset(%MonitoringCheck{}, %{
device_id: device.id,
agent_token_id: agent.id,
status: "success",
response_time_ms: 42.5,
checked_at: DateTime.utc_now()
})
assert changeset.valid?
end
end
describe "Monitoring.create_monitoring_check/1" do
test "inserts a monitoring check record" do
{device, agent} = create_device_with_agent()
now = DateTime.truncate(DateTime.utc_now(), :second)
assert {:ok, check} =
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent.id,
status: "success",
response_time_ms: 12.5,
checked_at: now
})
assert check.device_id == device.id
assert check.agent_token_id == agent.id
assert check.status == "success"
assert check.response_time_ms == 12.5
assert check.checked_at == now
end
test "inserts without agent_token_id" do
{device, _agent} = create_device_with_agent()
now = DateTime.truncate(DateTime.utc_now(), :second)
assert {:ok, check} =
Monitoring.create_monitoring_check(%{
device_id: device.id,
status: "failure",
checked_at: now
})
assert check.device_id == device.id
assert is_nil(check.agent_token_id)
assert check.status == "failure"
assert is_nil(check.response_time_ms)
end
test "rejects invalid attrs" do
assert {:error, changeset} = Monitoring.create_monitoring_check(%{})
refute changeset.valid?
end
end
defp create_device_with_agent do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent, _token} = Agents.create_agent_token(org.id, "Test Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Test Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{device, agent}
end
end

441
test/towerops/workers/agent_latency_evaluator_test.exs
View file

@ -6,7 +6,9 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
import Towerops.OrganizationsFixtures
alias Towerops.Agents
alias Towerops.Agents.Stats
alias Towerops.Monitoring
alias Towerops.Repo
alias Towerops.Sites
alias Towerops.Workers.AgentLatencyEvaluator
@ -15,17 +17,17 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
assert :ok = perform_job(AgentLatencyEvaluator, %{})
end
test "reassigns device to faster agent when 20%+ improvement available" do
test "reassigns device to faster cloud agent when 20%+ improvement available" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, device} =
@ -40,43 +42,42 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
})
# Create checks: slow = 100ms, fast = 50ms (50% improvement)
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify site was reassigned
updated_site = Sites.get_site!(site.id)
assert updated_site.agent_token_id == fast_agent.id
# Device-level assignment should be created to fast cloud agent
assignment = Agents.get_device_assignment(device.id)
assert assignment.agent_token_id == fast_cloud.id
end
test "does not reassign when improvement is below 20% threshold" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent1, _} = Agents.create_agent_token(org.id, "Agent 1")
{:ok, agent2, _} = Agents.create_agent_token(org.id, "Agent 2")
{:ok, cloud1, _} = Agents.create_cloud_poller("Cloud 1")
{:ok, cloud2, _} = Agents.create_cloud_poller("Cloud 2")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: agent1.id
agent_token_id: cloud1.id
})
{:ok, device} =
@ -90,42 +91,40 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks: agent1 = 100ms, agent2 = 85ms (15% improvement, below 20%)
for _ <- 1..15 do
Monitoring.create_check(%{
# Create checks: cloud1 = 100ms, cloud2 = 85ms (15% improvement, below 20%)
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent1.id,
status: :success,
agent_token_id: cloud1.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent2.id,
status: :success,
agent_token_id: cloud2.id,
status: "success",
response_time_ms: 85,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify site was NOT reassigned
updated_site = Sites.get_site!(site.id)
assert updated_site.agent_token_id == agent1.id
# No device assignment should be created
assert Agents.get_device_assignment(device.id) == nil
end
test "reassigns organization-level assignment" do
test "reassigns organization-level cloud poller to device-level" do
user = user_fixture()
{:ok, slow_agent, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_agent, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
org =
organization_fixture(user.id, %{
default_agent_token_id: slow_agent.id
default_agent_token_id: slow_cloud.id
})
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
@ -141,31 +140,33 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks showing fast agent is better
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 150,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify organization was reassigned
# Should create device-level assignment, not change org default
assignment = Agents.get_device_assignment(device.id)
assert assignment.agent_token_id == fast_cloud.id
# Organization default should remain unchanged
updated_org = Towerops.Organizations.get_organization!(org.id)
assert updated_org.default_agent_token_id == fast_agent.id
assert updated_org.default_agent_token_id == slow_cloud.id
end
test "creates device assignment for global default assignment" do
@ -175,7 +176,6 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Global")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Global")
# Set slow cloud as global default
{:ok, _} = Towerops.Settings.set_global_default_cloud_poller(slow_cloud.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
@ -191,33 +191,29 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks showing fast cloud is better
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_cloud.id,
status: :success,
status: "success",
response_time_ms: 200,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_cloud.id,
status: :success,
status: "success",
response_time_ms: 60,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify device assignment was created
assignment = Agents.get_device_assignment(device.id)
assert assignment.agent_token_id == fast_cloud.id
# Cleanup global default
{:ok, _} = Towerops.Settings.set_global_default_cloud_poller(nil)
end
@ -241,7 +237,6 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks from both cloud agents (bulk insert for speed)
now = DateTime.truncate(DateTime.utc_now(), :second)
for_result =
@ -251,7 +246,7 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
id: Ecto.UUID.generate(),
device_id: device.id,
agent_token_id: slow_cloud.id,
status: :success,
status: "success",
response_time_ms: 100.0,
checked_at: now,
inserted_at: now
@ -260,7 +255,7 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
id: Ecto.UUID.generate(),
device_id: device.id,
agent_token_id: fast_cloud.id,
status: :success,
status: "success",
response_time_ms: 50.0,
checked_at: now,
inserted_at: now
@ -270,12 +265,10 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
checks = List.flatten(for_result)
Repo.insert_all(Towerops.Monitoring.Check, checks)
Repo.insert_all(Towerops.Monitoring.MonitoringCheck, checks)
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify device assignment was created to the fast agent
assignment = Agents.get_device_assignment(device.id)
assert assignment != nil, "Expected device assignment to be created"
assert assignment.agent_token_id == fast_cloud.id
@ -284,8 +277,8 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
test "does not reassign direct device assignments" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
@ -301,46 +294,44 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
})
# Create direct device assignment
{:ok, _} = Agents.assign_device_to_agent(slow_agent.id, device.id)
{:ok, _} = Agents.assign_device_to_agent(slow_cloud.id, device.id)
# Create checks showing fast agent is better
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify device assignment was NOT changed
# Device assignment should remain with slow cloud (direct assignment not changed)
assignment = Agents.get_device_assignment(device.id)
assert assignment.agent_token_id == slow_agent.id
assert assignment.agent_token_id == slow_cloud.id
end
test "requires minimum check count" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, device} =
@ -354,51 +345,49 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create only 5 checks (below default minimum of 10)
for _ <- 1..5 do
Monitoring.create_check(%{
# Create only 3 checks (below minimum of 5)
for _ <- 1..3 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify site was NOT reassigned (insufficient data)
updated_site = Sites.get_site!(site.id)
assert updated_site.agent_token_id == slow_agent.id
# No assignment created (insufficient data)
assert Agents.get_device_assignment(device.id) == nil
end
test "handles multiple devices in single evaluation" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site1} =
Sites.create_site(%{
name: "Site 1",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, site2} =
Sites.create_site(%{
name: "Site 2",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, device1} =
@ -423,49 +412,47 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks for both devices
for device_id <- [device1.id, device2.id] do
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device_id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device_id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
end
# Trigger evaluation
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Verify both sites were reassigned
updated_site1 = Sites.get_site!(site1.id)
updated_site2 = Sites.get_site!(site2.id)
# Both devices should get device-level assignments
assignment1 = Agents.get_device_assignment(device1.id)
assignment2 = Agents.get_device_assignment(device2.id)
assert updated_site1.agent_token_id == fast_agent.id
assert updated_site2.agent_token_id == fast_agent.id
assert assignment1.agent_token_id == fast_cloud.id
assert assignment2.agent_token_id == fast_cloud.id
end
test "broadcasts PubSub event when device is reassigned" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, device} =
@ -479,49 +466,44 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks showing significant improvement
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Subscribe to PubSub
Phoenix.PubSub.subscribe(Towerops.PubSub, "devices:latency")
# Trigger evaluation
assert :ok = perform_job(AgentLatencyEvaluator, %{})
# Verify PubSub broadcast was sent
assert_received {:device_reassigned, candidate}
assert candidate.device_id == device.id
assert candidate.best_agent_token_id == fast_agent.id
assert candidate.current_agent_token_id == slow_agent.id
assert candidate.best_agent_token_id == fast_cloud.id
assert candidate.improvement_percent >= 20
end
test "does not broadcast when no reassignment occurs" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent1, _} = Agents.create_agent_token(org.id, "Agent 1")
{:ok, cloud1, _} = Agents.create_cloud_poller("Cloud 1")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: agent1.id
agent_token_id: cloud1.id
})
{:ok, device} =
@ -535,38 +517,35 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks with only one agent (no alternative)
for _ <- 1..15 do
Monitoring.create_check(%{
# Only one cloud agent - no alternative to compare
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: agent1.id,
status: :success,
agent_token_id: cloud1.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
end
# Subscribe to PubSub
Phoenix.PubSub.subscribe(Towerops.PubSub, "devices:latency")
# Trigger evaluation
assert :ok = perform_job(AgentLatencyEvaluator, %{})
# Verify no broadcast was sent
refute_received {:device_reassigned, _}
end
test "handles reassignment failures gracefully" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_agent, _} = Agents.create_agent_token(org.id, "Slow Agent")
{:ok, fast_agent, _} = Agents.create_agent_token(org.id, "Fast Agent")
{:ok, slow_cloud, _} = Agents.create_cloud_poller("Slow Cloud")
{:ok, fast_cloud, _} = Agents.create_cloud_poller("Fast Cloud")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: slow_agent.id
agent_token_id: slow_cloud.id
})
{:ok, device} =
@ -580,30 +559,198 @@ defmodule Towerops.Workers.AgentLatencyEvaluatorTest do
snmp_version: "2c"
})
# Create checks showing improvement
for _ <- 1..15 do
Monitoring.create_check(%{
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_agent.id,
status: :success,
agent_token_id: slow_cloud.id,
status: "success",
response_time_ms: 100,
checked_at: DateTime.utc_now()
})
Monitoring.create_check(%{
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_agent.id,
status: :success,
agent_token_id: fast_cloud.id,
status: "success",
response_time_ms: 50,
checked_at: DateTime.utc_now()
})
end
# Delete the site to cause a failure during reassignment
Sites.delete_site(site)
# Delete the device to cause a failure during reassignment
Towerops.Devices.delete_device(device)
# Worker should complete successfully despite the error
assert :ok = perform_job(AgentLatencyEvaluator, %{})
end
test "ignores local agent latency data" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, slow_local, _} = Agents.create_agent_token(org.id, "Slow Local")
{:ok, fast_local, _} = Agents.create_agent_token(org.id, "Fast Local")
{:ok, site} =
Sites.create_site(%{
name: "Local Site",
organization_id: org.id,
agent_token_id: slow_local.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: slow_local.id,
status: "success",
response_time_ms: 100.0,
checked_at: DateTime.utc_now()
})
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: fast_local.id,
status: "success",
response_time_ms: 30.0,
checked_at: DateTime.utc_now()
})
end
Phoenix.PubSub.subscribe(Towerops.PubSub, "devices:latency")
assert :ok = AgentLatencyEvaluator.perform(%Oban.Job{args: %{}})
# Local agents should not trigger reassignment
refute_receive {:device_reassigned, _}, 200
end
end
describe "find_reassignment_candidates with cloud_only" do
test "filters to cloud pollers only" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, cloud1, _} = Agents.create_cloud_poller("Cloud 1")
{:ok, cloud2, _} = Agents.create_cloud_poller("Cloud 2")
{:ok, local, _} = Agents.create_agent_token(org.id, "Local Agent")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: cloud1.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: cloud1.id,
status: "success",
response_time_ms: 100.0,
checked_at: DateTime.utc_now()
})
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: cloud2.id,
status: "success",
response_time_ms: 30.0,
checked_at: DateTime.utc_now()
})
# Local agent has best latency but should be excluded
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: local.id,
status: "success",
response_time_ms: 10.0,
checked_at: DateTime.utc_now()
})
end
candidates =
Stats.find_reassignment_candidates(
cloud_only: true,
min_checks_per_agent: 5,
hours_ago: 48
)
assert length(candidates) == 1
candidate = hd(candidates)
# Should pick cloud2 as best, not local agent
assert candidate.best_agent_token_id == cloud2.id
end
test "returns no candidates when cloud_only and only local agents" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, local1, _} = Agents.create_agent_token(org.id, "Local 1")
{:ok, local2, _} = Agents.create_agent_token(org.id, "Local 2")
{:ok, site} =
Sites.create_site(%{
name: "Test Site",
organization_id: org.id,
agent_token_id: local1.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
for _ <- 1..10 do
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: local1.id,
status: "success",
response_time_ms: 100.0,
checked_at: DateTime.utc_now()
})
Monitoring.create_monitoring_check(%{
device_id: device.id,
agent_token_id: local2.id,
status: "success",
response_time_ms: 30.0,
checked_at: DateTime.utc_now()
})
end
candidates =
Stats.find_reassignment_candidates(
cloud_only: true,
min_checks_per_agent: 5,
hours_ago: 48
)
assert candidates == []
end
end
end

116
test/towerops/workers/cloud_latency_probe_worker_test.exs Normal file
View file

@ -0,0 +1,116 @@
defmodule Towerops.Workers.CloudLatencyProbeWorkerTest do
use Towerops.DataCase
import Towerops.AccountsFixtures
import Towerops.OrganizationsFixtures
alias Towerops.Agents
alias Towerops.Repo
alias Towerops.Sites
alias Towerops.Workers.CloudLatencyProbeWorker
describe "perform/1" do
test "broadcasts latency probe jobs for each online cloud poller" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
cloud_poller
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
# Subscribe to the probe topic for this cloud poller
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{cloud_poller.id}:latency_probe")
assert :ok == CloudLatencyProbeWorker.perform(%Oban.Job{})
# Should receive probe jobs containing the device
assert_receive {:latency_probe_jobs, devices}, 1000
device_ids = Enum.map(devices, & &1.id)
assert device.id in device_ids
end
test "does nothing when no cloud pollers are online" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, _device} =
Towerops.Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
assert :ok == CloudLatencyProbeWorker.perform(%Oban.Job{})
end
test "does nothing when no cloud-polled devices exist" do
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
cloud_poller
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{cloud_poller.id}:latency_probe")
assert :ok == CloudLatencyProbeWorker.perform(%Oban.Job{})
# Should not receive any probe jobs
refute_receive {:latency_probe_jobs, _}, 200
end
test "skips devices assigned to local agents" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, cloud_poller, _token} = Agents.create_cloud_poller("Cloud 1")
{:ok, local_agent, _token} = Agents.create_agent_token(org.id, "Local Agent")
cloud_poller
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Local Router",
ip_address: "192.168.1.1",
site_id: site.id,
organization_id: org.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
# Assign to local agent — should not be probed
{:ok, _} = Agents.assign_device_to_agent(local_agent.id, device.id)
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{cloud_poller.id}:latency_probe")
assert :ok == CloudLatencyProbeWorker.perform(%Oban.Job{})
refute_receive {:latency_probe_jobs, _}, 200
end
end
end