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

@@ -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

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

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<dyn std::error::Error>> {
     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<dyn std::error::Error>> {
                 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<dyn std::error::Error>> {
                         }
                     }
                     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");

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<Histogram> = 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<HistogramVec> = 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<IntCounterVec> = 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<IntCounterVec> = 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);
+    }
+}

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