2e7d419f05
- Add metrics.py with pool utilization, task backlog, stuck rate, avg processing time, eval completion rate - Add alerts.py with alert rules (pool utilization > 90%, task backlog > 50, stuck rate > 10%) - Implement alert history and webhook notifications - Add metrics and alerts APIs - Add database migration for alert history table - Add 11 unit tests for metrics, 10 unit tests for alerts, 8 integration tests - Update migration tests to include new alert history table All 853 tests passing.
119 lines
3.2 KiB
Python
119 lines
3.2 KiB
Python
"""Metrics calculation for cron pool monitoring (监控指标).
|
|
|
|
Calculates:
|
|
- Pool utilization (busy/total)
|
|
- Task backlog (pending tasks count)
|
|
- Stuck rate (stuck/total)
|
|
- Average task processing time
|
|
- Eval completion rate
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
|
from agenteval.storage.db import (
|
|
IntelligentEvalDB,
|
|
IntelligentEvalTaskQueueDB,
|
|
OpenClawCronPoolDB,
|
|
utc_now,
|
|
)
|
|
|
|
|
|
def calculate_pool_utilization(session: Session) -> float:
|
|
"""Calculate pool utilization (busy/total).
|
|
|
|
Returns:
|
|
Utilization rate (0.0 to 1.0)
|
|
"""
|
|
crons = session.exec(select(OpenClawCronPoolDB)).all()
|
|
if not crons:
|
|
return 0.0
|
|
|
|
busy = sum(1 for c in crons if c.status == "busy")
|
|
return busy / len(crons)
|
|
|
|
|
|
def calculate_task_backlog(session: Session) -> int:
|
|
"""Calculate task backlog (pending tasks count).
|
|
|
|
Returns:
|
|
Number of pending tasks
|
|
"""
|
|
pending = session.exec(
|
|
select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.status == "pending")
|
|
).all()
|
|
return len(pending)
|
|
|
|
|
|
def calculate_stuck_rate(session: Session) -> float:
|
|
"""Calculate stuck rate (stuck/total).
|
|
|
|
Returns:
|
|
Stuck rate (0.0 to 1.0)
|
|
"""
|
|
crons = session.exec(select(OpenClawCronPoolDB)).all()
|
|
if not crons:
|
|
return 0.0
|
|
|
|
stuck = sum(1 for c in crons if c.status == "stuck")
|
|
return stuck / len(crons)
|
|
|
|
|
|
def calculate_avg_processing_time(session: Session) -> Optional[float]:
|
|
"""Calculate average task processing time (in seconds).
|
|
|
|
Returns:
|
|
Average processing time in seconds, or None if no completed tasks
|
|
"""
|
|
completed_tasks = session.exec(
|
|
select(IntelligentEvalTaskQueueDB).where(
|
|
IntelligentEvalTaskQueueDB.status == "completed",
|
|
IntelligentEvalTaskQueueDB.assigned_at.isnot(None),
|
|
IntelligentEvalTaskQueueDB.completed_at.isnot(None),
|
|
)
|
|
).all()
|
|
|
|
if not completed_tasks:
|
|
return None
|
|
|
|
total_seconds = 0
|
|
for task in completed_tasks:
|
|
if task.assigned_at and task.completed_at:
|
|
duration = (task.completed_at - task.assigned_at).total_seconds()
|
|
total_seconds += duration
|
|
|
|
return total_seconds / len(completed_tasks)
|
|
|
|
|
|
def calculate_eval_completion_rate(session: Session) -> float:
|
|
"""Calculate evaluation completion rate.
|
|
|
|
Returns:
|
|
Completion rate (0.0 to 1.0)
|
|
"""
|
|
all_evals = session.exec(select(IntelligentEvalDB)).all()
|
|
if not all_evals:
|
|
return 0.0
|
|
|
|
completed = sum(1 for e in all_evals if e.status == IntelligentEvalStatus.COMPLETED.value)
|
|
return completed / len(all_evals)
|
|
|
|
|
|
def get_all_metrics(session: Session) -> dict:
|
|
"""Get all metrics.
|
|
|
|
Returns:
|
|
Dict with all metrics
|
|
"""
|
|
return {
|
|
"pool_utilization": calculate_pool_utilization(session),
|
|
"task_backlog": calculate_task_backlog(session),
|
|
"stuck_rate": calculate_stuck_rate(session),
|
|
"avg_processing_time_seconds": calculate_avg_processing_time(session),
|
|
"eval_completion_rate": calculate_eval_completion_rate(session),
|
|
"timestamp": utc_now().isoformat(),
|
|
}
|