From b8eea1bd456d683d6d0ae052c68a3e778e749640 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 8 May 2026 10:36:49 -0500 Subject: [PATCH] prom: wire rust workers into prometheus prop-grid-rs already exposed prop_grid_rs_chain_step_duration_seconds + tasks_in_flight on :9100 but the comment still pointed at the old prom server (10.0.15.25). hrrr-point-rs had no metrics listener at all. - metrics.rs: new POINT_BATCH_DURATION / POINT_BATCHES_TOTAL / POINTS_PROCESSED_TOTAL series + record_point_batch helper, plus unit tests serialized via a static Mutex (the prometheus crate's global registry races otherwise) - hrrr_point_worker.rs: spawn metrics::serve on METRICS_ADDR (default 0.0.0.0:9100), wrap process_batch in InFlightGuard, record outcome + per-point counts on every batch - deployment-hrrr-point-rs.yaml: prometheus.io/{scrape,port,path,job} annotations, METRICS_ADDR env, ports.metrics, /health readiness probe - deployment-grid-rs.yaml: refresh stale 10.0.15.25 comment, add prometheus.io/job for clean dashboard up{} selectors Also fix Microwaveprop.PromExTest: `assert plugins != []` triggered Elixir 1.19 type-checker warning ("non_empty_list != empty_list always true"); replaced with membership assertions for the InstrumentPlugin + Plugins.Beam. --- k8s/deployment-grid-rs.yaml | 9 +- k8s/deployment-hrrr-point-rs.yaml | 30 ++++ .../prop_grid_rs/src/bin/hrrr_point_worker.rs | 30 +++- rust/prop_grid_rs/src/metrics.rs | 143 +++++++++++++++++- test/microwaveprop/prom_ex_test.exs | 5 +- 5 files changed, 209 insertions(+), 8 deletions(-) diff --git a/k8s/deployment-grid-rs.yaml b/k8s/deployment-grid-rs.yaml index 7c832456..e51325f3 100644 --- a/k8s/deployment-grid-rs.yaml +++ b/k8s/deployment-grid-rs.yaml @@ -36,11 +36,16 @@ spec: 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). + # Scraped by the Prometheus at 10.0.15.31 via the + # `kubernetes-pods` job in prometheus.yml.j2 (apiserver proxy). + # Metrics endpoint is on container port 9100 (see METRICS_ADDR + # env below). `prometheus.io/job` makes the dashboard's + # `up{job="prop-grid-rs"}` selector work without per-pod + # relabel. prometheus.io/scrape: "true" prometheus.io/port: "9100" prometheus.io/path: "/metrics" + prometheus.io/job: "prop-grid-rs" spec: # No node pinning — scheduler picks any host. Anti-affinity was # removed so HPA can stack multiple replicas on the same node if diff --git a/k8s/deployment-hrrr-point-rs.yaml b/k8s/deployment-hrrr-point-rs.yaml index 824611f7..96d3cf6e 100644 --- a/k8s/deployment-hrrr-point-rs.yaml +++ b/k8s/deployment-hrrr-point-rs.yaml @@ -30,6 +30,17 @@ spec: labels: app: hrrr-point-rs tier: hrrr-points + annotations: + # Scraped by the Prometheus at 10.0.15.31 via the + # `kubernetes-pods` job in prometheus.yml.j2 (apiserver proxy). + # The hrrr_point_worker binary spawns the same metrics::serve + # listener as prop-grid-rs (METRICS_ADDR, default 0.0.0.0:9100) + # and reports prop_grid_rs_point_batch_* + points_processed_* + # series. + prometheus.io/scrape: "true" + prometheus.io/port: "9100" + prometheus.io/path: "/metrics" + prometheus.io/job: "hrrr-point-rs" spec: # Anti-affinity against prop-grid-rs was removed; the scheduler # is free to colocate. Contention on wgrib2 + NFS hasn't been an @@ -67,9 +78,28 @@ spec: # chain-step idx fetch warms both workers. - name: HRRR_IDX_CACHE_DIR value: "/data/hrrr_idx" + # /metrics + /health listener; same default as prop-grid-rs. + - name: METRICS_ADDR + value: "0.0.0.0:9100" envFrom: - secretRef: name: prop-secrets + ports: + - name: metrics + containerPort: 9100 + protocol: TCP + # Readiness mirrors prop-grid-rs: /health returns 200 once the + # axum metrics listener is bound. Liveness intentionally + # omitted — an idle worker (no claimable tasks) is healthy, + # and the kubelet will restart on process crash anyway. + readinessProbe: + httpGet: + path: /health + port: 9100 + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 6 resources: requests: cpu: 100m diff --git a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs index 7de137f2..40cd73fa 100644 --- a/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs +++ b/rust/prop_grid_rs/src/bin/hrrr_point_worker.rs @@ -4,11 +4,12 @@ //! requested point, and upserts into `hrrr_profiles`. FOR UPDATE SKIP //! LOCKED allows N replicas with no coordination. +use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; -use prop_grid_rs::{db, fetcher::HrrrClient, hrrr_points, telemetry}; +use prop_grid_rs::{db, fetcher::HrrrClient, hrrr_points, metrics, telemetry}; use tracing::{error, info, warn}; const IDLE_SLEEP: Duration = Duration::from_secs(10); @@ -36,7 +37,23 @@ async fn main() -> Result<(), Box> { let shutdown = Arc::new(AtomicBool::new(false)); spawn_signal_handler(shutdown.clone()); - info!(pg_conns = max_conn, "hrrr-point-worker started"); + // Spawn the shared `/metrics` + `/health` HTTP listener on the same + // port the prop-grid-rs worker uses (METRICS_ADDR, default + // 0.0.0.0:9100). Pre-creating the metric statics so the first + // scrape lists every series even before any task runs. + 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!( + pg_conns = max_conn, + metrics_addr = %metrics_addr, + "hrrr-point-worker started" + ); let mut claim_failures: u32 = 0; while !shutdown.load(Ordering::Relaxed) { @@ -48,9 +65,16 @@ async fn main() -> Result<(), Box> { let points = task.points.clone(); let started = Instant::now(); + let _flight = metrics::InFlightGuard::new(); match hrrr_points::process_batch(&client, &pool, valid_time, &points).await { Ok(stats) => { let elapsed = started.elapsed(); + metrics::record_point_batch( + elapsed, + true, + stats.points_requested as u64, + stats.profiles_inserted as u64, + ); info!( %task_id, valid_time = %valid_time, @@ -64,6 +88,8 @@ async fn main() -> Result<(), Box> { } } Err(e) => { + let elapsed = started.elapsed(); + metrics::record_point_batch(elapsed, false, points.len() as u64, 0); warn!(%task_id, error = %e, "hrrr batch failed"); if let Err(dbe) = db::fail_hrrr_task(&pool, task_id, &e.to_string()).await { error!(%task_id, error = %dbe, "failed to mark task failed"); diff --git a/rust/prop_grid_rs/src/metrics.rs b/rust/prop_grid_rs/src/metrics.rs index 62a51805..e8e2de4b 100644 --- a/rust/prop_grid_rs/src/metrics.rs +++ b/rust/prop_grid_rs/src/metrics.rs @@ -1,7 +1,8 @@ //! 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`). +//! Scraped by the Prometheus server at 10.0.15.31. Exposed on +//! `METRICS_ADDR` (default `0.0.0.0:9100`). Both binaries +//! (`worker` and `hrrr_point_worker`) share this module. use std::sync::LazyLock; use std::time::Duration; @@ -59,6 +60,41 @@ pub static DECODE_DURATION: LazyLock = LazyLock::new(|| { .expect("register histogram") }); +/// Per-batch wall time of `hrrr_point_worker::process_batch`. Labelled +/// by outcome so retry loops can be tracked separately from the happy +/// path. +pub static POINT_BATCH_DURATION: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "prop_grid_rs_point_batch_duration_seconds", + "Wall time per hrrr_fetch_tasks batch (fetch + decode + upsert)", + &["outcome"], + vec![0.5, 1.0, 2.5, 5.0, 10.0, 20.0, 30.0, 60.0, 120.0] + ) + .expect("register histogram") +}); + +/// Counter of point batches by outcome (success/failure). +pub static POINT_BATCHES_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "prop_grid_rs_point_batches_total", + "Total hrrr_fetch_tasks batches processed", + &["outcome"] + ) + .expect("register counter") +}); + +/// Counter of points by lifecycle stage (`requested` from the task, +/// `inserted` into `hrrr_profiles`). The ratio gives the cache-fill +/// rate without needing a separate gauge. +pub static POINTS_PROCESSED_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "prop_grid_rs_points_processed_total", + "Per-point counts across hrrr_fetch_tasks batches", + &["kind"] + ) + .expect("register counter") +}); + /// Install the lazy statics so the `/metrics` endpoint lists every /// series from the first scrape, even before any chain step runs. pub fn init() { @@ -66,6 +102,9 @@ pub fn init() { LazyLock::force(&CHAIN_STEPS_TOTAL); LazyLock::force(&TASKS_IN_FLIGHT); LazyLock::force(&DECODE_DURATION); + LazyLock::force(&POINT_BATCH_DURATION); + LazyLock::force(&POINT_BATCHES_TOTAL); + LazyLock::force(&POINTS_PROCESSED_TOTAL); } /// RAII helper: increment on construction, decrement on drop. Guarantees @@ -99,6 +138,20 @@ pub fn record_chain_step(duration: Duration, success: bool) { CHAIN_STEPS_TOTAL.with_label_values(&[outcome]).inc(); } +pub fn record_point_batch(duration: Duration, success: bool, requested: u64, inserted: u64) { + let outcome = if success { "success" } else { "failure" }; + POINT_BATCH_DURATION + .with_label_values(&[outcome]) + .observe(duration.as_secs_f64()); + POINT_BATCHES_TOTAL.with_label_values(&[outcome]).inc(); + POINTS_PROCESSED_TOTAL + .with_label_values(&["requested"]) + .inc_by(requested); + POINTS_PROCESSED_TOTAL + .with_label_values(&["inserted"]) + .inc_by(inserted); +} + async fn metrics_handler(State(()): State<()>) -> impl IntoResponse { let mut buf = Vec::new(); let encoder = TextEncoder::new(); @@ -138,3 +191,89 @@ pub async fn serve(addr: std::net::SocketAddr) { } } } + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::*; + + // The prometheus crate keeps a process-wide registry so parallel + // tests would race on counter deltas. Serialize via a static Mutex. + static SERIAL: Mutex<()> = Mutex::new(()); + + #[test] + fn record_point_batch_success_increments_counters_and_observes_duration() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + init(); + let baseline_success = POINT_BATCHES_TOTAL.with_label_values(&["success"]).get(); + let baseline_requested = POINTS_PROCESSED_TOTAL.with_label_values(&["requested"]).get(); + let baseline_inserted = POINTS_PROCESSED_TOTAL.with_label_values(&["inserted"]).get(); + let baseline_hist_count = POINT_BATCH_DURATION + .with_label_values(&["success"]) + .get_sample_count(); + + record_point_batch(Duration::from_millis(250), true, 12, 9); + + assert_eq!( + POINT_BATCHES_TOTAL.with_label_values(&["success"]).get(), + baseline_success + 1 + ); + assert_eq!( + POINTS_PROCESSED_TOTAL.with_label_values(&["requested"]).get(), + baseline_requested + 12 + ); + assert_eq!( + POINTS_PROCESSED_TOTAL.with_label_values(&["inserted"]).get(), + baseline_inserted + 9 + ); + assert_eq!( + POINT_BATCH_DURATION + .with_label_values(&["success"]) + .get_sample_count(), + baseline_hist_count + 1 + ); + } + + #[test] + fn record_point_batch_failure_uses_failure_label() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + init(); + let baseline = POINT_BATCHES_TOTAL.with_label_values(&["failure"]).get(); + + record_point_batch(Duration::from_millis(50), false, 5, 0); + + assert_eq!( + POINT_BATCHES_TOTAL.with_label_values(&["failure"]).get(), + baseline + 1 + ); + } + + #[test] + fn record_chain_step_increments_counter() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + init(); + let baseline = CHAIN_STEPS_TOTAL.with_label_values(&["success"]).get(); + + record_chain_step(Duration::from_secs(15), true); + + assert_eq!( + CHAIN_STEPS_TOTAL.with_label_values(&["success"]).get(), + baseline + 1 + ); + } + + #[test] + fn in_flight_guard_inc_dec_pairs() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + init(); + let baseline = TASKS_IN_FLIGHT.get(); + + { + let _g = InFlightGuard::new(); + assert_eq!(TASKS_IN_FLIGHT.get(), baseline + 1); + } + + assert_eq!(TASKS_IN_FLIGHT.get(), baseline); + } +} diff --git a/test/microwaveprop/prom_ex_test.exs b/test/microwaveprop/prom_ex_test.exs index aec50ae0..5a73f836 100644 --- a/test/microwaveprop/prom_ex_test.exs +++ b/test/microwaveprop/prom_ex_test.exs @@ -10,10 +10,11 @@ defmodule Microwaveprop.PromExTest do end describe "plugins/0" do - test "returns a non-empty plugin list" do + test "includes the custom InstrumentPlugin alongside the PromEx defaults" do plugins = Microwaveprop.PromEx.plugins() assert is_list(plugins) - assert plugins != [] + assert Microwaveprop.PromEx.InstrumentPlugin in plugins + assert PromEx.Plugins.Beam in plugins end end end