Johan LEROY
f8d08c8686
`execution_timeout` plafonne une tentative, pas la tâche. Avec deux reprises, quinze minutes par tentative autorisaient quarante-neuf minutes par tâche et quatre-vingt-dix-huit pour l'enchaînement, quand le commentaire annonçait une somme tenant sous le pas horaire. Le plafond passe à cinq minutes, ce qui borne le pire cas à trente-huit minutes, et le test d'intégrité calcule désormais ce pire cas plutôt que la somme des plafonds : reprises et délais d'attente compris, c'est la durée qu'un `max_active_runs=1` fait payer à l'exécution suivante. La CI vérifie aussi `app.cli generate-recommendations --help` sans réseau. C'est la seconde commande du DAG, et son import tire FastAPI, les repositories et les services, donc une part de l'environnement `/opt/backend` que la détection seule ne touche pas. `10-infra.md` nomme enfin ce que le décalage de quinze minutes ne garantit pas : le plafond de `ml_score` valant trente minutes, un scoring qui déborde prive la règle `anomaly` de la prédiction de l'heure, qu'elle ne retrouvera au passage suivant que si sa fenêtre la couvre encore.
66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
"""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)
|
|
# `execution_timeout` vaut par tentative : c'est le pire cas des deux tâches enchaînées, reprises
|
|
# et délais compris, qui doit tenir sous le pas horaire. Les tests d'intégrité en font le calcul.
|
|
PLAFOND_PAR_TACHE = timedelta(minutes=5)
|
|
|
|
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
|