بايثون لأتمتة DevOps

معالجة الأخطاء والتسجيل في سكريبتات العمليات

18 دقيقة الدرس 5 من 28

معالجة الأخطاء والتسجيل في سكريبتات العمليات

سكريبت عمليات يتعطل بصمت عند الساعة الثالثة صباحاً — بينما يظل خط أنابيب النشر معلقاً ينتظر كود الخروج — أسوأ من غياب السكريبت كلياً. الأتمتة الجاهزة للإنتاج يجب أن تفشل بصوت عالٍ مع سياق كافٍ، وأن تتعافى حيثما أمكن، وأن تترك أثراً منظماً يستطيع مهندس الاستجابة للطوارئ قراءته دون الرجوع إلى الكود المصدري. تغطي هذه الدرس الركائز الثلاث التي تحقق ذلك: نموذج الاستثناءات في Python، وحدة logging، ومخرجات السجلات المنظمة (JSON).

هرم الاستثناءات الذي تحتاجه فعلاً

شجرة استثناءات Python كبيرة، لكن سكريبتات العمليات تتعامل مع مجموعة صغيرة ومتوقعة منها. فهم الهرم يخبرك بأي المعالجات تكتب وأيها تتركه يتصاعد.

OSError (وأسماؤها المستعارة IOError وFileNotFoundError وPermissionError وTimeoutError) — يغطي كل عمليات نظام الملفات والمقابس الشبكية. دائماً التقطه حول عمليات I/O للملفات واستدعاءات subprocess.
subprocess.CalledProcessError — يُرفع بواسطة subprocess.run(..., check=True) حين ينهي الاستدعاء الفرعي بكود خروج غير صفري. سماته .returncode و.stdout و.stderr هي نقطة التشخيص الأولى.
KeyError / ValueError — في الغالب خطأ في التكوين أو تحليل استجابة API. اكشفهما فوراً؛ التقاطهما وإخماتهما يخفي عيوباً حقيقية.
requests.exceptions.RequestException — الفئة الأساسية لكل أخطاء HTTP في requests (رفض الاتصال، انتهاء المهلة، حالة سيئة بعد .raise_for_status()). معالج واحد يغطي كل العائلة.
Exception — شبكة الأمان الشاملة. استخدمها فقط في المستوى الأعلى من السكريبت كملاذ أخير، ودائماً سجّل الـ traceback الكامل قبل الخروج بكود غير صفري.

فكرة أساسية: لا تستخدم except: مجردة أبداً. فهي تلتقط SystemExit وKeyboardInterrupt، مما يمنع الإيقاف النظيف. التقط دائماً Exception على الأقل، أو الأفضل، نوع الاستثناء المحدد الذي تتوقعه.

كتابة معالجات استثناءات متينة

النمط أدناه هو أساس كل سكريبت عمليات على نطاق واسع. ثلاثة أشياء تميزه عن معالجة الأخطاء الهاوية: يسجّل الـ traceback الكامل، يضع كود خروج ذا معنى، ولا يبتلع خطأً لا يستطيع التعافي منه بصمت.

#!/usr/bin/env python3
"""restart_service.py — إعادة تشغيل خدمة systemd بأمان مع محاولات إعادة."""

import subprocess
import logging
import sys
import time

log = logging.getLogger(__name__)

MAX_RETRIES = 3
RETRY_DELAY = 5  # ثوان


def restart_service(name: str) -> None:
    """أعد تشغيل خدمة systemd؛ ارفع CalledProcessError عند الفشل."""
    subprocess.run(
        ["systemctl", "restart", name],
        check=True,
        capture_output=True,
        text=True,
    )
    log.info("service_restarted", extra={"service": name})


def restart_with_retries(name: str) -> None:
    last_exc: Exception | None = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            restart_service(name)
            return
        except subprocess.CalledProcessError as exc:
            last_exc = exc
            log.warning(
                "restart_failed",
                extra={
                    "service": name,
                    "attempt": attempt,
                    "returncode": exc.returncode,
                    "stderr": exc.stderr.strip(),
                },
            )
            if attempt < MAX_RETRIES:
                time.sleep(RETRY_DELAY)
    raise RuntimeError(
        f"فشلت إعادة تشغيل {name!r} بعد {MAX_RETRIES} محاولات"
    ) from last_exc


if __name__ == "__main__":
    try:
        restart_with_retries(sys.argv[1])
    except (IndexError, ValueError) as exc:
        log.error("bad_arguments", extra={"error": str(exc)})
        sys.exit(2)      # exit 2 = خطأ في الاستخدام
    except Exception:
        log.critical("unhandled_exception", exc_info=True)
        sys.exit(1)

ممارسة احترافية: اتفاقيات كود الخروج مهمة. 0 = نجاح، 1 = خطأ في وقت التشغيل، 2 = وسائط خاطئة (نفس اتفاقية أدوات Unix). أنظمة CI/CD والسكريبتات المراقِبة تستطيع التفرع على هذه الأكواد دون تحليل stderr.

وحدة logging: تهيئة تتناسب مع النطاق الواسع

وحدة logging المدمجة في Python مجربة وقادرة على الإنتاج. معظم مهندسي العمليات لا يستثمرونها بالكامل — يستدعون logging.basicConfig(level=logging.INFO) ويتوقفون. هذا النهج يضيع السياق المنظم ويجعل تجميع السجلات في أدوات مثل Datadog أو Splunk أو CloudWatch مؤلماً.

النمط الصحيح هو تهيئة المُسجِّل الجذري مرة واحدة عند بدء التشغيل باستخدام dictConfig. هذا يفصل ماذا تسجّل (كود المكتبة) عن كيف تنسّقه (نقطة الدخول). وحدات المكتبة لا تهيّئ معالجات أبداً — تستدعي فقط logging.getLogger(__name__).

# logging_config.py — استدع setup_logging() مرة واحدة في __main__

import logging
import logging.config
import os


def setup_logging(level: str = "INFO") -> None:
    """هيّئ المُسجِّل الجذري. استدعِها مرة واحدة من نقطة دخول السكريبت."""
    log_level = getattr(logging, level.upper(), logging.INFO)

    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "human": {
                "format": "%(asctime)s  %(levelname)-8s  %(name)s  %(message)s",
                "datefmt": "%Y-%m-%dT%H:%M:%S",
            },
            "json": {
                "()": "logging_config.JsonFormatter",
            },
        },
        "handlers": {
            "stderr": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stderr",
                "formatter": "json" if os.getenv("LOG_FORMAT") == "json" else "human",
                "level": log_level,
            },
        },
        "root": {
            "handlers": ["stderr"],
            "level": log_level,
        },
    })

السجلات المنظمة: المعيار الإنتاجي

سجلات النص المقروء للإنسان ممتعة على حاسوب المطور. في الإنتاج فهي عبء. منصات تجميع السجلات تستوعب JSON، وتفهرس كل حقل، وتتيح تشغيل استعلامات كـ service:nginx status:500 | stats count by region. إذا كانت سجلاتك نصاً عادياً، فأنت تدفع تكلفة التحليل — والتحليل هش.

معامل extra في كل استدعاء log.* هو الآلية. مرر قاموساً من أزواج مفتاح-قيمة هناك؛ يقوم Formatter مخصص بتحويل LogRecord كاملاً (بما فيه تلك الحقول الإضافية) إلى JSON. النمط شفاف للمستدعي لكنه يحوّل كل تعليمة تسجيل إلى حدث منظم.

# logging_config.py (تتمة) — مُنسِّق JSON

import json
import logging
import traceback


class JsonFormatter(logging.Formatter):
    """أصدر كائن JSON واحداً لكل سجل، مناسباً لتجميع السجلات."""

    RESERVED = frozenset(logging.LogRecord(
        "", 0, "", 0, "", (), None
    ).__dict__.keys())

    def format(self, record: logging.LogRecord) -> str:
        payload: dict = {
            "ts":      self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level":   record.levelname,
            "logger":  record.name,
            "msg":     record.getMessage(),
        }

        # ادمج أي حقول extra= قدّمها المستدعي
        for key, value in record.__dict__.items():
            if key not in self.RESERVED and not key.startswith("_"):
                payload[key] = value

        # أضف معلومات الاستثناء حين تتوفر
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
            payload["traceback"] = traceback.format_exception(*record.exc_info)

        return json.dumps(payload, default=str)


# --- الاستخدام في أي وحدة ---
log = logging.getLogger(__name__)

log.info("deploy_started", extra={
    "service": "payments-api",
    "version": "v2.4.1",
    "region": "us-east-1",
    "triggered_by": "github_actions",
})
# المخرج (JSON):
# {"ts":"2025-03-15T14:22:01","level":"INFO","logger":"deploy",
#  "msg":"deploy_started","service":"payments-api","version":"v2.4.1",
#  "region":"us-east-1","triggered_by":"github_actions"}

يُصدر نفس المُسجِّل JSON لمنصة تجميع السجلات في الإنتاج ونصاً مقروءاً محلياً — يتحكم بذلك متغير بيئة واحد.

التقاط السياق مع LoggerAdapter

حين يدير سكريبت موارد متعددة (خمس نسخ EC2، عشر خدمات)، يجب أن يحمل كل سطر سجل معرّف المورد دون الحاجة إلى تكراره في كل استدعاء extra=. يحل logging.LoggerAdapter هذا بأناقة بحقن قاموس سياق ثابت في كل سجل يُصدر عبره.

أنشئ adapter واحداً لكل مورد: log = logging.LoggerAdapter(base_logger, {"instance_id": iid, "region": region})
استدع log.info("health_check_passed") — يدمج الـ adapter سياقه تلقائياً.
هذا يزيل الجهد الذهني لتتبع أي مورد ينتمي إليه سطر سجل عام عند قراءة آلاف الإدخالات في نظام تجميع السجلات.

ما يجب ألا تسجله

التسجيل المنظم قوي بما يكفي لتسريب الأسرار عن طريق الخطأ. طبّق هذه القواعد في مراجعة الكود:

لا تسجّل مفاتيح AWS أو مفاتيح API أو التوكنات — حتى جزئياً. استخدم غلافاً للتنقيح أو مرر المعرفات فقط (معرف المفتاح، لا السر).
لا تسجّل بيانات تعريف المستخدم الشخصية (البريد الإلكتروني، IP، الاسم) إلا إذا كانت سياسة تصنيف البيانات تسمح بذلك صراحةً مع ضوابط الاحتفاظ المناسبة.
لا تسجّل نصوص الطلب/الاستجابة الكاملة من واجهات APIs الخارجية — فهي غالباً تحتوي على أسرار مضمّنة من المستدعين.

فخ إنتاجي: log.debug("response body: %s", response.text) يبدو بريئاً في التطوير لكنه يمكن أن يتسرب بميغابايتات من JSON الحساس إلى منصة تجميع السجلات في الثانية تحت الحمل الكبير. راجع كل تعليمة DEBUG قبل الشحن. كثير من الفرق تضع مستوى السجل الإنتاجي على INFO وتحذف استدعاءات DEBUG في CI بقاعدة linter.

الجمع معاً: قالب سكريبت عمليات متكامل

النمط أدناه هو الهيكل الذي يجب أن يبدأ منه كل سكريبت عمليات جديد في أي فريق DevOps احترافي. يربط كل ما في هذا الدرس: تسجيل dictConfig، معالج استثناءات علوي، وأحداث سجلات منظمة مع سياق.

#!/usr/bin/env python3
"""template.py — هيكل سكريبت عمليات جاهز للإنتاج."""

import argparse
import logging
import sys

log = logging.getLogger(__name__)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--log-level", default="INFO",
                        choices=["DEBUG", "INFO", "WARNING", "ERROR"])
    return parser.parse_args()


def main(args: argparse.Namespace) -> int:
    """أعد كود الخروج: 0 نجاح، 1 خطأ، 2 مدخلات خاطئة."""
    log.info("script_start", extra={"region": args.region})

    try:
        pass  # --- منطقك هنا ---
    except ValueError as exc:
        log.error("invalid_input", extra={"error": str(exc)})
        return 2
    except OSError as exc:
        log.error("io_error", extra={"error": str(exc), "errno": exc.errno})
        return 1
    except Exception:
        log.critical("unhandled_exception", exc_info=True)
        return 1

    log.info("script_complete")
    return 0


if __name__ == "__main__":
    args = parse_args()
    logging.basicConfig(level=args.log_level)
    sys.exit(main(args))

ممارسة احترافية: في Google، تتبع جميع أدوات العمليات الداخلية نسخة من هذا القالب. الفكرة الجوهرية أن main() تُعيد عدداً صحيحاً ولا تستدعي sys.exit() مباشرة — ذلك متروك لكتلة if __name__ == "__main__". هذا يجعل الدالة قابلة للاختبار الوحدوي: يستطيع الاختبار استدعاء main(args) والتحقق من كود الإعادة دون التشعب بعملية.