يتقاطع في هذا الدرس الختامي كل مفهوم تعلمناه في هذا الدليل — تحليل المعاملات، واستدعاء العمليات الفرعية، ومعالجة JSON وYAML، وعملاء HTTP، ومعالجة الأخطاء والتسجيل، وحزم SDKs السحابية، والتزامن، والاختبار — لينصهر في منتج واحد قابل للشحن: أداة سطر أوامر تُدعى infracheck. تقوم هذه الأداة بمراجعة صحة البنية التحتية عبر استعلام APIs حية وتنتج تقريرًا منظمًا وسهل القراءة. هذا هو الشكل الحقيقي لأدوات تبنيها فرق SRE في كبرى شركات التقنية وتشغّلها من خطوط أنابيب CI يوميًا.
ما تفعله infracheck
infracheck أداة CLI صغيرة لكنها مكتملة: تقبل بيئة مستهدفة (dev / staging / prod)، تستعلم مجموعة قابلة للتهيئة من نقاط النهاية والموارد السحابية، تصنّف كل فحص كـ PASS / WARN / FAIL، وتطبع تقريرًا منسقًا — مع وضع إخراج JSON اختياري مناسب للتحويل إلى لوحات تحكم أو أنظمة تنبيه. يعكس التصميم الأدوات الداخلية الحقيقية: فحوصات قابلة للتركيب، وخلفيات قابلة للتوصيل، وإخراج يقرأه الآلة، ورمز خروج غير صفري عند الفشل حتى تنكسر خطوط CI بشكل صحيح.
معمارية infracheck: تُطلق واجهة CLI الفحوصات بالتوازي عبر مجموعة خيوط؛ يُنسّق Reporter النتائج المجمّعة.
هيكل المشروع
نظّم المشروع كحزمة Python حقيقية من اليوم الأول. حتى الأداة الصغيرة تستحق pyproject.toml ومجلد tests/ — هذا الاستثمار يؤتي ثماره فور أن يلمس المشروع مهندس ثانٍ.
كل فحص عبارة عن دالة قابلة للاستدعاء تُعيد dataclass من نوع CheckResult. هذه الواجهة الموحدة هي ما يجعل Runner قادرًا على جمع نتائج فحوصات HTTP وAWS وDNS وTLS دون أن يعرف تفاصيل تنفيذ أي منها — نمط strategy كلاسيكي.
# infracheck/runner.py
from __future__ import annotations
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, List
logger = logging.getLogger(__name__)
class Status(str, Enum):
PASS = "PASS"
WARN = "WARN"
FAIL = "FAIL"
@dataclass
class CheckResult:
name: str
status: Status
message: str
duration_ms: float = 0.0
metadata: dict = field(default_factory=dict)
CheckFn = Callable[[], CheckResult]
class CheckRunner:
def __init__(self, checks: List[CheckFn], max_workers: int = 10):
self._checks = checks
self._max_workers = max_workers
def run(self) -> List[CheckResult]:
results: List[CheckResult] = []
with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
futures = {pool.submit(fn): fn.__name__ for fn in self._checks}
for future in as_completed(futures):
name = futures[future]
try:
results.append(future.result(timeout=30))
except Exception as exc:
logger.error("Check %s raised: %s", name, exc)
results.append(CheckResult(
name=name,
status=Status.FAIL,
message=f"Unhandled exception: {exc}",
))
return sorted(results, key=lambda r: r.name)
كتابة فحص: نقطة نهاية HTTP
فحص HTTP هو الأبسط والأكثر شيوعًا: اضرب نقطة نهاية health، تحقق من رمز الحالة واختياريًا من مفتاح في الجسم. لاحظ النمط — قس الزمن الفعلي، والتقط الاستثناءات بمستوى التفصيل المناسب، وأعد نتيجة ذات نوع بدلًا من الإثارة.
# infracheck/checks/http.py
import time
import requests
from infracheck.runner import CheckResult, Status
def make_http_check(name: str, url: str, timeout: int = 10,
expected_status: int = 200,
body_contains: str | None = None) -> callable:
"""مصنع: يُعيد دالة فحص بدون معاملات."""
def check() -> CheckResult:
t0 = time.monotonic()
try:
resp = requests.get(url, timeout=timeout,
headers={"User-Agent": "infracheck/1.0"})
duration_ms = (time.monotonic() - t0) * 1000
if resp.status_code != expected_status:
return CheckResult(
name=name, status=Status.FAIL, duration_ms=duration_ms,
message=f"Expected {expected_status}, got {resp.status_code}",
)
if body_contains and body_contains not in resp.text:
return CheckResult(
name=name, status=Status.WARN, duration_ms=duration_ms,
message=f"Body missing '{body_contains}'",
)
latency_status = Status.WARN if duration_ms > 2000 else Status.PASS
return CheckResult(
name=name, status=latency_status, duration_ms=duration_ms,
message=f"{resp.status_code} in {duration_ms:.0f}ms",
)
except requests.Timeout:
return CheckResult(name=name, status=Status.FAIL,
duration_ms=(time.monotonic() - t0) * 1000,
message=f"Timed out after {timeout}s")
except requests.ConnectionError as exc:
return CheckResult(name=name, status=Status.FAIL,
duration_ms=(time.monotonic() - t0) * 1000,
message=f"Connection error: {exc}")
check.__name__ = name
return check
نقطة الدخول CLI
طبقة CLI، المبنية بـ Click (من الدرس السادس)، تقرأ إعدادات بيئة من YAML، تجمّع قائمة الفحوصات، تستدعي Runner، وتفوّض التنسيق إلى Reporter. رمز الخروج 1 عند أي FAIL أمر بالغ الأهمية — هو ما يُسبّب كسر خطوة في خط أنابيب CI وتنبيه المهندس المناوب.
# infracheck/cli.py
import sys
import click
import yaml
from infracheck.runner import CheckRunner, Status
from infracheck.checks.http import make_http_check
from infracheck.reporter import TextReporter, JsonReporter
@click.command()
@click.option("--env", required=True,
type=click.Choice(["dev", "staging", "prod"]),
help="البيئة المستهدفة للمراجعة.")
@click.option("--config", "config_path", default="infracheck.yaml",
show_default=True, help="مسار ملف إعدادات YAML.")
@click.option("--output", default="text",
type=click.Choice(["text", "json"]),
help="تنسيق الإخراج.")
@click.option("--fail-on-warn", is_flag=True, default=False,
help="الخروج بـ 1 عند WARN أيضًا.")
def main(env: str, config_path: str, output: str, fail_on_warn: bool) -> None:
"""infracheck — مراجعة صحة البنية التحتية وتقديم تقرير."""
with open(config_path) as fh:
cfg = yaml.safe_load(fh)
env_cfg = cfg.get("environments", {}).get(env, {})
checks = []
for item in env_cfg.get("http_checks", []):
checks.append(make_http_check(
name=item["name"],
url=item["url"],
timeout=item.get("timeout", 10),
expected_status=item.get("expected_status", 200),
body_contains=item.get("body_contains"),
))
runner = CheckRunner(checks, max_workers=cfg.get("max_workers", 10))
results = runner.run()
reporter = JsonReporter() if output == "json" else TextReporter()
reporter.print(results)
has_fail = any(r.status == Status.FAIL for r in results)
has_warn = any(r.status == Status.WARN for r in results)
if has_fail or (fail_on_warn and has_warn):
sys.exit(1)
if __name__ == "__main__":
main()
ملف الإعداد وpyproject.toml
ملف YAML يفصل منطق الأداة عن القيم الخاصة بكل بيئة. لا تُضمّن URLs أو مهل أو بيانات اعتماد في الكود المصدري أبدًا — الإعداد الخارجي يُمكّن نفس الثنائي من العمل مع dev وprod دون إعادة ترجمة.
بمجرد التثبيت (pip install -e .)، تعمل infracheck كأمر CLI من أول مرتبة من أي خطوة في GitHub Actions أو مرحلة Jenkins. رمز الخروج غير الصفري عند الفشل يُفشل الخطوة تلقائيًا، مما يوقف عملية النشر أو ينبّه فريق المناوبة عبر تكامل CI للتنبيه.
مبدأ تصميم أساسي: تُنتج infracheck مخرجَين في آنٍ واحد — نص مقروء للمهندس الذي يطالع سجل CI، وملف JSON يقرأه الآلة للأتمتة اللاحقة. فصل "منطق التشغيل" عن "منطق العرض" (فئة Reporter) هو ما يجعل ذلك ممكنًا دون تكرار الكود.
ممارسة إنتاجية: ثبّت اعتماديات الأداة في ملف requirements-lock.txt مُولَّد بـ pip-compile (pip-tools). الإعداد العائم requests>=2.31 في pyproject.toml مقبول للحزمة المنشورة، لكن CI يجب أن يثبّت من ملف القفل حتى لا يُعطل إصدار صاعد جديد مهمة المراجعة الليلية في الساعة الثانية صباحًا.
فشل شائع: الأدوات التي تخرج بـ 0 عند استثناءات غير معالجة. إذا أثار boto3.client() خطأ NoCredentialsError وابتلع معالج الاستثناءات العلوي هذا الخطأ وخرج بـ 0، سيُبلّغ خط أنابيب CI عن "البنية التحتية سليمة" بينما لم تعمل فحوصات AWS قط. سجّل دائمًا تتبع الاستثناء، أصدر نتيجة FAIL، وأوصل رمز خروج غير صفري. الـ CheckRunner أعلاه يفعل ذلك بصحة — الـ try/except الخارجي في run() يحوّل أي استثناء إلى نتيجة FAIL بدلًا من تجاهله صامتًا.
ما الذي بنيته
عبر هذا الدليل انتقلت من بيئة Python فارغة إلى أداة CLI جاهزة للإنتاج تجمع كل ما يحتاجه مهندس DevOps محترف: سلوك مدفوع بالإعداد، وتنفيذ متوازٍ، ونتائج ذات نوع محدد، وإخراج منسّق، ومعالجة صحيحة للأخطاء، وقابلية التثبيت، وتكامل CI. infracheck ليست لعبة — بإضافة وحدات فحص إضافية (اتصال قاعدة البيانات، جاهزية Pods في Kubernetes، انتهاء صلاحية الشهادات، نشر DNS)، تصبح من النوع الذي تشغّله فرق SRE في Google وCloudflare وStripe باستمرار ضد كل بيئة، على مدار الساعة.
نستخدم ملفات تعريف الارتباط لتشغيل هذا الموقع وتحليل الزيارات وعرض إعلانات مخصّصة. يمكنك قبول كل ملفات تعريف الارتباط أو رفض غير الأساسية منها.
سياسة الخصوصية