#init_docker
#en este archivo va el comando que lavnata docker
#y ejecuta el cliente.py
#_______________________
#client/cliente.py
import logging
import httpx
logger = logging.getLogger(__name__)
PATH = "http://localhost:8000/runs"
def enviar_datos() -> dict | None:
try:
respuesta = httpx.post(PATH, timeout=30.0)
respuesta.raise_for_status()
resultado = respuesta.json()
logger.INFO("Respuesta:", resultado)
return resultado
except httpx.HTTPStatusError as error:
logger.ERROR("Error HTTP:", error.response.status_code)
logger.INFO("Detalle:", error.response.text)
except httpx.RequestError as error:
logger.ERROR("Error de conexión:", error)
except ValueError as error: # respuesta que no es JSON
logger.ERROR("La respuesta no es JSON válido:", error)
return None
if __name__ == "__main__":
enviar_datos()
#_______________________
# app/life_span.py
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
from contextlib import asynccontextmanager
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
from app.shutdown_completo import shutdown
from data_base.sql_create_tables import create_tables
from routes import histoty, temporal_id, work, alertas
from temporal.cliente_temporal_connet import TemporalClientFactory
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
try:
await create_tables()
except Exception:
logger.exception("error en crear tablas")
raise
try:
app.state.temporal = await TemporalClientFactory.client_conn()
except Exception:
logger.exception("error al conectar temporal")
raise
yield
finally:
await shutdown(getattr(app.state, "temporal", None))
app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
Instrumentator().instrument(app).expose(app)
app.include_router(temporal_id.router)
app.include_router(work.router)
app.include_router(histoty.router)
app.include_router(alertas.router)
#_______________________
#temporal/cliente_temporal_connet.py
from temporalio.client import Client
from temporalio.contrib.pydantic import pydantic_data_converter
import logging
from core.config import TEMPORAL_HOST, TEMPORAL_NAMESPACE
from temporalio.runtime import Runtime, TelemetryConfig, PrometheusConfig
class TemporalClientFactory:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.runtime = Runtime(telemetry=TelemetryConfig(metrics=PrometheusConfig(bind_address="0.0.0.0:9464")))
async def client_conn():
try:
client = await Client.connect(TEMPORAL_HOST, data_converter=pydantic_data_converter, runtime=runtime, namespace=TEMPORAL_NAMESPACE)
self.logger.info("Cliente conectado a Temporal en %s", TEMPORAL_HOST)
return client
except RPCError as e:
raise("Temporal rechazó la conexión: %s", e)
except ValueError as e:
raise("TEMPORAL_HOST mal configurado: %s", e)
except (ConnectionRefusedError, OSError) as e:
raise("No se pudo alcanzar el host de Temporal: %s", e)
except TimeoutError as e:
raise("Timeout al conectar con Temporal: %s", e)
except Exception:
raise("error inesperado al conectar con el cliente")
async def close(self, client: Client | None):
if client is None:
return
try:
if hasattr(client, "close"):
await client.close()
except Exception as e:
raise(f"error al cerrar cliente temporal: {e}")
#_______________________
# app/shutdown_completo.py
import logging
from data_base.engine_sql import close_engine
from temporal.cliente_temporal_connet import TemporalClientFactory
logger = logging.getLogger(__name__)
async def shutdown(client=None):
try:
await TemporalClientFactory.close(client)
except Exception:
logger.exception("error al cerrar temporal")
try:
await close_engine()
except Exception:
logger.exception("error al cerrar motor de SQL")
#_______________________
# routes/work.py
import logging
from fastapi import APIRouter, HTTPException, Request
from temporalio.common import WorkflowIDConflictPolicy
from temporalio.service import RPCError
from core.config import TASK_QUEUE
from temporal.workflows import MiWorkflow
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/runs", status_code=202)
async def create_run(request: Request):
client = request.app.state.temporal
try:
handle = await client.start_workflow(
MiWorkflow.run,
id="rates-monitor",
task_queue=TASK_QUEUE,
id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
)
except RPCError as e:
logger.exception("Error de comunicación con Temporal")
raise HTTPException(status_code=502, detail="No se pudo contactar con Temporal") from e
except (TypeError, ValueError) as e:
logger.exception("Parámetros inválidos al iniciar el workflow")
raise HTTPException(status_code=500, detail="Parámetros inválidos del workflow") from e
return {"workflow_id": handle.id, "run_id": handle.result_run_id}
#_______________________
# app/worker.py
import asyncio
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
from prometheus_client import start_http_server
from app.shutdown_completo import shutdown
from data_base.sql_create_tables import create_tables
from temporal.register_worker import Registry_worker
logger = logging.getLogger(__name__)
METRICS_PORT = 9465 # métricas propias del sistema (el 9464 es el del SDK de Temporal)
async def main():
# Abre un servidor HTTP en un hilo aparte que publica los Gauge en /metrics
start_http_server(METRICS_PORT)
logger.info("Métricas del worker publicadas en el puerto %d", METRICS_PORT)
registry = Registry_worker()
try:
await create_tables()
await registry.create_worker()
finally:
await shutdown(registry.client)
if __name__ == "__main__":
asyncio.run(main())
#_______________________
#metricas/gauges.py
from prometheus_client import Gauge
cpu = Gauge("cpu_percent", "Uso de CPU (%)")
memory = Gauge("memory_percent", "Uso de memoria (%)")
memory_available = Gauge("memory_available", "Memoria disponible (bytes)")
disco = Gauge("disk_percent", "Uso de disco (%)")
disk_used = Gauge("disk_used", "Disco utilizado (bytes)")
# Marca de tiempo (Unix) de la última lectura: sirve para alertar si el monitor deja de medir
ultima_lectura = Gauge("ultima_lectura_unix", "Timestamp de la última lectura de métricas")
#_______________________
# routes/histoty.py
import math
from datetime import datetime, timedelta, timezone
import httpx
from fastapi import APIRouter, HTTPException, Query
from core.config import PROMETHEUS_URL
router = APIRouter()
METRICAS = ("cpu_percent", "memory_percent", "memory_available", "disk_percent", "disk_used")
_UNIDADES = {"s": 1, "m": 60, "h": 3600, "d": 86400}
_MAX_PUNTOS = 10000 # Prometheus rechaza más de 11.000 por serie
@router.get("/historial")
async def obtener_historial(
minutos: int = Query(default=60, ge=1, le=10080),
paso: str = Query(default="15s", pattern=r"^\d+[smhd]$"),
):
ahora = datetime.now(timezone.utc)
inicio = ahora - timedelta(minutes=minutos)
paso_seg = int(paso[:-1]) * _UNIDADES[paso[-1]]
paso_seg = max(paso_seg, math.ceil(minutos * 60 / _MAX_PUNTOS), 1)
consulta = '{__name__=~"' + "|".join(METRICAS) + '",job="worker-app"}'
params = {
"query": consulta,
"start": inicio.timestamp(),
"end": ahora.timestamp(),
"step": f"{paso_seg}s",
}
try:
async with httpx.AsyncClient(timeout=20) as client:
response = await client.get(f"{PROMETHEUS_URL}/api/v1/query_range", params=params)
response.raise_for_status()
resultado = response.json()
except httpx.HTTPError as exc:
raise HTTPException(status_code=502, detail="No fue posible consultar Prometheus") from exc
if resultado.get("status") != "success":
raise HTTPException(status_code=502, detail="Prometheus no pudo ejecutar la consulta")
return {
"desde": inicio.isoformat(),
"hasta": ahora.isoformat(),
"minutos": minutos,
"paso": f"{paso_seg}s",
"metricas": resultado["data"]["result"],
}
#_______________________
#prometheus/prometheus.yml
global:
scrape_interval: 15s
evaluation_interval: 15s
rule_files:
- /etc/prometheus/alert_rules.yml
alerting:
alertmanagers:
- static_configs:
- targets: ["alertmanager:9093"]
scrape_configs:
- job_name: prometheus
static_configs:
- targets: ["localhost:9090"]
- job_name: api # métricas HTTP de la API (/metrics)
metrics_path: /metrics
static_configs:
- targets: ["api:8000"]
- job_name: worker # métricas internas del SDK de Temporal
static_configs:
- targets: ["worker:9464"]
- job_name: worker-app # tus métricas de CPU/RAM/disco
# scrape_interval: 5s # opcional: la lectura ocurre cada ~6 s
static_configs:
- targets: ["worker:9465"]
- job_name: temporal
metrics_path: /metrics
static_configs:
- targets: ["temporal:8000"]
#_______________________
#prometheus/alert_rules.yml
groups:
- name: workflow-engine
rules:
- alert: ServicioCaido
expr: up == 0
for: 1m
labels:
severity: critical
annotations:
summary: "{{ $labels.job }} no responde"
description: "El target {{ $labels.instance }} lleva más de 1 minuto caído."
- alert: ApiErrores5xx
expr: |
sum(rate(http_requests_total{job="api", status="5xx"}[5m]))
/
sum(rate(http_requests_total{job="api"}[5m])) > 0.05
for: 5m
labels:
severity: warning
annotations:
summary: "Más del 5% de respuestas 5xx en la API"
- alert: ApiLenta
expr: |
histogram_quantile(0.95,
sum(rate(http_request_duration_seconds_bucket{job="api"}[5m])) by (le)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "p95 de la API supera 2 segundos"
- alert: ActivitiesFallando
expr: increase(temporal_activity_execution_failed_total{job="worker"}[10m]) > 10
for: 0m
labels:
severity: warning
annotations:
summary: "Muchas activities fallando en los últimos 10 minutos"
- alert: WorkflowsFallidos
expr: increase(temporal_workflow_failed_total{job="worker"}[10m]) > 0
for: 0m
labels:
severity: critical
annotations:
summary: "Hay workflows que terminaron en FAILED"
- alert: MonitorSinDatos
expr: time() - ultima_lectura_unix{job="worker-app"} > 60
for: 1m
labels:
severity: critical
annotations:
summary: "El worker lleva más de 60 s sin registrar métricas"
#_______________________
# alertmanager/alertmanager.yml
global:
resolve_timeout: 5m
route:
receiver: default
group_by: ["alertname", "job"]
group_wait: 30s
group_interval: 5m
repeat_interval: 4h
routes:
- matchers:
- severity="critical"
receiver: criticas
repeat_interval: 1h
receivers:
- name: default
webhook_configs:
- url: "http://api:8000/alerts"
- name: criticas
webhook_configs:
- url: "http://api:8000/alerts"
inhibit_rules:
- source_matchers:
- severity="critical"
target_matchers:
- severity="warning"
equal: ["alertname", "job"]
#_______________________
# routes/alertas.py
import logging
from fastapi import APIRouter
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/alerts")
async def recibir_alertas(payload: dict):
for alerta in payload.get("alerts", []):
nombre = alerta.get("labels", {}).get("alertname", "desconocida")
severidad = alerta.get("labels", {}).get("severity", "?")
resumen = alerta.get("annotations", {}).get("summary", "")
logger.warning("[%s] %s (%s): %s", alerta.get("status"), nombre, severidad, resumen)
return {"ok": True}
#_______________________
# temporal/register_worker.py
import logging
from temporalio.service import RPCError
from temporalio.worker import Worker
from activitys.activity_comparations import comparation
from activitys.activity_gemini import analisis_gemini
from activitys.activity_psutil import _psutil
from activitys.guardar_analisis_sql import upsert_analisis
from activitys.guardar_metricas_sql import upsert_metrics_data
from core.config import TASK_QUEUE, WORKFLOW_CONCURRENCY
from temporal.cliente_temporal_connet import TemporalClientFactory
from temporal.workflows import MiWorkflow
class Registry_worker:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.client = None
async def create_worker(self):
try:
self.client = await TemporalClientFactory.client_conn(with_metrics=True)
worker = Worker(
self.client,
task_queue=TASK_QUEUE,
workflows=[MiWorkflow],
activities=[
_psutil,
comparation,
upsert_metrics_data,
analisis_gemini,
upsert_analisis,
],
max_concurrent_activities=WORKFLOW_CONCURRENCY,
)
self.logger.info("Worker escuchando la cola %s", TASK_QUEUE)
await worker.run()
except ValueError as e:
self.logger.error(
"Configuración inválida del worker (¿falta @activity.defn o @workflow.defn?): %s", e
)
raise
except RPCError as e:
self.logger.error("Worker perdió conexión con Temporal: %s", e)
raise
except Exception:
self.logger.exception("error inesperado en el worker")
raise
#_______________________
# temporal/workflows.py
from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy
from temporalio.exceptions import ActivityError
ITERACIONES_POR_EJECUCION = 500 # ~7.500 eventos, muy por debajo del límite
INTERVALO = timedelta(seconds=5)
ENFRIAMIENTO_GEMINI = timedelta(minutes=5)
RETRY = RetryPolicy(maximum_attempts=3, non_retryable_error_types=["ValueError"])
@workflow.defn
class MiWorkflow:
async def _ejecutar(self, nombre: str, arg=None, timeout=timedelta(minutes=1)):
"""Ejecuta una activity; si falla, lo registra y devuelve None (no mata el monitor)."""
try:
if arg is None:
return await workflow.execute_activity(
nombre, start_to_close_timeout=timeout, retry_policy=RETRY
)
return await workflow.execute_activity(
nombre, arg, start_to_close_timeout=timeout, retry_policy=RETRY
)
except ActivityError:
workflow.logger.exception("Falló la activity %s", nombre)
return None
@workflow.run
async def run(self) -> None:
ultimo_analisis = None
for _ in range(ITERACIONES_POR_EJECUCION):
metrics_system = await self._ejecutar("_psutil", timeout=timedelta(seconds=30))
if metrics_system is None:
await workflow.sleep(INTERVALO)
continue
comparacion = await self._ejecutar("comparation", metrics_system)
await self._ejecutar("upsert_metrics_data", {"metrics": metrics_system})
if comparacion and comparacion["alerta"]:
ahora = workflow.now()
if ultimo_analisis is None or ahora - ultimo_analisis >= ENFRIAMIENTO_GEMINI:
analysis = await self._ejecutar(
"analisis_gemini",
{"metrics": metrics_system, "alertas": comparacion["alertas"]},
timeout=timedelta(minutes=2),
)
ultimo_analisis = ahora
if analysis is not None:
await self._ejecutar(
"upsert_analisis",
{"metrics": metrics_system, "analisis_alerta": analysis},
)
await workflow.sleep(INTERVALO)
# Reinicia el historial y sigue monitoreando
workflow.continue_as_new()
#_______________________
# activitys/activity_psutil.py
import asyncio
import time
import psutil
from temporalio import activity
from metricas.gauges import (
cpu,
disco,
disk_used,
memory,
memory_available,
ultima_lectura,
)
def _leer_metricas() -> dict:
cpu_pct = psutil.cpu_percent(interval=1)
memoria = psutil.virtual_memory()
disco_uso = psutil.disk_usage("/")
return {
"cpu_percent": round(cpu_pct),
"memory_percent": round(memoria.percent),
"memory_available": int(memoria.available),
"disk_percent": round(disco_uso.percent),
"disk_used": int(disco_uso.used),
}
@activity.defn
async def _psutil() -> dict:
# La lectura bloquea ~1 s: se ejecuta en un hilo para no frenar el event loop
datos = await asyncio.to_thread(_leer_metricas)
cpu.set(datos["cpu_percent"])
memory.set(datos["memory_percent"])
memory_available.set(datos["memory_available"])
disco.set(datos["disk_percent"])
disk_used.set(datos["disk_used"])
ultima_lectura.set(time.time())
return datos
#_______________________
#activitys/activity_comparations.py
from temporalio import activity
from core.config import MAX_PERCENT
@activity.defn
async def comparation(data_system: dict) -> dict:
recursos = ("cpu_percent", "memory_percent", "disk_percent")
alertas = [
f"{r}={data_system[r]}%" for r in recursos if data_system[r] >= MAX_PERCENT
]
return {"alerta": bool(alertas), "alertas": alertas}
#_______________________
# activitys/guardar_metricas_sql.py
import logging
from temporalio import activity
from data_base.sql_create_tables import insert_metrics_data
logger = logging.getLogger(__name__)
@activity.defn
async def upsert_metrics_data(data: dict):
metrics = data.get("metrics")
if not metrics:
raise ValueError("métricas vacías")
await insert_metrics_data(metrics)
logger.info("metricas insertadas correctamente")
#_______________________
# activitys/activity_gemini.py
import json
import logging
from google import genai
from google.genai import errors, types
from temporalio import activity
from core.config import GEMINI_API_KEY, GEMINI_MODEL
logger = logging.getLogger(__name__)
client = genai.Client(api_key=GEMINI_API_KEY)
PROMPT = """
Actúa como un ingeniero especializado en monitorización de sistemas, análisis de rendimiento y Python.
Tu tarea es analizar las métricas de rendimiento de una máquina que recibirás mediante un diccionario de Python en formato JSON.
Las métricas pueden incluir:
* `cpu_percent`: porcentaje de utilización de CPU.
* `memory_percent`: porcentaje de utilización de RAM.
* `memory_available`: memoria RAM disponible en bytes.
* `disk_percent`: porcentaje de espacio ocupado en disco.
* `disk_used`: espacio utilizado en disco en bytes.
**Reglas de análisis:**
1. Evalúa individualmente CPU, RAM y disco.
2. Considera que un recurso presenta un nivel de alerta cuando su porcentaje es igual o superior al 90 %.
3. Clasifica cada recurso como:
* `normal`: menos del 70 %.
* `advertencia`: desde el 70 % hasta menos del 90 %.
* `critico`: igual o superior al 90 %.
4. Identifica qué recurso presenta mayor utilización.
5. Explica brevemente los posibles efectos de los valores observados sobre el rendimiento del sistema.
6. No declares que la máquina está saturada basándote únicamente en un promedio general. Analiza cada recurso por separado.
7. Si faltan métricas, hay valores inválidos o los datos son insuficientes, indícalo sin inventar valores.
8. No inventes información sobre procesos, temperatura, carga de CPU, swap o latencia de disco si no se proporcionan esos datos.
9. Devuelve exclusivamente un objeto JSON válido, sin Markdown ni texto fuera del JSON.
**Formato de respuesta obligatorio:**
{
"estado_general": "normal | advertencia | critico",
"recurso_mas_utilizado": "cpu | ram | disco | desconocido",
"metricas": {
"cpu": {
"porcentaje": 0,
"estado": "normal | advertencia | critico",
"analisis": ""
},
"ram": {
"porcentaje": 0,
"estado": "normal | advertencia | critico",
"analisis": ""
},
"disco": {
"porcentaje": 0,
"estado": "normal | advertencia | critico",
"analisis": ""
}
},
"recomendacion": "",
"datos_insuficientes": false
}
El campo `estado_general` debe reflejar el estado más grave entre los recursos evaluados. Si alguno está en estado crítico, el estado general será crítico. Si ninguno está crítico, pero alguno está en advertencia, será advertencia. De lo contrario, será normal.
Analiza las siguientes métricas:
"""
@activity.defn
async def analisis_gemini(data: dict) -> dict:
data_metric = data.get("metrics")
if not data_metric:
raise ValueError("datos de metricas vacios")
contents = PROMPT + "\n" + json.dumps(data_metric, ensure_ascii=False)
try:
response = await client.aio.models.generate_content(
model=GEMINI_MODEL,
contents=contents,
config=types.GenerateContentConfig(response_mime_type="application/json"),
)
except errors.APIError as e:
raise RuntimeError(f"Error de Gemini: {e}") from e
if not response.text:
raise RuntimeError("Gemini no devolvió texto")
try:
respuesta = json.loads(response.text)
except json.JSONDecodeError as e:
# RuntimeError (reintentable): un JSON malformado del modelo suele ser puntual
raise RuntimeError("Gemini devolvió un JSON inválido") from e
if not isinstance(respuesta, dict):
raise RuntimeError("Gemini devolvió un formato inválido")
return respuesta
#_______________________
# activitys/guardar_analisis_sql.py
import logging
from temporalio import activity
from data_base.sql_create_tables import insert_metrics_analisis
logger = logging.getLogger(__name__)
@activity.defn
async def upsert_analisis(data: dict):
if "metrics" not in data or "analisis_alerta" not in data:
raise ValueError("faltan 'metrics' o 'analisis_alerta'")
await insert_metrics_analisis(
{"metricas": data["metrics"], "analisis": data["analisis_alerta"]}
)
logger.info("metricas de análisis insertadas correctamente")
#_______________________
# data_base/sql_create_tables.py
import logging
from datetime import datetime
from sqlalchemy import (
BigInteger,
Boolean,
DateTime,
Float,
ForeignKey,
Index,
Integer,
JSON,
String,
UniqueConstraint,
func,
insert,
text,
)
from sqlalchemy.exc import CompileError, IntegrityError, SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from data_base.engine_sql import creation_engine
logger = logging.getLogger(__name__)
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
password_hash: Mapped[str] = mapped_column(String(255))
rol: Mapped[str] = mapped_column(String(20), default="user", server_default="user")
activo: Mapped[bool] = mapped_column(Boolean, default=True, server_default=text("true"))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
class Host(Base):
__tablename__ = "hosts"
__table_args__ = (UniqueConstraint("owner_id", "name", name="uq_hosts_owner_name"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
owner_id: Mapped[int] = mapped_column(
Integer, ForeignKey("users.id", ondelete="CASCADE"), index=True
)
name: Mapped[str] = mapped_column(String(100))
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
class Table_data_metrics(Base):
"""Tabla PARTICIONADA por día (created_at). La crea la migración con SQL manual."""
__tablename__ = "metrics"
__table_args__ = (
Index("ix_metrics_host_created", "host_id", "created_at"),
{"postgresql_partition_by": "RANGE (created_at)"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), primary_key=True, server_default=func.now()
)
host_id: Mapped[int] = mapped_column(Integer, ForeignKey("hosts.id", ondelete="CASCADE"))
cpu_percent: Mapped[int] = mapped_column(Integer, default=0)
memory_percent: Mapped[int] = mapped_column(Integer, default=0)
memory_available: Mapped[int] = mapped_column(BigInteger, default=0)
disk_percent: Mapped[int] = mapped_column(Integer, default=0)
disk_used: Mapped[int] = mapped_column(BigInteger, default=0)
class Metrics1m(Base):
"""Resumen por minuto (promedio y máximo) para conservar historia a largo plazo."""
__tablename__ = "metrics_1m"
host_id: Mapped[int] = mapped_column(
Integer, ForeignKey("hosts.id", ondelete="CASCADE"), primary_key=True
)
minuto: Mapped[datetime] = mapped_column(DateTime(timezone=True), primary_key=True)
muestras: Mapped[int] = mapped_column(Integer)
cpu_avg: Mapped[float] = mapped_column(Float)
cpu_max: Mapped[int] = mapped_column(Integer)
memory_avg: Mapped[float] = mapped_column(Float)
memory_max: Mapped[int] = mapped_column(Integer)
disk_avg: Mapped[float] = mapped_column(Float)
disk_max: Mapped[int] = mapped_column(Integer)
memory_available_min: Mapped[int] = mapped_column(BigInteger)
disk_used_max: Mapped[int] = mapped_column(BigInteger)
class Table_data_analisis(Base):
__tablename__ = "metrics_analisis"
__table_args__ = (Index("ix_analisis_host_created", "host_id", "created_at"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
host_id: Mapped[int] = mapped_column(Integer, ForeignKey("hosts.id", ondelete="CASCADE"))
metricas: Mapped[dict] = mapped_column(JSON)
analisis: Mapped[dict] = mapped_column(JSON)
class Engine_tables:
def __init__(self):
self.Engine = creation_engine.engine()
self._session_factory = async_sessionmaker(self.Engine, expire_on_commit=False)
def get_session(self) -> AsyncSession:
"""Obtiene una sesión asíncrona."""
return self._session_factory()
async def _insertar(self, modelo, data: dict, nombre: str):
# Insert "Core" (no ORM): en una tabla particionada con PK (id, created_at)
# evita que el ORM tenga que recuperar los valores generados por la base.
try:
async with self.get_session() as session:
async with session.begin():
await session.execute(insert(modelo).values(**data))
except (CompileError, IntegrityError) as e:
# claves que no son columnas, o host_id inexistente: no sirve reintentar
raise ValueError(f"Datos inválidos para {nombre}: {e.__class__.__name__}") from e
except SQLAlchemyError as e:
raise RuntimeError(f"Error al insertar en {nombre}") from e
async def insert_metrics_data(self, data: dict):
await self._insertar(Table_data_metrics, data, "metrics")
async def insert_metrics_analisis(self, data: dict):
await self._insertar(Table_data_analisis, data, "metrics_analisis")
_tablas = Engine_tables()
insert_metrics_data = _tablas.insert_metrics_data
insert_metrics_analisis = _tablas.insert_metrics_analisis
async def get_db():
"""Dependencia de FastAPI: una sesión por petición."""
async with _tablas.get_session() as session:
yield session
#_______________________
#data_base/engine_sql.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from core.config import DB_URL, SQL_ECHO, MAX_WORKERS, MAX_OVERFLOWS
import asyncio
import logging
class Creation_engine:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.ENGINE = create_async_engine(
DB_URL,
echo=SQL_ECHO,
pool_size=MAX_WORKERS,
max_overflow=MAX_OVERFLOWS,
pool_pre_ping=True
)
def engine(self):
return self.ENGINE
async def close_engine(self):
try:
await self.ENGINE.dispose()
except asyncio.CancelledError:
self.logger.warning("El cierre del motor fue cancelado.")
raise
except Exception:
self.logger.exception("Error al cerrar el motor de la base de datos.")
raise
# Instancia única compartida por todo el proceso
creation_engine = Creation_engine()
async def close_engine():
await creation_engine.close_engine()
#URL = "postgresql+asyncpg://usuario:password@localhost:5432/mi_base_datos"
#postgresql+psycopg = Motor de base de datos y controlador de Python.
#usuario:password = Credenciales de acceso.
#localhost:5432 = Dirección del servidor y puerto de PostgreSQL.
#mi_base_datos = Nombre de la base de datos a la que accederás.
#_______________________
# Dockerfile
FROM python:3.12-slim
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=/srv
WORKDIR /srv
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN useradd -m appuser
USER appuser
#_______________________
#docker_compose
services:
db:
image: postgres:16
restart: unless-stopped
environment:
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: ${POSTGRES_DB}
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
interval: 5s
timeout: 5s
retries: 10
temporal:
image: temporalio/auto-setup:1.25.2
restart: unless-stopped
depends_on:
db:
condition: service_healthy
environment:
DB: postgres12
DB_PORT: 5432
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PWD: ${POSTGRES_PASSWORD}
POSTGRES_SEEDS: db
PROMETHEUS_ENDPOINT: 0.0.0.0:8000 # métricas del servidor Temporal
ports:
- "127.0.0.1:7233:7233"
# RIESGO: no verifiqué que `tctl` exista en esta imagen. Si el servicio
# nunca pasa a "healthy", revisa este comando con `docker compose logs temporal`.
healthcheck:
test: ["CMD-SHELL", "tctl --address temporal:7233 cluster health | grep -q SERVING"]
interval: 10s
timeout: 5s
retries: 15
temporal-ui:
image: temporalio/ui:2.31.2
restart: unless-stopped
depends_on:
- temporal
environment:
TEMPORAL_ADDRESS: temporal:7233
ports:
- "127.0.0.1:8080:8080"
api:
build: .
command: uvicorn app.life_span:app --host 0.0.0.0 --port 8000
restart: unless-stopped
env_file: .env
depends_on:
db:
condition: service_healthy
temporal:
condition: service_healthy
ports:
- "127.0.0.1:8000:8000"
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/metrics', timeout=3)"]
interval: 10s
timeout: 5s
retries: 10
start_period: 15s
worker:
build: .
command: python -m app.worker
restart: unless-stopped
env_file: .env
depends_on:
db:
condition: service_healthy
temporal:
condition: service_healthy
api:
condition: service_healthy # la API ya creó las tablas
expose:
- "9464" # métricas del SDK de Temporal (red interna)
- "9465" # solo si adoptas la pregunta 38 (métricas propias del worker)
prometheus:
image: prom/prometheus:v2.55.1
restart: unless-stopped
command:
- --config.file=/etc/prometheus/prometheus.yml
- --storage.tsdb.retention.time=7d
volumes:
- ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
- ./prometheus/alert_rules.yml:/etc/prometheus/alert_rules.yml:ro
- promdata:/prometheus
ports:
- "127.0.0.1:9090:9090"
depends_on:
- alertmanager
alertmanager:
image: prom/alertmanager:v0.27.0
restart: unless-stopped
command:
- --config.file=/etc/alertmanager/alertmanager.yml
volumes:
- ./alertmanager/alertmanager.yml:/etc/alertmanager/alertmanager.yml:ro
# - ./alertmanager/smtp_password:/etc/alertmanager/smtp_password:ro # si usas correo
- alertdata:/alertmanager
extra_hosts:
- "host.docker.internal:host-gateway" # necesario en Linux para host.docker.internal
ports:
- "127.0.0.1:9093:9093"
volumes:
pgdata:
promdata:
alertdata:
#_______________________
#_______________________
#core/config.py
from pathlib import Path
import os
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parent.parent / ".env") # raíz del repo, no core/
def get_env(name: str, default: str | None = None) -> str:
value = os.getenv(name, default)
if value is None or value == "":
raise RuntimeError(f"Falta la variable de entorno requerida: {name}")
return value
SQL_ECHO = get_env("SQL_ECHO", "False").lower() == "true"
MAX_WORKERS = int(get_env("MAX_WORKERS"))
MAX_OVERFLOWS = int(get_env("MAX_OVERFLOWS"))
TASK_QUEUE = get_env("TASK_QUEUE")
WORKFLOW_CONCURRENCY = int(get_env("WORKFLOW_CONCURRENCY"))
TEMPORAL_HOST = get_env("TEMPORAL_HOST")
TEMPORAL_NAMESPACE = get_env("TEMPORAL_NAMESPACE")
MAX_PERCENT = int(get_env("MAX_PERCENT"))
GEMINI_API_KEY = get_env("GEMINI_API_KEY")
GEMINI_MODEL = get_env("GEMINI_MODEL", "gemini-2.5-flash") # verifica que siga disponible
DB_URL = f"postgresql+asyncpg://{POSTGRES_USER}:{POSTGRES_PASSWORD}@db:5432/proyect_framework"
# añade al final de core/config.py
JWT_SECRET = get_env("JWT_SECRET")
JWT_ALGORITHM = "HS256"
JWT_EXPIRE_MINUTES = int(get_env("JWT_EXPIRE_MINUTES", "60"))
HOST_ID = int(get_env("HOST_ID")) # id del host (tabla hosts) que mide este worker
RAW_RETENTION_DAYS = int(get_env("RAW_RETENTION_DAYS", "7"))
AVG_RETENTION_DAYS = int(get_env("AVG_RETENTION_DAYS", "365"))
PARTITIONS_AHEAD_DAYS = int(get_env("PARTITIONS_AHEAD_DAYS", "7"))
ROLLUP_LOOKBACK_HOURS = int(get_env("ROLLUP_LOOKBACK_HOURS", "24"))
#_______________________
#.env
TEMPORAL_HOST=temporal:7233
TEMPORAL_NAMESPACE="default"
TASK_QUEUE="framework_2"
POSTGRES_USER=app_user2#cambiar por el usuario real
POSTGRES_PASSWORD=<NUEVA_CONTRASEÑA_LARGA> #cambiar por real
POSTGRES_DB=proyect_framework #cambira por real
SQL_ECHO=False
MAX_WORKERS=10
WORKFLOW_CONCURRENCY=5
MAX_OVERFLOWS=20
MAX_PERCENT=90
GEMINI_API_KEY=#colocar el real
# añade a .env
JWT_SECRET=<genera con: python -c "import secrets; print(secrets.token_urlsafe(48))">
JWT_EXPIRE_MINUTES=60
HOST_ID=1
RAW_RETENTION_DAYS=7
AVG_RETENTION_DAYS=365
PARTITIONS_AHEAD_DAYS=7
ROLLUP_LOOKBACK_HOURS=24
#_______________________
#.dockerignore
.env
.git
__pycache__
#_______________________
# requirements.txt (luego fija versiones con: pip freeze)
fastapi
uvicorn[standard]
temporalio
pydantic>=2
sqlalchemy[asyncio]>=2
asyncpg
psutil
httpx
prometheus-client
prometheus-fastapi-instrumentator
python-dotenv
google-genai
argon2-cffi
pyjwt
email-validator
alembic
#_______________________
# tests/conftest.py (config.py exige estas variables al importar)
import os
for k, v in {
"DB_URL": "postgresql+asyncpg://u:p@localhost:5432/test",
"MAX_WORKERS": "5", "MAX_OVERFLOWS": "5", "TASK_QUEUE": "test",
"WORKFLOW_CONCURRENCY": "2", "TEMPORAL_HOST": "localhost:7233",
"TEMPORAL_NAMESPACE": "default", "MAX_PERCENT": "90",
"GEMINI_API_KEY": "x", "API_KEY": "x",
}.items():
os.environ.setdefault(k, v)
#_______________________
# tests/test_comparation.py
from activitys.activity_comparations import comparation
async def test_sin_alerta():
r = await comparation({"cpu_percent": 10, "memory_percent": 20, "disk_percent": 30})
assert r == {"alerta": False, "alertas": []}
async def test_con_alerta():
r = await comparation({"cpu_percent": 95, "memory_percent": 20, "disk_percent": 30})
assert r["alerta"] is True
assert r["alertas"] == ["cpu_percent=95%"]
#_______________________
# tests/test_rate_params.py
import pytest
from pydantic import ValidationError
from metricas.metrica_entrada import RateParams
def test_valido():
RateParams(cpu_percent=10, memory_percent=20, memory_available=1,
disk_percent=30, disk_used=1)
def test_fuera_de_rango():
with pytest.raises(ValidationError):
RateParams(cpu_percent=150, memory_percent=20, memory_available=1,
disk_percent=30, disk_used=1)
#_______________________
# tests/test_workflow.py
from temporalio import activity
from temporalio.client import WorkflowExecutionStatus
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from temporal.workflows import MiWorkflow
DATOS = {"cpu_percent": 99, "memory_percent": 10, "memory_available": 1,
"disk_percent": 10, "disk_used": 1}
async def test_workflow_sobrevive_si_gemini_falla():
llamadas = {"psutil": 0}
@activity.defn(name="_psutil")
async def psutil_mock() -> dict:
llamadas["psutil"] += 1
return DATOS
@activity.defn(name="comparation")
async def comparation_mock(d: dict) -> dict:
return {"alerta": True, "alertas": ["cpu_percent=99%"]}
@activity.defn(name="upsert_metrics_data")
async def upsert_m(d: dict): ...
@activity.defn(name="publicar_metricas")
async def publicar(d: dict): ...
@activity.defn(name="analisis_gemini")
async def gemini_falla(d: dict):
raise RuntimeError("Gemini caído")
@activity.defn(name="upsert_analisis")
async def upsert_a(d: dict): ...
async with await WorkflowEnvironment.start_time_skipping() as env:
async with Worker(env.client, task_queue="t", workflows=[MiWorkflow],
activities=[psutil_mock, comparation_mock, upsert_m,
publicar, gemini_falla, upsert_a]):
handle = await env.client.start_workflow(
MiWorkflow.run, id="wf-test", task_queue="t")
await env.sleep(60) # el tiempo avanza "saltando" las esperas
assert llamadas["psutil"] >= 3
assert (await handle.describe()).status == WorkflowExecutionStatus.RUNNING
await handle.terminate()
#_______________________
touch app/__init__.py core/__init__.py data_base/__init__.py routes/__init__.py \
temporal/__init__.py activities/__init__.py metricas/__init__.py
To embed this project on your website, copy the following code and paste it into your website's HTML: