Coverage for bzfs_main / loggers.py: 100%

1# Copyright 2024 Wolfgang Hoschek AT mac DOT com 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# 

15"""Logging helpers that build default and syslog-enabled loggers; Centralizes logger setup so that all bzfs tools share 

16uniform formatting and configuration. 

17 

18Each bzfs.Job has its own separate Logger object such that multiple Jobs can run concurrently in the same Python process 

19without interfering with each other's logging semantics. This is achieved by passing a unique ``logger_name_suffix`` per job 

20to ``get_logger()`` so each job receives an isolated Logger instance. Callers are responsible for closing any loggers they 

21own via ``reset_logger()``. 

22""" 

23from __future__ import ( 

24 annotations, 

25) 

26import argparse 

27import contextlib 

28import logging 

29import sys 

30from datetime import ( 

31 datetime, 

32) 

33from logging import ( 

34 Logger, 

35) 

36from typing import ( 

37 TYPE_CHECKING, 

38 Any, 

39 Final, 

40 final, 

41) 

42 

43from bzfs_main.util.utils import ( 

44 LOG_STDERR, 

45 LOG_STDOUT, 

46 LOG_TRACE, 

47) 

48 

49if TYPE_CHECKING: # pragma: no cover - for type hints only 

50 from bzfs_main.configuration import ( 

51 LogParams, 

52 ) 

53 

54 

55def _get_logger_name() -> str: 

56 """Returns the canonical logger name used throughout bzfs.""" 

57 return "bzfs_main.bzfs" 

58 

59 

60def _resolve_logger_name(logger_name_suffix: str) -> str: 

61 """Returns the logger name for the given optional logger suffix. 

62 

63 Each bzfs.Job instance gets a distinct Logger name and hence uses a separate private (thread-safe) logging.Logger object. 

64 """ 

65 logger_name: str = _get_logger_name() 

66 return logger_name + "." + logger_name_suffix if logger_name_suffix else logger_name 

67 

68 

69def reset_logger(log: Logger) -> None: 

70 """Removes and closes logging handlers (and closes their files) and resets logger to default state.""" 

71 for handler in log.handlers.copy(): 

72 log.removeHandler(handler) 

73 with contextlib.suppress(BrokenPipeError): 

74 handler.flush() 

75 handler.close() 

76 for _filter in log.filters.copy(): 

77 log.removeFilter(_filter) 

78 log.setLevel(logging.NOTSET) 

79 log.propagate = True 

80 

81 

82def get_logger( 

83 log_params: LogParams, args: argparse.Namespace, log: Logger | None = None, logger_name_suffix: str = "" 

84) -> Logger: 

85 """Returns a logger configured from CLI arguments or an optional base logger.""" 

86 if log is not None: 

87 assert isinstance(log, Logger) 

88 return log # use third party provided logger object 

89 return _get_default_logger(log_params, args, logger_name_suffix=logger_name_suffix) 

90 

91 

92def _get_default_logger(log_params: LogParams, args: argparse.Namespace, logger_name_suffix: str = "") -> Logger: 

93 """Creates the default logger with stream, file and optional syslog handlers.""" 

94 logger_name: str = _resolve_logger_name(logger_name_suffix) 

95 log = Logger(logger_name) # noqa: LOG001 do not register logger with Logger.manager to avoid potential memory leak 

96 log.setLevel(log_params.log_level) 

97 log.propagate = False # don't propagate log messages up to the root logger to avoid emitting duplicate messages 

98 

99 handler: logging.StreamHandler = logging.StreamHandler(stream=sys.stdout) 

100 handler.terminator = "" 

101 handler.setFormatter(get_default_log_formatter(default_terminator="\n", log_params=log_params)) 

102 handler.setLevel(log_params.log_level) 

103 log.addHandler(handler) 

104 

105 handler = logging.FileHandler(log_params.log_file, encoding="utf-8") 

106 handler.terminator = "" 

107 handler.setFormatter(get_default_log_formatter(default_terminator="\n")) 

108 handler.setLevel(log_params.log_level) 

109 log.addHandler(handler) 

110 

111 address = args.log_syslog_address 

112 if address: # optionally, also log to local or remote syslog 

113 from logging import handlers # lazy import for startup perf 

114 

115 address, socktype = _get_syslog_address(address, args.log_syslog_socktype) 

116 log_syslog_prefix = str(args.log_syslog_prefix).strip().replace("%", "") # sanitize 

117 syslog_handler = handlers.SysLogHandler(address=address, facility=args.log_syslog_facility, socktype=socktype) 

118 syslog_handler.setFormatter(get_default_log_formatter(prefix=log_syslog_prefix + " ")) 

119 syslog_handler.setLevel(args.log_syslog_level) 

120 log.addHandler(syslog_handler) 

121 if syslog_handler.level < log.getEffectiveLevel(): 

122 log_level_name: str = logging.getLevelName(log.getEffectiveLevel()) 

123 log.warning( 

124 "%s", 

125 f"No messages with priority lower than {log_level_name} will be sent to syslog because syslog " 

126 f"log level {args.log_syslog_level} is lower than overall log level {log_level_name}.", 

127 ) 

128 return log 

129 

130 

131LOG_LEVEL_PREFIXES: Final[dict[int, str]] = { 

132 logging.CRITICAL: "[C] CRITICAL:", 

133 logging.ERROR: "[E] ERROR:", 

134 logging.WARNING: "[W]", 

135 logging.INFO: "[I]", 

136 logging.DEBUG: "[D]", 

137 LOG_TRACE: "[T]", 

138} 

139 

140 

141def get_default_log_formatter( 

142 prefix: str = "", default_terminator: str = "", log_params: LogParams | None = None 

143) -> logging.Formatter: 

144 """Returns a formatter for bzfs logs with optional prefix and column padding.""" 

145 level_prefixes_: dict[int, str] = LOG_LEVEL_PREFIXES.copy() 

146 log_stderr_: int = LOG_STDERR 

147 log_stdout_: int = LOG_STDOUT 

148 cols: int = 0 if log_params is None else log_params.terminal_columns 

149 

150 @final 

151 class DefaultLogFormatter(logging.Formatter): 

152 """Formatter adding timestamps and padding for progress output.""" 

153 

154 def format(self, record: logging.LogRecord) -> str: 

155 """Formats the given record, adding timestamp and level prefix and padding.""" 

156 levelno: int = record.levelno 

157 if levelno != log_stderr_ and levelno != log_stdout_: # emit stdout and stderr "as-is" (no formatting) 

158 timestamp: str = datetime.now().isoformat(sep=" ", timespec="seconds") # 2024-09-03 12:26:15 

159 ts_level: str = f"{timestamp} {level_prefixes_.get(levelno, '')} " 

160 msg: str = record.msg 

161 i: int = msg.find("%s") 

162 msg = ts_level + msg 

163 if i >= 1: 

164 i += len(ts_level) 

165 msg = msg[0:i].ljust(54) + msg[i:] # right-pad msg if record.msg contains "%s" unless at start 

166 if record.exc_info or record.exc_text or record.stack_info: 

167 original_msg = record.msg 

168 try: 

169 record.msg = msg 

170 msg = super().format(record) 

171 finally: 

172 record.msg = original_msg # restore original msg so other logging handlers see unmodified state 

173 elif record.args: 

174 msg = msg % record.args 

175 else: 

176 msg = super().format(record) 

177 

178 msg = prefix + msg 

179 msg = msg.ljust(cols) # w/ progress line, "overwrite" trailing chars of previous msg with spaces 

180 terminator: str = getattr(record, "terminator", default_terminator) 

181 return msg + terminator 

182 

183 return DefaultLogFormatter() 

184 

185 

186def get_simple_logger(program: str, logger_name_suffix: str = "") -> Logger: 

187 """Returns a minimal logger for simple tools.""" 

188 

189 level_prefixes_: dict[int, str] = LOG_LEVEL_PREFIXES.copy() 

190 logger_name = program + "." + logger_name_suffix if logger_name_suffix else program 

191 

192 @final 

193 class LevelFormatter(logging.Formatter): 

194 """Injects level prefix and program name into log records.""" 

195 

196 def format(self, record: logging.LogRecord) -> str: 

197 """Attaches extra fields before delegating to base formatter.""" 

198 record.level_prefix = level_prefixes_.get(record.levelno, "") 

199 record.program = program 

200 return super().format(record) 

201 

202 log = Logger(logger_name) # noqa: LOG001 do not register logger with Logger.manager to avoid potential memory leak 

203 log.setLevel(logging.INFO) 

204 log.propagate = False 

205 handler = logging.StreamHandler() 

206 handler.setFormatter( 

207 LevelFormatter(fmt="%(asctime)s %(level_prefix)s [%(program)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") 

208 ) 

209 log.addHandler(handler) 

210 return log 

211 

212 

213def set_logging_runtime_defaults() -> None: 

214 """Applies process-wide logging runtime defaults for production use.""" 

215 logging.addLevelName(LOG_TRACE, "TRACE") 

216 logging.addLevelName(LOG_STDERR, "STDERR") 

217 logging.addLevelName(LOG_STDOUT, "STDOUT") 

218 # perf: tell logging framework not to gather unnecessary expensive info for each log record 

219 logging.logProcesses = False 

220 logging.logThreads = False 

221 logging.logMultiprocessing = False 

222 logging.raiseExceptions = False # avoid noisy tracebacks from logging handler errors like BrokenPipeError in production 

223 

224 

225def _get_syslog_address(address: str, log_syslog_socktype: str) -> tuple[str | tuple[str, int], Any]: 

226 """Normalizes syslog address to tuple form and returns socket type.""" 

227 import socket # lazy import for startup perf 

228 

229 address = address.strip() 

230 socktype: socket.SocketKind | None = None 

231 if ":" in address: 

232 host, port_str = address.rsplit(":", 1) 

233 addr = (host.strip(), int(port_str.strip())) 

234 scktype: socket.SocketKind = socket.SOCK_DGRAM if log_syslog_socktype == "UDP" else socket.SOCK_STREAM # for TCP 

235 return addr, scktype 

236 return address, socktype