AIOps en GCP: Arquitectura de Observabilidad Inteligente y Auto-Remediación

import base64
import json
import logging
import os
from flask import Flask, request
from google.cloud import bigquery
import vertexai
from vertexai.generative_models import GenerativeModel

# Configuración de Logging y Entorno
logging.basicConfig(level=logging.INFO)
PROJECT_ID = os.getenv("GCP_PROJECT", "production-core-aiops")
LOCATION = os.getenv("GCP_REGION", "europe-west1")

vertexai.init(project=PROJECT_ID, location=LOCATION)
model = GenerativeModel("gemini-1.5-pro-preview-0409")
bq_client = bigquery.Client(project=PROJECT_ID)

app = Flask(__name__)

def fetch_incident_logs(resource_name: str, lookback_minutes: int = 10) -> str:
    """Extrae logs estructurados desde BigQuery Log Analytics para correlación contextual."""
    query = f"""
    SELECT timestamp, severity, log_id, text_payload, json_payload
    FROM `{PROJECT_ID}.global._Default._AllLogs`
    WHERE resource.type = 'k8s_container'
      AND resource.labels.pod_name LIKE CONCAT('%', @pod_name, '%')
      AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL @minutes MINUTE)
    ORDER BY timestamp DESC
    LIMIT 20
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("pod_name", "STRING", resource_name),
            bigquery.ScalarQueryParameter("minutes", "INT64", lookback_minutes),
        ]
    )
    query_job = bq_client.query(query, job_config=job_config)
    results = [dict(row) for row in query_job]
    return json.dumps(results, default=str)

@app.route("/", methods=["POST"])
def process_alert():
    """Handler del endpoint Pub/Sub invocado por Eventarc ante un incidente."""
    envelope = request.get_json()
    if not envelope or "message" not in envelope:
        return ("Bad Request: Formato PubSub inválido", 400)

    pubsub_msg = envelope["message"]
    if "data" not in pubsub_msg:
        return ("Bad Request: Payload vacío", 400)

    alert_payload = json.loads(base64.b64decode(pubsub_msg["data"]).decode("utf-8"))
    incident = alert_payload.get("incident", {})
    resource_id = incident.get("resource_id", "unknown-service")
    summary = incident.get("summary", "Sin resumen disponible")

    logging.info(f"[AIOps Event] Procesando incidente en recurso: {resource_id}")

    # Paso 1: Enriquecer con logs forenses
    log_context = fetch_incident_logs(resource_id)

    # Paso 2: Análisis cognitivo con Vertex AI
    prompt = f"""
    Eres un Site Reliability Engineer (SRE) Staff en GCP. Analiza la alerta y sus logs:
    Incidente: {summary}
    Contexto de Logs: {log_context}
    
    Genera un JSON con el siguiente esquema exacto:
    {{
      "root_cause": "explicación concisa",
      "severity_level": "CRITICAL|HIGH|MEDIUM|LOW",
      "recommended_action": "RESTART_POD|SCALE_REPLICAS|DRAIN_NODE|NO_ACTION",
      "confidence_score": 0.95
    }}
    """
    response = model.generate_content(prompt)
    
    try:
        clean_text = response.text.replace("json", "").replace("", "").strip()
        analysis = json.loads(clean_text)
        logging.info(f"[AIOps Diagnosis] Causa: {analysis['root_cause']} | Acción: {analysis['recommended_action']}")
        
        # Paso 3: Ejecución de Remediación Idempotente
        if analysis["confidence_score"] >= 0.90 and analysis["recommended_action"] == "RESTART_POD":
            logging.info(f"[Auto-Remediation] Despachando webhook de escalado para {resource_id}")
            # Lógica de mutación segura mediante API de Kubernetes Engine
            
        return ("OK", 200)
    except Exception as e:
        logging.error(f"Fallo en la inferencia o parseo: {str(e)}")
        return ("Internal Error", 500)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))