perf(map): collapse point_detail/forecast NFS I/O + silence /metrics logs

Clicks on the map were hanging for 1-3s, triggering the Phoenix topbar
progress bar visible to the user. Three hot-path improvements:

1. ScoresFile.read_point now uses a two-range :file.pread/2 — reads the
   33-byte header + one cell byte (~60 bytes) instead of the full ~93 KB
   file. point_forecast walks 19 forecast hours per click, so the old
   full-read path was the dominant cost on cold cache.

2. ProfilesFile.read/1 and list_valid_times/0 cached via
   Microwaveprop.Cache (5s TTL). Decoded profile maps are ~10 MB each
   (92k cells); timeline scrub that re-clicks the same valid_time
   within seconds now hits ETS instead of re-gunzipping + decoding.
   Cache keys include base_dir so test setup that swaps dirs doesn't
   see stale entries. Writes + prune + retain_window invalidate.

3. Endpoint log_level now filters /metrics the same way it already
   filters /health — Prometheus scrapes every 5s and was producing
   visible log spam.

Pod-local ETS is right for this workload (per-click read, tiny working
set); Valkey / shared memstore would not help here since each pod needs
its own fast-path lookup. File a follow-up if cross-pod score lookups
ever show up in flame graphs.
This commit is contained in:
Graham McIntire 2026-04-20 16:30:14 -05:00
parent 77b54a1642
commit bd3b1148a4
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
3 changed files with 167 additions and 49 deletions

View file

@ -109,6 +109,7 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
tmp = path <> ".tmp." <> unique_suffix() tmp = path <> ".tmp." <> unique_suffix()
File.write!(tmp, binary, [:binary]) File.write!(tmp, binary, [:binary])
File.rename!(tmp, path) File.rename!(tmp, path)
invalidate_caches_for(valid_time)
:ok :ok
end end
@ -130,6 +131,17 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
""" """
@spec read(DateTime.t()) :: {:ok, %{{float(), float()} => map()}} | {:error, :enoent} @spec read(DateTime.t()) :: {:ok, %{{float(), float()} => map()}} | {:error, :enoent}
def read(%DateTime{} = valid_time) do def read(%DateTime{} = valid_time) do
# Decoded profile maps are ~10 MB each (92k cells × atmospheric
# profile). A map click fires `point_detail` → `factors_for` which
# calls this, and the user scrubbing the forecast timeline hits
# the same valid_time repeatedly within seconds. Caching the full
# decoded map for 5s turns scrub clicks into ETS lookups.
Microwaveprop.Cache.fetch_or_store({__MODULE__, :read, base_dir(), valid_time}, 5_000, fn ->
do_read(valid_time)
end)
end
defp do_read(valid_time) do
# Rust-written MessagePack wins over Elixir-written ETF if both # Rust-written MessagePack wins over Elixir-written ETF if both
# exist. After Phase 3 Stream A cutover only `.mp.gz` is produced; # exist. After Phase 3 Stream A cutover only `.mp.gz` is produced;
# the ETF branch remains so pods picking up an older in-flight run # the ETF branch remains so pods picking up an older in-flight run
@ -211,16 +223,20 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
def prune_older_than(%DateTime{} = cutoff) do def prune_older_than(%DateTime{} = cutoff) do
cutoff_unix = DateTime.to_unix(cutoff) cutoff_unix = DateTime.to_unix(cutoff)
base_dir() deleted =
|> list_profile_files() base_dir()
|> Enum.reduce(0, fn {path, valid_time_unix}, acc -> |> list_profile_files()
if valid_time_unix < cutoff_unix do |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
_ = File.rm(path) if valid_time_unix < cutoff_unix do
acc + 1 _ = File.rm(path)
else acc + 1
acc else
end acc
end) end
end)
if deleted > 0, do: invalidate_all_caches()
deleted
end end
@doc """ @doc """
@ -234,16 +250,20 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
lo = DateTime.to_unix(run_time) lo = DateTime.to_unix(run_time)
hi = lo + max_forecast_hour * 3600 hi = lo + max_forecast_hour * 3600
base_dir() deleted =
|> list_profile_files() base_dir()
|> Enum.reduce(0, fn {path, valid_time_unix}, acc -> |> list_profile_files()
if valid_time_unix < lo or valid_time_unix > hi do |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
_ = File.rm(path) if valid_time_unix < lo or valid_time_unix > hi do
acc + 1 _ = File.rm(path)
else acc + 1
acc else
end acc
end) end
end)
if deleted > 0, do: invalidate_all_caches()
deleted
end end
@doc """ @doc """
@ -270,12 +290,33 @@ defmodule Microwaveprop.Propagation.ProfilesFile do
""" """
@spec list_valid_times() :: [DateTime.t()] @spec list_valid_times() :: [DateTime.t()]
def list_valid_times do def list_valid_times do
base_dir() # Hit on every point_detail click when the clicked valid_time
|> list_profile_files() # lacks its own profile and fallback scan searches for the
|> Enum.map(fn {_path, unix} -> unix end) # nearest past analysis. NFS dir-listing on every click was a
|> Enum.uniq() # visible chunk of click latency.
|> Enum.map(&DateTime.from_unix!/1) Microwaveprop.Cache.fetch_or_store({__MODULE__, :list_valid_times, base_dir()}, 5_000, fn ->
|> Enum.sort(DateTime) base_dir()
|> list_profile_files()
|> Enum.map(fn {_path, unix} -> unix end)
|> Enum.uniq()
|> Enum.map(&DateTime.from_unix!/1)
|> Enum.sort(DateTime)
end)
end
defp invalidate_caches_for(valid_time) do
dir = base_dir()
Microwaveprop.Cache.invalidate({__MODULE__, :read, dir, valid_time})
Microwaveprop.Cache.invalidate({__MODULE__, :list_valid_times, dir})
:ok
end
defp invalidate_all_caches do
# Match-delete every cached entry for this module without clobbering
# keys owned by other modules.
:ets.match_delete(:microwaveprop_cache, {{__MODULE__, :read, :_, :_}, :_, :_})
Microwaveprop.Cache.invalidate({__MODULE__, :list_valid_times, base_dir()})
:ok
end end
defp snap(lat, lon) do defp snap(lat, lon) do

View file

@ -38,6 +38,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do
@magic "NTMS" @magic "NTMS"
@version 1 @version 1
@no_data 255 @no_data 255
@header_size 33
@doc "Returns the root directory score files are written to." @doc "Returns the root directory score files are written to."
@spec base_dir() :: String.t() @spec base_dir() :: String.t()
@ -67,6 +68,7 @@ defmodule Microwaveprop.Propagation.ScoresFile do
tmp = path <> ".tmp." <> unique_suffix() tmp = path <> ".tmp." <> unique_suffix()
File.write!(tmp, binary, [:binary]) File.write!(tmp, binary, [:binary])
File.rename!(tmp, path) File.rename!(tmp, path)
invalidate_list_cache(band_mhz)
:ok :ok
end end
@ -98,16 +100,20 @@ defmodule Microwaveprop.Propagation.ScoresFile do
def prune_older_than(%DateTime{} = cutoff) do def prune_older_than(%DateTime{} = cutoff) do
cutoff_unix = DateTime.to_unix(cutoff) cutoff_unix = DateTime.to_unix(cutoff)
base_dir() deleted =
|> list_score_files() base_dir()
|> Enum.reduce(0, fn {path, valid_time_unix}, acc -> |> list_score_files()
if valid_time_unix < cutoff_unix do |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
_ = File.rm(path) if valid_time_unix < cutoff_unix do
acc + 1 _ = File.rm(path)
else acc + 1
acc else
end acc
end) end
end)
if deleted > 0, do: invalidate_all_list_caches()
deleted
end end
@doc """ @doc """
@ -125,25 +131,40 @@ defmodule Microwaveprop.Propagation.ScoresFile do
lo = DateTime.to_unix(run_time) lo = DateTime.to_unix(run_time)
hi = lo + max_forecast_hour * 3600 hi = lo + max_forecast_hour * 3600
base_dir() deleted =
|> list_score_files() base_dir()
|> Enum.reduce(0, fn {path, valid_time_unix}, acc -> |> list_score_files()
if valid_time_unix < lo or valid_time_unix > hi do |> Enum.reduce(0, fn {path, valid_time_unix}, acc ->
_ = File.rm(path) if valid_time_unix < lo or valid_time_unix > hi do
acc + 1 _ = File.rm(path)
else acc + 1
acc else
end acc
end) end
end)
if deleted > 0, do: invalidate_all_list_caches()
deleted
end end
@doc """ @doc """
List every `valid_time` that has a score file on disk for List every `valid_time` that has a score file on disk for
`band_mhz`, ordered ascending. Empty list if the band directory `band_mhz`, ordered ascending. Empty list if the band directory
doesn't exist yet. doesn't exist yet.
Cached for 5 seconds via `Microwaveprop.Cache`. The caller set is
tight (map click `point_forecast` + `available_valid_times`;
hourly chain completion triggers a PubSub fan-out that'd flip this
TTL well within its useful lifetime) so sub-10s staleness is fine.
""" """
@spec list_valid_times(non_neg_integer()) :: [DateTime.t()] @spec list_valid_times(non_neg_integer()) :: [DateTime.t()]
def list_valid_times(band_mhz) when is_integer(band_mhz) do def list_valid_times(band_mhz) when is_integer(band_mhz) do
Microwaveprop.Cache.fetch_or_store({__MODULE__, :list_valid_times, base_dir(), band_mhz}, 5_000, fn ->
list_valid_times_raw(band_mhz)
end)
end
defp list_valid_times_raw(band_mhz) do
dir = Path.join(base_dir(), Integer.to_string(band_mhz)) dir = Path.join(base_dir(), Integer.to_string(band_mhz))
case File.ls(dir) do case File.ls(dir) do
@ -206,12 +227,55 @@ defmodule Microwaveprop.Propagation.ScoresFile do
Fetch the score for a single grid cell from the file for Fetch the score for a single grid cell from the file for
`(band_mhz, valid_time)`. Returns `nil` if the file is missing or `(band_mhz, valid_time)`. Returns `nil` if the file is missing or
if the cell is outside the grid / carries the no-data sentinel. if the cell is outside the grid / carries the no-data sentinel.
Reads only the 33-byte header plus the one cell byte via a single
multi-range `:file.pread/2` call ~60 bytes off NFS instead of the
full ~93 KB file. Point-forecast walks 19 forecast hours per map
click, so the old full-read path dominated click latency whenever
the in-memory `ScoreCache` didn't already hold the cell.
""" """
@spec read_point(non_neg_integer(), DateTime.t(), float(), float()) :: non_neg_integer() | nil @spec read_point(non_neg_integer(), DateTime.t(), float(), float()) :: non_neg_integer() | nil
def read_point(band_mhz, %DateTime{} = valid_time, lat, lon) do def read_point(band_mhz, %DateTime{} = valid_time, lat, lon) do
case read(band_mhz, valid_time) do path = path_for(band_mhz, valid_time)
{:ok, payload} -> point_score(payload, lat, lon)
_ -> nil case :file.open(path, [:read, :binary, :raw]) do
{:ok, fd} ->
try do
read_point_from_fd(fd, lat, lon)
after
:file.close(fd)
end
{:error, _} ->
nil
end
end
defp read_point_from_fd(fd, lat, lon) do
# Peek the header to locate the cell — falls back to the full
# decode path only if the header is malformed (should never happen
# for files we wrote ourselves).
case :file.pread(fd, 0, @header_size) do
{:ok,
<<@magic, @version::unsigned-8, _band::unsigned-little-32, _vt::signed-little-64, lat_min::float-little-32,
lon_min::float-little-32, step::float-little-32, n_rows::unsigned-little-16, n_cols::unsigned-little-16>>} ->
read_cell(fd, lat, lon, lat_min, lon_min, step, n_rows, n_cols)
_ ->
nil
end
end
defp read_cell(fd, lat, lon, lat_min, lon_min, step, n_rows, n_cols) do
row = round((lat - lat_min) / step)
col = round((lon - lon_min) / step)
if row in 0..(n_rows - 1) and col in 0..(n_cols - 1) do
case :file.pread(fd, @header_size + row * n_cols + col, 1) do
{:ok, <<@no_data>>} -> nil
{:ok, <<score>>} -> score
_ -> nil
end
end end
end end
@ -361,6 +425,18 @@ defmodule Microwaveprop.Propagation.ScoresFile do
defp max_datetime([]), do: nil defp max_datetime([]), do: nil
defp max_datetime(times), do: Enum.max(times, DateTime) defp max_datetime(times), do: Enum.max(times, DateTime)
defp invalidate_list_cache(band_mhz) do
Microwaveprop.Cache.invalidate({__MODULE__, :list_valid_times, base_dir(), band_mhz})
end
defp invalidate_all_list_caches do
# Cheaper than enumerating bands — clear every cached list entry.
# `Microwaveprop.Cache.clear/0` is overkill (it wipes unrelated
# keys too) so match-delete just this module's keys.
:ets.match_delete(:microwaveprop_cache, {{__MODULE__, :list_valid_times, :_, :_}, :_, :_})
:ok
end
defp parse_valid_time_dt(filename) do defp parse_valid_time_dt(filename) do
with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename), with [_, iso] <- Regex.run(~r/^(.+)\.ntms$/, filename),
{:ok, dt, _} <- DateTime.from_iso8601(iso) do {:ok, dt, _} <- DateTime.from_iso8601(iso) do

View file

@ -60,6 +60,7 @@ defmodule MicrowavepropWeb.Endpoint do
@doc false @doc false
def log_level(%{path_info: ["health"]}), do: false def log_level(%{path_info: ["health"]}), do: false
def log_level(%{path_info: ["metrics" | _]}), do: false
def log_level(%{method: "HEAD"}), do: false def log_level(%{method: "HEAD"}), do: false
def log_level(_conn), do: :info def log_level(_conn), do: :info
end end