6d32653675
All checks were successful
CI / test (push) Successful in 3m56s
AlertManager gains an optional openclaw_client. check_alerts records each newly created alert and AlertManager.maybe_autoscale (called from the async router for each alert) invokes cron_pool.scale_up(1). scale_up itself caps at MAX_POOL_SIZE so repeated invocations are safe. Removed the xfail guard in test_alert_autoscale_link; rewrote the test to use task_backlog (duration_minutes=0) so a single check_alerts call fires an alert and triggers auto-scale.
157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
"""API routes for OpenClaw cron pool management."""
|
|
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
from sqlmodel import Session
|
|
|
|
from agenteval.intelligent_eval import cron_pool
|
|
from agenteval.intelligent_eval.openclaw_client import OpenClawClient
|
|
from agenteval.web.deps import get_db
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class ScaleRequest(BaseModel):
|
|
target_size: int = Field(ge=1, le=50)
|
|
|
|
|
|
class HeartbeatRequest(BaseModel):
|
|
status: str # idle / busy
|
|
current_eval_id: str | None = None
|
|
|
|
|
|
@router.get("/cron-pool")
|
|
async def get_cron_pool_status(session: Session = Depends(get_db)) -> dict:
|
|
"""Get cron pool status."""
|
|
status = cron_pool.get_pool_status(session)
|
|
return {"pool": status}
|
|
|
|
|
|
@router.post("/cron-pool/scale")
|
|
async def scale_cron_pool(request: ScaleRequest, session: Session = Depends(get_db)) -> dict:
|
|
"""Manually scale cron pool to target size."""
|
|
client = OpenClawClient()
|
|
return await cron_pool.scale_to(request.target_size, session, client)
|
|
|
|
|
|
@router.post("/cron-pool/sync")
|
|
async def sync_cron_states(session: Session = Depends(get_db)) -> dict:
|
|
"""Sync cron states from OpenClaw to platform DB."""
|
|
client = OpenClawClient()
|
|
synced = await cron_pool.sync_cron_states(session, client)
|
|
return {"success": True, "synced": synced}
|
|
|
|
|
|
@router.post("/cron-pool/auto-scale")
|
|
async def auto_scale_pool(session: Session = Depends(get_db)) -> dict:
|
|
"""Trigger auto-scaling based on current load."""
|
|
client = OpenClawClient()
|
|
scaled_up, scaled_down = await cron_pool.auto_scale(session, client)
|
|
return {
|
|
"success": True,
|
|
"scaled_up": scaled_up,
|
|
"scaled_down": scaled_down,
|
|
}
|
|
|
|
|
|
@router.post("/crons/{cron_id}/heartbeat")
|
|
async def report_heartbeat(
|
|
cron_id: str,
|
|
request: HeartbeatRequest,
|
|
session: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Report cron heartbeat. Updates last_active_at and current status."""
|
|
try:
|
|
cron_pool.heartbeat(cron_id, request.status, request.current_eval_id, session)
|
|
except LookupError as e:
|
|
raise HTTPException(status_code=404, detail=str(e)) from e
|
|
return {"success": True}
|
|
|
|
|
|
@router.get("/cron-pool/metrics")
|
|
async def get_cron_pool_metrics(session: Session = Depends(get_db)) -> dict:
|
|
"""Get cron pool metrics."""
|
|
from agenteval.intelligent_eval.metrics import get_all_metrics
|
|
|
|
metrics = get_all_metrics(session)
|
|
return {"metrics": metrics}
|
|
|
|
|
|
@router.post("/cron-pool/check-alerts")
|
|
async def check_alerts(session: Session = Depends(get_db)) -> dict:
|
|
"""Check alert rules and create alerts if triggered."""
|
|
from agenteval.intelligent_eval.alerts import AlertManager
|
|
|
|
client = OpenClawClient()
|
|
manager = AlertManager(session, openclaw_client=client)
|
|
alerts = manager.check_rules()
|
|
# T7: link each newly-created alert to auto-scale (best-effort).
|
|
for alert in alerts:
|
|
await manager.maybe_autoscale(alert)
|
|
|
|
return {
|
|
"success": True,
|
|
"alerts_triggered": len(alerts),
|
|
"alerts": [
|
|
{
|
|
"id": alert.id,
|
|
"alert_type": alert.alert_type,
|
|
"severity": alert.severity,
|
|
"message": alert.message,
|
|
"metric_value": alert.metric_value,
|
|
"threshold": alert.threshold,
|
|
"created_at": alert.created_at.isoformat(),
|
|
}
|
|
for alert in alerts
|
|
],
|
|
}
|
|
|
|
|
|
@router.get("/cron-pool/alerts")
|
|
async def get_alert_history(
|
|
limit: int = 100,
|
|
unresolved_only: bool = False,
|
|
session: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Get alert history."""
|
|
from agenteval.intelligent_eval.alerts import AlertManager
|
|
|
|
manager = AlertManager(session)
|
|
|
|
if unresolved_only:
|
|
alerts = manager.get_unresolved_alerts()
|
|
else:
|
|
alerts = manager.get_alert_history(limit=limit)
|
|
|
|
return {
|
|
"alerts": [
|
|
{
|
|
"id": alert.id,
|
|
"alert_type": alert.alert_type,
|
|
"severity": alert.severity,
|
|
"message": alert.message,
|
|
"metric_value": alert.metric_value,
|
|
"threshold": alert.threshold,
|
|
"created_at": alert.created_at.isoformat(),
|
|
"resolved_at": alert.resolved_at.isoformat() if alert.resolved_at else None,
|
|
"webhook_sent": alert.webhook_sent,
|
|
}
|
|
for alert in alerts
|
|
]
|
|
}
|
|
|
|
|
|
@router.post("/cron-pool/alerts/{alert_id}/resolve")
|
|
async def resolve_alert(alert_id: str, session: Session = Depends(get_db)) -> dict:
|
|
"""Resolve an alert."""
|
|
from agenteval.intelligent_eval.alerts import AlertManager
|
|
|
|
manager = AlertManager(session)
|
|
resolved = manager.resolve_alert(alert_id)
|
|
|
|
if not resolved:
|
|
raise HTTPException(status_code=404, detail=f"alert {alert_id} not found")
|
|
|
|
return {"success": True}
|