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

@@ -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.
   """

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, [])

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