Graham McIntire
e181c17f78
Fixes three critical issues affecting agent-based SNMP polling:
1. MikroTik gauge sensors incorrectly showing 10x values (223°C instead of 22.3°C)
- Root cause: get_int_value() rejected zero values, causing mtxrGaugeUnit
to be treated as nil and bypassing divisor logic
- Fixed: Accept zero values and apply correct divisor=10 for temp/voltage/current/power
2. Empty SNMP community strings sent to agents causing authentication failures
- Root cause: Devices with source="device" but null community weren't falling
back to organization/site inheritance
- Fixed: Enhanced resolve_snmp_community() to fall back to inheritance chain
even when source="device" but community is empty
3. Continuous MikroTik API commands sent every 60s instead of once per 24h
- Root cause: MikroTik jobs built on every poll cycle, not just discovery
- Fixed: Only build MikroTik jobs during discovery phase
Migrations update existing data to fix incorrect divisor values and community
source tracking. Added comprehensive debug logging to diagnose future issues.
945 lines
27 KiB
Elixir
945 lines
27 KiB
Elixir
defmodule Towerops.Devices do
|
|
@moduledoc """
|
|
The Devices context.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Agents
|
|
alias Towerops.Devices.Device, as: DeviceSchema
|
|
alias Towerops.Devices.Event
|
|
alias Towerops.Organizations.SubscriptionLimits
|
|
alias Towerops.Repo
|
|
alias Towerops.Sites
|
|
alias Towerops.Workers.DeviceMonitorWorker
|
|
alias Towerops.Workers.DevicePollerWorker
|
|
alias Towerops.Workers.DiscoveryWorker
|
|
|
|
@doc """
|
|
Returns the list of devices for a site.
|
|
Ordered by custom display_order (if set), then alphabetically by name.
|
|
"""
|
|
def list_site_devices(site_id) do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.site_id == ^site_id,
|
|
order_by: [asc: e.display_order, asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices for an organization (via sites).
|
|
Ordered by custom display_order (if set), then alphabetically by name.
|
|
|
|
Supports filtering by:
|
|
- site_id: Filter by specific site
|
|
- status: Filter by status ("up", "down", "unknown")
|
|
"""
|
|
def list_organization_devices(organization_id, filters \\ %{}) do
|
|
query =
|
|
from(e in DeviceSchema,
|
|
join: s in assoc(e, :site),
|
|
where: s.organization_id == ^organization_id,
|
|
order_by: [asc: e.display_order, asc: e.name],
|
|
preload: [site: s]
|
|
)
|
|
|
|
query =
|
|
if site_id = filters["site_id"] do
|
|
where(query, [e], e.site_id == ^site_id)
|
|
else
|
|
query
|
|
end
|
|
|
|
query =
|
|
if status = filters["status"] do
|
|
where(query, [e], e.status == ^status)
|
|
else
|
|
query
|
|
end
|
|
|
|
Repo.all(query)
|
|
end
|
|
|
|
@doc """
|
|
Returns all devices for multiple organizations (for GDPR data access).
|
|
"""
|
|
def list_devices_for_organizations(organization_ids) when is_list(organization_ids) do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
join: s in assoc(e, :site),
|
|
join: o in assoc(s, :organization),
|
|
where: s.organization_id in ^organization_ids,
|
|
order_by: [asc: s.organization_id, asc: e.name],
|
|
preload: [site: {s, organization: o}]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices for an organization.
|
|
"""
|
|
def count_organization_devices(organization_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema,
|
|
join: s in assoc(e, :site),
|
|
where: s.organization_id == ^organization_id
|
|
),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices for a site.
|
|
"""
|
|
def count_site_devices(site_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema, where: e.site_id == ^site_id),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the count of devices that is down for a site.
|
|
"""
|
|
def count_site_devices_down(site_id) do
|
|
Repo.aggregate(
|
|
from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == :down),
|
|
:count
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices with monitoring enabled.
|
|
"""
|
|
def list_monitored_devices do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.monitoring_enabled == true,
|
|
order_by: [asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of all devices with SNMP enabled.
|
|
"""
|
|
def list_snmp_enabled_devices do
|
|
Repo.all(
|
|
from(e in DeviceSchema,
|
|
where: e.snmp_enabled == true,
|
|
order_by: [asc: e.name]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Lists all devices with MikroTik API enabled and configured for backup.
|
|
Returns devices that have MikroTik API enabled, credentials, and an assigned agent.
|
|
"""
|
|
def list_mikrotik_devices_with_api do
|
|
# Get all MikroTik devices with API enabled and credentials configured
|
|
# Agent assignment is resolved hierarchically in backup_device/1 (includes global default)
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
left_join: s in assoc(d, :site),
|
|
left_join: o in assoc(s, :organization),
|
|
left_join: aa in assoc(d, :agent_assignments),
|
|
where: d.mikrotik_enabled == true,
|
|
where: not is_nil(d.mikrotik_username) or not is_nil(s.mikrotik_username) or not is_nil(o.mikrotik_username),
|
|
order_by: [asc: d.name],
|
|
preload: [site: {s, organization: o}, agent_assignments: aa],
|
|
distinct: d.id
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Resolves the agent token ID for a device using hierarchical resolution:
|
|
1. Direct agent assignment (via agent_assignments table)
|
|
2. Site-level agent assignment
|
|
3. Organization default agent
|
|
4. Global default cloud poller
|
|
|
|
Returns nil if no agent is assigned.
|
|
"""
|
|
def resolve_agent_token_id(device) do
|
|
# Ensure site and organization are preloaded
|
|
device = Repo.preload(device, site: :organization, agent_assignments: [])
|
|
|
|
# Try each level in order
|
|
get_direct_agent(device) ||
|
|
get_site_agent(device) ||
|
|
get_org_default_agent(device) ||
|
|
get_global_default_agent()
|
|
end
|
|
|
|
defp get_direct_agent(device) do
|
|
case Enum.find(device.agent_assignments, & &1.enabled) do
|
|
%{agent_token_id: id} when not is_nil(id) -> id
|
|
_ -> nil
|
|
end
|
|
end
|
|
|
|
defp get_site_agent(device), do: device.site.agent_token_id
|
|
|
|
defp get_org_default_agent(device), do: device.site.organization.default_agent_token_id
|
|
|
|
defp get_global_default_agent, do: Towerops.Settings.get_global_default_cloud_poller()
|
|
|
|
@doc """
|
|
Gets a single device by ID, returns nil if not found.
|
|
"""
|
|
def get_device(id) do
|
|
DeviceSchema
|
|
|> Repo.get(id)
|
|
|> case do
|
|
nil -> nil
|
|
device -> Repo.preload(device, [:site])
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Gets a single device by ID, raises if not found.
|
|
"""
|
|
def get_device!(id) do
|
|
DeviceSchema
|
|
|> Repo.get!(id)
|
|
|> Repo.preload([:site])
|
|
end
|
|
|
|
@doc """
|
|
Gets device with full details including SNMP device, interfaces, and sensors.
|
|
"""
|
|
def get_device_with_details(id) do
|
|
DeviceSchema
|
|
|> Repo.get(id)
|
|
|> case do
|
|
nil ->
|
|
nil
|
|
|
|
device ->
|
|
Repo.preload(device, [:site, snmp_device: [:interfaces, :sensors]])
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Finds a device by IP address within a site, optionally excluding a specific device ID.
|
|
Returns nil if no device found.
|
|
"""
|
|
def get_device_by_ip(ip_address, site_id, exclude_id \\ nil)
|
|
|
|
def get_device_by_ip(_ip_address, nil, _exclude_id), do: nil
|
|
|
|
def get_device_by_ip(ip_address, site_id, exclude_id) do
|
|
query =
|
|
from(d in DeviceSchema,
|
|
where: d.ip_address == ^ip_address and d.site_id == ^site_id,
|
|
preload: [:site]
|
|
)
|
|
|
|
query =
|
|
if exclude_id do
|
|
from(d in query, where: d.id != ^exclude_id)
|
|
else
|
|
query
|
|
end
|
|
|
|
Repo.one(query)
|
|
end
|
|
|
|
@doc """
|
|
Gets a single device belonging to a specific site.
|
|
"""
|
|
def get_site_device!(site_id, device_id) do
|
|
Repo.one!(from(e in DeviceSchema, where: e.id == ^device_id, where: e.site_id == ^site_id, preload: [:site]))
|
|
end
|
|
|
|
@doc """
|
|
Gets SNMP configuration for device with hierarchical fallback.
|
|
|
|
Falls back in this order:
|
|
1. Device-level configuration
|
|
2. Site-level configuration
|
|
3. Organization-level configuration
|
|
|
|
Returns a map with:
|
|
- version: SNMP version ("1", "2c", or "3")
|
|
- community: SNMP community string
|
|
- source: Where the config came from (:device, :site, or :organization)
|
|
"""
|
|
def get_snmp_config(device_id) when is_binary(device_id) do
|
|
device =
|
|
DeviceSchema
|
|
|> Repo.get!(device_id)
|
|
|> Repo.preload(site: :organization)
|
|
|
|
get_snmp_config(device)
|
|
end
|
|
|
|
def get_snmp_config(%DeviceSchema{} = device) do
|
|
device = ensure_snmp_config_loaded(device)
|
|
resolve_snmp_config(device)
|
|
end
|
|
|
|
defp ensure_snmp_config_loaded(device) do
|
|
# Ensure site and organization associations are loaded
|
|
needs_preload =
|
|
!Ecto.assoc_loaded?(device.site) ||
|
|
!Ecto.assoc_loaded?(device.site.organization)
|
|
|
|
if needs_preload do
|
|
Repo.preload(device, site: :organization)
|
|
else
|
|
device
|
|
end
|
|
end
|
|
|
|
defp resolve_snmp_config(device) do
|
|
# Resolve community and version independently with hierarchical fallback
|
|
# This allows setting version at device level while inheriting community from org
|
|
community =
|
|
device.snmp_community ||
|
|
device.site.snmp_community ||
|
|
device.site.organization.snmp_community
|
|
|
|
version =
|
|
device.snmp_version ||
|
|
device.site.snmp_version ||
|
|
device.site.organization.snmp_version ||
|
|
"2c"
|
|
|
|
source = determine_snmp_source(device)
|
|
|
|
%{
|
|
version: version,
|
|
community: community,
|
|
source: source
|
|
}
|
|
end
|
|
|
|
defp determine_snmp_source(device) do
|
|
# Determine source based on where community string came from (primary credential)
|
|
# If no community anywhere, source is :default
|
|
cond do
|
|
device.snmp_community != nil -> :device
|
|
device.site.snmp_community != nil -> :site
|
|
device.site.organization.snmp_community != nil -> :organization
|
|
true -> :default
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Gets MikroTik API configuration for a device with hierarchical fallback.
|
|
|
|
Returns a map with username, password (decrypted), port, use_ssl, enabled, and source.
|
|
|
|
## Examples
|
|
|
|
iex> get_mikrotik_config(device)
|
|
%{
|
|
username: "admin",
|
|
password: "secret",
|
|
port: 8729,
|
|
use_ssl: true,
|
|
enabled: true,
|
|
source: :device
|
|
}
|
|
"""
|
|
def get_mikrotik_config(device_id) when is_binary(device_id) do
|
|
device =
|
|
DeviceSchema
|
|
|> Repo.get!(device_id)
|
|
|> Repo.preload(site: :organization)
|
|
|
|
get_mikrotik_config(device)
|
|
end
|
|
|
|
def get_mikrotik_config(%DeviceSchema{} = device) do
|
|
device = ensure_mikrotik_config_loaded(device)
|
|
resolve_mikrotik_config(device)
|
|
end
|
|
|
|
defp ensure_mikrotik_config_loaded(device) do
|
|
# Ensure site and organization associations are loaded
|
|
needs_preload =
|
|
!Ecto.assoc_loaded?(device.site) ||
|
|
!Ecto.assoc_loaded?(device.site.organization)
|
|
|
|
if needs_preload do
|
|
Repo.preload(device, site: :organization)
|
|
else
|
|
device
|
|
end
|
|
end
|
|
|
|
defp resolve_mikrotik_config(device) do
|
|
# Resolve each field independently with hierarchical fallback
|
|
%{
|
|
username: resolve_mikrotik_username(device),
|
|
password: resolve_mikrotik_password(device),
|
|
port: resolve_mikrotik_port(device),
|
|
ssh_port: resolve_mikrotik_ssh_port(device),
|
|
use_ssl: resolve_mikrotik_use_ssl(device),
|
|
enabled: resolve_mikrotik_enabled(device),
|
|
source: determine_mikrotik_source(device)
|
|
}
|
|
end
|
|
|
|
defp resolve_mikrotik_username(device) do
|
|
device.mikrotik_username ||
|
|
device.site.mikrotik_username ||
|
|
device.site.organization.mikrotik_username
|
|
end
|
|
|
|
defp resolve_mikrotik_password(device) do
|
|
device.mikrotik_password ||
|
|
device.site.mikrotik_password ||
|
|
device.site.organization.mikrotik_password
|
|
end
|
|
|
|
defp resolve_mikrotik_port(device) do
|
|
device.mikrotik_port ||
|
|
device.site.mikrotik_port ||
|
|
device.site.organization.mikrotik_port ||
|
|
8729
|
|
end
|
|
|
|
defp resolve_mikrotik_ssh_port(device) do
|
|
device.mikrotik_ssh_port ||
|
|
device.site.mikrotik_ssh_port ||
|
|
device.site.organization.mikrotik_ssh_port ||
|
|
22
|
|
end
|
|
|
|
defp resolve_mikrotik_use_ssl(device) do
|
|
if device.mikrotik_use_ssl == nil do
|
|
device.site.mikrotik_use_ssl ||
|
|
device.site.organization.mikrotik_use_ssl ||
|
|
true
|
|
else
|
|
device.mikrotik_use_ssl
|
|
end
|
|
end
|
|
|
|
defp resolve_mikrotik_enabled(device) do
|
|
device.mikrotik_enabled ||
|
|
device.site.mikrotik_enabled ||
|
|
device.site.organization.mikrotik_enabled ||
|
|
false
|
|
end
|
|
|
|
defp determine_mikrotik_source(device) do
|
|
# Determine source based on where username came from (primary credential)
|
|
cond do
|
|
device.mikrotik_username != nil -> :device
|
|
device.site.mikrotik_username != nil -> :site
|
|
device.site.organization.mikrotik_username != nil -> :organization
|
|
true -> :default
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Propagates MikroTik credential changes from a site to all devices in that
|
|
site that inherit credentials from the site (credential_source = "site").
|
|
|
|
Updates username, password, port, use_ssl, and enabled fields.
|
|
"""
|
|
def propagate_site_mikrotik_change(site_id, attrs) do
|
|
# Find all devices in this site that inherit credentials from site
|
|
devices_to_update =
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
where: d.site_id == ^site_id,
|
|
where: d.mikrotik_credential_source == "site"
|
|
)
|
|
)
|
|
|
|
# Update each device's MikroTik credentials
|
|
Enum.each(devices_to_update, fn device ->
|
|
device
|
|
|> Ecto.Changeset.change(attrs)
|
|
|> Repo.update()
|
|
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device.id}:assignments",
|
|
{:assignments_changed, :mikrotik_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Propagates MikroTik credential changes from an organization to all devices
|
|
in sites that have no site-level credentials (credential_source = "site"
|
|
and site.mikrotik_username is nil).
|
|
|
|
Updates username, password, port, use_ssl, and enabled fields.
|
|
"""
|
|
def propagate_organization_mikrotik_change(organization_id, attrs) do
|
|
# Find all devices in this org's sites that:
|
|
# 1. Inherit from site (credential_source = "site")
|
|
# 2. Their site has no MikroTik username (so they fall back to org)
|
|
devices_to_update =
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
join: s in assoc(d, :site),
|
|
where: s.organization_id == ^organization_id,
|
|
where: d.mikrotik_credential_source == "site",
|
|
where: is_nil(s.mikrotik_username) or s.mikrotik_username == ""
|
|
)
|
|
)
|
|
|
|
# Update each device's MikroTik credentials
|
|
Enum.each(devices_to_update, fn device ->
|
|
device
|
|
|> Ecto.Changeset.change(attrs)
|
|
|> Repo.update()
|
|
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device.id}:assignments",
|
|
{:assignments_changed, :mikrotik_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Creates device.
|
|
|
|
## Options
|
|
* `:bypass_limits` - When true, bypasses subscription device limits (for superusers)
|
|
"""
|
|
def create_device(attrs, opts \\ []) do
|
|
bypass_limits = Keyword.get(opts, :bypass_limits, false)
|
|
|
|
# Check device quota before insertion (unless bypassing)
|
|
with {:ok, changeset} <- check_device_quota(attrs, bypass_limits),
|
|
{:ok, device} <- Repo.insert(changeset) do
|
|
# Start monitoring/polling if enabled
|
|
_ =
|
|
if device.monitoring_enabled do
|
|
DeviceMonitorWorker.start_monitoring(device.id)
|
|
end
|
|
|
|
_ =
|
|
if device.snmp_enabled do
|
|
DevicePollerWorker.start_polling(device.id)
|
|
end
|
|
|
|
{:ok, device}
|
|
end
|
|
end
|
|
|
|
defp check_device_quota(attrs, bypass_limits) do
|
|
changeset = DeviceSchema.changeset(%DeviceSchema{}, attrs)
|
|
|
|
if bypass_limits do
|
|
{:ok, changeset}
|
|
else
|
|
do_check_quota(changeset)
|
|
end
|
|
end
|
|
|
|
defp do_check_quota(changeset) do
|
|
case Ecto.Changeset.fetch_change(changeset, :site_id) do
|
|
{:ok, site_id} ->
|
|
validate_device_quota(changeset, site_id)
|
|
|
|
:error ->
|
|
# No site_id in changeset, let normal validation handle it
|
|
{:ok, changeset}
|
|
end
|
|
end
|
|
|
|
defp validate_device_quota(changeset, site_id) do
|
|
site = site_id |> Sites.get_site!() |> Repo.preload(:organization)
|
|
organization = site.organization
|
|
|
|
case SubscriptionLimits.check_device_limit(organization) do
|
|
{:ok, :within_limit} ->
|
|
{:ok, changeset}
|
|
|
|
{:error, :at_limit, _current, max} ->
|
|
error_changeset =
|
|
Ecto.Changeset.add_error(
|
|
changeset,
|
|
:base,
|
|
"You've reached your plan limit of #{max} devices. Upgrade to add more."
|
|
)
|
|
|
|
{:error, error_changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates device.
|
|
"""
|
|
def update_device(%DeviceSchema{} = device, attrs) do
|
|
old_monitoring = device.monitoring_enabled
|
|
old_snmp = device.snmp_enabled
|
|
old_snmp_version = device.snmp_version
|
|
old_snmp_port = device.snmp_port
|
|
|
|
changeset = DeviceSchema.changeset(device, attrs)
|
|
|
|
# Validate MikroTik SSL requirement for cloud pollers
|
|
changeset = validate_mikrotik_ssl_with_cloud_poller(changeset, device)
|
|
|
|
case Repo.update(changeset) do
|
|
{:ok, updated_device} = result ->
|
|
_ = handle_monitoring_changes(updated_device, old_monitoring)
|
|
_ = handle_snmp_changes(updated_device, old_snmp, old_snmp_version, old_snmp_port)
|
|
result
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
defp validate_mikrotik_ssl_with_cloud_poller(changeset, device) do
|
|
# Check if MikroTik is enabled and SSL is being disabled
|
|
mikrotik_enabled = Ecto.Changeset.get_field(changeset, :mikrotik_enabled)
|
|
use_ssl_change = Ecto.Changeset.get_change(changeset, :mikrotik_use_ssl)
|
|
|
|
# Only validate if MikroTik is enabled and SSL is explicitly being set to false
|
|
if mikrotik_enabled && use_ssl_change == false do
|
|
validate_cloud_poller_requires_ssl(changeset, device)
|
|
else
|
|
changeset
|
|
end
|
|
end
|
|
|
|
defp validate_cloud_poller_requires_ssl(changeset, device) do
|
|
# Get effective agent to check if it's a cloud poller
|
|
device_with_site = Repo.preload(device, site: :organization)
|
|
{effective_agent_id, _source} = Agents.get_effective_agent_token_with_source(device_with_site)
|
|
|
|
using_cloud_poller =
|
|
if effective_agent_id do
|
|
agent = Agents.get_agent_token!(effective_agent_id)
|
|
agent.is_cloud_poller
|
|
else
|
|
# No agent means cloud polling by default
|
|
true
|
|
end
|
|
|
|
if using_cloud_poller do
|
|
Ecto.Changeset.add_error(
|
|
changeset,
|
|
:mikrotik_use_ssl,
|
|
"SSL is required when using cloud pollers. Plain MikroTik API sends credentials unencrypted over the internet."
|
|
)
|
|
else
|
|
changeset
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Resolves the effective SNMP community string for a device.
|
|
|
|
Resolution order:
|
|
1. If device has its own community string (source = "device" AND community is non-empty), use it
|
|
2. Otherwise, inherit from site's community string
|
|
3. If site has no community, inherit from organization's community string
|
|
4. If no community found anywhere, return nil
|
|
|
|
Note: Even if source is "device", we fall back to inheritance if the device community is empty.
|
|
This handles cases where the source field is incorrect or the community was cleared.
|
|
"""
|
|
def resolve_snmp_community(%DeviceSchema{} = device) do
|
|
device = Repo.preload(device, site: :organization)
|
|
|
|
# Try device-specific community first (if source is "device" AND it's non-empty)
|
|
device_community =
|
|
if device.snmp_community_source == "device" && present?(device.snmp_community) do
|
|
device.snmp_community
|
|
end
|
|
|
|
# Fall back to inheritance chain
|
|
community =
|
|
device_community ||
|
|
device.site.snmp_community ||
|
|
device.site.organization.snmp_community
|
|
|
|
# Debug logging to track empty community strings
|
|
if is_nil(community) or community == "" do
|
|
require Logger
|
|
|
|
Logger.warning("Empty SNMP community resolved for device",
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
community_source: device.snmp_community_source,
|
|
device_community: device.snmp_community,
|
|
site_community: device.site.snmp_community,
|
|
org_community: device.site.organization.snmp_community,
|
|
site_id: device.site_id,
|
|
org_id: device.site.organization_id
|
|
)
|
|
end
|
|
|
|
community
|
|
end
|
|
|
|
# Helper to check if a string is present (not nil and not empty)
|
|
defp present?(value) when is_binary(value), do: String.trim(value) != ""
|
|
defp present?(_), do: false
|
|
|
|
@doc """
|
|
Propagates SNMP community string changes from a site to all devices
|
|
that inherit from it (source = "site").
|
|
|
|
This updates the actual snmp_community value on devices so they have
|
|
the resolved value cached and ready for agent communication.
|
|
"""
|
|
def propagate_site_community_change(site_id, new_community) do
|
|
# Find all devices in this site that inherit community from site
|
|
devices_to_update =
|
|
Repo.all(from(d in DeviceSchema, where: d.site_id == ^site_id, where: d.snmp_community_source == "site"))
|
|
|
|
# Update each device's community string
|
|
Enum.each(devices_to_update, fn device ->
|
|
device
|
|
|> Ecto.Changeset.change(%{snmp_community: new_community})
|
|
|> Repo.update()
|
|
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device.id}:assignments",
|
|
{:assignments_changed, :community_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Propagates SNMP community string changes from an organization to all
|
|
devices in sites that have no site-level community (source = "site"
|
|
and site.snmp_community is nil).
|
|
"""
|
|
def propagate_organization_community_change(organization_id, new_community) do
|
|
# Find all devices in this org's sites that:
|
|
# 1. Inherit from site (source = "site")
|
|
# 2. Their site has no community string (so they fall back to org)
|
|
devices_to_update =
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
join: s in assoc(d, :site),
|
|
where: s.organization_id == ^organization_id,
|
|
where: d.snmp_community_source == "site",
|
|
where: is_nil(s.snmp_community) or s.snmp_community == ""
|
|
)
|
|
)
|
|
|
|
# Update each device's community string
|
|
Enum.each(devices_to_update, fn device ->
|
|
device
|
|
|> Ecto.Changeset.change(%{snmp_community: new_community})
|
|
|> Repo.update()
|
|
|
|
# Broadcast assignment change to trigger agent job list refresh
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"device:#{device.id}:assignments",
|
|
{:assignments_changed, :community_updated}
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Deletes device.
|
|
"""
|
|
def delete_device(%DeviceSchema{} = device) do
|
|
# Stop monitoring and polling jobs before deleting
|
|
_ = DeviceMonitorWorker.stop_monitoring(device.id)
|
|
_ = DevicePollerWorker.stop_polling(device.id)
|
|
|
|
Repo.delete(device)
|
|
end
|
|
|
|
@doc """
|
|
Returns an `%Ecto.Changeset{}` for tracking device changes.
|
|
"""
|
|
def change_device(%DeviceSchema{} = device, attrs \\ %{}) do
|
|
DeviceSchema.changeset(device, attrs)
|
|
end
|
|
|
|
@doc """
|
|
Updates the status of device.
|
|
"""
|
|
def update_device_status(%DeviceSchema{} = device, new_status) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
changes = %{
|
|
status: new_status,
|
|
last_checked_at: now
|
|
}
|
|
|
|
changes =
|
|
if device.status == new_status do
|
|
changes
|
|
else
|
|
Map.put(changes, :last_status_change_at, now)
|
|
end
|
|
|
|
device
|
|
|> Ecto.Changeset.change(changes)
|
|
|> Repo.update()
|
|
end
|
|
|
|
@doc """
|
|
Updates the last SNMP poll timestamp for device.
|
|
Used for distributed coordination to prevent duplicate polling across pods.
|
|
"""
|
|
def update_snmp_poll_time(%DeviceSchema{} = device) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
device
|
|
|> Ecto.Changeset.change(%{last_snmp_poll_at: now})
|
|
|> Repo.update()
|
|
end
|
|
|
|
# Private helpers for monitoring/polling management
|
|
|
|
defp handle_monitoring_changes(device, old_monitoring) do
|
|
cond do
|
|
device.monitoring_enabled && !old_monitoring ->
|
|
_ = DeviceMonitorWorker.start_monitoring(device.id)
|
|
# Trigger discovery when monitoring is enabled (if SNMP is also enabled)
|
|
if device.snmp_enabled, do: DiscoveryWorker.enqueue(device.id)
|
|
|
|
!device.monitoring_enabled && old_monitoring ->
|
|
_ = DeviceMonitorWorker.stop_monitoring(device.id)
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp handle_snmp_changes(device, old_snmp, old_snmp_version, old_snmp_port) do
|
|
should_discover = should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port)
|
|
|
|
cond do
|
|
device.snmp_enabled && !old_snmp ->
|
|
_ = DevicePollerWorker.start_polling(device.id)
|
|
if should_discover, do: DiscoveryWorker.enqueue(device.id)
|
|
|
|
!device.snmp_enabled && old_snmp ->
|
|
_ = DevicePollerWorker.stop_polling(device.id)
|
|
|
|
should_discover ->
|
|
_ = DiscoveryWorker.enqueue(device.id)
|
|
|
|
true ->
|
|
:ok
|
|
end
|
|
end
|
|
|
|
defp should_trigger_discovery?(device, old_snmp, old_snmp_version, old_snmp_port) do
|
|
cond do
|
|
# SNMP was just enabled
|
|
device.snmp_enabled && !old_snmp ->
|
|
true
|
|
|
|
# SNMP version changed while enabled
|
|
device.snmp_enabled && device.snmp_version != old_snmp_version ->
|
|
true
|
|
|
|
# SNMP port changed while enabled
|
|
device.snmp_enabled && device.snmp_port != old_snmp_port ->
|
|
true
|
|
|
|
# Otherwise no discovery needed
|
|
true ->
|
|
false
|
|
end
|
|
end
|
|
|
|
## Events
|
|
|
|
@doc """
|
|
Creates an device event.
|
|
"""
|
|
def create_event(attrs) do
|
|
%Event{}
|
|
|> Event.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of events for device, ordered by most recent first.
|
|
"""
|
|
def list_devices_events(device_id, limit \\ 100) do
|
|
Repo.all(
|
|
from(e in Event,
|
|
where: e.device_id == ^device_id,
|
|
order_by: [desc: e.occurred_at],
|
|
limit: ^limit
|
|
)
|
|
)
|
|
end
|
|
|
|
## Display Order Management
|
|
|
|
@doc """
|
|
Reorders a device to a new position within its site.
|
|
|
|
Updates display_order for all devices in the site to maintain continuous ordering (1, 2, 3, ...).
|
|
The new_position is 1-based (1 = first position).
|
|
|
|
Returns {:ok, device} on success, {:error, changeset} on failure.
|
|
"""
|
|
def reorder_device(device_id, new_position) when is_integer(new_position) and new_position > 0 do
|
|
Repo.transaction(fn ->
|
|
device = Repo.get!(DeviceSchema, device_id)
|
|
site_id = device.site_id
|
|
|
|
# Get all devices in site excluding the one being moved, in current order
|
|
other_devices =
|
|
Repo.all(
|
|
from(d in DeviceSchema,
|
|
where: d.site_id == ^site_id and d.id != ^device_id,
|
|
order_by: [asc: d.display_order, asc: d.name]
|
|
)
|
|
)
|
|
|
|
# Insert at new position (1-based index, convert to 0-based for List.insert_at)
|
|
new_order = List.insert_at(other_devices, new_position - 1, device)
|
|
|
|
# Update display_order for all devices
|
|
new_order
|
|
|> Enum.with_index(1)
|
|
|> Enum.each(fn {d, order} ->
|
|
Repo.update!(Ecto.Changeset.change(d, display_order: order))
|
|
end)
|
|
|
|
Repo.get!(DeviceSchema, device_id)
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Reset all devices in organization to alphabetical order.
|
|
|
|
Clears display_order field for all devices, allowing them to fall back to alphabetical sorting.
|
|
"""
|
|
def reset_organization_device_order(organization_id) do
|
|
Repo.update_all(
|
|
from(d in DeviceSchema,
|
|
join: s in assoc(d, :site),
|
|
where: s.organization_id == ^organization_id
|
|
),
|
|
set: [display_order: nil, updated_at: DateTime.truncate(DateTime.utc_now(), :second)]
|
|
)
|
|
end
|
|
end
|