From 48dc4b852e3afa7455dec3dcef3f43fcde72328f Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 8 Feb 2026 13:44:21 -0600 Subject: [PATCH] feat: add Phoenix support for agent-side ping monitoring Enables Phoenix to send PING jobs to agents and receive monitoring check results, fixing the "unknown" status for devices assigned to local (non-cloud) agents. Changes: - Rename JobType::MONITOR to JobType::PING in protobuf for clarity - Add validate_monitoring_check_message() to Validator module - Add "monitoring_check" channel handler to receive ping results - Implement store_monitoring_check() to save results to database - Add build_ping_job() to generate PING jobs for agents - Update build_jobs_for_device() to include ping jobs when monitoring_enabled=true Phoenix now sends PING jobs alongside SNMP/MikroTik jobs, and agents send back MonitoringCheck results that are stored in the monitoring_checks table with proper agent_token_id tracking. Co-Authored-By: Claude Sonnet 4.5 --- lib/towerops/agent/validator.ex | 13 +++ lib/towerops_web/channels/agent_channel.ex | 105 ++++++++++++++++++++- priv/proto/agent.proto | 1 + 3 files changed, 118 insertions(+), 1 deletion(-) diff --git a/lib/towerops/agent/validator.ex b/lib/towerops/agent/validator.ex index 9689db7b..80c602c0 100644 --- a/lib/towerops/agent/validator.ex +++ b/lib/towerops/agent/validator.ex @@ -167,6 +167,19 @@ defmodule Towerops.Agent.Validator do end end + @doc """ + Validate MonitoringCheck message from binary protobuf. + + Decodes the binary and validates device_id, status enum, response_time range, and timestamp. + """ + @spec validate_monitoring_check_message(binary()) :: validation_result(MonitoringCheck.t()) + def validate_monitoring_check_message(binary) when is_binary(binary) do + with {:ok, check} <- safe_decode(MonitoringCheck, binary), + :ok <- validate_monitoring_check(check) do + {:ok, check} + end + end + ## Private Validation Functions # Safe decode with error handling diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index 8a003e27..a9bdd191 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -33,6 +33,7 @@ defmodule ToweropsWeb.AgentChannel do alias Towerops.Devices alias Towerops.Devices.BackupRequests alias Towerops.Devices.MikrotikBackups + alias Towerops.Monitoring alias Towerops.Snmp alias Towerops.Snmp.AgentDiscovery alias Towerops.Snmp.Discovery @@ -389,6 +390,43 @@ defmodule ToweropsWeb.AgentChannel do end end + @spec handle_in(String.t(), %{required(String.t()) => base64_string()}, socket()) :: + {:noreply, socket()} + def handle_in("monitoring_check", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do + with {:ok, binary} <- safe_base64_decode(binary_b64), + {:ok, check} <- Validator.validate_monitoring_check_message(binary) do + maybe_debug_log(socket, "Received monitoring check from agent", + device_id: check.device_id, + status: check.status, + response_time_ms: check.response_time_ms, + binary_size: byte_size(binary_b64) + ) + + # Store monitoring check result in database + _ = store_monitoring_check(check, socket) + + {:noreply, socket} + else + {:error, {type, message}} -> + Logger.error("Invalid monitoring check from agent: #{type} - #{message}", + agent_token_id: socket.assigns.agent_token_id, + error_type: type, + error_message: message, + binary_size: byte_size(binary_b64) + ) + + {:noreply, socket} + + {:error, :base64_decode_failed} -> + Logger.error("Failed to decode monitoring check (invalid base64)", + agent_token_id: socket.assigns.agent_token_id, + binary_size: byte_size(binary_b64) + ) + + {:noreply, socket} + end + end + # Private helpers @spec handle_validated_mikrotik_result(MikrotikResult.t(), socket()) :: {:noreply, socket()} @@ -436,8 +474,14 @@ defmodule ToweropsWeb.AgentChannel do build_mikrotik_job(device) end + # Build ping job if monitoring is enabled + ping_job = + if device.monitoring_enabled do + build_ping_job(device) + end + # Return list of jobs (filter out nil) - Enum.reject([snmp_job, mikrotik_job], &is_nil/1) + Enum.reject([snmp_job, mikrotik_job, ping_job], &is_nil/1) end defp needs_discovery?(device) do @@ -589,6 +633,22 @@ defmodule ToweropsWeb.AgentChannel do } end + defp build_ping_job(device) do + %AgentJob{ + job_id: "ping:#{device.id}", + job_type: :PING, + device_id: device.id, + # Ping only needs IP address - use minimal SnmpDevice message + snmp_device: %SnmpDevice{ + ip: to_string(device.ip_address), + port: 0, + version: "", + community: "" + }, + queries: [] + } + end + defp build_live_polling_job(device, sensor_oids, reply_topic) do snmp_credentials = resolve_snmp_credentials(device) @@ -846,6 +906,49 @@ defmodule ToweropsWeb.AgentChannel do end end + defp store_monitoring_check(check, socket) do + organization_id = socket.assigns.organization_id + agent_token_id = socket.assigns.agent_token_id + + with {:ok, device} <- fetch_device(check.device_id), + :ok <- verify_device_organization(device, organization_id) do + # Convert protobuf timestamp to DateTime + checked_at = DateTime.from_unix!(check.timestamp, :second) + + # Convert status string to atom + status = String.to_existing_atom(check.status) + + attrs = %{ + device_id: check.device_id, + agent_token_id: agent_token_id, + status: status, + response_time_ms: check.response_time_ms, + checked_at: checked_at + } + + case Monitoring.create_check(attrs) do + {:ok, _check} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to store monitoring check", + device_id: check.device_id, + errors: inspect(changeset.errors) + ) + + {:error, :storage_failed} + end + else + {:error, :device_not_found} -> + Logger.error("Device not found for monitoring check: #{check.device_id}") + {:error, :device_not_found} + + {:error, :wrong_organization} -> + Logger.error("Device #{check.device_id} not in agent's organization") + {:error, :wrong_organization} + end + end + defp fetch_device(device_id) do case Devices.get_device_with_details(device_id) do nil -> {:error, :device_not_found} diff --git a/priv/proto/agent.proto b/priv/proto/agent.proto index 45149f1c..44959c6e 100644 --- a/priv/proto/agent.proto +++ b/priv/proto/agent.proto @@ -114,6 +114,7 @@ enum JobType { POLL = 1; MIKROTIK = 2; TEST_CREDENTIALS = 3; + PING = 4; } enum QueryType {