49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""
|
|
Centralized Cloud Logging setup.
|
|
Uses CloudLoggingHandler (background thread) so logging does not add latency
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional, Dict, Literal
|
|
|
|
import google.cloud.logging
|
|
from google.cloud.logging.handlers import CloudLoggingHandler
|
|
|
|
from .config import cfg
|
|
|
|
|
|
def _setup_logger() -> logging.Logger:
|
|
"""Create or return the singleton evaluation logger."""
|
|
logger = logging.getLogger(cfg.log_name)
|
|
if any(isinstance(h, CloudLoggingHandler) for h in logger.handlers):
|
|
return logger
|
|
|
|
try:
|
|
client = google.cloud.logging.Client(project=cfg.project_id)
|
|
handler = CloudLoggingHandler(client, name=cfg.log_name) # async transport
|
|
logger.addHandler(handler)
|
|
logger.setLevel(getattr(logging, cfg.log_level.upper()))
|
|
except Exception as e:
|
|
# Fallback to console if Cloud Logging is unavailable (local dev)
|
|
logging.basicConfig(level=getattr(logging, cfg.log_level.upper()))
|
|
logger = logging.getLogger(cfg.log_name)
|
|
logger.warning("Cloud Logging setup failed; using console. Error: %s", e)
|
|
|
|
return logger
|
|
|
|
|
|
_eval_log = _setup_logger()
|
|
|
|
|
|
def log_structured_entry(message: str, severity: Literal["INFO", "WARNING", "ERROR"], custom_log: Optional[Dict] = None) -> None:
|
|
"""
|
|
Emit a JSON-structured log row.
|
|
|
|
Args:
|
|
message: Short label for the row (e.g., "Final agent turn").
|
|
severity: "INFO" | "WARNING" | "ERROR"
|
|
custom_log: A dict with your structured payload.
|
|
"""
|
|
level = getattr(logging, severity.upper(), logging.INFO)
|
|
_eval_log.log(level, message, extra={"json_fields": {"message": message, "custom": custom_log or {}}})
|