559 lines
17 KiB
Elixir
559 lines
17 KiB
Elixir
defmodule Towerops.Agents.Stats do
|
|
@moduledoc """
|
|
Statistical queries and analytics for agent monitoring and health.
|
|
|
|
Provides functions to monitor agent health, performance, and usage across the organization.
|
|
"""
|
|
|
|
import Ecto.Query, warn: false
|
|
|
|
alias Towerops.Agents.AgentAssignment
|
|
alias Towerops.Agents.AgentToken
|
|
alias Towerops.Devices.Device
|
|
alias Towerops.Monitoring.Check
|
|
alias Towerops.Repo
|
|
alias Towerops.Snmp.SensorReading
|
|
|
|
@doc """
|
|
Get agent health statistics for an organization.
|
|
|
|
Returns a list of agents with their health metrics:
|
|
- Online status (last_seen_at within 5 minutes)
|
|
- Number of assigned devices
|
|
- Last metric submission time
|
|
- Version information
|
|
|
|
## Examples
|
|
|
|
iex> get_organization_agent_health(org_id)
|
|
[
|
|
%{
|
|
id: "uuid",
|
|
name: "DC1 Agent",
|
|
online: true,
|
|
last_seen_at: ~U[2026-01-14 12:00:00Z],
|
|
device_count: 15,
|
|
last_metric_at: ~U[2026-01-14 12:00:30Z],
|
|
version: "0.1.0",
|
|
hostname: "agent-dc1"
|
|
}
|
|
]
|
|
"""
|
|
def get_organization_agent_health(organization_id) do
|
|
five_minutes_ago = DateTime.add(DateTime.utc_now(), -5, :minute)
|
|
|
|
query =
|
|
from(a in AgentToken,
|
|
where: a.organization_id == ^organization_id and a.enabled == true,
|
|
select: %{
|
|
id: a.id,
|
|
name: a.name,
|
|
online: a.last_seen_at > ^five_minutes_ago,
|
|
last_seen_at: a.last_seen_at,
|
|
version: fragment("?->>'version'", a.metadata),
|
|
hostname: fragment("?->>'hostname'", a.metadata)
|
|
},
|
|
order_by: [desc: a.last_seen_at]
|
|
)
|
|
|
|
agents = Repo.all(query)
|
|
|
|
# device counts for all agents in a single query
|
|
device_counts = get_batch_device_counts(Enum.map(agents, & &1.id))
|
|
|
|
# Get last metric times for all agents in a single query
|
|
last_metrics = get_batch_last_metric_times(Enum.map(agents, & &1.id))
|
|
|
|
# Combine data for each agent
|
|
Enum.map(agents, fn agent ->
|
|
agent
|
|
|> Map.put(:device_count, Map.get(device_counts, agent.id, 0))
|
|
|> Map.put(:last_metric_at, Map.get(last_metrics, agent.id))
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Get devices count breakdown by agent assignment source.
|
|
|
|
Returns counts for device assigned via:
|
|
- Direct assignment
|
|
- Site default
|
|
- Organization default
|
|
- No agent (cloud polling)
|
|
|
|
## Examples
|
|
|
|
iex> get_device_assignment_breakdown(org_id)
|
|
%{
|
|
direct: 50,
|
|
site: 30,
|
|
organization: 15,
|
|
cloud: 5
|
|
}
|
|
"""
|
|
def get_device_assignment_breakdown(organization_id) do
|
|
# Use a single SQL query with CASE to determine assignment source
|
|
# device into memory and calling get_effective_agent_token_with_source
|
|
results =
|
|
Repo.all(
|
|
from(e in Device,
|
|
join: s in assoc(e, :site),
|
|
join: o in assoc(s, :organization),
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == e.id and aa.enabled == true,
|
|
where: s.organization_id == ^organization_id and e.snmp_enabled == true,
|
|
select: %{
|
|
source:
|
|
fragment(
|
|
"""
|
|
CASE
|
|
WHEN ? IS NOT NULL THEN 'device'
|
|
WHEN ? IS NOT NULL THEN 'site'
|
|
WHEN ? IS NOT NULL THEN 'organization'
|
|
ELSE 'cloud'
|
|
END
|
|
""",
|
|
aa.agent_token_id,
|
|
s.agent_token_id,
|
|
o.default_agent_token_id
|
|
)
|
|
}
|
|
)
|
|
)
|
|
|
|
# Count by assignment source
|
|
Enum.reduce(results, %{direct: 0, site: 0, organization: 0, cloud: 0}, fn %{source: source}, acc ->
|
|
case source do
|
|
"device" -> %{acc | direct: acc.direct + 1}
|
|
"site" -> %{acc | site: acc.site + 1}
|
|
"organization" -> %{acc | organization: acc.organization + 1}
|
|
"cloud" -> %{acc | cloud: acc.cloud + 1}
|
|
end
|
|
end)
|
|
end
|
|
|
|
@doc """
|
|
Get metric submission stats for an agent over the last 24 hours.
|
|
|
|
Returns:
|
|
- Total metrics submitted
|
|
- Sensor readings count
|
|
- Interface stats count
|
|
- Average metrics per hour
|
|
- Last successful submission
|
|
|
|
## Examples
|
|
|
|
iex> get_agent_metric_stats(agent_id)
|
|
%{
|
|
total_metrics: 1440,
|
|
sensor_readings: 960,
|
|
interface_stats: 480,
|
|
avg_per_hour: 60,
|
|
last_submission: ~U[2026-01-14 12:00:00Z]
|
|
}
|
|
"""
|
|
def get_agent_metric_stats(agent_token_id) do
|
|
twenty_four_hours_ago = DateTime.add(DateTime.utc_now(), -24, :hour)
|
|
|
|
# Get sensor readings
|
|
sensor_count =
|
|
Repo.one(
|
|
from(sr in SensorReading,
|
|
join: s in assoc(sr, :sensor),
|
|
join: d in assoc(s, :snmp_device),
|
|
join: e in assoc(d, :device),
|
|
where: sr.checked_at > ^twenty_four_hours_ago and e.id in subquery(polling_targets_subquery(agent_token_id)),
|
|
select: count(sr.id)
|
|
)
|
|
) || 0
|
|
|
|
# Get interface stats
|
|
interface_count =
|
|
Repo.one(
|
|
from(is in Towerops.Snmp.InterfaceStat,
|
|
join: i in assoc(is, :interface),
|
|
join: d in assoc(i, :snmp_device),
|
|
join: e in assoc(d, :device),
|
|
where: is.checked_at > ^twenty_four_hours_ago and e.id in subquery(polling_targets_subquery(agent_token_id)),
|
|
select: count(is.id)
|
|
)
|
|
) || 0
|
|
|
|
total = sensor_count + interface_count
|
|
|
|
# Get last submission time
|
|
last_submission =
|
|
Repo.one(
|
|
from(sr in SensorReading,
|
|
join: s in assoc(sr, :sensor),
|
|
join: d in assoc(s, :snmp_device),
|
|
join: e in assoc(d, :device),
|
|
where: e.id in subquery(polling_targets_subquery(agent_token_id)),
|
|
select: max(sr.checked_at)
|
|
)
|
|
)
|
|
|
|
%{
|
|
total_metrics: total,
|
|
sensor_readings: sensor_count,
|
|
interface_stats: interface_count,
|
|
avg_per_hour: if(total > 0, do: Float.round(total / 24, 1), else: 0),
|
|
last_submission: last_submission
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Get list of offline agents (not seen in last 5 minutes).
|
|
|
|
Returns list of agent tokens that haven't reported in recently.
|
|
|
|
## Examples
|
|
|
|
iex> get_offline_agents(org_id)
|
|
[
|
|
%{id: "uuid", name: "DC2 Agent", last_seen_at: ~U[2026-01-14 10:00:00Z]}
|
|
]
|
|
"""
|
|
def get_offline_agents(organization_id) do
|
|
five_minutes_ago = DateTime.add(DateTime.utc_now(), -5, :minute)
|
|
|
|
Repo.all(
|
|
from(a in AgentToken,
|
|
where:
|
|
a.organization_id == ^organization_id and a.enabled == true and
|
|
(is_nil(a.last_seen_at) or a.last_seen_at < ^five_minutes_ago),
|
|
select: %{id: a.id, name: a.name, last_seen_at: a.last_seen_at},
|
|
order_by: [asc: a.last_seen_at]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Get agents with high load (>50 devices).
|
|
|
|
Identifies agents that may need to be split or have their devices redistributed.
|
|
|
|
## Examples
|
|
|
|
iex> get_high_load_agents(org_id)
|
|
[
|
|
%{id: "uuid", name: "DC1 Agent", device_count: 75}
|
|
]
|
|
"""
|
|
def get_high_load_agents(organization_id, threshold \\ 50) do
|
|
query =
|
|
from(a in AgentToken,
|
|
where: a.organization_id == ^organization_id and a.enabled == true,
|
|
select: %{id: a.id, name: a.name}
|
|
)
|
|
|
|
agents = Repo.all(query)
|
|
|
|
# device count for each agent, filter by threshold, and sort
|
|
agents
|
|
|> Enum.map(fn agent ->
|
|
device_count = get_agent_device_count(agent.id)
|
|
Map.put(agent, :device_count, device_count)
|
|
end)
|
|
|> Enum.filter(fn agent -> agent.device_count > threshold end)
|
|
|> Enum.sort_by(& &1.device_count, :desc)
|
|
end
|
|
|
|
@doc """
|
|
Get devices that have no agent assigned and are not polling.
|
|
|
|
Identifies orphaned devices that may need agent assignment.
|
|
|
|
## Examples
|
|
|
|
iex> get_unmonitored_devices(org_id)
|
|
[
|
|
%{id: "uuid", name: "Switch-01", site_name: "DC1"}
|
|
]
|
|
"""
|
|
def get_unmonitored_devices(organization_id) do
|
|
# device has no agent assigned
|
|
# device into memory
|
|
Repo.all(
|
|
from(e in Device,
|
|
join: s in assoc(e, :site),
|
|
join: o in assoc(s, :organization),
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == e.id and aa.enabled == true,
|
|
# Device has no agent via any method
|
|
where:
|
|
s.organization_id == ^organization_id and
|
|
e.snmp_enabled == true and
|
|
e.monitoring_enabled == true and
|
|
is_nil(aa.agent_token_id) and
|
|
is_nil(s.agent_token_id) and
|
|
is_nil(o.default_agent_token_id),
|
|
select: %{
|
|
id: e.id,
|
|
name: e.name,
|
|
ip_address: e.ip_address,
|
|
site_name: s.name
|
|
}
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Get average latency for a device grouped by agent over the last 24 hours.
|
|
|
|
Returns a list of agents with their average response times for the given device,
|
|
filtered to successful checks only and requiring minimum check count for statistical validity.
|
|
|
|
## Examples
|
|
|
|
iex> get_device_latency_by_agent(device_id, min_checks: 10)
|
|
[
|
|
%{agent_token_id: "uuid-1", avg_latency_ms: 12.5, check_count: 120},
|
|
%{agent_token_id: "uuid-2", avg_latency_ms: 45.2, check_count: 115}
|
|
]
|
|
"""
|
|
def get_device_latency_by_agent(device_id, opts \\ []) do
|
|
min_checks = Keyword.get(opts, :min_checks, 10)
|
|
hours_ago = Keyword.get(opts, :hours_ago, 24)
|
|
time_threshold = DateTime.add(DateTime.utc_now(), -hours_ago, :hour)
|
|
|
|
Repo.all(
|
|
from(c in Check,
|
|
where:
|
|
c.device_id == ^device_id and
|
|
c.status == :success and
|
|
not is_nil(c.agent_token_id) and
|
|
c.checked_at > ^time_threshold,
|
|
group_by: c.agent_token_id,
|
|
having: count(c.id) >= ^min_checks,
|
|
select: %{
|
|
agent_token_id: c.agent_token_id,
|
|
avg_latency_ms: avg(c.response_time_ms),
|
|
check_count: count(c.id)
|
|
},
|
|
order_by: [asc: avg(c.response_time_ms)]
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Get device assignment information combined with latency statistics.
|
|
|
|
Returns device assignment details with current agent and latency data for potential reassignment.
|
|
|
|
## Examples
|
|
|
|
iex> get_device_assignment_with_latency(device_id)
|
|
%{
|
|
device_id: "uuid",
|
|
current_agent_token_id: "uuid-1",
|
|
assignment_source: :site,
|
|
latency_stats: [
|
|
%{agent_token_id: "uuid-1", avg_latency_ms: 12.5, check_count: 120},
|
|
%{agent_token_id: "uuid-2", avg_latency_ms: 8.3, check_count: 115}
|
|
]
|
|
}
|
|
"""
|
|
def get_device_assignment_with_latency(device_id) do
|
|
device =
|
|
Device
|
|
|> Repo.get!(device_id)
|
|
|> Repo.preload(site: [organization: :default_agent_token])
|
|
|
|
{current_agent_id, source} = Towerops.Agents.get_effective_agent_token_with_source(device)
|
|
|
|
latency_stats = get_device_latency_by_agent(device_id)
|
|
|
|
%{
|
|
device_id: device_id,
|
|
current_agent_token_id: current_agent_id,
|
|
assignment_source: source,
|
|
latency_stats: latency_stats
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Find devices that are candidates for latency-based reassignment.
|
|
|
|
Identifies devices where an alternative agent has significantly better latency (default: 20% improvement).
|
|
Only considers devices with automatic assignments (site, organization, global, or none).
|
|
|
|
## Examples
|
|
|
|
iex> find_reassignment_candidates(min_improvement_percent: 20, min_checks_per_agent: 10)
|
|
[
|
|
%{
|
|
device_id: "uuid",
|
|
device_name: "Switch-01",
|
|
current_agent_token_id: "uuid-1",
|
|
current_latency_ms: 45.2,
|
|
best_agent_token_id: "uuid-2",
|
|
best_latency_ms: 28.3,
|
|
improvement_percent: 37.4,
|
|
assignment_source: :site
|
|
}
|
|
]
|
|
"""
|
|
def find_reassignment_candidates(opts \\ []) do
|
|
min_improvement_percent = Keyword.get(opts, :min_improvement_percent, 20)
|
|
min_checks = Keyword.get(opts, :min_checks_per_agent, 10)
|
|
hours_ago = Keyword.get(opts, :hours_ago, 24)
|
|
time_threshold = DateTime.add(DateTime.utc_now(), -hours_ago, :hour)
|
|
|
|
# Get all devices with multiple agent latency measurements
|
|
devices_with_agents =
|
|
Repo.all(
|
|
from(c in Check,
|
|
where: c.status == :success and not is_nil(c.agent_token_id) and c.checked_at > ^time_threshold,
|
|
group_by: [c.device_id, c.agent_token_id],
|
|
having: count(c.id) >= ^min_checks,
|
|
select: %{
|
|
device_id: c.device_id,
|
|
agent_token_id: c.agent_token_id,
|
|
avg_latency_ms: avg(c.response_time_ms),
|
|
check_count: count(c.id)
|
|
}
|
|
)
|
|
)
|
|
|
|
# Group by device and evaluate candidates
|
|
devices_with_agents
|
|
|> Enum.group_by(& &1.device_id)
|
|
|> Enum.filter(fn {_device_id, agents} -> length(agents) >= 2 end)
|
|
|> Enum.flat_map(fn {device_id, agents} ->
|
|
evaluate_device_reassignment(device_id, agents, min_improvement_percent)
|
|
end)
|
|
end
|
|
|
|
# Private helper functions
|
|
|
|
# Evaluates if a device should be reassigned based on latency comparison
|
|
defp evaluate_device_reassignment(device_id, agents, min_improvement_percent) do
|
|
device =
|
|
Device
|
|
|> Repo.get!(device_id)
|
|
|> Repo.preload(site: [organization: :default_agent_token])
|
|
|
|
{current_agent_id, source} = Towerops.Agents.get_effective_agent_token_with_source(device)
|
|
|
|
# Only consider automatic assignments for reassignment
|
|
if source in [:site, :organization, :global, :none] do
|
|
build_reassignment_candidate(device, current_agent_id, source, agents, min_improvement_percent)
|
|
else
|
|
[]
|
|
end
|
|
end
|
|
|
|
# Builds reassignment candidate if improvement threshold is met
|
|
defp build_reassignment_candidate(device, current_agent_id, source, agents, min_improvement_percent) do
|
|
# Special case: if no current agent (:none or :global with no agent),
|
|
# just pick the best agent without requiring improvement
|
|
if is_nil(current_agent_id) do
|
|
build_initial_assignment_candidate(device, source, agents)
|
|
else
|
|
build_reassignment_with_improvement(device, current_agent_id, source, agents, min_improvement_percent)
|
|
end
|
|
end
|
|
|
|
defp build_initial_assignment_candidate(device, source, agents) do
|
|
best = Enum.min_by(agents, & &1.avg_latency_ms, fn -> nil end)
|
|
|
|
if best do
|
|
[
|
|
%{
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
current_agent_token_id: nil,
|
|
current_latency_ms: nil,
|
|
best_agent_token_id: best.agent_token_id,
|
|
best_latency_ms: Float.round(best.avg_latency_ms, 2),
|
|
improvement_percent: 100.0,
|
|
assignment_source: source
|
|
}
|
|
]
|
|
else
|
|
[]
|
|
end
|
|
end
|
|
|
|
defp build_reassignment_with_improvement(device, current_agent_id, source, agents, min_improvement_percent) do
|
|
current_latency = Enum.find(agents, fn a -> a.agent_token_id == current_agent_id end)
|
|
|
|
with %{} <- current_latency,
|
|
best when not is_nil(best) <- find_best_alternative(agents, current_agent_id),
|
|
improvement when improvement >= min_improvement_percent <-
|
|
calculate_improvement_percent(current_latency, best) do
|
|
[
|
|
%{
|
|
device_id: device.id,
|
|
device_name: device.name,
|
|
current_agent_token_id: current_agent_id,
|
|
current_latency_ms: Float.round(current_latency.avg_latency_ms, 2),
|
|
best_agent_token_id: best.agent_token_id,
|
|
best_latency_ms: Float.round(best.avg_latency_ms, 2),
|
|
improvement_percent: improvement,
|
|
assignment_source: source
|
|
}
|
|
]
|
|
else
|
|
_ -> []
|
|
end
|
|
end
|
|
|
|
defp find_best_alternative(agents, current_agent_id) do
|
|
agents
|
|
|> Enum.reject(fn a -> a.agent_token_id == current_agent_id end)
|
|
|> Enum.min_by(& &1.avg_latency_ms, fn -> nil end)
|
|
end
|
|
|
|
defp calculate_improvement_percent(current, best) do
|
|
Float.round((current.avg_latency_ms - best.avg_latency_ms) / current.avg_latency_ms * 100, 1)
|
|
end
|
|
|
|
defp polling_targets_subquery(agent_token_id) do
|
|
# This is a simplified version - in production you'd want to use
|
|
# Agents.list_agent_polling_targets/1 logic here
|
|
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 and e.snmp_enabled == true,
|
|
select: e.id
|
|
)
|
|
end
|
|
|
|
defp get_agent_device_count(agent_token_id) do
|
|
# device by using list_agent_polling_targets
|
|
# This properly handles hierarchical assignment
|
|
agent_token_id
|
|
|> Towerops.Agents.list_agent_polling_targets()
|
|
|> length()
|
|
end
|
|
|
|
defp get_batch_device_counts(agent_token_ids) do
|
|
# For now, use the existing function per agent since list_agent_polling_targets
|
|
# handles complex hierarchical logic in application code.
|
|
# NOTE: This could be optimized with a pure SQL solution if performance becomes a concern.
|
|
Map.new(agent_token_ids, fn id -> {id, get_agent_device_count(id)} end)
|
|
end
|
|
|
|
defp get_batch_last_metric_times(agent_token_ids) do
|
|
# Get last metric time for all agents in a single query
|
|
# device ID, then map to agent via assignments
|
|
results =
|
|
Repo.all(
|
|
from(sr in SensorReading,
|
|
join: s in assoc(sr, :sensor),
|
|
join: d in assoc(s, :snmp_device),
|
|
join: e in assoc(d, :device),
|
|
join: aa in AgentAssignment,
|
|
on: aa.device_id == e.id and aa.enabled == true,
|
|
where: aa.agent_token_id in ^agent_token_ids,
|
|
group_by: aa.agent_token_id,
|
|
select: {aa.agent_token_id, max(sr.checked_at)}
|
|
)
|
|
)
|
|
|
|
Map.new(results)
|
|
end
|
|
end
|