diff --git a/lib/towerops/snmp.ex b/lib/towerops/snmp.ex index 523c1cfa..f3d9230c 100644 --- a/lib/towerops/snmp.ex +++ b/lib/towerops/snmp.ex @@ -17,6 +17,7 @@ defmodule Towerops.Snmp do alias Towerops.Snmp.Neighbor alias Towerops.Snmp.Sensor alias Towerops.Snmp.SensorReading + alias Towerops.Snmp.StateSensor @doc """ Tests SNMP connectivity to a device. @@ -94,7 +95,7 @@ defmodule Towerops.Snmp do def get_device_with_associations(device_id) do Device |> where([d], d.device_id == ^device_id) - |> preload([:interfaces, :sensors]) + |> preload([:interfaces, :sensors, :state_sensors]) |> Repo.one() end @@ -187,6 +188,27 @@ defmodule Towerops.Snmp do |> Repo.update() end + # State sensor queries + + @doc """ + Updates a state sensor. + """ + def update_state_sensor(%StateSensor{} = state_sensor, attrs) do + state_sensor + |> StateSensor.changeset(attrs) + |> Repo.update() + end + + @doc """ + Lists all state sensors for a device. + """ + def list_state_sensors(device_id) do + StateSensor + |> where([s], s.snmp_device_id == ^device_id) + |> order_by([s], s.sensor_index) + |> Repo.all() + end + # Sensor readings queries @doc """ diff --git a/lib/towerops/snmp/device.ex b/lib/towerops/snmp/device.ex index 4ffba380..933269f6 100644 --- a/lib/towerops/snmp/device.ex +++ b/lib/towerops/snmp/device.ex @@ -13,6 +13,7 @@ defmodule Towerops.Snmp.Device do alias Towerops.Devices.Device, as: DeviceSchema alias Towerops.Snmp.Interface alias Towerops.Snmp.Sensor + alias Towerops.Snmp.StateSensor @primary_key {:id, :binary_id, autogenerate: true} @foreign_key_type :binary_id @@ -34,6 +35,7 @@ defmodule Towerops.Snmp.Device do has_many :interfaces, Interface, foreign_key: :snmp_device_id has_many :sensors, Sensor, foreign_key: :snmp_device_id + has_many :state_sensors, StateSensor, foreign_key: :snmp_device_id timestamps(type: :utc_datetime) end @@ -56,6 +58,7 @@ defmodule Towerops.Snmp.Device do device: NotLoaded.t() | Devices.t(), interfaces: NotLoaded.t() | [Interface.t()], sensors: NotLoaded.t() | [Sensor.t()], + state_sensors: NotLoaded.t() | [StateSensor.t()], inserted_at: DateTime.t(), updated_at: DateTime.t() } diff --git a/lib/towerops/snmp/poller_worker.ex b/lib/towerops/snmp/poller_worker.ex index ba472b81..2618ec51 100644 --- a/lib/towerops/snmp/poller_worker.ex +++ b/lib/towerops/snmp/poller_worker.ex @@ -154,6 +154,9 @@ defmodule Towerops.Snmp.PollerWorker do Task.Supervisor.async_nolink(PollerTaskSupervisor, fn -> poll_device_sensors(device, snmp_device, client_opts, now) end), + Task.Supervisor.async_nolink(PollerTaskSupervisor, fn -> + poll_device_state_sensors(device, snmp_device, client_opts, now) + end), Task.Supervisor.async_nolink(PollerTaskSupervisor, fn -> poll_device_interfaces(device, snmp_device, client_opts, now) end), @@ -192,6 +195,21 @@ defmodule Towerops.Snmp.PollerWorker do Logger.error("Error polling sensors for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}") end + defp poll_device_state_sensors(device, snmp_device, client_opts, now) do + poll_state_sensors(snmp_device.state_sensors, client_opts, now, device.id) + Logger.debug("Polled #{length(snmp_device.state_sensors)} state sensors for #{device.name}") + + # Broadcast state sensor update event to device-specific topic + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "device:#{device.id}", + {:state_sensors_updated, device.id} + ) + rescue + error -> + Logger.error("Error polling state sensors for #{device.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}") + end + defp poll_device_interfaces(device, snmp_device, client_opts, now) do poll_interfaces(snmp_device.interfaces, client_opts, now) Logger.debug("Polled #{length(snmp_device.interfaces)} interfaces for #{device.name}") @@ -260,6 +278,115 @@ defmodule Towerops.Snmp.PollerWorker do |> Stream.run() end + defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id) do + # Poll state sensors in parallel + state_sensors + |> Task.async_stream( + fn state_sensor -> + result = poll_state_sensor_value(state_sensor, client_opts) + handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id) + end, + max_concurrency: 2, + timeout: 40_000, + on_timeout: :kill_task + ) + |> Stream.run() + end + + defp poll_state_sensor_value(state_sensor, client_opts) do + case Client.get(client_opts, state_sensor.sensor_oid) do + {:ok, raw_value} when is_integer(raw_value) -> + {:ok, raw_value} + + {:ok, _non_integer} -> + {:error, :non_integer} + + {:error, reason} -> + {:error, reason} + end + end + + defp handle_state_sensor_poll_result(state_sensor, {:ok, new_state_value}, timestamp, device_id) do + old_status = state_sensor.status + new_status = entity_state_to_status(new_state_value) + new_descr = entity_state_to_descr(new_state_value) + + # Update the state sensor + Snmp.update_state_sensor(state_sensor, %{ + state_value: new_state_value, + state_descr: new_descr, + status: new_status, + last_checked_at: timestamp + }) + + # Detect and broadcast state changes + if old_status != new_status do + broadcast_state_sensor_change(state_sensor, old_status, new_status, device_id, timestamp) + end + end + + defp handle_state_sensor_poll_result(state_sensor, {:error, reason}, timestamp, _device_id) do + Logger.debug("Failed to poll state sensor #{state_sensor.sensor_descr}: #{inspect(reason)}") + + Snmp.update_state_sensor(state_sensor, %{ + status: "unknown", + last_checked_at: timestamp + }) + end + + defp broadcast_state_sensor_change(state_sensor, old_status, new_status, device_id, timestamp) do + severity = determine_state_change_severity(old_status, new_status) + event_type = determine_state_change_event_type(new_status) + + event = %{ + device_id: device_id, + event_type: event_type, + severity: severity, + message: "#{state_sensor.sensor_descr} status changed from #{old_status} to #{new_status}", + metadata: %{ + state_sensor_id: state_sensor.id, + sensor_name: state_sensor.sensor_descr, + entity_type: state_sensor.entity_type, + old_status: old_status, + new_status: new_status + }, + occurred_at: timestamp + } + + # Broadcast to device-specific topic + Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{device_id}", {:device_event, event}) + + # Also broadcast to global events topic + Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + + Logger.info("State sensor change: #{event.message}") + end + + defp determine_state_change_severity(_old_status, "critical"), do: "critical" + defp determine_state_change_severity(_old_status, "warning"), do: "warning" + defp determine_state_change_severity("critical", "ok"), do: "info" + defp determine_state_change_severity("warning", "ok"), do: "info" + defp determine_state_change_severity(_old_status, _new_status), do: "info" + + defp determine_state_change_event_type("critical"), do: "state_sensor_critical" + defp determine_state_change_event_type("warning"), do: "state_sensor_warning" + defp determine_state_change_event_type("ok"), do: "state_sensor_normal" + defp determine_state_change_event_type(_), do: "state_sensor_unknown" + + # ENTITY-STATE-MIB operational status mapping + # 1=unknown, 2=disabled, 3=enabled, 4=testing + defp entity_state_to_status(1), do: "unknown" + defp entity_state_to_status(2), do: "ok" + defp entity_state_to_status(3), do: "warning" + defp entity_state_to_status(4), do: "unknown" + defp entity_state_to_status(_), do: "unknown" + + defp entity_state_to_descr(1), do: "unknown" + defp entity_state_to_descr(2), do: "enabled" + defp entity_state_to_descr(3), do: "disabled" + defp entity_state_to_descr(4), do: "testing" + defp entity_state_to_descr(_), do: "unknown" + defp poll_sensor_value(sensor, client_opts) do if sensor.metadata["calculation"] == "percentage" do poll_percentage_sensor(sensor, client_opts)