From d04b6c7023980298e4bba91dd1e1fb89eaa56956 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 3 Mar 2026 14:14:51 +0100 Subject: [PATCH] fix(fxt,services): remove mock fallbacks and add gRPC health checks - trade_ml.rs: Replace 3 mock data fallbacks (submit, predictions, performance) with proper error propagation. Commands now fail honestly when the API Gateway is unreachable instead of silently returning fake data. Mark 3 integration tests as #[ignore]. - monitoring_service: Add tonic-health with set_serving for MonitoringServiceServer. Enables grpc_health_probe readiness checks. - ml_training_service: Add tonic-health with set_serving for MlTrainingServiceServer. Wired into both TLS and non-TLS paths. - data_acquisition_service: Add tonic-health with set_serving for DataAcquisitionServiceServer. - ml/cuda_streams: Fix pre-existing unused variable clippy warning. All 8 services now have standard gRPC health checking enabled. Co-Authored-By: Claude Opus 4.6 --- Cargo.lock | 3 + bin/fxt/src/commands/trade_ml.rs | 130 +++--------------- crates/ml/src/ensemble/cuda_streams.rs | 1 + services/data_acquisition_service/Cargo.toml | 1 + services/data_acquisition_service/src/main.rs | 7 + services/ml_training_service/Cargo.toml | 1 + services/ml_training_service/src/main.rs | 9 ++ services/monitoring_service/Cargo.toml | 1 + services/monitoring_service/src/main.rs | 7 + 9 files changed, 46 insertions(+), 114 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 02ff7406b..44e6a809e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3044,6 +3044,7 @@ dependencies = [ "tokio-stream", "tokio-util", "tonic", + "tonic-health", "tonic-prost", "tonic-prost-build", "tonic-reflection", @@ -6174,6 +6175,7 @@ dependencies = [ "tokio-stream", "tokio-util", "tonic", + "tonic-health", "tonic-prost", "tonic-prost-build", "tonic-reflection", @@ -6249,6 +6251,7 @@ dependencies = [ "tokio", "tokio-stream", "tonic", + "tonic-health", "tonic-prost", "tonic-prost-build", "tracing", diff --git a/bin/fxt/src/commands/trade_ml.rs b/bin/fxt/src/commands/trade_ml.rs index 906afe85d..bb740788a 100644 --- a/bin/fxt/src/commands/trade_ml.rs +++ b/bin/fxt/src/commands/trade_ml.rs @@ -189,9 +189,7 @@ impl TradeMlArgs { /// * `api_gateway_url` - API Gateway URL /// * `jwt_token` - JWT authentication token /// - /// # Production Implementation /// Connects to API Gateway via gRPC and submits ML order request. - /// Falls back to mock data if connection fails (for testing). async fn submit_ml_order( &self, symbol: &str, @@ -207,25 +205,8 @@ impl TradeMlArgs { .get_ml_prediction(symbol, model, api_gateway_url, jwt_token) .await; - let (predicted_action, confidence, model_display) = match prediction_result { - Ok(pred) => pred, - Err(e) => { - println!( - "{}", - format!( - "\u{26a0}\u{fe0f} Warning: Failed to get ML prediction: {}", - e - ) - .yellow() - ); - println!("{}", "Using mock prediction for demonstration".yellow()); - ( - "BUY".to_owned(), - 0.85, - model.unwrap_or("Ensemble").to_owned(), - ) - }, - }; + let (predicted_action, confidence, model_display) = prediction_result + .map_err(|e| anyhow::anyhow!("ML prediction failed: {e}\n\nEnsure the API Gateway is reachable and ML models are loaded."))?; // Step 2: Submit order based on ML prediction let order_side = match predicted_action.as_str() { @@ -256,18 +237,8 @@ impl TradeMlArgs { ) .await; - // Step 3: Display results (with mock fallback for testing) - let order_id = match order_result { - Ok(order_id) => order_id, - Err(e) => { - println!( - "{}", - format!("\u{26a0}\u{fe0f} Warning: Failed to submit order: {}", e).yellow() - ); - println!("{}", "Using mock order ID for demonstration".yellow()); - uuid::Uuid::new_v4().to_string() - }, - }; + let order_id = order_result + .map_err(|e| anyhow::anyhow!("Order submission failed: {e}\n\nEnsure the API Gateway and Trading Service are running."))?; println!("{}", "\u{2705} ML order submitted successfully!".green()); println!(); @@ -445,11 +416,8 @@ impl TradeMlArgs { jwt_token: &str, ) -> Result<()> { use crate::proto::trading::{ - trading_service_client::TradingServiceClient, GetMlPredictionsRequest, MlPrediction, + trading_service_client::TradingServiceClient, GetMlPredictionsRequest, }; - use chrono::Utc; - - // Try to connect to API Gateway with fallback to mock data let predictions_result = async { let mut client = TradingServiceClient::connect(api_gateway_url.to_owned()) .await @@ -477,39 +445,8 @@ impl TradeMlArgs { } .await; - let predictions_response = match predictions_result { - Ok(predictions) => predictions, - Err(e) => { - println!( - "{}", - format!( - "\u{26a0}\u{fe0f} Warning: Failed to get predictions: {}", - e - ) - .yellow() - ); - println!("{}", "Using mock predictions for demonstration".yellow()); - // Generate mock predictions - vec![ - MlPrediction { - timestamp: Utc::now().to_rfc3339(), - model_id: model.unwrap_or("MAMBA2").to_owned(), - symbol: symbol.to_owned(), - predicted_action: "BUY".to_owned(), - confidence: 0.85, - actual_return: Some(0.023), - }, - MlPrediction { - timestamp: Utc::now().to_rfc3339(), - model_id: model.unwrap_or("DQN").to_owned(), - symbol: symbol.to_owned(), - predicted_action: "SELL".to_owned(), - confidence: 0.72, - actual_return: Some(-0.012), - }, - ] - }, - }; + let predictions_response = predictions_result + .map_err(|e| anyhow::anyhow!("Failed to fetch predictions: {e}\n\nEnsure the API Gateway and Trading Service are running."))?; // Display header println!(); @@ -619,10 +556,8 @@ impl TradeMlArgs { jwt_token: &str, ) -> Result<()> { use crate::proto::trading::{ - trading_service_client::TradingServiceClient, GetMlPerformanceRequest, ModelPerformance, + trading_service_client::TradingServiceClient, GetMlPerformanceRequest, }; - - // Try to connect to API Gateway with fallback to mock data let performance_result = async { let mut client = TradingServiceClient::connect(api_gateway_url.to_owned()) .await @@ -648,42 +583,8 @@ impl TradeMlArgs { } .await; - let models = match performance_result { - Ok(models) => models, - Err(e) => { - println!( - "{}", - format!( - "\u{26a0}\u{fe0f} Warning: Failed to get performance metrics: {}", - e - ) - .yellow() - ); - println!( - "{}", - "Using mock performance data for demonstration".yellow() - ); - // Generate mock performance data - vec![ - ModelPerformance { - model_id: model.unwrap_or("MAMBA2").to_owned(), - accuracy: 0.725, - total_predictions: 150, - sharpe_ratio: 1.82, - avg_return: 0.023, - max_drawdown: 0.031, - }, - ModelPerformance { - model_id: model.unwrap_or("DQN").to_owned(), - accuracy: 0.682, - total_predictions: 200, - sharpe_ratio: 1.45, - avg_return: 0.018, - max_drawdown: 0.045, - }, - ] - }, - }; + let models = performance_result + .map_err(|e| anyhow::anyhow!("Failed to fetch performance metrics: {e}\n\nEnsure the API Gateway and Trading Service are running."))?; // Display ML Model Performance header println!("\n{}", "ML Model Performance (Last 30 days)".bold()); @@ -1291,8 +1192,8 @@ mod tests { use super::*; #[tokio::test] + #[ignore] // Requires running API Gateway async fn test_submit_command_parses() { - // Test that command structure is correct let args = TradeMlArgs { command: TradeMlCommand::Submit { symbol: "ES.FUT".to_owned(), @@ -1301,12 +1202,12 @@ mod tests { }, }; - // Should execute without panic - let result = args.execute("http://localhost:50051", "mock-token").await; + let result = args.execute("http://localhost:50051", "test-token").await; result.unwrap(); } #[tokio::test] + #[ignore] // Requires running API Gateway async fn test_predictions_command_parses() { let args = TradeMlArgs { command: TradeMlCommand::Predictions { @@ -1316,11 +1217,12 @@ mod tests { }, }; - let result = args.execute("http://localhost:50051", "mock-token").await; + let result = args.execute("http://localhost:50051", "test-token").await; result.unwrap(); } #[tokio::test] + #[ignore] // Requires running API Gateway async fn test_performance_command_parses() { let args = TradeMlArgs { command: TradeMlCommand::Performance { @@ -1328,7 +1230,7 @@ mod tests { }, }; - let result = args.execute("http://localhost:50051", "mock-token").await; + let result = args.execute("http://localhost:50051", "test-token").await; result.unwrap(); } diff --git a/crates/ml/src/ensemble/cuda_streams.rs b/crates/ml/src/ensemble/cuda_streams.rs index 2fa814ed2..64f213f11 100644 --- a/crates/ml/src/ensemble/cuda_streams.rs +++ b/crates/ml/src/ensemble/cuda_streams.rs @@ -59,6 +59,7 @@ impl CudaStreamPool { /// work to complete before starting. /// /// On CPU: creates a no-op pool that passes through all operations. + #[allow(unused_variables)] // `count` used only in cuda feature pub fn new(device: &Device, count: usize) -> Result { #[cfg(feature = "cuda")] { diff --git a/services/data_acquisition_service/Cargo.toml b/services/data_acquisition_service/Cargo.toml index f4f144246..9843b3f6c 100644 --- a/services/data_acquisition_service/Cargo.toml +++ b/services/data_acquisition_service/Cargo.toml @@ -21,6 +21,7 @@ clap.workspace = true # gRPC and protocol buffers - USE WORKSPACE tonic.workspace = true tonic-prost.workspace = true +tonic-health.workspace = true tonic-reflection.workspace = true prost.workspace = true prost-types.workspace = true diff --git a/services/data_acquisition_service/src/main.rs b/services/data_acquisition_service/src/main.rs index e2a37a73c..3bfb80587 100644 --- a/services/data_acquisition_service/src/main.rs +++ b/services/data_acquisition_service/src/main.rs @@ -97,11 +97,18 @@ async fn main() -> Result<(), Box> { // Configure gRPC server address let addr: SocketAddr = format!("0.0.0.0:{}", args.port).parse()?; + // Setup gRPC health check service + let (health_reporter, health_service) = tonic_health::server::health_reporter(); + health_reporter + .set_serving::>() + .await; + info!("Data Acquisition Service listening on {}", addr); // Start gRPC server with graceful shutdown Server::builder() .layer(GrpcMetricsLayer::new("data-acquisition-service")) + .add_service(health_service) .add_service(DataAcquisitionServiceServer::new(service)) .serve_with_shutdown(addr, shutdown_signal()) .await?; diff --git a/services/ml_training_service/Cargo.toml b/services/ml_training_service/Cargo.toml index 9334bb140..9c838cfe4 100644 --- a/services/ml_training_service/Cargo.toml +++ b/services/ml_training_service/Cargo.toml @@ -23,6 +23,7 @@ rust_decimal.workspace = true # gRPC and protocol buffers - USE WORKSPACE tonic.workspace = true tonic-prost.workspace = true +tonic-health.workspace = true tonic-reflection.workspace = true prost.workspace = true prost-types.workspace = true diff --git a/services/ml_training_service/src/main.rs b/services/ml_training_service/src/main.rs index 1431f445c..4cd8eadd6 100644 --- a/services/ml_training_service/src/main.rs +++ b/services/ml_training_service/src/main.rs @@ -449,6 +449,13 @@ async fn serve(args: ServeArgs) -> Result<()> { // Build server with reflection let service = MlTrainingServiceServer::new(training_service); + // Setup gRPC health check service + let (health_reporter, health_service) = tonic_health::server::health_reporter(); + health_reporter + .set_serving::>() + .await; + info!("gRPC health service configured - ml_training_service marked as SERVING"); + // Wave 67 Agent 3: HTTP/2 streaming performance optimizations // Apply same optimizations as Trading Service for consistency let enable_http2_opts = std::env::var("ENABLE_HTTP2_OPTIMIZATIONS") @@ -484,10 +491,12 @@ async fn serve(args: ServeArgs) -> Result<()> { info!("TLS enabled - configuring mTLS for gRPC server"); server_builder .tls_config(tls.to_server_tls_config())? + .add_service(health_service) .add_service(service) } else { info!("TLS disabled - running gRPC server without encryption"); server_builder + .add_service(health_service) .add_service(service) }; diff --git a/services/monitoring_service/Cargo.toml b/services/monitoring_service/Cargo.toml index a431d0b3e..88eab6737 100644 --- a/services/monitoring_service/Cargo.toml +++ b/services/monitoring_service/Cargo.toml @@ -11,6 +11,7 @@ description = "Monitoring Service - Prometheus-to-gRPC bridge for live training tokio.workspace = true tonic.workspace = true tonic-prost.workspace = true +tonic-health.workspace = true prost.workspace = true prost-types.workspace = true serde.workspace = true diff --git a/services/monitoring_service/src/main.rs b/services/monitoring_service/src/main.rs index 512038b61..92b0dda4a 100644 --- a/services/monitoring_service/src/main.rs +++ b/services/monitoring_service/src/main.rs @@ -78,9 +78,16 @@ async fn main() -> Result<()> { .parse() .map_err(|e| anyhow::anyhow!("Invalid listen address: {e}"))?; + // Setup gRPC health check service + let (health_reporter, health_service) = tonic_health::server::health_reporter(); + health_reporter + .set_serving::>() + .await; + info!("gRPC server listening on {}", addr); tonic::transport::Server::builder() + .add_service(health_service) .add_service(MonitoringServiceServer::new(svc)) .serve(addr) .await?;