andwati/telecom-platform
1
0
Fork 0
mirror of https://github.com/andwati/telecom-platform.git synced 2026-10-20 20:04:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Re-enable circuit breaker implementation and add conditional eBPF compilation support

Browse Source 
- Uncomment CircuitBreaker and CircuitBreakerError imports in charging-engine
- Re-enable redis_circuit_breaker and postgres_circuit_breaker fields in ChargingEngine struct
- Add circuit breaker initialization with environment variable configuration (REDIS_FAILURE_THRESHOLD, REDIS_TIMEOUT, POSTGRES_FAILURE_THRESHOLD, POSTGRES_TIMEOUT)
- Add #[cfg(feature = "ebpf")] conditional compilation directives throughout
This commit is contained in:
nutcas3
2026-04-26 01:50:29 +03:00
parent 05f95a5c75
commit 95def1bcab
2 changed files with 150 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/charging-engine/src/charging/engine.rs
+25 -14
View File
@@ -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<Self> {
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,
})
}
apps/packet-gateway/src/main.rs
+125 -82
View File
@@ -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");