Graham McIntire
b53a53b199
- Add @type definitions to all schema modules: - User, UserCredential, Membership, Invitation - Organization, Site, AgentAssignment - InterfaceStat, SensorReading, Alert - Fix all compilation warnings with stronger Elixir types: - Remove unused require Logger in log_filter.ex - Remove unused parse_float(nil) clause - Add pin operators (^) for variables in binary pattern matches - Fix Dialyzer errors (25 → 0): - Remove unreachable pattern match clauses - Fix unmatched return values with _ = prefix - Update @spec for deliver_alert_notification/1 - Properly handle all Phoenix.PubSub.subscribe and Task.start return values - Explicitly ignore if statement return values where needed All files now pass mix compile --warnings-as-errors and mix dialyzer.
945 lines
30 KiB
Elixir
945 lines
30 KiB
Elixir
defmodule Towerops.Snmp.PollerWorker do
|
|
@moduledoc """
|
|
GenServer that regularly polls SNMP data for a device.
|
|
|
|
This worker continuously collects:
|
|
- Sensor readings (temperature, voltage, CPU, memory, etc.)
|
|
- Interface statistics (bandwidth, errors, discards, etc.)
|
|
- Neighbor discovery (LLDP/CDP topology information)
|
|
|
|
Runs independently of the connectivity monitoring in EquipmentMonitor.
|
|
Poll interval is configurable per equipment (default: 60 seconds).
|
|
"""
|
|
use GenServer
|
|
|
|
alias Towerops.Equipment
|
|
alias Towerops.Snmp
|
|
alias Towerops.Snmp.Client
|
|
alias Towerops.Snmp.NeighborDiscovery
|
|
alias Towerops.Snmp.PollerRegistry
|
|
alias Towerops.Snmp.Sensor
|
|
|
|
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)
|
|
|
|
# Subscribe to discovery completion events for this equipment
|
|
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "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 && (device.sensors != [] || device.interfaces != []) 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_info({:discovery_completed, _equipment_id}, state) do
|
|
Logger.info("Discovery completed, triggering immediate poll")
|
|
# Trigger immediate poll after discovery
|
|
send(self(), :poll_data)
|
|
{:noreply, state}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(_msg, state) do
|
|
# Ignore other messages (like equipment status changes)
|
|
{: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
|
|
poll_equipment_device(equipment_id, equipment)
|
|
end
|
|
end
|
|
|
|
defp poll_equipment_device(equipment_id, equipment) do
|
|
device = Snmp.get_device_with_associations(equipment_id)
|
|
|
|
if device do
|
|
poll_interval = get_poll_interval(equipment)
|
|
execute_device_poll(device, equipment, poll_interval)
|
|
schedule_next_poll(poll_interval)
|
|
end
|
|
end
|
|
|
|
defp execute_device_poll(device, equipment, poll_interval) do
|
|
if should_skip_poll?(equipment, poll_interval, 5) do
|
|
Logger.debug("Skipping poll for #{equipment.name} - recently polled by another process")
|
|
else
|
|
perform_device_data_collection(device, equipment)
|
|
end
|
|
end
|
|
|
|
defp perform_device_data_collection(device, equipment) do
|
|
Equipment.update_snmp_poll_time(equipment)
|
|
|
|
client_opts = build_client_opts(equipment)
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
poll_device_sensors(device, equipment, client_opts, now)
|
|
poll_device_interfaces(device, equipment, client_opts, now)
|
|
check_device_interface_changes(device, equipment, client_opts, now)
|
|
poll_device_neighbors(device, equipment, client_opts)
|
|
end
|
|
|
|
defp poll_device_sensors(device, equipment, client_opts, now) 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
|
|
|
|
defp poll_device_interfaces(device, equipment, client_opts, now) 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
|
|
|
|
defp check_device_interface_changes(device, equipment, client_opts, now) do
|
|
check_interface_changes(device.interfaces, equipment, client_opts, now)
|
|
rescue
|
|
error ->
|
|
Logger.error(
|
|
"Error checking interface changes for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}"
|
|
)
|
|
end
|
|
|
|
defp poll_device_neighbors(device, equipment, client_opts) do
|
|
# Add equipment_id to interfaces for neighbor discovery
|
|
interfaces_with_equipment = Enum.map(device.interfaces, &Map.put(&1, :equipment_id, equipment.id))
|
|
|
|
{:ok, neighbors} = NeighborDiscovery.discover_neighbors(client_opts, interfaces_with_equipment)
|
|
|
|
# Delete stale neighbors (not seen in last 5 minutes)
|
|
cutoff = DateTime.add(DateTime.utc_now(), -5, :minute)
|
|
Snmp.delete_stale_neighbors(equipment.id, cutoff)
|
|
|
|
# Upsert each discovered neighbor
|
|
Enum.each(neighbors, fn neighbor_data ->
|
|
Snmp.upsert_neighbor(neighbor_data)
|
|
end)
|
|
|
|
Logger.debug("Polled and saved #{length(neighbors)} neighbors for #{equipment.name}")
|
|
rescue
|
|
error ->
|
|
Logger.error("Error polling neighbors for #{equipment.name}: #{inspect(error)}\n#{Exception.format_stacktrace()}")
|
|
end
|
|
|
|
defp poll_sensors(sensors, client_opts, timestamp) do
|
|
Enum.each(sensors, fn sensor ->
|
|
result = poll_sensor_value(sensor, client_opts)
|
|
handle_sensor_poll_result(sensor, result, timestamp)
|
|
end)
|
|
end
|
|
|
|
defp poll_sensor_value(sensor, client_opts) do
|
|
if sensor.metadata["calculation"] == "percentage" do
|
|
poll_percentage_sensor(sensor, client_opts)
|
|
else
|
|
poll_simple_sensor(sensor, client_opts)
|
|
end
|
|
end
|
|
|
|
defp handle_sensor_poll_result(sensor, {:ok, value}, timestamp) do
|
|
Snmp.create_sensor_reading(%{
|
|
sensor_id: sensor.id,
|
|
value: value,
|
|
status: "ok",
|
|
checked_at: timestamp
|
|
})
|
|
|
|
detect_sensor_changes(sensor, value, timestamp)
|
|
|
|
Snmp.update_sensor(sensor, %{
|
|
last_value: value,
|
|
last_checked_at: timestamp
|
|
})
|
|
end
|
|
|
|
defp handle_sensor_poll_result(sensor, {:error, :non_numeric}, timestamp) do
|
|
Logger.debug("Non-numeric SNMP value for sensor #{sensor.sensor_descr}")
|
|
|
|
Snmp.create_sensor_reading(%{
|
|
sensor_id: sensor.id,
|
|
value: nil,
|
|
status: "unknown",
|
|
checked_at: timestamp
|
|
})
|
|
end
|
|
|
|
defp handle_sensor_poll_result(sensor, {:error, reason}, timestamp) do
|
|
log_sensor_error(sensor, reason)
|
|
|
|
Snmp.create_sensor_reading(%{
|
|
sensor_id: sensor.id,
|
|
value: nil,
|
|
status: "error",
|
|
checked_at: timestamp
|
|
})
|
|
end
|
|
|
|
defp log_sensor_error(sensor, :no_such_object) do
|
|
Logger.debug("Sensor #{sensor.sensor_descr} OID not found on device")
|
|
end
|
|
|
|
defp log_sensor_error(sensor, :no_such_instance) do
|
|
Logger.debug("Sensor #{sensor.sensor_descr} instance not found on device")
|
|
end
|
|
|
|
defp log_sensor_error(sensor, :end_of_mib_view) do
|
|
Logger.debug("Sensor #{sensor.sensor_descr} end of MIB view")
|
|
end
|
|
|
|
defp log_sensor_error(sensor, reason) do
|
|
Logger.warning("Failed to poll sensor #{sensor.sensor_descr}: #{inspect(reason)}")
|
|
end
|
|
|
|
defp poll_simple_sensor(sensor, client_opts) do
|
|
case Client.get(client_opts, sensor.sensor_oid) do
|
|
{:ok, raw_value} ->
|
|
decoded_value = decode_snmp_value(raw_value)
|
|
|
|
if is_number(decoded_value) do
|
|
# Calculate actual value using divisor
|
|
value = decoded_value / sensor.sensor_divisor
|
|
{:ok, value}
|
|
else
|
|
{:error, :non_numeric}
|
|
end
|
|
|
|
{:error, reason} ->
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
defp poll_percentage_sensor(sensor, client_opts) do
|
|
# For percentage sensors (like storage), fetch both used and size
|
|
size_oid = sensor.metadata["size_oid"]
|
|
used_oid = sensor.sensor_oid
|
|
|
|
with {:ok, used_raw} <- Client.get(client_opts, used_oid),
|
|
{:ok, size_raw} <- Client.get(client_opts, size_oid),
|
|
used when is_number(used) <- decode_snmp_value(used_raw),
|
|
size when is_number(size) <- decode_snmp_value(size_raw),
|
|
true <- size > 0 do
|
|
percentage = used / size * 100
|
|
{:ok, percentage}
|
|
else
|
|
{:error, reason} -> {:error, reason}
|
|
false -> {:error, :division_by_zero}
|
|
_ -> {:error, :non_numeric}
|
|
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 check_interface_changes(interfaces, equipment, client_opts, timestamp) do
|
|
Enum.each(interfaces, fn interface ->
|
|
case get_interface_attributes(client_opts, interface.if_index) do
|
|
{:ok, current_attrs} ->
|
|
detect_and_log_changes(interface, current_attrs, equipment.id, timestamp)
|
|
|
|
{:error, reason} ->
|
|
Logger.debug("Failed to get interface attributes for #{interface.if_name}: #{inspect(reason)}")
|
|
end
|
|
end)
|
|
end
|
|
|
|
defp get_interface_attributes(client_opts, if_index) do
|
|
oids = [
|
|
# ifSpeed
|
|
"1.3.6.1.2.1.2.2.1.5.#{if_index}",
|
|
# ifPhysAddress
|
|
"1.3.6.1.2.1.2.2.1.6.#{if_index}",
|
|
# ifAdminStatus
|
|
"1.3.6.1.2.1.2.2.1.7.#{if_index}",
|
|
# ifOperStatus
|
|
"1.3.6.1.2.1.2.2.1.8.#{if_index}"
|
|
]
|
|
|
|
case Client.get_multiple(client_opts, oids) do
|
|
{:ok, [speed, phys_addr, admin_status, oper_status]} ->
|
|
{:ok,
|
|
%{
|
|
if_speed: parse_interface_integer(speed),
|
|
if_phys_address: format_mac_address(phys_addr),
|
|
if_admin_status: parse_if_status(admin_status),
|
|
if_oper_status: parse_if_status(oper_status)
|
|
}}
|
|
|
|
{:error, reason} = error ->
|
|
Logger.debug("Failed to get interface attributes: #{inspect(reason)}")
|
|
error
|
|
end
|
|
end
|
|
|
|
@spec detect_and_log_changes(
|
|
Towerops.Snmp.Interface.t(),
|
|
map(),
|
|
Ecto.UUID.t(),
|
|
DateTime.t()
|
|
) :: :ok | nil
|
|
defp detect_and_log_changes(interface, current_attrs, equipment_id, timestamp) do
|
|
events =
|
|
[]
|
|
|> maybe_add_oper_status_event(interface, current_attrs, equipment_id, timestamp)
|
|
|> maybe_add_admin_status_event(interface, current_attrs, equipment_id, timestamp)
|
|
|> maybe_add_speed_change_event(interface, current_attrs, equipment_id, timestamp)
|
|
|> maybe_add_mac_change_event(interface, current_attrs, equipment_id, timestamp)
|
|
|
|
if events != [] do
|
|
broadcast_interface_events(events)
|
|
Snmp.update_interface(interface, current_attrs)
|
|
end
|
|
end
|
|
|
|
defp maybe_add_oper_status_event(events, interface, current_attrs, equipment_id, timestamp) do
|
|
if interface.if_oper_status == current_attrs.if_oper_status do
|
|
events
|
|
else
|
|
event = build_oper_status_event(interface, current_attrs, equipment_id, timestamp)
|
|
[{:event, event} | events]
|
|
end
|
|
end
|
|
|
|
defp build_oper_status_event(interface, current_attrs, equipment_id, timestamp) do
|
|
message =
|
|
if interface.if_oper_status do
|
|
"Interface #{interface.if_name} changed from #{String.upcase(interface.if_oper_status)} to #{String.upcase(current_attrs.if_oper_status)}"
|
|
else
|
|
"Interface #{interface.if_name} is now #{String.upcase(current_attrs.if_oper_status)}"
|
|
end
|
|
|
|
%{
|
|
equipment_id: equipment_id,
|
|
event_type: if(current_attrs.if_oper_status == "up", do: "interface_up", else: "interface_down"),
|
|
severity: if(current_attrs.if_oper_status == "up", do: "info", else: "warning"),
|
|
message: message,
|
|
metadata: %{
|
|
interface_id: interface.id,
|
|
interface_name: interface.if_name,
|
|
old_status: interface.if_oper_status,
|
|
new_status: current_attrs.if_oper_status
|
|
},
|
|
occurred_at: timestamp
|
|
}
|
|
end
|
|
|
|
defp maybe_add_admin_status_event(events, interface, current_attrs, equipment_id, timestamp) do
|
|
if interface.if_admin_status == current_attrs.if_admin_status do
|
|
events
|
|
else
|
|
event = build_admin_status_event(interface, current_attrs, equipment_id, timestamp)
|
|
[{:event, event} | events]
|
|
end
|
|
end
|
|
|
|
defp build_admin_status_event(interface, current_attrs, equipment_id, timestamp) do
|
|
%{
|
|
equipment_id: equipment_id,
|
|
event_type: "interface_admin_status_change",
|
|
severity: "info",
|
|
message:
|
|
"Interface #{interface.if_name} admin status changed from #{interface.if_admin_status} to #{current_attrs.if_admin_status}",
|
|
metadata: %{
|
|
interface_id: interface.id,
|
|
interface_name: interface.if_name,
|
|
old_status: interface.if_admin_status,
|
|
new_status: current_attrs.if_admin_status
|
|
},
|
|
occurred_at: timestamp
|
|
}
|
|
end
|
|
|
|
defp maybe_add_speed_change_event(events, interface, current_attrs, equipment_id, timestamp) do
|
|
if interface.if_speed != current_attrs.if_speed && current_attrs.if_speed != nil do
|
|
event = build_speed_change_event(interface, current_attrs, equipment_id, timestamp)
|
|
[{:event, event} | events]
|
|
else
|
|
events
|
|
end
|
|
end
|
|
|
|
defp build_speed_change_event(interface, current_attrs, equipment_id, timestamp) do
|
|
is_initial = interface.if_speed == nil
|
|
severity = determine_speed_change_severity(is_initial, interface.if_speed, current_attrs.if_speed)
|
|
message = format_speed_change_message(interface.if_name, is_initial, interface.if_speed, current_attrs.if_speed)
|
|
|
|
%{
|
|
equipment_id: equipment_id,
|
|
event_type: "interface_speed_change",
|
|
severity: severity,
|
|
message: message,
|
|
metadata: %{
|
|
interface_id: interface.id,
|
|
interface_name: interface.if_name,
|
|
old_speed: interface.if_speed,
|
|
new_speed: current_attrs.if_speed
|
|
},
|
|
occurred_at: timestamp
|
|
}
|
|
end
|
|
|
|
defp determine_speed_change_severity(true, _old_speed, _new_speed), do: "info"
|
|
defp determine_speed_change_severity(false, old_speed, new_speed) when new_speed < old_speed, do: "warning"
|
|
defp determine_speed_change_severity(false, _old_speed, _new_speed), do: "info"
|
|
|
|
defp format_speed_change_message(if_name, true, _old_speed, new_speed) do
|
|
"Interface #{if_name} speed detected: #{format_speed(new_speed)}"
|
|
end
|
|
|
|
defp format_speed_change_message(if_name, false, old_speed, new_speed) do
|
|
"Interface #{if_name} speed changed from #{format_speed(old_speed)} to #{format_speed(new_speed)}"
|
|
end
|
|
|
|
defp maybe_add_mac_change_event(events, interface, current_attrs, equipment_id, timestamp) do
|
|
if interface.if_phys_address != current_attrs.if_phys_address && current_attrs.if_phys_address != nil do
|
|
event = build_mac_change_event(interface, current_attrs, equipment_id, timestamp)
|
|
[{:event, event} | events]
|
|
else
|
|
events
|
|
end
|
|
end
|
|
|
|
defp build_mac_change_event(interface, current_attrs, equipment_id, timestamp) do
|
|
is_initial = interface.if_phys_address == nil
|
|
|
|
message =
|
|
format_mac_change_message(interface.if_name, is_initial, interface.if_phys_address, current_attrs.if_phys_address)
|
|
|
|
%{
|
|
equipment_id: equipment_id,
|
|
event_type: "interface_mac_change",
|
|
severity: if(is_initial, do: "info", else: "warning"),
|
|
message: message,
|
|
metadata: %{
|
|
interface_id: interface.id,
|
|
interface_name: interface.if_name,
|
|
old_mac: interface.if_phys_address,
|
|
new_mac: current_attrs.if_phys_address
|
|
},
|
|
occurred_at: timestamp
|
|
}
|
|
end
|
|
|
|
defp format_mac_change_message(if_name, true, _old_mac, new_mac) do
|
|
"Interface #{if_name} MAC address detected: #{new_mac}"
|
|
end
|
|
|
|
defp format_mac_change_message(if_name, false, old_mac, new_mac) do
|
|
"Interface #{if_name} MAC address changed from #{old_mac} to #{new_mac}"
|
|
end
|
|
|
|
defp broadcast_interface_events(events) do
|
|
Enum.each(events, fn {:event, event_attrs} ->
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"equipment:events",
|
|
{:equipment_event, event_attrs}
|
|
)
|
|
|
|
Logger.debug("Broadcast event: #{event_attrs.message}")
|
|
end)
|
|
end
|
|
|
|
defp parse_interface_integer(value) when is_integer(value), do: value
|
|
defp parse_interface_integer(_), do: nil
|
|
|
|
defp parse_if_status(1), do: "up"
|
|
defp parse_if_status(2), do: "down"
|
|
defp parse_if_status(3), do: "testing"
|
|
defp parse_if_status(_), do: "unknown"
|
|
|
|
defp format_mac_address(<<>>) when is_binary(<<>>), do: nil
|
|
defp format_mac_address(nil), do: nil
|
|
|
|
defp format_mac_address(mac) when is_binary(mac) do
|
|
mac
|
|
|> :binary.bin_to_list()
|
|
|> Enum.map_join(":", &String.pad_leading(Integer.to_string(&1, 16), 2, "0"))
|
|
|> String.downcase()
|
|
end
|
|
|
|
defp format_mac_address(_), do: nil
|
|
|
|
defp format_speed(speed_bps) when is_integer(speed_bps) do
|
|
cond do
|
|
speed_bps >= 1_000_000_000 ->
|
|
"#{Float.round(speed_bps / 1_000_000_000, 1)} Gbps"
|
|
|
|
speed_bps >= 1_000_000 ->
|
|
"#{Float.round(speed_bps / 1_000_000, 1)} Mbps"
|
|
|
|
speed_bps >= 1_000 ->
|
|
"#{Float.round(speed_bps / 1_000, 1)} Kbps"
|
|
|
|
true ->
|
|
"#{speed_bps} bps"
|
|
end
|
|
end
|
|
|
|
defp format_speed(_), do: "Unknown"
|
|
|
|
defp get_interface_stats(client_opts, if_index) do
|
|
oids = [
|
|
if_in_octets: "1.3.6.1.2.1.2.2.1.10.#{if_index}",
|
|
if_out_octets: "1.3.6.1.2.1.2.2.1.16.#{if_index}",
|
|
if_in_errors: "1.3.6.1.2.1.2.2.1.14.#{if_index}",
|
|
if_out_errors: "1.3.6.1.2.1.2.2.1.20.#{if_index}",
|
|
if_in_discards: "1.3.6.1.2.1.2.2.1.13.#{if_index}",
|
|
if_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} -> {key, decode_snmp_value(value)}
|
|
_ -> {key, nil}
|
|
end
|
|
end)
|
|
|
|
{:ok, Map.new(results)}
|
|
end
|
|
|
|
# Decode SNMP values, handling Counter64 and other binary types
|
|
# Note: Client.get already unwraps SNMPKit tuples via extract_snmp_value
|
|
defp decode_snmp_value(value) when is_number(value) do
|
|
value
|
|
end
|
|
|
|
# Handle binary Counter64 values (8 bytes, big-endian unsigned integer)
|
|
defp decode_snmp_value(value) when is_binary(value) do
|
|
size = byte_size(value)
|
|
|
|
result =
|
|
case size do
|
|
8 ->
|
|
# Standard Counter64 (8 bytes, big-endian unsigned integer)
|
|
<<counter::unsigned-big-integer-size(64)>> = value
|
|
counter
|
|
|
|
16 ->
|
|
# Some SNMP implementations return 16-byte values
|
|
# Try reading last 8 bytes as Counter64
|
|
<<_prefix::binary-size(8), counter::unsigned-big-integer-size(64)>> = value
|
|
counter
|
|
|
|
s when s > 8 ->
|
|
# Large binary, try reading last 8 bytes
|
|
offset = s - 8
|
|
<<_prefix::binary-size(^offset), counter::unsigned-big-integer-size(64)>> = value
|
|
counter
|
|
|
|
_ ->
|
|
# Unknown binary format or too small
|
|
Logger.warning("Unknown SNMP binary value format, size: #{size}")
|
|
nil
|
|
end
|
|
|
|
result
|
|
rescue
|
|
error ->
|
|
Logger.error("Failed to decode SNMP binary value (size: #{byte_size(value)}): #{inspect(error)}")
|
|
nil
|
|
end
|
|
|
|
defp decode_snmp_value(other) do
|
|
Logger.warning("Unexpected SNMP value type: #{inspect(other)}")
|
|
nil
|
|
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 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
|
|
|
|
@spec detect_sensor_changes(Sensor.t(), float(), DateTime.t()) :: :ok
|
|
defp detect_sensor_changes(sensor, current_value, timestamp) do
|
|
# Skip change detection if there's no previous value to compare against
|
|
if sensor.last_value == nil do
|
|
:ok
|
|
else
|
|
equipment_id = get_equipment_id_from_sensor(sensor)
|
|
thresholds = extract_thresholds(sensor.metadata)
|
|
|
|
events =
|
|
[]
|
|
|> maybe_add_threshold_event(sensor, current_value, timestamp, equipment_id, thresholds)
|
|
|> maybe_add_change_event(sensor, current_value, timestamp, equipment_id)
|
|
|
|
broadcast_sensor_events(events)
|
|
end
|
|
end
|
|
|
|
@spec extract_thresholds(map()) :: %{
|
|
warning_high: float() | nil,
|
|
critical_high: float() | nil,
|
|
warning_low: float() | nil,
|
|
critical_low: float() | nil
|
|
}
|
|
defp extract_thresholds(metadata) do
|
|
%{
|
|
warning_high: metadata["warning_high"],
|
|
critical_high: metadata["critical_high"],
|
|
warning_low: metadata["warning_low"],
|
|
critical_low: metadata["critical_low"]
|
|
}
|
|
end
|
|
|
|
defp maybe_add_threshold_event(events, sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
threshold_event = check_threshold_violation(sensor, current_value, timestamp, equipment_id, thresholds)
|
|
|
|
if threshold_event, do: [threshold_event | events], else: events
|
|
end
|
|
|
|
@spec check_threshold_violation(
|
|
Sensor.t(),
|
|
float(),
|
|
DateTime.t(),
|
|
Ecto.UUID.t(),
|
|
map()
|
|
) :: map() | nil
|
|
defp check_threshold_violation(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
threshold_checks = [
|
|
&check_critical_high/5,
|
|
&check_critical_low/5,
|
|
&check_warning_high/5,
|
|
&check_warning_low/5
|
|
]
|
|
|
|
Enum.find_value(threshold_checks, fn check_fn ->
|
|
check_fn.(sensor, current_value, timestamp, equipment_id, thresholds)
|
|
end) || check_returned_to_normal(sensor, current_value, timestamp, equipment_id, thresholds)
|
|
end
|
|
|
|
@spec check_critical_high(
|
|
Sensor.t(),
|
|
float(),
|
|
DateTime.t(),
|
|
Ecto.UUID.t(),
|
|
map()
|
|
) :: map() | nil
|
|
defp check_critical_high(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
if thresholds.critical_high && current_value >= thresholds.critical_high do
|
|
build_threshold_event(
|
|
sensor,
|
|
current_value,
|
|
timestamp,
|
|
equipment_id,
|
|
"critical_high",
|
|
thresholds.critical_high,
|
|
"critical",
|
|
"critically high"
|
|
)
|
|
end
|
|
end
|
|
|
|
@spec check_critical_low(
|
|
Sensor.t(),
|
|
float(),
|
|
DateTime.t(),
|
|
Ecto.UUID.t(),
|
|
map()
|
|
) :: map() | nil
|
|
defp check_critical_low(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
if thresholds.critical_low && current_value <= thresholds.critical_low do
|
|
build_threshold_event(
|
|
sensor,
|
|
current_value,
|
|
timestamp,
|
|
equipment_id,
|
|
"critical_low",
|
|
thresholds.critical_low,
|
|
"critical",
|
|
"critically low"
|
|
)
|
|
end
|
|
end
|
|
|
|
@spec check_warning_high(
|
|
Sensor.t(),
|
|
float(),
|
|
DateTime.t(),
|
|
Ecto.UUID.t(),
|
|
map()
|
|
) :: map() | nil
|
|
defp check_warning_high(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
if thresholds.warning_high && current_value >= thresholds.warning_high do
|
|
build_threshold_event(
|
|
sensor,
|
|
current_value,
|
|
timestamp,
|
|
equipment_id,
|
|
"warning_high",
|
|
thresholds.warning_high,
|
|
"warning",
|
|
"high"
|
|
)
|
|
end
|
|
end
|
|
|
|
@spec check_warning_low(
|
|
Sensor.t(),
|
|
float(),
|
|
DateTime.t(),
|
|
Ecto.UUID.t(),
|
|
map()
|
|
) :: map() | nil
|
|
defp check_warning_low(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
if thresholds.warning_low && current_value <= thresholds.warning_low do
|
|
build_threshold_event(
|
|
sensor,
|
|
current_value,
|
|
timestamp,
|
|
equipment_id,
|
|
"warning_low",
|
|
thresholds.warning_low,
|
|
"warning",
|
|
"low"
|
|
)
|
|
end
|
|
end
|
|
|
|
defp build_threshold_event(
|
|
sensor,
|
|
current_value,
|
|
timestamp,
|
|
equipment_id,
|
|
threshold_type,
|
|
threshold_value,
|
|
severity,
|
|
description
|
|
) do
|
|
event_type = if severity == "critical", do: "sensor_threshold_critical", else: "sensor_threshold_warning"
|
|
|
|
build_sensor_event(
|
|
equipment_id,
|
|
sensor,
|
|
event_type,
|
|
severity,
|
|
"#{sensor.sensor_descr} is #{description}: #{format_sensor_value(current_value, sensor.sensor_unit)} (threshold: #{threshold_value}#{sensor.sensor_unit})",
|
|
%{
|
|
current_value: current_value,
|
|
previous_value: sensor.last_value,
|
|
threshold_type: threshold_type,
|
|
threshold_value: threshold_value
|
|
},
|
|
timestamp
|
|
)
|
|
end
|
|
|
|
defp check_returned_to_normal(sensor, current_value, timestamp, equipment_id, thresholds) do
|
|
was_over_threshold = value_was_over_threshold?(sensor.last_value, thresholds)
|
|
is_now_normal = value_is_normal?(current_value, thresholds)
|
|
|
|
if was_over_threshold && is_now_normal do
|
|
build_sensor_event(
|
|
equipment_id,
|
|
sensor,
|
|
"sensor_threshold_normal",
|
|
"info",
|
|
"#{sensor.sensor_descr} returned to normal: #{format_sensor_value(current_value, sensor.sensor_unit)}",
|
|
%{current_value: current_value, previous_value: sensor.last_value},
|
|
timestamp
|
|
)
|
|
end
|
|
end
|
|
|
|
defp value_was_over_threshold?(last_value, thresholds) do
|
|
(thresholds.warning_high && last_value >= thresholds.warning_high) ||
|
|
(thresholds.critical_high && last_value >= thresholds.critical_high) ||
|
|
(thresholds.warning_low && last_value <= thresholds.warning_low) ||
|
|
(thresholds.critical_low && last_value <= thresholds.critical_low)
|
|
end
|
|
|
|
defp value_is_normal?(current_value, thresholds) do
|
|
(thresholds.warning_high == nil || current_value < thresholds.warning_high) &&
|
|
(thresholds.critical_high == nil || current_value < thresholds.critical_high) &&
|
|
(thresholds.warning_low == nil || current_value > thresholds.warning_low) &&
|
|
(thresholds.critical_low == nil || current_value > thresholds.critical_low)
|
|
end
|
|
|
|
defp maybe_add_change_event(events, sensor, current_value, timestamp, equipment_id) do
|
|
# Only check for spikes/drops on percentage sensors
|
|
if sensor.sensor_unit == "%" do
|
|
change_event = check_significant_change(sensor, current_value, timestamp, equipment_id)
|
|
if change_event, do: [change_event | events], else: events
|
|
else
|
|
events
|
|
end
|
|
end
|
|
|
|
defp check_significant_change(sensor, current_value, timestamp, equipment_id) do
|
|
change_percent = abs(current_value - sensor.last_value)
|
|
|
|
cond do
|
|
# Spike: increase of 30% or more
|
|
change_percent >= 30 && current_value > sensor.last_value ->
|
|
build_sensor_event(
|
|
equipment_id,
|
|
sensor,
|
|
"sensor_value_spike",
|
|
"warning",
|
|
"#{sensor.sensor_descr} spiked from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}",
|
|
%{current_value: current_value, previous_value: sensor.last_value, change: change_percent},
|
|
timestamp
|
|
)
|
|
|
|
# Drop: decrease of 30% or more
|
|
change_percent >= 30 && current_value < sensor.last_value ->
|
|
build_sensor_event(
|
|
equipment_id,
|
|
sensor,
|
|
"sensor_value_drop",
|
|
"info",
|
|
"#{sensor.sensor_descr} dropped from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}",
|
|
%{current_value: current_value, previous_value: sensor.last_value, change: change_percent},
|
|
timestamp
|
|
)
|
|
|
|
true ->
|
|
nil
|
|
end
|
|
end
|
|
|
|
defp broadcast_sensor_events(events) do
|
|
Enum.each(events, fn event ->
|
|
_ = Phoenix.PubSub.broadcast(Towerops.PubSub, "equipment:events", {:equipment_event, event})
|
|
Logger.debug("Sensor event: #{event.message}")
|
|
end)
|
|
end
|
|
|
|
defp build_sensor_event(equipment_id, sensor, event_type, severity, message, metadata, timestamp) do
|
|
base_metadata = %{
|
|
sensor_id: sensor.id,
|
|
sensor_name: sensor.sensor_descr,
|
|
sensor_type: sensor.sensor_type,
|
|
sensor_unit: sensor.sensor_unit
|
|
}
|
|
|
|
%{
|
|
equipment_id: equipment_id,
|
|
event_type: event_type,
|
|
severity: severity,
|
|
message: message,
|
|
metadata: Map.merge(base_metadata, metadata),
|
|
occurred_at: timestamp
|
|
}
|
|
end
|
|
|
|
defp get_equipment_id_from_sensor(sensor) do
|
|
# Get the device to find equipment_id
|
|
case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do
|
|
nil -> nil
|
|
device -> device.equipment_id
|
|
end
|
|
end
|
|
|
|
defp format_sensor_value(value, unit) when is_number(value) do
|
|
"#{Float.round(value, 1)}#{unit}"
|
|
end
|
|
|
|
defp format_sensor_value(_, _), do: "N/A"
|
|
|
|
defp via_tuple(equipment_id) do
|
|
{:via, Registry, {PollerRegistry, equipment_id}}
|
|
end
|
|
end
|