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

feat(training): move metrics to common::metrics, fix all clippy errors

Browse Source 
Move Prometheus training metrics from example-local baseline_common/
to common::metrics::{server,training_metrics} following the existing
grpc_metrics.rs pattern. Fix 29 let_underscore_must_use clippy errors
in push_metrics.rs, 3 shadow lint errors in training binaries, and
demote gradient clipping log from warn to debug.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-01 15:56:43 +01:00
parent 415480929d
commit 7744becdd5
11 changed files with 360 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/common/src/metrics/mod.rs
View File

@@ -5,6 +5,8 @@
pub mod grpc_metrics;
pub mod registry;
pub mod server;
pub mod training_metrics;
// Re-export commonly used functions
pub use registry::{

62
crates/common/src/metrics/server.rs Normal file
View File

@@ -0,0 +1,62 @@
//! Lightweight Prometheus HTTP metrics server for standalone binaries.
//!
//! Services that already embed Axum/tonic can expose `/metrics` through their
//! existing HTTP stack. Training binaries and CLI tools that don't have an
//! HTTP server use this instead.
use std::io::{BufRead, BufReader, Write as IoWrite};
use std::net::TcpListener;
use super::gather_metrics;
/// Spawn a background daemon thread serving Prometheus metrics over HTTP.
///
/// Responds to `GET /metrics` with the global registry output in Prometheus
/// text exposition format. Any other request gets a 404. The thread is
/// detached and dies with the process.
pub fn start_metrics_server(port: u16) {
std::thread::spawn(move || {
let addr = format!("0.0.0.0:{port}");
let listener = match TcpListener::bind(&addr) {
Ok(l) => {
tracing::info!("Prometheus metrics server listening on {addr}");
l
}
Err(e) => {
tracing::warn!("Failed to bind metrics server on {addr}: {e}");
return;
}
};
for stream in listener.incoming() {
let Ok(mut stream) = stream else {
continue;
};
// Read the HTTP request line (e.g. "GET /metrics HTTP/1.1")
let mut reader = BufReader::new(&stream);
let mut request_line = String::new();
if reader.read_line(&mut request_line).is_err() {
continue;
}
let is_metrics = request_line.starts_with("GET /metrics");
drop(reader);
if is_metrics {
let body = gather_metrics();
let response = format!(
"HTTP/1.1 200 OK\r\n\
Content-Type: text/plain; version=0.0.4\r\n\
Content-Length: {}\r\n\r\n\
{}",
body.len(),
body,
);
_ = stream.write_all(response.as_bytes());
} else {
_ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n");
}
}
});
}

235
crates/common/src/metrics/training_metrics.rs Normal file
View File

@@ -0,0 +1,235 @@
//! Prometheus metrics for ML training binaries.
//!
//! Registers the 18 metrics expected by the Training Cockpit Grafana dashboard
//! and provides typed helper functions so callers never mis-spell metric names.
//!
//! # Usage
//!
//! ```rust,ignore
//! use common::metrics::training_metrics;
//! use common::metrics::server;
//!
//! training_metrics::init();
//! server::start_metrics_server(9094);
//! training_metrics::set_active_workers(1.0);
//! // … training loop …
//! training_metrics::set_epoch("kan", "0", 1.0);
//! training_metrics::set_epoch_loss("kan", "0", 0.0042);
//! ```
use super::{
increment_counter_vec, observe_histogram_vec, register_counter_vec, register_gauge,
register_gauge_vec, register_histogram_vec, set_gauge, set_gauge_vec,
};
// ---------------------------------------------------------------------------
// Registration
// ---------------------------------------------------------------------------
/// Register all 18 training metrics with the global Prometheus registry.
///
/// Safe to call multiple times — the registry silently ignores duplicates.
pub fn init() {
let mf: &[&str] = &["model", "fold"];
let m: &[&str] = &["model"];
// Gauges (model + fold)
_ = register_gauge_vec("foxhunt_training_current_epoch", "Current epoch number", mf);
_ = register_gauge_vec("foxhunt_training_epoch_loss", "Training loss at end of epoch", mf);
_ = register_gauge_vec(
"foxhunt_training_validation_loss",
"Validation loss at end of epoch",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_batches_per_second",
"Training throughput in batches per second",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_batches_processed",
"Total batches processed this run",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_iteration_seconds",
"Last epoch wall time in seconds",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_eval_accuracy",
"Eval accuracy (supervised models)",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_eval_precision",
"Eval precision (supervised models)",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_eval_recall",
"Eval recall (supervised models)",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_eval_f1",
"Eval F1 score (supervised models)",
mf,
);
_ = register_gauge_vec(
"foxhunt_training_checkpoint_size_bytes",
"Size of last checkpoint file in bytes",
mf,
);
// Counters (model + fold)
_ = register_counter_vec(
"foxhunt_training_checkpoint_saves_total",
"Successful checkpoint saves",
mf,
);
_ = register_counter_vec(
"foxhunt_training_checkpoint_failures_total",
"Failed checkpoint saves",
mf,
);
_ = register_counter_vec(
"foxhunt_training_nan_detected_total",
"NaN loss events detected",
mf,
);
_ = register_counter_vec(
"foxhunt_training_gradient_explosion_total",
"Gradient explosion events",
mf,
);
_ = register_counter_vec(
"foxhunt_training_feature_errors_total",
"Feature extraction failures",
mf,
);
// Histograms (model only)
_ = register_histogram_vec(
"foxhunt_training_checkpoint_duration_seconds",
"Checkpoint save latency",
m,
vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0],
);
_ = register_histogram_vec(
"foxhunt_training_data_load_seconds",
"Data loading latency per fold",
m,
vec![0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0],
);
// Special gauge (no labels)
_ = register_gauge(
"foxhunt_training_active_workers",
"Number of active training workers",
);
}
// ---------------------------------------------------------------------------
// Typed helpers
// ---------------------------------------------------------------------------
pub fn set_epoch(model: &str, fold: &str, epoch: f64) {
set_gauge_vec("foxhunt_training_current_epoch", &[model, fold], epoch);
}
pub fn set_epoch_loss(model: &str, fold: &str, loss: f64) {
set_gauge_vec("foxhunt_training_epoch_loss", &[model, fold], loss);
}
pub fn set_validation_loss(model: &str, fold: &str, loss: f64) {
set_gauge_vec("foxhunt_training_validation_loss", &[model, fold], loss);
}
pub fn set_batches_per_second(model: &str, fold: &str, bps: f64) {
set_gauge_vec("foxhunt_training_batches_per_second", &[model, fold], bps);
}
pub fn set_batches_processed(model: &str, fold: &str, total: f64) {
set_gauge_vec("foxhunt_training_batches_processed", &[model, fold], total);
}
pub fn set_iteration_seconds(model: &str, fold: &str, secs: f64) {
set_gauge_vec("foxhunt_training_iteration_seconds", &[model, fold], secs);
}
pub fn set_eval_metrics(
model: &str,
fold: &str,
accuracy: f64,
precision: f64,
recall: f64,
f1: f64,
) {
set_gauge_vec("foxhunt_training_eval_accuracy", &[model, fold], accuracy);
set_gauge_vec("foxhunt_training_eval_precision", &[model, fold], precision);
set_gauge_vec("foxhunt_training_eval_recall", &[model, fold], recall);
set_gauge_vec("foxhunt_training_eval_f1", &[model, fold], f1);
}
pub fn record_checkpoint_save(model: &str, fold: &str, duration_secs: f64, size_bytes: f64) {
increment_counter_vec(
"foxhunt_training_checkpoint_saves_total",
&[model, fold],
1.0,
);
set_gauge_vec(
"foxhunt_training_checkpoint_size_bytes",
&[model, fold],
size_bytes,
);
observe_histogram_vec(
"foxhunt_training_checkpoint_duration_seconds",
&[model],
duration_secs,
);
}
pub fn record_checkpoint_failure(model: &str, fold: &str) {
increment_counter_vec(
"foxhunt_training_checkpoint_failures_total",
&[model, fold],
1.0,
);
}
pub fn record_nan_detected(model: &str, fold: &str) {
increment_counter_vec(
"foxhunt_training_nan_detected_total",
&[model, fold],
1.0,
);
}
pub fn record_gradient_explosion(model: &str, fold: &str) {
increment_counter_vec(
"foxhunt_training_gradient_explosion_total",
&[model, fold],
1.0,
);
}
pub fn record_feature_error(model: &str, fold: &str) {
increment_counter_vec(
"foxhunt_training_feature_errors_total",
&[model, fold],
1.0,
);
}
pub fn record_data_load(model: &str, duration_secs: f64) {
observe_histogram_vec(
"foxhunt_training_data_load_seconds",
&[model],
duration_secs,
);
}
pub fn set_active_workers(value: f64) {
set_gauge("foxhunt_training_active_workers", value);
}

258
crates/ml/examples/baseline_common/metrics.rs
View File

@@ -1,258 +0,0 @@
//! Prometheus metrics for training binaries.
//!
//! Registers 18 dashboard-expected metrics and exposes them via a lightweight
//! HTTP endpoint on `/metrics` (default port 9094, matching K8s annotations).
use std::io::{BufRead, BufReader, Write as IoWrite};
use std::net::TcpListener;
/// Register all 18 training metrics with the global Prometheus registry.
///
/// Safe to call multiple times — `common::metrics` silently ignores duplicate
/// registrations.
pub fn init_training_metrics() {
use common::metrics::{
register_counter_vec, register_gauge, register_gauge_vec, register_histogram_vec,
};
let mf = &["model", "fold"];
let m = &["model"];
// Gauges (model + fold)
let _ = register_gauge_vec("foxhunt_training_current_epoch", "Current epoch number", mf);
let _ = register_gauge_vec("foxhunt_training_epoch_loss", "Training loss at end of epoch", mf);
let _ = register_gauge_vec(
"foxhunt_training_validation_loss",
"Validation loss at end of epoch",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_batches_per_second",
"Training throughput in batches per second",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_batches_processed",
"Total batches processed this run",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_iteration_seconds",
"Last epoch wall time in seconds",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_eval_accuracy",
"Eval accuracy (supervised models)",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_eval_precision",
"Eval precision (supervised models)",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_eval_recall",
"Eval recall (supervised models)",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_eval_f1",
"Eval F1 score (supervised models)",
mf,
);
let _ = register_gauge_vec(
"foxhunt_training_checkpoint_size_bytes",
"Size of last checkpoint file in bytes",
mf,
);
// Counters (model + fold)
let _ = register_counter_vec(
"foxhunt_training_checkpoint_saves_total",
"Successful checkpoint saves",
mf,
);
let _ = register_counter_vec(
"foxhunt_training_checkpoint_failures_total",
"Failed checkpoint saves",
mf,
);
let _ = register_counter_vec(
"foxhunt_training_nan_detected_total",
"NaN loss events detected",
mf,
);
let _ = register_counter_vec(
"foxhunt_training_gradient_explosion_total",
"Gradient explosion events",
mf,
);
let _ = register_counter_vec(
"foxhunt_training_feature_errors_total",
"Feature extraction failures",
mf,
);
// Histograms (model only)
let _ = register_histogram_vec(
"foxhunt_training_checkpoint_duration_seconds",
"Checkpoint save latency",
m,
vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0],
);
let _ = register_histogram_vec(
"foxhunt_training_data_load_seconds",
"Data loading latency per fold",
m,
vec![0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0],
);
// Special gauge (no labels)
let _ = register_gauge("foxhunt_training_active_workers", "Number of active training workers");
}
/// Spawn a background daemon thread serving Prometheus metrics over HTTP.
///
/// Responds to `GET /metrics` with the global registry output. Any other
/// request gets a 404. The thread is detached and dies with the process.
pub fn start_metrics_server(port: u16) {
std::thread::spawn(move || {
let addr = format!("0.0.0.0:{port}");
let listener = match TcpListener::bind(&addr) {
Ok(l) => {
tracing::info!("Prometheus metrics server listening on {addr}");
l
}
Err(e) => {
tracing::warn!("Failed to bind metrics server on {addr}: {e}");
return;
}
};
for stream in listener.incoming() {
let Ok(mut stream) = stream else {
continue;
};
// Read request line (e.g. "GET /metrics HTTP/1.1")
let mut reader = BufReader::new(&stream);
let mut request_line = String::new();
if reader.read_line(&mut request_line).is_err() {
continue;
}
let is_metrics = request_line.starts_with("GET /metrics");
drop(reader);
if is_metrics {
let body = common::metrics::gather_metrics();
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\n\r\n{}",
body.len(),
body
);
let _ = stream.write_all(response.as_bytes());
} else {
let _ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n");
}
}
});
}
// ---------------------------------------------------------------------------
// Helper functions wrapping common::metrics
// ---------------------------------------------------------------------------
pub fn set_epoch(model: &str, fold: &str, epoch: f64) {
common::metrics::set_gauge_vec("foxhunt_training_current_epoch", &[model, fold], epoch);
}
pub fn set_epoch_loss(model: &str, fold: &str, loss: f64) {
common::metrics::set_gauge_vec("foxhunt_training_epoch_loss", &[model, fold], loss);
}
pub fn set_validation_loss(model: &str, fold: &str, loss: f64) {
common::metrics::set_gauge_vec("foxhunt_training_validation_loss", &[model, fold], loss);
}
pub fn set_batches_per_second(model: &str, fold: &str, bps: f64) {
common::metrics::set_gauge_vec("foxhunt_training_batches_per_second", &[model, fold], bps);
}
pub fn set_batches_processed(model: &str, fold: &str, total: f64) {
common::metrics::set_gauge_vec("foxhunt_training_batches_processed", &[model, fold], total);
}
pub fn set_iteration_seconds(model: &str, fold: &str, secs: f64) {
common::metrics::set_gauge_vec("foxhunt_training_iteration_seconds", &[model, fold], secs);
}
pub fn set_eval_metrics(model: &str, fold: &str, accuracy: f64, precision: f64, recall: f64, f1: f64) {
common::metrics::set_gauge_vec("foxhunt_training_eval_accuracy", &[model, fold], accuracy);
common::metrics::set_gauge_vec("foxhunt_training_eval_precision", &[model, fold], precision);
common::metrics::set_gauge_vec("foxhunt_training_eval_recall", &[model, fold], recall);
common::metrics::set_gauge_vec("foxhunt_training_eval_f1", &[model, fold], f1);
}
pub fn record_checkpoint_save(model: &str, fold: &str, duration_secs: f64, size_bytes: f64) {
common::metrics::increment_counter_vec(
"foxhunt_training_checkpoint_saves_total",
&[model, fold],
1.0,
);
common::metrics::set_gauge_vec(
"foxhunt_training_checkpoint_size_bytes",
&[model, fold],
size_bytes,
);
common::metrics::observe_histogram_vec(
"foxhunt_training_checkpoint_duration_seconds",
&[model],
duration_secs,
);
}
pub fn record_checkpoint_failure(model: &str, fold: &str) {
common::metrics::increment_counter_vec(
"foxhunt_training_checkpoint_failures_total",
&[model, fold],
1.0,
);
}
pub fn record_nan_detected(model: &str, fold: &str) {
common::metrics::increment_counter_vec(
"foxhunt_training_nan_detected_total",
&[model, fold],
1.0,
);
}
pub fn record_gradient_explosion(model: &str, fold: &str) {
common::metrics::increment_counter_vec(
"foxhunt_training_gradient_explosion_total",
&[model, fold],
1.0,
);
}
pub fn record_feature_error(model: &str, fold: &str) {
common::metrics::increment_counter_vec(
"foxhunt_training_feature_errors_total",
&[model, fold],
1.0,
);
}
pub fn record_data_load(model: &str, duration_secs: f64) {
common::metrics::observe_histogram_vec(
"foxhunt_training_data_load_seconds",
&[model],
duration_secs,
);
}
pub fn set_active_workers(value: f64) {
common::metrics::set_gauge("foxhunt_training_active_workers", value);
}

2
crates/ml/examples/baseline_common/mod.rs
View File

@@ -6,8 +6,6 @@
#[allow(dead_code)]
pub mod completion;
#[allow(dead_code)]
pub mod metrics;
use std::path::{Path, PathBuf};

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

@@ -47,9 +47,7 @@ use ml::hyperopt::adapters::ppo::PPOTrainer;
use ml::hyperopt::paths::{generate_run_id, TrainingPaths};
use ml::hyperopt::ArgminOptimizer;
#[allow(unreachable_pub)]
mod baseline_common;
use baseline_common::metrics;
use common::metrics::{server as metrics_server, training_metrics};
/// Hyperparameter optimization runner for DQN/PPO on Databento market data
#[derive(Parser, Debug)]
@@ -256,9 +254,9 @@ fn main() -> Result<()> {
.with_target(false)
.init();
metrics::init_training_metrics();
metrics::start_metrics_server(9094);
metrics::set_active_workers(1.0);
training_metrics::init();
metrics_server::start_metrics_server(9094);
training_metrics::set_active_workers(1.0);
let args = Args::parse();
@@ -288,7 +286,7 @@ fn main() -> Result<()> {
info!("Device: CUDA GPU");
// Resolve parallel: 0 = auto-detect. Reserve 2 cores for CUDA driver overhead.
let parallel = if args.parallel == 0 {
let mut parallel = if args.parallel == 0 {
cpus.saturating_sub(2).max(1)
} else {
args.parallel
@@ -297,7 +295,7 @@ fn main() -> Result<()> {
// Cap by VRAM budget — no point having more threads than concurrent GPU trials
let budget = ml::hyperopt::HardwareBudget::detect();
let vram_cap = budget.plan_hyperopt(300.0, 0.0002, 64.0, 4096.0).max_concurrent_trials;
let parallel = parallel.min(vram_cap);
parallel = parallel.min(vram_cap);
info!("Parallel: {} threads (CPU: {}, VRAM cap: {})", parallel, cpus, vram_cap);
// Configure rayon thread pool for parallel trial evaluation
@@ -406,6 +404,6 @@ fn main() -> Result<()> {
info!("========================================");
info!("{}", output_str);
metrics::set_active_workers(0.0);
training_metrics::set_active_workers(0.0);
Ok(())
}

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

@@ -37,9 +37,7 @@ use std::path::PathBuf;
use std::time::Instant;
use tracing::{error, info, warn, Level};
#[allow(unreachable_pub)]
mod baseline_common;
use baseline_common::metrics;
use common::metrics::{server as metrics_server, training_metrics};
use ml::hyperopt::adapters::diffusion::DiffusionTrainer;
use ml::hyperopt::adapters::kan::KANTrainer;
@@ -503,9 +501,9 @@ fn main() -> Result<()> {
.with_target(false)
.init();
metrics::init_training_metrics();
metrics::start_metrics_server(9094);
metrics::set_active_workers(1.0);
training_metrics::init();
metrics_server::start_metrics_server(9094);
training_metrics::set_active_workers(1.0);
let args = Args::parse();
@@ -602,6 +600,6 @@ fn main() -> Result<()> {
info!("========================================");
info!("{}", output_str);
metrics::set_active_workers(0.0);
training_metrics::set_active_workers(0.0);
Ok(())
}

16
crates/ml/examples/train_baseline_rl.rs
View File

@@ -34,8 +34,8 @@ use ml::dqn::{DQNConfig, Experience, DQN};
#[allow(unreachable_pub)]
mod baseline_common;
use baseline_common::completion::{write_failure_marker, write_success_marker, CompletionMetrics};
use baseline_common::metrics;
use baseline_common::{load_all_bars, spread_cost_bps};
use common::metrics::{server as metrics_server, training_metrics as metrics};
use ml::features::extraction::extract_ml_features;
use ml::ppo::ppo::{PPOConfig, PPO};
use ml::ppo::trajectories::{Trajectory, TrajectoryBatch, TrajectoryStep};
@@ -1038,8 +1038,8 @@ fn main() -> Result<()> {
)
.init();
metrics::init_training_metrics();
metrics::start_metrics_server(9094);
metrics::init();
metrics_server::start_metrics_server(9094);
metrics::set_active_workers(1.0);
let args = Args::parse();
@@ -1054,20 +1054,20 @@ fn main() -> Result<()> {
match result {
Ok(results) => {
for result in &results {
let best_val = result
for training_result in &results {
let best_val = training_result
.fold_results
.iter()
.map(|(_, loss)| *loss)
.fold(f64::MAX, f64::min);
let metrics = CompletionMetrics {
model: result.model_name.clone(),
model: training_result.model_name.clone(),
symbol: args.symbol.clone(),
best_val_loss: (best_val < f64::MAX).then_some(best_val),
sharpe_ratio: None,
epochs_completed: result.total_epochs,
folds_completed: result.fold_results.len(),
epochs_completed: training_result.total_epochs,
folds_completed: training_result.fold_results.len(),
};
write_success_marker(&args.output_dir, &metrics);
}

16
crates/ml/examples/train_baseline_supervised.rs
View File

@@ -42,8 +42,8 @@ use ml::walk_forward::{generate_walk_forward_windows, NormStats, WalkForwardConf
#[allow(unreachable_pub)]
mod baseline_common;
use baseline_common::completion::{write_failure_marker, write_success_marker, CompletionMetrics};
use baseline_common::metrics;
use baseline_common::{load_all_bars, spread_cost_bps};
use common::metrics::{server as metrics_server, training_metrics as metrics};
// Model adapter imports -- using verified paths from existing examples
use ml::diffusion::{DiffusionConfig, DiffusionTrainableAdapter};
@@ -804,8 +804,8 @@ fn main() -> Result<()> {
)
.init();
metrics::init_training_metrics();
metrics::start_metrics_server(9094);
metrics::init();
metrics_server::start_metrics_server(9094);
metrics::set_active_workers(1.0);
let args = Args::parse();
@@ -820,20 +820,20 @@ fn main() -> Result<()> {
match result {
Ok(results) => {
for result in &results {
let best_val = result
for training_result in &results {
let best_val = training_result
.fold_results
.iter()
.map(|(_, loss)| *loss)
.fold(f64::MAX, f64::min);
let metrics = CompletionMetrics {
model: result.model_name.clone(),
model: training_result.model_name.clone(),
symbol: args.symbol.clone(),
best_val_loss: (best_val < f64::MAX).then_some(best_val),
sharpe_ratio: None,
epochs_completed: result.total_epochs,
folds_completed: result.fold_results.len(),
epochs_completed: training_result.total_epochs,
folds_completed: training_result.fold_results.len(),
};
write_success_marker(&args.output_dir, &metrics);
}

5
crates/ml/src/optimizers/adam.rs
View File

@@ -158,9 +158,8 @@ impl Adam {
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
if actual_grad_norm > max_norm {
tracing::warn!(
"Gradient clipping: {:.4} -> {:.4} - this is expected occasionally but should be rare. \
If frequent, consider reducing learning rate.",
tracing::debug!(
"Gradient clipping: {:.4} -> {:.4}",
actual_grad_norm,
clipped_grad_norm
);

60
crates/ml/src/training/push_metrics.rs
View File

@@ -33,23 +33,23 @@ impl TrainingMetricsPusher {
let mut body = String::new();
// Epoch progress
let _ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch");
let _ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge");
let _ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch);
_ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch");
_ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge");
_ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch);
// Loss
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_epoch_loss Training loss value"
);
let _ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge");
let _ = writeln!(
_ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge");
_ = writeln!(
body,
"foxhunt_training_epoch_loss{{split=\"train\"}} {}",
state.train_loss
);
if let Some(val_loss) = state.val_loss {
let _ = writeln!(
_ = writeln!(
body,
"foxhunt_training_epoch_loss{{split=\"val\"}} {}",
val_loss
@@ -58,21 +58,21 @@ impl TrainingMetricsPusher {
// Accuracy
if let Some(accuracy) = state.accuracy {
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_eval_accuracy Model evaluation accuracy"
);
let _ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge");
let _ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy);
_ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge");
_ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy);
}
// Batches processed
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_batches_processed Total batches processed"
);
let _ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter");
let _ = writeln!(
_ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter");
_ = writeln!(
body,
"foxhunt_training_batches_processed {}",
state.batches_processed
@@ -80,12 +80,12 @@ impl TrainingMetricsPusher {
// Training speed
if state.batches_per_second > 0.0 {
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_batches_per_second Training throughput"
);
let _ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge");
let _ = writeln!(
_ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge");
_ = writeln!(
body,
"foxhunt_training_batches_per_second {}",
state.batches_per_second
@@ -93,12 +93,12 @@ impl TrainingMetricsPusher {
}
// Learning rate
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_learning_rate Current learning rate"
);
let _ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge");
let _ = writeln!(
_ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge");
_ = writeln!(
body,
"foxhunt_training_learning_rate {}",
state.learning_rate
@@ -106,27 +106,27 @@ impl TrainingMetricsPusher {
// NaN / gradient events
if state.nan_count > 0 {
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_nan_detected_total NaN gradient events"
);
let _ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter");
let _ = writeln!(
_ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter");
_ = writeln!(
body,
"foxhunt_training_nan_detected_total {}",
state.nan_count
);
}
if state.gradient_explosion_count > 0 {
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_gradient_explosion_total Gradient clipping events"
);
let _ = writeln!(
_ = writeln!(
body,
"# TYPE foxhunt_training_gradient_explosion_total counter"
);
let _ = writeln!(
_ = writeln!(
body,
"foxhunt_training_gradient_explosion_total {}",
state.gradient_explosion_count
@@ -134,15 +134,15 @@ impl TrainingMetricsPusher {
}
// Checkpoint saves
let _ = writeln!(
_ = writeln!(
body,
"# HELP foxhunt_training_checkpoint_saves_total Checkpoints saved"
);
let _ = writeln!(
_ = writeln!(
body,
"# TYPE foxhunt_training_checkpoint_saves_total counter"
);
let _ = writeln!(
_ = writeln!(
body,
"foxhunt_training_checkpoint_saves_total {}",
state.checkpoint_saves
@@ -174,7 +174,9 @@ impl TrainingMetricsPusher {
"{}/metrics/job/{}/model/{}",
self.pushgateway_url, self.job_id, self.model_name
);
let _ = self.client.delete(&url).send().await;
if let Err(e) = self.client.delete(&url).send().await {
eprintln!("Failed to delete metrics from Pushgateway: {e}");
}
}
}