67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
|
|
from Logger.LokiLogger import LokiLogger
|
|
from Logger import PrometheusMetric
|
|
import time
|
|
|
|
|
|
def prueba_log():
|
|
logger = LokiLogger(
|
|
service_name="suite_logs_example_3",
|
|
# add_labels={
|
|
# "job": "log_testing",
|
|
# "env": "development",
|
|
# "component": "main"
|
|
# },
|
|
min_level="DEBUG"
|
|
)
|
|
|
|
try:
|
|
logger.info("Iniciando pruebas del sistema de logging")
|
|
|
|
logger.trace("Inicio del proceso de ETL")
|
|
logger.debug("Carga de datos completada")
|
|
logger.info("Pipeline ejecutado correctamente")
|
|
logger.warn("Latencia superior a lo esperado")
|
|
logger.error("Error al conectar con base de datos")
|
|
logger.fatal("Fallo crítico en nodo principal")
|
|
logger.critical("Memoria insuficiente para procesamiento")
|
|
logger.unknown("Log sin nivel detectado")
|
|
|
|
logger.info("Pruebas de logging completadas exitosamente")
|
|
|
|
except Exception as e:
|
|
logger.exception(e)
|
|
|
|
|
|
def prueba_metricas(duracion_segundos: int = 180):
|
|
"""
|
|
Emite métricas de prueba durante un tiempo para permitir que Alloy las scrappee.
|
|
"""
|
|
metrics = PrometheusMetric(prefix="pruebas", port=9102)
|
|
|
|
requests_total = metrics.counter(
|
|
"requests_total", "Solicitudes de ejemplo", labels=["endpoint"]
|
|
)
|
|
latency_seconds = metrics.histogram(
|
|
"latency_seconds",
|
|
"Latencia simulada",
|
|
labels=["endpoint"],
|
|
buckets=[0.1, 0.5, 1, 2, 5],
|
|
)
|
|
workers_active = metrics.gauge("workers_active", "Workers activos", labels=["queue"])
|
|
|
|
start = time.time()
|
|
i = 0
|
|
while time.time() - start < duracion_segundos:
|
|
requests_total.inc(endpoint="/demo")
|
|
latency_seconds.observe(0.2 + (i % 5) * 0.1, endpoint="/demo")
|
|
workers_active.set(3 + (i % 4), queue="default")
|
|
i += 1
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
prueba_log()
|
|
# Mantiene el servidor de métricas vivo al menos 3 min para permitir scrapes
|
|
prueba_metricas()
|