AIOps aplicado a CI/CD: cuando el pipeline decide que algo va mal.

Hoy en día parece que todo lleva inteligencia artificial: tu móvil tiene IA, tu coche tiene IA… y si hacemos caso a algunos anuncios, hasta el secador de pelo de casa “aprende de ti” para secarte mejor el flequillo.

La palabra IA se ha convertido en un comodín. Sirve igual para describir modelos de deep learning que para sistemas de reglas con un par de if. Y, en medio de todo ese ruido, cada vez es más difícil responder a una pregunta sencilla pero importante: ¿qué significa realmente usar IA en sistemas de ingeniería?

En el mundo de la infraestructura, el CI/CD y la operación de plataformas, la IA no suele tener nada que ver con redes neuronales gigantes ni con modelos que “piensan”. Aquí aparece de una forma mucho más humilde y, curiosamente, mucho más interesante: como una herramienta para detectar comportamientos anómalos y tomar decisiones automáticas.

Este artículo no va de promesas grandilocuentes ni de dashboards llenos de gráficos futuristas. Va de un problema muy concreto: cómo detectar anomalías en pipelines que, en apariencia, funcionan perfectamente.

El caso práctico: detectar pipelines que se quedan “pensando”

El caso que hemos montado es bastante terrenal: detectar pipelines que tardan demasiado en hacer cosas que, en teoría, deberían ser rápidas. Nada de ciencia ficción ni de modelos que predicen el futuro. Simplemente identificar esos momentos en los que un pipeline se queda pensando más de la cuenta… y fingimos que es normal.

En lugar de decidir a mano cuánto es “demasiado tiempo” con el clásico número sacado de la manga, hemos dejado que el propio sistema observe cómo se comportan los pipelines habitualmente. Si lo normal es tardar unos segundos y, de repente, uno decide tomarse varios minutos (para reflexionar sobre el sentido de la vida, por ejemplo), el sistema levanta la mano.

No porque haya superado un umbral mágico, sino porque no encaja con lo que suele pasar.

El resultado es un pipeline que se vigila a sí mismo. Cuando se comporta como siempre, pasa sin problemas. Cuando empieza a ponerse creativo con los tiempos de espera, falla de forma automática. Sin discusiones, sin dashboards que nadie mira y sin personas decidiendo si “esta vez lo dejamos pasar”.

Al final, no se trata de hacer pipelines más listos, sino de hacerlos un poco menos confiados. Y, sobre todo, de dejar que los datos sean los que digan cuándo algo deja de ser normal.

¿Qué herramientas hemos utilizado para conseguirlo?

Para llegar al éxito de nuestra solución, tenemos que contar algunas herramientas que completen el engranaje de ejecución del pipeline, detección de anomalías y entrenamiento del modelo para su automatización.

Argo Workflows: la ejecución y el control del pipeline

Argo Workflows es el motor de ejecución. Aquí viven los pipelines reales: en este caso, workflows que simulan trabajos con duraciones variables y que, al finalizar, ejecutan un paso de validación automática.

Un punto clave del diseño es el uso de onExit mediante el cual:

• El workflow ejecuta su lógica principal.
• Al terminar, se lanza un paso de evaluación AIOps.
• Si ese paso falla, el workflow completo se marca como fallido.

Esto permite que la detección de anomalías forme parte del ciclo de vida del pipeline, no de un sistema externo que solo observa.

Un ejemplo de cómo aplicarlo sería construir un workflow que simule un pipeline. La configuración del pipeline tendrá un sleep de un segundo, pero añadiremos "la variable" (no sé cómo explicarlo mejor) de que, en algunas ocasiones, el sleep sea de 300 segundos.

Prometheus: la fuente de verdad del comportamiento

Prometheus actúa como fuente de datos históricos. No toma decisiones pero recoge métricas clave, como la duración de los workflows, y las expone de forma consistente.

• Construir el dataset de entrenamiento del modelo.
• Entender cómo se comporta el sistema en condiciones normales.
• Separar señal real de ruido.

Un punto importante es que Prometheus no se usa como motor de ML, sino como base de datos de series temporales.

Exportando métricas desde Prometheus para aprender

Hemos creado un exporter propio porque necesitábamos una métrica simple y fiable con la duración total de cada workflow. Argo Workflows no expone ese dato de forma directa y, para un sistema de AIOps, la duración es una señal clave del comportamiento de un pipeline.

El exporter se limita a consultar el estado de los workflows, calcular su tiempo de ejecución y exponerlo como una métrica Prometheus. No toma decisiones ni aplica lógica: solo convierte estado interno de la plataforma en observabilidad reutilizable.

De este modo, mantenemos separadas las responsabilidades: el exporter genera datos, Prometheus los almacena y la capa de IA decide cómo interpretarlos.

from kubernetes import client, config
from prometheus_client import Gauge, start_http_server
from kubernetes.config.config_exception import ConfigException
from datetime import datetime
import time

print("Exporter starting...", flush=True)

try:
   config.load_incluster_config()
   print("Using in-cluster config", flush=True)
except ConfigException:
   config.load_kube_config()
   print("Using local kubeconfig", flush=True)

api = client.CustomObjectsApi()

DURATION = Gauge(
   "argo_pipeline_duration_seconds",
   "Workflow duration",
   ["workflow"]
)

def parse(ts):
   return datetime.fromisoformat(ts.replace("Z", "+00:00"))

start_http_server(8000)
print("Metrics server listening on :8000", flush=True)

while True:
   wfs = api.list_namespaced_custom_object(
       group="argoproj.io",
       version="v1alpha1",
       namespace="argo",
       plural="workflows"
   )

   count = 0
   for wf in wfs["items"]:
       status = wf.get("status", {})
       if status.get("phase") == "Succeeded":
           start = parse(status["startedAt"])
           end = parse(status["finishedAt"])
           duration = (end - start).total_seconds()
           DURATION.labels(
               workflow=wf["metadata"]["name"]
           ).set(duration)
           count += 1

   print(f"Updated {count} workflows", flush=True)
   time.sleep(30)

Con la query argo_pipeline_duration_seconds{workflow=~"sleep-random-.*" podremos ver los datos exportados en Prometheus.

El trainer: cómo enseñamos al sistema qué es normal

Hemos implementado un trainer sencillo y explícito cuyo único objetivo es aprender cuál es la duración “normal” de nuestros workflows a partir de datos históricos reales.

El proceso es simple:

• Cargamos un dataset con duraciones históricas.
• Validamos que haya suficientes datos para entrenar.
• Calculamos estadísticas básicas para tener visibilidad del comportamiento.
• Entrenamos un modelo de detección de anomalías (Isolation Forest).
• Guardamos el modelo entrenado en un volumen persistente.

El entrenamiento se ejecuta como un Job/CronJob en Kubernetes, lo que nos permite refrescar el modelo periódicamente sin impactar en la ejecución de los pipelines. No hay lógica compleja ni dependencias pesadas: el valor está en aprender la distribución real del sistema, no en la sofisticación del algoritmo.

De esta forma, la definición de “normal” evoluciona con el tiempo y se adapta automáticamente a los cambios del entorno.

import pandas as pd
from sklearn.ensemble import IsolationForest
import joblib
import os
import time

def log(msg):
   print(f"[TRAINER] {msg}", flush=True)

DATASET_PATH = "/data/dataset.csv"
MODEL_PATH = "/models/sleep-random.pkl"
MIN_SAMPLES = 20

log("Starting model training")

if not os.path.exists(DATASET_PATH):
   log("Dataset not found, aborting training")
   exit(1)

df = pd.read_csv(DATASET_PATH)

log(f"Loaded dataset with {len(df)} samples")

if len(df) < MIN_SAMPLES:
   log(f"Not enough samples (<{MIN_SAMPLES}), skipping training")
   exit(0)

min_d = df["duration"].min()
max_d = df["duration"].max()
mean_d = df["duration"].mean()

log(f"Duration stats → min={min_d:.2f}s max={max_d:.2f}s mean={mean_d:.2f}s")

X = df[["duration"]]

log("Training IsolationForest model")

model = IsolationForest(
   contamination=0.02,
   random_state=42
)

start = time.time()
model.fit(X)
elapsed = time.time() - start

log(f"Model trained in {elapsed:.2f}s")

# 4️⃣ Guardado del modelo
joblib.dump(model, MODEL_PATH)
log(f"Model saved to {MODEL_PATH}")

ts = int(time.time())
versioned_path = f"/models/sleep-random-{ts}.pkl"
joblib.dump(model, versioned_path)
log(f"Versioned model saved to {versioned_path}")

log("Training job completed successfully")

El workflow de prueba: poniendo a trabajar al sistema

Para validar toda la arquitectura, hemos creado un workflow de Argo Workflows sencillo pero intencionadamente variable. Su objetivo no es hacer trabajo real, sino generar ejecuciones con comportamientos distintos que nos permitan comprobar si el sistema de AIOps funciona como esperamos.

El workflow ejecuta un único step que duerme un tiempo aleatorio:

• La mayoría de las veces simula una ejecución rápida.
• De forma puntual fuerza una ejecución anormalmente lenta.

De esta manera, introducimos ruido controlado y creamos una señal clara que el sistema puede aprender.

La parte clave está en el onExit. Cuando el workflow termina, se ejecuta un paso adicional que:

• Carga el modelo de machine learning entrenado.
• Evalúa la duración total del workflow.
• Decide si esa ejecución es normal o anómala.

Si el modelo detecta una anomalía, el workflow falla automáticamente. Si no, se da por válido.

Esto convierte la detección de anomalías en parte del propio pipeline, no en un análisis externo a posteriori.

Este workflow nos sirve como banco de pruebas para ejecutar el sistema varias veces, observar cómo evoluciona el modelo y verificar que la detección automática funciona de forma consistente en condiciones reales.

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
 generateName: sleep-random-
 namespace: argo
spec:
 serviceAccountName: mmartin
 entrypoint: main
 onExit: aiops-check

 templates:
   - name: main
     steps:
       - - name: random-sleep
           template: sleep

   - name: sleep
     container:
       image: alpine:3.19
       command: [sh, -c]
       args:
         - |
           R=$((RANDOM % 5))
           if [ "$R" -eq 0 ]; then
             SLEEP_TIME=300
             echo "🔥 ANOMALOUS RUN: sleeping ${SLEEP_TIME}s"
           else
             SLEEP_TIME=1
             echo "Normal run: sleeping ${SLEEP_TIME}s"
           fi

           sleep ${SLEEP_TIME}

   - name: aiops-check
     container:
       image: quitos90/argo-aiops-ml-check:0.1
       imagePullPolicy: Always
       command: ["python", "-u", "/app/aiops-check-ml.py"]
       env:
         - name: STEP_DURATION
           value: "{{workflow.duration}}"
       volumeMounts:
         - name: models
           mountPath: /models

 volumes:
   - name: models
     persistentVolumeClaim:
       claimName: aiops-models

Ejecutamos el job unas cuantas veces y…

Voilà! vemos los resultados en Argo Workflows.

Si vemos las dos ejecuciones que han fallado, son las que duran más de 5 minutos y, si miramos en detalle, veremos que son los que hemos forzado a ser las anomalías.

Con esto, y aunque el pipeline a simple vista ha terminado correctamente ya que no ha fallado en su tarea, podemos ver que hay algo raro que está ocurriendo. En este caso está tardando 5 minutos cuando la mayoría de pipelines se completan en aproximadamente 40 segundos.

¿Por qué no usar simplemente un timeout?

La solución más habitual para controlar pipelines lentos es añadir un timeout fijo. Es simple, fácil de entender y funciona… hasta que deja de hacerlo. El enfoque que hemos seguido con AIOps aborda limitaciones que los timeouts no pueden resolver.

Un timeout es estático

Un timeout define un límite rígido: si el pipeline tarda más, falla. Esto obliga a elegir un valor arbitrario que casi nunca es perfecto:

• Demasiado bajo → falsos positivos.
• Demasiado alto → problemas reales pasan desapercibidos.

Nuestro enfoque, en cambio, aprende cuál es el comportamiento normal del sistema y se adapta con el tiempo.

Conclusiones

Después de todo el recorrido, la principal conclusión es clara: la dificultad de aplicar AIOps no está en el modelo, sino en entrenarlo correctamente. El algoritmo utilizado es sencillo, pero su comportamiento depende por completo de la calidad y representatividad de los datos con los que aprende.

Para que un sistema de este tipo funcione, es imprescindible contar con buena observabilidad y suficiente histórico. Sin métricas fiables y sin contexto temporal, cualquier intento de “inteligencia” acaba degenerando en reglas arbitrarias o falsos positivos constantes. La IA no corrige una falta de visibilidad; solo amplifica lo que ya existe.

Este ejercicio deja una lección importante: antes de pensar en modelos más complejos o en arquitecturas más sofisticadas, es fundamental invertir en entender el comportamiento real del sistema. En AIOps, la inteligencia empieza mucho antes del entrenamiento y suele estar más cerca de la observabilidad que del machine learning.