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_UVICORN_ACCESS_LOGGER = "uvicorn.access" 

24_HEALTH_ACCESS_PATH = "/api/health" 

25 

26 

27def setup_server_log_file() -> Path: 

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

29 log_dir = cfg.data_root / _LOG_DIR_NAME 

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

31 log_path = log_dir / _SERVER_LOG_FILE_NAME 

32 

33 root = logging.getLogger() 

34 for handler in root.handlers: 

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

36 return log_path 

37 

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

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

40 file_handler.setLevel(logging.INFO) 

41 root.addHandler(file_handler) 

42 if root.level > logging.INFO: 

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

44 for handler in root.handlers: 

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

46 handler.setLevel(root.level) 

47 root.setLevel(logging.INFO) 

48 return log_path 

49 

50 

51class _FaultLog: 

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

53 

54 def __init__(self) -> None: 

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

56 

57 def enable(self) -> Path: 

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

59 log_dir = cfg.data_root / _LOG_DIR_NAME 

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

61 fault_path = log_dir / _FAULT_LOG_FILE_NAME 

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

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

64 faulthandler.enable(file=self._handle) 

65 return fault_path 

66 

67 

68_fault_log = _FaultLog() 

69 

70 

71def enable_fault_log() -> Path: 

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

73 return _fault_log.enable() 

74 

75 

76class _ExceptHook: 

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

78 

79 def __init__(self) -> None: 

80 self.installed = False 

81 

82 def install(self) -> None: 

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

84 if self.installed: 

85 return 

86 previous = sys.excepthook 

87 

88 def _hook( 

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

90 ) -> None: 

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

92 previous(exc_type, exc, tb) 

93 

94 sys.excepthook = _hook 

95 self.installed = True 

96 

97 

98_except_hook = _ExceptHook() 

99 

100 

101def install_excepthook() -> None: 

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

103 _except_hook.install() 

104 

105 

106def setup_server_logging() -> None: 

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

108 setup_server_log_file() 

109 enable_fault_log() 

110 install_excepthook() 

111 

112 

113class _HealthAccessFilter(logging.Filter): 

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

115 

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

117 return _HEALTH_ACCESS_PATH not in record.getMessage() 

118 

119 

120def install_health_access_filter() -> None: 

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

122 access_logger = logging.getLogger(_UVICORN_ACCESS_LOGGER) 

123 for filt in access_logger.filters: 

124 if isinstance(filt, _HealthAccessFilter): 

125 return 

126 access_logger.addFilter(_HealthAccessFilter())