import atexit import logging import logging.handlers import os from pathlib import ( Path, ) import queue import sys import threading from typing import ( Any, ) # Create a log queue log_queue: "queue.Queue[Any]" = queue.Queue() # Store the current listener to stop it on exit _current_listener: logging.handlers.QueueListener | None = None # Store the handlers for proper cleanup _current_handlers: list[logging.Handler] = [] # Event to track when the listener is ready _listener_ready = threading.Event() # Default format for log messages DEFAULT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" def _parse_debug_modules(debug_str: str) -> dict[str, int]: """ Parse the LIBP2P_DEBUG environment variable to determine module-specific log levels. Format examples: - "DEBUG" # All modules at DEBUG level - "libp2p.identity.identify:DEBUG" # Only identify module at DEBUG - "identity.identify:DEBUG" # Same as above, libp2p prefix is optional - "libp2p.identity:DEBUG,libp2p.transport:INFO" # Multiple modules """ module_levels: dict[str, int] = {} # Handle empty or whitespace-only string if not debug_str or debug_str.isspace(): return module_levels # If it's a plain log level without any colons, apply to all if ":" not in debug_str and debug_str.upper() in logging._nameToLevel: return {"": getattr(logging, debug_str.upper())} # Handle module-specific levels for part in debug_str.split(","): if ":" not in part: continue module, level = part.split(":") level = level.upper() if level not in logging._nameToLevel: continue # Handle module name module = module.strip() # Remove libp2p prefix if present, it will be added back when creating logger module = module.replace("libp2p.", "") # Convert any remaining dots to ensure proper format module = module.replace("/", ".").strip(".") module_levels[module] = getattr(logging, level) return module_levels def setup_logging() -> None: """ Set up logging configuration based on environment variables. Environment Variables: LIBP2P_DEBUG Controls logging levels. Examples: - "DEBUG" (all modules at DEBUG level) - "libp2p.identity.identify:DEBUG" (only identify module at DEBUG) - "identity.identify:DEBUG" (same as above, libp2p prefix optional) - "libp2p.identity:DEBUG,libp2p.transport:INFO" (multiple modules) LIBP2P_DEBUG_FILE If set, specifies the file path for log output. When this variable is set, logs will only be written to the specified file. If not set, logs will be written to both a default file (in the system's temp directory) and to stderr (console output). The logging system uses Python's native hierarchical logging: - Loggers are organized in a hierarchy using dots (e.g., libp2p.identity.identify) - Child loggers inherit their parent's level unless explicitly set - The root libp2p logger controls the default level """ global _current_listener, _listener_ready, _current_handlers # Reset the event _listener_ready.clear() # Stop existing listener if any if _current_listener is not None: _current_listener.stop() _current_listener = None # Close and clear existing handlers for handler in _current_handlers: if isinstance(handler, logging.FileHandler): handler.close() _current_handlers.clear() # Get the log level from environment variable debug_str = os.environ.get("LIBP2P_DEBUG", "") if not debug_str: # If LIBP2P_DEBUG is not set, disable logging root_logger = logging.getLogger("libp2p") root_logger.handlers.clear() root_logger.setLevel(logging.WARNING) root_logger.propagate = False _listener_ready.set() # Signal that we're done return # Parse module-specific levels module_levels = _parse_debug_modules(debug_str) # If no valid levels specified, default to WARNING if not module_levels: root_logger = logging.getLogger("libp2p") root_logger.handlers.clear() root_logger.setLevel(logging.WARNING) root_logger.propagate = False _listener_ready.set() # Signal that we're done return # Create formatter formatter = logging.Formatter(DEFAULT_LOG_FORMAT) # Configure handlers handlers: list[logging.StreamHandler[Any] | logging.FileHandler] = [] # Console handler console_handler = logging.StreamHandler(sys.stderr) console_handler.setFormatter(formatter) handlers.append(console_handler) # File handler (if configured) log_file = os.environ.get("LIBP2P_DEBUG_FILE") if log_file: # Ensure the directory exists log_path = Path(log_file) log_path.parent.mkdir(parents=True, exist_ok=True) else: # Use cross-platform temp file creation from libp2p.utils.paths import create_temp_file log_file = str(create_temp_file(prefix="py-libp2p_", suffix=".log")) # Print the log file path so users know where to find it print(f"Logging to: {log_file}", file=sys.stderr) try: file_handler = logging.FileHandler( log_file, mode="w" ) # Use 'w' mode to clear file file_handler.setFormatter(formatter) handlers.append(file_handler) except Exception as e: print(f"Error creating file handler: {e}", file=sys.stderr) raise # Create a QueueHandler and QueueListener queue_handler = logging.handlers.QueueHandler(log_queue) # Configure root logger for libp2p root_logger = logging.getLogger("libp2p") root_logger.handlers.clear() root_logger.addHandler(queue_handler) root_logger.propagate = False # Set default level based on configuration if "" in module_levels: # Global level specified root_logger.setLevel(module_levels[""]) else: # Default to INFO for module-specific logging root_logger.setLevel(logging.INFO) # Configure module-specific levels for module, level in module_levels.items(): if module: # Skip the default "" key logger = logging.getLogger(f"libp2p.{module}") logger.handlers.clear() logger.addHandler(queue_handler) logger.setLevel(level) logger.propagate = False # Prevent message duplication # Store handlers globally for cleanup _current_handlers.extend(handlers) # Start the listener AFTER configuring all loggers _current_listener = logging.handlers.QueueListener( log_queue, *handlers, respect_handler_level=True ) _current_listener.start() # Signal that the listener is ready _listener_ready.set() # Register cleanup function @atexit.register def cleanup_logging() -> None: """Clean up logging resources on exit.""" global _current_listener, _current_handlers if _current_listener is not None: _current_listener.stop() _current_listener = None # Close all file handlers to ensure proper cleanup on Windows for handler in _current_handlers: if isinstance(handler, logging.FileHandler): handler.close() _current_handlers.clear()