haproxy-openmanager/backend/haproxy_client.py

import asyncio
import aiohttp
import socket
import csv
import io
from typing import Dict, List, Optional, Any
import logging

logger = logging.getLogger(__name__)

class HAProxyClient:
    """Client for connecting to HAProxy instances and fetching stats"""
    
    def __init__(self, cluster_or_host, port: Optional[int] = None, connection_type: str = 'stats_socket', 
                 stats_socket_path: str = '/run/haproxy/admin.sock',
                 stats_username: Optional[str] = None,
                 stats_password: Optional[str] = None):
        # Support both dict (cluster) and string (host) parameters
        if isinstance(cluster_or_host, dict):
            # Initialize from cluster dict
            cluster = cluster_or_host
            self.host = cluster.get('host', 'localhost')
            self.port = cluster.get('port', 8404)
            self.connection_type = cluster.get('connection_type', 'local')
            self.stats_socket_path = cluster.get('stats_socket_path', '/run/haproxy/admin.sock')
            self.stats_username = cluster.get('stats_username')
            self.stats_password = cluster.get('stats_password')
        else:
            # Initialize from individual parameters (backward compatibility)
            self.host = cluster_or_host
            self.port = port or 8404
            self.connection_type = connection_type
            self.stats_socket_path = stats_socket_path
            self.stats_username = stats_username
            self.stats_password = stats_password
        
    async def get_stats(self) -> Dict[str, Any]:
        """Get HAProxy statistics"""
        try:
            if self.connection_type == 'stats_socket':
                # Try to get real stats, but for demo purposes always use fallback
                # return await self._get_stats_via_socket()
                return self._get_fallback_stats()
            elif self.connection_type == 'local':
                # Try to get real stats, but for demo purposes always use fallback
                # return await self._get_stats_via_http()
                return self._get_fallback_stats()
            else:
                # For other connection types, use HTTP stats page
                # return await self._get_stats_via_http()
                return self._get_fallback_stats()
        except Exception as e:
            logger.error(f"Failed to get HAProxy stats: {e}")
            return self._get_fallback_stats()
    
    async def _get_stats_via_http(self) -> Dict[str, Any]:
        """Get stats via HTTP stats page"""
        try:
            url = f"http://{self.host}:{self.port}/stats?stats;csv"
            auth = None
            if self.stats_username and self.stats_password:
                auth = aiohttp.BasicAuth(self.stats_username, self.stats_password)
            
            # v1.11.0: instrumented for completeness. NOTE the CSV body is
            # deliberately NOT handed to the span — a full stats dump is large,
            # changes every poll, and has no diagnostic value in an audit row;
            # status + duration is what matters. `auth` is likewise never logged:
            # aiohttp.BasicAuth is a NamedTuple whose repr contains the cleartext
            # password.
            from utils.http_instrumentation import outbound_span, TARGET_HAPROXY_STATS

            async with aiohttp.ClientSession() as session:
                async with outbound_span(
                    target=TARGET_HAPROXY_STATS, method="GET", url=url,
                    capture_body=False, capture_response_body=False,
                ) as span:
                    async with session.get(url, auth=auth, timeout=aiohttp.ClientTimeout(total=10)) as response:
                        span.set_response(response.status, getattr(response, "headers", None))
                        if response.status == 200:
                            csv_data = await response.text()
                            return self._parse_csv_stats(csv_data)
                        else:
                            logger.warning(f"HTTP stats request failed with status {response.status}")
                            return self._get_fallback_stats()
        except Exception as e:
            logger.error(f"HTTP stats request failed: {e}")
            return self._get_fallback_stats()
    
    async def _get_stats_via_socket(self) -> Dict[str, Any]:
        """Get stats via Unix socket (for Docker containers)"""
        try:
            # For Docker containers, we'll use HTTP as socket access is more complex
            # In a real deployment, you'd use actual socket connections
            return await self._get_stats_via_http()
        except Exception as e:
            logger.error(f"Socket stats request failed: {e}")
            return self._get_fallback_stats()
    
    def _parse_csv_stats(self, csv_data: str) -> Dict[str, Any]:
        """Parse HAProxy CSV stats format"""
        try:
            lines = csv_data.strip().split('\n')
            if not lines:
                return self._get_fallback_stats()
            
            # Skip comment lines and get header
            data_lines = [line for line in lines if not line.startswith('#')]
            if not data_lines:
                return self._get_fallback_stats()
            
            reader = csv.DictReader(io.StringIO('\n'.join(data_lines)))
            rows = list(reader)
            
            return self._process_haproxy_stats(rows)
        
        except Exception as e:
            logger.error(f"Failed to parse CSV stats: {e}")
            return self._get_fallback_stats()
    
    def _process_haproxy_stats(self, rows: List[Dict]) -> Dict[str, Any]:
        """Process HAProxy stats into our format"""
        try:
            backends = []
            servers = []
            frontend_stats = {"requests_total": 0, "requests_rate": 0, "status": "OPEN"}
            
            for row in rows:
                pxname = row.get('# pxname', '')
                svname = row.get('svname', '')
                status = row.get('status', 'UP')
                
                # Skip empty rows
                if not pxname or not svname:
                    continue
                
                # Process backend stats
                if svname == 'BACKEND':
                    total_servers = int(row.get('act', 0)) + int(row.get('bck', 0))
                    active_servers = int(row.get('act', 0))
                    
                    backends.append({
                        "name": pxname,
                        "status": "UP" if status in ['UP', 'OPEN'] else "DOWN",
                        "active_servers": active_servers,
                        "total_servers": max(total_servers, 1),  # Avoid division by zero
                        "requests": int(row.get('stot', 0)),
                        "response_time": int(float(row.get('rtime', 0)) or 0)
                    })
                
                # Process individual server stats
                elif svname != 'FRONTEND' and svname != 'BACKEND':
                    servers.append({
                        "name": svname,
                        "status": "UP" if status == 'UP' else "DOWN",
                        "weight": int(row.get('weight', 100)),
                        "requests": int(row.get('stot', 0)),
                        "backend": pxname,
                        "address": f"{row.get('addr', 'unknown')}:{row.get('port', '0')}"
                    })
                
                # Process frontend stats
                elif svname == 'FRONTEND':
                    frontend_stats["requests_total"] += int(row.get('stot', 0))
                    frontend_stats["requests_rate"] += float(row.get('req_rate', 0) or 0)
                    if status not in ['UP', 'OPEN']:
                        frontend_stats["status"] = "MAINT"
            
            return {
                "frontend": frontend_stats,
                "backends": backends,
                "servers": servers
            }
        
        except Exception as e:
            logger.error(f"Failed to process HAProxy stats: {e}")
            return self._get_fallback_stats()
    
    def _get_fallback_stats(self) -> Dict[str, Any]:
        """Return fallback stats when connection fails"""
        import random
        from datetime import datetime
        
        # Generate realistic demo data for development
        current_time = datetime.now()
        base_requests = 1000 + random.randint(0, 500)
        
        backends = []
        servers = []
        
        # Generate sample backends with different health statuses
        backend_names = ["web_servers", "api_servers", "db_pool", "cache_cluster"]
        for i, backend_name in enumerate(backend_names):
            total_servers = random.randint(2, 4)
            active_servers = random.randint(max(1, total_servers-1), total_servers)
            status = "UP" if active_servers > 0 else "DOWN"
            
            backend_requests = max(0, base_requests + random.randint(-200, 200))
            response_time = random.randint(50, 300) if status == "UP" else 0
            
            backends.append({
                "name": backend_name,
                "status": status,
                "active_servers": active_servers,
                "total_servers": total_servers,
                "requests": backend_requests,
                "response_time": response_time,
                "health_percentage": int((active_servers / total_servers) * 100)
            })
            
            # Generate servers for this backend
            for j in range(total_servers):
                server_name = f"{backend_name}-{j+1}"
                server_status = "UP" if j < active_servers else "DOWN"
                server_weight = random.randint(50, 150)
                server_requests = int(backend_requests / total_servers) + random.randint(-50, 50)
                server_port = 80 if "web" in backend_name else (3000 if "api" in backend_name else (5432 if "db" in backend_name else 6379))
                
                servers.append({
                    "name": server_name,
                    "status": server_status,
                    "weight": server_weight,
                    "requests": server_requests if server_status == "UP" else 0,
                    "backend": backend_name,
                    "address": f"192.168.1.{10 + len(servers)}:{server_port}",
                    "check_status": "L4OK" if server_status == "UP" else "L4TOUT",
                    "last_check": "0ms" if server_status == "UP" else "timeout"
                })
        
        return {
            "frontend": {
                "requests_total": base_requests,
                "requests_rate": round(random.uniform(10.0, 50.0), 2),
                "status": "OPEN"
            },
            "backends": backends,
            "servers": servers,
            "traffic_distribution": [
                {"name": backend["name"], "value": backend["requests"], "percentage": int((backend["requests"] / base_requests) * 100)} 
                for backend in backends
            ]
        }
    
    async def test_connection(self) -> Dict[str, Any]:
        """Test connection to HAProxy instance"""
        try:
            stats = await self.get_stats()
            if stats["frontend"]["status"] != "MAINT":
                return {
                    "status": "connected",
                    "error": None,
                    "version": "2.4.0"  # Could be parsed from stats if available
                }
            else:
                return {
                    "status": "error",
                    "error": "HAProxy appears to be in maintenance mode",
                    "version": None
                }
        except Exception as e:
            return {
                "status": "error", 
                "error": str(e),
                "version": None
            }
    
    async def get_config(self, config_path: str = "/etc/haproxy/haproxy.cfg") -> str:
        """Get HAProxy configuration file content"""
        try:
            if self.connection_type == 'local':
                # For local connections, read from local file system
                import os
                if os.path.exists(config_path):
                    with open(config_path, 'r') as f:
                        return f.read()
                else:
                    # Try common paths
                    for path in ["/usr/local/etc/haproxy/haproxy.cfg", "/etc/haproxy.cfg", "/opt/haproxy/haproxy.cfg"]:
                        if os.path.exists(path):
                            with open(path, 'r') as f:
                                return f.read()
                    raise Exception(f"HAProxy config file not found at {config_path}")
            else:
                # For remote connections, this would need SSH access
                # For now, return a demo config
                return """# HAProxy Configuration
# This is a demo configuration - implement SSH access for remote configs
global
    daemon
    log stdout local0

defaults
    mode http
    log global
    option httplog
    timeout connect 5s
    timeout client 50s
    timeout server 50s

frontend stats
    bind *:8404
    stats enable
    stats uri /stats
    stats admin if TRUE

frontend main
    bind *:80
    default_backend servers

backend servers
    balance roundrobin
    server web1 127.0.0.1:8001 check
    server web2 127.0.0.1:8002 check
"""
        except Exception as e:
            logger.error(f"Failed to get HAProxy config: {e}")
            raise Exception(f"Could not retrieve HAProxy configuration: {str(e)}")