Kubeflow et MLflow : industrialiser un modèle qui analyse les données de son propre cluster
Entraîner un modèle sur les métriques et les journaux de son cluster, le déployer, le surveiller et le réentraîner quand il dérive — la chaîne MLOps complète, déclarative et reproductible.
Un cluster Kubernetes produit une quantité considérable de signaux : métriques de ressources, événements, journaux, traces. La plupart des équipes les consultent après coup, quand l'incident a déjà eu lieu.
L'idée développée ici consiste à entraîner un modèle sur ces signaux pour détecter les anomalies avant qu'elles ne deviennent des pannes — et surtout à industrialiser tout le cycle, car un modèle qui n'est ni surveillé ni réentraîné se dégrade en quelques semaines.
Deux outils se partagent le travail. Kubeflow Pipelines orchestre les étapes d'entraînement. MLflow conserve les expérimentations, versionne les modèles et gère leur promotion. Ils se complètent plus qu'ils ne se concurrencent.
Répartition des responsabilités
| Préoccupation | Outil | Raison |
|---|---|---|
| Orchestration des étapes | Kubeflow Pipelines | Exécution native sur Kubernetes, reprise sur erreur |
| Suivi des expérimentations | MLflow Tracking | Comparaison des exécutions, paramètres, métriques |
| Registre de modèles | MLflow Registry | Étapes de promotion, versionnement sémantique |
| Service d'inférence | KServe | Mise à l'échelle jusqu'à zéro, déploiement progressif |
| Détection de dérive | Evidently + Prometheus | Comparaison des distributions dans le temps |
Kubeflow sans MLflow vous laisse sans historique exploitable des expérimentations. MLflow sans Kubeflow vous oblige à orchestrer à la main. L'association des deux couvre le cycle complet.
Collecter les données du cluster
La source se trouve déjà là : Prometheus conserve les métriques, Loki les journaux. Le travail consiste à les transformer en jeu de données d'entraînement.
"""Extraction des métriques du cluster vers un jeu de données horodaté."""
from datetime import datetime, timedelta, timezone
import pandas as pd
from prometheus_api_client import PrometheusConnect
REQUETES = {
"cpu_utilisation": 'sum by (pod) (rate(container_cpu_usage_seconds_total{namespace=~"prod-.*"}[5m]))',
"memoire_octets": 'sum by (pod) (container_memory_working_set_bytes{namespace=~"prod-.*"})',
"redemarrages": 'sum by (pod) (increase(kube_pod_container_status_restarts_total[1h]))',
"latence_p99": 'histogram_quantile(0.99, sum by (pod, le) (rate(http_request_duration_seconds_bucket[5m])))',
"taux_erreur": 'sum by (pod) (rate(http_requests_total{code=~"5.."}[5m])) / sum by (pod) (rate(http_requests_total[5m]))',
"attente_ordonnanceur":'sum by (pod) (rate(container_cpu_cfs_throttled_seconds_total[5m]))',
}
def extraire(url: str, jours: int = 30, pas: str = "5m") -> pd.DataFrame:
"""Construit un tableau pivoté : une ligne par pod et par pas de temps."""
client = PrometheusConnect(url=url, disable_ssl=False)
fin = datetime.now(timezone.utc)
debut = fin - timedelta(days=jours)
morceaux = []
for nom, requete in REQUETES.items():
brut = client.custom_query_range(
query=requete, start_time=debut, end_time=fin, step=pas
)
for serie in brut:
pod = serie["metric"].get("pod", "inconnu")
df = pd.DataFrame(serie["values"], columns=["horodatage", nom])
df["pod"] = pod
df["horodatage"] = pd.to_datetime(df["horodatage"], unit="s", utc=True)
df[nom] = pd.to_numeric(df[nom], errors="coerce")
morceaux.append(df.set_index(["horodatage", "pod"])[[nom]])
jeu = pd.concat(morceaux, axis=1).sort_index()
# Les valeurs manquantes se comblent par propagation avant, jamais par zéro :
# un zéro artificiel se confond avec une charge nulle réelle.
return jeu.groupby(level="pod").ffill().dropna()Le commentaire sur les valeurs manquantes n'est pas anodin. Remplir par zéro apprend au modèle qu'une absence de mesure équivaut à une absence d'activité — ce qui provoque des faux positifs massifs lors des coupures de collecte.
Le pipeline Kubeflow
Chaque étape s'exécute dans son propre conteneur. Les artefacts transitent par le stockage objet du cluster.
"""Pipeline d'entraînement du détecteur d'anomalies."""
from kfp import dsl, compiler
from kfp.dsl import Dataset, Input, Metrics, Model, Output
IMAGE = "registry.interne/mlops-socle:2026.08"
@dsl.component(base_image=IMAGE, packages_to_install=["prometheus-api-client==0.5.7"])
def extraction(url_prometheus: str, jours: int, sortie: Output[Dataset]):
import pandas as pd
from extraction_cluster import extraire
jeu = extraire(url_prometheus, jours=jours)
jeu.to_parquet(sortie.path)
print(f"{len(jeu)} observations extraites sur {jours} jours")
@dsl.component(base_image=IMAGE)
def preparation(entree: Input[Dataset], apprentissage: Output[Dataset],
validation: Output[Dataset]):
"""Découpage chronologique : jamais aléatoire sur une série temporelle."""
import pandas as pd
jeu = pd.read_parquet(entree.path).sort_index()
seuil = int(len(jeu) * 0.8)
jeu.iloc[:seuil].to_parquet(apprentissage.path)
jeu.iloc[seuil:].to_parquet(validation.path)
@dsl.component(base_image=IMAGE, packages_to_install=["mlflow==2.16.2"])
def entrainement(apprentissage: Input[Dataset], validation: Input[Dataset],
contamination: float, uri_mlflow: str,
modele: Output[Model], mesures: Output[Metrics]):
import mlflow, pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.metrics import precision_score, recall_score
from sklearn.preprocessing import RobustScaler
mlflow.set_tracking_uri(uri_mlflow)
mlflow.set_experiment("detection-anomalies-cluster")
app = pd.read_parquet(apprentissage.path)
val = pd.read_parquet(validation.path)
with mlflow.start_run() as execution:
mlflow.log_params({
"contamination": contamination,
"n_estimators": 300,
"observations_apprentissage": len(app),
})
# RobustScaler plutôt que StandardScaler : les métriques d'infrastructure
# comportent des valeurs extrêmes qui écraseraient une normalisation classique.
echelle = RobustScaler().fit(app)
detecteur = IsolationForest(
n_estimators=300, contamination=contamination,
random_state=42, n_jobs=-1,
).fit(echelle.transform(app))
scores = detecteur.decision_function(echelle.transform(val))
anomalies = (scores < 0).sum()
taux = float(anomalies) / len(val)
mlflow.log_metrics({
"taux_anomalies_validation": taux,
"score_moyen": float(scores.mean()),
})
mesures.log_metric("taux_anomalies_validation", taux)
mlflow.sklearn.log_model(
sk_model=detecteur,
artifact_path="modele",
registered_model_name="detecteur-anomalies-cluster",
input_example=val.head(3),
)
print(f"Exécution {execution.info.run_id} — taux d'anomalies : {taux:.3%}")
@dsl.component(base_image=IMAGE, packages_to_install=["mlflow==2.16.2"])
def promotion(uri_mlflow: str, seuil_max: float, mesures: Input[Metrics]):
"""Ne promeut que si le taux d'anomalies reste plausible."""
import mlflow
from mlflow.tracking import MlflowClient
client = MlflowClient(tracking_uri=uri_mlflow)
versions = client.search_model_versions("name='detecteur-anomalies-cluster'")
derniere = max(versions, key=lambda v: int(v.version))
taux = float(client.get_run(derniere.run_id).data.metrics["taux_anomalies_validation"])
if taux > seuil_max:
raise ValueError(
f"Taux d'anomalies de {taux:.3%} au-dessus du seuil de {seuil_max:.3%} : "
"promotion refusée, le modèle sur-détecte."
)
client.set_registered_model_alias(
name="detecteur-anomalies-cluster", alias="production", version=derniere.version
)
print(f"Version {derniere.version} promue en production")
@dsl.pipeline(name="detection-anomalies-cluster")
def pipeline(url_prometheus: str = "http://prometheus.observabilite.svc:9090",
jours: int = 30, contamination: float = 0.02,
uri_mlflow: str = "http://mlflow.mlops.svc:5000",
seuil_max: float = 0.08):
brut = extraction(url_prometheus=url_prometheus, jours=jours)
decoupe = preparation(entree=brut.outputs["sortie"])
entraine = entrainement(
apprentissage=decoupe.outputs["apprentissage"],
validation=decoupe.outputs["validation"],
contamination=contamination, uri_mlflow=uri_mlflow,
)
entraine.set_cpu_limit("4").set_memory_limit("16Gi")
promotion(uri_mlflow=uri_mlflow, seuil_max=seuil_max,
mesures=entraine.outputs["mesures"])
if __name__ == "__main__":
compiler.Compiler().compile(pipeline, "pipeline-anomalies.yaml")L'étape de promotion constitue le garde-fou essentiel. Un modèle qui signale 40 % d'anomalies n'est pas un détecteur : c'est un générateur de bruit qui fera ignorer toutes les alertes. Le seuil bloque sa mise en production.
Le découpage chronologique, erreur classique
La ligne jeu.iloc[:seuil] mérite un développement. Sur une série temporelle, un découpage aléatoire laisse fuiter de l'information du futur vers le passé : le modèle apprend sur des observations postérieures à celles qu'il doit prédire. Les métriques de validation deviennent flatteuses et le modèle s'effondre en production.
Le découpage chronologique respecte la flèche du temps. Les mesures obtenues sont plus modestes, mais elles reflètent le comportement réel.
Exécution planifiée et réentraînement
Un modèle entraîné une fois se dégrade. Les charges évoluent, de nouveaux services apparaissent, les profils changent. Le réentraînement périodique n'est pas une option.
apiVersion: kubeflow.org/v2beta1
kind: RecurringRun
metadata:
name: reentrainement-hebdomadaire
namespace: mlops
spec:
description: "Réentraînement du détecteur sur les 30 derniers jours"
pipelineVersionReference:
pipelineId: detection-anomalies-cluster
trigger:
cronSchedule:
cron: "0 3 * * 1" # lundi, 3 h du matin
startTime: "2026-09-01T00:00:00Z"
maxConcurrency: 1
noCatchup: true # pas de rattrapage après une interruption
runtimeConfig:
parameters:
jours: 30
contamination: 0.02
seuil_max: 0.08Surveiller la dérive
Deux dérives se distinguent. La dérive des données : la distribution des entrées change. La dérive du concept : la relation entre entrées et sortie change.
Sur un cluster, la première domine. L'arrivée d'un nouveau service modifie les distributions de latence et de consommation sans que le phénomène sous-jacent change.
"""Surveillance de dérive exposée en métriques Prometheus."""
from evidently.metric_preset import DataDriftPreset
from evidently.report import Report
from prometheus_client import Gauge, start_http_server
DERIVE = Gauge("mlops_derive_donnees_ratio",
"Proportion de variables ayant dérivé", ["modele"])
COLONNES_DERIVEES = Gauge("mlops_colonne_derivee",
"Variable en dérive", ["modele", "variable"])
def evaluer(reference, courant, modele: str) -> float:
rapport = Report(metrics=[DataDriftPreset()])
rapport.run(reference_data=reference, current_data=courant)
resultat = rapport.as_dict()["metrics"][0]["result"]
DERIVE.labels(modele=modele).set(resultat["share_of_drifted_columns"])
for variable, detail in resultat["drift_by_columns"].items():
COLONNES_DERIVEES.labels(modele=modele, variable=variable).set(
1 if detail["drift_detected"] else 0
)
return resultat["share_of_drifted_columns"]L'alerte se déclenche ensuite dans Prometheus, au même endroit que le reste de la supervision.
groups:
- name: mlops
rules:
- alert: DeriveDonneesDetectee
expr: mlops_derive_donnees_ratio{modele="detecteur-anomalies-cluster"} > 0.3
for: 6h
labels: { severity: warning }
annotations:
summary: "Dérive sur {{ $value | humanizePercentage }} des variables"
description: "Le modèle {{ $labels.modele }} observe des distributions éloignées de sa référence. Déclencher un réentraînement."Ce que produit la chaîne complète
Le lundi matin, le pipeline s'exécute. Il extrait les trente derniers jours de métriques, réentraîne, évalue. Si le taux d'anomalies reste plausible, la nouvelle version passe en production. Sinon, l'ancienne reste en place et l'échec remonte dans les alertes.
En semaine, le service d'inférence évalue les métriques en continu et publie un score. Quand la dérive dépasse 30 % des variables pendant six heures, une alerte invite à réentraîner sans attendre le lundi suivant.
Rien de tout cela ne se fait à la main. Le pipeline, la planification, les seuils de promotion et les règles d'alerte vivent dans le dépôt Git, appliqués par ArgoCD. Changer le seuil de contamination, c'est ouvrir une demande de fusion — pas se connecter à une interface.
C'est précisément ce qui distingue une chaîne MLOps d'un carnet de notes exécuté à la main : la reproductibilité, la traçabilité, et la capacité à revenir en arrière.