Kubeflow and MLflow: industrialising a model that analyses its own cluster data
Training a model on your cluster metrics and logs, deploying it, monitoring it and retraining it when it drifts — the complete MLOps chain, declarative and reproducible.
A Kubernetes cluster produces a considerable amount of signal: resource metrics, events, logs, traces. Most teams look at them after the fact, once the incident has already happened.
The idea developed here is to train a model on those signals to catch anomalies before they become outages — and above all to industrialise the whole cycle, because a model that is neither monitored nor retrained degrades within weeks.
Two tools share the work. Kubeflow Pipelines orchestrates the training steps. MLflow keeps the experiments, versions the models and manages their promotion. They complement each other rather than compete.
Splitting the responsibilities
| Concern | Tool | Reason |
|---|---|---|
| Step orchestration | Kubeflow Pipelines | Native execution on Kubernetes, failure recovery |
| Experiment tracking | MLflow Tracking | Run comparison, parameters, metrics |
| Model registry | MLflow Registry | Promotion stages, semantic versioning |
| Inference service | KServe | Scale to zero, progressive rollout |
| Drift detection | Evidently + Prometheus | Distribution comparison over time |
Kubeflow without MLflow leaves you with no usable experiment history. MLflow without Kubeflow forces you to orchestrate by hand. Pairing them covers the full cycle.
Collecting the cluster data
The source is already there: Prometheus holds the metrics, Loki the logs. The work is turning them into a training dataset.
"""Extracting cluster metrics into a timestamped dataset."""
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]))',
"p99_latency": 'histogram_quantile(0.99, sum by (pod, le) (rate(http_request_duration_seconds_bucket[5m])))',
"error_rate": '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 extract(url: str, days: int = 30, step: str = "5m") -> pd.DataFrame:
"""Builds a pivoted frame: one row per pod and per time step."""
client = PrometheusConnect(url=url, disable_ssl=False)
end = datetime.now(timezone.utc)
start = fin - timedelta(days=jours)
morceaux = []
for nom, query in REQUETES.items():
brut = client.custom_query_range(
query=query, start_time=start, end_time=end, step=step
)
for serie in brut:
pod = serie["metric"].get("pod", "inconnu")
df = pd.DataFrame(serie["values"], columns=["timestamp", nom])
df["pod"] = pod
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="s", utc=True)
df[nom] = pd.to_numeric(df[nom], errors="coerce")
morceaux.append(df.set_index(["timestamp", "pod"])[[nom]])
dataset = pd.concat(morceaux, axis=1).sort_index()
# Missing values are forward-filled, never zero-filled:
# an artificial zero looks exactly like genuine idle load.
return dataset.groupby(level="pod").ffill().dropna()The comment about missing values is not incidental. Filling with zero teaches the model that a missing measurement equals an absence of activity — which produces massive false positives whenever collection breaks.
The Kubeflow pipeline
Every step runs in its own container. Artefacts travel through the cluster object storage.
"""Training pipeline for the anomaly detector."""
from kfp import dsl, compiler
from kfp.dsl import Dataset, Input, Metrics, Model, Output
IMAGE = "registry.interne/mlops-platform:2026.08"
@dsl.component(base_image=IMAGE, packages_to_install=["prometheus-api-client==0.5.7"])
def extraction(url_prometheus: str, jours: int, output: Output[Dataset]):
import pandas as pd
from extraction_cluster import extract
dataset = extract(url_prometheus, jours=jours)
dataset.to_parquet(output.path)
print(f"{len(dataset)} observations extracted over {days} days")
@dsl.component(base_image=IMAGE)
def preparation(entry: Input[Dataset], training: Output[Dataset],
validation: Output[Dataset]):
"""Chronological split: never random on a time series."""
import pandas as pd
dataset = pd.read_parquet(entry.path).sort_index()
threshold = int(len(dataset) * 0.8)
dataset.iloc[:threshold].to_parquet(training.path)
dataset.iloc[threshold:].to_parquet(validation.path)
@dsl.component(base_image=IMAGE, packages_to_install=["mlflow==2.16.2"])
def entrainement(training: Input[Dataset], validation: Input[Dataset],
contamination: float, uri_mlflow: str,
model: Output[Model], measures: 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(training.path)
val = pd.read_parquet(validation.path)
with mlflow.start_run() as execution:
mlflow.log_params({
"contamination": contamination,
"n_estimators": 300,
"observations_training": len(app),
})
# RobustScaler rather than StandardScaler: infrastructure metrics
# carry outliers that would flatten a classic normalisation.
scaler = RobustScaler().fit(app)
detector = IsolationForest(
n_estimators=300, contamination=contamination,
random_state=42, n_jobs=-1,
).fit(scaler.transform(app))
scores = detector.decision_function(scaler.transform(val))
anomalies = (scores < 0).sum()
rate = float(anomalies) / len(val)
mlflow.log_metrics({
"validation_anomaly_rate": taux,
"score_moyen": float(scores.mean()),
})
measures.log_metric("validation_anomaly_rate", taux)
mlflow.sklearn.log_model(
sk_model=detector,
artifact_path="model",
registered_model_name="detector-anomalies-cluster",
input_example=val.head(3),
)
print(f"Exécution {execution.info.run_id} — taux d'anomalies : {rate:.3%}")
@dsl.component(base_image=IMAGE, packages_to_install=["mlflow==2.16.2"])
def promotion(uri_mlflow: str, max_threshold: float, measures: Input[Metrics]):
"""Promotes only if the anomaly rate stays plausible."""
import mlflow
from mlflow.tracking import MlflowClient
client = MlflowClient(tracking_uri=uri_mlflow)
versions = client.search_model_versions("name='detector-anomalies-cluster'")
derniere = max(versions, key=lambda v: int(v.version))
rate = float(client.get_run(derniere.run_id).data.metrics["validation_anomaly_rate"])
if taux > max_threshold:
raise ValueError(
f"Anomaly rate of {rate:.3%} above the {max_threshold:.3%} threshold: "
"promotion refused, the model over-detects."
)
client.set_registered_model_alias(
name="detector-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",
max_threshold: float = 0.08):
brut = extraction(url_prometheus=url_prometheus, jours=jours)
decoupe = preparation(entry=brut.outputs["output"])
entraine = entrainement(
training=decoupe.outputs["training"],
validation=decoupe.outputs["validation"],
contamination=contamination, uri_mlflow=uri_mlflow,
)
entraine.set_cpu_limit("4").set_memory_limit("16Gi")
promotion(uri_mlflow=uri_mlflow, max_threshold=max_threshold,
measures=entraine.outputs["measures"])
if __name__ == "__main__":
compiler.Compiler().compile(pipeline, "pipeline-anomalies.yaml")The promotion step is the essential guard. A model reporting 40 % anomalies is not a detector: it is a noise generator that will make everyone ignore every alert. The threshold blocks it from reaching production.
The chronological split, a classic mistake
The dataset.iloc[:threshold] line deserves a paragraph. On a time series, a random split leaks information from the future into the past: the model learns from observations that come after the ones it must predict. Validation metrics turn flattering and the model collapses in production.
A chronological split respects the arrow of time. The figures you get are more modest, but they reflect actual behaviour.
Scheduled runs and retraining
A model trained once degrades. Workloads evolve, new services appear, profiles change. Periodic retraining is not optional.
apiVersion: kubeflow.org/v2beta1
kind: RecurringRun
metadata:
name: reentrainement-hebdomadaire
namespace: mlops
spec:
description: "Retraining the detector on the last 30 days"
pipelineVersionReference:
pipelineId: detection-anomalies-cluster
trigger:
cronSchedule:
cron: "0 3 * * 1" # Monday, 3 a.m.
startTime: "2026-09-01T00:00:00Z"
maxConcurrency: 1
noCatchup: true # no catch-up after an outage
runtimeConfig:
parameters:
jours: 30
contamination: 0.02
max_threshold: 0.08Watching for drift
Two kinds of drift stand apart. Data drift: the input distribution changes. Concept drift: the relationship between inputs and output changes.
On a cluster, the first dominates. A new service arriving shifts the latency and consumption distributions without the underlying phenomenon changing at all.
"""Drift monitoring exposed as Prometheus metrics."""
from evidently.metric_preset import DataDriftPreset
from evidently.report import Report
from prometheus_client import Gauge, start_http_server
DERIVE = Gauge("mlops_data_drift_ratio",
"Proportion de features ayant dérivé", ["model"])
COLONNES_DERIVEES = Gauge("mlops_colonne_derivee",
"Variable en dérive", ["model", "variable"])
def evaluate(reference, courant, model: str) -> float:
rapport = Report(metrics=[DataDriftPreset()])
rapport.run(reference_data=reference, current_data=courant)
result = rapport.as_dict()["metrics"][0]["result"]
DERIVE.labels(model=model).set(result["share_of_drifted_columns"])
for variable, detail in result["drift_by_columns"].items():
COLONNES_DERIVEES.labels(model=model, variable=variable).set(
1 if detail["drift_detected"] else 0
)
return result["share_of_drifted_columns"]The alert then fires in Prometheus, in the same place as the rest of your monitoring.
groups:
- name: mlops
rules:
- alert: DeriveDonneesDetectee
expr: mlops_data_drift_ratio{model="detector-anomalies-cluster"} > 0.3
for: 6h
labels: { severity: warning }
annotations:
summary: "Drift on {{ $value | humanizePercentage }} of the features"
description: "Model {{ $labels.model }} sees distributions far from its reference. Trigger a retraining run."What the complete chain produces
On Monday morning the pipeline runs. It extracts the last thirty days of metrics, retrains, evaluates. If the anomaly rate stays plausible, the new version goes to production. Otherwise the previous one stays in place and the failure surfaces as an alert.
During the week, the inference service scores metrics continuously and publishes a value. When drift exceeds 30 % of the features for six hours, an alert invites retraining without waiting for the following Monday.
None of this happens by hand. The pipeline, the schedule, the promotion thresholds and the alerting rules live in the Git repository, applied by Argo CD. Changing the contamination threshold means opening a merge request — not logging into a console.
That is precisely what separates an MLOps chain from a notebook run by hand: reproducibility, traceability, and the ability to roll back.