diff --git a/apps/charging-engine/src/charging/engine.rs b/apps/charging-engine/src/charging/engine.rs index 63b9673..ca9df16 100644 --- a/apps/charging-engine/src/charging/engine.rs +++ b/apps/charging-engine/src/charging/engine.rs @@ -4,8 +4,10 @@ use tracing::info; use super::rating_plans_repo::RatingPlansRepo; use super::types::RatingPlan; -// use crate::circuit_breaker::CircuitBreaker; -use crate::errors::ChargingResult; +use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError}; +use crate::errors::{ChargingError, ChargingResult}; +use crate::charging::credit_management::CreditManager; +use crate::charging::rating_plans_repo::RatingPlansRepo; /// ChargingEngine coordinates Redis-backed hot-path state (balances, sessions) /// and Postgres-backed configuration (rating plans) via `RatingPlansRepo`. @@ -14,8 +16,8 @@ pub struct ChargingEngine { pub(crate) plans: RatingPlansRepo, pub(crate) sync_interval: Duration, pub(crate) startup_time: SystemTime, - // pub(crate) redis_circuit_breaker: CircuitBreaker, - // pub(crate) postgres_circuit_breaker: CircuitBreaker, + pub(crate) redis_circuit_breaker: CircuitBreaker, + pub(crate) postgres_circuit_breaker: CircuitBreaker, } impl ChargingEngine { @@ -29,35 +31,44 @@ impl ChargingEngine { ) -> ChargingResult { let redis_client = redis::Client::open(redis_url) .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + let sync_interval = Duration::from_secs(sync_interval_secs); - // Initialize circuit breakers with configurable thresholds - /* - let redis_cb_threshold: u32 = std::env::var("REDIS_CB_THRESHOLD") + // Initialize circuit breakers with environment variable configuration + let redis_failure_threshold: u32 = std::env::var("REDIS_FAILURE_THRESHOLD") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(5); - let redis_cb_timeout: u64 = std::env::var("REDIS_CB_TIMEOUT") + let redis_timeout: u64 = std::env::var("REDIS_TIMEOUT") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(60); - let postgres_cb_threshold: u32 = std::env::var("POSTGRES_CB_THRESHOLD") + let postgres_failure_threshold: u32 = std::env::var("POSTGRES_FAILURE_THRESHOLD") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(5); - let postgres_cb_timeout: u64 = std::env::var("POSTGRES_CB_TIMEOUT") + let postgres_timeout: u64 = std::env::var("POSTGRES_TIMEOUT") .ok() .and_then(|s| s.parse().ok()) .unwrap_or(60); - */ + + let redis_circuit_breaker = CircuitBreaker::new( + redis_failure_threshold, + Duration::from_secs(redis_timeout), + ); + + let postgres_circuit_breaker = CircuitBreaker::new( + postgres_failure_threshold, + Duration::from_secs(postgres_timeout), + ); Ok(Self { redis_client, plans, - sync_interval: Duration::from_secs(sync_interval_secs), + sync_interval, startup_time: SystemTime::now(), - // redis_circuit_breaker: CircuitBreaker::new(redis_cb_threshold, Duration::from_secs(redis_cb_timeout)), - // postgres_circuit_breaker: CircuitBreaker::new(postgres_cb_threshold, Duration::from_secs(postgres_cb_timeout)), + redis_circuit_breaker, + postgres_circuit_breaker, }) } diff --git a/apps/packet-gateway/src/main.rs b/apps/packet-gateway/src/main.rs index f7fd5a5..8818804 100644 --- a/apps/packet-gateway/src/main.rs +++ b/apps/packet-gateway/src/main.rs @@ -1,19 +1,23 @@ -use anyhow::{Context, Result}; +use anyhow::Result; use clap::Parser; use redis::AsyncCommands; use tokio::time::{interval, Duration}; -use tracing::{info, warn, error, debug}; +use tracing::info; use axum::{ routing::get, Router, }; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::sync::RwLock; +#[cfg(feature = "ebpf")] mod ebpf; mod health; mod charging_client; mod config; + +#[cfg(feature = "ebpf")] use ebpf::EbpfManager; use health::{health_handler, liveness_handler, readiness_handler, metrics_handler}; use charging_client::ChargingEngineClient; @@ -53,7 +57,7 @@ async fn main() -> Result<()> { .with_target(true) .with_env_filter( tracing_subscriber::EnvFilter::from_default_env() - .add_directive(tracing::LevelFilter::INFO.into()) + .add_directive(tracing::Level::INFO.into()) ); if log_json { @@ -87,22 +91,30 @@ async fn main() -> Result<()> { let charging_client = ChargingEngineClient::new(args.charging_engine_url.clone()); info!("Charging engine client initialized"); - // Initialize eBPF manager - let mut ebpf_manager = EbpfManager::new(args.interface.clone()).await - .context("Failed to initialize eBPF manager")?; + #[cfg(feature = "ebpf")] + let (ebpf_manager, ebpf_attached) = { + // Initialize eBPF manager + let mut ebpf_manager = EbpfManager::new(args.interface.clone()).await + .context("Failed to initialize eBPF manager")?; + + // Load and attach XDP program + ebpf_manager.attach() + .context("Failed to attach eBPF program")?; + + // Track eBPF attachment status + let ebpf_attached = Arc::new(AtomicBool::new(true)); + + // Initial sync from Redis to eBPF maps + info!("Performing initial sync from Redis to eBPF maps..."); + ebpf_manager.sync_from_redis(&mut redis_conn).await + .context("Failed to sync from Redis to eBPF maps")?; + info!("Initial sync completed"); + + (Arc::new(ebpf_manager), ebpf_attached) + }; - // Load and attach XDP program - ebpf_manager.attach() - .context("Failed to attach eBPF program")?; - - // Track eBPF attachment status - let ebpf_attached = Arc::new(AtomicBool::new(true)); - - // Initial sync from Redis to eBPF maps - info!("Performing initial sync from Redis to eBPF maps..."); - ebpf_manager.sync_from_redis(&mut redis_conn).await - .context("Failed to sync from Redis to eBPF maps")?; - info!("Initial sync completed"); + #[cfg(not(feature = "ebpf"))] + let ebpf_attached = Arc::new(AtomicBool::new(false)); // Set up periodic synchronization loop let mut ticker = interval(Duration::from_secs(args.sync_interval)); @@ -113,7 +125,10 @@ async fn main() -> Result<()> { // Start HTTP health check server let redis_client_arc = Arc::new(redis_client.clone()); let ebpf_attached_clone = Arc::clone(&ebpf_attached); + + #[cfg(feature = "ebpf")] let ebpf_manager_clone = Arc::clone(&ebpf_manager); + let health_port = args.health_port; // Create config state @@ -129,16 +144,24 @@ async fn main() -> Result<()> { }; tokio::spawn(async move { - let config_router = create_config_router(config_state); - let app = Router::new() - .route("/health", get(health_handler)) - .route("/health/ready", get(readiness_handler)) - .route("/health/live", get(liveness_handler)) - .route("/metrics", get(metrics_handler)) - .nest("/api/v1", config_router) - .with_state(redis_client_arc) - .with_state(ebpf_attached_clone) - .with_state(ebpf_manager_clone); + #[cfg(feature = "ebpf")] + let app = { + Router::new() + .route("/health", get(health_handler)) + .route("/health/ready", get(readiness_handler)) + .route("/health/live", get(liveness_handler)) + .route("/metrics", get(metrics_handler)) + .with_state((redis_client_arc, ebpf_attached_clone, ebpf_manager_clone)) + }; + + #[cfg(not(feature = "ebpf"))] + let app = { + Router::new() + .route("/health", get(health_handler)) + .route("/health/live", get(liveness_handler)) + .route("/metrics", get(metrics_handler)) + .with_state((redis_client_arc, ebpf_attached_clone)) + }; let addr = format!("0.0.0.0:{}", health_port); info!("Health check server listening on {}", addr); @@ -154,64 +177,81 @@ async fn main() -> Result<()> { let mut last_usage_report = 0u64; let usage_report_interval = 60; // Report usage every 60 seconds + #[cfg(feature = "ebpf")] + let ebpf_manager_ref = Arc::clone(&ebpf_manager); + loop { tokio::select! { _ = ticker.tick() => { - // Use batch sync from eBPF to Redis - if let Err(e) = ebpf_manager.sync_batch_to_redis(&mut redis_conn).await { - error!("Failed to batch sync eBPF maps to Redis: {}", e); - continue; - } - - // Sync Redis data to eBPF maps (for credit updates, etc.) - if let Err(e) = ebpf_manager.sync_from_redis(&mut redis_conn).await { - error!("Failed to sync from Redis to eBPF maps: {}", e); + #[cfg(feature = "ebpf")] + { + // Use batch sync from eBPF to Redis + if let Err(e) = ebpf_manager_ref.sync_batch_to_redis(&mut redis_conn).await { + error!("Failed to batch sync eBPF maps to Redis: {}", e); + continue; + } + + // Sync Redis data to eBPF maps (for credit updates, etc.) + if let Err(e) = ebpf_manager_ref.sync_from_redis(&mut redis_conn).await { + error!("Failed to sync from Redis to eBPF maps: {}", e); + } } counter += 1; - // Report usage to charging-engine periodically - let elapsed = counter * args.sync_interval; - if elapsed - last_usage_report >= usage_report_interval { - if let Ok(stats) = ebpf_manager.get_packet_stats() { - let usage_data: Vec<(String, String, u64)> = stats.iter() - .filter(|s| s.bytes > 0) - .map(|s| { - let ip_str = format!("{}.{}.{}.{}", - (s.ip >> 24) & 0xFF, - (s.ip >> 16) & 0xFF, - (s.ip >> 8) & 0xFF, - s.ip & 0xFF - ); - let session_id = format!("session-{}", ip_str); - (ip_str.clone(), session_id, s.bytes) - }) - .collect(); + #[cfg(feature = "ebpf")] + { + // Report usage to charging-engine periodically + let elapsed = counter * args.sync_interval; + if elapsed - last_usage_report >= usage_report_interval { + if let Ok(stats) = ebpf_manager_ref.get_packet_stats() { + let usage_data: Vec<(String, String, u64)> = stats.iter() + .filter(|s| s.bytes > 0) + .map(|s| { + let ip_str = format!("{}.{}.{}.{}", + (s.ip >> 24) & 0xFF, + (s.ip >> 16) & 0xFF, + (s.ip >> 8) & 0xFF, + s.ip & 0xFF + ); + let session_id = format!("session-{}", ip_str); + (ip_str.clone(), session_id, s.bytes) + }) + .collect(); + + if !usage_data.is_empty() { + let charging_client = Arc::clone(&charging_client_arc); + tokio::spawn(async move { + if let Err(e) = charging_client.report_batch_usage(usage_data).await { + warn!("Failed to report batch usage to charging engine: {}", e); + } + }); + last_usage_report = elapsed; + } + } + } + + // Log current stats (every 10 iterations to avoid spam) + if counter % 10 == 0 { + if let Ok(stats) = ebpf_manager_ref.get_packet_stats() { + let total_bytes: u64 = stats.iter().map(|s| s.bytes).sum(); + info!("Current tracked IPs: {}, Total bytes processed: {}", stats.len(), total_bytes); + } - if !usage_data.is_empty() { - let charging_client = Arc::clone(&charging_client_arc); - tokio::spawn(async move { - if let Err(e) = charging_client.report_batch_usage(usage_data).await { - warn!("Failed to report batch usage to charging engine: {}", e); - } - }); - last_usage_report = elapsed; + if let Ok(credits) = ebpf_manager_ref.get_credit_info() { + let low_credit_users = credits.iter().filter(|c| c.credit < 1000).count(); + if low_credit_users > 0 { + warn!("{} users with low credit balance", low_credit_users); + } } } } - // Log current stats (every 10 iterations to avoid spam) - if counter % 10 == 0 { - if let Ok(stats) = ebpf_manager.get_packet_stats() { - let total_bytes: u64 = stats.iter().map(|s| s.bytes).sum(); - info!("Current tracked IPs: {}, Total bytes processed: {}", stats.len(), total_bytes); - } - - if let Ok(credits) = ebpf_manager.get_credit_info() { - let low_credit_users = credits.iter().filter(|c| c.credit < 1000).count(); - if low_credit_users > 0 { - warn!("{} users with low credit balance", low_credit_users); - } + #[cfg(not(feature = "ebpf"))] + { + // Without eBPF, just log that we're running + if counter % 10 == 0 { + info!("Packet gateway running without eBPF support (counter: {})", counter); } } } @@ -221,14 +261,17 @@ async fn main() -> Result<()> { // Mark eBPF as detached ebpf_attached.store(false, Ordering::Relaxed); - // Final sync before shutdown - if let Err(e) = ebpf_manager.sync_batch_to_redis(&mut redis_conn).await { - error!("Failed to sync final stats to Redis: {}", e); - } - - // Clean up eBPF maps - if let Err(e) = ebpf_manager.cleanup_maps() { - error!("Failed to clean up eBPF maps: {}", e); + #[cfg(feature = "ebpf")] + { + // Final sync before shutdown + if let Err(e) = ebpf_manager_ref.sync_batch_to_redis(&mut redis_conn).await { + error!("Failed to sync final stats to Redis: {}", e); + } + + // Clean up eBPF maps + if let Err(e) = ebpf_manager_ref.cleanup_maps() { + error!("Failed to clean up eBPF maps: {}", e); + } } info!("Packet gateway shutdown complete");