add sensor value change events for all sensor types
extract sensor change detection from DevicePollerWorker into shared SensorChangeDetector module. add sensor_value_changed event for non-% sensors when value differs from last_value. agent channel now detects changes and updates last_value, matching phoenix poller behavior.
This commit is contained in:
parent
45357e334a
commit
abbe117f5e
6 changed files with 631 additions and 245 deletions
|
|
@ -1,6 +1,23 @@
|
|||
CHANGELOG - towerops-web
|
||||
========================
|
||||
|
||||
2026-02-11 - feat: add sensor value change events for all sensor types
|
||||
- New file: lib/towerops/snmp/sensor_change_detector.ex
|
||||
Extracted shared sensor change detection logic from DevicePollerWorker.
|
||||
Detects threshold violations, spike/drop for % sensors, and new
|
||||
sensor_value_changed events for all non-% sensors when value changes.
|
||||
- File: lib/towerops/devices/event.ex
|
||||
Added "sensor_value_changed" to allowed event types.
|
||||
- File: lib/towerops/workers/device_poller_worker.ex
|
||||
Replaced inline detect_sensor_changes with SensorChangeDetector module.
|
||||
Removed ~240 lines of private functions now in shared module.
|
||||
- File: lib/towerops_web/channels/agent_channel.ex
|
||||
Added change detection and last_value/last_checked_at update to
|
||||
process_sensor_reading/3 so agent-polled devices get same events
|
||||
as Phoenix-polled devices.
|
||||
- New file: test/towerops/snmp/sensor_change_detector_test.exs
|
||||
11 tests covering value changes, spike/drop, thresholds, nil handling.
|
||||
|
||||
2026-02-11 - fix: live polling uses effective agent from site/org cascade
|
||||
- File: lib/towerops_web/live/graph_live/show.ex
|
||||
- Bug: Live poll mode used Agents.get_device_assignment/1 which only checks
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ defmodule Towerops.Devices.Event do
|
|||
| :sensor_threshold_normal
|
||||
| :sensor_value_spike
|
||||
| :sensor_value_drop
|
||||
| :sensor_value_changed
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
|
@ -56,7 +57,8 @@ defmodule Towerops.Devices.Event do
|
|||
"sensor_threshold_critical",
|
||||
"sensor_threshold_normal",
|
||||
"sensor_value_spike",
|
||||
"sensor_value_drop"
|
||||
"sensor_value_drop",
|
||||
"sensor_value_changed"
|
||||
])
|
||||
|> validate_inclusion(:severity, ["info", "warning", "critical"])
|
||||
|> foreign_key_constraint(:device_id)
|
||||
|
|
|
|||
291
lib/towerops/snmp/sensor_change_detector.ex
Normal file
291
lib/towerops/snmp/sensor_change_detector.ex
Normal file
|
|
@ -0,0 +1,291 @@
|
|||
defmodule Towerops.Snmp.SensorChangeDetector do
|
||||
@moduledoc """
|
||||
Detects sensor value changes and broadcasts events.
|
||||
|
||||
Shared between DevicePollerWorker and AgentChannel to ensure consistent
|
||||
event detection across both Phoenix-polled and agent-polled devices.
|
||||
"""
|
||||
|
||||
alias Towerops.Snmp.Sensor
|
||||
|
||||
require Logger
|
||||
|
||||
@doc """
|
||||
Detects sensor changes and broadcasts events via PubSub.
|
||||
|
||||
For percentage sensors: emits spike/drop events for >=30% swings.
|
||||
For all other sensors: emits sensor_value_changed when value differs from last_value.
|
||||
For all sensors with thresholds: emits threshold warning/critical/normal events.
|
||||
|
||||
No-ops when `sensor.last_value` is nil (first reading).
|
||||
"""
|
||||
@spec detect_and_broadcast(Sensor.t(), number(), DateTime.t()) :: :ok
|
||||
def detect_and_broadcast(%Sensor{last_value: nil}, _current_value, _timestamp), do: :ok
|
||||
|
||||
def detect_and_broadcast(%Sensor{} = sensor, current_value, timestamp) do
|
||||
device_id = get_device_id_from_sensor(sensor)
|
||||
|
||||
if device_id do
|
||||
thresholds = extract_thresholds(sensor.metadata)
|
||||
|
||||
events =
|
||||
[]
|
||||
|> maybe_add_threshold_event(sensor, current_value, timestamp, device_id, thresholds)
|
||||
|> maybe_add_change_event(sensor, current_value, timestamp, device_id)
|
||||
|
||||
broadcast_events(events)
|
||||
end
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
# Threshold detection
|
||||
|
||||
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, device_id, thresholds) do
|
||||
threshold_event = check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds)
|
||||
|
||||
if threshold_event, do: [threshold_event | events], else: events
|
||||
end
|
||||
|
||||
defp check_threshold_violation(sensor, current_value, timestamp, device_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, device_id, thresholds)
|
||||
end) || check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds)
|
||||
end
|
||||
|
||||
defp check_critical_high(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.critical_high && current_value >= thresholds.critical_high do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"critical_high",
|
||||
thresholds.critical_high,
|
||||
"critical",
|
||||
"critically high"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_critical_low(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.critical_low && current_value <= thresholds.critical_low do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"critical_low",
|
||||
thresholds.critical_low,
|
||||
"critical",
|
||||
"critically low"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_warning_high(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.warning_high && current_value >= thresholds.warning_high do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"warning_high",
|
||||
thresholds.warning_high,
|
||||
"warning",
|
||||
"high"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_warning_low(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.warning_low && current_value <= thresholds.warning_low do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"warning_low",
|
||||
thresholds.warning_low,
|
||||
"warning",
|
||||
"low"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
threshold_type,
|
||||
threshold_value,
|
||||
severity,
|
||||
description
|
||||
) do
|
||||
event_type = if severity == "critical", do: "sensor_threshold_critical", else: "sensor_threshold_warning"
|
||||
|
||||
build_sensor_event(
|
||||
device_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, device_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(
|
||||
device_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
|
||||
|
||||
# Change detection
|
||||
|
||||
defp maybe_add_change_event(events, sensor, current_value, timestamp, device_id) do
|
||||
if sensor.sensor_unit == "%" do
|
||||
# Percentage sensors use spike/drop events for large swings
|
||||
change_event = check_significant_change(sensor, current_value, timestamp, device_id)
|
||||
if change_event, do: [change_event | events], else: events
|
||||
else
|
||||
# All other sensors emit value_changed when value differs
|
||||
if current_value == sensor.last_value do
|
||||
events
|
||||
else
|
||||
event =
|
||||
build_sensor_event(
|
||||
device_id,
|
||||
sensor,
|
||||
"sensor_value_changed",
|
||||
"info",
|
||||
"#{sensor.sensor_descr} changed 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},
|
||||
timestamp
|
||||
)
|
||||
|
||||
[event | events]
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp check_significant_change(sensor, current_value, timestamp, device_id) do
|
||||
change_percent = abs(current_value - sensor.last_value)
|
||||
|
||||
cond do
|
||||
change_percent >= 30 && current_value > sensor.last_value ->
|
||||
build_sensor_event(
|
||||
device_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
|
||||
)
|
||||
|
||||
change_percent >= 30 && current_value < sensor.last_value ->
|
||||
build_sensor_event(
|
||||
device_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
|
||||
|
||||
# Event building and broadcasting
|
||||
|
||||
defp build_sensor_event(device_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
|
||||
}
|
||||
|
||||
%{
|
||||
device_id: device_id,
|
||||
event_type: event_type,
|
||||
severity: severity,
|
||||
message: message,
|
||||
metadata: Map.merge(base_metadata, metadata),
|
||||
occurred_at: timestamp
|
||||
}
|
||||
end
|
||||
|
||||
defp broadcast_events(events) do
|
||||
Enum.each(events, fn event ->
|
||||
_ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{event.device_id}", {:device_event, event})
|
||||
_ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event})
|
||||
|
||||
Logger.debug("Sensor event: #{event.message}")
|
||||
end)
|
||||
end
|
||||
|
||||
defp get_device_id_from_sensor(sensor) do
|
||||
case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do
|
||||
nil -> nil
|
||||
device -> device.device_id
|
||||
end
|
||||
end
|
||||
|
||||
defp format_sensor_value(value, unit) when is_number(value) do
|
||||
"#{Float.round(value / 1.0, 1)}#{unit}"
|
||||
end
|
||||
|
||||
defp format_sensor_value(_, _), do: "N/A"
|
||||
end
|
||||
|
|
@ -28,6 +28,7 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
alias Towerops.Snmp.Client
|
||||
alias Towerops.Snmp.MacDiscovery
|
||||
alias Towerops.Snmp.NeighborDiscovery
|
||||
alias Towerops.Snmp.SensorChangeDetector
|
||||
alias Towerops.Workers.PollingOffset
|
||||
|
||||
require Logger
|
||||
|
|
@ -731,7 +732,7 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
checked_at: timestamp
|
||||
})
|
||||
|
||||
detect_sensor_changes(sensor, value, timestamp)
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, value, timestamp)
|
||||
|
||||
Snmp.update_sensor(sensor, %{
|
||||
last_value: value,
|
||||
|
|
@ -1266,247 +1267,4 @@ defmodule Towerops.Workers.DevicePollerWorker do
|
|||
seconds_since_poll < min_interval
|
||||
end
|
||||
end
|
||||
|
||||
defp detect_sensor_changes(sensor, current_value, timestamp) do
|
||||
if sensor.last_value == nil do
|
||||
:ok
|
||||
else
|
||||
device_id = get_device_id_from_sensor(sensor)
|
||||
thresholds = extract_thresholds(sensor.metadata)
|
||||
|
||||
events =
|
||||
[]
|
||||
|> maybe_add_threshold_event(sensor, current_value, timestamp, device_id, thresholds)
|
||||
|> maybe_add_change_event(sensor, current_value, timestamp, device_id)
|
||||
|
||||
broadcast_sensor_events(events)
|
||||
end
|
||||
end
|
||||
|
||||
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, device_id, thresholds) do
|
||||
threshold_event = check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds)
|
||||
|
||||
if threshold_event, do: [threshold_event | events], else: events
|
||||
end
|
||||
|
||||
defp check_threshold_violation(sensor, current_value, timestamp, device_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, device_id, thresholds)
|
||||
end) || check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds)
|
||||
end
|
||||
|
||||
defp check_critical_high(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.critical_high && current_value >= thresholds.critical_high do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"critical_high",
|
||||
thresholds.critical_high,
|
||||
"critical",
|
||||
"critically high"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_critical_low(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.critical_low && current_value <= thresholds.critical_low do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"critical_low",
|
||||
thresholds.critical_low,
|
||||
"critical",
|
||||
"critically low"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_warning_high(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.warning_high && current_value >= thresholds.warning_high do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"warning_high",
|
||||
thresholds.warning_high,
|
||||
"warning",
|
||||
"high"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp check_warning_low(sensor, current_value, timestamp, device_id, thresholds) do
|
||||
if thresholds.warning_low && current_value <= thresholds.warning_low do
|
||||
build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
"warning_low",
|
||||
thresholds.warning_low,
|
||||
"warning",
|
||||
"low"
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp build_threshold_event(
|
||||
sensor,
|
||||
current_value,
|
||||
timestamp,
|
||||
device_id,
|
||||
threshold_type,
|
||||
threshold_value,
|
||||
severity,
|
||||
description
|
||||
) do
|
||||
event_type = if severity == "critical", do: "sensor_threshold_critical", else: "sensor_threshold_warning"
|
||||
|
||||
build_sensor_event(
|
||||
device_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, device_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(
|
||||
device_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, device_id) do
|
||||
if sensor.sensor_unit == "%" do
|
||||
change_event = check_significant_change(sensor, current_value, timestamp, device_id)
|
||||
if change_event, do: [change_event | events], else: events
|
||||
else
|
||||
events
|
||||
end
|
||||
end
|
||||
|
||||
defp check_significant_change(sensor, current_value, timestamp, device_id) do
|
||||
change_percent = abs(current_value - sensor.last_value)
|
||||
|
||||
cond do
|
||||
change_percent >= 30 && current_value > sensor.last_value ->
|
||||
build_sensor_event(
|
||||
device_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
|
||||
)
|
||||
|
||||
change_percent >= 30 && current_value < sensor.last_value ->
|
||||
build_sensor_event(
|
||||
device_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, "device:#{event.device_id}", {:device_event, event})
|
||||
_ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event})
|
||||
|
||||
Logger.debug("Sensor event: #{event.message}")
|
||||
end)
|
||||
end
|
||||
|
||||
defp build_sensor_event(device_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
|
||||
}
|
||||
|
||||
%{
|
||||
device_id: device_id,
|
||||
event_type: event_type,
|
||||
severity: severity,
|
||||
message: message,
|
||||
metadata: Map.merge(base_metadata, metadata),
|
||||
occurred_at: timestamp
|
||||
}
|
||||
end
|
||||
|
||||
defp get_device_id_from_sensor(sensor) do
|
||||
case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do
|
||||
nil -> nil
|
||||
device -> device.device_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"
|
||||
end
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
alias Towerops.Snmp
|
||||
alias Towerops.Snmp.AgentDiscovery
|
||||
alias Towerops.Snmp.Discovery
|
||||
alias Towerops.Snmp.SensorChangeDetector
|
||||
|
||||
require Logger
|
||||
|
||||
|
|
@ -1547,6 +1548,13 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
status: "ok",
|
||||
checked_at: timestamp
|
||||
})
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, final_value, timestamp)
|
||||
|
||||
Snmp.update_sensor(sensor, %{
|
||||
last_value: final_value,
|
||||
last_checked_at: timestamp
|
||||
})
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
310
test/towerops/snmp/sensor_change_detector_test.exs
Normal file
310
test/towerops/snmp/sensor_change_detector_test.exs
Normal file
|
|
@ -0,0 +1,310 @@
|
|||
defmodule Towerops.Snmp.SensorChangeDetectorTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
alias Towerops.Devices
|
||||
alias Towerops.Organizations
|
||||
alias Towerops.Repo
|
||||
alias Towerops.Sites
|
||||
alias Towerops.Snmp.Device, as: SnmpDevice
|
||||
alias Towerops.Snmp.Sensor
|
||||
alias Towerops.Snmp.SensorChangeDetector
|
||||
|
||||
setup do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device} =
|
||||
Devices.create_device(%{
|
||||
name: "Test Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id,
|
||||
organization_id: organization.id,
|
||||
snmp_enabled: true
|
||||
})
|
||||
|
||||
{:ok, snmp_device} =
|
||||
Repo.insert(%SnmpDevice{
|
||||
device_id: device.id,
|
||||
sys_name: "test-device",
|
||||
sys_descr: "Test Device"
|
||||
})
|
||||
|
||||
%{device: device, snmp_device: snmp_device}
|
||||
end
|
||||
|
||||
defp create_sensor(snmp_device, attrs \\ %{}) do
|
||||
defaults = %{
|
||||
snmp_device_id: snmp_device.id,
|
||||
sensor_type: "frequency",
|
||||
sensor_index: "1",
|
||||
sensor_oid: "1.3.6.1.2.1.99.1.1.1.4.1",
|
||||
sensor_descr: "Power Supply 1 Frequency",
|
||||
sensor_unit: "Hz",
|
||||
sensor_divisor: 1,
|
||||
monitored: true,
|
||||
metadata: %{}
|
||||
}
|
||||
|
||||
{:ok, sensor} = Repo.insert(struct(Sensor, Map.merge(defaults, attrs)))
|
||||
sensor
|
||||
end
|
||||
|
||||
describe "detect_and_broadcast/3" do
|
||||
test "emits sensor_value_changed for non-% sensor when value changes", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"})
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 49.9, timestamp)
|
||||
|
||||
assert_receive {:device_event, event}
|
||||
assert event.event_type == "sensor_value_changed"
|
||||
assert event.severity == "info"
|
||||
assert event.device_id == device.id
|
||||
assert event.metadata.current_value == 49.9
|
||||
assert event.metadata.previous_value == 50.0
|
||||
assert event.metadata.sensor_id == sensor.id
|
||||
assert String.contains?(event.message, "changed from")
|
||||
assert String.contains?(event.message, "50.0Hz")
|
||||
assert String.contains?(event.message, "49.9Hz")
|
||||
end
|
||||
|
||||
test "emits sensor_value_changed for temperature sensor", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "temperature",
|
||||
sensor_descr: "CPU Temp",
|
||||
sensor_unit: "C",
|
||||
last_value: 45.0
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 46.0, timestamp)
|
||||
|
||||
assert_receive {:device_event, event}
|
||||
assert event.event_type == "sensor_value_changed"
|
||||
assert String.contains?(event.message, "CPU Temp")
|
||||
assert String.contains?(event.message, "45.0C")
|
||||
assert String.contains?(event.message, "46.0C")
|
||||
end
|
||||
|
||||
test "does not emit event when value is unchanged", %{snmp_device: snmp_device} do
|
||||
sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"})
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
# Subscribe to both topics
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 50.0, timestamp)
|
||||
|
||||
refute_receive {:device_event, _}, 100
|
||||
end
|
||||
|
||||
test "does not emit sensor_value_changed when last_value is nil", %{snmp_device: snmp_device} do
|
||||
sensor = create_sensor(snmp_device, %{last_value: nil, sensor_unit: "Hz"})
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 50.0, timestamp)
|
||||
|
||||
refute_receive {:device_event, _}, 100
|
||||
end
|
||||
|
||||
test "emits spike event for % sensor with >=30% increase", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "percent",
|
||||
sensor_descr: "CPU Usage",
|
||||
sensor_unit: "%",
|
||||
last_value: 20.0,
|
||||
metadata: %{}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 55.0, timestamp)
|
||||
|
||||
assert_receive {:device_event, event}
|
||||
assert event.event_type == "sensor_value_spike"
|
||||
assert event.severity == "warning"
|
||||
end
|
||||
|
||||
test "emits drop event for % sensor with >=30% decrease", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "percent",
|
||||
sensor_descr: "CPU Usage",
|
||||
sensor_unit: "%",
|
||||
last_value: 80.0,
|
||||
metadata: %{}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 40.0, timestamp)
|
||||
|
||||
assert_receive {:device_event, event}
|
||||
assert event.event_type == "sensor_value_drop"
|
||||
assert event.severity == "info"
|
||||
end
|
||||
|
||||
test "does not emit spike/drop for % sensor with <30% change", %{snmp_device: snmp_device} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "percent",
|
||||
sensor_descr: "CPU Usage",
|
||||
sensor_unit: "%",
|
||||
last_value: 50.0,
|
||||
metadata: %{}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 55.0, timestamp)
|
||||
|
||||
# Should not get spike/drop events for small % changes
|
||||
# (no sensor_value_changed for % sensors either, they use spike/drop)
|
||||
refute_receive {:device_event, _}, 100
|
||||
end
|
||||
|
||||
test "emits threshold warning event", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "temperature",
|
||||
sensor_descr: "Chassis Temp",
|
||||
sensor_unit: "C",
|
||||
last_value: 60.0,
|
||||
metadata: %{"warning_high" => 70, "critical_high" => 90}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 75.0, timestamp)
|
||||
|
||||
# Should get both threshold warning AND value changed
|
||||
events = collect_events(2)
|
||||
|
||||
event_types = Enum.map(events, & &1.event_type)
|
||||
assert "sensor_threshold_warning" in event_types
|
||||
assert "sensor_value_changed" in event_types
|
||||
end
|
||||
|
||||
test "emits threshold critical event", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "temperature",
|
||||
sensor_descr: "Chassis Temp",
|
||||
sensor_unit: "C",
|
||||
last_value: 60.0,
|
||||
metadata: %{"warning_high" => 70, "critical_high" => 90}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 95.0, timestamp)
|
||||
|
||||
events = collect_events(2)
|
||||
|
||||
event_types = Enum.map(events, & &1.event_type)
|
||||
assert "sensor_threshold_critical" in event_types
|
||||
assert "sensor_value_changed" in event_types
|
||||
end
|
||||
|
||||
test "emits threshold normal event when returning from warning", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor =
|
||||
create_sensor(snmp_device, %{
|
||||
sensor_type: "temperature",
|
||||
sensor_descr: "Chassis Temp",
|
||||
sensor_unit: "C",
|
||||
last_value: 75.0,
|
||||
metadata: %{"warning_high" => 70, "critical_high" => 90}
|
||||
})
|
||||
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 65.0, timestamp)
|
||||
|
||||
events = collect_events(2)
|
||||
|
||||
event_types = Enum.map(events, & &1.event_type)
|
||||
assert "sensor_threshold_normal" in event_types
|
||||
assert "sensor_value_changed" in event_types
|
||||
end
|
||||
|
||||
test "broadcasts to both device and global event topics", %{
|
||||
device: device,
|
||||
snmp_device: snmp_device
|
||||
} do
|
||||
sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"})
|
||||
timestamp = DateTime.utc_now()
|
||||
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events")
|
||||
|
||||
SensorChangeDetector.detect_and_broadcast(sensor, 49.0, timestamp)
|
||||
|
||||
# Should receive on both topics
|
||||
assert_receive {:device_event, event1}
|
||||
assert_receive {:device_event, event2}
|
||||
assert event1.event_type == "sensor_value_changed"
|
||||
assert event2.event_type == "sensor_value_changed"
|
||||
end
|
||||
end
|
||||
|
||||
# Helper to collect multiple events
|
||||
defp collect_events(count) do
|
||||
Enum.map(1..count, fn _ ->
|
||||
receive do
|
||||
{:device_event, event} -> event
|
||||
after
|
||||
500 -> flunk("Expected #{count} events but timed out")
|
||||
end
|
||||
end)
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue