الوحدة 8 — التسجيل ومجسّات الصحّة

خدمة تنبّؤ بلا مراقبة كسائق بلا لوحة قيادة: يقود حتّى ينفد الوقود بلا سابق إنذار. هذه الوحدة تُركّب لوحة قيادة الخدمة: سجلّات مُهيكلة تربط كلّ رسالة بمعرّف طلب فريد، مجسّات صحّة صادقة تُعلم Kubernetes متى نُعيد التشغيل ومتى نستقبل الحركة، ومقاييس تُظهر الكمون والأخطاء بمرور الوقت. بدون هذه الأدوات، تشخيص عطل في الإنتاج يستغرق ساعات بدل دقائق.

لماذا السجلّات المُهيكلة

logging.info("prediction pour customer 42") تبدو كافية، لكنّها كارثة عند التشخيص. الفرق مع سجلّ مُهيكل:

سجلّ نصّيّ عاديّ:

2026-09-06 14:32:11 INFO prediction pour customer 42 : 0.87

سجلّ مُهيكل JSON:

{"ts":"2026-09-06T14:32:11Z","level":"INFO","request_id":"a1b2","customer_id":42,"proba":0.87,"latency_ms":18,"model_version":"v1.3.0","client":"crm-nightly"}

الفارق العمليّ: الاستعلام. لتشخيص «لماذا تنبّؤات عميل CRM انخفضت الأسبوع الماضي؟» في سجلّ نصّيّ يعني grep معقّد على غيغابايتات نصّ. في سجلّات JSON، مسبار مركزيّ (Elasticsearch، Loki، Datadog) يُجيب في ثوانٍ: client:"crm-nightly" AND proba > 0.7 AND ts:[2026-08-30 TO 2026-09-05].

معرّف الطلب: الخيط الذي يربط كلّ شيء

كلّ طلب يحصل على معرّف فريد يعبر جميع السجلّات المتعلّقة به. هذا يسمح بتتبّع طلب واحد عبر عدّة خدمات (FastAPI، قاعدة بيانات، Celery). التطبيق ببرمجيّة وسطى:

import uuid
import logging
import time
from contextvars import ContextVar
from fastapi import Request

request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")


@app.middleware("http")
async def middleware_request_id(request: Request, call_next):
    rid = request.headers.get("x-request-id", str(uuid.uuid4())[:8])
    token = request_id_ctx.set(rid)
    debut = time.perf_counter()
    try:
        reponse = await call_next(request)
    finally:
        latence_ms = int((time.perf_counter() - debut) * 1000)
        logger.info("acces", extra={
            "request_id": rid,
            "path": request.url.path,
            "method": request.method,
            "latency_ms": latence_ms,
            "status": getattr(reponse, "status_code", 500),
        })
        request_id_ctx.reset(token)
    reponse.headers["x-request-id"] = rid
    return reponse

النقاط الحاسمة:

• x-request-id من الترويسة إن وُجد: يسمح للعميل بربط سجلّاته بسجلّات الخادم. لو لم يوجد، نُولّده.
• ContextVar بدل متغيّر عامّ: يعمل بشكل صحيح تحت async والخيوط المتوازية.
• الترويسة تُعاد للعميل: يستطيع فريق الدعم قراءتها من متصفّح المستخدم عند خطأ.

مُنسّق JSON

الشيفرة أعلاه تعتمد على مُنسّق يكتب JSON. تنفيذ بسيط:

import json
import logging


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        base = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%SZ"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "request_id": request_id_ctx.get(),
        }
        # حقول إضافيّة من extra=
        for cle, valeur in record.__dict__.items():
            if cle not in ("msg", "args", "levelname", "name", "created") \
                    and not cle.startswith("_"):
                base.setdefault(cle, valeur)
        return json.dumps(base, ensure_ascii=False)


handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(handlers=[handler], level=logging.INFO)

الآن كلّ logger.info(...) يُنتج سطرًا JSON صالحًا، يستقبله جامع السجلّات ويُفهرسه بلا عمل إضافيّ.

Liveness مقابل Readiness: الخلط الشائع

Kubernetes يُميّز مسارَي فحص لا يخلطهما:

• liveness: هل العمليّة حيّة؟ إن فشل، Kubernetes يقتل الحاوية ويُعيد تشغيلها. يجب أن يفشل فقط عندما تكون الخدمة معطّلة تعطّلًا لا رجعة فيه (تجميد، تسريب ذاكرة كارثيّ).
• readiness: هل الخدمة جاهزة لاستقبال الحركة؟ إن فشل، Kubernetes يوقف توجيه الطلبات إليها دون إعادة تشغيلها. يفشل مؤقّتًا عند التحميل، أثناء تحديث نموذج، أو عند انقطاع تبعيّة (Redis).

الفخّ الشائع: خلط الاثنَين. مسار /health واحد يفحص كلّ شيء (النموذج، قاعدة البيانات، MLflow، Redis) ويعطى لـliveness. النتيجة: عند فشل عابر لقاعدة البيانات، Kubernetes يعيد تشغيل الخدمة لا لزوم لذلك؛ حلقة إعادة تشغيل تصنع تعطّل خدمة كامل.

القاعدة: liveness أدنى ما يمكن (العمليّة تستجيب)، readiness يفحص التبعيّات الحقيقيّة.

@app.get("/livez")
def livez() -> dict:
    return {"status": "alive"}  # يكفي أن يُنفَّذ


@app.get("/readyz")
def readyz() -> dict:
    if app.state.model is None:
        raise HTTPException(503, "modele non charge")
    # تحقّق سريع من كلّ تبعيّة حرجة
    if not verifier_redis():
        raise HTTPException(503, "redis inaccessible")
    return {"status": "ready", "model_version": app.state.model_version}

في بيان Kubernetes:

livenessProbe:
  httpGet: { path: /livez, port: 8000 }
  periodSeconds: 30
  failureThreshold: 3
readinessProbe:
  httpGet: { path: /readyz, port: 8000 }
  periodSeconds: 10
  failureThreshold: 2

المسار الصحّيّ الكاذب

سجّل مسار /health يُعيد 200 OK دومًا كارثة كلاسيكيّة: الخدمة تبدو صحّية بينما تعيد أخطاء 500 لكلّ طلب حقيقيّ. النموذج ينهار بصمت وKubernetes لا يعرف. الاختبار الجيّد للـreadiness يجب أن يشمل تنبّؤًا وهميًّا فعليًّا:

@app.get("/readyz")
def readyz() -> dict:
    if app.state.model is None:
        raise HTTPException(503, "modele non charge")
    # تنبّؤ اختباريّ بمُدخل معياريّ
    try:
        _ = app.state.model.predict_proba(app.state.exemple_test)
    except Exception as e:
        raise HTTPException(503, f"modele en erreur : {e}")
    return {"status": "ready"}

الآن، فساد النموذج في الذاكرة (نادر لكن يحدث بعد ساعات طويلة من التشغيل) يُكتشف تلقائيًّا.

المقاييس: latency وerror rate بمرور الوقت

المكتبة prometheus-fastapi-instrumentator تُضيف نقطة نهاية /metrics بأربعة أسطر:

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

هذا يعرض تلقائيًّا: عدد الطلبات (http_requests_total)، الكمون بالمئينات (http_request_duration_seconds)، الأخطاء حسب رمز الحالة. Prometheus يجمع هذه المقاييس كلّ 15 ثانية، وGrafana يرسم لوحات تظهر المئين p95 وp99 للكمون، ومعدّل الأخطاء بمرور الوقت. تنبيه (alert) يُطلَق تلقائيًّا إذا تجاوز p95 ثلاث مئات ميلّي ثانية، أو إذا تجاوز معدّل 5xx واحدًا بالمئة.

مقاييس مخصّصة لكلّ نموذج تُضاف يدويًّا:

from prometheus_client import Counter, Histogram

TENTATIVES_TOTAL = Counter("churn_predictions_total", "توقّعات churn", ["result"])
PROBA_DISTRIBUTION = Histogram("churn_proba", "توزيع احتمالات churn",
                                buckets=[0.1, 0.3, 0.5, 0.7, 0.9])


@app.post("/predict")
def predire(req: ChurnRequest) -> ChurnResponse:
    ...
    TENTATIVES_TOTAL.labels(result="churn" if proba >= 0.5 else "reste").inc()
    PROBA_DISTRIBUTION.observe(proba)
    return ...

مؤشّر الأداء الحاسم في الإنتاج: مراقبة توزيع الاحتمالات بمرور الوقت. لو أنّ متوسّط proba انزلق فجأة من 0.3 إلى 0.6 دون تغيير حقيقيّ في بيانات المشتركين، ذلك مؤشّر على انحراف بيانات (data drift) يتطلّب إعادة تدريب النموذج (الوحدة 8 من دورة MLOps).

ما يجب ألّا يظهر في السجلّات

قاعدة أمنيّة لا تسامح فيها: لا تُسجّل أبدًا:

• محتوى الطلب الكامل (قد يحوي معلومات شخصيّة).
• مفاتيح API أو JWT ولو مُقتطعة.
• عناوين البريد وأرقام الهواتف والعناوين البريديّة.
• تتبّع الاستثناء بأسماء ملفّات ومسارات (يُسجَّل داخليًّا في نظام التشخيص، لا في التدفّق العامّ).

القاعدة العمليّة: قبل الإنتاج، افتح عيّنة من السجلّات وتحقّق يدويًّا. لو كان مطوّر غريب يستطيع استخراج معلومة عن مستخدم واحد بقراءة السجلّات، القانون الأوروبيّ (GDPR) والقانون التونسيّ (loi 63-2004) يُطبَّقان بغرامات فعليّة.

الخلاصة

• السجلّات المُهيكلة JSON بمعرّف طلب فريد تُحوّل ساعات تشخيص إلى ثوانٍ بفضل الاستعلام على حقل بدل grep نصّيّ.
• liveness أدنى ما يمكن (العمليّة حيّة)، readiness يفحص التبعيّات الحقيقيّة؛ خلطهما يُنتج حلقات إعادة تشغيل لا لزوم لها.
• المسار الصحّيّ الذي يُعيد 200 OK دومًا كاذب؛ اختبار جيّد يشمل تنبّؤًا فعليًّا على مُدخل معياريّ.
• Prometheus + Grafana يُعطيان الكمون بالمئينات، معدّل الأخطاء، وتوزيع احتمالات النموذج الذي يكشف انحراف البيانات مبكّرًا.
• لا معلومات شخصيّة ولا مفاتيح في السجلّات، أبدًا؛ التزام قانونيّ بأثر ماليّ ملموس.