diff --git a/Cargo.lock b/Cargo.lock index 78d4bbc7a..e58a4a7b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -244,7 +244,7 @@ dependencies = [ "once_cell", "prometheus", "prost 0.14.1", - "prost-build 0.14.1", + "prost-build", "qrcode", "rand 0.8.5", "redis", @@ -1531,9 +1531,10 @@ dependencies = [ "model_loader", "num-traits", "num_cpus", + "once_cell", "prometheus", "prost 0.14.1", - "prost-build 0.14.1", + "prost-build", "rand 0.8.5", "rayon", "reqwest 0.12.23", @@ -3549,7 +3550,7 @@ dependencies = [ "hdrhistogram", "ml", "prost 0.14.1", - "prost-types 0.14.1", + "prost-types", "rand 0.8.5", "reqwest 0.12.23", "risk", @@ -4703,6 +4704,8 @@ version = "1.0.0" dependencies = [ "anyhow", "chrono", + "futures", + "jsonwebtoken", "prost 0.14.1", "reqwest 0.12.23", "serde", @@ -4711,7 +4714,8 @@ dependencies = [ "thiserror 1.0.69", "tokio", "tonic", - "tonic-build 0.12.3", + "tonic-prost", + "tonic-prost-build", "tracing", "tracing-subscriber", "uuid", @@ -5075,7 +5079,7 @@ dependencies = [ "plotters", "prometheus", "prost 0.14.1", - "prost-build 0.14.1", + "prost-build", "rand 0.8.5", "serde", "serde_json", @@ -5491,11 +5495,12 @@ dependencies = [ "ml-data", "num_cpus", "object_store", + "once_cell", "pbkdf2", "prometheus", "prost 0.14.1", - "prost-build 0.14.1", - "prost-types 0.14.1", + "prost-build", + "prost-types", "rand 0.8.5", "reqwest 0.12.23", "risk", @@ -6641,26 +6646,6 @@ dependencies = [ "prost-derive 0.14.1", ] -[[package]] -name = "prost-build" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" -dependencies = [ - "heck 0.5.0", - "itertools 0.14.0", - "log", - "multimap", - "once_cell", - "petgraph 0.7.1", - "prettyplease", - "prost 0.13.5", - "prost-types 0.13.5", - "regex", - "syn 2.0.106", - "tempfile", -] - [[package]] name = "prost-build" version = "0.14.1" @@ -6675,7 +6660,7 @@ dependencies = [ "petgraph 0.7.1", "prettyplease", "prost 0.14.1", - "prost-types 0.14.1", + "prost-types", "pulldown-cmark", "pulldown-cmark-to-cmark", "regex", @@ -6722,15 +6707,6 @@ dependencies = [ "syn 2.0.106", ] -[[package]] -name = "prost-types" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16" -dependencies = [ - "prost 0.13.5", -] - [[package]] name = "prost-types" version = "0.14.1" @@ -9542,20 +9518,6 @@ dependencies = [ "webpki-roots 1.0.2", ] -[[package]] -name = "tonic-build" -version = "0.12.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9557ce109ea773b399c9b9e5dca39294110b74f1f342cb347a80d1fce8c26a11" -dependencies = [ - "prettyplease", - "proc-macro2", - "prost-build 0.13.5", - "prost-types 0.13.5", - "quote", - "syn 2.0.106", -] - [[package]] name = "tonic-build" version = "0.14.2" @@ -9600,12 +9562,12 @@ checksum = "b4a16cba4043dc3ff43fcb3f96b4c5c154c64cbd18ca8dce2ab2c6a451d058a2" dependencies = [ "prettyplease", "proc-macro2", - "prost-build 0.14.1", - "prost-types 0.14.1", + "prost-build", + "prost-types", "quote", "syn 2.0.106", "tempfile", - "tonic-build 0.14.2", + "tonic-build", ] [[package]] @@ -9615,7 +9577,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34da53e8387581d66db16ff01f98a70b426b091fdf76856e289d5c1bd386ed7b" dependencies = [ "prost 0.14.1", - "prost-types 0.14.1", + "prost-types", "tokio", "tokio-stream", "tonic", @@ -9871,7 +9833,7 @@ dependencies = [ "once_cell", "prometheus", "prost 0.14.1", - "prost-build 0.14.1", + "prost-build", "rand 0.8.5", "redis", "reqwest 0.12.23", diff --git a/certs/production/ca/ca-cert.srl b/certs/production/ca/ca-cert.srl index e44763224..48497da1b 100644 --- a/certs/production/ca/ca-cert.srl +++ b/certs/production/ca/ca-cert.srl @@ -1 +1 @@ -5B28085BAEC3B89B98347D9C8A85629C780242E1 +5B28085BAEC3B89B98347D9C8A85629C780242E5 diff --git a/certs/production/foxhunt-cert.pem.2048.backup b/certs/production/foxhunt-cert.pem.2048.backup new file mode 100644 index 000000000..5896a896c --- /dev/null +++ b/certs/production/foxhunt-cert.pem.2048.backup @@ -0,0 +1,27 @@ +-----BEGIN CERTIFICATE----- +MIIEhjCCAm4CFFsoCFuuw7ibmDR9nIqFYpx4AkLcMA0GCSqGSIb3DQEBCwUAMHwx +CzAJBgNVBAYTAlVTMQswCQYDVQQIDAJOWTEQMA4GA1UEBwwHTmV3WW9yazEbMBkG +A1UECgwSRm94aHVudCBQcm9kdWN0aW9uMREwDwYDVQQLDAhTZWN1cml0eTEeMBwG +A1UEAwwVRm94aHVudCBQcm9kdWN0aW9uIENBMB4XDTI1MDkwNzExNTkxMVoXDTI2 +MDkwNzExNTkxMVowgYIxCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJOWTEQMA4GA1UE +BwwHTmV3WW9yazEbMBkGA1UECgwSRm94aHVudCBQcm9kdWN0aW9uMRkwFwYDVQQL +DBBUcmFkaW5nIFBsYXRmb3JtMRwwGgYDVQQDDBN0cmFkaW5nLmZveGh1bnQuY29t +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxT3Z1kgU17xC7NaRMv9y +9fb5VpNODPTu8Skbg8sIbGaFDSWRExjgNgktAJgb2QBx9dhfSU5g2kOqTgkhQIQh +3HrMFFvkFbjPLJsOxAju3ecVXABAtL7xOpRttgQftvmZMcwEhSX+2hQ7vkVp9U0w +jhX3KISk5WNcx3/vdRx3JaT/QbergyhkX9BDTa+rMq5ctVY0rbdb5UWo4kqOlUg0 +tnf6fsiie4qOIA4eF+QCZnrFoa8J9jwoWOvuYCJxCzWaEWySxOUco20cTE6qJrNs +6UYfadWrUNp2a8FS+KmieE9Htb/9dm/d6NjHJUTcYZyHDuGtCHAkCJn7R24194B5 +jwIDAQABMA0GCSqGSIb3DQEBCwUAA4ICAQCLUr0g2tvUwfjsb0MA1U1dkAvuPJ1O +HXCcyyW8B4XIjpQw3VQjm+QiNJ2Mfr/rWHvtBbYAunKJV90h2eY/WSDQJP9sZrKo +pGfpbDhHIiGhpB7C9uOT/myIUzoxZjBfOdMI8SUTZbgNC1q+3sZgiTqnO63/rIyM +2Fgdzjar+a+TSuVfqXIQlAMX4n+h3mW0TQ/rzPnqYUDYHMxFVhhNEtMMf6siqjX+ +VkO/KYeEhOBwr14yK5h3m/xNMNLWtdbG7kGF2KFTNQAUckLKZIacv+SEe/OMpQ4n +5Sw5v4/bIbgLmmH4gUEPL/3JghYcekWCA5JyA5TVr2cTccPb5uVYoGiii/Ua3mp6 +r5KHyQButoWat7BAag/4fk84Z/ykn6R1dz1W+8t9daG+bo1gSwYl3iqJApvFUWBe +ltMQQvkfTVqotLUx8ceGbg8O9Od3g97t89JmnhAYjZ+NkxjcbKhiD0WAVhn5Mg/C +sLfjDkoBjtBy6rfAozfPwU/eEk+pR3RqOJGd4/wlhRFYHIzQKX+dsDbOnS9TJrhf +JP8HiWwMC/jndvmKXtHMZ7agaF4oqCaCaI+QODfYwU0oarx9mLLP+LrnnEqcBfYV +tj4kx7bgupN4dStygNHgZjm7wlzoZSXNKDrUfkRmllBr6xe4Yjo7BzYte7YBtbfW +ozxL3tzWzJPUWQ== +-----END CERTIFICATE----- diff --git a/certs/production/foxhunt-key.pem.2048.backup b/certs/production/foxhunt-key.pem.2048.backup new file mode 100644 index 000000000..43356ffcb --- /dev/null +++ b/certs/production/foxhunt-key.pem.2048.backup @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDFPdnWSBTXvELs +1pEy/3L19vlWk04M9O7xKRuDywhsZoUNJZETGOA2CS0AmBvZAHH12F9JTmDaQ6pO +CSFAhCHceswUW+QVuM8smw7ECO7d5xVcAEC0vvE6lG22BB+2+ZkxzASFJf7aFDu+ +RWn1TTCOFfcohKTlY1zHf+91HHclpP9Bt6uDKGRf0ENNr6syrly1VjStt1vlRaji +So6VSDS2d/p+yKJ7io4gDh4X5AJmesWhrwn2PChY6+5gInELNZoRbJLE5RyjbRxM +Tqoms2zpRh9p1atQ2nZrwVL4qaJ4T0e1v/12b93o2MclRNxhnIcO4a0IcCQImftH +bjX3gHmPAgMBAAECggEAUlOjKYQTowpy2r0ihOlHuC96Wi8I3HKdiy5PzD5CCVDv +RxH/kDuJVjSVQpK6yOQdLPK/ru7WkTO+EW1RIkckhQ0CQpsqRADMBH/B9YQPA7MQ +4MJAP9EVChDrDBaDkFF1JmpUFqlJywe3+g6qD9fEpQG5FIHs+8vx7VlSfwy/FIgi +Qp5vk+D3P8mzFCRbqTtO8hBep/1zWyigWWJinnmfcgw9ZLrC1cwqrWFoXwt8KCbt +5nTL5UkT0pQIHWYqwjvJNt+8PT2MabD7dG5GtChshaI7VpxNngBtlbhN7/8SjNqF +RatW2pEYJXLLdWvvB3SpxGiK/eZ4jegHDwbdGVHh3QKBgQD0GPEyh0wgEDiDar39 +PakoDDLHrBa/ehk+B5Z7b6fVuWGzpGDPujf7T68jGNZxIEInk0bOOYhaaB6M9rik +mEVDnRZX+XMwO9ajxGPleHiL5vSZ5J+S3+IBjfQp0teTunjg/6/wCaTRbiSoFzuM +0s/wXIUR7J1HsMvQfkWYPd4ynQKBgQDO3AJtOfpOiGZl8A+MooxDRUsLHKnUp0sZ +efh9qJkO28+N6J168CmQdulLKtCwF0XdMMUm0xVus4maisKRdm3dfP+qipeD+qP6 +hdUmDYXW9kkbwp3Rx+z0N4mV34q/zpy1exVu9OG+/VmtCXJtVoKRwTTN/yPQJl4j +6Bzr3ia/GwKBgGVfKmQSnkG+msmw/b/66JIyJrCXvka83VjOa+dNYoU23OVvAamj +K3aaDCDmQsBuZirrVf91EzSAgxeFqv7oIfPHfC0C8px4bVZLzDTTSDTJsJFsyfyE +L+b3Xgw0fMTkLyk55oYoayNyu5R3OYrtTeADfFwKSPMp8CrhFeP45ZtdAoGBALgW +3WX9Cb/n7VDckcnzQmX/fag26wCfnNrgS/7wEr5BafV9ROCKaELzP+yv/qYnH/qX +k2xYv6MTzeqlgD/ZqC5LNm+8BNjkLhgeEDFcKXbDGQ7GZJKznDE+V4f0N1M4+Qcc +sVdcQKYqayrt1FJvgaVazrZgT6vG6VKFF2/k9JBvAoGAbbArW6Tj7lNYXcDbeaZe +GS4f3sCRUV+bz7bGEQhRp9qX6VlxiLpiNUfdDFVaMp59RoYGH/hKXW1Ar+PvyFQR +4dNZNowmIZPVWblk61IrgeCiTTWqE1OtBgwFfK5ENEFeGzi07HhINuJATvQyEIuA +5qY6IbOIIsg+3ntTm/o3cu0= +-----END PRIVATE KEY----- diff --git a/services/api_gateway/src/main.rs b/services/api_gateway/src/main.rs index 8a5d2dd98..bbd403fd6 100644 --- a/services/api_gateway/src/main.rs +++ b/services/api_gateway/src/main.rs @@ -221,21 +221,24 @@ async fn main() -> Result<()> { .await; } - // Start Prometheus metrics HTTP endpoint on port 9091 - tokio::spawn(async { - use axum::{Router, routing::get}; - use prometheus::{Encoder, TextEncoder}; + // Initialize and start Prometheus metrics HTTP endpoint on port 9091 + let gateway_metrics = api_gateway::metrics::GatewayMetrics::new() + .expect("Failed to initialize gateway metrics"); - async fn metrics_handler() -> String { - let encoder = TextEncoder::new(); - let metric_families = prometheus::gather(); - let mut buffer = vec![]; - encoder.encode(&metric_families, &mut buffer).unwrap(); - String::from_utf8(buffer).unwrap() - } + // Add service info metric (always present) + use prometheus::{register_gauge_with_registry, Opts}; + let service_info = register_gauge_with_registry!( + Opts::new("api_gateway_service_info", "API Gateway service information") + .const_label("version", env!("CARGO_PKG_VERSION")) + .const_label("service", "api_gateway"), + gateway_metrics.registry().as_ref() + ).expect("Failed to register service info"); + service_info.set(1.0); - let metrics_app = Router::new() - .route("/metrics", get(metrics_handler)); + let metrics_registry = gateway_metrics.registry(); + + tokio::spawn(async move { + let metrics_app = api_gateway::metrics::metrics_router(metrics_registry); let metrics_addr = "0.0.0.0:9091"; info!("Prometheus metrics endpoint listening on http://{}", metrics_addr); diff --git a/services/backtesting_service/Cargo.toml b/services/backtesting_service/Cargo.toml index 4dd79c034..17ba462de 100644 --- a/services/backtesting_service/Cargo.toml +++ b/services/backtesting_service/Cargo.toml @@ -38,6 +38,7 @@ axum.workspace = true # Performance monitoring prometheus.workspace = true +once_cell.workspace = true # Database - USE WORKSPACE sqlx.workspace = true diff --git a/services/backtesting_service/src/lib.rs b/services/backtesting_service/src/lib.rs index 004b8c7c7..ae5314b09 100644 --- a/services/backtesting_service/src/lib.rs +++ b/services/backtesting_service/src/lib.rs @@ -28,6 +28,9 @@ pub mod strategy_engine; /// TLS configuration pub mod tls_config; +/// Simple Prometheus metrics +pub mod simple_metrics; + /// Generated gRPC code #[allow(missing_docs)] pub mod foxhunt { diff --git a/services/backtesting_service/src/main.rs b/services/backtesting_service/src/main.rs index 7a9f12a24..bbc9787b3 100644 --- a/services/backtesting_service/src/main.rs +++ b/services/backtesting_service/src/main.rs @@ -191,6 +191,19 @@ async fn main() -> Result<()> { .max_concurrent_streams(Some(10_000)); // Increased from 1,024 to 10,000 for production scale } + // Initialize Prometheus metrics + backtesting_service::simple_metrics::init_metrics(); + let service_start = std::time::Instant::now(); + + // Spawn uptime updater + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + loop { + interval.tick().await; + backtesting_service::simple_metrics::update_uptime(service_start); + } + }); + // Start Prometheus metrics HTTP endpoint on port 9093 tokio::spawn(async { use axum::{Router, routing::get}; diff --git a/services/backtesting_service/src/simple_metrics.rs b/services/backtesting_service/src/simple_metrics.rs new file mode 100644 index 000000000..e207fd217 --- /dev/null +++ b/services/backtesting_service/src/simple_metrics.rs @@ -0,0 +1,52 @@ +//! Simple Prometheus metrics for backtesting service +//! +//! Provides basic service metrics using the global Prometheus registry + +use once_cell::sync::Lazy; +use prometheus::{register_gauge, register_counter, Gauge, Counter}; + +/// Service uptime in seconds +pub static SERVICE_UPTIME: Lazy = Lazy::new(|| { + register_gauge!( + "backtesting_service_uptime_seconds", + "Service uptime in seconds" + ).unwrap() +}); + +/// Total number of backtests started +pub static BACKTESTS_STARTED: Lazy = Lazy::new(|| { + register_counter!( + "backtesting_backtests_started_total", + "Total number of backtests started" + ).unwrap() +}); + +/// Total number of backtests completed +pub static BACKTESTS_COMPLETED: Lazy = Lazy::new(|| { + register_counter!( + "backtesting_backtests_completed_total", + "Total number of backtests completed" + ).unwrap() +}); + +/// Total number of backtest errors +pub static BACKTEST_ERRORS: Lazy = Lazy::new(|| { + register_counter!( + "backtesting_errors_total", + "Total number of backtest errors" + ).unwrap() +}); + +/// Initialize metrics (registers them with global registry) +pub fn init_metrics() { + // Force lazy initialization by accessing each metric + let _ = &*SERVICE_UPTIME; + let _ = &*BACKTESTS_STARTED; + let _ = &*BACKTESTS_COMPLETED; + let _ = &*BACKTEST_ERRORS; +} + +/// Update service uptime metric +pub fn update_uptime(start_time: std::time::Instant) { + SERVICE_UPTIME.set(start_time.elapsed().as_secs_f64()); +} diff --git a/services/integration_tests/Cargo.toml b/services/integration_tests/Cargo.toml index 875b65711..a6794b12c 100644 --- a/services/integration_tests/Cargo.toml +++ b/services/integration_tests/Cargo.toml @@ -8,9 +8,11 @@ publish = false [dependencies] # Async runtime tokio = { workspace = true, features = ["rt-multi-thread", "macros", "time"] } +futures.workspace = true # gRPC client tonic = { workspace = true } +tonic-prost.workspace = true prost = { workspace = true } # HTTP client for metrics scraping @@ -20,6 +22,9 @@ reqwest = { workspace = true, features = ["json"] } serde = { workspace = true, features = ["derive"] } serde_json.workspace = true +# JWT for authentication +jsonwebtoken.workspace = true + # Error handling anyhow.workspace = true thiserror.workspace = true @@ -39,4 +44,4 @@ chrono = { workspace = true } serial_test.workspace = true [build-dependencies] -tonic-build = "0.12" +tonic-prost-build.workspace = true diff --git a/services/integration_tests/build.rs b/services/integration_tests/build.rs index 37da8cfc2..1fa131519 100644 --- a/services/integration_tests/build.rs +++ b/services/integration_tests/build.rs @@ -1,6 +1,6 @@ fn main() -> Result<(), Box> { - // Compile TLI proto files for client testing - tonic_build::configure() + // Compile TLI proto files for client testing (Tonic 0.14+) + tonic_prost_build::configure() .build_server(false) .build_client(true) .compile_protos( @@ -13,5 +13,11 @@ fn main() -> Result<(), Box> { &["../../tli/proto"], )?; + // Tell cargo to recompile if proto files change + println!("cargo:rerun-if-changed=../../tli/proto/trading.proto"); + println!("cargo:rerun-if-changed=../../tli/proto/ml.proto"); + println!("cargo:rerun-if-changed=../../tli/proto/config.proto"); + println!("cargo:rerun-if-changed=../../tli/proto/health.proto"); + Ok(()) } diff --git a/services/integration_tests/tests/backtesting_service_e2e.rs b/services/integration_tests/tests/backtesting_service_e2e.rs index 6a05c008f..427a88cb7 100644 --- a/services/integration_tests/tests/backtesting_service_e2e.rs +++ b/services/integration_tests/tests/backtesting_service_e2e.rs @@ -176,7 +176,7 @@ async fn test_e2e_backtest_status() -> Result<()> { let status = status_response.into_inner(); assert_eq!(status.backtest_id, backtest_id); - assert!(status.status != BacktestStatus::BacktestStatusUnspecified as i32); + assert!(status.status != BacktestStatus::Unspecified as i32); println!("✓ Backtest status retrieved"); println!(" Status: {:?}", status.status); @@ -273,7 +273,7 @@ async fn test_e2e_backtest_results() -> Result<()> { if let Ok(response) = client.get_backtest_status(status_request).await { let status = response.into_inner(); - if status.status == BacktestStatus::BacktestStatusCompleted as i32 { + if status.status == BacktestStatus::Completed as i32 { completed = true; println!("✓ Backtest completed"); break; @@ -388,7 +388,7 @@ async fn test_e2e_backtest_progress_subscription() -> Result<()> { // Receive progress updates (with timeout) let mut updates_received = 0; - while let Ok(Some(update_result)) = timeout( + while let Ok(update_result) = timeout( StdDuration::from_secs(10), stream.message() ).await { @@ -402,7 +402,7 @@ async fn test_e2e_backtest_progress_subscription() -> Result<()> { ); // Stop after receiving 5 updates or if completed - if updates_received >= 5 || update.status == BacktestStatus::BacktestStatusCompleted as i32 { + if updates_received >= 5 || update.status == BacktestStatus::Completed as i32 { break; } } @@ -453,7 +453,7 @@ async fn test_e2e_backtest_filtering_by_status() -> Result<()> { limit: 10, offset: 0, strategy_name: None, - status_filter: Some(BacktestStatus::BacktestStatusCompleted as i32), + status_filter: Some(BacktestStatus::Completed as i32), }); let response = client.list_backtests(request).await?; @@ -464,7 +464,7 @@ async fn test_e2e_backtest_filtering_by_status() -> Result<()> { println!(" Count: {}", list_result.backtests.len()); for backtest in &list_result.backtests { - assert_eq!(backtest.status, BacktestStatus::BacktestStatusCompleted as i32); + assert_eq!(backtest.status, BacktestStatus::Completed as i32); } Ok(()) diff --git a/services/integration_tests/tests/ml_training_service_e2e.rs b/services/integration_tests/tests/ml_training_service_e2e.rs index 6c4912eca..6fb0ea2c1 100644 --- a/services/integration_tests/tests/ml_training_service_e2e.rs +++ b/services/integration_tests/tests/ml_training_service_e2e.rs @@ -283,7 +283,7 @@ async fn test_e2e_filter_training_jobs_by_status() -> Result<()> { let request = Request::new(ListTrainingJobsRequest { model_name: None, - status: Some(TrainingStatus::TrainingStatusCompleted as i32), + status: Some(TrainingStatus::Completed as i32), start_time_after: None, start_time_before: None, tags: vec![], @@ -299,7 +299,7 @@ async fn test_e2e_filter_training_jobs_by_status() -> Result<()> { println!(" Count: {}", list_result.jobs.len()); for job in &list_result.jobs { - assert_eq!(job.status, TrainingStatus::TrainingStatusCompleted as i32); + assert_eq!(job.status, TrainingStatus::Completed as i32); } Ok(()) @@ -359,7 +359,7 @@ async fn test_e2e_watch_training_progress() -> Result<()> { // Receive progress updates (with timeout) let mut updates_received = 0; - while let Ok(Some(update_result)) = timeout( + while let Ok(update_result) = timeout( StdDuration::from_secs(10), stream.message() ).await { @@ -385,7 +385,7 @@ async fn test_e2e_watch_training_progress() -> Result<()> { } // Stop after 5 updates or if completed - if updates_received >= 5 || update.status == TrainingStatus::TrainingStatusCompleted as i32 { + if updates_received >= 5 || update.status == TrainingStatus::Completed as i32 { break; } } @@ -439,7 +439,7 @@ async fn test_e2e_stream_resource_metrics() -> Result<()> { // Receive resource metric updates (with timeout) let mut updates_received = 0; - while let Ok(Some(update_result)) = timeout( + while let Ok(update_result) = timeout( StdDuration::from_secs(5), stream.message() ).await { diff --git a/services/integration_tests/tests/service_health_resilience_e2e.rs b/services/integration_tests/tests/service_health_resilience_e2e.rs index 2a259fd4d..51ea9bfbc 100644 --- a/services/integration_tests/tests/service_health_resilience_e2e.rs +++ b/services/integration_tests/tests/service_health_resilience_e2e.rs @@ -112,7 +112,7 @@ async fn test_e2e_system_health_all_services() -> Result<()> { // At least API Gateway should be healthy let api_gateway_healthy = system_status.services.iter() - .any(|s| s.name.contains("gateway") && s.status == SystemStatus::SystemStatusHealthy as i32); + .any(|s| s.name.contains("gateway") && s.status == SystemStatus::Healthy as i32); assert!(api_gateway_healthy, "API Gateway should be healthy"); @@ -228,7 +228,7 @@ async fn test_e2e_health_status_transitions() -> Result<()> { // Monitor status changes for a short period let mut events_received = 0; - while let Ok(Some(event_result)) = timeout( + while let Ok(event_result) = timeout( StdDuration::from_secs(5), stream.message() ).await { @@ -279,7 +279,7 @@ async fn test_e2e_degraded_service_detection() -> Result<()> { println!("✓ Checking for degraded services"); let degraded_services: Vec<_> = system_status.services.iter() - .filter(|s| s.status == SystemStatus::SystemStatusDegraded as i32) + .filter(|s| s.status == SystemStatus::Degraded as i32) .collect(); println!(" Degraded Services: {}", degraded_services.len()); @@ -290,7 +290,7 @@ async fn test_e2e_degraded_service_detection() -> Result<()> { // System should handle degraded services gracefully assert!( - system_status.overall_status != SystemStatus::SystemStatusCritical as i32, + system_status.overall_status != SystemStatus::Critical as i32, "Overall system should not be critical" ); @@ -321,8 +321,8 @@ async fn test_e2e_trading_service_available_backtesting_optional() -> Result<()> // Trading service should work (core functionality) let order_request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.01, price: None, stop_price: None, @@ -387,7 +387,7 @@ async fn test_e2e_partial_service_failure_handling() -> Result<()> { let system_status = status_response.into_inner(); let healthy_services = system_status.services.iter() - .filter(|s| s.status == SystemStatus::SystemStatusHealthy as i32) + .filter(|s| s.status == SystemStatus::Healthy as i32) .count(); let total_services = system_status.services.len(); @@ -398,7 +398,7 @@ async fn test_e2e_partial_service_failure_handling() -> Result<()> { // System should remain operational with partial failures assert!( - system_status.overall_status != SystemStatus::SystemStatusUnknown as i32, + system_status.overall_status != SystemStatus::Unknown as i32, "System should report known status" ); @@ -429,8 +429,8 @@ async fn test_e2e_circuit_breaker_validation() -> Result<()> { for i in 0..20 { let request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.001, price: None, stop_price: None, @@ -531,8 +531,8 @@ async fn test_e2e_retry_logic_validation() -> Result<()> { while retry_count < max_retries { let request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.001, price: None, stop_price: None, diff --git a/services/integration_tests/tests/trading_service_e2e.rs b/services/integration_tests/tests/trading_service_e2e.rs index e74639bf6..a243d3b9d 100644 --- a/services/integration_tests/tests/trading_service_e2e.rs +++ b/services/integration_tests/tests/trading_service_e2e.rs @@ -104,8 +104,8 @@ async fn test_e2e_order_submission_market_order() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.1, price: None, stop_price: None, @@ -135,8 +135,8 @@ async fn test_e2e_order_submission_limit_order() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: "ETH/USD".to_string(), - side: OrderSide::OrderSideSell as i32, - order_type: OrderType::OrderTypeLimit as i32, + side: OrderSide::Sell as i32, + order_type: OrderType::Limit as i32, quantity: 1.5, price: Some(3500.0), stop_price: None, @@ -170,8 +170,8 @@ async fn test_e2e_order_submission_without_auth() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.1, price: None, stop_price: None, @@ -202,8 +202,8 @@ async fn test_e2e_order_cancellation() -> Result<()> { // First, submit an order let submit_request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeLimit as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Limit as i32, quantity: 0.1, price: Some(50000.0), stop_price: None, @@ -245,8 +245,8 @@ async fn test_e2e_order_status_query() -> Result<()> { // Submit an order first let submit_request = Request::new(SubmitOrderRequest { symbol: "ETH/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.5, price: None, stop_price: None, @@ -373,8 +373,8 @@ async fn test_e2e_market_data_subscription() -> Result<()> { let request = Request::new(SubscribeMarketDataRequest { symbols: vec!["BTC/USD".to_string(), "ETH/USD".to_string()], data_types: vec![ - MarketDataType::MarketDataTypeTrades as i32, - MarketDataType::MarketDataTypeQuotes as i32, + MarketDataType::Trades as i32, + MarketDataType::Quotes as i32, ], }); @@ -384,7 +384,7 @@ async fn test_e2e_market_data_subscription() -> Result<()> { // Receive first 5 market data events (with timeout) let mut events_received = 0; - while let Ok(Some(event_result)) = timeout( + while let Ok(event_result) = timeout( StdDuration::from_secs(5), stream.message() ).await { @@ -422,8 +422,8 @@ async fn test_e2e_order_updates_subscription() -> Result<()> { // Submit an order to generate an update let submit_request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.01, price: None, stop_price: None, @@ -435,7 +435,7 @@ async fn test_e2e_order_updates_subscription() -> Result<()> { println!("✓ Test order submitted"); // Wait for order update (with timeout) - if let Ok(Some(update_result)) = timeout( + if let Ok(update_result) = timeout( StdDuration::from_secs(3), stream.message() ).await { @@ -464,8 +464,8 @@ async fn test_e2e_concurrent_order_submissions() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: if i % 2 == 0 { "BTC/USD" } else { "ETH/USD" }.to_string(), - side: if i % 2 == 0 { OrderSide::OrderSideBuy } else { OrderSide::OrderSideSell } as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: if i % 2 == 0 { OrderSide::Buy } else { OrderSide::Sell } as i32, + order_type: OrderType::Market as i32, quantity: 0.01, price: None, stop_price: None, @@ -548,8 +548,8 @@ async fn test_e2e_invalid_symbol_handling() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: "INVALID_SYMBOL_XYZ".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: 0.1, price: None, stop_price: None, @@ -581,8 +581,8 @@ async fn test_e2e_negative_quantity_validation() -> Result<()> { let request = Request::new(SubmitOrderRequest { symbol: "BTC/USD".to_string(), - side: OrderSide::OrderSideBuy as i32, - order_type: OrderType::OrderTypeMarket as i32, + side: OrderSide::Buy as i32, + order_type: OrderType::Market as i32, quantity: -0.5, // Invalid negative quantity price: None, stop_price: None, diff --git a/services/load_tests/throughput_tests.rs b/services/load_tests/tests/throughput_tests.rs similarity index 100% rename from services/load_tests/throughput_tests.rs rename to services/load_tests/tests/throughput_tests.rs diff --git a/services/ml_training_service/Cargo.toml b/services/ml_training_service/Cargo.toml index bf3507b86..397d19a16 100644 --- a/services/ml_training_service/Cargo.toml +++ b/services/ml_training_service/Cargo.toml @@ -45,6 +45,7 @@ tracing-subscriber.workspace = true metrics.workspace = true metrics-exporter-prometheus.workspace = true prometheus.workspace = true +once_cell.workspace = true # Utilities - USE WORKSPACE base64.workspace = true diff --git a/services/ml_training_service/src/lib.rs b/services/ml_training_service/src/lib.rs index f146c4582..ad8dbaaa3 100644 --- a/services/ml_training_service/src/lib.rs +++ b/services/ml_training_service/src/lib.rs @@ -15,6 +15,7 @@ pub mod schema_types; pub mod service; pub mod storage; pub mod technical_indicators; +pub mod simple_metrics; /// Error types for the ML training service pub mod errors { diff --git a/services/ml_training_service/src/main.rs b/services/ml_training_service/src/main.rs index 81dfc6aac..ef5ac2f88 100644 --- a/services/ml_training_service/src/main.rs +++ b/services/ml_training_service/src/main.rs @@ -369,6 +369,19 @@ async fn serve(args: ServeArgs) -> Result<()> { info!("gRPC reflection enabled for development"); } + // Initialize Prometheus metrics + ml_training_service::simple_metrics::init_metrics(); + let service_start = std::time::Instant::now(); + + // Spawn uptime updater + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + loop { + interval.tick().await; + ml_training_service::simple_metrics::update_uptime(service_start); + } + }); + // Start Prometheus metrics HTTP endpoint on port 9094 tokio::spawn(async { use axum::{Router, routing::get}; diff --git a/services/ml_training_service/src/simple_metrics.rs b/services/ml_training_service/src/simple_metrics.rs new file mode 100644 index 000000000..e19847fe1 --- /dev/null +++ b/services/ml_training_service/src/simple_metrics.rs @@ -0,0 +1,52 @@ +//! Simple Prometheus metrics for ML training service +//! +//! Provides basic service metrics using the global Prometheus registry + +use once_cell::sync::Lazy; +use prometheus::{register_gauge, register_counter, Gauge, Counter}; + +/// Service uptime in seconds +pub static SERVICE_UPTIME: Lazy = Lazy::new(|| { + register_gauge!( + "ml_training_service_uptime_seconds", + "Service uptime in seconds" + ).unwrap() +}); + +/// Total number of training jobs started +pub static TRAINING_JOBS_STARTED: Lazy = Lazy::new(|| { + register_counter!( + "ml_training_jobs_started_total", + "Total number of training jobs started" + ).unwrap() +}); + +/// Total number of training jobs completed +pub static TRAINING_JOBS_COMPLETED: Lazy = Lazy::new(|| { + register_counter!( + "ml_training_jobs_completed_total", + "Total number of training jobs completed" + ).unwrap() +}); + +/// Total number of training errors +pub static TRAINING_ERRORS: Lazy = Lazy::new(|| { + register_counter!( + "ml_training_errors_total", + "Total number of training errors" + ).unwrap() +}); + +/// Initialize metrics (registers them with global registry) +pub fn init_metrics() { + // Force lazy initialization by accessing each metric + let _ = &*SERVICE_UPTIME; + let _ = &*TRAINING_JOBS_STARTED; + let _ = &*TRAINING_JOBS_COMPLETED; + let _ = &*TRAINING_ERRORS; +} + +/// Update service uptime metric +pub fn update_uptime(start_time: std::time::Instant) { + SERVICE_UPTIME.set(start_time.elapsed().as_secs_f64()); +} diff --git a/services/trading_service/src/lib.rs b/services/trading_service/src/lib.rs index d3c0d9b8c..b1730639c 100644 --- a/services/trading_service/src/lib.rs +++ b/services/trading_service/src/lib.rs @@ -95,6 +95,9 @@ pub mod model_loader_stub; /// Prometheus metrics for ML model monitoring pub mod ml_metrics; +/// Prometheus metrics server for trading operations +pub mod metrics_server; + /// Streaming infrastructure and HTTP/2 optimizations pub mod streaming; diff --git a/services/trading_service/src/main.rs b/services/trading_service/src/main.rs index 13e08a6c9..48a3ed6c1 100644 --- a/services/trading_service/src/main.rs +++ b/services/trading_service/src/main.rs @@ -349,31 +349,20 @@ async fn main() -> Result<()> { } // Start Prometheus metrics HTTP endpoint on port 9092 - tokio::spawn(async { - use axum::{Router, routing::get}; - use prometheus::{Encoder, TextEncoder}; + use trading_service::metrics_server::{TradingMetricsServer, MetricsServerConfig}; - async fn metrics_handler() -> String { - let encoder = TextEncoder::new(); - let metric_families = prometheus::gather(); - let mut buffer = vec![]; - encoder.encode(&metric_families, &mut buffer).unwrap(); - String::from_utf8(buffer).unwrap() + let metrics_config = MetricsServerConfig { + bind_address: "0.0.0.0".to_string(), + bind_port: 9092, + scrape_timeout_ms: 500, + max_metrics_per_scrape: 10000, + }; + + let metrics_server = TradingMetricsServer::new(metrics_config); + tokio::spawn(async move { + if let Err(e) = metrics_server.start().await { + tracing::error!("Metrics server failed: {}", e); } - - let metrics_app = Router::new() - .route("/metrics", get(metrics_handler)); - - let metrics_addr = "0.0.0.0:9092"; - info!("Prometheus metrics endpoint listening on http://{}", metrics_addr); - - let listener = tokio::net::TcpListener::bind(metrics_addr) - .await - .expect("Failed to bind metrics endpoint"); - - axum::serve(listener, metrics_app) - .await - .expect("Metrics server failed"); }); // Apply authentication interceptor to all gRPC services