Python基礎——logging模塊詳解

logging簡介

logging是python的內置庫，主要用於進行格式化內容輸出，可將格式化內容輸出到文件，也可輸出到屏幕。我們在開發過程中常用print函數來進行調試，但是實際應用部署時我們要將日誌的信息要輸出到文件中，方便後續查找以及備份。在我們使用日誌管理時，我們還可以將日誌格式化成json對象轉存到ELK中方便圖形化查看及管理。前面說的這些，我們都可以通過logging所包含的功能以及提供擴展的方式來完成。

logging工作流程

這是從網上查到一個關於python logging模塊的工作流程圖，非常遺憾沒有找到出處。

可以看到圖中，幾個關鍵的類、

Logger

用於記錄日誌的對象。

通過流程圖可以看到

判斷是否enabled，實質就是看記錄的level（logger.info，logger.debug等）和當前logger對象設置的level是否滿足（比如logger設置的lever是Info，記錄時使用的logger.debug，那麼就會不滿足，所以不會記錄日誌）

查看logger的過濾器是否滿足。filter通過之後，交給logger的handler來記錄日誌，一個logger是可以設置多個handler。

交給Handlers實際記錄日誌

注：在整個應用中可以有多個logger，使用logging.getLogger時通過指定name來獲取對象，實際logging中還存在一個Manager類，由Manager來進行多logger的單例模式管理。

Handler

用於記錄日誌到具體的文件或者輸出流或其他的管道。

不同的Handler可能有不同的處理，但是底層原理還是做這三件事情。

Filter

用於過濾用戶記錄日誌時，是否滿足記錄標準

Formatter

用於格式化，輸出的字符串格式

這是原生自帶的格式

%(name)s Name of the logger (logging channel) %(levelno)s Numeric logging level for the message (DEBUG, INFO, WARNING, ERROR, CRITICAL) %(levelname)s Text logging level for the message ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") %(pathname)s Full pathname of the source file where the logging call was issued (if available) %(filename)s Filename portion of pathname %(module)s Module (name portion of filename) %(lineno)d Source line number where the logging call was issued (if available) %(funcName)s Function name %(created)f Time when the LogRecord was created (time.time() return value) %(asctime)s Textual time when the LogRecord was created %(msecs)d Millisecond portion of the creation time %(relativeCreated)d Time in milliseconds when the LogRecord was created, relative to the time the logging module was loaded (typically at application startup time) %(thread)d Thread ID (if available) %(threadName)s Thread name (if available) %(process)d Process ID (if available) %(message)s The result of record.getMessage(), computed just as the record is emitted

LogRecord

我們每一次的 logger.info， logger.debug， logger.error等實際上都是進行一次LogRecord的處理

包含一些基本的輸出日誌內容，以及內容中參數，還有附帶的函數名稱，行數等信息

logging源碼閱讀

這是我這邊基於字典的一個logging配置

import logging import logging.config import os basedir = os.path.abspath(os.path.dirname(__file__)) log_path = os.path.join(basedir, '..', 'logs') service_name = "test_log" if not os.path.isdir(log_path): os.mkdir(log_path) class InfoFilter(logging.Filter): def filter(self, record): """ 只篩選出INFO級別的日誌 """ if logging.INFO <= record.levelno < logging.ERROR: return super().filter(record) else: return 0 class ErrorFilter(logging.Filter): def filter(self, record): """ 只篩選出ERROR級別的日誌 """ if logging.ERROR <= record.levelno < logging.CRITICAL: return super().filter(record) else: return 0 LOG_CONFIG_DICT = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'simple': { 'class': 'logging.Formatter', 'format': '%(asctime)s %(levelname)s %(name)s %(filename)s %(module)s %(funcName)s ' '%(lineno)d %(thread)d %(threadName)s %(process)d %(processName)s %(message)s' }, # json模式, 方便ELK收集處理 'json': { 'class': 'logging.Formatter', 'format': '{"time:":"%(asctime)s","level":"%(levelname)s","logger_name":"%(name)s",' '"file_name":"%(filename)s","module":"%(module)s","func_name":"%(funcName)s",' '"line_number":"%(lineno)d","thread_id":"%(thread)d","thread_name":"%(threadName)s",' '"process_id":"%(process)d","process_name":"%(processName)s","message":"%(message)s"}'} }, # 過濾器 'filters': { 'info_filter': { '()': InfoFilter }, 'error_filter': { '()': ErrorFilter } }, # 處理器 'handlers': { # 控制檯輸出 'console': { 'class': 'logging.StreamHandler', 'level': 'INFO', 'formatter': 'simple' }, # info文件輸出 'info_file': { 'level': 'INFO', 'formatter': 'json', 'class': 'logging.handlers.TimedRotatingFileHandler', 'filename': '{0}/{1}_info.log'.format(log_path, service_name), 'when': "d", 'interval': 1, 'encoding': 'utf8', 'backupCount': 30, 'filters': ['info_filter'] }, # error文件輸出 'error_file': { 'level': 'ERROR', 'formatter': 'json', 'class': 'logging.handlers.TimedRotatingFileHandler', 'filename': '{0}/{1}_error.log'.format(log_path, service_name), 'when': "d", 'interval': 1, 'encoding': 'utf8', 'backupCount': 30, 'filters': ['error_filter'] } }, # 記錄器 'loggers': { 'full_logger': { 'handlers': ['console', 'info_file', 'error_file'], 'level': 'INFO' }, 'only_console_logger': { 'handlers': ['console'], 'level': 'INFO' }, 'only_file_logger': { 'handlers': ['info_file', 'error_file'] } } } logging.config.dictConfig(LOG_CONFIG_DICT)

下面我們就基於config第一次配置時整個logging的工作原理。結合代碼進行分析logging的工作之路。

我的python版本是Python3.6

從logging.config開始

dictConfigClass = DictConfigurator def dictConfig(config): """Configure logging using a dictionary.""" dictConfigClass(config).configure()

實質上是實例化logging.config.DictConfigurator類的對象，然後執行其configure方法。

class DictConfigurator(BaseConfigurator): """ Configure logging using a dictionary-like object to describe the configuration. """ def configure(self): """Do the configuration.""" config = self.config if 'version' not in config: raise ValueError("dictionary doesn't specify a version") if config['version'] != 1: raise ValueError("Unsupported version: %s" % config['version']) incremental = config.pop('incremental', False) EMPTY_DICT = {} logging._acquireLock() try: if incremental: handlers = config.get('handlers', EMPTY_DICT) for name in handlers: if name not in logging._handlers: raise ValueError('No handler found with ' 'name %r' % name) else: try: handler = logging._handlers[name] handler_config = handlers[name] level = handler_config.get('level', None) if level: handler.setLevel(logging._checkLevel(level)) except Exception as e: raise ValueError('Unable to configure handler ' '%r: %s' % (name, e)) loggers = config.get('loggers', EMPTY_DICT) for name in loggers: try: self.configure_logger(name, loggers[name], True) except Exception as e: raise ValueError('Unable to configure logger ' '%r: %s' % (name, e)) root = config.get('root', None) if root: try: self.configure_root(root, True) except Exception as e: raise ValueError('Unable to configure root ' 'logger: %s' % e) else: disable_existing = config.pop('disable_existing_loggers', True) logging._handlers.clear() del logging._handlerList[:] # Do formatters first - they don't refer to anything else formatters = config.get('formatters', EMPTY_DICT) for name in formatters: try: formatters[name] = self.configure_formatter( formatters[name]) except Exception as e: raise ValueError('Unable to configure ' 'formatter %r: %s' % (name, e)) # Next, do filters - they don't refer to anything else, either filters = config.get('filters', EMPTY_DICT) for name in filters: try: filters[name] = self.configure_filter(filters[name]) except Exception as e: raise ValueError('Unable to configure ' 'filter %r: %s' % (name, e)) handlers = config.get('handlers', EMPTY_DICT) deferred = [] for name in sorted(handlers): try: handler = self.configure_handler(handlers[name]) handler.name = name handlers[name] = handler except Exception as e: if 'target not configured yet' in str(e): deferred.append(name) else: raise ValueError('Unable to configure handler ' '%r: %s' % (name, e)) # Now do any that were deferred for name in deferred: try: handler = self.configure_handler(handlers[name]) handler.name = name handlers[name] = handler except Exception as e: raise ValueError('Unable to configure handler ' '%r: %s' % (name, e)) root = logging.root existing = list(root.manager.loggerDict.keys()) existing.sort() child_loggers = [] #now set up the new ones... loggers = config.get('loggers', EMPTY_DICT) for name in loggers: if name in existing: i = existing.index(name) + 1 # look after name prefixed = name + "." pflen = len(prefixed) num_existing = len(existing) while i < num_existing: if existing[i][:pflen] == prefixed: child_loggers.append(existing[i]) i += 1 existing.remove(name) try: self.configure_logger(name, loggers[name]) except Exception as e: raise ValueError('Unable to configure logger ' '%r: %s' % (name, e)) _handle_existing_loggers(existing, child_loggers, disable_existing) # And finally, do the root logger root = config.get('root', None) if root: try: self.configure_root(root) except Exception as e: raise ValueError('Unable to configure root ' 'logger: %s' % e) finally: logging._releaseLock()

我們對**incremental**不進行設置，即使用全量的方式進行配置，對於所有的handler重置並處理

看下來基本上分為4步來走。

Formatter配置

源碼部分

def configure_formatter(self, config): """Configure a formatter from a dictionary.""" if '()' in config: factory = config['()'] # for use in exception handler try: result = self.configure_custom(config) except TypeError as te: if "'format'" not in str(te): raise config['fmt'] = config.pop('format') config['()'] = factory result = self.configure_custom(config) else: fmt = config.get('format', None) dfmt = config.get('datefmt', None) style = config.get('style', '%') cname = config.get('class', None) if not cname: c = logging.Formatter else: c = _resolve(cname) result = c(fmt, dfmt, style) return result

可以使用自己的Formatter，默認使用logging.Formatter

Filter配置

def configure_filter(self, config): """Configure a filter from a dictionary.""" if '()' in config: result = self.configure_custom(config) else: name = config.get('name', '') result = logging.Filter(name) return result

設置filter,可以參考我上面配置的字典，可以自己定義Filter也可以使用系統內置Filter

Handler配置

def configure_handler(self, config): """Configure a handler from a dictionary.""" config_copy = dict(config) # for restoring in case of error formatter = config.pop('formatter', None) if formatter: try: formatter = self.config['formatters'][formatter] except Exception as e: raise ValueError('Unable to set formatter ' '%r: %s' % (formatter, e)) level = config.pop('level', None) filters = config.pop('filters', None) if '()' in config: c = config.pop('()') if not callable(c): c = self.resolve(c) factory = c else: cname = config.pop('class') klass = self.resolve(cname) #Special case for handler which refers to another handler if issubclass(klass, logging.handlers.MemoryHandler) and\ 'target' in config: try: th = self.config['handlers'][config['target']] if not isinstance(th, logging.Handler): config.update(config_copy) # restore for deferred cfg raise TypeError('target not configured yet') config['target'] = th except Exception as e: raise ValueError('Unable to set target handler ' '%r: %s' % (config['target'], e)) elif issubclass(klass, logging.handlers.SMTPHandler) and\ 'mailhost' in config: config['mailhost'] = self.as_tuple(config['mailhost']) elif issubclass(klass, logging.handlers.SysLogHandler) and\ 'address' in config: config['address'] = self.as_tuple(config['address']) factory = klass props = config.pop('.', None) kwargs = dict([(k, config[k]) for k in config if valid_ident(k)]) try: result = factory(**kwargs) except TypeError as te: if "'stream'" not in str(te): raise kwargs['strm'] = kwargs.pop('stream') result = factory(**kwargs) if formatter: result.setFormatter(formatter) if level is not None: result.setLevel(logging._checkLevel(level)) if filters: self.add_filters(result, filters) if props: for name, value in props.items(): setattr(result, name, value) return result

可以看到，前面順序先配置formatter和filter是因為這裡需要引用的到

值得注意的是，我們初始化時傳遞的一個字典，在整個配置過程中，字典裡面的值會隨著我們每次的配置變化而變化，所以我們在每個元素配置之後，在使用上一個字典元素時，就是配置完成之後的元素，為了方便理解，將配置filter之前和配置filter之後，config中的filter變化列出來

config == > before filters {'info_filter': {'()': }, 'error_filter': {'()': }} config == > after filters {'info_filter': , 'error_filter': }

配置前是我們配置文件中的內容，配置完成之後filter已經是一組對象了，所以在配置handler時我們就可以直接使用對象add_filter了。

Logger配置

def common_logger_config(self, logger, config, incremental=False): level = config.get('level', None) if level is not None: logger.setLevel(logging._checkLevel(level)) if not incremental: #Remove any existing handlers for h in logger.handlers[:]: logger.removeHandler(h) handlers = config.get('handlers', None) if handlers: self.add_handlers(logger, handlers) filters = config.get('filters', None) if filters: self.add_filters(logger, filters) def configure_logger(self, name, config, incremental=False): """Configure a non-root logger from a dictionary.""" logger = logging.getLogger(name) self.common_logger_config(logger, config, incremental) propagate = config.get('propagate', None) if propagate is not None: logger.propagate = propagate

這就比較容易理解了，將我們上面配置過的filters和handlers添加到我們的logger中。

這裡需要注意的一點是logger = logging.getLogger(name)，看下logger.getLogger源碼

root = RootLogger(WARNING) Logger.root = root Logger.manager = Manager(Logger.root) def getLogger(name=None): if name: return Logger.manager.getLogger(name) else: return root class Manager(object): def __init__(self, rootnode): self.root = rootnode self.disable = 0 self.emittedNoHandlerWarning = False self.loggerDict = {} self.loggerClass = None self.logRecordFactory = None def getLogger(self, name): rv = None if not isinstance(name, str): raise TypeError('A logger name must be a string') _acquireLock() try: if name in self.loggerDict: rv = self.loggerDict[name] if isinstance(rv, PlaceHolder): ph = rv rv = (self.loggerClass or _loggerClass)(name) rv.manager = self self.loggerDict[name] = rv self._fixupChildren(ph, rv) self._fixupParents(rv) else: rv = (self.loggerClass or _loggerClass)(name) rv.manager = self self.loggerDict[name] = rv self._fixupParents(rv) finally: _releaseLock() return rv

可以看到，logging使用Mangaer進行logger的單例管理

截止到這裡，基本上我們使用前的準備，logging都替我們準備好了，下面就是我們的使用

獲取logger並記錄日誌

class Logger(Filterer): def info(self, msg, *args, **kwargs): if self.isEnabledFor(INFO): self._log(INFO, msg, args, **kwargs) def error(self, msg, *args, **kwargs): if self.isEnabledFor(ERROR): self._log(ERROR, msg, args, **kwargs) def isEnabledFor(self, level): if self.manager.disable >= level: return False return level >= self.getEffectiveLevel() def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False): sinfo = None if _srcfile: try: fn, lno, func, sinfo = self.findCaller(stack_info) except ValueError: # pragma: no cover fn, lno, func = "(unknown file)", 0, "(unknown function)" else: # pragma: no cover fn, lno, func = "(unknown file)", 0, "(unknown function)" if exc_info: if isinstance(exc_info, BaseException): exc_info = (type(exc_info), exc_info, exc_info.__traceback__) elif not isinstance(exc_info, tuple): exc_info = sys.exc_info() record = self.makeRecord(self.name, level, fn, lno, msg, args, exc_info, func, extra, sinfo) self.handle(record) def handle(self, record): if (not self.disabled) and self.filter(record): self.callHandlers(record) def callHandlers(self, record): c = self found = 0 while c: for hdlr in c.handlers: found = found + 1 if record.levelno >= hdlr.level: hdlr.handle(record) if not c.propagate: c = None #break out else: c = c.parent if (found == 0): if lastResort: if record.levelno >= lastResort.level: lastResort.handle(record) elif raiseExceptions and not self.manager.emittedNoHandlerWarning: sys.stderr.write("No handlers could be found for logger" " "%s"\n" % self.name) self.manager.emittedNoHandlerWarning = True class StreamHandler(Handler): terminator = '\n' def __init__(self, stream=None): """ Initialize the handler. If stream is not specified, sys.stderr is used. """ Handler.__init__(self) if stream is None: stream = sys.stderr self.stream = stream def flush(self): self.acquire() try: if self.stream and hasattr(self.stream, "flush"): self.stream.flush() finally: self.release() def emit(self, record): try: msg = self.format(record) stream = self.stream stream.write(msg) stream.write(self.terminator) self.flush() except Exception: self.handleError(record) def __repr__(self): level = getLevelName(self.level) name = getattr(self.stream, 'name', '') if name: name += ' ' return '' % (self.__class__.__name__, name, level)

根據代碼可以看到符合我們流程圖看到的流程，我們細化一下就是

1. 查看記錄日誌是否滿足過濾器，然後準備logrecord中的信息並生成logrecord

2. 將logrecord交給所有handlers處理

3. 在handler中確定是否滿足handler過濾器，滿足的話按照配置的Formatter生成字符串

4. 將內容寫入到具體的文件或者輸出流

logger = logging.getLogger("full_logger") logger.info("111") # 根據我們的配置 這裡會輸出到流中，以及記錄到test_log_info.log文件中 logger.error("error") # 根據我們的配置 這裡會輸出到流中，以及記錄到test_log_error.info文件中