From 3544d117a85f530f0a6c33fcfd0970bf6e6e1efe Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 4 Jan 2026 12:56:51 -0600 Subject: [PATCH] Implement SNMP polling for continuous data collection Add background workers to regularly poll SNMP sensors and interfaces: - Create PollerWorker GenServer to poll SNMP devices - Poll all sensors for temperature, voltage, CPU, memory, etc. - Poll all interfaces for bandwidth, errors, and discards - Save time-series data to snmp_sensor_readings and snmp_interface_stats - Integrate with monitoring supervisor to auto-start pollers - Use same interval as connectivity checks (minimum 30 seconds) - Add list_snmp_enabled_equipment function to Equipment context Pollers start automatically on app boot for all SNMP-enabled equipment and run independently of connectivity monitoring. --- lib/towerops/equipment.ex | 12 ++ lib/towerops/monitoring/supervisor.ex | 40 +++++- lib/towerops/snmp/poller_worker.ex | 193 ++++++++++++++++++++++++++ 3 files changed, 244 insertions(+), 1 deletion(-) create mode 100644 lib/towerops/snmp/poller_worker.ex diff --git a/lib/towerops/equipment.ex b/lib/towerops/equipment.ex index 1354d460..bdce2cba 100644 --- a/lib/towerops/equipment.ex +++ b/lib/towerops/equipment.ex @@ -41,6 +41,18 @@ defmodule Towerops.Equipment do ) end + @doc """ + Returns the list of all equipment with SNMP enabled. + """ + def list_snmp_enabled_equipment do + Repo.all( + from(e in EquipmentSchema, + where: e.snmp_enabled == true, + order_by: [asc: e.name] + ) + ) + end + @doc """ Gets a single equipment. """ diff --git a/lib/towerops/monitoring/supervisor.ex b/lib/towerops/monitoring/supervisor.ex index cd505baf..2e956627 100644 --- a/lib/towerops/monitoring/supervisor.ex +++ b/lib/towerops/monitoring/supervisor.ex @@ -5,6 +5,9 @@ defmodule Towerops.Monitoring.Supervisor do use Supervisor alias Towerops.Monitoring.EquipmentMonitor + alias Towerops.Snmp.PollerRegistry + alias Towerops.Snmp.PollerSupervisor + alias Towerops.Snmp.PollerWorker def start_link(init_arg) do Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__) @@ -15,8 +18,12 @@ defmodule Towerops.Monitoring.Supervisor do children = [ # Registry for naming monitor processes {Registry, keys: :unique, name: Towerops.Monitoring.Registry}, + # Registry for naming SNMP poller processes + {Registry, keys: :unique, name: PollerRegistry}, # DynamicSupervisor for monitor workers - {DynamicSupervisor, name: Towerops.Monitoring.DynamicSupervisor, strategy: :one_for_one} + {DynamicSupervisor, name: Towerops.Monitoring.DynamicSupervisor, strategy: :one_for_one}, + # DynamicSupervisor for SNMP poller workers + {DynamicSupervisor, name: PollerSupervisor, strategy: :one_for_one} ] # Start all monitors after supervisor is initialized (but not in test mode) @@ -27,6 +34,7 @@ defmodule Towerops.Monitoring.Supervisor do # Wait briefly for the supervisor tree to be fully initialized Process.sleep(100) start_all_monitors() + start_all_snmp_pollers() end) end @@ -64,6 +72,36 @@ defmodule Towerops.Monitoring.Supervisor do end) end + @doc """ + Starts an SNMP poller for a specific equipment. + """ + def start_snmp_poller(equipment_id) do + spec = {PollerWorker, equipment_id} + DynamicSupervisor.start_child(PollerSupervisor, spec) + end + + @doc """ + Stops an SNMP poller for a specific equipment. + """ + def stop_snmp_poller(equipment_id) do + case Registry.lookup(PollerRegistry, equipment_id) do + [{pid, _}] -> + DynamicSupervisor.terminate_child(PollerSupervisor, pid) + + [] -> + :ok + end + end + + @doc """ + Starts SNMP pollers for all equipment that has SNMP enabled. + """ + def start_all_snmp_pollers do + Enum.each(Towerops.Equipment.list_snmp_enabled_equipment(), fn equipment -> + start_snmp_poller(equipment.id) + end) + end + # Check if running in test mode by inspecting the Repo pool configuration defp test_mode? do config = Application.get_env(:towerops, Towerops.Repo, []) diff --git a/lib/towerops/snmp/poller_worker.ex b/lib/towerops/snmp/poller_worker.ex new file mode 100644 index 00000000..5a69eb14 --- /dev/null +++ b/lib/towerops/snmp/poller_worker.ex @@ -0,0 +1,193 @@ +defmodule Towerops.Snmp.PollerWorker do + @moduledoc """ + GenServer that regularly polls SNMP sensors and interfaces for a device. + + This worker continuously collects time-series data for: + - Sensor readings (temperature, voltage, CPU, memory, etc.) + - Interface statistics (bandwidth, errors, discards, etc.) + + Runs independently of the connectivity monitoring in EquipmentMonitor. + """ + use GenServer + + alias Towerops.Equipment + alias Towerops.Snmp + alias Towerops.Snmp.Client + alias Towerops.Snmp.PollerRegistry + + require Logger + + @default_poll_interval 60 + + # Client API + + @doc """ + Starts a poller for the given equipment ID. + """ + def start_link(equipment_id) do + GenServer.start_link(__MODULE__, equipment_id, name: via_tuple(equipment_id)) + end + + @doc """ + Triggers an immediate poll for the equipment. + """ + def trigger_poll(equipment_id) do + case Registry.lookup(PollerRegistry, equipment_id) do + [{_pid, _}] -> + GenServer.cast(via_tuple(equipment_id), :poll_now) + + [] -> + # Process doesn't exist, perform poll directly in background + Task.start(fn -> perform_poll(equipment_id) end) + end + end + + # Server Callbacks + + @impl true + def init(equipment_id) do + equipment = Equipment.get_equipment!(equipment_id) + + if equipment.snmp_enabled do + # Get the device to check if it has sensors/interfaces + device = Snmp.get_device_with_associations(equipment_id) + + if device && (length(device.sensors) > 0 || length(device.interfaces) > 0) do + # Perform immediate poll when starting + send(self(), :poll_data) + end + end + + {:ok, %{equipment_id: equipment_id}} + end + + @impl true + def handle_info(:poll_data, state) do + _ = perform_poll(state.equipment_id) + {:noreply, state} + end + + @impl true + def handle_cast(:poll_now, state) do + _ = perform_poll(state.equipment_id) + {:noreply, state} + end + + # Private Functions + + defp perform_poll(equipment_id) do + equipment = Equipment.get_equipment!(equipment_id) + + if equipment.snmp_enabled do + device = Snmp.get_device_with_associations(equipment_id) + + if device do + client_opts = build_client_opts(equipment) + now = DateTime.truncate(DateTime.utc_now(), :second) + + # Poll all sensors + _ = poll_sensors(device.sensors, client_opts, now) + + # Poll all interfaces + _ = poll_interfaces(device.interfaces, client_opts, now) + + # Schedule next poll + schedule_next_poll(get_poll_interval(equipment)) + end + end + end + + defp poll_sensors(sensors, client_opts, timestamp) do + Enum.each(sensors, fn sensor -> + case Client.get(client_opts, sensor.sensor_oid) do + {:ok, raw_value} when is_number(raw_value) -> + # Calculate actual value using divisor + value = raw_value / sensor.sensor_divisor + + # Determine status based on value (could be enhanced with thresholds) + status = "ok" + + # Save the reading + Snmp.create_sensor_reading(%{ + sensor_id: sensor.id, + value: value, + status: status, + checked_at: timestamp + }) + + {:ok, _non_numeric} -> + Logger.debug("Non-numeric SNMP value for sensor #{sensor.sensor_descr}") + + {:error, reason} -> + Logger.warning("Failed to poll sensor #{sensor.sensor_descr}: #{inspect(reason)}") + + # Save error status + Snmp.create_sensor_reading(%{ + sensor_id: sensor.id, + value: nil, + status: "error", + checked_at: timestamp + }) + end + end) + end + + defp poll_interfaces(interfaces, client_opts, timestamp) do + Enum.each(interfaces, fn interface -> + # Poll interface stats: ifInOctets, ifOutOctets, ifInErrors, ifOutErrors + {:ok, stat_data} = get_interface_stats(client_opts, interface.if_index) + + stats = + Map.merge(stat_data, %{ + interface_id: interface.id, + checked_at: timestamp + }) + + Snmp.create_interface_stat(stats) + end) + end + + defp get_interface_stats(client_opts, if_index) do + oids = [ + in_octets: "1.3.6.1.2.1.2.2.1.10.#{if_index}", + out_octets: "1.3.6.1.2.1.2.2.1.16.#{if_index}", + in_errors: "1.3.6.1.2.1.2.2.1.14.#{if_index}", + out_errors: "1.3.6.1.2.1.2.2.1.20.#{if_index}", + in_discards: "1.3.6.1.2.1.2.2.1.13.#{if_index}", + out_discards: "1.3.6.1.2.1.2.2.1.19.#{if_index}" + ] + + results = + Enum.map(oids, fn {key, oid} -> + case Client.get(client_opts, oid) do + {:ok, value} when is_number(value) -> {key, value} + _ -> {key, nil} + end + end) + + {:ok, Map.new(results)} + end + + defp build_client_opts(equipment) do + [ + ip: equipment.ip_address, + community: equipment.snmp_community, + version: equipment.snmp_version, + port: equipment.snmp_port || 161, + timeout: 5000 + ] + end + + defp get_poll_interval(equipment) do + # Use check_interval_seconds, but minimum of 30 seconds for SNMP polling + max(equipment.check_interval_seconds || @default_poll_interval, 30) + end + + defp schedule_next_poll(interval_seconds) do + Process.send_after(self(), :poll_data, interval_seconds * 1000) + end + + defp via_tuple(equipment_id) do + {:via, Registry, {PollerRegistry, equipment_id}} + end +end