Graham McIntire
56b724b5a4
Some checks failed
Build and Push / Build and Push Docker Image (push) Has been cancelled
The ANALYZE queries run via Task.Supervised which doesn't hold sandbox ownership. rescue only catches exceptions; sandbox-pool checkout failures come as exit signals. Add catch :exit so the Task logs a warning instead of crashing.
219 lines
8.5 KiB
Elixir
219 lines
8.5 KiB
Elixir
defmodule Microwaveprop.Weather do
|
|
@moduledoc false
|
|
|
|
alias Ecto.UUID
|
|
alias Microwaveprop.Repo
|
|
alias Microwaveprop.Weather.GefsProfile
|
|
alias Microwaveprop.Weather.Grid
|
|
alias Microwaveprop.Weather.Hrrr
|
|
alias Microwaveprop.Weather.Iemre
|
|
alias Microwaveprop.Weather.Narr
|
|
alias Microwaveprop.Weather.Soundings
|
|
alias Microwaveprop.Weather.Surface
|
|
|
|
require Logger
|
|
|
|
# ── Delegates to sub-facades ──
|
|
|
|
# HRRR
|
|
defdelegate find_nearest_hrrr(lat, lon, timestamp), to: Hrrr
|
|
defdelegate hrrr_profiles_for_path(contact), to: Hrrr
|
|
defdelegate hrrr_profiles_for_contacts(contacts), to: Hrrr
|
|
defdelegate find_nearest_native_profile(lat, lon, timestamp), to: Hrrr
|
|
defdelegate best_profile_for_contact(contact), to: Hrrr
|
|
defdelegate hrrr_for_contact(contact), to: Hrrr
|
|
defdelegate hrrr_data_fully_present?(contact), to: Hrrr
|
|
defdelegate has_hrrr_profile?(lat, lon, valid_time), to: Hrrr
|
|
defdelegate hrrr_points_present_batch(points), to: Hrrr
|
|
defdelegate round_to_hrrr_grid(lat, lon), to: Hrrr
|
|
defdelegate purge_grid_point_profiles(), to: Hrrr
|
|
defdelegate upsert_hrrr_profile(attrs), to: Hrrr
|
|
defdelegate upsert_hrrr_profiles_batch(profiles, opts \\ []), to: Hrrr
|
|
defdelegate nearest_native_duct_ghz(lat, lon, timestamp), to: Hrrr
|
|
defdelegate nearest_native_duct_info(lat, lon, timestamp), to: Hrrr
|
|
defdelegate backfill_hrrr_scalars(opts \\ []), to: Hrrr
|
|
defdelegate reconcile_hrrr_statuses(), to: Hrrr
|
|
defdelegate profiles_along_path(contact), to: Hrrr
|
|
|
|
# NARR
|
|
defdelegate narr_profiles_for_path(contact), to: Narr
|
|
defdelegate find_nearest_narr(lat, lon, timestamp), to: Narr
|
|
defdelegate narr_for_contact(contact), to: Narr
|
|
|
|
# IEMRE
|
|
defdelegate iemre_for_contact(contact), to: Iemre
|
|
defdelegate iemre_for_path(contact), to: Iemre
|
|
defdelegate find_nearest_iemre(lat, lon, timestamp), to: Iemre
|
|
defdelegate round_to_iemre_grid(lat, lon), to: Iemre
|
|
defdelegate upsert_iemre_observation(attrs), to: Iemre
|
|
defdelegate has_iemre_observation?(lat, lon, date), to: Iemre
|
|
defdelegate reconcile_iemre_statuses(), to: Iemre
|
|
|
|
# Soundings
|
|
defdelegate upsert_sounding(station, attrs), to: Soundings
|
|
defdelegate has_sounding?(station_id, observed_at), to: Soundings
|
|
defdelegate station_ids_with_soundings(station_ids, sounding_times), to: Soundings
|
|
defdelegate sounding_times_around(dt), to: Soundings
|
|
defdelegate weather_for_contact(contact_params, opts \\ []), to: Soundings
|
|
defdelegate soundings_with_widening_radius(params), to: Soundings
|
|
defdelegate nearest_sounding_to(lat, lon, timestamp, opts \\ []), to: Soundings
|
|
|
|
# Surface
|
|
defdelegate find_or_create_station(attrs), to: Surface
|
|
defdelegate upsert_surface_observation(station, attrs), to: Surface
|
|
defdelegate upsert_surface_observations(station, rows), to: Surface
|
|
defdelegate has_surface_observations?(station_id, start_dt, end_dt), to: Surface
|
|
defdelegate station_day_covered?(station_id, date), to: Surface
|
|
defdelegate station_day_pairs_covered(pairs), to: Surface
|
|
defdelegate station_ids_with_surface_observations(station_ids, start_dt, end_dt), to: Surface
|
|
defdelegate nearby_stations(lat, lon, station_type, radius_km), to: Surface
|
|
defdelegate sync_stations!(), to: Surface
|
|
defdelegate reconcile_weather_statuses(), to: Surface
|
|
defdelegate upsert_solar_index(attrs), to: Surface
|
|
defdelegate upsert_solar_indices_batch(records), to: Surface
|
|
defdelegate get_solar_index(date), to: Surface
|
|
defdelegate existing_solar_dates(), to: Surface
|
|
|
|
# Grid
|
|
defdelegate latest_grid_valid_time(), to: Grid
|
|
defdelegate latest_weather_grid(bounds), to: Grid
|
|
defdelegate load_weather_grid(bounds), to: Grid
|
|
defdelegate available_weather_valid_times(), to: Grid
|
|
defdelegate available_hrdps_valid_times(), to: Grid
|
|
defdelegate weather_grid_hrdps_at(valid_time, bounds), to: Grid
|
|
defdelegate weather_grid_at(valid_time, bounds), to: Grid
|
|
defdelegate warm_grid_cache_and_broadcast(valid_time), to: Grid
|
|
defdelegate warm_grid_cache_from_latest_profile(), to: Grid
|
|
defdelegate materialize_scalar_file(valid_time), to: Grid
|
|
defdelegate build_grid_cache_rows(grid_data, valid_time, bounds \\ nil), to: Grid
|
|
defdelegate weather_point_detail(lat, lon, valid_time), to: Grid
|
|
|
|
# ── Remaining inline: GEFS ──
|
|
|
|
@spec upsert_gefs_profile(map()) :: {:ok, GefsProfile.t()} | {:error, Ecto.Changeset.t()}
|
|
def upsert_gefs_profile(attrs) do
|
|
changeset = GefsProfile.changeset(%GefsProfile{}, attrs)
|
|
|
|
if changeset.valid? do
|
|
Repo.insert(changeset,
|
|
on_conflict: :nothing,
|
|
conflict_target: [:lat, :lon, :valid_time]
|
|
)
|
|
else
|
|
{:error, changeset}
|
|
end
|
|
end
|
|
|
|
@spec upsert_gefs_profiles_batch([map()]) :: {non_neg_integer(), nil}
|
|
def upsert_gefs_profiles_batch(profiles) do
|
|
Microwaveprop.Instrument.span(
|
|
[:db, :upsert_gefs_profiles],
|
|
%{count: length(profiles)},
|
|
fn -> do_upsert_gefs_profiles_batch(profiles) end
|
|
)
|
|
end
|
|
|
|
defp do_upsert_gefs_profiles_batch(profiles) do
|
|
now = DateTime.truncate(DateTime.utc_now(), :second)
|
|
|
|
profiles
|
|
|> Enum.chunk_every(500)
|
|
|> Enum.reduce({0, nil}, fn chunk, {total_count, _} ->
|
|
entries =
|
|
Enum.map(chunk, fn attrs ->
|
|
Map.merge(attrs, %{
|
|
id: UUID.generate(),
|
|
inserted_at: now,
|
|
updated_at: now
|
|
})
|
|
end)
|
|
|
|
{count, rows} =
|
|
Repo.insert_all(GefsProfile, entries,
|
|
on_conflict: :nothing,
|
|
conflict_target: [:lat, :lon, :valid_time]
|
|
)
|
|
|
|
{total_count + count, rows}
|
|
end)
|
|
end
|
|
|
|
# ── DB Maintenance ──
|
|
|
|
@doc """
|
|
Kick off `ANALYZE` on every public-schema table that hasn't been
|
|
auto-analyzed recently. Returns immediately with the list of tables
|
|
it will visit; the actual work runs in a supervised Task on the
|
|
current node so a broken `kubectl exec` doesn't abort it mid-way.
|
|
|
|
Callable from a release shell:
|
|
|
|
bin/microwaveprop rpc 'Microwaveprop.Weather.analyze_all()'
|
|
|
|
Surfaces the HRRR partition problem: most partitions have zero
|
|
analyze stats, so the planner picks nested-loop joins that scan
|
|
81 M rows. Running ANALYZE once gives it live_tup and column
|
|
histograms to work with.
|
|
|
|
`skip_recent` skips anything auto-analyzed in the last N seconds
|
|
(default 6 h) so re-running is cheap. Progress is logged via
|
|
`Logger.info` — tail `kubectl -n prop logs deploy/prop-backfill`
|
|
to watch.
|
|
"""
|
|
@spec analyze_all(keyword()) :: %{queued: non_neg_integer(), tables: [String.t()]}
|
|
def analyze_all(opts \\ []) do
|
|
skip_recent = Keyword.get(opts, :skip_recent_seconds, 6 * 3600)
|
|
|
|
# `last_autoanalyze` is only set by the autovacuum daemon;
|
|
# `last_analyze` is set by manual ANALYZE (including our Task
|
|
# below). Take the newer of the two — otherwise a re-run within
|
|
# `skip_recent` would re-analyze everything we just did.
|
|
q = """
|
|
SELECT relname
|
|
FROM pg_stat_user_tables
|
|
WHERE schemaname = 'public'
|
|
AND COALESCE(
|
|
GREATEST(last_analyze, last_autoanalyze),
|
|
'epoch'::timestamptz
|
|
) < now() - ($1 || ' seconds')::interval
|
|
ORDER BY n_live_tup ASC
|
|
"""
|
|
|
|
%{rows: rows} = Repo.query!(q, [Integer.to_string(skip_recent)])
|
|
tables = Enum.map(rows, fn [n] -> n end)
|
|
|
|
# Fire-and-forget: run each ANALYZE on the pod's own node so the
|
|
# caller returns before any single query finishes. Small tables
|
|
# first keeps progress visible early.
|
|
_ =
|
|
Task.Supervisor.start_child({:via, PartitionSupervisor, {Microwaveprop.TaskSupervisor, self()}}, fn ->
|
|
Enum.each(tables, fn t ->
|
|
Logger.info("Weather.analyze_all: ANALYZE #{t}")
|
|
started = System.monotonic_time(:millisecond)
|
|
|
|
try do
|
|
_ = Repo.query!("ANALYZE #{quote_ident(t)}", [], timeout: :infinity)
|
|
elapsed = System.monotonic_time(:millisecond) - started
|
|
Logger.info("Weather.analyze_all: ANALYZE #{t} done in #{elapsed}ms")
|
|
rescue
|
|
e -> Logger.warning("Weather.analyze_all: ANALYZE #{t} failed: #{inspect(e)}")
|
|
catch
|
|
:exit, reason ->
|
|
Logger.warning("Weather.analyze_all: ANALYZE #{t} sandbox exit: #{inspect(reason)}")
|
|
end
|
|
end)
|
|
|
|
Logger.info("Weather.analyze_all: complete (#{length(tables)} tables)")
|
|
end)
|
|
|
|
%{queued: length(tables), tables: tables}
|
|
end
|
|
|
|
# Defensive: pg_stat_user_tables returns legitimate identifiers from
|
|
# the catalog, but we still quote just in case a partition contains a
|
|
# character the shell would interpret.
|
|
defp quote_ident(name) when is_binary(name) do
|
|
escaped = String.replace(name, ~s("), ~s(""))
|
|
~s("#{escaped}")
|
|
end
|
|
end
|