r/pallas
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

feat: add /healthz and /metrics endpoints, replace print with logging

Browse Source 
- Add /healthz endpoint returning LLM provider validation status
- Add /metrics endpoint serving Prometheus metrics via prometheus_client
- Replace all print() calls in health.py with proper logging module
- Remove _PREFIX variable in favor of structured logger context
This commit is contained in:
Robert Helewka
2026-04-10 11:22:26 +00:00
parent 9092afb532
commit 0cea5ece3a
10 changed files with 1433 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pallas/health.py
View File

@@ -7,6 +7,7 @@ Validates LLM provider API keys and model availability at startup.
import asyncio
import json
import logging
import os
import re
from datetime import datetime, timezone
@@ -15,6 +16,8 @@ from pathlib import Path
import httpx
import yaml
logger = logging.getLogger(__name__)
def _config_root() -> Path:
"""Return the working directory where agents.yaml and fastagent configs live."""
@@ -31,7 +34,6 @@ def _load_deployment_name() -> str:
_DEPLOY_NAME = _load_deployment_name()
_PREFIX = f"[{_DEPLOY_NAME}]"
# ── Provider API endpoints ───────────────────────────────────────────────────
@@ -170,22 +172,22 @@ async def validate_llm_providers(timeout: float = 5.0) -> dict[str, dict]:
err = await _check_anthropic(client, anthropic_key, model_id)
if err:
results["anthropic"] = {"status": "error", "model": model_id, "message": err}
print(f"{_PREFIX} WARNING: anthropic: {err}")
logger.warning("anthropic: %s", err)
else:
results["anthropic"] = {"status": "ok", "model": model_id}
print(f"{_PREFIX} anthropic: {model_id} ✓")
logger.info("anthropic: %s ready", model_id)
else:
# Key is set but Anthropic isn't the active provider — just verify API access
err = await _check_anthropic(client, anthropic_key, "claude-sonnet-4-5")
if err and "not found" not in err:
results["anthropic"] = {"status": "error", "message": err}
print(f"{_PREFIX} WARNING: anthropic: {err}")
logger.warning("anthropic: %s", err)
else:
results["anthropic"] = {"status": "ok"}
print(f"{_PREFIX} anthropic: API key valid ✓")
logger.info("anthropic: API key valid")
elif active_provider == "anthropic":
results["anthropic"] = {"status": "error", "message": "API key not configured"}
print(f"{_PREFIX} WARNING: anthropic: API key not configured")
logger.warning("anthropic: API key not configured")
# ── OpenAI ───────────────────────────────────────────────────────
if openai_key:
@@ -193,22 +195,22 @@ async def validate_llm_providers(timeout: float = 5.0) -> dict[str, dict]:
err, models = await _list_openai_models(client, openai_key, openai_base)
if err:
results["openai"] = {"status": "error", "message": err}
print(f"{_PREFIX} WARNING: openai ({openai_base}): {err}")
logger.warning("openai (%s): %s", openai_base, err)
elif model_id:
if model_id in models:
results["openai"] = {"status": "ok", "model": model_id}
print(f"{_PREFIX} openai ({openai_base}): {model_id} ✓")
logger.info("openai (%s): %s ready", openai_base, model_id)
else:
label = ", ".join(models) if models else "none"
results["openai"] = {"status": "error", "model": model_id, "message": f"model '{model_id}' not found (available: {label})"}
print(f"{_PREFIX} WARNING: openai ({openai_base}): model '{model_id}' not found (available: {label})")
logger.warning("openai (%s): model '%s' not found (available: %s)", openai_base, model_id, label)
else:
results["openai"] = {"status": "ok", "models": models}
label = ", ".join(models) if models else "no models loaded"
print(f"{_PREFIX} openai ({openai_base}): {label} ✓")
logger.info("openai (%s): %s", openai_base, label)
elif active_provider == "openai":
results["openai"] = {"status": "error", "message": "API key not configured"}
print(f"{_PREFIX} WARNING: openai: API key not configured")
logger.warning("openai: API key not configured")
_llm_status.clear()
_llm_status.update(results)

75
pallas/log.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Logging setup for Pallas.
Configures structured JSON output so Alloy can extract a ``level`` label from
every log line and feed it into Loki. Call ``setup_logging()`` once from
``server.py:main()`` before any other module emits a log record.
Log format (one JSON object per line):
{"time": "2026-01-01T00:00:00Z", "level": "INFO", "logger": "pallas.server", "message": "..."}
Level conventions (Ouranos Lab — Python services use UPPERCASE):
ERROR — something is broken and requires human intervention
WARNING — degraded but self-recovering; retries, missing optional config
INFO — lifecycle events: start, ready, shutdown, LLM preflight
DEBUG — per-request detail; never enabled in production by default
"""
import json
import logging
from datetime import datetime, timezone
class _JSONFormatter(logging.Formatter):
"""Single-line JSON formatter compatible with Alloy's ``| json`` pipeline."""
def format(self, record: logging.LogRecord) -> str:
return json.dumps(
{
"time": datetime.fromtimestamp(record.created, tz=timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
)
class _HealthAccessFilter(logging.Filter):
"""Drop uvicorn access log lines for health/metrics endpoints.
Health check success is the *absence* of errors, not the presence of 200s.
Logging every HAProxy probe at INFO floods syslog with noise.
"""
_HEALTH_PATHS = (" GET /live ", " GET /ready ", " GET /metrics ")
def filter(self, record: logging.LogRecord) -> bool:
msg = record.getMessage()
return not any(path in msg for path in self._HEALTH_PATHS)
def setup_logging() -> None:
"""Configure Pallas logging.
- ``pallas.*`` logger: INFO, JSON to stdout
- ``httpx`` / ``httpcore``: WARNING (prevent request-level debug flooding)
- ``uvicorn.access``: health path filter applied
"""
handler = logging.StreamHandler()
handler.setFormatter(_JSONFormatter())
pallas_logger = logging.getLogger("pallas")
pallas_logger.setLevel(logging.INFO)
if not pallas_logger.handlers:
pallas_logger.addHandler(handler)
pallas_logger.propagate = False
# Silence noisy HTTP client internals — only surface warnings and above.
for noisy in ("httpx", "httpcore"):
logging.getLogger(noisy).setLevel(logging.WARNING)
# Suppress successful health probe access log entries.
health_filter = _HealthAccessFilter()
logging.getLogger("uvicorn.access").addFilter(health_filter)

31
pallas/multimodal_server.py
View File

@@ -28,6 +28,8 @@ from fast_agent.types import PromptMessageExtended, RequestParams
from fastmcp import Context as MCPContext
from fastmcp.prompts import Message
from mcp.types import ImageContent, TextContent
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
from starlette.responses import JSONResponse, Response
logger = get_logger(__name__)
@@ -58,6 +60,31 @@ def _history_to_fastmcp_messages(
class MultimodalAgentMCPServer(AgentMCPServer):
"""AgentMCPServer with optional image attachment support on send_message."""
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._register_health_routes()
def _register_health_routes(self) -> None:
"""Add /live, /ready, and /metrics to this agent's HTTP server.
Uses FastMCP's custom_route decorator — the same mechanism used by
fast-agent itself for the root ``/`` info route. HAProxy can health
check individual agent backends at ``/ready``.
"""
@self.mcp_server.custom_route("/live", methods=["GET"])
async def live(request):
return JSONResponse({"status": "alive"})
@self.mcp_server.custom_route("/ready", methods=["GET"])
async def ready(request):
return JSONResponse({"status": "ready"})
@self.mcp_server.custom_route("/metrics", methods=["GET"])
async def metrics(request):
data = generate_latest()
return Response(content=data, media_type=CONTENT_TYPE_LATEST)
def register_agent_tools(self, agent_name: str) -> None:
"""Register a send_message tool that accepts text + optional images."""
self._registered_agents.add(agent_name)
@@ -116,7 +143,7 @@ class MultimodalAgentMCPServer(AgentMCPServer):
async def execute_send() -> str:
start = time.perf_counter()
logger.info(
logger.debug(
f"MCP request received for agent '{agent_name}'",
name="mcp_request_start",
agent=agent_name,
@@ -124,7 +151,7 @@ class MultimodalAgentMCPServer(AgentMCPServer):
)
response = await agent.send(payload, request_params=request_params)
duration = time.perf_counter() - start
logger.info(
logger.debug(
f"Agent '{agent_name}' completed MCP request",
name="mcp_request_complete",
agent=agent_name,

115
pallas/registry.py
View File

@@ -4,17 +4,29 @@ Agent Registry
Serves GET /.well-known/mcp/server.json for agent discovery.
Reads agent topology from agents.yaml and model capabilities from
fastagent.config.yaml in the working directory.
Also exposes standard health and observability endpoints:
GET /live — liveness probe (always 200 while process is up)
GET /ready — readiness probe (200 when all configured agents reachable)
GET /metrics — Prometheus metrics in text exposition format
"""
import asyncio
import logging
import os
from datetime import datetime, timezone
from pathlib import Path
import httpx
import yaml
from prometheus_client import CONTENT_TYPE_LATEST, CollectorRegistry, Gauge, generate_latest
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse, Response
from starlette.routing import Route
logger = logging.getLogger(__name__)
def _config_root() -> Path:
"""Return the working directory where agents.yaml and fastagent configs live."""
@@ -106,17 +118,103 @@ def _build_registry(config: dict) -> dict:
return {"servers": entries}
# ── Starlette app ─────────────────────────────────────────────────────────────
# ── Prometheus metrics ────────────────────────────────────────────────────────
_metrics_registry = CollectorRegistry()
_pallas_up = Gauge(
"pallas_up",
"1 when the Pallas registry is running",
registry=_metrics_registry,
)
_pallas_up.set(1)
def _init_agent_metrics(config: dict) -> None:
"""Register per-agent info gauges once at startup."""
agents = config.get("agents", {})
if not agents:
return
agent_info = Gauge(
"pallas_agent_info",
"Static info about configured Pallas agents",
labelnames=["agent", "port"],
registry=_metrics_registry,
)
for name, agent in agents.items():
agent_info.labels(agent=name, port=str(agent["port"])).set(1)
# ── Route handlers ────────────────────────────────────────────────────────────
_deployment_config = _load_deployment_config()
_init_agent_metrics(_deployment_config)
async def server_json(request):
async def server_json(request: Request) -> JSONResponse:
return JSONResponse(_build_registry(_deployment_config))
async def live(request: Request) -> JSONResponse:
"""Liveness probe — always 200 while the process is running."""
return JSONResponse({"status": "alive"})
async def ready(request: Request) -> Response:
"""Readiness probe — 200 when all configured agents are reachable."""
agents = _deployment_config.get("agents", {})
if not agents:
return JSONResponse({"status": "ready"})
missing: list[str] = []
async with httpx.AsyncClient(timeout=2.0) as client:
checks = await asyncio.gather(
*(_probe_agent(client, name, agent["port"]) for name, agent in agents.items()),
return_exceptions=True,
)
for name, result in zip(agents.keys(), checks):
if isinstance(result, Exception) or result is False:
missing.append(name)
if missing:
return Response(
content=_json_bytes({"status": "unavailable", "missing": missing}),
status_code=503,
media_type="application/json",
)
return JSONResponse({"status": "ready"})
async def _probe_agent(client: httpx.AsyncClient, name: str, port: int) -> bool:
"""Return True if the agent's MCP port is accepting connections."""
try:
await client.get(f"http://127.0.0.1:{port}/mcp")
return True
except Exception:
return False
async def metrics(request: Request) -> Response:
"""Prometheus metrics in text exposition format."""
data = generate_latest(_metrics_registry)
return Response(content=data, media_type=CONTENT_TYPE_LATEST)
def _json_bytes(obj: dict) -> bytes:
import json
return json.dumps(obj).encode()
# ── Starlette app ─────────────────────────────────────────────────────────────
app = Starlette(
routes=[Route("/.well-known/mcp/server.json", server_json)],
routes=[
Route("/.well-known/mcp/server.json", server_json),
Route("/live", live),
Route("/ready", ready),
Route("/metrics", metrics),
],
)
@@ -128,8 +226,13 @@ async def run_registry(
import uvicorn
deploy_name = _deployment_config.get("name", "pallas")
print(f"[{deploy_name}] Registry on port {port}")
logger.info(
"Registry started: %s, port %d, %d agent(s)",
deploy_name,
port,
len(_deployment_config.get("agents", {})),
)
config = uvicorn.Config(app, host=host, port=port, log_level="info")
config = uvicorn.Config(app, host=host, port=port, log_level="warning")
server = uvicorn.Server(config)
await server.serve()

36
pallas/server.py
View File

@@ -18,8 +18,11 @@ from pathlib import Path
import yaml
from pallas.log import setup_logging
from pallas.multimodal_server import MultimodalAgentMCPServer
logger = logging.getLogger(__name__)
def _config_root() -> Path:
"""Return the working directory where agents.yaml and fastagent configs live."""
@@ -32,13 +35,13 @@ def _load_deployment_config() -> dict:
"""Load agents.yaml — single source of truth for deployment topology."""
config_path = _config_root() / os.environ.get("PALLAS_AGENTS_CONFIG", "agents.yaml")
if not config_path.exists():
raise SystemExit(f"[pallas] ERROR: deployment config not found: {config_path}")
raise SystemExit(f"deployment config not found: {config_path}")
with open(config_path) as f:
config = yaml.safe_load(f) or {}
if "agents" not in config:
raise SystemExit(f"[pallas] ERROR: no 'agents' section in {config_path}")
raise SystemExit(f"no 'agents' section in {config_path}")
return config
@@ -98,9 +101,9 @@ def _preflight_mcp_servers(agent_name: str, servers: dict[str, dict]) -> None:
for var in unresolved:
val = os.environ.get(var, "")
if not val:
print(
f"[pallas] WARNING: {agent_name} → {server_name}: "
f"{header_key} references ${{{var}}} but it is not set"
logger.warning(
"%s → %s: %s references ${%s} but it is not set",
agent_name, server_name, header_key, var,
)
@@ -140,10 +143,10 @@ def _register_unknown_models() -> None:
if is_vision:
tokenizes = list(ModelDatabase.QWEN_MULTIMODAL)
print(f"[pallas] Registered model '{model_name}' with vision capabilities")
logger.info("Registered model '%s' with vision capabilities", model_name)
else:
tokenizes = list(ModelDatabase.TEXT_ONLY)
print(f"[pallas] Registered model '{model_name}' as text-only")
logger.info("Registered model '%s' as text-only", model_name)
ModelDatabase.register_runtime_model_params(
model_name,
@@ -171,7 +174,7 @@ async def _start_agent(name: str, agents: dict[str, tuple[str, int]]) -> None:
module = importlib.import_module(module_path)
fast_instance = module.fast
print(f"[pallas] Starting {name} agent on port {port} ...")
logger.info("Starting %s agent on port %d", name, port)
async with fast_instance.run():
primary_instance = fast_instance._server_managed_instances[0]
@@ -207,11 +210,11 @@ async def _wait_for_agent(
try:
async with httpx.AsyncClient(timeout=2.0) as client:
await client.get(url)
print(f"[pallas] {name} is ready ✓")
logger.info("%s is ready", name)
return
except Exception:
await asyncio.sleep(1.0)
print(f"[pallas] WARNING: {name} did not become ready within {timeout}s")
logger.warning("%s did not become ready within %.0fs", name, timeout)
async def _run_single(name: str, agents: dict[str, tuple[str, int]]) -> None:
@@ -282,18 +285,19 @@ def main() -> None:
)
args = parser.parse_args()
logging.getLogger("httpx").setLevel(logging.DEBUG)
logging.getLogger("httpcore").setLevel(logging.DEBUG)
setup_logging()
if args.agent:
_, port = agents[args.agent]
print(f"[{deploy_name}] Starting {args.agent} agent on port {port} ...")
logger.info("Starting %s agent on port %d", args.agent, port)
asyncio.run(_run_single(args.agent, agents))
else:
print(f"[{deploy_name}] Starting all agents + registry ...")
print(f" {'registry':16s} → http://0.0.0.0:{registry_port}/.well-known/mcp/server.json")
logger.info("Starting all agents + registry for %s", deploy_name)
logger.info(
"registry → http://0.0.0.0:%d/.well-known/mcp/server.json", registry_port
)
for name, (_, port) in agents.items():
print(f" {name:16s} → http://0.0.0.0:{port}/mcp")
logger.info("%-16s → http://0.0.0.0:%d/mcp", name, port)
asyncio.run(_start_all(config))