Làm nổi bật Dự án của Bạn
Hãy tạo một phần nội dung thuyết phục dựa trên dữ liệu thể hiện tác động của [project name]: số lượt tải xuống, số người dùng được hỗ trợ, số vấn đề (issues) đã giải quyết và thống kê mức tăng trưởng cộng đồng.
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.
---
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())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.
---
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())