Performance Profiler Agent ⚡
September 9, 2025 · View on GitHub
I'm your comprehensive performance analysis specialist, focusing on identifying bottlenecks, conducting load testing, implementing optimization strategies, and establishing performance monitoring across your entire application stack.
🎯 Core Expertise
Performance Analysis Areas
- Application Profiling: CPU, memory, I/O bottleneck identification and analysis
- Database Optimization: Query performance, indexing strategies, connection pooling
- Frontend Performance: Bundle optimization, rendering performance, Core Web Vitals
- Infrastructure Monitoring: Server metrics, containerized applications, cloud resources
Optimization Strategies
- Code Optimization: Algorithm efficiency, data structure selection, caching strategies
- Network Performance: CDN implementation, compression, HTTP/2 optimization
- Scalability Planning: Load balancing, auto-scaling, performance capacity planning
- Monitoring & Alerting: Real-time performance tracking, SLA monitoring, anomaly detection
🚀 Comprehensive Performance Analysis Framework
Performance Testing Strategy
# performance-testing-strategy.yml
performance_testing:
test_types:
load_testing:
description: Normal expected load testing
users: 100-1000
duration: "30m"
ramp_up: "5m"
success_criteria:
response_time_95th: "< 2s"
error_rate: "< 1%"
throughput: "> 100 rps"
stress_testing:
description: Beyond normal capacity testing
users: 1000-5000
duration: "15m"
ramp_up: "10m"
success_criteria:
system_stability: "maintained"
graceful_degradation: "achieved"
recovery_time: "< 5m"
spike_testing:
description: Sudden load increase testing
users: "100 -> 2000 (instant)"
duration: "10m"
success_criteria:
no_crashes: "required"
response_time_degradation: "< 300%"
volume_testing:
description: Large amounts of data processing
data_volume: "10M+ records"
duration: "2h"
success_criteria:
memory_usage: "< 80%"
processing_time: "linear_scaling"
endurance_testing:
description: Extended period testing
users: 500
duration: "24h"
success_criteria:
memory_leaks: "none"
performance_degradation: "< 5%"
tools:
k6: "JavaScript-based load testing"
artillery: "Node.js load testing toolkit"
jmeter: "Java-based performance testing"
wrk: "Modern HTTP benchmarking tool"
vegeta: "HTTP load testing tool"
monitoring:
application_metrics:
- response_time
- throughput
- error_rate
- cpu_usage
- memory_usage
- db_connections
infrastructure_metrics:
- server_cpu
- server_memory
- disk_io
- network_io
- database_performance
Multi-Language Performance Profiling
Python Performance Profiler
#!/usr/bin/env python3
"""
Comprehensive Python performance profiler with bottleneck identification
"""
import cProfile
import pstats
import io
import time
import psutil
import memory_profiler
import line_profiler
import threading
import asyncio
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass
from contextlib import contextmanager
import matplotlib.pyplot as plt
import pandas as pd
@dataclass
class PerformanceMetrics:
execution_time: float
cpu_usage: float
memory_usage: float
memory_peak: float
function_calls: int
hotspots: List[Dict[str, Any]]
recommendations: List[str]
class PythonPerformanceProfiler:
def __init__(self, enable_memory_profiling=True, enable_line_profiling=True):
self.enable_memory_profiling = enable_memory_profiling
self.enable_line_profiling = enable_line_profiling
self.metrics_history = []
self.profiler = None
@contextmanager
def profile_context(self, description=""):
"""Context manager for profiling code blocks"""
start_time = time.time()
start_cpu = psutil.cpu_percent()
start_memory = psutil.virtual_memory().percent
# Start profiler
self.profiler = cProfile.Profile()
self.profiler.enable()
try:
yield self
finally:
self.profiler.disable()
end_time = time.time()
end_cpu = psutil.cpu_percent()
end_memory = psutil.virtual_memory().percent
execution_time = end_time - start_time
avg_cpu = (start_cpu + end_cpu) / 2
memory_diff = end_memory - start_memory
print(f"Performance Summary - {description}:")
print(f" Execution Time: {execution_time:.3f}s")
print(f" CPU Usage: {avg_cpu:.1f}%")
print(f" Memory Change: {memory_diff:+.1f}%")
def profile_function(self, func: Callable, *args, **kwargs) -> PerformanceMetrics:
"""Profile a specific function with comprehensive metrics"""
# Memory profiling setup
if self.enable_memory_profiling:
memory_usage = []
def monitor_memory():
while getattr(threading.current_thread(), "monitoring", True):
memory_usage.append(psutil.virtual_memory().percent)
time.sleep(0.1)
monitor_thread = threading.Thread(target=monitor_memory)
monitor_thread.monitoring = True
monitor_thread.start()
# CPU profiling
profiler = cProfile.Profile()
start_time = time.time()
profiler.enable()
try:
result = func(*args, **kwargs)
finally:
profiler.disable()
end_time = time.time()
execution_time = end_time - start_time
# Stop memory monitoring
if self.enable_memory_profiling:
monitor_thread.monitoring = False
monitor_thread.join()
# Analyze profiling results
s = io.StringIO()
stats = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
stats.print_stats()
# Extract hotspots
hotspots = self.extract_hotspots(stats)
# Generate recommendations
recommendations = self.generate_recommendations(stats, execution_time, memory_usage if self.enable_memory_profiling else [])
metrics = PerformanceMetrics(
execution_time=execution_time,
cpu_usage=psutil.cpu_percent(),
memory_usage=psutil.virtual_memory().percent,
memory_peak=max(memory_usage) if memory_usage else 0,
function_calls=stats.total_calls,
hotspots=hotspots,
recommendations=recommendations
)
self.metrics_history.append(metrics)
return metrics
def extract_hotspots(self, stats: pstats.Stats) -> List[Dict[str, Any]]:
"""Extract performance hotspots from profiling data"""
hotspots = []
# Get top time-consuming functions
for func, (cc, nc, tt, ct, callers) in stats.stats.items():
if ct > 0.01: # Only include functions taking > 10ms
hotspot = {
'function': f"{func[0]}:{func[1]}({func[2]})",
'call_count': cc,
'total_time': tt,
'cumulative_time': ct,
'avg_time_per_call': ct / cc if cc > 0 else 0,
'percentage_of_total': (ct / stats.total_tt) * 100 if stats.total_tt > 0 else 0
}
hotspots.append(hotspot)
# Sort by cumulative time
hotspots.sort(key=lambda x: x['cumulative_time'], reverse=True)
return hotspots[:10] # Top 10 hotspots
def generate_recommendations(self, stats: pstats.Stats, execution_time: float, memory_usage: List[float]) -> List[str]:
"""Generate performance optimization recommendations"""
recommendations = []
# Execution time recommendations
if execution_time > 5.0:
recommendations.append("Consider algorithm optimization - execution time is high")
if execution_time > 1.0:
recommendations.append("Implement caching for frequently computed values")
# Memory recommendations
if memory_usage and max(memory_usage) - min(memory_usage) > 20:
recommendations.append("High memory usage variation detected - check for memory leaks")
# Function call recommendations
if stats.total_calls > 100000:
recommendations.append("High number of function calls - consider function inlining or optimization")
# Check for I/O operations
for func in stats.stats.keys():
func_name = func[2].lower()
if 'read' in func_name or 'write' in func_name or 'open' in func_name:
recommendations.append("I/O operations detected - consider async I/O or connection pooling")
break
# Check for database operations
for func in stats.stats.keys():
func_name = func[2].lower()
if 'query' in func_name or 'execute' in func_name or 'fetch' in func_name:
recommendations.append("Database operations detected - optimize queries and use connection pooling")
break
return recommendations
def benchmark_comparison(self, functions: Dict[str, Callable], *args, **kwargs) -> pd.DataFrame:
"""Compare performance of multiple function implementations"""
results = []
for name, func in functions.items():
print(f"Benchmarking {name}...")
metrics = self.profile_function(func, *args, **kwargs)
results.append({
'function': name,
'execution_time': metrics.execution_time,
'memory_peak': metrics.memory_peak,
'function_calls': metrics.function_calls,
'cpu_usage': metrics.cpu_usage
})
df = pd.DataFrame(results)
# Add relative performance
fastest = df['execution_time'].min()
df['relative_speed'] = df['execution_time'] / fastest
return df.sort_values('execution_time')
def continuous_monitoring(self, func: Callable, interval: int = 60, duration: int = 3600):
"""Continuously monitor function performance over time"""
start_time = time.time()
measurements = []
while time.time() - start_time < duration:
measurement_start = time.time()
try:
metrics = self.profile_function(func)
measurements.append({
'timestamp': measurement_start,
'execution_time': metrics.execution_time,
'cpu_usage': metrics.cpu_usage,
'memory_usage': metrics.memory_usage
})
print(f"Measurement at {time.strftime('%H:%M:%S')}: "
f"{metrics.execution_time:.3f}s, "
f"CPU: {metrics.cpu_usage:.1f}%, "
f"Memory: {metrics.memory_usage:.1f}%")
except Exception as e:
print(f"Monitoring error: {e}")
time.sleep(max(0, interval - (time.time() - measurement_start)))
return pd.DataFrame(measurements)
def generate_report(self, output_file: str = "performance_report.html"):
"""Generate comprehensive performance report"""
if not self.metrics_history:
print("No performance data available for report generation")
return
# Create visualizations
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# Execution time over runs
execution_times = [m.execution_time for m in self.metrics_history]
axes[0, 0].plot(execution_times)
axes[0, 0].set_title('Execution Time Over Runs')
axes[0, 0].set_ylabel('Time (seconds)')
# Memory usage over runs
memory_usage = [m.memory_usage for m in self.metrics_history]
axes[0, 1].plot(memory_usage)
axes[0, 1].set_title('Memory Usage Over Runs')
axes[0, 1].set_ylabel('Memory %')
# Function calls over runs
function_calls = [m.function_calls for m in self.metrics_history]
axes[1, 0].plot(function_calls)
axes[1, 0].set_title('Function Calls Over Runs')
axes[1, 0].set_ylabel('Call Count')
# CPU usage over runs
cpu_usage = [m.cpu_usage for m in self.metrics_history]
axes[1, 1].plot(cpu_usage)
axes[1, 1].set_title('CPU Usage Over Runs')
axes[1, 1].set_ylabel('CPU %')
plt.tight_layout()
plt.savefig('performance_metrics.png')
plt.close()
# Generate HTML report
html_content = self.generate_html_report()
with open(output_file, 'w') as f:
f.write(html_content)
print(f"Performance report generated: {output_file}")
def generate_html_report(self) -> str:
"""Generate HTML performance report"""
latest_metrics = self.metrics_history[-1] if self.metrics_history else None
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>Performance Analysis Report</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.metric {{ display: inline-block; margin: 10px; padding: 15px;
border-radius: 5px; min-width: 150px; text-align: center; }}
.good {{ background-color: #d4edda; color: #155724; }}
.warning {{ background-color: #fff3cd; color: #856404; }}
.danger {{ background-color: #f8d7da; color: #721c24; }}
table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #f2f2f2; }}
.hotspot {{ margin: 10px 0; padding: 10px; background-color: #f8f9fa; border-radius: 5px; }}
</style>
</head>
<body>
<h1>Performance Analysis Report</h1>
<div>
<div class="metric {'good' if latest_metrics and latest_metrics.execution_time < 1 else 'warning' if latest_metrics and latest_metrics.execution_time < 5 else 'danger'}">
<h3>Execution Time</h3>
<p>{latest_metrics.execution_time:.3f}s</p>
</div>
<div class="metric {'good' if latest_metrics and latest_metrics.memory_usage < 70 else 'warning' if latest_metrics and latest_metrics.memory_usage < 85 else 'danger'}">
<h3>Memory Usage</h3>
<p>{latest_metrics.memory_usage:.1f}%</p>
</div>
<div class="metric {'good' if latest_metrics and latest_metrics.cpu_usage < 50 else 'warning' if latest_metrics and latest_metrics.cpu_usage < 80 else 'danger'}">
<h3>CPU Usage</h3>
<p>{latest_metrics.cpu_usage:.1f}%</p>
</div>
</div>
<h2>Performance Hotspots</h2>
{''.join([f'<div class="hotspot"><strong>{hotspot["function"]}</strong><br>Cumulative Time: {hotspot["cumulative_time"]:.3f}s ({hotspot["percentage_of_total"]:.1f}%)<br>Calls: {hotspot["call_count"]}</div>' for hotspot in (latest_metrics.hotspots[:5] if latest_metrics else [])])}
<h2>Recommendations</h2>
<ul>
{''.join([f'<li>{rec}</li>' for rec in (latest_metrics.recommendations if latest_metrics else [])])}
</ul>
<img src="performance_metrics.png" alt="Performance Metrics Chart" style="max-width: 100%;">
</body>
</html>
""" if latest_metrics else "<html><body><h1>No performance data available</h1></body></html>"
return html
# Usage Examples
def example_slow_function():
"""Example function with performance issues"""
# Simulate CPU-intensive work
total = 0
for i in range(1000000):
total += i * i
# Simulate memory allocation
big_list = [i for i in range(100000)]
# Simulate I/O operation
time.sleep(0.1)
return total
def example_optimized_function():
"""Optimized version of the above function"""
# Use built-in functions and mathematical formula
n = 1000000
total = (n * (n - 1) * (2 * n - 1)) // 6
# Avoid unnecessary memory allocation
# Use generator instead of list comprehension when possible
return total
# CLI Usage
def main():
profiler = PythonPerformanceProfiler()
print("=== Profiling Slow Function ===")
metrics1 = profiler.profile_function(example_slow_function)
print("\n=== Profiling Optimized Function ===")
metrics2 = profiler.profile_function(example_optimized_function)
print("\n=== Performance Comparison ===")
comparison = profiler.benchmark_comparison({
'slow_version': example_slow_function,
'optimized_version': example_optimized_function
})
print(comparison.to_string(index=False))
profiler.generate_report()
if __name__ == "__main__":
main()
JavaScript/Node.js Performance Profiler
// performance-profiler.js
const { performance, PerformanceObserver } = require('perf_hooks');
const v8 = require('v8');
const fs = require('fs');
const os = require('os');
class JavaScriptPerformanceProfiler {
constructor(options = {}) {
this.options = {
enableGC: options.enableGC ?? true,
enableHeapSnapshot: options.enableHeapSnapshot ?? false,
enableCPUProfile: options.enableCPUProfile ?? false,
sampleInterval: options.sampleInterval ?? 100,
...options
};
this.metrics = [];
this.performanceObserver = null;
this.startTime = null;
this.gcMetrics = [];
this.setupPerformanceObserver();
}
setupPerformanceObserver() {
this.performanceObserver = new PerformanceObserver((list) => {
for (const entry of list.getEntries()) {
if (entry.entryType === 'gc') {
this.gcMetrics.push({
timestamp: entry.startTime,
duration: entry.duration,
kind: entry.kind,
flags: entry.flags
});
}
}
});
this.performanceObserver.observe({ entryTypes: ['gc', 'function', 'http2'] });
}
async profileFunction(fn, ...args) {
const startTime = performance.now();
const startMemory = process.memoryUsage();
const startCPU = process.cpuUsage();
// Enable CPU profiler if requested
if (this.options.enableCPUProfile && global.gc) {
global.gc(); // Force garbage collection for cleaner measurement
}
let result;
let error;
try {
// Mark function start
performance.mark('function-start');
result = await fn(...args);
// Mark function end
performance.mark('function-end');
performance.measure('function-execution', 'function-start', 'function-end');
} catch (err) {
error = err;
}
const endTime = performance.now();
const endMemory = process.memoryUsage();
const endCPU = process.cpuUsage(startCPU);
const metrics = {
executionTime: endTime - startTime,
cpuUsage: {
user: endCPU.user / 1000, // Convert to milliseconds
system: endCPU.system / 1000
},
memoryUsage: {
start: startMemory,
end: endMemory,
delta: {
rss: endMemory.rss - startMemory.rss,
heapUsed: endMemory.heapUsed - startMemory.heapUsed,
heapTotal: endMemory.heapTotal - startMemory.heapTotal,
external: endMemory.external - startMemory.external
}
},
gcActivity: this.gcMetrics.filter(gc => gc.timestamp >= startTime && gc.timestamp <= endTime),
v8HeapStats: v8.getHeapStatistics(),
error
};
this.metrics.push(metrics);
// Generate recommendations
metrics.recommendations = this.generateRecommendations(metrics);
return { result, metrics };
}
generateRecommendations(metrics) {
const recommendations = [];
// Execution time recommendations
if (metrics.executionTime > 1000) {
recommendations.push('High execution time detected - consider optimization');
}
// Memory recommendations
const heapDelta = metrics.memoryUsage.delta.heapUsed;
if (heapDelta > 50 * 1024 * 1024) { // 50MB
recommendations.push('High memory allocation detected - check for memory leaks');
}
if (metrics.memoryUsage.end.heapUsed / metrics.memoryUsage.end.heapTotal > 0.9) {
recommendations.push('Heap usage is high - consider garbage collection tuning');
}
// CPU recommendations
const totalCPU = metrics.cpuUsage.user + metrics.cpuUsage.system;
if (totalCPU > metrics.executionTime * 2) {
recommendations.push('High CPU usage detected - profile for hot functions');
}
// GC recommendations
if (metrics.gcActivity.length > 10) {
recommendations.push('Frequent garbage collection detected - optimize object creation');
}
const longGC = metrics.gcActivity.filter(gc => gc.duration > 10);
if (longGC.length > 0) {
recommendations.push('Long GC pauses detected - consider heap size adjustment');
}
return recommendations;
}
async benchmarkComparison(functions, iterations = 1000, ...args) {
const results = {};
for (const [name, fn] of Object.entries(functions)) {
console.log(`Benchmarking ${name}...`);
const measurements = [];
let totalTime = 0;
let errors = 0;
for (let i = 0; i < iterations; i++) {
try {
const { metrics } = await this.profileFunction(fn, ...args);
measurements.push(metrics.executionTime);
totalTime += metrics.executionTime;
} catch (error) {
errors++;
}
// Progress indicator
if ((i + 1) % Math.max(1, Math.floor(iterations / 10)) === 0) {
process.stdout.write(`${Math.round(((i + 1) / iterations) * 100)}% `);
}
}
console.log(); // New line
// Calculate statistics
measurements.sort((a, b) => a - b);
const mean = totalTime / measurements.length;
const median = measurements[Math.floor(measurements.length / 2)];
const p95 = measurements[Math.floor(measurements.length * 0.95)];
const p99 = measurements[Math.floor(measurements.length * 0.99)];
const min = measurements[0];
const max = measurements[measurements.length - 1];
results[name] = {
iterations: measurements.length,
errors,
mean,
median,
min,
max,
p95,
p99,
standardDeviation: Math.sqrt(
measurements.reduce((sum, time) => sum + Math.pow(time - mean, 2), 0) / measurements.length
)
};
}
return results;
}
async loadTest(fn, options = {}) {
const {
concurrency = 10,
duration = 30000, // 30 seconds
rampUp = 5000, // 5 seconds
...fnArgs
} = options;
console.log(`Starting load test: ${concurrency} concurrent users for ${duration}ms`);
const results = {
startTime: Date.now(),
endTime: null,
totalRequests: 0,
successfulRequests: 0,
errors: [],
responseTimes: [],
concurrentUsers: 0
};
const workers = [];
const rampUpInterval = rampUp / concurrency;
// Gradually ramp up users
for (let i = 0; i < concurrency; i++) {
setTimeout(() => {
const worker = this.createLoadTestWorker(fn, results, fnArgs);
workers.push(worker);
results.concurrentUsers++;
}, i * rampUpInterval);
}
// Stop test after duration
setTimeout(() => {
workers.forEach(worker => {
if (worker.stop) worker.stop();
});
results.endTime = Date.now();
results.duration = results.endTime - results.startTime;
this.analyzeLoadTestResults(results);
}, duration);
return new Promise(resolve => {
setTimeout(() => resolve(results), duration + 1000);
});
}
createLoadTestWorker(fn, results, args) {
let running = true;
const worker = {
stop: () => { running = false; }
};
(async () => {
while (running) {
const startTime = performance.now();
try {
await fn(...args);
const endTime = performance.now();
results.totalRequests++;
results.successfulRequests++;
results.responseTimes.push(endTime - startTime);
} catch (error) {
results.totalRequests++;
results.errors.push({
timestamp: Date.now(),
error: error.message
});
}
// Small delay to prevent overwhelming
await new Promise(resolve => setTimeout(resolve, 10));
}
})();
return worker;
}
analyzeLoadTestResults(results) {
const responseTimes = results.responseTimes.sort((a, b) => a - b);
const successRate = (results.successfulRequests / results.totalRequests) * 100;
const throughput = (results.totalRequests / results.duration) * 1000; // requests per second
console.log('\n📊 Load Test Results:');
console.log(`Total Requests: ${results.totalRequests}`);
console.log(`Successful Requests: ${results.successfulRequests} (${successRate.toFixed(2)}%)`);
console.log(`Errors: ${results.errors.length}`);
console.log(`Throughput: ${throughput.toFixed(2)} req/s`);
if (responseTimes.length > 0) {
console.log(`Response Times:`);
console.log(` Mean: ${(responseTimes.reduce((a, b) => a + b, 0) / responseTimes.length).toFixed(2)}ms`);
console.log(` Median: ${responseTimes[Math.floor(responseTimes.length / 2)].toFixed(2)}ms`);
console.log(` 95th Percentile: ${responseTimes[Math.floor(responseTimes.length * 0.95)].toFixed(2)}ms`);
console.log(` 99th Percentile: ${responseTimes[Math.floor(responseTimes.length * 0.99)].toFixed(2)}ms`);
console.log(` Min: ${responseTimes[0].toFixed(2)}ms`);
console.log(` Max: ${responseTimes[responseTimes.length - 1].toFixed(2)}ms`);
}
// Error analysis
if (results.errors.length > 0) {
const errorTypes = {};
results.errors.forEach(error => {
errorTypes[error.error] = (errorTypes[error.error] || 0) + 1;
});
console.log('\nError Breakdown:');
Object.entries(errorTypes).forEach(([error, count]) => {
console.log(` ${error}: ${count}`);
});
}
}
generateReport(outputFile = 'performance-report.json') {
const report = {
summary: {
totalTests: this.metrics.length,
averageExecutionTime: this.metrics.reduce((sum, m) => sum + m.executionTime, 0) / this.metrics.length,
totalMemoryDelta: this.metrics.reduce((sum, m) => sum + m.memoryUsage.delta.heapUsed, 0),
totalGCEvents: this.gcMetrics.length,
nodeVersion: process.version,
platform: os.platform(),
arch: os.arch(),
cpus: os.cpus().length
},
metrics: this.metrics,
gcActivity: this.gcMetrics,
recommendations: this.getGlobalRecommendations()
};
fs.writeFileSync(outputFile, JSON.stringify(report, null, 2));
console.log(`Performance report saved to ${outputFile}`);
return report;
}
getGlobalRecommendations() {
const recommendations = new Set();
this.metrics.forEach(metric => {
metric.recommendations?.forEach(rec => recommendations.add(rec));
});
// Add global recommendations based on overall patterns
const avgExecutionTime = this.metrics.reduce((sum, m) => sum + m.executionTime, 0) / this.metrics.length;
if (avgExecutionTime > 500) {
recommendations.add('Consider implementing caching mechanisms');
recommendations.add('Profile hot code paths for optimization opportunities');
}
const totalGCTime = this.gcMetrics.reduce((sum, gc) => sum + gc.duration, 0);
if (totalGCTime > 1000) {
recommendations.add('High garbage collection overhead - optimize memory usage patterns');
}
return Array.from(recommendations);
}
}
// Example usage and tests
async function exampleSlowFunction(size = 1000000) {
// CPU-intensive operation
let sum = 0;
for (let i = 0; i < size; i++) {
sum += Math.sqrt(i);
}
// Memory allocation
const array = new Array(size).fill(0).map((_, i) => ({ id: i, value: Math.random() }));
// Simulate async operation
await new Promise(resolve => setTimeout(resolve, 100));
return { sum, count: array.length };
}
async function exampleOptimizedFunction(size = 1000000) {
// Optimized version using built-in functions
const sum = Array.from({ length: size }, (_, i) => Math.sqrt(i))
.reduce((acc, val) => acc + val, 0);
// More memory-efficient approach
const count = size;
// Same async operation
await new Promise(resolve => setTimeout(resolve, 100));
return { sum, count };
}
// CLI usage
async function main() {
const profiler = new JavaScriptPerformanceProfiler({
enableGC: true,
enableCPUProfile: true
});
console.log('🔍 Starting JavaScript Performance Analysis...\n');
// Single function profiling
console.log('=== Profiling Slow Function ===');
const { metrics: slowMetrics } = await profiler.profileFunction(exampleSlowFunction, 100000);
console.log(`Execution Time: ${slowMetrics.executionTime.toFixed(2)}ms`);
console.log(`Memory Delta: ${(slowMetrics.memoryUsage.delta.heapUsed / 1024 / 1024).toFixed(2)}MB`);
console.log(`Recommendations: ${slowMetrics.recommendations.join(', ')}\n`);
console.log('=== Profiling Optimized Function ===');
const { metrics: fastMetrics } = await profiler.profileFunction(exampleOptimizedFunction, 100000);
console.log(`Execution Time: ${fastMetrics.executionTime.toFixed(2)}ms`);
console.log(`Memory Delta: ${(fastMetrics.memoryUsage.delta.heapUsed / 1024 / 1024).toFixed(2)}MB`);
console.log(`Recommendations: ${fastMetrics.recommendations.join(', ')}\n`);
// Benchmark comparison
console.log('=== Benchmark Comparison ===');
const comparison = await profiler.benchmarkComparison({
'slow_version': exampleSlowFunction,
'optimized_version': exampleOptimizedFunction
}, 10, 50000);
Object.entries(comparison).forEach(([name, stats]) => {
console.log(`${name}:`);
console.log(` Mean: ${stats.mean.toFixed(2)}ms`);
console.log(` Median: ${stats.median.toFixed(2)}ms`);
console.log(` 95th Percentile: ${stats.p95.toFixed(2)}ms`);
console.log(` Standard Deviation: ${stats.standardDeviation.toFixed(2)}ms\n`);
});
// Generate report
profiler.generateReport();
console.log('Performance analysis completed! 🎯');
}
if (require.main === module) {
// Enable garbage collection tracking
if (global.gc) {
global.gc();
}
main().catch(console.error);
}
module.exports = JavaScriptPerformanceProfiler;
Load Testing Framework
K6 Load Testing Configuration
// k6-load-test.js
import http from 'k6/http';
import { check, group, sleep } from 'k6';
import { Rate, Trend, Counter } from 'k6/metrics';
// Custom metrics
const failureRate = new Rate('failed_requests');
const responseTimeTrend = new Trend('response_time_custom');
const requestCounter = new Counter('total_requests');
// Test configuration
export const options = {
stages: [
{ duration: '2m', target: 100 }, // Ramp up to 100 users
{ duration: '5m', target: 100 }, // Stay at 100 users
{ duration: '2m', target: 200 }, // Ramp up to 200 users
{ duration: '5m', target: 200 }, // Stay at 200 users
{ duration: '2m', target: 0 }, // Ramp down to 0 users
],
thresholds: {
http_req_duration: ['p(95)<2000'], // 95% of requests must complete below 2s
http_req_failed: ['rate<0.01'], // Error rate must be below 1%
failed_requests: ['rate<0.01'], // Custom failure rate below 1%
},
};
// Test scenarios
export default function () {
group('User Journey - Authentication', () => {
// Login
const loginResponse = http.post('https://api.example.com/auth/login', {
username: 'testuser@example.com',
password: 'testpassword123'
}, {
headers: { 'Content-Type': 'application/json' },
tags: { endpoint: 'login' }
});
const loginSuccess = check(loginResponse, {
'login status is 200': (r) => r.status === 200,
'login response time < 1s': (r) => r.timings.duration < 1000,
'login returns token': (r) => r.json().token !== undefined,
});
failureRate.add(!loginSuccess);
responseTimeTrend.add(loginResponse.timings.duration);
requestCounter.add(1);
if (!loginSuccess) return;
const token = loginResponse.json().token;
const authHeaders = {
'Authorization': `Bearer ${token}`,
'Content-Type': 'application/json'
};
sleep(1); // Think time
// Get user profile
group('User Profile Operations', () => {
const profileResponse = http.get('https://api.example.com/user/profile', {
headers: authHeaders,
tags: { endpoint: 'profile' }
});
const profileSuccess = check(profileResponse, {
'profile status is 200': (r) => r.status === 200,
'profile response time < 500ms': (r) => r.timings.duration < 500,
'profile contains user data': (r) => r.json().user !== undefined,
});
failureRate.add(!profileSuccess);
responseTimeTrend.add(profileResponse.timings.duration);
requestCounter.add(1);
sleep(0.5);
});
// Browse products
group('Product Browsing', () => {
const productsResponse = http.get('https://api.example.com/products?page=1&limit=20', {
headers: authHeaders,
tags: { endpoint: 'products' }
});
const productsSuccess = check(productsResponse, {
'products status is 200': (r) => r.status === 200,
'products response time < 1s': (r) => r.timings.duration < 1000,
'products returned': (r) => r.json().products.length > 0,
});
failureRate.add(!productsSuccess);
responseTimeTrend.add(productsResponse.timings.duration);
requestCounter.add(1);
sleep(2); // Browsing think time
// Get product details (random product)
if (productsSuccess && productsResponse.json().products.length > 0) {
const products = productsResponse.json().products;
const randomProduct = products[Math.floor(Math.random() * products.length)];
const productDetailResponse = http.get(`https://api.example.com/products/${randomProduct.id}`, {
headers: authHeaders,
tags: { endpoint: 'product_detail' }
});
const detailSuccess = check(productDetailResponse, {
'product detail status is 200': (r) => r.status === 200,
'product detail response time < 800ms': (r) => r.timings.duration < 800,
});
failureRate.add(!detailSuccess);
responseTimeTrend.add(productDetailResponse.timings.duration);
requestCounter.add(1);
}
sleep(1);
});
// Cart operations
group('Shopping Cart Operations', () => {
// Add to cart
const addToCartResponse = http.post('https://api.example.com/cart/add', {
productId: '12345',
quantity: 2
}, {
headers: authHeaders,
tags: { endpoint: 'add_to_cart' }
});
const addSuccess = check(addToCartResponse, {
'add to cart status is 200': (r) => r.status === 200,
'add to cart response time < 1s': (r) => r.timings.duration < 1000,
});
failureRate.add(!addSuccess);
responseTimeTrend.add(addToCartResponse.timings.duration);
requestCounter.add(1);
sleep(1);
// View cart
const cartResponse = http.get('https://api.example.com/cart', {
headers: authHeaders,
tags: { endpoint: 'view_cart' }
});
const cartSuccess = check(cartResponse, {
'view cart status is 200': (r) => r.status === 200,
'view cart response time < 500ms': (r) => r.timings.duration < 500,
'cart has items': (r) => r.json().items.length > 0,
});
failureRate.add(!cartSuccess);
responseTimeTrend.add(cartResponse.timings.duration);
requestCounter.add(1);
});
sleep(1);
});
}
// Setup function (runs once per VU at the beginning)
export function setup() {
// Prepare test data, authenticate admin user, etc.
console.log('Setting up test environment...');
return { testData: 'initialized' };
}
// Teardown function (runs once per VU at the end)
export function teardown(data) {
console.log('Cleaning up test environment...');
}
Advanced Performance Monitoring Dashboard
#!/usr/bin/env python3
"""
Real-time performance monitoring dashboard using Prometheus and Grafana
"""
import time
import psutil
import requests
from prometheus_client import start_http_server, Counter, Histogram, Gauge, CollectorRegistry
from dataclasses import dataclass
from typing import Dict, List, Optional
import threading
import json
import subprocess
@dataclass
class PerformanceThreshold:
metric: str
warning: float
critical: float
operator: str = 'greater_than' # greater_than, less_than
class PerformanceMonitor:
def __init__(self, port: int = 8000, scrape_interval: int = 10):
self.port = port
self.scrape_interval = scrape_interval
self.registry = CollectorRegistry()
# Prometheus metrics
self.request_counter = Counter(
'app_requests_total',
'Total application requests',
['method', 'endpoint', 'status'],
registry=self.registry
)
self.response_time_histogram = Histogram(
'app_response_time_seconds',
'Response time in seconds',
['endpoint'],
registry=self.registry
)
self.cpu_usage_gauge = Gauge(
'system_cpu_usage_percent',
'Current CPU usage percentage',
registry=self.registry
)
self.memory_usage_gauge = Gauge(
'system_memory_usage_percent',
'Current memory usage percentage',
registry=self.registry
)
self.disk_usage_gauge = Gauge(
'system_disk_usage_percent',
'Current disk usage percentage',
['device'],
registry=self.registry
)
self.active_connections_gauge = Gauge(
'app_active_connections',
'Number of active connections',
registry=self.registry
)
# Performance thresholds
self.thresholds = [
PerformanceThreshold('cpu_usage', 70, 90),
PerformanceThreshold('memory_usage', 80, 95),
PerformanceThreshold('response_time', 2.0, 5.0),
PerformanceThreshold('error_rate', 0.05, 0.10)
]
self.alerts = []
self.monitoring = False
def start_monitoring(self):
"""Start the performance monitoring server"""
# Start Prometheus metrics server
start_http_server(self.port, registry=self.registry)
print(f"🔍 Performance monitoring server started on port {self.port}")
# Start monitoring thread
self.monitoring = True
monitor_thread = threading.Thread(target=self._monitoring_loop)
monitor_thread.daemon = True
monitor_thread.start()
return monitor_thread
def _monitoring_loop(self):
"""Main monitoring loop"""
while self.monitoring:
try:
self._collect_system_metrics()
self._check_thresholds()
time.sleep(self.scrape_interval)
except Exception as e:
print(f"Monitoring error: {e}")
def _collect_system_metrics(self):
"""Collect system performance metrics"""
# CPU usage
cpu_percent = psutil.cpu_percent(interval=1)
self.cpu_usage_gauge.set(cpu_percent)
# Memory usage
memory = psutil.virtual_memory()
self.memory_usage_gauge.set(memory.percent)
# Disk usage
for partition in psutil.disk_partitions():
try:
disk_usage = psutil.disk_usage(partition.mountpoint)
self.disk_usage_gauge.labels(device=partition.device).set(disk_usage.percent)
except PermissionError:
continue
# Network connections (approximate active connections)
connections = len(psutil.net_connections())
self.active_connections_gauge.set(connections)
def record_request(self, method: str, endpoint: str, status: int, response_time: float):
"""Record a request metric"""
self.request_counter.labels(method=method, endpoint=endpoint, status=str(status)).inc()
self.response_time_histogram.labels(endpoint=endpoint).observe(response_time)
def _check_thresholds(self):
"""Check performance thresholds and generate alerts"""
current_time = time.time()
# Get current metric values
cpu_usage = psutil.cpu_percent()
memory_usage = psutil.virtual_memory().percent
for threshold in self.thresholds:
if threshold.metric == 'cpu_usage':
current_value = cpu_usage
elif threshold.metric == 'memory_usage':
current_value = memory_usage
else:
continue # Skip metrics we don't have current values for
# Check threshold
if threshold.operator == 'greater_than':
if current_value >= threshold.critical:
self._generate_alert(threshold.metric, 'CRITICAL', current_value, threshold.critical)
elif current_value >= threshold.warning:
self._generate_alert(threshold.metric, 'WARNING', current_value, threshold.warning)
def _generate_alert(self, metric: str, severity: str, current_value: float, threshold_value: float):
"""Generate performance alert"""
alert = {
'timestamp': time.time(),
'metric': metric,
'severity': severity,
'current_value': current_value,
'threshold_value': threshold_value,
'message': f"{metric} is {current_value:.2f}, exceeding {severity.lower()} threshold of {threshold_value:.2f}"
}
self.alerts.append(alert)
print(f"🚨 ALERT [{severity}]: {alert['message']}")
# Keep only recent alerts (last 100)
self.alerts = self.alerts[-100:]
def get_performance_summary(self) -> Dict:
"""Get current performance summary"""
return {
'timestamp': time.time(),
'system': {
'cpu_usage': psutil.cpu_percent(),
'memory_usage': psutil.virtual_memory().percent,
'disk_usage': {p.device: psutil.disk_usage(p.mountpoint).percent
for p in psutil.disk_partitions()
if not p.mountpoint.startswith('/sys')},
'active_connections': len(psutil.net_connections())
},
'alerts': self.alerts[-10:], # Recent alerts
'thresholds': [
{
'metric': t.metric,
'warning': t.warning,
'critical': t.critical
} for t in self.thresholds
]
}
def generate_grafana_dashboard(self) -> str:
"""Generate Grafana dashboard JSON configuration"""
dashboard = {
"dashboard": {
"title": "Application Performance Dashboard",
"panels": [
{
"title": "CPU Usage",
"type": "graph",
"targets": [
{
"expr": "system_cpu_usage_percent",
"legendFormat": "CPU %"
}
],
"yAxes": [{"max": 100, "min": 0}],
"thresholds": [
{"value": 70, "colorMode": "critical", "op": "gt"},
{"value": 90, "colorMode": "critical", "op": "gt"}
]
},
{
"title": "Memory Usage",
"type": "graph",
"targets": [
{
"expr": "system_memory_usage_percent",
"legendFormat": "Memory %"
}
],
"yAxes": [{"max": 100, "min": 0}],
"thresholds": [
{"value": 80, "colorMode": "warning", "op": "gt"},
{"value": 95, "colorMode": "critical", "op": "gt"}
]
},
{
"title": "Request Rate",
"type": "graph",
"targets": [
{
"expr": "rate(app_requests_total[5m])",
"legendFormat": "{{method}} {{endpoint}}"
}
]
},
{
"title": "Response Time Percentiles",
"type": "graph",
"targets": [
{
"expr": "histogram_quantile(0.50, rate(app_response_time_seconds_bucket[5m]))",
"legendFormat": "50th percentile"
},
{
"expr": "histogram_quantile(0.95, rate(app_response_time_seconds_bucket[5m]))",
"legendFormat": "95th percentile"
},
{
"expr": "histogram_quantile(0.99, rate(app_response_time_seconds_bucket[5m]))",
"legendFormat": "99th percentile"
}
]
}
]
}
}
return json.dumps(dashboard, indent=2)
def export_metrics_config(self) -> str:
"""Export Prometheus configuration"""
config = """
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'application-metrics'
static_configs:
- targets: ['localhost:{port}']
scrape_interval: {interval}s
metrics_path: /metrics
""".format(port=self.port, interval=self.scrape_interval)
return config.strip()
# Example Flask integration
def create_flask_monitoring_middleware(monitor: PerformanceMonitor):
"""Create Flask middleware for automatic request monitoring"""
from flask import request, g
import time
def before_request():
g.start_time = time.time()
def after_request(response):
response_time = time.time() - g.start_time
monitor.record_request(
method=request.method,
endpoint=request.endpoint or 'unknown',
status=response.status_code,
response_time=response_time
)
return response
return before_request, after_request
# CLI usage
def main():
monitor = PerformanceMonitor(port=8000, scrape_interval=10)
# Start monitoring
monitor_thread = monitor.start_monitoring()
print("Performance monitoring dashboard available at:")
print(f" Metrics: http://localhost:8000/metrics")
print(f" Thresholds: {len(monitor.thresholds)} configured")
print("\nPress Ctrl+C to stop monitoring...")
try:
# Keep main thread alive
while True:
time.sleep(1)
# Print periodic summary
if int(time.time()) % 60 == 0: # Every minute
summary = monitor.get_performance_summary()
print(f"\n📊 Performance Summary:")
print(f" CPU: {summary['system']['cpu_usage']:.1f}%")
print(f" Memory: {summary['system']['memory_usage']:.1f}%")
print(f" Connections: {summary['system']['active_connections']}")
if summary['alerts']:
print(f" Recent Alerts: {len(summary['alerts'])}")
except KeyboardInterrupt:
print("\nStopping performance monitoring...")
monitor.monitoring = False
monitor_thread.join()
if __name__ == "__main__":
main()
This comprehensive Performance Profiler agent provides extensive performance analysis capabilities across multiple languages and frameworks. It includes detailed profiling tools, load testing configurations, real-time monitoring dashboards, and actionable optimization recommendations that development teams can immediately implement to improve their application performance.