feat: add historical wireless client tracking with TimescaleDB

- Create wireless_client_readings hypertable with compression and retention
- Add WirelessClientReading schema and batch insert function
- Update DevicePollerWorker to save historical readings
- Add signal_badge and snr_badge components to CoreComponents
This commit is contained in:
Graham McIntire 2026-03-09 17:42:17 -05:00
parent 05d2db4925
commit f09d1b296d
5 changed files with 508 additions and 17 deletions

View file

@ -27,6 +27,7 @@ defmodule Towerops.Snmp do
alias Towerops.Snmp.StorageReading
alias Towerops.Snmp.Vlan
alias Towerops.Snmp.WirelessClient
alias Towerops.Snmp.WirelessClientReading
require Logger
@ -2389,4 +2390,35 @@ defmodule Towerops.Snmp do
upsert_wireless_clients(device_id, organization_id, clients)
end)
end
@doc """
Batch inserts multiple wireless client readings using `Repo.insert_all/3`.
Accepts a list of attribute maps containing wireless client metrics.
Generates UUIDs and timestamps automatically. Returns `{count, nil}`.
## Examples
iex> create_wireless_client_readings_batch([
...> %{device_id: id, wireless_client_id: wc_id, organization_id: org_id,
...> mac_address: "aa:bb:cc:dd:ee:ff", signal_strength: -65,
...> snr: 30, checked_at: ~U[2024-01-01 00:00:00Z]}
...> ])
{1, nil}
"""
@spec create_wireless_client_readings_batch([map()]) :: {non_neg_integer(), nil}
def create_wireless_client_readings_batch([]), do: {0, nil}
def create_wireless_client_readings_batch(entries) when is_list(entries) do
now = DateTime.truncate(DateTime.utc_now(), :second)
rows =
Enum.map(entries, fn entry ->
entry
|> Map.put(:id, Ecto.UUID.generate())
|> Map.put(:inserted_at, now)
end)
Repo.insert_all(WirelessClientReading, rows)
end
end

View file

@ -0,0 +1,89 @@
defmodule Towerops.Snmp.WirelessClientReading do
@moduledoc """
Time-series wireless client reading schema.
Stores snapshots of wireless client metrics (signal strength, SNR, distance, rates)
collected during SNMP polling cycles. Stored in TimescaleDB hypertable for
historical trend analysis and proactive alerting.
"""
use Ecto.Schema
import Ecto.Changeset
alias Ecto.Association.NotLoaded
alias Towerops.Organizations.Organization
alias Towerops.Snmp.Device
alias Towerops.Snmp.WirelessClient
@primary_key {:id, :binary_id, autogenerate: true}
@foreign_key_type :binary_id
schema "wireless_client_readings" do
field :checked_at, :utc_datetime
field :mac_address, :string
field :ip_address, :string
field :signal_strength, :integer
field :snr, :integer
field :distance, :integer
field :tx_rate, :integer
field :rx_rate, :integer
field :uptime_seconds, :integer
field :metadata, :map, default: %{}
belongs_to :device, Device
belongs_to :wireless_client, WirelessClient
belongs_to :organization, Organization
timestamps(type: :utc_datetime, updated_at: false)
end
@type t :: %__MODULE__{
id: Ecto.UUID.t(),
checked_at: DateTime.t(),
mac_address: String.t(),
ip_address: String.t() | nil,
signal_strength: integer() | nil,
snr: integer() | nil,
distance: integer() | nil,
tx_rate: integer() | nil,
rx_rate: integer() | nil,
uptime_seconds: integer() | nil,
metadata: map(),
device_id: Ecto.UUID.t(),
device: NotLoaded.t() | Device.t(),
wireless_client_id: Ecto.UUID.t(),
wireless_client: NotLoaded.t() | WirelessClient.t(),
organization_id: Ecto.UUID.t(),
organization: NotLoaded.t() | Organization.t(),
inserted_at: DateTime.t()
}
@doc false
def changeset(reading, attrs) do
reading
|> cast(attrs, [
:device_id,
:wireless_client_id,
:organization_id,
:mac_address,
:ip_address,
:signal_strength,
:snr,
:distance,
:tx_rate,
:rx_rate,
:uptime_seconds,
:metadata,
:checked_at
])
|> validate_required([
:device_id,
:wireless_client_id,
:organization_id,
:mac_address,
:checked_at
])
|> foreign_key_constraint(:device_id)
|> foreign_key_constraint(:wireless_client_id)
|> foreign_key_constraint(:organization_id)
end
end

View file

@ -410,23 +410,7 @@ defmodule Towerops.Workers.DevicePollerWorker do
if vendor_profile do
{:ok, clients} = WirelessClientDiscovery.discover_wireless_clients(client_opts, vendor_profile)
if clients != [] do
cutoff = DateTime.add(DateTime.utc_now(), -5, :minute)
Snmp.delete_stale_and_upsert_wireless_clients(device.id, device.organization_id, clients, cutoff)
Logger.debug("Polled and saved #{length(clients)} wireless clients for #{device.name}")
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"device:#{device.id}",
{:wireless_clients_updated, device.id}
)
# Refresh subscriber-to-device links with new wireless client data
SubscriberMatching.refresh_links_for_device(device.id)
end
process_wireless_clients(device, clients)
end
rescue
error ->
@ -435,6 +419,58 @@ defmodule Towerops.Workers.DevicePollerWorker do
)
end
defp process_wireless_clients(_device, []), do: :ok
defp process_wireless_clients(device, clients) do
now = DateTime.truncate(DateTime.utc_now(), :second)
cutoff = DateTime.add(now, -5, :minute)
Snmp.delete_stale_and_upsert_wireless_clients(device.id, device.organization_id, clients, cutoff)
# Load upserted wireless clients to create historical readings
wireless_clients = Snmp.list_wireless_clients(device.id)
# Build and insert historical readings
save_wireless_client_readings(device, wireless_clients, now)
Logger.debug("Polled and saved #{length(clients)} wireless clients for #{device.name}")
_ =
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"wireless_clients:device:#{device.id}",
{:wireless_clients_updated, device.id, length(wireless_clients)}
)
# Refresh subscriber-to-device links with new wireless client data
SubscriberMatching.refresh_links_for_device(device.id)
end
defp save_wireless_client_readings(_device, [], _now), do: :ok
defp save_wireless_client_readings(device, wireless_clients, now) do
reading_entries =
Enum.map(wireless_clients, fn client ->
%{
device_id: device.id,
wireless_client_id: client.id,
organization_id: device.organization_id,
mac_address: client.mac_address,
ip_address: client.ip_address,
signal_strength: client.signal_strength,
snr: client.snr,
distance: client.distance,
tx_rate: client.tx_rate,
rx_rate: client.rx_rate,
uptime_seconds: client.uptime_seconds,
metadata: client.metadata || %{},
checked_at: now
}
end)
{count, _} = Snmp.create_wireless_client_readings_batch(reading_entries)
Logger.debug("Inserted #{count} wireless client readings for #{device.name}")
end
defp poll_device_processors(device, snmp_device, client_opts, now) do
processors = snmp_device.processors || []

View file

@ -859,6 +859,124 @@ defmodule ToweropsWeb.CoreComponents do
"""
end
@doc """
Renders a signal strength badge with color-coded quality indicator.
## Thresholds (dBm)
- Excellent: -55 or higher (green)
- Good: -56 to -65 (blue)
- Fair: -66 to -75 (yellow)
- Poor: -76 to -85 (orange)
- Critical: below -85 (red)
## Examples
<.signal_badge value={-60} />
<.signal_badge value={nil} />
"""
attr :value, :integer, doc: "Signal strength in dBm"
def signal_badge(%{value: nil} = assigns) do
~H"""
<span class="inline-flex items-center rounded-md bg-gray-50 px-2 py-1 text-xs font-medium text-gray-600 ring-1 ring-gray-500/10 ring-inset dark:bg-gray-400/10 dark:text-gray-400 dark:ring-gray-400/20">
</span>
"""
end
def signal_badge(assigns) do
{color, label} =
cond do
assigns.value >= -55 -> {"green", "Excellent"}
assigns.value >= -65 -> {"blue", "Good"}
assigns.value >= -75 -> {"yellow", "Fair"}
assigns.value >= -85 -> {"orange", "Poor"}
true -> {"red", "Critical"}
end
assigns = assign(assigns, color: color, label: label)
~H"""
<span
class={[
"inline-flex items-center rounded-md px-2 py-1 text-xs font-medium ring-1 ring-inset",
@color == "green" &&
"bg-green-50 text-green-700 ring-green-600/20 dark:bg-green-400/10 dark:text-green-400 dark:ring-green-400/20",
@color == "blue" &&
"bg-blue-50 text-blue-700 ring-blue-600/20 dark:bg-blue-400/10 dark:text-blue-400 dark:ring-blue-400/20",
@color == "yellow" &&
"bg-yellow-50 text-yellow-800 ring-yellow-600/20 dark:bg-yellow-400/10 dark:text-yellow-500 dark:ring-yellow-400/20",
@color == "orange" &&
"bg-orange-50 text-orange-700 ring-orange-600/20 dark:bg-orange-400/10 dark:text-orange-400 dark:ring-orange-400/20",
@color == "red" &&
"bg-red-50 text-red-700 ring-red-600/10 dark:bg-red-400/10 dark:text-red-400 dark:ring-red-400/20"
]}
title={@label}
>
{@value} dBm
</span>
"""
end
@doc """
Renders a Signal-to-Noise Ratio (SNR) badge with color-coded quality indicator.
## Thresholds (dB)
- Excellent: 35+ (green)
- Good: 25-34 (blue)
- Fair: 15-24 (yellow)
- Poor: 10-14 (orange)
- Critical: below 10 (red)
## Examples
<.snr_badge value={30} />
<.snr_badge value={nil} />
"""
attr :value, :integer, doc: "SNR in dB"
def snr_badge(%{value: nil} = assigns) do
~H"""
<span class="inline-flex items-center rounded-md bg-gray-50 px-2 py-1 text-xs font-medium text-gray-600 ring-1 ring-gray-500/10 ring-inset dark:bg-gray-400/10 dark:text-gray-400 dark:ring-gray-400/20">
</span>
"""
end
def snr_badge(assigns) do
{color, label} =
cond do
assigns.value >= 35 -> {"green", "Excellent"}
assigns.value >= 25 -> {"blue", "Good"}
assigns.value >= 15 -> {"yellow", "Fair"}
assigns.value >= 10 -> {"orange", "Poor"}
true -> {"red", "Critical"}
end
assigns = assign(assigns, color: color, label: label)
~H"""
<span
class={[
"inline-flex items-center rounded-md px-2 py-1 text-xs font-medium ring-1 ring-inset",
@color == "green" &&
"bg-green-50 text-green-700 ring-green-600/20 dark:bg-green-400/10 dark:text-green-400 dark:ring-green-400/20",
@color == "blue" &&
"bg-blue-50 text-blue-700 ring-blue-600/20 dark:bg-blue-400/10 dark:text-blue-400 dark:ring-blue-400/20",
@color == "yellow" &&
"bg-yellow-50 text-yellow-800 ring-yellow-600/20 dark:bg-yellow-400/10 dark:text-yellow-500 dark:ring-yellow-400/20",
@color == "orange" &&
"bg-orange-50 text-orange-700 ring-orange-600/20 dark:bg-orange-400/10 dark:text-orange-400 dark:ring-orange-400/20",
@color == "red" &&
"bg-red-50 text-red-700 ring-red-600/10 dark:bg-red-400/10 dark:text-red-400 dark:ring-red-400/20"
]}
title={@label}
>
{@value} dB
</span>
"""
end
@doc """
Translates the errors for a field from a keyword list of errors.
"""

View file

@ -0,0 +1,216 @@
defmodule Towerops.Repo.Migrations.AddWirelessClientReadingsHypertable do
@moduledoc """
Creates wireless_client_readings hypertable for historical wireless client metrics tracking.
Stores time-series snapshots of wireless client data (signal strength, SNR, distance, rates)
collected every 60 seconds during SNMP polling. Enables trend analysis, historical queries,
and proactive alerting for wireless performance issues.
TimescaleDB Configuration:
- 1-day chunks for efficient time-based queries
- Compression after 7 days (90-95% storage reduction)
- 90-day retention on raw data
- Continuous aggregates: hourly (1 year) and daily (5 years) for fast analytics
"""
use Ecto.Migration
@disable_ddl_transaction true
@disable_migration_lock true
def up do
# Create wireless_client_readings table
create table(:wireless_client_readings, primary_key: false) do
add :id, :uuid, primary_key: true
add :checked_at, :utc_datetime, null: false, primary_key: true
# Foreign keys (denormalized for query performance)
add :device_id, references(:snmp_devices, on_delete: :delete_all, type: :uuid), null: false
add :wireless_client_id,
references(:wireless_clients, on_delete: :delete_all, type: :uuid),
null: false
add :organization_id, references(:organizations, on_delete: :delete_all, type: :uuid),
null: false
# Denormalized client identifiers (avoid joins for common queries)
add :mac_address, :string, null: false
add :ip_address, :string
# Wireless metrics
add :signal_strength, :integer
add :snr, :integer
add :distance, :integer
add :tx_rate, :integer
add :rx_rate, :integer
add :uptime_seconds, :integer
# Additional metadata (vendor-specific fields, future expansion)
add :metadata, :map, default: %{}
# Timestamp (no updated_at - immutable time-series)
add :inserted_at, :utc_datetime, null: false
end
# Convert to TimescaleDB hypertable (if available)
if timescaledb_available?() do
execute("""
SELECT create_hypertable(
'wireless_client_readings',
'checked_at',
chunk_time_interval => INTERVAL '1 day',
if_not_exists => true
)
""")
# Create indexes for common query patterns
execute(
"CREATE INDEX IF NOT EXISTS wireless_client_readings_device_id_checked_at_idx ON wireless_client_readings (device_id, checked_at DESC)"
)
execute(
"CREATE INDEX IF NOT EXISTS wireless_client_readings_mac_address_checked_at_idx ON wireless_client_readings (mac_address, checked_at DESC)"
)
execute(
"CREATE INDEX IF NOT EXISTS wireless_client_readings_organization_id_checked_at_idx ON wireless_client_readings (organization_id, checked_at DESC)"
)
execute(
"CREATE INDEX IF NOT EXISTS wireless_client_readings_checked_at_idx ON wireless_client_readings (checked_at DESC)"
)
# Enable compression (segmented by device_id + mac_address for efficient queries)
execute(
"ALTER TABLE wireless_client_readings SET (timescaledb.compress, timescaledb.compress_segmentby = 'device_id,mac_address')"
)
execute(
"SELECT add_compression_policy('wireless_client_readings', INTERVAL '7 days', if_not_exists => true)"
)
# Add retention policy (90 days for raw data)
execute(
"SELECT add_retention_policy('wireless_client_readings', INTERVAL '90 days', if_not_exists => true)"
)
# Create continuous aggregate: hourly rollup
execute("""
CREATE MATERIALIZED VIEW IF NOT EXISTS wireless_client_readings_hourly
WITH (timescaledb.continuous) AS
SELECT
device_id,
mac_address,
wireless_client_id,
organization_id,
time_bucket('1 hour', checked_at) AS bucket,
AVG(signal_strength) AS avg_signal_strength,
MIN(signal_strength) AS min_signal_strength,
MAX(signal_strength) AS max_signal_strength,
AVG(snr) AS avg_snr,
MIN(snr) AS min_snr,
MAX(snr) AS max_snr,
AVG(distance) AS avg_distance,
AVG(tx_rate) AS avg_tx_rate,
AVG(rx_rate) AS avg_rx_rate,
COUNT(*) AS sample_count
FROM wireless_client_readings
GROUP BY device_id, mac_address, wireless_client_id, organization_id, bucket
WITH NO DATA
""")
# Create continuous aggregate: daily rollup
execute("""
CREATE MATERIALIZED VIEW IF NOT EXISTS wireless_client_readings_daily
WITH (timescaledb.continuous) AS
SELECT
device_id,
mac_address,
wireless_client_id,
organization_id,
time_bucket('1 day', checked_at) AS bucket,
AVG(signal_strength) AS avg_signal_strength,
MIN(signal_strength) AS min_signal_strength,
MAX(signal_strength) AS max_signal_strength,
AVG(snr) AS avg_snr,
MIN(snr) AS min_snr,
MAX(snr) AS max_snr,
AVG(distance) AS avg_distance,
AVG(tx_rate) AS avg_tx_rate,
AVG(rx_rate) AS avg_rx_rate,
COUNT(*) AS sample_count
FROM wireless_client_readings
GROUP BY device_id, mac_address, wireless_client_id, organization_id, bucket
WITH NO DATA
""")
# Add refresh policies for continuous aggregates
execute(
"SELECT add_continuous_aggregate_policy('wireless_client_readings_hourly', start_offset => INTERVAL '3 hours', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', if_not_exists => true)"
)
execute(
"SELECT add_continuous_aggregate_policy('wireless_client_readings_daily', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 day', schedule_interval => INTERVAL '1 day', if_not_exists => true)"
)
# Add retention policies for continuous aggregates
execute(
"SELECT add_retention_policy('wireless_client_readings_hourly', INTERVAL '1 year', if_not_exists => true)"
)
execute(
"SELECT add_retention_policy('wireless_client_readings_daily', INTERVAL '5 years', if_not_exists => true)"
)
else
# If TimescaleDB not available (test env), create regular indexes
create index(:wireless_client_readings, [:device_id, :checked_at])
create index(:wireless_client_readings, [:mac_address, :checked_at])
create index(:wireless_client_readings, [:organization_id, :checked_at])
create index(:wireless_client_readings, [:checked_at])
end
end
def down do
if timescaledb_available?() do
# Remove refresh policies
execute(
"SELECT remove_continuous_aggregate_policy('wireless_client_readings_hourly', if_exists => true)"
)
execute(
"SELECT remove_continuous_aggregate_policy('wireless_client_readings_daily', if_exists => true)"
)
# Remove retention policies from continuous aggregates
execute(
"SELECT remove_retention_policy('wireless_client_readings_hourly', if_exists => true)"
)
execute(
"SELECT remove_retention_policy('wireless_client_readings_daily', if_exists => true)"
)
# Drop continuous aggregates
execute("DROP MATERIALIZED VIEW IF EXISTS wireless_client_readings_hourly CASCADE")
execute("DROP MATERIALIZED VIEW IF EXISTS wireless_client_readings_daily CASCADE")
# Remove retention policy
execute("SELECT remove_retention_policy('wireless_client_readings', if_exists => true)")
# Remove compression policy
execute("SELECT remove_compression_policy('wireless_client_readings', if_exists => true)")
# Drop hypertable (CASCADE removes all chunks)
execute("DROP TABLE IF EXISTS wireless_client_readings CASCADE")
else
drop table(:wireless_client_readings)
end
end
defp timescaledb_available? do
case repo().query("SELECT 1 FROM pg_available_extensions WHERE name = 'timescaledb'") do
{:ok, %{num_rows: n}} when n > 0 -> true
_ -> false
end
end
end