mayor
3fff0db784
Adds @spec annotations to all public functions in agent-related modules for better static analysis and documentation. Changes: - Added 17 typespecs to Towerops.Agents context module - All CRUD operations (create, read, update, delete, revoke) - Assignment management functions - Agent token resolution and lookup functions - PubSub broadcast function - Added 9 typespecs to Towerops.Agents.Stats module - Agent health and metrics statistics - Device assignment breakdowns - Latency analysis and reassignment candidate detection - Added typespec to Towerops.Workers.StaleAgentWorker - find_stale_agents/0 function - Preserved existing typespecs in AgentChannel (socket types) Benefits: - Improved code documentation with clear function signatures - Better static analysis via dialyzer - Clearer API contracts for all agent management functions - Type-safe UUID handling throughout agent system
108 lines
3.1 KiB
Elixir
108 lines
3.1 KiB
Elixir
defmodule Towerops.Workers.StaleAgentWorker do
|
|
@moduledoc """
|
|
Oban worker that periodically checks for stale agents.
|
|
|
|
Runs every minute via Oban.Plugins.Cron to:
|
|
1. Find agents that haven't sent a heartbeat in over 10 minutes
|
|
2. Log warnings for operators
|
|
3. Broadcast alerts via PubSub for UI updates
|
|
|
|
Agents are considered stale if:
|
|
- They are enabled
|
|
- They have checked in at least once (last_seen_at is not nil)
|
|
- They haven't checked in for more than 10 minutes
|
|
"""
|
|
use Oban.Worker, queue: :maintenance
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Agents.AgentToken
|
|
alias Towerops.Repo
|
|
|
|
require Logger
|
|
|
|
# Consider agents stale if not seen in 10 minutes
|
|
@stale_threshold_minutes 10
|
|
|
|
# Consider agents "newly stale" if not seen in 10-15 minutes (for warning logs)
|
|
@newly_stale_threshold_minutes 15
|
|
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{}) do
|
|
stale_agents = find_stale_agents()
|
|
|
|
_ =
|
|
if !Enum.empty?(stale_agents) do
|
|
process_stale_agents(stale_agents)
|
|
end
|
|
|
|
:ok
|
|
end
|
|
|
|
defp process_stale_agents(stale_agents) do
|
|
# Separate newly stale (10-15 min) from long-term stale (>15 min)
|
|
{newly_stale, long_term_stale} =
|
|
Enum.split_with(stale_agents, &newly_stale?/1)
|
|
|
|
# Log warnings for newly stale agents
|
|
log_newly_stale_agents(newly_stale)
|
|
|
|
# Log debug for long-term stale agents (reduce log noise)
|
|
Enum.each(long_term_stale, fn agent -> log_stale_agent(agent, :debug) end)
|
|
|
|
# Broadcast all stale agents for UI updates
|
|
_ = Phoenix.PubSub.broadcast(Towerops.PubSub, "agents:health", {:agents_stale, stale_agents})
|
|
end
|
|
|
|
defp log_newly_stale_agents([]), do: :ok
|
|
|
|
defp log_newly_stale_agents(newly_stale) do
|
|
Enum.each(newly_stale, fn agent -> log_stale_agent(agent, :warning) end)
|
|
Logger.warning("Detected #{length(newly_stale)} newly stale agent(s)")
|
|
end
|
|
|
|
@doc """
|
|
Returns a list of stale agent tokens.
|
|
|
|
Public for testing and manual inspection.
|
|
"""
|
|
@spec find_stale_agents() :: [AgentToken.t()]
|
|
def find_stale_agents do
|
|
cutoff = DateTime.add(DateTime.utc_now(), -@stale_threshold_minutes, :minute)
|
|
|
|
AgentToken
|
|
|> where([t], t.enabled == true)
|
|
|> where([t], not is_nil(t.last_seen_at))
|
|
|> where([t], t.last_seen_at < ^cutoff)
|
|
|> preload(:organization)
|
|
|> Repo.all()
|
|
end
|
|
|
|
defp newly_stale?(agent) do
|
|
newly_stale_cutoff = DateTime.add(DateTime.utc_now(), -@newly_stale_threshold_minutes, :minute)
|
|
DateTime.after?(agent.last_seen_at, newly_stale_cutoff)
|
|
end
|
|
|
|
defp log_stale_agent(agent, level) do
|
|
minutes_since_seen =
|
|
DateTime.utc_now()
|
|
|> DateTime.diff(agent.last_seen_at, :second)
|
|
|> div(60)
|
|
|
|
message = "Agent '#{agent.name}' is stale - last seen #{minutes_since_seen} minutes ago"
|
|
|
|
metadata = [
|
|
agent_token_id: agent.id,
|
|
agent_name: agent.name,
|
|
organization_id: agent.organization_id,
|
|
last_seen_at: agent.last_seen_at,
|
|
last_ip: agent.last_ip,
|
|
minutes_since_seen: minutes_since_seen
|
|
]
|
|
|
|
case level do
|
|
:warning -> Logger.warning(message, metadata)
|
|
:debug -> Logger.debug(message, metadata)
|
|
end
|
|
end
|
|
end
|