From 9e20337ee4192721fd09900a99b0442d2e5e444f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:35:38 +0100 Subject: [PATCH] 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 --- Cargo.lock | 26 ++ Cargo.toml | 1 + services/monitoring_service/Cargo.toml | 36 +++ services/monitoring_service/build.rs | 10 + services/monitoring_service/src/main.rs | 89 ++++++ .../src/prometheus_client.rs | 118 ++++++++ services/monitoring_service/src/service.rs | 267 ++++++++++++++++++ 7 files changed, 547 insertions(+) create mode 100644 services/monitoring_service/Cargo.toml create mode 100644 services/monitoring_service/build.rs create mode 100644 services/monitoring_service/src/main.rs create mode 100644 services/monitoring_service/src/prometheus_client.rs create mode 100644 services/monitoring_service/src/service.rs diff --git a/Cargo.lock b/Cargo.lock index adc3b8148..8c3e88aa4 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 91fa58600..bd6769651 100644 --- a/Cargo.toml +++ b/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", diff --git a/services/monitoring_service/Cargo.toml b/services/monitoring_service/Cargo.toml new file mode 100644 index 000000000..a431d0b3e --- /dev/null +++ b/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" diff --git a/services/monitoring_service/build.rs b/services/monitoring_service/build.rs new file mode 100644 index 000000000..9bc2ed07d --- /dev/null +++ b/services/monitoring_service/build.rs @@ -0,0 +1,10 @@ +fn main() -> Result<(), Box> { + 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(()) +} diff --git a/services/monitoring_service/src/main.rs b/services/monitoring_service/src/main.rs new file mode 100644 index 000000000..512038b61 --- /dev/null +++ b/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(()) +} diff --git a/services/monitoring_service/src/prometheus_client.rs b/services/monitoring_service/src/prometheus_client.rs new file mode 100644 index 000000000..6c7eb4938 --- /dev/null +++ b/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, + value: (f64, String), // (timestamp, value_string) +} + +#[derive(Debug, Deserialize)] +struct PromData { + result: Vec, +} + +#[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> { + 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> { + 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> { + 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 { + 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::().ok()) + .sum(); + Ok(count as u32) + } +} + +fn parse_samples(results: Vec) -> Vec { + 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() +} diff --git a/services/monitoring_service/src/service.rs b/services/monitoring_service/src/service.rs new file mode 100644 index 000000000..9a56c9d69 --- /dev/null +++ b/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, + 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 { + 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, + ) -> Result, Status> { + let filter = &request.into_inner().model_filter; + let resp = Self::build_response(&self.prom, filter).await?; + Ok(Response::new(resp)) + } + + type StreamTrainingMetricsStream = + Pin> + Send>>; + + async fn stream_training_metrics( + &self, + request: Request, + ) -> Result, 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 { + 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); + } +}