graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(snmp): extract metadata submodules (vlans, ips, processors, storage, mempools, sensors, interfaces, neighbors)

Browse source 
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.
This commit is contained in:
Graham McIntire 2026-04-30 14:04:38 -05:00
parent 94f144806a
commit 9bb43f6007
9 changed files with 790 additions and 668 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

739
lib/towerops/snmp.ex
View file

@ -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 """

81
lib/towerops/snmp/interfaces.ex Normal file
View file

@ -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

35
lib/towerops/snmp/ip_addresses.ex Normal file
View file

@ -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

116
lib/towerops/snmp/mempools.ex Normal file
View file

@ -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

182
lib/towerops/snmp/neighbors.ex Normal file
View file

@ -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

65
lib/towerops/snmp/processors.ex Normal file
View file

@ -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

85
lib/towerops/snmp/sensors.ex Normal file
View file

@ -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

126
lib/towerops/snmp/storage_queries.ex Normal file
View file

@ -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

29
lib/towerops/snmp/vlans.ex Normal file
View file

@ -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