perf(nexrad): route fetch_decoded_frame through NexradCache
Telemetry puts nexrad_decode_png at 1.76 s/call × 37,847 calls/h — ~18.5 h of CPU across the cluster every real hour, the single biggest CPU consumer in the system. The map-click path (fetch_rain_cells) already cached decoded frames by 5-min bucket, but the per-contact CommonVolumeRadarWorker path (fetch_decoded_frame) went straight to network + decode on every call. Backfill means many contacts share a 5-min window, so the same 66 MB frame was being decoded dozens of times. Wire fetch_decoded_frame through NexradCache keyed on the rounded timestamp. Add a 20-entry size cap in NexradCache so backfill processing contacts in random timestamp order can't grow the ETS table to hundreds of GB. Each frame is ~66 MB, so 20 = ~1.3 GB worst case, well under typical pod memory. Expected impact: cuts sustained decode load by an order of magnitude depending on backfill temporal locality; map-click path is unchanged.
This commit is contained in:
parent
3ab71cf8fe
commit
bbf6481603
4 changed files with 129 additions and 18 deletions
|
|
@ -13,6 +13,13 @@ defmodule Microwaveprop.Weather.NexradCache do
|
||||||
|
|
||||||
@table :nexrad_frame_cache
|
@table :nexrad_frame_cache
|
||||||
|
|
||||||
|
# Each frame is ~66 MB (12,200 × 5,400 palette-index bytes). 20 frames
|
||||||
|
# caps at ~1.3 GB — comfortable headroom under a 2 GB pod limit, and
|
||||||
|
# enough to hit on the common backfill pattern where a worker chews
|
||||||
|
# through contacts in one timestamp window before moving on. Adjust
|
||||||
|
# if pod memory limits change or the n0q geometry changes.
|
||||||
|
@max_entries 20
|
||||||
|
|
||||||
@type pixels :: binary()
|
@type pixels :: binary()
|
||||||
@type width :: pos_integer()
|
@type width :: pos_integer()
|
||||||
|
|
||||||
|
|
@ -32,9 +39,26 @@ defmodule Microwaveprop.Weather.NexradCache do
|
||||||
@spec put(DateTime.t(), pixels(), width()) :: :ok
|
@spec put(DateTime.t(), pixels(), width()) :: :ok
|
||||||
def put(rounded_ts, pixels, width) do
|
def put(rounded_ts, pixels, width) do
|
||||||
:ets.insert(@table, {rounded_ts, pixels, width})
|
:ets.insert(@table, {rounded_ts, pixels, width})
|
||||||
|
enforce_size_cap()
|
||||||
:ok
|
:ok
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defp enforce_size_cap do
|
||||||
|
size = :ets.info(@table, :size)
|
||||||
|
|
||||||
|
if size > @max_entries do
|
||||||
|
# Drop the oldest-by-key entries until we're back under the cap.
|
||||||
|
# O(n log n) but n is ~20, so this runs in microseconds.
|
||||||
|
excess = size - @max_entries
|
||||||
|
|
||||||
|
@table
|
||||||
|
|> :ets.tab2list()
|
||||||
|
|> Enum.sort_by(fn {ts, _, _} -> DateTime.to_unix(ts) end)
|
||||||
|
|> Enum.take(excess)
|
||||||
|
|> Enum.each(fn {ts, _, _} -> :ets.delete(@table, ts) end)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
@spec prune_older_than(DateTime.t()) :: non_neg_integer()
|
||||||
def prune_older_than(cutoff_ts) do
|
def prune_older_than(cutoff_ts) do
|
||||||
match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}]
|
match_spec = [{{:"$1", :_, :_}, [{:<, :"$1", {:const, cutoff_ts}}], [true]}]
|
||||||
|
|
|
||||||
|
|
@ -76,37 +76,39 @@ defmodule Microwaveprop.Weather.NexradClient do
|
||||||
"""
|
"""
|
||||||
@spec fetch_rain_cells(float(), float(), float(), float()) :: {:ok, [{float(), float(), float()}]} | {:error, term()}
|
@spec fetch_rain_cells(float(), float(), float(), float()) :: {:ok, [{float(), float(), float()}]} | {:error, term()}
|
||||||
def fetch_rain_cells(lat, lon, radius_km \\ 300.0, min_dbz \\ 25.0) do
|
def fetch_rain_cells(lat, lon, radius_km \\ 300.0, min_dbz \\ 25.0) do
|
||||||
now = DateTime.utc_now()
|
case fetch_decoded_frame(DateTime.utc_now()) do
|
||||||
rounded = round_to_5min(now)
|
|
||||||
|
|
||||||
case NexradCache.fetch(rounded) do
|
|
||||||
{:ok, pixels, width} ->
|
{:ok, pixels, width} ->
|
||||||
{:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)}
|
{:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)}
|
||||||
|
|
||||||
:miss ->
|
error ->
|
||||||
case fetch_and_decode_frame(rounded) do
|
error
|
||||||
{:ok, pixels, width} ->
|
|
||||||
NexradCache.put(rounded, pixels, width)
|
|
||||||
NexradCache.prune_older_than(DateTime.add(rounded, -600, :second))
|
|
||||||
{:ok, extract_rain_cells(pixels, width, lat, lon, radius_km, min_dbz)}
|
|
||||||
|
|
||||||
error ->
|
|
||||||
error
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Fetch and decode the n0q PNG frame nearest the given timestamp.
|
Fetch and decode the n0q PNG frame nearest the given timestamp.
|
||||||
Returns `{:ok, pixels, width}` — a flat binary of palette-index bytes —
|
Returns `{:ok, pixels, width}` — a flat binary of palette-index bytes —
|
||||||
or `{:error, reason}`. Does NOT go through `NexradCache`; callers that
|
or `{:error, reason}`. Goes through `NexradCache`, keyed on the 5-min
|
||||||
want caching (e.g. real-time point queries) should call `fetch_rain_cells/4`.
|
rounded timestamp — concurrent radar workers (and any backfill
|
||||||
|
neighbors that share a 5-min window) reuse the same decoded buffer
|
||||||
|
instead of each paying the full ~5 MB download + ~1.7 s decode.
|
||||||
"""
|
"""
|
||||||
@spec fetch_decoded_frame(DateTime.t()) ::
|
@spec fetch_decoded_frame(DateTime.t()) ::
|
||||||
{:ok, binary(), non_neg_integer()} | {:error, term()}
|
{:ok, binary(), non_neg_integer()} | {:error, term()}
|
||||||
def fetch_decoded_frame(%DateTime{} = dt) do
|
def fetch_decoded_frame(%DateTime{} = dt) do
|
||||||
rounded = round_to_5min(dt)
|
rounded = round_to_5min(dt)
|
||||||
fetch_and_decode_frame(rounded)
|
|
||||||
|
case NexradCache.fetch(rounded) do
|
||||||
|
{:ok, pixels, width} ->
|
||||||
|
{:ok, pixels, width}
|
||||||
|
|
||||||
|
:miss ->
|
||||||
|
with {:ok, pixels, width} <- fetch_and_decode_frame(rounded) do
|
||||||
|
NexradCache.put(rounded, pixels, width)
|
||||||
|
NexradCache.prune_older_than(DateTime.add(rounded, -600, :second))
|
||||||
|
{:ok, pixels, width}
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp fetch_and_decode_frame(rounded) do
|
defp fetch_and_decode_frame(rounded) do
|
||||||
|
|
|
||||||
|
|
@ -44,4 +44,28 @@ defmodule Microwaveprop.Weather.NexradCacheTest do
|
||||||
assert {:ok, <<1>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
assert {:ok, <<1>>, 1} = NexradCache.fetch(~U[2026-04-12 12:00:00Z])
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "size cap" do
|
||||||
|
test "evicts the oldest-by-timestamp entries once the cap is exceeded" do
|
||||||
|
# Insert 25 entries; max is 20, so 5 oldest get dropped.
|
||||||
|
base = ~U[2026-04-12 00:00:00Z]
|
||||||
|
|
||||||
|
for i <- 1..25 do
|
||||||
|
ts = DateTime.add(base, i * 300, :second)
|
||||||
|
NexradCache.put(ts, <<i>>, 1)
|
||||||
|
end
|
||||||
|
|
||||||
|
# The 5 oldest (i = 1..5) should have been evicted.
|
||||||
|
for i <- 1..5 do
|
||||||
|
ts = DateTime.add(base, i * 300, :second)
|
||||||
|
assert NexradCache.fetch(ts) == :miss
|
||||||
|
end
|
||||||
|
|
||||||
|
# The most recent 20 should still be present.
|
||||||
|
for i <- 6..25 do
|
||||||
|
ts = DateTime.add(base, i * 300, :second)
|
||||||
|
assert {:ok, <<^i>>, 1} = NexradCache.fetch(ts)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
defmodule Microwaveprop.Weather.NexradClientTest do
|
defmodule Microwaveprop.Weather.NexradClientTest do
|
||||||
use ExUnit.Case, async: true
|
use ExUnit.Case, async: false
|
||||||
|
|
||||||
|
alias Microwaveprop.Weather.NexradCache
|
||||||
alias Microwaveprop.Weather.NexradClient
|
alias Microwaveprop.Weather.NexradClient
|
||||||
|
|
||||||
describe "round_to_5min/1" do
|
describe "round_to_5min/1" do
|
||||||
|
|
@ -139,6 +140,66 @@ defmodule Microwaveprop.Weather.NexradClientTest do
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "fetch_decoded_frame/1 caching" do
|
||||||
|
# Minimal 20×20 indexed-color PNG identical to the NexradWorkerTest
|
||||||
|
# fixture. Used here just to give the decoder something valid to
|
||||||
|
# chew on; caching semantics don't care about the contents.
|
||||||
|
defp minimal_png(width \\ 20, height \\ 20) do
|
||||||
|
signature = <<137, 80, 78, 71, 13, 10, 26, 10>>
|
||||||
|
ihdr_data = <<width::32, height::32, 8, 3, 0, 0, 0>>
|
||||||
|
ihdr_chunk = <<byte_size(ihdr_data)::32, "IHDR", ihdr_data::binary, 0::32>>
|
||||||
|
|
||||||
|
raw = for _ <- 1..height, into: <<>>, do: <<0, 0::size(width)-unit(8)>>
|
||||||
|
compressed = :zlib.compress(raw)
|
||||||
|
idat_chunk = <<byte_size(compressed)::32, "IDAT", compressed::binary, 0::32>>
|
||||||
|
iend_chunk = <<0::32, "IEND", 0::32>>
|
||||||
|
|
||||||
|
signature <> ihdr_chunk <> idat_chunk <> iend_chunk
|
||||||
|
end
|
||||||
|
|
||||||
|
setup do
|
||||||
|
NexradCache.clear()
|
||||||
|
on_exit(fn -> NexradCache.clear() end)
|
||||||
|
:ok
|
||||||
|
end
|
||||||
|
|
||||||
|
test "first call fetches+decodes; second call for same 5-min bucket is served from cache" do
|
||||||
|
test_pid = self()
|
||||||
|
|
||||||
|
Req.Test.stub(NexradClient, fn conn ->
|
||||||
|
send(test_pid, :http_fetched)
|
||||||
|
Plug.Conn.send_resp(conn, 200, minimal_png())
|
||||||
|
end)
|
||||||
|
|
||||||
|
ts = ~U[2022-08-20 14:07:00Z]
|
||||||
|
|
||||||
|
{:ok, pixels1, width1} = NexradClient.fetch_decoded_frame(ts)
|
||||||
|
{:ok, pixels2, width2} = NexradClient.fetch_decoded_frame(ts)
|
||||||
|
|
||||||
|
# The cache is keyed on the 5-min rounded timestamp, so a
|
||||||
|
# second call within the same bucket reuses the decoded pixels.
|
||||||
|
assert pixels1 == pixels2
|
||||||
|
assert width1 == width2
|
||||||
|
assert_received :http_fetched
|
||||||
|
refute_received :http_fetched
|
||||||
|
end
|
||||||
|
|
||||||
|
test "different 5-min buckets hit the network separately" do
|
||||||
|
test_pid = self()
|
||||||
|
|
||||||
|
Req.Test.stub(NexradClient, fn conn ->
|
||||||
|
send(test_pid, :http_fetched)
|
||||||
|
Plug.Conn.send_resp(conn, 200, minimal_png())
|
||||||
|
end)
|
||||||
|
|
||||||
|
{:ok, _, _} = NexradClient.fetch_decoded_frame(~U[2022-08-20 14:05:00Z])
|
||||||
|
{:ok, _, _} = NexradClient.fetch_decoded_frame(~U[2022-08-20 14:10:00Z])
|
||||||
|
|
||||||
|
assert_received :http_fetched
|
||||||
|
assert_received :http_fetched
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
describe "pixel_to_dbz/1" do
|
describe "pixel_to_dbz/1" do
|
||||||
test "value 0 maps to 0 (no echo)" do
|
test "value 0 maps to 0 (no echo)" do
|
||||||
assert NexradClient.pixel_to_dbz(0) == 0.0
|
assert NexradClient.pixel_to_dbz(0) == 0.0
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue