diff --git a/lib/towerops/snmp.ex b/lib/towerops/snmp.ex index 24309eda..51ee0b21 100644 --- a/lib/towerops/snmp.ex +++ b/lib/towerops/snmp.ex @@ -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 diff --git a/lib/towerops/snmp/wireless_client_reading.ex b/lib/towerops/snmp/wireless_client_reading.ex new file mode 100644 index 00000000..f4a64f99 --- /dev/null +++ b/lib/towerops/snmp/wireless_client_reading.ex @@ -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 diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index 74f06f05..e66e40d7 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -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 || [] diff --git a/lib/towerops_web/components/core_components.ex b/lib/towerops_web/components/core_components.ex index da064ac8..cfeb33f5 100644 --- a/lib/towerops_web/components/core_components.ex +++ b/lib/towerops_web/components/core_components.ex @@ -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""" + + — + + """ + 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""" + + {@value} dBm + + """ + 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""" + + — + + """ + 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""" + + {@value} dB + + """ + end + @doc """ Translates the errors for a field from a keyword list of errors. """ diff --git a/priv/repo/migrations/20260309223347_add_wireless_client_readings_hypertable.exs b/priv/repo/migrations/20260309223347_add_wireless_client_readings_hypertable.exs new file mode 100644 index 00000000..27aaabe0 --- /dev/null +++ b/priv/repo/migrations/20260309223347_add_wireless_client_readings_hypertable.exs @@ -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