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 collections.abc import Iterable 

9from logging.handlers import RotatingFileHandler 

10from pathlib import Path 

11from types import TracebackType 

12from typing import IO 

13 

14from lilbee.core.config import cfg, load_warnings 

15 

16logger = logging.getLogger(__name__) 

17 

18_LOG_DIR_NAME = "logs" 

19_SERVER_LOG_FILE_NAME = "server.log" 

20_FAULT_LOG_FILE_NAME = "server-fault.log" 

21_MAX_BYTES = 2_097_152 # 2 MiB 

22_BACKUP_COUNT = 3 

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

24_UVICORN_ACCESS_LOGGER = "uvicorn.access" 

25_HEALTH_ACCESS_PATH = "/api/health" 

26 

27 

28def setup_server_log_file(earlier: Iterable[str] = ()) -> Path: 

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

30 

31 A new handler starts with the *earlier* warnings, logged before any handler existed. 

32 """ 

33 log_dir = cfg.data_root / _LOG_DIR_NAME 

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

35 log_path = log_dir / _SERVER_LOG_FILE_NAME 

36 

37 root = logging.getLogger() 

38 for handler in root.handlers: 

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

40 return log_path 

41 

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

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

44 file_handler.setLevel(logging.INFO) 

45 root.addHandler(file_handler) 

46 for message in earlier: 

47 file_handler.handle( 

48 logging.makeLogRecord( 

49 { 

50 "name": logger.name, 

51 "levelno": logging.WARNING, 

52 "levelname": "WARNING", 

53 "msg": message, 

54 } 

55 ) 

56 ) 

57 if root.level > logging.INFO: 

58 root.setLevel(logging.INFO) 

59 return log_path 

60 

61 

62class _FaultLog: 

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

64 

65 def __init__(self) -> None: 

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

67 

68 def enable(self) -> Path: 

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

70 log_dir = cfg.data_root / _LOG_DIR_NAME 

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

72 fault_path = log_dir / _FAULT_LOG_FILE_NAME 

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

74 self._handle = fault_path.open("a", encoding="utf-8") 

75 faulthandler.enable(file=self._handle) 

76 return fault_path 

77 

78 

79_fault_log = _FaultLog() 

80 

81 

82def enable_fault_log() -> Path: 

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

84 return _fault_log.enable() 

85 

86 

87class _ExceptHook: 

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

89 

90 def __init__(self) -> None: 

91 self.installed = False 

92 

93 def install(self) -> None: 

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

95 if self.installed: 

96 return 

97 previous = sys.excepthook 

98 

99 def _hook( 

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

101 ) -> None: 

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

103 previous(exc_type, exc, tb) 

104 

105 sys.excepthook = _hook 

106 self.installed = True 

107 

108 

109_except_hook = _ExceptHook() 

110 

111 

112def install_excepthook() -> None: 

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

114 _except_hook.install() 

115 

116 

117def _show_info_on_console() -> None: 

118 """Lower each root console handler to INFO so startup and signal lines show at any CLI level.""" 

119 for handler in logging.getLogger().handlers: 

120 if ( 

121 isinstance(handler, logging.StreamHandler) 

122 and not isinstance(handler, logging.FileHandler) 

123 and handler.level > logging.INFO 

124 ): 

125 handler.setLevel(logging.INFO) 

126 

127 

128def setup_server_logging() -> None: 

129 """Console, file log, fault log, and excepthook for a ``serve`` process.""" 

130 setup_server_log_file(load_warnings) 

131 _show_info_on_console() 

132 enable_fault_log() 

133 install_excepthook() 

134 

135 

136class _HealthAccessFilter(logging.Filter): 

137 """Drop uvicorn access records for the health endpoint.""" 

138 

139 def filter(self, record: logging.LogRecord) -> bool: 

140 return _HEALTH_ACCESS_PATH not in record.getMessage() 

141 

142 

143def install_health_access_filter() -> None: 

144 """Quiet health-poll lines on the uvicorn access logger. Idempotent.""" 

145 access_logger = logging.getLogger(_UVICORN_ACCESS_LOGGER) 

146 for filt in access_logger.filters: 

147 if isinstance(filt, _HealthAccessFilter): 

148 return 

149 access_logger.addFilter(_HealthAccessFilter())