From efa3cc804da4855ca6d167f51d4baf74c4877ca0 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 20 Apr 2026 13:18:12 -0500 Subject: [PATCH] perf(rust): disk-backed HRRR idx cache + scale workers off talos5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shared NFS idx cache at /data/hrrr_idx lets all prop-grid-rs and hrrr-point-rs replicas deduplicate redundant NOAA S3 idx fetches. Opt-in via HRRR_IDX_CACHE_DIR; falls back to in-memory-only when unset. Also scales prop-grid-rs to 3 replicas (no talos5 pinning), hrrr-point-rs to 2 replicas, and drops hot pod replicas 4→3 so the physical-host anti-affinity still fits with talos5 cordoned off for Postgres. --- k8s/deployment-grid-rs.yaml | 33 ++--- k8s/deployment-hrrr-point-rs.yaml | 16 ++- k8s/deployment.yaml | 2 +- rust/prop_grid_rs/src/fetcher.rs | 192 +++++++++++++++++++++++++++--- 4 files changed, 201 insertions(+), 42 deletions(-) diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 8c36826c..a40cc56e 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -9,10 +9,11 @@ metadata: # commercial-link degradation, ProfilesFile write). f01..f18 land # here from the Rust worker. spec: - # 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 + # Three replicas claim from grid_tasks via FOR UPDATE SKIP LOCKED + # with no coordination. Anti-affinity spreads them across distinct + # hosts so any single node going down leaves the hourly chain with + # two workers still draining. + replicas: 3 minReadySeconds: 5 strategy: type: RollingUpdate @@ -34,10 +35,8 @@ spec: prometheus.io/port: "9100" prometheus.io/path: "/metrics" spec: - # 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. + # No node pinning — let the scheduler place replicas on any host + # the anti-affinity rule allows. talos5 is reserved for Postgres. affinity: podAntiAffinity: requiredDuringSchedulingIgnoredDuringExecution: @@ -45,19 +44,6 @@ spec: 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 - value: grid-rs - effect: NoSchedule imagePullSecrets: - name: forgejo-registry securityContext: @@ -80,6 +66,11 @@ spec: value: "http://skippy.w5isp.com:8080" - name: PROP_SCORES_DIR value: "/data/scores" + # Shared idx-file cache on NFS. All grid-rs replicas read/write + # through the same directory; idx bodies are immutable per HRRR + # run, so cross-replica hits eliminate redundant NOAA S3 calls. + - name: HRRR_IDX_CACHE_DIR + value: "/data/hrrr_idx" - name: RUST_LOG value: "info" # Single concurrent task per pod. 2 replicas → 2 concurrent diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index ca3c72e5..934402eb 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -8,7 +8,9 @@ metadata: # points JSONB array); this deployment drains via FOR UPDATE SKIP # LOCKED, fetches each HRRR cycle once, and upserts hrrr_profiles. spec: - replicas: 1 + # Two replicas for parallelism on QSO-submit spikes. FOR UPDATE SKIP + # LOCKED on hrrr_fetch_tasks lets both claim without coordination. + replicas: 2 strategy: type: RollingUpdate rollingUpdate: @@ -59,6 +61,10 @@ spec: value: "info" - name: PROP_GRID_RS_PG_CONNS value: "4" + # Shared NFS idx cache — same directory as prop-grid-rs so a + # chain-step idx fetch warms both workers. + - name: HRRR_IDX_CACHE_DIR + value: "/data/hrrr_idx" - name: OTEL_EXPORTER_OTLP_ENDPOINT value: "http://otel-collector.observability.svc.cluster.local:4317" envFrom: @@ -87,3 +93,11 @@ spec: - ALL seccompProfile: type: RuntimeDefault + volumeMounts: + - name: data + mountPath: /data + volumes: + - name: data + nfs: + server: 10.0.15.103 + path: /data diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 7b80e9f4..04db849a 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -4,7 +4,7 @@ metadata: name: prop namespace: prop spec: - replicas: 4 + replicas: 3 minReadySeconds: 5 strategy: type: RollingUpdate diff --git a/rust/prop_grid_rs/src/fetcher.rs b/rust/prop_grid_rs/src/fetcher.rs index 84c1519d..e5fd1bd3 100644 --- a/rust/prop_grid_rs/src/fetcher.rs +++ b/rust/prop_grid_rs/src/fetcher.rs @@ -7,6 +7,9 @@ //! * retry on HTTP 429/5xx with jittered exponential backoff use std::collections::HashMap; +use std::fs; +use std::io::Write; +use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::time::{Duration, Instant}; @@ -98,13 +101,15 @@ fn parse_idx_line(line: &str) -> Option { parts.next()?; // date let var = parts.next()?.to_string(); let level = parts.next()?.to_string(); - Some(IdxEntry { msg, offset, var, level }) + Some(IdxEntry { + msg, + offset, + var, + level, + }) } -pub fn byte_ranges_for_messages( - idx: &[IdxEntry], - wanted: &[(String, String)], -) -> Vec { +pub fn byte_ranges_for_messages(idx: &[IdxEntry], wanted: &[(String, String)]) -> Vec { let mut out = Vec::new(); for (var, level) in wanted { for (i, entry) in idx.iter().enumerate() { @@ -113,7 +118,10 @@ pub fn byte_ranges_for_messages( .get(i + 1) .map(|nxt| nxt.offset.saturating_sub(1)) .unwrap_or(entry.offset + 10_000_000); - out.push(ByteRange { start: entry.offset, end }); + out.push(ByteRange { + start: entry.offset, + end, + }); } } } @@ -166,15 +174,73 @@ pub fn surface_messages_owned() -> Vec<(String, String)> { // ── Idx cache ──────────────────────────────────────────────────────── +// Disk TTL is longer than the in-memory TTL because idx files are +// immutable per HRRR run — cross-replica sharing on NFS benefits from +// keeping hits alive across pod restarts and the next hourly cycle. +pub const IDX_DISK_TTL: Duration = Duration::from_secs(24 * 3_600); + #[derive(Debug, Clone)] struct IdxCacheEntry { body: String, inserted: Instant, } +#[derive(Debug)] +pub struct DiskIdxCache { + dir: PathBuf, + ttl: Duration, +} + +impl DiskIdxCache { + pub fn new(dir: impl AsRef, ttl: Duration) -> Self { + let dir = dir.as_ref().to_path_buf(); + let _ = fs::create_dir_all(&dir); + Self { dir, ttl } + } + + fn path_for(&self, url: &str) -> PathBuf { + // Sanitize URL into a safe filename. Collisions would require + // two distinct URLs to differ only in unsafe characters mapped + // to '_' — not possible for HRRR S3 URLs. + let mut sanitized = String::with_capacity(url.len()); + for c in url.chars() { + if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { + sanitized.push(c); + } else { + sanitized.push('_'); + } + } + self.dir.join(sanitized) + } + + pub fn get(&self, url: &str) -> Option { + let path = self.path_for(url); + let meta = fs::metadata(&path).ok()?; + let mtime = meta.modified().ok()?; + if mtime.elapsed().ok()? > self.ttl { + return None; + } + fs::read_to_string(&path).ok() + } + + pub fn put(&self, url: &str, body: &str) { + // Atomic write: tmp file + rename so concurrent readers across + // replicas never observe a partial body. + let path = self.path_for(url); + let tmp = path.with_extension("tmp"); + if let Ok(mut f) = fs::File::create(&tmp) { + if f.write_all(body.as_bytes()).is_ok() { + drop(f); + let _ = fs::rename(&tmp, &path); + } + } + } +} + #[derive(Debug, Default)] pub struct IdxCache { inner: Mutex>, + disk: Option, } impl IdxCache { @@ -182,19 +248,50 @@ impl IdxCache { Self::default() } - pub fn get(&self, url: &str) -> Option { - let map = self.inner.lock().ok()?; - let entry = map.get(url)?; - if entry.inserted.elapsed() < IDX_TTL { - Some(entry.body.clone()) - } else { - None + pub fn with_disk(dir: impl AsRef) -> Self { + Self { + inner: Mutex::new(HashMap::new()), + disk: Some(DiskIdxCache::new(dir, IDX_DISK_TTL)), } } + pub fn get(&self, url: &str) -> Option { + if let Ok(map) = self.inner.lock() { + if let Some(entry) = map.get(url) { + if entry.inserted.elapsed() < IDX_TTL { + return Some(entry.body.clone()); + } + } + } + if let Some(disk) = &self.disk { + if let Some(body) = disk.get(url) { + if let Ok(mut map) = self.inner.lock() { + map.insert( + url.to_string(), + IdxCacheEntry { + body: body.clone(), + inserted: Instant::now(), + }, + ); + } + return Some(body); + } + } + None + } + pub fn put(&self, url: String, body: String) { + if let Some(disk) = &self.disk { + disk.put(&url, &body); + } if let Ok(mut map) = self.inner.lock() { - map.insert(url, IdxCacheEntry { body, inserted: Instant::now() }); + map.insert( + url, + IdxCacheEntry { + body, + inserted: Instant::now(), + }, + ); } } } @@ -213,10 +310,17 @@ impl HrrrClient { .timeout(REQUEST_TIMEOUT) .pool_max_idle_per_host(MAX_PARALLEL_RANGES) .build()?; + // HRRR_IDX_CACHE_DIR opts into a disk-backed idx cache shared + // across replicas via NFS (e.g. `/data/hrrr_idx`). Unset leaves + // the in-memory-only behaviour intact. + let idx_cache = match std::env::var("HRRR_IDX_CACHE_DIR") { + Ok(dir) if !dir.is_empty() => IdxCache::with_disk(dir), + _ => IdxCache::new(), + }; Ok(Self { http, base: base.into(), - idx_cache: IdxCache::new(), + idx_cache, }) } @@ -397,8 +501,18 @@ mod tests { #[test] fn byte_ranges_cover_last_message_with_fallback() { let idx = vec![ - IdxEntry { msg: 1, offset: 0, var: "A".into(), level: "x".into() }, - IdxEntry { msg: 2, offset: 100, var: "B".into(), level: "y".into() }, + IdxEntry { + msg: 1, + offset: 0, + var: "A".into(), + level: "x".into(), + }, + IdxEntry { + msg: 2, + offset: 100, + var: "B".into(), + level: "y".into(), + }, ]; let wanted = vec![("A".into(), "x".into()), ("B".into(), "y".into())]; let r = byte_ranges_for_messages(&idx, &wanted); @@ -413,8 +527,14 @@ mod tests { fn adjacent_ranges_merge() { let merged = merge_ranges(vec![ ByteRange { start: 0, end: 99 }, - ByteRange { start: 100, end: 200 }, - ByteRange { start: 500, end: 600 }, + ByteRange { + start: 100, + end: 200, + }, + ByteRange { + start: 500, + end: 600, + }, ]); assert_eq!(merged.len(), 2); assert_eq!(merged[0].start, 0); @@ -442,4 +562,38 @@ mod tests { assert_eq!(cache.get("u").as_deref(), Some("body")); assert!(cache.get("missing").is_none()); } + + #[test] + fn disk_idx_cache_round_trips() { + let tmp = tempfile::TempDir::new().unwrap(); + let cache = DiskIdxCache::new(tmp.path(), Duration::from_secs(3_600)); + cache.put("https://example.com/a.grib2.idx", "body1"); + assert_eq!( + cache.get("https://example.com/a.grib2.idx").as_deref(), + Some("body1") + ); + assert!(cache.get("https://example.com/missing.idx").is_none()); + } + + #[test] + fn disk_idx_cache_respects_ttl() { + let tmp = tempfile::TempDir::new().unwrap(); + let cache = DiskIdxCache::new(tmp.path(), Duration::from_millis(10)); + cache.put("u", "body"); + std::thread::sleep(Duration::from_millis(30)); + assert!(cache.get("u").is_none()); + } + + #[test] + fn idx_cache_disk_fallback_rewarms_memory() { + let tmp = tempfile::TempDir::new().unwrap(); + // Seed disk via one cache instance... + let c1 = IdxCache::with_disk(tmp.path()); + c1.put("u".into(), "body".into()); + + // A fresh cache simulates a new pod — in-memory map is empty + // but the disk entry must still resolve. + let c2 = IdxCache::with_disk(tmp.path()); + assert_eq!(c2.get("u").as_deref(), Some("body")); + } }