Xây dựng chuỗi Lazy Generator Pipeline xử lý Server Log

Mục tiêu kiến thức

• Hiểu và ứng dụng hàm Generator (yield) để xử lý luồng dữ liệu lớn với bộ nhớ O(1).
• Xây dựng kiến trúc đường ống xử lý lười (Lazy Evaluation Data Pipeline) theo phong cách Pythonic.
• Tránh cạm bẫy chuyển đổi toàn bộ luồng thành danh sách (List Materialization) gây tràn RAM.

Mô tả bài toán

Hệ thống AI Serving tại AI Empire Academy ghi nhận hàng triệu dòng log truy cập mỗi giờ. Để phân tích các lỗi và phát hiện các IP gây nghẽn hệ thống mà không làm cạn kiệt RAM của máy chủ, ta cần xây dựng một chuỗi xử lý generator gồm 3 hàm độc lập:

• parse_log_stream(lines: Iterable[str]) -> Generator[dict, None, None]
• Nhận một iterable chứa các chuỗi log dạng raw.
• Mỗi dòng log có định dạng chuẩn: "<ip> [<timestamp>] "<method> <path>" <status_code> <latency_ms>"
• Ví dụ: "192.168.1.1 [2026-09-07T10:00:00] "GET /api/v1/predict" 200 45"
• Tách và yield ra từng dictionary:

{"ip": str, "timestamp": str, "method": str, "path": str, "status": int, "latency": int}

• Bỏ qua các dòng rỗng hoặc không đúng định dạng (không ném ngoại lệ làm gián đoạn luồng).

• filter_errors(records: Iterable[dict], min_status: int = 400) -> Generator[dict, None, None]
• Nhận iterable các dictionary log.
• Chỉ yield ra các bản ghi có status >= min_status.

• extract_slow_ips(records: Iterable[dict], min_latency: int = 100) -> Generator[str, None, None]
• Nhận iterable các dictionary log.
• Chỉ yield ra chuỗi ip của các bản ghi có latency >= min_latency.

Toàn bộ các hàm trên phải là Generator, tuyệt đối không được nạp toàn bộ dữ liệu vào list hay cấu trúc nạp sẵn.

Ràng buộc

• Bộ nhớ phụ trợ: O(1) cho mỗi bước pipeline.
• Dòng hợp lệ luôn có status và latency là số nguyên không âm.
• Thời gian chạy tối đa: 1000ms.
• Giới hạn bộ nhớ: 256MB.