feat(monitoring): add monitoring_service crate with Prometheus-to-gRPC bridge

Prometheus client queries training/hyperopt/GPU/K8s metrics, gRPC service
exposes unary + server-streaming RPCs, 4 unit tests, zero clippy warnings.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Cargo.lock

@@ -6144,6 +6144,32 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "monitoring_service"
+version = "1.0.0"
+dependencies = [
+ "anyhow",
+ "async-stream",
+ "axum 0.7.9",
+ "chrono",
+ "clap",
+ "common",
+ "prometheus",
+ "prost 0.14.1",
+ "prost-build",
+ "prost-types",
+ "reqwest 0.12.23",
+ "serde",
+ "serde_json",
+ "tokio",
+ "tokio-stream",
+ "tonic",
+ "tonic-prost",
+ "tonic-prost-build",
+ "tracing",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "moxcms"
 version = "0.7.6"

Cargo.toml

@@ -132,6 +132,7 @@ members = [
     "services/data_acquisition_service",
     "services/trading_agent_service",
     "services/api_gateway",
+    "services/monitoring_service",
     # Testing
     "testing/integration",
     "testing/e2e",

services/monitoring_service/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "monitoring_service"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+license.workspace = true
+description = "Monitoring Service - Prometheus-to-gRPC bridge for live training metrics"
+
+[dependencies]
+tokio.workspace = true
+tonic.workspace = true
+tonic-prost.workspace = true
+prost.workspace = true
+prost-types.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+anyhow.workspace = true
+clap.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+reqwest = { version = "0.12", features = ["rustls-tls", "json"], default-features = false }
+axum.workspace = true
+prometheus.workspace = true
+tokio-stream.workspace = true
+async-stream.workspace = true
+chrono.workspace = true
+common = { workspace = true }
+
+[build-dependencies]
+tonic-prost-build.workspace = true
+prost-build.workspace = true
+
+[[bin]]
+name = "monitoring_service"
+path = "src/main.rs"

services/monitoring_service/build.rs

@@ -0,0 +1,10 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tonic_prost_build::configure()
+        .build_server(true)
+        .build_client(false)
+        .compile_protos(&["proto/monitoring.proto"], &["proto"])?;
+
+    println!("cargo:rerun-if-changed=proto/monitoring.proto");
+
+    Ok(())
+}

services/monitoring_service/src/main.rs

@@ -0,0 +1,89 @@
+#![deny(
+    clippy::unwrap_used,
+    clippy::expect_used,
+    clippy::panic,
+    clippy::indexing_slicing
+)]
+
+mod prometheus_client;
+mod service;
+
+pub mod monitoring {
+    #![allow(
+        clippy::all,
+        clippy::pedantic,
+        clippy::restriction,
+        clippy::nursery
+    )]
+    tonic::include_proto!("monitoring");
+}
+
+use anyhow::Result;
+use clap::Parser;
+use tracing::info;
+
+use monitoring::monitoring_service_server::MonitoringServiceServer;
+use prometheus_client::PrometheusClient;
+use service::MonitoringServiceImpl;
+
+#[derive(Parser, Debug)]
+#[command(name = "monitoring-service")]
+#[command(about = "Prometheus-to-gRPC bridge for live training metrics")]
+struct Args {
+    /// Prometheus base URL
+    #[arg(
+        long,
+        env = "PROMETHEUS_URL",
+        default_value = "http://gitlab-prometheus-server.foxhunt.svc:80"
+    )]
+    prometheus_url: String,
+
+    /// gRPC listen port
+    #[arg(long, env = "GRPC_PORT", default_value = "50057")]
+    grpc_port: u16,
+
+    /// Prometheus metrics port (self-metrics)
+    #[arg(long, env = "METRICS_PORT", default_value = "9099")]
+    metrics_port: u16,
+
+    /// Default streaming interval in seconds
+    #[arg(long, env = "STREAM_INTERVAL_SECS", default_value = "3")]
+    stream_interval: u32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
+    if let Err(e) =
+        common::observability::init_observability("monitoring_service", otlp_endpoint.as_deref())
+    {
+        eprintln!("Observability init failed (non-fatal): {e}");
+    }
+
+    let args = Args::parse();
+
+    info!("Starting monitoring_service");
+    info!("  Prometheus: {}", args.prometheus_url);
+    info!("  gRPC port: {}", args.grpc_port);
+    info!("  Metrics port: {}", args.metrics_port);
+    info!("  Stream interval: {}s", args.stream_interval);
+
+    // Start self-metrics HTTP endpoint
+    common::metrics::server::start_metrics_server(args.metrics_port);
+
+    let prom = PrometheusClient::new(&args.prometheus_url);
+    let svc = MonitoringServiceImpl::new(prom, args.stream_interval);
+
+    let addr = format!("0.0.0.0:{}", args.grpc_port)
+        .parse()
+        .map_err(|e| anyhow::anyhow!("Invalid listen address: {e}"))?;
+
+    info!("gRPC server listening on {}", addr);
+
+    tonic::transport::Server::builder()
+        .add_service(MonitoringServiceServer::new(svc))
+        .serve(addr)
+        .await?;
+
+    Ok(())
+}

services/monitoring_service/src/prometheus_client.rs

@@ -0,0 +1,118 @@
+use anyhow::{Context, Result};
+use serde::Deserialize;
+use std::collections::HashMap;
+
+/// A single Prometheus instant-query result entry
+#[derive(Debug, Deserialize)]
+struct PromResult {
+    metric: HashMap<String, String>,
+    value: (f64, String), // (timestamp, value_string)
+}
+
+#[derive(Debug, Deserialize)]
+struct PromData {
+    result: Vec<PromResult>,
+}
+
+#[derive(Debug, Deserialize)]
+struct PromResponse {
+    status: String,
+    data: PromData,
+}
+
+/// Parsed metric value with model/fold labels
+#[derive(Debug, Clone)]
+pub struct MetricSample {
+    pub name: String,
+    pub model: String,
+    pub fold: String,
+    pub value: f64,
+}
+
+pub struct PrometheusClient {
+    http: reqwest::Client,
+    base_url: String,
+}
+
+impl PrometheusClient {
+    pub fn new(base_url: &str) -> Self {
+        let http = reqwest::Client::builder()
+            .timeout(std::time::Duration::from_secs(5))
+            .build()
+            .unwrap_or_else(|_| reqwest::Client::new());
+        Self {
+            http,
+            base_url: base_url.trim_end_matches('/').to_owned(),
+        }
+    }
+
+    /// Execute an instant query against Prometheus
+    async fn query(&self, promql: &str) -> Result<Vec<PromResult>> {
+        let url = format!("{}/api/v1/query", self.base_url);
+        let resp = self
+            .http
+            .get(&url)
+            .query(&[("query", promql)])
+            .send()
+            .await
+            .context("Prometheus HTTP request failed")?;
+
+        if !resp.status().is_success() {
+            anyhow::bail!("Prometheus returned HTTP {}", resp.status());
+        }
+
+        let body: PromResponse = resp
+            .json()
+            .await
+            .context("Failed to parse Prometheus response")?;
+
+        if body.status != "success" {
+            anyhow::bail!("Prometheus query status: {}", body.status);
+        }
+
+        Ok(body.data.result)
+    }
+
+    /// Fetch all training + hyperopt metrics
+    pub async fn fetch_training_metrics(&self) -> Result<Vec<MetricSample>> {
+        let results = self
+            .query(r#"{__name__=~"foxhunt_training_.*|foxhunt_hyperopt_.*"}"#)
+            .await?;
+        Ok(parse_samples(results))
+    }
+
+    /// Fetch GPU metrics from DCGM exporter
+    pub async fn fetch_gpu_metrics(&self) -> Result<Vec<MetricSample>> {
+        let results = self
+            .query(r#"{__name__=~"dcgm_gpu_utilization|dcgm_fb_used|dcgm_fb_free|dcgm_gpu_temp|dcgm_power_usage"}"#)
+            .await?;
+        Ok(parse_samples(results))
+    }
+
+    /// Fetch active K8s training job count
+    pub async fn fetch_active_jobs(&self) -> Result<u32> {
+        let results = self
+            .query(r#"kube_job_status_active{namespace="foxhunt",job_name=~"training-.*"}"#)
+            .await?;
+        let count: f64 = results
+            .iter()
+            .filter_map(|r| r.value.1.parse::<f64>().ok())
+            .sum();
+        Ok(count as u32)
+    }
+}
+
+fn parse_samples(results: Vec<PromResult>) -> Vec<MetricSample> {
+    results
+        .into_iter()
+        .filter_map(|r| {
+            let value: f64 = r.value.1.parse().ok()?;
+            Some(MetricSample {
+                name: r.metric.get("__name__")?.clone(),
+                model: r.metric.get("model").cloned().unwrap_or_default(),
+                fold: r.metric.get("fold").cloned().unwrap_or_default(),
+                value,
+            })
+        })
+        .collect()
+}

services/monitoring_service/src/service.rs

@@ -0,0 +1,267 @@
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio_stream::Stream;
+use tonic::{Request, Response, Status};
+use tracing::error;
+
+use crate::monitoring::{
+    monitoring_service_server::MonitoringService, GetLiveTrainingMetricsRequest,
+    GetLiveTrainingMetricsResponse, GpuSnapshot, StreamTrainingMetricsRequest, TrainingSession,
+};
+use crate::prometheus_client::{MetricSample, PrometheusClient};
+
+pub struct MonitoringServiceImpl {
+    prom: Arc<PrometheusClient>,
+    default_interval: u32,
+}
+
+impl MonitoringServiceImpl {
+    pub fn new(prom: PrometheusClient, default_interval: u32) -> Self {
+        Self {
+            prom: Arc::new(prom),
+            default_interval,
+        }
+    }
+
+    async fn build_response(
+        prom: &PrometheusClient,
+        model_filter: &str,
+    ) -> Result<GetLiveTrainingMetricsResponse, Status> {
+        let (training, gpu, jobs) = tokio::try_join!(
+            prom.fetch_training_metrics(),
+            prom.fetch_gpu_metrics(),
+            prom.fetch_active_jobs(),
+        )
+        .map_err(|e| Status::internal(format!("Prometheus query failed: {e}")))?;
+
+        let sessions = group_into_sessions(&training, model_filter);
+        let gpu_snapshot = build_gpu_snapshot(&gpu);
+
+        Ok(GetLiveTrainingMetricsResponse {
+            sessions,
+            gpu: Some(gpu_snapshot),
+            active_k8s_jobs: jobs,
+            timestamp: chrono::Utc::now().timestamp(),
+        })
+    }
+}
+
+#[tonic::async_trait]
+impl MonitoringService for MonitoringServiceImpl {
+    async fn get_live_training_metrics(
+        &self,
+        request: Request<GetLiveTrainingMetricsRequest>,
+    ) -> Result<Response<GetLiveTrainingMetricsResponse>, Status> {
+        let filter = &request.into_inner().model_filter;
+        let resp = Self::build_response(&self.prom, filter).await?;
+        Ok(Response::new(resp))
+    }
+
+    type StreamTrainingMetricsStream =
+        Pin<Box<dyn Stream<Item = Result<GetLiveTrainingMetricsResponse, Status>> + Send>>;
+
+    async fn stream_training_metrics(
+        &self,
+        request: Request<StreamTrainingMetricsRequest>,
+    ) -> Result<Response<Self::StreamTrainingMetricsStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            self.default_interval
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let filter = req.model_filter;
+        let prom = self.prom.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match Self::build_response(&prom, &filter).await {
+                    Ok(resp) => yield Ok(resp),
+                    Err(e) => {
+                        error!("Stream tick failed: {}", e);
+                        yield Err(e);
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
+}
+
+/// Group flat metric samples into TrainingSession structs keyed by (model, fold)
+fn group_into_sessions(samples: &[MetricSample], model_filter: &str) -> Vec<TrainingSession> {
+    let mut map: HashMap<(String, String), TrainingSession> = HashMap::new();
+
+    for s in samples {
+        if !model_filter.is_empty() && s.model != model_filter {
+            continue;
+        }
+
+        let key = (s.model.clone(), s.fold.clone());
+        let session = map.entry(key).or_insert_with(|| TrainingSession {
+            model: s.model.clone(),
+            fold: s.fold.clone(),
+            ..Default::default()
+        });
+
+        match s.name.as_str() {
+            "foxhunt_training_current_epoch" => session.current_epoch = s.value as f32,
+            "foxhunt_training_epoch_loss" => session.epoch_loss = s.value as f32,
+            "foxhunt_training_validation_loss" => session.validation_loss = s.value as f32,
+            "foxhunt_training_batches_per_second" => session.batches_per_second = s.value as f32,
+            "foxhunt_training_batches_processed" => session.batches_processed = s.value as f32,
+            "foxhunt_training_iteration_seconds" => session.iteration_seconds = s.value as f32,
+            "foxhunt_training_eval_accuracy" => session.eval_accuracy = s.value as f32,
+            "foxhunt_training_eval_precision" => session.eval_precision = s.value as f32,
+            "foxhunt_training_eval_recall" => session.eval_recall = s.value as f32,
+            "foxhunt_training_eval_f1" => session.eval_f1 = s.value as f32,
+            "foxhunt_training_checkpoint_size_bytes" => {
+                session.checkpoint_size_bytes = s.value as f32;
+            }
+            "foxhunt_training_checkpoint_saves_total" => {
+                session.checkpoint_saves = s.value as u32;
+            }
+            "foxhunt_training_checkpoint_failures_total" => {
+                session.checkpoint_failures = s.value as u32;
+            }
+            "foxhunt_training_nan_detected_total" => session.nan_detected = s.value as u32,
+            "foxhunt_training_gradient_explosion_total" => {
+                session.gradient_explosions = s.value as u32;
+            }
+            "foxhunt_training_feature_errors_total" => session.feature_errors = s.value as u32,
+            "foxhunt_hyperopt_trial_current" => {
+                session.hyperopt_trial_current = s.value as u32;
+                session.is_hyperopt = true;
+            }
+            "foxhunt_hyperopt_trial_total" => session.hyperopt_trial_total = s.value as u32,
+            "foxhunt_hyperopt_best_objective" => {
+                session.hyperopt_best_objective = s.value as f32;
+            }
+            "foxhunt_hyperopt_trials_failed_total" => {
+                session.hyperopt_trials_failed = s.value as u32;
+            }
+            "foxhunt_hyperopt_mode" => {
+                if s.value > 0.5 {
+                    session.is_hyperopt = true;
+                }
+            }
+            _ => {}
+        }
+    }
+
+    let mut sessions: Vec<_> = map.into_values().collect();
+    sessions.sort_by(|a, b| (&a.model, &a.fold).cmp(&(&b.model, &b.fold)));
+    sessions
+}
+
+fn build_gpu_snapshot(samples: &[MetricSample]) -> GpuSnapshot {
+    let mut snap = GpuSnapshot::default();
+    let mut fb_free: f32 = 0.0;
+    for s in samples {
+        match s.name.as_str() {
+            "dcgm_gpu_utilization" => snap.utilization_percent = s.value as f32,
+            "dcgm_fb_used" => snap.memory_used_mb = s.value as f32,
+            "dcgm_fb_free" => fb_free = s.value as f32,
+            "dcgm_gpu_temp" => snap.temperature_celsius = s.value as f32,
+            "dcgm_power_usage" => snap.power_watts = s.value as f32,
+            _ => {}
+        }
+    }
+    // dcgm_fb_free + dcgm_fb_used = total
+    snap.memory_total_mb = snap.memory_used_mb + fb_free;
+    snap
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_group_empty_samples() {
+        let sessions = group_into_sessions(&[], "");
+        assert!(sessions.is_empty());
+    }
+
+    #[test]
+    fn test_group_with_filter() {
+        let samples = vec![
+            MetricSample {
+                name: "foxhunt_training_current_epoch".to_owned(),
+                model: "dqn".to_owned(),
+                fold: "0".to_owned(),
+                value: 5.0,
+            },
+            MetricSample {
+                name: "foxhunt_training_current_epoch".to_owned(),
+                model: "ppo".to_owned(),
+                fold: "0".to_owned(),
+                value: 3.0,
+            },
+        ];
+        let sessions = group_into_sessions(&samples, "dqn");
+        assert_eq!(sessions.len(), 1);
+        assert_eq!(sessions.first().map(|s| s.model.as_str()), Some("dqn"));
+        assert_eq!(sessions.first().map(|s| s.current_epoch), Some(5.0));
+    }
+
+    #[test]
+    fn test_group_sets_hyperopt_flag() {
+        let samples = vec![MetricSample {
+            name: "foxhunt_hyperopt_mode".to_owned(),
+            model: "dqn".to_owned(),
+            fold: "".to_owned(),
+            value: 1.0,
+        }];
+        let sessions = group_into_sessions(&samples, "");
+        assert_eq!(sessions.len(), 1);
+        assert!(sessions.first().map(|s| s.is_hyperopt).unwrap_or(false));
+    }
+
+    #[test]
+    fn test_build_gpu_snapshot() {
+        let samples = vec![
+            MetricSample {
+                name: "dcgm_gpu_utilization".to_owned(),
+                model: String::new(),
+                fold: String::new(),
+                value: 87.0,
+            },
+            MetricSample {
+                name: "dcgm_fb_used".to_owned(),
+                model: String::new(),
+                fold: String::new(),
+                value: 38200.0,
+            },
+            MetricSample {
+                name: "dcgm_fb_free".to_owned(),
+                model: String::new(),
+                fold: String::new(),
+                value: 9800.0,
+            },
+            MetricSample {
+                name: "dcgm_gpu_temp".to_owned(),
+                model: String::new(),
+                fold: String::new(),
+                value: 62.0,
+            },
+            MetricSample {
+                name: "dcgm_power_usage".to_owned(),
+                model: String::new(),
+                fold: String::new(),
+                value: 245.0,
+            },
+        ];
+        let snap = build_gpu_snapshot(&samples);
+        assert_eq!(snap.utilization_percent, 87.0);
+        assert_eq!(snap.memory_used_mb, 38200.0);
+        assert_eq!(snap.memory_total_mb, 48000.0);
+        assert_eq!(snap.temperature_celsius, 62.0);
+        assert_eq!(snap.power_watts, 245.0);
+    }
+}