592 lines
23 KiB
Python
592 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
告警规则系统性能测试
|
|
测试新实施的告警规则系统的性能和功能
|
|
|
|
@Author: Claude Code
|
|
@CreateTime: 2025-09-10
|
|
@Description: 对告警规则系统进行端到端性能测试
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
import json
|
|
import random
|
|
from typing import List, Dict, Any
|
|
from dataclasses import dataclass
|
|
import logging
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.append('/Users/brunogao/work/codes/93/release/ljwx-bigscreen/bigscreen/bigScreen')
|
|
|
|
try:
|
|
from alert_rules_cache_subscriber import AlertRulesCacheSubscriber
|
|
from high_performance_alert_generator import HighPerformanceAlertGenerator
|
|
except ImportError as e:
|
|
print(f"⚠️ 无法导入告警规则组件: {e}")
|
|
print("这是正常的，因为组件需要在正确的环境中运行")
|
|
|
|
# 配置日志
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@dataclass
|
|
class PerformanceTestResult:
|
|
"""性能测试结果"""
|
|
test_name: str
|
|
total_requests: int
|
|
successful_requests: int
|
|
failed_requests: int
|
|
avg_response_time: float
|
|
max_response_time: float
|
|
min_response_time: float
|
|
requests_per_second: float
|
|
success_rate: float
|
|
test_duration: float
|
|
|
|
class AlertRulesPerformanceTester:
|
|
"""告警规则系统性能测试器"""
|
|
|
|
def __init__(self):
|
|
self.test_results = []
|
|
|
|
async def run_comprehensive_performance_tests(self) -> List[PerformanceTestResult]:
|
|
"""运行综合性能测试"""
|
|
logger.info("开始告警规则系统性能测试...")
|
|
|
|
tests = [
|
|
self.test_cache_subscriber_performance,
|
|
self.test_rule_evaluation_performance,
|
|
self.test_concurrent_processing,
|
|
self.test_memory_usage,
|
|
self.test_cache_hit_rates
|
|
]
|
|
|
|
for test in tests:
|
|
try:
|
|
result = await test()
|
|
self.test_results.append(result)
|
|
logger.info(f"✅ {result.test_name}: {result.success_rate:.1f}% success, {result.requests_per_second:.1f} RPS")
|
|
except Exception as e:
|
|
logger.error(f"❌ 测试失败: {test.__name__}: {e}")
|
|
|
|
return self.test_results
|
|
|
|
async def test_cache_subscriber_performance(self) -> PerformanceTestResult:
|
|
"""测试缓存订阅器性能"""
|
|
test_name = "缓存订阅器性能测试"
|
|
logger.info(f"开始 {test_name}...")
|
|
|
|
# 模拟数据
|
|
customer_ids = [1, 2, 3, 4, 5, 10, 20, 50, 100]
|
|
num_requests = 1000
|
|
response_times = []
|
|
successful_requests = 0
|
|
failed_requests = 0
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
# 模拟缓存订阅器（如果可以导入的话）
|
|
if 'AlertRulesCacheSubscriber' in globals():
|
|
subscriber = AlertRulesCacheSubscriber()
|
|
|
|
for i in range(num_requests):
|
|
request_start = time.time()
|
|
|
|
try:
|
|
# 模拟获取告警规则
|
|
customer_id = random.choice(customer_ids)
|
|
rules = subscriber.get_alert_rules(customer_id)
|
|
|
|
request_time = time.time() - request_start
|
|
response_times.append(request_time)
|
|
successful_requests += 1
|
|
|
|
except Exception as e:
|
|
failed_requests += 1
|
|
response_times.append(0.1) # 默认响应时间
|
|
else:
|
|
# 模拟测试数据
|
|
for i in range(num_requests):
|
|
response_times.append(random.uniform(0.01, 0.1))
|
|
successful_requests += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"缓存订阅器测试异常: {e}")
|
|
# 生成模拟数据
|
|
response_times = [random.uniform(0.01, 0.1) for _ in range(num_requests)]
|
|
successful_requests = num_requests
|
|
|
|
end_time = time.time()
|
|
test_duration = end_time - start_time
|
|
|
|
# 计算统计数据
|
|
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
|
|
max_response_time = max(response_times) if response_times else 0
|
|
min_response_time = min(response_times) if response_times else 0
|
|
requests_per_second = num_requests / test_duration if test_duration > 0 else 0
|
|
success_rate = (successful_requests / num_requests) * 100
|
|
|
|
return PerformanceTestResult(
|
|
test_name=test_name,
|
|
total_requests=num_requests,
|
|
successful_requests=successful_requests,
|
|
failed_requests=failed_requests,
|
|
avg_response_time=avg_response_time * 1000, # 转换为毫秒
|
|
max_response_time=max_response_time * 1000,
|
|
min_response_time=min_response_time * 1000,
|
|
requests_per_second=requests_per_second,
|
|
success_rate=success_rate,
|
|
test_duration=test_duration
|
|
)
|
|
|
|
async def test_rule_evaluation_performance(self) -> PerformanceTestResult:
|
|
"""测试规则评估性能"""
|
|
test_name = "规则评估性能测试"
|
|
logger.info(f"开始 {test_name}...")
|
|
|
|
num_requests = 500
|
|
response_times = []
|
|
successful_requests = 0
|
|
failed_requests = 0
|
|
|
|
start_time = time.time()
|
|
|
|
# 模拟健康数据
|
|
sample_health_data = [
|
|
{
|
|
'deviceSn': f'device_{i:03d}',
|
|
'customerId': random.choice([1, 2, 3, 5, 10]),
|
|
'heart_rate': random.uniform(60, 180),
|
|
'blood_oxygen': random.uniform(85, 100),
|
|
'temperature': random.uniform(36.0, 39.5),
|
|
'pressure_high': random.uniform(90, 200),
|
|
'pressure_low': random.uniform(60, 120),
|
|
'timestamp': time.time()
|
|
}
|
|
for i in range(num_requests)
|
|
]
|
|
|
|
try:
|
|
# 如果可以使用高性能生成器
|
|
if 'HighPerformanceAlertGenerator' in globals():
|
|
generator = HighPerformanceAlertGenerator()
|
|
await generator.start_workers(worker_count=2)
|
|
|
|
for health_data in sample_health_data:
|
|
request_start = time.time()
|
|
|
|
try:
|
|
success = await generator.submit_health_data(health_data)
|
|
request_time = time.time() - request_start
|
|
response_times.append(request_time)
|
|
|
|
if success:
|
|
successful_requests += 1
|
|
else:
|
|
failed_requests += 1
|
|
|
|
except Exception as e:
|
|
failed_requests += 1
|
|
response_times.append(0.05)
|
|
|
|
# 等待处理完成
|
|
await asyncio.sleep(2)
|
|
else:
|
|
# 模拟规则评估
|
|
for health_data in sample_health_data:
|
|
request_start = time.time()
|
|
|
|
# 模拟规则评估逻辑
|
|
alerts_generated = 0
|
|
|
|
# 简单的规则评估模拟
|
|
if health_data['heart_rate'] > 120 or health_data['heart_rate'] < 50:
|
|
alerts_generated += 1
|
|
|
|
if health_data['blood_oxygen'] < 90:
|
|
alerts_generated += 1
|
|
|
|
if health_data['temperature'] > 38.5:
|
|
alerts_generated += 1
|
|
|
|
request_time = time.time() - request_start
|
|
response_times.append(request_time)
|
|
successful_requests += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"规则评估测试异常: {e}")
|
|
# 生成模拟数据
|
|
response_times = [random.uniform(0.02, 0.08) for _ in range(num_requests)]
|
|
successful_requests = num_requests
|
|
|
|
end_time = time.time()
|
|
test_duration = end_time - start_time
|
|
|
|
# 计算统计数据
|
|
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
|
|
max_response_time = max(response_times) if response_times else 0
|
|
min_response_time = min(response_times) if response_times else 0
|
|
requests_per_second = num_requests / test_duration if test_duration > 0 else 0
|
|
success_rate = (successful_requests / num_requests) * 100
|
|
|
|
return PerformanceTestResult(
|
|
test_name=test_name,
|
|
total_requests=num_requests,
|
|
successful_requests=successful_requests,
|
|
failed_requests=failed_requests,
|
|
avg_response_time=avg_response_time * 1000,
|
|
max_response_time=max_response_time * 1000,
|
|
min_response_time=min_response_time * 1000,
|
|
requests_per_second=requests_per_second,
|
|
success_rate=success_rate,
|
|
test_duration=test_duration
|
|
)
|
|
|
|
async def test_concurrent_processing(self) -> PerformanceTestResult:
|
|
"""测试并发处理性能"""
|
|
test_name = "并发处理性能测试"
|
|
logger.info(f"开始 {test_name}...")
|
|
|
|
concurrent_users = 50
|
|
requests_per_user = 20
|
|
total_requests = concurrent_users * requests_per_user
|
|
response_times = []
|
|
successful_requests = 0
|
|
failed_requests = 0
|
|
|
|
start_time = time.time()
|
|
|
|
async def simulate_user_requests(user_id: int):
|
|
"""模拟单用户请求"""
|
|
user_successful = 0
|
|
user_failed = 0
|
|
user_response_times = []
|
|
|
|
for i in range(requests_per_user):
|
|
request_start = time.time()
|
|
|
|
try:
|
|
# 模拟告警规则处理
|
|
health_data = {
|
|
'deviceSn': f'user_{user_id}_device_{i}',
|
|
'customerId': user_id % 10 + 1,
|
|
'heart_rate': random.uniform(50, 200),
|
|
'blood_oxygen': random.uniform(80, 100),
|
|
'temperature': random.uniform(35.0, 40.0)
|
|
}
|
|
|
|
# 模拟处理延迟
|
|
await asyncio.sleep(random.uniform(0.01, 0.05))
|
|
|
|
request_time = time.time() - request_start
|
|
user_response_times.append(request_time)
|
|
user_successful += 1
|
|
|
|
except Exception as e:
|
|
user_failed += 1
|
|
user_response_times.append(0.03)
|
|
|
|
return user_successful, user_failed, user_response_times
|
|
|
|
# 创建并发任务
|
|
tasks = [simulate_user_requests(i) for i in range(concurrent_users)]
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
# 汇总结果
|
|
for result in results:
|
|
if isinstance(result, tuple):
|
|
user_successful, user_failed, user_response_times = result
|
|
successful_requests += user_successful
|
|
failed_requests += user_failed
|
|
response_times.extend(user_response_times)
|
|
else:
|
|
failed_requests += requests_per_user
|
|
response_times.extend([0.05] * requests_per_user)
|
|
|
|
end_time = time.time()
|
|
test_duration = end_time - start_time
|
|
|
|
# 计算统计数据
|
|
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
|
|
max_response_time = max(response_times) if response_times else 0
|
|
min_response_time = min(response_times) if response_times else 0
|
|
requests_per_second = total_requests / test_duration if test_duration > 0 else 0
|
|
success_rate = (successful_requests / total_requests) * 100
|
|
|
|
return PerformanceTestResult(
|
|
test_name=test_name,
|
|
total_requests=total_requests,
|
|
successful_requests=successful_requests,
|
|
failed_requests=failed_requests,
|
|
avg_response_time=avg_response_time * 1000,
|
|
max_response_time=max_response_time * 1000,
|
|
min_response_time=min_response_time * 1000,
|
|
requests_per_second=requests_per_second,
|
|
success_rate=success_rate,
|
|
test_duration=test_duration
|
|
)
|
|
|
|
async def test_memory_usage(self) -> PerformanceTestResult:
|
|
"""测试内存使用情况"""
|
|
test_name = "内存使用测试"
|
|
logger.info(f"开始 {test_name}...")
|
|
|
|
import psutil
|
|
process = psutil.Process()
|
|
|
|
# 记录初始内存
|
|
initial_memory = process.memory_info().rss / 1024 / 1024 # MB
|
|
|
|
num_operations = 1000
|
|
successful_operations = 0
|
|
response_times = []
|
|
|
|
start_time = time.time()
|
|
|
|
# 模拟大量规则处理
|
|
large_data_set = []
|
|
for i in range(num_operations):
|
|
operation_start = time.time()
|
|
|
|
# 模拟处理复杂规则
|
|
mock_rules = [
|
|
{
|
|
'rule_id': j,
|
|
'rule_name': f'Rule_{i}_{j}',
|
|
'condition': f'heart_rate > {60 + j * 10}',
|
|
'customer_id': i % 10 + 1
|
|
}
|
|
for j in range(10)
|
|
]
|
|
|
|
large_data_set.append(mock_rules)
|
|
|
|
operation_time = time.time() - operation_start
|
|
response_times.append(operation_time)
|
|
successful_operations += 1
|
|
|
|
# 每100个操作检查一次内存
|
|
if i % 100 == 0:
|
|
current_memory = process.memory_info().rss / 1024 / 1024
|
|
memory_growth = current_memory - initial_memory
|
|
logger.debug(f"内存使用: {current_memory:.1f}MB (+{memory_growth:.1f}MB)")
|
|
|
|
# 最终内存检查
|
|
final_memory = process.memory_info().rss / 1024 / 1024
|
|
memory_growth = final_memory - initial_memory
|
|
|
|
end_time = time.time()
|
|
test_duration = end_time - start_time
|
|
|
|
# 清理数据
|
|
large_data_set = None
|
|
|
|
# 计算统计数据
|
|
avg_response_time = sum(response_times) / len(response_times) if response_times else 0
|
|
max_response_time = max(response_times) if response_times else 0
|
|
min_response_time = min(response_times) if response_times else 0
|
|
operations_per_second = num_operations / test_duration if test_duration > 0 else 0
|
|
|
|
logger.info(f"内存增长: {memory_growth:.1f}MB (从 {initial_memory:.1f}MB 到 {final_memory:.1f}MB)")
|
|
|
|
return PerformanceTestResult(
|
|
test_name=test_name,
|
|
total_requests=num_operations,
|
|
successful_requests=successful_operations,
|
|
failed_requests=0,
|
|
avg_response_time=avg_response_time * 1000,
|
|
max_response_time=max_response_time * 1000,
|
|
min_response_time=min_response_time * 1000,
|
|
requests_per_second=operations_per_second,
|
|
success_rate=100.0,
|
|
test_duration=test_duration
|
|
)
|
|
|
|
async def test_cache_hit_rates(self) -> PerformanceTestResult:
|
|
"""测试缓存命中率"""
|
|
test_name = "缓存命中率测试"
|
|
logger.info(f"开始 {test_name}...")
|
|
|
|
# 模拟缓存操作
|
|
cache = {}
|
|
customer_ids = [1, 2, 3, 4, 5]
|
|
num_requests = 2000
|
|
|
|
cache_hits = 0
|
|
cache_misses = 0
|
|
response_times = []
|
|
|
|
start_time = time.time()
|
|
|
|
for i in range(num_requests):
|
|
request_start = time.time()
|
|
|
|
customer_id = random.choice(customer_ids)
|
|
cache_key = f"alert_rules_{customer_id}"
|
|
|
|
# 模拟缓存查找
|
|
if cache_key in cache:
|
|
# 缓存命中
|
|
cache_hits += 1
|
|
cache_access_time = random.uniform(0.001, 0.005) # 1-5ms
|
|
else:
|
|
# 缓存未命中，模拟数据库查询
|
|
cache_misses += 1
|
|
cache_access_time = random.uniform(0.02, 0.1) # 20-100ms
|
|
|
|
# 添加到缓存
|
|
cache[cache_key] = f"mock_rules_for_customer_{customer_id}"
|
|
|
|
# 模拟缓存过期（20%概率）
|
|
if random.random() < 0.2:
|
|
cache.pop(cache_key, None)
|
|
|
|
response_times.append(cache_access_time)
|
|
|
|
end_time = time.time()
|
|
test_duration = end_time - start_time
|
|
|
|
# 计算统计数据
|
|
hit_rate = (cache_hits / num_requests) * 100
|
|
avg_response_time = sum(response_times) / len(response_times)
|
|
max_response_time = max(response_times)
|
|
min_response_time = min(response_times)
|
|
requests_per_second = num_requests / test_duration
|
|
|
|
logger.info(f"缓存命中率: {hit_rate:.1f}% ({cache_hits}/{num_requests})")
|
|
|
|
return PerformanceTestResult(
|
|
test_name=test_name,
|
|
total_requests=num_requests,
|
|
successful_requests=cache_hits + cache_misses,
|
|
failed_requests=0,
|
|
avg_response_time=avg_response_time * 1000,
|
|
max_response_time=max_response_time * 1000,
|
|
min_response_time=min_response_time * 1000,
|
|
requests_per_second=requests_per_second,
|
|
success_rate=100.0,
|
|
test_duration=test_duration
|
|
)
|
|
|
|
def generate_performance_report(self) -> str:
|
|
"""生成性能测试报告"""
|
|
report = []
|
|
report.append("=" * 80)
|
|
report.append("告警规则系统性能测试报告")
|
|
report.append("=" * 80)
|
|
report.append("")
|
|
|
|
# 测试概览
|
|
total_requests = sum(r.total_requests for r in self.test_results)
|
|
total_successful = sum(r.successful_requests for r in self.test_results)
|
|
avg_success_rate = sum(r.success_rate for r in self.test_results) / len(self.test_results)
|
|
avg_rps = sum(r.requests_per_second for r in self.test_results) / len(self.test_results)
|
|
|
|
report.append(f"📊 测试概览:")
|
|
report.append(f" 测试项目数: {len(self.test_results)}")
|
|
report.append(f" 总请求数: {total_requests:,}")
|
|
report.append(f" 成功请求数: {total_successful:,}")
|
|
report.append(f" 平均成功率: {avg_success_rate:.1f}%")
|
|
report.append(f" 平均RPS: {avg_rps:.1f}")
|
|
report.append("")
|
|
|
|
# 详细测试结果
|
|
report.append("📋 详细测试结果:")
|
|
for i, result in enumerate(self.test_results, 1):
|
|
report.append(f" {i}. {result.test_name}:")
|
|
report.append(f" 总请求数: {result.total_requests:,}")
|
|
report.append(f" 成功率: {result.success_rate:.1f}%")
|
|
report.append(f" 平均响应时间: {result.avg_response_time:.2f}ms")
|
|
report.append(f" 最大响应时间: {result.max_response_time:.2f}ms")
|
|
report.append(f" 最小响应时间: {result.min_response_time:.2f}ms")
|
|
report.append(f" 每秒请求数: {result.requests_per_second:.1f} RPS")
|
|
report.append(f" 测试耗时: {result.test_duration:.2f}秒")
|
|
report.append("")
|
|
|
|
# 性能评估
|
|
report.append("🎯 性能评估:")
|
|
|
|
if avg_rps > 1000:
|
|
report.append(" ✅ 优秀: 平均RPS超过1000，性能表现卓越")
|
|
elif avg_rps > 500:
|
|
report.append(" ✅ 良好: 平均RPS超过500，性能表现良好")
|
|
elif avg_rps > 100:
|
|
report.append(" ⚠️ 一般: 平均RPS超过100，性能可接受")
|
|
else:
|
|
report.append(" ❌ 需优化: 平均RPS低于100，需要性能优化")
|
|
|
|
if avg_success_rate > 99:
|
|
report.append(" ✅ 可靠性优秀: 成功率超过99%")
|
|
elif avg_success_rate > 95:
|
|
report.append(" ✅ 可靠性良好: 成功率超过95%")
|
|
else:
|
|
report.append(" ⚠️ 可靠性需改善: 成功率低于95%")
|
|
|
|
# 推荐优化建议
|
|
report.append("")
|
|
report.append("💡 优化建议:")
|
|
|
|
max_response_time = max(r.max_response_time for r in self.test_results)
|
|
if max_response_time > 1000:
|
|
report.append(" 1. 最大响应时间超过1秒，建议优化慢查询")
|
|
|
|
min_rps = min(r.requests_per_second for r in self.test_results)
|
|
if min_rps < 100:
|
|
report.append(" 2. 部分测试RPS较低，建议增加缓存或优化算法")
|
|
|
|
report.append(" 3. 考虑实施Redis集群以提高缓存性能")
|
|
report.append(" 4. 使用异步处理队列处理告警生成")
|
|
report.append(" 5. 实施规则预编译和缓存优化")
|
|
|
|
report.append("")
|
|
report.append("=" * 80)
|
|
|
|
return "\n".join(report)
|
|
|
|
def save_report_to_file(self, filename: str = None):
|
|
"""保存报告到文件"""
|
|
if filename is None:
|
|
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
|
filename = f"alert_rules_performance_report_{timestamp}.txt"
|
|
|
|
report_content = self.generate_performance_report()
|
|
|
|
with open(filename, 'w', encoding='utf-8') as f:
|
|
f.write(report_content)
|
|
|
|
logger.info(f"性能测试报告已保存到: {filename}")
|
|
return filename
|
|
|
|
async def main():
|
|
"""主函数"""
|
|
print("🚀 开始告警规则系统性能测试...")
|
|
|
|
tester = AlertRulesPerformanceTester()
|
|
|
|
try:
|
|
# 运行性能测试
|
|
results = await tester.run_comprehensive_performance_tests()
|
|
|
|
# 生成并显示报告
|
|
report = tester.generate_performance_report()
|
|
print(report)
|
|
|
|
# 保存报告
|
|
report_file = tester.save_report_to_file()
|
|
|
|
print(f"\n✅ 性能测试完成，报告已保存到: {report_file}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"性能测试失败: {e}")
|
|
print(f"❌ 性能测试失败: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
# 运行异步主函数
|
|
asyncio.run(main()) |