feat(etl): ordonnance la détection d'alertes et les recommandations par un DAG Airflow

Le DAG `alertes` enchaîne `app.detection.internal_alerts` puis `app.cli generate-recommendations`, à la quinzième minute de chaque heure. Le décalage laisse finir `ml_score`, qui écrit à l'heure pile les prédictions dont la règle `anomaly` a besoin, sans créer de dépendance entre les deux DAGs : quatre règles de détection sur cinq ne touchent pas au modèle, et un modèle jamais entraîné ne doit pas priver le parc de ses alertes. L'image Airflow porte un second environnement uv, `/opt/backend/.venv`, puisque la logique vit dans le backend (ADR 0006) et qu'aucune route HTTP ne l'expose. Le `UV_PROJECT_ENVIRONMENT` global hérité de l'issue #115 disparaît : il vaut pour tous les projets, donc `uv run` depuis `/opt/ml` résolvait le venv du backend. uv prend `<projet>/.venv` par défaut, se placer dans le dossier suffit. La CI vérifie maintenant que les deux environnements s'importent sans réseau. Le conteneur reçoit `DATABASE_URL` en asyncpg et une `APP_SECRET_KEY` distincte de celle de l'API, alimentée par `AIRFLOW_APP_SECRET_KEY` : la détection ne signe aucun jeton, et Airflow permet d'exécuter du code depuis son interface.
2026-09-21 12:13:55 +02:00
parent 3cd9a6b272
commit ae58a896d9
7 changed files with 175 additions and 17 deletions
@@ -1,7 +1,8 @@
-# Image Airflow EnerVision : ajoute le projet ml/ dans son propre environnement Python 3.14,
-# distinct du Python 3.12 qui fait tourner Airflow lui-meme (apache-airflow 2.10 ne supporte pas
-# 3.14), pour que les DAGs puissent lancer `uv run python -m enervision_ml.train`/`.score` en
-# sous-processus. Airflow ne devient jamais un consommateur direct de LightGBM/MLflow.
+# Image Airflow EnerVision : ajoute ml/ et apps/backend/ dans leurs propres environnements Python
+# 3.14, distincts du Python 3.12 qui fait tourner Airflow lui-meme (apache-airflow 2.10 ne supporte
+# pas 3.14), pour que les DAGs puissent lancer `uv run python -m enervision_ml.train`/`.score`,
+# `app.detection.internal_alerts` et `app.cli` en sous-processus. Airflow ne devient jamais un
+# consommateur direct de LightGBM, de MLflow ou du SQLAlchemy du backend. Cf. ADR 0008.
 FROM apache/airflow:2.10.4-python3.12

 # LightGBM est compile contre libgomp (OpenMP), absent de l'image de base (minimale, sans
@@ -16,7 +17,8 @@ RUN apt-get update \
 # `ml_train` et `ml_score` (le modele ecrit par l'un, lu par l'autre). Un volume nomme herite des
 # permissions du repertoire qu'il recouvre a son premier montage ; sans ce chown prealable, il
 # serait cree root:root et illisible par le conteneur, qui tourne en `airflow` (uid 50000).
-RUN mkdir -p /opt/ml/state && chown -R airflow:root /opt/ml
+# `/opt/backend` ne porte aucun volume, mais `WORKDIR` le creerait root meme sous `USER airflow`.
+RUN mkdir -p /opt/ml/state /opt/backend && chown -R airflow:root /opt/ml /opt/backend
 USER airflow

 # L'image de base embarque deja un `uv`, mais trop ancien (0.4.29) pour le format de verrou de
@@ -24,9 +26,10 @@ USER airflow
 # (apps/backend/Dockerfile).
 COPY --from=ghcr.io/astral-sh/uv:0.11.26 /uv /home/airflow/.local/bin/uv

+# Piege : pas de `UV_PROJECT_ENVIRONMENT` global. Il vaudrait pour les deux projets, et `uv run`
+# dans l'un resoudrait le venv de l'autre. Par defaut, uv prend `<projet>/.venv`, donc le bon.
 ENV UV_COMPILE_BYTECODE=1 \
-    UV_LINK_MODE=copy \
-    UV_PROJECT_ENVIRONMENT=/opt/ml/.venv
+    UV_LINK_MODE=copy

 WORKDIR /opt/ml

@@ -36,4 +39,13 @@ RUN uv sync --locked --no-install-project --no-dev
 COPY --chown=airflow:root ml/enervision_ml ./enervision_ml
 RUN uv sync --locked --no-dev

+WORKDIR /opt/backend
+
+# `packages = ["app"]` : le reste de apps/backend (alembic, tests) n'a rien a faire dans l'image.
+COPY --chown=airflow:root apps/backend/pyproject.toml apps/backend/uv.lock ./
+RUN uv sync --locked --no-install-project --no-dev
+
+COPY --chown=airflow:root apps/backend/app ./app
+RUN uv sync --locked --no-dev
+
 WORKDIR /opt/airflow
@@ -0,0 +1,65 @@
+"""DAG de détection des alertes et de génération des recommandations (issue #116).
+
+Ordonnance ce que `docs/architecture/20-backend.md` et l'ADR 0006 décrivent encore comme lancé à
+la main. Toute la logique reste dans `apps/backend`, ce DAG ne fait que l'appeler, sur le patron
+de `ml_score` (cf. `docs/architecture/10-infra.md`, section Airflow, et l'ADR 0008 pour
+l'environnement `/opt/backend` que l'image embarque désormais).
+
+Planifié à la quinzième minute plutôt qu'à l'heure pile : la règle `anomaly` compare une lecture
+à la `prediction` du même instant, que `ml_score` (`@hourly`) vient d'écrire. Aucune dépendance
+déclarée entre les deux DAGs pour autant, quatre règles sur cinq ne touchent pas au modèle et un
+modèle jamais entraîné ne doit pas priver le parc de ses alertes.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+from airflow.models.dag import DAG
+from airflow.operators.bash import BashOperator
+
+# Le backend a son propre environnement uv dans l'image (ADR 0008). `--no-sync` et
+# `env -u VIRTUAL_ENV` : cf. `ml_train.py`, même raisonnement.
+COMMANDE_BACKEND = "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m"
+
+# Les deux tâches sont idempotentes en base (`ON CONFLICT DO NOTHING` sur
+# `uq_alert_source_reference` et `uq_recommendation_alert_rule`) : reprendre ne duplique rien.
+TENTATIVES = 2
+DELAI_ENTRE_TENTATIVES = timedelta(minutes=2)
+# La somme des deux plafonds reste sous le pas horaire : une exécution pendue ne doit pas
+# empiéter sur la suivante.
+PLAFOND_PAR_TACHE = timedelta(minutes=15)
+
+with DAG(
+    dag_id="alertes",
+    description=(
+        "Détecte les alertes internes puis génère les recommandations "
+        "(app.detection.internal_alerts, app.cli)."
+    ),
+    schedule="15 * * * *",
+    start_date=datetime(2026, 1, 1),
+    catchup=False,
+    # Deux exécutions simultanées analyseraient la même fenêtre de 48h, et la génération relit
+    # l'intégralité de la table `alert` à chaque passage.
+    max_active_runs=1,
+    tags=["alertes"],
+) as dag:
+    detection = BashOperator(
+        task_id="detection",
+        bash_command=f"{COMMANDE_BACKEND} app.detection.internal_alerts",
+        retries=TENTATIVES,
+        retry_delay=DELAI_ENTRE_TENTATIVES,
+        execution_timeout=PLAFOND_PAR_TACHE,
+    )
+
+    recommandations = BashOperator(
+        task_id="recommandations",
+        bash_command=f"{COMMANDE_BACKEND} app.cli generate-recommendations",
+        retries=TENTATIVES,
+        retry_delay=DELAI_ENTRE_TENTATIVES,
+        execution_timeout=PLAFOND_PAR_TACHE,
+    )
+
+    # `recommendation.alert_id` est une clé étrangère `NOT NULL` : la génération n'a rien à lire
+    # tant que la détection n'a pas écrit.
+    detection >> recommandations
@@ -9,6 +9,14 @@ from airflow.models.dagbag import DagBag

 DAGS_FOLDER = Path(__file__).resolve().parent.parent / "dags"

+DAG_IDS = ["ml_train", "ml_score", "alertes"]
+TACHES = [
+    ("ml_train", "train"),
+    ("ml_score", "score"),
+    ("alertes", "detection"),
+    ("alertes", "recommandations"),
+]
+

@pytest.fixture(scope="module")
 def dagbag() -> DagBag:
@@ -20,7 +28,7 @@ def test_dags_folder_has_no_import_error(dagbag: DagBag) -> None:


 def test_every_expected_dag_is_discovered(dagbag: DagBag) -> None:
-    assert set(dagbag.dag_ids) == {"ml_train", "ml_score"}
+    assert set(dagbag.dag_ids) == set(DAG_IDS)


 def test_ml_train_has_no_schedule(dagbag: DagBag) -> None:
@@ -32,6 +40,12 @@ def test_ml_score_runs_every_hour(dagbag: DagBag) -> None:
    assert dagbag.dags["ml_score"].timetable.summary == "0 * * * *"


+def test_alertes_runs_after_the_hourly_scoring(dagbag: DagBag) -> None:
+    # Le decalage n'est pas cosmetique : la regle `anomaly` compare une lecture a la `prediction`
+    # du meme instant, que `ml_score` ecrit a l'heure pile.
+    assert dagbag.dags["alertes"].timetable.summary == "15 * * * *"
+
+
 def test_ml_train_task_calls_the_training_module(dagbag: DagBag) -> None:
    tache = dagbag.dags["ml_train"].get_task("train")
    assert "enervision_ml.train" in tache.bash_command
@@ -42,6 +56,28 @@ def test_ml_score_task_calls_the_scoring_module(dagbag: DagBag) -> None:
    assert "enervision_ml.score" in tache.bash_command


+def test_alertes_detection_task_calls_the_backend_detection(dagbag: DagBag) -> None:
+    tache = dagbag.dags["alertes"].get_task("detection")
+    assert "app.detection.internal_alerts" in tache.bash_command
+
+
+def test_alertes_recommendation_task_calls_the_backend_cli(dagbag: DagBag) -> None:
+    tache = dagbag.dags["alertes"].get_task("recommandations")
+    assert "app.cli generate-recommendations" in tache.bash_command
+
+
+@pytest.mark.parametrize("task_id", ["detection", "recommandations"])
+def test_alertes_tasks_run_in_the_backend_environment(dagbag: DagBag, task_id: str) -> None:
+    # Le backend a son propre venv dans l'image, distinct de celui de ml/ (ADR 0008).
+    assert "/opt/backend" in dagbag.dags["alertes"].get_task(task_id).bash_command
+
+
+def test_alertes_generates_recommendations_after_detecting(dagbag: DagBag) -> None:
+    # `recommendation.alert_id` est une cle etrangere `NOT NULL` : la generation n'a rien a lire
+    # tant que la detection n'a pas ecrit.
+    assert dagbag.dags["alertes"].get_task("detection").downstream_task_ids == {"recommandations"}
+
+
 def test_ml_score_reuses_the_model_path_written_by_ml_train(dagbag: DagBag) -> None:
    entrainement = dagbag.dags["ml_train"].get_task("train").bash_command
    scoring = dagbag.dags["ml_score"].get_task("score").bash_command
@@ -51,14 +87,14 @@ def test_ml_score_reuses_the_model_path_written_by_ml_train(dagbag: DagBag) -> N
    assert chemin_modele in scoring


-@pytest.mark.parametrize("dag_id", ["ml_train", "ml_score"])
+@pytest.mark.parametrize("dag_id", DAG_IDS)
 def test_no_two_runs_of_a_dag_overlap(dagbag: DagBag, dag_id: str) -> None:
    # Deux entrainements ecriraient le meme fichier modele, deux scorings inseriraient en meme
-    # temps dans `prediction`.
+    # temps dans `prediction`, deux detections analyseraient la meme fenetre.
    assert dagbag.dags[dag_id].max_active_runs == 1


-@pytest.mark.parametrize(("dag_id", "task_id"), [("ml_train", "train"), ("ml_score", "score")])
+@pytest.mark.parametrize(("dag_id", "task_id"), TACHES)
 def test_every_task_has_an_execution_timeout(dagbag: DagBag, dag_id: str, task_id: str) -> None:
    # Sans plafond, une connexion pendue immobilise un slot du scheduler indefiniment.
    assert dagbag.dags[dag_id].get_task(task_id).execution_timeout is not None
@@ -70,13 +106,29 @@ def test_ml_score_execution_timeout_stays_below_its_hourly_step(dagbag: DagBag)
    assert timeout < timedelta(hours=1)


+def test_alertes_execution_timeouts_stay_below_its_hourly_step(dagbag: DagBag) -> None:
+    # Les deux taches s'enchainent : c'est leur somme qui doit tenir dans le pas horaire.
+    plafonds = [
+        dagbag.dags["alertes"].get_task(task_id).execution_timeout
+        for task_id in ("detection", "recommandations")
+    ]
+    assert all(plafond is not None for plafond in plafonds)
+    assert sum(plafonds, timedelta()) < timedelta(hours=1)
+
+
 def test_ml_score_retries_after_a_transient_failure(dagbag: DagBag) -> None:
    assert dagbag.dags["ml_score"].get_task("score").retries >= 1


-@pytest.mark.parametrize(("dag_id", "task_id"), [("ml_train", "train"), ("ml_score", "score")])
+@pytest.mark.parametrize("task_id", ["detection", "recommandations"])
+def test_alertes_retries_after_a_transient_failure(dagbag: DagBag, task_id: str) -> None:
+    # Les deux commandes sont idempotentes en base, une reprise ne duplique rien.
+    assert dagbag.dags["alertes"].get_task(task_id).retries >= 1
+
+
+@pytest.mark.parametrize(("dag_id", "task_id"), TACHES)
 def test_tasks_never_resync_the_baked_environment(
    dagbag: DagBag, dag_id: str, task_id: str
 ) -> None:
-    # Sans `--no-sync`, `uv run` reconstruit `enervision-ml` a chaque execution.
+    # Sans `--no-sync`, `uv run` reconstruit le projet a chaque execution.
    assert "--no-sync" in dagbag.dags[dag_id].get_task(task_id).bash_command