From 354cd5c1165b8cb40ee511d635e57a7c3efe93c2 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:18:50 +0100 Subject: [PATCH 1/9] feat(metrics): add 6 hyperopt Prometheus metrics to training_metrics Co-Authored-By: Claude Opus 4.6 --- crates/common/src/metrics/training_metrics.rs | 109 +++++++++++++++++- 1 file changed, 107 insertions(+), 2 deletions(-) diff --git a/crates/common/src/metrics/training_metrics.rs b/crates/common/src/metrics/training_metrics.rs index fbcd18488..168740625 100644 --- a/crates/common/src/metrics/training_metrics.rs +++ b/crates/common/src/metrics/training_metrics.rs @@ -1,6 +1,6 @@ //! Prometheus metrics for ML training binaries. //! -//! Registers the 18 metrics expected by the Training Cockpit Grafana dashboard +//! Registers the 24 metrics expected by the Training Cockpit Grafana dashboard //! and provides typed helper functions so callers never mis-spell metric names. //! //! # Usage @@ -26,7 +26,7 @@ use super::{ // Registration // --------------------------------------------------------------------------- -/// Register all 18 training metrics with the global Prometheus registry. +/// Register all 24 training metrics with the global Prometheus registry. /// /// Safe to call multiple times — the registry silently ignores duplicates. pub fn init() { @@ -128,6 +128,43 @@ pub fn init() { "foxhunt_training_active_workers", "Number of active training workers", ); + + // Hyperopt gauges (model only) + _ = register_gauge_vec( + "foxhunt_hyperopt_trial_current", + "Current trial number (1-indexed)", + m, + ); + _ = register_gauge_vec( + "foxhunt_hyperopt_trial_total", + "Total trials configured", + m, + ); + _ = register_gauge_vec( + "foxhunt_hyperopt_best_objective", + "Best objective value so far (Sharpe)", + m, + ); + _ = register_gauge_vec( + "foxhunt_hyperopt_mode", + "1=hyperopt active, 0=training only", + m, + ); + + // Hyperopt counters (model only) + _ = register_counter_vec( + "foxhunt_hyperopt_trials_failed_total", + "Failed trials count", + m, + ); + + // Hyperopt histogram (model only) + _ = register_histogram_vec( + "foxhunt_hyperopt_trial_duration_seconds", + "Duration of each trial", + m, + vec![5.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0], + ); } // --------------------------------------------------------------------------- @@ -233,3 +270,71 @@ pub fn record_data_load(model: &str, duration_secs: f64) { pub fn set_active_workers(value: f64) { set_gauge("foxhunt_training_active_workers", value); } + +pub fn set_hyperopt_trial(model: &str, current: f64, total: f64) { + set_gauge_vec("foxhunt_hyperopt_trial_current", &[model], current); + set_gauge_vec("foxhunt_hyperopt_trial_total", &[model], total); +} + +pub fn set_hyperopt_best_objective(model: &str, value: f64) { + set_gauge_vec("foxhunt_hyperopt_best_objective", &[model], value); +} + +pub fn record_hyperopt_trial_duration(model: &str, secs: f64) { + observe_histogram_vec("foxhunt_hyperopt_trial_duration_seconds", &[model], secs); +} + +pub fn record_hyperopt_trial_failure(model: &str) { + increment_counter_vec("foxhunt_hyperopt_trials_failed_total", &[model], 1.0); +} + +pub fn set_hyperopt_mode(model: &str, active: bool) { + set_gauge_vec( + "foxhunt_hyperopt_mode", + &[model], + if active { 1.0 } else { 0.0 }, + ); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_init_is_idempotent() { + init(); + init(); // calling twice must not panic + } + + #[test] + fn test_set_hyperopt_trial() { + init(); + set_hyperopt_trial("dqn", 5.0, 20.0); + // No panic = success (gauge_vec values not directly readable without label matching) + } + + #[test] + fn test_set_hyperopt_mode() { + init(); + set_hyperopt_mode("dqn", true); + set_hyperopt_mode("dqn", false); + } + + #[test] + fn test_record_hyperopt_trial_failure() { + init(); + record_hyperopt_trial_failure("ppo"); + } + + #[test] + fn test_set_hyperopt_best_objective() { + init(); + set_hyperopt_best_objective("tft", 2.34); + } + + #[test] + fn test_record_hyperopt_trial_duration() { + init(); + record_hyperopt_trial_duration("dqn", 45.3); + } +} From 33d3cf2dd2445a84631a5e03e81ff2275d45a78b Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:21:29 +0100 Subject: [PATCH 2/9] feat(monitoring): add monitoring.proto with gRPC service definition Co-Authored-By: Claude Opus 4.6 --- .../monitoring_service/proto/monitoring.proto | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 services/monitoring_service/proto/monitoring.proto diff --git a/services/monitoring_service/proto/monitoring.proto b/services/monitoring_service/proto/monitoring.proto new file mode 100644 index 000000000..c3f90f4ab --- /dev/null +++ b/services/monitoring_service/proto/monitoring.proto @@ -0,0 +1,74 @@ +syntax = "proto3"; +package monitoring; + +service MonitoringService { + // Single snapshot of all active training sessions + rpc GetLiveTrainingMetrics(GetLiveTrainingMetricsRequest) + returns (GetLiveTrainingMetricsResponse); + + // Server-streaming: pushes updates every N seconds + rpc StreamTrainingMetrics(StreamTrainingMetricsRequest) + returns (stream GetLiveTrainingMetricsResponse); +} + +message GetLiveTrainingMetricsRequest { + string model_filter = 1; // optional: "dqn", "ppo", etc. +} + +message StreamTrainingMetricsRequest { + string model_filter = 1; + uint32 interval_seconds = 2; // 0 = server default (3s) +} + +message GetLiveTrainingMetricsResponse { + repeated TrainingSession sessions = 1; + GpuSnapshot gpu = 2; + uint32 active_k8s_jobs = 3; + int64 timestamp = 4; +} + +message TrainingSession { + string model = 1; + string fold = 2; + bool is_hyperopt = 3; + + // Epoch/progress + float current_epoch = 4; + float epoch_loss = 5; + float validation_loss = 6; + + // Throughput + float batches_per_second = 7; + float batches_processed = 8; + float iteration_seconds = 9; + + // Eval metrics + float eval_accuracy = 10; + float eval_precision = 11; + float eval_recall = 12; + float eval_f1 = 13; + + // Checkpoint + float checkpoint_size_bytes = 14; + uint32 checkpoint_saves = 15; + uint32 checkpoint_failures = 16; + + // Health counters + uint32 nan_detected = 17; + uint32 gradient_explosions = 18; + uint32 feature_errors = 19; + + // Hyperopt fields (populated when is_hyperopt=true) + uint32 hyperopt_trial_current = 20; + uint32 hyperopt_trial_total = 21; + float hyperopt_best_objective = 22; + uint32 hyperopt_trials_failed = 23; +} + +message GpuSnapshot { + float utilization_percent = 1; + float memory_used_mb = 2; + float memory_total_mb = 3; + float temperature_celsius = 4; + float power_watts = 5; +} From c4693403d5306182f90e02c23831a8efc00a9a57 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:25:56 +0100 Subject: [PATCH 3/9] feat(ml): wire hyperopt Prometheus metrics into training binaries Co-Authored-By: Claude Opus 4.6 --- crates/ml/examples/hyperopt_baseline_rl.rs | 15 +++++++ .../examples/hyperopt_baseline_supervised.rs | 45 +++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/crates/ml/examples/hyperopt_baseline_rl.rs b/crates/ml/examples/hyperopt_baseline_rl.rs index 52a258dc1..fc5618044 100644 --- a/crates/ml/examples/hyperopt_baseline_rl.rs +++ b/crates/ml/examples/hyperopt_baseline_rl.rs @@ -166,6 +166,8 @@ fn run_dqn_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("dqn", 0.0, args.trials as f64); + let start = Instant::now(); let result = if parallel > 1 { info!("Using parallel optimization ({} threads)", parallel); @@ -179,6 +181,9 @@ fn run_dqn_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) }; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("dqn", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("dqn", result.best_objective); + info!("DQN hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -230,6 +235,8 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("ppo", 0.0, args.trials as f64); + let start = Instant::now(); let result = if parallel > 1 { info!("Using parallel optimization ({} threads)", parallel); @@ -243,6 +250,9 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) }; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("ppo", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("ppo", result.best_objective); + info!("PPO hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -281,6 +291,10 @@ fn main() -> Result<()> { let args = Args::parse(); + // Signal hyperopt mode active — will be cleared at exit + let hyperopt_model_label = args.model.clone(); + training_metrics::set_hyperopt_mode(&hyperopt_model_label, true); + info!("========================================"); info!(" Hyperopt Baseline Runner"); info!("========================================"); @@ -430,6 +444,7 @@ fn main() -> Result<()> { info!("========================================"); info!("{}", output_str); + training_metrics::set_hyperopt_mode(&hyperopt_model_label, false); training_metrics::set_active_workers(0.0); Ok(()) } diff --git a/crates/ml/examples/hyperopt_baseline_supervised.rs b/crates/ml/examples/hyperopt_baseline_supervised.rs index 742338b02..dfff07887 100644 --- a/crates/ml/examples/hyperopt_baseline_supervised.rs +++ b/crates/ml/examples/hyperopt_baseline_supervised.rs @@ -138,12 +138,17 @@ fn run_tft_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("tft", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("TFT hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("tft", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("tft", result.best_objective); + info!("TFT hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -191,12 +196,17 @@ fn run_mamba2_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("mamba2", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("Mamba2 hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("mamba2", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("mamba2", result.best_objective); + info!("Mamba2 hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -245,12 +255,17 @@ fn run_liquid_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("liquid", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("Liquid hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("liquid", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("liquid", result.best_objective); + info!("Liquid hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -299,12 +314,17 @@ fn run_tggn_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("tggn", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("TGGN hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("tggn", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("tggn", result.best_objective); + info!("TGGN hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -353,12 +373,17 @@ fn run_tlob_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("tlob", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("TLOB hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("tlob", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("tlob", result.best_objective); + info!("TLOB hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -407,12 +432,17 @@ fn run_kan_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("kan", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("KAN hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("kan", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("kan", result.best_objective); + info!("KAN hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -461,12 +491,17 @@ fn run_xlstm_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("xlstm", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("xLSTM hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("xlstm", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("xlstm", result.best_objective); + info!("xLSTM hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -515,12 +550,17 @@ fn run_diffusion_hyperopt(args: &Args) -> Result { .seed(args.seed) .build(); + training_metrics::set_hyperopt_trial("diffusion", 0.0, args.trials as f64); + let start = Instant::now(); let result = optimizer .optimize(trainer) .context("Diffusion hyperopt optimization failed")?; let elapsed = start.elapsed().as_secs_f64(); + training_metrics::set_hyperopt_trial("diffusion", result.all_trials.len() as f64, args.trials as f64); + training_metrics::set_hyperopt_best_objective("diffusion", result.best_objective); + info!("Diffusion hyperopt complete:"); info!(" Best objective: {:.6}", result.best_objective); info!(" Total trials: {}", result.all_trials.len()); @@ -559,6 +599,10 @@ fn main() -> Result<()> { let args = Args::parse(); + // Signal hyperopt mode active — will be cleared at exit + let hyperopt_model_label = args.model.clone(); + training_metrics::set_hyperopt_mode(&hyperopt_model_label, true); + info!("========================================"); info!(" Hyperopt Baseline Supervised Runner"); info!("========================================"); @@ -652,6 +696,7 @@ fn main() -> Result<()> { info!("========================================"); info!("{}", output_str); + training_metrics::set_hyperopt_mode(&hyperopt_model_label, false); training_metrics::set_active_workers(0.0); Ok(()) } From 9e20337ee4192721fd09900a99b0442d2e5e444f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:35:38 +0100 Subject: [PATCH 4/9] 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); + } +} From dd14d58fc35a4609c42b3a1a55c613a7fd9369dd Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:38:08 +0100 Subject: [PATCH 5/9] infra: add K8s deployment manifest for monitoring-service Co-Authored-By: Claude Opus 4.6 --- infra/k8s/services/monitoring-service.yaml | 117 +++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100644 infra/k8s/services/monitoring-service.yaml diff --git a/infra/k8s/services/monitoring-service.yaml b/infra/k8s/services/monitoring-service.yaml new file mode 100644 index 000000000..b074adb2d --- /dev/null +++ b/infra/k8s/services/monitoring-service.yaml @@ -0,0 +1,117 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: monitoring-service + namespace: foxhunt + labels: + app.kubernetes.io/name: monitoring-service + app.kubernetes.io/part-of: foxhunt +spec: + replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxSurge: 0 + maxUnavailable: 1 + selector: + matchLabels: + app.kubernetes.io/name: monitoring-service + template: + metadata: + annotations: + gitlab.com/prometheus_scrape: "true" + gitlab.com/prometheus_port: "9099" + gitlab.com/prometheus_path: "/metrics" + labels: + app.kubernetes.io/name: monitoring-service + app.kubernetes.io/part-of: foxhunt + spec: + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + seccompProfile: + type: RuntimeDefault + imagePullSecrets: + - name: scw-registry + nodeSelector: + k8s.scaleway.com/pool-name: foxhunt + containers: + - name: monitoring-service + image: rg.fr-par.scw.cloud/foxhunt-ci/foxhunt-runtime:latest + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + command: ["/binaries/monitoring_service"] + ports: + - containerPort: 50057 + name: grpc + - containerPort: 9099 + name: metrics + env: + - name: PROMETHEUS_URL + value: "http://gitlab-prometheus-server.foxhunt.svc:80" + - name: GRPC_PORT + value: "50057" + - name: METRICS_PORT + value: "9099" + - name: STREAM_INTERVAL_SECS + value: "3" + - name: RUST_LOG + value: info + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: "http://tempo.foxhunt.svc.cluster.local:4317" + volumeMounts: + - name: binaries + mountPath: /binaries + readOnly: true + - name: tmp + mountPath: /tmp + readinessProbe: + tcpSocket: + port: 50057 + initialDelaySeconds: 5 + periodSeconds: 10 + livenessProbe: + tcpSocket: + port: 50057 + initialDelaySeconds: 10 + periodSeconds: 15 + failureThreshold: 5 + resources: + requests: + cpu: 50m + memory: 64Mi + limits: + cpu: 200m + memory: 128Mi + volumes: + - name: binaries + persistentVolumeClaim: + claimName: foxhunt-binaries + readOnly: true + - name: tmp + emptyDir: + sizeLimit: 10Mi +--- +apiVersion: v1 +kind: Service +metadata: + name: monitoring-service + namespace: foxhunt + labels: + app.kubernetes.io/name: monitoring-service + app.kubernetes.io/part-of: foxhunt +spec: + selector: + app.kubernetes.io/name: monitoring-service + ports: + - port: 50057 + targetPort: 50057 + name: grpc + - port: 9099 + targetPort: 9099 + name: metrics From 06cf6bf39e5da049077e8131ab1d6ae1a2ce9026 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:41:54 +0100 Subject: [PATCH 6/9] feat(fxt): add 'fxt train monitor' subcommand with live TUI Adds a new CLI subcommand that connects to the monitoring service via gRPC to display live training metrics. Supports single-snapshot mode (--once), model filtering (--model), and configurable streaming interval. Renders per-model epoch/loss table, GPU telemetry, hyperopt progress, and health counters. Co-Authored-By: Claude Opus 4.6 --- bin/fxt/build.rs | 2 + bin/fxt/proto/monitoring.proto | 74 +++++++++++ bin/fxt/src/commands/train/mod.rs | 23 ++++ bin/fxt/src/commands/train/monitor.rs | 173 ++++++++++++++++++++++++++ bin/fxt/src/lib.rs | 9 ++ 5 files changed, 281 insertions(+) create mode 100644 bin/fxt/proto/monitoring.proto create mode 100644 bin/fxt/src/commands/train/monitor.rs diff --git a/bin/fxt/build.rs b/bin/fxt/build.rs index c79fb51af..f46baa51d 100644 --- a/bin/fxt/build.rs +++ b/bin/fxt/build.rs @@ -22,6 +22,7 @@ fn main() -> Result<(), Box> { "proto/ml_training.proto", "proto/trading_agent.proto", "proto/broker_gateway.proto", + "proto/monitoring.proto", ], &["proto"], )?; @@ -34,6 +35,7 @@ fn main() -> Result<(), Box> { println!("cargo:rerun-if-changed=proto/ml_training.proto"); println!("cargo:rerun-if-changed=proto/trading_agent.proto"); println!("cargo:rerun-if-changed=proto/broker_gateway.proto"); + println!("cargo:rerun-if-changed=proto/monitoring.proto"); Ok(()) } diff --git a/bin/fxt/proto/monitoring.proto b/bin/fxt/proto/monitoring.proto new file mode 100644 index 000000000..c3f90f4ab --- /dev/null +++ b/bin/fxt/proto/monitoring.proto @@ -0,0 +1,74 @@ +syntax = "proto3"; +package monitoring; + +service MonitoringService { + // Single snapshot of all active training sessions + rpc GetLiveTrainingMetrics(GetLiveTrainingMetricsRequest) + returns (GetLiveTrainingMetricsResponse); + + // Server-streaming: pushes updates every N seconds + rpc StreamTrainingMetrics(StreamTrainingMetricsRequest) + returns (stream GetLiveTrainingMetricsResponse); +} + +message GetLiveTrainingMetricsRequest { + string model_filter = 1; // optional: "dqn", "ppo", etc. +} + +message StreamTrainingMetricsRequest { + string model_filter = 1; + uint32 interval_seconds = 2; // 0 = server default (3s) +} + +message GetLiveTrainingMetricsResponse { + repeated TrainingSession sessions = 1; + GpuSnapshot gpu = 2; + uint32 active_k8s_jobs = 3; + int64 timestamp = 4; +} + +message TrainingSession { + string model = 1; + string fold = 2; + bool is_hyperopt = 3; + + // Epoch/progress + float current_epoch = 4; + float epoch_loss = 5; + float validation_loss = 6; + + // Throughput + float batches_per_second = 7; + float batches_processed = 8; + float iteration_seconds = 9; + + // Eval metrics + float eval_accuracy = 10; + float eval_precision = 11; + float eval_recall = 12; + float eval_f1 = 13; + + // Checkpoint + float checkpoint_size_bytes = 14; + uint32 checkpoint_saves = 15; + uint32 checkpoint_failures = 16; + + // Health counters + uint32 nan_detected = 17; + uint32 gradient_explosions = 18; + uint32 feature_errors = 19; + + // Hyperopt fields (populated when is_hyperopt=true) + uint32 hyperopt_trial_current = 20; + uint32 hyperopt_trial_total = 21; + float hyperopt_best_objective = 22; + uint32 hyperopt_trials_failed = 23; +} + +message GpuSnapshot { + float utilization_percent = 1; + float memory_used_mb = 2; + float memory_total_mb = 3; + float temperature_celsius = 4; + float power_watts = 5; +} diff --git a/bin/fxt/src/commands/train/mod.rs b/bin/fxt/src/commands/train/mod.rs index a45791fdf..fcc2a863d 100644 --- a/bin/fxt/src/commands/train/mod.rs +++ b/bin/fxt/src/commands/train/mod.rs @@ -7,6 +7,7 @@ //! - `status` - Query individual training job status (stub for Wave 2 Agent 5) pub mod list; +pub mod monitor; pub mod progress_tracker; pub mod start; pub mod status; @@ -15,6 +16,7 @@ pub mod status; pub use progress_tracker::{JobProgress, JobStatus, ProgressTracker}; pub use list::ListCommand; +pub use monitor::MonitorCommand; pub use start::StartCommand; pub use status::StatusCommand; @@ -54,6 +56,24 @@ pub enum TrainCommand { #[clap(flatten)] status_args: StatusCommand, }, + + /// Live training metrics monitor (Prometheus via gRPC) + #[clap( + long_about = "Stream live training metrics from the monitoring service.\n\n\ + Shows:\n\ + - Per-model epoch, loss, validation loss, throughput\n\ + - GPU utilization, VRAM, temperature, power\n\ + - Hyperopt trial progress and best objective\n\ + - Health counters (NaN, gradient explosions, checkpoints)\n\n\ + Examples:\n\ + fxt train monitor\n\ + fxt train monitor --once\n\ + fxt train monitor --model dqn --interval 5" + )] + Monitor { + #[clap(flatten)] + monitor_args: MonitorCommand, + }, } /// Execute train command @@ -66,5 +86,8 @@ pub async fn execute_train_command( TrainCommand::List { list_args } => list_args.run(api_gateway_url, jwt_token).await, TrainCommand::Start { start_args } => start_args.run(api_gateway_url, jwt_token).await, TrainCommand::Status { status_args } => status_args.run(api_gateway_url, jwt_token).await, + TrainCommand::Monitor { monitor_args } => { + monitor_args.run(api_gateway_url, jwt_token).await + } } } diff --git a/bin/fxt/src/commands/train/monitor.rs b/bin/fxt/src/commands/train/monitor.rs new file mode 100644 index 000000000..94496fbac --- /dev/null +++ b/bin/fxt/src/commands/train/monitor.rs @@ -0,0 +1,173 @@ +//! fxt train monitor -- live training metrics from monitoring service +//! +//! Connects to the MonitoringService gRPC endpoint and displays a live TUI +//! with per-model epoch progress, GPU telemetry, and health counters. +//! Use `--once` for a single snapshot instead of continuous streaming. + +use anyhow::{Context, Result}; +use clap::Args; +use colored::Colorize; +use std::io::{self, Write}; + +use crate::proto::monitoring::{ + monitoring_service_client::MonitoringServiceClient, GetLiveTrainingMetricsRequest, + GetLiveTrainingMetricsResponse, StreamTrainingMetricsRequest, TrainingSession, +}; + +/// Live training metrics monitor +#[derive(Args, Debug)] +pub struct MonitorCommand { + /// Print a single snapshot and exit (no live TUI) + #[arg(long)] + pub once: bool, + + /// Filter to a specific model (e.g. "dqn", "ppo", "tft") + #[arg(long)] + pub model: Option, + + /// Streaming interval in seconds (default: 3) + #[arg(long, default_value = "3")] + pub interval: u32, +} + +impl MonitorCommand { + /// Execute the monitor command -- connect to monitoring service and display metrics + pub async fn run(&self, api_gateway_url: &str, _jwt_token: &str) -> Result<()> { + // monitoring_service runs on a different port -- derive URL from env or default + let monitoring_url = std::env::var("MONITORING_SERVICE_URL") + .unwrap_or_else(|_| api_gateway_url.replace(":50050", ":50057")); + + let mut client = MonitoringServiceClient::connect(monitoring_url) + .await + .context("Failed to connect to monitoring service")?; + + let model_filter = self.model.clone().unwrap_or_default(); + + if self.once { + let response = client + .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest { + model_filter, + })) + .await + .context("GetLiveTrainingMetrics failed")?; + render_snapshot(&response.into_inner()); + } else { + println!("Connecting to monitoring service..."); + let response = client + .stream_training_metrics(tonic::Request::new(StreamTrainingMetricsRequest { + model_filter, + interval_seconds: self.interval, + })) + .await + .context("StreamTrainingMetrics failed")?; + + let mut stream = response.into_inner(); + while let Some(msg) = stream.message().await? { + render_tui(&msg); + } + } + + Ok(()) + } +} + +fn render_snapshot(resp: &GetLiveTrainingMetricsResponse) { + if resp.sessions.is_empty() { + println!("No active training sessions found."); + return; + } + + if let Some(gpu) = &resp.gpu { + println!( + "GPU: {:.0}% util | {:.1}/{:.1} GB VRAM | {:.0}C | {:.0}W", + gpu.utilization_percent, + gpu.memory_used_mb / 1024.0, + gpu.memory_total_mb / 1024.0, + gpu.temperature_celsius, + gpu.power_watts, + ); + } + println!("Active K8s training jobs: {}", resp.active_k8s_jobs); + println!(); + + print_session_table(&resp.sessions); + print_hyperopt_summary(&resp.sessions); + print_health_summary(&resp.sessions); +} + +fn render_tui(resp: &GetLiveTrainingMetricsResponse) { + // Clear screen + print!("\x1B[2J\x1B[H"); + // Intentionally ignoring flush errors on stdout (terminal output, non-critical) + drop(io::stdout().flush()); + + println!("{}", "=".repeat(72).bright_black()); + println!( + "{}", + " Foxhunt Training Monitor" + .bright_white() + .bold() + ); + println!("{}", "=".repeat(72).bright_black()); + println!(); + + render_snapshot(resp); + + println!(); + println!("{}", "Ctrl+C to exit".bright_black()); +} + +fn print_session_table(sessions: &[TrainingSession]) { + println!( + "{:<10} {:<6} {:<7} {:<10} {:<10} {:<9} {:<10}", + "Model".bright_cyan(), + "Fold".bright_cyan(), + "Epoch".bright_cyan(), + "Loss".bright_cyan(), + "Val Loss".bright_cyan(), + "Batch/s".bright_cyan(), + "Eval Acc".bright_cyan(), + ); + println!("{}", "-".repeat(72).bright_black()); + + for s in sessions { + let acc = if s.eval_accuracy > 0.0 { + format!("{:.1}%", s.eval_accuracy * 100.0) + } else { + "-".to_owned() + }; + println!( + "{:<10} {:<6} {:<7.0} {:<10.4} {:<10.4} {:<9.1} {:<10}", + s.model, s.fold, s.current_epoch, s.epoch_loss, s.validation_loss, + s.batches_per_second, acc, + ); + } +} + +fn print_hyperopt_summary(sessions: &[TrainingSession]) { + let hyperopt: Vec<_> = sessions.iter().filter(|s| s.is_hyperopt).collect(); + if hyperopt.is_empty() { + return; + } + println!(); + for s in &hyperopt { + println!( + "Hyperopt ({}): trial {}/{} | best Sharpe {:.2} | {} failures", + s.model.bright_magenta(), + s.hyperopt_trial_current, + s.hyperopt_trial_total, + s.hyperopt_best_objective, + s.hyperopt_trials_failed, + ); + } +} + +fn print_health_summary(sessions: &[TrainingSession]) { + let total_nan: u32 = sessions.iter().map(|s| s.nan_detected).sum(); + let total_grad: u32 = sessions.iter().map(|s| s.gradient_explosions).sum(); + let total_ckpt: u32 = sessions.iter().map(|s| s.checkpoint_saves).sum(); + println!( + "Health: {} NaN | {} grad explosions | {} checkpoints", + total_nan, total_grad, total_ckpt + ); +} diff --git a/bin/fxt/src/lib.rs b/bin/fxt/src/lib.rs index aa427a1eb..8e9244650 100644 --- a/bin/fxt/src/lib.rs +++ b/bin/fxt/src/lib.rs @@ -247,4 +247,13 @@ pub mod proto { pub mod broker_gateway { tonic::include_proto!("broker_gateway"); } + + /// Monitoring service protobuf definitions + /// + /// Contains message types and service interfaces for live training metrics, + /// GPU telemetry, and streaming monitoring data from Prometheus. + #[allow(clippy::shadow_reuse, clippy::impl_trait_in_params, clippy::empty_structs_with_brackets)] + pub mod monitoring { + tonic::include_proto!("monitoring"); + } } From 349174992779484d95a36f0f167cacaee07de109 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:50:19 +0100 Subject: [PATCH 7/9] feat(web-gateway): integrate monitoring service for live training metrics Wire monitoring_service gRPC into web-gateway: proto compilation, config/state/client plumbing, REST endpoint at /api/monitoring/live-metrics, and a 3s polling bridge that broadcasts training_progress over WebSocket. Co-Authored-By: Claude Opus 4.6 --- crates/web-gateway/build.rs | 2 + crates/web-gateway/src/config.rs | 15 ++++++ crates/web-gateway/src/grpc/clients.rs | 6 +++ crates/web-gateway/src/grpc/streams.rs | 55 ++++++++++++++++++++ crates/web-gateway/src/lib.rs | 10 ++++ crates/web-gateway/src/main.rs | 2 + crates/web-gateway/src/routes/auth.rs | 1 + crates/web-gateway/src/routes/backtesting.rs | 1 + crates/web-gateway/src/routes/config.rs | 1 + crates/web-gateway/src/routes/ml.rs | 1 + crates/web-gateway/src/routes/mod.rs | 2 + crates/web-gateway/src/routes/monitoring.rs | 28 ++++++++++ crates/web-gateway/src/routes/performance.rs | 1 + crates/web-gateway/src/routes/risk.rs | 1 + crates/web-gateway/src/routes/trading.rs | 1 + crates/web-gateway/src/routes/training.rs | 1 + crates/web-gateway/src/routes/tune.rs | 1 + crates/web-gateway/src/state.rs | 7 +++ 18 files changed, 136 insertions(+) create mode 100644 crates/web-gateway/src/routes/monitoring.rs diff --git a/crates/web-gateway/build.rs b/crates/web-gateway/build.rs index 0e97abb20..29ae527db 100644 --- a/crates/web-gateway/build.rs +++ b/crates/web-gateway/build.rs @@ -13,6 +13,7 @@ fn main() -> Result<(), Box> { "../../bin/fxt/proto/config.proto", "../../bin/fxt/proto/ml_training.proto", "../../bin/fxt/proto/trading_agent.proto", + "../../bin/fxt/proto/monitoring.proto", ], &["../../bin/fxt/proto"], )?; @@ -23,6 +24,7 @@ fn main() -> Result<(), Box> { println!("cargo:rerun-if-changed=../../bin/fxt/proto/config.proto"); println!("cargo:rerun-if-changed=../../bin/fxt/proto/ml_training.proto"); println!("cargo:rerun-if-changed=../../bin/fxt/proto/trading_agent.proto"); + println!("cargo:rerun-if-changed=../../bin/fxt/proto/monitoring.proto"); Ok(()) } diff --git a/crates/web-gateway/src/config.rs b/crates/web-gateway/src/config.rs index d0381c1a3..67a4f8822 100644 --- a/crates/web-gateway/src/config.rs +++ b/crates/web-gateway/src/config.rs @@ -10,6 +10,8 @@ pub struct GatewayConfig { pub backtesting_service_url: String, /// gRPC ML training service endpoint pub ml_training_service_url: String, + /// gRPC monitoring service endpoint + pub monitoring_service_url: String, /// CORS allowed origins (comma-separated) pub cors_origins: Vec, /// JWT secret for token validation @@ -23,6 +25,7 @@ impl Default for GatewayConfig { trading_service_url: "https://localhost:50051".to_owned(), backtesting_service_url: "https://localhost:50052".to_owned(), ml_training_service_url: "https://localhost:50053".to_owned(), + monitoring_service_url: "https://localhost:50057".to_owned(), cors_origins: vec!["http://localhost:5173".to_owned()], jwt_secret: String::new(), } @@ -40,6 +43,8 @@ impl GatewayConfig { .unwrap_or_else(|_| "https://localhost:50052".to_owned()), ml_training_service_url: std::env::var("ML_TRAINING_SERVICE_URL") .unwrap_or_else(|_| "https://localhost:50053".to_owned()), + monitoring_service_url: std::env::var("MONITORING_SERVICE_URL") + .unwrap_or_else(|_| "https://localhost:50057".to_owned()), cors_origins: std::env::var("CORS_ORIGINS") .unwrap_or_else(|_| "http://localhost:5173".to_owned()) .split(',') @@ -78,6 +83,12 @@ mod tests { assert_eq!(cfg.ml_training_service_url, "https://localhost:50053"); } + #[test] + fn test_default_monitoring_service_url() { + let cfg = GatewayConfig::default(); + assert_eq!(cfg.monitoring_service_url, "https://localhost:50057"); + } + #[test] fn test_default_cors_origins() { let cfg = GatewayConfig::default(); @@ -105,6 +116,10 @@ mod tests { from_env.ml_training_service_url, default.ml_training_service_url ); + assert_eq!( + from_env.monitoring_service_url, + default.monitoring_service_url + ); assert_eq!(from_env.cors_origins, default.cors_origins); } } diff --git a/crates/web-gateway/src/grpc/clients.rs b/crates/web-gateway/src/grpc/clients.rs index 49a98fac7..d3c247b00 100644 --- a/crates/web-gateway/src/grpc/clients.rs +++ b/crates/web-gateway/src/grpc/clients.rs @@ -1,6 +1,7 @@ use tonic::transport::Channel; use crate::proto::ml_training::ml_training_service_client::MlTrainingServiceClient; +use crate::proto::monitoring::monitoring_service_client::MonitoringServiceClient; use crate::proto::trading::backtesting_service_client::BacktestingServiceClient; use crate::proto::trading::trading_service_client::TradingServiceClient; @@ -18,3 +19,8 @@ pub fn backtesting_client(channel: &Channel) -> BacktestingServiceClient MlTrainingServiceClient { MlTrainingServiceClient::new(channel.clone()) } + +/// Create a `MonitoringServiceClient` from the shared channel +pub fn monitoring_client(channel: &Channel) -> MonitoringServiceClient { + MonitoringServiceClient::new(channel.clone()) +} diff --git a/crates/web-gateway/src/grpc/streams.rs b/crates/web-gateway/src/grpc/streams.rs index e82d24ec2..a981321bf 100644 --- a/crates/web-gateway/src/grpc/streams.rs +++ b/crates/web-gateway/src/grpc/streams.rs @@ -19,6 +19,7 @@ use crate::ws::messages::ServerMessage; /// on failure with exponential backoff. pub fn start_grpc_stream_bridges( trading_channel: Option, + monitoring_channel: Option, ws_broadcast: broadcast::Sender, jwt_secret: String, ) { @@ -55,11 +56,20 @@ pub fn start_grpc_stream_bridges( }); } + // Clone for training metrics poller before heartbeat takes ownership + let tx_training_progress = ws_broadcast.clone(); + // Heartbeat for WebSocket connectivity verification let tx_heartbeat = ws_broadcast; tokio::spawn(async move { heartbeat_loop(tx_heartbeat).await; }); + + if let Some(channel) = monitoring_channel { + tokio::spawn(async move { + poll_training_metrics(channel, tx_training_progress).await; + }); + } } /// Bridge `SubscribeMarketData` gRPC stream to WebSocket broadcast @@ -193,6 +203,51 @@ async fn heartbeat_loop(tx: broadcast::Sender) { } } +/// Poll monitoring_service every 3s and broadcast training_progress to WebSocket clients +#[allow(clippy::infinite_loop)] +async fn poll_training_metrics(channel: Channel, tx: broadcast::Sender) { + use crate::grpc::clients::monitoring_client; + use crate::proto::monitoring::GetLiveTrainingMetricsRequest; + + let mut backoff = Duration::from_secs(1); + let max_backoff = Duration::from_secs(60); + + loop { + let mut client = monitoring_client(&channel); + match client + .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest { + model_filter: String::new(), + })) + .await + { + Ok(resp) => { + backoff = Duration::from_secs(1); + let data = serde_json::to_value(resp.into_inner()).unwrap_or_default(); + let msg = ServerMessage::TrainingProgress { data }; + if let Ok(json) = serde_json::to_string(&msg) { + drop(tx.send(json)); + } + } + Err(e) => { + if is_unimplemented(&e) { + info!( + "Monitoring service not available, disabling training_progress poller" + ); + return; + } + warn!( + "Training metrics poll failed: {}, retrying in {:?}", + e, backoff + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(max_backoff); + continue; + } + } + tokio::time::sleep(Duration::from_secs(3)).await; + } +} + /// Check whether a gRPC error indicates the endpoint is not implemented. /// Matches both the standard Unimplemented status code and error messages /// containing "not implemented" (which proxies may return with a different code). diff --git a/crates/web-gateway/src/lib.rs b/crates/web-gateway/src/lib.rs index ac7f3da28..53d2cca24 100644 --- a/crates/web-gateway/src/lib.rs +++ b/crates/web-gateway/src/lib.rs @@ -46,4 +46,14 @@ pub mod proto { )] tonic::include_proto!("grpc.health.v1"); } + + pub mod monitoring { + #![allow( + clippy::all, + clippy::pedantic, + clippy::restriction, + clippy::nursery + )] + tonic::include_proto!("monitoring"); + } } diff --git a/crates/web-gateway/src/main.rs b/crates/web-gateway/src/main.rs index 94b3bf8fc..bce0ecfb1 100644 --- a/crates/web-gateway/src/main.rs +++ b/crates/web-gateway/src/main.rs @@ -44,6 +44,7 @@ async fn main() -> Result<()> { info!(" Trading service: {}", config.trading_service_url); info!(" Backtesting service: {}", config.backtesting_service_url); info!(" ML Training service: {}", config.ml_training_service_url); + info!(" Monitoring service: {}", config.monitoring_service_url); info!(" CORS origins: {:?}", config.cors_origins); let state = AppState::new(config.clone()).await?; @@ -51,6 +52,7 @@ async fn main() -> Result<()> { // Start gRPC stream bridge tasks (forward gRPC streams to WebSocket broadcast) start_grpc_stream_bridges( state.trading_channel.clone(), + state.monitoring_channel.clone(), state.ws_broadcast.clone(), config.jwt_secret.clone(), ); diff --git a/crates/web-gateway/src/routes/auth.rs b/crates/web-gateway/src/routes/auth.rs index 33d5b107c..2734adfbb 100644 --- a/crates/web-gateway/src/routes/auth.rs +++ b/crates/web-gateway/src/routes/auth.rs @@ -101,6 +101,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, }; Router::new() diff --git a/crates/web-gateway/src/routes/backtesting.rs b/crates/web-gateway/src/routes/backtesting.rs index f10906f16..ce8c88a8e 100644 --- a/crates/web-gateway/src/routes/backtesting.rs +++ b/crates/web-gateway/src/routes/backtesting.rs @@ -172,6 +172,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/config.rs b/crates/web-gateway/src/routes/config.rs index f1e2b737c..f25a94f97 100644 --- a/crates/web-gateway/src/routes/config.rs +++ b/crates/web-gateway/src/routes/config.rs @@ -91,6 +91,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/ml.rs b/crates/web-gateway/src/routes/ml.rs index a9320b810..83bf1efea 100644 --- a/crates/web-gateway/src/routes/ml.rs +++ b/crates/web-gateway/src/routes/ml.rs @@ -125,6 +125,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/mod.rs b/crates/web-gateway/src/routes/mod.rs index dc4bbfcea..eb86a9ed6 100644 --- a/crates/web-gateway/src/routes/mod.rs +++ b/crates/web-gateway/src/routes/mod.rs @@ -16,6 +16,7 @@ pub mod auth; pub mod backtesting; pub mod config; pub mod ml; +pub mod monitoring; pub mod performance; pub mod risk; pub mod trading; @@ -48,6 +49,7 @@ pub fn create_router(state: AppState) -> Router { // Compute-heavy routes (strict rate limit) let compute_routes = Router::new() .nest("/training", training::router()) + .nest("/monitoring", monitoring::router()) .nest("/backtest", backtesting::router()) .nest("/tune", tune::router()) .layer(middleware::from_fn_with_state(compute_limiter, rate_limit_middleware)); diff --git a/crates/web-gateway/src/routes/monitoring.rs b/crates/web-gateway/src/routes/monitoring.rs new file mode 100644 index 000000000..75a491d8c --- /dev/null +++ b/crates/web-gateway/src/routes/monitoring.rs @@ -0,0 +1,28 @@ +use axum::{extract::State, routing::get, Json, Router}; + +use crate::error::AppError; +use crate::grpc::clients::monitoring_client; +use crate::proto::monitoring::GetLiveTrainingMetricsRequest; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new().route("/live-metrics", get(live_metrics)) +} + +async fn live_metrics( + State(state): State, +) -> Result, AppError> { + let channel = state + .monitoring_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Monitoring service not configured")))?; + let mut client = monitoring_client(channel); + let response = client + .get_live_training_metrics(tonic::Request::new(GetLiveTrainingMetricsRequest { + model_filter: String::new(), + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/crates/web-gateway/src/routes/performance.rs b/crates/web-gateway/src/routes/performance.rs index a01aeeec9..590435acf 100644 --- a/crates/web-gateway/src/routes/performance.rs +++ b/crates/web-gateway/src/routes/performance.rs @@ -113,6 +113,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/risk.rs b/crates/web-gateway/src/routes/risk.rs index 61b9c941d..3086df976 100644 --- a/crates/web-gateway/src/routes/risk.rs +++ b/crates/web-gateway/src/routes/risk.rs @@ -134,6 +134,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/trading.rs b/crates/web-gateway/src/routes/trading.rs index f691e9443..5258f99be 100644 --- a/crates/web-gateway/src/routes/trading.rs +++ b/crates/web-gateway/src/routes/trading.rs @@ -264,6 +264,7 @@ mod tests { trading_channel: channel, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/training.rs b/crates/web-gateway/src/routes/training.rs index beebace0a..0fa33188f 100644 --- a/crates/web-gateway/src/routes/training.rs +++ b/crates/web-gateway/src/routes/training.rs @@ -153,6 +153,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/routes/tune.rs b/crates/web-gateway/src/routes/tune.rs index aa975d1a4..c86de18ee 100644 --- a/crates/web-gateway/src/routes/tune.rs +++ b/crates/web-gateway/src/routes/tune.rs @@ -193,6 +193,7 @@ mod tests { trading_channel: None, backtesting_channel: None, ml_training_channel: None, + monitoring_channel: None, ws_broadcast, } } diff --git a/crates/web-gateway/src/state.rs b/crates/web-gateway/src/state.rs index 685a4602d..ae20a466d 100644 --- a/crates/web-gateway/src/state.rs +++ b/crates/web-gateway/src/state.rs @@ -15,6 +15,7 @@ pub struct AppState { pub trading_channel: Option, pub backtesting_channel: Option, pub ml_training_channel: Option, + pub monitoring_channel: Option, /// Broadcast sender for WebSocket events pub ws_broadcast: broadcast::Sender, } @@ -36,11 +37,16 @@ impl AppState { .ok() .map(|c| c.timeout(GRPC_TIMEOUT).connect_lazy()); + let monitoring_channel = Channel::from_shared(config.monitoring_service_url.clone()) + .ok() + .map(|c| c.timeout(GRPC_TIMEOUT).connect_lazy()); + Ok(Self { config: Arc::new(config), trading_channel, backtesting_channel, ml_training_channel, + monitoring_channel, ws_broadcast, }) } @@ -58,6 +64,7 @@ mod tests { assert!(state.trading_channel.is_some()); assert!(state.backtesting_channel.is_some()); assert!(state.ml_training_channel.is_some()); + assert!(state.monitoring_channel.is_some()); } #[tokio::test] From 928d228db85e0cfc007214990f9f6ad9ca478d1a Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 21:52:28 +0100 Subject: [PATCH 8/9] ci: add monitoring_service to compile-services job Co-Authored-By: Claude Opus 4.6 --- .gitlab-ci.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 52e5e5af0..a3ea1730c 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -434,10 +434,11 @@ compile-services: -p trading_agent_service -p data_acquisition_service -p web-gateway + -p monitoring_service - sccache --show-stats || true - mkdir -p build-out/services - | - for bin in trading_service api_gateway broker_gateway_service ml_training_service backtesting_service trading_agent_service data_acquisition_service web-gateway; do + for bin in trading_service api_gateway broker_gateway_service ml_training_service backtesting_service trading_agent_service data_acquisition_service web-gateway monitoring_service; do cp $TARGET_DIR/$bin build-out/services/ strip build-out/services/$bin done From 1a62060daba327f2ee340204b5820c3da0e90336 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 22:06:25 +0100 Subject: [PATCH 9/9] fix(clippy): resolve all workspace clippy warnings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix pre-existing and new clippy issues across 5 files: - service_auth.rs: .to_string() → .to_owned() on &str, backtick doc items - streams.rs: backtick doc items, allow cognitive_complexity on reconnect_loop - main.rs: .to_owned(), allow infinite_loop, drop must_use, rename shadow - start.rs: eliminate shadow_reuse on endpoint binding - enhanced_ml.rs: remove unnecessary f64 cast Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/commands/train/start.rs | 12 +++++------- crates/web-gateway/src/grpc/service_auth.rs | 14 +++++++------- crates/web-gateway/src/grpc/streams.rs | 3 ++- crates/web-gateway/src/main.rs | 9 +++++---- .../trading_service/src/services/enhanced_ml.rs | 2 +- 5 files changed, 20 insertions(+), 20 deletions(-) diff --git a/bin/fxt/src/commands/train/start.rs b/bin/fxt/src/commands/train/start.rs index bb156b779..b229a7447 100644 --- a/bin/fxt/src/commands/train/start.rs +++ b/bin/fxt/src/commands/train/start.rs @@ -67,15 +67,13 @@ impl StartCommand { ); // Connect to API Gateway - let endpoint = Channel::from_shared(api_gateway_url.to_owned()) + let mut endpoint = Channel::from_shared(api_gateway_url.to_owned()) .context("Invalid API Gateway URL")?; - let endpoint = if api_gateway_url.starts_with("https://") { - endpoint + if api_gateway_url.starts_with("https://") { + endpoint = endpoint .tls_config(tonic::transport::ClientTlsConfig::new().with_enabled_roots()) - .context("Failed to configure TLS")? - } else { - endpoint - }; + .context("Failed to configure TLS")?; + } let channel = endpoint .connect() .await diff --git a/crates/web-gateway/src/grpc/service_auth.rs b/crates/web-gateway/src/grpc/service_auth.rs index 0cded3105..3a5186ac3 100644 --- a/crates/web-gateway/src/grpc/service_auth.rs +++ b/crates/web-gateway/src/grpc/service_auth.rs @@ -2,7 +2,7 @@ //! //! When web-gateway proxies HTTP requests to the api-gateway via gRPC, the //! api-gateway requires a valid JWT Bearer token. This module generates -//! short-lived service tokens using the shared JWT_SECRET so internal +//! short-lived service tokens using the shared `JWT_SECRET` so internal //! gRPC calls are authenticated without forwarding user credentials. use std::time::{SystemTime, UNIX_EPOCH}; @@ -39,15 +39,15 @@ fn service_bearer_value( let claims = ServiceClaims { jti: uuid::Uuid::new_v4().to_string(), - sub: "web-gateway-service".to_string(), + sub: "web-gateway-service".to_owned(), iat: now, exp: now + 3599, // just under 1h (api-gateway enforces max 1h token age) nbf: now, - iss: "foxhunt-api-gateway".to_string(), - aud: "foxhunt-services".to_string(), - roles: vec!["service".to_string()], - permissions: vec!["api.access".to_string()], - token_type: "access".to_string(), + iss: "foxhunt-api-gateway".to_owned(), + aud: "foxhunt-services".to_owned(), + roles: vec!["service".to_owned()], + permissions: vec!["api.access".to_owned()], + token_type: "access".to_owned(), session_id: uuid::Uuid::new_v4().to_string(), }; diff --git a/crates/web-gateway/src/grpc/streams.rs b/crates/web-gateway/src/grpc/streams.rs index a981321bf..afdc4e6dd 100644 --- a/crates/web-gateway/src/grpc/streams.rs +++ b/crates/web-gateway/src/grpc/streams.rs @@ -203,7 +203,7 @@ async fn heartbeat_loop(tx: broadcast::Sender) { } } -/// Poll monitoring_service every 3s and broadcast training_progress to WebSocket clients +/// Poll `monitoring_service` every 3s and broadcast `training_progress` to WebSocket clients #[allow(clippy::infinite_loop)] async fn poll_training_metrics(channel: Channel, tx: broadcast::Sender) { use crate::grpc::clients::monitoring_client; @@ -261,6 +261,7 @@ fn is_unimplemented(e: &tonic::Status) -> bool { /// Reconnect wrapper with exponential backoff for a gRPC stream bridge task. /// Stops retrying if the server returns Unimplemented (endpoint doesn't exist). +#[allow(clippy::cognitive_complexity)] async fn reconnect_loop(stream_name: &str, connect_fn: F) where F: Fn() -> Fut, diff --git a/crates/web-gateway/src/main.rs b/crates/web-gateway/src/main.rs index bce0ecfb1..529523c82 100644 --- a/crates/web-gateway/src/main.rs +++ b/crates/web-gateway/src/main.rs @@ -23,7 +23,7 @@ pub struct RequestId(pub String); async fn main() -> Result<()> { // Initialize observability (JSON logging + OpenTelemetry tracing via OTLP) let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") - .unwrap_or_else(|_| "http://localhost:4317".to_string()); + .unwrap_or_else(|_| "http://localhost:4317".to_owned()); if let Err(e) = common::observability::init_observability("web_gateway", Some(&otlp_endpoint)) { eprintln!("Failed to initialize observability: {}", e); } @@ -82,6 +82,7 @@ async fn main() -> Result<()> { let service_start = std::time::Instant::now(); // Spawn uptime updater + #[allow(clippy::infinite_loop)] tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); loop { @@ -103,11 +104,11 @@ async fn main() -> Result<()> { let encoder = TextEncoder::new(); let metric_families = prometheus::gather(); let mut buffer = vec![]; - let _ = encoder.encode(&metric_families, &mut buffer); + drop(encoder.encode(&metric_families, &mut buffer)); String::from_utf8(buffer).unwrap_or_else(|_| String::new()) } - let app = Router::new().route("/metrics", get(metrics_handler)); + let metrics_app = Router::new().route("/metrics", get(metrics_handler)); let addr = format!("0.0.0.0:{}", metrics_port); tracing::info!("Prometheus metrics endpoint listening on http://{}", addr); @@ -118,7 +119,7 @@ async fn main() -> Result<()> { return; } }; - if let Err(e) = axum::serve(metrics_listener, app).await { + if let Err(e) = axum::serve(metrics_listener, metrics_app).await { tracing::error!("Metrics server failed: {}", e); } }); diff --git a/services/trading_service/src/services/enhanced_ml.rs b/services/trading_service/src/services/enhanced_ml.rs index a35464375..68ebba3fd 100644 --- a/services/trading_service/src/services/enhanced_ml.rs +++ b/services/trading_service/src/services/enhanced_ml.rs @@ -1741,7 +1741,7 @@ impl MLModel for RealPPOModel { // Convert factored action exposure to prediction value (0.0-1.0 range) // target_exposure() returns -1.0..+1.0, map to 0.0..1.0 - let prediction_value = (action.target_exposure() as f64 + 1.0) / 2.0; + let prediction_value = (action.target_exposure() + 1.0) / 2.0; // Confidence from log probability (higher log_prob = higher confidence) // log_prob is negative, so we transform it: confidence = exp(log_prob)