From 1f0bf132d38cb81eca9ea8cfd463d5d06fd4431c Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 16:51:28 +0100 Subject: [PATCH 1/3] feat: add custom alerts by config --- .env.example | 3 + README.md | 1 - config/config.exs | 5 + config/runtime.exs | 30 +++ docker/docker-compose.tel.yml | 1 + etc/grafana/dashboards/meteosix_wrf.json | 28 +++ etc/grafana/datasource.yml | 1 + .../provisioning/alerting/contact_points.yaml | 10 + .../alerting/notification_policies.yaml | 10 + etc/grafana/provisioning/alerting/rules.yaml | 81 ++++++++ lib/chronodash/alerting/dispatcher.ex | 47 +++++ lib/chronodash/alerting/manager.ex | 183 ++++++++++++++++++ lib/chronodash/alerting/provider.ex | 7 + lib/chronodash/alerting/providers/discord.ex | 31 +++ lib/chronodash/application.ex | 13 +- lib/chronodash/polling/worker.ex | 18 +- 16 files changed, 464 insertions(+), 5 deletions(-) create mode 100644 etc/grafana/provisioning/alerting/contact_points.yaml create mode 100644 etc/grafana/provisioning/alerting/notification_policies.yaml create mode 100644 etc/grafana/provisioning/alerting/rules.yaml create mode 100644 lib/chronodash/alerting/dispatcher.ex create mode 100644 lib/chronodash/alerting/manager.ex create mode 100644 lib/chronodash/alerting/provider.ex create mode 100644 lib/chronodash/alerting/providers/discord.ex diff --git a/.env.example b/.env.example index 5b9e602..8fea1e5 100644 --- a/.env.example +++ b/.env.example @@ -7,3 +7,6 @@ DB_PASSWORD=your_database_password_here # MeteoSIX # Get your key at https://www.meteogalicia.gal/web/servizos/meteosix/solicitudeClave.action METEOSIX_API_KEY=YOUR_KEY_HERE + +# Alerting +DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/... diff --git a/README.md b/README.md index 1cb59e9..e0062bb 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,6 @@ Chronodash is a multi-datasource monitoring and monitoring application built with **Elixir**, **Phoenix**, and the **Ash Framework**. It is designed to collect time-series metrics from various providers (starting with MeteoSIX) and visualize them in **Grafana** using **TimescaleDB**. ---- ## Key Architecture Concepts diff --git a/config/config.exs b/config/config.exs index 6c073a5..7ca38a1 100644 --- a/config/config.exs +++ b/config/config.exs @@ -111,6 +111,11 @@ config :chronodash, :polling_jobs, [ } ] +# Alerts configuration +config :chronodash, :alerts, + rules: [], + channels: %{} + # Finch configuration config :chronodash, :http_client, Chronodash.HttpClient.Finch diff --git a/config/runtime.exs b/config/runtime.exs index 211e54a..35b42aa 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -24,6 +24,36 @@ config :chronodash, :meteosix, api_key: System.get_env("METEOSIX_API_KEY") || "KEY", base_url: "https://servizos.meteogalicia.gal/apiv5" +config :chronodash, :alerts, + rules: [ + # Alert if wind speed > 30 km/h at any point in the next 24h + %{ + id: "high_wind_warning", + metric: "wind", + threshold: 30.0, + condition: :max_gt, + window_hours: 24, + cooldown_hours: 12, + message: "⚠️ High Wind Warning in %{location} in the next 24h: %{value} km/h", + channels: [:discord] + }, + # Alert if no precipitation is expected in the next 6 hours + %{ + id: "no_rain_forecast", + metric: "precipitation_amount", + threshold: 1.0, + # Maximum in the window is less than 1.0 + condition: :max_lt, + window_hours: 6, + cooldown_hours: 6, + message: "☀️ Good news! No significant rain predicted in %{location} for the next 6 hours.", + channels: [:discord] + } + ], + channels: %{ + discord: %{url: System.get_env("DISCORD_WEBHOOK_URL")} + } + if config_env() == :prod do database_url = System.get_env("DATABASE_URL") || diff --git a/docker/docker-compose.tel.yml b/docker/docker-compose.tel.yml index 57689b7..9e48776 100644 --- a/docker/docker-compose.tel.yml +++ b/docker/docker-compose.tel.yml @@ -17,6 +17,7 @@ services: volumes: - "../etc/grafana/datasource.yml:/etc/grafana/provisioning/datasources/datasource.yml" - "../etc/grafana/dashboard_provider.yml:/etc/grafana/provisioning/dashboards/main.yaml" + - "../etc/grafana/provisioning/alerting:/etc/grafana/provisioning/alerting" - "../etc/grafana/dashboards:/var/lib/grafana/dashboards" environment: - GF_SECURITY_ADMIN_PASSWORD=admin diff --git a/etc/grafana/dashboards/meteosix_wrf.json b/etc/grafana/dashboards/meteosix_wrf.json index 539ef15..3f31554 100644 --- a/etc/grafana/dashboards/meteosix_wrf.json +++ b/etc/grafana/dashboards/meteosix_wrf.json @@ -89,6 +89,34 @@ "y": 0 }, "id": 2, + "alert": { + "conditions": [ + { + "evaluator": { + "params": [5], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": ["A", "5m", "now"] + }, + "reducer": { + "params": [], + "type": "max" + }, + "type": "query" + } + ], + "executionErrorState": "alerting", + "for": "5m", + "frequency": "1m", + "handler": 1, + "name": "High Precipitation Alert", + "noDataState": "no_data", + "notifications": [] + }, "options": { "legend": { "displayMode": "list", diff --git a/etc/grafana/datasource.yml b/etc/grafana/datasource.yml index c7eefcb..9399148 100644 --- a/etc/grafana/datasource.yml +++ b/etc/grafana/datasource.yml @@ -9,6 +9,7 @@ datasources: editable: true - name: PostgreSQL + uid: postgres_db type: postgres access: proxy url: postgres:5432 diff --git a/etc/grafana/provisioning/alerting/contact_points.yaml b/etc/grafana/provisioning/alerting/contact_points.yaml new file mode 100644 index 0000000..48e150e --- /dev/null +++ b/etc/grafana/provisioning/alerting/contact_points.yaml @@ -0,0 +1,10 @@ +apiVersion: 1 + +contactPoints: + - orgId: 1 + name: Discord + receivers: + - uid: discord_alert + type: discord + settings: + url: https://discord.com/api/webhooks/1477276961423491164/gcEFNjDjVcVIHQmq3aRbeP2sqcACmlk7m81SYDWRwxwrGGYJXBGqxOtWRYQhRxA8MRyx diff --git a/etc/grafana/provisioning/alerting/notification_policies.yaml b/etc/grafana/provisioning/alerting/notification_policies.yaml new file mode 100644 index 0000000..d6df6e7 --- /dev/null +++ b/etc/grafana/provisioning/alerting/notification_policies.yaml @@ -0,0 +1,10 @@ +apiVersion: 1 + +# Notification policies for Grafana alerting +policies: + - orgId: 1 + receiver: Discord + group_by: ['alertname', 'severity'] + group_wait: 30s + group_interval: 5m + repeat_interval: 4h diff --git a/etc/grafana/provisioning/alerting/rules.yaml b/etc/grafana/provisioning/alerting/rules.yaml new file mode 100644 index 0000000..a386d32 --- /dev/null +++ b/etc/grafana/provisioning/alerting/rules.yaml @@ -0,0 +1,81 @@ +apiVersion: 1 + +groups: + - orgId: 1 + name: Weather Alerts + folder: MeteoSIX Alerts + interval: 1m + rules: + - uid: high_wind_alert + title: High Wind Warning + condition: C + data: + - refId: A + datasourceUid: postgres_db + relativeTimeRange: + from: 3600 + to: 0 + model: + rawSql: "SELECT value, timestamp FROM observations WHERE metric_type = 'wind_module' AND timestamp > now() AND timestamp < now() + interval '24 hours' ORDER BY timestamp ASC" + format: table + - refId: B + datasourceUid: __expr__ + model: + expression: A + type: reduce + reducer: max + - refId: C + datasourceUid: __expr__ + model: + expression: B + type: threshold + conditions: + - evaluator: + params: [30] + type: gt + operator: + type: and + noDataState: OK + execErrState: Error + for: 0s + annotations: + summary: "High wind predicted: {{ index $values \"B\" }} km/h" + labels: + severity: warning + + - uid: high_precip_alert + title: High Precipitation Warning + condition: C + data: + - refId: A + datasourceUid: postgres_db + relativeTimeRange: + from: 3600 + to: 0 + model: + rawSql: "SELECT value, timestamp FROM observations WHERE metric_type = 'precipitation_amount' AND timestamp > now() AND timestamp < now() + interval '6 hours' ORDER BY timestamp ASC" + format: table + - refId: B + datasourceUid: __expr__ + model: + expression: A + type: reduce + reducer: max + - refId: C + datasourceUid: __expr__ + model: + expression: B + type: threshold + conditions: + - evaluator: + params: [5] + type: gt + operator: + type: and + noDataState: OK + execErrState: Error + for: 0s + annotations: + summary: "Heavy rain predicted: {{ index $values \"B\" }} mm" + labels: + severity: critical diff --git a/lib/chronodash/alerting/dispatcher.ex b/lib/chronodash/alerting/dispatcher.ex new file mode 100644 index 0000000..0061184 --- /dev/null +++ b/lib/chronodash/alerting/dispatcher.ex @@ -0,0 +1,47 @@ +defmodule Chronodash.Alerting.Dispatcher do + @moduledoc """ + Dispatches alert messages to configured channel providers. + """ + require Logger + + alias Chronodash.Alerting.Providers + + @providers %{ + discord: Providers.Discord + } + + # ============================================================================ + # Public API + # ============================================================================ + + @doc """ + Sends a message to a list of channels. + """ + def dispatch(message, channel_keys) when is_list(channel_keys) do + config = Application.get_env(:chronodash, :alerts, []) + channel_configs = Keyword.get(config, :channels, %{}) + + Enum.each(channel_keys, fn key -> + case Map.get(channel_configs, key) do + nil -> + Logger.warning("Attempted to dispatch to unknown channel: #{key}") + + chan_config -> + do_dispatch(message, key, chan_config) + end + end) + end + + defp do_dispatch(message, key, config) do + case Map.get(@providers, key) do + nil -> + Logger.error("No provider implementation for channel type: #{key}") + + provider -> + # We wrap in Task.start to ensure alerting doesn't block the caller + Task.start(fn -> + provider.deliver(message, config) + end) + end + end +end diff --git a/lib/chronodash/alerting/manager.ex b/lib/chronodash/alerting/manager.ex new file mode 100644 index 0000000..27fc56e --- /dev/null +++ b/lib/chronodash/alerting/manager.ex @@ -0,0 +1,183 @@ +defmodule Chronodash.Alerting.Manager do + @moduledoc """ + Subscribes to telemetry, evaluates alert rules, and manages alert cooldowns. + """ + use GenServer + require Logger + alias Chronodash.Alerting.Dispatcher + + @table_name :alert_cooldowns + + # ============================================================================ + # PUBLIC API + # ============================================================================ + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @doc """ + Telemetry handler callback. Redirects to GenServer for state-aware evaluation. + """ + def handle_telemetry_event(event, measurements, metadata, _config) do + GenServer.cast(__MODULE__, {:evaluate, event, measurements, metadata}) + end + + # ============================================================================ + # GenServer Callbacks + # ============================================================================ + + @impl true + def init(_opts) do + :ets.new(@table_name, [:set, :protected, :named_table]) + + :telemetry.attach_many( + "chronodash-alert-manager", + [ + [:chronodash, :polling, :observation], + [:chronodash, :polling, :forecast_received] + ], + &__MODULE__.handle_telemetry_event/4, + nil + ) + + {:ok, %{}} + end + + @impl true + def handle_cast({:evaluate, [:chronodash, :polling, :observation], %{value: value}, metadata}, state) do + get_instant_rules() + |> Enum.each(&process_instant_rule(&1, value, metadata)) + + {:noreply, state} + end + + @impl true + def handle_cast({:evaluate, [:chronodash, :polling, :forecast_received], _measurements, metadata}, state) do + get_window_rules() + |> Enum.each(&process_window_rule(&1, metadata)) + + {:noreply, state} + end + + # ============================================================================ + # Internal Functions - Processing + # ============================================================================ + + defp process_instant_rule(rule, value, metadata) do + if not in_cooldown?(rule, metadata.location) and should_alert_point?(rule, value, metadata) do + trigger_alert(rule, value, metadata) + end + end + + defp process_window_rule(rule, metadata) do + if not in_cooldown?(rule, metadata.location) do + {triggered, peak_value} = evaluate_forecast(rule, metadata.points) + if triggered, do: trigger_alert(rule, peak_value, metadata) + end + end + + # ============================================================================ + # Internal Functions - Evaluation + # ============================================================================ + + defp evaluate_forecast(rule, points) do + window_end = DateTime.add(DateTime.utc_now(), rule.window_hours, :hour) + + points + |> filter_relevant_points(rule, window_end) + |> check_forecast_condition(rule.condition, rule.threshold) + end + + defp filter_relevant_points(points, rule, window_end) do + for p <- points, + DateTime.compare(p.timestamp, window_end) != :gt, + m <- p.metrics, + metric_name_match?(m.name, rule.metric), + do: normalize_raw_value(m.value) + end + + defp metric_name_match?(m_name, rule_metric) do + to_string(m_name) == to_string(rule_metric) or + (to_string(rule_metric) == "wind" and m_name == "wind_module") + end + + defp check_forecast_condition([], _cond, _thresh), do: {false, nil} + defp check_forecast_condition(values, :max_gt, thresh) do + max = Enum.max(values) + {max > thresh, max} + end + defp check_forecast_condition(values, :max_lt, thresh) do + max = Enum.max(values) + {max < thresh, max} + end + defp check_forecast_condition(values, :min_lt, thresh) do + min = Enum.min(values) + {min < thresh, min} + end + defp check_forecast_condition(values, :gt_any, thresh) do + val = Enum.find(values, &(&1 > thresh)) + {!is_nil(val), val} + end + defp check_forecast_condition(_, _, _), do: {false, nil} + + # ============================================================================ + # Helper Functions + # ============================================================================ + + defp in_cooldown?(rule, location) do + cooldown_ms = Map.get(rule, :cooldown_hours, 24) * 3600 * 1000 + now = System.system_time(:millisecond) + + case :ets.lookup(@table_name, {rule.id, location}) do + [{{_id, _loc}, last_triggered_at}] -> (now - last_triggered_at) < cooldown_ms + [] -> false + end + end + + defp should_alert_point?(rule, value, metadata) do + metric_match = to_string(rule.metric) == to_string(metadata.variable) + if metric_match, do: evaluate_condition(rule.condition, value, rule.threshold), else: false + end + + defp trigger_alert(rule, value, metadata) do + :ets.insert(@table_name, {{rule.id, metadata.location}, System.system_time(:millisecond)}) + message = format_message(rule.message, value, metadata) + Dispatcher.dispatch(message, rule.channels) + Logger.info("Alert triggered: #{rule.id} for #{metadata.location}. Entering cooldown.") + end + + defp normalize_raw_value(v) when is_number(v), do: v * 1.0 + defp normalize_raw_value(v) when is_binary(v) do + case Float.parse(v) do + {f, _} -> f + _ -> 0.0 + end + end + defp normalize_raw_value(_), do: 0.0 + + defp evaluate_condition(:gt, val, thresh), do: val > thresh + defp evaluate_condition(:lt, val, thresh), do: val < thresh + defp evaluate_condition(:eq, val, thresh), do: val == thresh + defp evaluate_condition(_, _, _), do: false + + defp format_message(template, value, metadata) do + template + |> String.replace("%{location}", to_string(metadata.location)) + |> String.replace("%{value}", to_string(value)) + |> String.replace("%{variable}", to_string(metadata.variable)) + end + + defp get_instant_rules do + get_rules() |> Enum.reject(&Map.has_key?(&1, :window_hours)) + end + + defp get_window_rules do + get_rules() |> Enum.filter(&Map.has_key?(&1, :window_hours)) + end + + defp get_rules do + config = Application.get_env(:chronodash, :alerts, []) + Keyword.get(config, :rules, []) + end +end diff --git a/lib/chronodash/alerting/provider.ex b/lib/chronodash/alerting/provider.ex new file mode 100644 index 0000000..2331120 --- /dev/null +++ b/lib/chronodash/alerting/provider.ex @@ -0,0 +1,7 @@ +defmodule Chronodash.Alerting.Provider do + @moduledoc """ + Behaviour for alert delivery channels (Discord, Telegram, etc). + """ + + @callback deliver(message :: String.t(), config :: map()) :: :ok | {:error, any()} +end diff --git a/lib/chronodash/alerting/providers/discord.ex b/lib/chronodash/alerting/providers/discord.ex new file mode 100644 index 0000000..65cf645 --- /dev/null +++ b/lib/chronodash/alerting/providers/discord.ex @@ -0,0 +1,31 @@ +defmodule Chronodash.Alerting.Providers.Discord do + @moduledoc """ + Discord implementation for alert delivery. + + Expects a configuration map with: + - `:url` - The Discord webhook URL to send messages to. + """ + + @behaviour Chronodash.Alerting.Provider + require Logger + + @impl true + def deliver(message, %{url: url}) when is_binary(url) do + case Req.post(url, json: %{content: message}) do + {:ok, %Req.Response{status: status}} when status in 200..299 -> + :ok + + {:ok, %Req.Response{status: status, body: body}} -> + Logger.error("Discord delivery failed with status #{status}: #{inspect(body)}") + {:error, :http_status} + + {:error, reason} -> + Logger.error("Discord delivery failed: #{inspect(reason)}") + {:error, reason} + end + end + + def deliver(_message, _config) do + {:error, :invalid_config} + end +end diff --git a/lib/chronodash/application.ex b/lib/chronodash/application.ex index 6ed0c84..1f205fa 100644 --- a/lib/chronodash/application.ex +++ b/lib/chronodash/application.ex @@ -21,7 +21,9 @@ defmodule Chronodash.Application do {Finch, name: Chronodash.Finch}, # Polling Services Chronodash.Polling.Supervisor, - Chronodash.Polling.Scheduler + Chronodash.Polling.Scheduler, + # Alerting Services + Chronodash.Alerting.Manager ] Chronodash.Release.create_and_migrate() @@ -29,7 +31,14 @@ defmodule Chronodash.Application do # See https://hexdocs.pm/elixir/Supervisor.html # for other strategies and supported options opts = [strategy: :one_for_one, name: Chronodash.Supervisor] - Supervisor.start_link(children, opts) + + case Supervisor.start_link(children, opts) do + {:ok, pid} -> + {:ok, pid} + + error -> + error + end end # Tell Phoenix to update the endpoint configuration diff --git a/lib/chronodash/polling/worker.ex b/lib/chronodash/polling/worker.ex index a47cbbc..3803a5a 100644 --- a/lib/chronodash/polling/worker.ex +++ b/lib/chronodash/polling/worker.ex @@ -47,10 +47,13 @@ defmodule Chronodash.Polling.Worker do end defp handle_success(forecast, state) do - # 1. Find or create the location in our DB + # 1. Emit full forecast event for alerting (window rules) + emit_forecast_telemetry(forecast, state) + + # 2. Find or create the location in our DB location = get_or_create_location(forecast) - # 2. Save all forecast points to the DB + # 3. Save all forecast points to the DB # TODO: If you eventually poll thousands of locations, you could batch the Ash.bulk_create calls # (e.g., save in groups of 500 using Enum.chunk_every/2) to avoid long-running transactions. process_forecast_points(forecast, location, state) @@ -133,6 +136,17 @@ defmodule Chronodash.Polling.Worker do end end + defp emit_forecast_telemetry(forecast, state) do + metadata = + Map.merge(state.metadata, %{ + location: forecast.location_name, + variable: forecast.metric_type, + points: forecast.points + }) + + :telemetry.execute([:chronodash, :polling, :forecast_received], %{count: 1}, metadata) + end + defp emit_observation(metric, timestamp, location_name, state) do # We emit a metric for each data point measurements = %{value: normalize_value(metric.value)} From e3859a687ae6e266fd90a5eada8e6d31eb107897 Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 17:06:22 +0100 Subject: [PATCH 2/3] chore: remove unnecesary harcoded string --- etc/grafana/provisioning/alerting/contact_points.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/etc/grafana/provisioning/alerting/contact_points.yaml b/etc/grafana/provisioning/alerting/contact_points.yaml index 48e150e..11a639a 100644 --- a/etc/grafana/provisioning/alerting/contact_points.yaml +++ b/etc/grafana/provisioning/alerting/contact_points.yaml @@ -7,4 +7,4 @@ contactPoints: - uid: discord_alert type: discord settings: - url: https://discord.com/api/webhooks/1477276961423491164/gcEFNjDjVcVIHQmq3aRbeP2sqcACmlk7m81SYDWRwxwrGGYJXBGqxOtWRYQhRxA8MRyx + url: ${DISCORD_WEBHOOK_URL} From 21fff5818d320a298007c0ad7a7df36dbc4268f8 Mon Sep 17 00:00:00 2001 From: yagogarea Date: Sat, 28 Feb 2026 17:08:13 +0100 Subject: [PATCH 3/3] style: apply format --- lib/chronodash/alerting/manager.ex | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/lib/chronodash/alerting/manager.ex b/lib/chronodash/alerting/manager.ex index 27fc56e..d46c3a3 100644 --- a/lib/chronodash/alerting/manager.ex +++ b/lib/chronodash/alerting/manager.ex @@ -45,7 +45,10 @@ defmodule Chronodash.Alerting.Manager do end @impl true - def handle_cast({:evaluate, [:chronodash, :polling, :observation], %{value: value}, metadata}, state) do + def handle_cast( + {:evaluate, [:chronodash, :polling, :observation], %{value: value}, metadata}, + state + ) do get_instant_rules() |> Enum.each(&process_instant_rule(&1, value, metadata)) @@ -53,7 +56,10 @@ defmodule Chronodash.Alerting.Manager do end @impl true - def handle_cast({:evaluate, [:chronodash, :polling, :forecast_received], _measurements, metadata}, state) do + def handle_cast( + {:evaluate, [:chronodash, :polling, :forecast_received], _measurements, metadata}, + state + ) do get_window_rules() |> Enum.each(&process_window_rule(&1, metadata)) @@ -99,26 +105,31 @@ defmodule Chronodash.Alerting.Manager do defp metric_name_match?(m_name, rule_metric) do to_string(m_name) == to_string(rule_metric) or - (to_string(rule_metric) == "wind" and m_name == "wind_module") + (to_string(rule_metric) == "wind" and m_name == "wind_module") end defp check_forecast_condition([], _cond, _thresh), do: {false, nil} + defp check_forecast_condition(values, :max_gt, thresh) do max = Enum.max(values) {max > thresh, max} end + defp check_forecast_condition(values, :max_lt, thresh) do max = Enum.max(values) {max < thresh, max} end + defp check_forecast_condition(values, :min_lt, thresh) do min = Enum.min(values) {min < thresh, min} end + defp check_forecast_condition(values, :gt_any, thresh) do val = Enum.find(values, &(&1 > thresh)) {!is_nil(val), val} end + defp check_forecast_condition(_, _, _), do: {false, nil} # ============================================================================ @@ -130,7 +141,7 @@ defmodule Chronodash.Alerting.Manager do now = System.system_time(:millisecond) case :ets.lookup(@table_name, {rule.id, location}) do - [{{_id, _loc}, last_triggered_at}] -> (now - last_triggered_at) < cooldown_ms + [{{_id, _loc}, last_triggered_at}] -> now - last_triggered_at < cooldown_ms [] -> false end end @@ -148,12 +159,14 @@ defmodule Chronodash.Alerting.Manager do end defp normalize_raw_value(v) when is_number(v), do: v * 1.0 + defp normalize_raw_value(v) when is_binary(v) do case Float.parse(v) do {f, _} -> f _ -> 0.0 end end + defp normalize_raw_value(_), do: 0.0 defp evaluate_condition(:gt, val, thresh), do: val > thresh