Module logging — Hệ thống ghi log chuyên nghiệp Standard Library

print() debug là khoản nợ kỹ thuật không bao giờ xuất hiện trong sprint planning.

Hình dung kịch bản: 3 giờ sáng, hệ thống production sập. Bạn mở terminal, grep qua hàng triệu dòng log — toàn print("here"), print(data), không timestamp, không context, không severity level. Bạn không biết request nào gây lỗi, user nào bị ảnh hưởng, service nào gọi service nào.

Module logging trong standard library tồn tại vì một lý do đơn giản: print() không scale. Khi hệ thống có 10 service, 1000 request/giây, và 5 developer cùng debug — bạn cần một hệ thống ghi log có cấu trúc, có phân cấp, có thể lọc, xoay vòng, và tích hợp với hạ tầng observability.

---

Bức tranh tư duy

Hãy hình dung hệ thống logging như hệ thống thông báo nội bộ của một công ty:

Thành phần	Tương tự trong công ty
Logger	Nhân viên — người tạo ra thông báo
Handler	Kênh truyền — email, Slack, bảng tin, hệ thống cảnh báo
Formatter	Mẫu báo cáo — format chuẩn để mọi người đọc hiểu
Filter	Quy tắc phân loại — chỉ gửi thông báo phù hợp cho từng kênh

Một nhân viên (Logger) tạo báo cáo. Báo cáo được format theo mẫu chuẩn (Formatter), lọc theo quy tắc (Filter), rồi gửi qua nhiều kênh (Handler): email cho sếp, Slack cho team, bảng tin cho toàn bộ phòng ban.

Code gọi logger.info("msg")
        │
        ▼
┌──────────────────────────────────────────────────┐
│  Logger (name="myapp.auth")                      │
│  ├── level check: INFO >= logger.level?          │
│  ├── filter check: filter.filter(record)?        │
│  └── propagate lên parent: "myapp" → root        │
└────────────────────┬─────────────────────────────┘
                     │ LogRecord
        ┌────────────┼────────────┐
        ▼            ▼            ▼
┌──────────────┐ ┌──────────┐ ┌───────────────────┐
│ StreamHandler│ │FileHandler│ │RotatingFileHandler│
│ (console)    │ │ (app.log)│ │ (app.log, 10MB)   │
└──────┬───────┘ └────┬─────┘ └─────────┬─────────┘
       │              │                 │
       ▼              ▼                 ▼
┌──────────────┐ ┌──────────┐ ┌───────────────────┐
│  Formatter   │ │Formatter │ │   JSONFormatter    │
│  (text)      │ │ (text)   │ │   (structured)     │
└──────┬───────┘ └────┬─────┘ └─────────┬─────────┘
       │              │                 │
       ▼              ▼                 ▼
   Terminal        app.log         app.log.json

Logger hierarchy hoạt động theo dot notation — giống hệ thống phòng ban:

root                    ← Logger gốc, bắt mọi log
├── myapp               ← Logger ứng dụng
│   ├── myapp.auth      ← Logger module auth
│   ├── myapp.api       ← Logger module api
│   └── myapp.db        ← Logger module database
└── uvicorn             ← Logger framework
    └── uvicorn.access  ← Logger access log

Khi myapp.auth ghi log, LogRecord propagate lên myapp, rồi lên root. Mỗi logger trên đường đi đều có thể xử lý hoặc bỏ qua record đó.

---

Cốt lõi kỹ thuật

Log Levels — Khi nào dùng level nào

Mỗi log message có một severity level. Chọn đúng level là kỹ năng quan trọng nhất của logging:

Level	Giá trị	Khi nào sử dụng
DEBUG	10	Chi tiết kỹ thuật cho developer — biến, state, flow
INFO	20	Sự kiện nghiệp vụ bình thường — user login, order created
WARNING	30	Điều bất thường nhưng hệ thống vẫn hoạt động — disk 80%, retry
ERROR	40	Lỗi cần xử lý — request thất bại, service không phản hồi
CRITICAL	50	Hệ thống không thể tiếp tục — database mất kết nối, out of memory

import logging

logger = logging.getLogger(__name__)

# DEBUG — chỉ hiện khi cần debug chi tiết
logger.debug("Query params: user_id=%s, page=%d", user_id, page)

# INFO — sự kiện nghiệp vụ bình thường
logger.info("User %s logged in successfully", username)

# WARNING — bất thường nhưng chưa lỗi
logger.warning("Disk usage at %d%%, threshold is 80%%", disk_pct)

# ERROR — lỗi cần attention
logger.error("Payment failed for order %s: %s", order_id, error_msg)

# CRITICAL — hệ thống gặp vấn đề nghiêm trọng
logger.critical("Database connection pool exhausted, service degraded")

Quy tắc vàng

Production nên chạy ở level INFO. Chỉ bật DEBUG khi cần điều tra vấn đề cụ thể. Nếu bạn thấy mình đang log DEBUG cho mọi thứ — bạn đang dùng sai level.

Logger Hierarchy — getLogger(__name__)

Luôn dùng __name__ khi tạo logger. Python tự động tạo hierarchy dựa trên tên module:

# file: myapp/auth/service.py
import logging

# __name__ == "myapp.auth.service"
# Tự động tạo hierarchy: root → myapp → myapp.auth → myapp.auth.service
logger = logging.getLogger(__name__)

def authenticate(username: str) -> bool:
    logger.info("Authentication attempt for user: %s", username)
    # ...

Propagation — LogRecord đi từ child lên parent:

import logging

# Tắt propagation khi muốn logger con xử lý độc lập
child_logger = logging.getLogger("myapp.auth")
child_logger.propagate = False  # Không gửi lên "myapp" và root

# Kiểm tra hierarchy
root = logging.getLogger()           # root logger
app = logging.getLogger("myapp")     # parent
auth = logging.getLogger("myapp.auth")  # child

print(auth.parent)  # <Logger myapp (WARNING)>

Handlers — Nơi log được ghi ra

Handler quyết định đích đến của log message:

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)

# --- StreamHandler: ghi ra console ---
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
logger.addHandler(console_handler)

# --- FileHandler: ghi ra file ---
file_handler = logging.FileHandler("app.log", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
logger.addHandler(file_handler)

# --- RotatingFileHandler: xoay vòng theo kích thước ---
# Tối đa 10MB/file, giữ 5 bản backup
rotating_handler = RotatingFileHandler(
    "app_rotating.log",
    maxBytes=10 * 1024 * 1024,  # 10MB
    backupCount=5,
    encoding="utf-8",
)
rotating_handler.setLevel(logging.WARNING)
logger.addHandler(rotating_handler)

# Các handler khác: TimedRotatingFileHandler (xoay theo thời gian),
# QueueHandler + QueueListener (non-blocking cho async) — xem phần Thực chiến.

Formatters — Định dạng log output

Formatter kiểm soát nội dung hiển thị của mỗi dòng log:

import logging

# Format cơ bản cho console
console_fmt = logging.Formatter(
    fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%H:%M:%S",
)

# Format chi tiết cho file
file_fmt = logging.Formatter(
    fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

Bảng format attributes thường dùng:

Attribute	Ý nghĩa	Ví dụ
%(asctime)s	Timestamp	2024-01-15 03:22:41
%(levelname)s	Tên level	ERROR
%(name)s	Tên logger	myapp.auth.service
%(funcName)s	Tên function	authenticate
%(lineno)d	Số dòng code	42
%(message)s	Nội dung log	User login failed

Custom Formatter — JSON cho production:

import json
import logging
from datetime import datetime, timezone


class JSONFormatter(logging.Formatter):
    """Formatter xuất JSON cho ELK stack / Datadog / CloudWatch."""

    def format(self, record: logging.LogRecord) -> str:
        log_data: dict[str, object] = {
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        if record.exc_info and record.exc_info[1] is not None:
            log_data["exception"] = {
                "type": record.exc_info[0].__name__ if record.exc_info[0] else "Unknown",
                "message": str(record.exc_info[1]),
                "traceback": self.formatException(record.exc_info),
            }

        return json.dumps(log_data, ensure_ascii=False, default=str)

Filters — Kiểm soát log nào được ghi

Filter cho phép logic phân loại tinh vi hơn level:

import logging
import contextvars

# Context variable lưu request_id xuyên suốt request lifecycle
request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "request_id", default="no-request-id"
)


class RequestContextFilter(logging.Filter):
    """Tự động inject request_id vào mọi LogRecord."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()  # type: ignore[attr-defined]
        return True


# Gắn filter vào logger
logger = logging.getLogger("myapp")
logger.addFilter(RequestContextFilter())

# Mở rộng: viết SensitiveDataFilter chặn log chứa password, token, api_key
# bằng cách kiểm tra record.getMessage() và thay bằng [REDACTED].

Dict Config — Cấu hình tập trung

Trong production, luôn dùng dictConfig thay vì gọi addHandler thủ công:

import logging.config

LOGGING_CONFIG: dict = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {
        "request_context": {
            "()": "myapp.logging_setup.RequestContextFilter",
        },
    },
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        },
        "json": {
            "()": "myapp.logging_setup.JSONFormatter",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "filters": ["request_context"],
            "stream": "ext://sys.stdout",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "json",
            "filters": ["request_context"],
            "filename": "logs/app.log",
            "maxBytes": 10485760,  # 10MB
            "backupCount": 5,
            "encoding": "utf-8",
        },
    },
    "loggers": {
        "myapp": {
            "level": "DEBUG",
            "handlers": ["console", "file"],
            "propagate": False,
        },
    },
    "root": {
        "level": "WARNING",
        "handlers": ["console"],
    },
}

# Áp dụng cấu hình khi khởi động ứng dụng
logging.config.dictConfig(LOGGING_CONFIG)

---

Thực chiến — Production Logging cho FastAPI

Kịch bản: Hệ thống logging cho FastAPI microservice, xuất JSON structured log để ELK stack thu thập và phân tích. Kết hợp JSONFormatter, RequestContextFilter từ phần Cốt lõi kỹ thuật.

Setup logging module

# file: myapp/logging_setup.py
"""Centralized logging configuration cho production."""

import logging
import os
from queue import Queue
from logging.handlers import QueueHandler, QueueListener, RotatingFileHandler

# Import RequestContextFilter, JSONFormatter đã định nghĩa ở trên
from myapp.filters import RequestContextFilter
from myapp.formatters import JSONFormatter

_log_queue: Queue[logging.LogRecord] = Queue(maxsize=10000)
_listener: QueueListener | None = None


def setup_logging(*, log_level: str = "INFO", log_file: str = "logs/app.log") -> None:
    """Khởi tạo logging system cho toàn bộ ứng dụng."""
    context_filter = RequestContextFilter()

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.setFormatter(
        logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s | [%(request_id)s] %(message)s",
            datefmt="%H:%M:%S",
        )
    )
    console_handler.addFilter(context_filter)

    os.makedirs(os.path.dirname(log_file), exist_ok=True)
    file_handler = RotatingFileHandler(
        log_file, maxBytes=50 * 1024 * 1024, backupCount=10, encoding="utf-8",
    )
    file_handler.setFormatter(JSONFormatter())
    file_handler.addFilter(context_filter)

    # QueueHandler — non-blocking wrapper
    global _listener
    queue_handler = QueueHandler(_log_queue)
    _listener = QueueListener(
        _log_queue, console_handler, file_handler, respect_handler_level=True,
    )
    _listener.start()

    root_logger = logging.getLogger()
    root_logger.setLevel(getattr(logging, log_level.upper()))
    root_logger.addHandler(queue_handler)


def shutdown_logging() -> None:
    global _listener
    if _listener is not None:
        _listener.stop()
        _listener = None

Middleware gắn request context

# file: myapp/middleware.py
import uuid
import logging
import time
import contextvars
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response

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

logger = logging.getLogger(__name__)


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        req_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request_id_var.set(req_id)
        start_time = time.perf_counter()

        logger.info("Request started: %s %s", request.method, request.url.path)

        try:
            response = await call_next(request)
        except Exception:
            logger.exception("Unhandled exception during request processing")
            raise

        duration_ms = (time.perf_counter() - start_time) * 1000
        logger.info(
            "Request completed: %s %s → %d (%.1fms)",
            request.method, request.url.path, response.status_code, duration_ms,
        )
        response.headers["X-Request-ID"] = req_id
        return response

Tích hợp vào FastAPI

# file: myapp/main.py
import logging
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI
from myapp.logging_setup import setup_logging, shutdown_logging
from myapp.middleware import RequestLoggingMiddleware

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    setup_logging(log_level="INFO", log_file="logs/app.log")
    logger.info("Application starting up")
    yield
    logger.info("Application shutting down")
    shutdown_logging()

app = FastAPI(title="MyApp", lifespan=lifespan)
app.add_middleware(RequestLoggingMiddleware)

---

Sai lầm điển hình

❌ SAI: Log dữ liệu nhạy cảm

import logging

logger = logging.getLogger(__name__)

# ❌ TUYỆT ĐỐI KHÔNG — password, token xuất hiện trong log file
def login(username: str, password: str, api_token: str) -> None:
    logger.info("Login: user=%s, pass=%s, token=%s", username, password, api_token)

✅ ĐÚNG: Che dữ liệu nhạy cảm

import logging

logger = logging.getLogger(__name__)

def login(username: str, password: str, api_token: str) -> None:
    logger.info("Login attempt: user=%s", username)
    # Nếu cần debug auth: log token prefix, không log toàn bộ
    logger.debug("Token prefix: %s...", api_token[:8])

---

❌ SAI: Dùng f-string trong log — luôn bị evaluate

import logging

logger = logging.getLogger(__name__)

# ❌ f-string được evaluate NGAY CẢ KHI level bị filter
# Nếu logger.level = WARNING, dòng dưới vẫn tốn CPU tính f-string
logger.debug(f"Processing items: {expensive_serialize(items)}")

✅ ĐÚNG: Dùng lazy formatting với %s

import logging

logger = logging.getLogger(__name__)

# ✅ %s chỉ format KHI log thực sự được ghi
# Nếu logger.level = WARNING, expensive_serialize() KHÔNG được gọi
logger.debug("Processing items: %s", expensive_serialize(items))

# Hoặc kiểm tra level trước khi tính toán nặng
if logger.isEnabledFor(logging.DEBUG):
    logger.debug("Full state: %s", compute_debug_snapshot())

---

❌ SAI: Quên exc_info=True trong exception handler

import logging

logger = logging.getLogger(__name__)

try:
    result = call_external_service()
except Exception as e:
    # ❌ Mất toàn bộ traceback — chỉ có message, không biết lỗi ở đâu
    logger.error("Service call failed: %s", e)

✅ ĐÚNG: Luôn dùng logger.exception() hoặc exc_info=True

import logging

logger = logging.getLogger(__name__)

try:
    result = call_external_service()
except ConnectionError:
    # ✅ logger.exception() tự động thêm traceback
    logger.exception("Service call failed — connection error")
except Exception:
    # ✅ exc_info=True cũng tương đương, dùng với level khác ERROR
    logger.critical("Unexpected error in service call", exc_info=True)

---

Under the Hood

LogRecord lifecycle

Khi bạn gọi logger.info("msg"), bên trong xảy ra chuỗi sự kiện:

# Pseudocode mô phỏng internal flow
def info(self, msg, *args, **kwargs):
    # 1. Kiểm tra level — O(1)
    if not self.isEnabledFor(INFO):
        return  # ← f-string đã bị evaluate trước khi đến đây!

    # 2. Tạo LogRecord — chứa tất cả metadata
    record = LogRecord(
        name=self.name, level=INFO, pathname=__file__,
        lineno=current_line, msg=msg, args=args,
        exc_info=kwargs.get("exc_info"),
    )

    # 3. Chạy qua filters của logger
    for f in self.filters:
        if not f.filter(record):
            return

    # 4. Gửi cho handlers CỦA LOGGER NÀY
    for handler in self.handlers:
        if record.levelno >= handler.level:
            handler.handle(record)  # → formatter.format() → emit()

    # 5. Propagate lên parent (nếu propagate=True)
    if self.propagate and self.parent:
        self.parent.handle(record)  # Lặp lại bước 4 cho parent

Thread-safety

Module logging thread-safe by design — mỗi Handler có threading.RLock riêng. LogRecord được tạo riêng cho mỗi thread (không share state). Khi nhiều thread log đồng thời, lock contention tăng — QueueHandler giải quyết bằng cách serialize qua queue, xử lý I/O trong background thread riêng.

---

✅ Checklist triển khai

Thiết lập cơ bản

• [ ] Luôn dùng logging.getLogger(__name__) — không dùng root logger trực tiếp
• [ ] Cấu hình logging MỘT LẦN khi app startup (dictConfig hoặc setup function)
• [ ] Chọn đúng level: DEBUG cho dev, INFO cho production, WARNING cho thư viện

Formatting và output

• [ ] Dùng %s placeholder thay vì f-string trong log calls
• [ ] JSON formatter cho production — tương thích ELK / Datadog
• [ ] Bao gồm request_id trong mọi log để trace request flow

Error handling

• [ ] Luôn dùng logger.exception() hoặc exc_info=True trong except blocks
• [ ] Không bao giờ log password, token, API key, hoặc PII data
• [ ] Kiểm tra isEnabledFor() trước khi tính toán expensive debug data

Production readiness

• [ ] RotatingFileHandler hoặc TimedRotatingFileHandler — không để file log phình vô hạn
• [ ] QueueHandler cho async code — không block event loop
• [ ] disable_existing_loggers: False trong dictConfig để không tắt logger thư viện
• [ ] Gắn Filter cho request context (request_id, user_id) xuyên suốt request lifecycle

---

🧠 Quiz — Bài tập luyện tập

Bài 1 — Thiết lập logging cơ bản

Viết function setup_app_logging() cấu hình:

• Console handler ở level INFO, format: "%(asctime)s [%(levelname)s] %(message)s"
• File handler ở level DEBUG, ghi vào "debug.log" với RotatingFileHandler (5MB, 3 backup)
• Root logger ở level DEBUG

Xem gợi ý

Dùng logging.getLogger() (không tên) để lấy root logger. Tạo hai handler riêng biệt, mỗi handler có formatter và level riêng. Nhớ addHandler() cho root logger.

import logging
from logging.handlers import RotatingFileHandler


def setup_app_logging() -> None:
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)

    # Console handler
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(
        logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
    )
    root.addHandler(console)

    # File handler
    file_h = RotatingFileHandler(
        "debug.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=3,
        encoding="utf-8",
    )
    file_h.setLevel(logging.DEBUG)
    file_h.setFormatter(
        logging.Formatter(
            "%(asctime)s | %(levelname)-8s | %(name)s | "
            "%(funcName)s:%(lineno)d | %(message)s"
        )
    )
    root.addHandler(file_h)

Bài 2 — JSON Structured Logging với correlation ID

Xây dựng hệ thống logging cho web service với yêu cầu:

1. JSON output tương thích Elasticsearch
2. Correlation ID từ header X-Correlation-ID (tạo mới nếu không có)
3. Mỗi log entry chứa: timestamp, level, logger, message, correlation_id, duration_ms (nếu có)
4. Dùng contextvars để truyền correlation_id xuyên async context

Xem gợi ý

Kết hợp contextvars.ContextVar, custom logging.Filter, và JSONFormatter từ phần Cốt lõi kỹ thuật. Middleware set context variable, filter inject vào LogRecord.

import json
import logging
import uuid
import contextvars
from datetime import datetime, timezone

correlation_id_var: contextvars.ContextVar[str] = contextvars.ContextVar(
    "correlation_id", default="-"
)


class CorrelationFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.correlation_id = correlation_id_var.get()  # type: ignore[attr-defined]
        return True


class ELKFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "@timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "-"),
        }
        if hasattr(record, "duration_ms"):
            entry["duration_ms"] = record.duration_ms  # type: ignore[attr-defined]
        if record.exc_info and record.exc_info[1]:
            entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(entry, ensure_ascii=False, default=str)


# Usage trong async middleware
async def logging_middleware(request, call_next):
    corr_id = request.headers.get(
        "X-Correlation-ID", str(uuid.uuid4())
    )
    correlation_id_var.set(corr_id)

    logger = logging.getLogger("myapp.http")
    logger.info("Request: %s %s", request.method, request.url.path)

    response = await call_next(request)

    logger.info(
        "Response: %d", response.status_code,
        extra={"duration_ms": response.headers.get("X-Duration-Ms")},
    )
    return response

---

📚 Bài liên quan

Kiến thức nền tảng

• Functions — Hàm trong Python — Hiểu decorator và context manager trước khi viết logging utilities
• pathlib & File I/O — Quản lý log file paths và directories

Nâng cao

• Context Managers — Tạo logging context tự động cleanup
• Decorators — Viết @log_execution_time decorator
• Concurrency — Thread-safety, async logging, QueueHandler patterns

Thực hành

• Test Architecture — Kiểm thử logging output, caplog fixture trong pytest

Tài liệu tham khảo

• Python logging module — docs.python.org
• Logging HOWTO — docs.python.org
• Logging Cookbook — docs.python.org