graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
towerops/lib/towerops_web/channels/agent_channel.ex
mayor bf4fee4846
feat: Add comprehensive agent polling for all discoverable data 
Extended agent polling to collect all available SNMP data during regular
polls, not just sensors and interfaces. This ensures data stays current
and reduces reliance on periodic discovery runs.

Changes:
- Added polling queries for: neighbors (LLDP/CDP), ARP tables, MAC tables,
  IP addresses, HOST-RESOURCES processors and storage
- Process additional data using Replay adapter pattern to reuse existing
  parsing logic from discovery modules
- Made discovery helper functions public for reuse in polling context:
  save_neighbors/2, save_arp_entries/3, sync_ip_addresses/2,
  sync_processors/2, sync_storage/2
- Optimized by fetching interfaces once and reusing across operations
- Fixed compilation warnings by using proper guard syntax (list != [])

Technical details:
- Detection threshold: polls with more OIDs than sensors+interfaces
  are processed for additional data
- Background processing: spawns async process to avoid blocking metric
  recording
- Replay adapter: allows reusing discovery parsing logic without
  re-querying SNMP device
2026-02-06 11:18:36 -06:00

1260 lines
39 KiB
Elixir
Raw Blame History

defmodule ToweropsWeb.AgentChannel do
@moduledoc """
Phoenix channel for bidirectional agent communication.
Handles WebSocket connections from remote SNMP polling agents, distributing
polling jobs and collecting metrics via Protocol Buffers binary encoding.
## Messages from Server → Agent (binary protobuf)
- "jobs" - List of SNMP jobs to execute
- "job" - Single new job (pushed when device added)
## Messages from Agent → Server (binary protobuf)
- "result" - SNMP query results
- "heartbeat" - Agent metadata (version, hostname, uptime)
- "error" - Job execution errors
All messages use Protocol Buffers binary encoding for efficiency.
"""
use ToweropsWeb, :channel
alias Towerops.Agent.AgentJob
alias Towerops.Agent.AgentJobList
alias Towerops.Agent.MikrotikCommand
alias Towerops.Agent.MikrotikDevice
alias Towerops.Agent.MikrotikResult
alias Towerops.Agent.SnmpDevice
alias Towerops.Agent.SnmpQuery
alias Towerops.Agent.Validator
alias Towerops.Agents
alias Towerops.Devices
alias Towerops.Devices.BackupRequests
alias Towerops.Devices.MikrotikBackups
alias Towerops.Snmp
alias Towerops.Snmp.AgentDiscovery
require Logger
@typep socket :: Phoenix.Socket.t()
@typep base64_string :: String.t()
@typep protobuf_binary :: binary()
@impl true
@spec join(String.t(), map(), Phoenix.Socket.t()) ::
{:ok, Phoenix.Socket.t()} | {:error, map()}
def join("agent:" <> agent_id, %{"token" => token} = payload, socket) do
Logger.info("Agent join attempt",
topic: "agent:#{agent_id}",
has_token: not is_nil(token),
token_length: if(is_binary(token), do: byte_size(token), else: 0),
payload_keys: Map.keys(payload)
)
# Verify agent token from join payload
case Agents.verify_agent_token(token) do
{:ok, agent_token} ->
socket =
socket
|> assign(:agent_token_id, agent_token.id)
|> assign(:organization_id, agent_token.organization_id)
|> assign(:debug_enabled, agent_token.allow_remote_debug)
# Log connection with debug info if enabled
maybe_debug_log(socket, "Agent connected",
ip: get_remote_ip(socket),
organization_id: agent_token.organization_id
)
# Subscribe to assignment changes for this agent
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:assignments")
# Subscribe to discovery requests for this agent
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:discovery")
# Subscribe to backup requests for this agent
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:backup")
# Subscribe to credential test requests for this agent
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:credential_test")
# Subscribe to live poll requests for this agent
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:live_poll")
# Update last_seen_at and IP on join
remote_ip = get_remote_ip(socket)
_ = Agents.update_agent_token_heartbeat(agent_token.id, remote_ip, %{})
# Send initial job list
send(self(), :send_jobs)
{:ok, socket}
{:error, _} ->
{:error, %{reason: "unauthorized"}}
end
end
def join("agent:" <> _agent_id, _payload, _socket) do
{:error, %{reason: "missing token"}}
end
@impl true
@spec handle_info(:send_jobs, Phoenix.Socket.t()) :: {:noreply, Phoenix.Socket.t()}
def handle_info(:send_jobs, socket) do
jobs = build_jobs_for_agent(socket.assigns.agent_token_id)
job_list = %AgentJobList{jobs: jobs}
binary = AgentJobList.encode(job_list)
Logger.info("Sending #{length(jobs)} jobs to agent: #{inspect(Enum.map(jobs, & &1.job_id))}")
push(socket, "jobs", %{binary: Base.encode64(binary)})
{:noreply, socket}
end
# Handle PubSub broadcast when device assignments change
def handle_info({:assignments_changed, _event}, socket) do
Logger.info("Agent assignments changed, sending updated job list",
agent_token_id: socket.assigns.agent_token_id
)
send(self(), :send_jobs)
{:noreply, socket}
end
# Handle PubSub broadcast when discovery is requested for a device
def handle_info({:discovery_requested, device_id}, socket) do
Logger.info("Discovery requested for device, sending discovery job to agent",
agent_token_id: socket.assigns.agent_token_id,
device_id: device_id
)
case Devices.get_device_with_details(device_id) do
nil ->
Logger.error("Device not found for discovery request", device_id: device_id)
{:noreply, socket}
device ->
# Build and send discovery job to agent
job = build_discovery_job(device)
job_list = %AgentJobList{jobs: [job]}
binary = AgentJobList.encode(job_list)
push(socket, "discovery_job", %{binary: Base.encode64(binary)})
{:noreply, socket}
end
end
# Handle PubSub broadcast when backup is requested for a device
def handle_info({:backup_requested, job}, socket) do
Logger.info("Backup requested for device, sending backup job to agent",
agent_token_id: socket.assigns.agent_token_id,
device_id: job.device_id,
job_id: job.job_id
)
job_list = %AgentJobList{jobs: [job]}
binary = AgentJobList.encode(job_list)
push(socket, "backup_job", %{binary: Base.encode64(binary)})
{:noreply, socket}
end
# Handle PubSub broadcast when credential test is requested
def handle_info({:credential_test_requested, test_id, snmp_config}, socket) do
Logger.info("Credential test requested, sending test job to agent",
agent_token_id: socket.assigns.agent_token_id,
test_id: test_id
)
# Build SNMP device from config
snmp_device = %SnmpDevice{
ip: snmp_config[:ip],
port: snmp_config[:port],
version: snmp_config[:version],
community: snmp_config[:community],
v3_security_level: snmp_config[:v3_security_level],
v3_username: snmp_config[:v3_username],
v3_auth_protocol: snmp_config[:v3_auth_protocol],
v3_auth_password: snmp_config[:v3_auth_password],
v3_priv_protocol: snmp_config[:v3_priv_protocol],
v3_priv_password: snmp_config[:v3_priv_password]
}
# Build test credentials job
job = %AgentJob{
job_id: test_id,
job_type: :TEST_CREDENTIALS,
device_id: test_id,
snmp_device: snmp_device,
queries: [],
mikrotik_device: nil,
mikrotik_commands: []
}
job_list = %AgentJobList{jobs: [job]}
binary = AgentJobList.encode(job_list)
push(socket, "jobs", %{binary: Base.encode64(binary)})
{:noreply, socket}
end
# Handle PubSub broadcast when live polling is requested for a device
def handle_info({:live_poll_requested, device_id, sensor_oids, reply_topic}, socket) do
maybe_debug_log(socket, "Live poll requested for device",
device_id: device_id,
sensor_count: length(sensor_oids),
reply_topic: reply_topic
)
case Devices.get_device_with_details(device_id) do
nil ->
Logger.error("Device not found for live poll request", device_id: device_id)
{:noreply, socket}
device ->
# Build live polling job with only requested sensor OIDs
job = build_live_polling_job(device, sensor_oids, reply_topic)
job_list = %AgentJobList{jobs: [job]}
binary = AgentJobList.encode(job_list)
maybe_debug_log(socket, "Sending live poll job to agent",
device_id: device_id,
device_name: device.name,
oid_count: length(sensor_oids)
)
push(socket, "jobs", %{binary: Base.encode64(binary)})
{:noreply, socket}
end
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
with {:ok, binary} <- safe_base64_decode(binary_b64),
{:ok, result} <- Validator.validate_snmp_result(binary) do
maybe_debug_log(socket, "Received SNMP result from agent",
device_id: result.device_id,
job_type: result.job_type,
job_id: result.job_id,
binary_size: byte_size(binary_b64),
oid_count: map_size(result.oid_values)
)
# Check if this is a live poll result
if String.starts_with?(result.job_id, "live_poll:") do
handle_live_poll_result(result, socket)
else
_ = process_snmp_result(socket.assigns.organization_id, result, socket)
end
{:noreply, socket}
else
{:error, {type, message}} ->
Logger.error("Invalid SNMP result 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 SNMP result (invalid base64)",
agent_token_id: socket.assigns.agent_token_id,
binary_size: byte_size(binary_b64)
)
{:noreply, socket}
end
end
@spec handle_in(String.t(), %{required(String.t()) => base64_string()}, socket()) ::
{:noreply, socket()}
def handle_in("heartbeat", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
with {:ok, binary} <- safe_base64_decode(binary_b64),
{:ok, heartbeat} <- Validator.validate_heartbeat(binary) do
metadata = %{
"version" => heartbeat.version,
"hostname" => heartbeat.hostname,
"uptime_seconds" => heartbeat.uptime_seconds
}
_ =
Agents.update_agent_token_heartbeat(
socket.assigns.agent_token_id,
heartbeat.ip_address,
metadata
)
{:noreply, socket}
else
{:error, {type, message}} ->
Logger.error("Invalid heartbeat from agent",
agent_token_id: socket.assigns.agent_token_id,
error_type: type,
error_message: message
)
{:noreply, socket}
end
end
def handle_in("error", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
with {:ok, binary} <- safe_base64_decode(binary_b64),
{:ok, error} <- Validator.validate_agent_error(binary) do
maybe_debug_log(socket, "Agent job error",
device_id: error.device_id,
error_message: error.error_message,
binary_size: byte_size(binary_b64)
)
Logger.error("Agent job error",
agent_token_id: socket.assigns.agent_token_id,
device_id: error.device_id,
error: error.error_message
)
{:noreply, socket}
else
{:error, {type, message}} ->
Logger.error("Invalid error message from agent",
agent_token_id: socket.assigns.agent_token_id,
error_type: type,
error_message: message
)
{:noreply, socket}
end
end
def handle_in("mikrotik_result", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
with {:ok, binary} <- safe_base64_decode(binary_b64),
{:ok, result} <- Validator.validate_mikrotik_result(binary) do
handle_validated_mikrotik_result(result, socket)
else
{:error, {type, message}} ->
Logger.error("Invalid MikroTik result from agent",
agent_token_id: socket.assigns.agent_token_id,
error_type: type,
error_message: message
)
{:noreply, socket}
end
end
def handle_in("credential_test_result", %{"binary" => binary_b64}, socket) when is_binary(binary_b64) do
with {:ok, binary} <- safe_base64_decode(binary_b64),
{:ok, result} <- Validator.validate_credential_test_result(binary) do
maybe_debug_log(socket, "Received credential test result from agent",
test_id: result.test_id,
success: result.success,
has_error: result.error_message != ""
)
# Broadcast result to the requesting LiveView via PubSub
# The test_id contains the device_id, so we can broadcast to that topic
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"credential_test:#{result.test_id}",
{:credential_test_result, result}
)
{:noreply, socket}
else
{:error, {type, message}} ->
Logger.error(
"Invalid credential test result 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 credential test result (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()}
defp handle_validated_mikrotik_result(result, socket) do
_binary = MikrotikResult.encode(result)
maybe_debug_log(socket, "Received MikroTik result from agent",
device_id: result.device_id,
job_id: result.job_id,
has_error: result.error != "",
sentence_count: length(result.sentences)
)
# Check if this is a backup job by the job_id prefix
if String.starts_with?(result.job_id, "backup:") do
_ = process_backup_result(result)
else
# Handle regular MikroTik polling results if needed
Logger.debug("Received non-backup MikroTik result", job_id: result.job_id)
end
{:noreply, socket}
end
@spec build_jobs_for_agent(Ecto.UUID.t()) :: [AgentJob.t()]
defp build_jobs_for_agent(agent_token_id) do
agent_token_id
|> Agents.list_agent_polling_targets()
|> Enum.flat_map(&build_jobs_for_device/1)
end
defp build_jobs_for_device(device) do
# Build SNMP job (discovery or polling)
snmp_job =
if needs_discovery?(device) do
build_discovery_job(device)
else
build_polling_job(device)
end
# Build MikroTik job only during discovery (not during regular polling)
# MikroTik commands are discovery-type operations that should run once per 24h
mikrotik_job =
if needs_discovery?(device) and mikrotik_device?(device) do
build_mikrotik_job(device)
end
# Return list of jobs (filter out nil)
Enum.reject([snmp_job, mikrotik_job], &is_nil/1)
end
defp needs_discovery?(device) do
# Need discovery if no SNMP device or last discovery was >24 hours ago
is_nil(device.snmp_device) or
is_nil(device.last_discovery_at) or
DateTime.diff(DateTime.utc_now(), device.last_discovery_at, :hour) > 24
end
defp mikrotik_device?(device) do
# Check if device is MikroTik based on SNMP discovery data
device.snmp_device &&
(String.contains?(device.snmp_device.manufacturer || "", "MikroTik") ||
String.contains?(device.snmp_device.sys_descr || "", "RouterOS"))
end
# Resolves SNMP credentials for a device.
# For SNMPv3, returns a map with resolved credentials via cascade.
# For v1/v2c, returns a simple map with community string and version.
defp resolve_snmp_credentials(device) do
if device.snmp_version == "3" do
Devices.get_snmpv3_config(device)
else
%{
community: Devices.resolve_snmp_community(device),
version: device.snmp_version
}
end
end
# Checks if SNMP credentials are present (not nil/empty).
defp credentials_present?(%{community: community}) when is_binary(community) do
community != ""
end
defp credentials_present?(%{username: username}) when is_binary(username) do
username != ""
end
defp credentials_present?(_), do: false
# Builds the SnmpDevice protobuf message with appropriate credentials.
# For v1/v2c: Uses community string
# For v3: Uses SNMPv3 credentials (security_level, username, auth/priv protocols/passwords)
defp build_snmp_device_message(device, snmp_config) do
if device.snmp_version == "3" do
build_v3_snmp_device(device, snmp_config)
else
build_v2c_snmp_device(device, snmp_config)
end
end
defp build_v3_snmp_device(device, snmp_config) do
%SnmpDevice{
ip: device.ip_address,
version: device.snmp_version,
port: device.snmp_port || 161,
transport: device.snmp_transport || "udp",
community: "",
v3_security_level: snmp_config.security_level || "",
v3_username: snmp_config.username || "",
v3_auth_protocol: snmp_config.auth_protocol || "",
v3_auth_password: snmp_config.auth_password || "",
v3_priv_protocol: snmp_config.priv_protocol || "",
v3_priv_password: snmp_config.priv_password || ""
}
end
defp build_v2c_snmp_device(device, snmp_config) do
community = snmp_config.community || ""
%SnmpDevice{
ip: device.ip_address,
version: device.snmp_version,
port: device.snmp_port || 161,
transport: device.snmp_transport || "udp",
community: community
}
end
defp build_discovery_job(device) do
snmp_credentials = resolve_snmp_credentials(device)
maybe_debug_log(%{assigns: %{agent_token_id: "system"}}, "Building discovery job",
device_id: device.id,
device_name: device.name,
snmp_version: device.snmp_version,
credentials_present: credentials_present?(snmp_credentials),
credential_source:
if(device.snmp_version == "3",
do: Map.get(snmp_credentials, :source),
else: device.snmp_community_source
)
)
%AgentJob{
job_id: "discover:#{device.id}",
job_type: :DISCOVER,
device_id: device.id,
snmp_device: build_snmp_device_message(device, snmp_credentials),
queries: build_discovery_queries()
}
end
defp build_polling_job(device) do
_snmp_device = device.snmp_device
snmp_credentials = resolve_snmp_credentials(device)
maybe_debug_log(%{assigns: %{agent_token_id: "system"}}, "Building polling job",
device_id: device.id,
device_name: device.name,
snmp_version: device.snmp_version,
credentials_present: credentials_present?(snmp_credentials),
credential_source:
if(device.snmp_version == "3",
do: Map.get(snmp_credentials, :source),
else: device.snmp_community_source
),
site_id: device.site_id,
site_loaded: not is_nil(Map.get(device, :site))
)
%AgentJob{
job_id: "poll:#{device.id}",
job_type: :POLL,
device_id: device.id,
snmp_device: build_snmp_device_message(device, snmp_credentials),
queries: build_polling_queries(device)
}
end
defp build_live_polling_job(device, sensor_oids, reply_topic) do
snmp_credentials = resolve_snmp_credentials(device)
%AgentJob{
job_id: "live_poll:#{device.id}:#{reply_topic}",
job_type: :POLL,
device_id: device.id,
snmp_device: build_snmp_device_message(device, snmp_credentials),
queries: [
%SnmpQuery{
query_type: :GET,
oids: sensor_oids
}
]
}
end
# Handle live poll result - broadcast to reply topic instead of saving
defp handle_live_poll_result(result, socket) do
# Extract reply topic from job_id: "live_poll:device_id:reply_topic"
case String.split(result.job_id, ":", parts: 3) do
["live_poll", _device_id, reply_topic] ->
maybe_debug_log(socket, "Broadcasting live poll result",
reply_topic: reply_topic,
oid_count: map_size(result.oid_values)
)
# Broadcast OID values directly to the waiting LiveView
Phoenix.PubSub.broadcast(
Towerops.PubSub,
reply_topic,
{:live_poll_result, result.oid_values}
)
_ ->
Logger.warning("Invalid live poll job_id format: #{result.job_id}")
end
end
defp build_discovery_queries do
[
# System info (GET)
%SnmpQuery{
query_type: :GET,
oids: [
# sysDescr
"1.3.6.1.2.1.1.1.0",
# sysObjectID
"1.3.6.1.2.1.1.2.0",
# sysUpTime
"1.3.6.1.2.1.1.3.0",
# sysContact
"1.3.6.1.2.1.1.4.0",
# sysName
"1.3.6.1.2.1.1.5.0",
# sysLocation
"1.3.6.1.2.1.1.6.0"
]
},
# Interface table (WALK)
%SnmpQuery{
query_type: :WALK,
# ifTable
oids: ["1.3.6.1.2.1.2.2.1"]
},
# Interface extensions (WALK)
%SnmpQuery{
query_type: :WALK,
# ifXTable
oids: ["1.3.6.1.2.1.31.1.1.1"]
},
# Sensor discovery - ENTITY-SENSOR-MIB (WALK)
%SnmpQuery{
query_type: :WALK,
oids: [
# ENTITY-SENSOR-MIB - sensor tables
"1.3.6.1.2.1.99.1.1.1",
# ENTITY-MIB - physical entity descriptions (needed for sensor names)
"1.3.6.1.2.1.47.1.1.1.1.2",
# entPhysicalName
"1.3.6.1.2.1.47.1.1.1.1.7",
# entPhysicalClass
"1.3.6.1.2.1.47.1.1.1.1.5"
]
},
# Vendor-specific MIBs (WALK)
%SnmpQuery{
query_type: :WALK,
oids: [
# Cisco
"1.3.6.1.4.1.9",
# MikroTik
"1.3.6.1.4.1.14988",
# Ubiquiti
"1.3.6.1.4.1.41112"
]
},
# Neighbor discovery (WALK)
%SnmpQuery{
query_type: :WALK,
oids: [
# LLDP
"1.0.8802.1.1.2.1.4.1.1",
# Cisco CDP
"1.3.6.1.4.1.9.9.23"
]
},
# IP address discovery (WALK)
%SnmpQuery{
query_type: :WALK,
oids: [
# IP-MIB ipAddrTable (IPv4 - deprecated but widely supported)
"1.3.6.1.2.1.4.20",
# IP-MIB ipAddressTable (RFC 4293 - unified IPv4/IPv6)
"1.3.6.1.2.1.4.34"
]
}
]
end
defp build_polling_queries(device) do
sensor_oids = Enum.map(device.snmp_device.sensors, & &1.sensor_oid)
interface_oids =
Enum.flat_map(device.snmp_device.interfaces, fn iface ->
idx = iface.if_index
[
# ifInOctets
"1.3.6.1.2.1.2.2.1.10.#{idx}",
# ifOutOctets
"1.3.6.1.2.1.2.2.1.16.#{idx}",
# ifInErrors
"1.3.6.1.2.1.2.2.1.14.#{idx}",
# ifOutErrors
"1.3.6.1.2.1.2.2.1.20.#{idx}",
# ifInDiscards
"1.3.6.1.2.1.2.2.1.13.#{idx}",
# ifOutDiscards
"1.3.6.1.2.1.2.2.1.19.#{idx}"
]
end)
# Neighbor discovery OIDs
neighbor_oids = [
# LLDP
"1.0.8802.1.1.2.1.4.1.1",
# Cisco CDP
"1.3.6.1.4.1.9.9.23"
]
# ARP table OIDs
arp_oids = [
# IP-MIB ipNetToMediaTable (IPv4 ARP)
"1.3.6.1.2.1.4.22",
# IP-MIB ipNetToPhysicalTable (RFC 4293 - unified IPv4/IPv6)
"1.3.6.1.2.1.4.35"
]
# MAC address table OIDs
mac_oids = [
# BRIDGE-MIB dot1dTpFdbTable
"1.3.6.1.2.1.17.4.3"
]
# IP address OIDs (for polling changes)
ip_address_oids = [
# IP-MIB ipAddrTable (IPv4)
"1.3.6.1.2.1.4.20",
# IP-MIB ipAddressTable (IPv6)
"1.3.6.1.2.1.4.34"
]
# HOST-RESOURCES-MIB for processors and storage
host_resources_oids = [
# hrProcessorTable
"1.3.6.1.2.1.25.3.3",
# hrStorageTable
"1.3.6.1.2.1.25.2.3"
]
[
%SnmpQuery{
query_type: :GET,
oids: sensor_oids ++ interface_oids
},
%SnmpQuery{
query_type: :WALK,
oids: neighbor_oids ++ arp_oids ++ mac_oids ++ ip_address_oids ++ host_resources_oids
}
]
end
defp build_mikrotik_job(device) do
mikrotik_config = Devices.get_mikrotik_config(device)
# Only build job if MikroTik API is enabled and has credentials
if mikrotik_config.enabled && mikrotik_config.username do
%AgentJob{
job_id: "mikrotik:#{device.id}",
job_type: :MIKROTIK,
device_id: device.id,
mikrotik_device: %MikrotikDevice{
ip: device.ip_address,
username: mikrotik_config.username,
password: mikrotik_config.password,
port: mikrotik_config.port,
use_ssl: mikrotik_config.use_ssl
},
mikrotik_commands: build_mikrotik_commands()
}
end
end
defp build_mikrotik_commands do
[
# System identity
%MikrotikCommand{
command: "/system/identity/print",
args: %{}
},
# System resources
%MikrotikCommand{
command: "/system/resource/print",
args: %{}
}
]
end
defp process_snmp_result(organization_id, result, socket) do
with {:ok, device} <- fetch_device(result.device_id),
:ok <- verify_device_organization(device, organization_id) do
process_job_result(device, result, socket)
else
{:error, :device_not_found} ->
Logger.error("Device not found: #{result.device_id}")
{:error, :wrong_organization} ->
Logger.error("Device #{result.device_id} not in agent's organization")
end
end
defp fetch_device(device_id) do
case Devices.get_device_with_details(device_id) do
nil -> {:error, :device_not_found}
device -> {:ok, device}
end
end
defp verify_device_organization(device, organization_id) do
# Cloud pollers (organization_id = nil) can access any device
if is_nil(organization_id) or device.organization_id == organization_id do
:ok
else
{:error, :wrong_organization}
end
end
defp process_job_result(device, %{job_type: :DISCOVER} = result, socket) do
process_discovery_result(device, result, socket)
end
defp process_job_result(device, %{job_type: :POLL} = result, socket) do
process_polling_result(device, result, socket)
end
defp process_discovery_result(device, result, socket) do
# Agent has completed discovery and sent back SNMP data
# Update last_discovery_at to signal completion to DiscoveryWorker
Logger.info("Discovery results received from agent for #{device.name}, processing data")
# Update device timestamp to mark discovery as complete
case Devices.update_device(device, %{last_discovery_at: DateTime.utc_now()}) do
{:ok, updated_device} ->
Logger.info("Updated last_discovery_at for device #{device.name}")
# Process the SNMP data from agent's discovery queries
# The agent sends back oid_values map with all the collected data
process_discovery_data(updated_device, result, socket)
{:error, changeset} ->
Logger.error("Failed to update last_discovery_at for device #{device.name}: #{inspect(changeset)}")
end
end
defp process_discovery_data(device, result, socket) do
oid_values = Map.new(result.oid_values)
Logger.info("Processing discovery with #{map_size(oid_values)} OIDs",
device_id: device.id,
device_name: device.name
)
# Debug log full OID values if debug is enabled
maybe_debug_log(socket, "Discovery OID values received",
device_id: device.id,
device_name: device.name,
oid_count: map_size(oid_values),
sample_oids: oid_values |> Enum.take(10) |> Map.new(),
all_oids: if(socket.assigns[:debug_enabled], do: oid_values, else: :redacted)
)
# Process full discovery using agent data
case AgentDiscovery.process_agent_discovery(device, oid_values) do
{:ok, _discovered_device} ->
Logger.info("Agent discovery completed",
device_id: device.id,
device_name: device.name
)
maybe_debug_log(socket, "Discovery processing succeeded",
device_id: device.id,
device_name: device.name
)
# Broadcast for real-time UI updates
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"device:#{device.id}",
{:discovery_completed, device.id}
)
:ok
{:error, reason} ->
Logger.error("Agent discovery failed: #{inspect(reason)}",
device_id: device.id,
device_name: device.name
)
maybe_debug_log(socket, "Discovery processing failed",
device_id: device.id,
device_name: device.name,
error_reason: inspect(reason)
)
# Don't update last_discovery_at - DiscoveryWorker will retry
# with direct discovery as fallback
{:error, reason}
end
end
defp process_polling_result(device, result, _socket) do
snmp_device = device.snmp_device
oid_values = Map.new(result.oid_values)
timestamp = DateTime.from_unix!(result.timestamp, :second)
Logger.info(
"Processing polling result for #{device.name}: #{map_size(oid_values)} OIDs, #{length(snmp_device.sensors)} sensors, #{length(snmp_device.interfaces)} interfaces"
)
# Process sensor readings (quick path - already in memory)
Enum.each(snmp_device.sensors, &process_sensor_reading(&1, oid_values, timestamp))
# Process interface stats (quick path - already in memory)
Enum.each(device.snmp_device.interfaces, &process_interface_stats(&1, oid_values, timestamp))
# Process complex SNMP data (neighbors, ARP, MAC, IP addresses, processors, storage)
# using Replay adapter to reuse existing parsing logic
if map_size(oid_values) > length(snmp_device.sensors) + (length(snmp_device.interfaces) * 6) do
process_additional_polling_data(device, oid_values)
end
end
# Process neighbors, ARP, MAC, IP addresses, processors, storage from agent polling
defp process_additional_polling_data(device, oid_values) do
# Build Replay adapter opts to reuse existing parsing logic
client_opts = [
ip: device.ip_address,
adapter: Towerops.Snmp.Adapters.Replay,
oid_map: oid_values
]
# Reuse existing polling functions with Replay adapter
spawn(fn ->
alias Towerops.Snmp.Profiles.Base
alias Towerops.Snmp.NeighborDiscovery
# Fetch interfaces once for reuse across multiple operations
interfaces = Towerops.Snmp.list_interfaces(device.snmp_device.id)
# Process neighbors (LLDP/CDP)
case NeighborDiscovery.discover_neighbors(client_opts, interfaces) do
{:ok, neighbors} when neighbors != [] ->
Towerops.Snmp.Discovery.save_neighbors(device.id, neighbors)
_ ->
:ok
end
# Process ARP entries
case Towerops.Snmp.ArpDiscovery.discover_arp_table(client_opts) do
{:ok, arp_entries} when arp_entries != [] ->
Towerops.Snmp.Discovery.save_arp_entries(device.id, arp_entries, interfaces)
_ ->
:ok
end
# Process MAC addresses
case Towerops.Snmp.MacDiscovery.discover_mac_table(client_opts) do
{:ok, mac_addresses} when mac_addresses != [] ->
Towerops.Snmp.upsert_mac_addresses(device.id, mac_addresses, interfaces)
_ ->
:ok
end
# Process IP addresses
case Base.discover_all_ip_addresses(client_opts) do
{:ok, ip_addresses} when ip_addresses != [] ->
discovered_device = %{
device_id: device.id,
interfaces: interfaces
}
Towerops.Snmp.Discovery.sync_ip_addresses(discovered_device, ip_addresses)
_ ->
:ok
end
# Process processors
case Base.discover_processors(client_opts) do
{:ok, processors} when processors != [] ->
discovered_device = %{device_id: device.id}
Towerops.Snmp.Discovery.sync_processors(discovered_device, processors)
_ ->
:ok
end
# Process storage
case Base.discover_storage(client_opts) do
{:ok, storage} when storage != [] ->
discovered_device = %{device_id: device.id}
Towerops.Snmp.Discovery.sync_storage(discovered_device, storage)
_ ->
:ok
end
end)
end
defp process_sensor_reading(sensor, oid_values, timestamp) do
case Map.get(oid_values, sensor.sensor_oid) do
nil ->
:ok
value ->
with parsed_value when not is_nil(parsed_value) <- parse_float(value) do
final_value = parsed_value / sensor.sensor_divisor
Snmp.create_sensor_reading(%{
sensor_id: sensor.id,
value: final_value,
status: "ok",
checked_at: timestamp
})
end
end
end
defp process_interface_stats(iface, oid_values, timestamp) do
idx = iface.if_index
Snmp.create_interface_stat(%{
interface_id: iface.id,
if_in_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.10.#{idx}")),
if_out_octets: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.16.#{idx}")),
if_in_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.14.#{idx}")),
if_out_errors: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.20.#{idx}")),
if_in_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.13.#{idx}")),
if_out_discards: parse_integer(Map.get(oid_values, "1.3.6.1.2.1.2.2.1.19.#{idx}")),
checked_at: timestamp
})
end
defp parse_integer(nil), do: nil
defp parse_integer(value) when is_integer(value), do: value
defp parse_integer(value) when is_binary(value) do
case Integer.parse(value) do
{int, _} -> int
:error -> nil
end
end
defp parse_integer(_), do: nil
defp parse_float(value) when is_float(value), do: value
defp parse_float(value) when is_integer(value), do: value / 1.0
defp parse_float(value) when is_binary(value) do
case Float.parse(value) do
{float, _} -> float
:error -> nil
end
end
defp parse_float(_), do: nil
defp get_remote_ip(socket) do
case socket.transport_pid do
nil ->
nil
pid when is_pid(pid) ->
case :ranch.get_addr(pid) do
{ip, _port} when is_tuple(ip) -> format_ip(ip)
_ -> nil
end
end
rescue
_ -> nil
end
defp format_ip({a, b, c, d}), do: "#{a}.#{b}.#{c}.#{d}"
defp format_ip({a, b, c, d, e, f, g, h}),
do:
"#{Integer.to_string(a, 16)}:#{Integer.to_string(b, 16)}:#{Integer.to_string(c, 16)}:#{Integer.to_string(d, 16)}:#{Integer.to_string(e, 16)}:#{Integer.to_string(f, 16)}:#{Integer.to_string(g, 16)}:#{Integer.to_string(h, 16)}"
# Debug logging helper - only logs when debug is enabled for this agent
# Uses :info level since production logger filters out :debug
defp maybe_debug_log(socket, message, metadata) do
if socket.assigns[:debug_enabled] do
Logger.info(
message,
Keyword.merge(
[
agent_token_id: socket.assigns.agent_token_id,
debug_enabled: true
],
metadata
)
)
end
end
defp process_backup_result(result) do
case BackupRequests.get_request_by_job_id(result.job_id) do
nil ->
Logger.warning("Backup request not found for job_id", job_id: result.job_id)
request ->
handle_backup_request(request, result)
end
end
defp handle_backup_request(request, %{error: ""} = result) do
case extract_config_from_result(result) do
{:ok, config_text} ->
# Manual backups always create a snapshot, even if config unchanged
# Automated backups (daily_cron) use deduplication to save storage
should_create_backup =
request.source == "manual" || backup_needed?(result.device_id, config_text)
if should_create_backup do
create_and_save_backup(request, result, config_text)
else
mark_backup_skipped(request, result)
end
{:error, reason} ->
mark_backup_failed(request, result, "Failed to extract config: #{reason}")
end
end
defp handle_backup_request(request, result) do
mark_backup_failed(request, result, result.error)
end
defp create_and_save_backup(request, result, config_text) do
case MikrotikBackups.create_backup(result.device_id, config_text, request.source) do
{:ok, _backup} ->
BackupRequests.update_request_status(request.id, "success", nil)
Logger.info("Backup created", device_id: result.device_id, source: request.source)
{:error, reason} ->
mark_backup_failed(request, result, "Failed to create backup: #{inspect(reason)}")
end
end
defp backup_needed?(device_id, config_text) do
{:ok, needs_backup?} = MikrotikBackups.needs_backup?(device_id, config_text)
needs_backup?
end
defp mark_backup_skipped(request, result) do
BackupRequests.update_request_status(request.id, "success", nil)
Logger.info("Backup skipped (no changes)", device_id: result.device_id)
end
defp mark_backup_failed(request, result, error_message) do
BackupRequests.update_request_status(request.id, "failed", error_message)
Logger.error("Backup failed", device_id: result.device_id, error: error_message)
end
defp extract_config_from_result(result) do
# SSH backups return a single sentence with "config" attribute (new method)
# API backups use /file/read which returns chunks with "data" attribute (old method)
# Try SSH backup format first (single sentence with "config" attribute)
config_direct =
Enum.find_value(result.sentences, fn sentence -> Map.get(sentence.attributes, "config") end)
case config_direct do
nil ->
# Try chunked API backup format
extract_config_from_chunked_data(result)
config_text when is_binary(config_text) and config_text != "" ->
{:ok, config_text}
_ ->
{:error, "Config attribute found but empty or invalid"}
end
end
defp extract_config_from_chunked_data(result) do
# The old API backup workflow uses /file/read which returns chunks with "data" attribute
data_chunks =
result.sentences
|> Enum.filter(fn sentence -> Map.has_key?(sentence.attributes, "data") end)
|> Enum.map(fn sentence -> Map.get(sentence.attributes, "data") end)
case data_chunks do
[] ->
# Fallback: try other methods
extract_config_from_fallback(result)
chunks ->
# Combine all chunks in order
config_text = Enum.join(chunks, "")
if config_text == "" do
{:error, "No config data in result - all chunks were empty"}
else
{:ok, config_text}
end
end
end
defp extract_config_from_fallback(result) do
# Fallback: try to find any sentence with configuration-like content
# Look for sentences with common RouterOS config markers
config_text =
Enum.find_value(result.sentences, fn sentence ->
find_config_in_attributes(sentence.attributes)
end)
case config_text do
nil -> {:error, "No config data in result - received #{length(result.sentences)} sentences"}
text -> {:ok, text}
end
end
defp find_config_in_attributes(attributes) do
Enum.find_value(attributes, fn {_key, value} ->
if routeros_config?(value), do: value
end)
end
defp routeros_config?(value) when is_binary(value) do
String.contains?(value, ["# ", "/", "set "]) and String.length(value) > 100
end
defp routeros_config?(_), do: false
## Safe Encoding/Decoding Helpers
# Safely decode Base64-encoded protobuf binary.
# Returns {:ok, binary} on success or {:error, {type, message}} on failure.
@spec safe_base64_decode(base64_string()) :: {:ok, protobuf_binary()} | {:error, {atom(), String.t()}}
defp safe_base64_decode(base64_string) when is_binary(base64_string) do
case Base.decode64(base64_string) do
{:ok, binary} ->
{:ok, binary}
:error ->
{:error, {:invalid_base64, "Message is not valid Base64-encoded data"}}
end
rescue
e in ArgumentError ->
Logger.error("Base64 decode error: #{Exception.message(e)}")
{:error, {:invalid_base64, "Failed to decode Base64 data"}}
end
end
Reference in a new issue View git blame Copy permalink