diff --git a/backlog/tasks/ml-11 - Telemetry.Storage-GenServer-funnels-synchronous-SQLite-writes.md b/backlog/tasks/ml-11 - Telemetry.Storage-GenServer-funnels-synchronous-SQLite-writes.md index 379c08d1..fb0c2fc3 100644 --- a/backlog/tasks/ml-11 - Telemetry.Storage-GenServer-funnels-synchronous-SQLite-writes.md +++ b/backlog/tasks/ml-11 - Telemetry.Storage-GenServer-funnels-synchronous-SQLite-writes.md @@ -1,9 +1,10 @@ --- id: ML-11 title: Telemetry.Storage GenServer funnels synchronous SQLite writes -status: To Do +status: Done assignee: [] created_date: '2026-04-20 08:49' +updated_date: '2026-04-24 09:29' labels: [] dependencies: [] references: @@ -47,7 +48,7 @@ Whichever option is chosen, replace the bare `catch _, _` with either a targeted - Existing dashboard queries continue to work -- [ ] #1 Telemetry write path does not block on the GenServer mailbox -- [ ] #2 Error path surfaces at least once per distinct failure (not silently swallowed) -- [ ] #3 Existing dashboard queries continue to work +- [x] #1 Telemetry write path does not block on the GenServer mailbox +- [x] #2 Error path surfaces at least once per distinct failure (not silently swallowed) +- [x] #3 Existing dashboard queries continue to work diff --git a/config/config.exs b/config/config.exs index 9c785113..99d3b284 100644 --- a/config/config.exs +++ b/config/config.exs @@ -122,7 +122,9 @@ config :music_library, MusicLibrary.BackgroundRepo, priv: "priv/background_repo" config :music_library, MusicLibrary.TelemetryRepo, priv: "priv/telemetry_repo", log: false -config :music_library, MusicLibraryWeb.Telemetry.Storage, buffer_size: 32_768 +config :music_library, MusicLibraryWeb.Telemetry.Storage, + retention_limit: 32_768, + flush_interval_ms: 5_000 config :swoosh, :api_client, Swoosh.ApiClient.Req diff --git a/docs/architecture.md b/docs/architecture.md index ecac1a3a..f18789c8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -33,7 +33,7 @@ MusicLibrary.Application (one_for_one) ├── MusicLibrary.BackgroundRepo # Oban SQLite repo (separate DB) ├── MusicLibrary.TelemetryRepo # Telemetry metrics SQLite repo ├── MusicLibraryWeb.Telemetry # Telemetry supervisor -│ ├── Telemetry.Storage # Metrics storage (SQLite-backed, persistent) +│ ├── Telemetry.Storage # Buffered metrics storage (in-memory, 5s flush to SQLite, force-flush on read) │ └── :telemetry_poller # 30s periodic measurements ├── Oban # Background job engine ├── Ecto.Migrator # Migrations (skipped in release; run by Coolify post-deploy) @@ -420,3 +420,7 @@ All events are namespaced with `music_library:` prefix. - **`use MusicLibraryWeb, :live_view`** imports: Phoenix.LiveView, Fluxon, Gettext, LiveToast, CoreComponents, Phoenix.HTML, verified routes, JS alias - **`use MusicLibraryWeb, :live_component`** additionally imports `put_toast!/2` +- **Telemetry buffering**: `Telemetry.Storage` keeps incoming datapoints in an in-memory map + keyed by metric and flushes to SQLite on a 5 s timer, on shutdown, and synchronously when + `metrics_history/1` is called (only for the requested metric). Keeps the cast path O(1) and + the dashboard read path consistent. diff --git a/docs/production-infrastructure.md b/docs/production-infrastructure.md index 7cc2fd6b..db088aa8 100644 --- a/docs/production-infrastructure.md +++ b/docs/production-infrastructure.md @@ -205,7 +205,12 @@ checks and post-deploy verification. ### Telemetry SQLite-backed persistent metrics (`MusicLibraryWeb.Telemetry.Storage`) with 30-second -polling interval. Tracks: +polling interval. Events are buffered in GenServer state keyed by metric and flushed to +SQLite every 5 seconds inside a single transaction; reads via `metrics_history/1` +force-flush only the requested metric so the dashboard sees fresh data without waiting +for the next tick. Per-metric retention is capped at 32 768 rows +(`:retention_limit`), pruned after each flush. Flush failures are logged at `:warning` +and the offending batch is dropped. Tracks: - Database query times (total, query, queue) - External API request latency (Finch) diff --git a/lib/music_library_web/telemetry/storage.ex b/lib/music_library_web/telemetry/storage.ex index a1f01cde..1888c2c9 100644 --- a/lib/music_library_web/telemetry/storage.ex +++ b/lib/music_library_web/telemetry/storage.ex @@ -1,36 +1,74 @@ defmodule MusicLibraryWeb.Telemetry.Storage do - @moduledoc false + @moduledoc """ + Telemetry metrics storage backed by SQLite with an in-memory write buffer. + + Telemetry events arrive via `:telemetry` handlers and are forwarded through + a fast cast to this GenServer. Each cast prepends a datapoint to an + in-state buffer keyed by metric; no disk I/O happens on the cast path. + + Buffered datapoints are flushed to SQLite on three triggers: + + * a periodic timer (`:flush_interval_ms`, default 5s) which drains the + full buffer inside a single transaction; + * a call to `metrics_history/1`, which force-flushes only the buffer + entry for the requested metric so readers see fresh data without + waiting for the next tick; + * `terminate/2`, so graceful shutdown does not lose buffered datapoints. + + Per metric key, at most `:retention_limit` rows (default 32 768) are kept; + older rows are pruned after each flush. + + Flush failures are logged at `:warning` level; the offending batch is + dropped (telemetry is not authoritative) and the process keeps running. + """ use GenServer - @buffer_size Application.compile_env( - :music_library, - [__MODULE__, :buffer_size], - 50 - ) + require Logger + + @retention_limit Application.compile_env!(:music_library, [__MODULE__, :retention_limit]) + @flush_interval_ms Application.compile_env!(:music_library, [__MODULE__, :flush_interval_ms]) + @insert_chunk_size 200 def metrics_history(metric) do GenServer.call(__MODULE__, {:data, metric}) end def start_link(args) do - GenServer.start_link(__MODULE__, args, name: __MODULE__) + {metrics, opts} = + case args do + {metrics, opts} when is_list(metrics) and is_list(opts) -> {metrics, opts} + metrics when is_list(metrics) -> {metrics, [name: __MODULE__]} + end + + {server_opts, init_opts} = Keyword.split(opts, [:name]) + GenServer.start_link(__MODULE__, {metrics, init_opts}, server_opts) end @impl true - def init(metrics) do + def init({metrics, opts}) do Process.flag(:trap_exit, true) for metric <- metrics do attach_handler(metric) end - {:ok, metrics} + state = %{ + metrics: metrics, + buffer: %{}, + retention_limit: Keyword.get(opts, :retention_limit, @retention_limit), + flush_interval_ms: Keyword.get(opts, :flush_interval_ms, @flush_interval_ms), + flush_ref: nil + } + + {:ok, %{state | flush_ref: schedule_flush(state.flush_interval_ms)}} end @impl true - def terminate(_, metrics) do - for metric <- metrics do + def terminate(_reason, state) do + _ = flush_all(state) + + for metric <- state.metrics do :telemetry.detach({__MODULE__, metric, self()}) end @@ -55,24 +93,38 @@ defmodule MusicLibraryWeb.Telemetry.Storage do @impl true def handle_cast({:telemetry_metric, data, metric}, state) do key = metric_key(metric) - label = if is_map(data), do: Map.get(data, :label) - measurement = if is_map(data), do: Map.get(data, :measurement, 0), else: 0 - - time = - if is_map(data), - do: Map.get(data, :time, System.system_time(:microsecond)), - else: System.system_time(:microsecond) - - insert_and_prune(key, label, measurement, time) - - {:noreply, state} + entry = datapoint_from_data(data) + buffer = Map.update(state.buffer, key, [entry], &[entry | &1]) + {:noreply, %{state | buffer: buffer}} end @impl true def handle_call({:data, metric}, _from, state) do key = metric_key(metric) - datapoints = fetch_datapoints(key) - {:reply, datapoints, state} + state = flush_key(state, key) + {:reply, fetch_datapoints(key), state} + end + + @impl true + def handle_info(:flush, state) do + state = flush_all(state) + {:noreply, %{state | flush_ref: schedule_flush(state.flush_interval_ms)}} + end + + defp schedule_flush(interval_ms) do + Process.send_after(self(), :flush, interval_ms) + end + + defp datapoint_from_data(data) when is_map(data) do + %{ + label: Map.get(data, :label), + measurement: Map.get(data, :measurement, 0), + time: Map.get(data, :time, System.system_time(:microsecond)) + } + end + + defp datapoint_from_data(_data) do + %{label: nil, measurement: 0, time: System.system_time(:microsecond)} end defp metric_key(metric) do @@ -86,27 +138,84 @@ defmodule MusicLibraryWeb.Telemetry.Storage do ) end - defp insert_and_prune(key, label, measurement, time) do - MusicLibrary.TelemetryRepo.query( - "INSERT INTO telemetry_datapoints (metric_key, label, measurement, time) VALUES (?1, ?2, ?3, ?4)", - [key, label, measurement, time] - ) + defp flush_all(%{buffer: buffer} = state) when map_size(buffer) == 0, do: state - MusicLibrary.TelemetryRepo.query( - """ - DELETE FROM telemetry_datapoints - WHERE metric_key = ?1 - AND id NOT IN ( - SELECT id FROM telemetry_datapoints - WHERE metric_key = ?1 - ORDER BY id DESC - LIMIT ?2 + defp flush_all(state) do + keys = Map.keys(state.buffer) + + try do + MusicLibrary.TelemetryRepo.transaction(fn -> + Enum.each(keys, &persist_key(state, &1)) + end) + rescue + error -> + Logger.warning( + "[Telemetry.Storage] flush failed: " <> Exception.format(:error, error, __STACKTRACE__) ) - """, - [key, @buffer_size] - ) - catch - _, _ -> :ok + catch + kind, reason -> + Logger.warning("[Telemetry.Storage] flush #{kind}: #{inspect(reason)}") + end + + %{state | buffer: %{}} + end + + defp flush_key(state, key) do + case Map.get(state.buffer, key) do + nil -> + state + + [] -> + %{state | buffer: Map.delete(state.buffer, key)} + + _entries -> + try do + MusicLibrary.TelemetryRepo.transaction(fn -> + persist_key(state, key) + end) + rescue + error -> + Logger.warning( + "[Telemetry.Storage] flush failed for #{key}: " <> + Exception.format(:error, error, __STACKTRACE__) + ) + catch + kind, reason -> + Logger.warning("[Telemetry.Storage] flush #{kind} for #{key}: #{inspect(reason)}") + end + + %{state | buffer: Map.delete(state.buffer, key)} + end + end + + defp persist_key(state, key) do + case Map.get(state.buffer, key, []) do + [] -> + :ok + + entries -> + entries + |> Enum.reverse() + |> Enum.map(&Map.put(&1, :metric_key, key)) + |> Enum.chunk_every(@insert_chunk_size) + |> Enum.each(&MusicLibrary.TelemetryRepo.insert_all("telemetry_datapoints", &1)) + + MusicLibrary.TelemetryRepo.query!( + """ + DELETE FROM telemetry_datapoints + WHERE metric_key = ?1 + AND id NOT IN ( + SELECT id FROM telemetry_datapoints + WHERE metric_key = ?1 + ORDER BY id DESC + LIMIT ?2 + ) + """, + [key, state.retention_limit] + ) + + :ok + end end defp fetch_datapoints(key) do @@ -119,10 +228,17 @@ defmodule MusicLibraryWeb.Telemetry.Storage do %{label: label, measurement: measurement, time: time} end) - _ -> + {:error, reason} -> + Logger.warning("[Telemetry.Storage] fetch failed for #{key}: #{inspect(reason)}") [] end - catch - _, _ -> [] + rescue + error -> + Logger.warning( + "[Telemetry.Storage] fetch raised for #{key}: " <> + Exception.format(:error, error, __STACKTRACE__) + ) + + [] end end diff --git a/test/music_library_web/telemetry/storage_test.exs b/test/music_library_web/telemetry/storage_test.exs new file mode 100644 index 00000000..2ea5f492 --- /dev/null +++ b/test/music_library_web/telemetry/storage_test.exs @@ -0,0 +1,194 @@ +defmodule MusicLibraryWeb.Telemetry.StorageTest do + use ExUnit.Case, async: false + + import ExUnit.CaptureLog + import Telemetry.Metrics, only: [summary: 2] + + alias MusicLibrary.TelemetryRepo + alias MusicLibraryWeb.Telemetry.Storage + + setup do + id = System.unique_integer([:positive]) + metric = summary("music_library.test.metric_#{id}.duration", tags: []) + pid = start_supervised!({Storage, {[], [flush_interval_ms: 60_000]}}) + + on_exit(fn -> + TelemetryRepo.query!( + "DELETE FROM telemetry_datapoints WHERE metric_key LIKE ?1", + ["%metric_#{id}%"] + ) + end) + + %{pid: pid, metric: metric, id: id} + end + + describe "handle_cast/2" do + test "buffers datapoints in-memory without persisting", %{pid: pid, metric: metric} do + cast_point(pid, metric, 1.0, 100) + cast_point(pid, metric, 2.0, 200) + + %{buffer: buffer} = :sys.get_state(pid) + key = metric_key(metric) + + assert [%{measurement: 2.0, time: 200}, %{measurement: 1.0, time: 100}] = + Map.get(buffer, key) + + assert db_count(key) == 0 + end + + test "many casts return quickly (no sync I/O on cast path)", %{pid: pid, metric: metric} do + {time_us, :ok} = + :timer.tc(fn -> + for _ <- 1..500, do: cast_point(pid, metric, 1.0, 0) + _ = :sys.get_state(pid) + :ok + end) + + assert time_us < 200_000, "500 casts took #{time_us}us" + end + end + + describe "handle_call({:data, metric}, ...)" do + test "force-flushes the requested metric and returns persisted rows", %{ + pid: pid, + metric: metric + } do + cast_point(pid, metric, 42.0, 1000) + + result = GenServer.call(pid, {:data, metric}) + + assert [%{measurement: m, time: 1000, label: nil}] = result + assert_in_delta m, 42.0, 0.001 + assert db_count(metric_key(metric)) == 1 + end + + test "only flushes the requested metric, leaves others buffered", %{ + pid: pid, + metric: metric_a, + id: id + } do + metric_b = summary("music_library.test.metric_#{id}_b.duration", tags: []) + + cast_point(pid, metric_a, 1.0, 1) + cast_point(pid, metric_b, 2.0, 2) + + _ = GenServer.call(pid, {:data, metric_a}) + + assert db_count(metric_key(metric_a)) == 1 + assert db_count(metric_key(metric_b)) == 0 + + %{buffer: buffer} = :sys.get_state(pid) + refute Map.has_key?(buffer, metric_key(metric_a)) + assert [%{measurement: 2.0}] = Map.get(buffer, metric_key(metric_b)) + end + + test "returns empty list when nothing has been cast", %{pid: pid, metric: metric} do + assert GenServer.call(pid, {:data, metric}) == [] + end + end + + describe "periodic :flush" do + test "drains the full buffer to SQLite", %{pid: pid, metric: metric, id: id} do + metric_b = summary("music_library.test.metric_#{id}_b.duration", tags: []) + cast_point(pid, metric, 1.5, 10) + cast_point(pid, metric_b, 2.5, 20) + + send(pid, :flush) + _ = :sys.get_state(pid) + + assert db_count(metric_key(metric)) == 1 + assert db_count(metric_key(metric_b)) == 1 + + %{buffer: buffer} = :sys.get_state(pid) + assert buffer == %{} + end + + test "prunes rows above retention_limit per metric", %{id: id} do + metric = summary("music_library.test.metric_#{id}_r.duration", tags: []) + + pid = + start_supervised!({Storage, {[], [retention_limit: 3, flush_interval_ms: 60_000]}}, + id: :storage_retention + ) + + for n <- 1..10, do: cast_point(pid, metric, n / 1.0, n) + send(pid, :flush) + _ = :sys.get_state(pid) + + key = metric_key(metric) + assert db_count(key) == 3 + + {:ok, %{rows: rows}} = + TelemetryRepo.query( + "SELECT time FROM telemetry_datapoints WHERE metric_key = ?1 ORDER BY time ASC", + [key] + ) + + assert rows == [[8], [9], [10]] + end + end + + describe "terminate/2" do + test "flushes the buffer on shutdown", %{pid: pid, metric: metric} do + cast_point(pid, metric, 7.0, 5000) + + :ok = stop_supervised(MusicLibraryWeb.Telemetry.Storage) + + {:ok, %{rows: rows}} = + TelemetryRepo.query( + "SELECT measurement, time FROM telemetry_datapoints WHERE metric_key = ?1", + [metric_key(metric)] + ) + + assert [[_, 5000]] = rows + refute Process.alive?(pid) + end + end + + describe "error handling" do + test "logs a warning and keeps running when the flush raises", %{pid: pid, metric: metric} do + bad = %{measurement: {:not_a_number}, time: 1} + GenServer.cast(pid, {:telemetry_metric, bad, metric}) + + log = + capture_log(fn -> + send(pid, :flush) + _ = :sys.get_state(pid) + end) + + assert log =~ "[Telemetry.Storage] flush" + assert Process.alive?(pid) + + %{buffer: buffer} = :sys.get_state(pid) + assert buffer == %{} + end + end + + defp cast_point(pid, metric, measurement, time) do + GenServer.cast( + pid, + {:telemetry_metric, %{measurement: measurement, time: time}, metric} + ) + end + + defp db_count(key) do + {:ok, %{rows: [[count]]}} = + TelemetryRepo.query( + "SELECT COUNT(*) FROM telemetry_datapoints WHERE metric_key = ?1", + [key] + ) + + count + end + + defp metric_key(metric) do + Enum.join( + [ + inspect(metric.__struct__), + Enum.join(metric.name, "."), + Enum.join(metric.tags, ".") + ], + ":" + ) + end +end