fix(hrrr-points): bind jsonb[] correctly, warn on empty fetch, 2 Gi limit
Three fixes after hrrr-point-rs restarts on the first live drain: 1. Type mismatch (hard error): the `profile` column on `hrrr_profiles` is `jsonb[]` (one element per pressure level), but the Rust worker was binding a single jsonb array-of-objects cast as `::jsonb`. Every insert failed with `column "profile" is of type jsonb[] but expression is of type jsonb`. Switched to `Vec<sqlx::types::Json<Value>>`, dropped the explicit `::jsonb` cast, and enabled sqlx's `json` feature so the encoder maps the array into `_jsonb` correctly. 2. Silent zero-inserts: four consecutive 2019-09-22 batches completed with `profiles_inserted: 0` and no diagnostic. Most likely the upstream archive doesn't keep cycles that old, but without a log it looks identical to a snap-mismatch bug. Added a WARN when both surface and pressure grids come back empty so fetch-miss vs. projection-miss is distinguishable. 3. OOMKilled: the pod took four OOM restarts in ten minutes at 1 Gi. A single CONUS decode holds the ~40 MB blob, the ~200 MB wgrib2 working set, AND the full 92k-cell merged map until all requested points drain. 1 Gi has no headroom. Bumped to 2 Gi, matching the rest of the per-container budgets in this namespace.
This commit is contained in:
parent
48da629147
commit
a1d2e9f94a
3 changed files with 29 additions and 12 deletions
|
|
@ -69,10 +69,13 @@ spec:
|
||||||
limits:
|
limits:
|
||||||
# Per-batch peak is a single CONUS surface + pressure
|
# Per-batch peak is a single CONUS surface + pressure
|
||||||
# decode (~40 MB blob, ~200 MB wgrib2 working set plus
|
# decode (~40 MB blob, ~200 MB wgrib2 working set plus
|
||||||
# the merged CellValues map). 1 Gi gives headroom with
|
# the full 92k-cell merged CellValues map kept in memory
|
||||||
# room for occasional retries.
|
# until all requested points are drained). 1 Gi OOMKilled
|
||||||
|
# repeatedly — four restarts in ten minutes — so bumped
|
||||||
|
# to 2 Gi. The scheduler-on-worker2 placement has plenty
|
||||||
|
# of headroom.
|
||||||
cpu: "2"
|
cpu: "2"
|
||||||
memory: 1Gi
|
memory: 2Gi
|
||||||
securityContext:
|
securityContext:
|
||||||
allowPrivilegeEscalation: false
|
allowPrivilegeEscalation: false
|
||||||
runAsNonRoot: true
|
runAsNonRoot: true
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,7 @@ path = "src/bin/hrrr_point_worker.rs"
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
tokio = { version = "1", features = ["full"] }
|
tokio = { version = "1", features = ["full"] }
|
||||||
sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "macros"] }
|
sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "macros", "json"] }
|
||||||
reqwest = { version = "0.12", features = ["rustls-tls", "stream"], default-features = false }
|
reqwest = { version = "0.12", features = ["rustls-tls", "stream"], default-features = false }
|
||||||
bytes = "1"
|
bytes = "1"
|
||||||
anyhow = "1"
|
anyhow = "1"
|
||||||
|
|
|
||||||
|
|
@ -75,6 +75,17 @@ pub async fn process_batch(
|
||||||
.await
|
.await
|
||||||
.expect("blocking join")?;
|
.expect("blocking join")?;
|
||||||
|
|
||||||
|
// Empty grids almost always mean the HRRR archive doesn't have this
|
||||||
|
// cycle (common for backfills older than the rolling retention
|
||||||
|
// window). Log so "0 inserted" doesn't look like a silent success.
|
||||||
|
if sfc_grid.is_empty() && prs_grid.is_empty() {
|
||||||
|
tracing::warn!(
|
||||||
|
valid_time = %valid_time,
|
||||||
|
points = points.len(),
|
||||||
|
"hrrr fetch returned empty surface and pressure grids — upstream archive likely missing this cycle"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
// For each requested point, pull the merged cell and build an
|
// For each requested point, pull the merged cell and build an
|
||||||
// hrrr_profiles row. Skip the point silently if extract_grid
|
// hrrr_profiles row. Skip the point silently if extract_grid
|
||||||
// produced nothing for its snapped key (e.g. off-CONUS).
|
// produced nothing for its snapped key (e.g. off-CONUS).
|
||||||
|
|
@ -124,10 +135,14 @@ async fn upsert_profile(
|
||||||
.copied()
|
.copied()
|
||||||
.map(|v| v as f64);
|
.map(|v| v as f64);
|
||||||
|
|
||||||
// Pressure-level profile list for SoundingParams.derive compatibility
|
// Pressure-level profile list. Postgres column `profile` is jsonb[]
|
||||||
// on the Elixir read side. Same shape Elixir's HrrrClient.build_profile
|
// (ARRAY of jsonb), one element per level, matching Elixir's
|
||||||
// emits: [{"pres_mb", "hght_m", "tmpc", "dwpc"}, …].
|
// HrrrClient.build_profile output shape:
|
||||||
let levels: Vec<serde_json::Value> = fetcher::GRID_PRESSURE_LEVELS
|
// [{"pres_mb", "hght_m", "tmpc", "dwpc"}, …]
|
||||||
|
// We build a Vec<Json<Value>> so sqlx encodes it as jsonb[] rather
|
||||||
|
// than wrapping the whole list in a single jsonb (which yields the
|
||||||
|
// "expression is of type jsonb" type-mismatch error at runtime).
|
||||||
|
let levels: Vec<sqlx::types::Json<serde_json::Value>> = fetcher::GRID_PRESSURE_LEVELS
|
||||||
.iter()
|
.iter()
|
||||||
.filter_map(|&p| {
|
.filter_map(|&p| {
|
||||||
let pres_mb = p as f64;
|
let pres_mb = p as f64;
|
||||||
|
|
@ -141,10 +156,9 @@ async fn upsert_profile(
|
||||||
if let Some(dv) = d {
|
if let Some(dv) = d {
|
||||||
m.insert("dwpc".into(), serde_json::json!((dv as f64) - 273.15));
|
m.insert("dwpc".into(), serde_json::json!((dv as f64) - 273.15));
|
||||||
}
|
}
|
||||||
Some(serde_json::Value::Object(m))
|
Some(sqlx::types::Json(serde_json::Value::Object(m)))
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
let profile_json = serde_json::Value::Array(levels);
|
|
||||||
|
|
||||||
let id = Uuid::new_v4();
|
let id = Uuid::new_v4();
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
|
|
@ -154,7 +168,7 @@ async fn upsert_profile(
|
||||||
surface_temp_c, surface_dewpoint_c, surface_pressure_mb,
|
surface_temp_c, surface_dewpoint_c, surface_pressure_mb,
|
||||||
is_grid_point, inserted_at, updated_at)
|
is_grid_point, inserted_at, updated_at)
|
||||||
VALUES
|
VALUES
|
||||||
($1, $2, $3, $4, $2, $5::jsonb, $6, $7, $8, $9, $10, false, NOW(), NOW())
|
($1, $2, $3, $4, $2, $5, $6, $7, $8, $9, $10, false, NOW(), NOW())
|
||||||
ON CONFLICT (lat, lon, valid_time) DO UPDATE
|
ON CONFLICT (lat, lon, valid_time) DO UPDATE
|
||||||
SET profile = EXCLUDED.profile,
|
SET profile = EXCLUDED.profile,
|
||||||
hpbl_m = EXCLUDED.hpbl_m,
|
hpbl_m = EXCLUDED.hpbl_m,
|
||||||
|
|
@ -169,7 +183,7 @@ async fn upsert_profile(
|
||||||
.bind(valid_time.naive_utc())
|
.bind(valid_time.naive_utc())
|
||||||
.bind(lat)
|
.bind(lat)
|
||||||
.bind(lon)
|
.bind(lon)
|
||||||
.bind(serde_json::to_string(&profile_json).unwrap_or_else(|_| "[]".to_string()))
|
.bind(&levels)
|
||||||
.bind(hpbl_m)
|
.bind(hpbl_m)
|
||||||
.bind(pwat_mm)
|
.bind(pwat_mm)
|
||||||
.bind(surface_temp_c)
|
.bind(surface_temp_c)
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue