🎯 Mục tiêu: Viết Python script gọi REST API của nhiều service (health endpoint), tổng hợp kết quả, in report màu với trạng thái UP/DOWN và response time.
🧰 Công cụ / nền tảng: Python 3.11+, requests, concurrent.futures, terminal.
📦 Chuẩn bị: Python 3.11+; chạy python -m venv .venv && source .venv/bin/activate && pip install requests.
▶️ Các bước:
# 1. Tạo môi trường
mkdir devops-m05-lab && cd devops-m05-lab
python -m venv .venv
source .venv/bin/activate # Linux/macOS
# .venv\Scripts\Activate.ps1 # PowerShell Windows
pip install requests
echo "requests==2.32.3" > requirements.txt
# 2. Tạo file check-service-health.py
cat > check-service-health.py << 'PYEOF'
#!/usr/bin/env python3
"""
check-service-health.py — Kiểm tra health của nhiều REST API endpoint đồng thời.
Usage: python check-service-health.py [--timeout 5] [--output report.json]
"""
import argparse
import json
import logging
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import requests
# ---- Cấu hình logging ----
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
# ---- Màu ANSI cho terminal ----
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
RESET = "\033[0m"
BOLD = "\033[1m"
# ---- Danh sách service cần kiểm tra ----
SERVICES = [
{"name": "GitHub API", "url": "https://api.github.com", "expected_status": 200},
{"name": "JSONPlaceholder", "url": "https://jsonplaceholder.typicode.com/posts/1", "expected_status": 200},
{"name": "HTTPBin Health", "url": "https://httpbin.org/status/200", "expected_status": 200},
{"name": "Fake Down Service", "url": "https://httpbin.org/status/503", "expected_status": 200},
]
def check_service(service: dict, timeout: int) -> dict:
"""Gọi API và trả về kết quả kiểm tra."""
name = service["name"]
url = service["url"]
expected = service.get("expected_status", 200)
start = time.monotonic()
try:
resp = requests.get(url, timeout=timeout, allow_redirects=True)
elapsed_ms = int((time.monotonic() - start) * 1000)
is_up = resp.status_code == expected
return {
"name": name,
"url": url,
"status": "UP" if is_up else "DEGRADED",
"http_code": resp.status_code,
"response_ms": elapsed_ms,
"error": None,
}
except requests.exceptions.ConnectionError as e:
elapsed_ms = int((time.monotonic() - start) * 1000)
logger.warning("Connection error for %s: %s", name, e)
return {"name": name, "url": url, "status": "DOWN", "http_code": None, "response_ms": elapsed_ms, "error": str(e)}
except requests.exceptions.Timeout:
return {"name": name, "url": url, "status": "DOWN", "http_code": None, "response_ms": timeout * 1000, "error": "Timeout"}
def print_report(results: list[dict]) -> None:
"""In report dạng bảng màu ra stdout."""
print(f"\n{BOLD}{'='*60}{RESET}")
print(f"{BOLD} Service Health Report — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}{RESET}")
print(f"{BOLD}{'='*60}{RESET}")
print(f" {'Service':<25} {'Status':<10} {'HTTP':>5} {'ms':>6}")
print(f" {'-'*25} {'-'*10} {'-'*5} {'-'*6}")
for r in results:
color = GREEN if r["status"] == "UP" else (YELLOW if r["status"] == "DEGRADED" else RED)
http = str(r["http_code"]) if r["http_code"] else "N/A"
print(f" {r['name']:<25} {color}{r['status']:<10}{RESET} {http:>5} {r['response_ms']:>5}ms")
up = sum(1 for r in results if r["status"] == "UP")
total = len(results)
color = GREEN if up == total else (YELLOW if up > 0 else RED)
print(f"\n {color}Summary: {up}/{total} services UP{RESET}\n")
def main():
parser = argparse.ArgumentParser(description="Check REST API health endpoints")
parser.add_argument("--timeout", type=int, default=5, help="Request timeout in seconds")
parser.add_argument("--output", type=str, default=None, help="Save JSON report to file")
args = parser.parse_args()
logger.info("Checking %d services (timeout=%ds)...", len(SERVICES), args.timeout)
results = []
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(check_service, svc, args.timeout): svc for svc in SERVICES}
for future in as_completed(futures):
results.append(future.result())
results.sort(key=lambda x: x["name"])
print_report(results)
if args.output:
report = {"checked_at": datetime.now().isoformat(), "results": results}
with open(args.output, "w", encoding="utf-8") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
logger.info("Report saved to %s", args.output)
# Exit code 1 nếu có service DOWN (hữu ích cho CI pipeline)
any_down = any(r["status"] == "DOWN" for r in results)
sys.exit(1 if any_down else 0)
if __name__ == "__main__":
main()
PYEOF
# 3. Chạy script
python check-service-health.py --timeout 5
# 4. Lưu report ra JSON
python check-service-health.py --output health-report.json
# 5. Xem report JSON
python -m json.tool health-report.json | head -30
# 6. Xem exit code (0 = all up, 1 = có DOWN)
echo "Exit code: $?"
🖥️ VS Code:
Mở check-service-health.py → chọn Python interpreter (.venv) ở góc dưới bên phải → Run (F5) hoặc nhấn play button → xem output màu trong terminal.
✅ Kết quả mong đợi: Bảng report hiển thị 4 service: 3 màu xanh UP, 1 màu đỏ DOWN (httpbin 503); echo $? trả về 1; file health-report.json chứa đầy đủ kết quả với response time tính bằng ms.
🧹 Cleanup: deactivate && cd .. && rm -rf devops-m05-lab