Get summary of monitoring results.
(self)
| 117 | }) |
| 118 | |
| 119 | def get_summary(self) -> Dict[str, Any]: |
| 120 | """Get summary of monitoring results.""" |
| 121 | if not self.metrics["response_times"]: |
| 122 | return {"error": "No data collected"} |
| 123 | |
| 124 | response_times = [m["latency"] for m in self.metrics["response_times"]] |
| 125 | |
| 126 | return { |
| 127 | "response_time_percentiles": { |
| 128 | "p50": statistics.median(response_times), |
| 129 | "p95": statistics.quantiles(response_times, n=20)[18] if len(response_times) > 20 else max(response_times), |
| 130 | "p99": statistics.quantiles(response_times, n=100)[98] if len(response_times) > 100 else max(response_times) |
| 131 | }, |
| 132 | "error_rate": len(self.metrics["error_counts"]) / len(self.metrics["response_times"]) if self.metrics["response_times"] else 0, |
| 133 | "peak_memory_mb": max(m["memory_mb"] for m in self.metrics["memory_usage"]) if self.metrics["memory_usage"] else 0, |
| 134 | "avg_cpu_percent": statistics.mean(m["cpu_percent"] for m in self.metrics["cpu_usage"]) if self.metrics["cpu_usage"] else 0, |
| 135 | "max_queue_depth": max(m["depth"] for m in self.metrics["queue_depths"]) if self.metrics["queue_depths"] else 0, |
| 136 | "message_loss_count": self.metrics["message_loss"] |
| 137 | } |
| 138 | |
| 139 | |
| 140 | class LoadTestClient: |