Graham McIntire
e107706915
Removed n_live_tup > 0 filter that hid tables when pg_stats reports 0 rows during VACUUM. Added dead tuple column with warning highlight to monitor vacuum progress. Also includes Styler reformatting and unused dep cleanup.
470 lines
16 KiB
Elixir
470 lines
16 KiB
Elixir
defmodule MicrowavepropWeb.BackfillLive do
|
|
@moduledoc false
|
|
use MicrowavepropWeb, :live_view
|
|
use LiveStash
|
|
|
|
import Ecto.Query
|
|
|
|
alias Microwaveprop.Radio.Contact
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
|
|
|
|
@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()
|
|
|
|
{limit, _} =
|
|
case LiveStash.recover_state(socket) do
|
|
{:recovered, socket} -> {socket.assigns[:limit] || 500, socket}
|
|
_ -> {500, socket}
|
|
end
|
|
|
|
{:ok,
|
|
assign(socket,
|
|
page_title: "Backfill",
|
|
limit: limit,
|
|
stats: stats,
|
|
unprocessed: unprocessed,
|
|
db_stats: db_stats,
|
|
last_enqueued: nil,
|
|
enqueuing: false,
|
|
refresh_timer: nil
|
|
)}
|
|
end
|
|
|
|
@impl true
|
|
def handle_event("enqueue", %{"limit" => limit_str}, socket) do
|
|
limit = String.to_integer(limit_str)
|
|
|
|
# Run in a task to not block the LiveView
|
|
pid = self()
|
|
|
|
Task.start(fn ->
|
|
count = enqueue_batch(limit)
|
|
send(pid, {:enqueued, count})
|
|
end)
|
|
|
|
{:noreply, socket |> assign(enqueuing: true, limit: limit) |> LiveStash.stash_assigns([:limit])}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info({:enqueued, count}, socket) do
|
|
{:noreply, assign(socket, last_enqueued: count, enqueuing: false)}
|
|
end
|
|
|
|
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
|
|
ref = Process.send_after(self(), :refresh_stats, 1000)
|
|
assign(socket, refresh_timer: ref)
|
|
end
|
|
|
|
@enrichable [:pending, :failed]
|
|
|
|
defp enqueue_batch(limit) do
|
|
contacts =
|
|
Repo.all(
|
|
from(c in Contact,
|
|
where:
|
|
(not is_nil(c.pos1) or not is_nil(c.grid1)) and
|
|
(c.hrrr_status in ^@enrichable or c.weather_status in ^@enrichable or c.terrain_status in ^@enrichable or
|
|
c.iemre_status in ^@enrichable),
|
|
order_by: [desc: c.qso_timestamp],
|
|
limit: ^limit
|
|
)
|
|
)
|
|
|
|
# ensure_positions! is called inside enqueue_for_contact
|
|
Enum.each(contacts, &ContactWeatherEnqueueWorker.enqueue_for_contact/1)
|
|
length(contacts)
|
|
end
|
|
|
|
defp count_unprocessed do
|
|
incomplete = [:pending, :queued, :processing, :failed]
|
|
|
|
terrain =
|
|
Repo.one(from(c in Contact, where: c.terrain_status in ^incomplete and not is_nil(c.pos1), select: count()))
|
|
|
|
hrrr =
|
|
Repo.one(from(c in Contact, where: c.hrrr_status in ^incomplete and not is_nil(c.pos1), select: count()))
|
|
|
|
weather =
|
|
Repo.one(from(c in Contact, where: c.weather_status in ^incomplete and not is_nil(c.pos1), select: count()))
|
|
|
|
%{
|
|
terrain: terrain,
|
|
hrrr: hrrr,
|
|
weather: weather,
|
|
total: max(terrain, max(hrrr, weather))
|
|
}
|
|
end
|
|
|
|
defp fetch_stats 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(fn {worker, states} ->
|
|
short_name = worker |> String.split(".") |> List.last()
|
|
total = Enum.reduce(states, 0, fn {_, c}, acc -> acc + c end)
|
|
executing = Enum.find_value(states, 0, fn {s, c} -> if s == "executing", do: c end)
|
|
available = Enum.find_value(states, 0, fn {s, c} -> if s == "available", do: c end)
|
|
retryable = Enum.find_value(states, 0, fn {s, c} -> if s == "retryable", do: c end)
|
|
|
|
%{
|
|
worker: short_name,
|
|
total: total,
|
|
executing: executing,
|
|
available: available,
|
|
retryable: retryable
|
|
}
|
|
end)
|
|
|> 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 fetch_db_stats 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_%'
|
|
ORDER BY pg_total_relation_size(c.oid) 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)
|
|
|
|
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,
|
|
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.reverse/1)
|
|
|> String.reverse()
|
|
|> then(fn s -> String.replace(s, ~r/^,/, "") end)
|
|
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}>
|
|
<.header>
|
|
Backfill Dashboard
|
|
<:subtitle>Enqueue and monitor contact enrichment jobs</:subtitle>
|
|
</.header>
|
|
|
|
<div class="grid grid-cols-1 md:grid-cols-3 gap-4 my-6">
|
|
<div class="bg-base-200 rounded-box p-4">
|
|
<div class="text-xs opacity-60">Enrichment Progress</div>
|
|
<div class="text-2xl font-bold">{@unprocessed.total} remaining</div>
|
|
<div class="text-xs opacity-60 mt-1">
|
|
{@db_stats.totals.contacts - @unprocessed.total} / {@db_stats.totals.contacts} complete
|
|
</div>
|
|
<progress
|
|
class="progress progress-primary w-full mt-2"
|
|
value={@db_stats.totals.contacts - @unprocessed.total}
|
|
max={@db_stats.totals.contacts}
|
|
/>
|
|
<div class="flex gap-4 mt-2 text-xs opacity-60">
|
|
<span>Terrain: {@unprocessed.terrain}</span>
|
|
<span>HRRR: {@unprocessed.hrrr}</span>
|
|
<span>Weather: {@unprocessed.weather}</span>
|
|
</div>
|
|
</div>
|
|
<div class="bg-base-200 rounded-box p-4">
|
|
<div class="text-xs opacity-60">Active/Queued Jobs</div>
|
|
<div class="text-2xl font-bold">
|
|
{Enum.reduce(@stats.by_worker, 0, fn w, acc -> acc + w.total end)}
|
|
</div>
|
|
</div>
|
|
<div class="bg-base-200 rounded-box p-4">
|
|
<div class="text-xs opacity-60">Completed (last hour)</div>
|
|
<div class="text-2xl font-bold">{@stats.completed_1h}</div>
|
|
</div>
|
|
</div>
|
|
|
|
<div class="bg-base-200 rounded-box p-4 mb-6">
|
|
<form phx-submit="enqueue" class="flex items-end gap-4">
|
|
<div>
|
|
<label class="label text-sm">Contacts to enqueue</label>
|
|
<input
|
|
type="number"
|
|
name="limit"
|
|
value={@limit}
|
|
min="1"
|
|
max="5000"
|
|
class="input input-bordered w-32"
|
|
/>
|
|
</div>
|
|
<button class="btn btn-primary" disabled={@enqueuing}>
|
|
<%= if @enqueuing do %>
|
|
<span class="loading loading-spinner loading-sm"></span> Enqueuing...
|
|
<% else %>
|
|
Enqueue Backfill
|
|
<% end %>
|
|
</button>
|
|
<%= if @last_enqueued do %>
|
|
<span class="text-sm text-success">
|
|
Enqueued {@last_enqueued} contacts
|
|
</span>
|
|
<% end %>
|
|
</form>
|
|
</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"}] 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="bg-base-200 rounded-box p-3">
|
|
<div class="text-xs opacity-60">Total DB Size</div>
|
|
<div class="text-lg font-bold font-mono">{@db_stats.db_size}</div>
|
|
</div>
|
|
<div class="bg-base-200 rounded-box p-3">
|
|
<div class="text-xs opacity-60">HRRR Profiles</div>
|
|
<div class="text-lg font-bold font-mono">
|
|
{format_number(@db_stats.totals.hrrr_profiles)}
|
|
</div>
|
|
<div class="text-xs opacity-60">{@db_stats.averages.hrrr_per_contact} per contact</div>
|
|
</div>
|
|
<div class="bg-base-200 rounded-box p-3">
|
|
<div class="text-xs opacity-60">Surface Observations</div>
|
|
<div class="text-lg font-bold font-mono">
|
|
{format_number(@db_stats.totals.surface_observations)}
|
|
</div>
|
|
<div class="text-xs opacity-60">{@db_stats.averages.obs_per_contact} per contact</div>
|
|
</div>
|
|
<div class="bg-base-200 rounded-box p-3">
|
|
<div class="text-xs opacity-60">Soundings</div>
|
|
<div class="text-lg font-bold font-mono">{format_number(@db_stats.totals.soundings)}</div>
|
|
<div class="text-xs opacity-60">{@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
|