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.

Record GTFS-realtime feed health, and expose it as metrics

d446902 · Gabriel Morell · 2026-08-24 13:26

27 files +347 -8
Message 
{commit_body(@commit)}

Files changed

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

added apps/room_gtfs/lib/room_gtfs/feed_health.ex
+168 −0
@@ -0,0 +1,168 @@
1 +defmodule RoomGtfs.FeedHealth do
2 + @moduledoc """
3 + The last thing each realtime feed did, so prod is debuggable without a shell.
4 +
5 + ## Why this and not the logs
6 +
7 + The logs already say when a feed fails, and say it every thirty seconds
8 + forever. What they cannot answer is the question you actually have:
9 +
10 + which of my feeds are working right now, and since when
11 +
12 + Answering that from Loki means knowing every source's name, grepping for the
13 + absence of errors, and trusting that a feed nobody has logged about is
14 + healthy rather than silently not being polled at all. Two of the three real
15 + problems in this fleet were invisible that way: a source whose realtime URLs
16 + pointed at a `.proto` definition rather than a feed only showed up as a
17 + `:decode_failed` line that named a truncated URL, and a feed that had simply
18 + stopped being fetched would produce no log line whatsoever.
19 +
20 + So every fetch attempt records its outcome here, keyed by source and kind,
21 + and `RoomObservatory` turns that into metrics. A feed that stops being polled
22 + shows as a last-success age that climbs, which is the one symptom silence
23 + cannot produce.
24 +
25 + ## Shape
26 +
27 + An ETS table of `{{source_id, kind}, record}`, where kind is `:sa`, `:tu` or
28 + `:vp`. Written by the realtime workers, read by the metrics poller and the
29 + source page. Public and unsupervised on purpose -- see `ensure_table/0`: a
30 + missing table must never be the reason a poll fails.
31 + """
32 +
33 + @table :room_gtfs_feed_health
34 +
35 + @typedoc "sa = service alerts, tu = trip updates, vp = vehicle positions"
36 + @type kind :: :sa | :tu | :vp
37 +
38 + @doc """
39 + Note the outcome of one fetch.
40 +
41 + `result` is what the fetch returned. Success records the entity count so a
42 + feed that starts answering with nothing is distinguishable from one that has
43 + stopped answering; failure records the reason and leaves the previous success
44 + in place, because "failing now, last worked eleven minutes ago" is the useful
45 + reading and "failing now" alone is not.
46 + """
47 + def record(source_id, source_name, kind, url, result, duration_ms) do
48 + ensure_table()
49 + now = System.system_time(:second)
50 + previous = get(source_id, kind)
51 +
52 + record =
53 + case result do
54 + {:ok, feed} ->
55 + %{
56 + ok: true,
57 + entities: length(feed.entity),
58 + reason: nil,
59 + last_success_at: now,
60 + extensions: count_extensions(feed)
61 + }
62 +
63 + {:error, reason} ->
64 + %{
65 + ok: false,
66 + entities: previous[:entities],
67 + reason: describe(reason),
68 + last_success_at: previous[:last_success_at],
69 + extensions: previous[:extensions]
70 + }
71 + end
72 +
73 + record =
74 + Map.merge(record, %{
75 + source_id: source_id,
76 + source_name: source_name,
77 + kind: kind,
78 + url: url,
79 + last_attempt_at: now,
80 + duration_ms: duration_ms
81 + })
82 +
83 + :ets.insert(@table, {{source_id, kind}, record})
84 + record
85 + end
86 +
87 + @doc "One feed's last known state, or nil."
88 + def get(source_id, kind) do
89 + ensure_table()
90 +
91 + case :ets.lookup(@table, {source_id, kind}) do
92 + [{_key, record}] -> record
93 + [] -> nil
94 + end
95 + end
96 +
97 + @doc "Every feed's last known state, newest attempt first."
98 + def all do
99 + ensure_table()
100 +
101 + @table
102 + |> :ets.tab2list()
103 + |> Enum.map(fn {_key, record} -> record end)
104 + |> Enum.sort_by(& &1.last_attempt_at, :desc)
105 + end
106 +
107 + @doc "Forget a source's feeds, for when one is deleted or reconfigured."
108 + def forget(source_id) do
109 + ensure_table()
110 + Enum.each([:sa, :tu, :vp], &:ets.delete(@table, {source_id, &1}))
111 + end
112 +
113 + # Created on first use rather than owned by a supervisor. The alternative is
114 + # another process in the tree whose only failure mode is taking realtime
115 + # polling down with it; a table that races two workers into creating it loses
116 + # one insert, once, at boot.
117 + defp ensure_table do
118 + case :ets.whereis(@table) do
119 + :undefined ->
120 + :ets.new(@table, [:named_table, :public, :set, read_concurrency: true])
121 + :ok
122 +
123 + _tid ->
124 + :ok
125 + end
126 + rescue
127 + ArgumentError -> :ok
128 + end
129 +
130 + # How many extension blobs the decoder could not name. Zero is the healthy
131 + # reading: it means every field in the feed has a definition behind it. A
132 + # number that appears after an agency changes its feed is the earliest
133 + # warning that we are dropping something.
134 + defp count_extensions(feed) do
135 + Enum.reduce(feed.entity, length(feed.header.__unknown_fields__), fn e, acc ->
136 + acc + length(e.__unknown_fields__) + unknown_in(e.trip_update) +
137 + unknown_in(e.vehicle) + unknown_in(e.alert)
138 + end)
139 + end
140 +
141 + defp unknown_in(nil), do: 0
142 +
143 + defp unknown_in(%{__unknown_fields__: fields} = message) do
144 + length(fields) + nested_unknown(message)
145 + end
146 +
147 + defp unknown_in(_other), do: 0
148 +
149 + defp nested_unknown(%{stop_time_update: updates}) when is_list(updates) do
150 + Enum.sum(Enum.map(updates, &length(&1.__unknown_fields__)))
151 + end
152 +
153 + defp nested_unknown(%{informed_entity: entities}) when is_list(entities) do
154 + Enum.sum(Enum.map(entities, &length(&1.__unknown_fields__)))
155 + end
156 +
157 + defp nested_unknown(_message), do: 0
158 +
159 + # Reasons reach metrics as a label, so they have to be a small closed set
160 + # rather than whatever an HTTP client happened to raise.
161 + defp describe(:bad_status), do: "bad_status"
162 + defp describe(:decode_failed), do: "decode_failed"
163 + defp describe(:not_protobuf), do: "not_protobuf"
164 + defp describe(%HTTPoison.Error{reason: reason}), do: "http_#{reason}"
165 + defp describe({:fetch_crashed, _reason}), do: "crashed"
166 + defp describe(reason) when is_atom(reason), do: Atom.to_string(reason)
167 + defp describe(_reason), do: "unknown"
168 +end
modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+17 −3
@@ -325,6 +325,20 @@ defmodule RoomGtfs.Worker.RT do
325 325 Phoenix.PubSub.broadcast(RoomSanctum.PubSub, "gtfs", {:gtfs, id, :disabled})
326 326 end
327 327
328 + # Every realtime fetch goes through here so that the outcome is recorded
329 + # whether it worked or not. The logs already say when one fails; what they
330 + # cannot say is that a feed stopped being fetched at all, which is the failure
331 + # that looks exactly like silence. See RoomGtfs.FeedHealth.
332 + defp fetch_recorded(state, kind, url) do
333 + started = System.monotonic_time(:millisecond)
334 + result = FeedCache.get(url)
335 + took = System.monotonic_time(:millisecond) - started
336 +
337 + RoomGtfs.FeedHealth.record(state.id, state.inst && state.inst.name, kind, url, result, took)
338 +
339 + result
340 + end
341 +
328 342 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
329 343
330 344 def handle_call({:query_alerts, stop, route_ids}, _from, state) do
@@ -682,7 +696,7 @@ defmodule RoomGtfs.Worker.RT do
682 696 state
683 697
684 698 val ->
685 − case FeedCache.get(val) do
699 + case fetch_recorded(state, :sa, val) do
686 700 {:ok, data_sa} when is_struct(data_sa, TransitRealtime.FeedMessage) ->
687 701 state |> Map.put(:rt_sa, data_sa)
688 702
@@ -701,7 +715,7 @@ defmodule RoomGtfs.Worker.RT do
701 715 state
702 716
703 717 val ->
704 − case FeedCache.get(val) do
718 + case fetch_recorded(state, :tu, val) do
705 719 {:ok, data_tu} when is_struct(data_tu, TransitRealtime.FeedMessage) ->
706 720 state |> Map.put(:rt_tu, data_tu)
707 721
@@ -720,7 +734,7 @@ defmodule RoomGtfs.Worker.RT do
720 734 state
721 735
722 736 val ->
723 − case FeedCache.get(val) do
737 + case fetch_recorded(state, :vp, val) do
724 738 {:ok, data_vp} when is_struct(data_vp, TransitRealtime.FeedMessage) ->
725 739 new_state = state |> Map.put(:rt_vp, data_vp) |> ensure_stops()
726 740
added apps/room_observatory/lib/room_observatory/plugins/gtfs_realtime.ex
+140 −0

Click to load diff…

modified apps/room_observatory/lib/room_observatory/prom_ex.ex
+5 −1

Click to load diff…

modified apps/room_observatory/mix.exs
+5 −1

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/components/query_geospatial_map.ex
+6 −1

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/map_view.ex
+4 −1

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/map_view.html.heex
+1 −1

Click to load diff…

modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/stations.ex
+1 −0

Click to load diff…

Parents: bf3e0f1