420 lines
14 KiB
Python
420 lines
14 KiB
Python
"""
|
|
Performance monitoring dashboard for MaxClaw Agent
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Any, Optional
|
|
from pathlib import Path
|
|
import threading
|
|
|
|
# Add current directory to path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
class PerformanceDashboard:
|
|
"""Performance monitoring dashboard"""
|
|
|
|
def __init__(self, data_dir: Optional[str] = None):
|
|
"""
|
|
Initialize performance dashboard
|
|
|
|
Args:
|
|
data_dir: Directory for storing performance data
|
|
"""
|
|
if data_dir is None:
|
|
self.data_dir = Path.home() / ".maxclaw" / "performance"
|
|
else:
|
|
self.data_dir = Path(data_dir)
|
|
|
|
# Create data directory
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Performance metrics
|
|
self.metrics = {
|
|
"requests": [],
|
|
"errors": [],
|
|
"response_times": [],
|
|
"tool_usage": {},
|
|
"provider_usage": {},
|
|
"system_stats": []
|
|
}
|
|
|
|
# Load existing data
|
|
self._load_data()
|
|
|
|
# Start background monitoring
|
|
self._start_monitoring()
|
|
|
|
def _load_data(self):
|
|
"""Load existing performance data"""
|
|
data_file = self.data_dir / "performance_data.json"
|
|
|
|
if data_file.exists():
|
|
try:
|
|
with open(data_file, 'r') as f:
|
|
data = json.load(f)
|
|
self.metrics.update(data)
|
|
except Exception as e:
|
|
print(f"Warning: Could not load performance data: {e}")
|
|
|
|
def _save_data(self):
|
|
"""Save performance data to file"""
|
|
data_file = self.data_dir / "performance_data.json"
|
|
|
|
try:
|
|
with open(data_file, 'w') as f:
|
|
json.dump(self.metrics, f, indent=2)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save performance data: {e}")
|
|
|
|
def _start_monitoring(self):
|
|
"""Start background monitoring thread"""
|
|
def monitor():
|
|
while True:
|
|
time.sleep(60) # Collect data every minute
|
|
self._collect_system_stats()
|
|
self._save_data()
|
|
|
|
thread = threading.Thread(target=monitor, daemon=True)
|
|
thread.start()
|
|
|
|
def _collect_system_stats(self):
|
|
"""Collect system statistics"""
|
|
try:
|
|
import psutil
|
|
|
|
stats = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"cpu_percent": psutil.cpu_percent(interval=1),
|
|
"memory_percent": psutil.virtual_memory().percent,
|
|
"disk_usage": psutil.disk_usage('/').percent,
|
|
"network_connections": len(psutil.net_connections())
|
|
}
|
|
|
|
self.metrics["system_stats"].append(stats)
|
|
|
|
# Keep only last 1000 entries
|
|
if len(self.metrics["system_stats"]) > 1000:
|
|
self.metrics["system_stats"] = self.metrics["system_stats"][-1000:]
|
|
|
|
except ImportError:
|
|
# psutil not available, skip system stats
|
|
pass
|
|
except Exception as e:
|
|
print(f"Warning: Could not collect system stats: {e}")
|
|
|
|
def record_request(self, endpoint: str, method: str, status_code: int, duration: float):
|
|
"""Record an API request"""
|
|
request = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"endpoint": endpoint,
|
|
"method": method,
|
|
"status_code": status_code,
|
|
"duration": duration
|
|
}
|
|
|
|
self.metrics["requests"].append(request)
|
|
|
|
# Keep only last 10000 requests
|
|
if len(self.metrics["requests"]) > 10000:
|
|
self.metrics["requests"] = self.metrics["requests"][-10000:]
|
|
|
|
def record_error(self, error_type: str, error_message: str, context: Optional[str] = None):
|
|
"""Record an error"""
|
|
error = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"type": error_type,
|
|
"message": error_message,
|
|
"context": context
|
|
}
|
|
|
|
self.metrics["errors"].append(error)
|
|
|
|
# Keep only last 1000 errors
|
|
if len(self.metrics["errors"]) > 1000:
|
|
self.metrics["errors"] = self.metrics["errors"][-1000:]
|
|
|
|
def record_response_time(self, operation: str, duration: float):
|
|
"""Record response time for an operation"""
|
|
response_time = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"operation": operation,
|
|
"duration": duration
|
|
}
|
|
|
|
self.metrics["response_times"].append(response_time)
|
|
|
|
# Keep only last 5000 response times
|
|
if len(self.metrics["response_times"]) > 5000:
|
|
self.metrics["response_times"] = self.metrics["response_times"][-5000:]
|
|
|
|
def record_tool_usage(self, tool_name: str, success: bool, duration: float):
|
|
"""Record tool usage"""
|
|
if tool_name not in self.metrics["tool_usage"]:
|
|
self.metrics["tool_usage"][tool_name] = {
|
|
"total_calls": 0,
|
|
"successful_calls": 0,
|
|
"failed_calls": 0,
|
|
"total_duration": 0,
|
|
"avg_duration": 0
|
|
}
|
|
|
|
tool_stats = self.metrics["tool_usage"][tool_name]
|
|
tool_stats["total_calls"] += 1
|
|
tool_stats["total_duration"] += duration
|
|
|
|
if success:
|
|
tool_stats["successful_calls"] += 1
|
|
else:
|
|
tool_stats["failed_calls"] += 1
|
|
|
|
tool_stats["avg_duration"] = tool_stats["total_duration"] / tool_stats["total_calls"]
|
|
|
|
def record_provider_usage(self, provider: str, model: str, success: bool):
|
|
"""Record provider usage"""
|
|
key = f"{provider}:{model}"
|
|
|
|
if key not in self.metrics["provider_usage"]:
|
|
self.metrics["provider_usage"][key] = {
|
|
"total_calls": 0,
|
|
"successful_calls": 0,
|
|
"failed_calls": 0
|
|
}
|
|
|
|
provider_stats = self.metrics["provider_usage"][key]
|
|
provider_stats["total_calls"] += 1
|
|
|
|
if success:
|
|
provider_stats["successful_calls"] += 1
|
|
else:
|
|
provider_stats["failed_calls"] += 1
|
|
|
|
def get_summary(self, hours: int = 24) -> Dict[str, Any]:
|
|
"""
|
|
Get performance summary for the last N hours
|
|
|
|
Args:
|
|
hours: Number of hours to include in summary
|
|
|
|
Returns:
|
|
Dictionary with performance summary
|
|
"""
|
|
cutoff_time = datetime.now() - timedelta(hours=hours)
|
|
|
|
# Filter recent data
|
|
recent_requests = [
|
|
r for r in self.metrics["requests"]
|
|
if datetime.fromisoformat(r["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
recent_errors = [
|
|
e for e in self.metrics["errors"]
|
|
if datetime.fromisoformat(e["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
recent_response_times = [
|
|
rt for rt in self.metrics["response_times"]
|
|
if datetime.fromisoformat(rt["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
# Calculate statistics
|
|
total_requests = len(recent_requests)
|
|
total_errors = len(recent_errors)
|
|
|
|
if total_requests > 0:
|
|
error_rate = total_errors / total_requests
|
|
avg_response_time = sum(r["duration"] for r in recent_requests) / total_requests
|
|
else:
|
|
error_rate = 0
|
|
avg_response_time = 0
|
|
|
|
# Status code distribution
|
|
status_codes = {}
|
|
for request in recent_requests:
|
|
code = request["status_code"]
|
|
status_codes[code] = status_codes.get(code, 0) + 1
|
|
|
|
# Endpoint usage
|
|
endpoints = {}
|
|
for request in recent_requests:
|
|
endpoint = request["endpoint"]
|
|
endpoints[endpoint] = endpoints.get(endpoint, 0) + 1
|
|
|
|
# Tool usage summary
|
|
tool_summary = {}
|
|
for tool_name, stats in self.metrics["tool_usage"].items():
|
|
tool_summary[tool_name] = {
|
|
"total_calls": stats["total_calls"],
|
|
"success_rate": stats["successful_calls"] / stats["total_calls"] if stats["total_calls"] > 0 else 0,
|
|
"avg_duration": stats["avg_duration"]
|
|
}
|
|
|
|
# Provider usage summary
|
|
provider_summary = {}
|
|
for provider_key, stats in self.metrics["provider_usage"].items():
|
|
provider_summary[provider_key] = {
|
|
"total_calls": stats["total_calls"],
|
|
"success_rate": stats["successful_calls"] / stats["total_calls"] if stats["total_calls"] > 0 else 0
|
|
}
|
|
|
|
# System stats (if available)
|
|
system_stats = {}
|
|
if self.metrics["system_stats"]:
|
|
latest_stats = self.metrics["system_stats"][-1]
|
|
system_stats = {
|
|
"cpu_percent": latest_stats.get("cpu_percent", 0),
|
|
"memory_percent": latest_stats.get("memory_percent", 0),
|
|
"disk_usage": latest_stats.get("disk_usage", 0)
|
|
}
|
|
|
|
return {
|
|
"period_hours": hours,
|
|
"total_requests": total_requests,
|
|
"total_errors": total_errors,
|
|
"error_rate": error_rate,
|
|
"avg_response_time": avg_response_time,
|
|
"status_codes": status_codes,
|
|
"endpoints": endpoints,
|
|
"tool_usage": tool_summary,
|
|
"provider_usage": provider_summary,
|
|
"system_stats": system_stats,
|
|
"generated_at": datetime.now().isoformat()
|
|
}
|
|
|
|
def get_detailed_report(self, hours: int = 24) -> str:
|
|
"""
|
|
Get detailed performance report
|
|
|
|
Args:
|
|
hours: Number of hours to include in report
|
|
|
|
Returns:
|
|
Formatted report string
|
|
"""
|
|
summary = self.get_summary(hours)
|
|
|
|
report = f"📊 Performance Report (Last {hours} hours)\n"
|
|
report += "=" * 50 + "\n\n"
|
|
|
|
# Request statistics
|
|
report += f"📈 Request Statistics:\n"
|
|
report += f" Total Requests: {summary['total_requests']}\n"
|
|
report += f" Total Errors: {summary['total_errors']}\n"
|
|
report += f" Error Rate: {summary['error_rate']:.2%}\n"
|
|
report += f" Avg Response Time: {summary['avg_response_time']:.3f}s\n\n"
|
|
|
|
# Status codes
|
|
if summary['status_codes']:
|
|
report += f"📋 Status Code Distribution:\n"
|
|
for code, count in sorted(summary['status_codes'].items()):
|
|
report += f" {code}: {count}\n"
|
|
report += "\n"
|
|
|
|
# Top endpoints
|
|
if summary['endpoints']:
|
|
report += f"🔗 Top Endpoints:\n"
|
|
sorted_endpoints = sorted(summary['endpoints'].items(), key=lambda x: x[1], reverse=True)[:10]
|
|
for endpoint, count in sorted_endpoints:
|
|
report += f" {endpoint}: {count}\n"
|
|
report += "\n"
|
|
|
|
# Tool usage
|
|
if summary['tool_usage']:
|
|
report += f"🔧 Tool Usage:\n"
|
|
for tool_name, stats in summary['tool_usage'].items():
|
|
report += f" {tool_name}:\n"
|
|
report += f" Calls: {stats['total_calls']}\n"
|
|
report += f" Success Rate: {stats['success_rate']:.2%}\n"
|
|
report += f" Avg Duration: {stats['avg_duration']:.3f}s\n"
|
|
report += "\n"
|
|
|
|
# Provider usage
|
|
if summary['provider_usage']:
|
|
report += f"🤖 Provider Usage:\n"
|
|
for provider_key, stats in summary['provider_usage'].items():
|
|
report += f" {provider_key}:\n"
|
|
report += f" Calls: {stats['total_calls']}\n"
|
|
report += f" Success Rate: {stats['success_rate']:.2%}\n"
|
|
report += "\n"
|
|
|
|
# System stats
|
|
if summary['system_stats']:
|
|
report += f"💻 System Statistics:\n"
|
|
report += f" CPU Usage: {summary['system_stats']['cpu_percent']}%\n"
|
|
report += f" Memory Usage: {summary['system_stats']['memory_percent']}%\n"
|
|
report += f" Disk Usage: {summary['system_stats']['disk_usage']}%\n"
|
|
|
|
return report
|
|
|
|
def clear_old_data(self, days: int = 30):
|
|
"""
|
|
Clear data older than specified days
|
|
|
|
Args:
|
|
days: Number of days to keep data
|
|
"""
|
|
cutoff_time = datetime.now() - timedelta(days=days)
|
|
|
|
# Filter out old data
|
|
self.metrics["requests"] = [
|
|
r for r in self.metrics["requests"]
|
|
if datetime.fromisoformat(r["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
self.metrics["errors"] = [
|
|
e for e in self.metrics["errors"]
|
|
if datetime.fromisoformat(e["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
self.metrics["response_times"] = [
|
|
rt for rt in self.metrics["response_times"]
|
|
if datetime.fromisoformat(rt["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
self.metrics["system_stats"] = [
|
|
s for s in self.metrics["system_stats"]
|
|
if datetime.fromisoformat(s["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
# Save cleaned data
|
|
self._save_data()
|
|
|
|
print(f"✅ Cleared data older than {days} days")
|
|
|
|
|
|
# Global performance dashboard instance
|
|
performance_dashboard = PerformanceDashboard()
|
|
|
|
|
|
def get_performance_summary(hours: int = 24) -> Dict[str, Any]:
|
|
"""
|
|
Get performance summary
|
|
|
|
Args:
|
|
hours: Number of hours to include
|
|
|
|
Returns:
|
|
Performance summary dictionary
|
|
"""
|
|
return performance_dashboard.get_summary(hours)
|
|
|
|
|
|
def get_performance_report(hours: int = 24) -> str:
|
|
"""
|
|
Get formatted performance report
|
|
|
|
Args:
|
|
hours: Number of hours to include
|
|
|
|
Returns:
|
|
Formatted report string
|
|
"""
|
|
return performance_dashboard.get_detailed_report(hours)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Print performance report when run directly
|
|
print(get_performance_report()) |