Spaces:
Sleeping
Sleeping
File size: 5,722 Bytes
5acd81f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
# monitoring.py - System monitoring and metrics
import psutil
import time
import logging
from datetime import datetime
from typing import Dict, Any
import asyncio
import aiofiles
import json
class SystemMonitor:
"""System performance and health monitoring"""
def __init__(self, log_file: str = "logs/metrics.log"):
self.log_file = log_file
self.logger = logging.getLogger("system_monitor")
async def get_system_metrics(self) -> Dict[str, Any]:
"""Collect comprehensive system metrics"""
# CPU metrics
cpu_percent = psutil.cpu_percent(interval=1)
cpu_count = psutil.cpu_count()
# Memory metrics
memory = psutil.virtual_memory()
# Disk metrics
disk = psutil.disk_usage('/')
# Process metrics
process = psutil.Process()
process_memory = process.memory_info()
metrics = {
"timestamp": datetime.now().isoformat(),
"system": {
"cpu_percent": cpu_percent,
"cpu_count": cpu_count,
"memory_total": memory.total,
"memory_available": memory.available,
"memory_percent": memory.percent,
"disk_total": disk.total,
"disk_free": disk.free,
"disk_percent": disk.percent
},
"process": {
"pid": process.pid,
"memory_rss": process_memory.rss,
"memory_vms": process_memory.vms,
"cpu_percent": process.cpu_percent(),
"num_threads": process.num_threads(),
"create_time": process.create_time()
}
}
return metrics
async def log_metrics(self, metrics: Dict[str, Any]):
"""Log metrics to file"""
async with aiofiles.open(self.log_file, 'a') as f:
await f.write(json.dumps(metrics) + '\n')
async def check_health(self) -> Dict[str, str]:
"""Perform health checks"""
health_status = {
"overall": "healthy",
"components": {}
}
# Check CPU usage
cpu_percent = psutil.cpu_percent(interval=1)
if cpu_percent > 90:
health_status["components"]["cpu"] = "critical"
health_status["overall"] = "unhealthy"
elif cpu_percent > 70:
health_status["components"]["cpu"] = "warning"
else:
health_status["components"]["cpu"] = "healthy"
# Check memory usage
memory = psutil.virtual_memory()
if memory.percent > 90:
health_status["components"]["memory"] = "critical"
health_status["overall"] = "unhealthy"
elif memory.percent > 80:
health_status["components"]["memory"] = "warning"
else:
health_status["components"]["memory"] = "healthy"
# Check disk space
disk = psutil.disk_usage('/')
if disk.percent > 95:
health_status["components"]["disk"] = "critical"
health_status["overall"] = "unhealthy"
elif disk.percent > 85:
health_status["components"]["disk"] = "warning"
else:
health_status["components"]["disk"] = "healthy"
return health_status
class PerformanceProfiler:
"""Performance profiling for document processing"""
def __init__(self):
self.processing_times = []
self.error_rates = {}
self.throughput_metrics = {}
def record_processing_time(self, operation: str, duration: float, success: bool):
"""Record processing time and success rate"""
timestamp = time.time()
self.processing_times.append({
"operation": operation,
"duration": duration,
"success": success,
"timestamp": timestamp
})
# Update error rates
if operation not in self.error_rates:
self.error_rates[operation] = {"total": 0, "errors": 0}
self.error_rates[operation]["total"] += 1
if not success:
self.error_rates[operation]["errors"] += 1
def get_performance_summary(self) -> Dict[str, Any]:
"""Get performance summary"""
if not self.processing_times:
return {"message": "No performance data available"}
# Calculate averages by operation
operations = {}
for record in self.processing_times:
op = record["operation"]
if op not in operations:
operations[op] = []
operations[op].append(record["duration"])
summary = {}
for op, times in operations.items():
avg_time = sum(times) / len(times)
max_time = max(times)
min_time = min(times)
error_rate = 0
if op in self.error_rates:
total = self.error_rates[op]["total"]
errors = self.error_rates[op]["errors"]
error_rate = (errors / total) * 100 if total > 0 else 0
summary[op] = {
"avg_duration": round(avg_time, 2),
"max_duration": round(max_time, 2),
"min_duration": round(min_time, 2),
"total_operations": len(times),
"error_rate_percent": round(error_rate, 2)
}
return summary
|