Thực hành: gắn correlation ID vào log JSON trong Python để lần ra lỗi từ xa

Khách hàng chạy một lệnh grep theo một chuỗi 32 ký tự hex và gửi về đúng 2 dòng log của request bị lỗi, trong đó có cả traceback. Không cần tài khoản SSH hay cuộc gọi chia sẻ màn hình nào. Bài này hướng dẫn bạn dựng đúng cơ chế đó.

Với FDE, đây là tình huống quen thuộc. Hệ thống chạy trong mạng nội bộ của khách, nhóm bảo mật không cấp quyền truy cập, và thứ duy nhất bạn nhận được là một file log dài hàng chục nghìn dòng, trong đó log của hàng trăm request nằm xen kẽ nhau. Nếu mỗi dòng không cho biết nó thuộc request nào, bạn chỉ còn cách đoán.

Bạn sẽ tự làm từ đầu, chỉ dùng thư viện chuẩn: log dạng JSON, mỗi dòng có một correlation ID, và ID đó lấy từ header chuẩn khi request đi qua nhiều service. Bạn cần Python 3.7 trở lên, một file app.py và khoảng 30 phút.

Vì sao phải là JSON thay vì một dòng text đẹp?

Logging Cookbook của Python nêu rõ mục tiêu: xuất log ở dạng có cấu trúc để chương trình đọc được mà không phải viết những regex phức tạp. Cookbook cũng đưa ra một cách đơn giản là serialise mỗi event thành JSON.

Lợi ích này rất rõ khi bạn làm việc với khách hàng. Họ có thể nạp log vào bất kỳ công cụ nào đang có, còn bạn viết được script lọc theo field mà không lo regex hỏng chỉ vì ai đó thêm dấu hai chấm vào message.

Bước 1: một ContextVar để giữ ID của request

Cookbook cho biết từ Python 3.7, module contextvars cung cấp kho lưu trữ theo context, chạy được cho cả threading lẫn asyncio, và nhìn chung nên dùng nó thay cho thread-local. Một service có thể xử lý nhiều request cùng lúc, nên đây là chỗ phù hợp để đặt ID.

import contextvars, json, logging, uuid

correlation_id = contextvars.ContextVar("correlation_id", default="-")

Cách kiểm tra: correlation_id.get() phải trả về "-" khi chưa có request nào. Giá trị mặc định này hữu ích vì log lúc khởi động service vẫn có field, chỉ là không thuộc request nào.

Bước 2: một Filter ghi ID lên từng dòng log

Tài liệu Python nói bạn có thể dùng một Filter tự định nghĩa để thêm thông tin ngữ cảnh vào log. Filter được phép sửa LogRecord và thêm attribute, sau đó format string hoặc Formatter sẽ in attribute đó ra.

class CorrelationIdFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = correlation_id.get()
        return True  # never drop the record

Nhớ return True. Nếu trả về giá trị falsy, dòng log sẽ bị bỏ mà không có thông báo gì, và đây là một lỗi rất khó tìm.

Bước 3: Formatter xuất mỗi event thành một dòng JSON

class JsonFormatter(logging.Formatter):
    def format(self, record):
        event = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "-"),
        }
        if record.exc_info:
            event["exc"] = self.formatException(record.exc_info)
        return json.dumps(event, ensure_ascii=False)

def setup_logging():
    handler = logging.StreamHandler()
    handler.addFilter(CorrelationIdFilter())
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(logging.INFO)

Lời khuyên thực tế: gắn filter vào handler chứ đừng gắn vào một logger cụ thể. Như vậy mọi dòng mà handler này ghi ra đều đi qua filter và mang ID, dù dòng đó đến từ logger nào.

Đây là phiên bản rút gọn để học. Khi dùng thật, bạn nên thêm các field như tên service hay phiên bản build.

Bước 4: lấy ID từ traceparent, nếu không có thì tự sinh

Khi request đi qua nhiều service, ID cần đi theo request đó. Chuẩn W3C Trace Context định nghĩa các HTTP header và định dạng giá trị để truyền thông tin ngữ cảnh giữa các service. Header traceparent có dạng version-traceid-parentid-flags, ví dụ trong spec là 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01.

def id_from_headers(headers):
    parts = headers.get("traceparent", "").split("-")
    if len(parts) == 4 and len(parts[1]) == 32:
        return parts[1]          # trace-id
    return uuid.uuid4().hex      # no header: start a new trace

log = logging.getLogger("orders")

def handle_request(headers, order_id):
    token = correlation_id.set(id_from_headers(headers))
    try:
        log.info("received order %s", order_id)
        if order_id == "B-42":
            raise ValueError("missing warehouse code")
        log.info("done")
    except Exception:
        log.exception("order failed")
    finally:
        correlation_id.reset(token)

Phần kiểm tra header ở đây đã được giản lược và chỉ xét số phần cùng độ dài trace-id. Quy tắc đầy đủ nằm trong spec của W3C. Code chọn trace-id làm correlation ID vì đó là phần được truyền chung giữa các service, nên cùng một chuỗi grep dùng được ở log của mọi service mà request đi qua.

Tài liệu structlog khuyên đặt lại context ngay ở đầu request handler để ID của request trước không lọt sang request sau. Code ở trên làm việc này bằng cặp lệnh: set một giá trị mới khi vào, rồi reset(token) trong finally khi ra, để không request nào để lại ID cho request kế tiếp.

Bước 5: chạy thử rồi grep đúng một request

if __name__ == "__main__":
    setup_logging()
    handle_request({}, "A-1")
    handle_request(
        {"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"},
        "B-42",
    )

python app.py 2> app.log
grep 0af7651916cd43dd8448eb211c80319c app.log

StreamHandler mặc định ghi ra stderr, vì vậy lệnh trên dùng 2>. Kết quả grep phải có đúng 2 dòng của đơn B-42: dòng received order B-42 và dòng "level": "ERROR" kèm field exc chứa traceback. Nếu có dòng nào của A-1 xuất hiện, ID đã bị rò và bạn cần kiểm tra lại bước reset.

Bước 6: ID có giữ được khi chạy song song không?

Thêm vào app.py một worker giả lập request chạy bằng asyncio:

import asyncio

async def worker(n):
    token = correlation_id.set(f"req-{n}")
    try:
        await asyncio.sleep(0.01)
        log.info("worker %s done", n)
    finally:
        correlation_id.reset(token)

Đừng chỉ nhìn log bằng mắt. Biến lần kiểm tra thành test: gắn chính CorrelationIdFilter vào handler bắt log của pytest, chạy 3 worker song song, rồi assert từng dòng. Bản dưới đây đã rút gọn để học:

# test_app.py
import asyncio, logging
from app import CorrelationIdFilter, worker

def test_no_id_leak(caplog):
    caplog.handler.addFilter(CorrelationIdFilter())
    caplog.set_level(logging.INFO)

    async def main():
        await asyncio.gather(*(worker(i) for i in range(3)))
    asyncio.run(main())

    done = [r for r in caplog.records if "done" in r.getMessage()]
    assert len(done) == 3
    for r in done:
        assert r.correlation_id == f"req-{r.args[0]}"

pytest -q test_app.py

Test phải pass: mỗi dòng worker N done mang đúng req-N. Thử xoá dòng reset(token) hoặc đặt ID ở biến toàn cục thay vì ContextVar để thấy test đỏ, khi đó bạn biết test thực sự bắt được lỗi rò ID.

Trường hợp khó hơn là ứng dụng trộn sync và async. Tài liệu structlog cảnh báo rằng với Starlette hay FastAPI, context variable đặt trong context sync có thể không xuất hiện trong log của context async.

Nếu middleware đặt ID ở một chế độ còn endpoint chạy ở chế độ kia, log sẽ mang giá trị "-". Vì vậy cần thêm một test gọi cả endpoint def lẫn async def, và assert không dòng nào mang "-".

Nếu nhóm bạn dùng structlog thay cho bản viết tay, tài liệu của thư viện này khuyên đặt structlog.contextvars.merge_contextvars làm processor đầu tiên để context đã bind có mặt trong mọi event. Quy tắc đặt lại context ở đầu mỗi request vẫn áp dụng như trên.

Những lỗi hay gặp

Lỗi phổ biến nhất là quên đặt lại context, khiến ID của request trước xuất hiện trong log của request sau. Khi đó khách hàng gửi bạn những dòng log không liên quan, và bạn mất thời gian lần theo hướng sai.

Một lỗi khác là có ID trong log nhưng người dùng cuối không bao giờ thấy nó. Hãy trả ID về trong response hoặc in lên màn hình lỗi, để khi báo lỗi khách hàng có sẵn chuỗi cần grep.

Kỹ năng này được dùng thế nào khi làm việc với khách hàng

Trong một deployment thực tế, ngay tuần đầu bạn nên thống nhất với khách ba việc: log ghi ra đâu, ai trong nhóm họ có quyền chạy grep, và ID hiện ra ở chỗ nào trên giao diện. Khi có lỗi, bạn chỉ cần xin một ID thay vì xin cả file log, vừa nhanh hơn vừa ít rủi ro lộ dữ liệu không liên quan.

Về nghề nghiệp, khi đọc mô tả công việc FDE, hãy để ý các từ như “debug in customer environments” hoặc “observability”. Trong CV, nên viết cụ thể kiểu “thêm correlation ID theo W3C traceparent vào log JSON, rút ngắn thời gian tìm lỗi trên hệ thống khách không cho truy cập” thay vì chỉ ghi “biết logging”.

Khi phỏng vấn, chuẩn bị sẵn câu trả lời cho câu hỏi ID có bị mất giữa sync và async không. Câu hỏi này cho người phỏng vấn biết bạn đã chạy hệ thống thật hay mới chỉ đọc tài liệu.

Lần tới khách hàng nói “hệ thống lỗi lúc 3 giờ chiều”, bạn sẽ không phải hỏi thêm câu nào ngoài việc xin họ ID của request đó.