Graham McIntire
2528e78ff9
Adds 7 new composable Query modules following the existing DeviceQuery/AlertQuery pattern. Each exposes base/0 and for_organization/1,2 plus filters used by 2+ callers. New modules: - Towerops.OnCall.ScheduleQuery - Towerops.OnCall.EscalationPolicyQuery - Towerops.Maintenance.MaintenanceWindowQuery - Towerops.Organizations.MembershipQuery - Towerops.Organizations.InvitationQuery - Towerops.Agents.AgentTokenQuery - Towerops.Gaiia.DeviceSubscriberLinkQuery Refactors callsites in alerts, agents, devices, gaiia, maintenance, on_call, organizations, search, and subscription_limits to use existing or new Query modules instead of inline 'where: x.organization_id == ^...' clauses.
1127 lines
35 KiB
Elixir
1127 lines
35 KiB
Elixir
defmodule Towerops.Agents do
|
|
@moduledoc """
|
|
The Agents context for managing remote SNMP polling agents.
|
|
|
|
This context provides functions for creating and managing agent authentication tokens,
|
|
assigning devices to agents, and verifying agent credentials.
|
|
"""
|
|
|
|
import Ecto.Query, warn: false
|
|
|
|
alias Towerops.Agents.AgentAssignment
|
|
alias Towerops.Agents.AgentCache
|
|
alias Towerops.Agents.AgentToken
|
|
alias Towerops.Agents.AgentTokenQuery
|
|
alias Towerops.Agents.ReleaseChecker
|
|
alias Towerops.Devices.Device
|
|
alias Towerops.Repo
|
|
|
|
require Logger
|
|
|
|
## Token management
|
|
|
|
@doc """
|
|
Creates a new agent token for an organization.
|
|
|
|
Returns {:ok, agent_token, token_string} where token_string is the plain token
|
|
that should be shown to the user ONCE and never stored.
|
|
|
|
## Examples
|
|
|
|
iex> create_agent_token(org_id, "Datacenter Agent")
|
|
{:ok, %AgentToken{}, "abc123..."}
|
|
|
|
iex> create_agent_token(nil, "")
|
|
{:error, %Ecto.Changeset{}}
|
|
|
|
"""
|
|
@spec create_agent_token(String.t(), String.t()) :: {:ok, AgentToken.t(), String.t()} | {:error, Ecto.Changeset.t()}
|
|
def create_agent_token(organization_id, name) do
|
|
{token, changeset} = AgentToken.build_token(organization_id, name)
|
|
|
|
case Repo.insert(changeset) do
|
|
{:ok, agent_token} ->
|
|
broadcast_agent_change(:agent_created, agent_token)
|
|
{:ok, agent_token, token}
|
|
|
|
{:error, changeset} ->
|
|
{:error, changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Creates a new cloud poller agent token (superadmin only).
|
|
|
|
Cloud pollers are application-wide agents that are not tied to any organization
|
|
and can poll devices for any organization.
|
|
|
|
Returns {:ok, agent_token, token_string} where token_string is the plain token
|
|
that should be shown to the user ONCE and never stored.
|
|
|
|
## Examples
|
|
|
|
iex> create_cloud_poller("Cloud Poller 1")
|
|
{:ok, %AgentToken{is_cloud_poller: true}, "abc123..."}
|
|
|
|
"""
|
|
@spec create_cloud_poller(String.t()) :: {:ok, AgentToken.t(), String.t()} | {:error, Ecto.Changeset.t()}
|
|
def create_cloud_poller(name) do
|
|
{token, changeset} = AgentToken.build_token(nil, name, is_cloud_poller: true)
|
|
|
|
case Repo.insert(changeset) do
|
|
{:ok, agent_token} ->
|
|
broadcast_agent_change(:agent_created, agent_token)
|
|
{:ok, agent_token, token}
|
|
|
|
{:error, changeset} ->
|
|
{:error, changeset}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Lists all agent tokens for an organization.
|
|
|
|
Returns a list of agent tokens ordered by creation date (newest first).
|
|
Does not include cloud pollers.
|
|
|
|
## Examples
|
|
|
|
iex> list_organization_agent_tokens(org_id)
|
|
[%AgentToken{}, ...]
|
|
|
|
"""
|
|
@spec list_organization_agent_tokens(String.t()) :: [AgentToken.t()]
|
|
def list_organization_agent_tokens(organization_id) do
|
|
organization_id
|
|
|> AgentTokenQuery.for_organization()
|
|
|> AgentTokenQuery.excluding_cloud_pollers()
|
|
|> order_by([t], desc: t.inserted_at)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Lists all agent tokens across all organizations, including cloud pollers.
|
|
|
|
Returns a list of agent tokens with organization preloaded, ordered by
|
|
creation date (newest first). Used by the admin agents page.
|
|
|
|
## Examples
|
|
|
|
iex> list_all_agent_tokens()
|
|
[%AgentToken{organization: %Organization{}}, ...]
|
|
|
|
"""
|
|
@spec list_all_agent_tokens() :: [AgentToken.t()]
|
|
def list_all_agent_tokens do
|
|
AgentToken
|
|
|> order_by([t], desc: t.inserted_at)
|
|
|> preload(:organization)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Lists all cloud poller agent tokens (superadmin only).
|
|
|
|
Returns a list of cloud poller agent tokens ordered by creation date (newest first).
|
|
|
|
## Examples
|
|
|
|
iex> list_cloud_pollers()
|
|
[%AgentToken{is_cloud_poller: true}, ...]
|
|
|
|
"""
|
|
@spec list_cloud_pollers() :: [AgentToken.t()]
|
|
def list_cloud_pollers do
|
|
AgentToken
|
|
|> where([t], t.is_cloud_poller == true)
|
|
|> order_by([t], desc: t.inserted_at)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Counts the number of devices directly assigned to an agent.
|
|
|
|
This only counts devices with explicit assignments, not devices
|
|
that inherit the agent from site or organization defaults.
|
|
|
|
## Examples
|
|
|
|
iex> count_assigned_devices(agent_token_id)
|
|
5
|
|
|
|
"""
|
|
@spec count_assigned_devices(Ecto.UUID.t()) :: non_neg_integer()
|
|
def count_assigned_devices(agent_token_id) do
|
|
AgentAssignment
|
|
|> where([a], a.agent_token_id == ^agent_token_id and a.enabled == true)
|
|
|> Repo.aggregate(:count)
|
|
end
|
|
|
|
@doc """
|
|
Batch version of `count_assigned_devices/1`. Given a list of agent token ids,
|
|
returns a `%{agent_token_id => count}` map in a single query. Token ids
|
|
with no assignments are absent from the map (caller should default to 0).
|
|
"""
|
|
@spec count_assigned_devices_batch([Ecto.UUID.t()]) :: %{Ecto.UUID.t() => non_neg_integer()}
|
|
def count_assigned_devices_batch([]), do: %{}
|
|
|
|
def count_assigned_devices_batch(agent_token_ids) when is_list(agent_token_ids) do
|
|
AgentAssignment
|
|
|> where([a], a.agent_token_id in ^agent_token_ids and a.enabled == true)
|
|
|> group_by([a], a.agent_token_id)
|
|
|> select([a], {a.agent_token_id, count(a.id)})
|
|
|> Repo.all()
|
|
|> Map.new()
|
|
end
|
|
|
|
@doc """
|
|
Gets a single agent token by ID.
|
|
|
|
Raises `Ecto.NoResultsError` if the token does not exist.
|
|
|
|
## Examples
|
|
|
|
iex> get_agent_token!(id)
|
|
%AgentToken{}
|
|
|
|
iex> get_agent_token!("nonexistent")
|
|
** (Ecto.NoResultsError)
|
|
|
|
"""
|
|
@spec get_agent_token!(Ecto.UUID.t()) :: AgentToken.t()
|
|
def get_agent_token!(id), do: Repo.get!(AgentToken, id)
|
|
|
|
@doc """
|
|
Updates the heartbeat timestamp and metadata for an agent token.
|
|
|
|
This is called automatically when an agent makes an API request.
|
|
|
|
## Examples
|
|
|
|
iex> update_agent_token_heartbeat(token_id, "192.168.1.1", %{version: "0.1.0"})
|
|
{1, nil}
|
|
|
|
"""
|
|
@spec update_agent_token_heartbeat(Ecto.UUID.t(), String.t() | nil, map() | nil) ::
|
|
{non_neg_integer(), nil | [term()]}
|
|
def update_agent_token_heartbeat(agent_token_id, ip_address, metadata \\ nil) do
|
|
now = Towerops.Time.now()
|
|
|
|
# Only update metadata if explicitly provided (not nil and not empty)
|
|
updates =
|
|
case metadata do
|
|
nil -> [last_seen_at: now, last_ip: ip_address]
|
|
meta when map_size(meta) > 0 -> [last_seen_at: now, last_ip: ip_address, metadata: meta]
|
|
_ -> [last_seen_at: now, last_ip: ip_address]
|
|
end
|
|
|
|
result =
|
|
AgentToken
|
|
|> where([t], t.id == ^agent_token_id)
|
|
|> Repo.update_all(set: updates)
|
|
|
|
# Auto-dismiss agent offline insights when agent comes back online
|
|
dismiss_agent_offline_insights(agent_token_id)
|
|
|
|
result
|
|
end
|
|
|
|
defp dismiss_agent_offline_insights(agent_token_id) do
|
|
alias Towerops.Preseem.Insight
|
|
|
|
now = Towerops.Time.now()
|
|
|
|
{count, _} =
|
|
Insight
|
|
|> where(agent_token_id: ^agent_token_id, type: "agent_offline", status: "active")
|
|
|> Repo.update_all(set: [status: "dismissed", dismissed_at: now])
|
|
|
|
if count > 0 do
|
|
Logger.info("Auto-dismissed #{count} agent offline insight(s) for agent #{agent_token_id}")
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Verifies an agent token and returns the agent token record if valid.
|
|
|
|
Returns {:ok, agent_token} if the token is valid and enabled,
|
|
otherwise {:error, :invalid_token}.
|
|
|
|
## Examples
|
|
|
|
iex> verify_agent_token("valid_token")
|
|
{:ok, %AgentToken{}}
|
|
|
|
iex> verify_agent_token("invalid_token")
|
|
{:error, :invalid_token}
|
|
|
|
"""
|
|
@spec verify_agent_token(String.t()) :: {:ok, AgentToken.t()} | {:error, :invalid_token}
|
|
def verify_agent_token(token) do
|
|
with {:ok, token_string} <- AgentToken.verify_token(token),
|
|
%AgentToken{} = agent_token <-
|
|
Repo.get_by(AgentToken, token: token_string, enabled: true) do
|
|
{:ok, agent_token}
|
|
else
|
|
_ -> {:error, :invalid_token}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Updates an agent token with the given attributes.
|
|
|
|
Currently only supports updating the name and allow_remote_debug fields.
|
|
Other fields like the token itself cannot be changed.
|
|
|
|
## Examples
|
|
|
|
iex> update_agent_token(agent_token, %{name: "New Name"})
|
|
{:ok, %AgentToken{name: "New Name"}}
|
|
|
|
iex> update_agent_token(agent_token, %{name: ""})
|
|
{:error, %Ecto.Changeset{}}
|
|
|
|
"""
|
|
@spec update_agent_token(AgentToken.t(), map()) :: {:ok, AgentToken.t()} | {:error, Ecto.Changeset.t()}
|
|
def update_agent_token(%AgentToken{} = agent_token, attrs) do
|
|
result =
|
|
agent_token
|
|
|> AgentToken.update_changeset(attrs)
|
|
|> Repo.update()
|
|
|
|
# If token was disabled, disconnect any active agent channel
|
|
with {:ok, updated} <- result,
|
|
true <- agent_token.enabled == true and updated.enabled == false do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{updated.id}:lifecycle",
|
|
:token_disabled
|
|
)
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
@doc """
|
|
Requests an agent to restart by broadcasting on its lifecycle topic.
|
|
|
|
The agent channel picks this up and pushes a "restart" event to the agent,
|
|
which exits cleanly so Docker can restart it.
|
|
"""
|
|
@spec restart_agent(binary()) :: :ok | {:error, term()}
|
|
def restart_agent(agent_token_id) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token_id}:lifecycle",
|
|
:restart_requested
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Requests an agent to self-update by broadcasting the download URL and checksum
|
|
on its lifecycle topic.
|
|
|
|
The agent channel picks this up and pushes an "update" event to the agent,
|
|
which downloads the binary, verifies the checksum, and replaces itself.
|
|
"""
|
|
@spec update_agent(binary(), String.t(), String.t()) :: :ok | {:error, term()}
|
|
def update_agent(agent_token_id, url, checksum) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token_id}:lifecycle",
|
|
{:update_requested, url, checksum}
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns all agent tokens eligible for mass update.
|
|
|
|
An agent is updatable if it is enabled and was last seen within the past 10 minutes.
|
|
Includes both organization agents and cloud pollers.
|
|
"""
|
|
@spec list_updatable_agents() :: [AgentToken.t()]
|
|
def list_updatable_agents do
|
|
cutoff = DateTime.add(DateTime.utc_now(), -10, :minute)
|
|
|
|
AgentToken
|
|
|> where([t], t.enabled == true and not is_nil(t.last_seen_at) and t.last_seen_at > ^cutoff)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Broadcasts an update command to all connected agents.
|
|
|
|
1. Invalidates the release cache
|
|
2. Fetches the latest release from GitHub
|
|
3. Finds all updatable agents
|
|
4. Sends update command to each agent matching their architecture
|
|
|
|
Returns `{:ok, %{notified: N, skipped: N, version: String.t()}}` on success.
|
|
"""
|
|
@spec broadcast_mass_update() ::
|
|
{:ok, %{notified: non_neg_integer(), skipped: non_neg_integer(), version: String.t()}}
|
|
| {:error, term()}
|
|
def broadcast_mass_update do
|
|
ReleaseChecker.invalidate_cache()
|
|
|
|
case ReleaseChecker.get_latest_release() do
|
|
{:ok, %{version: version, assets: assets}} ->
|
|
agents = list_updatable_agents()
|
|
{notified, skipped} = notify_agents(agents, assets)
|
|
|
|
{:ok, %{notified: notified, skipped: skipped, version: version}}
|
|
|
|
{:error, reason} ->
|
|
Logger.error("Failed to fetch latest release for mass update: #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
|
|
defp notify_agents(agents, assets) do
|
|
Enum.reduce(agents, {0, 0}, fn agent, {notified, skipped} ->
|
|
arch = get_in(agent.metadata, ["arch"]) || "x86_64"
|
|
|
|
case ReleaseChecker.asset_for_arch(assets, arch) do
|
|
{:ok, asset} ->
|
|
_ = update_agent(agent.id, asset.url, asset.checksum)
|
|
{notified + 1, skipped}
|
|
|
|
{:error, :no_matching_asset} ->
|
|
Logger.warning("No matching asset for agent #{agent.id} (arch: #{arch})")
|
|
{notified, skipped + 1}
|
|
end
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Toggles remote debugging for an agent.
|
|
|
|
Creates an audit log entry for accountability.
|
|
|
|
## Examples
|
|
|
|
iex> toggle_agent_debug(agent_token, true, user)
|
|
{:ok, %AgentToken{allow_remote_debug: true}}
|
|
|
|
iex> toggle_agent_debug(agent_token, false, user)
|
|
{:ok, %AgentToken{allow_remote_debug: false}}
|
|
|
|
"""
|
|
@spec toggle_agent_debug(AgentToken.t(), boolean(), Towerops.Accounts.User.t()) ::
|
|
{:ok, AgentToken.t()} | {:error, Ecto.Changeset.t()}
|
|
def toggle_agent_debug(%AgentToken{} = agent_token, enabled, user) do
|
|
case update_agent_token(agent_token, %{allow_remote_debug: enabled}) do
|
|
{:ok, updated_agent} ->
|
|
# Create audit log
|
|
Towerops.Admin.create_audit_log(%{
|
|
superuser_id: user.id,
|
|
action: if(enabled, do: "agent_debug_enable", else: "agent_debug_disable"),
|
|
metadata: %{
|
|
agent_token_id: agent_token.id,
|
|
agent_name: agent_token.name,
|
|
organization_id: agent_token.organization_id
|
|
}
|
|
})
|
|
|
|
{:ok, updated_agent}
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Revokes an agent token by setting enabled to false.
|
|
|
|
## Examples
|
|
|
|
iex> revoke_agent_token(id)
|
|
{:ok, %AgentToken{enabled: false}}
|
|
|
|
"""
|
|
@spec revoke_agent_token(Ecto.UUID.t()) :: {:ok, AgentToken.t()} | {:error, Ecto.Changeset.t()}
|
|
def revoke_agent_token(id) do
|
|
AgentToken
|
|
|> Repo.get!(id)
|
|
|> Ecto.Changeset.change(enabled: false)
|
|
|> Repo.update()
|
|
end
|
|
|
|
@doc """
|
|
Deletes an agent token and cleans up all references.
|
|
|
|
This function:
|
|
1. Removes all direct device assignments for this agent
|
|
2. Removes this agent from any sites using it as default
|
|
3. Removes this agent from any organizations using it as default
|
|
4. Deletes the agent token
|
|
|
|
Devices that were assigned to this agent will fall back to:
|
|
- Site-level agent (if configured)
|
|
- Organization-level default agent (if configured)
|
|
- Cloud polling (if no fallback exists)
|
|
|
|
## Examples
|
|
|
|
iex> delete_agent_token(id)
|
|
{:ok, %AgentToken{}}
|
|
|
|
"""
|
|
@spec delete_agent_token(Ecto.UUID.t()) :: {:ok, AgentToken.t()} | {:error, any()}
|
|
def delete_agent_token(id) do
|
|
result =
|
|
Repo.transaction(fn ->
|
|
agent_token = Repo.get!(AgentToken, id)
|
|
|
|
# 1. Remove all direct device assignments
|
|
Repo.delete_all(from(a in AgentAssignment, where: a.agent_token_id == ^id))
|
|
|
|
# 2. Remove this agent from any sites using it as default
|
|
Repo.update_all(from(s in Towerops.Sites.Site, where: s.agent_token_id == ^id), set: [agent_token_id: nil])
|
|
|
|
# 3. Remove this agent from any organizations using it as default
|
|
Repo.update_all(from(o in Towerops.Organizations.Organization, where: o.default_agent_token_id == ^id),
|
|
set: [default_agent_token_id: nil]
|
|
)
|
|
|
|
# 4. Delete the agent token
|
|
Repo.delete!(agent_token)
|
|
|
|
agent_token
|
|
end)
|
|
|
|
case result do
|
|
{:ok, agent_token} ->
|
|
# Broadcast only after the transaction commits — otherwise a rollback
|
|
# would leave subscribers acting on a deletion that never happened.
|
|
_ =
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{id}:lifecycle",
|
|
:token_disabled
|
|
)
|
|
|
|
broadcast_agent_change(:agent_deleted, agent_token)
|
|
{:ok, agent_token}
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
## Cloud poller queries
|
|
|
|
@doc """
|
|
Lists all online cloud poller agent tokens.
|
|
|
|
Returns cloud pollers that are enabled and have been seen within the last 5 minutes.
|
|
"""
|
|
@spec list_online_cloud_pollers() :: [AgentToken.t()]
|
|
def list_online_cloud_pollers do
|
|
five_minutes_ago = DateTime.add(DateTime.utc_now(), -5, :minute)
|
|
|
|
AgentToken
|
|
|> where([t], t.is_cloud_poller == true and t.enabled == true)
|
|
|> where([t], not is_nil(t.last_seen_at) and t.last_seen_at > ^five_minutes_ago)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Lists all devices that should be polled by cloud pollers.
|
|
|
|
Returns devices whose effective agent is either a cloud poller or no agent at all,
|
|
filtered to only monitoring-enabled devices.
|
|
"""
|
|
@spec list_cloud_polled_devices() :: [Device.t()]
|
|
def list_cloud_polled_devices do
|
|
Repo.all(cloud_polled_devices_query())
|
|
end
|
|
|
|
defp cloud_polled_devices_query do
|
|
from(d in Device,
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == d.id and aa.enabled == true,
|
|
left_join: device_token in AgentToken,
|
|
on: device_token.id == aa.agent_token_id,
|
|
left_join: s in assoc(d, :site),
|
|
left_join: site_token in AgentToken,
|
|
on: site_token.id == s.agent_token_id,
|
|
left_join: o in assoc(d, :organization),
|
|
left_join: org_token in AgentToken,
|
|
on: org_token.id == o.default_agent_token_id,
|
|
where: d.monitoring_enabled == true,
|
|
where: ^cloud_polled_condition()
|
|
)
|
|
end
|
|
|
|
# Device is cloud-polled if its effective agent is a cloud poller or no agent exists.
|
|
# Split into separate dynamic conditions to satisfy credo complexity limits.
|
|
defp cloud_polled_condition do
|
|
device_cloud = device_assigned_to_cloud()
|
|
site_cloud = site_default_is_cloud()
|
|
org_cloud = org_default_is_cloud()
|
|
no_agent = no_agent_at_any_level()
|
|
|
|
dynamic([], ^device_cloud or ^site_cloud or ^org_cloud or ^no_agent)
|
|
end
|
|
|
|
defp device_assigned_to_cloud do
|
|
dynamic(
|
|
[d, aa, device_token, s, site_token, o, org_token],
|
|
not is_nil(aa.id) and device_token.is_cloud_poller == true
|
|
)
|
|
end
|
|
|
|
defp site_default_is_cloud do
|
|
dynamic(
|
|
[d, aa, device_token, s, site_token, o, org_token],
|
|
is_nil(aa.id) and not is_nil(s.agent_token_id) and site_token.is_cloud_poller == true
|
|
)
|
|
end
|
|
|
|
defp org_default_is_cloud do
|
|
dynamic(
|
|
[d, aa, device_token, s, site_token, o, org_token],
|
|
is_nil(aa.id) and (is_nil(s.id) or is_nil(s.agent_token_id)) and
|
|
not is_nil(o.default_agent_token_id) and org_token.is_cloud_poller == true
|
|
)
|
|
end
|
|
|
|
defp no_agent_at_any_level do
|
|
dynamic(
|
|
[d, aa, device_token, s, site_token, o, org_token],
|
|
is_nil(aa.id) and (is_nil(s.id) or is_nil(s.agent_token_id)) and
|
|
is_nil(o.default_agent_token_id)
|
|
)
|
|
end
|
|
|
|
## Assignment management
|
|
|
|
@doc """
|
|
Assigns device to an agent.
|
|
|
|
Creates an agent assignment that links device to an agent token.
|
|
Devices can only be assigned to one agent at a time due to unique constraint.
|
|
|
|
## Examples
|
|
|
|
iex> assign_device_to_agent(agent_id, device_id)
|
|
{:ok, %AgentAssignment{}}
|
|
|
|
iex> assign_device_to_agent(agent_id, already_assigned_device_id)
|
|
{:error, %Ecto.Changeset{}}
|
|
|
|
"""
|
|
@spec assign_device_to_agent(Ecto.UUID.t(), Ecto.UUID.t()) ::
|
|
{:ok, AgentAssignment.t()} | {:error, Ecto.Changeset.t()}
|
|
def assign_device_to_agent(agent_token_id, device_id) do
|
|
result =
|
|
%AgentAssignment{}
|
|
|> AgentAssignment.changeset(%{
|
|
agent_token_id: agent_token_id,
|
|
device_id: device_id
|
|
})
|
|
|> Repo.insert()
|
|
|
|
case result do
|
|
{:ok, assignment} ->
|
|
_ = broadcast_assignment_change(agent_token_id, :device_assigned)
|
|
{:ok, assignment}
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Removes the agent assignment for a piece of device.
|
|
|
|
## Examples
|
|
|
|
iex> unassign_device(device_id)
|
|
{1, nil}
|
|
|
|
"""
|
|
@spec unassign_device(Ecto.UUID.t()) :: {non_neg_integer(), nil | [term()]}
|
|
def unassign_device(device_id) do
|
|
# Get the current assignment to know which agent to notify
|
|
current_assignment = get_device_assignment(device_id)
|
|
|
|
result =
|
|
AgentAssignment
|
|
|> where([a], a.device_id == ^device_id)
|
|
|> Repo.delete_all()
|
|
|
|
# Notify the agent that was previously assigned
|
|
_ =
|
|
if current_assignment do
|
|
broadcast_assignment_change(current_assignment.agent_token_id, :device_unassigned)
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
@doc """
|
|
Lists all devices assigned to an agent.
|
|
|
|
Returns device records with preloaded site and SNMP device associations.
|
|
|
|
## Examples
|
|
|
|
iex> list_agent_devices(agent_token_id)
|
|
[%Device{site: %Site{}, snmp_device: %Device{sensors: [...], interfaces: [...]}}]
|
|
|
|
"""
|
|
@spec list_agent_devices(Ecto.UUID.t()) :: [Device.t()]
|
|
def list_agent_devices(agent_token_id) do
|
|
Repo.all(
|
|
from(e in Device,
|
|
join: a in AgentAssignment,
|
|
on: a.device_id == e.id,
|
|
where: a.agent_token_id == ^agent_token_id and a.enabled == true,
|
|
preload: [:site, snmp_device: [:sensors, :interfaces]]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Lists all devices that should be polled by an agent.
|
|
|
|
This includes devices that is:
|
|
1. Directly assigned to this agent
|
|
2. At a site that has this agent as default
|
|
3. In an organization that has this agent as default (and not overridden by site or device)
|
|
4. This agent is the global default cloud poller (and device has no assignment at any level)
|
|
|
|
Returns device records with preloaded associations, filtered to devices where
|
|
SNMP polling is enabled OR ICMP monitoring is enabled (or both).
|
|
|
|
## Examples
|
|
|
|
iex> list_agent_polling_targets(agent_token_id)
|
|
[%Device{snmp_enabled: true, site: %Site{}, snmp_device: %Device{sensors: [...], interfaces: [...]}}]
|
|
|
|
"""
|
|
@spec list_agent_polling_targets(Ecto.UUID.t()) :: [Device.t()]
|
|
def list_agent_polling_targets(agent_token_id) do
|
|
global_default = Towerops.Settings.get_global_default_cloud_poller()
|
|
is_global_default = agent_token_id == global_default
|
|
|
|
query =
|
|
from(e in Device,
|
|
left_join: s in assoc(e, :site),
|
|
join: o in assoc(e, :organization),
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == e.id and aa.enabled == true,
|
|
# Detect disabled assignments for this specific agent so cascade
|
|
# doesn't route devices that were explicitly disabled
|
|
left_join: disabled in AgentAssignment,
|
|
on:
|
|
disabled.device_id == e.id and disabled.agent_token_id == ^agent_token_id and
|
|
disabled.enabled == false,
|
|
where: e.snmp_enabled == true or e.monitoring_enabled == true,
|
|
preload: [
|
|
:agent_assignments,
|
|
:organization,
|
|
site: :organization,
|
|
snmp_device: [
|
|
sensors: ^from(s in Towerops.Snmp.Sensor, where: s.monitored == true),
|
|
interfaces: ^from(i in Towerops.Snmp.Interface, where: i.monitored == true)
|
|
]
|
|
]
|
|
)
|
|
|
|
query
|
|
|> where_agent_matches(agent_token_id, is_global_default)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Counts devices polled by an agent (matches the same cascade as
|
|
`list_agent_polling_targets/1`) without preloading any assocs.
|
|
|
|
Use this for UI counters instead of `length(list_agent_polling_targets(id))`,
|
|
which materializes every device + sensors + interfaces.
|
|
"""
|
|
@spec count_agent_polling_targets(Ecto.UUID.t()) :: non_neg_integer()
|
|
def count_agent_polling_targets(agent_token_id) do
|
|
global_default = Towerops.Settings.get_global_default_cloud_poller()
|
|
is_global_default = agent_token_id == global_default
|
|
|
|
query =
|
|
from(e in Device,
|
|
left_join: s in assoc(e, :site),
|
|
join: o in assoc(e, :organization),
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == e.id and aa.enabled == true,
|
|
left_join: disabled in AgentAssignment,
|
|
on:
|
|
disabled.device_id == e.id and disabled.agent_token_id == ^agent_token_id and
|
|
disabled.enabled == false,
|
|
where: e.snmp_enabled == true or e.monitoring_enabled == true
|
|
)
|
|
|
|
query
|
|
|> where_agent_matches(agent_token_id, is_global_default)
|
|
|> select([e], count(e.id))
|
|
|> Repo.one()
|
|
end
|
|
|
|
# Adds where clause to match devices assigned to this agent.
|
|
# Supports direct assignment, site cascade, org cascade, and global default.
|
|
# A disabled assignment for this agent blocks cascade to prevent polling
|
|
# devices that were explicitly disabled.
|
|
# credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
|
|
defp where_agent_matches(query, agent_token_id, is_global_default) do
|
|
from([e, s, o, aa, disabled] in query,
|
|
where:
|
|
aa.agent_token_id == ^agent_token_id or
|
|
(is_nil(aa.agent_token_id) and is_nil(disabled.id) and not is_nil(s.id) and
|
|
s.agent_token_id == ^agent_token_id) or
|
|
(is_nil(aa.agent_token_id) and is_nil(disabled.id) and
|
|
(is_nil(s.id) or is_nil(s.agent_token_id)) and
|
|
o.default_agent_token_id == ^agent_token_id) or
|
|
(^is_global_default and is_nil(aa.agent_token_id) and is_nil(disabled.id) and
|
|
(is_nil(s.id) or is_nil(s.agent_token_id)) and
|
|
is_nil(o.default_agent_token_id))
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Gets the agent assignment for a piece of device.
|
|
|
|
Returns the assignment record if it exists, nil otherwise.
|
|
|
|
## Examples
|
|
|
|
iex> get_device_assignment(device_id)
|
|
%AgentAssignment{}
|
|
|
|
iex> get_device_assignment(unassigned_device_id)
|
|
nil
|
|
|
|
"""
|
|
@spec get_device_assignment(Ecto.UUID.t()) :: AgentAssignment.t() | nil
|
|
def get_device_assignment(device_id) do
|
|
Repo.get_by(AgentAssignment, device_id: device_id)
|
|
end
|
|
|
|
@doc """
|
|
Updates or creates an device assignment.
|
|
|
|
If agent_token_id is nil, removes any existing assignment.
|
|
Otherwise, creates or updates the assignment.
|
|
|
|
## Examples
|
|
|
|
iex> update_device_assignment(device_id, agent_token_id)
|
|
{:ok, %AgentAssignment{}}
|
|
|
|
iex> update_device_assignment(device_id, nil)
|
|
{:ok, nil}
|
|
|
|
"""
|
|
@spec update_device_assignment(Ecto.UUID.t(), Ecto.UUID.t() | nil) ::
|
|
{:ok, AgentAssignment.t() | nil} | {:error, Ecto.Changeset.t()}
|
|
def update_device_assignment(device_id, nil) do
|
|
# Remove assignment if setting to nil
|
|
_ = unassign_device(device_id)
|
|
{:ok, nil}
|
|
end
|
|
|
|
def update_device_assignment(device_id, agent_token_id) do
|
|
case get_device_assignment(device_id) do
|
|
nil ->
|
|
# Create new assignment
|
|
assign_device_to_agent(agent_token_id, device_id)
|
|
|
|
existing ->
|
|
update_existing_assignment(existing, agent_token_id)
|
|
end
|
|
end
|
|
|
|
defp update_existing_assignment(existing, agent_token_id) do
|
|
old_agent_id = existing.agent_token_id
|
|
|
|
result =
|
|
existing
|
|
|> AgentAssignment.changeset(%{agent_token_id: agent_token_id})
|
|
|> Repo.update()
|
|
|
|
case result do
|
|
{:ok, assignment} ->
|
|
# Notify both the old and new agents if they're different
|
|
_ =
|
|
if old_agent_id != agent_token_id do
|
|
_ = broadcast_assignment_change(old_agent_id, :device_unassigned)
|
|
broadcast_assignment_change(agent_token_id, :device_assigned)
|
|
end
|
|
|
|
{:ok, assignment}
|
|
|
|
error ->
|
|
error
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Gets the effective agent token for a piece of device.
|
|
|
|
Resolves the agent token using the following hierarchy:
|
|
1. Device's direct agent assignment (highest priority)
|
|
2. Site's agent_token_id
|
|
3. Organization's default_agent_token_id
|
|
4. Global default cloud poller (application-wide fallback)
|
|
5. Cloud polling (nil - lowest priority)
|
|
|
|
Returns the agent_token_id if found, nil otherwise.
|
|
|
|
The device must be preloaded with `[:organization, site: [organization: :default_agent_token]]`.
|
|
|
|
## Examples
|
|
|
|
iex> device = Repo.preload(device, [:organization, site: [organization: :default_agent_token]])
|
|
iex> get_effective_agent_token(device)
|
|
"agent-token-uuid"
|
|
|
|
iex> get_effective_agent_token(equipment_without_any_assignment)
|
|
nil
|
|
|
|
"""
|
|
@spec get_effective_agent_token(Device.t()) :: Ecto.UUID.t() | nil
|
|
def get_effective_agent_token(%Device{} = device) do
|
|
# 1. Check device assignment
|
|
case get_device_assignment(device.id) do
|
|
%AgentAssignment{agent_token_id: agent_token_id} when not is_nil(agent_token_id) ->
|
|
agent_token_id
|
|
|
|
_ ->
|
|
get_fallback_agent_token(device)
|
|
end
|
|
end
|
|
|
|
# Checks site, organization, and global default in sequence
|
|
defp get_fallback_agent_token(device) do
|
|
{agent_token_id, _source} = get_fallback_agent_token_with_source(device)
|
|
agent_token_id
|
|
end
|
|
|
|
@doc """
|
|
Gets information about where the effective agent token comes from.
|
|
|
|
Returns a tuple of {agent_token_id, source} where source is one of:
|
|
- :device - Direct device assignment
|
|
- :site - Inherited from site
|
|
- :organization - Inherited from organization
|
|
- :global - Global default cloud poller
|
|
- :none - No agent assigned
|
|
|
|
## Examples
|
|
|
|
iex> get_effective_agent_token_with_source(device)
|
|
{"token-id", :device}
|
|
|
|
iex> get_effective_agent_token_with_source(equipment_no_assignment)
|
|
{nil, :none}
|
|
|
|
"""
|
|
@spec get_effective_agent_token_with_source(Device.t()) ::
|
|
{Ecto.UUID.t() | nil, :device | :site | :organization | :global | :none}
|
|
def get_effective_agent_token_with_source(%Device{} = device) do
|
|
# 1. Check device assignment
|
|
case get_device_assignment(device.id) do
|
|
%AgentAssignment{agent_token_id: agent_token_id} when not is_nil(agent_token_id) ->
|
|
{agent_token_id, :device}
|
|
|
|
_ ->
|
|
get_fallback_agent_token_with_source(device)
|
|
end
|
|
end
|
|
|
|
# Site-level agent (highest priority fallback)
|
|
defp get_fallback_agent_token_with_source(%{site: %{agent_token_id: id}} = _device) when not is_nil(id), do: {id, :site}
|
|
|
|
# Organization default via site
|
|
defp get_fallback_agent_token_with_source(%{site: %{organization: %{default_agent_token_id: id}}}) when not is_nil(id),
|
|
do: {id, :organization}
|
|
|
|
# Organization default directly (device has no site or site has no org reference)
|
|
defp get_fallback_agent_token_with_source(%{organization: %{default_agent_token_id: id}}) when not is_nil(id),
|
|
do: {id, :organization}
|
|
|
|
# Global default or none
|
|
defp get_fallback_agent_token_with_source(_device) do
|
|
case Towerops.Settings.get_global_default_cloud_poller() do
|
|
id when not is_nil(id) -> {id, :global}
|
|
_ -> {nil, :none}
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Checks if a device should be polled by Phoenix directly.
|
|
|
|
Returns false if the device is assigned to a non-cloud-poller agent
|
|
(i.e., a customer/local agent). Phoenix should never poll devices that
|
|
are assigned to local agents - those devices should only be polled by
|
|
their assigned agent.
|
|
|
|
Returns true if:
|
|
- No agent is assigned to the device
|
|
- The device is assigned to a cloud poller (Phoenix-hosted agent)
|
|
|
|
## Examples
|
|
|
|
iex> should_phoenix_poll_device?(device_with_local_agent)
|
|
false
|
|
|
|
iex> should_phoenix_poll_device?(device_with_cloud_poller)
|
|
true
|
|
|
|
iex> should_phoenix_poll_device?(device_no_agent)
|
|
true
|
|
|
|
"""
|
|
@spec should_phoenix_poll_device?(Device.t()) :: boolean()
|
|
def should_phoenix_poll_device?(%Device{id: device_id}) do
|
|
# Use a single atomic query to check assignment and cloud poller status
|
|
# This prevents race conditions where assignment changes between reads
|
|
# Checks device assignment first, then site default, then org default
|
|
query =
|
|
from d in Device,
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == d.id and aa.enabled == true,
|
|
left_join: device_token in AgentToken,
|
|
on: device_token.id == aa.agent_token_id,
|
|
left_join: s in assoc(d, :site),
|
|
left_join: site_token in AgentToken,
|
|
on: site_token.id == s.agent_token_id,
|
|
left_join: o in assoc(s, :organization),
|
|
left_join: org_token in AgentToken,
|
|
on: org_token.id == o.default_agent_token_id,
|
|
where: d.id == ^device_id,
|
|
select: %{
|
|
# Device-level assignment
|
|
has_device_assignment: not is_nil(aa.id),
|
|
device_is_cloud: device_token.is_cloud_poller,
|
|
# Site-level fallback
|
|
has_site_token: not is_nil(s.agent_token_id),
|
|
site_is_cloud: site_token.is_cloud_poller,
|
|
# Org-level fallback
|
|
has_org_token: not is_nil(o.default_agent_token_id),
|
|
org_is_cloud: org_token.is_cloud_poller
|
|
}
|
|
|
|
case Repo.one(query) do
|
|
nil ->
|
|
# Device doesn't exist
|
|
false
|
|
|
|
result ->
|
|
# Determine effective agent using precedence: device > site > org
|
|
cond do
|
|
# 1. Device-level assignment
|
|
result.has_device_assignment ->
|
|
# Device assigned - check if cloud poller (nil = deleted token, treat as cloud)
|
|
result.device_is_cloud != false
|
|
|
|
# 2. Site-level default
|
|
result.has_site_token ->
|
|
# Site has default - check if cloud poller
|
|
result.site_is_cloud != false
|
|
|
|
# 3. Organization-level default
|
|
result.has_org_token ->
|
|
# Org has default - check if cloud poller
|
|
result.org_is_cloud != false
|
|
|
|
# 4. No agent anywhere
|
|
true ->
|
|
# No agent assigned at any level - Phoenix polls
|
|
true
|
|
end
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Returns true if the device has any effective agent assigned (cloud or local)
|
|
at any level: device, site, organization, or global default.
|
|
|
|
Unlike `should_phoenix_poll_device?/1`, this makes no distinction between
|
|
cloud pollers and local agents. Any agent means the Go agent handles polling.
|
|
|
|
Used by the check executor to decide whether to skip server-side execution.
|
|
"""
|
|
@spec device_has_effective_agent?(Ecto.UUID.t()) :: boolean()
|
|
def device_has_effective_agent?(device_id) do
|
|
AgentCache.device_has_effective_agent?(device_id)
|
|
end
|
|
|
|
@doc """
|
|
Queries the database directly to check if a device has an effective agent.
|
|
|
|
This is the uncached version used by `AgentCache`. Callers should use
|
|
`device_has_effective_agent?/1` which reads from the ETS cache.
|
|
"""
|
|
@spec query_device_has_effective_agent?(Ecto.UUID.t()) :: boolean()
|
|
def query_device_has_effective_agent?(device_id) do
|
|
query =
|
|
from d in Device,
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == d.id and aa.enabled == true,
|
|
left_join: s in assoc(d, :site),
|
|
left_join: o in assoc(d, :organization),
|
|
where: d.id == ^device_id,
|
|
select: %{
|
|
has_device_assignment: not is_nil(aa.agent_token_id),
|
|
has_site_token: not is_nil(s.agent_token_id),
|
|
has_org_token: not is_nil(o.default_agent_token_id)
|
|
}
|
|
|
|
case Repo.one(query) do
|
|
nil ->
|
|
false
|
|
|
|
result ->
|
|
result.has_device_assignment or
|
|
result.has_site_token or
|
|
result.has_org_token or
|
|
AgentCache.global_default_cloud_poller() != nil
|
|
end
|
|
end
|
|
|
|
## PubSub notifications
|
|
|
|
@doc """
|
|
Broadcasts an assignment change to any connected agent channels.
|
|
|
|
This allows agents to immediately receive updated job lists when
|
|
devices are assigned or unassigned, without waiting for reconnection.
|
|
"""
|
|
@spec broadcast_assignment_change(Ecto.UUID.t() | nil, atom()) :: :ok
|
|
def broadcast_assignment_change(agent_token_id, event) when not is_nil(agent_token_id) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token_id}:assignments",
|
|
{:assignments_changed, event}
|
|
)
|
|
end
|
|
|
|
def broadcast_assignment_change(nil, _event), do: :ok
|
|
|
|
@doc """
|
|
Broadcasts agent lifecycle changes (created/deleted) to the admin agents topic.
|
|
|
|
Only the admin agents page subscribes to this topic, keeping these events
|
|
separate from the org-scoped "agents:health" topic.
|
|
"""
|
|
@spec broadcast_agent_change(atom(), AgentToken.t()) :: :ok
|
|
def broadcast_agent_change(event, %AgentToken{} = agent_token) when event in [:agent_created, :agent_deleted] do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"admin:agents",
|
|
{event, agent_token.id, agent_token.is_cloud_poller}
|
|
)
|
|
end
|
|
end
|