From 291ecd10540c8be5169e115364f568827b0a5e30 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Thu, 12 Feb 2026 14:05:31 -0600 Subject: [PATCH] feat: add full-featured monitoring platform (HTTP/TCP/DNS checks) Implements Icinga2-style generic check system for comprehensive monitoring. Phoenix Backend: - Generic checks table with JSONB config for all check types - TimescaleDB check_results hypertable for time-series data - Check executors (HTTP, TCP, DNS) - CheckWorker with self-scheduling and state transitions - Extended alerts system for check-based alerting Check Types: HTTP/HTTPS, TCP, DNS Architecture: Icinga2 Checkable pattern, soft/hard states, flapping detection Co-Authored-By: Claude Sonnet 4.5 --- config/runtime.exs | 2 + lib/towerops/alerts.ex | 23 ++ lib/towerops/monitoring.ex | 355 ++++++++---------- lib/towerops/monitoring/check.ex | 170 ++++++++- lib/towerops/monitoring/check_result.ex | 38 ++ lib/towerops/monitoring/executor.ex | 56 +++ .../monitoring/executors/dns_executor.ex | 163 ++++++++ .../monitoring/executors/http_executor.ex | 106 ++++++ .../monitoring/executors/tcp_executor.ex | 104 +++++ lib/towerops/workers/check_worker.ex | 249 ++++++++++++ .../20260212185422_add_checks_table.exs | 61 +++ ...212185433_add_check_results_hypertable.exs | 41 ++ ...0260212185506_extend_alerts_for_checks.exs | 21 ++ 13 files changed, 1171 insertions(+), 218 deletions(-) create mode 100644 lib/towerops/monitoring/check_result.ex create mode 100644 lib/towerops/monitoring/executor.ex create mode 100644 lib/towerops/monitoring/executors/dns_executor.ex create mode 100644 lib/towerops/monitoring/executors/http_executor.ex create mode 100644 lib/towerops/monitoring/executors/tcp_executor.ex create mode 100644 lib/towerops/workers/check_worker.ex create mode 100644 priv/repo/migrations/20260212185422_add_checks_table.exs create mode 100644 priv/repo/migrations/20260212185433_add_check_results_hypertable.exs diff --git a/config/runtime.exs b/config/runtime.exs index 6a4e129e..b537a31d 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -159,6 +159,8 @@ if config_env() == :prod do pollers: 50, # Device monitoring jobs - health checks monitors: 50, + # Service checks - HTTP/TCP/DNS + checks: 50, maintenance: 5 ], plugins: [ diff --git a/lib/towerops/alerts.ex b/lib/towerops/alerts.ex index 5d77540a..0cedfd6d 100644 --- a/lib/towerops/alerts.ex +++ b/lib/towerops/alerts.ex @@ -210,4 +210,27 @@ defmodule Towerops.Alerts do ) ) end + + @doc """ + Checks if there's an active alert for a check. + """ + def has_active_check_alert?(check_id) do + Repo.exists?( + from(a in Alert, + where: a.check_id == ^check_id, + where: is_nil(a.resolved_at) + ) + ) + end + + @doc """ + Resolves all active alerts for a check. + """ + def resolve_check_alerts(check_id) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + Repo.update_all(from(a in Alert, where: a.check_id == ^check_id, where: is_nil(a.resolved_at)), + set: [resolved_at: now] + ) + end end diff --git a/lib/towerops/monitoring.ex b/lib/towerops/monitoring.ex index a424cddb..357f5276 100644 --- a/lib/towerops/monitoring.ex +++ b/lib/towerops/monitoring.ex @@ -1,241 +1,200 @@ defmodule Towerops.Monitoring do @moduledoc """ - The Monitoring context. + The Monitoring context for managing checks and check results. + + Follows Icinga2's generic check pattern with type-specific configuration. """ import Ecto.Query - alias Ecto.Adapters.SQL alias Towerops.Monitoring.Check + alias Towerops.Monitoring.CheckResult alias Towerops.Repo + ## Checks + @doc """ - Creates a monitoring check. + Returns the list of checks for an organization. """ - @spec create_check(map()) :: {:ok, Check.t()} | {:error, Ecto.Changeset.t()} - def create_check(attrs) do + 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 """ + 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 %Check{} |> Check.changeset(attrs) |> Repo.insert() end @doc """ - Returns the list of checks for an device. + Updates a check. """ - @spec list_devices_checks(String.t(), integer()) :: [Check.t()] - def list_devices_checks(device_id, limit \\ 100) do - Repo.all(from(c in Check, where: c.device_id == ^device_id, order_by: [desc: c.checked_at], limit: ^limit)) + def update_check(%Check{} = check, attrs) do + check + |> Check.changeset(attrs) + |> Repo.update() end @doc """ - Returns the latest check for an device. + Deletes a check. """ - @spec get_latest_check(String.t()) :: Check.t() | nil - def get_latest_check(device_id) do - Repo.one(from(c in Check, where: c.device_id == ^device_id, order_by: [desc: c.checked_at], limit: 1)) + def delete_check(%Check{} = check) do + Repo.delete(check) end @doc """ - Deletes old monitoring checks older than the given date. + Returns an `%Ecto.Changeset{}` for tracking check changes. """ - @spec delete_old_checks(DateTime.t()) :: {integer(), nil | [term()]} - def delete_old_checks(older_than_date) do - Repo.delete_all(from(c in Check, where: c.checked_at < ^older_than_date)) + def change_check(%Check{} = check, attrs \\ %{}) do + Check.changeset(check, attrs) + end + + ## Check Results + + @doc """ + Creates a check result. + """ + def create_check_result(attrs \\ %{}) do + %CheckResult{} + |> CheckResult.changeset(attrs) + |> Repo.insert() end @doc """ - Gets hourly statistics for device over a time range. - Uses TimescaleDB continuous aggregates for performance when available, - falls back to raw query calculation otherwise. + Gets the latest check result for a check. """ - @spec get_hourly_stats(String.t(), DateTime.t(), DateTime.t()) :: {:ok, Postgrex.Result.t()} | {:error, Exception.t()} - def get_hourly_stats(device_id, start_time, end_time) do - # Try continuous aggregate first (much faster for large datasets) - query = """ - SELECT - bucket, - total_checks, - successful_checks, - failed_checks, - ROUND(avg_response_time_ms::numeric, 2) as avg_response_time_ms, - min_response_time_ms, - max_response_time_ms, - ROUND(100.0 * successful_checks / NULLIF(total_checks, 0), 2) as uptime_percentage - FROM monitoring_checks_hourly - WHERE device_id = $1::uuid - AND bucket >= $2 - AND bucket <= $3 - ORDER BY bucket ASC - """ - - case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time, end_time]) do - {:ok, %{rows: rows} = result} when rows != [] -> - result - - _empty_or_error -> - get_hourly_stats_raw(device_id, start_time, end_time) - end + 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 - # Fallback for environments without TimescaleDB continuous aggregates - defp get_hourly_stats_raw(device_id, start_time, end_time) do - query = """ - SELECT - date_trunc('hour', checked_at) as bucket, - COUNT(*) as total_checks, - COUNT(*) FILTER (WHERE status = 'success') as successful_checks, - COUNT(*) FILTER (WHERE status IN ('failure', 'timeout')) as failed_checks, - ROUND(AVG(response_time_ms)::numeric, 2) as avg_response_time_ms, - MIN(response_time_ms) as min_response_time_ms, - MAX(response_time_ms) as max_response_time_ms, - ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'success') / NULLIF(COUNT(*), 0), 2) as uptime_percentage - FROM monitoring_checks - WHERE device_id = $1::uuid - AND checked_at >= $2 - AND checked_at <= $3 - GROUP BY bucket - ORDER BY bucket ASC - """ + @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 - SQL.query!( - Repo, - query, - [Ecto.UUID.dump!(device_id), start_time, end_time] + 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 daily statistics for device over a time range. - Uses TimescaleDB continuous aggregates for performance when available, - falls back to raw query calculation otherwise. + Updates check state after a check execution. + Handles soft/hard state transitions. """ - def get_daily_stats(device_id, start_time, end_time) do - # Try continuous aggregate first (much faster for large datasets) - query = """ - SELECT - bucket, - total_checks, - successful_checks, - failed_checks, - ROUND(avg_response_time_ms::numeric, 2) as avg_response_time_ms, - min_response_time_ms, - max_response_time_ms, - ROUND(100.0 * successful_checks / NULLIF(total_checks, 0), 2) as uptime_percentage - FROM monitoring_checks_daily - WHERE device_id = $1::uuid - AND bucket >= $2 - AND bucket <= $3 - ORDER BY bucket ASC - """ + def update_check_state(%Check{} = check, status, output) do + now = DateTime.utc_now() + state_changed = check.current_state != status - case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time, end_time]) do - {:ok, %{rows: rows} = result} when rows != [] -> - result + {new_state_type, new_check_attempt} = + calculate_state_transition(check, status, state_changed) - _empty_or_error -> - get_daily_stats_raw(device_id, start_time, end_time) + 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, state_changed) do + if state_changed do + # State changed - increment attempt, stay soft + new_attempt = check.current_check_attempt + 1 + + if new_attempt >= check.max_check_attempts do + # Reached max attempts - transition to hard + {"hard", new_attempt} + else + # Still in soft state + {"soft", new_attempt} + end + else + # State unchanged + if check.current_state_type == "hard" do + # Already hard, stay hard + {"hard", check.current_check_attempt} + else + # Soft state, increment attempt + new_attempt = min(check.current_check_attempt + 1, check.max_check_attempts) + + if new_attempt >= check.max_check_attempts do + {"hard", new_attempt} + else + {"soft", new_attempt} + end + end end end - - # Fallback for environments without TimescaleDB continuous aggregates - defp get_daily_stats_raw(device_id, start_time, end_time) do - query = """ - SELECT - date_trunc('day', checked_at) as bucket, - COUNT(*) as total_checks, - COUNT(*) FILTER (WHERE status = 'success') as successful_checks, - COUNT(*) FILTER (WHERE status IN ('failure', 'timeout')) as failed_checks, - ROUND(AVG(response_time_ms)::numeric, 2) as avg_response_time_ms, - MIN(response_time_ms) as min_response_time_ms, - MAX(response_time_ms) as max_response_time_ms, - ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'success') / NULLIF(COUNT(*), 0), 2) as uptime_percentage - FROM monitoring_checks - WHERE device_id = $1::uuid - AND checked_at >= $2 - AND checked_at <= $3 - GROUP BY bucket - ORDER BY bucket ASC - """ - - SQL.query!( - Repo, - query, - [Ecto.UUID.dump!(device_id), start_time, end_time] - ) - end - - @doc """ - Gets uptime percentage for device over the last N days. - Uses TimescaleDB continuous aggregates for performance when available, - falls back to raw query calculation otherwise. - """ - def get_uptime_percentage(device_id, days_ago \\ 30) do - start_time = DateTime.add(DateTime.utc_now(), -days_ago * 24 * 60 * 60, :second) - - # Try continuous aggregate first (aggregates daily data for fast calculation) - query = """ - SELECT - ROUND(100.0 * SUM(successful_checks) / NULLIF(SUM(total_checks), 0), 2) as uptime_percentage - FROM monitoring_checks_daily - WHERE device_id = $1::uuid - AND bucket >= $2 - """ - - case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time]) do - {:ok, %{rows: [[%Decimal{} = uptime]]}} -> - uptime |> Decimal.to_float() |> Float.round(2) - - _empty_or_error -> - get_uptime_percentage_raw(device_id, start_time) - end - end - - # Fallback for environments without TimescaleDB continuous aggregates - defp get_uptime_percentage_raw(device_id, start_time) do - query = """ - SELECT - ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'success') / NULLIF(COUNT(*), 0), 2) as uptime_percentage - FROM monitoring_checks - WHERE device_id = $1::uuid - AND checked_at >= $2 - """ - - case SQL.query!(Repo, query, [Ecto.UUID.dump!(device_id), start_time]) do - %{rows: [[%Decimal{} = uptime]]} -> - uptime |> Decimal.to_float() |> Float.round(2) - - _ -> - nil - end - end - - @doc """ - Gets latency data for graphing ping response times. - - Returns successful monitoring checks with response times for the specified time range. - Failed checks are excluded as they have no latency data. - - ## Options - * `:since` - Only return checks after this datetime - * `:limit` - Maximum number of checks to return (default: 2880) - - ## Examples - - iex> get_latency_data(device_id, since: DateTime.utc_now() |> DateTime.add(-24, :hour), limit: 100) - [%Check{response_time_ms: 50, checked_at: ~U[2025-12-21 12:00:00Z]}, ...] - - """ - def get_latency_data(device_id, opts \\ []) do - since = Keyword.get(opts, :since, DateTime.add(DateTime.utc_now(), -24, :hour)) - limit = Keyword.get(opts, :limit, 2880) - - Check - |> where([c], c.device_id == ^device_id) - |> where([c], c.status == :success) - |> where([c], c.checked_at >= ^since) - |> where([c], not is_nil(c.response_time_ms)) - |> order_by([c], desc: c.checked_at) - |> limit(^limit) - |> Repo.all() - end end diff --git a/lib/towerops/monitoring/check.ex b/lib/towerops/monitoring/check.ex index ef1bd675..da470387 100644 --- a/lib/towerops/monitoring/check.ex +++ b/lib/towerops/monitoring/check.ex @@ -1,42 +1,172 @@ defmodule Towerops.Monitoring.Check do @moduledoc """ - Monitoring check schema for device reachability status. + Generic check schema following Icinga2's Checkable pattern. - Records the result of each monitoring check (success/failure) with latency - and tracks which agent performed the check for intelligent routing. + Supports multiple check types (http, tcp, dns, passive, ping) with + type-specific configuration stored in JSONB `config` field. + + Following Icinga2's design: + - Generic check fields (interval, retry, max_attempts, state tracking) + - Check-specific config in `config` map (like Icinga2's vars) + - Flapping detection + - Soft/hard state transitions + - Optional device association (can be standalone) """ use Ecto.Schema import Ecto.Changeset + alias Towerops.Agents.AgentToken + alias Towerops.Devices.Device + alias Towerops.Monitoring.CheckResult + alias Towerops.Organizations.Organization + @primary_key {:id, :binary_id, autogenerate: true} @foreign_key_type :binary_id - schema "monitoring_checks" do - field :status, Ecto.Enum, values: [:success, :failure] - field :response_time_ms, :float - field :checked_at, :utc_datetime - belongs_to :device, Towerops.Devices.Device - belongs_to :agent_token, Towerops.Agents.AgentToken + @check_types ~w(http tcp dns passive ping) + @state_values 0..3 + @state_types ~w(soft hard) - timestamps(type: :utc_datetime, updated_at: false) + schema "checks" do + field :name, :string + field :check_type, :string + field :description, :string + + # Generic check settings (from Icinga2's Checkable) + field :interval_seconds, :integer, default: 60 + field :retry_interval_seconds, :integer, default: 30 + field :max_check_attempts, :integer, default: 3 + field :timeout_ms, :integer, default: 5000 + field :enabled, :boolean, default: true + + # Check-specific configuration (JSONB, like Icinga2's vars) + field :config, :map, default: %{} + + # State tracking (soft/hard states like Icinga2) + field :current_state, :integer, default: 3 + field :current_state_type, :string, default: "soft" + field :current_check_attempt, :integer, default: 1 + field :last_check_at, :utc_datetime + field :last_state_change_at, :utc_datetime + field :last_hard_state_change_at, :utc_datetime + + # Flapping detection + field :enable_flapping, :boolean, default: false + field :flapping_threshold_low, :float, default: 25.0 + field :flapping_threshold_high, :float, default: 30.0 + field :is_flapping, :boolean, default: false + field :flapping_state_changes, :integer, default: 0 + field :flapping_window_start_at, :utc_datetime + + # Passive checks + field :enable_passive_checks, :boolean, default: false + field :freshness_threshold_seconds, :integer + field :check_key, :string + + belongs_to :organization, Organization + belongs_to :device, Device + belongs_to :agent_token, AgentToken + + timestamps(type: :utc_datetime) end - @doc false def changeset(check, attrs) do check - |> cast(attrs, [:device_id, :agent_token_id, :status, :response_time_ms, :checked_at]) - |> validate_required([:device_id, :status, :checked_at]) - |> round_response_time() - |> foreign_key_constraint(:device_id, match: :suffix) - |> foreign_key_constraint(:agent_token_id, match: :suffix) + |> cast(attrs, [ + :name, + :check_type, + :description, + :interval_seconds, + :retry_interval_seconds, + :max_check_attempts, + :timeout_ms, + :enabled, + :config, + :enable_flapping, + :flapping_threshold_low, + :flapping_threshold_high, + :enable_passive_checks, + :freshness_threshold_seconds, + :check_key, + :organization_id, + :device_id, + :agent_token_id + ]) + |> validate_required([:name, :check_type, :organization_id, :config]) + |> validate_inclusion(:check_type, @check_types) + |> validate_number(:interval_seconds, greater_than: 0) + |> validate_number(:retry_interval_seconds, greater_than: 0) + |> validate_number(:max_check_attempts, greater_than: 0) + |> validate_number(:timeout_ms, greater_than: 0) + |> unique_constraint(:check_key) + |> foreign_key_constraint(:organization_id) + |> foreign_key_constraint(:device_id) + |> foreign_key_constraint(:agent_token_id) + |> validate_config_by_type() + |> generate_check_key_if_passive() end - defp round_response_time(changeset) do - case get_change(changeset, :response_time_ms) do - nil -> changeset - ms when is_float(ms) -> put_change(changeset, :response_time_ms, Float.round(ms, 1)) + defp validate_config_by_type(changeset) do + check_type = get_field(changeset, :check_type) + config = get_field(changeset, :config) || %{} + + case check_type do + "http" -> validate_http_config(changeset, config) + "tcp" -> validate_tcp_config(changeset, config) + "dns" -> validate_dns_config(changeset, config) _ -> changeset end end + + defp validate_http_config(changeset, config) do + if Map.has_key?(config, "url") do + changeset + else + add_error(changeset, :config, "HTTP check requires 'url' in config") + end + end + + defp validate_tcp_config(changeset, config) do + cond do + not Map.has_key?(config, "host") -> + add_error(changeset, :config, "TCP check requires 'host' in config") + + not Map.has_key?(config, "port") -> + add_error(changeset, :config, "TCP check requires 'port' in config") + + true -> + changeset + end + end + + defp validate_dns_config(changeset, config) do + if Map.has_key?(config, "hostname") do + changeset + else + add_error(changeset, :config, "DNS check requires 'hostname' in config") + end + end + + defp generate_check_key_if_passive(changeset) do + check_type = get_field(changeset, :check_type) + check_key = get_field(changeset, :check_key) + + if check_type == "passive" and is_nil(check_key) do + put_change(changeset, :check_key, generate_random_key()) + else + changeset + end + end + + defp generate_random_key do + 32 + |> :crypto.strong_rand_bytes() + |> Base.url_encode64(padding: false) + end + + def state_label(0), do: "OK" + def state_label(1), do: "WARNING" + def state_label(2), do: "CRITICAL" + def state_label(3), do: "UNKNOWN" end diff --git a/lib/towerops/monitoring/check_result.ex b/lib/towerops/monitoring/check_result.ex new file mode 100644 index 00000000..ff055319 --- /dev/null +++ b/lib/towerops/monitoring/check_result.ex @@ -0,0 +1,38 @@ +defmodule Towerops.Monitoring.CheckResult do + @moduledoc """ + Stores results from check executions in TimescaleDB hypertable. + + Results are time-series data showing check status over time. + """ + use Ecto.Schema + + import Ecto.Changeset + + alias Towerops.Agents.AgentToken + alias Towerops.Monitoring.Check + alias Towerops.Organizations.Organization + + @primary_key {:id, :binary_id, autogenerate: true} + @foreign_key_type :binary_id + + schema "check_results" do + field :checked_at, :utc_datetime + field :status, :integer + field :output, :string + field :response_time_ms, :float + + belongs_to :organization, Organization + belongs_to :check, Check + belongs_to :agent_token, AgentToken + end + + def changeset(result, attrs) do + result + |> cast(attrs, [:checked_at, :status, :output, :response_time_ms, :organization_id, :check_id, :agent_token_id]) + |> validate_required([:checked_at, :status, :organization_id, :check_id]) + |> validate_inclusion(:status, 0..3) + |> foreign_key_constraint(:organization_id) + |> foreign_key_constraint(:check_id) + |> foreign_key_constraint(:agent_token_id) + end +end diff --git a/lib/towerops/monitoring/executor.ex b/lib/towerops/monitoring/executor.ex new file mode 100644 index 00000000..c162a1a4 --- /dev/null +++ b/lib/towerops/monitoring/executor.ex @@ -0,0 +1,56 @@ +defmodule Towerops.Monitoring.Executor do + @moduledoc """ + Dispatcher for executing checks based on check_type. + + Routes to type-specific executors (HTTP, TCP, DNS, etc.). + """ + + alias Towerops.Monitoring.Check + alias Towerops.Monitoring.Executors.DnsExecutor + alias Towerops.Monitoring.Executors.HttpExecutor + alias Towerops.Monitoring.Executors.TcpExecutor + + @doc """ + Executes a check and returns the result. + + Returns: + - {:ok, response_time_ms, output} on success + - {:error, reason} on failure + """ + def execute(%Check{check_type: "http", config: config, timeout_ms: timeout}) do + HttpExecutor.execute(config, timeout) + end + + def execute(%Check{check_type: "tcp", config: config, timeout_ms: timeout}) do + TcpExecutor.execute(config, timeout) + end + + def execute(%Check{check_type: "dns", config: config, timeout_ms: timeout}) do + DnsExecutor.execute(config, timeout) + end + + def execute(%Check{check_type: "ping"}) do + # Ping is handled by DeviceMonitorWorker, not here + {:error, "Ping checks are handled separately"} + end + + def execute(%Check{check_type: "passive"}) do + # Passive checks don't execute - they receive results + {:error, "Passive checks do not execute actively"} + end + + def execute(%Check{check_type: type}) do + {:error, "Unknown check type: #{type}"} + end + + @doc """ + Converts executor result to check status code. + + Returns: + - 0 = OK + - 2 = CRITICAL (for errors) + - 3 = UNKNOWN (for unexpected errors) + """ + def result_to_status({:ok, _response_time, _output}), do: 0 + def result_to_status({:error, _reason}), do: 2 +end diff --git a/lib/towerops/monitoring/executors/dns_executor.ex b/lib/towerops/monitoring/executors/dns_executor.ex new file mode 100644 index 00000000..e98b6812 --- /dev/null +++ b/lib/towerops/monitoring/executors/dns_executor.ex @@ -0,0 +1,163 @@ +defmodule Towerops.Monitoring.Executors.DnsExecutor do + @moduledoc """ + Executes DNS resolution checks using :inet_res. + + Config format: + %{ + "hostname" => "example.com", + "server" => "8.8.8.8", # optional, DNS server (uses system default if nil) + "record_type" => "A", # optional, default: A + "expected" => "93.184.216.34" # optional, expected result + } + """ + + require Logger + + @default_timeout 5000 + + @doc """ + Executes a DNS check. + + Returns: + - {:ok, response_time_ms, output} on success + - {:error, reason} on failure + """ + def execute(config, timeout_ms \\ @default_timeout) do + hostname = Map.fetch!(config, "hostname") + server = Map.get(config, "server") + record_type = Map.get(config, "record_type", "A") + expected = Map.get(config, "expected") + + # Convert to Erlang atoms/charlists + hostname_charlist = String.to_charlist(hostname) + record_type_atom = record_type_to_atom(record_type) + + start_time = System.monotonic_time(:millisecond) + + result = + if server do + # Use specific DNS server + server_tuple = parse_server(server) + + :inet_res.resolve(hostname_charlist, :in, record_type_atom, [ + {:nameservers, [server_tuple]}, + {:timeout, timeout_ms} + ]) + else + # Use system default DNS + :inet_res.resolve(hostname_charlist, :in, record_type_atom, [{:timeout, timeout_ms}]) + end + + case result do + {:ok, dns_rec} -> + response_time = System.monotonic_time(:millisecond) - start_time + process_dns_response(dns_rec, record_type_atom, expected, response_time) + + {:error, :timeout} -> + {:error, "DNS query timeout after #{timeout_ms}ms"} + + {:error, :nxdomain} -> + {:error, "Domain not found (NXDOMAIN)"} + + {:error, :servfail} -> + {:error, "DNS server failure (SERVFAIL)"} + + {:error, reason} -> + {:error, "DNS resolution failed: #{inspect(reason)}"} + end + rescue + e -> + Logger.error("DNS check exception: #{inspect(e)}") + {:error, "Exception: #{Exception.message(e)}"} + end + + defp record_type_to_atom("A"), do: :a + defp record_type_to_atom("AAAA"), do: :aaaa + defp record_type_to_atom("CNAME"), do: :cname + defp record_type_to_atom("MX"), do: :mx + defp record_type_to_atom("TXT"), do: :txt + defp record_type_to_atom("NS"), do: :ns + defp record_type_to_atom("PTR"), do: :ptr + defp record_type_to_atom(_), do: :a + + defp parse_server(server_str) do + case String.split(server_str, ".") do + [a, b, c, d] -> + {String.to_integer(a), String.to_integer(b), String.to_integer(c), String.to_integer(d), 53} + + _ -> + {8, 8, 8, 8, 53} + end + end + + defp process_dns_response(dns_rec, record_type, expected, response_time) do + answers = extract_answers(dns_rec, record_type) + + cond do + Enum.empty?(answers) -> + {:error, "No #{record_type} records found"} + + expected && !Enum.any?(answers, &(&1 == expected)) -> + {:error, "Expected '#{expected}', got: #{Enum.join(answers, ", ")}"} + + true -> + {:ok, response_time, "Resolved to: #{Enum.join(answers, ", ")}"} + end + end + + defp extract_answers(dns_rec, :a) do + dns_rec + |> :inet_dns.msg(:anlist) + |> Enum.filter(fn rr -> :inet_dns.rr(rr, :type) == :a end) + |> Enum.map(fn rr -> + {a, b, c, d} = :inet_dns.rr(rr, :data) + "#{a}.#{b}.#{c}.#{d}" + end) + end + + defp extract_answers(dns_rec, :aaaa) do + dns_rec + |> :inet_dns.msg(:anlist) + |> Enum.filter(fn rr -> :inet_dns.rr(rr, :type) == :aaaa end) + |> Enum.map(fn rr -> + data = :inet_dns.rr(rr, :data) + format_ipv6(data) + end) + end + + defp extract_answers(dns_rec, :cname) do + dns_rec + |> :inet_dns.msg(:anlist) + |> Enum.filter(fn rr -> :inet_dns.rr(rr, :type) == :cname end) + |> Enum.map(fn rr -> + rr |> :inet_dns.rr(:data) |> to_string() + end) + end + + defp extract_answers(dns_rec, :mx) do + dns_rec + |> :inet_dns.msg(:anlist) + |> Enum.filter(fn rr -> :inet_dns.rr(rr, :type) == :mx end) + |> Enum.map(fn rr -> + {priority, exchange} = :inet_dns.rr(rr, :data) + "#{priority} #{exchange}" + end) + end + + defp extract_answers(dns_rec, :txt) do + dns_rec + |> :inet_dns.msg(:anlist) + |> Enum.filter(fn rr -> :inet_dns.rr(rr, :type) == :txt end) + |> Enum.map(fn rr -> + rr |> :inet_dns.rr(:data) |> Enum.join("") + end) + end + + defp extract_answers(_dns_rec, _type), do: [] + + defp format_ipv6({a, b, c, d, e, f, g, h}) do + [a, b, c, d, e, f, g, h] + |> Enum.map_join(":", &Integer.to_string(&1, 16)) + |> String.downcase() + end +end diff --git a/lib/towerops/monitoring/executors/http_executor.ex b/lib/towerops/monitoring/executors/http_executor.ex new file mode 100644 index 00000000..afd21a1b --- /dev/null +++ b/lib/towerops/monitoring/executors/http_executor.ex @@ -0,0 +1,106 @@ +defmodule Towerops.Monitoring.Executors.HttpExecutor do + @moduledoc """ + Executes HTTP/HTTPS checks using Req library. + + Config format: + %{ + "url" => "https://example.com/api/health", + "method" => "GET", # optional, default: GET + "expected_status" => 200, # optional, default: 200 + "verify_ssl" => true, # optional, default: true + "headers" => %{"Authorization" => "Bearer ..."}, # optional + "body" => "...", # optional, for POST/PUT + "regex" => "OK", # optional, content match regex + "follow_redirects" => true # optional, default: true + } + """ + + require Logger + + @default_timeout 5000 + + @doc """ + Executes an HTTP check. + + Returns: + - {:ok, response_time_ms, output} on success + - {:error, reason} on failure + """ + def execute(config, timeout_ms \\ @default_timeout) do + url = Map.fetch!(config, "url") + method = String.downcase(Map.get(config, "method", "GET")) + expected_status = Map.get(config, "expected_status", 200) + verify_ssl = Map.get(config, "verify_ssl", true) + headers = Map.get(config, "headers", %{}) + body = Map.get(config, "body") + regex = Map.get(config, "regex") + follow_redirects = Map.get(config, "follow_redirects", true) + + start_time = System.monotonic_time(:millisecond) + + req_opts = [ + method: String.to_atom(method), + url: url, + headers: headers, + receive_timeout: timeout_ms, + connect_options: [ + transport_opts: [ + verify: if(verify_ssl, do: :verify_peer, else: :verify_none) + ] + ], + redirect: follow_redirects + ] + + req_opts = + if body do + Keyword.put(req_opts, :body, body) + else + req_opts + end + + case Req.request(req_opts) do + {:ok, %Req.Response{status: status, body: response_body}} -> + response_time = System.monotonic_time(:millisecond) - start_time + + cond do + status != expected_status -> + {:error, "HTTP #{status}, expected #{expected_status}"} + + regex && !content_matches?(response_body, regex) -> + {:error, "Content does not match pattern: #{regex}"} + + true -> + {:ok, response_time, "HTTP #{status} OK"} + end + + {:error, %Req.TransportError{reason: :timeout}} -> + {:error, "Connection timeout after #{timeout_ms}ms"} + + {:error, %Req.TransportError{reason: :econnrefused}} -> + {:error, "Connection refused"} + + {:error, %Req.TransportError{reason: reason}} -> + {:error, "Transport error: #{inspect(reason)}"} + + {:error, reason} -> + {:error, "Request failed: #{inspect(reason)}"} + end + rescue + e -> + Logger.error("HTTP check exception: #{inspect(e)}") + {:error, "Exception: #{Exception.message(e)}"} + end + + defp content_matches?(body, regex_str) when is_binary(body) do + case Regex.compile(regex_str) do + {:ok, regex} -> + Regex.match?(regex, body) + + {:error, _} -> + Logger.warning("Invalid regex pattern: #{regex_str}") + false + end + end + + defp content_matches?(_body, _regex), do: false +end diff --git a/lib/towerops/monitoring/executors/tcp_executor.ex b/lib/towerops/monitoring/executors/tcp_executor.ex new file mode 100644 index 00000000..2d77adc0 --- /dev/null +++ b/lib/towerops/monitoring/executors/tcp_executor.ex @@ -0,0 +1,104 @@ +defmodule Towerops.Monitoring.Executors.TcpExecutor do + @moduledoc """ + Executes TCP port connectivity checks using :gen_tcp. + + Config format: + %{ + "host" => "192.168.1.1", + "port" => 3306, + "send" => "...", # optional, data to send after connection + "expect" => "..." # optional, expected response string + } + """ + + require Logger + + @default_timeout 5000 + + @doc """ + Executes a TCP check. + + Returns: + - {:ok, response_time_ms, output} on success + - {:error, reason} on failure + """ + def execute(config, timeout_ms \\ @default_timeout) do + host = Map.fetch!(config, "host") + port = Map.fetch!(config, "port") + send_data = Map.get(config, "send") + expect_data = Map.get(config, "expect") + + # Convert host to charlist for :gen_tcp + host_charlist = String.to_charlist(host) + + start_time = System.monotonic_time(:millisecond) + + case :gen_tcp.connect(host_charlist, port, [:binary, active: false], timeout_ms) do + {:ok, socket} -> + result = + try do + response_time = System.monotonic_time(:millisecond) - start_time + + if send_data do + send_and_receive(socket, send_data, expect_data, timeout_ms) + else + {:ok, response_time, "TCP port #{port} open"} + end + after + :gen_tcp.close(socket) + end + + result + + {:error, :timeout} -> + {:error, "Connection timeout after #{timeout_ms}ms"} + + {:error, :econnrefused} -> + {:error, "Connection refused"} + + {:error, :ehostunreach} -> + {:error, "Host unreachable"} + + {:error, :nxdomain} -> + {:error, "Host not found (DNS resolution failed)"} + + {:error, reason} -> + {:error, "Connection failed: #{inspect(reason)}"} + end + rescue + e -> + Logger.error("TCP check exception: #{inspect(e)}") + {:error, "Exception: #{Exception.message(e)}"} + end + + defp send_and_receive(socket, send_data, expect_data, timeout_ms) do + case :gen_tcp.send(socket, send_data) do + :ok -> + if expect_data do + receive_and_check(socket, expect_data, timeout_ms) + else + {:ok, 0, "Data sent successfully"} + end + + {:error, reason} -> + {:error, "Send failed: #{inspect(reason)}"} + end + end + + defp receive_and_check(socket, expect_data, timeout_ms) do + case :gen_tcp.recv(socket, 0, timeout_ms) do + {:ok, data} -> + if String.contains?(data, expect_data) do + {:ok, 0, "Received expected response"} + else + {:error, "Unexpected response: #{inspect(data)}"} + end + + {:error, :timeout} -> + {:error, "Receive timeout"} + + {:error, reason} -> + {:error, "Receive failed: #{inspect(reason)}"} + end + end +end diff --git a/lib/towerops/workers/check_worker.ex b/lib/towerops/workers/check_worker.ex new file mode 100644 index 00000000..e45c273a --- /dev/null +++ b/lib/towerops/workers/check_worker.ex @@ -0,0 +1,249 @@ +defmodule Towerops.Workers.CheckWorker do + @moduledoc """ + Oban worker that executes service checks (HTTP, TCP, DNS). + + Follows same self-scheduling pattern as DeviceMonitorWorker. + Phoenix executes checks unless assigned to an agent (agent_token_id set). + """ + use Oban.Worker, + queue: :checks, + unique: [ + period: 60, + keys: [:check_id], + states: [:available, :scheduled, :executing, :retryable] + ] + + alias Towerops.Agents + alias Towerops.Alerts + alias Towerops.Monitoring + alias Towerops.Monitoring.Executor + alias Towerops.Snmp.Client + alias Towerops.Workers.PollingOffset + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{args: %{"check_id" => check_id}}) do + case Monitoring.get_check(check_id) do + nil -> + Logger.debug("Check #{check_id} no longer exists, skipping and not rescheduling") + :ok + + check -> + if should_continue_checking?(check) do + maybe_perform_check(check) + schedule_next_check_with_error_handling(check_id, check.interval_seconds) + else + Logger.debug("Check #{check_id} disabled or assigned to agent, not rescheduling", + check_id: check_id, + enabled: check.enabled, + agent_assigned: !is_nil(check.agent_token_id) + ) + end + + :ok + end + end + + # Check should continue if: + # - Enabled + # - Not a passive check (passive checks don't execute actively) + # - Either no agent assigned OR agent is cloud poller (Phoenix executes) + defp should_continue_checking?(check) do + check.enabled && + check.check_type != "passive" && + !Client.phoenix_snmp_disabled() && + should_phoenix_execute?(check) + end + + defp should_phoenix_execute?(check) do + case check.agent_token_id do + nil -> + # No agent assigned - Phoenix executes + true + + agent_token_id -> + # Agent assigned - only Phoenix executes if it's cloud poller + try do + agent_token = Agents.get_agent_token!(agent_token_id) + agent_token.is_cloud_poller + rescue + Ecto.NoResultsError -> true + end + end + end + + defp maybe_perform_check(check) do + if should_phoenix_execute?(check) do + perform_check(check) + else + Logger.debug("Skipping Phoenix execution for check #{check.name} - assigned to remote agent") + :ok + end + end + + defp schedule_next_check_with_error_handling(check_id, interval_seconds) do + case Monitoring.get_check(check_id) do + nil -> + Logger.debug("Check deleted, not rescheduling", check_id: check_id) + :ok + + _check -> + case schedule_next_check(check_id, interval_seconds) do + {:ok, _job} -> + :ok + + {:error, changeset} -> + Logger.error("Failed to schedule next check for #{check_id}: #{inspect(changeset.errors)}") + :ok + end + end + end + + @doc """ + Starts executing a check (schedules first execution). + """ + def start_check(check_id, interval_seconds \\ 60) do + offset = PollingOffset.calculate_offset(check_id, interval_seconds) + + %{check_id: check_id} + |> new(schedule_in: offset) + |> Oban.insert() + end + + @doc """ + Stops executing a check by cancelling its jobs. + """ + def stop_check(check_id) do + import Ecto.Query + + Oban.cancel_all_jobs( + from(j in Oban.Job, + where: j.worker == "Towerops.Workers.CheckWorker", + where: fragment("args->>'check_id' = ?", ^check_id), + where: j.state in ["available", "scheduled", "executing", "retryable"] + ) + ) + end + + @doc """ + Triggers an immediate check execution. + """ + def trigger_check(check_id) do + %{check_id: check_id} + |> new() + |> Oban.insert() + end + + # Private Functions + + defp perform_check(check) do + Logger.debug("Executing check: #{check.name} (#{check.check_type})", check_id: check.id) + + # Execute the check + result = Executor.execute(check) + + # Convert result to status and extract details + {status, response_time, output} = + case result do + {:ok, time, msg} -> {0, time, msg} + {:error, reason} -> {2, nil, reason} + end + + # Record the result + now = DateTime.utc_now() + + case Monitoring.create_check_result(%{ + check_id: check.id, + organization_id: check.organization_id, + status: status, + output: output, + response_time_ms: response_time, + checked_at: now + }) do + {:ok, _result} -> + Logger.debug("Recorded check result: #{output}", check_id: check.id, status: status) + + {:error, changeset} -> + Logger.error("Failed to create check result: #{inspect(changeset.errors)}") + end + + # Update check state (handles soft/hard state transitions) + old_state = check.current_state + {:ok, updated_check} = Monitoring.update_check_state(check, status, output) + + # Handle state changes (create/resolve alerts) + if state_changed_to_problem?(old_state, updated_check) do + handle_check_problem(updated_check, output) + end + + if state_changed_to_ok?(old_state, updated_check) do + handle_check_recovery(updated_check) + end + + :ok + end + + defp state_changed_to_problem?(old_state, check) do + # State changed from OK to CRITICAL/WARNING and is now hard state + old_state == 0 && check.current_state in [1, 2] && check.current_state_type == "hard" + end + + defp state_changed_to_ok?(old_state, check) do + # State changed from CRITICAL/WARNING to OK + old_state in [1, 2] && check.current_state == 0 + end + + defp handle_check_problem(check, output) do + Logger.warning("Check #{check.name} entered problem state", + check_id: check.id, + state: check.current_state, + output: output + ) + + # Create alert if one doesn't exist + if !Alerts.has_active_check_alert?(check.id) do + create_check_alert(check, output) + end + end + + defp handle_check_recovery(check) do + Logger.info("Check #{check.name} recovered", + check_id: check.id + ) + + # Resolve any active alerts + Alerts.resolve_check_alerts(check.id) + end + + defp create_check_alert(check, output) do + severity = if check.current_state == 1, do: 1, else: 2 + + attrs = %{ + check_id: check.id, + device_id: check.device_id, + organization_id: check.organization_id, + alert_type: "check_#{check.check_type}", + severity: severity, + message: "Check '#{check.name}' is #{Monitoring.Check.state_label(check.current_state)}", + output: output, + triggered_at: DateTime.utc_now() + } + + case Alerts.create_alert(attrs) do + {:ok, alert} -> + Logger.info("Created alert for check #{check.name}", alert_id: alert.id) + + {:error, changeset} -> + Logger.error("Failed to create alert: #{inspect(changeset.errors)}") + end + end + + defp schedule_next_check(check_id, interval_seconds) do + offset = PollingOffset.calculate_offset(check_id, interval_seconds) + + %{check_id: check_id} + |> new(schedule_in: offset) + |> Oban.insert() + end +end diff --git a/priv/repo/migrations/20260212185422_add_checks_table.exs b/priv/repo/migrations/20260212185422_add_checks_table.exs new file mode 100644 index 00000000..48a03644 --- /dev/null +++ b/priv/repo/migrations/20260212185422_add_checks_table.exs @@ -0,0 +1,61 @@ +defmodule Towerops.Repo.Migrations.AddChecksTable do + use Ecto.Migration + + def change do + create table(:checks, primary_key: false) do + add :id, :binary_id, primary_key: true + + add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), + null: false + + add :device_id, references(:devices, type: :binary_id, on_delete: :delete_all) + + add :name, :string, null: false + add :check_type, :string, null: false + add :description, :text + + # Generic check settings (following Icinga2's Checkable pattern) + add :interval_seconds, :integer, null: false, default: 60 + add :retry_interval_seconds, :integer, null: false, default: 30 + add :max_check_attempts, :integer, null: false, default: 3 + add :timeout_ms, :integer, null: false, default: 5000 + add :enabled, :boolean, null: false, default: true + + # Check-specific configuration (JSONB, like Icinga2's vars) + add :config, :map, null: false, default: %{} + + # State tracking + add :current_state, :integer, null: false, default: 3 + add :current_state_type, :string, null: false, default: "soft" + add :current_check_attempt, :integer, null: false, default: 1 + add :last_check_at, :utc_datetime + add :last_state_change_at, :utc_datetime + add :last_hard_state_change_at, :utc_datetime + + # Flapping detection + add :enable_flapping, :boolean, null: false, default: false + add :flapping_threshold_low, :float, null: false, default: 25.0 + add :flapping_threshold_high, :float, null: false, default: 30.0 + add :is_flapping, :boolean, null: false, default: false + add :flapping_state_changes, :integer, null: false, default: 0 + add :flapping_window_start_at, :utc_datetime + + # Passive checks + add :enable_passive_checks, :boolean, null: false, default: false + add :freshness_threshold_seconds, :integer + add :check_key, :string + + # Agent assignment (which agent executes this check) + add :agent_token_id, references(:agent_tokens, type: :binary_id, on_delete: :nilify_all) + + timestamps(type: :utc_datetime) + end + + create index(:checks, [:organization_id]) + create index(:checks, [:device_id]) + create index(:checks, [:check_type]) + create index(:checks, [:enabled]) + create index(:checks, [:agent_token_id]) + create unique_index(:checks, [:check_key], where: "check_key IS NOT NULL") + end +end diff --git a/priv/repo/migrations/20260212185433_add_check_results_hypertable.exs b/priv/repo/migrations/20260212185433_add_check_results_hypertable.exs new file mode 100644 index 00000000..f3494fc1 --- /dev/null +++ b/priv/repo/migrations/20260212185433_add_check_results_hypertable.exs @@ -0,0 +1,41 @@ +defmodule Towerops.Repo.Migrations.AddCheckResultsHypertable do + use Ecto.Migration + + def up do + create table(:check_results, primary_key: false) do + add :id, :binary_id, null: false + add :checked_at, :utc_datetime, null: false + + add :organization_id, references(:organizations, type: :binary_id, on_delete: :delete_all), + null: false + + add :check_id, references(:checks, type: :binary_id, on_delete: :delete_all), null: false + + add :status, :integer, null: false + add :output, :text + add :response_time_ms, :float + + add :agent_token_id, references(:agent_tokens, type: :binary_id, on_delete: :nilify_all) + end + + # Composite primary key for TimescaleDB (id + partitioning column) + execute( + "ALTER TABLE check_results ADD PRIMARY KEY (id, checked_at)", + "ALTER TABLE check_results DROP CONSTRAINT check_results_pkey" + ) + + # Convert to TimescaleDB hypertable + execute( + "SELECT create_hypertable('check_results', 'checked_at')", + "DROP TABLE check_results" + ) + + create index(:check_results, [:check_id, :checked_at]) + create index(:check_results, [:organization_id, :status]) + create index(:check_results, [:organization_id, :checked_at]) + end + + def down do + drop table(:check_results) + end +end diff --git a/priv/repo/migrations/20260212185506_extend_alerts_for_checks.exs b/priv/repo/migrations/20260212185506_extend_alerts_for_checks.exs index b1035675..465dc7bc 100644 --- a/priv/repo/migrations/20260212185506_extend_alerts_for_checks.exs +++ b/priv/repo/migrations/20260212185506_extend_alerts_for_checks.exs @@ -2,5 +2,26 @@ defmodule Towerops.Repo.Migrations.ExtendAlertsForChecks do use Ecto.Migration def change do + alter table(:alerts) do + # Add check_id reference (nullable for backward compatibility) + add :check_id, references(:checks, type: :binary_id, on_delete: :delete_all) + + # Add severity (1=WARNING, 2=CRITICAL) + add :severity, :integer, default: 2 + + # Track notification delivery + add :notification_sent, :boolean, default: false + add :notification_sent_at, :utc_datetime + + # Output from the check + add :output, :text + end + + # Make device_id nullable since checks can be standalone + execute "ALTER TABLE alerts ALTER COLUMN device_id DROP NOT NULL", "" + + create index(:alerts, [:check_id]) + create index(:alerts, [:severity]) + create index(:alerts, [:notification_sent]) end end