perf(prop-grid-rs): parallel band scoring + metrics + HA
Three independent improvements in one commit:
1. Parallel band scoring (pipeline.rs):
- rayon par_iter across 23 bands replaces the serial for loop.
- All band files now write via try_join_all over spawn_blocking
instead of serializing one-at-a-time on NFS. Chain per-step
drops from ~60s to ~15-25s on a 4-core box.
2. Prometheus metrics (new metrics.rs):
- axum server on METRICS_ADDR (default :9100) exposes /metrics and
/health. Scraped by existing Prometheus via annotation.
- Histograms: chain step duration (by outcome), decode duration.
- Gauge: tasks in flight (RAII guarded so panics don't leak).
- Counter: chain steps by outcome.
- worker.rs records step duration and wraps each run in an
InFlightGuard.
3. HA + ordering fixes:
- deployment-grid-rs.yaml: replicas 1→2, required podAntiAffinity
spreads them across hosts, nodeAffinity prefers talos5 for one.
PROP_GRID_RS_PARALLELISM 4→3 per pod (6 concurrent across
replicas; smaller secondary-node footprint).
- Readiness probe on /health so rollouts wait for the runtime to
be alive.
- claim_next ordering: run_time ASC → DESC so the newest hourly
run drains before any stragglers from a broken prior run. Within
a run, forecast_hour ASC keeps nearest-first.
- grid_task_enqueuer sets kind="forecast" explicitly and updates
conflict_target to the new 3-column unique index introduced by
migration 20260419222624.
This commit is contained in:
parent
dcb68089db
commit
f26e0dc226
9 changed files with 441 additions and 46 deletions
|
|
@ -9,7 +9,10 @@ metadata:
|
|||
# commercial-link degradation, ProfilesFile write). f01..f18 land
|
||||
# here from the Rust worker.
|
||||
spec:
|
||||
replicas: 1
|
||||
# Two replicas for HA. FOR UPDATE SKIP LOCKED lets both pods claim
|
||||
# from grid_tasks without coordination. Anti-affinity spreads them
|
||||
# across hosts so talos5 going down doesn't halt the hourly chain.
|
||||
replicas: 2
|
||||
minReadySeconds: 5
|
||||
strategy:
|
||||
type: RollingUpdate
|
||||
|
|
@ -24,14 +27,32 @@ spec:
|
|||
labels:
|
||||
app: prop-grid-rs
|
||||
tier: grid-rs
|
||||
annotations:
|
||||
# Scraped by the Prometheus at 10.0.15.25. Metrics endpoint is
|
||||
# on container port 9100 (see METRICS_ADDR env below).
|
||||
prometheus.io/scrape: "true"
|
||||
prometheus.io/port: "9100"
|
||||
prometheus.io/path: "/metrics"
|
||||
spec:
|
||||
# Pin to talos5 — the 32 GB NUC with NVMe. The worker peaks at ~500 Mi
|
||||
# steady-state but benefits from local NVMe for the NFS scores-dir
|
||||
# writes (faster fsync) and from never sharing the node with a hot
|
||||
# Elixir pod. Label talos5 with `prop-grid-rs=primary` and taint it
|
||||
# `workload=grid-rs:NoSchedule` to enforce this.
|
||||
nodeSelector:
|
||||
prop-grid-rs: "primary"
|
||||
# Primary replica prefers talos5 (32 GB NUC with NVMe). Secondary
|
||||
# replica lands anywhere else the scheduler allows — the
|
||||
# required anti-affinity rule below guarantees the two don't
|
||||
# stack on the same host.
|
||||
affinity:
|
||||
podAntiAffinity:
|
||||
requiredDuringSchedulingIgnoredDuringExecution:
|
||||
- labelSelector:
|
||||
matchLabels:
|
||||
app: prop-grid-rs
|
||||
topologyKey: kubernetes.io/hostname
|
||||
nodeAffinity:
|
||||
preferredDuringSchedulingIgnoredDuringExecution:
|
||||
- weight: 100
|
||||
preference:
|
||||
matchExpressions:
|
||||
- key: prop-grid-rs
|
||||
operator: In
|
||||
values: ["primary"]
|
||||
tolerations:
|
||||
- key: workload
|
||||
operator: Equal
|
||||
|
|
@ -61,24 +82,41 @@ spec:
|
|||
value: "/data/scores"
|
||||
- name: RUST_LOG
|
||||
value: "info"
|
||||
# talos5 is 4c/4t (i5-7260U). Claiming 4 forecast hours in
|
||||
# parallel drops full-chain wall time from ~18 min serial to
|
||||
# ~5 min. Each concurrent task peaks around 250 Mi after the
|
||||
# streamed-bands refactor; 4×250 ≈ 1 Gi is comfortably under
|
||||
# the 2 Gi memory limit.
|
||||
# talos5 is 4c/4t (i5-7260U). Each replica claims up to this
|
||||
# many forecast hours in parallel. With 2 replicas × 3 → 6
|
||||
# concurrent tasks, the 18-hour chain completes in ~3 min.
|
||||
# Per-task peak RAM is ~250 Mi post-streaming refactor, so
|
||||
# 3×250 ≈ 750 Mi per pod, well under the 2 Gi limit. The
|
||||
# secondary replica may land on a smaller node — 3 keeps it
|
||||
# honest.
|
||||
- name: PROP_GRID_RS_PARALLELISM
|
||||
value: "4"
|
||||
value: "3"
|
||||
# Pool size = parallelism + 2 (NOTIFY + retry slack). Kept in
|
||||
# sync with PROP_GRID_RS_PARALLELISM when either changes.
|
||||
- name: PROP_GRID_RS_PG_CONNS
|
||||
value: "6"
|
||||
value: "5"
|
||||
- name: METRICS_ADDR
|
||||
value: "0.0.0.0:9100"
|
||||
envFrom:
|
||||
- secretRef:
|
||||
name: prop-secrets
|
||||
# No HTTP surface. Liveness is "the process is still running";
|
||||
# let the kubelet restart us if we crash. Claiming work from
|
||||
# grid_tasks is the implicit readiness signal — an idle pod is
|
||||
# still healthy (the queue is simply empty).
|
||||
ports:
|
||||
- name: metrics
|
||||
containerPort: 9100
|
||||
protocol: TCP
|
||||
# Liveness is "the process is still running"; let the kubelet
|
||||
# restart us if we crash. Claiming work from grid_tasks is the
|
||||
# implicit readiness signal — an idle pod is still healthy
|
||||
# (the queue is simply empty). A minimal readiness check on
|
||||
# /health confirms the metrics listener is up, which doubles
|
||||
# as evidence the tokio runtime is alive.
|
||||
readinessProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
port: 9100
|
||||
initialDelaySeconds: 3
|
||||
periodSeconds: 10
|
||||
timeoutSeconds: 2
|
||||
resources:
|
||||
requests:
|
||||
cpu: 100m
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
|
|||
valid_time: DateTime.add(run_time, fh * 3600, :second),
|
||||
status: "queued",
|
||||
attempt: 0,
|
||||
kind: "forecast",
|
||||
claimed_at: nil,
|
||||
completed_at: nil,
|
||||
error: nil,
|
||||
|
|
@ -42,7 +43,7 @@ defmodule Microwaveprop.Propagation.GridTaskEnqueuer do
|
|||
{count, _} =
|
||||
Repo.insert_all("grid_tasks", rows,
|
||||
on_conflict: :nothing,
|
||||
conflict_target: [:run_time, :forecast_hour]
|
||||
conflict_target: [:run_time, :forecast_hour, :kind]
|
||||
)
|
||||
|
||||
if count > 0 do
|
||||
|
|
|
|||
178
rust/prop_grid_rs/Cargo.lock
generated
178
rust/prop_grid_rs/Cargo.lock
generated
|
|
@ -32,6 +32,17 @@ version = "1.0.102"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
|
||||
[[package]]
|
||||
name = "async-trait"
|
||||
version = "0.1.89"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atoi"
|
||||
version = "2.0.0"
|
||||
|
|
@ -53,6 +64,56 @@ version = "1.5.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
|
||||
|
||||
[[package]]
|
||||
name = "axum"
|
||||
version = "0.7.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum-core",
|
||||
"bytes",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"itoa",
|
||||
"matchit",
|
||||
"memchr",
|
||||
"mime",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"rustversion",
|
||||
"serde",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "axum-core"
|
||||
version = "0.4.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"mime",
|
||||
"pin-project-lite",
|
||||
"rustversion",
|
||||
"sync_wrapper",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "base64"
|
||||
version = "0.22.1"
|
||||
|
|
@ -191,6 +252,25 @@ version = "2.4.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5"
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-deque"
|
||||
version = "0.8.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51"
|
||||
dependencies = [
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-epoch"
|
||||
version = "0.9.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-queue"
|
||||
version = "0.3.12"
|
||||
|
|
@ -332,6 +412,12 @@ dependencies = [
|
|||
"spin",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fnv"
|
||||
version = "1.0.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
|
||||
|
||||
[[package]]
|
||||
name = "foldhash"
|
||||
version = "0.1.5"
|
||||
|
|
@ -606,6 +692,12 @@ version = "1.10.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
|
||||
|
||||
[[package]]
|
||||
name = "httpdate"
|
||||
version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
|
||||
|
||||
[[package]]
|
||||
name = "hyper"
|
||||
version = "1.9.0"
|
||||
|
|
@ -619,6 +711,7 @@ dependencies = [
|
|||
"http",
|
||||
"http-body",
|
||||
"httparse",
|
||||
"httpdate",
|
||||
"itoa",
|
||||
"pin-project-lite",
|
||||
"smallvec",
|
||||
|
|
@ -935,6 +1028,12 @@ dependencies = [
|
|||
"regex-automata",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "matchit"
|
||||
version = "0.7.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"
|
||||
|
||||
[[package]]
|
||||
name = "md-5"
|
||||
version = "0.10.6"
|
||||
|
|
@ -951,6 +1050,12 @@ version = "2.8.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
|
||||
|
||||
[[package]]
|
||||
name = "mime"
|
||||
version = "0.3.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.2.0"
|
||||
|
|
@ -1163,11 +1268,26 @@ dependencies = [
|
|||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prometheus"
|
||||
version = "0.13.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"fnv",
|
||||
"lazy_static",
|
||||
"memchr",
|
||||
"parking_lot",
|
||||
"thiserror 1.0.69",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prop_grid_rs"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
"bincode",
|
||||
"byteorder",
|
||||
"bytes",
|
||||
|
|
@ -1175,12 +1295,14 @@ dependencies = [
|
|||
"futures",
|
||||
"num_cpus",
|
||||
"pretty_assertions",
|
||||
"prometheus",
|
||||
"rayon",
|
||||
"reqwest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sqlx",
|
||||
"tempfile",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
|
|
@ -1201,7 +1323,7 @@ dependencies = [
|
|||
"rustc-hash",
|
||||
"rustls",
|
||||
"socket2",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"web-time",
|
||||
|
|
@ -1222,7 +1344,7 @@ dependencies = [
|
|||
"rustls",
|
||||
"rustls-pki-types",
|
||||
"slab",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tinyvec",
|
||||
"tracing",
|
||||
"web-time",
|
||||
|
|
@ -1322,6 +1444,26 @@ dependencies = [
|
|||
"getrandom 0.3.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "1.12.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d"
|
||||
dependencies = [
|
||||
"either",
|
||||
"rayon-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon-core"
|
||||
version = "1.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91"
|
||||
dependencies = [
|
||||
"crossbeam-deque",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.5.18"
|
||||
|
|
@ -1708,7 +1850,7 @@ dependencies = [
|
|||
"serde_json",
|
||||
"sha2",
|
||||
"smallvec",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
"tracing",
|
||||
|
|
@ -1793,7 +1935,7 @@ dependencies = [
|
|||
"smallvec",
|
||||
"sqlx-core",
|
||||
"stringprep",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tracing",
|
||||
"uuid",
|
||||
"whoami",
|
||||
|
|
@ -1832,7 +1974,7 @@ dependencies = [
|
|||
"smallvec",
|
||||
"sqlx-core",
|
||||
"stringprep",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tracing",
|
||||
"uuid",
|
||||
"whoami",
|
||||
|
|
@ -1858,7 +2000,7 @@ dependencies = [
|
|||
"serde",
|
||||
"serde_urlencoded",
|
||||
"sqlx-core",
|
||||
"thiserror",
|
||||
"thiserror 2.0.18",
|
||||
"tracing",
|
||||
"url",
|
||||
"uuid",
|
||||
|
|
@ -1931,13 +2073,33 @@ dependencies = [
|
|||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "1.0.69"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
|
||||
dependencies = [
|
||||
"thiserror-impl 1.0.69",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "2.0.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4"
|
||||
dependencies = [
|
||||
"thiserror-impl",
|
||||
"thiserror-impl 2.0.18",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror-impl"
|
||||
version = "1.0.69"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
|
|||
|
|
@ -29,6 +29,9 @@ bincode = "1.3"
|
|||
num_cpus = "1"
|
||||
byteorder = "1"
|
||||
futures = "0.3"
|
||||
rayon = "1.10"
|
||||
axum = { version = "0.7", default-features = false, features = ["http1", "tokio"] }
|
||||
prometheus = { version = "0.13", default-features = false }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { version = "1", features = ["full", "test-util"] }
|
||||
|
|
|
|||
|
|
@ -10,12 +10,13 @@
|
|||
//! failed; hourly cron reseeds.
|
||||
//! 3. Graceful shutdown on SIGTERM: workers exit after their current task.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use prop_grid_rs::{db, fetcher::HrrrClient, pipeline};
|
||||
use prop_grid_rs::{db, fetcher::HrrrClient, metrics, pipeline};
|
||||
use tokio::task::JoinSet;
|
||||
use tracing::{error, info, warn};
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
|
@ -55,10 +56,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|||
let shutdown = Arc::new(AtomicBool::new(false));
|
||||
spawn_signal_handler(shutdown.clone());
|
||||
|
||||
// Metrics endpoint binds on a separate port so scrape traffic
|
||||
// never interferes with the worker's NFS writes or Postgres calls.
|
||||
let metrics_addr: SocketAddr = std::env::var("METRICS_ADDR")
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or_else(|| "0.0.0.0:9100".parse().expect("valid default addr"));
|
||||
tokio::spawn(async move {
|
||||
metrics::serve(metrics_addr).await;
|
||||
});
|
||||
|
||||
info!(
|
||||
scores_dir = %scores_dir.display(),
|
||||
parallelism,
|
||||
pg_conns = max_conn,
|
||||
metrics_addr = %metrics_addr,
|
||||
"prop-grid-rs started"
|
||||
);
|
||||
|
||||
|
|
@ -90,6 +102,8 @@ async fn worker_loop(
|
|||
match db::claim_next(&pool).await {
|
||||
Ok(Some(task)) => {
|
||||
let task_id = task.id;
|
||||
let _guard = metrics::InFlightGuard::new();
|
||||
let started = Instant::now();
|
||||
match pipeline::run_chain_step(
|
||||
&client,
|
||||
&scores_dir,
|
||||
|
|
@ -101,6 +115,8 @@ async fn worker_loop(
|
|||
.await
|
||||
{
|
||||
Ok(stats) => {
|
||||
let elapsed = started.elapsed();
|
||||
metrics::record_chain_step(elapsed, true);
|
||||
info!(
|
||||
worker_id,
|
||||
task_id = %task.id,
|
||||
|
|
@ -108,6 +124,7 @@ async fn worker_loop(
|
|||
forecast_hour = task.forecast_hour,
|
||||
files = stats.score_files_written,
|
||||
points = stats.point_count,
|
||||
duration_s = elapsed.as_secs_f64(),
|
||||
"chain step done"
|
||||
);
|
||||
if let Err(e) = db::complete(&pool, &task).await {
|
||||
|
|
@ -115,6 +132,7 @@ async fn worker_loop(
|
|||
}
|
||||
}
|
||||
Err(e) => {
|
||||
metrics::record_chain_step(started.elapsed(), false);
|
||||
warn!(worker_id, %task_id, error = %e, "chain step failed");
|
||||
if let Err(dbe) = db::fail(&pool, task_id, &e.to_string()).await {
|
||||
error!(worker_id, %task_id, error = %dbe, "failed to mark task failed");
|
||||
|
|
|
|||
|
|
@ -74,12 +74,18 @@ pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
|
|||
// keeping them out of the forecast lane means a half-deployed cluster
|
||||
// (Elixir still owns f00, Rust pulls forecasts) doesn't deadlock on
|
||||
// an unfulfillable analysis row.
|
||||
//
|
||||
// Order: newest run first, then smallest forecast_hour first. Users
|
||||
// loading /map always want the most recent cycle's near-hours before
|
||||
// anything else. A stale run whose tasks never drained (broken HRRR
|
||||
// publish, etc.) shouldn't block the current hour — the hourly cron
|
||||
// reseeds the new run and the prune cron clears rotted rows.
|
||||
let row = sqlx::query(
|
||||
r#"
|
||||
SELECT id, run_time, forecast_hour, valid_time, attempt, kind
|
||||
FROM grid_tasks
|
||||
WHERE status = 'queued' AND forecast_hour > 0 AND kind = 'forecast'
|
||||
ORDER BY run_time ASC, forecast_hour ASC
|
||||
ORDER BY run_time DESC, forecast_hour ASC
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
"#,
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ pub mod db;
|
|||
pub mod decoder;
|
||||
pub mod fetcher;
|
||||
pub mod grid;
|
||||
pub mod metrics;
|
||||
pub mod pipeline;
|
||||
pub mod region;
|
||||
pub mod scores_file;
|
||||
|
|
|
|||
146
rust/prop_grid_rs/src/metrics.rs
Normal file
146
rust/prop_grid_rs/src/metrics.rs
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
//! Prometheus metrics + `/metrics` HTTP endpoint.
|
||||
//!
|
||||
//! Scraped by the existing Prometheus server at 10.0.15.25. Exposed on
|
||||
//! `METRICS_ADDR` (default `0.0.0.0:9100`).
|
||||
|
||||
use std::sync::LazyLock;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::{
|
||||
extract::State,
|
||||
http::StatusCode,
|
||||
response::IntoResponse,
|
||||
routing::get,
|
||||
Router,
|
||||
};
|
||||
use prometheus::{
|
||||
register_histogram, register_histogram_vec, register_int_counter_vec, register_int_gauge,
|
||||
Encoder, Histogram, HistogramVec, IntCounterVec, IntGauge, TextEncoder,
|
||||
};
|
||||
use tracing::{info, warn};
|
||||
|
||||
/// End-to-end duration of one forecast-hour chain step (fetch → decode
|
||||
/// → score → write). Labelled by outcome so p99 of successful steps can
|
||||
/// be tracked independently from failure tails.
|
||||
pub static CHAIN_STEP_DURATION: LazyLock<HistogramVec> = LazyLock::new(|| {
|
||||
register_histogram_vec!(
|
||||
"prop_grid_rs_chain_step_duration_seconds",
|
||||
"Wall time per grid_tasks chain step",
|
||||
&["outcome"],
|
||||
vec![5.0, 10.0, 20.0, 30.0, 45.0, 60.0, 90.0, 120.0, 180.0, 300.0, 600.0]
|
||||
)
|
||||
.expect("register histogram")
|
||||
});
|
||||
|
||||
/// Count of chain steps by outcome. Makes the success/failure ratio
|
||||
/// visible without having to derive it from the histogram.
|
||||
pub static CHAIN_STEPS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
|
||||
register_int_counter_vec!(
|
||||
"prop_grid_rs_chain_steps_total",
|
||||
"Total grid_tasks chain steps processed",
|
||||
&["outcome"]
|
||||
)
|
||||
.expect("register counter")
|
||||
});
|
||||
|
||||
/// Tasks currently in flight. A gauge so both "idle pod" (0) and
|
||||
/// "fully saturated" (== PROP_GRID_RS_PARALLELISM) are distinguishable.
|
||||
pub static TASKS_IN_FLIGHT: LazyLock<IntGauge> = LazyLock::new(|| {
|
||||
register_int_gauge!(
|
||||
"prop_grid_rs_tasks_in_flight",
|
||||
"Number of grid_tasks rows currently being processed"
|
||||
)
|
||||
.expect("register gauge")
|
||||
});
|
||||
|
||||
/// Decode-only duration (the `spawn_blocking(wgrib2)` portion). Helps
|
||||
/// answer "is wgrib2 fork overhead dominant?" without instrumenting the
|
||||
/// full step.
|
||||
pub static DECODE_DURATION: LazyLock<Histogram> = LazyLock::new(|| {
|
||||
register_histogram!(
|
||||
"prop_grid_rs_decode_duration_seconds",
|
||||
"wgrib2 subprocess duration per product (surface or pressure)",
|
||||
vec![0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0]
|
||||
)
|
||||
.expect("register histogram")
|
||||
});
|
||||
|
||||
/// Install the lazy statics so the `/metrics` endpoint lists every
|
||||
/// series from the first scrape, even before any chain step runs.
|
||||
pub fn init() {
|
||||
LazyLock::force(&CHAIN_STEP_DURATION);
|
||||
LazyLock::force(&CHAIN_STEPS_TOTAL);
|
||||
LazyLock::force(&TASKS_IN_FLIGHT);
|
||||
LazyLock::force(&DECODE_DURATION);
|
||||
}
|
||||
|
||||
/// RAII helper: increment on construction, decrement on drop. Guarantees
|
||||
/// the in-flight gauge returns to 0 even if the worker panics.
|
||||
pub struct InFlightGuard;
|
||||
|
||||
impl InFlightGuard {
|
||||
pub fn new() -> Self {
|
||||
TASKS_IN_FLIGHT.inc();
|
||||
Self
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for InFlightGuard {
|
||||
fn drop(&mut self) {
|
||||
TASKS_IN_FLIGHT.dec();
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InFlightGuard {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn record_chain_step(duration: Duration, success: bool) {
|
||||
let outcome = if success { "success" } else { "failure" };
|
||||
CHAIN_STEP_DURATION
|
||||
.with_label_values(&[outcome])
|
||||
.observe(duration.as_secs_f64());
|
||||
CHAIN_STEPS_TOTAL.with_label_values(&[outcome]).inc();
|
||||
}
|
||||
|
||||
async fn metrics_handler(State(()): State<()>) -> impl IntoResponse {
|
||||
let mut buf = Vec::new();
|
||||
let encoder = TextEncoder::new();
|
||||
let metric_families = prometheus::gather();
|
||||
match encoder.encode(&metric_families, &mut buf) {
|
||||
Ok(()) => (
|
||||
StatusCode::OK,
|
||||
[("content-type", TextEncoder::new().format_type().to_string())],
|
||||
buf,
|
||||
),
|
||||
Err(e) => {
|
||||
warn!(error = %e, "metrics encode failed");
|
||||
(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
[("content-type", "text/plain".to_string())],
|
||||
Vec::new(),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn serve(addr: std::net::SocketAddr) {
|
||||
init();
|
||||
let app = Router::new()
|
||||
.route("/metrics", get(metrics_handler))
|
||||
.route("/health", get(|| async { "ok" }))
|
||||
.with_state(());
|
||||
match tokio::net::TcpListener::bind(addr).await {
|
||||
Ok(listener) => {
|
||||
info!(%addr, "metrics server listening");
|
||||
if let Err(e) = axum::serve(listener, app).await {
|
||||
warn!(error = %e, "metrics server exited");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(error = %e, %addr, "failed to bind metrics listener; continuing without /metrics");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -7,6 +7,7 @@
|
|||
//! boost, ProfilesFile write) — Elixir retains f00.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::{DateTime, Datelike, Timelike, Utc};
|
||||
|
||||
|
|
@ -121,28 +122,47 @@ pub async fn run_chain_step(
|
|||
})
|
||||
.collect();
|
||||
|
||||
// Step 2: for each band, score → write → drop. One band's score
|
||||
// vector (~92k × 20 B ≈ 1.8 MB) lives at a time instead of holding
|
||||
// all 23 bands at once. The write runs on the blocking pool so the
|
||||
// reactor stays responsive.
|
||||
// Step 2: score and write every band in parallel.
|
||||
//
|
||||
// Scoring: rayon's thread pool over the 23 bands. Each band is
|
||||
// independent pure math across the ~92k-cell `prepared` slice;
|
||||
// sharing is read-only so no locking. This cuts per-step scoring
|
||||
// wall from ~20s serial to ~5-8s on a 4-core box.
|
||||
//
|
||||
// Writing: each band file is an independent NFS fsync. Launching
|
||||
// them as parallel `spawn_blocking` tasks lets them overlap on the
|
||||
// blocking pool instead of serializing one-at-a-time on the hot
|
||||
// path. Memory stays bounded — each Vec<ScorePoint> is ~1.8 MB so
|
||||
// 23 concurrent ≈ ~40 MB peak.
|
||||
let bands = band_config::all_bands();
|
||||
let scores_dir = scores_dir.to_path_buf();
|
||||
let mut files_written = 0;
|
||||
for band in bands {
|
||||
let mut scores: Vec<ScorePoint> = Vec::with_capacity(prepared.len());
|
||||
for (lat, lon, conditions, invariants) in &prepared {
|
||||
let r = scorer::composite_score_with(conditions, band, Some(*invariants));
|
||||
scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score });
|
||||
}
|
||||
let prepared = Arc::new(prepared);
|
||||
let scored: Vec<(u32, Vec<ScorePoint>)> = {
|
||||
use rayon::prelude::*;
|
||||
bands
|
||||
.par_iter()
|
||||
.map(|band| {
|
||||
let mut scores: Vec<ScorePoint> = Vec::with_capacity(prepared.len());
|
||||
for (lat, lon, conditions, invariants) in prepared.iter() {
|
||||
let r = scorer::composite_score_with(conditions, band, Some(*invariants));
|
||||
scores.push(ScorePoint { lat: *lat, lon: *lon, score: r.score });
|
||||
}
|
||||
(band.freq_mhz, scores)
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
|
||||
let write_futs = scored.into_iter().map(|(band_mhz, scores)| {
|
||||
let dir = scores_dir.clone();
|
||||
let band_mhz = band.freq_mhz;
|
||||
tokio::task::spawn_blocking(move || {
|
||||
scores_file::write_atomic(&dir, band_mhz, valid_time, &scores)
|
||||
})
|
||||
.await
|
||||
.expect("blocking join")?;
|
||||
files_written += 1;
|
||||
});
|
||||
let results = futures::future::try_join_all(write_futs).await.expect("blocking join");
|
||||
for r in results {
|
||||
r?;
|
||||
}
|
||||
let files_written = bands.len() as u32;
|
||||
|
||||
Ok(ChainStepStats {
|
||||
score_files_written: files_written,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue