536 lines
18 KiB
Elixir
536 lines
18 KiB
Elixir
defmodule Towerops.Snmp.Discovery do
|
|
@moduledoc """
|
|
Core SNMP discovery orchestrator.
|
|
|
|
Handles the complete discovery workflow:
|
|
1. Test SNMP connectivity
|
|
2. Discover system information
|
|
3. Select appropriate device profile (Cisco, NetSNMP, or Base)
|
|
4. Discover interfaces and sensors using profile
|
|
5. Save discovered data to database
|
|
|
|
Can be run manually or scheduled as a background job.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Devices
|
|
alias Towerops.Devices.Device, as: DeviceSchema
|
|
alias Towerops.Profiles
|
|
alias Towerops.Profiles.DeviceProfile
|
|
alias Towerops.Repo
|
|
alias Towerops.Snmp.Client
|
|
alias Towerops.Snmp.Device
|
|
alias Towerops.Snmp.Interface
|
|
alias Towerops.Snmp.NeighborDiscovery
|
|
alias Towerops.Snmp.Profiles.Base
|
|
alias Towerops.Snmp.Profiles.Cisco
|
|
alias Towerops.Snmp.Profiles.Dynamic
|
|
alias Towerops.Snmp.Profiles.Mikrotik
|
|
alias Towerops.Snmp.Profiles.NetSnmp
|
|
alias Towerops.Snmp.Profiles.Ubiquiti
|
|
alias Towerops.Snmp.Sensor
|
|
|
|
require Logger
|
|
|
|
@type system_info :: %{
|
|
optional(:sys_descr) => String.t(),
|
|
optional(:sys_object_id) => String.t(),
|
|
optional(:sys_name) => String.t(),
|
|
optional(:sys_uptime) => non_neg_integer(),
|
|
optional(:sys_contact) => String.t(),
|
|
optional(:sys_location) => String.t(),
|
|
optional(atom()) => term()
|
|
}
|
|
|
|
@type device_info :: %{
|
|
optional(:manufacturer) => String.t(),
|
|
optional(:model) => String.t(),
|
|
optional(:firmware_version) => String.t() | nil,
|
|
optional(:serial_number) => String.t() | nil,
|
|
optional(atom()) => term()
|
|
}
|
|
|
|
@type interface_data :: %{
|
|
required(:if_index) => non_neg_integer(),
|
|
required(:if_descr) => String.t(),
|
|
optional(:if_name) => String.t() | nil,
|
|
optional(:if_alias) => String.t() | nil,
|
|
optional(:if_type) => non_neg_integer() | nil,
|
|
optional(:if_mtu) => non_neg_integer() | nil,
|
|
optional(:if_speed) => non_neg_integer() | nil,
|
|
optional(:if_phys_address) => String.t() | nil,
|
|
optional(:if_admin_status) => non_neg_integer() | nil,
|
|
optional(:if_oper_status) => non_neg_integer() | nil,
|
|
optional(atom()) => term()
|
|
}
|
|
|
|
@type sensor_data :: %{
|
|
required(:sensor_type) => String.t(),
|
|
required(:sensor_index) => String.t(),
|
|
required(:sensor_oid) => String.t(),
|
|
required(:sensor_descr) => String.t(),
|
|
required(:sensor_unit) => String.t(),
|
|
required(:sensor_divisor) => number(),
|
|
optional(:last_value) => float() | nil,
|
|
optional(:status) => String.t(),
|
|
optional(atom()) => term()
|
|
}
|
|
|
|
@type profile :: module() | {:dynamic, DeviceProfile.t()}
|
|
|
|
@type discovery_summary :: %{
|
|
success: non_neg_integer(),
|
|
failed: non_neg_integer(),
|
|
errors: [term()]
|
|
}
|
|
|
|
@doc """
|
|
Runs discovery for a single device.
|
|
Returns {:ok, device} or {:error, reason}.
|
|
|
|
## Examples
|
|
|
|
iex> discover_device(device)
|
|
{:ok, %Device{manufacturer: "Cisco", model: "C2960"}}
|
|
|
|
iex> discover_device(device_without_snmp)
|
|
{:error, :snmp_not_enabled}
|
|
"""
|
|
@spec discover_device(DeviceSchema.t()) :: {:ok, Device.t()} | {:error, term()}
|
|
def discover_device(%DeviceSchema{} = device) do
|
|
if device.snmp_enabled do
|
|
Logger.info("Starting SNMP discovery for device: #{device.name} (#{device.ip_address})")
|
|
|
|
client_opts = build_client_opts(device)
|
|
|
|
with {:ok, _} <- Client.test_connection(client_opts),
|
|
{:ok, system_info} <- discover_system(client_opts),
|
|
{:ok, device} <- update_device_name_from_snmp(device, system_info),
|
|
profile = select_profile(system_info),
|
|
{:ok, device_info} <- build_device_info(client_opts, system_info, profile),
|
|
{:ok, interfaces} <- discover_interfaces(client_opts, profile),
|
|
{:ok, sensors} <- discover_sensors(client_opts, profile),
|
|
{:ok, discovered_device} <- save_discovery_results(device, device_info, interfaces, sensors),
|
|
{:ok, neighbors} <- NeighborDiscovery.discover_neighbors(client_opts, discovered_device.interfaces),
|
|
:ok <- save_neighbors(discovered_device.device_id, neighbors) do
|
|
_ = update_device_discovery_time(device)
|
|
Logger.info("SNMP discovery completed successfully for: #{device.name}")
|
|
|
|
# Determine if this was first discovery or rediscovery
|
|
# Check if device existed before (had last_discovery_at set)
|
|
is_rediscovery = device.last_discovery_at != nil
|
|
|
|
# Log discovery event
|
|
log_discovery_event(device, discovered_device, is_rediscovery)
|
|
|
|
# Broadcast discovery completion for real-time updates
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device.id}",
|
|
{:discovery_completed, device.id}
|
|
)
|
|
|
|
{:ok, discovered_device}
|
|
else
|
|
{:error, reason} = error ->
|
|
Logger.error("SNMP discovery failed for #{device.name}: #{inspect(reason)}")
|
|
error
|
|
end
|
|
else
|
|
{:error, :snmp_not_enabled}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Runs discovery for all SNMP-enabled devices in an organization.
|
|
Returns a summary of successful and failed discoveries.
|
|
"""
|
|
@spec discover_all(String.t()) :: {:ok, discovery_summary()}
|
|
def discover_all(org_id) do
|
|
device_list =
|
|
DeviceSchema
|
|
|> join(:inner, [e], s in assoc(e, :site))
|
|
|> where([e, s], s.organization_id == ^org_id and e.snmp_enabled == true)
|
|
|> Repo.all()
|
|
|
|
Logger.info("Starting SNMP discovery for #{length(device_list)} devices in org #{org_id}")
|
|
|
|
results =
|
|
device_list
|
|
|> Task.async_stream(
|
|
&discover_device/1,
|
|
max_concurrency: 5,
|
|
timeout: 60_000,
|
|
on_timeout: :kill_task
|
|
)
|
|
|> Enum.reduce(%{success: 0, failed: 0, errors: []}, fn
|
|
{:ok, {:ok, _device}}, acc ->
|
|
%{acc | success: acc.success + 1}
|
|
|
|
{:ok, {:error, reason}}, acc ->
|
|
%{acc | failed: acc.failed + 1, errors: [reason | acc.errors]}
|
|
|
|
{:exit, :timeout}, acc ->
|
|
%{acc | failed: acc.failed + 1, errors: [:timeout | acc.errors]}
|
|
end)
|
|
|
|
Logger.info("Discovery completed: #{results.success} succeeded, #{results.failed} failed")
|
|
{:ok, results}
|
|
end
|
|
|
|
# Private functions
|
|
|
|
@spec build_client_opts(DeviceSchema.t()) :: Client.connection_opts()
|
|
defp build_client_opts(device) do
|
|
# Get SNMP config with hierarchical fallback (device -> site -> organization)
|
|
snmp_config = Devices.get_snmp_config(device)
|
|
|
|
[
|
|
ip: device.ip_address,
|
|
community: snmp_config.community,
|
|
version: snmp_config.version,
|
|
port: device.snmp_port || 161
|
|
]
|
|
end
|
|
|
|
@spec discover_system(Client.connection_opts()) :: {:ok, system_info()} | {:error, term()}
|
|
defp discover_system(client_opts) do
|
|
# Use Base profile for initial system discovery
|
|
Base.discover_system_info(client_opts)
|
|
end
|
|
|
|
@doc """
|
|
Selects the appropriate SNMP profile based on system information.
|
|
|
|
First tries to match against database-stored device profiles.
|
|
If no database match is found, falls back to hard-coded profile modules.
|
|
|
|
## Examples
|
|
|
|
iex> select_profile(%{sys_descr: "Cisco IOS"})
|
|
Towerops.Snmp.Profiles.Cisco
|
|
|
|
iex> select_profile(%{sys_object_id: "1.3.6.1.4.1.41112"})
|
|
Towerops.Snmp.Profiles.Ubiquiti
|
|
"""
|
|
@spec select_profile(system_info()) :: profile()
|
|
def select_profile(system_info) do
|
|
# First try database profiles
|
|
case Profiles.match_profile(system_info) do
|
|
%DeviceProfile{} = profile ->
|
|
Logger.debug("Selected database profile: #{profile.name}")
|
|
{:dynamic, profile}
|
|
|
|
nil ->
|
|
# Fall back to hardcoded profiles
|
|
select_hardcoded_profile(system_info)
|
|
end
|
|
end
|
|
|
|
defp select_hardcoded_profile(system_info) do
|
|
sys_descr = Map.get(system_info, :sys_descr, "")
|
|
sys_object_id = Map.get(system_info, :sys_object_id, "")
|
|
|
|
cond do
|
|
String.contains?(sys_descr, "RouterOS") ->
|
|
Logger.debug("Selected MikroTik profile")
|
|
Mikrotik
|
|
|
|
String.contains?(sys_descr, ["Cisco", "IOS"]) ->
|
|
Logger.debug("Selected Cisco profile")
|
|
Cisco
|
|
|
|
# Ubiquiti devices (check before generic Linux since they run Linux)
|
|
# OID 10002 = old Ubiquiti/UniFi devices, 41112 = newer airOS devices
|
|
String.contains?(sys_object_id, ["10002", "41112"]) or
|
|
String.contains?(sys_descr, ["airOS", "AirFiber", "airMAX", "Ubiquiti"]) ->
|
|
Logger.debug("Selected Ubiquiti profile")
|
|
Ubiquiti
|
|
|
|
String.contains?(sys_descr, "Linux") ->
|
|
Logger.debug("Selected NetSNMP profile")
|
|
NetSnmp
|
|
|
|
true ->
|
|
Logger.debug("Selected Base profile")
|
|
Base
|
|
end
|
|
end
|
|
|
|
@spec build_device_info(Client.connection_opts(), system_info(), profile()) ::
|
|
{:ok, device_info()}
|
|
defp build_device_info(client_opts, system_info, {:dynamic, profile}) do
|
|
# Use dynamic profile to identify device
|
|
identified_info = Dynamic.identify_device(profile, client_opts, system_info)
|
|
{:ok, identified_info}
|
|
rescue
|
|
error ->
|
|
Logger.error("Failed to identify device with dynamic profile: #{inspect(error)}")
|
|
{:ok, system_info}
|
|
end
|
|
|
|
defp build_device_info(client_opts, system_info, profile) when is_atom(profile) do
|
|
Logger.info(
|
|
"build_device_info: system_info has _raw_discovery_data: #{inspect(Map.has_key?(system_info, :_raw_discovery_data))}"
|
|
)
|
|
|
|
# Get profile-specific system info if profile has discover_system_info/1
|
|
enriched_system_info =
|
|
if function_exported?(profile, :discover_system_info, 1) do
|
|
case profile.discover_system_info(client_opts) do
|
|
{:ok, profile_info} ->
|
|
Logger.info("Enriched system_info with profile-specific data")
|
|
Map.merge(system_info, profile_info)
|
|
|
|
{:error, _reason} ->
|
|
Logger.warning("Failed to get profile-specific system info, using base info")
|
|
system_info
|
|
end
|
|
else
|
|
system_info
|
|
end
|
|
|
|
# Let the hard-coded profile identify the device
|
|
identified_info = profile.identify_device(enriched_system_info)
|
|
|
|
# Collect raw debug data if profile supports it
|
|
identified_info_with_debug =
|
|
if function_exported?(profile, :collect_raw_debug_data, 1) do
|
|
raw_data = profile.collect_raw_debug_data(client_opts)
|
|
Logger.info("Collected raw debug data: #{map_size(raw_data)} keys")
|
|
|
|
identified_info
|
|
|> Map.put(:_raw_discovery_data, raw_data)
|
|
|> Map.put(:_last_discovery_at, DateTime.utc_now())
|
|
else
|
|
identified_info
|
|
end
|
|
|
|
Logger.info(
|
|
"build_device_info: identified_info has _raw_discovery_data: #{inspect(Map.has_key?(identified_info_with_debug, :_raw_discovery_data))}"
|
|
)
|
|
|
|
{:ok, identified_info_with_debug}
|
|
rescue
|
|
_error ->
|
|
Logger.error("Failed to identify device with profile, falling back to Base profile")
|
|
# Fall back to Base profile identification to ensure manufacturer/model are set
|
|
{:ok, Base.identify_device(system_info)}
|
|
end
|
|
|
|
@spec discover_interfaces(Client.connection_opts(), profile()) ::
|
|
{:ok, [interface_data()]}
|
|
defp discover_interfaces(client_opts, {:dynamic, profile}) do
|
|
case Dynamic.discover_interfaces(profile, client_opts) do
|
|
{:ok, interfaces} ->
|
|
Logger.debug("Discovered #{length(interfaces)} interfaces")
|
|
{:ok, interfaces}
|
|
|
|
{:error, _} ->
|
|
Logger.warning("Interface discovery failed, continuing without interfaces")
|
|
{:ok, []}
|
|
end
|
|
end
|
|
|
|
defp discover_interfaces(client_opts, profile) when is_atom(profile) do
|
|
case profile.discover_interfaces(client_opts) do
|
|
{:ok, interfaces} ->
|
|
Logger.debug("Discovered #{length(interfaces)} interfaces")
|
|
{:ok, interfaces}
|
|
|
|
{:error, _} ->
|
|
Logger.warning("Interface discovery failed, continuing without interfaces")
|
|
{:ok, []}
|
|
end
|
|
end
|
|
|
|
@spec discover_sensors(Client.connection_opts(), profile()) :: {:ok, [sensor_data()]}
|
|
defp discover_sensors(client_opts, {:dynamic, profile}) do
|
|
{:ok, sensors} = Dynamic.discover_sensors(profile, client_opts)
|
|
Logger.debug("Discovered #{length(sensors)} sensors")
|
|
{:ok, sensors}
|
|
end
|
|
|
|
defp discover_sensors(client_opts, profile) when is_atom(profile) do
|
|
case profile.discover_sensors(client_opts) do
|
|
{:ok, sensors} ->
|
|
Logger.debug("Discovered #{length(sensors)} sensors")
|
|
{:ok, sensors}
|
|
|
|
{:error, _} ->
|
|
Logger.warning("Sensor discovery failed, continuing without sensors")
|
|
{:ok, []}
|
|
end
|
|
end
|
|
|
|
@spec save_discovery_results(
|
|
DeviceSchema.t(),
|
|
device_info(),
|
|
[interface_data()],
|
|
[sensor_data()]
|
|
) :: {:ok, Device.t()} | {:error, term()}
|
|
defp save_discovery_results(device, device_info, interfaces, sensors) do
|
|
Repo.transaction(fn ->
|
|
# Upsert Device
|
|
snmp_device = upsert_device(device, device_info)
|
|
|
|
# Delete old interfaces/sensors and insert new ones
|
|
_ = delete_old_data(snmp_device)
|
|
new_interfaces = insert_interfaces(snmp_device, interfaces)
|
|
_ = insert_sensors(snmp_device, sensors)
|
|
|
|
# Return device with interfaces loaded and device_id added
|
|
# Use snmp_device.device_id which references the Equipment table
|
|
%{snmp_device | interfaces: Enum.map(new_interfaces, &Map.put(&1, :device_id, snmp_device.device_id))}
|
|
end)
|
|
end
|
|
|
|
@spec upsert_device(DeviceSchema.t(), device_info()) :: Device.t()
|
|
defp upsert_device(device, device_info) do
|
|
Logger.info("upsert_device called with device_info keys: #{inspect(Map.keys(device_info))}")
|
|
|
|
# Extract raw discovery data if present (using underscore prefix to distinguish from regular fields)
|
|
raw_discovery_data = Map.get(device_info, :_raw_discovery_data)
|
|
last_discovery_at = Map.get(device_info, :_last_discovery_at, DateTime.utc_now())
|
|
|
|
Logger.info("Extracted raw_discovery_data: #{inspect(raw_discovery_data != nil)}")
|
|
|
|
# Remove internal keys before saving
|
|
device_attrs =
|
|
device_info
|
|
|> Map.delete(:_raw_discovery_data)
|
|
|> Map.delete(:_last_discovery_at)
|
|
|> Map.put(:device_id, device.id)
|
|
|> Map.put(:raw_discovery_data, raw_discovery_data)
|
|
|> Map.put(:last_discovery_at, last_discovery_at)
|
|
|
|
Logger.info("device_attrs has raw_discovery_data: #{inspect(Map.has_key?(device_attrs, :raw_discovery_data))}")
|
|
|
|
case Repo.get_by(Device, device_id: device.id) do
|
|
nil ->
|
|
%Device{}
|
|
|> Device.changeset(device_attrs)
|
|
|> Repo.insert!()
|
|
|
|
existing_device ->
|
|
existing_device
|
|
|> Device.changeset(device_attrs)
|
|
|> Repo.update!()
|
|
end
|
|
end
|
|
|
|
@spec delete_old_data(Device.t()) :: {non_neg_integer(), nil | [term()]}
|
|
defp delete_old_data(device) do
|
|
# Delete old interfaces and sensors (cascade will handle stats/readings)
|
|
Repo.delete_all(from i in Interface, where: i.snmp_device_id == ^device.id)
|
|
Repo.delete_all(from s in Sensor, where: s.snmp_device_id == ^device.id)
|
|
end
|
|
|
|
@spec insert_interfaces(Device.t(), [interface_data()]) :: [Interface.t()]
|
|
defp insert_interfaces(device, interfaces) do
|
|
Enum.map(interfaces, fn interface_data ->
|
|
%Interface{}
|
|
|> Interface.changeset(Map.put(interface_data, :snmp_device_id, device.id))
|
|
|> Repo.insert!()
|
|
end)
|
|
end
|
|
|
|
@spec insert_sensors(Device.t(), [sensor_data()]) :: [Sensor.t()]
|
|
defp insert_sensors(device, sensors) do
|
|
# Deduplicate sensors by sensor_index (keep first occurrence)
|
|
# This prevents duplicate key errors when profiles generate duplicate indices
|
|
sensors
|
|
|> Enum.uniq_by(& &1.sensor_index)
|
|
|> Enum.map(fn sensor_data ->
|
|
%Sensor{}
|
|
|> Sensor.changeset(Map.put(sensor_data, :snmp_device_id, device.id))
|
|
|> Repo.insert!()
|
|
end)
|
|
end
|
|
|
|
@spec update_device_discovery_time(DeviceSchema.t()) ::
|
|
{:ok, DeviceSchema.t()} | {:error, Ecto.Changeset.t()}
|
|
defp update_device_discovery_time(device) do
|
|
device
|
|
|> Ecto.Changeset.change(last_discovery_at: DateTime.truncate(DateTime.utc_now(), :second))
|
|
|> Repo.update()
|
|
end
|
|
|
|
@spec save_neighbors(binary(), [map()]) :: :ok
|
|
defp save_neighbors(device_id, neighbors) do
|
|
# Delete stale neighbors (not seen in last 5 minutes)
|
|
cutoff = DateTime.add(DateTime.utc_now(), -5, :minute)
|
|
Towerops.Snmp.delete_stale_neighbors(device_id, cutoff)
|
|
|
|
# Upsert each discovered neighbor
|
|
Enum.each(neighbors, fn neighbor_data ->
|
|
Towerops.Snmp.upsert_neighbor(neighbor_data)
|
|
end)
|
|
|
|
Logger.info("Saved #{length(neighbors)} neighbors for device #{device_id}")
|
|
:ok
|
|
end
|
|
|
|
@spec update_device_name_from_snmp(DeviceSchema.t(), system_info()) ::
|
|
{:ok, DeviceSchema.t()} | {:error, term()}
|
|
defp update_device_name_from_snmp(device, system_info) do
|
|
# Only update device name if it's currently empty and we got a sysName from SNMP
|
|
if (is_nil(device.name) or device.name == "") and Map.has_key?(system_info, :sys_name) do
|
|
sys_name = Map.get(system_info, :sys_name)
|
|
|
|
if sys_name && sys_name != "" do
|
|
Logger.info("Updating device name from SNMP sysName: #{sys_name}")
|
|
|
|
device
|
|
|> Ecto.Changeset.change(name: sys_name)
|
|
|> Repo.update()
|
|
else
|
|
{:ok, device}
|
|
end
|
|
else
|
|
{:ok, device}
|
|
end
|
|
end
|
|
|
|
defp log_discovery_event(device, discovered_device, is_rediscovery) do
|
|
event_type = if is_rediscovery, do: "device_rediscovered", else: "device_discovered"
|
|
|
|
message =
|
|
if is_rediscovery do
|
|
"Device rediscovered: #{device.name} (#{discovered_device.manufacturer} #{discovered_device.model})"
|
|
else
|
|
"Device discovered: #{device.name} (#{discovered_device.manufacturer} #{discovered_device.model})"
|
|
end
|
|
|
|
metadata = %{
|
|
manufacturer: discovered_device.manufacturer,
|
|
model: discovered_device.model,
|
|
firmware_version: discovered_device.firmware_version,
|
|
serial_number: discovered_device.serial_number,
|
|
interface_count: length(discovered_device.interfaces || []),
|
|
sensor_count:
|
|
Repo.aggregate(
|
|
from(s in Sensor, where: s.snmp_device_id == ^discovered_device.id),
|
|
:count
|
|
)
|
|
}
|
|
|
|
event_attrs = %{
|
|
device_id: device.id,
|
|
event_type: event_type,
|
|
severity: "info",
|
|
message: message,
|
|
metadata: metadata,
|
|
occurred_at: DateTime.utc_now()
|
|
}
|
|
|
|
# Publish event for EventLogger to persist
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:events",
|
|
{:device_event, event_attrs}
|
|
)
|
|
end
|
|
end
|