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
用於記錄日誌到具體的文件或者輸出流或其他的管道。
- 查看記錄日誌是否滿足過濾器
- 滿足過濾器，按照設置的Formatter生成字符串
- 將內容寫入到具體的文件或者輸出流
不同的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 emittedLogRecord
我們每一次的 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
- 配置Filter
- 配置Handler
- 配置Logger
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文件中分享到:
