egutierrez/suite_logs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Nombre del servicio directo en el logger

Browse Source
This commit is contained in:
egutierrez
2025-10-31 15:32:33 +01:00
parent 281ec46fc7
commit 64e1c0e970
1 changed files with 38 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Logger/LokiLogger.py
+38 -44
View File
@@ -3,11 +3,13 @@ import json
import requests
from typing import Optional, Dict, Any
class LokiLogger:
"""
Logger compatible con Grafana Loki / Alloy.
Envía logs en formato JSON a través del endpoint HTTP de Loki.
"""
ALLOWED_LEVELS = (
"TRACE",
"DEBUG",
@@ -24,25 +26,29 @@ class LokiLogger:
endpoint: str = "http://127.0.0.1:3101/loki/api/v1/push",
default_labels: Optional[Dict[str, str]] = None,
timeout: float = 5.0,
min_level: str = "DEBUG"
min_level: str = "DEBUG",
service_name: Optional[str] = None,
):
"""
:param endpoint: URL completa del endpoint de Loki / Alloy.
:param default_labels: etiquetas estáticas por defecto para todos los logs,
ej: {"job": "my-service", "env": "prod"}
:param default_labels: etiquetas estáticas comunes (ej: {"env": "dev"})
:param timeout: timeout en segundos para la petición HTTP
:param min_level: nivel mínimo para enviar logs. Ej: "INFO" solo enviará INFO o más graves.
:param min_level: nivel mínimo para enviar logs
:param service_name: nombre del servicio (se usará también como 'job')
"""
self.endpoint = endpoint
self.default_labels = default_labels or {}
self.timeout = timeout
self.service_name = service_name or "unknown-service"
min_level = min_level.upper()
if min_level not in self.ALLOWED_LEVELS:
raise ValueError(f"min_level debe estar en {self.ALLOWED_LEVELS}")
self.min_level = min_level
# Asigna prioridad (menor valor = más detallado)
# Base labels: incluye el job automáticamente
self.default_labels = dict(default_labels or {})
self.default_labels["job"] = self.service_name
self._level_order = {
"TRACE": 0,
"DEBUG": 1,
@@ -66,15 +72,11 @@ class LokiLogger:
self,
level: str,
message: str,
service: str,
labels: Optional[Dict[str, str]] = None,
metadata: Optional[Dict[str, Any]] = None
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""
Envía un log a Loki con los campos mínimos + metadata opcional.
"""
"""Envía un log a Loki con los campos mínimos + metadata opcional."""
level = level.upper()
# Soporte retroactivo
if level == "WARNING":
level = "WARN"
@@ -82,70 +84,62 @@ class LokiLogger:
raise ValueError(f"Nivel no válido: {level}. Debe estar en {self.ALLOWED_LEVELS}")
if not self._should_log(level):
return # No enviar porque está por debajo del umbral
return
# Combinar etiquetas
final_labels = dict(self.default_labels)
if labels:
final_labels.update(labels)
# Payload base
payload_metadata = {
"service": service,
"level": level
"service": self.service_name,
"level": level,
}
if metadata:
payload_metadata.update(metadata)
# Construir línea JSON
log_line = json.dumps({
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"message": message,
**payload_metadata
**payload_metadata,
})
body = {
"streams": [
{
"stream": final_labels,
"values": [
[self._current_ns(), log_line]
]
}
{"stream": final_labels, "values": [[self._current_ns(), log_line]]}
]
}
# Enviar al endpoint Loki
try:
resp = requests.post(self.endpoint, json=body, timeout=self.timeout)
resp.raise_for_status()
except Exception as e:
print(f"Failed to send log to Loki: {e}", flush=True)
# Métodos por nivel estándar
def trace(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("TRACE", message, service, labels, metadata)
# Métodos estándar por nivel
def trace(self, message, labels=None, metadata=None):
self.log("TRACE", message, labels, metadata)
def debug(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("DEBUG", message, service, labels, metadata)
def debug(self, message, labels=None, metadata=None):
self.log("DEBUG", message, labels, metadata)
def info(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("INFO", message, service, labels, metadata)
def info(self, message, labels=None, metadata=None):
self.log("INFO", message, labels, metadata)
def warn(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("WARN", message, service, labels, metadata)
def warn(self, message, labels=None, metadata=None):
self.log("WARN", message, labels, metadata)
def warning(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("WARN", message, service, labels, metadata)
def warning(self, message, labels=None, metadata=None):
self.log("WARN", message, labels, metadata)
def error(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("ERROR", message, service, labels, metadata)
def error(self, message, labels=None, metadata=None):
self.log("ERROR", message, labels, metadata)
def fatal(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("FATAL", message, service, labels, metadata)
def fatal(self, message, labels=None, metadata=None):
self.log("FATAL", message, labels, metadata)
def critical(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("CRITICAL", message, service, labels, metadata)
def critical(self, message, labels=None, metadata=None):
self.log("CRITICAL", message, labels, metadata)
def unknown(self, message: str, service: str, labels: Optional[Dict[str, str]] = None, metadata: Optional[Dict[str, Any]] = None):
self.log("UNKNOWN", message, service, labels, metadata)
def unknown(self, message, labels=None, metadata=None):
self.log("UNKNOWN", message, labels, metadata)