From 5fa329e4e049487cafc236461351e691e134d291 Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sat, 25 Apr 2026 00:48:27 +0300 Subject: [PATCH] feat: Refactor monitoring sync module to use multiplexed Redis connections and explicit AsyncCommands trait for improved connection pooling - Update all Redis operations to use get_multiplexed_async_connection instead of get_async_connection for better connection pooling - Replace context error handling with map_err for RedisConnection/RedisOperation errors - Add explicit redis::AsyncCommands trait qualification for all Redis operations (keys/get/set/del/ttl/expire) - Add chrono::Datelike import --- apps/charging-engine/src/monitoring/sync.rs | 74 +++++++++++---------- 1 file changed, 40 insertions(+), 34 deletions(-) diff --git a/apps/charging-engine/src/monitoring/sync.rs b/apps/charging-engine/src/monitoring/sync.rs index c9520cc..e80cd17 100644 --- a/apps/charging-engine/src/monitoring/sync.rs +++ b/apps/charging-engine/src/monitoring/sync.rs @@ -1,10 +1,12 @@ use tokio::time::{interval, sleep}; use tracing::{info, warn, error, debug}; use std::time::Duration; +use chrono::Datelike; use crate::errors::{ChargingError, ChargingResult, ErrorContext}; +use crate::monitoring::types::{SystemStats, HealthStatus}; -impl super::ChargingEngine { +impl crate::charging::ChargingEngine { pub async fn start_background_sync(&self) -> ChargingResult<()> { let mut interval_timer = interval(self.sync_interval); @@ -39,20 +41,20 @@ impl super::ChargingEngine { } async fn cleanup_expired_blocks(&self) -> ChargingResult<()> { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; let pattern = "block:*".to_string(); - let keys: Vec = conn.keys(&pattern).await - .context("Failed to get block keys")?; + let keys: Vec = redis::AsyncCommands::keys(&mut conn, &pattern).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; let mut cleaned_count = 0; for key in keys { - let ttl: i64 = conn.ttl(&key).await.unwrap_or(-1); + let ttl: i64 = redis::AsyncCommands::ttl(&mut conn, &key).await.unwrap_or(-1); if ttl == -1 || ttl == -2 { // Key has no expiration or doesn't exist - let _: () = conn.del(&key).await.unwrap_or(()); + let _: () = redis::AsyncCommands::del(&mut conn, &key).await.unwrap_or(()); cleaned_count += 1; } } @@ -65,44 +67,45 @@ impl super::ChargingEngine { } async fn update_usage_statistics(&self) -> ChargingResult<()> { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; // Get total active sessions let pattern = "session:*".to_string(); - let keys: Vec = conn.keys(&pattern).await - .context("Failed to get session keys")?; + let keys: Vec = redis::AsyncCommands::keys(&mut conn, &pattern).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; let active_sessions = keys.len(); // Store statistics let stats_key = "stats:active_sessions"; - let _: () = conn.set(stats_key, active_sessions).await.unwrap_or(()); + let _: () = redis::AsyncCommands::set(&mut conn, stats_key, active_sessions).await.unwrap_or(()); debug!("Updated usage statistics: {} active sessions", active_sessions); Ok(()) } async fn check_low_balance_alerts(&self) -> ChargingResult<()> { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; let pattern = "credit:*".to_string(); - let keys: Vec = conn.keys(&pattern).await - .context("Failed to get credit keys")?; + let keys: Vec = redis::AsyncCommands::keys(&mut conn, &pattern).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; let low_balance_threshold = 100_000_000; // 100MB in bytes let mut low_balance_count = 0; for key in keys { - if let Ok(balance) = conn.get::<_, u64>(&key).await { + if let Ok(balance) = redis::AsyncCommands::get::<_, u64>(&mut conn, &key).await { if balance < low_balance_threshold { low_balance_count += 1; warn!("Low balance alert for {}: {} bytes remaining", key, balance); // Store alert let alert_key = format!("alert:low_balance:{}", key); - let _: () = conn.setex(&alert_key, 3600, "low_balance").await.unwrap_or(()); + let _: () = redis::AsyncCommands::set(&mut conn, &alert_key, "low_balance").await.unwrap_or(()); + let _: () = redis::AsyncCommands::expire(&mut conn, &alert_key, 3600).await.unwrap_or(()); } } } @@ -115,11 +118,11 @@ impl super::ChargingEngine { } async fn apply_monthly_fees_if_needed(&self) -> ChargingResult<()> { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; let last_fee_key = "last_monthly_fee"; - let last_fee: Option = conn.get(last_fee_key).await.unwrap_or(None); + let last_fee: Option = redis::AsyncCommands::get(&mut conn, last_fee_key).await.unwrap_or(None); let now = chrono::Utc::now(); let current_month = format!("{}-{}", now.year(), now.month()); @@ -134,20 +137,23 @@ impl super::ChargingEngine { let processed = self.apply_monthly_fees().await?; // Update last fee timestamp - let _: () = conn.set(last_fee_key, current_month).await.unwrap_or(()); + let _: () = redis::AsyncCommands::set(&mut conn, last_fee_key, current_month.clone()).await.unwrap_or(()); info!("Applied monthly fees to {} accounts for {}", processed, current_month); Ok(()) } pub async fn get_system_statistics(&self) -> ChargingResult { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; - let active_sessions: u64 = conn.get("stats:active_sessions").await.unwrap_or(0); - let total_accounts: u64 = conn.keys("account:*").await.unwrap_or_default().len() as u64; - let blocked_users: u64 = conn.keys("block:*").await.unwrap_or_default().len() as u64; - let low_balance_alerts: u64 = conn.keys("alert:low_balance:*").await.unwrap_or_default().len() as u64; + let active_sessions: u64 = redis::AsyncCommands::get(&mut conn, "stats:active_sessions").await.unwrap_or(0); + let account_keys: Vec = redis::AsyncCommands::keys(&mut conn, "account:*").await.unwrap_or_default(); + let total_accounts: u64 = account_keys.len() as u64; + let block_keys: Vec = redis::AsyncCommands::keys(&mut conn, "block:*").await.unwrap_or_default(); + let blocked_users: u64 = block_keys.len() as u64; + let alert_keys: Vec = redis::AsyncCommands::keys(&mut conn, "alert:low_balance:*").await.unwrap_or_default(); + let low_balance_alerts: u64 = alert_keys.len() as u64; let stats = SystemStats { active_sessions, @@ -160,7 +166,7 @@ impl super::ChargingEngine { Ok(stats) } - async fn get_uptime(&self) -> ChargingResult { + pub async fn get_uptime(&self) -> ChargingResult { // Calculate actual uptime since startup match self.startup_time.elapsed() { Ok(duration) => Ok(duration.as_secs()), @@ -175,11 +181,11 @@ impl super::ChargingEngine { } pub async fn health_check(&self) -> ChargingResult { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; // Test Redis connection - let _: String = conn.get("health_check").await.unwrap_or_else(|_| "ok".to_string()); + let _: String = redis::AsyncCommands::get(&mut conn, "health_check").await.unwrap_or_else(|_| "ok".to_string()); let status = HealthStatus { redis_connected: true, @@ -233,8 +239,8 @@ impl super::ChargingEngine { } async fn get_redis_memory_usage(&self) -> ChargingResult { - let mut conn = self.redis_client.get_async_connection().await - .context("Failed to get Redis connection")?; + let mut conn = self.redis_client.get_multiplexed_async_connection().await + .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; // Get Redis memory info let info: String = redis::cmd("INFO").query_async(&mut conn).await.unwrap_or_default();