Coverage for src / lilbee / cli / commands / serve_logging.py: 100%

1"""Server log-file routing for ``lilbee serve``: rotating file under the data root.""" 

2 

3from __future__ import annotations 

4 

5import faulthandler 

6import logging 

7import sys 

8from logging.handlers import RotatingFileHandler 

9from pathlib import Path 

10from types import TracebackType 

11from typing import IO 

12 

13from lilbee.core.config import cfg 

14 

15logger = logging.getLogger(__name__) 

16 

17_LOG_DIR_NAME = "logs" 

18_SERVER_LOG_FILE_NAME = "server.log" 

19_FAULT_LOG_FILE_NAME = "server-fault.log" 

20_MAX_BYTES = 2_097_152 # 2 MiB 

21_BACKUP_COUNT = 3 

22_LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s" 

23 

24 

25def setup_server_log_file() -> Path: 

26 """Install a RotatingFileHandler at ``cfg.data_root/logs/server.log``. Idempotent.""" 

27 log_dir = cfg.data_root / _LOG_DIR_NAME 

28 log_dir.mkdir(parents=True, exist_ok=True) 

29 log_path = log_dir / _SERVER_LOG_FILE_NAME 

30 

31 root = logging.getLogger() 

32 for handler in root.handlers: 

33 if isinstance(handler, RotatingFileHandler) and Path(handler.baseFilename) == log_path: 

34 return log_path 

35 

36 file_handler = RotatingFileHandler(log_path, maxBytes=_MAX_BYTES, backupCount=_BACKUP_COUNT) 

37 file_handler.setFormatter(logging.Formatter(_LOG_FORMAT)) 

38 file_handler.setLevel(logging.INFO) 

39 root.addHandler(file_handler) 

40 if root.level > logging.INFO: 

41 # NOTSET stream handlers delegate to root; pin them so stderr verbosity is unchanged. 

42 for handler in root.handlers: 

43 if handler is not file_handler and handler.level == logging.NOTSET: 

44 handler.setLevel(root.level) 

45 root.setLevel(logging.INFO) 

46 return log_path 

47 

48 

49class _FaultLog: 

50 """Holds the fault-log handle alive; faulthandler writes to the raw fd.""" 

51 

52 def __init__(self) -> None: 

53 self._handle: IO[str] | None = None 

54 

55 def enable(self) -> Path: 

56 """Open the fault log if needed and point faulthandler at it.""" 

57 log_dir = cfg.data_root / _LOG_DIR_NAME 

58 log_dir.mkdir(parents=True, exist_ok=True) 

59 fault_path = log_dir / _FAULT_LOG_FILE_NAME 

60 if self._handle is None or self._handle.closed: 

61 self._handle = fault_path.open("a") 

62 faulthandler.enable(file=self._handle) 

63 return fault_path 

64 

65 

66_fault_log = _FaultLog() 

67 

68 

69def enable_fault_log() -> Path: 

70 """Point ``faulthandler`` at ``cfg.data_root/logs/server-fault.log``. Idempotent.""" 

71 return _fault_log.enable() 

72 

73 

74class _ExceptHook: 

75 """Tracks whether the logging excepthook is installed.""" 

76 

77 def __init__(self) -> None: 

78 self.installed = False 

79 

80 def install(self) -> None: 

81 """Install the logging excepthook once, chaining to the previous hook.""" 

82 if self.installed: 

83 return 

84 previous = sys.excepthook 

85 

86 def _hook( 

87 exc_type: type[BaseException], exc: BaseException, tb: TracebackType | None 

88 ) -> None: 

89 logger.critical("unhandled exception", exc_info=(exc_type, exc, tb)) 

90 previous(exc_type, exc, tb) 

91 

92 sys.excepthook = _hook 

93 self.installed = True 

94 

95 

96_except_hook = _ExceptHook() 

97 

98 

99def install_excepthook() -> None: 

100 """Log unhandled exceptions before the previous hook runs. Idempotent.""" 

101 _except_hook.install() 

102 

103 

104def setup_server_logging() -> None: 

105 """File log, fault log, and excepthook for a ``serve`` process.""" 

106 setup_server_log_file() 

107 enable_fault_log() 

108 install_excepthook()