diff --git a/config/dev.exs b/config/dev.exs index 4cbf48ad..c0b48381 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -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 diff --git a/config/runtime.exs b/config/runtime.exs index b537a31d..10ac33ac 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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 diff --git a/lib/towerops/agents.ex b/lib/towerops/agents.ex index a51850af..ec824c7a 100644 --- a/lib/towerops/agents.ex +++ b/lib/towerops/agents.ex @@ -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 """ diff --git a/lib/towerops/agents/stats.ex b/lib/towerops/agents/stats.ex index 2d423173..dccc891c 100644 --- a/lib/towerops/agents/stats.ex +++ b/lib/towerops/agents/stats.ex @@ -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 diff --git a/lib/towerops/monitoring.ex b/lib/towerops/monitoring.ex index 6b1adb2e..d6e3f5c0 100644 --- a/lib/towerops/monitoring.ex +++ b/lib/towerops/monitoring.ex @@ -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 diff --git a/lib/towerops/monitoring/executors/http_executor.ex b/lib/towerops/monitoring/executors/http_executor.ex index afd21a1b..c19aac69 100644 --- a/lib/towerops/monitoring/executors/http_executor.ex +++ b/lib/towerops/monitoring/executors/http_executor.ex @@ -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 diff --git a/lib/towerops/monitoring/monitoring_check.ex b/lib/towerops/monitoring/monitoring_check.ex new file mode 100644 index 00000000..f226a11b --- /dev/null +++ b/lib/towerops/monitoring/monitoring_check.ex @@ -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 diff --git a/lib/towerops/workers/agent_latency_evaluator.ex b/lib/towerops/workers/agent_latency_evaluator.ex index 3d3d56f4..0efbcb7c 100644 --- a/lib/towerops/workers/agent_latency_evaluator.ex +++ b/lib/towerops/workers/agent_latency_evaluator.ex @@ -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, diff --git a/lib/towerops/workers/cloud_latency_probe_worker.ex b/lib/towerops/workers/cloud_latency_probe_worker.ex new file mode 100644 index 00000000..af828c69 --- /dev/null +++ b/lib/towerops/workers/cloud_latency_probe_worker.ex @@ -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 diff --git a/lib/towerops/workers/device_monitor_worker.ex b/lib/towerops/workers/device_monitor_worker.ex index 754c8735..e88bf8f6 100644 --- a/lib/towerops/workers/device_monitor_worker.ex +++ b/lib/towerops/workers/device_monitor_worker.ex @@ -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 diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index b176a300..5f09fa83 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -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} -> diff --git a/lib/towerops_web/live/device_live/show.ex b/lib/towerops_web/live/device_live/show.ex index db4e4aaa..f7b8863b 100644 --- a/lib/towerops_web/live/device_live/show.ex +++ b/lib/towerops_web/live/device_live/show.ex @@ -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) diff --git a/lib/towerops_web/live/device_live/show.html.heex b/lib/towerops_web/live/device_live/show.html.heex index 3fe1e008..986557f0 100644 --- a/lib/towerops_web/live/device_live/show.html.heex +++ b/lib/towerops_web/live/device_live/show.html.heex @@ -2184,7 +2184,6 @@ - <% _ -> %>

Tab not found

diff --git a/test/towerops/agents/cloud_poller_test.exs b/test/towerops/agents/cloud_poller_test.exs new file mode 100644 index 00000000..62b2daf6 --- /dev/null +++ b/test/towerops/agents/cloud_poller_test.exs @@ -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 diff --git a/test/towerops/agents/stats_test.exs b/test/towerops/agents/stats_test.exs index 85e07f95..58bddf19 100644 --- a/test/towerops/agents/stats_test.exs +++ b/test/towerops/agents/stats_test.exs @@ -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() }) diff --git a/test/towerops/monitoring/monitoring_check_test.exs b/test/towerops/monitoring/monitoring_check_test.exs new file mode 100644 index 00000000..219bf138 --- /dev/null +++ b/test/towerops/monitoring/monitoring_check_test.exs @@ -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 diff --git a/test/towerops/workers/agent_latency_evaluator_test.exs b/test/towerops/workers/agent_latency_evaluator_test.exs index b9e862d1..8d681610 100644 --- a/test/towerops/workers/agent_latency_evaluator_test.exs +++ b/test/towerops/workers/agent_latency_evaluator_test.exs @@ -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 diff --git a/test/towerops/workers/cloud_latency_probe_worker_test.exs b/test/towerops/workers/cloud_latency_probe_worker_test.exs new file mode 100644 index 00000000..e2b34d46 --- /dev/null +++ b/test/towerops/workers/cloud_latency_probe_worker_test.exs @@ -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