Graham McIntire
c1fca6276a
Broadcast agent lifecycle events (created/deleted) on a separate "admin:agents" topic so the admin page updates dynamically when agents are added or removed from any organization.
107 lines
3.4 KiB
Elixir
107 lines
3.4 KiB
Elixir
defmodule ToweropsWeb.Admin.AgentLive.Index do
|
|
@moduledoc false
|
|
use ToweropsWeb, :live_view
|
|
|
|
import ToweropsWeb.AgentLive.Helpers
|
|
|
|
alias Towerops.Agents
|
|
|
|
@impl true
|
|
def mount(_params, _session, socket) do
|
|
if connected?(socket) do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health")
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "admin:agents")
|
|
:timer.send_interval(1000, :tick)
|
|
end
|
|
|
|
agent_tokens = Agents.list_all_agent_tokens()
|
|
{cloud_pollers, org_agents} = Enum.split_with(agent_tokens, & &1.is_cloud_poller)
|
|
device_counts = calculate_device_counts(agent_tokens)
|
|
|
|
{:ok,
|
|
socket
|
|
|> assign(:page_title, "All Agents")
|
|
|> assign(:timezone, socket.assigns.current_scope.timezone)
|
|
|> stream(:cloud_pollers, cloud_pollers)
|
|
|> stream(:org_agents, org_agents)
|
|
|> assign(:has_cloud_pollers, cloud_pollers != [])
|
|
|> assign(:has_org_agents, org_agents != [])
|
|
|> assign(:device_counts, device_counts)
|
|
|> assign(:now, DateTime.utc_now())}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(:tick, socket) do
|
|
{:noreply, assign(socket, :now, DateTime.utc_now())}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agent_connected, agent_token_id, _organization_id}, socket) do
|
|
{:noreply, refresh_agent(socket, agent_token_id)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agent_disconnected, agent_token_id, _organization_id}, socket) do
|
|
{:noreply, refresh_agent(socket, agent_token_id)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agent_heartbeat, agent_token_id, _organization_id}, socket) do
|
|
{:noreply, refresh_agent(socket, agent_token_id)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agents_stale, _stale_agents}, socket) do
|
|
{:noreply, reload_all(socket)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agent_created, agent_token_id, _is_cloud_poller}, socket) do
|
|
{:noreply, refresh_agent(socket, agent_token_id)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:agent_deleted, _agent_token_id, _is_cloud_poller}, socket) do
|
|
{:noreply, reload_all(socket)}
|
|
end
|
|
|
|
defp refresh_agent(socket, agent_token_id) do
|
|
agent_token = agent_token_id |> Agents.get_agent_token!() |> Towerops.Repo.preload(:organization)
|
|
|
|
direct = Agents.count_assigned_devices(agent_token_id)
|
|
total = length(Agents.list_agent_polling_targets(agent_token_id))
|
|
|
|
{stream_name, section_assign} =
|
|
if agent_token.is_cloud_poller,
|
|
do: {:cloud_pollers, :has_cloud_pollers},
|
|
else: {:org_agents, :has_org_agents}
|
|
|
|
socket
|
|
|> stream_insert(stream_name, agent_token)
|
|
|> assign(section_assign, true)
|
|
|> assign(:device_counts, Map.put(socket.assigns.device_counts, agent_token_id, %{direct: direct, total: total}))
|
|
rescue
|
|
Ecto.NoResultsError -> socket
|
|
end
|
|
|
|
defp reload_all(socket) do
|
|
agent_tokens = Agents.list_all_agent_tokens()
|
|
{cloud_pollers, org_agents} = Enum.split_with(agent_tokens, & &1.is_cloud_poller)
|
|
device_counts = calculate_device_counts(agent_tokens)
|
|
|
|
socket
|
|
|> stream(:cloud_pollers, cloud_pollers, reset: true)
|
|
|> stream(:org_agents, org_agents, reset: true)
|
|
|> assign(:has_cloud_pollers, cloud_pollers != [])
|
|
|> assign(:has_org_agents, org_agents != [])
|
|
|> assign(:device_counts, device_counts)
|
|
end
|
|
|
|
defp calculate_device_counts(agent_tokens) do
|
|
Map.new(agent_tokens, fn t ->
|
|
direct = Agents.count_assigned_devices(t.id)
|
|
total = length(Agents.list_agent_polling_targets(t.id))
|
|
{t.id, %{direct: direct, total: total}}
|
|
end)
|
|
end
|
|
end
|