sdk-py-trace 0.2.0

Function-level tracing SDK for HICAS workers

sdk-trace

SDK gắn decorator @trace vào hàm Python để tự động thu thập trace và gửi lên hicas-trace-server.

---

Cài đặt

pip install -e /path/to/sdk-trace
# hoặc sau khi publish:
pip install sdk-trace

Yêu cầu: Python ≥ 3.10

---

Bắt đầu nhanh

1. Cấu hình (1 lần duy nhất)

Đặt 3 biến môi trường — SDK tự đọc khi import, không cần gọi thêm gì trong code:

export SDK_TRACE_URL=http://localhost:8899
export SDK_TRACE_API_KEY=ht-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
export SDK_TRACE_PROJECT=ten-project

Hoặc trong file .env:

SDK_TRACE_URL=http://localhost:8899
SDK_TRACE_API_KEY=ht-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
SDK_TRACE_PROJECT=ten-project

2. Gắn decorator

from sdk_trace import trace

@trace(deep=True)
def chay_pipeline(task_id: str) -> dict:
    data = lay_du_lieu(task_id)      # hàm này cũng được trace tự động
    result = xu_ly(data)             # và hàm này nữa
    return result

chay_pipeline("TASK-001")
# → trace tự động gửi lên server, không cần làm gì thêm

3. Xem kết quả

Mở http://localhost:8899 trong trình duyệt.

---

Các dạng @trace

@trace — trace hàm đơn lẻ

@trace
def tinh_tong(a: int, b: int) -> int:
    return a + b

@trace(deep=True) — trace toàn bộ cây gọi hàm

Tự động bắt mọi hàm con trong project mà không cần gắn @trace từng cái:

def _phan_tich(van_ban: str) -> list[str]:
    return van_ban.lower().split()           # không có @trace

def _dem_tu(tokens: list) -> dict:
    freq = {}
    for t in tokens:
        freq[t] = freq.get(t, 0) + 1
    return freq                              # không có @trace

@trace(deep=True)                            # ← chỉ cần gắn ở đây
def phan_tich_van_ban(van_ban: str) -> dict:
    tokens = _phan_tich(van_ban)
    freq = _dem_tu(tokens)
    return {"tokens": len(tokens), "freq": freq}

@trace với nhãn tuỳ chọn

@trace(label="bước 1: chuẩn hoá dữ liệu")
def chuan_hoa(df):
    ...

Hàm async

@trace(deep=True)
async def xu_ly_async(doc_id: str) -> dict:
    data = await fetch_doc(doc_id)
    return analyze(data)

---

Tham số @trace

Tham số	Kiểu	Mặc định	Mô tả
deep	bool	False	Tự bắt mọi hàm con trong project
label	str | None	None	Nhãn hiển thị thêm trong UI
exclude	set[str]	None	Tên hàm muốn ẩn hoàn toàn (khi deep=True)
max_calls_per_site	int | None	None	Giới hạn lần log trong vòng lặp

Ví dụ kiểm soát vòng lặp

Khi deep=True bắt gặp vòng lặp gọi cùng 1 hàm nhiều lần, dùng max_calls_per_site để tránh phình trace:

@trace(
    deep=True,
    max_calls_per_site=3,   # log chi tiết 3 lần đầu, sau đó gộp: "⋯ +997 lần gọi lặp lại"
    exclude={"_log", "_validate_schema"},   # ẩn hẳn 2 helper này
)
def xu_ly_batch(items: list) -> list:
    return [_xu_ly_mot(item) for item in items]  # 1000 items → vẫn trace được

---

Lồng nhiều hàm có @trace

Khi nhiều hàm đều có @trace, chỉ hàm ngoài cùng (top-level) flush lên server:

@trace
def buoc_1(x): return x * 2

@trace
def buoc_2(x): return x + 10

@trace
def pipeline(x):        # top-level → flush sau khi xong
    a = buoc_1(x)       # được ghi vào cùng 1 trace
    b = buoc_2(a)
    return b

pipeline(5)             # → 1 trace duy nhất, 3 event

---

Điều gì được ghi vào trace?

Mỗi lần gọi hàm (event) ghi lại:

Trường	Nội dung
func_name / qualname	Tên hàm
args	Snapshot các tham số đầu vào (tên, kiểu, preview, full value)
result	Snapshot giá trị trả về
result_diff	So sánh kết quả với tham số đầu (tự động)
exception	Tên và message exception (nếu có)
duration_ms	Thời gian thực thi
depth / parent	Vị trí trong cây gọi hàm

Snapshot ghi preview ngắn + full value (hiển thị qua nút "xem đầy đủ" trong UI):

• list → type, số phần tử, preview 8 phần tử đầu, full 200 phần tử
• dict → type, số key, preview 10 key đầu, full 100 key
• str → preview 120 ký tự, full 2000 ký tự
• Các kiểu khác → repr() rút gọn

---

Cách thủ công (không dùng env vars)

Nếu không muốn dùng env vars, có thể cấu hình và flush thủ công:

from sdk_trace import trace, configure, flush_sync, get_tracer

# Cấu hình 1 lần
configure(
    base_url="http://localhost:8899",
    api_key="ht-...",
    project="my-project",
)

@trace(deep=True)
def pipeline(x):
    return process(x)

# Sau khi gọi hàm, flush thủ công nếu không muốn auto-flush
pipeline(data)
# (auto-flush đã chạy rồi vì configure() đã set global client)

# Hoặc dùng async
import asyncio
from sdk_trace import flush

async def main():
    pipeline(data)
    url = await flush()
    print(f"Trace: {url}")

---

Chạy test

cd sdk-trace
SDK_TRACE_URL=http://localhost:8899 \
SDK_TRACE_API_KEY=ht-... \
SDK_TRACE_PROJECT=test \
python test_sdk.py

Kết quả mong đợi: 16/16 passed 🎉