Graham McIntire
91e3181bbc
Prefix fire-and-forget calls (PubSub.broadcast, Oban enqueue, Logger, update_*, etc.) with `_ =` so dialyzer knows the return value is intentionally discarded. Wrap a few if/case expressions whose branches produce mixed types the same way. No behavior changes — only explicit acknowledgement of discarded returns. Warnings: 242 → 88.
662 lines
19 KiB
Elixir
662 lines
19 KiB
Elixir
defmodule Towerops.Monitoring do
|
|
@moduledoc """
|
|
The Monitoring context for managing checks and check results.
|
|
|
|
Follows Icinga2's generic check pattern with type-specific configuration.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Towerops.Agents.AgentAssignment
|
|
alias Towerops.Devices.Device
|
|
alias Towerops.Monitoring.Check
|
|
alias Towerops.Monitoring.CheckResult
|
|
alias Towerops.Monitoring.MonitoringCheck
|
|
alias Towerops.Repo
|
|
alias Towerops.Workers.PollingOffset
|
|
|
|
## Checks
|
|
|
|
@doc """
|
|
Returns the list of checks for an organization.
|
|
"""
|
|
def list_checks(organization_id, opts \\ []) do
|
|
query =
|
|
from c in Check,
|
|
where: c.organization_id == ^organization_id,
|
|
order_by: [asc: c.name]
|
|
|
|
query
|
|
|> maybe_filter_by_device(opts[:device_id])
|
|
|> maybe_filter_by_check_type(opts[:check_type])
|
|
|> maybe_filter_by_enabled(opts[:enabled])
|
|
|> Repo.all()
|
|
end
|
|
|
|
defp maybe_filter_by_device(query, nil), do: query
|
|
|
|
defp maybe_filter_by_device(query, device_id) do
|
|
from c in query, where: c.device_id == ^device_id
|
|
end
|
|
|
|
defp maybe_filter_by_check_type(query, nil), do: query
|
|
|
|
defp maybe_filter_by_check_type(query, check_type) do
|
|
from c in query, where: c.check_type == ^check_type
|
|
end
|
|
|
|
defp maybe_filter_by_enabled(query, nil), do: query
|
|
|
|
defp maybe_filter_by_enabled(query, enabled) do
|
|
from c in query, where: c.enabled == ^enabled
|
|
end
|
|
|
|
@doc """
|
|
Returns the list of service checks for devices assigned to a specific agent.
|
|
|
|
Uses the full agent assignment cascade (device → site → org → global default)
|
|
to find checks, not just the check's own agent_token_id field.
|
|
Only returns http/tcp/dns/ssl checks (service checks that agents can execute).
|
|
"""
|
|
def list_checks_for_agent(agent_token_id, opts \\ []) do
|
|
global_default = Towerops.Settings.get_global_default_cloud_poller()
|
|
is_global_default = agent_token_id == global_default
|
|
|
|
query =
|
|
from c in Check,
|
|
left_join: d in assoc(c, :device),
|
|
left_join: s in assoc(d, :site),
|
|
left_join: o in assoc(d, :organization),
|
|
left_join: aa in AgentAssignment,
|
|
on: aa.device_id == d.id and aa.enabled == true,
|
|
left_join: disabled in AgentAssignment,
|
|
on:
|
|
disabled.device_id == d.id and disabled.agent_token_id == ^agent_token_id and
|
|
disabled.enabled == false,
|
|
where: c.check_type in ["http", "tcp", "dns", "ssl"],
|
|
order_by: [asc: c.name],
|
|
distinct: c.id
|
|
|
|
query
|
|
|> where_check_agent_matches(agent_token_id, is_global_default)
|
|
|> maybe_filter_by_enabled(opts[:enabled])
|
|
|> Repo.all()
|
|
end
|
|
|
|
# Builds the WHERE clause for matching checks to an agent via the full cascade:
|
|
# direct check assignment → device assignment → site → org default → global default
|
|
defp where_check_agent_matches(query, agent_token_id, is_global_default) do
|
|
condition =
|
|
[c]
|
|
|> dynamic(c.agent_token_id == ^agent_token_id)
|
|
|> or_device_assignment(agent_token_id)
|
|
|> or_site_assignment(agent_token_id)
|
|
|> or_org_default(agent_token_id)
|
|
|> or_global_default(is_global_default)
|
|
|
|
from q in query, where: ^condition
|
|
end
|
|
|
|
defp or_device_assignment(condition, agent_token_id) do
|
|
dynamic(
|
|
[c, d, s, o, aa, disabled],
|
|
^condition or
|
|
(not is_nil(d.id) and aa.agent_token_id == ^agent_token_id and is_nil(disabled.id))
|
|
)
|
|
end
|
|
|
|
defp or_site_assignment(condition, agent_token_id) do
|
|
dynamic(
|
|
[c, d, s, o, aa, disabled],
|
|
^condition or
|
|
(not is_nil(d.id) and 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)
|
|
)
|
|
end
|
|
|
|
defp or_org_default(condition, agent_token_id) do
|
|
dynamic(
|
|
[c, d, s, o, aa, disabled],
|
|
^condition or
|
|
(not is_nil(d.id) 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
|
|
o.default_agent_token_id == ^agent_token_id)
|
|
)
|
|
end
|
|
|
|
defp or_global_default(condition, false), do: condition
|
|
|
|
defp or_global_default(condition, true) do
|
|
dynamic(
|
|
[c, d, s, o, aa, disabled],
|
|
^condition or
|
|
(not is_nil(d.id) 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 a single check.
|
|
"""
|
|
def get_check(id), do: Repo.get(Check, id)
|
|
|
|
@doc """
|
|
Gets a single check, raises if not found.
|
|
"""
|
|
def get_check!(id), do: Repo.get!(Check, id)
|
|
|
|
@doc """
|
|
Gets a check by check_key (for passive checks).
|
|
"""
|
|
def get_check_by_key(check_key) do
|
|
Repo.get_by(Check, check_key: check_key)
|
|
end
|
|
|
|
@doc """
|
|
Creates a check.
|
|
|
|
For auto-discovery checks, uses upsert logic to update existing checks
|
|
instead of creating duplicates.
|
|
"""
|
|
def create_check(attrs \\ %{}) do
|
|
# Use upsert for auto-discovery checks to avoid duplicates
|
|
result =
|
|
if Map.get(attrs, :source_type) == "auto_discovery" and Map.has_key?(attrs, :source_id) do
|
|
# For partial unique indexes, find existing record and update it
|
|
case Repo.get_by(Check,
|
|
device_id: attrs.device_id,
|
|
check_type: attrs.check_type,
|
|
source_id: attrs.source_id
|
|
) do
|
|
nil ->
|
|
%Check{}
|
|
|> Check.changeset(attrs)
|
|
|> Repo.insert()
|
|
|
|
existing_check ->
|
|
existing_check
|
|
|> Check.changeset(attrs)
|
|
|> Repo.update()
|
|
end
|
|
else
|
|
%Check{}
|
|
|> Check.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
_ =
|
|
case result do
|
|
{:ok, check} -> broadcast_check_change(check)
|
|
_ -> :ok
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
@doc """
|
|
Updates a check.
|
|
"""
|
|
def update_check(%Check{} = check, attrs) do
|
|
result =
|
|
check
|
|
|> Check.changeset(attrs)
|
|
|> Repo.update()
|
|
|
|
_ =
|
|
case result do
|
|
{:ok, updated} -> broadcast_check_change(updated)
|
|
_ -> :ok
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
@doc """
|
|
Deletes a check.
|
|
"""
|
|
def delete_check(%Check{} = check) do
|
|
result = Repo.delete(check)
|
|
|
|
_ =
|
|
case result do
|
|
{:ok, deleted} -> broadcast_check_change(deleted)
|
|
_ -> :ok
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
defp broadcast_check_change(%Check{agent_token_id: nil}), do: :ok
|
|
|
|
defp broadcast_check_change(%Check{agent_token_id: agent_token_id}) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"checks:agent:#{agent_token_id}",
|
|
:checks_changed
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Returns an `%Ecto.Changeset{}` for tracking check changes.
|
|
"""
|
|
def change_check(%Check{} = check, attrs \\ %{}) do
|
|
Check.changeset(check, attrs)
|
|
end
|
|
|
|
@doc """
|
|
Schedules a check for execution.
|
|
|
|
Queues a CheckExecutorWorker job with polling offset to distribute load.
|
|
"""
|
|
def schedule_check(%Check{} = check) do
|
|
alias Towerops.Workers.CheckExecutorWorker
|
|
|
|
offset = PollingOffset.calculate_offset(check.id, check.interval_seconds)
|
|
|
|
%{check_id: check.id}
|
|
|> CheckExecutorWorker.new(schedule_in: offset)
|
|
|> Oban.insert()
|
|
end
|
|
|
|
## Check Results
|
|
|
|
@doc """
|
|
Creates a check result.
|
|
"""
|
|
def create_check_result(attrs \\ %{}) do
|
|
%CheckResult{}
|
|
|> CheckResult.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Gets the latest check result for a check.
|
|
"""
|
|
def get_latest_check_result(check_id) do
|
|
Repo.one(from(cr in CheckResult, where: cr.check_id == ^check_id, order_by: [desc: cr.checked_at], limit: 1))
|
|
end
|
|
|
|
@doc """
|
|
Gets check results for a check within a time range.
|
|
"""
|
|
def get_check_results(check_id, opts \\ []) do
|
|
from_time = opts[:from] || DateTime.add(DateTime.utc_now(), -3600, :second)
|
|
to_time = opts[:to] || DateTime.utc_now()
|
|
limit = opts[:limit] || 1000
|
|
|
|
Repo.all(
|
|
from(cr in CheckResult,
|
|
where: cr.check_id == ^check_id,
|
|
where: cr.checked_at >= ^from_time,
|
|
where: cr.checked_at <= ^to_time,
|
|
order_by: [desc: cr.checked_at],
|
|
limit: ^limit
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Get check status data for a device since a given time.
|
|
Returns a list of check results with timestamps and statuses.
|
|
"""
|
|
def get_device_check_data(device_id, since) do
|
|
check_ids =
|
|
Check
|
|
|> where(device_id: ^device_id)
|
|
|> select([c], c.id)
|
|
|> Repo.all()
|
|
|
|
if Enum.empty?(check_ids) do
|
|
[]
|
|
else
|
|
CheckResult
|
|
|> where([r], r.check_id in ^check_ids)
|
|
|> where([r], r.checked_at >= ^since)
|
|
|> order_by(asc: :checked_at)
|
|
|> select([r], %{t: r.checked_at, status: r.status})
|
|
|> limit(1000)
|
|
|> Repo.all()
|
|
|> Enum.map(fn r ->
|
|
%{t: DateTime.to_iso8601(r.t), status: r.status}
|
|
end)
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Gets graph data for a check, combining check_results and legacy sensor readings.
|
|
|
|
For SNMP checks (auto_discovery source), queries both:
|
|
- check_results (new data)
|
|
- snmp_sensor_readings (historical backfill)
|
|
|
|
Returns list of %{timestamp: DateTime.t(), value: float()} sorted by timestamp.
|
|
"""
|
|
def get_check_graph_data(check_id, from_time, to_time) do
|
|
check = get_check!(check_id)
|
|
|
|
# Query new check_results table
|
|
recent_results =
|
|
Repo.all(
|
|
from(r in CheckResult,
|
|
where: r.check_id == ^check_id,
|
|
where: r.checked_at >= ^from_time,
|
|
where: r.checked_at <= ^to_time,
|
|
where: not is_nil(r.value),
|
|
order_by: [asc: r.checked_at],
|
|
select: %{timestamp: r.checked_at, value: r.value}
|
|
)
|
|
)
|
|
|
|
# If SNMP check with source_id, also query old snmp_sensor_readings for backfill
|
|
historical_results =
|
|
if check.check_type == "snmp_sensor" && check.source_id do
|
|
alias Towerops.Snmp.SensorReading
|
|
# Find earliest check_result timestamp to avoid overlap
|
|
earliest_check_result =
|
|
recent_results
|
|
|> Enum.map(& &1.timestamp)
|
|
|> Enum.min(DateTime, fn -> to_time end)
|
|
|
|
# Query historical sensor readings before check_results cutover
|
|
Repo.all(
|
|
from(r in SensorReading,
|
|
where: r.sensor_id == ^check.source_id,
|
|
where: r.checked_at >= ^from_time,
|
|
where: r.checked_at < ^earliest_check_result,
|
|
where: not is_nil(r.value),
|
|
order_by: [asc: r.checked_at],
|
|
select: %{timestamp: r.checked_at, value: r.value}
|
|
)
|
|
)
|
|
else
|
|
[]
|
|
end
|
|
|
|
# Combine and sort by timestamp
|
|
Enum.sort_by(historical_results ++ recent_results, & &1.timestamp, DateTime)
|
|
end
|
|
|
|
@doc """
|
|
Gets graph data for a check including status field.
|
|
|
|
Returns list of %{timestamp: DateTime.t(), value: float() | nil, status: integer()} sorted by timestamp.
|
|
Used for DNS checks that need to show both availability and response time.
|
|
"""
|
|
def get_check_graph_data_with_status(check_id, from_time, to_time) do
|
|
Repo.all(
|
|
from(r in CheckResult,
|
|
where: r.check_id == ^check_id,
|
|
where: r.checked_at >= ^from_time,
|
|
where: r.checked_at <= ^to_time,
|
|
order_by: [asc: r.checked_at],
|
|
select: %{timestamp: r.checked_at, value: r.value, status: r.status}
|
|
)
|
|
)
|
|
end
|
|
|
|
@doc """
|
|
Updates check state after a check execution.
|
|
Handles soft/hard state transitions.
|
|
"""
|
|
def update_check_state(%Check{} = check, status, _output) do
|
|
now = DateTime.utc_now()
|
|
state_changed = check.current_state != status
|
|
|
|
{new_state_type, new_check_attempt} =
|
|
calculate_state_transition(check, status, state_changed)
|
|
|
|
attrs = %{
|
|
current_state: status,
|
|
current_state_type: new_state_type,
|
|
current_check_attempt: new_check_attempt,
|
|
last_check_at: now
|
|
}
|
|
|
|
attrs =
|
|
if state_changed do
|
|
Map.put(attrs, :last_state_change_at, now)
|
|
else
|
|
attrs
|
|
end
|
|
|
|
attrs =
|
|
if new_state_type == "hard" and state_changed do
|
|
Map.put(attrs, :last_hard_state_change_at, now)
|
|
else
|
|
attrs
|
|
end
|
|
|
|
result =
|
|
check
|
|
|> Check.state_changeset(attrs)
|
|
|> Repo.update()
|
|
|
|
_ =
|
|
case result do
|
|
{:ok, updated} -> broadcast_check_change(updated)
|
|
_ -> :ok
|
|
end
|
|
|
|
result
|
|
end
|
|
|
|
defp calculate_state_transition(check, _new_status, true = _state_changed) do
|
|
new_attempt = check.current_check_attempt + 1
|
|
state_type = if new_attempt >= check.max_check_attempts, do: "hard", else: "soft"
|
|
{state_type, new_attempt}
|
|
end
|
|
|
|
defp calculate_state_transition(%{current_state_type: "hard"} = check, _new_status, false) do
|
|
{"hard", check.current_check_attempt}
|
|
end
|
|
|
|
defp calculate_state_transition(check, _new_status, false) do
|
|
new_attempt = min(check.current_check_attempt + 1, check.max_check_attempts)
|
|
state_type = if new_attempt >= check.max_check_attempts, do: "hard", else: "soft"
|
|
{state_type, new_attempt}
|
|
end
|
|
|
|
@doc """
|
|
Stops all check jobs for a device by canceling scheduled Oban jobs.
|
|
|
|
Used when deleting a device to clean up scheduled check executions.
|
|
"""
|
|
def stop_device_checks(device_id) do
|
|
# Get all check IDs for this device
|
|
check_ids =
|
|
Repo.all(
|
|
from c in Check,
|
|
where: c.device_id == ^device_id,
|
|
select: c.id
|
|
)
|
|
|
|
# Cancel all scheduled jobs for these checks
|
|
Enum.each(check_ids, fn check_id ->
|
|
Oban.cancel_all_jobs(
|
|
from(j in Oban.Job,
|
|
where: j.queue == "check_executors",
|
|
where: j.state in ["available", "scheduled", "retryable"],
|
|
where: fragment("? @> ?", j.args, ^%{"check_id" => check_id})
|
|
)
|
|
)
|
|
end)
|
|
|
|
:ok
|
|
end
|
|
|
|
@doc """
|
|
Disables all checks for a device.
|
|
|
|
Used when SNMP is disabled on a device to stop polling without deleting checks.
|
|
Checks can be re-enabled when SNMP is re-enabled.
|
|
"""
|
|
def disable_device_checks(device_id) do
|
|
Repo.update_all(from(c in Check, where: c.device_id == ^device_id, where: c.enabled == true),
|
|
set: [enabled: false, updated_at: DateTime.utc_now()]
|
|
)
|
|
|
|
# Also cancel any scheduled jobs
|
|
stop_device_checks(device_id)
|
|
|
|
:ok
|
|
end
|
|
|
|
## Monitoring Checks (ping results from agents)
|
|
|
|
@doc """
|
|
Creates a monitoring check record in the monitoring_checks table.
|
|
"""
|
|
def create_monitoring_check(attrs \\ %{}) do
|
|
%MonitoringCheck{}
|
|
|> MonitoringCheck.changeset(attrs)
|
|
|> Repo.insert()
|
|
end
|
|
|
|
@doc """
|
|
Returns a health summary for multiple devices in a single batch query.
|
|
|
|
Queries the `checks` table to find the worst-case `current_state` and
|
|
latest response time across all enabled checks for each device.
|
|
|
|
Returns `%{device_id => %{worst_status: 0..3, latest_response_ms: float | nil, last_check_at: DateTime.t() | nil}}`.
|
|
Devices with no checks will not appear in the map.
|
|
"""
|
|
def get_device_health_summary(device_ids) when is_list(device_ids) do
|
|
if device_ids == [] do
|
|
%{}
|
|
else
|
|
Check
|
|
|> where([c], c.device_id in ^device_ids and c.enabled == true)
|
|
|> group_by([c], c.device_id)
|
|
|> select([c], %{
|
|
device_id: c.device_id,
|
|
worst_status: max(c.current_state),
|
|
last_check_at: max(c.last_check_at)
|
|
})
|
|
|> Repo.all()
|
|
|> Map.new(fn row -> {row.device_id, row} end)
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Returns latest response times for multiple devices from monitoring_checks.
|
|
|
|
Fetches the most recent monitoring_check per device.
|
|
Returns `%{device_id => %{response_time_ms: float | nil, checked_at: DateTime.t()}}`.
|
|
"""
|
|
def get_device_latest_response_times(device_ids) when is_list(device_ids) do
|
|
if device_ids == [] do
|
|
%{}
|
|
else
|
|
# Use a lateral join / DISTINCT ON to get latest per device
|
|
MonitoringCheck
|
|
|> where([mc], mc.device_id in ^device_ids)
|
|
|> distinct([mc], [mc.device_id])
|
|
|> order_by([mc], [mc.device_id, desc: mc.checked_at])
|
|
|> select([mc], %{
|
|
device_id: mc.device_id,
|
|
response_time_ms: mc.response_time_ms,
|
|
checked_at: mc.checked_at
|
|
})
|
|
|> Repo.all()
|
|
|> Map.new(fn row -> {row.device_id, row} end)
|
|
end
|
|
end
|
|
|
|
## Auto ICMP ping checks
|
|
|
|
@doc """
|
|
Ensures a default ICMP ping check exists for the given device.
|
|
|
|
Creates an auto_discovery ping check if none exists, or updates the host
|
|
config if the device IP has changed.
|
|
"""
|
|
def ensure_default_ping_check(%Device{} = device) do
|
|
existing =
|
|
Repo.one(
|
|
from(c in Check,
|
|
where: c.device_id == ^device.id,
|
|
where: c.check_type == "ping",
|
|
where: c.source_type == "auto_discovery",
|
|
limit: 1
|
|
)
|
|
)
|
|
|
|
case existing do
|
|
nil ->
|
|
with {:ok, check} <-
|
|
create_check(%{
|
|
organization_id: device.organization_id,
|
|
device_id: device.id,
|
|
name: "ICMP Ping",
|
|
check_type: "ping",
|
|
config: %{"host" => device.ip_address, "count" => 3},
|
|
enabled: true,
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
source_type: "auto_discovery"
|
|
}) do
|
|
# Best-effort scheduling; JobHealthCheckWorker recovers orphaned jobs
|
|
_ = schedule_check(check)
|
|
{:ok, check}
|
|
end
|
|
|
|
check ->
|
|
if check.config["host"] == device.ip_address do
|
|
{:ok, check}
|
|
else
|
|
update_check(check, %{config: Map.put(check.config, "host", device.ip_address)})
|
|
end
|
|
end
|
|
end
|
|
|
|
@doc """
|
|
Finds an existing auto-discovered check or creates a new one.
|
|
|
|
Uses `(device_id, check_type, source_id)` as the uniqueness key.
|
|
If a check already exists, updates its name and config. Otherwise creates it
|
|
and schedules a CheckExecutorWorker job.
|
|
|
|
Attrs must include: `:device_id`, `:check_type`, `:source_id`, `:source_type`,
|
|
`:organization_id`, `:name`, `:config`.
|
|
"""
|
|
def ensure_discovery_check(attrs) do
|
|
device_id = Map.fetch!(attrs, :device_id)
|
|
check_type = Map.fetch!(attrs, :check_type)
|
|
source_id = Map.fetch!(attrs, :source_id)
|
|
|
|
existing =
|
|
Repo.one(
|
|
from(c in Check,
|
|
where: c.device_id == ^device_id,
|
|
where: c.check_type == ^check_type,
|
|
where: c.source_id == ^source_id,
|
|
limit: 1
|
|
)
|
|
)
|
|
|
|
case existing do
|
|
nil ->
|
|
with {:ok, check} <- create_check(attrs) do
|
|
_ = schedule_check(check)
|
|
{:ok, check}
|
|
end
|
|
|
|
check ->
|
|
name = Map.get(attrs, :name, check.name)
|
|
config = Map.get(attrs, :config, check.config)
|
|
|
|
if check.name == name and check.config == config do
|
|
{:ok, check}
|
|
else
|
|
update_check(check, %{name: name, config: config})
|
|
end
|
|
end
|
|
end
|
|
|
|
## Legacy ping-based monitoring compatibility (stub)
|
|
|
|
@doc false
|
|
def get_latency_data(_device_or_site_id, _opts), do: []
|
|
end
|