Logging trong Python: cấu hình cho production

1. Tổng quan

print() đủ dùng khi bạn viết script nhỏ. Lên production thì khác: bạn cần lọc log theo level, gom log vào hệ thống tập trung, lần theo một request qua nhiều dòng log, và việc ghi log không được làm chậm ứng dụng.

Bài này hướng dẫn cấu hình module logging chuẩn của Python cho production. Đọc xong, bạn sẽ dựng được một cấu hình gồm logger theo module, dictConfig, log dạng JSON, request_id gắn tự động, ghi log bất đồng bộ bằng QueueHandler và test log bằng pytest. Mọi ví dụ dùng Python 3.12 và chỉ cần thư viện chuẩn. Riêng phần test cần thêm pytest.

2. Vì sao print và basicConfig không đủ

Xem một service xử lý đơn hàng viết theo kiểu “nhanh cho xong”:

import logging

logging.basicConfig(level=logging.INFO)

def place_order(order_id: str, amount: float) -> None:
    print(f"Placing order {order_id}")
    logging.info(f"Order {order_id} amount={amount}")

place_order("ORD-1001", 250_000)

Placing order ORD-1001
INFO:root:Order ORD-1001 amount=250000

Đoạn code này có ba vấn đề:

• print không có level, không có timestamp và không tắt được qua cấu hình.
• logging.info(...) ghi vào root logger. Bạn không biết dòng log đến từ module nào và không chỉnh level riêng cho từng phần được.
• basicConfig chỉ có tác dụng ở lần gọi đầu tiên, nếu root logger chưa có handler. Thư viện nào gọi nó trước bạn thì cấu hình của bạn bị bỏ qua mà không có cảnh báo nào.

3. Logger theo module với getLogger(__name__)

Quy tắc đầu tiên: mỗi module tạo một logger riêng, đặt tên theo __name__.

# shop/orders.py
import logging

logger = logging.getLogger(__name__)  # tên logger: "shop.orders"

class OrderService:
    def place_order(self, order_id: str, quantity: int) -> None:
        logger.info("Placing order %s with quantity %d", order_id, quantity)
        if quantity > 50:
            logger.warning("Large order %s: quantity=%d", order_id, quantity)

Tên logger tạo thành một cây phân cấp theo dấu chấm. shop.orders là con của shop, và shop là con của root. Một record đi từ logger con lên logger cha (gọi là propagation). Vì vậy bạn chỉ cần gắn handler ở root hoặc ở shop, rồi chỉnh level riêng cho từng nhánh. Ví dụ, bạn có thể đặt shop.payments ở DEBUG và giữ phần còn lại ở INFO.

Bạn cũng để ý cách truyền tham số: dùng %s kèm đối số thay vì f-string. Khi level bị tắt, chuỗi sẽ không được format, nên bạn không tốn chi phí cho những log không bao giờ được in. Ngoài ra, công cụ gom log có thể nhóm các dòng theo cùng message template.

Lưu ý: Trong thư viện (package bạn phát hành cho người khác dùng), đừng cấu hình handler. Chỉ gọi getLogger(__name__) và để ứng dụng quyết định. Nếu muốn, thêm logging.getLogger("your_lib").addHandler(logging.NullHandler()).

4. Cấu hình tập trung bằng dictConfig

Ở production, bạn nên gom toàn bộ cấu hình vào một chỗ, gọi một lần duy nhất khi ứng dụng khởi động. logging.config.dictConfig phù hợp cho việc này.

# shop/logging_config.py
import logging.config
import os

def setup_logging() -> None:
    level = os.getenv("LOG_LEVEL", "INFO")
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "plain": {
                "format": "%(asctime)s %(levelname)-8s %(name)s - %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "formatter": "plain",
            },
        },
        "loggers": {
            "shop": {"level": level},
            "urllib3": {"level": "WARNING"},
        },
        "root": {"level": "WARNING", "handlers": ["console"]},
    })

Các điểm chính trong cấu hình:

• "disable_existing_loggers": False là bắt buộc trong thực tế. Giá trị mặc định True sẽ vô hiệu hóa mọi logger đã tạo trước khi dictConfig chạy, kể cả logger ở đầu các module bạn đã import. Kết quả là log biến mất mà không có lỗi nào.
• Handler chỉ gắn ở root. Logger shop chỉ đặt level, còn record sẽ propagate lên root để được in ra.
• Root để ở WARNING nhằm chặn log INFO ồn ào từ thư viện bên thứ ba. Code của bạn trong nhánh shop vẫn log ở INFO.
• Level đọc từ biến môi trường LOG_LEVEL. Bạn bật DEBUG khi điều tra sự cố mà không cần sửa code.

Gọi hàm ở entry point:

# main.py
from shop.logging_config import setup_logging
from shop.orders import OrderService

setup_logging()
OrderService().place_order("ORD-1001", quantity=80)

2026-10-18 09:15:02,114 INFO     shop.orders - Placing order ORD-1001 with quantity 80
2026-10-18 09:15:02,114 WARNING  shop.orders - Large order ORD-1001: quantity=80

Ghi ra stdout là lựa chọn đúng khi chạy trong container. Docker hoặc Kubernetes sẽ thu log từ stream này, nên bạn không cần tự quản lý file và việc xoay vòng file.

5. Log có cấu trúc dạng JSON

Hệ thống gom log như Loki, Elasticsearch hay CloudWatch tìm kiếm trên JSON tốt hơn nhiều so với text tự do. Bạn có thể tự viết formatter mà không cần thư viện ngoài:

# shop/json_formatter.py
import json
import logging
from datetime import datetime, timezone

_RESERVED = set(vars(logging.LogRecord("", 0, "", 0, "", None, None))) | {"message", "asctime"}

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, value in record.__dict__.items():
            if key not in _RESERVED and not key.startswith("_"):
                payload[key] = value
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)

Cách hoạt động:

• _RESERVED chứa các thuộc tính mặc định của một LogRecord, ví dụ pathname, lineno, args.
• Mọi thuộc tính khác được coi là field tùy biến và đưa vào JSON. Các field này thường đến từ tham số extra= hoặc từ filter (xem mục 6).
• default=str giúp json.dumps không lỗi khi gặp kiểu như Decimal hay UUID.

Khai báo formatter trong dictConfig bằng key đặc biệt "()":

"formatters": {
    "json": {"()": "shop.json_formatter.JsonFormatter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
    },
},

Thêm field nghiệp vụ bằng extra:

logger.info("Payment captured", extra={"order_id": "ORD-1001", "amount": 250000})

{"ts": "2026-10-18T02:15:02.114+00:00", "level": "INFO", "logger": "shop.payments", "message": "Payment captured", "order_id": "ORD-1001", "amount": 250000}

Lưu ý: Key trong extra không được trùng tên thuộc tính có sẵn của LogRecord. Ví dụ, extra={"name": "..."} hay extra={"message": "..."} sẽ ném KeyError: "Attempt to overwrite 'name' in LogRecord".

6. Gắn request_id tự động bằng contextvars

Truyền extra={"request_id": ...} vào từng lời gọi log thì vừa dài dòng vừa dễ quên. Cách tốt hơn là lưu request_id vào một ContextVar, rồi để một Filter gắn nó vào mọi record.

# shop/request_context.py
import logging
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True  # không chặn record nào, chỉ bổ sung dữ liệu

ContextVar an toàn với cả thread lẫn asyncio. Mỗi task hoặc request giữ giá trị riêng, nên không có chuyện request này “mượn” ID của request khác.

Gắn filter vào handler, không gắn vào logger. Filter gắn ở logger chỉ chạy cho record tạo ra trực tiếp từ logger đó, chứ không chạy cho record propagate từ logger con lên.

"filters": {
    "request_id": {"()": "shop.request_context.RequestIdFilter"},
},
"handlers": {
    "console": {
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
        "formatter": "json",
        "filters": ["request_id"],
    },
},

Đặt giá trị ở đầu mỗi request, ví dụ trong middleware:

import uuid
from shop.request_context import request_id_var

def handle_request(order_id: str) -> None:
    token = request_id_var.set(uuid.uuid4().hex[:12])
    try:
        OrderService().place_order(order_id, quantity=3)
    finally:
        request_id_var.reset(token)

{"ts": "...", "level": "INFO", "logger": "shop.orders", "message": "Placing order ORD-1002 with quantity 3", "request_id": "9f2c1ab07d3e"}

Giờ bạn chỉ cần lọc theo request_id là thấy toàn bộ hành trình của một request.

7. Ghi log exception đúng cách

logger = logging.getLogger("shop.payments")

class PaymentService:
    def charge(self, order_id: str, amount: int) -> bool:
        try:
            raise ConnectionError("gateway timeout")
        except ConnectionError:
            logger.exception("Charge failed for order %s", order_id)
            return False

logger.exception(...) ghi ở level ERROR và tự đính kèm stack trace. Bạn chỉ gọi nó bên trong khối except. Nếu muốn ghi ở level khác mà vẫn có stack trace, dùng logger.warning("...", exc_info=True).

Hai lỗi hay gặp:

• logger.error(f"Charge failed: {e}") làm mất stack trace. Khi điều tra, bạn chỉ thấy đúng một dòng “gateway timeout” mà không biết nó xảy ra ở đâu.
• Vừa log vừa raise lại ở mọi tầng. Cùng một exception sẽ xuất hiện năm lần trong log. Hãy log ở tầng xử lý exception, tức là nơi quyết định không ném tiếp nữa.

8. Không để logging chặn ứng dụng: QueueHandler

StreamHandler ghi ra stdout thường đủ nhanh. Nhưng handler gửi qua mạng (syslog, HTTP, SMTP) hoặc ghi lên ổ đĩa chậm sẽ chặn chính thread đang xử lý request. Để tránh việc này, đưa record vào hàng đợi và để một thread nền ghi ra đích thật:

import atexit
import logging
import logging.handlers
import queue

def setup_async_logging(target: logging.Handler) -> None:
    log_queue: queue.Queue = queue.Queue(-1)
    queue_handler = logging.handlers.QueueHandler(log_queue)

    listener = logging.handlers.QueueListener(
        log_queue, target, respect_handler_level=True
    )
    listener.start()
    atexit.register(listener.stop)  # flush phần còn lại khi tắt

    root = logging.getLogger()
    root.handlers = [queue_handler]

Luồng xử lý như sau:

• Thread ứng dụng chỉ làm một việc là put record vào queue. Thao tác này gần như tức thời.
• QueueListener chạy trên thread riêng, lấy record ra và chuyển cho target, ví dụ một SysLogHandler.
• atexit.register(listener.stop) bảo đảm log cuối cùng được ghi xong trước khi process thoát.

Lưu ý: Filter như RequestIdFilter phải gắn vào QueueHandler, không gắn vào handler đích. Handler đích chạy trên thread của listener, mà ContextVar ở thread đó không mang giá trị của request gốc.

9. Test log với pytest caplog

Log cảnh báo nghiệp vụ, ví dụ đơn hàng số lượng lớn, là hành vi đáng được test. Thêm pytest vào dev dependency:

pip install pytest

# tests/test_orders.py
import logging
from shop.orders import OrderService

def test_large_order_logs_warning(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2001", quantity=80)

    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(warnings) == 1
    assert warnings[0].name == "shop.orders"
    assert warnings[0].getMessage() == "Large order ORD-2001: quantity=80"

def test_small_order_does_not_warn(caplog):
    with caplog.at_level(logging.INFO, logger="shop.orders"):
        OrderService().place_order("ORD-2002", quantity=2)

    assert all(r.levelno < logging.WARNING for r in caplog.records)

tests/test_orders.py ..                                  [100%]
2 passed in 0.03s

caplog.at_level tạm hạ level của logger shop.orders trong phạm vi khối with. caplog.records chứa các LogRecord gốc, nên bạn kiểm tra được level, tên logger và cả field extra nếu có.

Lưu ý: caplog gắn handler vào root logger. Nếu cấu hình của bạn đặt "propagate": False cho logger shop, test sẽ không bắt được record nào.

10. Lỗi hay gặp và khi nào không nên làm

• Cấu hình logging ở cấp module (gọi basicConfig hay dictConfig lúc import). Chỉ cấu hình ở entry point, một lần duy nhất.
• Handler bị gắn hai lần, ví dụ gắn ở cả shop và root trong khi propagate vẫn là True. Mỗi dòng log sẽ in hai lần. Bạn chỉ nên gắn handler ở một tầng.
• Ghi dữ liệu nhạy cảm: mật khẩu, token, số thẻ. Đừng log nguyên object request. Nếu bắt buộc, hãy viết một Filter để che (mask) các field này.
• Dùng RotatingFileHandler trong ứng dụng nhiều process (Gunicorn với nhiều worker). Các process cùng xoay vòng một file sẽ gây mất log. Hãy ghi ra stdout, hoặc để mỗi worker đẩy qua QueueHandler về một process ghi duy nhất.
• Log quá nhiều ở INFO trong vòng lặp nóng. Chi phí lưu trữ và độ nhiễu tăng rất nhanh. Những log như vậy nên để ở DEBUG.
• Dùng JSON khi phát triển local. JSON khó đọc bằng mắt, vì vậy hãy chọn formatter theo biến môi trường: plain cho dev, json cho production.

11. Kết luận

Một cấu hình logging tốt cho production không phức tạp, chỉ cần nhất quán. Dùng getLogger(__name__) trong mọi module. Gom cấu hình vào một lần gọi dictConfig với disable_existing_loggers: False. Ghi JSON ra stdout, gắn request_id qua ContextVar và Filter, và dùng logger.exception để giữ stack trace. Khi handler đích chậm, chuyển sang QueueHandler và QueueListener.

Cuối cùng, hãy coi log là một phần hành vi của hệ thống và test nó bằng caplog. Khi có sự cố lúc 2 giờ sáng, những dòng log có cấu trúc này sẽ giúp bạn tìm ra nguyên nhân nhanh hơn nhiều.