Graham McIntire
3604896726
- Schema: Era5Profile → NarrProfile; table renamed via migration rename_era5_profiles_to_narr_profiles (ALTER TABLE + rename indexes, atomic and instant, no row rewrite) - Weather helpers: find_nearest_era5/era5_for_contact/era5_profiles_for_path → find_nearest_narr/narr_for_contact/narr_profiles_for_path - BackfillEnqueueWorker: :era5 → :narr type (virtual, no narr_status column); reconcile_stale_queued_to_unavailable and status_priority_order now skip virtual types via @virtual_types. Fixes the prod crash where the cron tried to update a non-existent era5_status column. - ContactWeatherEnqueueWorker: build_era5_jobs → build_narr_jobs, era5_jobs_for_contact → narr_jobs_for_contact - ContactLive: @era5 → @narr, maybe_enqueue_era5 → maybe_enqueue_narr; UI label "ERA5 (0.25°)" → "NARR (32 km)" - Cron (runtime.exs): args types list now "narr" instead of "era5" - /status: progress row, status-by-type card, totals.narr_profiles, and table-count lookup all target narr_profiles - Drop obsolete Era5CdsJob schema + era5_cds_jobs table (inspection artifact from the retired CDS pipeline; 34 orphan rows in prod) - Misc docstring/comment cleanups (skew_t, about, wgrib2, propagation_train) Includes a regression test for the virtual-type crash.
512 lines
17 KiB
Elixir
512 lines
17 KiB
Elixir
defmodule MicrowavepropWeb.StatusLive do
|
|
@moduledoc false
|
|
use MicrowavepropWeb, :live_view
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Cache
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
|
|
@impl true
|
|
def mount(_params, _session, socket) do
|
|
if connected?(socket) do
|
|
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:contact_status")
|
|
Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "db:oban_jobs")
|
|
end
|
|
|
|
stats = fetch_stats()
|
|
unprocessed = count_unprocessed()
|
|
db_stats = fetch_db_stats()
|
|
|
|
{:ok,
|
|
assign(socket,
|
|
page_title: "Status",
|
|
stats: stats,
|
|
unprocessed: unprocessed,
|
|
db_stats: db_stats,
|
|
refresh_timer: nil
|
|
)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:contact_status_changed, _data}, socket) do
|
|
{:noreply, schedule_refresh(socket)}
|
|
end
|
|
|
|
def handle_info({:oban_job_changed, _data}, socket) do
|
|
{:noreply, schedule_refresh(socket)}
|
|
end
|
|
|
|
def handle_info(:refresh_stats, socket) do
|
|
{:noreply,
|
|
assign(socket,
|
|
stats: fetch_stats(),
|
|
unprocessed: count_unprocessed(),
|
|
db_stats: fetch_db_stats(),
|
|
refresh_timer: nil
|
|
)}
|
|
end
|
|
|
|
defp schedule_refresh(%{assigns: %{refresh_timer: ref}} = socket) when is_reference(ref) do
|
|
# Already scheduled, skip
|
|
socket
|
|
end
|
|
|
|
defp schedule_refresh(socket) do
|
|
# 2s debounce — bulk enrichment runs produce hundreds of status_changed
|
|
# events per second, and each stats refresh fires 8+ count queries.
|
|
ref = Process.send_after(self(), :refresh_stats, 2_000)
|
|
assign(socket, refresh_timer: ref)
|
|
end
|
|
|
|
defp count_unprocessed do
|
|
Cache.fetch_or_store({__MODULE__, :count_unprocessed}, 2_000, fn ->
|
|
count_unprocessed_raw()
|
|
end)
|
|
end
|
|
|
|
defp count_unprocessed_raw do
|
|
incomplete = [:pending, :queued, :processing, :failed]
|
|
done = [:complete, :unavailable]
|
|
|
|
total = Repo.one(from(c in Contact, where: not is_nil(c.pos1), select: count()))
|
|
terrain = count_incomplete(:terrain_status, incomplete)
|
|
hrrr = count_incomplete(:hrrr_status, incomplete)
|
|
weather = count_incomplete(:weather_status, incomplete)
|
|
iemre = count_incomplete(:iemre_status, incomplete)
|
|
|
|
narr_candidates =
|
|
Repo.one(from(c in Contact, where: c.hrrr_status == :unavailable and not is_nil(c.pos1), select: count()))
|
|
|
|
narr_done = count_narr_done()
|
|
|
|
all_done = count_all_done(done)
|
|
|
|
%{
|
|
terrain: terrain,
|
|
hrrr: hrrr,
|
|
weather: weather,
|
|
iemre: iemre,
|
|
narr_candidates: narr_candidates,
|
|
narr_done: narr_done,
|
|
remaining: max(terrain, max(hrrr, max(weather, iemre))),
|
|
complete: all_done,
|
|
total: total
|
|
}
|
|
end
|
|
|
|
# NARR is the fallback for pre-2014 contacts (HRRR archive starts 2014).
|
|
# "Done" means a narr_profiles row within 0.15° + 30 min of the contact,
|
|
# matching `Weather.find_nearest_narr/3` lookup tolerance.
|
|
defp count_narr_done do
|
|
%{rows: [[done]]} =
|
|
Repo.query!("""
|
|
SELECT count(*)
|
|
FROM contacts c
|
|
WHERE c.hrrr_status = 'unavailable'
|
|
AND c.pos1 IS NOT NULL
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM narr_profiles p
|
|
WHERE p.lat BETWEEN ((c.pos1->>'lat')::float - 0.15) AND ((c.pos1->>'lat')::float + 0.15)
|
|
AND p.lon BETWEEN ((c.pos1->>'lon')::float - 0.15) AND ((c.pos1->>'lon')::float + 0.15)
|
|
AND p.valid_time BETWEEN (c.qso_timestamp - INTERVAL '30 minutes') AND (c.qso_timestamp + INTERVAL '30 minutes')
|
|
)
|
|
""")
|
|
|
|
done
|
|
end
|
|
|
|
defp count_incomplete(status_field, statuses) do
|
|
Repo.one(
|
|
from(c in Contact,
|
|
where: field(c, ^status_field) in ^statuses and not is_nil(c.pos1),
|
|
select: count()
|
|
)
|
|
)
|
|
end
|
|
|
|
defp count_all_done(done) do
|
|
Repo.one(
|
|
from(c in Contact,
|
|
where:
|
|
not is_nil(c.pos1) and
|
|
c.hrrr_status in ^done and
|
|
c.weather_status in ^done and
|
|
c.terrain_status in ^done and
|
|
c.iemre_status in ^done,
|
|
select: count()
|
|
)
|
|
)
|
|
end
|
|
|
|
defp fetch_stats do
|
|
Cache.fetch_or_store({__MODULE__, :stats}, 2_000, fn -> fetch_stats_raw() end)
|
|
end
|
|
|
|
defp fetch_stats_raw do
|
|
jobs =
|
|
Repo.all(
|
|
from(j in "oban_jobs",
|
|
where: j.state in ["available", "executing", "scheduled", "retryable"],
|
|
group_by: [j.worker, j.state],
|
|
select: {j.worker, j.state, count(j.id)}
|
|
)
|
|
)
|
|
|
|
by_worker =
|
|
jobs
|
|
|> Enum.group_by(&elem(&1, 0), fn {_, state, count} -> {state, count} end)
|
|
|> Enum.map(&summarize_worker/1)
|
|
|> Enum.sort_by(& &1.total, :desc)
|
|
|
|
completed_1h =
|
|
Repo.one(
|
|
from(j in "oban_jobs",
|
|
where: j.state == "completed" and j.completed_at > ago(1, "hour"),
|
|
select: count(j.id)
|
|
)
|
|
)
|
|
|
|
%{by_worker: by_worker, completed_1h: completed_1h}
|
|
end
|
|
|
|
defp summarize_worker({worker, states}) do
|
|
short_name = worker |> String.split(".") |> List.last()
|
|
|
|
%{
|
|
worker: short_name,
|
|
total: Enum.reduce(states, 0, fn {_, c}, acc -> acc + c end),
|
|
executing: state_count(states, "executing"),
|
|
available: state_count(states, "available"),
|
|
retryable: state_count(states, "retryable")
|
|
}
|
|
end
|
|
|
|
defp state_count(states, name) do
|
|
Enum.find_value(states, 0, fn {s, c} -> if s == name, do: c end)
|
|
end
|
|
|
|
defp fetch_db_stats do
|
|
# pg_class + pg_stat_user_tables join is slow — 5s cache is fine since
|
|
# table row counts barely change in a 5-second window.
|
|
Cache.fetch_or_store({__MODULE__, :db_stats}, 5_000, fn -> fetch_db_stats_raw() end)
|
|
end
|
|
|
|
defp fetch_db_stats_raw do
|
|
%{rows: rows} =
|
|
Repo.query!("""
|
|
SELECT
|
|
c.relname as table,
|
|
s.n_live_tup as rows,
|
|
s.n_dead_tup as dead,
|
|
pg_total_relation_size(c.oid) as bytes
|
|
FROM pg_class c
|
|
JOIN pg_namespace n ON n.oid = c.relnamespace
|
|
JOIN pg_stat_user_tables s ON s.relid = c.oid
|
|
WHERE n.nspname = 'public'
|
|
AND c.relkind = 'r'
|
|
AND c.relname NOT LIKE 'oban_%'
|
|
AND c.relname NOT LIKE 'schema_%'
|
|
AND NOT EXISTS (SELECT 1 FROM pg_inherits WHERE inhrelid = c.oid)
|
|
UNION ALL
|
|
SELECT
|
|
parent.relname,
|
|
COALESCE(SUM(cs.n_live_tup), 0)::bigint,
|
|
COALESCE(SUM(cs.n_dead_tup), 0)::bigint,
|
|
COALESCE(SUM(pg_total_relation_size(child.oid)), 0)::bigint
|
|
FROM pg_class parent
|
|
JOIN pg_namespace n ON n.oid = parent.relnamespace
|
|
JOIN pg_inherits inh ON inh.inhparent = parent.oid
|
|
JOIN pg_class child ON child.oid = inh.inhrelid
|
|
LEFT JOIN pg_stat_user_tables cs ON cs.relid = child.oid
|
|
WHERE n.nspname = 'public' AND parent.relkind = 'p'
|
|
GROUP BY parent.relname
|
|
ORDER BY bytes DESC
|
|
LIMIT 20
|
|
""")
|
|
|
|
tables =
|
|
Enum.map(rows, fn [table, row_count, dead, bytes] ->
|
|
%{table: table, rows: row_count, dead: dead, size: format_bytes(bytes)}
|
|
end)
|
|
|
|
%{rows: total_rows} =
|
|
Repo.query!("SELECT pg_database_size(current_database())")
|
|
|
|
[[db_bytes]] = total_rows
|
|
|
|
# Enrichment status breakdown
|
|
%{rows: status_rows} =
|
|
Repo.query!("""
|
|
SELECT
|
|
'hrrr' as type, hrrr_status as status, count(*) as cnt FROM contacts GROUP BY hrrr_status
|
|
UNION ALL
|
|
SELECT
|
|
'weather', weather_status, count(*) FROM contacts GROUP BY weather_status
|
|
UNION ALL
|
|
SELECT
|
|
'terrain', terrain_status, count(*) FROM contacts GROUP BY terrain_status
|
|
UNION ALL
|
|
SELECT
|
|
'iemre', iemre_status, count(*) FROM contacts GROUP BY iemre_status
|
|
ORDER BY 1, 2
|
|
""")
|
|
|
|
statuses =
|
|
Enum.group_by(status_rows, fn [type, _, _] -> type end, fn [_, status, count] ->
|
|
{status, count}
|
|
end)
|
|
|
|
# Per-table row counts from pg_stat estimates (already fetched above, avoids slow count(*))
|
|
table_counts = Map.new(tables, fn %{table: t, rows: r} -> {t, r} end)
|
|
|
|
# NARR doesn't have its own status column — synthesize from HRRR unavailable
|
|
# + narr_profiles row count.
|
|
narr_count = Map.get(table_counts, "narr_profiles", 0)
|
|
|
|
hrrr_unavailable =
|
|
statuses |> Map.get("hrrr", []) |> Enum.find_value(0, fn {s, c} -> if s == "unavailable", do: c end)
|
|
|
|
statuses =
|
|
Map.put(statuses, "narr", [
|
|
{"candidates (HRRR unavail)", hrrr_unavailable},
|
|
{"profiles fetched", narr_count}
|
|
])
|
|
|
|
hrrr_count = Map.get(table_counts, "hrrr_profiles", 0)
|
|
terrain_count = Map.get(table_counts, "terrain_profiles", 0)
|
|
obs_count = Map.get(table_counts, "surface_observations", 0)
|
|
sounding_count = Map.get(table_counts, "soundings", 0)
|
|
iemre_count = Map.get(table_counts, "iemre_observations", 0)
|
|
contact_count = max(Map.get(table_counts, "contacts", 0), 1)
|
|
|
|
%{
|
|
db_size: format_bytes(db_bytes),
|
|
tables: tables,
|
|
statuses: statuses,
|
|
totals: %{
|
|
hrrr_profiles: hrrr_count,
|
|
narr_profiles: Map.get(table_counts, "narr_profiles", 0),
|
|
terrain_profiles: terrain_count,
|
|
surface_observations: obs_count,
|
|
soundings: sounding_count,
|
|
iemre_observations: iemre_count,
|
|
contacts: contact_count
|
|
},
|
|
averages: %{
|
|
hrrr_per_contact: Float.round(hrrr_count / contact_count, 1),
|
|
obs_per_contact: Float.round(obs_count / contact_count, 1),
|
|
soundings_per_contact: Float.round(sounding_count / contact_count, 1)
|
|
}
|
|
}
|
|
end
|
|
|
|
defp format_number(n) when is_integer(n) do
|
|
n
|
|
|> Integer.to_string()
|
|
|> String.graphemes()
|
|
|> Enum.reverse()
|
|
|> Enum.chunk_every(3)
|
|
|> Enum.map_join(",", &Enum.join/1)
|
|
|> String.reverse()
|
|
|> String.replace(~r/^,/, "")
|
|
end
|
|
|
|
defp format_number(n) when is_float(n), do: format_number(round(n))
|
|
defp format_number(n), do: to_string(n)
|
|
|
|
defp status_class("complete"), do: "text-success"
|
|
defp status_class("pending"), do: "text-warning"
|
|
defp status_class("queued"), do: "text-info"
|
|
defp status_class("processing"), do: "text-info"
|
|
defp status_class("failed"), do: "text-error"
|
|
defp status_class("unavailable"), do: "opacity-40"
|
|
defp status_class(_), do: ""
|
|
|
|
defp format_bytes(bytes) when bytes >= 1_073_741_824, do: "#{Float.round(bytes / 1_073_741_824, 1)} GB"
|
|
|
|
defp format_bytes(bytes) when bytes >= 1_048_576, do: "#{Float.round(bytes / 1_048_576, 1)} MB"
|
|
|
|
defp format_bytes(bytes) when bytes >= 1024, do: "#{Float.round(bytes / 1024, 1)} KB"
|
|
|
|
defp format_bytes(bytes), do: "#{bytes} B"
|
|
|
|
@impl true
|
|
def render(assigns) do
|
|
~H"""
|
|
<Layouts.app flash={@flash} current_scope={@current_scope}>
|
|
<.header>
|
|
Status
|
|
<:subtitle>Enrichment progress, job queues, and database health</:subtitle>
|
|
</.header>
|
|
|
|
<div class="grid grid-cols-1 md:grid-cols-4 gap-4 my-6">
|
|
<div class="stat bg-base-200 rounded-box overflow-hidden md:col-span-2">
|
|
<div class="stat-title">Enrichment Progress</div>
|
|
<div class="stat-value text-2xl">
|
|
{format_number(@unprocessed.complete)} / {format_number(@unprocessed.total)}
|
|
</div>
|
|
<div class="stat-desc">fully enriched</div>
|
|
<div class="flex flex-col gap-1.5 mt-3">
|
|
<%= for {key, label, complete, denom} <- [
|
|
{"hrrr", "HRRR", @unprocessed.total - @unprocessed.hrrr, @unprocessed.total},
|
|
{"weather", "Weather", @unprocessed.total - @unprocessed.weather, @unprocessed.total},
|
|
{"terrain", "Terrain", @unprocessed.total - @unprocessed.terrain, @unprocessed.total},
|
|
{"iemre", "IEMRE", @unprocessed.total - @unprocessed.iemre, @unprocessed.total},
|
|
{"narr", "NARR", @unprocessed.narr_done, @unprocessed.narr_candidates}
|
|
] do %>
|
|
<% pct = if denom > 0, do: round(complete * 100 / denom), else: 0 %>
|
|
<div class="flex items-center gap-2 text-xs" data-progress-key={key}>
|
|
<span class="w-14 shrink-0">{label}</span>
|
|
<progress
|
|
class="progress progress-primary flex-1 min-w-0"
|
|
value={pct}
|
|
max={100}
|
|
/>
|
|
<span class="shrink-0 font-mono opacity-60 whitespace-nowrap">
|
|
{format_number(complete)} / {format_number(denom)}
|
|
</span>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
</div>
|
|
<div class="stats stats-vertical bg-base-200 rounded-box md:col-span-2">
|
|
<div class="stat">
|
|
<div class="stat-title">Active/Queued Jobs</div>
|
|
<div class="stat-value text-2xl">
|
|
{Enum.reduce(@stats.by_worker, 0, fn w, acc -> acc + w.total end)}
|
|
</div>
|
|
</div>
|
|
<div class="stat">
|
|
<div class="stat-title">Completed (last hour)</div>
|
|
<div class="stat-value text-2xl">{format_number(@stats.completed_1h)}</div>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
|
|
<h2 class="text-base font-semibold mb-2">Job Queue Status</h2>
|
|
<%= if @stats.by_worker == [] do %>
|
|
<p class="text-sm text-base-content/50 italic">No active jobs.</p>
|
|
<% else %>
|
|
<div class="overflow-x-auto">
|
|
<table class="table table-sm table-zebra">
|
|
<thead>
|
|
<tr>
|
|
<th>Worker</th>
|
|
<th>Executing</th>
|
|
<th>Pending</th>
|
|
<th>Retryable</th>
|
|
<th>Total</th>
|
|
</tr>
|
|
</thead>
|
|
<tbody>
|
|
<%= for w <- @stats.by_worker do %>
|
|
<tr>
|
|
<td class="font-semibold">{w.worker}</td>
|
|
<td>
|
|
<%= if w.executing > 0 do %>
|
|
<span class="flex items-center gap-1">
|
|
<span class="loading loading-spinner loading-xs"></span> {w.executing}
|
|
</span>
|
|
<% else %>
|
|
0
|
|
<% end %>
|
|
</td>
|
|
<td>{w.available}</td>
|
|
<td>
|
|
<%= if w.retryable > 0 do %>
|
|
<span class="text-warning">{w.retryable}</span>
|
|
<% else %>
|
|
0
|
|
<% end %>
|
|
</td>
|
|
<td>{w.total}</td>
|
|
</tr>
|
|
<% end %>
|
|
</tbody>
|
|
</table>
|
|
</div>
|
|
<% end %>
|
|
|
|
<div class="divider" />
|
|
|
|
<h2 class="text-base font-semibold mb-2">Enrichment Status by Type</h2>
|
|
<div class="grid grid-cols-2 md:grid-cols-4 gap-4 mb-6">
|
|
<%= for {type, label} <- [{"hrrr", "HRRR"}, {"weather", "Weather"}, {"terrain", "Terrain"}, {"iemre", "IEMRE"}, {"narr", "NARR"}] do %>
|
|
<div class="bg-base-200 rounded-box p-3 text-xs">
|
|
<div class="font-semibold mb-1">{label}</div>
|
|
<%= for {status, count} <- Map.get(@db_stats.statuses, type, []) do %>
|
|
<div class="flex justify-between">
|
|
<span class={status_class(status)}>{status}</span>
|
|
<span class="font-mono">{format_number(count)}</span>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
|
|
<h2 class="text-base font-semibold mb-2">Database</h2>
|
|
<div class="grid grid-cols-2 md:grid-cols-4 gap-4 mb-4">
|
|
<div class="stat bg-base-200 rounded-box overflow-hidden">
|
|
<div class="stat-title">Total DB Size</div>
|
|
<div class="stat-value text-xl font-mono">{@db_stats.db_size}</div>
|
|
</div>
|
|
<div class="stat bg-base-200 rounded-box overflow-hidden">
|
|
<div class="stat-title">HRRR Profiles</div>
|
|
<div class="stat-value text-xl font-mono">
|
|
{format_number(@db_stats.totals.hrrr_profiles)}
|
|
</div>
|
|
<div class="stat-desc">{@db_stats.averages.hrrr_per_contact} per contact</div>
|
|
</div>
|
|
<div class="stat bg-base-200 rounded-box overflow-hidden">
|
|
<div class="stat-title">Surface Observations</div>
|
|
<div class="stat-value text-xl font-mono">
|
|
{format_number(@db_stats.totals.surface_observations)}
|
|
</div>
|
|
<div class="stat-desc">{@db_stats.averages.obs_per_contact} per contact</div>
|
|
</div>
|
|
<div class="stat bg-base-200 rounded-box overflow-hidden">
|
|
<div class="stat-title">Soundings</div>
|
|
<div class="stat-value text-xl font-mono">
|
|
{format_number(@db_stats.totals.soundings)}
|
|
</div>
|
|
<div class="stat-desc">{@db_stats.averages.soundings_per_contact} per contact</div>
|
|
</div>
|
|
</div>
|
|
|
|
<h2 class="text-base font-semibold mb-2">Table Sizes</h2>
|
|
<div class="overflow-x-auto mb-4">
|
|
<table class="table table-xs table-zebra">
|
|
<thead>
|
|
<tr>
|
|
<th>Table</th>
|
|
<th class="text-right">Rows</th>
|
|
<th class="text-right">Dead</th>
|
|
<th class="text-right">Size</th>
|
|
</tr>
|
|
</thead>
|
|
<tbody>
|
|
<%= for t <- @db_stats.tables do %>
|
|
<tr>
|
|
<td class="font-mono text-xs">{t.table}</td>
|
|
<td class="text-right font-mono text-xs">{format_number(t.rows)}</td>
|
|
<td class={[
|
|
"text-right font-mono text-xs",
|
|
t.dead > 0 && "text-warning"
|
|
]}>
|
|
{format_number(t.dead)}
|
|
</td>
|
|
<td class="text-right font-mono text-xs">{t.size}</td>
|
|
</tr>
|
|
<% end %>
|
|
</tbody>
|
|
</table>
|
|
</div>
|
|
|
|
<p class="text-xs text-base-content/50 mt-4">Live updates via database notifications.</p>
|
|
</Layouts.app>
|
|
"""
|
|
end
|
|
end
|