haproxy-openmanager/backend/utils/haproxy_stats_cache.py

"""
HAProxy Stats Cache Manager
Handles caching of HAProxy statistics in Redis for performance optimization
"""

import json
import logging
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
from database.connection import redis_client

logger = logging.getLogger(__name__)

# Redis key prefixes
STATS_PREFIX = "haproxy:stats"
METRICS_PREFIX = "haproxy:metrics"

# TTL values (in seconds)
# Agent sends heartbeat every 30s, so TTL must be 3x to avoid race conditions
TTL_REALTIME = 90  # 90 seconds for real-time stats (3x agent heartbeat)
TTL_SHORT = 600  # 10 minutes for short-term data
TTL_MEDIUM = 3600  # 1 hour for medium-term data
TTL_LONG = 86400  # 24 hours for long-term data


class HAProxyStatsCache:
    """Manager for HAProxy statistics caching in Redis"""
    
    def __init__(self):
        self.redis = redis_client
    
    def _get_key(self, key_type: str, cluster_id: int, suffix: str = "") -> str:
        """Generate Redis key with consistent format"""
        base_key = f"{STATS_PREFIX}:{cluster_id}:{key_type}"
        return f"{base_key}:{suffix}" if suffix else base_key
    
    def _safe_json_encode(self, data: Any) -> str:
        """Safely encode data to JSON"""
        try:
            return json.dumps(data, default=str)
        except Exception as e:
            logger.error(f"JSON encoding error: {e}")
            return json.dumps({})
    
    def _safe_json_decode(self, data: str) -> Any:
        """Safely decode JSON data"""
        try:
            return json.loads(data) if data else None
        except Exception as e:
            logger.error(f"JSON decoding error: {e}")
            return None
    
    # ============ Summary Stats ============
    
    def cache_summary_stats(self, cluster_id: int, stats: Dict[str, Any], ttl: int = TTL_REALTIME):
        """Cache summary statistics for a cluster"""
        try:
            key = self._get_key("summary", cluster_id)
            value = self._safe_json_encode(stats)
            self.redis.setex(key, ttl, value)
            logger.debug(f"Cached summary stats for cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to cache summary stats: {e}")
    
    def get_summary_stats(self, cluster_id: int) -> Optional[Dict[str, Any]]:
        """Get cached summary statistics"""
        try:
            key = self._get_key("summary", cluster_id)
            data = self.redis.get(key)
            return self._safe_json_decode(data) if data else None
        except Exception as e:
            logger.error(f"Failed to get summary stats: {e}")
            return None
    
    # ============ Frontend Stats ============
    
    def cache_frontend_stats(self, cluster_id: int, frontend_name: str, stats: Dict[str, Any], ttl: int = TTL_REALTIME):
        """Cache frontend statistics"""
        try:
            key = self._get_key("frontend", cluster_id, frontend_name)
            value = self._safe_json_encode(stats)
            self.redis.setex(key, ttl, value)
            
            # Also add to frontend list
            list_key = self._get_key("frontends", cluster_id)
            self.redis.sadd(list_key, frontend_name)
            self.redis.expire(list_key, ttl)
            
            bytes_in = stats.get('bytes_in', 0)
            bytes_out = stats.get('bytes_out', 0)
            logger.info(f"✍️  CACHE WRITE: Frontend '{frontend_name}' (cluster {cluster_id}) - bytes_in={bytes_in}, bytes_out={bytes_out}, key='{key}'")
        except Exception as e:
            logger.error(f"Failed to cache frontend stats: {e}")
    
    def get_frontend_stats(self, cluster_id: int, frontend_name: str) -> Optional[Dict[str, Any]]:
        """Get cached frontend statistics"""
        try:
            key = self._get_key("frontend", cluster_id, frontend_name)
            data = self.redis.get(key)
            return self._safe_json_decode(data) if data else None
        except Exception as e:
            logger.error(f"Failed to get frontend stats: {e}")
            return None
    
    def get_all_frontend_stats(self, cluster_id: int) -> Dict[str, Dict[str, Any]]:
        """Get all cached frontend statistics for a cluster"""
        try:
            list_key = self._get_key("frontends", cluster_id)
            frontend_names = self.redis.smembers(list_key)
            
            logger.info(f"🔍 CACHE READ: Cluster {cluster_id} - list_key='{list_key}', found {len(frontend_names)} frontend names")
            
            result = {}
            for name in frontend_names:
                name_str = name.decode('utf-8') if isinstance(name, bytes) else name
                stats = self.get_frontend_stats(cluster_id, name_str)
                if stats:
                    bytes_in = stats.get('bytes_in', 0)
                    bytes_out = stats.get('bytes_out', 0)
                    logger.debug(f"🔍 CACHE READ: Frontend '{name_str}' - bytes_in={bytes_in}, bytes_out={bytes_out}")
                    result[name_str] = stats
                else:
                    logger.warning(f"🔍 CACHE READ: Frontend '{name_str}' - NO STATS FOUND")
            
            return result
        except Exception as e:
            logger.error(f"Failed to get all frontend stats: {e}")
            return {}
    
    # ============ Backend Stats ============
    
    def cache_backend_stats(self, cluster_id: int, backend_name: str, stats: Dict[str, Any], ttl: int = TTL_REALTIME):
        """Cache backend statistics"""
        try:
            key = self._get_key("backend", cluster_id, backend_name)
            value = self._safe_json_encode(stats)
            self.redis.setex(key, ttl, value)
            
            # Also add to backend list
            list_key = self._get_key("backends", cluster_id)
            self.redis.sadd(list_key, backend_name)
            self.redis.expire(list_key, ttl)
            
            logger.debug(f"Cached backend stats for {backend_name} in cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to cache backend stats: {e}")
    
    def get_backend_stats(self, cluster_id: int, backend_name: str) -> Optional[Dict[str, Any]]:
        """Get cached backend statistics"""
        try:
            key = self._get_key("backend", cluster_id, backend_name)
            data = self.redis.get(key)
            return self._safe_json_decode(data) if data else None
        except Exception as e:
            logger.error(f"Failed to get backend stats: {e}")
            return None
    
    def get_all_backend_stats(self, cluster_id: int) -> Dict[str, Dict[str, Any]]:
        """Get all cached backend statistics for a cluster"""
        try:
            list_key = self._get_key("backends", cluster_id)
            backend_names = self.redis.smembers(list_key)
            
            result = {}
            for name in backend_names:
                name_str = name.decode('utf-8') if isinstance(name, bytes) else name
                stats = self.get_backend_stats(cluster_id, name_str)
                if stats:
                    result[name_str] = stats
            
            return result
        except Exception as e:
            logger.error(f"Failed to get all backend stats: {e}")
            return {}
    
    # ============ Server Stats ============
    
    def cache_server_stats(self, cluster_id: int, backend_name: str, server_name: str, 
                          stats: Dict[str, Any], ttl: int = TTL_REALTIME):
        """Cache server statistics"""
        try:
            key = self._get_key("server", cluster_id, f"{backend_name}:{server_name}")
            value = self._safe_json_encode(stats)
            self.redis.setex(key, ttl, value)
            
            # Also add to server list for the backend
            list_key = self._get_key("servers", cluster_id, backend_name)
            self.redis.sadd(list_key, server_name)
            self.redis.expire(list_key, ttl)
            
            logger.debug(f"Cached server stats for {backend_name}/{server_name} in cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to cache server stats: {e}")
    
    def get_server_stats(self, cluster_id: int, backend_name: str, server_name: str) -> Optional[Dict[str, Any]]:
        """Get cached server statistics"""
        try:
            key = self._get_key("server", cluster_id, f"{backend_name}:{server_name}")
            data = self.redis.get(key)
            return self._safe_json_decode(data) if data else None
        except Exception as e:
            logger.error(f"Failed to get server stats: {e}")
            return None
    
    def get_all_server_stats(self, cluster_id: int) -> List[Dict[str, Any]]:
        """Get all cached server statistics for a cluster"""
        try:
            # Get all backends
            backends_key = self._get_key("backends", cluster_id)
            backend_names = self.redis.smembers(backends_key)
            
            result = []
            for backend_name in backend_names:
                backend_str = backend_name.decode('utf-8') if isinstance(backend_name, bytes) else backend_name
                
                # Get servers for this backend
                servers_key = self._get_key("servers", cluster_id, backend_str)
                server_names = self.redis.smembers(servers_key)
                
                for server_name in server_names:
                    server_str = server_name.decode('utf-8') if isinstance(server_name, bytes) else server_name
                    stats = self.get_server_stats(cluster_id, backend_str, server_str)
                    if stats:
                        stats['backend'] = backend_str
                        stats['name'] = server_str
                        result.append(stats)
            
            return result
        except Exception as e:
            logger.error(f"Failed to get all server stats: {e}")
            return []
    
    # ============ Time Series Metrics ============
    
    def add_timeseries_metric(self, cluster_id: int, metric_name: str, value: float, 
                             timestamp: Optional[datetime] = None, ttl: int = TTL_LONG):
        """Add a time series metric point"""
        try:
            if timestamp is None:
                timestamp = datetime.utcnow()
            
            key = self._get_key("timeseries", cluster_id, metric_name)
            score = timestamp.timestamp()
            member = json.dumps({"value": value, "timestamp": timestamp.isoformat()})
            
            # Add to sorted set
            self.redis.zadd(key, {member: score})
            self.redis.expire(key, ttl)
            
            # Cleanup old entries (keep last 24 hours)
            cutoff = (datetime.utcnow() - timedelta(hours=24)).timestamp()
            self.redis.zremrangebyscore(key, 0, cutoff)
            
            logger.debug(f"Added timeseries metric {metric_name} for cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to add timeseries metric: {e}")
    
    def get_timeseries_metrics(self, cluster_id: int, metric_name: str, 
                              start_time: Optional[datetime] = None,
                              end_time: Optional[datetime] = None,
                              limit: Optional[int] = None) -> List[Dict[str, Any]]:
        """Get time series metrics for a given time range with optional limit"""
        try:
            key = self._get_key("timeseries", cluster_id, metric_name)
            
            # Default to last 24 hours if no time range specified
            if end_time is None:
                end_time = datetime.utcnow()
            if start_time is None:
                start_time = end_time - timedelta(hours=24)
            
            min_score = start_time.timestamp()
            max_score = end_time.timestamp()
            
            # Get data from sorted set (in reverse order to get most recent first if limiting)
            if limit and limit > 0:
                # Get most recent N points
                data = self.redis.zrevrangebyscore(key, max_score, min_score, start=0, num=limit)
                data = list(reversed(data))  # Reverse back to chronological order
            else:
                data = self.redis.zrangebyscore(key, min_score, max_score)
            
            result = []
            for item in data:
                try:
                    decoded = item.decode('utf-8') if isinstance(item, bytes) else item
                    point = json.loads(decoded)
                    result.append(point)
                except Exception as e:
                    logger.warning(f"Failed to decode timeseries point: {e}")
            
            return result
        except Exception as e:
            logger.error(f"Failed to get timeseries metrics: {e}")
            return []
    
    # ============ Raw Stats Storage ============
    
    def cache_raw_stats(self, cluster_id: int, agent_name: str, raw_stats: str, ttl: int = TTL_SHORT):
        """Cache raw HAProxy stats CSV output"""
        try:
            key = self._get_key("raw", cluster_id, agent_name)
            self.redis.setex(key, ttl, raw_stats)
            logger.debug(f"Cached raw stats for agent {agent_name} in cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to cache raw stats: {e}")
    
    def get_raw_stats(self, cluster_id: int, agent_name: str) -> Optional[str]:
        """Get cached raw HAProxy stats"""
        try:
            key = self._get_key("raw", cluster_id, agent_name)
            data = self.redis.get(key)
            return data.decode('utf-8') if data else None
        except Exception as e:
            logger.error(f"Failed to get raw stats: {e}")
            return None
    
    # ============ Aggregation Helpers ============
    
    def aggregate_cluster_stats(self, cluster_id: int) -> Dict[str, Any]:
        """Aggregate all stats for a cluster"""
        try:
            # Get all frontend stats
            frontends = self.get_all_frontend_stats(cluster_id)
            
            # Get all backend stats
            backends = self.get_all_backend_stats(cluster_id)
            
            # Get all server stats
            servers = self.get_all_server_stats(cluster_id)
            
            # Calculate aggregated metrics
            total_requests = sum(f.get('requests_total', 0) for f in frontends.values())
            total_sessions = sum(f.get('current_sessions', 0) for f in frontends.values())
            total_bytes_in = sum(f.get('bytes_in', 0) for f in frontends.values())
            total_bytes_out = sum(f.get('bytes_out', 0) for f in frontends.values())
            
            # Server health
            servers_total = len(servers)
            servers_up = len([s for s in servers if s.get('status') == 'UP'])
            
            # Calculate average response time
            response_times = [s.get('response_time', 0) for s in servers if s.get('response_time', 0) > 0]
            avg_response_time = sum(response_times) / len(response_times) if response_times else 0
            
            return {
                'cluster_id': cluster_id,
                'timestamp': datetime.utcnow().isoformat(),
                'frontends': {
                    'count': len(frontends),
                    'data': list(frontends.values())
                },
                'backends': {
                    'count': len(backends),
                    'data': list(backends.values())
                },
                'servers': {
                    'total': servers_total,
                    'up': servers_up,
                    'down': servers_total - servers_up,
                    'data': servers
                },
                'metrics': {
                    'total_requests': total_requests,
                    'total_sessions': total_sessions,
                    'total_bytes_in': total_bytes_in,
                    'total_bytes_out': total_bytes_out,
                    'avg_response_time': round(avg_response_time, 2)
                }
            }
        except Exception as e:
            logger.error(f"Failed to aggregate cluster stats: {e}")
            return {}
    
    # ============ Cleanup ============
    
    def clear_cluster_stats(self, cluster_id: int):
        """Clear all cached stats for a cluster"""
        try:
            pattern = f"{STATS_PREFIX}:{cluster_id}:*"
            keys = self.redis.keys(pattern)
            if keys:
                self.redis.delete(*keys)
                logger.info(f"Cleared {len(keys)} cache keys for cluster {cluster_id}")
        except Exception as e:
            logger.error(f"Failed to clear cluster stats: {e}")


# Global instance
haproxy_stats_cache = HAProxyStatsCache()