From 9bb43f6007bda6a517618576e7c1c9332c95b3f3 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 30 Apr 2026 14:04:38 -0500 Subject: [PATCH] refactor(snmp): extract metadata submodules (vlans, ips, processors, storage, mempools, sensors, interfaces, neighbors) Splits cohesive sub-domains out of lib/towerops/snmp.ex into dedicated submodules under lib/towerops/snmp/. Public API preserved via defdelegate in Towerops.Snmp. - Towerops.Snmp.Vlans, IpAddresses, Processors, StorageQueries - Towerops.Snmp.Mempools, Sensors, Interfaces, Neighbors Neighbors uses Towerops.Devices.DeviceQuery.for_organization/1 to remove inline org_id filters. snmp.ex shrinks from 2742 to 2159 lines. --- lib/towerops/snmp.ex | 739 +++------------------------ lib/towerops/snmp/interfaces.ex | 81 +++ lib/towerops/snmp/ip_addresses.ex | 35 ++ lib/towerops/snmp/mempools.ex | 116 +++++ lib/towerops/snmp/neighbors.ex | 182 +++++++ lib/towerops/snmp/processors.ex | 65 +++ lib/towerops/snmp/sensors.ex | 85 +++ lib/towerops/snmp/storage_queries.ex | 126 +++++ lib/towerops/snmp/vlans.ex | 29 ++ 9 files changed, 790 insertions(+), 668 deletions(-) create mode 100644 lib/towerops/snmp/interfaces.ex create mode 100644 lib/towerops/snmp/ip_addresses.ex create mode 100644 lib/towerops/snmp/mempools.ex create mode 100644 lib/towerops/snmp/neighbors.ex create mode 100644 lib/towerops/snmp/processors.ex create mode 100644 lib/towerops/snmp/sensors.ex create mode 100644 lib/towerops/snmp/storage_queries.ex create mode 100644 lib/towerops/snmp/vlans.ex diff --git a/lib/towerops/snmp.ex b/lib/towerops/snmp.ex index b82e5e8e..443a90f2 100644 --- a/lib/towerops/snmp.ex +++ b/lib/towerops/snmp.ex @@ -16,23 +16,21 @@ defmodule Towerops.Snmp do alias Towerops.Snmp.EntityPhysical alias Towerops.Snmp.EntityPhysicalReading alias Towerops.Snmp.Interface - alias Towerops.Snmp.IpAddress + alias Towerops.Snmp.Interfaces + alias Towerops.Snmp.IpAddresses alias Towerops.Snmp.MacAddress - alias Towerops.Snmp.Mempool - alias Towerops.Snmp.MempoolReading + alias Towerops.Snmp.Mempools alias Towerops.Snmp.Monitoring alias Towerops.Snmp.Neighbor + alias Towerops.Snmp.Neighbors alias Towerops.Snmp.PrinterSupply - alias Towerops.Snmp.Processor - alias Towerops.Snmp.ProcessorReading + alias Towerops.Snmp.Processors alias Towerops.Snmp.Queries - alias Towerops.Snmp.Sensor - alias Towerops.Snmp.StateSensor - alias Towerops.Snmp.Storage - alias Towerops.Snmp.StorageReading + alias Towerops.Snmp.Sensors + alias Towerops.Snmp.StorageQueries alias Towerops.Snmp.Transceiver alias Towerops.Snmp.TransceiverReading - alias Towerops.Snmp.Vlan + alias Towerops.Snmp.Vlans alias Towerops.Snmp.WirelessClient alias Towerops.Snmp.WirelessClientReading @@ -288,128 +286,27 @@ defmodule Towerops.Snmp do |> Repo.one() end - # Interface queries + ## Interface metadata delegates (implementation in `Towerops.Snmp.Interfaces`). - @doc """ - Lists all interfaces for a device. - """ - def list_interfaces(device_id) do - Interface - |> where([i], i.snmp_device_id == ^device_id) - |> order_by([i], i.if_index) - |> Repo.all() - end + defdelegate list_interfaces(device_id), to: Interfaces + defdelegate list_monitored_interfaces(device_id), to: Interfaces + defdelegate get_interface(interface_id), to: Interfaces + defdelegate update_interface(interface, attrs), to: Interfaces + defdelegate set_manual_capacity(interface_id, capacity_bps), to: Interfaces + defdelegate clear_manual_capacity(interface_id), to: Interfaces - @doc """ - Lists only monitored interfaces for a device. - """ - def list_monitored_interfaces(device_id) do - Interface - |> where([i], i.snmp_device_id == ^device_id and i.monitored == true) - |> order_by([i], i.if_index) - |> Repo.all() - end + ## Sensor metadata delegates (implementation in `Towerops.Snmp.Sensors`). - @doc """ - Gets a specific interface. - """ - def get_interface(interface_id) do - Repo.get(Interface, interface_id) - end + defdelegate list_sensors(device_id), to: Sensors + defdelegate list_monitored_sensors(device_id), to: Sensors + defdelegate list_sensors_by_type(device_id), to: Sensors + defdelegate get_sensor(sensor_id), to: Sensors + defdelegate update_sensor(sensor, attrs), to: Sensors + defdelegate update_state_sensor(state_sensor, attrs), to: Sensors + defdelegate list_state_sensors(device_id), to: Sensors - @doc """ - Updates an interface. - """ - def update_interface(%Interface{} = interface, attrs) do - interface - |> Interface.changeset(attrs) - |> Repo.update() - end - - # Sensor queries - - @doc """ - Lists all sensors for a device. - """ - def list_sensors(device_id) do - Sensor - |> where([s], s.snmp_device_id == ^device_id) - |> order_by([s], [s.sensor_type, s.sensor_index]) - |> Repo.all() - end - - @doc """ - Lists only monitored sensors for a device. - """ - def list_monitored_sensors(device_id) do - Sensor - |> where([s], s.snmp_device_id == ^device_id and s.monitored == true) - |> order_by([s], [s.sensor_type, s.sensor_index]) - |> Repo.all() - end - - @doc """ - Lists sensors grouped by type. - """ - def list_sensors_by_type(device_id) do - sensors = - Sensor - |> where([s], s.snmp_device_id == ^device_id) - |> order_by([s], [s.sensor_type, s.sensor_index]) - |> Repo.all() - - Enum.group_by(sensors, & &1.sensor_type) - end - - @doc """ - Gets a specific sensor. - """ - def get_sensor(sensor_id) do - Repo.get(Sensor, sensor_id) - end - - @doc """ - Updates a sensor. - """ - def update_sensor(%Sensor{} = sensor, attrs) do - sensor - |> Sensor.changeset(attrs) - |> Repo.update() - end - - # State sensor queries - - @doc """ - Updates a state sensor. - """ - def update_state_sensor(%StateSensor{} = state_sensor, attrs) do - state_sensor - |> StateSensor.changeset(attrs) - |> Repo.update() - end - - @doc """ - Lists all state sensors for a device. - """ - def list_state_sensors(device_id) do - StateSensor - |> where([s], s.snmp_device_id == ^device_id) - |> order_by([s], s.sensor_index) - |> Repo.all() - end - - # Sensor readings queries - - @doc """ - Gets recent sensor readings for a sensor. - - ## Options - - - `:limit` - Maximum number of readings to return (default: 100) - - `:since` - Only return readings after this datetime - """ - - ## Time-series query delegates (implementation in `Towerops.Snmp.Queries`). + ## Sensor / interface time-series query delegates + ## (implementation in `Towerops.Snmp.Queries`). defdelegate get_sensor_readings(sensor_id, opts \\ []), to: Queries defdelegate get_latest_sensor_reading(sensor_id), to: Queries @@ -418,36 +315,6 @@ defmodule Towerops.Snmp do defdelegate get_latest_interface_stat(interface_id), to: Queries defdelegate get_latest_interface_stats_batch(interface_ids), to: Queries - @doc """ - Sets a manual capacity override on an interface. - """ - def set_manual_capacity(interface_id, capacity_bps) do - case Repo.get(Interface, interface_id) do - nil -> - {:error, :not_found} - - interface -> - interface - |> Interface.changeset(%{configured_capacity_bps: capacity_bps, capacity_source: "manual"}) - |> Repo.update() - end - end - - @doc """ - Clears the capacity override on an interface, resetting both fields to nil. - """ - def clear_manual_capacity(interface_id) do - case Repo.get(Interface, interface_id) do - nil -> - {:error, :not_found} - - interface -> - interface - |> Ecto.Changeset.change(%{configured_capacity_bps: nil, capacity_source: nil}) - |> Repo.update() - end - end - ## Time-series write delegates (implementation in `Towerops.Snmp.Monitoring`). defdelegate create_sensor_reading(attrs), to: Monitoring @@ -455,174 +322,14 @@ defmodule Towerops.Snmp do defdelegate create_interface_stat(attrs), to: Monitoring defdelegate create_interface_stats_batch(entries), to: Monitoring - # Neighbor queries + ## Neighbor delegates (implementation in `Towerops.Snmp.Neighbors`). - @doc """ - Lists all neighbors for a piece of device. - """ - def list_neighbors(device_id) do - Neighbor - |> where([n], n.device_id == ^device_id) - |> preload(:interface) - |> order_by([n], [n.protocol, n.remote_system_name]) - |> Repo.all() - end - - @doc """ - Lists all neighbors for device with enriched data linking to known device. - Adds a `matched_equipment` field to each neighbor if we can identify the remote device. - """ - def list_neighbors_with_equipment(device_id) do - neighbors = list_neighbors(device_id) - - # device to search within the same org - device = Towerops.Devices.get_device!(device_id) - - Enum.map(neighbors, fn neighbor -> - matched_device = find_matching_equipment(neighbor, device.organization_id) - Map.put(neighbor, :matched_device, matched_device) - end) - end - - # device in the organization - defp find_matching_equipment(neighbor, organization_id) do - # Try matching strategies in order: chassis ID, port ID, system name - try_match_by_chassis(neighbor, organization_id) || - try_match_by_port(neighbor, organization_id) || - try_match_by_name(neighbor, organization_id) - end - - defp try_match_by_chassis(neighbor, organization_id) do - if neighbor.remote_chassis_id && mac_address?(neighbor.remote_chassis_id) do - find_device_by_mac(neighbor.remote_chassis_id, organization_id) - end - end - - defp try_match_by_port(neighbor, organization_id) do - if neighbor.remote_port_id && mac_address?(neighbor.remote_port_id) do - find_device_by_mac(neighbor.remote_port_id, organization_id) - end - end - - defp try_match_by_name(neighbor, organization_id) do - if neighbor.remote_system_name do - find_device_by_name(neighbor.remote_system_name, organization_id) - end - end - - # Check if a string looks like a MAC address - defp mac_address?(str) when is_binary(str) do - # Match patterns like "AA:BB:CC:DD:EE:FF" or "aa-bb-cc-dd-ee-ff" - String.match?(str, ~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/) - end - - defp mac_address?(_), do: false - - defp find_device_by_mac(mac_address, organization_id) do - # MAC address might be in format "aa:bb:cc:dd:ee:ff", "aa-bb-cc-dd-ee-ff", or "aabbccddeeff" - # Normalize to just the hex digits for comparison - normalized_mac = mac_address |> String.downcase() |> String.replace(~r/[:-]/, "") - - Repo.one( - from(i in Interface, - join: d in Device, - on: i.snmp_device_id == d.id, - join: e in DeviceSchema, - on: d.device_id == e.id, - join: s in assoc(e, :site), - where: s.organization_id == ^organization_id, - where: - fragment("REPLACE(REPLACE(LOWER(?), ':', ''), '-', '')", i.if_phys_address) == - ^normalized_mac, - select: e, - limit: 1 - ) - ) - end - - defp find_device_by_name(system_name, organization_id) do - # Try exact match first, then partial match - normalized_name = String.downcase(system_name) - - from(e in DeviceSchema, - join: s in assoc(e, :site), - where: s.organization_id == ^organization_id, - where: fragment("LOWER(?)", e.name) == ^normalized_name, - select: e, - limit: 1 - ) - |> Repo.one() - |> case do - nil -> - # Try partial match as fallback - Repo.one( - from(e in DeviceSchema, - join: s in assoc(e, :site), - where: s.organization_id == ^organization_id, - where: fragment("LOWER(?) LIKE ?", e.name, ^"%#{Towerops.QueryHelpers.sanitize_like(normalized_name)}%"), - select: e, - limit: 1 - ) - ) - - device -> - device - end - end - - @doc """ - Gets a specific neighbor. - """ - def get_neighbor(neighbor_id) do - Repo.get(Neighbor, neighbor_id) - end - - @doc """ - Creates or updates a neighbor record. - - Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple - pollers try to upsert the same neighbor simultaneously. - """ - def upsert_neighbor(attrs) do - %Neighbor{} - |> Neighbor.changeset(attrs) - |> Repo.insert( - on_conflict: {:replace_all_except, [:id, :inserted_at]}, - conflict_target: [:interface_id, :remote_chassis_id, :protocol], - returning: true - ) - end - - @doc """ - Deletes stale neighbors that haven't been seen since the given datetime. - """ - def delete_stale_neighbors(device_id, cutoff_datetime) do - Neighbor - |> where([n], n.device_id == ^device_id) - |> where([n], n.last_discovered_at < ^cutoff_datetime) - |> Repo.delete_all() - end - - @doc """ - Atomically deletes stale neighbors and upserts new ones in a single transaction. - Prevents race conditions where concurrent delete/upsert could remove fresh data. - """ - def delete_stale_and_upsert_neighbors(device_id, neighbors, cutoff) do - Repo.transaction(fn -> - delete_stale_neighbors(device_id, cutoff) - Enum.each(neighbors, &upsert_neighbor_safe(device_id, &1)) - end) - end - - defp upsert_neighbor_safe(device_id, neighbor_data) do - case upsert_neighbor(neighbor_data) do - {:ok, _neighbor} -> - :ok - - {:error, changeset} -> - Logger.error("Failed to upsert neighbor for device #{device_id}: #{inspect(changeset.errors)}") - end - end + defdelegate list_neighbors(device_id), to: Neighbors + defdelegate list_neighbors_with_equipment(device_id), to: Neighbors + defdelegate get_neighbor(neighbor_id), to: Neighbors + defdelegate upsert_neighbor(attrs), to: Neighbors + defdelegate delete_stale_neighbors(device_id, cutoff_datetime), to: Neighbors + defdelegate delete_stale_and_upsert_neighbors(device_id, neighbors, cutoff), to: Neighbors @doc """ Lists all discovered devices across an organization that haven't been added yet. @@ -943,6 +650,13 @@ defmodule Towerops.Snmp do ) end + # Check if a string looks like a MAC address + defp mac_address?(str) when is_binary(str) do + String.match?(str, ~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/) + end + + defp mac_address?(_), do: false + # Extract identifiers from ARP entry defp extract_arp_identifiers(arp) do Enum.reject( @@ -1354,250 +1068,49 @@ defmodule Towerops.Snmp do defp safe_protocol_to_atom("cdp"), do: :cdp defp safe_protocol_to_atom(_unknown), do: :unknown - # VLAN queries + ## VLAN delegates (implementation in `Towerops.Snmp.Vlans`). - @doc """ - Lists all VLANs for a device. - """ - def list_vlans(snmp_device_id) do - Vlan - |> where([v], v.snmp_device_id == ^snmp_device_id) - |> order_by([v], v.vlan_id) - |> Repo.all() - end + defdelegate list_vlans(snmp_device_id), to: Vlans + defdelegate get_vlan(vlan_id), to: Vlans - @doc """ - Gets a specific VLAN. - """ - def get_vlan(vlan_id) do - Repo.get(Vlan, vlan_id) - end + ## IP address delegates (implementation in `Towerops.Snmp.IpAddresses`). - # IP Address queries + defdelegate list_ip_addresses(snmp_device_id), to: IpAddresses + defdelegate list_interface_ip_addresses(interface_id), to: IpAddresses - @doc """ - Lists all IP addresses for a device (across all its interfaces). - """ - def list_ip_addresses(snmp_device_id) do - IpAddress - |> join(:inner, [ip], i in Interface, on: ip.snmp_interface_id == i.id) - |> where([ip, i], i.snmp_device_id == ^snmp_device_id) - |> preload(:snmp_interface) - |> order_by([ip], [ip.ip_type, ip.ip_address]) - |> Repo.all() - end - - @doc """ - Lists IP addresses for a specific interface. - """ - def list_interface_ip_addresses(interface_id) do - IpAddress - |> where([ip], ip.snmp_interface_id == ^interface_id) - |> order_by([ip], [ip.ip_type, ip.ip_address]) - |> Repo.all() - end - - # Processor queries - - @doc """ - Lists all processors for a device. - """ - def list_processors(snmp_device_id) do - Processor - |> where([p], p.snmp_device_id == ^snmp_device_id) - |> order_by([p], p.processor_index) - |> Repo.all() - end - - @doc """ - Gets a specific processor. - """ - def get_processor(processor_id) do - Repo.get(Processor, processor_id) - end - - @doc """ - Updates a processor. - """ - def update_processor(processor, attrs) do - processor - |> Processor.changeset(attrs) - |> Repo.update() - end - - # Processor reading queries - - @doc """ - Gets recent processor readings for a processor. - - ## Options - - - `:limit` - Maximum number of readings to return (default: 100) - - `:since` - Only return readings after this datetime - """ - def get_processor_readings(processor_id, opts \\ []) do - limit = Keyword.get(opts, :limit, 100) - since = Keyword.get(opts, :since) - - query = - ProcessorReading - |> where([r], r.processor_id == ^processor_id) - |> order_by([r], desc: r.checked_at) - |> limit(^limit) - - query = - if since do - where(query, [r], r.checked_at >= ^since) - else - query - end - - Repo.all(query) - end + ## Processor metadata + reading delegates + ## (queries in `Towerops.Snmp.Processors`, reading inserts in + ## `Towerops.Snmp.Monitoring`). + defdelegate list_processors(snmp_device_id), to: Processors + defdelegate get_processor(processor_id), to: Processors + defdelegate update_processor(processor, attrs), to: Processors + defdelegate get_processor_readings(processor_id, opts \\ []), to: Processors defdelegate create_processor_reading(attrs), to: Monitoring defdelegate create_processor_readings_batch(entries), to: Monitoring - # Storage queries + ## Storage metadata + reading delegates + ## (implementation in `Towerops.Snmp.StorageQueries`). - @doc """ - Lists all storage entries for a device. - """ - def list_storage(snmp_device_id) do - Storage - |> where([s], s.snmp_device_id == ^snmp_device_id) - |> order_by([s], s.storage_index) - |> Repo.all() - end + defdelegate list_storage(snmp_device_id), to: StorageQueries + defdelegate get_storage(storage_id), to: StorageQueries + defdelegate update_storage(storage, attrs), to: StorageQueries + defdelegate get_storage_readings(storage_id, opts \\ []), to: StorageQueries + defdelegate get_latest_storage_reading(storage_id), to: StorageQueries + defdelegate get_latest_storage_readings_batch(storage_ids), to: StorageQueries + defdelegate create_storage_reading(attrs), to: StorageQueries + defdelegate create_storage_readings_batch(entries), to: StorageQueries - @doc """ - Gets a specific storage entry. - """ - def get_storage(storage_id) do - Repo.get(Storage, storage_id) - end + ## Mempool metadata + reading delegates + ## (implementation in `Towerops.Snmp.Mempools`). - @doc """ - Updates a storage entry. - """ - def update_storage(storage, attrs) do - storage - |> Storage.changeset(attrs) - |> Repo.update() - end - - # Storage reading queries - - @doc """ - Gets recent storage readings for a storage entry. - - ## Options - - - `:limit` - Maximum number of readings to return (default: 100) - - `:since` - Only return readings after this datetime - """ - def get_storage_readings(storage_id, opts \\ []) do - limit = Keyword.get(opts, :limit, 100) - since = Keyword.get(opts, :since) - - query = - StorageReading - |> where([r], r.storage_id == ^storage_id) - |> order_by([r], desc: r.checked_at) - |> limit(^limit) - - query = - if since do - where(query, [r], r.checked_at >= ^since) - else - query - end - - Repo.all(query) - end - - @doc """ - Gets the latest storage reading for a storage entry. - """ - def get_latest_storage_reading(storage_id) do - StorageReading - |> where([r], r.storage_id == ^storage_id) - |> order_by([r], desc: r.checked_at) - |> limit(1) - |> Repo.one() - end - - @doc """ - Gets the latest storage readings for multiple storage entries in a single query. - - Returns a map of %{storage_id => reading} for efficient batch loading. - Storage entries without readings will not be present in the map. - - ## Examples - - iex> get_latest_storage_readings_batch([storage_id1, storage_id2]) - %{storage_id1 => %StorageReading{}, storage_id2 => %StorageReading{}} - """ - def get_latest_storage_readings_batch(storage_ids) when is_list(storage_ids) do - if Enum.empty?(storage_ids) do - %{} - else - # Use DISTINCT ON to get only the latest reading per storage - query = - from(r in StorageReading, - where: r.storage_id in ^storage_ids, - distinct: r.storage_id, - order_by: [asc: r.storage_id, desc: r.checked_at] - ) - - query - |> Repo.all() - |> Map.new(fn reading -> {reading.storage_id, reading} end) - end - end - - @doc """ - Records a new storage reading. - """ - def create_storage_reading(attrs) do - %StorageReading{} - |> StorageReading.changeset(attrs) - |> Repo.insert() - end - - @doc """ - Batch inserts multiple storage readings using `Repo.insert_all/3`. - - Accepts a list of attribute maps with the same keys as `create_storage_reading/1`. - Generates UUIDs and timestamps automatically. Returns `{count, nil}`. - """ - @spec create_storage_readings_batch([map()]) :: {non_neg_integer(), nil} - def create_storage_readings_batch([]), do: {0, nil} - - def create_storage_readings_batch(entries) when is_list(entries) do - now = Towerops.Time.now() - - rows = - Enum.map(entries, fn entry -> - entry - |> Map.put(:id, Ecto.UUID.generate()) - |> Map.put(:inserted_at, now) - end) - - Repo.insert_all(StorageReading, rows) - end - - # Mempool queries - - @doc """ - Lists all memory pools for a device. - """ - def list_mempools(snmp_device_id) do - Mempool - |> where([m], m.snmp_device_id == ^snmp_device_id) - |> order_by([m], m.mempool_index) - |> Repo.all() - end + defdelegate list_mempools(snmp_device_id), to: Mempools + defdelegate update_mempool(mempool, attrs), to: Mempools + defdelegate get_mempool_readings(mempool_id, opts \\ []), to: Mempools + defdelegate get_latest_mempool_reading(mempool_id), to: Mempools + defdelegate get_latest_mempool_readings_batch(mempool_ids), to: Mempools + defdelegate create_mempool_reading(attrs), to: Mempools + defdelegate create_mempool_readings_batch(entries), to: Mempools @doc """ Lists all entity physical components for a given SNMP device. @@ -1858,116 +1371,6 @@ defmodule Towerops.Snmp do Repo.insert_all(EntityPhysicalReading, rows) end - @doc """ - Updates a mempool with the given attributes. - """ - def update_mempool(mempool, attrs) do - mempool - |> Mempool.changeset(attrs) - |> Repo.update() - end - - # Mempool reading queries - - @doc """ - Gets recent mempool readings for a memory pool. - - ## Options - - - `:limit` - Maximum number of readings to return (default: 100) - - `:since` - Only return readings after this datetime - """ - def get_mempool_readings(mempool_id, opts \\ []) do - limit = Keyword.get(opts, :limit, 100) - since = Keyword.get(opts, :since) - - query = - MempoolReading - |> where([r], r.mempool_id == ^mempool_id) - |> order_by([r], desc: r.checked_at) - |> limit(^limit) - - query = - if since do - where(query, [r], r.checked_at >= ^since) - else - query - end - - Repo.all(query) - end - - @doc """ - Gets the latest mempool reading for a memory pool. - """ - def get_latest_mempool_reading(mempool_id) do - MempoolReading - |> where([r], r.mempool_id == ^mempool_id) - |> order_by([r], desc: r.checked_at) - |> limit(1) - |> Repo.one() - end - - @doc """ - Gets the latest mempool readings for multiple memory pools in a single query. - - Returns a map of %{mempool_id => reading} for efficient batch loading. - Mempools without readings will not be present in the map. - - ## Examples - - iex> get_latest_mempool_readings_batch([mempool_id1, mempool_id2]) - %{mempool_id1 => %MempoolReading{}, mempool_id2 => %MempoolReading{}} - """ - def get_latest_mempool_readings_batch(mempool_ids) when is_list(mempool_ids) do - if Enum.empty?(mempool_ids) do - %{} - else - # Use DISTINCT ON to get only the latest reading per mempool - query = - from(r in MempoolReading, - where: r.mempool_id in ^mempool_ids, - distinct: r.mempool_id, - order_by: [asc: r.mempool_id, desc: r.checked_at] - ) - - query - |> Repo.all() - |> Map.new(fn reading -> {reading.mempool_id, reading} end) - end - end - - @doc """ - Records a new mempool reading. - """ - def create_mempool_reading(attrs) do - %MempoolReading{} - |> MempoolReading.changeset(attrs) - |> Repo.insert() - end - - @doc """ - Batch inserts multiple mempool readings using `Repo.insert_all/3`. - - Accepts a list of attribute maps with the same keys as `create_mempool_reading/1`. - Generates UUIDs and timestamps automatically. Returns `{count, nil}`. - """ - @spec create_mempool_readings_batch([map()]) :: {non_neg_integer(), nil} - def create_mempool_readings_batch([]), do: {0, nil} - - def create_mempool_readings_batch(entries) when is_list(entries) do - now = Towerops.Time.now() - - rows = - Enum.map(entries, fn entry -> - entry - |> Map.put(:id, Ecto.UUID.generate()) - |> Map.put(:inserted_at, now) - end) - - Repo.insert_all(MempoolReading, rows) - end - # ARP Entry queries @doc """ diff --git a/lib/towerops/snmp/interfaces.ex b/lib/towerops/snmp/interfaces.ex new file mode 100644 index 00000000..fa208b65 --- /dev/null +++ b/lib/towerops/snmp/interfaces.ex @@ -0,0 +1,81 @@ +defmodule Towerops.Snmp.Interfaces do + @moduledoc """ + Read-side queries and mutations for interfaces discovered on SNMP devices. + + Time-series interface stat queries live in `Towerops.Snmp.Queries`. This + module covers interface metadata and capacity overrides. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Interface + + @doc """ + Lists all interfaces for a device. + """ + def list_interfaces(device_id) do + Interface + |> where([i], i.snmp_device_id == ^device_id) + |> order_by([i], i.if_index) + |> Repo.all() + end + + @doc """ + Lists only monitored interfaces for a device. + """ + def list_monitored_interfaces(device_id) do + Interface + |> where([i], i.snmp_device_id == ^device_id and i.monitored == true) + |> order_by([i], i.if_index) + |> Repo.all() + end + + @doc """ + Gets a specific interface. + """ + def get_interface(interface_id) do + Repo.get(Interface, interface_id) + end + + @doc """ + Updates an interface. + """ + def update_interface(%Interface{} = interface, attrs) do + interface + |> Interface.changeset(attrs) + |> Repo.update() + end + + @doc """ + Sets a manual capacity override on an interface. + """ + def set_manual_capacity(interface_id, capacity_bps) do + case Repo.get(Interface, interface_id) do + nil -> + {:error, :not_found} + + interface -> + interface + |> Interface.changeset(%{configured_capacity_bps: capacity_bps, capacity_source: "manual"}) + |> Repo.update() + end + end + + @doc """ + Clears the capacity override on an interface, resetting both fields to nil. + """ + def clear_manual_capacity(interface_id) do + case Repo.get(Interface, interface_id) do + nil -> + {:error, :not_found} + + interface -> + interface + |> Ecto.Changeset.change(%{configured_capacity_bps: nil, capacity_source: nil}) + |> Repo.update() + end + end +end diff --git a/lib/towerops/snmp/ip_addresses.ex b/lib/towerops/snmp/ip_addresses.ex new file mode 100644 index 00000000..398982a3 --- /dev/null +++ b/lib/towerops/snmp/ip_addresses.ex @@ -0,0 +1,35 @@ +defmodule Towerops.Snmp.IpAddresses do + @moduledoc """ + Read-side queries for IP addresses associated with SNMP-discovered interfaces. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Interface + alias Towerops.Snmp.IpAddress + + @doc """ + Lists all IP addresses for a device (across all its interfaces). + """ + def list_ip_addresses(snmp_device_id) do + IpAddress + |> join(:inner, [ip], i in Interface, on: ip.snmp_interface_id == i.id) + |> where([ip, i], i.snmp_device_id == ^snmp_device_id) + |> preload(:snmp_interface) + |> order_by([ip], [ip.ip_type, ip.ip_address]) + |> Repo.all() + end + + @doc """ + Lists IP addresses for a specific interface. + """ + def list_interface_ip_addresses(interface_id) do + IpAddress + |> where([ip], ip.snmp_interface_id == ^interface_id) + |> order_by([ip], [ip.ip_type, ip.ip_address]) + |> Repo.all() + end +end diff --git a/lib/towerops/snmp/mempools.ex b/lib/towerops/snmp/mempools.ex new file mode 100644 index 00000000..64491b94 --- /dev/null +++ b/lib/towerops/snmp/mempools.ex @@ -0,0 +1,116 @@ +defmodule Towerops.Snmp.Mempools do + @moduledoc """ + Read-side queries and mutations for memory pools discovered on SNMP devices, + plus persistence for mempool time-series readings. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Mempool + alias Towerops.Snmp.MempoolReading + + @doc """ + Lists all memory pools for a device. + """ + def list_mempools(snmp_device_id) do + Mempool + |> where([m], m.snmp_device_id == ^snmp_device_id) + |> order_by([m], m.mempool_index) + |> Repo.all() + end + + @doc """ + Updates a mempool with the given attributes. + """ + def update_mempool(mempool, attrs) do + mempool + |> Mempool.changeset(attrs) + |> Repo.update() + end + + @doc """ + Gets recent mempool readings for a memory pool. + + ## Options + + - `:limit` - Maximum number of readings to return (default: 100) + - `:since` - Only return readings after this datetime + """ + def get_mempool_readings(mempool_id, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + since = Keyword.get(opts, :since) + + MempoolReading + |> where([r], r.mempool_id == ^mempool_id) + |> order_by([r], desc: r.checked_at) + |> limit(^limit) + |> maybe_filter_since(since) + |> Repo.all() + end + + defp maybe_filter_since(query, nil), do: query + defp maybe_filter_since(query, since), do: where(query, [r], r.checked_at >= ^since) + + @doc """ + Gets the latest mempool reading for a memory pool. + """ + def get_latest_mempool_reading(mempool_id) do + MempoolReading + |> where([r], r.mempool_id == ^mempool_id) + |> order_by([r], desc: r.checked_at) + |> limit(1) + |> Repo.one() + end + + @doc """ + Gets the latest mempool readings for multiple memory pools in a single query. + + Returns a map of %{mempool_id => reading} for efficient batch loading. + Mempools without readings will not be present in the map. + """ + def get_latest_mempool_readings_batch([]), do: %{} + + def get_latest_mempool_readings_batch(mempool_ids) when is_list(mempool_ids) do + from(r in MempoolReading, + where: r.mempool_id in ^mempool_ids, + distinct: r.mempool_id, + order_by: [asc: r.mempool_id, desc: r.checked_at] + ) + |> Repo.all() + |> Map.new(fn reading -> {reading.mempool_id, reading} end) + end + + @doc """ + Records a new mempool reading. + """ + def create_mempool_reading(attrs) do + %MempoolReading{} + |> MempoolReading.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Batch inserts multiple mempool readings using `Repo.insert_all/3`. + + Accepts a list of attribute maps with the same keys as `create_mempool_reading/1`. + Generates UUIDs and timestamps automatically. Returns `{count, nil}`. + """ + @spec create_mempool_readings_batch([map()]) :: {non_neg_integer(), nil} + def create_mempool_readings_batch([]), do: {0, nil} + + def create_mempool_readings_batch(entries) when is_list(entries) do + now = Towerops.Time.now() + + rows = + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + end) + + Repo.insert_all(MempoolReading, rows) + end +end diff --git a/lib/towerops/snmp/neighbors.ex b/lib/towerops/snmp/neighbors.ex new file mode 100644 index 00000000..b452998a --- /dev/null +++ b/lib/towerops/snmp/neighbors.ex @@ -0,0 +1,182 @@ +defmodule Towerops.Snmp.Neighbors do + @moduledoc """ + Read-side queries, upserts, and stale cleanup for LLDP/CDP neighbor records + discovered on SNMP devices. + + Includes matching helpers that try to identify the remote device on the + other end of a neighbor entry by chassis MAC, port MAC, or system name — + scoped to the same organization. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Devices.DeviceQuery + alias Towerops.Repo + alias Towerops.Snmp.Device + alias Towerops.Snmp.Interface + alias Towerops.Snmp.Neighbor + + require Logger + + @doc """ + Lists all neighbors for a piece of device. + """ + def list_neighbors(device_id) do + Neighbor + |> where([n], n.device_id == ^device_id) + |> preload(:interface) + |> order_by([n], [n.protocol, n.remote_system_name]) + |> Repo.all() + end + + @doc """ + Lists all neighbors for device with enriched data linking to known device. + Adds a `matched_equipment` field to each neighbor if we can identify the remote device. + """ + def list_neighbors_with_equipment(device_id) do + neighbors = list_neighbors(device_id) + + # device to search within the same org + device = Towerops.Devices.get_device!(device_id) + + Enum.map(neighbors, fn neighbor -> + matched_device = find_matching_equipment(neighbor, device.organization_id) + Map.put(neighbor, :matched_device, matched_device) + end) + end + + @doc """ + Gets a specific neighbor. + """ + def get_neighbor(neighbor_id) do + Repo.get(Neighbor, neighbor_id) + end + + @doc """ + Creates or updates a neighbor record. + + Uses atomic PostgreSQL ON CONFLICT to prevent race conditions when multiple + pollers try to upsert the same neighbor simultaneously. + """ + def upsert_neighbor(attrs) do + %Neighbor{} + |> Neighbor.changeset(attrs) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :inserted_at]}, + conflict_target: [:interface_id, :remote_chassis_id, :protocol], + returning: true + ) + end + + @doc """ + Deletes stale neighbors that haven't been seen since the given datetime. + """ + def delete_stale_neighbors(device_id, cutoff_datetime) do + Neighbor + |> where([n], n.device_id == ^device_id) + |> where([n], n.last_discovered_at < ^cutoff_datetime) + |> Repo.delete_all() + end + + @doc """ + Atomically deletes stale neighbors and upserts new ones in a single transaction. + Prevents race conditions where concurrent delete/upsert could remove fresh data. + """ + def delete_stale_and_upsert_neighbors(device_id, neighbors, cutoff) do + Repo.transaction(fn -> + delete_stale_neighbors(device_id, cutoff) + Enum.each(neighbors, &upsert_neighbor_safe(device_id, &1)) + end) + end + + defp upsert_neighbor_safe(device_id, neighbor_data) do + case upsert_neighbor(neighbor_data) do + {:ok, _neighbor} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to upsert neighbor for device #{device_id}: #{inspect(changeset.errors)}") + end + end + + # device in the organization + defp find_matching_equipment(neighbor, organization_id) do + # Try matching strategies in order: chassis ID, port ID, system name + try_match_by_chassis(neighbor, organization_id) || + try_match_by_port(neighbor, organization_id) || + try_match_by_name(neighbor, organization_id) + end + + defp try_match_by_chassis(neighbor, organization_id) do + if neighbor.remote_chassis_id && mac_address?(neighbor.remote_chassis_id) do + find_device_by_mac(neighbor.remote_chassis_id, organization_id) + end + end + + defp try_match_by_port(neighbor, organization_id) do + if neighbor.remote_port_id && mac_address?(neighbor.remote_port_id) do + find_device_by_mac(neighbor.remote_port_id, organization_id) + end + end + + defp try_match_by_name(neighbor, organization_id) do + if neighbor.remote_system_name do + find_device_by_name(neighbor.remote_system_name, organization_id) + end + end + + # Check if a string looks like a MAC address + defp mac_address?(str) when is_binary(str) do + # Match patterns like "AA:BB:CC:DD:EE:FF" or "aa-bb-cc-dd-ee-ff" + String.match?(str, ~r/^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$/) + end + + defp mac_address?(_), do: false + + defp find_device_by_mac(mac_address, organization_id) do + # MAC address might be in format "aa:bb:cc:dd:ee:ff", "aa-bb-cc-dd-ee-ff", or "aabbccddeeff" + # Normalize to just the hex digits for comparison + normalized_mac = mac_address |> String.downcase() |> String.replace(~r/[:-]/, "") + + organization_id + |> DeviceQuery.for_organization() + |> join(:inner, [e], d in Device, on: d.device_id == e.id) + |> join(:inner, [e, d], i in Interface, on: i.snmp_device_id == d.id) + |> where( + [e, d, i], + fragment("REPLACE(REPLACE(LOWER(?), ':', ''), '-', '')", i.if_phys_address) == + ^normalized_mac + ) + |> select([e], e) + |> limit(1) + |> Repo.one() + end + + defp find_device_by_name(system_name, organization_id) do + # Try exact match first, then partial match + normalized_name = String.downcase(system_name) + + organization_id + |> DeviceQuery.for_organization() + |> where([e], fragment("LOWER(?)", e.name) == ^normalized_name) + |> limit(1) + |> Repo.one() + |> case do + nil -> + # Try partial match as fallback + organization_id + |> DeviceQuery.for_organization() + |> where( + [e], + fragment("LOWER(?) LIKE ?", e.name, ^"%#{Towerops.QueryHelpers.sanitize_like(normalized_name)}%") + ) + |> limit(1) + |> Repo.one() + + device -> + device + end + end +end diff --git a/lib/towerops/snmp/processors.ex b/lib/towerops/snmp/processors.ex new file mode 100644 index 00000000..cafe59af --- /dev/null +++ b/lib/towerops/snmp/processors.ex @@ -0,0 +1,65 @@ +defmodule Towerops.Snmp.Processors do + @moduledoc """ + Read-side queries and mutations for processors discovered on SNMP devices. + + Time-series reading inserts live in `Towerops.Snmp.Monitoring`. This module + covers processor metadata and recent reading lookups. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Processor + alias Towerops.Snmp.ProcessorReading + + @doc """ + Lists all processors for a device. + """ + def list_processors(snmp_device_id) do + Processor + |> where([p], p.snmp_device_id == ^snmp_device_id) + |> order_by([p], p.processor_index) + |> Repo.all() + end + + @doc """ + Gets a specific processor. + """ + def get_processor(processor_id) do + Repo.get(Processor, processor_id) + end + + @doc """ + Updates a processor. + """ + def update_processor(processor, attrs) do + processor + |> Processor.changeset(attrs) + |> Repo.update() + end + + @doc """ + Gets recent processor readings for a processor. + + ## Options + + - `:limit` - Maximum number of readings to return (default: 100) + - `:since` - Only return readings after this datetime + """ + def get_processor_readings(processor_id, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + since = Keyword.get(opts, :since) + + ProcessorReading + |> where([r], r.processor_id == ^processor_id) + |> order_by([r], desc: r.checked_at) + |> limit(^limit) + |> maybe_filter_since(since) + |> Repo.all() + end + + defp maybe_filter_since(query, nil), do: query + defp maybe_filter_since(query, since), do: where(query, [r], r.checked_at >= ^since) +end diff --git a/lib/towerops/snmp/sensors.ex b/lib/towerops/snmp/sensors.ex new file mode 100644 index 00000000..4440af6b --- /dev/null +++ b/lib/towerops/snmp/sensors.ex @@ -0,0 +1,85 @@ +defmodule Towerops.Snmp.Sensors do + @moduledoc """ + Read-side queries and mutations for sensors and state sensors discovered on + SNMP devices. + + Time-series reading queries live in `Towerops.Snmp.Queries`. This module + covers sensor metadata. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Sensor + alias Towerops.Snmp.StateSensor + + @doc """ + Lists all sensors for a device. + """ + def list_sensors(device_id) do + Sensor + |> where([s], s.snmp_device_id == ^device_id) + |> order_by([s], [s.sensor_type, s.sensor_index]) + |> Repo.all() + end + + @doc """ + Lists only monitored sensors for a device. + """ + def list_monitored_sensors(device_id) do + Sensor + |> where([s], s.snmp_device_id == ^device_id and s.monitored == true) + |> order_by([s], [s.sensor_type, s.sensor_index]) + |> Repo.all() + end + + @doc """ + Lists sensors grouped by type. + """ + def list_sensors_by_type(device_id) do + sensors = + Sensor + |> where([s], s.snmp_device_id == ^device_id) + |> order_by([s], [s.sensor_type, s.sensor_index]) + |> Repo.all() + + Enum.group_by(sensors, & &1.sensor_type) + end + + @doc """ + Gets a specific sensor. + """ + def get_sensor(sensor_id) do + Repo.get(Sensor, sensor_id) + end + + @doc """ + Updates a sensor. + """ + def update_sensor(%Sensor{} = sensor, attrs) do + sensor + |> Sensor.changeset(attrs) + |> Repo.update() + end + + @doc """ + Updates a state sensor. + """ + def update_state_sensor(%StateSensor{} = state_sensor, attrs) do + state_sensor + |> StateSensor.changeset(attrs) + |> Repo.update() + end + + @doc """ + Lists all state sensors for a device. + """ + def list_state_sensors(device_id) do + StateSensor + |> where([s], s.snmp_device_id == ^device_id) + |> order_by([s], s.sensor_index) + |> Repo.all() + end +end diff --git a/lib/towerops/snmp/storage_queries.ex b/lib/towerops/snmp/storage_queries.ex new file mode 100644 index 00000000..2dc421b9 --- /dev/null +++ b/lib/towerops/snmp/storage_queries.ex @@ -0,0 +1,126 @@ +defmodule Towerops.Snmp.StorageQueries do + @moduledoc """ + Read-side queries and mutations for storage entities discovered on SNMP devices, + plus persistence for storage time-series readings. + + Named `StorageQueries` to avoid clashing with `Towerops.Snmp.Storage` (the + Ecto schema). Public callers use `Towerops.Snmp` delegates. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Storage + alias Towerops.Snmp.StorageReading + + @doc """ + Lists all storage entries for a device. + """ + def list_storage(snmp_device_id) do + Storage + |> where([s], s.snmp_device_id == ^snmp_device_id) + |> order_by([s], s.storage_index) + |> Repo.all() + end + + @doc """ + Gets a specific storage entry. + """ + def get_storage(storage_id) do + Repo.get(Storage, storage_id) + end + + @doc """ + Updates a storage entry. + """ + def update_storage(storage, attrs) do + storage + |> Storage.changeset(attrs) + |> Repo.update() + end + + @doc """ + Gets recent storage readings for a storage entry. + + ## Options + + - `:limit` - Maximum number of readings to return (default: 100) + - `:since` - Only return readings after this datetime + """ + def get_storage_readings(storage_id, opts \\ []) do + limit = Keyword.get(opts, :limit, 100) + since = Keyword.get(opts, :since) + + StorageReading + |> where([r], r.storage_id == ^storage_id) + |> order_by([r], desc: r.checked_at) + |> limit(^limit) + |> maybe_filter_since(since) + |> Repo.all() + end + + defp maybe_filter_since(query, nil), do: query + defp maybe_filter_since(query, since), do: where(query, [r], r.checked_at >= ^since) + + @doc """ + Gets the latest storage reading for a storage entry. + """ + def get_latest_storage_reading(storage_id) do + StorageReading + |> where([r], r.storage_id == ^storage_id) + |> order_by([r], desc: r.checked_at) + |> limit(1) + |> Repo.one() + end + + @doc """ + Gets the latest storage readings for multiple storage entries in a single query. + + Returns a map of %{storage_id => reading} for efficient batch loading. + Storage entries without readings will not be present in the map. + """ + def get_latest_storage_readings_batch([]), do: %{} + + def get_latest_storage_readings_batch(storage_ids) when is_list(storage_ids) do + from(r in StorageReading, + where: r.storage_id in ^storage_ids, + distinct: r.storage_id, + order_by: [asc: r.storage_id, desc: r.checked_at] + ) + |> Repo.all() + |> Map.new(fn reading -> {reading.storage_id, reading} end) + end + + @doc """ + Records a new storage reading. + """ + def create_storage_reading(attrs) do + %StorageReading{} + |> StorageReading.changeset(attrs) + |> Repo.insert() + end + + @doc """ + Batch inserts multiple storage readings using `Repo.insert_all/3`. + + Accepts a list of attribute maps with the same keys as `create_storage_reading/1`. + Generates UUIDs and timestamps automatically. Returns `{count, nil}`. + """ + @spec create_storage_readings_batch([map()]) :: {non_neg_integer(), nil} + def create_storage_readings_batch([]), do: {0, nil} + + def create_storage_readings_batch(entries) when is_list(entries) do + now = Towerops.Time.now() + + rows = + Enum.map(entries, fn entry -> + entry + |> Map.put(:id, Ecto.UUID.generate()) + |> Map.put(:inserted_at, now) + end) + + Repo.insert_all(StorageReading, rows) + end +end diff --git a/lib/towerops/snmp/vlans.ex b/lib/towerops/snmp/vlans.ex new file mode 100644 index 00000000..0e42a999 --- /dev/null +++ b/lib/towerops/snmp/vlans.ex @@ -0,0 +1,29 @@ +defmodule Towerops.Snmp.Vlans do + @moduledoc """ + Read-side queries for VLANs discovered on SNMP devices. + + Extracted from `Towerops.Snmp` to keep the context module focused. + """ + + import Ecto.Query + + alias Towerops.Repo + alias Towerops.Snmp.Vlan + + @doc """ + Lists all VLANs for a device. + """ + def list_vlans(snmp_device_id) do + Vlan + |> where([v], v.snmp_device_id == ^snmp_device_id) + |> order_by([v], v.vlan_id) + |> Repo.all() + end + + @doc """ + Gets a specific VLAN. + """ + def get_vlan(vlan_id) do + Repo.get(Vlan, vlan_id) + end +end