GITGUD

neiam neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Count sanctum's sources and the rows they leave behind

3f6ff12 · Gabriel Morell · 2026-08-24 03:57

7 files +340 -1
Message 
{commit_body(@commit)}

Files changed

added apps/room_observatory/lib/room_observatory/plugins/sanctum.ex
+335 −0
@@ -0,0 +1,335 @@
1 +defmodule RoomObservatory.Plugins.Sanctum do
2 + @moduledoc """
3 + What room_sanctum is actually holding: how many sources of each kind are
4 + configured, and how many rows their ingestion has left behind.
5 +
6 + Nothing here is derived from a request or a job. These are inventory
7 + questions — "how many GTFS feeds are we running", "is gtfs_stop_times still
8 + growing" — that only the database can answer, so all three metrics are
9 + polled rather than driven by `:telemetry` events the app already emits.
10 +
11 + ## Three metrics, two poll rates, and why they are not one metric
12 +
13 + `phx_sanctum_sources_count{type, enabled}` is a `GROUP BY` over
14 + `cfg_sources`, which has tens of rows. Cheap at any rate.
15 +
16 + `phx_sanctum_storage_rows_estimate{table}` is **not** `count(*)`. It reads
17 + `pg_class.reltuples`, the planner's estimate, for every storage table in one
18 + query that touches no user table at all. An exact count of gtfs_stop_times is
19 + a full sequential scan over millions of rows; doing that every 30 seconds
20 + would make the metric the most expensive thing the database does. The
21 + estimate is refreshed by autovacuum's ANALYZE, so it lags a bulk GTFS import
22 + by minutes and is off by a percent or two in steady state — which is the
23 + right trade for a number whose purpose is watching a trend line.
24 +
25 + `phx_sanctum_storage_source_rows{table, source_id, source, type}` *is* the
26 + exact `count(*)`, grouped by source, and it is the one that answers "which
27 + feed is eating the disk". It is also the one that scans. It polls every 15
28 + minutes by default and lives in its own polling group, so its slow query
29 + runs in its own poller process and cannot delay the cheap metrics above.
30 +
31 + ## Tables with no source, and rows with no source
32 +
33 + `icarus_flight_watches` is keyed on a flight instance rather than on a
34 + source — several queries watching the same arrival share one watch — so it
35 + appears in the estimate metric and is absent from the per-source one.
36 +
37 + `storage_mail` hangs off `cfg_mailboxes`, not off a source directly, so it
38 + reaches its source through that join.
39 +
40 + A row whose `source_id` is NULL is reported under `source_id="none"`. That is
41 + deliberate rather than filtered out: orphaned rows left behind by a deleted
42 + source are exactly the kind of thing this metric exists to surface.
43 +
44 + ## Where the numbers go stale
45 +
46 + A gauge keyed by label keeps its last value until the process restarts, and
47 + neither Prometheus nor PromEx has a way to say "this series no longer
48 + exists". Two consequences worth knowing before trusting a panel:
49 +
50 + * Deleting a source leaves its `storage_source_rows` series frozen at its
51 + final value. Zeroes are filled in for every *existing* source of a
52 + table's kind on every poll, so an emptied feed does drop to zero — but a
53 + deleted one does not, until the pod restarts.
54 +
55 + * If a kind has no sources at all, its tables are not counted. That is the
56 + point: no GTFS feeds means no reason to scan gtfs_stop_times. The
57 + estimate metric still reports those tables, so orphaned rows stay
58 + visible there.
59 + """
60 +
61 + use PromEx.Plugin
62 +
63 + require Logger
64 +
65 + alias RoomSanctum.Repo
66 +
67 + @inventory_event [:room_observatory, :sanctum, :inventory]
68 + @rows_event [:room_observatory, :sanctum, :storage, :rows]
69 + @source_rows_event [:room_observatory, :sanctum, :storage, :source_rows]
70 +
71 + @inventory_poll_rate :timer.seconds(30)
72 + @source_rows_poll_rate :timer.minutes(15)
73 +
74 + # Long enough that a full scan of gtfs_stop_times on a cold cache does not
75 + # trip the 15s default and leave a permanent hole in the metric.
76 + @count_timeout :timer.minutes(2)
77 +
78 + # Every table under RoomSanctum.Storage, with the source type that fills it
79 + # and how a row reaches its source. `:column` is a plain `source_id`;
80 + # `:via_mailbox` joins through cfg_mailboxes; `:none` means the table has no
81 + # source at all and is estimate-only.
82 + @tables [
83 + %{table: "airnow_hourly_data", type: :aqi, via: :column},
84 + %{table: "airnow_hourly_observations", type: :aqi, via: :column},
85 + %{table: "airnow_monitoring_sites", type: :aqi, via: :column},
86 + %{table: "airnow_reporting_area", type: :aqi, via: :column},
87 + %{table: "gbfs_alerts", type: :gbfs, via: :column},
88 + %{table: "gbfs_ebikes_stations", type: :gbfs, via: :column},
89 + %{table: "gbfs_free_bike_status", type: :gbfs, via: :column},
90 + %{table: "gbfs_geofencing_zones", type: :gbfs, via: :column},
91 + %{table: "gbfs_station_information", type: :gbfs, via: :column},
92 + %{table: "gbfs_station_status", type: :gbfs, via: :column},
93 + %{table: "gbfs_system_information", type: :gbfs, via: :column},
94 + %{table: "gbfs_system_pricing_plans", type: :gbfs, via: :column},
95 + %{table: "gbfs_vehicle_types", type: :gbfs, via: :column},
96 + %{table: "gtfs_agencies", type: :gtfs, via: :column},
97 + %{table: "gtfs_calendars", type: :gtfs, via: :column},
98 + %{table: "gtfs_directions", type: :gtfs, via: :column},
99 + %{table: "gtfs_routes", type: :gtfs, via: :column},
100 + %{table: "gtfs_shapes", type: :gtfs, via: :column},
101 + %{table: "gtfs_stops", type: :gtfs, via: :column},
102 + %{table: "gtfs_stop_times", type: :gtfs, via: :column},
103 + %{table: "gtfs_trips", type: :gtfs, via: :column},
104 + %{table: "icalendar_entries", type: :calendar, via: :column},
105 + %{table: "storage_mail", type: :mailbox, via: :via_mailbox},
106 + %{table: "icarus_flight_watches", type: nil, via: :none}
107 + ]
108 +
109 + @impl true
110 + def polling_metrics(opts) do
111 + prefix = Keyword.get(opts, :metric_prefix, [:room_observatory, :sanctum])
112 +
113 + [
114 + inventory_metrics(prefix, Keyword.get(opts, :poll_rate, @inventory_poll_rate)),
115 + source_rows_metrics(prefix, Keyword.get(opts, :source_rows_poll_rate, @source_rows_poll_rate))
116 + ]
117 + end
118 +
119 + defp inventory_metrics(prefix, poll_rate) do
120 + Polling.build(
121 + :sanctum_inventory_polling_events,
122 + poll_rate,
123 + {__MODULE__, :execute_inventory_metrics, []},
124 + [
125 + last_value(
126 + prefix ++ [:sources, :count],
127 + event_name: @inventory_event,
128 + measurement: :sources,
129 + description: "Configured sources in cfg_sources, by kind and whether they are enabled.",
130 + tags: [:type, :enabled]
131 + ),
132 + last_value(
133 + prefix ++ [:storage, :rows, :estimate],
134 + event_name: @rows_event,
135 + measurement: :rows,
136 + description:
137 + "Planner row estimate (pg_class.reltuples) per storage table. Refreshed by ANALYZE, so it lags a bulk import.",
138 + tags: [:table]
139 + )
140 + ],
141 + detach_on_error: false
142 + )
143 + end
144 +
145 + defp source_rows_metrics(prefix, poll_rate) do
146 + Polling.build(
147 + :sanctum_storage_source_rows_polling_events,
148 + poll_rate,
149 + {__MODULE__, :execute_source_rows_metrics, []},
150 + [
151 + last_value(
152 + prefix ++ [:storage, :source_rows],
153 + event_name: @source_rows_event,
154 + measurement: :rows,
155 + description: "Exact row count per storage table, attributed to the source that filled it.",
156 + tags: [:table, :source_id, :source, :type]
157 + )
158 + ],
159 + detach_on_error: false
160 + )
161 + end
162 +
163 + @doc false
164 + def execute_inventory_metrics do
165 + emit_source_counts()
166 + emit_row_estimates()
167 + rescue
168 + error -> log_skipped(:inventory, error)
169 + catch
170 + :exit, reason -> log_skipped(:inventory, reason)
171 + end
172 +
173 + @doc false
174 + def execute_source_rows_metrics do
175 + case sources_by_id() do
176 + empty when map_size(empty) == 0 -> :ok
177 + sources -> Enum.each(@tables, &safely_emit_source_rows(&1, sources))
178 + end
179 + rescue
180 + error -> log_skipped(:storage_source_rows, error)
181 + catch
182 + :exit, reason -> log_skipped(:storage_source_rows, reason)
183 + end
184 +
185 + # Per table rather than per group: 24 counts are 24 separate queries, and one
186 + # that fails — a table dropped by a migration, a statement timeout on the
187 + # largest — should cost its own series, not the 23 behind it in the loop.
188 + defp safely_emit_source_rows(table, sources) do
189 + emit_source_rows(table, sources)
190 + rescue
191 + error -> log_skipped(table.table, error)
192 + catch
193 + :exit, reason -> log_skipped(table.table, reason)
194 + end
195 +
196 + # -- sources -------------------------------------------------------------
197 +
198 + defp emit_source_counts do
199 + %{rows: rows} = Repo.query!("SELECT type, enabled, count(*) FROM cfg_sources GROUP BY 1, 2")
200 +
201 + counts = Map.new(rows, fn [type, enabled, count] -> {{type_label(type), enabled}, count} end)
202 +
203 + # Zero-filled across the full enum rather than only the types present, so a
204 + # kind that drops to zero reads as zero instead of vanishing from the graph.
205 + types = Enum.uniq(known_types() ++ Enum.map(rows, fn [type, _, _] -> type_label(type) end))
206 +
207 + for type <- types, enabled <- [true, false] do
208 + :telemetry.execute(
209 + @inventory_event,
210 + %{sources: Map.get(counts, {type, enabled}, 0)},
211 + %{type: type, enabled: enabled}
212 + )
213 + end
214 + end
215 +
216 + defp known_types do
217 + RoomSanctum.Configuration.Source
218 + |> Ecto.Enum.values(:type)
219 + |> Enum.map(&Atom.to_string/1)
220 + end
221 +
222 + defp type_label(nil), do: "unknown"
223 + defp type_label(type), do: to_string(type)
224 +
225 + # -- estimates -----------------------------------------------------------
226 +
227 + @estimate_query """
228 + SELECT c.relname::text, GREATEST(c.reltuples, 0)::bigint
229 + FROM pg_class c
230 + JOIN pg_namespace n ON n.oid = c.relnamespace
231 + WHERE n.nspname = current_schema()
232 + AND c.relkind = 'r'
233 + AND c.relname::text = ANY($1)
234 + """
235 +
236 + defp emit_row_estimates do
237 + names = Enum.map(@tables, & &1.table)
238 + %{rows: rows} = Repo.query!(@estimate_query, [names])
239 + estimates = Map.new(rows, fn [table, estimate] -> {table, estimate} end)
240 +
241 + # A table missing from pg_class has not been migrated yet; report 0 rather
242 + # than leaving a gap, so the series exists from the first scrape.
243 + for table <- names do
244 + :telemetry.execute(@rows_event, %{rows: Map.get(estimates, table, 0)}, %{table: table})
245 + end
246 + end
247 +
248 + # -- exact per-source counts ---------------------------------------------
249 +
250 + defp sources_by_id do
251 + %{rows: rows} = Repo.query!("SELECT id, name, type FROM cfg_sources")
252 +
253 + Map.new(rows, fn [id, name, type] ->
254 + {id, %{name: name || "(unnamed)", type: type_label(type)}}
255 + end)
256 + end
257 +
258 + defp emit_source_rows(%{via: :none}, _sources), do: :ok
259 +
260 + defp emit_source_rows(%{type: type} = table, sources) do
261 + expected = for {id, source} <- sources, source.type == to_string(type), do: {id, source}
262 +
263 + # No feeds of this kind means nothing to attribute, and scanning millions of
264 + # leftover rows to say so is the cost this guard exists to avoid.
265 + if expected == [] do
266 + :ok
267 + else
268 + counts = count_by_source(table)
269 +
270 + for [source_id, count] <- counts do
271 + emit_source_row(table.table, source_id, count, sources, type)
272 + end
273 +
274 + counted = MapSet.new(counts, fn [source_id, _] -> source_id end)
275 +
276 + for {id, source} <- expected, not MapSet.member?(counted, id) do
277 + emit(table.table, to_string(id), source.name, source.type, 0)
278 + end
279 + end
280 + end
281 +
282 + defp count_by_source(%{table: table, via: :column}) do
283 + %{rows: rows} =
284 + Repo.query!(~s|SELECT source_id, count(*) FROM "#{table}" GROUP BY 1|, [],
285 + timeout: @count_timeout
286 + )
287 +
288 + rows
289 + end
290 +
291 + # The foreign key is `mail`, not `mail_id` — the migration named it that way
292 + # while `belongs_to :mail` in the schema implies the suffixed name. Raw SQL
293 + # here has to follow the column that actually exists.
294 + defp count_by_source(%{via: :via_mailbox}) do
295 + %{rows: rows} =
296 + Repo.query!(
297 + """
298 + SELECT m.source_id, count(*)
299 + FROM storage_mail s
300 + JOIN cfg_mailboxes m ON m.id = s.mail
301 + GROUP BY 1
302 + """,
303 + [],
304 + timeout: @count_timeout
305 + )
306 +
307 + rows
308 + end
309 +
310 + defp emit_source_row(table, nil, count, _sources, type) do
311 + emit(table, "none", "(unassigned)", type_label(type), count)
312 + end
313 +
314 + defp emit_source_row(table, source_id, count, sources, type) do
315 + source = Map.get(sources, source_id, %{name: "(deleted)", type: type_label(type)})
316 + emit(table, to_string(source_id), source.name, source.type, count)
317 + end
318 +
319 + defp emit(table, source_id, source, type, count) do
320 + :telemetry.execute(@source_rows_event, %{rows: count}, %{
321 + table: table,
322 + source_id: source_id,
323 + source: source,
324 + type: type
325 + })
326 + end
327 +
328 + defp log_skipped(group, reason) do
329 + Logger.warning(
330 + "RoomObservatory.Plugins.Sanctum skipped #{group} metrics: #{inspect(reason)}"
331 + )
332 +
333 + :ok
334 + end
335 +end
modified apps/room_observatory/lib/room_observatory/prom_ex.ex
+5 −1
@@ -72,7 +72,11 @@ defmodule RoomObservatory.PromEx do
72 72 # label the Ecto plugin already attaches.
73 73 {Plugins.Ecto, repos: [RoomSanctum.Repo, RoomHermes.Repo], metric_prefix: [:phx, :ecto]},
74 74 # Oban runs under room_sanctum only (see config/config.exs).
75 − {Plugins.Oban, metric_prefix: [:phx, :oban]}
75 + {Plugins.Oban, metric_prefix: [:phx, :oban]},
76 + # Not a PromEx plugin — see the module. Counts configured sources and the
77 + # rows they have accumulated, which nothing in the app emits telemetry
78 + # for, so it asks the database on a timer.
79 + {RoomObservatory.Plugins.Sanctum, metric_prefix: [:phx, :sanctum]}
76 80 ]
77 81 end
78 82 end

Parents: fb200bd