今天我們要學習如何在 Flask 專案中建立 Middleware 來實作 API 監控功能。Flask 作為輕量級且靈活的 Web 框架，提供了多種方式來實作 middleware 功能，我們將使用 Flask 的 request hooks 來建立一個完整的監控系統。
Flask 的 middleware 實作方式與其他框架略有不同，主要透過以下機制：
Flask 提供了多種請求生命週期的攔截點：
@app.before_request # 每個請求之前執行
@app.after_request # 每個請求之後執行（需要回傳 response）
@app.teardown_request # 請求結束時執行（包括異常情況）
@app.errorhandler(Exception)
def handle_exception(e):
# 處理所有未捕獲的異常
return jsonify({"error": "Internal Server Error"}), 500
我們建立一個 FlaskMonitoringMiddleware 類別來統一管理所有監控功能：
class FlaskMonitoringMiddleware:
def __init__(self, app: Flask):
# 註冊各種 hooks
app.before_request(self._before_request)
app.after_request(self._after_request)
app.teardown_request(self._teardown_request)
app.errorhandler(Exception)(self._handle_exception)
def _before_request(self):
"""請求開始前記錄開始時間和系統狀態"""
request.start_time = time.time()
request.start_metrics = self._get_system_metrics()
def _after_request(self, response):
"""請求結束後收集資料並推送到 Loki"""
# 計算請求處理時間
duration = time.time() - getattr(request, 'start_time', time.time())
# 收集監控資料
monitoring_data = {
"request": self._collect_request_data(),
"response": self._collect_response_data(response),
"performance": {
"duration_ms": round(duration * 1000, 2),
"start_metrics": getattr(request, 'start_metrics', {}),
"end_metrics": self._get_system_metrics()
}
}
# 推送到 Loki
self._push_to_loki(monitoring_data)
return response
建立一個異步的 Loki 客戶端來避免阻塞主要請求處理：
class FlaskLokiClient:
def __init__(self, url="http://localhost:3100"):
self.push_url = f"{url}/loki/api/v1/push"
self.queue = Queue(maxsize=1000)
self.worker_thread = threading.Thread(target=self._worker_loop, daemon=True)
self.worker_thread.start()
def _worker_loop(self):
"""背景執行緒處理日誌推送"""
while True:
try:
log_item = self.queue.get(timeout=1.0)
self._sync_push(log_item)
except Empty:
continue
def push_log_async(self, message, labels, level="info"):
"""非阻塞日誌推送"""
try:
self.queue.put_nowait({
"message": message,
"labels": labels,
"level": level,
"timestamp": time.time()
})
except:
pass # 避免監控系統影響主要應用
使用 psutil 來監控系統資源：
class PerformanceMonitor:
def __init__(self):
self.process = psutil.Process()
self.start_time = time.time()
def get_system_metrics(self):
"""獲取當前系統效能指標"""
try:
memory_info = self.process.memory_info()
return {
"cpu_percent": self.process.cpu_percent(),
"memory_rss_mb": round(memory_info.rss / 1024 / 1024, 2),
"memory_percent": self.process.memory_percent(),
"threads": self.process.num_threads(),
"uptime_seconds": round(time.time() - self.start_time, 2)
}
except:
return {}
建立一個簡單的 HTML 儀表板來即時查看系統狀態：
@app.route('/monitoring/dashboard')
def monitoring_dashboard():
return """
<!DOCTYPE html>
<html>
<head>
<title>Flask Middleware 監控儀表板</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.metric-card {
border: 1px solid #ddd;
padding: 15px;
margin: 10px 0;
border-radius: 5px;
}
</style>
</head>
<body>
<h1>Flask Middleware 監控儀表板</h1>
<div id="health-status" class="metric-card">
<h3>系統狀態</h3>
<div id="health-info">載入中...</div>
</div>
<script>
function updateDashboard() {
fetch('/monitoring/health')
.then(response => response.json())
.then(data => {
document.getElementById('health-info').innerHTML = `
<p><strong>狀態:</strong> ${data.status}</p>
<p><strong>CPU:</strong> ${data.system_metrics.cpu_percent}%</p>
<p><strong>記憶體:</strong> ${data.system_metrics.memory_percent}%</p>
<p><strong>執行緒:</strong> ${data.system_metrics.threads}</p>
`;
});
}
updateDashboard();
setInterval(updateDashboard, 5000);
</script>
</body>
</html>
"""
現在我們將使用 Locust 來測試我們建立的 Flask Middleware 監控系統，驗證它在各種負載情況下的表現。
我們設計了三種不同的測試場景：
# 使用 Poetry 啟動 Flask 應用程式（在另一個終端機）
poetry run python main.py
# 執行基本負載測試
poetry run locust -f demo.py --host=http://localhost:5001 --users=10 --spawn-rate=2 --run-time=60s --headless --only-summary
# 執行高負載測試
poetry run locust -f demo.py --host=http://localhost:5001 --users=20 --spawn-rate=4 --run-time=30s --headless --only-summary HighLoadTester
# 執行錯誤測試
poetry run locust -f demo.py --host=http://localhost:5001 --users=5 --spawn-rate=1 --run-time=30s --headless --only-summary InvalidRequestTester
訪問 http://localhost:5001/monitoring/dashboard 來查看即時系統監控資料，包括：
如果您有設定 Grafana，可以使用以下 LogQL 查詢來分析監控資料：
# 查看所有請求
{service="flask-middleware-demo"}
# 查看錯誤請求
{service="flask-middleware-demo", level="error"}
# 計算平均回應時間
avg_over_time({service="flask-middleware-demo"}
| json
| unwrap performance_duration_ms [5m]) by (endpoint)
# 查看最慢的端點
topk(5,
avg_over_time({service="flask-middleware-demo"}
| json
| unwrap performance_duration_ms [10m]) by (endpoint)
)
執行 60 秒測試，10 個使用者，每秒新增 2 個使用者：
Type Name # reqs # fails | Avg Min Max Med | req/s failures/s
--------|----------------------------------------------------------------------------|-------|-------------|-------|-------|-------|-------|--------|-----------
GET /api/products 101 0(0.00%) | 35 15 58 35 | 1.69 0.00
POST /api/products/1/purchase 42 4(9.52%) | 40 2 105 38 | 0.70 0.07
GET /api/products/featured 74 0(0.00%) | 313 115 505 320 | 1.24 0.00
GET /api/orders 12 0(0.00%) | 25 15 36 24 | 0.20 0.00
GET /simulate-error 5 0(0.00%) | 151 117 184 160 | 0.08 0.00
測試結果分析：
專門測試無效請求的處理：
Type Name # reqs # fails | Avg Min Max Med | req/s failures/s
--------|----------------------------------------------------------------------------|-------|-------------|-------|-------|-------|-------|--------|-----------
GET /api/products/999 7 0(0.00%) | 28 20 39 28 | 0.24 0.00
GET /api/products/abc 8 0(0.00%) | 6 4 9 6 | 0.27 0.00
POST /api/products/1/purchase 15 2(13.33%) | 4 2 8 5 | 0.51 0.07
GET /invalid/path 3 0(0.00%) | 3 2 4 4 | 0.10 0.00
錯誤處理驗證：
20 個並發使用者，30 秒高強度測試：
Type Name # reqs # fails | Avg Min Max Med | req/s failures/s
--------|----------------------------------------------------------------------------|-------|-------------|-------|-------|-------|-------|--------|-----------
GET /api/products 824 0(0.00%) | 33 13 58 32 | 27.46 0.00
POST /api/products/*/purchase 199 8(4.02%) | 72 28 106 71 | 6.63 0.27
GET /api/products/featured 389 0(0.00%) | 316 106 508 315 | 12.97 0.00
高負載測試結果：
Flask 使用 before_request、after_request 和 teardown_request hooks 來攔截請求生命週期的各個階段。
使用背景執行緒和佇列來處理日誌推送，避免影響主要請求處理效能。
整合 psutil 來監控 CPU、記憶體使用情況，提供完整的系統健康狀態。
透過 errorhandler 裝飾器來捕獲和記錄所有未處理的異常。
Flask Middleware 提供了一個輕量級但功能完整的監控解決方案：
優點：
適用場景：
透過今天的實作，我們學會了如何在 Flask 中建立一個完整的 API 監控系統，包括請求追蹤、效能監控、錯誤處理和即時儀表板。
iThome鐵人賽
看影片追技術