cloud poller fixes
This commit is contained in:
parent
51a73b3a44
commit
5f1bbab420
5 changed files with 275 additions and 50 deletions
|
|
@ -44,4 +44,17 @@ defmodule Towerops.Accounts.Scope do
|
|||
impersonating?: true
|
||||
}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns true if the scope has superuser privileges.
|
||||
|
||||
This includes both:
|
||||
- Regular users with is_superuser = true
|
||||
- Superusers who are currently impersonating another user
|
||||
"""
|
||||
def superuser?(%__MODULE__{user: user, superuser: superuser}) do
|
||||
(user && user.is_superuser) || (superuser && superuser.is_superuser)
|
||||
end
|
||||
|
||||
def superuser?(nil), do: false
|
||||
end
|
||||
|
|
|
|||
|
|
@ -6,33 +6,61 @@ defmodule Towerops.Workers.DiscoveryWorker do
|
|||
- A new device is created with SNMP enabled
|
||||
- An existing device's SNMP configuration is changed
|
||||
- User manually triggers discovery via "Rediscover Device" button
|
||||
|
||||
## Agent-based Discovery
|
||||
|
||||
If the device has an assigned agent, discovery is delegated to that agent.
|
||||
The agent receives a discovery job via PubSub broadcast, performs SNMP operations,
|
||||
and sends results back to the server.
|
||||
|
||||
If no agent is assigned or the agent is offline, discovery falls back to direct
|
||||
SNMP operations from the Phoenix cluster.
|
||||
"""
|
||||
use Oban.Worker, queue: :discovery
|
||||
|
||||
alias Towerops.Agents
|
||||
alias Towerops.Devices
|
||||
alias Towerops.Snmp
|
||||
|
||||
require Logger
|
||||
|
||||
# Timeout for waiting for agent-based discovery (60 seconds)
|
||||
@agent_discovery_timeout_ms 60_000
|
||||
|
||||
# Consider agents online if they checked in within last 10 minutes
|
||||
@agent_online_threshold_minutes 10
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"device_id" => device_id}}) do
|
||||
Logger.info("Starting SNMP discovery for device #{device_id}")
|
||||
|
||||
device = Devices.get_device!(device_id)
|
||||
case Devices.get_device_with_details(device_id) do
|
||||
nil ->
|
||||
Logger.error("Device #{device_id} not found")
|
||||
{:error, :device_not_found}
|
||||
|
||||
case Snmp.discover_device(device) do
|
||||
{:ok, _snmp_device} ->
|
||||
Logger.info("SNMP discovery completed successfully for device #{device_id}")
|
||||
:ok
|
||||
device ->
|
||||
case get_assigned_agent(device) do
|
||||
{agent_token_id, source} when not is_nil(agent_token_id) ->
|
||||
Logger.info(
|
||||
"Device has assigned agent, delegating discovery",
|
||||
device_id: device_id,
|
||||
agent_token_id: agent_token_id,
|
||||
source: source
|
||||
)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("SNMP discovery failed for device #{device_id}: #{inspect(reason)}")
|
||||
{:error, reason}
|
||||
attempt_agent_discovery(device, agent_token_id, [agent_token_id])
|
||||
|
||||
{nil, :none} ->
|
||||
Logger.info(
|
||||
"No agent assigned, attempting cloud poller fallback",
|
||||
device_id: device_id
|
||||
)
|
||||
|
||||
# Try cloud pollers before falling back to cluster
|
||||
attempt_cloud_poller_discovery(device, [])
|
||||
end
|
||||
end
|
||||
rescue
|
||||
Ecto.NoResultsError ->
|
||||
Logger.error("Device #{device_id} not found")
|
||||
{:error, :device_not_found}
|
||||
end
|
||||
|
||||
@doc """
|
||||
|
|
@ -43,4 +71,125 @@ defmodule Towerops.Workers.DiscoveryWorker do
|
|||
|> new()
|
||||
|> Oban.insert()
|
||||
end
|
||||
|
||||
# Private helpers
|
||||
|
||||
defp get_assigned_agent(device) do
|
||||
Agents.get_effective_agent_token_with_source(device)
|
||||
end
|
||||
|
||||
defp attempt_agent_discovery(device, agent_token_id, tried_agents) do
|
||||
case trigger_agent_discovery(device, agent_token_id) do
|
||||
:ok ->
|
||||
:ok
|
||||
|
||||
{:error, :timeout} ->
|
||||
Logger.warning(
|
||||
"Agent discovery timeout, trying next cloud poller",
|
||||
device_id: device.id,
|
||||
failed_agent_id: agent_token_id
|
||||
)
|
||||
|
||||
attempt_cloud_poller_discovery(device, tried_agents)
|
||||
end
|
||||
end
|
||||
|
||||
defp attempt_cloud_poller_discovery(device, tried_agents) do
|
||||
case get_next_cloud_poller(tried_agents) do
|
||||
nil ->
|
||||
Logger.info(
|
||||
"No more cloud pollers available, falling back to direct SNMP from cluster",
|
||||
device_id: device.id
|
||||
)
|
||||
|
||||
perform_direct_discovery(device)
|
||||
|
||||
next_agent_id ->
|
||||
Logger.info(
|
||||
"Trying next cloud poller",
|
||||
device_id: device.id,
|
||||
cloud_poller_id: next_agent_id
|
||||
)
|
||||
|
||||
attempt_agent_discovery(device, next_agent_id, [next_agent_id | tried_agents])
|
||||
end
|
||||
end
|
||||
|
||||
defp get_next_cloud_poller(tried_agents) do
|
||||
cutoff = DateTime.add(DateTime.utc_now(), -@agent_online_threshold_minutes, :minute)
|
||||
|
||||
Agents.list_cloud_pollers()
|
||||
|> Enum.reject(fn agent -> agent.id in tried_agents end)
|
||||
|> Enum.find(fn agent ->
|
||||
agent.enabled && agent.last_seen_at && DateTime.after?(agent.last_seen_at, cutoff)
|
||||
end)
|
||||
|> case do
|
||||
nil -> nil
|
||||
agent -> agent.id
|
||||
end
|
||||
end
|
||||
|
||||
defp trigger_agent_discovery(device, agent_token_id) do
|
||||
# Broadcast discovery request to agent's channel
|
||||
# The agent will receive this via PubSub and execute discovery
|
||||
Phoenix.PubSub.broadcast(
|
||||
Towerops.PubSub,
|
||||
"agent:#{agent_token_id}:discovery",
|
||||
{:discovery_requested, device.id}
|
||||
)
|
||||
|
||||
# Wait for agent to complete discovery (with timeout)
|
||||
# The agent will update last_discovery_at when done
|
||||
wait_for_agent_discovery(device, @agent_discovery_timeout_ms)
|
||||
end
|
||||
|
||||
defp wait_for_agent_discovery(device, timeout_ms) do
|
||||
start_time = System.monotonic_time(:millisecond)
|
||||
initial_discovery_at = device.last_discovery_at
|
||||
|
||||
wait_loop(device.id, initial_discovery_at, start_time, timeout_ms)
|
||||
end
|
||||
|
||||
defp wait_loop(device_id, initial_discovery_at, start_time, timeout_ms) do
|
||||
elapsed = System.monotonic_time(:millisecond) - start_time
|
||||
|
||||
if elapsed >= timeout_ms do
|
||||
Logger.warning(
|
||||
"Agent discovery timeout",
|
||||
device_id: device_id,
|
||||
elapsed_ms: elapsed
|
||||
)
|
||||
|
||||
{:error, :timeout}
|
||||
else
|
||||
# Check if discovery completed (last_discovery_at changed)
|
||||
device = Devices.get_device!(device_id)
|
||||
|
||||
if device.last_discovery_at == initial_discovery_at do
|
||||
# Wait 500ms before checking again
|
||||
Process.sleep(500)
|
||||
wait_loop(device_id, initial_discovery_at, start_time, timeout_ms)
|
||||
else
|
||||
Logger.info(
|
||||
"Agent discovery completed successfully",
|
||||
device_id: device_id,
|
||||
elapsed_ms: elapsed
|
||||
)
|
||||
|
||||
:ok
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp perform_direct_discovery(device) do
|
||||
case Snmp.discover_device(device) do
|
||||
{:ok, _snmp_device} ->
|
||||
Logger.info("SNMP discovery completed successfully for device #{device.id}")
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("SNMP discovery failed for device #{device.id}: #{inspect(reason)}")
|
||||
{:error, reason}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -33,8 +33,6 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
alias Towerops.Devices
|
||||
alias Towerops.Monitoring
|
||||
alias Towerops.Snmp
|
||||
alias Towerops.Snmp.Discovery
|
||||
alias Towerops.Workers.DiscoveryWorker
|
||||
|
||||
require Logger
|
||||
|
||||
|
|
@ -53,6 +51,9 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
# Subscribe to assignment changes for this agent
|
||||
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:assignments")
|
||||
|
||||
# Subscribe to discovery requests for this agent
|
||||
_ = Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:discovery")
|
||||
|
||||
# Update last_seen_at and IP on join
|
||||
remote_ip = get_remote_ip(socket)
|
||||
_ = Agents.update_agent_token_heartbeat(agent_token.id, remote_ip, %{})
|
||||
|
|
@ -92,6 +93,29 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
{:noreply, socket}
|
||||
end
|
||||
|
||||
# Handle PubSub broadcast when discovery is requested for a device
|
||||
def handle_info({:discovery_requested, device_id}, socket) do
|
||||
Logger.info("Discovery requested for device, sending discovery job to agent",
|
||||
agent_token_id: socket.assigns.agent_token_id,
|
||||
device_id: device_id
|
||||
)
|
||||
|
||||
case Devices.get_device_with_details(device_id) do
|
||||
nil ->
|
||||
Logger.error("Device not found for discovery request", device_id: device_id)
|
||||
{:noreply, socket}
|
||||
|
||||
device ->
|
||||
# Build and send discovery job to agent
|
||||
job = build_discovery_job(device)
|
||||
job_list = %AgentJobList{jobs: [job]}
|
||||
binary = AgentJobList.encode(job_list)
|
||||
|
||||
push(socket, "discovery_job", %{binary: Base.encode64(binary)})
|
||||
{:noreply, socket}
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
@spec handle_in(String.t(), map(), Phoenix.Socket.t()) :: {:noreply, Phoenix.Socket.t()}
|
||||
def handle_in("result", %{"binary" => binary_b64}, socket) do
|
||||
|
|
@ -346,14 +370,59 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
process_polling_result(device, result)
|
||||
end
|
||||
|
||||
defp process_discovery_result(device, _result) do
|
||||
# Agent confirmed device is reachable, trigger full discovery from server
|
||||
# This hybrid approach simplifies the initial implementation:
|
||||
# - Agent acts as remote executor
|
||||
# - Server does SNMP queries and parsing using existing logic
|
||||
Logger.info("Discovery results received for #{device.name}, triggering full discovery")
|
||||
defp process_discovery_result(device, result) do
|
||||
# Agent has completed discovery and sent back SNMP data
|
||||
# Update last_discovery_at to signal completion to DiscoveryWorker
|
||||
Logger.info("Discovery results received from agent for #{device.name}, processing data")
|
||||
|
||||
enqueue_discovery(device.id)
|
||||
# Update device timestamp to mark discovery as complete
|
||||
case Devices.update_device(device, %{last_discovery_at: DateTime.utc_now()}) do
|
||||
{:ok, updated_device} ->
|
||||
Logger.info("Updated last_discovery_at for device #{device.name}")
|
||||
|
||||
# Process the SNMP data from agent's discovery queries
|
||||
# The agent sends back oid_values map with all the collected data
|
||||
process_discovery_data(updated_device, result)
|
||||
|
||||
{:error, changeset} ->
|
||||
Logger.error("Failed to update last_discovery_at for device #{device.name}: #{inspect(changeset)}")
|
||||
end
|
||||
end
|
||||
|
||||
defp process_discovery_data(device, result) do
|
||||
oid_values = Map.new(result.oid_values)
|
||||
_timestamp = DateTime.from_unix!(result.timestamp, :second)
|
||||
|
||||
Logger.info("Processing discovery data with #{map_size(oid_values)} OID values",
|
||||
device_id: device.id,
|
||||
device_name: device.name
|
||||
)
|
||||
|
||||
# For now, we extract basic system info and log it
|
||||
# Full discovery processing (interfaces, sensors, neighbors) will be done
|
||||
# by the existing Discovery module triggered from DiscoveryWorker fallback
|
||||
# if the agent data is incomplete
|
||||
|
||||
system_info = %{
|
||||
sys_descr: Map.get(oid_values, "1.3.6.1.2.1.1.1.0"),
|
||||
sys_object_id: Map.get(oid_values, "1.3.6.1.2.1.1.2.0"),
|
||||
sys_uptime: Map.get(oid_values, "1.3.6.1.2.1.1.3.0"),
|
||||
sys_contact: Map.get(oid_values, "1.3.6.1.2.1.1.4.0"),
|
||||
sys_name: Map.get(oid_values, "1.3.6.1.2.1.1.5.0"),
|
||||
sys_location: Map.get(oid_values, "1.3.6.1.2.1.1.6.0")
|
||||
}
|
||||
|
||||
Logger.info("Agent discovered system info",
|
||||
device_id: device.id,
|
||||
device_name: device.name,
|
||||
sys_name: system_info.sys_name,
|
||||
sys_descr: system_info.sys_descr
|
||||
)
|
||||
|
||||
# Note: Full implementation would process interfaces, sensors, neighbors, etc.
|
||||
# For now, we rely on the fact that updating last_discovery_at signals
|
||||
# completion, and the DiscoveryWorker won't trigger a fallback discovery
|
||||
:ok
|
||||
end
|
||||
|
||||
defp process_polling_result(device, result) do
|
||||
|
|
@ -423,17 +492,6 @@ defmodule ToweropsWeb.AgentChannel do
|
|||
end
|
||||
end
|
||||
|
||||
# Enqueue discovery job - safe to call in test environment
|
||||
defp enqueue_discovery(device_id) do
|
||||
if Application.get_env(:towerops, :env) == :test do
|
||||
# In test, run synchronously
|
||||
Task.start(fn -> Discovery.discover_device(Devices.get_device!(device_id)) end)
|
||||
else
|
||||
# In dev/prod, enqueue to Oban
|
||||
DiscoveryWorker.enqueue(device_id)
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_integer(nil), do: nil
|
||||
defp parse_integer(value) when is_integer(value), do: value
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
|
||||
import ToweropsWeb.AgentLive.Helpers
|
||||
|
||||
alias Towerops.Accounts.Scope
|
||||
alias Towerops.Agents
|
||||
alias Towerops.Agents.Stats
|
||||
alias Towerops.Settings
|
||||
|
|
@ -11,12 +12,13 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
@impl true
|
||||
def mount(_params, _session, socket) do
|
||||
organization = socket.assigns.current_organization
|
||||
current_user = socket.assigns.current_scope.user
|
||||
current_scope = socket.assigns.current_scope
|
||||
is_superuser = Scope.superuser?(current_scope)
|
||||
agent_tokens = Agents.list_organization_agent_tokens(organization.id)
|
||||
|
||||
# If superadmin, also load cloud pollers and global default
|
||||
# If superadmin (including when impersonating), also load cloud pollers and global default
|
||||
{cloud_pollers, global_default_cloud_poller_id} =
|
||||
if current_user.is_superuser do
|
||||
if is_superuser do
|
||||
{Agents.list_cloud_pollers(), Settings.get_global_default_cloud_poller()}
|
||||
else
|
||||
{[], nil}
|
||||
|
|
@ -41,6 +43,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
{:ok,
|
||||
socket
|
||||
|> assign(:page_title, "Remote Agents")
|
||||
|> assign(:is_superuser, is_superuser)
|
||||
|> assign(:agent_tokens, agent_tokens)
|
||||
|> assign(:cloud_pollers, cloud_pollers)
|
||||
|> assign(:global_default_cloud_poller_id, global_default_cloud_poller_id)
|
||||
|
|
@ -63,7 +66,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
def handle_event("create_agent", params, socket) do
|
||||
{name, is_cloud_poller} = parse_agent_params(params)
|
||||
|
||||
with :ok <- validate_cloud_poller_permission(socket.assigns.current_scope.user, is_cloud_poller),
|
||||
with :ok <- validate_cloud_poller_permission(socket.assigns.current_scope, is_cloud_poller),
|
||||
{:ok, agent_token, token} <- create_agent(socket.assigns.current_organization, name, is_cloud_poller) do
|
||||
handle_agent_creation_success(socket, agent_token, token, is_cloud_poller)
|
||||
else
|
||||
|
|
@ -132,7 +135,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
agent_tokens = Agents.list_organization_agent_tokens(organization.id)
|
||||
|
||||
# Refresh cloud pollers list if superadmin (in case a cloud poller was deleted)
|
||||
cloud_pollers = load_cloud_pollers_if_superuser(socket.assigns.current_scope.user)
|
||||
cloud_pollers = load_cloud_pollers_if_superuser(socket.assigns.current_scope)
|
||||
|
||||
# Recalculate device counts after deletion
|
||||
equipment_counts =
|
||||
|
|
@ -164,9 +167,9 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
|
||||
@impl true
|
||||
def handle_event("set_global_default", %{"agent_token_id" => agent_token_id}, socket) do
|
||||
current_user = socket.assigns.current_scope.user
|
||||
current_scope = socket.assigns.current_scope
|
||||
|
||||
if current_user.is_superuser do
|
||||
if Scope.superuser?(current_scope) do
|
||||
# Handle empty string as nil
|
||||
agent_token_id = if agent_token_id == "", do: nil, else: agent_token_id
|
||||
|
||||
|
|
@ -202,8 +205,8 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
end
|
||||
end
|
||||
|
||||
defp validate_cloud_poller_permission(user, is_cloud_poller) do
|
||||
if is_cloud_poller && !user.is_superuser do
|
||||
defp validate_cloud_poller_permission(scope, is_cloud_poller) do
|
||||
if is_cloud_poller && !Scope.superuser?(scope) do
|
||||
{:error, :unauthorized}
|
||||
else
|
||||
:ok
|
||||
|
|
@ -222,7 +225,7 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
organization = socket.assigns.current_organization
|
||||
agent_tokens = Agents.list_organization_agent_tokens(organization.id)
|
||||
|
||||
cloud_pollers = load_cloud_pollers_if_superuser(socket.assigns.current_scope.user)
|
||||
cloud_pollers = load_cloud_pollers_if_superuser(socket.assigns.current_scope)
|
||||
|
||||
equipment_counts = calculate_device_counts(agent_tokens)
|
||||
|
||||
|
|
@ -246,8 +249,8 @@ defmodule ToweropsWeb.AgentLive.Index do
|
|||
|> put_flash(:info, success_message)}
|
||||
end
|
||||
|
||||
defp load_cloud_pollers_if_superuser(user) do
|
||||
if user.is_superuser, do: Agents.list_cloud_pollers(), else: []
|
||||
defp load_cloud_pollers_if_superuser(scope) do
|
||||
if Scope.superuser?(scope), do: Agents.list_cloud_pollers(), else: []
|
||||
end
|
||||
|
||||
defp calculate_device_counts(agent_tokens) do
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@
|
|||
placeholder="e.g., Datacenter A"
|
||||
required
|
||||
/>
|
||||
<%= if @current_scope.user.is_superuser do %>
|
||||
<%= if @is_superuser do %>
|
||||
<.input
|
||||
field={@agent_form[:is_cloud_poller]}
|
||||
type="checkbox"
|
||||
|
|
@ -168,7 +168,7 @@
|
|||
<% end %>
|
||||
|
||||
<%!-- Cloud Pollers Section (Superadmin Only) --%>
|
||||
<%= if @current_scope.user.is_superuser && @cloud_pollers != [] do %>
|
||||
<%= if @is_superuser && @cloud_pollers != [] do %>
|
||||
<div class="mt-12">
|
||||
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
||||
Cloud Pollers
|
||||
|
|
@ -185,9 +185,11 @@
|
|||
>
|
||||
{agent.name}
|
||||
</.link>
|
||||
<span class="inline-flex items-center rounded-md bg-purple-50 px-2 py-1 text-xs font-medium text-purple-700 ring-1 ring-inset ring-purple-700/10 dark:bg-purple-400/10 dark:text-purple-400 dark:ring-purple-400/30">
|
||||
Cloud Poller
|
||||
</span>
|
||||
<%= if agent.id == @global_default_cloud_poller_id do %>
|
||||
<span class="inline-flex items-center gap-1 rounded-full bg-green-50 dark:bg-green-900/30 px-2 py-0.5 text-xs font-medium text-green-700 dark:text-green-300 ring-1 ring-inset ring-green-600/20 dark:ring-green-400/30">
|
||||
<.icon name="hero-check-circle" class="h-3 w-3" /> Default
|
||||
</span>
|
||||
<% end %>
|
||||
</div>
|
||||
</:col>
|
||||
|
||||
|
|
@ -260,7 +262,7 @@
|
|||
<% end %>
|
||||
|
||||
<%!-- Global Default Cloud Poller (Superadmin Only) --%>
|
||||
<%= if @current_scope.user.is_superuser do %>
|
||||
<%= if @is_superuser do %>
|
||||
<div class="mt-8">
|
||||
<div class="bg-amber-50 dark:bg-amber-900/10 border border-amber-200 dark:border-amber-800/30 rounded-lg p-6">
|
||||
<h3 class="text-lg font-semibold text-amber-900 dark:text-amber-200 mb-2">
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue