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

fix(fxt,services): remove mock fallbacks and add gRPC health checks

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 14:14:51 +01:00
parent 8352a0a75e
commit d04b6c7023
9 changed files with 46 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.lock generated
View File

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

130
bin/fxt/src/commands/trade_ml.rs
View File

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

1
crates/ml/src/ensemble/cuda_streams.rs
View File

@@ -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<Self, MLError> {
#[cfg(feature = "cuda")]
{

1
services/data_acquisition_service/Cargo.toml
View File

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

7
services/data_acquisition_service/src/main.rs
View File

@@ -97,11 +97,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
// 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::<DataAcquisitionServiceServer<DataAcquisitionServiceImpl>>()
.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?;

1
services/ml_training_service/Cargo.toml
View File

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

9
services/ml_training_service/src/main.rs
View File

@@ -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::<MlTrainingServiceServer<MLTrainingServiceImpl>>()
.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)
};

1
services/monitoring_service/Cargo.toml
View File

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

7
services/monitoring_service/src/main.rs
View File

@@ -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::<MonitoringServiceServer<MonitoringServiceImpl>>()
.await;
info!("gRPC server listening on {}", addr);
tonic::transport::Server::builder()
.add_service(health_service)
.add_service(MonitoringServiceServer::new(svc))
.serve(addr)
.await?;