Backend (Rust): - Create proto/auth.proto with Login and RefreshToken RPCs - Implement AuthGrpcService in services/api with JWT issuance - Register as unauthenticated service in main.rs (pre-auth) - Update JWT issuer from foxhunt-api-gateway to foxhunt-api Dashboard (TypeScript): - Install @connectrpc/connect-web + @bufbuild/protobuf - Generate TypeScript proto clients via buf (src/gen/) - Create grpc.ts transport with JWT interceptor - Rewrite all useApi hooks to typed gRPC calls - Replace WebSocket with useGrpcStream server-streaming hooks - Migrate all 6 pages + 6 components to grpc-web - Delete api.ts, websocket.ts, useWebSocket.ts Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
780 lines
32 KiB
Rust
780 lines
32 KiB
Rust
#![deny(clippy::unwrap_used, clippy::expect_used)]
|
|
|
|
//! Foxhunt API Service
|
|
//!
|
|
//! Unified gRPC gateway with tonic-web for browser access and 6-layer authentication.
|
|
//! Optimized for HFT requirements with <10us authentication overhead.
|
|
|
|
use anyhow::Result;
|
|
use clap::Parser;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
use tracing::{error, info, warn};
|
|
|
|
use common::metrics::GrpcMetricsLayer;
|
|
use tonic_web::GrpcWebLayer;
|
|
use tower_http::cors::{AllowHeaders, CorsLayer};
|
|
|
|
// Import all needed types from the library
|
|
use api::auth::jwt::JwtConfig;
|
|
use api::auth::{
|
|
AuditLogger, AuthInterceptor, AuthzService, JwtService, RateLimiter, RevocationService,
|
|
};
|
|
|
|
#[derive(Parser, Debug)]
|
|
#[command(name = "api", about = "Foxhunt API Service")]
|
|
struct Args {
|
|
/// gRPC server bind address
|
|
#[arg(long, env = "GATEWAY_BIND_ADDR", default_value = "0.0.0.0:50051")]
|
|
bind_addr: String,
|
|
|
|
/// JWT secret (or use JWT_SECRET_FILE for production)
|
|
#[arg(long, env = "JWT_SECRET")]
|
|
jwt_secret: Option<String>,
|
|
|
|
/// JWT issuer
|
|
#[arg(long, env = "JWT_ISSUER", default_value = "foxhunt-trading")]
|
|
jwt_issuer: String,
|
|
|
|
/// JWT audience
|
|
#[arg(long, env = "JWT_AUDIENCE", default_value = "trading-api")]
|
|
jwt_audience: String,
|
|
|
|
/// Redis URL for JWT revocation
|
|
#[arg(long, env = "REDIS_URL", default_value = "redis://localhost:6379")]
|
|
redis_url: String,
|
|
|
|
/// Rate limit (requests per second per user)
|
|
#[arg(long, env = "RATE_LIMIT_RPS", default_value = "100")]
|
|
rate_limit_rps: u32,
|
|
|
|
/// Enable audit logging
|
|
#[arg(long, env = "ENABLE_AUDIT_LOGGING", default_value = "true")]
|
|
enable_audit_logging: bool,
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
// Install ring as the default rustls CryptoProvider (required by kube-rs for in-cluster TLS)
|
|
if let Err(_already) = rustls::crypto::ring::default_provider().install_default() {
|
|
// Another provider already installed -- fine
|
|
}
|
|
|
|
let args = Args::parse();
|
|
|
|
// 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());
|
|
if let Err(e) = common::observability::init_observability("api", Some(&otlp_endpoint)) {
|
|
eprintln!("Failed to initialize observability: {}", e);
|
|
}
|
|
|
|
info!("Starting Foxhunt API Service");
|
|
info!("Bind address: {}", args.bind_addr);
|
|
info!("JWT issuer: {}", args.jwt_issuer);
|
|
info!("JWT audience: {}", args.jwt_audience);
|
|
info!("Redis URL: {}", args.redis_url);
|
|
info!("Rate limit: {} req/s per user", args.rate_limit_rps);
|
|
info!("Audit logging: {}", args.enable_audit_logging);
|
|
|
|
// Load JWT configuration (Vault-based with fallback)
|
|
info!("Loading JWT configuration...");
|
|
let jwt_config = match load_jwt_config().await {
|
|
Ok(config) => {
|
|
info!("JWT configuration loaded successfully");
|
|
config
|
|
},
|
|
Err(e) => {
|
|
error!("Failed to load JWT configuration: {}", e);
|
|
return Err(e);
|
|
},
|
|
};
|
|
|
|
// Initialize authentication components
|
|
info!("Initializing authentication services...");
|
|
|
|
let jwt_service = JwtService::new(
|
|
jwt_config.jwt_secret.clone(),
|
|
jwt_config.jwt_issuer.clone(),
|
|
jwt_config.jwt_audience.clone(),
|
|
);
|
|
info!("JWT service initialized with cached decoding key");
|
|
|
|
let revocation_service = RevocationService::new(&args.redis_url)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to connect to Redis for revocation service: {}", e))?;
|
|
info!("JWT revocation service connected to Redis");
|
|
|
|
let authz_service = AuthzService::new();
|
|
info!("Authorization service initialized with permission cache");
|
|
|
|
let rate_limiter = RateLimiter::new(args.rate_limit_rps)
|
|
.map_err(|e| anyhow::anyhow!("Failed to create rate limiter: {}", e))?;
|
|
info!("Rate limiter initialized ({} req/s)", args.rate_limit_rps);
|
|
|
|
let audit_logger = AuditLogger::new(args.enable_audit_logging);
|
|
info!("Audit logger initialized");
|
|
|
|
// Create authentication interceptor
|
|
let auth_interceptor = AuthInterceptor::new(
|
|
jwt_service,
|
|
revocation_service,
|
|
authz_service,
|
|
rate_limiter,
|
|
audit_logger,
|
|
);
|
|
info!("6-layer authentication interceptor ready");
|
|
|
|
info!("API Service initialization complete");
|
|
info!("Ready to accept gRPC requests with <10us auth overhead");
|
|
|
|
// Initialize backend service proxies
|
|
info!("Connecting to backend services...");
|
|
|
|
let trading_backend_url = std::env::var("TRADING_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50052".to_string());
|
|
let backtesting_backend_url = std::env::var("BACKTESTING_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50053".to_string());
|
|
let ml_training_backend_url = std::env::var("ML_TRAINING_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50054".to_string());
|
|
let broker_gw_url = std::env::var("BROKER_GATEWAY_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50056".to_string());
|
|
let data_acq_url = std::env::var("DATA_ACQUISITION_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50057".to_string());
|
|
let ml_service_url = std::env::var("ML_SERVICE_URL")
|
|
.unwrap_or_else(|_| trading_backend_url.clone());
|
|
let trading_agent_url = std::env::var("TRADING_AGENT_SERVICE_URL")
|
|
.unwrap_or_else(|_| "http://localhost:50055".to_string());
|
|
let prometheus_url = std::env::var("PROMETHEUS_URL")
|
|
.unwrap_or_else(|_| "http://prometheus.foxhunt.svc:80".to_string());
|
|
let monitoring_stream_interval: u32 = std::env::var("MONITORING_STREAM_INTERVAL")
|
|
.ok()
|
|
.and_then(|v| v.parse().ok())
|
|
.unwrap_or(3);
|
|
|
|
// Load TLS certificate paths for backtesting service (mTLS)
|
|
let backtesting_tls_ca_cert = std::env::var("BACKTESTING_TLS_CA_CERT").ok();
|
|
let backtesting_tls_client_cert = std::env::var("BACKTESTING_TLS_CLIENT_CERT").ok();
|
|
let backtesting_tls_client_key = std::env::var("BACKTESTING_TLS_CLIENT_KEY").ok();
|
|
|
|
// Load TLS certificate paths for ML training service (mTLS)
|
|
let ml_training_tls_ca_cert = std::env::var("ML_TRAINING_TLS_CA_CERT").ok();
|
|
let ml_training_tls_client_cert = std::env::var("ML_TRAINING_TLS_CLIENT_CERT").ok();
|
|
let ml_training_tls_client_key = std::env::var("ML_TRAINING_TLS_CLIENT_KEY").ok();
|
|
|
|
// Initialize trading service proxy
|
|
let mut trading_proxy = api::grpc::TradingServiceProxy::new_lazy(&trading_backend_url)
|
|
.map_err(|e| anyhow::anyhow!("Failed to create trading service proxy: {}", e))?;
|
|
info!(
|
|
"Trading service proxy initialized ({})",
|
|
trading_backend_url
|
|
);
|
|
|
|
// Initialize backtesting service proxy (optional - graceful degradation)
|
|
info!("Attempting to initialize Backtesting Service proxy...");
|
|
info!(" Backend URL: {}", backtesting_backend_url);
|
|
info!(" CA cert: {:?}", backtesting_tls_ca_cert);
|
|
info!(" Client cert: {:?}", backtesting_tls_client_cert);
|
|
info!(" Client key: {:?}", backtesting_tls_client_key);
|
|
|
|
let backtesting_proxy = match api::grpc::BacktestingServiceProxy::new(
|
|
&backtesting_backend_url,
|
|
backtesting_tls_ca_cert.as_deref(),
|
|
backtesting_tls_client_cert.as_deref(),
|
|
backtesting_tls_client_key.as_deref(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(proxy) => {
|
|
info!(
|
|
"Backtesting service proxy initialized ({})",
|
|
backtesting_backend_url
|
|
);
|
|
Some(Arc::new(proxy))
|
|
},
|
|
Err(e) => {
|
|
error!("Backtesting service initialization failed!");
|
|
error!(" Error type: {:?}", e);
|
|
error!(" Error message: {}", e);
|
|
warn!("Backtesting service unavailable: {}. API will run without backtesting endpoints.", e);
|
|
warn!(" Backtesting endpoints will return 503 Service Unavailable");
|
|
None
|
|
},
|
|
};
|
|
|
|
// Spawn background health check task for backtesting service
|
|
if let Some(proxy) = backtesting_proxy.as_ref() {
|
|
let proxy_clone = Arc::clone(proxy);
|
|
tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(Duration::from_secs(10));
|
|
loop {
|
|
interval.tick().await;
|
|
proxy_clone.background_health_check().await;
|
|
}
|
|
});
|
|
info!("Backtesting service health check task started (10s interval)");
|
|
}
|
|
|
|
// Initialize ML training service proxy (optional - graceful degradation)
|
|
info!("Attempting to initialize ML Training Service proxy...");
|
|
info!(" Backend URL: {}", ml_training_backend_url);
|
|
info!(" CA cert: {:?}", ml_training_tls_ca_cert);
|
|
info!(" Client cert: {:?}", ml_training_tls_client_cert);
|
|
info!(" Client key: {:?}", ml_training_tls_client_key);
|
|
|
|
let ml_config = api::grpc::MlTrainingBackendConfig {
|
|
address: ml_training_backend_url.clone(),
|
|
connect_timeout_ms: 5000,
|
|
request_timeout_ms: 30000,
|
|
circuit_breaker_failures: 5,
|
|
circuit_breaker_reset_secs: 30,
|
|
tls_ca_cert_path: ml_training_tls_ca_cert.clone(),
|
|
tls_client_cert_path: ml_training_tls_client_cert.clone(),
|
|
tls_client_key_path: ml_training_tls_client_key.clone(),
|
|
};
|
|
let ml_training_proxy = match api::grpc::setup_ml_training_proxy(ml_config).await {
|
|
Ok(proxy) => {
|
|
info!(
|
|
"ML training service proxy initialized ({})",
|
|
ml_training_backend_url
|
|
);
|
|
Some(proxy)
|
|
},
|
|
Err(e) => {
|
|
error!("ML Training service initialization failed!");
|
|
error!(" Error type: {:?}", e);
|
|
error!(" Error message: {}", e);
|
|
warn!(
|
|
"ML Training service unavailable: {}. API will run without ML endpoints.",
|
|
e
|
|
);
|
|
warn!(" ML training endpoints will return 503 Service Unavailable");
|
|
None
|
|
},
|
|
};
|
|
|
|
// Initialize monitoring service handler (direct Prometheus scraping, no backend needed)
|
|
let monitoring_backends = vec![
|
|
("trading-service".to_string(), trading_backend_url.clone()),
|
|
("backtesting-service".to_string(), backtesting_backend_url.clone()),
|
|
("ml-training-service".to_string(), ml_training_backend_url.clone()),
|
|
("broker-gateway".to_string(), broker_gw_url.clone()),
|
|
("data-acquisition-service".to_string(), data_acq_url.clone()),
|
|
("trading-agent-service".to_string(), trading_agent_url.clone()),
|
|
];
|
|
let monitoring_handler = api::MonitoringServiceHandler::new(
|
|
&prometheus_url,
|
|
monitoring_stream_interval,
|
|
)
|
|
.with_backends(monitoring_backends);
|
|
info!(
|
|
"Monitoring service handler initialized (Prometheus={}, interval={}s, backends=6)",
|
|
prometheus_url, monitoring_stream_interval
|
|
);
|
|
|
|
// Initialize direct proxy services (trading, risk, config, ML, broker_gw, data_acq, trading_agent)
|
|
let trading_channel = tonic::transport::Channel::from_shared(trading_backend_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Invalid TRADING_SERVICE_URL: {e}"))?
|
|
.connect_lazy();
|
|
|
|
let trading_direct_proxy = api::TradingDirectProxy::new(
|
|
api::trading_backend::trading_service_client::TradingServiceClient::new(
|
|
trading_channel.clone(),
|
|
),
|
|
);
|
|
let risk_proxy = api::RiskServiceProxy::new(
|
|
api::risk::risk_service_client::RiskServiceClient::new(trading_channel.clone()),
|
|
);
|
|
let config_direct_proxy = api::ConfigServiceProxy::new(
|
|
api::config_backend::config_service_client::ConfigServiceClient::new(
|
|
trading_channel,
|
|
),
|
|
);
|
|
|
|
let ml_channel = tonic::transport::Channel::from_shared(ml_service_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Invalid ML_SERVICE_URL: {e}"))?
|
|
.connect_lazy();
|
|
let ml_proxy = api::MlServiceProxy::new(
|
|
api::ml_inference::ml_service_client::MlServiceClient::new(ml_channel),
|
|
);
|
|
|
|
let broker_channel = tonic::transport::Channel::from_shared(broker_gw_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Invalid BROKER_GATEWAY_SERVICE_URL: {e}"))?
|
|
.connect_lazy();
|
|
let broker_proxy = api::BrokerGatewayProxy::new(
|
|
api::broker_gateway::broker_gateway_service_client::BrokerGatewayServiceClient::new(
|
|
broker_channel,
|
|
),
|
|
);
|
|
|
|
let data_channel = tonic::transport::Channel::from_shared(data_acq_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Invalid DATA_ACQUISITION_SERVICE_URL: {e}"))?
|
|
.connect_lazy();
|
|
let data_proxy = api::DataAcquisitionProxy::new(
|
|
api::data_acquisition::data_acquisition_service_client::DataAcquisitionServiceClient::new(
|
|
data_channel,
|
|
),
|
|
);
|
|
|
|
let agent_channel = tonic::transport::Channel::from_shared(trading_agent_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Invalid TRADING_AGENT_SERVICE_URL: {e}"))?
|
|
.connect_lazy();
|
|
let agent_proxy = api::TradingAgentProxy::new(
|
|
api::trading_agent::trading_agent_service_client::TradingAgentServiceClient::new(
|
|
agent_channel,
|
|
),
|
|
);
|
|
|
|
info!("Direct proxy services initialized:");
|
|
info!(" - trading.TradingService -> {}", trading_backend_url);
|
|
info!(" - risk.RiskService -> {}", trading_backend_url);
|
|
info!(" - config.ConfigService -> {}", trading_backend_url);
|
|
info!(" - ml.MLService -> {}", ml_service_url);
|
|
info!(" - broker_gateway.BrokerGatewayService -> {}", broker_gw_url);
|
|
info!(" - data_acquisition.DataAcquisitionService -> {}", data_acq_url);
|
|
info!(" - trading_agent.TradingAgentService -> {}", trading_agent_url);
|
|
|
|
// Setup backend health monitoring for SubscribeSystemStatus.
|
|
// A background task checks gRPC health on each backend and pushes
|
|
// updates through a watch channel consumed by the streaming RPC.
|
|
let (health_tx, health_rx) = api::BackendHealthState::new();
|
|
trading_proxy.set_health_state(health_rx);
|
|
{
|
|
use tonic_health::pb::health_client::HealthClient;
|
|
|
|
// Build persistent lazy channels (reused across checks)
|
|
let backends: Vec<(&str, tonic::transport::Channel)> = vec![
|
|
("trading-service", tonic::transport::Channel::from_shared(trading_backend_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid trading URL: {}", e))?.connect_lazy()),
|
|
("ml-training-service", tonic::transport::Channel::from_shared(ml_training_backend_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid ml-training URL: {}", e))?.connect_lazy()),
|
|
("backtesting-service", tonic::transport::Channel::from_shared(backtesting_backend_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid backtesting URL: {}", e))?.connect_lazy()),
|
|
("broker-gateway", tonic::transport::Channel::from_shared(broker_gw_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid broker-gateway URL: {}", e))?.connect_lazy()),
|
|
("data-acquisition-service", tonic::transport::Channel::from_shared(data_acq_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid data-acquisition URL: {}", e))?.connect_lazy()),
|
|
("trading-agent-service", tonic::transport::Channel::from_shared(trading_agent_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("invalid trading-agent URL: {}", e))?.connect_lazy()),
|
|
];
|
|
let backends: Vec<(String, tonic::transport::Channel)> = backends
|
|
.into_iter()
|
|
.map(|(n, c)| (n.to_string(), c))
|
|
.collect();
|
|
|
|
tokio::spawn(async move {
|
|
let mut interval = tokio::time::interval(Duration::from_secs(5));
|
|
loop {
|
|
interval.tick().await;
|
|
let mut entries = Vec::with_capacity(backends.len());
|
|
for (name, channel) in &backends {
|
|
let mut hc = HealthClient::new(channel.clone());
|
|
let req = tonic_health::pb::HealthCheckRequest {
|
|
service: String::new(), // overall server health
|
|
};
|
|
let (healthy, msg) = match tokio::time::timeout(
|
|
Duration::from_millis(2000),
|
|
hc.check(req),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(_)) => (true, "Serving"),
|
|
// Unimplemented means gRPC server is UP but has no health service
|
|
Ok(Err(s)) if s.code() == tonic::Code::Unimplemented => {
|
|
(true, "Serving (no health check)")
|
|
}
|
|
Ok(Err(_)) => (false, "Error"),
|
|
Err(_) => (false, "Unreachable"),
|
|
};
|
|
entries.push(api::ServiceHealthEntry {
|
|
name: name.clone(),
|
|
healthy,
|
|
message: msg.to_string(),
|
|
});
|
|
}
|
|
let _ = health_tx.send(entries);
|
|
}
|
|
});
|
|
info!("Backend health monitor started (5s interval)");
|
|
}
|
|
|
|
// Initialize configuration manager (requires database)
|
|
let database_url = std::env::var("DATABASE_URL")
|
|
.map_err(|_| anyhow::anyhow!("DATABASE_URL environment variable must be set"))?;
|
|
let db_pool = sqlx::PgPool::connect(&database_url)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to connect to database: {}", e))?;
|
|
info!("Database connection established");
|
|
|
|
// Create Redis connection for config manager
|
|
let redis_client = redis::Client::open(args.redis_url.clone())
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Redis client: {}", e))?;
|
|
let redis_conn = redis::aio::ConnectionManager::new(redis_client)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create Redis connection manager: {}", e))?;
|
|
|
|
let mut config_manager = api::ConfigurationManager::new(db_pool.clone(), redis_conn)
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to create configuration manager: {}", e))?;
|
|
|
|
// Start NOTIFY listener for hot-reload
|
|
config_manager
|
|
.start_listening()
|
|
.await
|
|
.map_err(|e| anyhow::anyhow!("Failed to start configuration listener: {}", e))?;
|
|
info!("Configuration manager initialized with hot-reload");
|
|
|
|
// Load TLS configuration if enabled (Wave H1 Security Enforcement)
|
|
let tls_enabled = std::env::var("TLS_ENABLED")
|
|
.unwrap_or_else(|_| "false".to_string())
|
|
.parse::<bool>()
|
|
.unwrap_or(false);
|
|
|
|
let tls_config = if tls_enabled {
|
|
info!("TLS/mTLS enabled - initializing TLS 1.3 configuration");
|
|
|
|
let cert_path = std::env::var("TLS_CERT_PATH")
|
|
.unwrap_or_else(|_| "./certs/server-cert.pem".to_string());
|
|
let key_path =
|
|
std::env::var("TLS_KEY_PATH").unwrap_or_else(|_| "./certs/server-key.pem".to_string());
|
|
let ca_path =
|
|
std::env::var("TLS_CA_PATH").unwrap_or_else(|_| "./certs/ca/ca-cert.pem".to_string());
|
|
let require_client_cert = std::env::var("TLS_REQUIRE_CLIENT_CERT")
|
|
.unwrap_or_else(|_| "true".to_string())
|
|
.parse::<bool>()
|
|
.unwrap_or(true);
|
|
let enable_revocation = std::env::var("MTLS_ENABLE_REVOCATION_CHECK")
|
|
.unwrap_or_else(|_| "false".to_string())
|
|
.parse::<bool>()
|
|
.unwrap_or(false);
|
|
let crl_url = std::env::var("MTLS_CRL_URL").ok();
|
|
|
|
let tls = api::auth::mtls::ApiGatewayTlsConfig::from_files(
|
|
&cert_path,
|
|
&key_path,
|
|
&ca_path,
|
|
require_client_cert,
|
|
enable_revocation,
|
|
crl_url,
|
|
)
|
|
.await?;
|
|
|
|
info!(
|
|
"TLS configuration loaded - Protocol: TLS 1.3, mTLS: {}, Revocation: {}",
|
|
require_client_cert, enable_revocation
|
|
);
|
|
Some(tls)
|
|
} else {
|
|
info!("TLS disabled - running in development mode (set TLS_ENABLED=true for production)");
|
|
None
|
|
};
|
|
|
|
// Build gRPC server with all services
|
|
use api::auth_proto::auth_service_server::AuthServiceServer;
|
|
use api::foxhunt::tli::{
|
|
backtesting_service_server::BacktestingServiceServer,
|
|
trading_service_server::TradingServiceServer,
|
|
};
|
|
use api::ml_training::ml_training_service_server::MlTrainingServiceServer;
|
|
use api::monitoring::monitoring_service_server::MonitoringServiceServer;
|
|
|
|
let addr: std::net::SocketAddr = args
|
|
.bind_addr
|
|
.parse()
|
|
.map_err(|e| anyhow::anyhow!("Failed to parse bind address '{}': {}", args.bind_addr, e))?;
|
|
|
|
info!("Starting gRPC server on {}", addr);
|
|
|
|
// Setup graceful shutdown
|
|
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
|
|
|
|
// Spawn signal handler for graceful shutdown
|
|
tokio::spawn(async move {
|
|
match tokio::signal::ctrl_c().await {
|
|
Ok(()) => {
|
|
info!("Received shutdown signal, draining connections...");
|
|
}
|
|
Err(e) => {
|
|
error!("Failed to listen for ctrl-c signal: {}", e);
|
|
}
|
|
}
|
|
let _ = shutdown_tx.send(());
|
|
});
|
|
|
|
// Setup health check service
|
|
let (health_reporter, health_service) = tonic_health::server::health_reporter();
|
|
|
|
// Always register trading service (required)
|
|
health_reporter
|
|
.set_serving::<TradingServiceServer<api::grpc::TradingServiceProxy>>()
|
|
.await;
|
|
|
|
// Conditionally register optional services
|
|
if backtesting_proxy.is_some() {
|
|
health_reporter
|
|
.set_serving::<BacktestingServiceServer<api::grpc::BacktestingServiceProxy>>()
|
|
.await;
|
|
}
|
|
if ml_training_proxy.is_some() {
|
|
health_reporter
|
|
.set_serving::<MlTrainingServiceServer<api::grpc::MlTrainingProxy>>()
|
|
.await;
|
|
}
|
|
// Monitoring handler is always available (direct Prometheus, no backend dependency)
|
|
health_reporter
|
|
.set_serving::<MonitoringServiceServer<api::grpc::MonitoringServiceHandler>>()
|
|
.await;
|
|
|
|
// Initialize and start Prometheus metrics HTTP endpoint on port 9091
|
|
let gateway_metrics = api::metrics::GatewayMetrics::new()
|
|
.map_err(|e| anyhow::anyhow!("Failed to initialize gateway metrics: {}", e))?;
|
|
|
|
// Add service info metric (always present)
|
|
use prometheus::{register_gauge_with_registry, Opts};
|
|
let service_info = register_gauge_with_registry!(
|
|
Opts::new(
|
|
"api_service_info",
|
|
"API service information"
|
|
)
|
|
.const_label("version", env!("CARGO_PKG_VERSION"))
|
|
.const_label("service", "api"),
|
|
gateway_metrics.registry().as_ref()
|
|
)
|
|
.map_err(|e| anyhow::anyhow!("Failed to register service info: {}", e))?;
|
|
service_info.set(1.0);
|
|
|
|
let metrics_registry = gateway_metrics.registry();
|
|
|
|
tokio::spawn(async move {
|
|
// Use combined router for metrics AND health/resilience endpoints
|
|
let combined_app = api::metrics::combined_router(metrics_registry);
|
|
|
|
let metrics_addr = "0.0.0.0:9091";
|
|
info!(
|
|
"Prometheus metrics endpoint listening on http://{}",
|
|
metrics_addr
|
|
);
|
|
info!("Health endpoints available:");
|
|
info!(" - GET http://{}/health/liveness", metrics_addr);
|
|
info!(" - GET http://{}/health/readiness", metrics_addr);
|
|
info!(" - GET http://{}/health/startup", metrics_addr);
|
|
info!(
|
|
" - GET http://{}/resilience/circuit-breaker/status",
|
|
metrics_addr
|
|
);
|
|
info!(
|
|
" - GET http://{}/resilience/rate-limit/status",
|
|
metrics_addr
|
|
);
|
|
info!(" - GET http://{}/resilience/timeout/config", metrics_addr);
|
|
info!(" - GET http://{}/resilience/retry/config", metrics_addr);
|
|
|
|
let listener = match tokio::net::TcpListener::bind(metrics_addr).await {
|
|
Ok(l) => l,
|
|
Err(e) => {
|
|
error!("Failed to bind metrics endpoint on {}: {}", metrics_addr, e);
|
|
return;
|
|
}
|
|
};
|
|
|
|
if let Err(e) = axum::serve(listener, combined_app).await {
|
|
error!("Metrics server failed: {}", e);
|
|
}
|
|
});
|
|
|
|
// Configure CORS for gRPC-Web browser access
|
|
let cors_origins = std::env::var("CORS_ORIGINS")
|
|
.unwrap_or_else(|_| "http://localhost:5173".to_string());
|
|
let origins: Vec<http::HeaderValue> = cors_origins
|
|
.split(',')
|
|
.filter_map(|s| s.trim().parse().ok())
|
|
.collect();
|
|
|
|
let cors = CorsLayer::new()
|
|
.allow_origin(origins)
|
|
.allow_headers(AllowHeaders::any())
|
|
.allow_methods([http::Method::POST, http::Method::OPTIONS])
|
|
.expose_headers([
|
|
http::HeaderName::from_static("grpc-status"),
|
|
http::HeaderName::from_static("grpc-message"),
|
|
]);
|
|
|
|
// Build server with HTTP/2 optimizations and gRPC-Web support
|
|
let mut server_builder = if let Some(ref tls) = tls_config {
|
|
tonic::transport::Server::builder()
|
|
.accept_http1(true) // Required for grpc-web
|
|
.layer(GrpcMetricsLayer::new("api"))
|
|
.layer(cors.clone())
|
|
.layer(GrpcWebLayer::new())
|
|
.tls_config(tls.to_server_tls_config())?
|
|
.max_concurrent_streams(Some(10_000))
|
|
.http2_keepalive_interval(Some(Duration::from_secs(30)))
|
|
.http2_keepalive_timeout(Some(Duration::from_secs(10)))
|
|
} else {
|
|
tonic::transport::Server::builder()
|
|
.accept_http1(true) // Required for grpc-web
|
|
.layer(GrpcMetricsLayer::new("api"))
|
|
.layer(cors.clone())
|
|
.layer(GrpcWebLayer::new())
|
|
.max_concurrent_streams(Some(10_000))
|
|
.http2_keepalive_interval(Some(Duration::from_secs(30)))
|
|
.http2_keepalive_timeout(Some(Duration::from_secs(10)))
|
|
}
|
|
.layer(tower::ServiceBuilder::new().layer(tower::layer::util::Identity::new())); // Placeholder for auth interceptor layer
|
|
|
|
// Add gRPC reflection service (unauthenticated -- for grpcurl/debugging)
|
|
let reflection_service = tonic_reflection::server::Builder::configure()
|
|
.register_encoded_file_descriptor_set(api::TLI_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::MONITORING_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::ML_TRAINING_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::TRADING_AGENT_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::TRADING_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::RISK_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::CONFIG_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::BROKER_GATEWAY_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::DATA_ACQUISITION_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::ML_FILE_DESCRIPTOR_SET)
|
|
.register_encoded_file_descriptor_set(api::AUTH_FILE_DESCRIPTOR_SET)
|
|
.build_v1()
|
|
.map_err(|e| anyhow::anyhow!("Failed to build reflection service: {}", e))?;
|
|
|
|
// Create unauthenticated auth service (Login / RefreshToken -- pre-authentication)
|
|
let auth_grpc_service = api::auth::grpc_login::AuthGrpcService::new(Arc::new(jwt_config));
|
|
|
|
// Add health + reflection services, then unauthenticated auth service
|
|
let mut router = server_builder
|
|
.add_service(health_service)
|
|
.add_service(reflection_service)
|
|
.add_service(AuthServiceServer::new(auth_grpc_service));
|
|
|
|
// Always add trading service (required) with authentication
|
|
router = router.add_service(TradingServiceServer::with_interceptor(
|
|
trading_proxy,
|
|
auth_interceptor.clone(),
|
|
));
|
|
|
|
// Track service availability for logging
|
|
let backtesting_available = backtesting_proxy.is_some();
|
|
let ml_training_available = ml_training_proxy.is_some();
|
|
|
|
// Conditionally add optional services with authentication
|
|
if let Some(backtesting) = backtesting_proxy.as_ref() {
|
|
// Clone the entire Arc - tonic services can work with Arc-wrapped implementations
|
|
router = router.add_service(BacktestingServiceServer::with_interceptor(
|
|
Arc::clone(backtesting),
|
|
auth_interceptor.clone(),
|
|
));
|
|
}
|
|
if let Some(ml_training) = ml_training_proxy {
|
|
router = router.add_service(MlTrainingServiceServer::with_interceptor(
|
|
ml_training,
|
|
auth_interceptor.clone(),
|
|
));
|
|
}
|
|
|
|
// Monitoring handler is always available (direct Prometheus scraping, no backend)
|
|
router = router.add_service(MonitoringServiceServer::with_interceptor(
|
|
monitoring_handler,
|
|
auth_interceptor.clone(),
|
|
));
|
|
|
|
// Register all direct proxy services with auth interceptor
|
|
{
|
|
use api::trading_backend::trading_service_server::TradingServiceServer as TradingDirectServer;
|
|
use api::risk::risk_service_server::RiskServiceServer;
|
|
use api::broker_gateway::broker_gateway_service_server::BrokerGatewayServiceServer;
|
|
use api::data_acquisition::data_acquisition_service_server::DataAcquisitionServiceServer;
|
|
use api::ml_inference::ml_service_server::MlServiceServer;
|
|
use api::config_backend::config_service_server::ConfigServiceServer as ConfigBackendServer;
|
|
use api::trading_agent::trading_agent_service_server::TradingAgentServiceServer;
|
|
|
|
router = router
|
|
.add_service(TradingDirectServer::with_interceptor(
|
|
trading_direct_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(RiskServiceServer::with_interceptor(
|
|
risk_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(BrokerGatewayServiceServer::with_interceptor(
|
|
broker_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(DataAcquisitionServiceServer::with_interceptor(
|
|
data_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(MlServiceServer::with_interceptor(
|
|
ml_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(ConfigBackendServer::with_interceptor(
|
|
config_direct_proxy,
|
|
auth_interceptor.clone(),
|
|
))
|
|
.add_service(TradingAgentServiceServer::with_interceptor(
|
|
agent_proxy,
|
|
auth_interceptor.clone(),
|
|
));
|
|
}
|
|
|
|
// Log startup information
|
|
info!("API Service listening on {}", addr);
|
|
info!(" - gRPC-Web: enabled (CORS origins: {})", cors_origins);
|
|
info!(" - Trading Service: {} (REQUIRED)", trading_backend_url);
|
|
info!(
|
|
" - Backtesting Service: {} ({})",
|
|
backtesting_backend_url,
|
|
if backtesting_available {
|
|
"AVAILABLE"
|
|
} else {
|
|
"UNAVAILABLE"
|
|
}
|
|
);
|
|
info!(
|
|
" - ML Training Service: {} ({})",
|
|
ml_training_backend_url,
|
|
if ml_training_available {
|
|
"AVAILABLE"
|
|
} else {
|
|
"UNAVAILABLE"
|
|
}
|
|
);
|
|
info!(
|
|
" - Monitoring: direct Prometheus scraping ({})",
|
|
prometheus_url
|
|
);
|
|
info!(" - trading.TradingService -> {}", trading_backend_url);
|
|
info!(" - risk.RiskService -> {}", trading_backend_url);
|
|
info!(" - config.ConfigService -> {}", trading_backend_url);
|
|
info!(" - ml.MLService -> {}", ml_service_url);
|
|
info!(" - broker_gateway.BrokerGatewayService -> {}", broker_gw_url);
|
|
info!(" - data_acquisition.DataAcquisitionService -> {}", data_acq_url);
|
|
info!(" - trading_agent.TradingAgentService -> {}", trading_agent_url);
|
|
info!(" - Health checks: enabled");
|
|
info!(" - Authentication: 6-layer (<10us overhead)");
|
|
info!(" - Rate limiting: {} req/s per user", args.rate_limit_rps);
|
|
|
|
// Start server with graceful shutdown
|
|
let server = router.serve_with_shutdown(addr, async {
|
|
shutdown_rx.await.ok();
|
|
});
|
|
|
|
// Start server
|
|
server.await?;
|
|
|
|
info!("API Service shutdown complete");
|
|
Ok(())
|
|
}
|
|
|
|
/// Load JWT configuration from Vault (production) or environment (development)
|
|
///
|
|
/// Priority:
|
|
/// 1. Vault (secret/foxhunt/jwt) - Production
|
|
/// 2. JWT_SECRET_FILE - File-based secret
|
|
/// 3. JWT_SECRET env var - Development fallback
|
|
async fn load_jwt_config() -> Result<JwtConfig> {
|
|
JwtConfig::new().await
|
|
}
|