Python logging 제대로 하는 방법
Python 로깅에 대한 정리와 Best Practice 예시
Python logging¶
Python 로깅의 workflow는 아래와 같다.
stateDiagram-v2
logging_flow : Logging Flow
state logging_flow {
logging_call : logging call
[*] --> logging_call
state enabled_level_logger <<choice>>
logging_call --> enabled_level_logger : enabled level?
create_LogRecord : create LogRecord
enabled_level_logger --> create_LogRecord : Yes
enabled_level_logger --> STOP : No
state rejected_by_filter <<choice>>
create_LogRecord --> rejected_by_filter : rejected by filter?
pass_to_handler : pass LogRecord to Handler
rejected_by_filter --> pass_to_handler : No
rejected_by_filter --> STOP : Yes
state is_propagate <<choice>>
pass_to_handler --> is_propagate : is propagate True?
is_propagate --> STOP : No
state parent_exist <<choice>>
is_propagate --> parent_exist : is there a parent logger?
set_current_to_parent : set current logger to parent
parent_exist --> STOP : No
parent_exist --> set_current_to_parent : Yes
set_current_to_parent --> pass_to_handler
state handler_exist <<choice>>
pass_to_handler --> handler_exist : does handler exits in hierarchy?
use_lastResort_handler : use lastResort handler
handler_exist --> use_lastResort_handler : No
logging_handler : logging handler
use_lastResort_handler --> logging_handler
handler_exist --> logging_handler : Yes
}
handler_flow : Handler Flow
state handler_flow {
state enabled_level_handler <<choice>>
stop2: STOP
enabled_level_handler --> stop2 : No
state rejected_by_filter_handler <<choice>>
logging_handler --> enabled_level_handler : enabled level?
enabled_level_handler --> rejected_by_filter_handler : rejected by filter?
rejected_by_filter_handler --> stop2 : Yes
EMIT : Emit (includes formatting)
rejected_by_filter_handler --> EMIT : No
EMIT --> [*]
}Handler¶
Python에서 기본 제공하는 로깅 모듈을 사용하면 시스템 로그를 아주 간편하게 남길 수 있는데, Handler를 사용하여 로깅을 위한 여러가지 설정을 쉽게 관리할 수 있다.
TimedRotatingFileHandler¶
Python이 기본 제공하는 다양한 Log Handler 중에 TimedRotatingFileHandler를 사용하면 일정한 시간 간격으로 로그가 새로운 파일로 나뉘어서 생성되도록 관리할 수 있다.
TimedRotatingFileHandler Handler의 경우 아래 두 설정이 중요하다.
when: time rotate의 기준 시점backupCount: 로그를 남길 파일 개수, 로그 파일이 해당 설정의 수보다 많을 경우 자동 삭제
when 값 |
interval 유형 | atTime 입력 시 |
|---|---|---|
| S | 초 | 영향 없음 |
| M | 분 | 영향 없음 |
| H | 시간 | 영향 없음 |
| D | 일 | 영향 없음 |
| W0 - W6 | 요일 (0=월요일) | 최초 롤오버 시간 계산에 사용 |
| midnight | atTime 미입력 시 자정에 롤오버1 |
최초 롤오버 시간 계산에 사용 |
Tip
file_handler.suffix = "%Y%m%d.log"와 같이 suffix 속성을 설정할 경우 롤오버 시 생성되는 파일의 파일명 규칙을 수정할 수 있지만, 이 경우 backupCount 속성이 제대로 작동하지 않게 된다.
QueueHandler¶
로깅을 위한 IO는 프로그램의 속도를 저하시키는 원인이 되기 때문에 웹 서비스와 같이 빠른 속도가 중요한 프로그램을 만들 때에는 로깅을 메인 쓰레드와 분리된 새로운 쓰레드에서 수행해 줄 필요가 있는데, 이를 위해서 QueueHandler와 QueueListener를 사용한다.
Along with the QueueListener class, QueueHandler can be used to let handlers do their work on a separate thread from the one which does the logging.
LogRecord¶
로그가 출력될 때 실제로는 LogRecord 클래스의 인스턴스가 생성되고, 해당 인스턴스에 각종 정보들이 담긴 후 사용자가 설정한 내용들만 추려서 출력된다.
LogRecord 클래스의 요소들은 공식 문서에서 확인할 수 있는데, 주로 사용할만한 속성들은 아래와 같다.
| Attribute name | Format | Description |
|---|---|---|
| asctime | %(asctime)s | 로그가 생성된 시간 |
| filename | %(filename)s | 로그를 발생시킨 파일의 이름 |
| levelname | %(levelname)s | 로그 레벨 이름 |
| lineno | %(lineno)d | 소스코드에서 로그를 발생시킨 라인 넘버 |
| message | %(message)s | 로그 메세지 |
| module | %(module)s | 로그를 발생시킨 모듈 이름 |
| name | %(name)s | 로거의 이름 |
| process | %(process)d | 프로세스 ID(가능할 경우에만) |
| thread | %(thread)d | 쓰레드 ID(가능할 경우에만) |
Python Log의 asctime
엄밀히 말하면 asctime은 LogRecord 객체의 요소는 아니다. LogRecord 객체는 time.time()2으로 생성시간을 저장한 후, Formatter가 생성시간을 time.strftime을 사용해서 입력받은 포맷대로 생성해준다.
class LogRecord(object):
...
def __init__(self, name, level, pathname, lineno,
msg, args, exc_info, func=None, sinfo=None, **kwargs):
...
ct = time.time()
...
self.created = ct
...
class Formatter(object):
...
converter = time.localtime
def __init__(self, fmt=None, datefmt=None, style='%', validate=True, *,
defaults=None):
...
def formatTime(self, record, datefmt=None):
...
ct = self.converter(record.created)
if datefmt:
s = time.strftime(datefmt, ct)
else:
s = time.strftime(self.default_time_format, ct)
if self.default_msec_format:
s = self.default_msec_format % (s, record.msecs)
return s
...
Filter¶
필터는 로그 레벨에 따라 로그의 출력을 걸러주기 위해 사용하는 객체로, LogRecord 인스턴스를 필터링하는 규칙을 직접 만들 수 있다.
logging 모듈의 기본 필터는 로그 레벨로 필터링하고, 그 중에서도 지정된 레벨보다 상위 레벨 전체의 로그를 필터링한다.
로그 필터를 직접 만들어 주입해주면 특정 정보의 로그만 필터링 해줄 수 있다.
src/log/filter.py
from logging import Filter, LogRecord
class MyFilter(Filter): # (1)!
def __init__(self, levels: list[int]):
self.__level = levels
def filter(self, logRecord: LogRecord):
return logRecord.levelno in self.__level
- 아래 note에 작성했듯이
class MyFilter(object)로 만들어도 전혀 문제 없다.
Filterer
참고로 Python 공식 문서 Logging facility for Python에서는 필터는 굳이 표준 라이브러리의 클래스를 상속해서 만들 필요 없이, 단순히 filter 메서드를 가진 객체는 아무 것이나 사용해도 된다고 한다.
You don’t actually need to subclass
Filter: you can pass any instance which has afiltermethod with the same semantics.
실제로 logging.Logger, logging.Handler 클래스가 상속하고 있는 logging.Filterer 클래스를 살펴보면 아래와 같이 addFilter 메서드는 filters 리스트에 필터 객체를 추가해주기만 하며, filter 메서드는 filters 리스트에 속한 필터들의 filter 메서드를 호출하는 역할만 한다.
class Filterer(object):
...
def __init__(self):
...
self.filters = []
def addFilter(self, filter):
...
if not (filter in self.filters):
self.filters.append(filter)
...
def filter(self, record):
...
rv = True
for f in self.filters:
if hasattr(f, 'filter'):
result = f.filter(record)
else:
result = f(record) # assume callable - will raise if not
if not result:
rv = False
break
return rv
...
Formatter¶
Formatter 클래스를 직접 만들어 사용한다면, 로그를 jsonl과 같이 특이한 형태의 파일로도 출력 만들 수 있다.
src/log/formatter.py
import datetime as dt
import json
from logging import Formatter, LogRecord
KST = dt.timezone(offset=dt.timedelta(hours=9))
class JsonFormatter(Formatter):
def __init__(
self,
*,
fmt_keys: dict[str, str] | None = None,
) -> None:
super().__init__()
self.fmt_keys = fmt_keys if fmt_keys is not None else {}
# override
def format(self, record: LogRecord) -> str:
message = self._prepare_log(record=record)
return json.dumps(obj=message, default=str)
def _prepare_log(self, record: LogRecord):
always_fields = {
"message": record.getMessage(),
"timestamp": (
dt.datetime.fromtimestamp(record.created, tz=KST) # (1)!
).isoformat(),
}
if record.exc_info is not None:
always_fields["exc_info"] = self.formatException(record.exc_info)
if record.stack_info is not None:
always_fields["stack_info"] = self.formatStack(record.stack_info)
message = {
key: (
msg_val
if (msg_val := always_fields.pop(val, None)) is not None
else getattr(record, val)
)
for key, val in self.fmt_keys.items()
}
message.update(always_fields)
return message
- UTC 기준으로 로그를 생성하고 싶다면
dt.datetime.fromtimestamp(record.created, tz=dt.timezone.utc)으로 만들면 된다.
src/log/formatter.py
import datetime as dt
import json
from logging import Formatter, LogRecord
from typing import override
KST = dt.timezone(dt.timedelta(hours=9))
class JsonFormatter(Formatter):
def __init__(
self,
*,
fmt_keys: dict[str, str] | None = None,
) -> None:
super().__init__()
self.fmt_keys = fmt_keys if fmt_keys is not None else {}
@override
def format(self, record: LogRecord) -> str:
message = self._prepare_log(record=record)
return json.dumps(message, default=str)
def _prepare_log(self, record: LogRecord):
always_fields = {
"message": record.getMessage(),
"timestamp": (
dt.datetime.fromtimestamp(timestamp=record.created, tz=KST) # (1)!
).isoformat(),
}
if record.exc_info is not None:
always_fields["exc_info"] = self.formatException(record.exc_info)
if record.stack_info is not None:
always_fields["stack_info"] = self.formatStack(record.stack_info)
message = {
key: (
msg_val
if (msg_val := always_fields.pop(val, None)) is not None
else getattr(record, val)
)
for key, val in self.fmt_keys.items()
}
message.update(always_fields)
return message
- UTC 기준으로 로그를 생성하고 싶다면
dt.datetime.fromtimestamp(timestamp=record.created, tz=dt.timezone.utc)으로 만들면 된다.
Best Practice¶
Code를 통한 로그 설정¶
src/log/__init__.py
import logging
import queue
from logging import Formatter, StreamHandler
from logging.handlers import QueueHandler, QueueListener, TimedRotatingFileHandler
from pathlib import Path
from src.log import filter, formatter
# set log directory
log_dir = Path("logs")
try:
log_dir.mkdir()
except FileExistsError:
...
# create Logger instance
logger = logging.getLogger(name="logger")
logger.setLevel(level=logging.DEBUG)
# set log format
simple_formatter = Formatter(
fmt="%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
)
detailed_formatter = Formatter(
fmt="%(asctime)s - %(levelname)s - [%(module)s:%(lineno)d] %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
)
# StreamHandler
stream_handler = StreamHandler()
stream_handler.setFormatter(fmt=simple_formatter)
# TimedRotatingFileHandler
file_handler = TimedRotatingFileHandler(
filename=log_dir / "app.log",
when="midnight", # rotate at every midnight
backupCount=3, # define number of log files, 0 to save all log files
encoding="utf-8",
)
file_handler.setFormatter(fmt=detailed_formatter)
# JsonlHandler
json_handler = TimedRotatingFileHandler(
filename=log_dir / "app.log.jsonl",
when="midnight",
backupCount=3,
encoding="utf-8",
)
fmt_keys = {
"level": "levelname",
"message": "message",
"logger": "name",
"module": "module",
"function": "funcName",
"line": "lineno",
}
json_handler.setFormatter(fmt=formatter.JsonFormatter(fmt_keys=fmt_keys))
# DebugHandler
debug_handler = TimedRotatingFileHandler(
filename=log_dir / "debug.log",
when="midnight",
backupCount=3,
encoding="utf-8",
)
debug_handler.setFormatter(fmt=detailed_formatter)
debug_handler.addFilter(
filter=filter.MyFilter(
levels=[
logging.DEBUG,
logging.ERROR,
logging.CRITICAL,
]
)
)
# QueueHandler
log_queue = queue.Queue() # (1)!
queue_handler = QueueHandler(queue=log_queue)
logger.addHandler(hdlr=queue_handler)
# QueueListener
log_listener = QueueListener(
log_queue,
stream_handler,
file_handler,
json_handler,
debug_handler,
)
- 멀티프로세싱 환경에서 QueueHandler를 사용할 경우 multiprocessing.Queue를 사용해야 한다.
실제 애플리케이션에서의 로그 활용
main.py
from src.log import log_listener, logger
def main():
log_listener.start()
logger.debug("debug message")
logger.info("info message")
logger.warning("warn message")
logger.error("error message")
logger.critical("critical message")
try:
raise Exception
except Exception as e:
logger.exception(e) # logging error traceback
log_listener.stop()
if __name__ == "__main__":
main()
dictConfig를 통한 로그 설정¶
로그 설정을 위한 실제 config 파일로 로그 설정을 JSON이나 YAML 등 외부 파일로 다루면 추후 사용자가 로그 기능을 쉽게 수정할 수 있다는 장점이 있다. 사용자를 믿을 수 있는지는 모르겠지만..
log_config.json
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
"simple": {
"format": "%(asctime)s - %(levelname)s - %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%S%z"
},
"detailed": {
"format": "%(asctime)s - %(levelname)s - [%(module)s:%(lineno)d] %(message)s",
"datefmt": "%Y-%m-%dT%H:%M:%S%z"
},
"json": {
"()": "src.log.formatter.JsonFormatter", // (1)!
"fmt_keys": {
"level": "levelname",
"message": "message",
"logger": "name",
"module": "module",
"function": "funcName",
"line": "lineno"
}
}
},
"handlers": {
"stream": {
"class": "logging.StreamHandler",
"level": "DEBUG",
"formatter": "simple",
"stream": "ext://sys.stderr"
},
"rotating_file": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "DEBUG",
"formatter": "detailed",
"filename": "logs/app.log",
"when": "midnight",
"backupCount": 3,
"encoding": "utf-8"
},
"rotating_json": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "DEBUG",
"formatter": "json",
"filename": "logs/app.log.jsonl",
"when": "midnight",
"backupCount": 3,
"encoding": "utf-8"
},
"debug_handler": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "DEBUG",
"formatter": "detailed",
"filename": "logs/debug.log",
"when": "midnight",
"backupCount": 3,
"encoding": "utf-8"
},
"queue_handler": {
"class": "logging.handlers.QueueHandler",
"handlers": [
"stream",
"rotating_file",
"rotating_json",
"debug_handler"
],
"respect_handler_level": true
}
},
"loggers": {
"root": {
"level": "DEBUG",
"handlers": ["queue_handler"]
}
}
}
- 사용자 class를 사용할 때는 키 값을
()으로 설정하지 않으면 key 들이 하드코딩으로 주입된다.
Warning
3.11 버전까지는 QueueHandler와 QueueListener의 설정이 dictConfig를 통해 쉽게 주입하기 어렵다는 문제가 있다.
Tip
3.12 버전부터는 QueueHandler에 로그 Queue를 자동으로 주입받고, QueueListener 역시 dictConfig를 통해 handlers를 주입받을 수 있도록 하는 내부적인 변경이 생겨 dictConfig를 통해 QueueHandler도 쉽게 다룰 수 있게 되었다.
log_config.json에서 입력받은 로그 설정을 애플리케이션에 주입하기 위한 코드
src/log/__init__.py
import atexit
import json
import logging
import logging.config
from pathlib import Path
from src.log import filter
logger = logging.getLogger(name="logger")
def set_logger():
log_config_file = Path(r"config\log_config.json")
with open(file=log_config_file, encoding="utf-8") as f:
log_config = json.load(fp=f)
logging.config.dictConfig(config=log_config)
queue_handler = logging.getHandlerByName(name="queue_handler")
if queue_handler is not None:
queue_handler.listener.start()
atexit.register(func=queue_handler.listener.stop)
debug_handler = logging.getHandlerByName(name="debug_handler")
if debug_handler is not None:
debug_handler.addFilter(
filter=filter.MyFilter(
levels=[
logging.DEBUG,
logging.ERROR,
logging.CRITICAL,
]
)
)
실제 애플리케이션에서의 로그 활용
main.py
from src.log import log_listener, logger
def main():
set_logger()
logger.debug("debug message")
logger.info("info message")
logger.warning("warn message")
logger.error("error message")
logger.critical("critical message")
try:
raise Exception
except Exception as e:
logger.exception(e) # log for error catch
if __name__ == "__main__":
main()
로그 출력 결과¶
두 가지 설정 방식은 아래와 같이 로그들을 출력해준다.
standard out
2024-01-28T16:22:31+0900 - DEBUG - debug message
2024-01-28T16:22:31+0900 - INFO - info message
2024-01-28T16:22:31+0900 - WARNING - warn message
2024-01-28T16:22:31+0900 - ERROR - error message
2024-01-28T16:22:31+0900 - CRITICAL - critical message
2024-01-28T16:22:31+0900 - ERROR -
Traceback (most recent call last):
File "C:\projects\python312\main.py", line 14, in main
raise Exception
Exception
app.log
2024-01-28T16:22:31+0900 - DEBUG - [main:7] debug message
2024-01-28T16:22:31+0900 - INFO - [main:8] info message
2024-01-28T16:22:31+0900 - WARNING - [main:9] warn message
2024-01-28T16:22:31+0900 - ERROR - [main:10] error message
2024-01-28T16:22:31+0900 - CRITICAL - [main:11] critical message
2024-01-28T16:22:31+0900 - ERROR - [main:16]
Traceback (most recent call last):
File "C:\projects\python312\main.py", line 14, in main
raise Exception
Exception
app.log.jsonl
{"level": "DEBUG", "message": "debug message", "logger": "logger", "module": "main", "function": "main", "line": 7, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
{"level": "INFO", "message": "info message", "logger": "logger", "module": "main", "function": "main", "line": 8, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
{"level": "WARNING", "message": "warn message", "logger": "logger", "module": "main", "function": "main", "line": 9, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
{"level": "ERROR", "message": "error message", "logger": "logger", "module": "main", "function": "main", "line": 10, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
{"level": "CRITICAL", "message": "critical message", "logger": "logger", "module": "main", "function": "main", "line": 11, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
{"level": "ERROR", "message": "\nTraceback (most recent call last):\n File \"C:\\projects\\python312\\main.py\", line 14, in main\n raise Exception\nException", "logger": "logger", "module": "main", "function": "main", "line": 16, "timestamp": "2024-01-28T16:22:31.069276+00:00"}
debug.log
2024-01-28T16:22:31+0900 - DEBUG - [main:7] debug message
2024-01-28T16:22:31+0900 - ERROR - [main:8] error message
2024-01-28T16:22:31+0900 - CRITICAL - [main:9] critical message
2024-01-28T16:22:31+0900 - ERROR - [main:14]
Traceback (most recent call last):
File "C:\projects\python311\main.py", line 12, in main
raise Exception
Exception
로거 프리셋¶
디버그용 로거 설정¶
프로그램 디버깅만을 위한 디버그 전용 로거 설정 방법
src/log/__init__.py
import logging
import queue
from logging.handlers import QueueHandler, QueueListener, TimedRotatingFileHandler
from pathlib import Path
from src.log import filter
# set log directory
log_dir = Path("logs")
try:
log_dir.mkdir()
except FileExistsError:
...
# create Logger instance
logger = logging.getLogger("logger")
logger.setLevel(logging.DEBUG)
# set log format
detailed_formatter = logging.Formatter(
fmt="%(asctime)s - %(levelname)s - [%(module)s:%(lineno)d] %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
)
# DebugHandler
debug_handler = TimedRotatingFileHandler(
filename=log_dir / "debug.log",
when="midnight",
backupCount=3,
encoding="utf-8",
)
debug_handler.setFormatter(detailed_formatter)
debug_handler.addFilter(
filter.MyFilter(
levels=[
logging.DEBUG,
logging.ERROR,
logging.CRITICAL,
]
)
)
# QueueHandler
log_queue = queue.Queue()
queue_handler = QueueHandler(log_queue)
logger.addHandler(queue_handler)
# QueueListener
log_listener = QueueListener(log_queue, debug_handler)
간단한 로거 설정¶
src/log/__init__.py
import logging
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path
# set log directory
log_dir = Path("logs")
try:
log_dir.mkdir()
except FileExistsError:
...
# create Logger instance
logger = logging.getLogger("logger")
logger.setLevel(logging.DEBUG)
# set log format
formatter = logging.Formatter(
fmt="%(asctime)s - %(levelname)s - [%(module)s:%(lineno)d] %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S"
)
# TimedRotatingFileHandler
file_handler = TimedRotatingFileHandler(
filename=log_dir / "debug_log.log",
when="midnight", # rotate at every midnight
backupCount=3, # define number of log files, 0 to save all log files
encoding="utf-8",
)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)