""" DSS Structured Logger - JSON-based logging for AI-consumable audit trails Provides structured, machine-readable logging in JSONL format (one JSON object per line). All DSS operations are logged with consistent fields for analysis, debugging, and compliance. Features: - JSONL format (newline-delimited JSON) for easy parsing - Structured log entries with standardized fields - Context tracking (session_id, tool_name, operation) - Performance metrics (duration, timestamps) - Log rotation and cleanup - Integration with DSSRuntime Usage: from core.structured_logger import get_logger, LogContext logger = get_logger("dss.tool.sync_figma") with LogContext(session_id="abc123", tool="dss_sync_figma"): logger.info("Starting Figma sync", extra={"file_key": "xyz"}) # ... operation ... logger.info("Figma sync complete", extra={"tokens_extracted": 42}) """ import json import logging import os import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Optional from contextlib import contextmanager import threading # Thread-local storage for context _context = threading.local() class DSSJSONFormatter(logging.Formatter): """ Custom JSON formatter for structured logging. Outputs each log record as a single-line JSON object with standardized fields: - timestamp: ISO 8601 UTC timestamp - level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) - logger: Logger name (e.g., "dss.tool.sync_figma") - message: Human-readable log message - context: Additional contextual data (session_id, tool_name, etc.) - extra: Tool-specific extra data """ def format(self, record: logging.LogRecord) -> str: """Format log record as single-line JSON""" # Build base log entry log_entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), } # Add context from thread-local storage if hasattr(_context, "session_id"): log_entry["session_id"] = _context.session_id if hasattr(_context, "tool_name"): log_entry["tool"] = _context.tool_name if hasattr(_context, "operation"): log_entry["operation"] = _context.operation # Add extra fields from record if hasattr(record, "extra_data"): log_entry["extra"] = record.extra_data # Add exception info if present if record.exc_info: log_entry["exception"] = { "type": record.exc_info[0].__name__ if record.exc_info[0] else None, "message": str(record.exc_info[1]) if record.exc_info[1] else None, "traceback": self.formatException(record.exc_info) if record.exc_info else None, } # Add location info for ERROR and above if record.levelno >= logging.ERROR: log_entry["location"] = { "file": record.pathname, "line": record.lineno, "function": record.funcName, } return json.dumps(log_entry, default=str) class DSSLogger(logging.Logger): """ Extended logger with structured logging support. Wraps standard Python logger with methods that accept extra data as keyword arguments for structured logging. """ def _log_with_extra(self, level: int, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Internal method to log with extra structured data""" if extra: # Store extra data in a custom attribute extra_record = {"extra_data": extra} super()._log(level, msg, (), extra=extra_record, **kwargs) else: super()._log(level, msg, (), **kwargs) def debug(self, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Log DEBUG message with optional extra data""" self._log_with_extra(logging.DEBUG, msg, extra, **kwargs) def info(self, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Log INFO message with optional extra data""" self._log_with_extra(logging.INFO, msg, extra, **kwargs) def warning(self, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Log WARNING message with optional extra data""" self._log_with_extra(logging.WARNING, msg, extra, **kwargs) def error(self, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Log ERROR message with optional extra data""" self._log_with_extra(logging.ERROR, msg, extra, **kwargs) def critical(self, msg: str, extra: Optional[Dict[str, Any]] = None, **kwargs): """Log CRITICAL message with optional extra data""" self._log_with_extra(logging.CRITICAL, msg, extra, **kwargs) # Configure custom logger class logging.setLoggerClass(DSSLogger) def get_logger(name: str, log_file: Optional[str] = None) -> DSSLogger: """ Get or create a structured logger instance. Args: name: Logger name (e.g., "dss.tool.sync_figma") log_file: Optional custom log file path (defaults to .dss/logs/dss-operations.jsonl) Returns: DSSLogger instance configured for structured logging Example: logger = get_logger("dss.tool.extract_tokens") logger.info("Starting token extraction", extra={"source": "css"}) """ logger = logging.getLogger(name) # Only configure if not already configured if not logger.handlers: # Determine log file path if log_file is None: dss_home = os.environ.get("DSS_HOME", ".dss") log_dir = Path(dss_home) / "logs" log_dir.mkdir(parents=True, exist_ok=True) log_file = str(log_dir / "dss-operations.jsonl") # Create file handler with JSON formatter file_handler = logging.FileHandler(log_file, mode="a", encoding="utf-8") file_handler.setFormatter(DSSJSONFormatter()) logger.addHandler(file_handler) # Also add console handler for development (can be disabled in production) if os.environ.get("DSS_LOG_CONSOLE", "false").lower() == "true": console_handler = logging.StreamHandler(sys.stderr) console_handler.setFormatter(DSSJSONFormatter()) logger.addHandler(console_handler) # Set log level from environment or default to INFO log_level = os.environ.get("DSS_LOG_LEVEL", "INFO").upper() logger.setLevel(getattr(logging, log_level, logging.INFO)) # Prevent propagation to root logger logger.propagate = False return logger @contextmanager def LogContext(session_id: Optional[str] = None, tool: Optional[str] = None, operation: Optional[str] = None): """ Context manager for adding structured context to log entries. All log entries within this context will include the provided fields (session_id, tool_name, operation). Args: session_id: Unique session identifier tool: Tool name (e.g., "dss_sync_figma") operation: Operation being performed (e.g., "token_extraction") Example: with LogContext(session_id="abc123", tool="dss_sync_figma"): logger.info("Starting sync") # This log will include session_id and tool fields """ # Store previous context prev_session_id = getattr(_context, "session_id", None) prev_tool_name = getattr(_context, "tool_name", None) prev_operation = getattr(_context, "operation", None) # Set new context if session_id: _context.session_id = session_id if tool: _context.tool_name = tool if operation: _context.operation = operation try: yield finally: # Restore previous context if prev_session_id: _context.session_id = prev_session_id elif hasattr(_context, "session_id"): delattr(_context, "session_id") if prev_tool_name: _context.tool_name = prev_tool_name elif hasattr(_context, "tool_name"): delattr(_context, "tool_name") if prev_operation: _context.operation = prev_operation elif hasattr(_context, "operation"): delattr(_context, "operation") class PerformanceLogger: """ Helper for logging operation performance metrics. Automatically measures duration and logs performance data. Example: perf = PerformanceLogger("token_extraction") perf.start() # ... operation ... perf.end(extra={"tokens_found": 42}) """ def __init__(self, operation: str, logger: Optional[DSSLogger] = None): """ Initialize performance logger. Args: operation: Operation name logger: Optional logger (defaults to root DSS logger) """ self.operation = operation self.logger = logger or get_logger("dss.performance") self.start_time = None self.end_time = None def start(self): """Mark operation start time""" self.start_time = datetime.now(timezone.utc) self.logger.debug(f"Started: {self.operation}", extra={ "operation": self.operation, "start_time": self.start_time.isoformat(), }) def end(self, extra: Optional[Dict[str, Any]] = None): """ Mark operation end time and log performance. Args: extra: Additional metrics to log """ self.end_time = datetime.now(timezone.utc) if self.start_time is None: self.logger.warning(f"Performance logger end() called without start() for: {self.operation}") return duration_ms = (self.end_time - self.start_time).total_seconds() * 1000 perf_data = { "operation": self.operation, "duration_ms": round(duration_ms, 2), "start_time": self.start_time.isoformat(), "end_time": self.end_time.isoformat(), } if extra: perf_data.update(extra) self.logger.info(f"Completed: {self.operation}", extra=perf_data) def configure_log_rotation(log_dir: Optional[Path] = None, max_bytes: int = 10 * 1024 * 1024, backup_count: int = 5): """ Configure log rotation for DSS log files. Args: log_dir: Log directory (defaults to .dss/logs/) max_bytes: Max size per log file (default: 10MB) backup_count: Number of backup files to keep (default: 5) Note: This uses RotatingFileHandler. For production, consider using a log rotation service like logrotate. """ from logging.handlers import RotatingFileHandler if log_dir is None: dss_home = os.environ.get("DSS_HOME", ".dss") log_dir = Path(dss_home) / "logs" log_dir.mkdir(parents=True, exist_ok=True) log_file = log_dir / "dss-operations.jsonl" # Get root DSS logger logger = logging.getLogger("dss") # Remove existing file handlers for handler in logger.handlers[:]: if isinstance(handler, logging.FileHandler): logger.removeHandler(handler) # Add rotating file handler rotating_handler = RotatingFileHandler( str(log_file), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8" ) rotating_handler.setFormatter(DSSJSONFormatter()) logger.addHandler(rotating_handler) logger.info("Log rotation configured", extra={ "max_bytes": max_bytes, "backup_count": backup_count, "log_file": str(log_file), }) # Example usage (can be removed in production) if __name__ == "__main__": # Example 1: Basic logging logger = get_logger("dss.example") logger.info("DSS operation started", extra={"user": "admin"}) # Example 2: Context-based logging with LogContext(session_id="session-123", tool="dss_sync_figma"): logger.info("Syncing Figma file", extra={"file_key": "abc123"}) logger.info("Sync complete", extra={"tokens_extracted": 42}) # Example 3: Performance logging perf = PerformanceLogger("token_extraction", logger) perf.start() # Simulate work import time time.sleep(0.1) perf.end(extra={"tokens_found": 100}) print(f"\nLogs written to: {Path('.dss/logs/dss-operations.jsonl').absolute()}")