Prompt Monitoring

---
name: log-error-pattern-triage
description: Phân loại các tệp nhật ký (log) ứng dụng, máy chủ và truy cập lớn hoặc hỗn độn - gom nhóm hàng nghìn dòng log thành một danh sách xếp hạng các mẫu lỗi kèm số lượng, thời điểm thấy đầu tiên và cuối cùng, các đợt tăng đột biến (spikes) và các mẫu mới xuất hiện so với bản log cơ sở (baseline) lúc hệ thống ổn định, sau đó tách biệt nguyên nhân gốc rễ khỏi các triệu chứng dây chuyền phía sau và viết báo cáo phân loại sự cố ngắn gọn kèm các bước kiểm tra tiếp theo. Sử dụng khi người dùng dán hoặc tải tệp log lên, hỏi "có chuyện gì đang xảy ra trong các log này?", "tại sao lỗi lại tăng vọt lúc 10:09?" hoặc cần bản tóm tắt sự cố sơ bộ.
---

# Phân Loại Và Chẩn Đoán Mẫu Lỗi Trong Nhật Ký Hệ Thống (Log)

Bạn biến một "bức tường" dày đặc các dòng log thành một danh sách vấn đề ngắn gọn, có xếp hạng ưu tiên cùng bước hành động tiếp theo rõ ràng. Bạn không bao giờ dán trả lại toàn bộ tệp log; bạn đếm, gom nhóm, so sánh và giải thích.

## Các tệp trong kỹ năng này

- `scripts/cluster_logs.py` - gom nhóm các bản ghi log thành những mẫu đã che biến số (masked patterns), xếp hạng chúng, phát hiện các đợt tăng đột biến (spikes) và đánh dấu các mẫu MỚI (`NEW`) so với tệp log cơ sở (chỉ dùng thư viện chuẩn Python 3)
- `references/log-normalization.md` - cách chuyển các dòng log thành mẫu chuẩn, những thành phần nào được che bằng placeholder và cách xử lý các định dạng mà kịch bản chưa nhận diện được
- `references/triage-heuristics.md` - phương pháp xếp hạng các mẫu lỗi, phân biệt nguyên nhân gốc rễ với triệu chứng và quyết định cần kiểm tra gì tiếp theo
- `templates/triage-report.md` - định dạng báo cáo phân loại sự cố
- `examples/example-checkout-incident.md` - bài giải mẫu phân loại một sự cố tăng đột biến lỗi quá thời gian chờ (timeout) khi thanh toán

## Quy trình thực hiện

### 1. Lấy đúng lát cắt dữ liệu log cần thiết
Hỏi (hoặc xác nhận) tên dịch vụ, khoảng thời gian xung quanh lúc xảy ra sự cố kèm theo múi giờ, và nếu có thể thì lấy thêm một tệp log từ khoảng thời gian hệ thống hoạt động ổn định có cùng độ dài để làm mốc cơ sở (baseline). Nếu tệp log quá khổng lồ, hãy chỉ làm việc trên khoảng thời gian quan trọng; một lát cắt 15 phút xoay quanh sự cố thường là đủ.

Loại bỏ các thông tin bí mật trước khi chia sẻ: mã thông báo (tokens), mật khẩu, cookie phiên làm việc và dữ liệu cá nhân. Nếu bạn thấy bất kỳ thông tin nhạy cảm nào trong dữ liệu đầu vào, hãy cảnh báo cho người dùng biết và tuyệt đối không lặp lại chúng.

### 2. Chạy công cụ gom nhóm log
```bash
python3 scripts/cluster_logs.py app.log
python3 scripts/cluster_logs.py app.log --baseline yesterday.log --top 20
python3 scripts/cluster_logs.py access.log --min-level INFO
kubectl logs deploy/api --since=30m | python3 scripts/cluster_logs.py - --json
```
Công cụ sẽ in ra mỗi dòng một mẫu lỗi kèm theo mức độ (level), số lần xuất hiện (count), tỷ trọng (share), thời điểm xuất hiện đầu tiên và cuối cùng, các nhãn cờ (`NEW` = không có trong bản baseline, `SPIKE` = có một phút tăng vọt ít nhất gấp 3 lần tốc độ thông thường), theo sau là một dòng log mẫu thực tế và dòng cuối cùng của dấu vết ngăn xếp (stack trace) cho mỗi mẫu. Mã thoát (exit code) 1 nghĩa là tìm thấy ít nhất một mẫu lỗi ở mức ERROR hoặc FATAL.

Nếu không thể chạy kịch bản, hãy gom nhóm các dòng log thủ công bằng các quy tắc che biến số trong `references/log-normalization.md` và nêu rõ rằng số liệu đếm chỉ mang tính xấp xỉ.

### 3. Phân loại và chẩn đoán (Triage)
Áp dụng `references/triage-heuristics.md`:
1. Sắp xếp các mẫu theo mức độ tác động: ưu tiên FATAL và NEW+SPIKE lên đầu, sau đó xét theo số lượng, rồi đến mức độ ảnh hưởng trực tiếp tới người dùng cuối.
2. Dựng một dòng thời gian (timeline) ngắn gọn từ thời điểm xuất hiện đầu tiên của các mẫu. Mẫu lỗi mới xuất hiện sớm nhất trong một đợt bùng phát thường nằm gần nguyên nhân gốc rễ hơn; các mẫu bắt đầu xuất hiện sau đó vài giây thường là triệu chứng hệ quả.
3. Tách biệt rõ các ứng viên nguyên nhân gốc rễ, các triệu chứng và nhiễu nền (background noise) vốn đã tồn tại sẵn trong bản baseline.
4. Đối với mỗi ứng viên nguyên nhân gốc rễ, hãy chỉ rõ bằng chứng và bước kiểm tra tiếp theo ít tốn công sức nhất (bảng điều khiển giám sát, trang trạng thái của dịch vụ phụ thuộc, bản so sánh thay đổi cấu hình, nhật ký triển khai bản mới hoặc một truy vấn cụ thể).

### 4. Báo cáo kết quả
Điền thông tin vào `templates/triage-report.md`, tương tự như trong `examples/example-checkout-incident.md`. Giữ phần tóm tắt gọn trong ba câu để một người quản lý có thể đọc hiểu ngay.

## Quy tắc
- Trích dẫn các dòng log mẫu có thật; tuyệt đối không tự bịa ra dòng log, số lần xuất hiện hay mốc thời gian.
- Nêu rõ múi giờ của tệp log và giữ nhất quán xuyên suốt báo cáo.
- Không khẳng định chắc chắn 100% nguyên nhân gốc rễ nếu chỉ dựa vào log; hãy dùng cụm từ "có khả năng cao nhất" (most likely) và liệt kê những gì cần kiểm tra để xác nhận nó.
- Đánh giá nhiễu nền một cách trung thực: nếu một mẫu lỗi cũng xuất hiện trong bản baseline với tần suất tương tự thì đó không phải là nguyên nhân của sự cố lần này.
- Tuyệt đối không bao giờ đề xuất xóa log hoặc tắt chức năng ghi log để làm cho lỗi biến mất.
FILE:references/log-normalization.md
# Chuẩn hóa nhật ký: từ các dòng log riêng lẻ đến các mẫu lỗi (patterns)

Cơ chế gom nhóm hoạt động bằng cách biến mỗi thông điệp log thành một khuôn mẫu (template): những từ cố định được giữ nguyên, còn các phần biến thiên sẽ được thay thế bằng ký hiệu giữ chỗ (placeholder). Hai dòng log có cùng khuôn mẫu chính là cùng một vấn đề đang lặp lại nhiều lần.

## Những thành phần được kịch bản che đi (mask)

| Thành phần biến thiên | Ví dụ | Ký hiệu giữ chỗ (Placeholder) |
| --- | --- | --- |
| UUID | `0288ddd8-5e8c-45f9-a0e7-486d8fa66b03` | `<uuid>` |
| Địa chỉ email | `li@example.com` | `<email>` |
| Địa chỉ IPv4 có kèm cổng (tùy chọn) | `192.0.2.44:5432` | `<ip>` |
| Giá trị thập lục phân (hex) và mã hex dài | `0x7f3a`, `9f1c2e7a4b3d` | `<hex>` |
| Chuỗi truy vấn URL (query string) | `?page=2&sort=price` | `?<query>` |
| Giá trị nằm trong dấu ngoặc kép/đơn | `'cart:42'`, `"Bob"` | `<str>` |
| Con số có kèm đơn vị (tùy chọn) | `5000ms`, `89%`, `17` | `<n>` |
| Mã định danh trong ngoặc vuông có chứa chữ số | `[http-nio-8080-exec-9]`, `[req-ab12]` | `[<id>]` |

Dấu thời gian (timestamps) và mức độ nghiêm trọng (levels) được phân tích trước tiên và tách ra khỏi nội dung thông điệp. Đối với các dòng syslog, tên máy chủ (hostname) được lược bỏ để cùng một sự cố xảy ra trên `web-01` và `web-02` được gom chung vào một nhóm. Đối với nhật ký truy cập (access logs), khuôn mẫu sẽ có dạng `METHOD /path -> HTTPstatus`, trong đó các phân đoạn đường dẫn dạng số được che đi để `/api/orders/123` và `/api/orders/456` gom chung thành một mẫu.

## Các định dạng được hỗ trợ

- Các dòng văn bản thuần có dấu thời gian chuẩn ISO-8601: `2026-10-09T10:09:00.123Z ERROR [thread] logger - message`
- Syslog: `Oct  9 03:12:44 web-02 kernel: message`
- Nhật ký truy cập kết hợp (combined access logs) của Nginx và Apache: `... [09/Oct/2026:09:58:02 +0300] "POST /api/checkout HTTP/1.1" 502 ...`
- Các dòng JSON có trường `level` hoặc `severity`, `msg` hoặc `message`, `time` hoặc `timestamp`, và trường tùy chọn `error`

## Các bản ghi kéo dài nhiều dòng (Multi-line entries)

Dấu vết ngăn xếp (stack traces) thuộc về dòng log ngay phía trên chúng. Kịch bản sẽ tự động ghép nối các dòng thụt lề, `Traceback (most recent call last)`, `Caused by:`, các khung gọi hàm Java dạng `at ...(File.java:12)`, `... 12 more`, và các dòng ngoại lệ đứng riêng lẻ như `java.net.SocketTimeoutException: Read timed out`. Dòng được ghép nối cuối cùng sẽ hiển thị tại mục "trace ends" vì nó thường chỉ ra khung gọi hàm sâu nhất hoặc ngoại lệ thực sự gây lỗi.

## Mức độ nghiêm trọng (Levels)

Các nhãn mức độ ghi tường minh sẽ được ưu tiên (`TRACE/DEBUG`, `INFO/NOTICE`, `WARN/WARNING`, `ERROR/ERR/SEVERE`, `CRITICAL/FATAL/PANIC`). Một dòng không có nhãn mức độ sẽ được phân loại dựa theo từ ngữ bên trong: các từ chỉ sự thất bại (failed, out of memory, timed out, refused, denied, killed) được tính là ERROR, còn các từ chỉ việc thử lại hoặc ngừng hỗ trợ (retry, deprecation) được tính là WARN. Mã trạng thái HTTP 5xx trong access log được tính là ERROR, còn mã 4xx ngoài 404 được tính là WARN.

## Khi việc gom nhóm gặp sai lệch

- **Quá nhiều mẫu nhỏ lẻ**: có một từ biến thiên chưa được che đi (tên người dùng, tên máy chủ nằm ngay trong câu thông báo, tên tệp tin). Hãy nêu rõ điều đó và tự gom nhóm các dòng đó lại trong báo cáo, ví dụ: "2 mẫu: tấn công vét cạn SSH từ 2 địa chỉ IP với các tên người dùng khác nhau".
- **Một mẫu khổng lồ che khuất hai vấn đề khác nhau**: thông điệp quá chung chung ("request failed"). Hãy xem xét các dòng mẫu cụ thể và phần đuôi stack trace, hoặc chạy lại trên một khung thời gian hẹp hơn.
- **Định dạng lạ chưa nhận diện được**: nếu hầu hết các bản ghi đều báo không có dấu thời gian, hãy chuyển đổi định dạng tệp log trước (ví dụ dùng `jq -c` cho JSON lồng nhau) hoặc mô tả định dạng và gom nhóm thủ công.
- **Các dòng bị cắt ngắn**: khuôn mẫu được cắt ở mức tối đa 160 ký tự; dòng mẫu thực tế cắt ở mức 200 ký tự.
FILE:references/triage-heuristics.md
# Kinh nghiệm chẩn đoán và phân loại sự cố (Triage heuristics)

## Xếp hạng mẫu lỗi theo mức độ tác động, không chỉ theo số lượng

1. **Các mẫu FATAL hoặc sập tiến trình (crash)** (tiến trình tự thoát, tràn bộ nhớ - out of memory, panic): chỉ cần xuất hiện một lần cũng đã rất nghiêm trọng.
2. **Đi kèm cả hai nhãn NEW và SPIKE**: có thứ gì đó vừa thay đổi. Đây thường chính là tâm điểm của sự cố.
3. **Lỗi ảnh hưởng trực tiếp tới người dùng** (lỗi 5xx trên các endpoint phục vụ khách hàng, thanh toán thất bại, đăng nhập thất bại) được ưu tiên hơn các lỗi nội bộ (trượt bộ nhớ đệm - cache misses, các lần thử lại mà sau đó đã thành công).
4. **Số lần xuất hiện và tỷ trọng**: trong cùng một cấp độ ưu tiên, mẫu nào nhiều hơn sẽ xếp trước.
5. **Nhiễu nền xếp cuối cùng**: các mẫu vốn đã xuất hiện trong bản baseline với tần suất tương đương là nhiễu nền của hệ thống, không phải sự cố hiện tại.

## Nguyên nhân gốc rễ hay triệu chứng?

| Manh mối | Nghiêng về nguyên nhân gốc rễ | Nghiêng về triệu chứng hệ quả |
| --- | --- | --- |
| Thời điểm | mẫu lỗi mới xuất hiện đầu tiên trong đợt bùng phát | bắt đầu xuất hiện vài giây sau một mẫu lỗi khác |
| Vị trí phát sinh | chỉ đích danh một dịch vụ phụ thuộc, cấu hình, giới hạn tài nguyên hoặc đợt triển khai (deploy) | lớp bọc thông báo chung chung ("request failed", "checkout failed") |
| Dấu vết ngăn xếp (Stack trace) | khung gọi hàm sâu nhất nằm ở thư viện client hoặc lời gọi tài nguyên | dấu vết kết thúc ngay trong mã controller của bạn sau khi gọi sang một thành phần khác |
| Tỷ lệ số lượng | số lần xuất hiện khớp với số lệnh gọi dịch vụ đầu nguồn bị thất bại | số lần xuất hiện bằng tổng của một vài mẫu lỗi khác cộng lại |
| So với bản cơ sở (Baseline) | hoàn toàn không có trước đó | đã xuất hiện từ trước nhưng ở tần suất thấp hơn |

Một chuỗi sự cố điển hình: quá thời gian chờ dịch vụ phụ thuộc (nguyên nhân) -> bộ xử lý yêu cầu thất bại (triệu chứng) -> cơ chế tự thử lại làm tăng tải (yếu tố khuếch đại) -> bão hòa vùng chứa kết nối - connection pool (triệu chứng thứ cấp).

## Nguyên nhân điển hình đằng sau các mẫu lỗi phổ biến

- **Quá thời gian chờ (Timeouts) tới một dịch vụ phụ thuộc**: dịch vụ phụ thuộc bị sập hoặc phản hồi chậm, thay đổi về mạng, cấu hình ngưỡng timeout quá thấp sau khi deploy, hoặc cạn kiệt vùng chứa kết nối (connection pool) ở phía bên gọi.
- **Connection pool chạm ngưỡng hoặc gần 100 phần trăm**: các truy vấn chậm hoặc lời gọi dịch vụ hạ nguồn (downstream) chậm làm giữ kết nối quá lâu, rò rỉ kết nối (leak), hoặc lưu lượng truy cập tăng đột biến.
- **Tràn bộ nhớ (Out of memory) và tiến trình bị buộc dừng (killed)**: dữ liệu đầu vào quá lớn, rò rỉ bộ nhớ, giới hạn container bị hạ thấp, hoặc chạy quá nhiều tiến trình worker trên cùng một máy chủ.
- **Từ chối quyền truy cập (Permission denied) / hệ thống tệp chỉ đọc (read-only file system)**: bản deploy làm thay đổi người dùng chạy tiến trình hoặc điểm gắn kết ổ đĩa (volume mount), đầy ổ cứng, hoặc khóa bí mật (secrets) vừa được xoay vòng.
- **Lỗi 429 hoặc bị giới hạn tần suất (throttling)**: một trình khách (client) hoặc tác vụ ngầm đang gọi dồn dập vào một endpoint, hoặc do chính cơn bão thử lại (retry storm) từ hệ thống của bạn.
- **Lỗi SMTP hoặc gửi email thất bại**: thường do chạm giới hạn tần suất hoặc sự cố từ nhà cung cấp dịch vụ email; hiếm khi là sự cố chính trừ khi sản phẩm cốt lõi của bạn là dịch vụ gửi email.

## Các bước kiểm tra tiếp theo ít tốn kém nhất

- Các đợt triển khai (deploys) và thay đổi cấu hình trong vòng 30 phút trước khi xuất hiện mẫu lỗi mới đầu tiên.
- Trang thông báo trạng thái của dịch vụ phụ thuộc cùng bảng điều khiển theo dõi độ trễ và tỷ lệ lỗi của dịch vụ đó.
- Các chỉ số hạ tầng máy chủ ngay tại phút xảy ra đột biến: CPU, bộ nhớ, ổ đĩa, số lượng kết nối đang mở.
- Truy vết trọn vẹn một yêu cầu mẫu từ đầu đến cuối (dựa theo trace id hoặc request id).
- Xem mẫu lỗi đó có tự động biến mất hay không, và điều gì đã thay đổi ngay tại thời điểm đó.

## Cách dùng từ trong báo cáo

- Dùng cụm từ "Nguyên nhân có khả năng cao nhất" (Most likely cause) khi dữ liệu log cộng với trình tự thời gian cùng chỉ về một hướng nhưng chưa có bằng chứng độc lập nào xác nhận.
- Chỉ dùng từ "Đã xác nhận" (Confirmed) khi có bằng chứng độc lập (thông báo sự cố từ nhà cung cấp, hoàn tác bản cập nhật - rollback giúp hết lỗi, hoặc có số liệu đo lường chứng minh).
- Đưa ra con số cụ thể: "30 lỗi quá thời gian chờ thanh toán trong 2 phút, so với 0 lần trong bản baseline".
FILE:templates/triage-report.md
# Báo Cáo Phân Loại Sự Cố Qua Log: <service> <date>

**Khung thời gian:** <start> đến <end> (<time zone>) | **Số bản ghi đã đọc:** <n> | **Từ mức WARN trở lên:** <n> bản ghi thuộc <n> mẫu
**Bản cơ sở (Baseline):** <tệp và khung thời gian, hoặc "none">

## Tóm tắt (3 câu)
<Chuyện gì đã xảy ra, ảnh hưởng tới ai, từ lúc nào và nguyên nhân có khả năng cao nhất là gì, diễn đạt bằng ngôn ngữ dễ hiểu.>

## Bảng xếp hạng các mẫu lỗi

| # | Mức độ | Số lượng | Nhãn cờ | Mẫu lỗi (rút gọn) | Vai trò |
| --- | --- | --- | --- | --- | --- |
| 1 | ERROR | <n> | NEW, SPIKE | <pattern> | ứng viên nguyên nhân gốc rễ / triệu chứng / nhiễu nền |

## Dòng thời gian (Timeline)
- <hh:mm:ss> <mẫu lỗi mới đầu tiên>
- <hh:mm:ss> <sự kiện tiếp theo>
- <hh:mm:ss> <thời điểm phục hồi, hoặc "vẫn đang tiếp diễn ở cuối tệp log">

## Các ứng viên nguyên nhân gốc rễ
1. **<candidate>** - bằng chứng: <dòng log mẫu, số lượng, thời điểm>. Độ tin cậy: <thấp/trung bình/cao>.
   Bước kiểm tra tiếp theo: <một bước kiểm tra cụ thể>.

## Triệu chứng và tác động dây chuyền
- <pattern> gây ra bởi <candidate> vì <lý do>.

## Nhiễu nền (cũng xuất hiện trong bản baseline)
- <pattern> ở tần suất <rate> lần mỗi phút, tương đương với bản baseline.

## Các bước tiếp theo được khuyến nghị
1. <biện pháp giảm thiểu tức thời, nếu có>
2. <bước kiểm tra giúp xác nhận hoặc loại trừ ứng viên chính>
3. <công việc tiếp nối: cải thiện cảnh báo, ngưỡng timeout, cơ chế thử lại hoặc cách ghi log>

## Khoảng trống thông tin
- <các log còn thiếu, chưa rõ múi giờ, các dòng không thể phân tích cú pháp>
FILE:examples/example-checkout-incident.md
# Ví dụ: sự cố tăng đột biến lỗi quá thời gian chờ khi thanh toán đơn hàng

**Người dùng:** Lỗi thanh toán (checkout) tăng vọt vào khoảng 10:09 sáng nay (giờ UTC). Đây là lát cắt 15 phút từ tệp log API và tệp log của ngày hôm qua trong cùng khung giờ đó. Đã có chuyện gì xảy ra?

**Lệnh chạy:**
```bash
python3 scripts/cluster_logs.py app.log --baseline baseline.log
```

**Kết quả đầu ra của kịch bản (đã rút gọn):**
```
Read 138 entries (0 without a timestamp); 60 at WARN or above in 5 patterns.
Time range: 10-09 10:00:00 -> 10-09 10:14:48

 #  LEVEL COUNT  SHARE  FIRST          LAST           FLAGS      TEMPLATE
 1  ERROR    30  50.0%  10-09 10:09:00 10-09 10:10:56 NEW,SPIKE  [<id>] c.s.payments.PaymentClient - payment authorization timed out after <n> for order <n> (provider=psp-eu, attempt=<n>)
 2  ERROR    10  16.7%  10-09 10:09:00 10-09 10:10:48 NEW,SPIKE  [<id>] c.s.checkout.OrderController - checkout failed for session <uuid>
 3  ERROR     3   5.0%  10-09 10:03:30 10-09 10:12:30 -          [<id>] c.s.mail.ReceiptMailer - failed to send receipt to <email>: SMTP <n> try again later
 4  WARN     14  23.3%  10-09 10:00:42 10-09 10:14:42 -          [<id>] c.s.cache.RedisCache - cache miss for key <str> took <n>
 5  WARN      3   5.0%  10-09 10:02:45 10-09 10:11:45 -          [<id>] c.s.db.Hikari - connection pool at <n> (<n>/<n> active)

Details:
 2. ... trace ends: at com.shop.checkout.OrderController.submit(OrderController.java:77)
    spike: 5/min at 10:09 (median 0/min)
```

---

# Báo Cáo Phân Loại Sự Cố Qua Log: checkout API 2026-10-09

**Khung thời gian:** 10:00:00 đến 10:14:48 (UTC) | **Số bản ghi đã đọc:** 138 | **Từ mức WARN trở lên:** 60 bản ghi thuộc 5 mẫu
**Bản cơ sở (Baseline):** baseline.log, 2026-10-08 10:00 đến 10:10 UTC

## Tóm tắt (3 câu)
Từ 10:09:00 đến khoảng 10:11 UTC, các giao dịch thanh toán thẻ bị quá thời gian chờ (timeout) tại nhà cung cấp dịch vụ thanh toán psp-eu khiến khách hàng gặp lỗi thanh toán đơn hàng thất bại. Các lỗi quá thời gian chờ thanh toán này mới hoàn toàn so với ngày hôm qua và đạt đỉnh ở mức 15 lỗi mỗi phút, đồng thời mọi giao dịch checkout thất bại đều mang theo lỗi quá thời gian chờ đọc socket (socket read timeout) từ trình khách thanh toán. Nguyên nhân có khả năng cao nhất là do hệ thống của psp-eu bị chậm hoặc gặp sự cố gián đoạn; không có dấu hiệu nào trong tệp log này chỉ ra lỗi thuộc về mã nguồn hay cơ sở dữ liệu của chúng ta.

## Bảng xếp hạng các mẫu lỗi

| # | Mức độ | Số lượng | Nhãn cờ | Mẫu lỗi (rút gọn) | Vai trò |
| --- | --- | --- | --- | --- | --- |
| 1 | ERROR | 30 | NEW, SPIKE | payment authorization timed out after 5000ms (provider=psp-eu) | ứng viên nguyên nhân gốc rễ |
| 2 | ERROR | 10 | NEW, SPIKE | checkout failed for session ... (SocketTimeoutException) | triệu chứng của mẫu 1 |
| 3 | ERROR | 3 | - | failed to send receipt ... SMTP 421 | nhiễu nền (cũng có trong baseline) |
| 4 | WARN | 14 | - | cache miss for key 'cart:...' | nhiễu nền (cũng có trong baseline) |
| 5 | WARN | 3 | - | connection pool at 85 to 97 percent | cần theo dõi (cũng có trong baseline) |

## Dòng thời gian (Timeline)
- 10:09:00 xuất hiện lỗi quá thời gian chờ cấp phép thanh toán đầu tiên và lỗi checkout thất bại đầu tiên, trong cùng một giây
- 10:09 phút đạt đỉnh: 15 lỗi timeout mỗi phút
- 10:10:56 lỗi timeout thanh toán cuối cùng; không còn lỗi thanh toán nào nữa cho đến cuối tệp log lúc 10:14:48

## Các ứng viên nguyên nhân gốc rễ
1. **Nhà cung cấp thanh toán psp-eu phản hồi chậm hoặc không khả dụng** - bằng chứng: 30 lần quá thời gian chờ sau đúng 5000 ms, tất cả đều thuộc về provider=psp-eu, không có lần nào trong bản baseline; các dấu vết ngăn xếp đều kết thúc tại `PaymentClient.authorize`. Độ tin cậy: trung bình.
   Bước kiểm tra tiếp theo: kiểm tra trang trạng thái của psp-eu và bảng điều khiển độ trễ kết nối gọi ra ngoài của chúng ta trong khoảng 10:08 đến 10:12 UTC.

## Triệu chứng và tác động dây chuyền
- Lỗi "checkout failed for session" gây ra bởi ứng viên 1: bắt đầu cùng một giây, và dấu vết ngăn xếp của nó kết thúc tại `PaymentClient.authorize` thông qua `OrderController.submit`.
- Các lần tự thử lại (attempt=2 và 3 trong các dòng mẫu) có thể đã làm tăng thêm tải trong suốt đợt tăng đột biến.

## Nhiễu nền (cũng xuất hiện trong bản baseline)
- Lỗi gửi biên lai SMTP 421 (3 lần trong 15 phút) và các cảnh báo trượt cache (cache misses) cũng xuất hiện vào ngày hôm qua với tần suất tương đương.
- Các cảnh báo vùng chứa kết nối (connection pool) ở mức 85 đến 97 phần trăm cũng xuất hiện vào ngày hôm qua; đây không phải nguyên nhân sự cố lần này nhưng đã khá sát ngưỡng giới hạn.

## Các bước tiếp theo được khuyến nghị
1. Xác nhận với trang thông báo trạng thái của nhà cung cấp; nếu đúng là sự cố từ phía họ thì không cần hoàn tác bản cập nhật (rollback).
2. Đếm số đơn hàng bị thất bại trong khoảng từ 10:09 đến 10:11 và quyết định xem có nên gửi email hỗ trợ những khách hàng đó hay không.
3. Công việc tiếp nối: thiết lập cảnh báo khi lỗi timeout thanh toán vượt quá 5 lần mỗi phút, giới hạn số lần thử lại kèm thời gian chờ tăng dần (backoff) và kiểm tra lại dung lượng dự phòng của vùng chứa kết nối.

## Khoảng trống thông tin
- Không có log hoặc số liệu đo lường từ phía nhà cung cấp thanh toán; ngưỡng cắt timeout cứng ở mức 5000 ms đã che khuất mất việc nhà cung cấp thực sự bị chậm tới mức nào.
FILE:scripts/cluster_logs.py
#!/usr/bin/env python3
"""Group log lines into error patterns (templates) and rank them for triage.

Usage:
  python3 cluster_logs.py app.log [more.log ...] [options]
  cat app.log | python3 cluster_logs.py - [options]

Options:
  --min-level LEVEL   lowest level to include: DEBUG, INFO, WARN, ERROR (default WARN)
  --top N             show the N largest patterns (default 15)
  --baseline FILE     log from a known-good period; patterns not seen there are marked NEW
  --json              print machine-readable JSON instead of a table

Understands plain lines with an ISO-8601, syslog ("Oct 09 10:01:02") or
nginx ("[09/Oct/2026:10:01:02 +0300]") timestamp, and JSON lines with
level/msg/message/time/timestamp keys, and web server access logs (method,
path and status are kept; 5xx counts as ERROR, 4xx other than 404 as WARN).
Indented lines, "Traceback", "at ...", "Caused by" and bare
"pkg.SomeException: ..." lines are attached to the entry above them.
Lines without an explicit level are rated by wording ("failed", "out of
memory", "timed out" -> ERROR; "retrying", "deprecated" -> WARN).
Variable parts (UUIDs, hex ids, IPs, emails, numbers, quoted values, URL
query strings) are masked so repeats of the same problem group together.
Exit code: 0 no ERROR-level patterns, 1 ERROR or worse found, 2 usage/input error.
Standard library only.
"""
import json
import re
import statistics
import sys
from collections import OrderedDict
from datetime import datetime

LEVELS = {"TRACE": 0, "DEBUG": 0, "INFO": 1, "NOTICE": 1, "WARN": 2, "WARNING": 2,
          "ERROR": 3, "ERR": 3, "SEVERE": 3, "CRITICAL": 4, "CRIT": 4, "FATAL": 4, "PANIC": 4, "ALERT": 4, "EMERG": 4}
CANON = {0: "DEBUG", 1: "INFO", 2: "WARN", 3: "ERROR", 4: "FATAL"}
MONTHS = {m: i for i, m in enumerate(["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"], 1)}

TS_ISO = re.compile(r"(\d{4}-\d{2}-\d{2})[T ](\d{2}:\d{2}:\d{2})(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?")
TS_SYSLOG = re.compile(r"\b(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s+(\d{1,2}) (\d{2}:\d{2}:\d{2})")
TS_NGINX = re.compile(r"\[(\d{2})/(\w{3})/(\d{4}):(\d{2}:\d{2}:\d{2})[^\]]*\]")
LEVEL_RE = re.compile(r"(?<![\w-])(TRACE|DEBUG|INFO|NOTICE|WARNING|WARN|ERROR|ERR|SEVERE|CRITICAL|CRIT|FATAL|PANIC)(?![\w-])", re.I)
ERROR_HINT = re.compile(r"\b(out of memory|oom-?kill\w*|killed process|segfault|panic|fatal|failed|failure|"
                        r"exception|refused|timed out|timeout|denied|unreachable|code=killed)\b", re.I)
WARN_HINT = re.compile(r"\b(deprecated|retrying|retry|slow|degraded|throttl\w*)\b", re.I)
ACCESS_RE = re.compile(r'"(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS) (\S+) HTTP/[\d.]+" (\d{3}) ')
CONT_RE = re.compile(r"^(\s+\S|Traceback \(most recent call last\)|Caused by:|\s*at [\w$.<>]+\(|\s*\.\.\. \d+ more|[\w$]+(?:\.[\w$]+)+(?:Exception|Error)(?::|$))")

MASKS = [
    (re.compile(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", re.I), "<uuid>"),
    (re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"), "<email>"),
    (re.compile(r"\b\d{1,3}(?:\.\d{1,3}){3}(?::\d+)?\b"), "<ip>"),
    (re.compile(r"\b0x[0-9a-f]+\b", re.I), "<hex>"),
    (re.compile(r"\b(?=[0-9a-f]*\d)(?=[0-9a-f]*[a-f])[0-9a-f]{12,}\b", re.I), "<hex>"),
    (re.compile(r"\?[^\s\"']+"), "?<query>"),
    (re.compile(r"\"[^\"]{0,200}\"|'[^']{0,200}'"), "<str>"),
    (re.compile(r"(?<![\w<])[-+]?\d+(?:\.\d+)?(?:ms|s|kb|mb|gb|%)?(?![\w>])", re.I), "<n>"),
]


def usage(msg):
    print(f"error: {msg}\n", file=sys.stderr)
    print(__doc__.strip().split("\n\n")[1], file=sys.stderr)
    sys.exit(2)


def parse_time(line):
    m = TS_ISO.search(line)
    if m:
        return datetime.strptime(f"{m.group(1)} {m.group(2)}", "%Y-%m-%d %H:%M:%S"), m.span()
    m = TS_NGINX.search(line)
    if m and m.group(2) in MONTHS:
        d = datetime(int(m.group(3)), MONTHS[m.group(2)], int(m.group(1)))
        h, mi, s = map(int, m.group(4).split(":"))
        return d.replace(hour=h, minute=mi, second=s), m.span()
    m = TS_SYSLOG.search(line)
    if m:
        h, mi, s = map(int, m.group(3).split(":"))
        return datetime(1900, MONTHS[m.group(1)], int(m.group(2)), h, mi, s), m.span()
    return None, None


def parse_line(line):
    """Return (time, level_num, message) for the first line of an entry."""
    s = line.strip()
    if s.startswith("{"):
        try:
            obj = json.loads(s)
        except ValueError:
            obj = None
        if isinstance(obj, dict):
            lvl = str(obj.get("level") or obj.get("severity") or obj.get("lvl") or "INFO").upper()
            msg = str(obj.get("msg") or obj.get("message") or obj.get("error") or s)
            if obj.get("error") and obj.get("error") != msg:
                msg += f" error={obj['error']}"
            ts = str(obj.get("time") or obj.get("timestamp") or obj.get("ts") or "")
            t, _ = parse_time(ts)
            return t, LEVELS.get(lvl, 1), msg
    t, span = parse_time(s)
    rest = s[span[1]:] if span else s
    acc = ACCESS_RE.search(rest)
    if acc:  # web server access log: keep method, path and status, level from status
        method, path, status = acc.group(1), acc.group(2).split("?")[0], acc.group(3)
        lvl = 3 if status.startswith("5") else 2 if status.startswith("4") and status != "404" else 1
        return t, lvl, f"{method} {path} -> HTTP{status}"
    if span and TS_SYSLOG.match(s[span[0]:span[1]]):
        rest = rest.split(None, 1)[1] if len(rest.split(None, 1)) == 2 else rest  # drop syslog hostname
    m = LEVEL_RE.search(rest[:80])
    if m:
        lvl = LEVELS[m.group(1).upper()]
        rest = rest[:m.start()] + rest[m.end():]
    elif ERROR_HINT.search(rest):
        lvl = 3  # no explicit level, but the wording describes a failure
    elif WARN_HINT.search(rest):
        lvl = 2
    else:
        lvl = 1
    rest = re.sub(r":\s*:", ":", re.sub(r"^[\s:|-]+", "", rest))
    return t, lvl, rest


def template(msg):
    first = msg.split("\n", 1)[0]
    first = re.sub(r"\[[\w.:/-]*\d[\w.:/-]*\]", "[<id>]", first)  # [thread-12], [req-ab12]
    for rx, repl in MASKS:
        first = rx.sub(repl, first)
    return re.sub(r"\s+", " ", first).strip()[:160]


def read_entries(paths):
    entries = []
    for p in paths:
        try:
            fh = sys.stdin if p == "-" else open(p, encoding="utf-8", errors="replace")
        except OSError as e:
            usage(str(e))
        with fh:
            for raw in fh:
                line = raw.rstrip("\n")
                if not line.strip():
                    continue
                if entries and CONT_RE.match(line):
                    entries[-1]["extra"] += 1
                    entries[-1]["trace_tail"] = line.strip()
                    continue
                t, lvl, msg = parse_line(line)
                entries.append({"time": t, "level": lvl, "msg": msg, "extra": 0, "trace_tail": ""})
    return entries


def cluster(entries, min_level):
    groups = OrderedDict()
    for e in entries:
        if e["level"] < min_level:
            continue
        key = template(e["msg"])
        g = groups.setdefault(key, {"template": key, "count": 0, "level": 0, "first": None, "last": None,
                                    "sample": e["msg"].split("\n", 1)[0][:200], "trace_tail": "", "minutes": {}})
        g["count"] += 1
        g["level"] = max(g["level"], e["level"])
        if e["trace_tail"] and not g["trace_tail"]:
            g["trace_tail"] = e["trace_tail"][:160]
        if e["time"]:
            g["first"] = min(g["first"] or e["time"], e["time"])
            g["last"] = max(g["last"] or e["time"], e["time"])
            k = e["time"].strftime("%Y-%m-%d %H:%M")
            g["minutes"][k] = g["minutes"].get(k, 0) + 1
    return list(groups.values())


def spike(g, all_minutes):
    """Peak minute vs median of this pattern's per-minute counts over the whole time range."""
    if g["count"] < 5 or len(all_minutes) < 3:
        return None
    series = [g["minutes"].get(m, 0) for m in all_minutes]
    peak = max(series)
    med = statistics.median(series)
    if peak >= 5 and peak >= 3 * max(med, 1):
        at = all_minutes[series.index(peak)]
        return f"{peak}/min at {at[11:]} (median {med:g}/min)"
    return None


def fmt(t):
    return t.strftime("%m-%d %H:%M:%S") if t and t.year != 1900 else (t.strftime("%b %d %H:%M:%S") if t else "-")


def main(argv):
    args, paths = {"min": "WARN", "top": 15, "baseline": None, "json": False}, []
    it = iter(argv)
    for a in it:
        if a == "--min-level":
            args["min"] = next(it, "").upper()
        elif a == "--top":
            v = next(it, "")
            if not v.isdigit():
                usage("--top needs a number")
            args["top"] = int(v)
        elif a == "--baseline":
            args["baseline"] = next(it, None)
        elif a == "--json":
            args["json"] = True
        elif a.startswith("--"):
            usage(f"unknown option {a}")
        else:
            paths.append(a)
    if not paths:
        usage("give at least one log file, or - for stdin")
    if args["min"] not in LEVELS:
        usage(f"unknown level {args['min']}")
    min_level = LEVELS[args["min"]]

    entries = read_entries(paths)
    if not entries:
        print("error: no log lines found", file=sys.stderr)
        return 2
    groups = cluster(entries, min_level)
    known = None
    if args["baseline"]:
        known = {g["template"] for g in cluster(read_entries([args["baseline"]]), 0)}
    times = sorted(e["time"] for e in entries if e["time"])
    all_minutes = []
    if times:
        cur = times[0].replace(second=0)
        while cur <= times[-1] and len(all_minutes) < 10000:
            all_minutes.append(cur.strftime("%Y-%m-%d %H:%M"))
            cur = cur.fromtimestamp(cur.timestamp() + 60)
    for g in groups:
        g["new"] = known is not None and g["template"] not in known
        g["spike"] = spike(g, all_minutes)
    groups.sort(key=lambda g: (-g["level"], -g["count"]))
    shown = groups[: args["top"]]
    total = sum(g["count"] for g in groups)
    worst = max((g["level"] for g in groups), default=0)

    if args["json"]:
        out = {"entries_read": len(entries), "entries_at_or_above_min_level": total, "patterns": len(groups),
               "time_range": [fmt(times[0]), fmt(times[-1])] if times else None,
               "patterns_top": [{"level": CANON[g["level"]], "count": g["count"], "share_pct": round(100 * g["count"] / total, 1),
                                 "first": fmt(g["first"]), "last": fmt(g["last"]), "new": g["new"], "spike": g["spike"],
                                 "template": g["template"], "sample": g["sample"], "trace_tail": g["trace_tail"]} for g in shown]}
        print(json.dumps(out, indent=2))
        return 1 if worst >= 3 else 0

    unparsed = sum(1 for e in entries if e["time"] is None)
    print(f"Read {len(entries)} entries ({unparsed} without a timestamp); "
          f"{total} at {args['min']} or above in {len(groups)} patterns.")
    if times:
        print(f"Time range: {fmt(times[0])} -> {fmt(times[-1])}")
    if not groups:
        print("No entries at or above the minimum level.")
        return 0
    print()
    print(f"{'#':>2}  {'LEVEL':<5} {'COUNT':>5} {'SHARE':>6}  {'FIRST':<14} {'LAST':<14} FLAGS      TEMPLATE")
    for i, g in enumerate(shown, 1):
        flags = ",".join(f for f in ("NEW" if g["new"] else "", "SPIKE" if g["spike"] else "") if f) or "-"
        print(f"{i:>2}  {CANON[g['level']]:<5} {g['count']:>5} {100 * g['count'] / total:>5.1f}%  "
              f"{fmt(g['first']):<14} {fmt(g['last']):<14} {flags:<10} {g['template']}")
    print("\nDetails:")
    for i, g in enumerate(shown, 1):
        print(f"{i:>2}. sample: {g['sample']}")
        if g["trace_tail"]:
            print(f"    trace ends: {g['trace_tail']}")
        if g["spike"]:
            print(f"    spike: {g['spike']}")
    if len(groups) > len(shown):
        print(f"\n({len(groups) - len(shown)} smaller patterns not shown; use --top)")
    return 1 if worst >= 3 else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

role: >
  Bạn là một kỹ sư độ tin cậy hệ thống cấp cao (Senior Site Reliability Engineer - SRE) đã từng biên soạn và duy trì
  tài liệu hướng dẫn trực ca (on-call runbooks) cho các dịch vụ web, API, đường ống dữ liệu (data pipelines) và các tiến trình nền (background
  workers). Bạn viết runbook dành cho một kỹ sư đang mệt mỏi bị gọi dậy lúc 3 giờ sáng: các bước ngắn gọn,
  có thứ tự rõ ràng, chỉ đích danh nơi cần kiểm tra, ưu tiên các biện pháp giảm thiểu rủi ro an toàn trước tiên, và các quy tắc
  rõ ràng về thời điểm cần chuyển cấp hỗ trợ (escalate). Bạn không bao giờ tự bịa ra tên dashboard, câu lệnh hoặc
  đầu mối liên hệ mà người dùng không cung cấp; thay vào đó, bạn để lại các nhãn giữ chỗ (placeholders) được đánh dấu rõ ràng.

task: >
  Xây dựng bộ khung tài liệu trực ca (on-call runbook scaffold) cho dịch vụ được mô tả bên dưới:
  các cấp độ nghiêm trọng, tín hiệu phát hiện và cảnh báo, luồng phân loại xử lý ban đầu (triage flow), các kịch bản
  giảm thiểu (mitigation playbooks) cho từng sự cố phổ biến, quy trình quay lui (rollback), các mẫu thông báo
  giao tiếp, lộ trình chuyển cấp (escalation path), và danh sách kiểm tra sau sự cố, sẵn sàng để
  dán vào tài liệu của nhóm và điền thông tin chi tiết vào.

inputs:
  service_name: "checkout-api"
  what_it_does: "handles cart checkout and payment authorization for the web shop and mobile app"
  architecture: "stateless Node.js service on Kubernetes, 6 pods, PostgreSQL primary with one replica, Redis cache, calls an external payment provider"
  traffic_and_slo: "about 40 requests per second at peak, 99.9 percent availability and p95 latency under 400 ms"
  common_incidents: "payment provider timeouts, database connection pool exhaustion, bad deploys, Redis eviction storms, certificate expiry"
  dashboards_and_tools: "Grafana dashboard 'Checkout Overview', Loki logs, PagerDuty, Argo CD for deploys, status page on Statuspage"
  deploy_and_rollback: "Argo CD sync from main, roughly 10 deploys a week, feature flags in LaunchDarkly"
  escalation_contacts: "primary on-call (payments team), secondary on-call, payments team lead, database on-call from platform team, payment provider support line"
  comms_channels: "#inc-checkout Slack channel, customer support lead, public status page"
  business_hours_and_timezone: "Europe/Berlin, 24x7 on-call, weekly rotation"
  known_gaps: "no alert on certificate expiry yet, replica lag is not graphed"

instructions:
  - Nêu rõ các giả định ở những nơi thiếu thông tin đầu vào thay vì đặt câu hỏi ngược lại.
  - Định nghĩa 4 cấp độ nghiêm trọng (SEV1 đến SEV4) bao gồm mức độ ảnh hưởng đến khách hàng, ví dụ cụ thể đối với dịch vụ này, thời gian phản hồi, ai sẽ nhận cảnh báo gọi trực (paged), và việc trang trạng thái (status page) có cần cập nhật hay không.
  - Đối với khâu phát hiện sự cố, hãy liệt kê từng cảnh báo hoặc tín hiệu kèm theo mã giữ chỗ cho câu truy vấn metric hoặc log, gợi ý ngưỡng cảnh báo gắn liền với SLO, các nguyên nhân có khả năng cao, và bảng điều khiển (dashboard panel) đầu tiên cần mở lên.
  - Viết một luồng phân loại xử lý chung (generic triage flow) gồm tối đa 10 bước theo thứ tự cho 15 phút đầu tiên, bắt đầu từ việc xác nhận đã nhận cảnh báo (acknowledging the page) và xác nhận phạm vi ảnh hưởng.
  - Viết một kịch bản giảm thiểu (mitigation playbook) cho mỗi sự cố thường gặp gồm: triệu chứng, các bước kiểm tra nhanh, các biện pháp giảm thiểu an toàn được sắp xếp theo thứ tự từ ít rủi ro nhất đến nhiều rủi ro nhất, khi nào cần chuyển cấp (escalate), và cách xác nhận hệ thống đã phục hồi.
  - Viết quy trình quay lui (rollback procedure) cho một bản triển khai (deploy) bị lỗi, bao gồm cách tìm phiên bản hoạt động tốt gần nhất, cách thực hiện rollback, cách tắt cờ tính năng (feature flag), và những gì cần kiểm tra xác minh sau đó.
  - Cung cấp các mẫu thông báo ngắn gọn để mở đầu kênh xử lý sự cố (incident channel kickoff), bản cập nhật mỗi 30 phút, thông báo trên trang trạng thái (đang điều tra, đã xác định nguyên nhân, đang theo dõi, đã giải quyết), và ghi chú gửi cho bộ phận hỗ trợ khách hàng.
  - Đưa ra lộ trình chuyển cấp (escalation path) kèm theo điều kiện kích hoạt và giới hạn thời gian, sử dụng các vai trò mà tôi đã cung cấp.
  - Bổ sung danh sách kiểm tra sau sự cố (dòng thời gian, số liệu ảnh hưởng, các ticket theo dõi tiếp theo, tinh chỉnh cảnh báo, cập nhật runbook) và chương trình họp rút kinh nghiệm không đổ lỗi (blameless review agenda).
  - Chuyển các lỗ hổng đã biết (known gaps) thành các công việc cần theo dõi tiếp theo (follow-up tasks) kèm theo vai trò người phụ trách.
  - Sử dụng các dấu <PLACEHOLDER> cho mọi đường dẫn liên kết, câu truy vấn, câu lệnh hoặc tên gọi mà nhóm phải tự điền vào; không bao giờ tự bịa ra chúng.

output_format: Chỉ xuất YAML, không có văn xuôi bên ngoài YAML, sử dụng chính xác cấu trúc này

output_schema:
  assumptions: [string]
  service:
    name: string
    purpose: string
    dependencies: [string]
    slo: string
    owners: string
  severity_levels:
    - level: SEV1 | SEV2 | SEV3 | SEV4
      customer_impact: string
      examples: [string]
      response_time: string
      page: [string]
      status_page_update: boolean
  detection_signals:
    - name: string
      signal: string
      query_placeholder: string
      threshold_idea: string
      likely_causes: [string]
      first_panel_to_open: string
  triage_steps:
    - step: number
      action: string
      look_at: string
      decision: string
  playbooks:
    - incident: string
      symptoms: [string]
      quick_checks: [string]
      mitigations:
        - order: number
          action: string
          risk: low | medium | high
          command_placeholder: string
      escalate_when: string
      confirm_recovery: [string]
  rollback:
    find_last_good_version: [string]
    steps: [string]
    feature_flag_kill_switch: [string]
    verify_after: [string]
  comms_templates:
    incident_channel_kickoff: string
    update_every_30_min: string
    status_page:
      investigating: string
      identified: string
      monitoring: string
      resolved: string
    customer_support_note: string
  escalation_path:
    - role: string
      when: string
      after_minutes: number
      how_to_reach: string
  post_incident_checklist: [string]
  blameless_review_agenda: [string]
  follow_up_tasks:
    - task: string
      owner_role: string
      priority: high | medium | low

---
name: time-series-gap-anomaly-checker
description: Kiểm tra dữ liệu chuỗi thời gian dạng CSV (số liệu đo lường, chỉ số cảm biến, doanh số theo giờ) để phát hiện các vấn đề dữ liệu trước khi vẽ biểu đồ hoặc thiết lập cảnh báo: thiếu khoảng thời gian, dấu thời gian (timestamp) bị trùng lặp hoặc sai thứ tự, giá trị trống, đường đi ngang (flatlines) do trình xuất dữ liệu bị treo, các điểm đột biến (spikes) phát hiện qua trung vị/MAD bền vững và hiện tượng đặt lại bộ đếm (counter resets). Sử dụng khi bảng điều khiển (dashboard) có dấu hiệu bất thường, trước khi huấn luyện mô hình dự báo hoặc khi cần trả lời câu hỏi "dữ liệu này đã đầy đủ chưa?". Bao gồm công cụ kiểm tra bằng thư viện chuẩn Python có hỗ trợ cờ --self-test.
---

# Công cụ Kiểm tra Khoảng trống và Điểm bất thường trong Chuỗi thời gian

Bạn giúp các chuyên viên phân tích, kỹ sư SRE và kỹ sư dữ liệu quyết định xem một chuỗi thời gian có đáng tin cậy hay không. Bạn tìm ra những vị trí dữ liệu bị thiếu, lặp lại, bị treo cứng hoặc đáng ngờ, giải thích nguyên nhân khả dĩ của từng vấn đề, đồng thời tách biệt các sự cố thu thập dữ liệu (trên đường ống pipeline) khỏi các sự kiện thực tế (trên hệ thống đang được đo lường). Bạn không bao giờ âm thầm tự lấp đầy các khoảng trống hoặc xóa bỏ các điểm ngoại lai (outliers); bạn đề xuất phương án xử lý và nêu rõ phương án đó làm thay đổi điều gì.

## Các tệp trong kỹ năng này

- `scripts/ts_check.py` - đọc tệp CSV, suy luận khoảng thời gian lấy mẫu (sampling interval) và báo cáo các khoảng trống, dữ liệu trùng lặp, các dòng sai thứ tự, giá trị bị thiếu, đường đi ngang (flatlines), điểm đột biến (spikes) và hiện tượng đặt lại bộ đếm dưới dạng markdown hoặc JSON (chỉ sử dụng thư viện chuẩn Python 3)
- `references/anomaly-methods.md` - cách thức hoạt động của các phép kiểm tra, cách tinh chỉnh ngưỡng (thresholds) và những trường hợp mỗi phương pháp không phát huy tác dụng
- `references/gap-handling-playbook.md` - cách xử lý cho từng loại vấn đề: điền giá trị (fill), nội suy (interpolate), loại bỏ (drop) hoặc ghi chú (annotate)
- `templates/data-quality-report.md` - mẫu báo cáo phục vụ việc sửa lỗi dashboard, đánh giá sự cố hoặc bàn giao bộ dữ liệu
- `examples/api-metrics.csv` - dữ liệu số liệu API mỗi 5 phút trong suốt ba giờ, bao gồm một đợt gián đoạn thu thập dữ liệu, một dòng trùng lặp, một giá trị bị treo, một điểm đột biến độ trễ và một lần đặt lại bộ đếm

## Quy trình thực hiện

### 1. Tìm hiểu chuỗi dữ liệu
Hỏi xem mỗi cột đo lường đại lượng gì, đơn vị tính, khoảng thời gian lấy mẫu kỳ vọng là bao nhiêu, và liệu có cột nào là bộ đếm đơn điệu (monotonic counter) hay không (ví dụ: requests_total, bytes_sent). Hỏi xem các dấu thời gian không kèm múi giờ (naive timestamps) đang sử dụng múi giờ nào; kịch bản mặc định coi chúng là giờ UTC.

### 2. Chạy công cụ kiểm tra
```
python3 scripts/ts_check.py examples/api-metrics.csv --counter requests_total
python3 scripts/ts_check.py data.csv --time-col ts --cols temp_c,humidity --interval 10m
python3 scripts/ts_check.py data.csv --z 5 --window 12 --flatline-points 12 --json > report.json
python3 scripts/ts_check.py --self-test
```
Dấu thời gian có thể ở định dạng ISO 8601 hoặc thời gian Unix epoch tính bằng giây hoặc mili-giây. Trả về mã thoát (exit code) là 1 khi phát hiện bất kỳ vấn đề nào, và 2 khi đầu vào không hợp lệ, nhờ đó có thể dùng làm cổng kiểm soát (gate) trong đường ống dữ liệu.

### 3. Đọc kết quả
- **Độ đầy đủ (Completeness)** = số dòng hiện có / số dòng kỳ vọng trên toàn dải thời gian. Mức dưới 95% thường đồng nghĩa với việc các con số tổng hợp và trung bình đã bị sai lệch.
- **gap**: khoảng cách giữa các điểm vượt quá 1.5 lần khoảng lấy mẫu. Các khoảng trống dài xuất hiện vào những khung giờ cố định thường là do quá trình triển khai (deploy) hoặc tác vụ định kỳ (cron job); một khoảng trống dài đơn lẻ thường do bộ thu thập dữ liệu (collector) bị ngừng hoạt động.
- **duplicate / out_of_order**: thu thập dữ liệu bị đúp, cơ chế thử lại (retries) hoặc do gộp nhiều tệp với nhau.
- **flatline**: cùng một giá trị lặp đi lặp lại; đây là điều bình thường đối với tín hiệu bật/tắt (on/off), nhưng lại rất đáng ngờ đối với các phép đo vốn có độ nhiễu tự nhiên.
- **spike**: điểm z-score bền vững (robust z-score) so với trung vị trượt (rolling median). Hãy kiểm tra xem các cột lân cận có tăng đột biến cùng lúc hay không (nếu có thì là sự kiện thực tế, nếu không thì là lỗi nhiễu dữ liệu).
- **reset**: bộ đếm bị sụt giảm giá trị, thường do tiến trình khởi động lại; các phép tính tốc độ thay đổi (rate) bắt buộc phải xử lý trường hợp này.

### 4. Đề xuất phương án xử lý
Sử dụng `references/gap-handling-playbook.md`. Tinh chỉnh các ngưỡng bằng `references/anomaly-methods.md` nếu dữ liệu có tính mùa vụ, biến động theo từng đợt (bursty) hoặc rất thưa thớt.

### 5. Báo cáo
Điền vào mẫu `templates/data-quality-report.md` với các con số tóm tắt, những vấn đề hàng đầu kèm theo thời gian theo múi giờ của người dùng, nguyên nhân nghi ngờ và phương án xử lý đã thống nhất.

## Quy tắc

- Chỉ báo cáo các vấn đề; không tự ý sửa đổi dữ liệu nguồn. Ghi dữ liệu đã làm sạch ra một tệp mới và ghi lại tài liệu cho mọi thay đổi đã thực hiện.
- Không kết luận một điểm đột biến (spike) là sự cố hệ thống nếu chưa có bằng chứng củng cố (từ các số liệu khác, nhật ký hệ thống hoặc xác nhận của con người).
- Nêu rõ khoảng lấy mẫu và các ngưỡng bạn đã sử dụng để kết quả có thể tái lập lại được.
- Chuyển đổi giờ UTC sang múi giờ của người dùng khi trình bày kết quả.
FILE:references/anomaly-methods.md
# Phương pháp Phát hiện Bất thường và Tinh chỉnh Tham số

## Suy luận khoảng thời gian lấy mẫu
Khoảng thời gian lấy mẫu kỳ vọng là trung vị của khoảng cách giữa các dấu thời gian duy nhất đã được sắp xếp tăng dần. Cách tính này vẫn chính xác ngay cả khi có một vài khoảng trống và độ trễ dao động nhỏ (jitter). Hãy truyền tham số `--interval` khi chuỗi dữ liệu phần lớn là khoảng trống (dữ liệu sự kiện thưa thớt) hoặc khi bạn đã biết chắc chắn chu kỳ thu thập dữ liệu.

Khoảng trống (gap) là bất kỳ khoảng cách nào lớn hơn 1.5 lần khoảng lấy mẫu; vì vậy với dữ liệu 5 phút một lần, khoảng cách 7 phút không bị coi là khoảng trống nhưng khoảng cách 10 phút thì được tính là một khoảng trống (thiếu 1 điểm dữ liệu).

## Phát hiện điểm đột biến bền vững (trung vị trượt + MAD)
Đối với mỗi điểm, lấy tối đa `--window` điểm lân cận ở mỗi bên (không tính chính điểm đang xét), tính trung vị (median) và độ lệch tuyệt đối trung vị (MAD - median absolute deviation) của chúng. Điểm z-score bền vững được tính theo công thức:

    z = |value - median| / (1.4826 * MAD)

Hệ số 1.4826 giúp quy đổi MAD tương đương với độ lệch chuẩn đối với dữ liệu phân phối chuẩn. Ngưỡng mặc định bằng 4 tương ứng với xác suất khoảng 1 trên 15.000 đối với nhiễu phân phối chuẩn. Khi có hơn một nửa số điểm lân cận mang giá trị giống hệt nhau (MAD = 0), độ lệch tuyệt đối trung bình (mean absolute deviation) sẽ được sử dụng thay thế, và nếu toàn bộ vùng lân cận hoàn toàn phẳng thì lấy bằng 1% mức giá trị đó.

Tại sao không dùng giá trị trung bình (mean) và độ lệch chuẩn (standard deviation)? Chỉ cần một điểm đột biến lớn cũng đủ làm phình đại độ lệch chuẩn và khiến chính điểm đột biến đó bị che khuất. Ngược lại, trung vị và MAD hoàn toàn miễn nhiễm với hiện tượng này.

## Tinh chỉnh tham số
| Loại dữ liệu | Gợi ý cấu hình |
|---|---|
| Số liệu nhiều nhiễu (độ trễ, CPU) | `--z 5`, `--window 12` |
| Có tính mùa vụ theo ngày hoặc tuần | Kiểm tra từng chu kỳ mùa vụ một, hoặc so sánh với cùng khung giờ tuần trước trước khi tin vào các điểm đột biến |
| Dữ liệu kinh doanh tăng vọt theo đợt (đơn hàng) | Tăng `--z` cao hơn (6-8); các điểm vọt lên thường là sự kiện thật |
| Giá trị số nguyên hoặc độ phân giải thấp | Thường gặp MAD = 0; hãy tăng `--z` hoặc tổng hợp (aggregate) dữ liệu trước |
| Tín hiệu bật/tắt hoặc trạng thái | Bỏ qua cảnh báo flatline; chỉ kiểm tra các khoảng trống (gaps) |
| Bộ đếm (Counters) | Luôn truyền cờ `--counter`; kiểm tra tốc độ biến thiên (rates) sau khi đã xử lý các điểm reset |

## Các giới hạn đã biết
- Sự trượt giá trị từ từ (gradual drifts) và sự dịch chuyển nền giá trị (level shifts) không phải là điểm đột biến ngắn hạn; hãy vẽ biểu đồ chuỗi dữ liệu hoặc so sánh trung vị giữa các tuần.
- Một vài điểm đầu tiên và cuối cùng có ít điểm lân cận hơn, do đó điểm số của chúng kém tin cậy hơn.
- Các luồng sự kiện không đều đặn (lượt đăng nhập, lượt mua hàng) không có khoảng lấy mẫu cố định; hãy gom nhóm (aggregate) chúng vào các khung thời gian (buckets) trước.
FILE:references/gap-handling-playbook.md
# Cẩm nang Xử lý Khoảng trống và Điểm bất thường

| Vấn đề | Nguyên nhân khả dĩ | Phương án xử lý mặc định | Điều cần tránh |
|---|---|---|---|
| Khoảng trống ngắn (1-2 điểm) | quá thời gian thu thập (scrape timeout), chập chờn mạng thoáng qua | Nội suy tuyến tính cho các tín hiệu mượt; giữ nguyên giá trị null đối với số đếm | Điền số 0 vào các cột số đếm |
| Khoảng trống dài | trình thu thập ngừng chạy, đang triển khai hệ thống, máy chủ tắt | Giữ nguyên null và ghi chú lên biểu đồ; loại trừ khỏi phép tính SLO hoặc tính là trạng thái chưa xác định (unknown) | Nội suy bắc cầu qua nhiều giờ liền |
| Dấu thời gian trùng lặp | thu thập đúp, chạy lại lô dữ liệu (retried batch), gộp nhiều tệp | Giữ lại một dòng duy nhất (dòng đầu tiên hoặc dòng có nhiều trường dữ liệu nhất); kiểm tra xem các giá trị có khác nhau không | Cộng gộp các dòng trùng lặp |
| Dòng dữ liệu sai thứ tự thời gian | dữ liệu đến muộn, nối ghép các tệp | Sắp xếp lại theo thời gian trước khi thực hiện bất kỳ phép tính trượt (windowed calculation) nào | Mặc định thứ tự dòng trong tệp là thứ tự thời gian |
| Giá trị bị thiếu (Missing values) | cảm biến mất tín hiệu tạm thời, lỗi phân tích cú pháp, chuỗi "N/A" | Sửa lỗi bộ phân tích cú pháp (parser) trước; sau đó xử lý như một khoảng trống | Coi ô trống là giá trị 0 |
| Đường đi ngang (Flatline) | trình xuất dữ liệu bị treo, giá trị lấy từ bộ nhớ đệm, cảm biến bị đơ | Xác nhận chéo với nguồn dữ liệu thứ hai; đánh dấu là đáng ngờ | Tin tưởng dữ liệu này để kích hoạt cảnh báo |
| Điểm đột biến (Spike đơn lẻ) | lỗi nhiễu thoáng qua, sai đơn vị đo, đợt tăng tải thực tế | Kiểm tra các cột liên quan và nhật ký hệ thống; chỉ giới hạn trần (cap) hoặc loại bỏ khi có lý do được ghi chép rõ ràng | Xóa bỏ mà không để lại ghi chú |
| Đặt lại bộ đếm (Counter reset) | tiến trình khởi động lại, tràn bộ đếm (overflow) | Tính tốc độ (rate) theo từng phân đoạn (tốc độ sau khi reset = giá trị mới / khoảng thời gian) | Để xuất hiện tốc độ âm |

## Trước khi xây dựng mô hình hoặc dự báo
1. Sắp xếp, khử trùng lặp và chuẩn hóa về đúng khoảng thời gian lấy mẫu kỳ vọng (chèn giá trị null vào các vị trí bị khuyết).
2. Quyết định chính sách xử lý khoảng trống cho từng cột và ghi nhận lại.
3. Duy trì thêm một cột đánh dấu mặt nạ (`is_imputed`) để mô hình và người đọc biết những vị trí nào đã được điền dữ liệu nhân tạo.
4. Chạy lại công cụ kiểm tra trên tệp đã làm sạch và đính kèm cả hai bản báo cáo.

## Khi nào cần báo cáo nâng cấp độ ưu tiên (Escalate)
- Độ đầy đủ (Completeness) dưới 90% đối với một chỉ số được dùng để tính SLO hoặc lập hóa đơn thanh toán.
- Các khoảng trống lặp đi lặp lại vào cùng một thời điểm mỗi ngày (có một tác vụ ngầm nào đó đang làm sập bộ thu thập dữ liệu).
- Các điểm đột biến xuất hiện đồng loạt trên tất cả các máy chủ cùng một lúc (đây là sự cố hệ thống thực sự, không phải lỗi dữ liệu).
FILE:templates/data-quality-report.md
# Chất lượng Dữ liệu Chuỗi thời gian - <tên bộ dữ liệu hoặc dashboard>

**Nguồn:** <tệp / truy vấn>  **Phạm vi:** <bắt đầu> đến <kết thúc> (<múi giờ>)  **Khoảng lấy mẫu:** <kỳ vọng>
**Số dòng:** <n>  **Độ đầy đủ:** <phần trăm>  **Ngưỡng áp dụng:** z=<z>, window=<w>, flatline=<n>

## Tóm tắt
<hai câu: dữ liệu này có đáng tin cậy cho mục đích sử dụng dự kiến hay không, và điều gì cần phải thay đổi trước tiên>

## Các vấn đề phát hiện
| Loại vấn đề | Thời điểm (<múi giờ>) | Cột | Chi tiết | Nguyên nhân nghi ngờ | Phương án xử lý |
|---|---|---|---|---|---|
| | | | | | |

## Các quyết định xử lý
- Chính sách xử lý khoảng trống: <nội suy các khoảng trống ngắn / giữ nguyên null / loại trừ>
- Chính sách xử lý điểm ngoại lai: <giữ nguyên / giới hạn trần / loại bỏ kèm ghi chú>
- Xử lý bộ đếm: <tính tốc độ theo phân đoạn tại các điểm reset>

## Công việc tiếp theo
- [ ] <sửa bộ thu thập dữ liệu / bộ phân tích cú pháp / cảnh báo> - người phụ trách <tên> - hạn hoàn thành <ngày>

Được tạo bằng scripts/ts_check.py. Dữ liệu nguồn không bị sửa đổi.
FILE:examples/api-metrics.csv
timestamp,p95_latency_ms,error_rate,requests_total
2026-10-18T22:00:00Z,182,0.4,1200
2026-10-18T22:05:00Z,176,0.3,2435
2026-10-18T22:10:00Z,190,0.5,3705
2026-10-18T22:15:00Z,185,0.4,5010
2026-10-18T22:20:00Z,179,0.3,6210
2026-10-18T22:25:00Z,188,0.4,7445
2026-10-18T22:30:00Z,181,0.5,8715
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:40:00Z,177,0.3,11220
2026-10-18T22:45:00Z,184,0.4,12455
2026-10-18T22:50:00Z,186,0.4,13725
2026-10-18T22:55:00Z,180,0.4,15030
2026-10-18T23:00:00Z,191,0.4,16230
2026-10-18T23:05:00Z,,0.4,17465
2026-10-18T23:10:00Z,178,0.4,18735
2026-10-18T23:15:00Z,187,0.4,20040
2026-10-18T23:20:00Z,182,0.4,21240
2026-10-18T23:25:00Z,189,0.4,22475
2026-10-18T23:30:00Z,612,2.1,23745
2026-10-18T23:35:00Z,185,0.5,25050
2026-10-18T23:40:00Z,180,0.3,26250
2026-10-18T23:45:00Z,184,0.4,27485
2026-10-19T00:10:00Z,188,0.5,28755
2026-10-19T00:15:00Z,183,0.4,30060
2026-10-19T00:20:00Z,177,0.3,31260
2026-10-19T00:25:00Z,185,0.4,32495
2026-10-19T00:30:00Z,182,0.5,1180
2026-10-19T00:35:00Z,189,0.4,2485
2026-10-19T00:40:00Z,184,0.3,3685
2026-10-19T00:45:00Z,180,0.4,4920
2026-10-19T00:50:00Z,186,0.5,6190
2026-10-19T00:55:00Z,183,0.4,7495
FILE:scripts/ts_check.py
#!/usr/bin/env python3
"""Time series gap and anomaly checker for CSV metrics (Python 3 standard library only).

Reads a CSV with a timestamp column and one or more numeric columns, infers the sampling
interval (median spacing) unless --interval is given, and reports:
  gap          missing samples (spacing > 1.5 x interval), with how many points are missing
  duplicate    repeated timestamps
  out_of_order timestamps that go backwards
  missing      empty, NaN, or non-numeric values
  flatline     the same value repeated for >= --flatline-points samples (stuck sensor/exporter)
  spike        robust outliers: |value - rolling median| / (1.4826 * MAD) >= --z (default 4);
               if MAD is 0 the mean absolute deviation is used instead
  reset        a counter column that decreases (--counter col1,col2)

Timestamps: ISO 8601 (with Z or offset; naive = UTC) or Unix epoch seconds/milliseconds.

Usage:
  python3 ts_check.py metrics.csv
  python3 ts_check.py metrics.csv --time-col ts --cols cpu,latency_ms --interval 5m --z 5
  python3 ts_check.py metrics.csv --counter requests_total --json
  python3 ts_check.py --self-test
Exit code: 0 clean, 1 issues found, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import re
import statistics
import sys


class InputError(ValueError):
    pass


def parse_ts(raw):
    s = (raw or "").strip()
    if not s:
        raise InputError("empty timestamp")
    if re.fullmatch(r"-?\d+(\.\d+)?", s):
        v = float(s)
        return v / 1000.0 if v > 1e11 else v
    try:
        t = dt.datetime.fromisoformat(s.replace("Z", "+00:00").replace(" ", "T", 1))
    except ValueError:
        raise InputError(f"unparseable timestamp {s!r}")
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.timestamp()


def parse_interval(s):
    m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)?", s.strip())
    if not m:
        raise InputError(f"bad interval {s!r}; use e.g. 30s, 5m, 1h")
    mult = {"ms": 0.001, "s": 1, None: 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]
    return float(m.group(1)) * mult


def fmt_ts(ts):
    return dt.datetime.fromtimestamp(ts, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def fmt_dur(sec):
    for unit, size in (("d", 86400), ("h", 3600), ("m", 60)):
        if sec >= size:
            v = sec / size
            return f"{v:.0f}{unit}" if v == int(v) else f"{v:.1f}{unit}"
    return f"{sec:g}s"


def to_float(raw):
    s = (raw or "").strip()
    if s == "" or s.lower() in ("nan", "null", "none", "na", "n/a", "-"):
        return None
    try:
        v = float(s)
    except ValueError:
        return None
    return None if math.isnan(v) or math.isinf(v) else v


def load(text, time_col=None, cols=None):
    reader = csv.DictReader(io.StringIO(text))
    if not reader.fieldnames:
        raise InputError("CSV has no header row")
    fields = [f.strip() for f in reader.fieldnames]
    reader.fieldnames = fields
    if time_col is None:
        cands = [f for f in fields if f.lower() in ("timestamp", "time", "ts", "date", "datetime")]
        time_col = cands[0] if cands else fields[0]
    if time_col not in fields:
        raise InputError(f"time column {time_col!r} not in header {fields}")
    value_cols = cols or [f for f in fields if f != time_col]
    for c in value_cols:
        if c not in fields:
            raise InputError(f"column {c!r} not in header {fields}")
    rows = []
    for i, r in enumerate(reader, start=2):
        rows.append({"line": i, "ts": parse_ts(r.get(time_col)),
                     "raw": {c: r.get(c) for c in value_cols}})
    if len(rows) < 3:
        raise InputError("need at least 3 data rows")
    return time_col, value_cols, rows


def rolling_robust(values, idx, window):
    lo, hi = max(0, idx - window), min(len(values), idx + window + 1)
    neigh = [values[j] for j in range(lo, hi) if j != idx and values[j] is not None]
    if len(neigh) < 4:
        return None, None
    med = statistics.median(neigh)
    mad = statistics.median(abs(v - med) for v in neigh)
    scale = 1.4826 * mad
    if scale == 0:  # more than half the neighbours equal the median: fall back to mean absolute deviation
        scale = 1.2533 * statistics.fmean(abs(v - med) for v in neigh)
    if scale == 0:  # perfectly flat neighbourhood: 1% of the level
        scale = max(abs(med) * 0.01, 1e-9)
    return med, scale


def check(text, time_col=None, cols=None, interval=None, z=4.0, window=6, flatline_points=6, counters=()):
    time_col, value_cols, rows = load(text, time_col, cols)
    issues = []

    def add(kind, line, ts, col, detail):
        issues.append({"kind": kind, "line": line, "time": fmt_ts(ts), "column": col, "detail": detail})

    # order and duplicates (in file order)
    seen = {}
    for prev, cur in zip(rows, rows[1:]):
        if cur["ts"] < prev["ts"]:
            add("out_of_order", cur["line"], cur["ts"], time_col, f"goes back {fmt_dur(prev['ts'] - cur['ts'])} from line {prev['line']}")
    for r in rows:
        if r["ts"] in seen:
            add("duplicate", r["line"], r["ts"], time_col, f"same timestamp as line {seen[r['ts']]}")
        else:
            seen[r["ts"]] = r["line"]
    srt = sorted({r["ts"]: r for r in reversed(rows)}.values(), key=lambda r: r["ts"])  # first occurrence wins
    diffs = [b["ts"] - a["ts"] for a, b in zip(srt, srt[1:]) if b["ts"] > a["ts"]]
    if interval is None:
        if not diffs:
            raise InputError("cannot infer interval: all timestamps identical")
        interval = statistics.median(diffs)
    expected = int(round((srt[-1]["ts"] - srt[0]["ts"]) / interval)) + 1
    missing_points = 0
    for a, b in zip(srt, srt[1:]):
        d = b["ts"] - a["ts"]
        if d > 1.5 * interval:
            n = int(round(d / interval)) - 1
            missing_points += n
            add("gap", b["line"], a["ts"], time_col, f"{fmt_dur(d)} without data after {fmt_ts(a['ts'])} (~{n} missing)")

    stats = {}
    for c in value_cols:
        vals = [to_float(r["raw"][c]) for r in srt]
        for r, v in zip(srt, vals):
            if v is None:
                add("missing", r["line"], r["ts"], c, f"value {r['raw'][c]!r} is empty or not numeric")
        # flatline runs
        run_start = 0
        for i in range(1, len(vals) + 1):
            if i < len(vals) and vals[i] is not None and vals[i] == vals[run_start]:
                continue
            length = i - run_start
            if vals[run_start] is not None and length >= flatline_points:
                r0, r1 = srt[run_start], srt[i - 1]
                add("flatline", r0["line"], r0["ts"], c, f"value {vals[run_start]:g} repeated {length}x until {fmt_ts(r1['ts'])}")
            run_start = i
        if c in counters:
            for i in range(1, len(vals)):
                if vals[i] is not None and vals[i - 1] is not None and vals[i] < vals[i - 1]:
                    add("reset", srt[i]["line"], srt[i]["ts"], c, f"counter dropped {vals[i - 1]:g} -> {vals[i]:g} (restart or wrap)")
        else:
            for i, v in enumerate(vals):
                if v is None:
                    continue
                med, scale = rolling_robust(vals, i, window)
                if med is None:
                    continue
                score = abs(v - med) / scale
                if score >= z:
                    add("spike", srt[i]["line"], srt[i]["ts"], c, f"{v:g} vs local median {med:g} (robust z {score:.1f})")
        good = [v for v in vals if v is not None]
        stats[c] = {"count": len(good), "min": min(good) if good else None, "max": max(good) if good else None,
                    "median": statistics.median(good) if good else None}

    order = {"out_of_order": 0, "duplicate": 1, "gap": 2, "reset": 3, "missing": 4, "flatline": 5, "spike": 6}
    issues.sort(key=lambda x: (order[x["kind"]], x["time"], x["column"]))
    counts = {}
    for it in issues:
        counts[it["kind"]] = counts.get(it["kind"], 0) + 1
    return {
        "rows": len(rows), "time_column": time_col, "columns": value_cols,
        "start": fmt_ts(srt[0]["ts"]), "end": fmt_ts(srt[-1]["ts"]),
        "interval": fmt_dur(interval), "interval_seconds": interval,
        "expected_points": expected, "missing_points": missing_points,
        "completeness": round(100.0 * len(srt) / expected, 1) if expected else 100.0,
        "stats": stats, "counts": counts, "issues": issues,
    }


def to_markdown(r):
    lines = ["# Time series check", "",
             f"Rows: {r['rows']}  Range: {r['start']} to {r['end']}  Interval: {r['interval']}",
             f"Completeness: {r['completeness']}% ({r['missing_points']} of {r['expected_points']} expected points missing)", "",
             "| Column | Count | Min | Median | Max |", "|---|---|---|---|---|"]
    for c, s in r["stats"].items():
        f = lambda v: "-" if v is None else f"{v:g}"
        lines.append(f"| {c} | {s['count']} | {f(s['min'])} | {f(s['median'])} | {f(s['max'])} |")
    lines += ["", "## Issues"]
    if not r["issues"]:
        lines.append("No issues found.")
    else:
        lines.append("Totals: " + ", ".join(f"{k} {v}" for k, v in r["counts"].items()))
        lines += ["", "| Kind | Time (UTC) | Column | Line | Detail |", "|---|---|---|---|---|"]
        lines += [f"| {i['kind']} | {i['time']} | {i['column']} | {i['line']} | {i['detail']} |" for i in r["issues"]]
    return "\n".join(lines)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    base = 1760860800  # 2025-10-19T08:00:00Z
    def csv_of(points, header="timestamp,value"):
        return header + "\n" + "\n".join(f"{t},{v}" for t, v in points) + "\n"
    wave = [10 + (i % 3) for i in range(30)]
    clean = csv_of([(base + 60 * i, v) for i, v in enumerate(wave)])
    r = check(clean)
    ok("clean series has no issues", r["issues"] == [])
    ok("interval inferred as 1m", r["interval"] == "1m")
    ok("completeness 100%", r["completeness"] == 100.0)
    gap_pts = [(base + 60 * i, v) for i, v in enumerate(wave) if not 10 <= i < 15]
    r = check(csv_of(gap_pts))
    ok("gap detected", r["counts"].get("gap") == 1)
    ok("gap counts 5 missing points", r["missing_points"] == 5)
    spiky = list(wave); spiky[20] = 95
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(spiky)]))
    ok("spike detected", [i["line"] for i in r["issues"] if i["kind"] == "spike"] == [22])
    flat = list(wave); flat[5:13] = [7] * 8
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(flat)]), z=100)
    ok("flatline detected", r["counts"].get("flatline") == 1)
    miss = [(base + 60 * i, ("" if i == 4 else "NaN" if i == 9 else v)) for i, v in enumerate(wave)]
    ok("missing values detected", check(csv_of(miss)).get("counts").get("missing") == 2)
    dup = [(base + 60 * i, v) for i, v in enumerate(wave)]
    dup.insert(6, dup[5])
    ok("duplicate detected", check(csv_of(dup))["counts"].get("duplicate") == 1)
    ooo = [(base + 60 * i, v) for i, v in enumerate(wave)]
    ooo[7], ooo[8] = ooo[8], ooo[7]
    ok("out of order detected", check(csv_of(ooo))["counts"].get("out_of_order") == 1)
    counter = [(base + 60 * i, 100 * i if i < 20 else 100 * (i - 20)) for i in range(30)]
    r = check(csv_of(counter, "ts,requests_total"), counters=("requests_total",))
    ok("counter reset detected", r["counts"].get("reset") == 1 and "spike" not in r["counts"])
    iso_rows = "time,cpu\n" + "\n".join(f"2026-10-19T08:{i:02d}:00Z,{40 + (i % 2)}" for i in range(0, 60, 5)) + "\n"
    r = check(iso_rows)
    ok("ISO timestamps and 5m interval", r["interval"] == "5m" and r["issues"] == [])
    ms_rows = csv_of([((base + 30 * i) * 1000, v) for i, v in enumerate(wave)])
    ok("epoch milliseconds accepted", check(ms_rows)["interval"] == "30s")
    r = check(clean, interval=30.0)
    ok("explicit 30s interval flags every 1m step as a gap", r["counts"].get("gap") == 29 and r["completeness"] < 60)
    jitter = csv_of([(base + 60 * i + (3 if i % 2 else -2), v) for i, v in enumerate(wave)])
    ok("small timestamp jitter is not a gap", "gap" not in check(jitter)["counts"])
    multi = "timestamp,a,b\n" + "\n".join(f"{base + 60 * i},{wave[i]},{wave[i] * 2 if i != 12 else 500}" for i in range(30)) + "\n"
    r = check(multi, cols=["a", "b"])
    ok("multi-column spike only in b", {i["column"] for i in r["issues"]} == {"b"})
    for bad, label in (("timestamp,value\n1,2\n", "too few rows rejected"),
                       ("timestamp,value\nyesterday,1\ntoday,2\nnow,3\n", "bad timestamp rejected")):
        try:
            check(bad)
            ok(label, False)
        except InputError:
            ok(label, True)
    ok("markdown renders", "## Issues" in to_markdown(check(clean)))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Find gaps, duplicates, flatlines, spikes, and counter resets in CSV metrics.")
    ap.add_argument("csv", nargs="?", help="CSV file, or '-' for stdin")
    ap.add_argument("--time-col")
    ap.add_argument("--cols", help="comma-separated value columns (default: all except the time column)")
    ap.add_argument("--interval", help="expected spacing, e.g. 30s, 5m, 1h (default: median spacing)")
    ap.add_argument("--z", type=float, default=4.0, help="robust z-score threshold for spikes (default 4)")
    ap.add_argument("--window", type=int, default=6, help="neighbours on each side for the rolling median (default 6)")
    ap.add_argument("--flatline-points", type=int, default=6)
    ap.add_argument("--counter", default="", help="comma-separated monotonic counter columns (checked for resets, not spikes)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.csv:
        ap.error("csv path required (or --self-test)")
    try:
        text = sys.stdin.read() if a.csv == "-" else open(a.csv, encoding="utf-8-sig").read()
        r = check(text, a.time_col, [c.strip() for c in a.cols.split(",")] if a.cols else None,
                  parse_interval(a.interval) if a.interval else None, a.z, a.window, a.flatline_points,
                  tuple(c.strip() for c in a.counter.split(",") if c.strip()))
    except (OSError, InputError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    return 1 if r["issues"] else 0


if __name__ == "__main__":
    sys.exit(main())

---
name: slo-error-budget-burn-rate-calculator
description: Chuyển đổi số lượng yêu cầu thành các chỉ số trả lời cho SLO: SLI, ngân sách lỗi cho khung thời gian 28 ngày hoặc 30 ngày, ngân sách đã tiêu thụ và còn lại, tốc độ tiêu hao (burn rate) trong khoảng từ 5 phút đến 3 ngày, cảnh báo tốc độ tiêu hao đa khung thời gian (gọi trực khẩn cấp ở mức 14.4x và 6x, tạo ticket ở mức 3x và 1x, được quy đổi theo khung thời gian), và thời gian cho đến khi cạn kiệt ngân sách. Sử dụng khi thiết lập SLO, tinh chỉnh cảnh báo, quyết định đóng băng phát hành hoặc viết tóm tắt sự cố. Bao gồm một công cụ tính bằng Python dùng thư viện chuẩn với tùy chọn --self-test.
---

# Công Cụ Tính Ngân Sách Lỗi và Tốc Độ Tiêu Hao SLO

Bạn giúp các kỹ sư và đội ngũ trực sự cố (on-call) đánh giá độ tin cậy của hệ thống dựa trên số liệu thay vì cảm tính. Bạn chuyển đổi số lượng yêu cầu thành công/tổng số yêu cầu thành chỉ số SLI, quy đổi mục tiêu SLO thành ngân sách lỗi (error budget), chỉ ra tốc độ tiêu hao ngân sách qua nhiều khung thời gian, cho biết cảnh báo nào cần gọi trực khẩn cấp (page) hoặc mở ticket, đồng thời ước tính thời điểm ngân sách sẽ cạn kiệt nếu giữ nguyên tốc độ hiện tại. Bạn giải thích từng con số bằng ngôn ngữ dễ hiểu để các chủ sản phẩm (product owner) có thể nắm bắt và đưa ra quyết định.

## Các tệp trong kỹ năng này

- `scripts/slo_budget.py` - đọc tệp CSV chứa số lượng yêu cầu theo từng khoảng thời gian và in ra SLI, ngân sách đã tiêu thụ và còn lại, tốc độ tiêu hao, trạng thái cảnh báo, thời gian dự kiến cạn kiệt và khoảng thời gian tệ nhất; tùy chọn `--allowance` in ra thời gian ngừng hoạt động cho phép và ngưỡng cảnh báo cho bất kỳ mức SLO nào (chỉ sử dụng thư viện chuẩn của Python 3)
- `references/slo-math.md` - công thức tính SLI, ngân sách lỗi, tốc độ tiêu hao, mức tiêu thụ dự kiến và các con số tính sẵn cho những mục tiêu phổ biến
- `references/burn-rate-alerting.md` - chính sách cảnh báo đa khung thời gian, đa tốc độ tiêu hao, cách các ngưỡng thay đổi theo khung thời gian và lời khuyên tinh chỉnh
- `templates/slo-status-report.md` - mẫu báo cáo trạng thái SLO hàng tuần hoặc tóm tắt ngân sách sau sự cố
- `examples/checkout-api-requests.csv` - dữ liệu số lượng yêu cầu theo từng khoảng 15 phút trong hai ngày của một API thanh toán, có một đợt gián đoạn nhỏ và một sự cố kéo dài 2 giờ ở cuối kỳ

## Quy trình làm việc

### 1. Xác định SLI và SLO
Hỏi rõ thế nào được coi là một sự kiện "tốt" (mã trạng thái < 500, độ trễ dưới 300 ms, tác vụ chạy thành công), dữ liệu đếm lấy từ đâu, mục tiêu là bao nhiêu (99%, 99.5%, 99.9%, 99.95%), và khung thời gian tính là gì (28 hoặc 30 ngày trượt). Hãy phản biện lại những mục tiêu mà hệ thống chưa từng đạt được trong thực tế: luôn hiển thị chỉ số SLI lịch sử trước tiên.

### 2. Thu thập dữ liệu
Xuất số lượng sự kiện tốt và tổng số sự kiện (hoặc sự kiện lỗi và tổng số sự kiện) theo từng khoảng thời gian, ví dụ các khung 1 phút, 5 phút hoặc 15 phút, dưới dạng tệp CSV với các cột `timestamp,total,good`. Cột timestamp đánh dấu thời điểm bắt đầu của mỗi khoảng thời gian. Sử dụng múi giờ UTC.

### 3. Chạy công cụ tính
```
python3 scripts/slo_budget.py examples/checkout-api-requests.csv --slo 99.9 --window 30d
python3 scripts/slo_budget.py requests.csv --slo 99.5 --window 28d --json > slo.json
python3 scripts/slo_budget.py requests.csv --now 2026-10-19T22:00:00Z
python3 scripts/slo_budget.py --allowance --slo 99.95 --window 30d
python3 scripts/slo_budget.py --self-test
```
Mã thoát (exit code) bằng 1 có nghĩa là cảnh báo gọi trực khẩn cấp (page alert) đang kích hoạt hoặc ngân sách đã cạn kiệt; bằng 2 có nghĩa là dữ liệu đầu vào không hợp lệ (thiếu cột, số lỗi > tổng số, không phân tích được định dạng timestamp).

### 4. Diễn giải kết quả
- Tốc độ tiêu hao (burn rate) bằng 1.0 nghĩa là tiêu thụ vừa hết ngân sách khi kết thúc khung thời gian; mức 14.4 duy trì trong một giờ sẽ tiêu tốn 2% ngân sách của cả 30 ngày.
- Ngân sách đã tiêu thụ được phóng chiếu dựa trên lưu lượng truy cập ghi nhận được cho đến hiện tại; nếu dữ liệu chưa đủ một khung thời gian đầy đủ, hãy nêu rõ điều đó.
- Một cảnh báo chỉ kích hoạt khi cả khung thời gian dài và khung thời gian ngắn của nó đều vượt ngưỡng; khung thời gian ngắn giúp cảnh báo nhanh chóng tắt đi sau khi hệ thống phục hồi. Xem thêm trong `references/burn-rate-alerting.md`.
- Các khung thời gian ngắn hơn bước nhảy dữ liệu sẽ được làm tròn lên thành một bước nhảy; các khung thời gian dài hơn tổng thời lượng dữ liệu sẽ có trạng thái là "not evaluable" (không thể đánh giá).

### 5. Đưa ra khuyến nghị và báo cáo
Gắn kết các con số với một quyết định cụ thể: gọi trực khẩn cấp ngay, mở ticket, làm chậm các đợt phát hành tiềm ẩn rủi ro khi ngân sách còn lại thấp, hoặc nới lỏng một SLO vốn luôn bị vi phạm do đặc thù thiết kế. Điền thông tin vào `templates/slo-status-report.md`.

## Kết quả ví dụ
Đối với `examples/checkout-api-requests.csv` ở mức SLO 99.9% trong 30 ngày: SLI đạt 99.8768%, đã tiêu thụ 8.2% ngân sách, tốc độ tiêu hao trong 1h là 26.46 và trong 5m là 28.06 nên cảnh báo gọi trực nhanh được kích hoạt, tốc độ tiêu hao trong 6h là 5.83 nằm ngay dưới ngưỡng 6.0, cảnh báo ticket 3d ở trạng thái không thể đánh giá (not evaluable) vì chỉ có hai ngày dữ liệu, và với tốc độ của giờ cuối cùng thì ngân sách sẽ cạn sạch sau khoảng 1 ngày 16 giờ. Mã thoát là 1.

## Quy tắc

- Luôn nêu rõ SLO, khung thời gian, phạm vi dữ liệu và bất kỳ khoảng thời gian bị thiếu nào kèm theo mỗi kết quả.
- Không trộn lẫn nhiều định nghĩa sự kiện tốt vào chung một SLI; hãy tính các SLO riêng biệt cho độ sẵn sàng (availability) và độ trễ (latency).
- SLO là công cụ hỗ trợ ra quyết định, không phải công cụ quy trách nhiệm: hãy mô tả tác động lên người dùng, không nhằm vào cá nhân.
- Không tự bịa ra số liệu lưu lượng truy cập; nếu thiếu dữ liệu đếm, hãy yêu cầu cung cấp hoặc nêu rõ kết quả nào không thể tính toán được.
FILE:references/slo-math.md
# Toán Học Trong SLO

## Các định nghĩa
- **SLI** (service level indicator - chỉ số mức dịch vụ) = số sự kiện tốt / tổng số sự kiện, tính theo phần trăm.
- **SLO** (objective - mục tiêu mức dịch vụ) = mức SLI mục tiêu trong một khung thời gian, ví dụ: 99.9% yêu cầu thành công trong 30 ngày trượt.
- **Ngân sách lỗi (Error budget)** = 1 - SLO = tỷ lệ sự kiện được phép thất bại. Mức 99.9% -> 0.1%.
- **Ngân sách tính theo số sự kiện** = (1 - SLO) x tổng số sự kiện trong khung thời gian.
- **Tỷ lệ lỗi (Error rate)** trong một khung thời gian = số sự kiện lỗi / tổng số sự kiện trong khung thời gian đó.
- **Tốc độ tiêu hao (Burn rate)** = tỷ lệ lỗi / (1 - SLO). Mức 1.0 có nghĩa là ngân sách sẽ được sử dụng vừa hết vào thời điểm kết thúc khung thời gian.

## Mức tiêu thụ dự kiến với dữ liệu một phần
Với D giây dữ liệu thực tế và W giây của khung thời gian SLO:
- số yêu cầu dự kiến trong toàn khung = (số yêu cầu trong dữ liệu / D) x W
- số lỗi cho phép = (1 - SLO) x số yêu cầu dự kiến
- tỷ lệ đã tiêu thụ = số lỗi trong dữ liệu / số lỗi cho phép
- thời gian cho đến khi cạn kiệt theo tốc độ gần đây = (số lỗi cho phép - số lỗi tính đến hiện tại) / (số lỗi mỗi giây trong khung thời gian gần đây đó)

## Thời gian ngừng hoạt động hoàn toàn tối đa cho phép
| SLO | mỗi 30 ngày | mỗi 28 ngày | mỗi tuần | mỗi ngày |
|---|---|---|---|---|
| 99% | 7h 12m | 6h 43m 12s | 1h 40m 48s | 14m 24s |
| 99.5% | 3h 36m | 3h 21m 36s | 50m 24s | 7m 12s |
| 99.9% | 43m 12s | 40m 19s | 10m 5s | 1m 26s |
| 99.95% | 21m 36s | 20m 10s | 5m 2s | 43s |
| 99.99% | 4m 19s | 4m 2s | 1m 0s | 9s |

Các SLO dựa trên số lượng yêu cầu không phụ thuộc trực tiếp vào thời gian đồng hồ: 10 phút bị lỗi 50% sẽ tiêu tốn lượng ngân sách tương đương với 5 phút ngừng hoạt động hoàn toàn nếu có cùng lưu lượng truy cập, và một sự cố ngừng hoạt động lúc 3 giờ sáng sẽ tốn ít ngân sách hơn so với sự cố xảy ra vào giờ cao điểm.

## Ví dụ minh họa cách tính
SLO 99.9%, khung thời gian 30 ngày, 1,000,000 yêu cầu mỗi ngày.
- Ngân sách: 0.001 x 30,000,000 = 30,000 yêu cầu thất bại.
- Tỷ lệ lỗi 2% kéo dài trong một giờ ở mức lưu lượng 41,667 yêu cầu/giờ sẽ làm tiêu hao 833 lỗi = 2.8% ngân sách, tương ứng tốc độ tiêu hao là 20.
- Ở tốc độ tiêu hao bằng 20, toàn bộ ngân sách sẽ chỉ trụ được trong 30 ngày / 20 = 36 giờ.

## Các bẫy thường gặp
- Đếm các lần thử lại (retry) như những sự kiện tốt riêng biệt sẽ che giấu các lỗi mà người dùng thực sự gặp phải.
- Lưu lượng truy cập quá thấp (ban đêm, khu vực mới mở) khiến tỷ lệ lỗi bị nhiễu mạnh; hãy áp dụng số lượng yêu cầu tối thiểu hoặc dùng khung thời gian ngắn dài hơn.
- Các phép đo giả lập (synthetic probes) đo lường chính công cụ thăm dò chứ không phải người dùng thực tế; hãy gắn nhãn tách biệt cho chúng.
- Thay đổi SLO giữa chừng khung thời gian: hãy tính toán lại toàn bộ khung thời gian theo mục tiêu mới và ghi chú rõ điều này.
FILE:references/burn-rate-alerting.md
# Cảnh Báo Theo Tốc Độ Tiêu Hao (Burn-Rate Alerting)

## Chính sách mặc định (đa khung thời gian, đa tốc độ tiêu hao)
| Mức độ nghiêm trọng | Tỷ phần ngân sách | Khung dài | Khung ngắn | Ngưỡng tiêu hao (30d) | Ngưỡng tiêu hao (28d) |
|---|---|---|---|---|---|
| page | 2% | 1h | 5m | 14.4 | 13.44 |
| page | 5% | 6h | 30m | 6.0 | 5.6 |
| ticket | 10% | 1d | 2h | 3.0 | 2.8 |
| ticket | 10% | 3d | 6h | 1.0 | 0.93 |

Ngưỡng = tỷ phần ngân sách x khung thời gian SLO / khung dài. Ví dụ: 0.02 x 720h / 1h = 14.4.

## Tại sao cần kết hợp hai khung thời gian
- **Khung thời gian dài (long window)** chứng minh sự cố đủ nghiêm trọng (đã làm tiêu tốn một lượng ngân sách đáng kể).
- **Khung thời gian ngắn (short window)** (bằng khoảng 1/12 khung dài) chứng minh sự cố vẫn đang tiếp diễn, nhờ đó cảnh báo sẽ tự động tắt chỉ sau vài phút khi hệ thống phục hồi thay vì tiếp tục báo đỏ suốt cả tiếng đồng hồ.
- Chỉ phát cảnh báo khi cả hai khung thời gian cùng vượt ngưỡng.

## Ví dụ quy tắc cảnh báo (theo cú pháp Prometheus)
```
(
  sum(rate(http_requests_errors_total[1h])) / sum(rate(http_requests_total[1h])) > (14.4 * 0.001)
and
  sum(rate(http_requests_errors_total[5m])) / sum(rate(http_requests_total[5m])) > (14.4 * 0.001)
)
```

## Tinh chỉnh cảnh báo
- Các dịch vụ có lưu lượng thấp: yêu cầu số lượng yêu cầu tối thiểu trong khung ngắn, hoặc chỉ gọi trực khẩn cấp dựa trên khung 6h.
- Bị gọi trực quá nhiều lần: hãy kiểm tra xem SLO có đang đặt khắt khe hơn mức người dùng thực sự cần hay không trước khi nâng ngưỡng cảnh báo.
- Bỏ sót sự cố: các đợt tiêu hao âm ỉ dưới mức 1.0 theo thiết kế sẽ không bao giờ phát cảnh báo; hãy rà soát ngân sách hàng tuần.
- Bước nhảy dữ liệu thô hơn khung thời gian ngắn (ví dụ: dữ liệu gom theo từng khung 15 phút): khi đó khung ngắn sẽ bị quy về thành một khung dữ liệu duy nhất và việc phát hiện hệ thống phục hồi sẽ chậm hơn.

## Gợi ý chính sách quản lý ngân sách lỗi
- Đã tiêu thụ trên 50%: rà soát kỹ các thay đổi tiềm ẩn rủi ro, ưu tiên triển khai kiểu canary.
- Đã tiêu thụ trên 75%: đóng băng phát hành đối với các thay đổi không phải bản vá lỗi, ưu tiên công việc củng cố độ tin cậy lên hàng đầu.
- Cạn kiệt ngân sách: bắt buộc thực hiện họp rút kinh nghiệm sau sự cố (postmortem); đóng băng phát hành cho đến khi khung thời gian trượt phục hồi lại ngân sách.
FILE:templates/slo-status-report.md
# Trạng Thái SLO - <service> - <period or incident>

**SLI:** <what counts as good>  **SLO:** <target>% trên khung trượt <28d/30d>
**Dữ liệu:** <start> đến <end> UTC, bước nhảy <interval>, thiếu <n> khoảng thời gian
**Công cụ tính:** `slo_budget.py --slo <target> --window <window>` (mã thoát <code>)

## Ngân sách
| Chỉ số | Giá trị |
|---|---|
| SLI trong kỳ | |
| Ngân sách đã tiêu thụ | |
| Ngân sách còn lại | |
| Thời gian ngừng hoạt động hoàn toàn cho phép mỗi khung | |

## Tốc độ tiêu hao (Burn rates)
| 5m | 30m | 1h | 2h | 6h | 1d | 3d |
|---|---|---|---|---|---|---|
| | | | | | | |

## Cảnh báo
| Chính sách | Trạng thái | Mức tiêu hao dài / ngắn |
|---|---|---|
| page 1h/5m | | |
| page 6h/30m | | |
| ticket 1d/2h | | |
| ticket 3d/6h | | |

## Chuyện gì đã xảy ra
<khoảng thời gian tệ nhất, tác động tới người dùng, nguyên nhân nếu đã rõ>

## Quyết định
- [ ] Gọi trực khẩn cấp / mở sự cố
- [ ] Chính sách phát hành: <bình thường / thận trọng / đóng băng> vì <ngân sách còn lại>
- [ ] Tinh chỉnh SLO hoặc cảnh báo: <nội dung thay đổi và lý do>

## Tóm tắt dành cho các bên liên quan
> <service> đạt mức <SLI>% so với mục tiêu <target>%. Đã sử dụng <x>% ngân sách lỗi của khung thời gian này; với tốc độ hiện tại, ngân sách sẽ duy trì được trong <time>. Bước tiếp theo: <decision>.
FILE:examples/checkout-api-requests.csv
timestamp,total,good
2026-10-18T00:00:00Z,4757,4755
2026-10-18T00:15:00Z,4525,4523
2026-10-18T00:30:00Z,4313,4311
2026-10-18T00:45:00Z,4122,4120
2026-10-18T01:00:00Z,3951,3949
2026-10-18T01:15:00Z,3803,3801
2026-10-18T01:30:00Z,3678,3676
2026-10-18T01:45:00Z,3577,3576
2026-10-18T02:00:00Z,3500,3498
2026-10-18T02:15:00Z,3448,3446
2026-10-18T02:30:00Z,3421,3419
2026-10-18T02:45:00Z,3019,3018
2026-10-18T03:00:00Z,3044,3043
2026-10-18T03:15:00Z,3093,3092
2026-10-18T03:30:00Z,3169,3168
2026-10-18T03:45:00Z,3270,3268
2026-10-18T04:00:00Z,3396,3394
2026-10-18T04:15:00Z,3547,3545
2026-10-18T04:30:00Z,3722,3720
2026-10-18T04:45:00Z,3921,3919
2026-10-18T05:00:00Z,4143,4141
2026-10-18T05:15:00Z,4388,4386
2026-10-18T05:30:00Z,4253,4251
2026-10-18T05:45:00Z,4539,4537
2026-10-18T06:00:00Z,4845,4843
2026-10-18T06:15:00Z,5168,5166
2026-10-18T06:30:00Z,5509,5507
2026-10-18T06:45:00Z,5865,5863
2026-10-18T07:00:00Z,6236,6234
2026-10-18T07:15:00Z,6619,6616
2026-10-18T07:30:00Z,7013,7009
2026-10-18T07:45:00Z,7418,7414
2026-10-18T08:00:00Z,7831,7827
2026-10-18T08:15:00Z,7850,7847
2026-10-18T08:30:00Z,8274,8271
2026-10-18T08:45:00Z,8702,8699
2026-10-18T09:00:00Z,9132,9127
2026-10-18T09:15:00Z,9561,9556
2026-10-18T09:30:00Z,9989,9984
2026-10-18T09:45:00Z,10413,10408
2026-10-18T10:00:00Z,10832,10827
2026-10-18T10:15:00Z,11245,11240
2026-10-18T10:30:00Z,11650,11645
2026-10-18T10:45:00Z,12044,12038
2026-10-18T11:00:00Z,12028,12022
2026-10-18T11:15:00Z,12398,12392
2026-10-18T11:30:00Z,12754,12748
2026-10-18T11:45:00Z,13095,13089
2026-10-18T12:00:00Z,13418,13412
2026-10-18T12:15:00Z,13724,13719
2026-10-18T12:30:00Z,14010,14003
2026-10-18T12:45:00Z,14275,14268
2026-10-18T13:00:00Z,14520,14513
2026-10-18T13:15:00Z,14742,14735
2026-10-18T13:30:00Z,14941,14934
2026-10-18T13:45:00Z,14716,14710
2026-10-18T14:00:00Z,14867,14861
2026-10-18T14:15:00Z,14993,14985
2026-10-18T14:30:00Z,15094,15086
2026-10-18T14:45:00Z,15170,15163
2026-10-18T15:00:00Z,15220,15213
2026-10-18T15:15:00Z,15244,15237
2026-10-18T15:30:00Z,15242,15236
2026-10-18T15:45:00Z,15215,15209
2026-10-18T16:00:00Z,15163,15155
2026-10-18T16:15:00Z,14686,14679
2026-10-18T16:30:00Z,14585,14578
2026-10-18T16:45:00Z,14460,14453
2026-10-18T17:00:00Z,14312,14306
2026-10-18T17:15:00Z,14141,14135
2026-10-18T17:30:00Z,13950,13866
2026-10-18T17:45:00Z,13738,13656
2026-10-18T18:00:00Z,13506,13499
2026-10-18T18:15:00Z,13257,13251
2026-10-18T18:30:00Z,12990,12984
2026-10-18T18:45:00Z,12708,12702
2026-10-18T19:00:00Z,12012,12007
2026-10-18T19:15:00Z,11702,11697
2026-10-18T19:30:00Z,11382,11376
2026-10-18T19:45:00Z,11051,11045
2026-10-18T20:00:00Z,10712,10707
2026-10-18T20:15:00Z,10367,10362
2026-10-18T20:30:00Z,10017,10013
2026-10-18T20:45:00Z,9663,9659
2026-10-18T21:00:00Z,9308,9304
2026-10-18T21:15:00Z,8952,8947
2026-10-18T21:30:00Z,8598,8594
2026-10-18T21:45:00Z,7848,7844
2026-10-18T22:00:00Z,7503,7500
2026-10-18T22:15:00Z,7164,7161
2026-10-18T22:30:00Z,6833,6830
2026-10-18T22:45:00Z,6513,6510
2026-10-18T23:00:00Z,6204,6201
2026-10-18T23:15:00Z,5907,5904
2026-10-18T23:30:00Z,5625,5622
2026-10-18T23:45:00Z,5358,5356
2026-10-19T00:00:00Z,5109,5107
2026-10-19T00:15:00Z,4877,4875
2026-10-19T00:30:00Z,4265,4263
2026-10-19T00:45:00Z,4074,4072
2026-10-19T01:00:00Z,3903,3901
2026-10-19T01:15:00Z,3755,3753
2026-10-19T01:30:00Z,3630,3628
2026-10-19T01:45:00Z,3529,3527
2026-10-19T02:00:00Z,3452,3451
2026-10-19T02:15:00Z,3400,3399
2026-10-19T02:30:00Z,3373,3371
2026-10-19T02:45:00Z,3371,3369
2026-10-19T03:00:00Z,3396,3394
2026-10-19T03:15:00Z,3045,3044
2026-10-19T03:30:00Z,3121,3120
2026-10-19T03:45:00Z,3222,3221
2026-10-19T04:00:00Z,3348,3347
2026-10-19T04:15:00Z,3499,3497
2026-10-19T04:30:00Z,3674,3672
2026-10-19T04:45:00Z,3873,3871
2026-10-19T05:00:00Z,4095,4093
2026-10-19T05:15:00Z,4340,4338
2026-10-19T05:30:00Z,4605,4603
2026-10-19T05:45:00Z,4491,4489
2026-10-19T06:00:00Z,4797,4795
2026-10-19T06:15:00Z,5120,5117
2026-10-19T06:30:00Z,5461,5458
2026-10-19T06:45:00Z,5817,5814
2026-10-19T07:00:00Z,6188,6185
2026-10-19T07:15:00Z,6571,6568
2026-10-19T07:30:00Z,6965,6962
2026-10-19T07:45:00Z,7370,7366
2026-10-19T08:00:00Z,7783,7779
2026-10-19T08:15:00Z,8202,8198
2026-10-19T08:30:00Z,8226,8222
2026-10-19T08:45:00Z,8654,8650
2026-10-19T09:00:00Z,9084,9080
2026-10-19T09:15:00Z,9513,9509
2026-10-19T09:30:00Z,9941,9936
2026-10-19T09:45:00Z,10365,10360
2026-10-19T10:00:00Z,10784,10779
2026-10-19T10:15:00Z,11197,11192
2026-10-19T10:30:00Z,11602,11597
2026-10-19T10:45:00Z,11996,11991
2026-10-19T11:00:00Z,12380,12375
2026-10-19T11:15:00Z,12350,12344
2026-10-19T11:30:00Z,12706,12700
2026-10-19T11:45:00Z,13047,13041
2026-10-19T12:00:00Z,13370,13364
2026-10-19T12:15:00Z,13676,13670
2026-10-19T12:30:00Z,13962,13956
2026-10-19T12:45:00Z,14227,14221
2026-10-19T13:00:00Z,14472,14464
2026-10-19T13:15:00Z,14694,14687
2026-10-19T13:30:00Z,14893,14886
2026-10-19T13:45:00Z,15068,15061
2026-10-19T14:00:00Z,14819,14812
2026-10-19T14:15:00Z,14945,14939
2026-10-19T14:30:00Z,15046,15040
2026-10-19T14:45:00Z,15122,15114
2026-10-19T15:00:00Z,15172,15164
2026-10-19T15:15:00Z,15196,15189
2026-10-19T15:30:00Z,15194,15187
2026-10-19T15:45:00Z,15167,15160
2026-10-19T16:00:00Z,15115,15109
2026-10-19T16:15:00Z,15038,15032
2026-10-19T16:30:00Z,14937,14929
2026-10-19T16:45:00Z,14412,14405
2026-10-19T17:00:00Z,14264,14257
2026-10-19T17:15:00Z,14093,14087
2026-10-19T17:30:00Z,13902,13896
2026-10-19T17:45:00Z,13690,13684
2026-10-19T18:00:00Z,13458,13453
2026-10-19T18:15:00Z,13209,13202
2026-10-19T18:30:00Z,12942,12936
2026-10-19T18:45:00Z,12660,12654
2026-10-19T19:00:00Z,12364,12358
2026-10-19T19:15:00Z,11654,11649
2026-10-19T19:30:00Z,11334,11329
2026-10-19T19:45:00Z,11003,10999
2026-10-19T20:00:00Z,10664,10658
2026-10-19T20:15:00Z,10319,10314
2026-10-19T20:30:00Z,9969,9964
2026-10-19T20:45:00Z,9615,9611
2026-10-19T21:00:00Z,9260,9256
2026-10-19T21:15:00Z,8904,8900
2026-10-19T21:30:00Z,8550,8547
2026-10-19T21:45:00Z,8200,8196
2026-10-19T22:00:00Z,7455,7298
2026-10-19T22:15:00Z,7116,6959
2026-10-19T22:30:00Z,6785,6629
2026-10-19T22:45:00Z,6465,6310
2026-10-19T23:00:00Z,6156,6002
2026-10-19T23:15:00Z,5859,5707
2026-10-19T23:30:00Z,5577,5426
2026-10-19T23:45:00Z,5310,5161
FILE:scripts/slo_budget.py
#!/usr/bin/env python3
"""SLO error budget and burn rate calculator (Python 3 standard library only).

Input: CSV with one row per interval, the timestamp marking the START of the interval, and request
counts. Accepted columns (case-insensitive): timestamp|time|ts, total|requests|count, and either
good|success|ok or bad|errors|failed. Timestamps: ISO 8601 (Z or offset) or Unix seconds.

Math (see references/slo-math.md):
  error budget (fraction)  = 1 - SLO                          e.g. 99.9% -> 0.001
  error rate over window   = sum(bad) / sum(total)
  burn rate                = error rate / (1 - SLO)           1.0 = spends exactly the budget over the SLO window
  budget consumed so far   = sum(bad) / ((1 - SLO) * expected requests in the SLO window)
  alert threshold          = budget share * window hours / long window hours   (14.4 = 2% of 30d in 1h)

Default multiwindow, multi-burn-rate policy (Google SRE Workbook): page 2% in 1h (short 5m),
page 5% in 6h (short 30m), ticket 10% in 1d (short 2h), ticket 10% in 3d (short 6h). An alert fires
only when BOTH the long and the short window burn faster than the threshold. Windows shorter than
the data step are rounded up to one step; windows longer than the data are reported as not evaluable.

Usage:
  python3 slo_budget.py requests.csv --slo 99.9 --window 30d
  python3 slo_budget.py requests.csv --slo 99.5 --window 28d --json
  python3 slo_budget.py --allowance --slo 99.95 --window 30d     # downtime allowed, no CSV needed
  python3 slo_budget.py --self-test
Exit code: 0 OK, 1 a page alert fires or the budget is exhausted, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import json
import re
import statistics
import sys

POLICY = [
    # (severity, long window, short window, budget share)
    ("page", "1h", "5m", 0.02),
    ("page", "6h", "30m", 0.05),
    ("ticket", "1d", "2h", 0.10),
    ("ticket", "3d", "6h", 0.10),
]
ALIASES = {
    "ts": ("timestamp", "time", "ts", "date", "datetime"),
    "total": ("total", "requests", "count", "events"),
    "good": ("good", "success", "ok", "successes"),
    "bad": ("bad", "errors", "failed", "failures", "error"),
}


class SLOError(ValueError):
    pass


def parse_duration(s):
    m = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([smhdw])\s*", str(s).lower())
    if not m:
        raise SLOError(f"bad duration {s!r} (use e.g. 5m, 1h, 28d)")
    return float(m.group(1)) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[m.group(2)]


def fmt_dur(sec):
    sec = int(round(sec))
    if sec >= 86400:
        d, h = divmod(sec // 3600, 24)
        return f"{d}d {h}h" if h else f"{d}d"
    if sec >= 3600:
        h, m = divmod(sec // 60, 60)
        return f"{h}h {m}m" if m else f"{h}h"
    if sec >= 60:
        m, s = divmod(sec, 60)
        return f"{m}m {s}s" if s else f"{m}m"
    return f"{sec}s"


def parse_ts(v):
    v = v.strip()
    if re.fullmatch(r"\d+(\.\d+)?", v):
        return dt.datetime.fromtimestamp(float(v), dt.timezone.utc)
    t = dt.datetime.fromisoformat(v.replace("Z", "+00:00").replace(" ", "T"))
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.astimezone(dt.timezone.utc)


def load_rows(lines):
    rd = csv.DictReader(lines)
    if not rd.fieldnames:
        raise SLOError("empty CSV")
    cols = {}
    lower = {f.strip().lower(): f for f in rd.fieldnames}
    for key, names in ALIASES.items():
        for n in names:
            if n in lower:
                cols[key] = lower[n]
                break
    if "ts" not in cols or "total" not in cols or not ({"good", "bad"} & cols.keys()):
        raise SLOError(f"need timestamp, total and good or bad columns; got {rd.fieldnames}")
    rows = []
    for i, r in enumerate(rd, 2):
        try:
            ts = parse_ts(r[cols["ts"]])
            total = int(float(r[cols["total"]]))
            bad = int(float(r[cols["bad"]])) if "bad" in cols else total - int(float(r[cols["good"]]))
        except (ValueError, TypeError) as e:
            raise SLOError(f"line {i}: {e}")
        if total < 0 or bad < 0 or bad > total:
            raise SLOError(f"line {i}: counts must satisfy 0 <= bad <= total (total={total}, bad={bad})")
        rows.append((ts, total, bad))
    if len(rows) < 2:
        raise SLOError("need at least two rows")
    rows.sort(key=lambda r: r[0])
    return rows


def analyze(rows, slo_pct, window, now=None, policy=POLICY):
    if not 0 < slo_pct < 100:
        raise SLOError("SLO must be between 0 and 100 (exclusive), e.g. 99.9")
    budget = 1 - slo_pct / 100
    win = parse_duration(window)
    diffs = [(b[0] - a[0]).total_seconds() for a, b in zip(rows, rows[1:])]
    step = statistics.median(diffs)
    if step <= 0:
        raise SLOError("duplicate timestamps")
    first = rows[0][0]
    end = (now or rows[-1][0] + dt.timedelta(seconds=step))
    covered = (end - first).total_seconds()
    expected_rows = int(round(covered / step))
    missing = max(0, expected_rows - len(rows))

    def window_stats(seconds):
        seconds = max(seconds, step)
        if seconds > covered + 1e-6:
            return None
        start = end - dt.timedelta(seconds=seconds)
        tot = sum(t for ts, t, b in rows if start <= ts < end)
        bad = sum(b for ts, t, b in rows if start <= ts < end)
        rate = bad / tot if tot else 0.0
        return {"seconds": seconds, "total": tot, "bad": bad, "error_rate": rate, "burn_rate": rate / budget}

    total = sum(r[1] for r in rows)
    bad = sum(r[2] for r in rows)
    req_per_sec = total / covered
    expected_window_total = req_per_sec * win
    allowed_bad = budget * expected_window_total
    consumed = bad / allowed_bad if allowed_bad else 0.0
    alerts = []
    for sev, lw, sw, share in policy:
        thr = share * win / parse_duration(lw)
        L, S = window_stats(parse_duration(lw)), window_stats(parse_duration(sw))
        if L is None or S is None:
            alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr, "status": "not evaluable",
                           "long_burn": None, "short_burn": None})
            continue
        fires = L["burn_rate"] >= thr and S["burn_rate"] >= thr
        alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr,
                       "status": "FIRING" if fires else "ok",
                       "long_burn": L["burn_rate"], "short_burn": S["burn_rate"]})
    burn = {}
    for w in ("5m", "30m", "1h", "2h", "6h", "1d", "3d"):
        s = window_stats(parse_duration(w))
        burn[w] = None if s is None else s["burn_rate"]
    remaining_bad = allowed_bad - bad
    exhaust = {}
    for w in ("1h", "6h", "1d"):
        s = window_stats(parse_duration(w))
        if s and s["bad"] > 0 and remaining_bad > 0:
            exhaust[w] = remaining_bad / (s["bad"] / s["seconds"])
        elif s:
            exhaust[w] = 0.0 if remaining_bad <= 0 else None
    worst = max(rows, key=lambda r: (r[2] / r[1] if r[1] else 0, r[0]))
    return {
        "slo_percent": slo_pct, "window": window, "window_seconds": win, "step_seconds": step,
        "data_start": first.strftime("%Y-%m-%dT%H:%M:%SZ"), "data_end": end.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "rows": len(rows), "missing_intervals": missing,
        "total": total, "bad": bad, "sli_percent": 100 * (1 - bad / total) if total else 100.0,
        "allowed_bad_in_window": allowed_bad, "budget_consumed": consumed, "budget_remaining": 1 - consumed,
        "allowed_downtime_seconds": budget * win,
        "burn_rates": burn, "alerts": alerts, "exhaustion_seconds": exhaust,
        "worst_interval": {"start": worst[0].strftime("%Y-%m-%dT%H:%M:%SZ"), "total": worst[1], "bad": worst[2],
                           "error_rate": worst[2] / worst[1] if worst[1] else 0.0},
    }


def allowance_table(slo_pct, window):
    win = parse_duration(window)
    b = 1 - slo_pct / 100
    return {"slo_percent": slo_pct, "window": window, "allowed_downtime_seconds": b * win,
            "per_day_seconds": b * 86400, "per_week_seconds": b * 604800,
            "thresholds": [{"severity": s, "long": l, "short": sh, "share": sha, "burn_rate": sha * win / parse_duration(l)}
                           for s, l, sh, sha in POLICY]}


def to_markdown(r):
    pct = lambda x: f"{100 * x:.1f}%"
    out = [f"# SLO error budget report", "",
           f"**SLO:** {r['slo_percent']}% over {r['window']}  **Data:** {r['data_start']} to {r['data_end']} "
           f"({r['rows']} rows, step {fmt_dur(r['step_seconds'])}, {r['missing_intervals']} missing)",
           f"**SLI over data:** {r['sli_percent']:.4f}%  ({r['bad']:,} bad of {r['total']:,})",
           f"**Error budget:** {r['allowed_bad_in_window']:,.0f} bad requests per {r['window']} at current traffic "
           f"(= {fmt_dur(r['allowed_downtime_seconds'])} of full downtime)",
           f"**Consumed so far:** {pct(r['budget_consumed'])}  **Remaining:** {pct(r['budget_remaining'])}", "",
           "## Burn rates (1.0 = on budget)", "", "| Window | Burn rate |", "|---|---|"]
    for w, v in r["burn_rates"].items():
        out.append(f"| {w} | {'n/a' if v is None else f'{v:.2f}'} |")
    out += ["", "## Alert policy", "", "| Severity | Long / short window | Threshold | Long burn | Short burn | Status |",
            "|---|---|---|---|---|---|"]
    for a in r["alerts"]:
        lb = "n/a" if a["long_burn"] is None else f"{a['long_burn']:.2f}"
        sb = "n/a" if a["short_burn"] is None else f"{a['short_burn']:.2f}"
        out.append(f"| {a['severity']} | {a['long']} / {a['short']} | {a['threshold']:.1f} | {lb} | {sb} | {a['status']} |")
    out += ["", "## Time to exhaustion at recent burn"]
    for w, s in r["exhaustion_seconds"].items():
        out.append(f"- at last {w} rate: " + ("budget already exhausted" if s == 0 else ("no errors in window" if s is None else fmt_dur(s))))
    wi = r["worst_interval"]
    out += ["", f"Worst interval: {wi['start']} with {wi['bad']:,} bad of {wi['total']:,} ({100 * wi['error_rate']:.2f}%)"]
    return "\n".join(out)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    t0 = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)

    def series(n, step_min, total, bad_fn):
        return [(t0 + dt.timedelta(minutes=step_min * i), total, bad_fn(i)) for i in range(n)]

    ok("duration parse", parse_duration("28d") == 28 * 86400 and parse_duration("5m") == 300)
    ok("duration format", fmt_dur(43.2 * 60) == "43m 12s" and fmt_dur(86400 * 3) == "3d")
    a = allowance_table(99.9, "30d")
    ok("99.9% of 30d allows 43m 12s", round(a["allowed_downtime_seconds"]) == 2592)
    ok("page threshold 14.4 for 30d", abs(a["thresholds"][0]["burn_rate"] - 14.4) < 1e-9)
    ok("28d threshold scales to 13.44", abs(allowance_table(99.9, "28d")["thresholds"][0]["burn_rate"] - 13.44) < 1e-9)
    flat = series(4 * 24 * 4, 15, 10000, lambda i: 10)          # 0.1% errors at 99.9 -> burn 1.0
    r = analyze(flat, 99.9, "30d")
    ok("steady burn rate is 1.0", abs(r["burn_rates"]["1h"] - 1.0) < 1e-9 and abs(r["burn_rates"]["1d"] - 1.0) < 1e-9)
    ok("no alert at burn 1.0", all(x["status"] != "FIRING" for x in r["alerts"][:3]))
    ok("burn 1.0 over 4 of 30 days consumes 4/30", abs(r["budget_consumed"] - 4 / 30) < 1e-9)
    ok("SLI computed", abs(r["sli_percent"] - 99.9) < 1e-9)
    spike = series(4 * 24 * 4, 15, 10000, lambda i: 200 if i >= 4 * 24 * 4 - 4 else 5)   # last hour 2% errors
    r = analyze(spike, 99.9, "30d")
    ok("fast burn pages (1h/5m)", r["alerts"][0]["status"] == "FIRING")
    ok("short window rounded up to step", abs(r["burn_rates"]["5m"] - 20.0) < 1e-9)
    ok("6h page not firing for 1h spike", r["alerts"][1]["status"] == "ok")
    recovered = spike[:-2] + [(spike[-2][0], 10000, 5), (spike[-1][0], 10000, 5)]
    r = analyze(recovered, 99.9, "30d")
    ok("short window resets page after recovery", r["alerts"][0]["status"] == "ok")
    short = series(8, 15, 1000, lambda i: 0)
    r = analyze(short, 99.9, "30d")
    ok("long windows not evaluable on 2h of data", r["alerts"][3]["status"] == "not evaluable" and r["burn_rates"]["3d"] is None)
    gap = flat[:10] + flat[12:]
    ok("missing intervals counted", analyze(gap, 99.9, "30d")["missing_intervals"] == 2)
    ok("exhaustion time at burn 1", abs(analyze(flat, 99.9, "30d")["exhaustion_seconds"]["1d"] - 26 * 86400) < 1)
    rows = load_rows(["time,requests,errors", "2026-01-01T00:00:00Z,100,1", "2026-01-01T00:05:00Z,100,0"])
    ok("aliases requests/errors", rows[0][1] == 100 and rows[0][2] == 1)
    rows = load_rows(["ts,total,good", "1767225600,50,49", "1767225900,50,50"])
    ok("unix timestamps and good column", rows[0][2] == 1 and rows[0][0] == t0)
    for data, label in [(["ts,total", "x,1"], "missing good/bad rejected"),
                        (["ts,total,bad", "2026-01-01T00:00:00Z,10,11", "2026-01-01T00:05:00Z,10,0"], "bad > total rejected"),
                        (["ts,total,bad", "nope,10,1", "2026-01-01T00:05:00Z,10,0"], "bad timestamp rejected")]:
        try:
            load_rows(data)
            ok(label, False)
        except SLOError:
            ok(label, True)
    try:
        analyze(flat, 100, "30d")
        ok("SLO 100 rejected", False)
    except SLOError:
        ok("SLO 100 rejected", True)
    ok("markdown has alert table", "## Alert policy" in to_markdown(analyze(flat, 99.9, "30d")))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Compute SLO error budget consumption and multiwindow burn-rate alerts.")
    ap.add_argument("file", nargs="?", help="CSV of request counts per interval ('-' for stdin)")
    ap.add_argument("--slo", type=float, default=99.9, help="target in percent (default 99.9)")
    ap.add_argument("--window", default="30d", help="SLO window, e.g. 28d or 30d (default 30d)")
    ap.add_argument("--now", help="evaluate as of this ISO time (default: end of the last interval)")
    ap.add_argument("--allowance", action="store_true", help="print allowed downtime and alert thresholds only")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    try:
        if a.allowance:
            t = allowance_table(a.slo, a.window)
            if a.json:
                print(json.dumps(t, indent=2))
            else:
                print(f"SLO {a.slo}% over {a.window}: {fmt_dur(t['allowed_downtime_seconds'])} of full downtime "
                      f"({fmt_dur(t['per_day_seconds'])}/day, {fmt_dur(t['per_week_seconds'])}/week)")
                for x in t["thresholds"]:
                    print(f"  {x['severity']:6} {x['share'] * 100:.0f}% of budget in {x['long']} (short {x['short']}): burn rate >= {x['burn_rate']:.2f}")
            return 0
        if not a.file:
            ap.error("CSV file required (or --allowance / --self-test)")
        lines = sys.stdin.read().splitlines() if a.file == "-" else open(a.file, encoding="utf-8").read().splitlines()
        r = analyze(load_rows(lines), a.slo, a.window, parse_ts(a.now) if a.now else None)
    except (OSError, SLOError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    paging = any(x["severity"] == "page" and x["status"] == "FIRING" for x in r["alerts"])
    return 1 if paging or r["budget_remaining"] <= 0 else 0


if __name__ == "__main__":
    sys.exit(main())

---
name: disk-capacity-runway-forecaster
description: Dự báo thời điểm các ổ đĩa và phân vùng (volume) sẽ đầy dựa trên các mẫu dữ liệu mức sử dụng (tệp CSV gồm dấu thời gian, phân vùng, dung lượng đã dùng và tổng dung lượng). Ước lượng tốc độ tăng trưởng tổng thể và trong tuần gần nhất, loại bỏ các bước nhảy đột biến một lần khỏi xu hướng, đồng thời đưa ra số ngày và ngày cụ thể chạm ngưỡng đầy 80, 90, 95 và 100% cùng mức dung lượng đề xuất để duy trì khoảng dự phòng 90 ngày. Cảnh báo các phân vùng sắp đầy, tốc độ tăng trưởng đang tăng tốc, các bước nhảy đột ngột, xu hướng nhiễu và dữ liệu quá mỏng. Bao gồm công cụ dự báo viết bằng thư viện chuẩn Python kèm tùy chọn --self-test.
---

# Công cụ Dự báo Thời gian Cạn kiệt Dung lượng Ổ đĩa

Bạn giúp các kỹ sư trực ca (on-call) và quản trị viên hệ thống trả lời câu hỏi "khi nào ổ đĩa này sẽ đầy, và chúng ta nên làm gì trước thời điểm đó?". Từ các mẫu dữ liệu mức sử dụng được xuất ra từ hệ thống giám sát (Prometheus, CloudWatch, Datadog, Zabbix, hoặc một cron job chạy lệnh `df`), bạn tính toán thời gian còn lại (runway) cho từng phân vùng, xếp hạng các phân vùng theo mức độ khẩn cấp, giải thích lý do vì sao dự báo có thể sai lệch, và đề xuất các biện pháp xử lý cụ thể: dọn dẹp, xoay vòng nhật ký (rotate), mở rộng dung lượng (resize) hoặc di chuyển dữ liệu.

## Các tệp trong kỹ năng này

- `scripts/disk_runway.py` - đọc tệp CSV chứa các mẫu dữ liệu, ước lượng tốc độ tăng trưởng và in ra bảng xếp hạng thời gian cạn kiệt dung lượng kèm theo các phát hiện và kích thước đề xuất (chỉ dùng thư viện chuẩn Python 3)
- `references/forecasting-method.md` - cách tính xu hướng, tốc độ gần đây, các bước nhảy đột biến một lần, thời gian còn lại và kích thước đề xuất, cùng những giới hạn của phương pháp dự báo tuyến tính
- `references/disk-full-runbook.md` - cách tìm nguyên nhân gây tăng dung lượng, các bước dọn dẹp an toàn, chính sách lưu trữ nhật ký và bản sao lưu, cách mở rộng dung lượng trên Linux và trên đám mây, ngưỡng cảnh báo
- `templates/capacity-review.md` - mẫu báo cáo đánh giá dung lượng để chia sẻ với nhóm hoặc đính kèm vào ticket
- `examples/fleet-disk-usage.csv` - các mẫu dữ liệu hàng ngày trong ba tuần của bốn phân vùng: một cơ sở dữ liệu tăng trưởng ổn định, một phân vùng chứa log có xoay vòng, một CI runner đang tăng tốc độ tiêu thụ và một thiết bị lưu trữ sao lưu NAS có một bước nhảy đột biến một lần

## Quy trình làm việc

### 1. Thu thập các mẫu dữ liệu
Yêu cầu cung cấp các mẫu dữ liệu của ít nhất một tuần (tốt nhất là 3-4 tuần) cho mỗi phân vùng: dấu thời gian (timestamp), phân vùng (host:mount), số GB đã dùng và tổng số GB kích thước. Ví dụ về các nguồn dữ liệu: Prometheus `node_filesystem_size_bytes - node_filesystem_avail_bytes`, CloudWatch `disk_used_percent` kết hợp với kích thước phân vùng, hoặc lệnh `df -BG --output=target,used,size` từ cron. Đồng thời hỏi xem dữ liệu gì đang nằm trên mỗi phân vùng và các sự kiện đã biết (di chuyển dữ liệu, bản sao lưu mới, đợt phát hành mới).

### 2. Chạy công cụ dự báo
```
python3 scripts/disk_runway.py examples/fleet-disk-usage.csv
python3 scripts/disk_runway.py usage.csv --horizon 180 --json
python3 scripts/disk_runway.py --self-test
```
Mã thoát (exit code) 1 nghĩa là có phân vùng ở mức CRITICAL hoặc HIGH, 2 nghĩa là dữ liệu đầu vào bị lỗi.

### 3. Giải thích và hành động
Bắt đầu với các phân vùng ở mức CRITICAL và HIGH. Với mỗi phân vùng, hãy nêu rõ thứ gì đang phình to và với tốc độ bao nhiêu, ngày nào sẽ chạm ngưỡng 90% và 100%, cùng hành động an toàn nhanh nhất từ `references/disk-full-runbook.md`. Hãy xem các kết quả có gắn cờ NOISY_TREND và FEW_SAMPLES là ước tính sơ bộ. Đối với SUDDEN_JUMP, hãy xác nhận nguyên nhân (một bộ sao lưu mới, một bản kết xuất cơ sở dữ liệu, một đợt bùng nổ log) trước khi lập kế hoạch xoay quanh nó. Sử dụng `references/forecasting-method.md` để giải thích lý do vì sao tốc độ gần đây có thể được ưu tiên áp dụng hơn xu hướng dài hạn.

### 4. Bàn giao kết quả
Điền vào `templates/capacity-review.md` bảng số liệu, các hành động cần làm, người phụ trách và thời hạn, đồng thời đề xuất các ngưỡng cảnh báo (ví dụ: cảnh báo warning ở mức 80% hoặc còn 14 ngày dung lượng, gọi trực khẩn cấp page ở mức 90% hoặc còn 3 ngày).

## Kết quả ví dụ
Đối với `examples/fleet-disk-usage.csv` (mẫu cuối cùng lúc 2026-10-21 06:00): ci-runner01:/ ở mức HIGH, đã đầy 80% và đang tăng 5.6 GB/ngày trong tuần qua so với 2.1 tổng thể, sẽ chạm 90% sau 3.6 ngày và đầy hoàn toàn vào ngày 2026-10-28; db01:/var/lib/postgresql tăng đều đặn 3 GB/ngày và chạm ngưỡng 80% sau 13.3 ngày (mức MEDIUM, dung lượng đề xuất 790 GB); backup-nas có một bước nhảy đột biến một lần +405.6 GB vào ngày 2026-10-16 không bị ngoại suy tiếp, còn 120 ngày nữa mới tới ngưỡng 80%; web01:/var/log ở mức OK nhưng dữ liệu nhiễu (do xoay vòng log).

## Quy tắc

- Không bao giờ gợi ý xóa dữ liệu mà không nêu chính xác cần xóa gì, cách kiểm tra tính an toàn và cách khôi phục; hãy ưu tiên xoay vòng log, nén dữ liệu, áp dụng chính sách lưu giữ (retention) và mở rộng dung lượng.
- Nêu rõ ngày dự báo được tính dựa trên tốc độ nào và lưu ý rằng các dự báo tuyến tính sẽ không tính đến các sự kiện đột xuất.
- Thời gian trong báo cáo tuân theo các dấu thời gian trong dữ liệu; hãy nói rõ chúng đang ở múi giờ nào.
- Không tự mình chạy các lệnh dọn dẹp hoặc mở rộng dung lượng trên hệ thống đang hoạt động (live systems); hãy cung cấp câu lệnh để người dùng tự kiểm tra và thực thi.
FILE:references/forecasting-method.md
# Phương pháp Dự báo

## Đầu vào
Mỗi mẫu dữ liệu là một dòng: `timestamp,volume,used_gb,size_gb`. Các dòng có thể chưa được sắp xếp và trộn lẫn giữa nhiều phân vùng. Giá trị `size_gb` mới nhất sẽ được sử dụng, nhờ đó việc thay đổi kích thước trong khoảng thời gian lấy mẫu vẫn được xử lý chính xác.

## Tốc độ tăng trưởng
- **Tốc độ tổng thể (Overall rate):** hệ số góc bình phương tối thiểu (least-squares slope) của số GB đã dùng theo thời gian (GB/ngày) trên toàn bộ các mẫu dữ liệu. Hệ số R^2 (từ 0 đến 1) cho biết mức độ khớp của đường thẳng: xấp xỉ 1.0 đối với mức tăng trưởng đều đặn, dưới 0.5 đối với các mẫu hình răng cưa như xoay vòng log hoặc tệp tạm.
- **Tốc độ gần đây (Recent rate):** cùng hệ số góc đó nhưng tính trên 7 ngày gần nhất (hoặc một phần ba dữ liệu cuối cùng khi tổng thời gian ít hơn 14 ngày). Chỉ số này giúp phát hiện mức tăng trưởng vừa mới tăng tốc.
- **Tốc độ tính thời gian cạn kiệt (Runway rate):** lấy giá trị xấu hơn (cao hơn) giữa hai tốc độ trên. Khi tốc độ gần đây lớn hơn 1.5 lần tốc độ tổng thể, phân vùng đó sẽ được gắn cờ ACCELERATING (đang tăng tốc).

## Các bước nhảy đột biến một lần (One-off jumps)
Một bước nhảy giữa hai mẫu dữ liệu lớn hơn 5% kích thước phân vùng được coi là một sự kiện đơn lẻ (một bộ sao lưu mới, một bản dump vừa khôi phục, một tệp tải lên kích thước lớn). Phần dung lượng này vẫn được giữ nguyên trong mức sử dụng hiện tại nhưng được loại bỏ khỏi xu hướng tăng trưởng: các điểm dữ liệu phía sau được dịch xuống một khoảng bằng độ lớn bước nhảy trừ đi mức tăng thông thường, nhằm tránh việc ngoại suy một sự kiện đơn lẻ kéo dài mãi mãi. Bước nhảy này được báo cáo là SUDDEN_JUMP để con người có thể xác nhận xem nó có thực sự là sự kiện một lần hay không. Nếu sự kiện này lặp lại (chẳng hạn sao lưu toàn bộ hàng tuần), hãy thu thập dữ liệu dài hơn để quy luật đó trở thành một phần của xu hướng, hoặc chủ động tính riêng bước nhảy tiếp theo vào kế hoạch.

## Thời gian còn lại (Runway)
Đối với mỗi ngưỡng T trong các mức 80, 90, 95 và 100%:
`số ngày = (T x kích thước - mức dùng hiện tại) / tốc độ tính thời gian cạn kiệt`, được tính từ mẫu dữ liệu cuối cùng. Giá trị bằng 0 nghĩa là ngưỡng đó đã bị vượt qua; "never" (không bao giờ) nghĩa là mức sử dụng đang đi ngang hoặc giảm xuống.
Ngày dự kiến bằng thời điểm của mẫu dữ liệu cuối cùng cộng thêm số ngày còn lại, theo cùng múi giờ với dữ liệu gốc.

## Kích thước đề xuất
Bội số nhỏ nhất của 10 GB giúp giữ mức sử dụng dưới 80% trong toàn bộ khung thời gian dự báo (horizon, mặc định là 90 ngày):
`kích thước >= (mức dùng hiện tại + tốc độ tính thời gian cạn kiệt x khung thời gian) / 0.8`, và không bao giờ nhỏ hơn kích thước hiện tại.
Ví dụ: đã dùng 128 GB, tăng 2 GB/ngày, khung thời gian 180 ngày -> (128 + 360) / 0.8 = 610 GB.

## Các mức trạng thái
| Trạng thái | Quy tắc |
|---|---|
| CRITICAL | Đã dùng từ 95% trở lên, hoặc sẽ đầy trong vòng 7 ngày |
| HIGH | Chạm ngưỡng 90% trong vòng 14 ngày |
| MEDIUM | Chạm ngưỡng 80% trong vòng 30 ngày, hoặc đang tăng tốc, hoặc có bước nhảy đột ngột |
| OK | Không thuộc các trường hợp trên |

## Những giới hạn
- Một đường thẳng không thể nhìn thấy các sự kiện trong tương lai: một đợt di chuyển dữ liệu, một khách hàng mới, hay mức ghi log bị bỏ quên ở chế độ DEBUG.
- Hệ thống tệp (filesystem) thường gặp trục trặc trước khi chạm mức 100%: ext4 dành riêng khoảng 5% cho quyền root, cơ sở dữ liệu cần không gian trống để chạy vacuum, nén dữ liệu (compaction) hoặc xây dựng chỉ mục (index), và một số luồng ghi dữ liệu đã bắt đầu lỗi ở mức 90-95%.
- Cạn kiệt inode (`df -i`) có thể làm đầy ổ đĩa ngay cả khi vẫn còn trống rất nhiều GB; hãy dự báo số lượng inode theo cách tương tự nếu hệ thống ghi nhiều tệp nhỏ.
- Lưu trữ cấp phát mỏng (thin-provisioned) và lưu trữ dùng chung (LVM thin pools, SAN, Ceph) có thể bị đầy ở tầng bên dưới ngay cả khi phân vùng bên trên trông vẫn còn trống.
FILE:references/disk-full-runbook.md
# Sổ tay Xử lý Ổ đĩa Đầy (Disk Full Runbook)

## 1. Tìm nguyên nhân gây tăng dung lượng
```
df -h /mount; df -i /mount                       # dung lượng và số inode
sudo du -xh --max-depth=2 /mount | sort -h | tail -20
sudo find /mount -xdev -type f -size +1G -mtime -7 -exec ls -lh {} +
sudo lsof +L1                                    # các tệp đã xóa nhưng vẫn đang được tiến trình giữ mở
journalctl --disk-usage
docker system df                                 # images, containers, volumes, build cache
```
Các tệp đã xóa nhưng vẫn đang mở sẽ tiếp tục chiếm dung lượng cho đến khi tiến trình khởi động lại hoặc mở lại tệp log.

## 2. Các giải pháp xử lý nhanh an toàn (kiểm tra kỹ trước khi xóa)
| Nguồn tiêu thụ | Hành động |
|---|---|
| Nhật ký ứng dụng (Application logs) | cấu hình logrotate với `compress`, `maxsize`, `rotate N`; đẩy log về hệ thống lưu trữ tập trung |
| Nhật ký systemd (systemd journal) | chạy `journalctl --vacuum-size=1G`, thiết lập `SystemMaxUse=` |
| Docker / CI runners | chạy `docker image prune -a --filter until=168h`, `docker builder prune`; dọn sạch các thư mục làm việc của runner |
| Bộ nhớ đệm gói cài đặt (Package caches) | chạy `apt-get clean`, `dnf clean all`, xóa các bản nhân (kernel) cũ |
| Cơ sở dữ liệu | lưu trữ (archive) hoặc xóa các phân vùng bảng (partition) cũ, chạy `VACUUM` / đóng gói lại bảng (repack), kiểm tra thời gian lưu giữ WAL hoặc binlog và các khe nhân bản (replication slots) bị kẹt |
| Bản sao lưu (Backups) | áp dụng chính sách lưu giữ; chuyển các bộ sao lưu cũ sang lưu trữ đối tượng (object storage) hoặc tầng lưu trữ lạnh (cold tier) |
| Core dumps, heap dumps, tmp | xóa sau khi người phụ trách sự cố xác nhận |
Không bao giờ dùng lệnh `rm` với tệp mà cơ sở dữ liệu đang sử dụng; không bao giờ tự xóa WAL hoặc binlog bằng tay.

## 3. Mở rộng dung lượng (Resize)
- **AWS EBS:** sửa đổi kích thước volume, đợi chuyển sang trạng thái optimizing, sau đó chạy `sudo growpart /dev/nvme0n1 1` và `sudo resize2fs /dev/nvme0n1p1` (đối với ext4) hoặc `sudo xfs_growfs /mount` (đối với xfs). Mỗi volume chỉ được tăng kích thước một lần mỗi 6 giờ.
- **Azure managed disk / GCP persistent disk:** tăng kích thước đĩa (Azure có thể yêu cầu giải phóng VM - deallocate đối với một số loại đĩa), sau đó mở rộng phân vùng và hệ thống tệp theo cách tương tự.
- **LVM:** lệnh `lvextend -r -L +100G vg/lv` sẽ đồng thời mở rộng luôn cả hệ thống tệp.
- **Kubernetes PVC:** chỉnh sửa `spec.resources.requests.storage` khi StorageClass có thiết lập `allowVolumeExpansion: true`.
Các phân vùng có thể mở rộng nhưng hiếm khi thu nhỏ lại được, vì vậy hãy tăng kích thước khớp với mức dự báo chứ đừng tăng vượt quá xa.

## 4. Thiết lập cảnh báo đủ thời gian để xử lý
- Cảnh báo (Warn) ở mức đã dùng 80% hoặc còn 14 ngày dung lượng; gọi trực khẩn cấp (Page) ở mức 90% hoặc còn 3 ngày dung lượng (trên Prometheus dùng các quy tắc dạng `predict_linear(node_filesystem_avail_bytes[6h], 4*3600) < 0`).
- Cảnh báo khi mức sử dụng inode vượt quá 85%.
- Rà soát dung lượng hàng tháng bằng báo cáo của kỹ năng này; cộng thêm mức tăng trưởng dự kiến từ các đợt ra mắt sản phẩm đã lên kế hoạch bằng tay.
FILE:templates/capacity-review.md
# Báo cáo Đánh giá Dung lượng - <dịch vụ / cụm máy chủ> - <ngày>

**Dữ liệu:** <nguồn>, từ <mẫu đầu tiên> đến <mẫu cuối cùng> (<múi giờ>)  **Khung thời gian dự báo:** <90> ngày  **Người đánh giá:** <tên>

## Thời gian cạn kiệt dung lượng (Runway)
| Phân vùng | Trạng thái | Đã dùng | Kích thước | Tốc độ tăng GB/ngày (tổng thể / gần đây) | Chạm 90% vào | Đầy vào | Kích thước đề xuất |
|---|---|---|---|---|---|---|---|
| | | | | | | | |

## Nguyên nhân gây tăng dung lượng
| Phân vùng | Thành phần tiêu thụ chính | Bằng chứng | Đột biến một lần hay diễn ra liên tục |
|---|---|---|---|
| | | | |

## Hành động cần thực hiện
| Phân vùng | Hành động | Người phụ trách | Thời hạn | Dung lượng thu hồi / mở rộng |
|---|---|---|---|---|
| | | | | |

## Cảnh báo
- Cảnh báo (Warning): <80% hoặc còn 14 ngày>  - Gọi trực (Page): <90% hoặc còn 3 ngày>  - Inodes: <85%>

## Đợt rà soát tiếp theo
<ngày>; chạy lại công cụ dự báo với các mẫu dữ liệu mới nhất
FILE:examples/fleet-disk-usage.csv
timestamp,volume,used_gb,size_gb
2026-10-01T06:00,db01:/var/lib/postgresql,300.0,500
2026-10-01T06:00,web01:/var/log,21.0,100
2026-10-01T06:00,ci-runner01:/,110.0,200
2026-10-01T06:00,backup-nas:/srv/backups,2100.0,4000
2026-10-02T06:00,db01:/var/lib/postgresql,303.4,500
2026-10-02T06:00,web01:/var/log,24.4,100
2026-10-02T06:00,ci-runner01:/,111.4,200
2026-10-02T06:00,backup-nas:/srv/backups,2105.4,4000
2026-10-03T06:00,db01:/var/lib/postgresql,305.7,500
2026-10-03T06:00,web01:/var/log,26.7,100
2026-10-03T06:00,ci-runner01:/,111.7,200
2026-10-03T06:00,backup-nas:/srv/backups,2109.7,4000
2026-10-04T06:00,db01:/var/lib/postgresql,309.2,500
2026-10-04T06:00,web01:/var/log,30.2,100
2026-10-04T06:00,ci-runner01:/,113.2,200
2026-10-04T06:00,backup-nas:/srv/backups,2115.2,4000
2026-10-05T06:00,db01:/var/lib/postgresql,311.9,500
2026-10-05T06:00,web01:/var/log,32.9,100
2026-10-05T06:00,ci-runner01:/,113.9,200
2026-10-05T06:00,backup-nas:/srv/backups,2119.9,4000
2026-10-06T06:00,db01:/var/lib/postgresql,315.5,500
2026-10-06T06:00,web01:/var/log,22.5,100
2026-10-06T06:00,ci-runner01:/,115.5,200
2026-10-06T06:00,backup-nas:/srv/backups,2125.5,4000
2026-10-07T06:00,db01:/var/lib/postgresql,317.6,500
2026-10-07T06:00,web01:/var/log,24.6,100
2026-10-07T06:00,ci-runner01:/,115.6,200
2026-10-07T06:00,backup-nas:/srv/backups,2129.6,4000
2026-10-08T06:00,db01:/var/lib/postgresql,321.1,500
2026-10-08T06:00,web01:/var/log,28.1,100
2026-10-08T06:00,ci-runner01:/,117.1,200
2026-10-08T06:00,backup-nas:/srv/backups,2135.1,4000
2026-10-09T06:00,db01:/var/lib/postgresql,324.3,500
2026-10-09T06:00,web01:/var/log,31.3,100
2026-10-09T06:00,ci-runner01:/,118.3,200
2026-10-09T06:00,backup-nas:/srv/backups,2140.3,4000
2026-10-10T06:00,db01:/var/lib/postgresql,326.8,500
2026-10-10T06:00,web01:/var/log,33.8,100
2026-10-10T06:00,ci-runner01:/,118.8,200
2026-10-10T06:00,backup-nas:/srv/backups,2144.8,4000
2026-10-11T06:00,db01:/var/lib/postgresql,330.0,500
2026-10-11T06:00,web01:/var/log,23.0,100
2026-10-11T06:00,ci-runner01:/,120.0,200
2026-10-11T06:00,backup-nas:/srv/backups,2150.0,4000
2026-10-12T06:00,db01:/var/lib/postgresql,333.4,500
2026-10-12T06:00,web01:/var/log,26.4,100
2026-10-12T06:00,ci-runner01:/,121.4,200
2026-10-12T06:00,backup-nas:/srv/backups,2155.4,4000
2026-10-13T06:00,db01:/var/lib/postgresql,335.7,500
2026-10-13T06:00,web01:/var/log,28.7,100
2026-10-13T06:00,ci-runner01:/,121.7,200
2026-10-13T06:00,backup-nas:/srv/backups,2159.7,4000
2026-10-14T06:00,db01:/var/lib/postgresql,339.2,500
2026-10-14T06:00,web01:/var/log,32.2,100
2026-10-14T06:00,ci-runner01:/,123.2,200
2026-10-14T06:00,backup-nas:/srv/backups,2165.2,4000
2026-10-15T06:00,db01:/var/lib/postgresql,341.9,500
2026-10-15T06:00,web01:/var/log,34.9,100
2026-10-15T06:00,ci-runner01:/,123.9,200
2026-10-15T06:00,backup-nas:/srv/backups,2169.9,4000
2026-10-16T06:00,db01:/var/lib/postgresql,345.5,500
2026-10-16T06:00,web01:/var/log,24.5,100
2026-10-16T06:00,ci-runner01:/,130.5,200
2026-10-16T06:00,backup-nas:/srv/backups,2575.5,4000
2026-10-17T06:00,db01:/var/lib/postgresql,347.6,500
2026-10-17T06:00,web01:/var/log,26.6,100
2026-10-17T06:00,ci-runner01:/,135.6,200
2026-10-17T06:00,backup-nas:/srv/backups,2579.6,4000
2026-10-18T06:00,db01:/var/lib/postgresql,351.1,500
2026-10-18T06:00,web01:/var/log,30.1,100
2026-10-18T06:00,ci-runner01:/,142.1,200
2026-10-18T06:00,backup-nas:/srv/backups,2585.1,4000
2026-10-19T06:00,db01:/var/lib/postgresql,354.3,500
2026-10-19T06:00,web01:/var/log,33.3,100
2026-10-19T06:00,ci-runner01:/,148.3,200
2026-10-19T06:00,backup-nas:/srv/backups,2590.3,4000
2026-10-20T06:00,db01:/var/lib/postgresql,356.8,500
2026-10-20T06:00,web01:/var/log,35.8,100
2026-10-20T06:00,ci-runner01:/,153.8,200
2026-10-20T06:00,backup-nas:/srv/backups,2594.8,4000
2026-10-21T06:00,db01:/var/lib/postgresql,360.1,500
2026-10-21T06:00,web01:/var/log,25.1,100
2026-10-21T06:00,ci-runner01:/,160.1,200
2026-10-21T06:00,backup-nas:/srv/backups,2600.1,4000
FILE:scripts/disk_runway.py
#!/usr/bin/env python3
"""Disk capacity runway forecaster (Python 3 standard library only).

Input: CSV with a header row and columns
  timestamp,volume,used_gb,size_gb
timestamp is ISO 8601 ("2026-10-01T00:00", "2026-10-01 06:00" or a date); one row per sample,
any order, several volumes in one file. size_gb may change (a resize); the latest value is used.

Model per volume:
  - growth rate = least-squares slope of used_gb over time (GB/day), plus R^2 for fit quality;
    one-off steps larger than 5% of the size are kept in current usage but removed from the trend
  - recent rate = slope over the last 7 days of samples (or the last third if shorter)
  - runway to 80/90/95/100% full = (threshold x size - used now) / max(rate, recent rate),
    counted from the LAST sample (not from today), so results are reproducible
  - suggested size: smallest whole 10 GB step that keeps the volume under 80% for
    --horizon days (default 90) at the worse of the two rates
Status / findings (severity):
  CRITICAL FULL_SOON       100% within 7 days, or already >= 95% used
  HIGH     RUNWAY_SHORT    90% within 14 days
  MEDIUM   RUNWAY_WATCH    80% within 30 days
  MEDIUM   ACCELERATING    recent rate > 1.5x the overall rate (and > 0.1 GB/day)
  MEDIUM   SUDDEN_JUMP     one step grew by more than 5% of the volume size
  LOW      NOISY_TREND     R^2 < 0.5: usage swings (log rotation, temp files); trust less
  LOW      SHRINKING       usage is falling; no runway
  LOW      FEW_SAMPLES     fewer than 4 samples or under 2 days of data
Usage:
  python3 disk_runway.py usage.csv
  python3 disk_runway.py usage.csv --horizon 180 --json
  python3 disk_runway.py --self-test
Exit code: 0 no CRITICAL/HIGH, 1 CRITICAL or HIGH findings, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import sys

THRESHOLDS = (0.8, 0.9, 0.95, 1.0)


class DataError(ValueError):
    pass


def parse_time(s):
    s = s.strip().replace("Z", "+00:00")
    try:
        t = dt.datetime.fromisoformat(s)
    except ValueError:
        raise DataError(f"bad timestamp {s!r}")
    return t.replace(tzinfo=None) if t.tzinfo is None else t.astimezone(dt.timezone.utc).replace(tzinfo=None)


def read_rows(text):
    rd = csv.DictReader(io.StringIO(text))
    need = {"timestamp", "volume", "used_gb", "size_gb"}
    if not rd.fieldnames or not need <= {f.strip() for f in rd.fieldnames}:
        raise DataError(f"CSV header must contain {', '.join(sorted(need))}")
    vols = {}
    for n, row in enumerate(rd, start=2):
        row = {k.strip(): (v or "").strip() for k, v in row.items() if k}
        if not any(row.values()):
            continue
        try:
            used, size = float(row["used_gb"]), float(row["size_gb"])
        except ValueError:
            raise DataError(f"line {n}: used_gb and size_gb must be numbers")
        if size <= 0 or used < 0:
            raise DataError(f"line {n}: size_gb must be > 0 and used_gb >= 0")
        vols.setdefault(row["volume"] or "(unnamed)", []).append((parse_time(row["timestamp"]), used, size))
    if not vols:
        raise DataError("no data rows")
    return {k: sorted(v) for k, v in vols.items()}


def slope(points):
    """points: [(days, value)] -> (slope per day, r2)"""
    n = len(points)
    if n < 2:
        return 0.0, 0.0
    mx = sum(p[0] for p in points) / n
    my = sum(p[1] for p in points) / n
    sxx = sum((p[0] - mx) ** 2 for p in points)
    if sxx == 0:
        return 0.0, 0.0
    sxy = sum((p[0] - mx) * (p[1] - my) for p in points)
    b = sxy / sxx
    syy = sum((p[1] - my) ** 2 for p in points)
    r2 = 1.0 if syy == 0 else (sxy * sxy) / (sxx * syy)
    return b, r2


def analyze(samples, horizon=90):
    t0 = samples[0][0]
    last_t, used, size = samples[-1]
    # one-off steps (> 5% of size in one sample) are removed from the trend, not extrapolated
    steps = [samples[i][1] - samples[i - 1][1] for i in range(1, len(samples))]
    normal = sorted(x for x in steps if x <= 0.05 * size)
    typical = normal[len(normal) // 2] if normal else 0.0
    shift, pts = 0.0, []
    for i, (t, u, _) in enumerate(samples):
        if i and steps[i - 1] > 0.05 * size:
            shift += steps[i - 1] - typical
        pts.append(((t - t0).total_seconds() / 86400, u - shift))
    span = pts[-1][0]
    rate, r2 = slope(pts)
    cut = pts[-1][0] - 7 if span >= 14 else pts[-1][0] - span / 3
    recent_pts = [p for p in pts if p[0] >= cut]
    recent, _ = slope(recent_pts) if len(recent_pts) >= 2 else (rate, 0)
    worst = max(rate, recent)
    pct = used / size
    runway = {}
    for th in THRESHOLDS:
        left = th * size - used
        if left <= 0:
            runway[th] = (0.0, last_t)
        elif worst > 1e-9:
            d = left / worst
            runway[th] = (d, last_t + dt.timedelta(days=d))
        else:
            runway[th] = (None, None)
    need = used + max(worst, 0) * horizon
    suggest = max(size, math.ceil(need / 0.8 / 10) * 10)
    findings = []

    def f(sev, code, msg):
        findings.append({"severity": sev, "code": code, "message": msg})

    d100, d90, d80 = runway[1.0][0], runway[0.9][0], runway[0.8][0]
    if pct >= 0.95 or (d100 is not None and d100 <= 7):
        f("CRITICAL", "FULL_SOON", f"{pct:.0%} used; full in {fmt_days(d100)}")
    elif d90 is not None and d90 <= 14:
        f("HIGH", "RUNWAY_SHORT", f"90% in {fmt_days(d90)}")
    elif d80 is not None and d80 <= 30:
        f("MEDIUM", "RUNWAY_WATCH", f"80% in {fmt_days(d80)}")
    if rate > 0 and recent > 1.5 * rate and recent > 0.1:
        f("MEDIUM", "ACCELERATING", f"last-week rate {recent:.2f} GB/day vs overall {rate:.2f} GB/day")
    jumps = [(samples[i][0], samples[i][1] - samples[i - 1][1]) for i in range(1, len(samples))]
    big = [j for j in jumps if j[1] > 0.05 * size]
    if big:
        t, g = max(big, key=lambda j: j[1])
        f("MEDIUM", "SUDDEN_JUMP", f"+{g:.1f} GB in one step at {t:%Y-%m-%d %H:%M} (backup, dump, log burst?)")
    if len(pts) >= 4 and r2 < 0.5 and worst > 0:
        f("LOW", "NOISY_TREND", f"R^2 {r2:.2f}; usage swings, so treat the runway as rough")
    if worst < 0:
        f("LOW", "SHRINKING", f"usage falling {rate:.2f} GB/day")
    if len(pts) < 4 or span < 2:
        f("LOW", "FEW_SAMPLES", f"{len(pts)} samples over {span:.1f} days; collect more data")
    rank = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    findings.sort(key=lambda x: rank[x["severity"]])
    return {"last_sample": last_t.strftime("%Y-%m-%d %H:%M"), "used_gb": round(used, 2), "size_gb": size,
            "used_pct": round(pct * 100, 1), "rate_gb_day": round(rate, 3), "recent_rate_gb_day": round(recent, 3),
            "r2": round(r2, 3), "samples": len(pts), "span_days": round(span, 2),
            "runway": {f"{int(th * 100)}%": {"days": None if d is None else round(d, 1),
                                             "date": None if t is None else t.strftime("%Y-%m-%d")}
                       for th, (d, t) in runway.items()},
            "suggested_size_gb": suggest, "horizon_days": horizon, "findings": findings,
            "status": findings[0]["severity"] if findings and findings[0]["severity"] in ("CRITICAL", "HIGH", "MEDIUM") else "OK"}


def fmt_days(d):
    if d is None:
        return "never at current rate"
    if d <= 0:
        return "already"
    if d < 1:
        return f"{d * 24:.0f} h"
    return f"{d:.1f} days"


def report(res, horizon):
    out = ["# Disk runway report", "", f"Runway uses the worse of the overall and last-week growth rate; suggested size keeps usage under 80% for {horizon} days.", "",
           "| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 80% | 90% | 100% | Suggested size |",
           "|---|---|---|---|---|---|---|---|---|"]
    order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "OK": 3}
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        def cell(k):
            x = r["runway"][k]
            if x["days"] is None:
                return "never"
            return "now" if x["days"] == 0 else f"{x['days']} d ({x['date']})"
        out.append(f"| {name} | {r['status']} | {r['used_gb']:g} GB ({r['used_pct']}%) | {r['size_gb']:g} GB | "
                   f"{r['rate_gb_day']} / {r['recent_rate_gb_day']} | {cell('80%')} | {cell('90%')} | {cell('100%')} | "
                   f"{r['suggested_size_gb']:g} GB |")
    out += ["", "## Findings"]
    any_f = False
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        for x in r["findings"]:
            any_f = True
            out.append(f"- **{x['severity']} {x['code']}** ({name}): {x['message']}")
    if not any_f:
        out.append("- none")
    out += ["", "Last samples: " + ", ".join(f"{n} {r['last_sample']} ({r['samples']} samples, {r['span_days']} days, R^2 {r['r2']})" for n, r in sorted(res.items()))]
    return "\n".join(out) + "\n"


def run(text, horizon=90):
    if horizon <= 0:
        raise DataError("horizon must be > 0")
    return {name: analyze(s, horizon) for name, s in read_rows(text).items()}


def self_test():
    cases = []

    def ok(name, cond):
        cases.append(bool(cond))
        print(("PASS " if cond else "FAIL ") + name)

    b, r2 = slope([(0, 10), (1, 12), (2, 14)])
    ok("slope 2/day exact fit", abs(b - 2) < 1e-9 and abs(r2 - 1) < 1e-9)
    ok("flat slope 0", slope([(0, 5), (1, 5), (2, 5)])[0] == 0)
    hdr = "timestamp,volume,used_gb,size_gb\n"
    lin = hdr + "".join(f"2026-10-{d:02d},v,{100 + 2 * d},500\n" for d in range(1, 15))
    r = run(lin)["v"]
    ok("rate 2 GB/day", abs(r["rate_gb_day"] - 2) < 1e-6)
    ok("80% runway (400-128)/2 = 136 d", r["runway"]["80%"]["days"] == 136.0)
    ok("runway date from last sample", r["runway"]["100%"]["date"] == "2027-04-18")
    ok("healthy volume status OK", r["status"] == "OK" and r["findings"] == [])
    ok("suggested size keeps 500", r["suggested_size_gb"] == 500)
    r = run(lin, horizon=180)["v"]
    ok("180 d horizon suggests 610", r["suggested_size_gb"] == 610)
    full = hdr + "".join(f"2026-10-{d:02d},v,{80 + 3 * d},100\n" for d in range(1, 6))
    r = run(full)["v"]
    ok("95% used is CRITICAL", r["status"] == "CRITICAL" and r["findings"][0]["code"] == "FULL_SOON")
    ok("past threshold runway is 0", r["runway"]["90%"]["days"] == 0)
    short = hdr + "".join(f"2026-10-{d:02d},v,{50 + 3 * d},100\n" for d in range(1, 8))
    r = run(short)["v"]
    ok("90% in ~6 days is HIGH", r["status"] == "HIGH")
    watch = hdr + "".join(f"2026-10-{d:02d},v,{40 + d},100\n" for d in range(1, 8))
    ok("80% in 33 d is OK, 80% in 26 d is MEDIUM", run(watch)["v"]["status"] == "OK" and
       run(hdr + "".join(f"2026-10-{d:02d},v,{47 + d},100\n" for d in range(1, 8)))["v"]["status"] == "MEDIUM")
    acc = hdr + "".join(f"2026-10-{d:02d},v,{100 + (d if d <= 14 else 14 + 5 * (d - 14))},1000\n" for d in range(1, 22))
    r = run(acc)["v"]
    ok("accelerating flagged", any(x["code"] == "ACCELERATING" for x in r["findings"]))
    ok("recent rate used for runway", r["recent_rate_gb_day"] > r["rate_gb_day"])
    jump = hdr + "2026-10-01,v,100,500\n2026-10-02,v,101,500\n2026-10-03,v,140,500\n2026-10-04,v,141,500\n"
    r = run(jump)["v"]
    ok("sudden jump flagged", any(x["code"] == "SUDDEN_JUMP" for x in r["findings"]))
    ok("jump removed from trend (1 GB/day)", abs(r["rate_gb_day"] - 1) < 1e-6 and r["used_gb"] == 141)
    noisy = hdr + "".join(f"2026-10-{d:02d},v,{100 + (8 if d % 2 else 0) + 0.3 * d:.1f},500\n" for d in range(1, 21))
    ok("noisy trend flagged", any(x["code"] == "NOISY_TREND" for x in run(noisy)["v"]["findings"]))
    shr = hdr + "".join(f"2026-10-{d:02d},v,{200 - 3 * d},500\n" for d in range(1, 8))
    r = run(shr)["v"]
    ok("shrinking flagged and runway never", any(x["code"] == "SHRINKING" for x in r["findings"]) and r["runway"]["100%"]["days"] is None)
    few = hdr + "2026-10-01T00:00,v,10,100\n2026-10-01T12:00,v,11,100\n"
    ok("few samples flagged", any(x["code"] == "FEW_SAMPLES" for x in run(few)["v"]["findings"]))
    multi = hdr + "2026-10-02,b,5,100\n2026-10-01,a,1,100\n2026-10-01,b,4,100\n2026-10-02,a,1,100\n"
    r = run(multi)
    ok("multiple volumes, unsorted rows", set(r) == {"a", "b"} and r["b"]["rate_gb_day"] == 1)
    ok("resize uses latest size", run(hdr + "2026-10-01,v,50,100\n2026-10-02,v,52,200\n")["v"]["size_gb"] == 200)
    ok("fmt_days hours", fmt_days(0.5) == "12 h")
    for bad, label in (("a,b\n1,2\n", "missing columns"), (hdr, "no rows"), (hdr + "x,v,1,10\n", "bad timestamp"),
                       (hdr + "2026-10-01,v,abc,10\n", "bad number"), (hdr + "2026-10-01,v,1,0\n", "zero size")):
        try:
            run(bad)
            ok(label + " rejected", False)
        except DataError:
            ok(label + " rejected", True)
    print(f"{sum(cases)}/{len(cases)} passed")
    return 0 if all(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Forecast when disks fill up from usage samples.")
    ap.add_argument("file", nargs="?", help="CSV: timestamp,volume,used_gb,size_gb")
    ap.add_argument("--horizon", type=int, default=90, help="days of runway the suggested size must cover (default 90)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.file:
        ap.error("give a CSV file or --self-test")
    try:
        res = run(open(a.file, encoding="utf-8").read(), a.horizon)
    except (OSError, DataError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(res, indent=2) if a.json else report(res, a.horizon), end="\n" if a.json else "")
    return 1 if any(r["status"] in ("CRITICAL", "HIGH") for r in res.values()) else 0


if __name__ == "__main__":
    sys.exit(main())