rishithayanidhi/datacenter-cooling-optimization

1# Copyright (c) Meta Platforms, Inc. and affiliates.2# All rights reserved.3#4# This source code is licensed under the BSD-style license found in the5# LICENSE file in the root directory of this source tree.6 7"""8Logging service for the Data Center Cooling Environment server.9 10Handles container logs, request/response logging, and debugging information.11Logs are written to both an in-memory buffer (for API retrieval) and a rotating12file on disk (configurable via SERVER_LOG_FILE env var).13"""14 15import logging16import io17import os18import sys19from datetime import datetime20from pathlib import Path21from typing import Dict, List, Optional22from collections import deque23from enum import Enum24from logging.handlers import RotatingFileHandler25 26 27class LogLevel(str, Enum):28    """Log levels for filtering."""29    DEBUG = "DEBUG"30    INFO = "INFO"31    WARNING = "WARNING"32    ERROR = "ERROR"33    CRITICAL = "CRITICAL"34 35 36class LogBuffer:37    """38    In-memory circular buffer for storing logs.39    40    Keeps the most recent N log entries in memory for quick retrieval.41    """42    43    def __init__(self, max_size: int = 10000):44        """45        Initialize the log buffer.46        47        Args:48            max_size: Maximum number of log entries to store49        """50        self.max_size = max_size51        self.logs: deque = deque(maxlen=max_size)52        self.start_time = datetime.utcnow()53    54    def add(self, level: str, message: str, extra: Optional[Dict] = None):55        """Add a log entry."""56        entry = {57            "timestamp": datetime.utcnow().isoformat(),58            "level": level,59            "message": message,60            "extra": extra or {}61        }62        self.logs.append(entry)63    64    def get_all(self, level: Optional[str] = None) -> List[Dict]:65        """66        Retrieve all logs, optionally filtered by level.67        68        Args:69            level: Optional log level to filter by70            71        Returns:72            List of log entries73        """74        logs = list(self.logs)75        if level:76            logs = [log for log in logs if log["level"] == level.upper()]77        return logs78    79    def get_recent(self, count: int = 100, level: Optional[str] = None) -> List[Dict]:80        """81        Get the most recent N log entries.82        83        Args:84            count: Number of recent entries to retrieve85            level: Optional log level to filter by86            87        Returns:88            List of recent log entries89        """90        all_logs = self.get_all(level)91        return all_logs[-count:] if all_logs else []92    93    def clear(self):94        """Clear all logs."""95        self.logs.clear()96    97    def get_stats(self) -> Dict:98        """Get statistics about logged entries."""99        level_counts = {}100        for log in self.logs:101            level = log["level"]102            level_counts[level] = level_counts.get(level, 0) + 1103        104        return {105            "total_entries": len(self.logs),106            "max_size": self.max_size,107            "start_time": self.start_time.isoformat(),108            "uptime_seconds": (datetime.utcnow() - self.start_time).total_seconds(),109            "by_level": level_counts110        }111 112 113class ContainerLogger:114    """115    Logger for container-specific information and debugging.116 117    Writes to:118    - stdout (console)119    - in-memory circular buffer (for /logs API endpoint)120    - rotating log file on disk (path from SERVER_LOG_FILE env var)121    """122 123    def __init__(self, max_buffer_size: int = 10000):124        """Initialize container logger."""125        self.buffer = LogBuffer(max_size=max_buffer_size)126        self.logger = logging.getLogger("container")127        self._setup_logging()128 129    def _setup_logging(self) -> None:130        """Configure Python's built-in logging with console and file handlers."""131        if self.logger.handlers:132            return  # Already configured; avoid adding duplicate handlers133 134        formatter = logging.Formatter(135            "%(asctime)s - %(name)s - %(levelname)s - %(message)s",136            datefmt="%Y-%m-%dT%H:%M:%S",137        )138 139        # Console handler140        console_handler = logging.StreamHandler(sys.stdout)141        console_handler.setFormatter(formatter)142        console_handler.setLevel(logging.DEBUG)143        self.logger.addHandler(console_handler)144 145        # Rotating file handler — path and size from env vars146        log_file = os.getenv("SERVER_LOG_FILE", "server.log")147        max_bytes = int(os.getenv("LOG_MAX_BYTES", str(10 * 1024 * 1024)))  # 10 MB default148        backup_count = int(os.getenv("LOG_BACKUP_COUNT", "5"))149 150        try:151            log_dir = Path(log_file).parent152            if str(log_dir) != ".":153                log_dir.mkdir(parents=True, exist_ok=True)154 155            file_handler = RotatingFileHandler(156                log_file,157                maxBytes=max_bytes,158                backupCount=backup_count,159                encoding="utf-8",160            )161            file_handler.setFormatter(formatter)162            file_handler.setLevel(logging.DEBUG)163            self.logger.addHandler(file_handler)164        except Exception as exc:165            # Fall back to console-only if file handler setup fails166            self.logger.warning("Could not set up file logging (%s): %s", log_file, exc)167 168        self.logger.setLevel(logging.DEBUG)169    170    def debug(self, message: str, extra: Optional[Dict] = None):171        """Log debug message."""172        self.logger.debug(message)173        self.buffer.add("DEBUG", message, extra)174    175    def info(self, message: str, extra: Optional[Dict] = None):176        """Log info message."""177        self.logger.info(message)178        self.buffer.add("INFO", message, extra)179    180    def warning(self, message: str, extra: Optional[Dict] = None):181        """Log warning message."""182        self.logger.warning(message)183        self.buffer.add("WARNING", message, extra)184    185    def error(self, message: str, extra: Optional[Dict] = None):186        """Log error message."""187        self.logger.error(message)188        self.buffer.add("ERROR", message, extra)189    190    def critical(self, message: str, extra: Optional[Dict] = None):191        """Log critical message."""192        self.logger.critical(message)193        self.buffer.add("CRITICAL", message, extra)194    195    def get_logs(self, 196                 limit: int = 100, 197                 level: Optional[str] = None,198                 recent_only: bool = True) -> Dict:199        """200        Get formatted container logs.201        202        Args:203            limit: Maximum number of log entries to return204            level: Filter by log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)205            recent_only: If True, get recent logs; if False, get all206            207        Returns:208            Dictionary containing logs and metadata209        """210        if recent_only:211            logs = self.buffer.get_recent(count=limit, level=level)212        else:213            all_logs = self.buffer.get_all(level=level)214            logs = all_logs[-limit:] if all_logs else []215        216        return {217            "logs": logs,218            "count": len(logs),219            "stats": self.buffer.get_stats(),220            "filter": {221                "limit": limit,222                "level": level,223                "recent_only": recent_only224            }225        }226    227    def get_container_info(self) -> Dict:228        """Get container runtime information."""229        import socket230        return {231            "hostname": os.getenv("HOSTNAME", socket.gethostname()),232            "log_file": os.getenv("SERVER_LOG_FILE", "server.log"),233            "uptime": (datetime.utcnow() - self.buffer.start_time).total_seconds(),234            "log_buffer_size": self.buffer.max_size,235            "log_entries": len(self.buffer.logs),236            "start_time": self.buffer.start_time.isoformat(),237        }238    239    def get_health_status(self) -> Dict:240        """Get health status of the logger and container."""241        stats = self.buffer.get_stats()242        error_count = stats.get("by_level", {}).get("ERROR", 0)243        critical_count = stats.get("by_level", {}).get("CRITICAL", 0)244        245        health_status = "healthy"246        if critical_count > 0:247            health_status = "critical"248        elif error_count > 10:249            health_status = "degraded"250        251        return {252            "status": health_status,253            "error_count": error_count,254            "critical_count": critical_count,255            "total_logs": stats.get("total_entries", 0),256            "timestamp": datetime.utcnow().isoformat(),257        }258 259 260# Global logger instance261_container_logger: Optional[ContainerLogger] = None262 263 264def get_container_logger() -> ContainerLogger:265    """Get or create the global container logger instance."""266    global _container_logger267    if _container_logger is None:268        _container_logger = ContainerLogger()269    return _container_logger270 271 272def log_request(method: str, path: str, status_code: int, client_ip: str = "unknown"):273    """Log an HTTP request."""274    logger = get_container_logger()275    logger.info(276        f"HTTP {method} {path}",277        extra={278            "http_method": method,279            "http_path": path,280            "http_status": status_code,281            "client_ip": client_ip,282            "type": "http_request"283        }284    )285 286 287def log_environment_event(event_type: str, details: Optional[Dict] = None):288    """Log an environment-related event."""289    logger = get_container_logger()290    message = f"Environment: {event_type}"291    logger.info(message, extra={"event_type": event_type, "details": details or {}})292 293 294def log_websocket_event(event: str, client_id: str = "unknown", details: Optional[Dict] = None):295    """Log WebSocket events."""296    logger = get_container_logger()297    message = f"WebSocket {event} ({client_id})"298    logger.info(299        message,300        extra={301            "websocket_event": event,302            "client_id": client_id,303            "details": details or {}304        }305    )306