Nivel 100: Auditoría de Seguridad (Alerts History)

Objetivo

Aprender a auditar el comportamiento del sistema a través del tiempo. Verás cómo consultar el historial completo de alertas disparadas, incluso desde diferentes instancias de pipelines que comparten la misma base de datos de seguimiento.

Conceptos Clave

• Persistencia de Alertas: Las alertas no solo se disparan en consola; se guardan con metadatos completos en SQLite.

• Tracker Querying: Uso de get_fired_alerts() para extraer inteligencia operacional.

• Correlación de Eventos: Cómo múltiples ejecuciones (L100 y L100b) contribuyen a un historial común de salud del sistema.

¿Qué estamos probando?

Estamos probando la capacidad de “memoria” del motor. Configuramos un umbral de alerta muy sensible (duración > 1ms) para forzar su activación y luego validamos que el sistema es capaz de recordar y listar estas alertas con precisión, diferenciando entre distintas ejecuciones.

ACTIVAR MODO INTERACTIVO

Código Fuente

"""
DEMO LEVEL 100: Alerts Historial
--------------------------------
Adds: View history completo de alerts.
Continues: L99.

DIAGRAM:
get_fired_alerts() en pipeline diferente
"""

import time

from wpipe import Pipeline, step, Metric, Severity

@step(name="task")
def task(data: dict) -> None:

    """Task step.

    Args:

        data: Input data for the step.

    Returns:

        dict: Result of the step.

    """
    time.sleep(0.02)
    return {"ok": True}

if __name__ == "__main__":
    print(">>> Alerts history...")

    pipe = Pipeline(
        pipeline_name="viaje_l100_history",
        verbose=True,
        tracking_db="output/alerts100.db",
    )

    pipe.tracker.add_alert_threshold(
        metric=Metric.PIPELINE_DURATION,
        expression=">1",
        severity=Severity.WARNING,
    )

    pipe.set_steps([task])
    pipe.run({})

    print(">>> Running second pipeline...")
    pipe2 = Pipeline(
        pipeline_name="viaje_l100b",
        verbose=False,
        tracking_db="output/alerts100.db",
    )
    pipe2.set_steps([task])
    pipe2.run({})

    alerts = pipe.tracker.get_fired_alerts()
    print(f"\n🚨 Total in history: {len(alerts)}")

Resultado de Ejecución

---

>>> Alerts history...
[PIPELINE STATUS] Registered: PIPE-E91B7989
viaje_l100_history ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00
[PIPELINE STATUS] PIPE-E91B7989: COMPLETED
>>> Running second pipeline...
viaje_l100_history ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00
viaje_l100b        ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00

🚨 Total in history: 0