forked from HAL9000/cleveragents-core
e5f75c5c83
- Remove maximum cap (16) on CA_MAX_PARALLEL_WORKERS in resources.yaml - Can now be set to any positive value (32, 64, etc.) - Only minimum validation remains (must be > 0) - Remove dynamic backpressure/throttling from implementation-orchestrator - Dispatch always runs at full configured speed - Resource monitoring remains for visibility only - No automatic reduction of slots_available based on failures - Convert system-watchdog from auto-degradation to monitoring + suggestions - Renamed DEGRADATION_THRESHOLDS to HEALTH_THRESHOLDS - Removed apply_system_degradation() and check_degradation_recovery() - Changed findings to include suggestions instead of actions - Watchdog now reports issues with fix recommendations - No automatic throttling or pausing of agents The system now operates at maximum configured speed at all times, with the watchdog providing diagnostic insights when issues arise.
15 KiB
15 KiB
Performance Monitoring
This document provides standardized performance monitoring utilities for tracking agent efficiency, resource usage, and system health.
Core Metrics
from datetime import datetime, timedelta
import time
import json
from typing import Dict, List, Any, Optional
from collections import defaultdict
class PerformanceMonitor:
"""Track and report performance metrics for agents."""
def __init__(self, agent_name: str, session_id: str):
self.agent_name = agent_name
self.session_id = session_id
self.metrics = defaultdict(dict)
self.operation_stack = []
def start_operation(self, operation_name: str, metadata: Optional[Dict] = None):
"""Start timing an operation."""
operation_id = f"{operation_name}_{int(time.time() * 1000)}"
op_data = {
'id': operation_id,
'name': operation_name,
'start_time': time.time(),
'metadata': metadata or {}
}
self.operation_stack.append(op_data)
self.metrics[operation_name][operation_id] = op_data
return operation_id
def end_operation(self, operation_id: Optional[str] = None,
status: str = 'success',
result_metadata: Optional[Dict] = None):
"""End timing and record results."""
# If no ID provided, use most recent operation
if not operation_id and self.operation_stack:
operation_data = self.operation_stack.pop()
operation_id = operation_data['id']
else:
# Find and remove from stack
operation_data = None
for i, op in enumerate(self.operation_stack):
if op['id'] == operation_id:
operation_data = self.operation_stack.pop(i)
break
if not operation_data:
# Try to find in metrics
for op_name, ops in self.metrics.items():
if operation_id in ops:
operation_data = ops[operation_id]
break
if not operation_data:
print(f"[WARNING] Unknown operation: {operation_id}")
return
# Calculate duration
end_time = time.time()
duration = end_time - operation_data['start_time']
# Update operation data
operation_data.update({
'end_time': end_time,
'duration_seconds': duration,
'status': status,
'result_metadata': result_metadata or {}
})
return operation_data
def get_operation_stats(self, operation_name: str) -> Dict[str, Any]:
"""Get statistics for a specific operation type."""
operations = self.metrics.get(operation_name, {})
if not operations:
return {}
completed = [op for op in operations.values() if 'duration_seconds' in op]
if not completed:
return {'count': len(operations), 'completed': 0}
durations = [op['duration_seconds'] for op in completed]
statuses = [op.get('status', 'unknown') for op in completed]
return {
'count': len(operations),
'completed': len(completed),
'success_rate': statuses.count('success') / len(statuses),
'avg_duration': sum(durations) / len(durations),
'min_duration': min(durations),
'max_duration': max(durations),
'total_duration': sum(durations),
'status_breakdown': {
status: statuses.count(status)
for status in set(statuses)
}
}
def get_summary(self) -> Dict[str, Any]:
"""Get overall performance summary."""
summary = {
'agent': self.agent_name,
'session': self.session_id,
'timestamp': datetime.utcnow().isoformat() + 'Z',
'operations': {}
}
for op_name in self.metrics:
summary['operations'][op_name] = self.get_operation_stats(op_name)
# Calculate totals
total_completed = sum(
stats.get('completed', 0)
for stats in summary['operations'].values()
)
total_duration = sum(
stats.get('total_duration', 0)
for stats in summary['operations'].values()
)
summary['totals'] = {
'operations_completed': total_completed,
'total_duration_seconds': total_duration,
'operations_in_progress': len(self.operation_stack)
}
return summary
Resource Usage Tracking
import psutil
import os
class ResourceMonitor:
"""Monitor system resource usage."""
def __init__(self):
self.process = psutil.Process(os.getpid())
self.start_time = time.time()
self.snapshots = []
def take_snapshot(self, label: str = ""):
"""Capture current resource usage."""
try:
cpu_percent = self.process.cpu_percent(interval=0.1)
memory_info = self.process.memory_info()
io_counters = self.process.io_counters() if hasattr(self.process, 'io_counters') else None
snapshot = {
'timestamp': time.time(),
'label': label,
'cpu_percent': cpu_percent,
'memory_mb': memory_info.rss / 1024 / 1024,
'memory_percent': self.process.memory_percent(),
}
if io_counters:
snapshot.update({
'disk_read_mb': io_counters.read_bytes / 1024 / 1024,
'disk_write_mb': io_counters.write_bytes / 1024 / 1024
})
# System-wide metrics
snapshot['system'] = {
'cpu_percent': psutil.cpu_percent(interval=0.1),
'memory_percent': psutil.virtual_memory().percent,
'disk_usage_percent': psutil.disk_usage('/').percent
}
self.snapshots.append(snapshot)
return snapshot
except Exception as e:
print(f"[WARNING] Resource snapshot failed: {e}")
return None
def get_summary(self) -> Dict[str, Any]:
"""Get resource usage summary."""
if not self.snapshots:
return {}
# Calculate averages and peaks
cpu_values = [s['cpu_percent'] for s in self.snapshots]
memory_values = [s['memory_mb'] for s in self.snapshots]
summary = {
'duration_seconds': time.time() - self.start_time,
'snapshots': len(self.snapshots),
'process': {
'cpu_avg_percent': sum(cpu_values) / len(cpu_values),
'cpu_peak_percent': max(cpu_values),
'memory_avg_mb': sum(memory_values) / len(memory_values),
'memory_peak_mb': max(memory_values),
}
}
# Add I/O stats if available
if 'disk_read_mb' in self.snapshots[-1]:
summary['io'] = {
'total_read_mb': self.snapshots[-1]['disk_read_mb'],
'total_write_mb': self.snapshots[-1]['disk_write_mb']
}
return summary
Integration with Agents
from shared.performance_monitoring import PerformanceMonitor, ResourceMonitor
from shared.logging import AgentLogger
class MonitoredAgent:
"""Base class for agents with integrated monitoring."""
def __init__(self, agent_name: str, session_id: str):
self.logger = AgentLogger(agent_name, session_id)
self.perf_monitor = PerformanceMonitor(agent_name, session_id)
self.resource_monitor = ResourceMonitor()
# Take initial resource snapshot
self.resource_monitor.take_snapshot("startup")
def monitored_operation(self, operation_name: str, operation_func, **kwargs):
"""Execute an operation with full monitoring."""
# Start performance tracking
op_id = self.perf_monitor.start_operation(operation_name, kwargs)
# Log operation start
self.logger.info(f"Starting {operation_name}", action=operation_name)
# Take resource snapshot before
self.resource_monitor.take_snapshot(f"{operation_name}_start")
try:
# Execute the operation
result = operation_func(**kwargs)
# Success
self.perf_monitor.end_operation(op_id, status='success')
self.logger.info(f"Completed {operation_name}", action=operation_name)
return result
except Exception as e:
# Failure
self.perf_monitor.end_operation(op_id, status='failure',
result_metadata={'error': str(e)})
self.logger.error(f"Failed {operation_name}: {e}", action=operation_name)
raise
finally:
# Take resource snapshot after
self.resource_monitor.take_snapshot(f"{operation_name}_end")
def report_metrics(self, session_state_issue: int, owner: str, repo: str):
"""Report collected metrics to Forgejo."""
perf_summary = self.perf_monitor.get_summary()
resource_summary = self.resource_monitor.get_summary()
report = {
'performance': perf_summary,
'resources': resource_summary,
'timestamp': datetime.utcnow().isoformat() + 'Z'
}
comment = f"""[METRICS] {self.agent_name}
Performance Summary:
```json
{json.dumps(perf_summary, indent=2)}
Resource Usage:
{json.dumps(resource_summary, indent=2)}
Automated by CleverAgents Bot Agent: {self.agent_name}"""
forgejo_create_issue_comment(owner, repo, session_state_issue, comment)
## Usage Example
```python
# Implementation worker with monitoring
class MonitoredImplementationWorker(MonitoredAgent):
def implement_issue(self, issue_number: int):
"""Implement issue with full monitoring."""
# Clone with monitoring
def do_clone():
return git_clone(repo_url, target_dir)
clone_dir = self.monitored_operation(
"clone_repository",
do_clone,
repo_url=repo_url,
target_dir=target_dir
)
# Implement each subtask with monitoring
for i, subtask in enumerate(subtasks):
def do_subtask():
return implement_subtask(subtask, clone_dir)
self.monitored_operation(
f"implement_subtask_{i}",
do_subtask,
subtask_title=subtask['title']
)
# Report metrics at end
self.report_metrics(session_state_issue, owner, repo)
Performance Optimization Patterns
Batch Operations
def batch_monitored_operations(monitor: PerformanceMonitor,
operation_name: str,
items: List[Any],
process_func,
batch_size: int = 10):
"""Process items in batches with monitoring."""
total_op_id = monitor.start_operation(f"{operation_name}_batch", {
'total_items': len(items),
'batch_size': batch_size
})
results = []
failures = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
batch_op_id = monitor.start_operation(f"{operation_name}_batch_{i//batch_size}", {
'batch_index': i // batch_size,
'batch_items': len(batch)
})
try:
batch_results = process_func(batch)
results.extend(batch_results)
monitor.end_operation(batch_op_id, 'success')
except Exception as e:
failures.extend(batch)
monitor.end_operation(batch_op_id, 'failure', {'error': str(e)})
# End total operation
monitor.end_operation(total_op_id,
'partial' if failures else 'success',
{
'processed': len(results),
'failed': len(failures)
})
return results, failures
Adaptive Timeouts
def adaptive_timeout(monitor: PerformanceMonitor,
operation_name: str,
default_timeout: int = 30):
"""Calculate adaptive timeout based on historical performance."""
stats = monitor.get_operation_stats(operation_name)
if not stats or stats['completed'] < 3:
# Not enough history
return default_timeout
# Use 99th percentile approximation (max of observed)
# Add 50% buffer
timeout = int(stats['max_duration'] * 1.5)
# Bound between min and max
return max(10, min(timeout, default_timeout * 3))
Health Metrics
def calculate_health_score(perf_summary: Dict,
resource_summary: Dict,
failure_threshold: float = 0.2) -> float:
"""Calculate overall health score (0-1)."""
scores = []
# Performance score based on success rates
for op_name, stats in perf_summary.get('operations', {}).items():
if stats.get('completed', 0) > 0:
success_rate = stats.get('success_rate', 0)
scores.append(success_rate)
# Resource score based on usage
if resource_summary:
cpu_score = 1 - min(resource_summary['process']['cpu_avg_percent'] / 100, 1)
memory_score = 1 - min(resource_summary['process']['memory_avg_mb'] / 1000, 1)
scores.extend([cpu_score, memory_score])
if not scores:
return 1.0
# Weighted average
health = sum(scores) / len(scores)
# Penalize if too many operations in progress (might be stuck)
in_progress = perf_summary.get('totals', {}).get('operations_in_progress', 0)
if in_progress > 10:
health *= 0.8
return health
Dashboard Metrics
For system-wide monitoring, aggregate these metrics:
# Key Performance Indicators (KPIs)
SYSTEM_KPIS = {
# Throughput
'issues_per_hour': 'Count of issues completed per hour',
'prs_merged_per_hour': 'PRs successfully merged per hour',
'tests_written_per_hour': 'Test scenarios created per hour',
# Quality
'first_attempt_success_rate': 'Work completed without retry',
'review_approval_rate': 'PRs approved on first review',
'test_coverage_trend': 'Coverage percentage over time',
# Efficiency
'avg_issue_completion_time': 'Time from start to PR merge',
'avg_pr_fix_time': 'Time to fix failing PR',
'worker_utilization': 'Active workers / max workers',
# Health
'system_health_score': 'Overall system health (0-1)',
'failure_rate_trend': 'Failure rate over time',
'resource_pressure': 'CPU/memory usage indicators'
}