Add distributed coordination for SNMP polling

Prevents duplicate polling when multiple pods are running by using
database-based coordination.

Changes:
- Added last_snmp_poll_at timestamp to equipment table
- Added Equipment.update_snmp_poll_time/1 function
- Updated PollerWorker to check last_snmp_poll_at before polling
- Polls are skipped if equipment was polled within (interval - 5s)
- Updates timestamp before polling (optimistic locking)

This ensures that when K8s scales up/down or pods restart, only one
pod polls each piece of equipment at a time, preventing wasteful
duplicate SNMP queries and database writes.

The 5-second grace period accounts for clock drift and processing delays
between pods.

lib/towerops/equipment.ex

@@ -123,4 +123,16 @@ defmodule Towerops.Equipment do
     |> Ecto.Changeset.change(changes)
     |> Repo.update()
   end
+
+  @doc """
+  Updates the last SNMP poll timestamp for equipment.
+  Used for distributed coordination to prevent duplicate polling across pods.
+  """
+  def update_snmp_poll_time(%EquipmentSchema{} = equipment) do
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+
+    equipment
+    |> Ecto.Changeset.change(%{last_snmp_poll_at: now})
+    |> Repo.update()
+  end
 end

lib/towerops/equipment/equipment.ex

@@ -22,6 +22,7 @@ defmodule Towerops.Equipment.Equipment do
     field :snmp_community, :string
     field :snmp_port, :integer, default: 161
     field :last_discovery_at, :utc_datetime
+    field :last_snmp_poll_at, :utc_datetime
 
     belongs_to :site, Towerops.Sites.Site
 

lib/towerops/snmp/poller_worker.ex

@@ -82,33 +82,46 @@ defmodule Towerops.Snmp.PollerWorker 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)
+        # Check if recently polled by another pod (distributed coordination)
+        poll_interval = get_poll_interval(equipment)
+        grace_period = 5
 
-        # Poll all sensors
-        try do
-          poll_sensors(device.sensors, client_opts, now)
-          Logger.debug("Polled #{length(device.sensors)} sensors for #{equipment.name}")
-        rescue
-          error ->
-            Logger.error(
-              "Error polling sensors for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}"
-            )
+        if !should_skip_poll?(equipment, poll_interval, grace_period) do
+          # Update poll timestamp before polling (optimistic locking)
+          Equipment.update_snmp_poll_time(equipment)
+
+          client_opts = build_client_opts(equipment)
+          now = DateTime.truncate(DateTime.utc_now(), :second)
+
+          # Poll all sensors
+          try do
+            poll_sensors(device.sensors, client_opts, now)
+            Logger.debug("Polled #{length(device.sensors)} sensors for #{equipment.name}")
+          rescue
+            error ->
+              Logger.error(
+                "Error polling sensors for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}"
+              )
+          end
+
+          # Poll all interfaces
+          try do
+            poll_interfaces(device.interfaces, client_opts, now)
+            Logger.debug("Polled #{length(device.interfaces)} interfaces for #{equipment.name}")
+          rescue
+            error ->
+              Logger.error(
+                "Error polling interfaces for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}"
+              )
+          end
+        else
+          Logger.debug(
+            "Skipping poll for #{equipment.name} - recently polled by another process"
+          )
         end
 
-        # Poll all interfaces
-        try do
-          poll_interfaces(device.interfaces, client_opts, now)
-          Logger.debug("Polled #{length(device.interfaces)} interfaces for #{equipment.name}")
-        rescue
-          error ->
-            Logger.error(
-              "Error polling interfaces for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}"
-            )
-        end
-
-        # Schedule next poll
-        schedule_next_poll(get_poll_interval(equipment))
+        # Schedule next poll regardless of whether we polled or skipped
+        schedule_next_poll(poll_interval)
       end
     end
   end
@@ -266,6 +279,21 @@ defmodule Towerops.Snmp.PollerWorker do
     Process.send_after(self(), :poll_data, interval_seconds * 1000)
   end
 
+  defp should_skip_poll?(equipment, poll_interval_seconds, grace_period_seconds) do
+    case equipment.last_snmp_poll_at do
+      nil ->
+        false
+
+      last_poll ->
+        now = DateTime.utc_now()
+        seconds_since_poll = DateTime.diff(now, last_poll, :second)
+        min_interval = poll_interval_seconds - grace_period_seconds
+
+        # Skip if polled within the minimum interval
+        seconds_since_poll < min_interval
+    end
+  end
+
   defp via_tuple(equipment_id) do
     {:via, Registry, {PollerRegistry, equipment_id}}
   end

priv/repo/migrations/20260104192555_add_last_snmp_poll_to_equipment.exs

@@ -0,0 +1,11 @@
+defmodule Towerops.Repo.Migrations.AddLastSnmpPollToEquipment do
+  use Ecto.Migration
+
+  def change do
+    alter table(:equipment) do
+      add :last_snmp_poll_at, :utc_datetime
+    end
+
+    create index(:equipment, [:last_snmp_poll_at])
+  end
+end