From ae58a896d974a03a0e72f7f68e3a0588e5dad029 Mon Sep 17 00:00:00 2001 From: Johan LEROY Date: Mon, 21 Sep 2026 12:13:55 +0200 Subject: [PATCH] =?UTF-8?q?feat(etl):=20ordonnance=20la=20d=C3=A9tection?= =?UTF-8?q?=20d'alertes=20et=20les=20recommandations=20par=20un=20DAG=20Ai?= =?UTF-8?q?rflow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 `/.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. --- .env.example | 5 +++ .github/workflows/airflow.yml | 20 +++++++++-- Makefile | 5 ++- docker-compose.yml | 7 ++++ etl/airflow/Dockerfile | 26 ++++++++++---- etl/airflow/dags/alertes.py | 65 ++++++++++++++++++++++++++++++++++ etl/airflow/tests/test_dags.py | 64 +++++++++++++++++++++++++++++---- 7 files changed, 175 insertions(+), 17 deletions(-) create mode 100644 etl/airflow/dags/alertes.py diff --git a/.env.example b/.env.example index e0abef3..b3ff600 100644 --- a/.env.example +++ b/.env.example @@ -38,3 +38,8 @@ AIRFLOW_ADMIN_USERNAME=admin # comptes `app_user` d'EnerVision. AIRFLOW_ADMIN_PASSWORD=change_me AIRFLOW_ADMIN_EMAIL=admin@enervision.fr +# `APP_SECRET_KEY` du backend, que le DAG `alertes` lance en sous-processus. Distincte de +# celle de l'API : la détection ne signe aucun jeton, et Airflow exécute du code depuis son +# interface (cf. ADR 0008). Générer la vôtre : +# python -c "import secrets; print(secrets.token_urlsafe(48))" +AIRFLOW_APP_SECRET_KEY=change_me diff --git a/.github/workflows/airflow.yml b/.github/workflows/airflow.yml index a79f97f..9616da2 100644 --- a/.github/workflows/airflow.yml +++ b/.github/workflows/airflow.yml @@ -4,8 +4,9 @@ name: Airflow # (contrairement à backend.yml et ml.yml) : apache-airflow 2.10 ne supporte pas 3.14. Le 3.14 de # ml/ ne vit que dans l'image Docker, dans son propre environnement (cf. etl/airflow/Dockerfile). # -# Piège : l'image COPY les fichiers de dépendances et le code de ml/. Une modification de ml/ -# peut donc casser sa construction, d'où ces chemins dans les déclencheurs. +# Piège : l'image COPY les fichiers de dépendances et le code de ml/ et de apps/backend/. Une +# modification de l'un ou de l'autre peut donc casser sa construction, d'où ces chemins dans +# les déclencheurs, alors même que ce workflow ne teste ni le modèle ni l'API. on: push: @@ -14,6 +15,9 @@ on: - "ml/pyproject.toml" - "ml/uv.lock" - "ml/enervision_ml/**" + - "apps/backend/pyproject.toml" + - "apps/backend/uv.lock" + - "apps/backend/app/**" - ".github/workflows/airflow.yml" pull_request: paths: @@ -21,6 +25,9 @@ on: - "ml/pyproject.toml" - "ml/uv.lock" - "ml/enervision_ml/**" + - "apps/backend/pyproject.toml" + - "apps/backend/uv.lock" + - "apps/backend/app/**" - ".github/workflows/airflow.yml" permissions: @@ -73,7 +80,7 @@ jobs: - name: Récupère le dépôt uses: actions/checkout@v4 - - name: Construit l'image (contexte à la racine, elle COPY ml/) + - name: Construit l'image (contexte à la racine, elle COPY ml/ et apps/backend/) run: docker build -f etl/airflow/Dockerfile -t enervision-airflow:ci . # Vérifie ce qui ne casse qu'à l'exécution, pas à la construction : libgomp1 absent @@ -82,3 +89,10 @@ jobs: run: > docker run --rm --network none enervision-airflow:ci bash -c "cd /opt/ml && env -u VIRTUAL_ENV uv run --no-sync python -m enervision_ml.train --help" + + # `--help` sort par argparse avant `get_settings()` : ni base ni secret requis, et + # l'import du module prouve que l'environnement /opt/backend est complet. + - name: Vérifie que la détection d'alertes s'importe sans réseau + run: > + docker run --rm --network none enervision-airflow:ci + bash -c "cd /opt/backend && env -u VIRTUAL_ENV uv run --no-sync python -m app.detection.internal_alerts --help" diff --git a/Makefile b/Makefile index 66f4298..51b6e67 100644 --- a/Makefile +++ b/Makefile @@ -8,7 +8,7 @@ AIRFLOW := etl/airflow dev dev-backend dev-frontend \ lint format typecheck test test-cov test-integration check \ openapi docker-build db-up db-down db-reset db-logs db-psql migrate bootstrap-admin \ - ml-lint ml-typecheck ml-test ml-check ml-train ml-score recommendations \ + ml-lint ml-typecheck ml-test ml-check ml-train ml-score detect-alerts recommendations \ airflow-lint airflow-test airflow-check airflow-up airflow-down airflow-logs help: ## Liste les cibles disponibles @@ -83,6 +83,9 @@ ml-train: ## Entraine le modele LightGBM. CSV=chemin optionnel, sinon lit ML_DAT ml-score: ## Score le prochain pas horaire et l'ecrit dans `prediction`. CSV=chemin optionnel cd $(ML) && uv run python -m enervision_ml.score $(if $(CSV),--csv $(CSV),) +detect-alerts: ## Détecte les alertes internes depuis les lectures en base. SITE= et NOW= optionnels + cd $(BACKEND) && uv run python -m app.detection.internal_alerts $(if $(SITE),--site-id $(SITE),) $(if $(NOW),--now $(NOW),) + recommendations: ## Genere les recommandations depuis les alertes en base. SITE=identifiant optionnel cd $(BACKEND) && uv run python -m app.cli generate-recommendations $(if $(SITE),--site-id $(SITE),) diff --git a/docker-compose.yml b/docker-compose.yml index 5f6c9c0..73ffa64 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -27,6 +27,12 @@ x-airflow-common: &airflow-common # memes identifiants que le backend en attendant. ML_DATABASE_URL: postgresql+psycopg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB} MLFLOW_TRACKING_URI: sqlite:////opt/ml/state/mlflow.db + # Le DAG `alertes` lance le backend en sous-processus : il lit `DATABASE_URL`, en + # dialecte asyncpg, là où le pipeline ML lit `ML_DATABASE_URL`. + DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB} + # Clé distincte de celle de l'API : la détection ne signe ni ne vérifie aucun jeton, et + # Airflow permet d'exécuter du code depuis son interface (cf. ADR 0008). + APP_SECRET_KEY: ${AIRFLOW_APP_SECRET_KEY:-} volumes: - ./etl/airflow/dags:/opt/airflow/dags - ./etl/airflow/plugins:/opt/airflow/plugins @@ -129,6 +135,7 @@ services: set -euo pipefail : "$${AIRFLOW__CORE__FERNET_KEY:?AIRFLOW_FERNET_KEY manquant dans .env}" : "$${AIRFLOW__WEBSERVER__SECRET_KEY:?AIRFLOW_WEBSERVER_SECRET_KEY manquant dans .env}" + : "$${APP_SECRET_KEY:?AIRFLOW_APP_SECRET_KEY manquant dans .env}" exec airflow version airflow-webserver: diff --git a/etl/airflow/Dockerfile b/etl/airflow/Dockerfile index 058928f..b934588 100644 --- a/etl/airflow/Dockerfile +++ b/etl/airflow/Dockerfile @@ -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 `/.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 diff --git a/etl/airflow/dags/alertes.py b/etl/airflow/dags/alertes.py new file mode 100644 index 0000000..f221ab1 --- /dev/null +++ b/etl/airflow/dags/alertes.py @@ -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 diff --git a/etl/airflow/tests/test_dags.py b/etl/airflow/tests/test_dags.py index 58f6ab9..ccfbe5d 100644 --- a/etl/airflow/tests/test_dags.py +++ b/etl/airflow/tests/test_dags.py @@ -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