Graham McIntire
76b3dd6a93
add ssl check executor that connects via TLS, reads peer certificate expiry, and returns OK/WARNING/CRITICAL based on configurable warning_days threshold. includes protobuf config, liveview form fields, agent channel encoding, oban worker dispatch, and unit tests. Reviewed-on: graham/towerops-web#64
419 lines
11 KiB
Elixir
419 lines
11 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.Monitoring.Check
|
|
alias Towerops.Monitoring.CheckResult
|
|
alias Towerops.Monitoring.MonitoringCheck
|
|
alias Towerops.Repo
|
|
|
|
## 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 assigned to a specific agent.
|
|
|
|
Only returns http/tcp/dns checks (service checks that agents can execute).
|
|
"""
|
|
def list_checks_for_agent(agent_token_id, opts \\ []) do
|
|
query =
|
|
from c in Check,
|
|
where: c.agent_token_id == ^agent_token_id,
|
|
where: c.check_type in ["http", "tcp", "dns", "ssl"],
|
|
order_by: [asc: c.name]
|
|
|
|
query
|
|
|> maybe_filter_by_enabled(opts[:enabled])
|
|
|> Repo.all()
|
|
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.
|
|
"""
|
|
def create_check(attrs \\ %{}) do
|
|
result =
|
|
%Check{}
|
|
|> Check.changeset(attrs)
|
|
|> Repo.insert()
|
|
|
|
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
|
|
alias Towerops.Workers.PollingOffset
|
|
|
|
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 """
|
|
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 """
|
|
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
|
|
|
|
update_check(check, attrs)
|
|
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
|
|
|
|
## Legacy ping-based monitoring compatibility (stub)
|
|
|
|
@doc false
|
|
def get_latency_data(_device_or_site_id, _opts), do: []
|
|
end
|