feat(prop): Phase 2 cutover — Rust owns f01–f18
Rust `prop-grid-rs` has been validated end-to-end on talos5 on the streamed-bands image (main-1776635096-6d91461): real chain steps complete in ~7 s each and the peak-heap refactor landed. Changes: - `PropagationGridWorker.seed_chain` now enqueues a single f00 Oban job instead of f00–f18. The f00 analysis-hour keeps its enrichment (native-level duct, NEXRAD composite, commercial-link boost, ProfilesFile write) and the GridCache broadcast. - `GridTaskEnqueuer.seed/1` is called again so grid_tasks gets the 18 forecast-hour rows the Rust worker claims. - Test asserts only one Oban job is enqueued and the matching 18 grid_tasks rows land in the DB. - `deployment-grid-rs.yaml` flips `PROP_SCORES_DIR` from `/data/scores_shadow` to `/data/scores` so Rust writes to the live score tree. No file collision — Elixir only writes the f00 files. The hot pod memory shrink (6 Gi → 1 Gi) is deferred until one full hourly cycle runs clean under the new split.
This commit is contained in:
parent
54bc46ab60
commit
65693ed415
3 changed files with 35 additions and 30 deletions
|
|
@ -3,9 +3,11 @@ kind: Deployment
|
|||
metadata:
|
||||
name: prop-grid-rs
|
||||
namespace: prop
|
||||
# Phase 1 (shadow) default: write to /data/scores_shadow so Elixir keeps
|
||||
# owning the live /data/scores tree. Flip PROP_SCORES_DIR to /data/scores
|
||||
# at cutover after ShadowComparator shows < 0.5 mean delta for 72h.
|
||||
# Phase 2 cutover: Rust writes directly to /data/scores. Elixir only
|
||||
# runs the f00 analysis-hour step (which carries enrichment Rust
|
||||
# doesn't own yet — native-level duct merge, NEXRAD composite,
|
||||
# commercial-link degradation, ProfilesFile write). f01..f18 land
|
||||
# here from the Rust worker.
|
||||
spec:
|
||||
replicas: 1
|
||||
minReadySeconds: 5
|
||||
|
|
@ -56,7 +58,7 @@ spec:
|
|||
- name: HRRR_BASE_URL
|
||||
value: "http://skippy.w5isp.com:8080"
|
||||
- name: PROP_SCORES_DIR
|
||||
value: "/data/scores_shadow"
|
||||
value: "/data/scores"
|
||||
- name: RUST_LOG
|
||||
value: "info"
|
||||
- name: PROP_GRID_RS_PG_CONNS
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|||
alias Microwaveprop.Propagation
|
||||
alias Microwaveprop.Propagation.BandConfig
|
||||
alias Microwaveprop.Propagation.Grid
|
||||
alias Microwaveprop.Propagation.GridTaskEnqueuer
|
||||
alias Microwaveprop.Propagation.ProfilesFile
|
||||
alias Microwaveprop.Propagation.ScoreCache
|
||||
alias Microwaveprop.Weather
|
||||
|
|
@ -82,25 +83,17 @@ defmodule Microwaveprop.Workers.PropagationGridWorker do
|
|||
two_hours_ago = DateTime.add(DateTime.utc_now(), -2, :hour)
|
||||
run_time = two_hours_ago |> HrrrClient.nearest_hrrr_hour() |> DateTime.truncate(:second)
|
||||
|
||||
Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00-f#{@max_forecast_hour} (parallel)")
|
||||
Logger.info("PropagationGrid: seeding chain run_time=#{run_time}, f00 (+f01-f#{@max_forecast_hour} via grid_tasks)")
|
||||
|
||||
# Fan out all 19 forecast hours at once. Each step is independent
|
||||
# (different HRRR URL, different valid_time file output) so there's
|
||||
# no dependency that requires the old sequential chain pattern.
|
||||
# With 2 slots/pod × 3 pods = 6 concurrent workers the chain wall
|
||||
# time drops from ~48 min to ~10 min. Cleanup (retain_window,
|
||||
# prune) runs on PropagationPruneWorker's own 15-min cron.
|
||||
jobs =
|
||||
for fh <- 0..@max_forecast_hour do
|
||||
new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => fh})
|
||||
end
|
||||
# Post-cutover: Elixir only runs the f00 analysis-hour step. f00 carries
|
||||
# the expensive enrichment that Rust doesn't cover yet (native-level
|
||||
# duct merge, NEXRAD composite, commercial-link degradation) and writes
|
||||
# the ProfilesFile that /weather reads. f01..f18 go through the
|
||||
# `grid_tasks` handoff queue to the Rust `prop-grid-rs` worker, which
|
||||
# fetches + decodes + scores each forecast hour independently.
|
||||
Oban.insert_all([new(%{"run_time" => DateTime.to_iso8601(run_time), "forecast_hour" => 0})])
|
||||
|
||||
Oban.insert_all(jobs)
|
||||
|
||||
# `GridTaskEnqueuer.seed/1` feeds the Rust `prop-grid-rs` worker via
|
||||
# grid_tasks. Disabled until the Rust image is validated end-to-end
|
||||
# on talos5 — once cutover is approved, the Elixir fan-out above
|
||||
# shrinks to just fh=0 and this line comes back.
|
||||
_ = GridTaskEnqueuer.seed(run_time)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
|
|
|||
|
|
@ -45,24 +45,34 @@ defmodule Microwaveprop.Workers.PropagationGridWorkerTest do
|
|||
end
|
||||
|
||||
describe "perform/1 — chain seeding (empty args)" do
|
||||
test "enqueues all 19 forecast-hour jobs at once, all pointing at the same run_time" do
|
||||
test "enqueues only f00 in Oban; f01..f18 go to grid_tasks for Rust" do
|
||||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||||
import Ecto.Query
|
||||
|
||||
assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}})
|
||||
|
||||
jobs = all_enqueued(worker: PropagationGridWorker)
|
||||
assert length(jobs) == 19
|
||||
assert length(jobs) == 1
|
||||
|
||||
fhs = jobs |> Enum.map(& &1.args["forecast_hour"]) |> Enum.sort()
|
||||
assert fhs == Enum.to_list(0..18)
|
||||
[job] = jobs
|
||||
assert job.args["forecast_hour"] == 0
|
||||
|
||||
# Every child job shares the same run_time and it's normalized
|
||||
# to top-of-hour UTC.
|
||||
run_times = jobs |> Enum.map(& &1.args["run_time"]) |> Enum.uniq()
|
||||
assert [rt] = run_times
|
||||
{:ok, dt, _} = DateTime.from_iso8601(rt)
|
||||
# run_time is normalized to top-of-hour UTC.
|
||||
{:ok, dt, _} = DateTime.from_iso8601(job.args["run_time"])
|
||||
assert dt.minute == 0
|
||||
assert dt.second == 0
|
||||
assert DateTime.before?(dt, DateTime.utc_now())
|
||||
|
||||
# The Rust worker picks up f01..f18 from the grid_tasks table.
|
||||
task_fhs =
|
||||
Microwaveprop.Repo.all(
|
||||
from t in "grid_tasks",
|
||||
where: t.run_time == ^DateTime.truncate(dt, :second),
|
||||
select: t.forecast_hour,
|
||||
order_by: t.forecast_hour
|
||||
)
|
||||
|
||||
assert task_fhs == Enum.to_list(1..18)
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue