gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(monitoring): add monitoring service for live training metrics over gRPC

Browse Source 
New monitoring_service bridges Prometheus training/GPU metrics to gRPC,
enabling `fxt train monitor` TUI and web-dashboard live updates without
direct Prometheus dependency on clients.

Components:
- 6 new hyperopt Prometheus metrics (common::metrics::training_metrics)
- monitoring_service: Prometheus-to-gRPC bridge (port 50057)
- fxt train monitor: live TUI + --once snapshot mode
- web-gateway: REST endpoint + WS training_progress poller
- K8s deployment manifest + CI integration

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-02 22:07:42 +01:00
parent 148d4cbb43 1a62060dab
commit bb133d6619
39 changed files with 1343 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitlab-ci.yml
View File

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

26
Cargo.lock generated
View File

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

1
Cargo.toml
View File

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

2
bin/fxt/build.rs
View File

@@ -22,6 +22,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"proto/ml_training.proto",
"proto/trading_agent.proto",
"proto/broker_gateway.proto",
"proto/monitoring.proto",
],
&["proto"],
)?;
@@ -34,6 +35,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}

74
bin/fxt/proto/monitoring.proto Normal file
View File

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

23
bin/fxt/src/commands/train/mod.rs
View File

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

173
bin/fxt/src/commands/train/monitor.rs Normal file
View File

@@ -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<String>,
/// 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
);
}

12
bin/fxt/src/commands/train/start.rs
View File

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

9
bin/fxt/src/lib.rs
View File

@@ -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");
}
}

109
crates/common/src/metrics/training_metrics.rs
View File

@@ -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);
}
}

15
crates/ml/examples/hyperopt_baseline_rl.rs
View File

@@ -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(())
}

45
crates/ml/examples/hyperopt_baseline_supervised.rs
View File

@@ -138,12 +138,17 @@ fn run_tft_hyperopt(args: &Args) -> Result<Value> {
.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<Value> {
.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<Value> {
.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<Value> {
.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<Value> {
.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<Value> {
.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<Value> {
.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<Value> {
.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(())
}

2
crates/web-gateway/build.rs
View File

@@ -13,6 +13,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"../../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<dyn std::error::Error>> {
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(())
}

15
crates/web-gateway/src/config.rs
View File

@@ -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<String>,
/// 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);
}
}

6
crates/web-gateway/src/grpc/clients.rs
View File

@@ -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<Channel
pub fn ml_training_client(channel: &Channel) -> MlTrainingServiceClient<Channel> {
MlTrainingServiceClient::new(channel.clone())
}
/// Create a `MonitoringServiceClient` from the shared channel
pub fn monitoring_client(channel: &Channel) -> MonitoringServiceClient<Channel> {
MonitoringServiceClient::new(channel.clone())
}

14
crates/web-gateway/src/grpc/service_auth.rs
View File

@@ -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(),
};

56
crates/web-gateway/src/grpc/streams.rs
View File

@@ -19,6 +19,7 @@ use crate::ws::messages::ServerMessage;
/// on failure with exponential backoff.
pub fn start_grpc_stream_bridges(
trading_channel: Option<Channel>,
monitoring_channel: Option<Channel>,
ws_broadcast: broadcast::Sender<String>,
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<String>) {
}
}
/// 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<String>) {
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).
@@ -206,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<F, Fut>(stream_name: &str, connect_fn: F)
where
F: Fn() -> Fut,

10
crates/web-gateway/src/lib.rs
View File

@@ -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");
}
}

11
crates/web-gateway/src/main.rs
View File

@@ -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);
}
@@ -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(),
);
@@ -80,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 {
@@ -101,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);
@@ -116,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);
}
});

1
crates/web-gateway/src/routes/auth.rs
View File

@@ -101,6 +101,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
};
Router::new()

1
crates/web-gateway/src/routes/backtesting.rs
View File

@@ -172,6 +172,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/config.rs
View File

@@ -91,6 +91,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/ml.rs
View File

@@ -125,6 +125,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

2
crates/web-gateway/src/routes/mod.rs
View File

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

28
crates/web-gateway/src/routes/monitoring.rs Normal file
View File

@@ -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<AppState> {
Router::new().route("/live-metrics", get(live_metrics))
}
async fn live_metrics(
State(state): State<AppState>,
) -> Result<Json<serde_json::Value>, 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()))?,
))
}

1
crates/web-gateway/src/routes/performance.rs
View File

@@ -113,6 +113,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/risk.rs
View File

@@ -134,6 +134,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/trading.rs
View File

@@ -264,6 +264,7 @@ mod tests {
trading_channel: channel,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/training.rs
View File

@@ -153,6 +153,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

1
crates/web-gateway/src/routes/tune.rs
View File

@@ -193,6 +193,7 @@ mod tests {
trading_channel: None,
backtesting_channel: None,
ml_training_channel: None,
monitoring_channel: None,
ws_broadcast,
}
}

7
crates/web-gateway/src/state.rs
View File

@@ -15,6 +15,7 @@ pub struct AppState {
pub trading_channel: Option<Channel>,
pub backtesting_channel: Option<Channel>,
pub ml_training_channel: Option<Channel>,
pub monitoring_channel: Option<Channel>,
/// Broadcast sender for WebSocket events
pub ws_broadcast: broadcast::Sender<String>,
}
@@ -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]

117
infra/k8s/services/monitoring-service.yaml Normal file
View File

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

36
services/monitoring_service/Cargo.toml Normal file
View File

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

10
services/monitoring_service/build.rs Normal file
View File

@@ -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(())
}

74
services/monitoring_service/proto/monitoring.proto Normal file
View File

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

89
services/monitoring_service/src/main.rs Normal file
View File

@@ -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(())
}

118
services/monitoring_service/src/prometheus_client.rs Normal file
View File

@@ -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()
}

267
services/monitoring_service/src/service.rs Normal file
View File

@@ -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);
}
}

2
services/trading_service/src/services/enhanced_ml.rs
View File

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