測試與可觀測性

測試策略：金字塔、Trophy 與測試類型 基礎 高頻

測試的目的不是追求數字，而是讓團隊能快速且有信心地改動程式。每一種測試都在「信心」、「速度」、「穩定性」、「維護成本」之間取捨。

測試金字塔 vs Testing Trophy

測試類型

TDD 的取捨

Test-Driven Development：紅（先寫會失敗的測試）→ 綠（寫最少程式讓它通過）→ 重構。

• 好處：迫使你先想清楚介面與行為；自然產生可測的設計（依賴可注入）；留下完整的回歸測試；重構有安全網。
• 代價：探索期（不確定要做什麼、做 spike/原型）時寫測試是浪費；過度細的測試會綁死實作細節，讓重構反而更難。
• 務實做法：需求明確的純邏輯（計費、出價規則、解析器）與修 bug（先寫重現 bug 的測試）適合 TDD；探索性程式碼先寫、穩定後補測試。

pytest 實戰 進階 高頻

pytest 是 Python 事實上的標準測試框架：用純 assert（失敗時自動顯示比較細節）、fixture 依賴注入、強大的外掛生態。

Fixtures 與 scope

# conftest.py —— 同目錄與子目錄的測試自動可用，不需 import
import pytest

@pytest.fixture(scope="session")
def settings():
    return {"db_url": "sqlite:///:memory:", "line_api": "http://fake-line"}

@pytest.fixture
def repo():
    """function scope（預設）：每個測試拿到全新的物件，測試間互不影響。"""
    from app.repo import InMemoryCampaignRepo
    r = InMemoryCampaignRepo()
    yield r                 # yield 之前是 setup，之後是 teardown（即使測試失敗也會執行）
    r.clear()

@pytest.fixture
def campaign(repo):         # fixture 可以依賴其他 fixture
    return repo.create(name="bf", budget_micros=1_000_000)

function（預設）

每個測試一次。最隔離。

class / module / package

同一類別/模組/套件共用一次。

session

整個測試執行只建立一次，適合昂貴資源（啟動 DB container）。共用狀態要小心測試間互相污染——常見做法是 session 級建立 DB，function 級在 transaction 中執行並 rollback。

autouse=True

自動套用到範圍內所有測試，例如每個測試前重置全域設定。少用，因為隱性依賴降低可讀性。

parametrize：一份測試、多組資料

import pytest
from app.validation import is_valid_parentheses

@pytest.mark.parametrize(
    ("s", "expected"),
    [
        ("", True),
        ("()[]{}", True),
        ("([)]", False),
        ("(((", False),
        ("{[]}", True),
    ],
    ids=["empty", "flat", "interleaved", "unclosed", "nested"],
)
def test_valid_parentheses(s, expected):
    assert is_valid_parentheses(s) is expected

@pytest.mark.parametrize("status", [500, 502, 503, 504])
@pytest.mark.parametrize("attempt", [0, 3])          # 疊加 = 笛卡兒積，共 8 個測試
def test_retryable(status, attempt):
    ...

mock / patch：「patch where it's used」

這是 Python mock 最常見的錯誤，也是面試常考點。patch 替換的是某個命名空間裡的名稱綁定，而不是物件本身。

# app/notifier.py
def send_line(user_id: str, text: str) -> bool:
    ...  # 真的打 LINE API

# app/service.py
from app.notifier import send_line          # service 模組裡多了一個名稱 send_line 指向同一函式

def notify_winner(user_id: str) -> bool:
    return send_line(user_id, "Congrats!")

from unittest.mock import patch
from app import service

# 錯誤：替換的是 app.notifier.send_line，但 service 模組裡的名稱仍指向原函式 → 真的打 API
def test_wrong():
    with patch("app.notifier.send_line") as m:
        service.notify_winner("U1")
        m.assert_called_once()      # 失敗

# 正確：patch 「被使用的地方」
def test_right():
    with patch("app.service.send_line", autospec=True, return_value=True) as m:
        assert service.notify_winner("U1") is True
        m.assert_called_once_with("U1", "Congrats!")

• 若 service.py 寫的是 import app.notifier 並呼叫 app.notifier.send_line(...)，那就要 patch app.notifier.send_line——因為查找發生在 notifier 模組的命名空間。
• autospec=True：mock 會帶有原函式的簽章，參數錯誤時會報錯；否則 Mock 接受任何呼叫、任何屬性，typo 如 m.assert_called_onse() 在舊版甚至會靜默通過（Python 3.5+ 對 assert 開頭的拼錯屬性會報錯，但其他拼錯仍會）。
• 非同步函式用 AsyncMock（patch 在 Python 3.8+ 會自動對 async def 目標使用 AsyncMock）。
• 過度 mock 是警訊：如果一個測試要 patch 五個東西，代表被測程式耦合太多，應該重構成依賴注入。

monkeypatch：環境變數、屬性、字典

def test_reads_env(monkeypatch):
    monkeypatch.setenv("LINE_CHANNEL_TOKEN", "test-token")
    monkeypatch.delenv("HTTP_PROXY", raising=False)
    from app.config import load
    assert load().line_token == "test-token"

def test_disable_network(monkeypatch):
    import socket
    def guard(*a, **k):
        raise RuntimeError("network access in unit test!")
    monkeypatch.setattr(socket, "create_connection", guard)
    # 測試結束後 monkeypatch 會自動還原

時間：freezegun

from datetime import datetime, timedelta, timezone
from freezegun import freeze_time

def is_campaign_live(start: datetime, end: datetime) -> bool:
    return start <= datetime.now(timezone.utc) < end

@freeze_time("2026-11-27 03:59:59", tz_offset=0)
def test_not_live_one_second_before():
    start = datetime(2026, 11, 27, 4, 0, tzinfo=timezone.utc)
    assert not is_campaign_live(start, datetime(2026, 11, 28, tzinfo=timezone.utc))

def test_ttl_expiry():
    with freeze_time("2026-01-01 00:00:00") as frozen:
        cache = {"k": ("v", datetime.now().timestamp() + 60)}
        frozen.tick(timedelta(seconds=61))              # 時間前進 61 秒，不用真的 sleep
        assert datetime.now().timestamp() > cache["k"][1]

更好的設計是把「時鐘」當成依賴注入（clock: Callable[[], datetime]），測試時傳入固定值，就不需要 monkeypatch 全域時間。time.monotonic() 也要注意：freezegun 會凍結它，但部分 C 擴充或其他執行緒可能不受影響；替代品 time-machine 速度較快。

pytest-asyncio

# pyproject.toml:  [tool.pytest.ini_options]  asyncio_mode = "auto"
import asyncio
import pytest
import httpx
from app.main import app

@pytest.fixture
async def client():
    transport = httpx.ASGITransport(app=app)            # 不起真的 server，直接呼叫 ASGI app
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

async def test_create_campaign(client):
    r = await client.post("/v1/campaigns", json={"name": "bf", "budget_micros": 1},
                          headers={"Authorization": "Bearer t"})
    assert r.status_code == 201
    assert r.headers["location"].startswith("/v1/campaigns/")

async def test_timeout_is_enforced():
    async def slow():
        await asyncio.sleep(10)
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow(), timeout=0.01)

Coverage 與它的侷限

uv pip install pytest pytest-cov pytest-xdist
pytest -n auto --cov=app --cov-branch --cov-report=term-missing --cov-fail-under=80

• Line coverage 只看每一行有沒有被執行；branch coverage（--cov-branch）再看每個 if 的真/假兩個分支是否都走過，更有意義。
• 侷限：coverage 衡量「被執行」，不是「被驗證」。沒有 assert 的測試也能 100% 覆蓋。它也無法告訴你漏掉了哪些需求或輸入組合。
• Mutation testing（如 mutmut）故意改程式（把 < 換成 <=），看測試能否抓到，衡量測試的真實強度，但很慢，適合關鍵模組。
• 實務建議：對新程式碼（diff coverage）設門檻比對整體設門檻有效；不要為了數字去測 getter。

Test Doubles 與可測性設計 進階 高頻

Gerard Meszaros 在《xUnit Test Patterns》把所有「替身」統稱 test double，再細分成五種。面試常問 mock 與 stub 的差別，重點是它們驗證的東西不同。

Python 的 unittest.mock.Mock 其實可以扮演 stub、spy、mock 三種角色，取決於你怎麼用它（設 return_value 是 stub；事後讀 call_args_list 是 spy；用 assert_called_* 是 mock）。

依賴注入讓程式可測

難測的程式通常有這些特徵：在函式內部直接 new 出外部依賴、讀全域狀態、直接呼叫 datetime.now()/random、把 I/O 與邏輯混在一起。解法是把依賴從外部傳進來，並把純邏輯與 I/O 分離（「functional core, imperative shell」）。

# 難測：依賴寫死在裡面
import requests, datetime

def send_reminders():
    users = requests.get("https://crm/api/users").json()
    for u in users:
        if u["expires_at"] < datetime.datetime.now().isoformat():
            requests.post("https://api.line.me/v2/bot/message/push", json={...})

# 可測：依賴以參數注入，邏輯是純函式
from dataclasses import dataclass
from datetime import datetime
from typing import Callable, Iterable, Protocol

@dataclass
class User:
    id: str
    expires_at: datetime

class UserSource(Protocol):
    def list_users(self) -> Iterable[User]: ...

class Sender(Protocol):
    def push(self, user_id: str, text: str) -> None: ...

def users_to_remind(users: Iterable[User], now: datetime) -> list[User]:     # 純邏輯，最好測
    return [u for u in users if u.expires_at < now]

class ReminderService:
    def __init__(self, source: UserSource, sender: Sender, clock: Callable[[], datetime]):
        self.source, self.sender, self.clock = source, sender, clock

    def run(self) -> int:
        targets = users_to_remind(self.source.list_users(), self.clock())
        for u in targets:
            self.sender.push(u.id, "Your plan is expiring")
        return len(targets)

# ---- 測試：fake + spy，沒有任何 patch ----
class FakeSource:
    def __init__(self, users): self.users = users
    def list_users(self): return self.users

class SpySender:
    def __init__(self): self.sent = []
    def push(self, user_id, text): self.sent.append((user_id, text))

def test_only_expired_users_are_reminded():
    now = datetime(2026, 11, 3)
    src = FakeSource([User("U1", datetime(2026, 11, 1)), User("U2", datetime(2026, 12, 1))])
    sender = SpySender()
    assert ReminderService(src, sender, clock=lambda: now).run() == 1
    assert sender.sent == [("U1", "Your plan is expiring")]

在 FastAPI 中，Depends 就是依賴注入容器，測試時用 app.dependency_overrides 替換：

from fastapi.testclient import TestClient
from app.main import app, get_sender

def test_endpoint_uses_fake_sender():
    spy = SpySender()
    app.dependency_overrides[get_sender] = lambda: spy
    try:
        r = TestClient(app).post("/v1/reminders:run")
        assert r.status_code == 200
        assert len(spy.sent) == r.json()["sent"]
    finally:
        app.dependency_overrides.clear()          # 避免污染其他測試

整合測試、契約測試、Property-based 與 CI 穩定性 深入

Testcontainers：用真的 DB 做整合測試

用 SQLite 代替 MySQL 測試會漏掉大量問題：SQL 方言、型別行為、交易隔離等級、鎖、JSON 函式、字元集排序。Testcontainers 在測試中以 Docker 啟動真正的 MySQL/PostgreSQL/Redis/Kafka，測完自動銷毀。

# uv pip install "testcontainers[mysql]" sqlalchemy pymysql
import pytest
import sqlalchemy as sa
from testcontainers.mysql import MySqlContainer

@pytest.fixture(scope="session")
def engine():
    with MySqlContainer("mysql:8.0") as mysql:                # session 級：整輪測試只啟動一次
        eng = sa.create_engine(mysql.get_connection_url())
        with eng.begin() as conn:
            conn.execute(sa.text("""
                CREATE TABLE campaigns (
                  id BIGINT PRIMARY KEY AUTO_INCREMENT,
                  name VARCHAR(100) NOT NULL,
                  budget BIGINT NOT NULL,
                  version INT NOT NULL DEFAULT 1)"""))
        yield eng

@pytest.fixture
def conn(engine):
    """每個測試在 transaction 裡執行，結束 rollback：快速且互不干擾。"""
    with engine.connect() as c:
        tx = c.begin()
        yield c
        tx.rollback()

def test_optimistic_lock(conn):
    conn.execute(sa.text("INSERT INTO campaigns (id, name, budget) VALUES (1, 'bf', 300)"))
    upd = sa.text("UPDATE campaigns SET budget=:b, version=version+1 WHERE id=1 AND version=:v")
    assert conn.execute(upd, {"b": 500, "v": 1}).rowcount == 1
    assert conn.execute(upd, {"b": 700, "v": 1}).rowcount == 0     # 舊版本更新失敗

注意：transaction + rollback 的技巧無法測試「程式本身會 commit」或跨連線的行為（例如兩個連線同時更新的鎖競爭），這類測試要改用每個測試前 TRUNCATE 表。CI 中需要能跑 Docker（GitHub Actions 的 Linux runner 可以；某些 CI 需要 Docker-in-Docker 或掛 socket）。

Contract Testing（Pact）

問題：微服務 A 呼叫 B。B 的團隊把回應欄位 budget 改名為 budget_micros，B 自己的測試全過，部署後 A 爆掉。E2E 測試能抓到但又慢又脆弱。

• Consumer-driven：由消費方寫下「我實際用到哪些欄位」，提供方只需保證這些不被破壞——比整份 OpenAPI 更精準（提供方可以自由改沒人用的欄位）。
• Provider 驗證時常需要 provider states（「假設 campaign 1 存在」）來準備資料。
• 替代方案：以 OpenAPI 為契約，consumer 用 schema 驗證 mock、provider 在 CI 用 schemathesis 等工具對實作做 schema-based fuzzing；或用 oasdiff 偵測 breaking change。gRPC 則可用 buf breaking 檢查 proto 相容性。

Property-based testing（Hypothesis）

傳統測試是「舉例」：給定輸入 X 期望輸出 Y。Property-based 是「陳述性質」：對所有合法輸入，某個不變式都成立；框架自動產生數百組輸入，找到反例後還會縮減（shrink）成最小的反例。

# uv pip install hypothesis
from hypothesis import given, strategies as st

def encode_cursor(ts: int, id_: int) -> str: ...
def decode_cursor(c: str) -> tuple[int, int]: ...

@given(st.integers(min_value=0), st.integers(min_value=0))
def test_cursor_roundtrip(ts, id_):
    assert decode_cursor(encode_cursor(ts, id_)) == (ts, id_)          # 性質：互逆

records = st.lists(st.fixed_dictionaries({"id": st.integers(0, 50), "v": st.integers()}))

@given(records, records)
def test_merge_properties(left, right):
    from app.merge import merge_records
    out = merge_records(left, right, how="outer")
    ids = [r["id"] for r in out]
    assert len(ids) == len(set(ids))                                    # 性質：輸出 id 唯一
    assert set(ids) == {r["id"] for r in left} | {r["id"] for r in right}  # 性質：outer = 聯集

@given(st.lists(st.floats(min_value=0, max_value=1e6), min_size=1))
def test_p99_within_range(xs):
    from app.stats import percentile
    assert min(xs) <= percentile(xs, 99) <= max(xs)

常見可測性質：編解碼互逆（serialize/deserialize）、冪等（f(f(x)) == f(x)，例如正規化）、與簡單參考實作結果相同（oracle）、不變式（排序後長度不變、合併後總數守恆）、交換律/結合律（分散式彙總的 merge 函式必須滿足，才能任意順序合併部分結果）。

負載測試

負載測試（load）、壓力測試（stress）、浸泡測試（soak，長時間跑找記憶體洩漏）、尖峰測試（spike）的工具與方法論在第 8 章詳述。這裡強調與可觀測性的關聯：壓測時一定要同時看 metrics 與 traces，否則只知道「500 QPS 時 p99 變 2 秒」，不知道是 DB 連線池耗盡、GC、還是下游限流。另外要注意 coordinated omission：壓測工具若在等待慢回應時暫停送出新請求，會嚴重低估尾端延遲。

CI 中的 flaky test

Flaky test：程式沒變，同一個測試時過時不過。它的傷害是摧毀團隊對 CI 的信任——大家習慣按 rerun，真正的 bug 也被當成 flaky 忽略。

1. 偵測：CI 記錄每個測試的歷史結果，同一 commit 上「失敗後重跑通過」即標記為 flaky。
2. 隔離（quarantine）：把確認 flaky 的測試移到不阻擋合併的 job，並自動開 issue 指派負責人，設期限修復。
3. 有限重試：pytest-rerunfailures（--reruns 2）可以止血，但必須記錄與報告重試次數，不能變成永久遮羞布。
4. 修根本原因：大多數 flaky 是真實的競態問題，有時就是正式環境偶發 bug 的縮影。

# GitHub Actions 片段：快的測試先跑、整合測試分開、平行化
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv venv && uv pip install -r requirements-dev.txt
      - run: uv run ruff check . && uv run mypy app
      - run: uv run pytest -m "not integration" -n auto -p randomly --cov=app --cov-branch
      - run: uv run pytest -m integration --reruns 1 --junitxml=report.xml

可觀測性：三本柱、Profiles 與 Monitoring 的差異 基礎 高頻

Monitoring 是 observability 的子集：有了好的可觀測性資料，才能建立好的監控。

三本柱的價值在於相互關聯：metrics 告訴你「有問題」，透過 exemplar 或共同的 label 跳到 trace 找「在哪裡」，再用 trace_id 查 log 看「為什麼」。所以所有訊號都要帶一致的 resource attributes（service.name、service.version、deployment.environment、k8s.pod.name）與 trace_id。

Metrics：類型、Prometheus、PromQL 與 Cardinality 進階 高頻

四種 metric 類型

Counter

只增不減的累計值（process 重啟歸零）。例：請求總數、錯誤總數、送出位元組。永遠不要直接看 counter 的原始值，要用 rate()/increase() 看變化率；這兩個函式會自動處理重啟造成的歸零（counter reset）。

Gauge

可升可降的瞬時值。例：記憶體用量、佇列長度、進行中的請求數、連線池使用數、最後成功時間戳。

Histogram

把觀測值放進預先定義的 bucket，輸出 _bucket{le="0.1"}（累計：小於等於 0.1 的次數）、_sum、_count。分位數在查詢時於伺服器端計算（histogram_quantile），可以跨 instance 聚合。精確度取決於 bucket 設計（在 bucket 內做線性插值）。

Summary

在client 端直接計算分位數（例如 p50、p99）並輸出。值精確（滑動時間窗），但不能跨 instance 聚合（把 10 台的 p99 平均起來沒有數學意義），且分位數要事先決定。多數情況應該選 histogram。

Prometheus 的 pull 模型

• Pull 的好處：Prometheus 掌握抓取頻率，不會被目標灌爆；抓取失敗本身就是訊號（up == 0）；開發時可以直接用瀏覽器看 /metrics；不需要每個服務設定推送目的地。
• Pull 的限制：短命的 batch job 可能在被抓取前就結束 → 用 Pushgateway（但只適合 service-level 的 batch job，不要拿來把 pull 改成 push）；在 NAT/防火牆後面的目標不好抓 → 用 agent 模式或 remote write。
• 儲存：本地 TSDB 適合數週資料；長期與全域查詢用 Thanos、Cortex/Mimir、VictoriaMetrics。
• Exposition format 範例：

# HELP http_requests_total Total HTTP requests.
# TYPE http_requests_total counter
http_requests_total{method="POST",route="/v1/messages",status="202"} 18342
http_requests_total{method="POST",route="/v1/messages",status="429"} 57
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{route="/v1/messages",le="0.05"} 16020
http_request_duration_seconds_bucket{route="/v1/messages",le="0.1"} 18011
http_request_duration_seconds_bucket{route="/v1/messages",le="+Inf"} 18399
http_request_duration_seconds_sum{route="/v1/messages"} 702.4
http_request_duration_seconds_count{route="/v1/messages"} 18399

PromQL 常用查詢

# 每秒請求數（QPS），依 route 分組
sum by (route) (rate(http_requests_total[5m]))

# 錯誤率（5xx 比例）
sum(rate(http_requests_total{status=~"5.."}[5m]))
  /
sum(rate(http_requests_total[5m]))

# p99 延遲（跨所有 pod 聚合後再算分位數；le 必須保留在 by 裡）
histogram_quantile(0.99, sum by (le, route) (rate(http_request_duration_seconds_bucket[5m])))

# 平均延遲
sum(rate(http_request_duration_seconds_sum[5m])) / sum(rate(http_request_duration_seconds_count[5m]))

# 過去 1 小時的總請求量
sum(increase(http_requests_total[1h]))

# 與上週同一時間比較（廣告流量有很強的週期性）
sum(rate(http_requests_total[5m])) / sum(rate(http_requests_total[5m] offset 1w))

# 前 5 名最耗 CPU 的 pod
topk(5, sum by (pod) (rate(container_cpu_usage_seconds_total[5m])))

# 目標掛了
up{job="bidder"} == 0

# 預測 4 小時後磁碟是否會滿
predict_linear(node_filesystem_avail_bytes{mountpoint="/"}[6h], 4 * 3600) < 0

# 延遲 SLI：200ms 內完成的請求比例（le 必須剛好是一個 bucket 邊界）
sum(rate(http_request_duration_seconds_bucket{le="0.2"}[5m]))
  /
sum(rate(http_request_duration_seconds_count[5m]))

• rate()：區間內的每秒平均增加率，適合告警與圖表；irate()：只用最後兩個點，反應快但抖動大，只適合看瞬間尖峰的圖。
• rate 的區間至少要是 scrape interval 的 4 倍左右（例如 15s 抓取用 [1m] 以上），否則可能區間內不到兩個點而得到空值。
• 先 rate 再 sum：sum(rate(x[5m])) 正確；rate(sum(x)[5m]) 會因為某個 pod 重啟歸零而誤判。

Cardinality 爆炸

每一組不同的 label 值組合就是一條獨立的時間序列（time series）。序列數 = 各 label 基數的乘積。例如 route（50）× method（5）× status（20）× pod（100）= 50 萬條，再加一個 histogram 的 15 個 bucket 就是 750 萬條。

• 絕對不要當 label 的：user_id、request_id、trace_id、email、完整 URL（含 ID 的 path，例如 /campaigns/123）、時間戳、錯誤訊息全文、campaign_id（數量無上限時）。
• 後果：Prometheus 記憶體暴增 OOM、查詢變慢、商業 SaaS 帳單爆炸（多數依序列數計費）。
• 預防：用路由樣板（/campaigns/{id}）而非實際 path；status 可以歸類成 2xx/4xx/5xx；在 relabel 階段丟掉危險 label；設定 sample_limit；定期用 topk(10, count by (__name__) ({__name__=~".+"})) 找出序列最多的 metric。
• 高基數的分析需求（「哪個 campaign 的出價失敗最多？」）交給 log/trace 或事件分析系統（ClickHouse、BigQuery），而不是 metrics。

Python prometheus_client 範例

# uv pip install prometheus-client
import random, time
from prometheus_client import Counter, Gauge, Histogram, start_http_server

BID_REQUESTS = Counter(
    "bid_requests_total", "Bid requests received",
    ["exchange", "result"],                 # result: bid | nobid | timeout | error —— 有限集合
)
BID_LATENCY = Histogram(
    "bid_latency_seconds", "End-to-end bid handling latency",
    ["exchange"],
    # 依 SLO 設計 bucket：RTB 預算約 100ms，在 10–120ms 間要有足夠解析度
    buckets=(0.005, 0.01, 0.02, 0.03, 0.05, 0.07, 0.08, 0.09, 0.1, 0.12, 0.2, 0.5),
)
IN_FLIGHT = Gauge("bid_in_flight", "Requests currently being processed")
MODEL_VERSION = Gauge("model_info", "Loaded model version", ["version"])   # info 型 gauge，值恆為 1

def handle_bid(exchange: str) -> str:
    with IN_FLIGHT.track_inprogress(), BID_LATENCY.labels(exchange).time():
        time.sleep(random.uniform(0.005, 0.12))
        result = random.choices(["bid", "nobid", "timeout"], [0.6, 0.38, 0.02])[0]
    BID_REQUESTS.labels(exchange, result).inc()
    return result

if __name__ == "__main__":
    start_http_server(8000)                 # 在 :8000/metrics 暴露
    MODEL_VERSION.labels("ctr-2026-11-01").set(1)
    while True:
        handle_bid(random.choice(["exA", "exB"]))

RED、USE、Four Golden Signals 與 Dashboard 設計 進階 高頻

記法：RED 看服務「對外表現」，USE 看資源「為什麼」。告警通常建立在 RED/Golden Signals（症狀）上，排查時再往下看 USE（原因）。

Grafana Dashboard 設計原則

1. 由上而下說故事：最上面一列是使用者面向的 SLI（QPS、錯誤率、p99、SLO 剩餘 error budget），往下是各元件 RED，最下面是資源 USE。值班人員 5 秒內應該能判斷「現在有沒有事」。
2. 一個 dashboard 一個目的：服務總覽、單一服務深入、容量規劃分開，不要做一個 80 個 panel 的巨型 dashboard。
3. 使用 template variables（$env、$region、$exchange）讓同一份 dashboard 套用多個環境，避免複製貼上。
4. 標上 SLO 門檻線與部署事件 annotation：圖表上直接看到「部署後 p99 跳升」。
5. 單位、圖例、Y 軸從 0 開始（比率類），避免誤導；比較類圖表用相同 Y 軸範圍。
6. Dashboard as code：用 JSON/Jsonnet/Terraform 管理並放進 Git 審查，避免手改後無人知曉。
7. 每個告警連到對應的 dashboard 與 runbook。

Logging：結構化、等級、關聯 ID 與遮罩 進階 高頻

結構化 JSON log

純文字 log（"User 123 failed to send message to LINE: timeout"）對人好讀，對機器難查：要寫 regex 才能統計。結構化 log 每一筆都是 JSON 物件，欄位可以直接被索引、過濾、聚合：

{"ts":"2026-11-03T10:15:02.123Z","level":"error","service":"push-sender","version":"1.8.2",
 "event":"line_push_failed","message_id":"msg_01HF9Z","batch_no":42,"channel_id":"ch_01HF3",
 "upstream_status":503,"attempt":3,"duration_ms":10004,
 "request_id":"c0a8012f9d","trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7"}

• 用固定的 event 名稱（snake_case）而非每次不同的句子，方便計數與告警。
• 統一欄位命名（可參考 OpenTelemetry semantic conventions 或 Elastic Common Schema），不同服務一致才能跨服務查詢。
• 寫到 stdout/stderr，由平台（Kubernetes + Fluent Bit 等）收集——十二要素應用（Twelve-Factor）原則；應用不要自己管理 log 檔輪替。

Log level 策略

原則：ERROR 應該代表「有人需要看」，如果團隊每天有上萬筆 ERROR 卻沒人理，ERROR 就失去意義。高 QPS 服務的 access log 可以取樣（例如只記 1%，但錯誤與慢請求全記）。

Correlation / Request ID

每個進入系統的請求產生（或沿用上游的）唯一 ID，透過 header 往下游傳遞、寫進每一筆 log。在 Python 中用 contextvars 保存，async 任務會自動繼承，不會在並行的請求之間串號（這點 threading.local 在 asyncio 下做不到）。導入 OpenTelemetry 之後，可以直接用 trace_id 作為關聯 ID。

import contextvars, json, logging, re, sys, time

request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

SENSITIVE_KEYS = {"password", "token", "authorization", "access_token", "secret", "api_key"}
EMAIL_RE = re.compile(r"([A-Za-z0-9._%+-])[A-Za-z0-9._%+-]*@([A-Za-z0-9.-]+)")

def mask(value):
    if isinstance(value, dict):
        return {k: ("***" if k.lower() in SENSITIVE_KEYS else mask(v)) for k, v in value.items()}
    if isinstance(value, list):
        return [mask(v) for v in value]
    if isinstance(value, str):
        return EMAIL_RE.sub(r"\1***@\2", value)        # amy@example.com → a***@example.com
    return value

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
                  + f".{int(record.msecs):03d}Z",
            "level": record.levelname.lower(),
            "logger": record.name,
            "event": record.getMessage(),
            "request_id": request_id_var.get(),
        }
        payload.update(mask(getattr(record, "fields", {})))
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)

def setup_logging(level: str = "INFO") -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(level)

log = logging.getLogger("push")

if __name__ == "__main__":
    setup_logging()
    request_id_var.set("req-123")
    log.info("message_created", extra={"fields": {"message_id": "msg_1", "to": "amy@example.com",
                                                  "headers": {"Authorization": "Bearer abc"}}})
    # {"ts": "...", "level": "info", "logger": "push", "event": "message_created", "request_id": "req-123",
    #  "message_id": "msg_1", "to": "a***@example.com", "headers": {"Authorization": "***"}}

正式專案可用 structlog 或 python-json-logger，它們提供 processor 鏈、contextvars 綁定與效能優化。

敏感資料遮罩

• 不要記：密碼、token、API key、完整信用卡號、session cookie、OAuth code。
• 遮罩或雜湊：email、電話、IP（視法規）、裝置 ID、LINE user id 等個資（GDPR、台灣個資法）。需要關聯分析時可用帶 salt 的 hash 取代原值。
• 遮罩要在應用端輸出前做（如上面的 formatter），收集端（Fluent Bit / Logstash）再做第二道防線；不要依賴「之後在 Elasticsearch 刪除」。
• 小心例外訊息與 request body dump 會夾帶敏感資料；不要整包 log request body。

ELK / EFK / Loki 架構

# LogQL：push-sender 過去 5 分鐘 upstream 5xx 的錯誤數，依 channel 分組
sum by (channel_id) (
  count_over_time({service="push-sender", level="error"} | json | upstream_status >= 500 [5m])
)

中間加一層 Kafka 的理由：吸收尖峰、後端（Elasticsearch）維護或故障時不丟 log、同一份 log 可以被多個消費者使用（存檔到 S3、即時異常偵測）。

分散式追蹤：Span、Context Propagation、OpenTelemetry 與取樣 深入 高頻

核心概念

Trace

一個請求在整個分散式系統中的完整旅程，由一個 128-bit 的 trace_id 識別，是一棵 span 樹。

Span

一個有開始與結束時間的工作單位（一次 HTTP 處理、一次 DB 查詢、一次 RPC）。包含 span_id（64-bit）、parent span id、名稱、時間、attributes（如 http.route、db.system）、events（時間點事件，如例外）、status、links（關聯到其他 trace，例如批次處理時一個 span 對應多個來源訊息）。

Span kind

SERVER、CLIENT、PRODUCER、CONSUMER、INTERNAL，讓後端能畫出服務拓撲。

Context propagation

把 trace context 跨越 process 邊界傳下去（HTTP header、gRPC metadata、Kafka message header）。沒有傳遞，每個服務就各自產生不相關的 trace。

Baggage

隨 context 一起傳遞的鍵值對（例如 tenant_id），讓下游服務也能取得；注意會增加每個請求的 header 大小，且可能洩漏給第三方。

W3C Trace Context：traceparent

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
             │  │                                │                └ trace-flags（01 = sampled）
             │  │                                └ parent-id：呼叫方 span 的 id（16 hex = 64 bit）
             │  └ trace-id（32 hex = 128 bit，全零無效）
             └ version（目前為 00）
tracestate: vendorA=opaque-value,vendorB=xyz        # 廠商自訂資訊，選用

W3C Trace Context 是標準化的傳遞格式（取代各家私有的 X-B3-*、uber-trace-id 等），OpenTelemetry 預設使用它。sampled flag 讓下游遵循上游的取樣決定，確保一個 trace 要嘛完整、要嘛完全不記錄。

OpenTelemetry 架構

OpenTelemetry（OTel）是 CNCF 專案，由 OpenTracing 與 OpenCensus 合併而成，提供與廠商無關的 API、SDK 與傳輸協定（OTLP），涵蓋 traces、metrics、logs（profiles 訊號正在發展中）。

• API vs SDK 分離：函式庫只依賴 API（沒有設定 SDK 時是 no-op，零成本），應用程式決定要不要裝 SDK 與輸出到哪裡。
• Collector：通常以 agent（每節點 DaemonSet 或 sidecar）+ gateway（集中式叢集）兩層部署。好處：應用不用知道後端是 Jaeger 還是商業 SaaS；集中處理批次、重試、遮罩、取樣；切換廠商不用改程式。
• BatchSpanProcessor：在背景批次送出，避免每個 span 同步送出拖慢請求；正式環境不要用 SimpleSpanProcessor。

取樣策略：Head vs Tail

# otel-collector gateway：保留所有錯誤、所有超過 200ms 的 trace，其餘取樣 2%
receivers:
  otlp:
    protocols:
      grpc: { endpoint: 0.0.0.0:4317 }
processors:
  memory_limiter: { check_interval: 1s, limit_percentage: 80, spike_limit_percentage: 20 }
  tail_sampling:
    decision_wait: 10s
    num_traces: 200000
    policies:
      - name: errors
        type: status_code
        status_code: { status_codes: [ERROR] }
      - name: slow
        type: latency
        latency: { threshold_ms: 200 }
      - name: baseline
        type: probabilistic
        probabilistic: { sampling_percentage: 2 }
  attributes/scrub:
    actions:
      - key: http.request.header.authorization
        action: delete
  batch: {}
exporters:
  otlp/tempo:
    endpoint: tempo:4317
    tls: { insecure: true }
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, tail_sampling, attributes/scrub, batch]
      exporters: [otlp/tempo]

Jaeger vs Tempo

• Jaeger（CNCF 畢業專案，源自 Uber）：成熟的 UI 與服務相依圖；儲存後端常用 Elasticsearch/Cassandra/OpenSearch，可依 tag 搜尋，但索引成本高。新版 Jaeger 以 OTel Collector 為核心重構。
• Grafana Tempo：只依 trace_id 存取、內容存在物件儲存（S3/GCS），非常便宜；搜尋靠 TraceQL 掃描與 metrics/log 中的 exemplar、trace_id 跳轉。與 Grafana、Loki、Prometheus 整合緊密。

SLI / SLO / SLA、Error Budget 與告警設計 深入 高頻

SLI（Service Level Indicator）

量測服務水準的指標，最好定義成「好事件 / 總事件」的比例。例：「回應碼非 5xx 且在 100ms 內完成的出價請求比例」。

SLO（Service Level Objective）

SLI 在一段時間窗內的目標。例：「過去 30 天滾動視窗內，99.9% 的請求成功且在 100ms 內」。內部目標，用來做工程決策。

SLA（Service Level Agreement）

與客戶的合約，未達成有賠償（服務抵用金）。SLA 應該比 SLO 寬鬆，讓內部在違約前有緩衝。

Error budget

1 − SLO。99.9% 的 SLO 代表允許 0.1% 的失敗。這是可以「花」的額度：用來發布新功能、做實驗、維護。

Error budget policy

Error budget 把「穩定 vs 速度」的爭論變成數據決策：預算充足時可以加快發布、做風險較高的變更；預算耗盡時凍結功能發布，把工程資源轉向可靠性改善，直到預算恢復。這要事先與產品端達成共識並寫成政策。也要避免 SLO 定太高——100% 既不可能也不划算，使用者的網路、手機本身的可靠度遠低於 99.99%。

Burn rate 告警

Burn rate = 實際錯誤率 / SLO 允許錯誤率。Burn rate 1 表示剛好在時間窗結束時用完預算；burn rate 14.4 表示以 14.4 倍速燃燒，30 天的預算約 50 小時就會用完，而 1 小時就會用掉 30 天預算的 2%（14.4 × 1 / 720 = 2%）。

直接對「錯誤率 > 0.1%」告警的問題：門檻太敏感會因短暫尖峰半夜叫醒人（precision 低），時間窗太長則偵測太慢（detection time 長）。Google SRE Workbook 推薦多視窗、多燃燒率告警：

短視窗的作用是讓告警在問題已經解除後快速復原，而不是因為長視窗還記得過去的錯誤而持續響一小時。

# Prometheus recording + alerting rules（SLO 99.9%，錯誤預算 0.001）
groups:
  - name: push-api-slo
    rules:
      - record: job:slo_errors:ratio_rate5m
        expr: |
          sum(rate(http_requests_total{job="push-api",status=~"5.."}[5m]))
          / sum(rate(http_requests_total{job="push-api"}[5m]))
      - record: job:slo_errors:ratio_rate1h
        expr: |
          sum(rate(http_requests_total{job="push-api",status=~"5.."}[1h]))
          / sum(rate(http_requests_total{job="push-api"}[1h]))
      - record: job:slo_errors:ratio_rate30m
        expr: |
          sum(rate(http_requests_total{job="push-api",status=~"5.."}[30m]))
          / sum(rate(http_requests_total{job="push-api"}[30m]))
      - record: job:slo_errors:ratio_rate6h
        expr: |
          sum(rate(http_requests_total{job="push-api",status=~"5.."}[6h]))
          / sum(rate(http_requests_total{job="push-api"}[6h]))

      - alert: PushAPIErrorBudgetBurnFast
        expr: |
          (job:slo_errors:ratio_rate1h > (14.4 * 0.001) and job:slo_errors:ratio_rate5m > (14.4 * 0.001))
          or
          (job:slo_errors:ratio_rate6h > (6 * 0.001) and job:slo_errors:ratio_rate30m > (6 * 0.001))
        labels:
          severity: page
        annotations:
          summary: "push-api 正快速消耗 error budget"
          description: "1h 錯誤率 {{ $value | humanizePercentage }}，超過 SLO 允許速率"
          runbook_url: "https://wiki.example.com/runbooks/push-api-error-budget"
          dashboard: "https://grafana.example.com/d/push-api"

告警設計原則

1. 對症狀告警，不對原因告警（symptom-based）：使用者感受到的是「推播 API 錯誤率升高」「延遲變慢」，而不是「某台機器 CPU 85%」。CPU 高但使用者沒受影響就不該半夜叫人；原因類指標放 dashboard 供排查。例外：可預見且即將造成症狀的原因（磁碟 4 小時內會滿、憑證 7 天內到期）。
2. 每個 page 都必須可行動（actionable）且緊急：收到告警的人必須能做點什麼，而且必須現在做。不需要立即行動的改成 ticket 或 email。
3. 避免 alert fatigue：大量誤報讓人習慣忽略，真正的事故被淹沒。定期檢視：每個告警過去一個月觸發幾次、多少是誤報、有沒有人因此採取行動；沒有就刪除或調整。
4. 去重、分組、抑制：Alertmanager 的 group_by 把同一事故的多個告警合併成一則通知；inhibit_rules 在「整個叢集掛了」時抑制底下上百個服務的告警。
5. 使用 for: 持續時間避免瞬間尖峰誤報（但會延遲偵測，burn rate 告警是更好的解法）。
6. 每個告警附 runbook：說明這個告警代表什麼、影響範圍、如何確認（查哪個 dashboard、什麼查詢）、常見原因與處置步驟、何時升級給誰。runbook 能逐步自動化成腳本，這正是「開發輔助維運工具」的切入點。
7. 監控你的監控：Prometheus 本身掛掉時誰會知道？使用 dead man's switch（一個永遠在觸發的 Watchdog 告警，外部服務在「沒收到」時通知）。

On-call、事故處理、Postmortem 與 Health Check 進階

事故處理流程

1. 偵測：告警或使用者回報。記錄時間點（事後計算 MTTD，mean time to detect）。
2. 分級與宣告：依影響範圍定 severity（SEV1：核心功能全面中斷/營收受損）。開事故頻道，指定 Incident Commander（協調，不親自除錯）、Communications lead（對內外更新狀態）、Operations/Subject matter experts（實際排查）。
3. 止血優先於根因：先恢復服務——回滾最近的部署、關閉 feature flag、切流量、擴容、限流降級。不要在服務掛著時試圖找出完美的根本原因。統計顯示大比例事故與變更（部署、設定）相關，「最近改了什麼？」是第一個問題。
4. 持續溝通：固定頻率更新狀態（即使「仍在調查」），更新 status page。
5. 恢復與觀察：確認指標回到正常，觀察一段時間再宣告結束。記錄 MTTR（mean time to recover）。
6. Postmortem：事後數日內完成。

Blameless postmortem

「無究責」不代表沒有責任，而是假設每個人在當時的資訊下都做了合理的決定，把焦點放在系統與流程為什麼允許這個錯誤發生。如果怪罪個人，大家會隱瞞資訊，組織就學不到東西。

• 內容：摘要、影響（多少使用者、多久、多少營收/曝光損失）、時間軸（偵測、升級、緩解、恢復的每個時間點）、根本原因與觸發因素、哪些做得好、哪些做得不好、哪裡運氣好、改善行動項目（每項有負責人與期限）。
• 根因分析：用「5 Whys」往下追，但要注意真實事故往往是多個因素同時成立（瑞士起司模型），不是單一根因。
• 行動項目優先選擇能防止一整類問題的系統性改善（自動化檢查、guardrail、更好的告警），而不是「下次小心一點」。

On-call 健康度

• 每次輪值的 page 數量要可控（Google SRE 建議每 12 小時輪班平均不超過 2 個事件）。
• 交接時整理未解決問題；值班期間未被 page 的時間用來改善工具與告警品質。
• 追蹤 toil（重複、手動、可自動化的維運工作）比例，並持續自動化。

Health check：Liveness vs Readiness vs Startup

containers:
  - name: bidder
    image: registry.example.com/bidder:1.8.2
    ports: [{ containerPort: 8080 }]
    startupProbe:
      httpGet: { path: /healthz, port: 8080 }
      periodSeconds: 5
      failureThreshold: 30          # 最多給 150 秒載入模型
    livenessProbe:
      httpGet: { path: /healthz, port: 8080 }
      periodSeconds: 10
      timeoutSeconds: 2
      failureThreshold: 3
    readinessProbe:
      httpGet: { path: /readyz, port: 8080 }
      periodSeconds: 5
      failureThreshold: 2
    lifecycle:
      preStop:
        exec: { command: ["sleep", "10"] }   # 讓 endpoint 移除先傳播到 LB/kube-proxy，再開始關閉

優雅關閉（graceful shutdown）的順序：收到 SIGTERM → readiness 改為失敗 → 等待流量停止導入 → 完成進行中的請求 → 關閉連線池 → 結束。terminationGracePeriodSeconds（預設 30 秒）要大於這整段時間。

實戰：為 FastAPI 服務加上 Metrics + 結構化 Log + Tracing 深入

以下是一個可執行的完整範例，把本章的概念整合在一起：Prometheus metrics（RED + 上游呼叫 + exemplar）、帶 request_id/trace_id 的 JSON log、OpenTelemetry tracing（自動 instrument FastAPI 與 httpx，並建立自訂業務 span）、liveness/readiness 端點與優雅關閉。

# observability.py + app —— 為 FastAPI 服務加上 metrics、結構化 log、tracing
# uv pip install fastapi uvicorn httpx prometheus-client \
#   opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc \
#   opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-httpx
import asyncio, contextvars, json, logging, os, sys, time, uuid
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, HTTPException, Request, Response
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
from opentelemetry.trace import Status, StatusCode
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Gauge, Histogram, generate_latest

SERVICE = os.getenv("SERVICE_NAME", "push-api")
VERSION = os.getenv("SERVICE_VERSION", "1.0.0")

# ============================ 1. Tracing ============================
def setup_tracing() -> None:
    resource = Resource.create({
        "service.name": SERVICE,
        "service.version": VERSION,
        "deployment.environment": os.getenv("ENV", "dev"),
    })
    # 尊重上游的取樣決定；自己是根節點時取樣 10%（錯誤與慢請求交給 collector 的 tail sampling 保留）
    provider = TracerProvider(resource=resource,
                              sampler=ParentBased(TraceIdRatioBased(float(os.getenv("TRACE_RATIO", "0.1")))))
    if os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT"):
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
        exporter = OTLPSpanExporter()                 # 端點由環境變數決定，例如 http://otel-collector:4317
    else:
        exporter = ConsoleSpanExporter()              # 本機開發：印到 stdout
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

# ============================ 2. Structured logging ============================
request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
SENSITIVE = {"authorization", "token", "password", "access_token", "api_key"}

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        ctx = trace.get_current_span().get_span_context()
        doc = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)) + f".{int(record.msecs):03d}Z",
            "level": record.levelname.lower(),
            "service": SERVICE,
            "version": VERSION,
            "event": record.getMessage(),
            "request_id": request_id_var.get(),
        }
        if ctx.is_valid:                                  # 與 trace 關聯：log ↔ trace 互相跳轉
            doc["trace_id"] = format(ctx.trace_id, "032x")
            doc["span_id"] = format(ctx.span_id, "016x")
        for k, v in getattr(record, "fields", {}).items():
            doc[k] = "***" if k.lower() in SENSITIVE else v
        if record.exc_info:
            doc["exception"] = self.formatException(record.exc_info)
        return json.dumps(doc, ensure_ascii=False, default=str)

def setup_logging() -> None:
    h = logging.StreamHandler(sys.stdout)
    h.setFormatter(JsonFormatter())
    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), handlers=[h], force=True)
    logging.getLogger("uvicorn.access").disabled = True   # 改用下方自己的 access log（結構化、可取樣）
    logging.getLogger("httpx").setLevel(logging.WARNING)  # httpx 預設每個請求都記一筆 INFO

log = logging.getLogger(SERVICE)

def log_event(level: int, event: str, **fields) -> None:
    log.log(level, event, extra={"fields": fields})

# ============================ 3. Metrics ============================
REQUESTS = Counter("http_requests_total", "HTTP requests", ["method", "route", "status"])
LATENCY = Histogram("http_request_duration_seconds", "HTTP request latency", ["method", "route"],
                    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5))
IN_PROGRESS = Gauge("http_requests_in_progress", "In-flight HTTP requests", ["method"])
UPSTREAM = Histogram("upstream_request_duration_seconds", "Calls to upstream APIs",
                     ["upstream", "outcome"], buckets=(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10))
BUILD = Gauge("build_info", "Build information", ["version"])

# ============================ 4. App ============================
state = {"ready": False}

@asynccontextmanager
async def lifespan(app: FastAPI):
    BUILD.labels(VERSION).set(1)
    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=1.0))   # 共用連線池
    state["ready"] = True
    log_event(logging.INFO, "service_started")
    yield
    state["ready"] = False                         # 優雅關閉：先讓 readiness 失敗
    await app.state.http.aclose()
    log_event(logging.INFO, "service_stopped")

setup_logging()
setup_tracing()
app = FastAPI(lifespan=lifespan)
FastAPIInstrumentor.instrument_app(app, excluded_urls="healthz,readyz,metrics")  # 自動建立 SERVER span、解析 traceparent
HTTPXClientInstrumentor().instrument()             # 自動為 httpx 呼叫建立 CLIENT span 並注入 traceparent

SKIP = {"/metrics", "/healthz", "/readyz"}

@app.middleware("http")
async def observe(request: Request, call_next):
    rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex
    token = request_id_var.set(rid)
    method = request.method
    IN_PROGRESS.labels(method).inc()
    start = time.perf_counter()
    status = 500
    try:
        response = await call_next(request)
        status = response.status_code
        response.headers["X-Request-ID"] = rid
        return response
    finally:
        elapsed = time.perf_counter() - start
        IN_PROGRESS.labels(method).dec()
        # 用路由樣板（/v1/messages/{message_id}）而不是實際 path，避免 cardinality 爆炸
        route = getattr(request.scope.get("route"), "path", "unmatched")
        if request.url.path not in SKIP:
            REQUESTS.labels(method, route, str(status)).inc()
            ctx = trace.get_current_span().get_span_context()
            exemplar = {"trace_id": format(ctx.trace_id, "032x")} if ctx.is_valid and ctx.trace_flags.sampled else None
            LATENCY.labels(method, route).observe(elapsed, exemplar=exemplar)   # metrics → trace 跳轉
            if status >= 500 or elapsed > 1.0:     # 錯誤與慢請求全記，其餘不記每筆 access log
                log_event(logging.WARNING, "http_request", method=method, route=route,
                          status=status, duration_ms=round(elapsed * 1000, 1))
        request_id_var.reset(token)

@app.get("/metrics")
def metrics() -> Response:
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.get("/healthz")
def healthz():                                     # liveness：只代表 process 能回應
    return {"status": "ok"}

@app.get("/readyz")
def readyz():                                      # readiness：暖機完成、未在關閉中
    if not state["ready"]:
        raise HTTPException(503, "not ready")
    return {"status": "ready"}

LINE_URL = os.getenv("LINE_URL", "https://api.line.me/v2/bot/message/push")

@app.post("/v1/messages/{message_id}:send")
async def send(message_id: str, request: Request):
    with tracer.start_as_current_span("send_message") as span:           # 自訂業務 span
        span.set_attribute("app.message_id", message_id)                 # 高基數值放 span attribute，不放 metric label
        start = time.perf_counter()
        outcome = "ok"
        try:
            r = await request.app.state.http.post(LINE_URL, json={"to": "U" + "0" * 32,
                                                                   "messages": [{"type": "text", "text": "hi"}]})
            if r.status_code == 429:
                outcome = "throttled"
                raise HTTPException(503, "upstream throttled", headers={"Retry-After": "5"})
            r.raise_for_status()
        except httpx.HTTPError as e:
            outcome = "error"
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, "upstream failed"))
            log_event(logging.ERROR, "line_push_failed", message_id=message_id, error=type(e).__name__)
            raise HTTPException(502, "upstream error")
        finally:
            UPSTREAM.labels("line", outcome).observe(time.perf_counter() - start)
        log_event(logging.INFO, "line_push_sent", message_id=message_id)
        return {"id": message_id, "status": "sent"}

設計重點

• 路由樣板當 label：request.scope["route"].path 得到 /v1/messages/{message_id}:send，而非帶實際 ID 的 path；無法匹配的路由（404 掃描流量）歸到 unmatched，避免攻擊者用隨機 URL 製造無限序列。
• 高基數值放 span attribute 與 log（message_id），低基數維度放 metric label（method、route、status、outcome）。
• Exemplar：histogram 觀測值附上 trace_id，Grafana 可以從延遲圖上的點直接跳到那一筆 trace（需以 OpenMetrics 格式抓取並在 Prometheus 啟用 exemplar storage）。
• Log 與 trace 關聯：formatter 從目前的 span context 取 trace_id/span_id，Loki/Elasticsearch 中的 log 可以一鍵跳到 Tempo/Jaeger。
• Access log 取樣：只記錯誤與慢請求，正常請求靠 metrics 統計——高 QPS 服務的 log 成本因此下降幾個數量級。
• 取樣：SDK 端 ParentBased(TraceIdRatioBased(0.1))；若要保留所有錯誤 trace，需要把 ratio 設高並在 collector 做 tail sampling。
• 多 worker 部署：若用 gunicorn -w 4，prometheus_client 需設定 multiprocess mode；OTel 的 BatchSpanProcessor 背景執行緒要在 fork 之後初始化（例如在 gunicorn 的 post_fork hook），否則子 process 的 exporter 不會運作。
• 測試：用 httpx.MockTransport 替換 app.state.http 模擬 LINE 回 200/429/500，斷言 status code 與 /metrics 輸出中的計數，就是本章前半「依賴注入 + fake」的實際應用。

# docker-compose.yml：本機一鍵起完整的可觀測性堆疊
services:
  app:
    build: .
    environment:
      OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317
      SERVICE_VERSION: "1.0.0"
    ports: ["8000:8000"]
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.110.0
    volumes: ["./otel-collector.yaml:/etc/otelcol-contrib/config.yaml"]
  tempo:
    image: grafana/tempo:2.6.0
    command: ["-config.file=/etc/tempo.yaml"]
    volumes: ["./tempo.yaml:/etc/tempo.yaml"]
  prometheus:
    image: prom/prometheus:v2.54.1
    command: ["--config.file=/etc/prometheus/prometheus.yml", "--enable-feature=exemplar-storage"]
    volumes: ["./prometheus.yml:/etc/prometheus/prometheus.yml"]
  loki:
    image: grafana/loki:3.2.0
  grafana:
    image: grafana/grafana:11.2.0
    ports: ["3000:3000"]

# prometheus.yml
global:
  scrape_interval: 15s
scrape_configs:
  - job_name: push-api
    static_configs:
      - targets: ["app:8000"]
rule_files: ["slo-rules.yml"]

Appier 情境：Bidder、投遞異常與資料管線延遲 深入

1. Bidder 的 QPS、延遲與 timeout rate

以 RTB 廣告平台常見做法為例：交易所對每個出價請求有 tmax（例如約 100ms，含網路往返），超時的回應直接作廢。Bidder 的關鍵指標：

# 各交易所 timeout 比例
sum by (exchange) (rate(bid_requests_total{result="timeout"}[5m]))
  / sum by (exchange) (rate(bid_requests_total[5m]))

# 處理時間 p99 是否逼近 80ms（留 20ms 給網路）
histogram_quantile(0.99, sum by (le, exchange) (rate(bid_latency_seconds_bucket[5m]))) > 0.08

# 流量比上週同時段低 40% 以上（偵測「靜默」的流量流失）
sum by (exchange) (rate(bid_requests_total[10m]))
  < 0.6 * sum by (exchange) (rate(bid_requests_total[10m] offset 1w))

2. 廣告投遞率異常偵測

投遞（delivery / pacing）異常：某些活動突然完全沒曝光、或在一小時內花光一整天的預算。困難點在於正常值本身隨時間劇烈變化（日夜、週末、節慶），固定門檻不可行。

• 季節性基準：與上週同一時段比較（offset 1w），或用多週的中位數做基準，比與前一小時比較可靠。
• 統計方法：z-score（(x - avg_over_time(x[1d])) / stddev_over_time(x[1d])）、EWMA、Holt-Winters；更複雜的放在資料平台以 ML 模型做。
• Cardinality 分層：Prometheus 只監控總體與少數低基數維度（exchange、region、廣告格式）；每個 campaign 的投遞監控（數萬個）交給串流處理（Flink/Spark Streaming）或 OLAP（ClickHouse），每分鐘計算實際花費 vs 預期 pacing 曲線，偏離時發事件給告警系統。
• 業務告警的接收者：投遞異常可能不是工程問題（廣告主改了出價、素材被拒），告警要能路由給營運或客戶成功團隊，並附上最近的設定變更紀錄。

# 總曝光量的 z-score（以過去一天為基準），|z| > 3 視為異常
(
  sum(rate(impressions_total[10m]))
  - avg_over_time(sum(rate(impressions_total[10m]))[1d:10m])
)
/ stddev_over_time(sum(rate(impressions_total[10m]))[1d:10m])

3. 資料管線延遲（freshness）監控

報表、受眾分群、模型訓練都依賴事件管線。管線「沒有報錯但資料晚了三小時」是最常見也最難發現的事故類型。

# 批次 job 結束時回報（Pushgateway）
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import time

def report_success(job: str, rows: int) -> None:
    reg = CollectorRegistry()
    Gauge("job_last_success_timestamp_seconds", "Last success", registry=reg).set(time.time())
    Gauge("job_rows_processed", "Rows processed in last run", registry=reg).set(rows)
    push_to_gateway("pushgateway:9091", job=job, registry=reg)

面試自測題