microwind 基础设施分析器技能
概述
好的基础设施可以扩展。糟糕的基础设施在扩展时会崩溃。在问题级联之前分析。需要建立完善的基础设施分析和优化机制。
核心原则: 好的基础设施可以扩展。糟糕的基础设施在扩展时会崩溃。在问题级联之前分析。
何时使用
始终:
- 规划基础设施
- 为增长扩展
- 性能瓶颈调查
- 灾难恢复规划
- 成本优化分析
- 安全架构审查
触发短语:
- "分析基础设施架构"
- "优化资源配置"
- "检查基础设施安全"
- "基础设施性能问题"
- "灾难恢复计划"
- "基础设施成本分析"
基础设施分析功能
架构分析
- 系统拓扑检查
- 组件依赖分析
- 扩展性评估
- 可用性分析
- 容错能力检查
性能分析
- 资源使用监控
- 瓶颈识别
- 负载均衡分析
- 网络性能检查
- 存储性能评估
安全分析
- 安全配置检查
- 访问控制审查
- 网络安全分析
- 数据保护检查
- 合规性验证
常见基础设施问题
单点故障
问题:
系统存在单点故障风险
后果:
- 服务中断
- 数据丢失
- 业务影响
- 声誉损失
解决方案:
- 实现高可用架构
- 添加冗余组件
- 设置故障转移
- 建立监控告警
资源配置不当
问题:
资源配置不合理
后果:
- 性能瓶颈
- 资源浪费
- 成本增加
- 扩展困难
解决方案:
- 容量规划
- 自动扩缩容
- 资源监控
- 成本优化
安全配置缺陷
问题:
安全配置存在漏洞
后果:
- 数据泄露
- 系统入侵
- 合规违规
- 法律风险
解决方案:
- 安全审计
- 配置加固
- 访问控制
- 监控告警
基础设施优化策略
高可用设计
冗余架构:
- 多区域部署
- 负载均衡
- 故障转移
- 健康检查
容错机制:
- 断路器模式
- 重试机制
- 降级策略
- 熔断保护
性能优化
资源优化:
- 垂直扩展
- 水平扩展
- 缓存策略
- CDN加速
网络优化:
- 带宽规划
- 延迟优化
- 负载分布
- 流量管理
代码实现示例
基础设施配置分析器
import json
import yaml
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
import boto3
import requests
class InfrastructureType(Enum):
AWS = "aws"
AZURE = "azure"
GCP = "gcp"
HYBRID = "hybrid"
@dataclass
class InfrastructureComponent:
"""基础设施组件"""
name: str
type: str
region: str
configuration: Dict[str, Any]
dependencies: List[str]
metrics: Dict[str, Any]
@dataclass
class SecurityIssue:
"""安全问题"""
severity: str
component: str
description: str
recommendation: str
class InfrastructureAnalyzer:
"""基础设施分析器"""
def __init__(self, infrastructure_type: InfrastructureType):
self.infrastructure_type = infrastructure_type
self.components = []
self.security_issues = []
self.performance_metrics = {}
self.cost_analysis = {}
def analyze_infrastructure(self, config_path: str = None) -> Dict[str, Any]:
"""分析基础设施"""
# 加载配置
if config_path:
config = self._load_configuration(config_path)
else:
config = self._fetch_cloud_configuration()
# 分析组件
self.components = self._extract_components(config)
# 分析依赖关系
dependency_analysis = self._analyze_dependencies(self.components)
# 分析安全性
security_analysis = self._analyze_security(self.components)
# 分析性能
performance_analysis = self._analyze_performance(self.components)
# 分析成本
cost_analysis = self._analyze_costs(self.components)
# 分析可用性
availability_analysis = self._analyze_availability(self.components)
return {
'infrastructure_type': self.infrastructure_type.value,
'components_count': len(self.components),
'dependency_analysis': dependency_analysis,
'security_analysis': security_analysis,
'performance_analysis': performance_analysis,
'cost_analysis': cost_analysis,
'availability_analysis': availability_analysis,
'recommendations': self._generate_recommendations()
}
def _load_configuration(self, config_path: str) -> Dict[str, Any]:
"""加载配置文件"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
if config_path.endswith('.yaml') or config_path.endswith('.yml'):
return yaml.safe_load(f)
elif config_path.endswith('.json'):
return json.load(f)
else:
raise ValueError("不支持的配置文件格式")
except Exception as e:
print(f"加载配置文件失败: {str(e)}")
return {}
def _fetch_cloud_configuration(self) -> Dict[str, Any]:
"""从云服务获取配置"""
if self.infrastructure_type == InfrastructureType.AWS:
return self._fetch_aws_configuration()
elif self.infrastructure_type == InfrastructureType.AZURE:
return self._fetch_azure_configuration()
elif self.infrastructure_type == InfrastructureType.GCP:
return self._fetch_gcp_configuration()
else:
return {}
def _fetch_aws_configuration(self) -> Dict[str, Any]:
"""获取AWS配置"""
try:
session = boto3.Session()
ec2 = session.client('ec2')
rds = session.client('rds')
elb = session.client('elbv2')
config = {
'ec2_instances': self._get_ec2_instances(ec2),
'rds_instances': self._get_rds_instances(rds),
'load_balancers': self._get_load_balancers(elb),
'vpcs': self._get_vpcs(ec2)
}
return config
except Exception as e:
print(f"获取AWS配置失败: {str(e)}")
return {}
def _get_ec2_instances(self, ec2_client) -> List[Dict[str, Any]]:
"""获取EC2实例"""
instances = []
try:
response = ec2_client.describe_instances()
for reservation in response['Reservations']:
for instance in reservation['Instances']:
instances.append({
'id': instance['InstanceId'],
'type': instance['InstanceType'],
'state': instance['State']['Name'],
'region': instance['Placement']['AvailabilityZone'],
'security_groups': [sg['GroupId'] for sg in instance['SecurityGroups']],
'tags': {tag['Key']: tag['Value'] for tag in instance.get('Tags', [])}
})
except Exception as e:
print(f"获取EC2实例失败: {str(e)}")
return instances
def _get_rds_instances(self, rds_client) -> List[Dict[str, Any]]:
"""获取RDS实例"""
instances = []
try:
response = rds_client.describe_db_instances()
for instance in response['DBInstances']:
instances.append({
'id': instance['DBInstanceIdentifier'],
'engine': instance['Engine'],
'status': instance['DBInstanceStatus'],
'instance_class': instance['DBInstanceClass'],
'multi_az': instance.get('MultiAZ', False),
'storage_type': instance.get('StorageType', 'standard')
})
except Exception as e:
print(f"获取RDS实例失败: {str(e)}")
return instances
def _get_load_balancers(self, elb_client) -> List[Dict[str, Any]]:
"""获取负载均衡器"""
load_balancers = []
try:
response = elb_client.describe_load_balancers()
for lb in response['LoadBalancers']:
load_balancers.append({
'name': lb['LoadBalancerArn'],
'type': lb['Type'],
'state': lb['State']['Code'],
'scheme': lb['Scheme'],
'vpc_id': lb['VpcId']
})
except Exception as e:
print(f"获取负载均衡器失败: {str(e)}")
return load_balancers
def _get_vpcs(self, ec2_client) -> List[Dict[str, Any]]:
"""获取VPC"""
vpcs = []
try:
response = ec2_client.describe_vpcs()
for vpc in response['Vpcs']:
vpcs.append({
'id': vpc['VpcId'],
'cidr': vpc['CidrBlock'],
'state': vpc['State'],
'is_default': vpc['IsDefault']
})
except Exception as e:
print(f"获取VPC失败: {str(e)}")
return vpcs
def _extract_components(self, config: Dict[str, Any]) -> List[InfrastructureComponent]:
"""提取基础设施组件"""
components = []
# 提取EC2实例
for instance in config.get('ec2_instances', []):
component = InfrastructureComponent(
name=instance['id'],
type='EC2',
region=instance['region'],
configuration=instance,
dependencies=self._extract_instance_dependencies(instance),
metrics={}
)
components.append(component)
# 提取RDS实例
for instance in config.get('rds_instances', []):
component = InfrastructureComponent(
name=instance['id'],
type='RDS',
region=instance.get('region', 'unknown'),
configuration=instance,
dependencies=[],
metrics={}
)
components.append(component)
# 提取负载均衡器
for lb in config.get('load_balancers', []):
component = InfrastructureComponent(
name=lb['name'],
type='LoadBalancer',
region=lb.get('region', 'unknown'),
configuration=lb,
dependencies=[],
metrics={}
)
components.append(component)
return components
def _extract_instance_dependencies(self, instance: Dict[str, Any]) -> List[str]:
"""提取实例依赖关系"""
dependencies = []
# 检查安全组依赖
for sg_id in instance.get('security_groups', []):
dependencies.append(f"SecurityGroup:{sg_id}")
# 检查子网依赖
if 'SubnetId' in instance:
dependencies.append(f"Subnet:{instance['SubnetId']}")
return dependencies
def _analyze_dependencies(self, components: List[InfrastructureComponent]) -> Dict[str, Any]:
"""分析依赖关系"""
dependency_graph = {}
single_points_of_failure = []
circular_dependencies = []
# 构建依赖图
for component in components:
dependency_graph[component.name] = component.dependencies
# 检查单点故障
dependency_counts = {}
for dependencies in dependency_graph.values():
for dep in dependencies:
dependency_counts[dep] = dependency_counts.get(dep, 0) + 1
# 找出被多个组件依赖的关键组件
for dep, count in dependency_counts.items():
if count > 3: # 被3个以上组件依赖
single_points_of_failure.append({
'component': dep,
'dependent_count': count,
'risk_level': 'high'
})
# 检查循环依赖（简化版）
visited = set()
recursion_stack = set()
def has_circular_dependency(component: str) -> bool:
if component in recursion_stack:
return True
if component in visited:
return False
visited.add(component)
recursion_stack.add(component)
for dep in dependency_graph.get(component, []):
if has_circular_dependency(dep):
return True
recursion_stack.remove(component)
return False
for component in dependency_graph:
if has_circular_dependency(component):
circular_dependencies.append(component)
return {
'dependency_graph': dependency_graph,
'single_points_of_failure': single_points_of_failure,
'circular_dependencies': circular_dependencies,
'total_dependencies': len(dependency_counts)
}
def _analyze_security(self, components: List[InfrastructureComponent]) -> Dict[str, Any]:
"""分析安全性"""
security_issues = []
for component in components:
if component.type == 'EC2':
issues = self._analyze_ec2_security(component)
security_issues.extend(issues)
elif component.type == 'RDS':
issues = self._analyze_rds_security(component)
security_issues.extend(issues)
# 按严重程度分类
critical_issues = [issue for issue in security_issues if issue.severity == 'critical']
high_issues = [issue for issue in security_issues if issue.severity == 'high']
medium_issues = [issue for issue in security_issues if issue.severity == 'medium']
low_issues = [issue for issue in security_issues if issue.severity == 'low']
return {
'total_issues': len(security_issues),
'critical_issues': len(critical_issues),
'high_issues': len(high_issues),
'medium_issues': len(medium_issues),
'low_issues': len(low_issues),
'issues_by_component': self._group_issues_by_component(security_issues),
'security_score': self._calculate_security_score(security_issues)
}
def _analyze_ec2_security(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析EC2安全性"""
issues = []
config = component.configuration
# 检查安全组配置
security_groups = config.get('security_groups', [])
if len(security_groups) == 0:
issues.append(SecurityIssue(
severity='high',
component=component.name,
description='EC2实例没有配置安全组',
recommendation='为实例配置适当的安全组'
))
# 检查公开端口
if config.get('public_ip'):
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='EC2实例具有公开IP地址',
recommendation='确保只有必要的服务公开访问'
))
# 检查IAM角色
if not config.get('iam_instance_profile'):
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='EC2实例没有配置IAM角色',
recommendation='为实例配置适当的IAM角色以限制权限'
))
return issues
def _analyze_rds_security(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析RDS安全性"""
issues = []
config = component.configuration
# 检查多可用区
if not config.get('multi_az', False):
issues.append(SecurityIssue(
severity='high',
component=component.name,
description='RDS实例未启用多可用区',
recommendation='启用多可用区以提高可用性'
))
# 检查加密
if not config.get('storage_encrypted', False):
issues.append(SecurityIssue(
severity='high',
component=component.name,
description='RDS实例未启用存储加密',
recommendation='启用存储加密以保护数据'
))
# 检查公开访问
if config.get('publicly_accessible', False):
issues.append(SecurityIssue(
severity='critical',
component=component.name,
description='RDS实例允许公开访问',
recommendation='禁用公开访问，只允许VPC内部访问'
))
return issues
def _analyze_performance(self, components: List[InfrastructureComponent]) -> Dict[str, Any]:
"""分析性能"""
performance_issues = []
bottlenecks = []
for component in components:
if component.type == 'EC2':
issues = self._analyze_ec2_performance(component)
performance_issues.extend(issues)
elif component.type == 'RDS':
issues = self._analyze_rds_performance(component)
performance_issues.extend(issues)
# 识别性能瓶颈
bottleneck_types = {}
for issue in performance_issues:
issue_type = issue.description.split(':')[0]
bottleneck_types[issue_type] = bottleneck_types.get(issue_type, 0) + 1
return {
'total_issues': len(performance_issues),
'bottleneck_types': bottleneck_types,
'performance_score': self._calculate_performance_score(performance_issues),
'recommendations': [issue.recommendation for issue in performance_issues]
}
def _analyze_ec2_performance(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析EC2性能"""
issues = []
config = component.configuration
# 检查实例类型
instance_type = config.get('type', '')
if instance_type.startswith('t2.'):
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='EC2实例类型: 使用了突发性能实例',
recommendation='考虑使用更稳定的实例类型以保证性能'
))
# 检查状态
state = config.get('state', '')
if state != 'running':
issues.append(SecurityIssue(
severity='low',
component=component.name,
description=f'EC2实例状态: {state}',
recommendation='确保实例处于运行状态'
))
return issues
def _analyze_rds_performance(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析RDS性能"""
issues = []
config = component.configuration
# 检查实例类型
instance_class = config.get('instance_class', '')
if instance_class in ['db.t2.micro', 'db.t2.small']:
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='RDS实例类型: 使用了小规格实例',
recommendation='根据工作负载选择合适的实例规格'
))
# 检查存储类型
storage_type = config.get('storage_type', 'standard')
if storage_type == 'standard':
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='RDS存储类型: 使用了标准存储',
recommendation='考虑使用SSD存储以提高I/O性能'
))
return issues
def _analyze_costs(self, components: List[InfrastructureComponent]) -> Dict[str, Any]:
"""分析成本"""
cost_breakdown = {}
total_estimated_cost = 0
for component in components:
if component.type == 'EC2':
cost = self._estimate_ec2_cost(component)
elif component.type == 'RDS':
cost = self._estimate_rds_cost(component)
else:
cost = 0
cost_breakdown[component.name] = cost
total_estimated_cost += cost
# 成本优化建议
optimization_suggestions = self._generate_cost_optimization_suggestions(components)
return {
'total_estimated_cost': total_estimated_cost,
'cost_breakdown': cost_breakdown,
'optimization_suggestions': optimization_suggestions,
'cost_efficiency_score': self._calculate_cost_efficiency(components)
}
def _estimate_ec2_cost(self, component: InfrastructureComponent) -> float:
"""估算EC2成本"""
instance_type = component.configuration.get('type', '')
# 简化的成本估算，实际应该使用AWS定价API
cost_map = {
't2.micro': 0.0116,
't2.small': 0.023,
't2.medium': 0.046,
't3.micro': 0.0104,
't3.small': 0.0208,
't3.medium': 0.0416,
'm5.large': 0.096,
'm5.xlarge': 0.192,
'c5.large': 0.085,
'c5.xlarge': 0.17
}
return cost_map.get(instance_type, 0.1) * 24 * 30 # 月成本估算
def _estimate_rds_cost(self, component: InfrastructureComponent) -> float:
"""估算RDS成本"""
instance_class = component.configuration.get('instance_class', '')
# 简化的成本估算
cost_map = {
'db.t2.micro': 0.0116,
'db.t2.small': 0.029,
'db.t2.medium': 0.058,
'db.t3.micro': 0.0104,
'db.t3.small': 0.0416,
'db.t3.medium': 0.083,
'db.m5.large': 0.145,
'db.m5.xlarge': 0.29
}
base_cost = cost_map.get(instance_class, 0.15) * 24 * 30
# 多可用区额外费用
if component.configuration.get('multi_az', False):
base_cost *= 2
return base_cost
def _generate_cost_optimization_suggestions(self, components: List[InfrastructureComponent]) -> List[str]:
"""生成成本优化建议"""
suggestions = []
# 检查闲置实例
stopped_instances = [c for c in components if c.type == 'EC2' and c.configuration.get('state') == 'stopped']
if stopped_instances:
suggestions.append(f"发现{len(stopped_instances)}个停止的EC2实例，建议删除以节省成本")
# 检查过度配置
oversized_instances = [c for c in components if c.type == 'EC2' and c.configuration.get('type') in ['m5.2xlarge', 'c5.2xlarge']]
if oversized_instances:
suggestions.append(f"发现{len(oversized_instances)}个大规格实例，建议评估实际需求")
# 检查存储优化
rds_instances = [c for c in components if c.type == 'RDS']
if rds_instances:
suggestions.append("考虑使用RDS预留实例以降低成本")
return suggestions
def _analyze_availability(self, components: List[InfrastructureComponent]) -> Dict[str, Any]:
"""分析可用性"""
availability_issues = []
availability_score = 100
for component in components:
if component.type == 'EC2':
issues = self._analyze_ec2_availability(component)
availability_issues.extend(issues)
elif component.type == 'RDS':
issues = self._analyze_rds_availability(component)
availability_issues.extend(issues)
# 计算可用性评分
for issue in availability_issues:
if issue.severity == 'critical':
availability_score -= 20
elif issue.severity == 'high':
availability_score -= 10
elif issue.severity == 'medium':
availability_score -= 5
elif issue.severity == 'low':
availability_score -= 2
availability_score = max(0, availability_score)
return {
'availability_score': availability_score,
'availability_issues': availability_issues,
'redundancy_level': self._calculate_redundancy_level(components),
'fault_tolerance': self._assess_fault_tolerance(components)
}
def _analyze_ec2_availability(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析EC2可用性"""
issues = []
config = component.configuration
# 检查是否在单个可用区
if len(set([config.get('region', '')])) == 1:
issues.append(SecurityIssue(
severity='high',
component=component.name,
description='EC2实例部署在单个可用区',
recommendation='考虑跨多个可用区部署以提高可用性'
))
return issues
def _analyze_rds_availability(self, component: InfrastructureComponent) -> List[SecurityIssue]:
"""分析RDS可用性"""
issues = []
config = component.configuration
# 检查多可用区
if not config.get('multi_az', False):
issues.append(SecurityIssue(
severity='high',
component=component.name,
description='RDS实例未启用多可用区',
recommendation='启用多可用区部署'
))
# 检查备份
if not config.get('backup_retention_period', 0):
issues.append(SecurityIssue(
severity='medium',
component=component.name,
description='RDS实例未启用备份',
recommendation='启用自动备份'
))
return issues
def _calculate_redundancy_level(self, components: List[InfrastructureComponent]) -> str:
"""计算冗余级别"""
ec2_count = len([c for c in components if c.type == 'EC2'])
rds_count = len([c for c in components if c.type == 'RDS'])
if ec2_count >= 3 and rds_count >= 2:
return 'high'
elif ec2_count >= 2 and rds_count >= 1:
return 'medium'
else:
return 'low'
def _assess_fault_tolerance(self, components: List[InfrastructureComponent]) -> str:
"""评估容错能力"""
multi_az_rds = len([c for c in components if c.type == 'RDS' and c.configuration.get('multi_az', False)])
total_rds = len([c for c in components if c.type == 'RDS'])
if total_rds > 0 and multi_az_rds / total_rds >= 0.8:
return 'excellent'
elif multi_az_rds / total_rds >= 0.5:
return 'good'
else:
return 'poor'
def _group_issues_by_component(self, issues: List[SecurityIssue]) -> Dict[str, List[SecurityIssue]]:
"""按组件分组问题"""
grouped = {}
for issue in issues:
if issue.component not in grouped:
grouped[issue.component] = []
grouped[issue.component].append(issue)
return grouped
def _calculate_security_score(self, issues: List[SecurityIssue]) -> int:
"""计算安全评分"""
score = 100
for issue in issues:
if issue.severity == 'critical':
score -= 25
elif issue.severity == 'high':
score -= 15
elif issue.severity == 'medium':
score -= 8
elif issue.severity == 'low':
score -= 3
return max(0, score)
def _calculate_performance_score(self, issues: List[SecurityIssue]) -> int:
"""计算性能评分"""
score = 100
for issue in issues:
if issue.severity == 'critical':
score -= 20
elif issue.severity == 'high':
score -= 12
elif issue.severity == 'medium':
score -= 6
elif issue.severity == 'low':
score -= 2
return max(0, score)
def _calculate_cost_efficiency(self, components: List[InfrastructureComponent]) -> str:
"""计算成本效率"""
# 简化的成本效率评估
total_components = len(components)
oversized_components = len([c for c in components if 'large' in c.configuration.get('type', '') or 'xlarge' in c.configuration.get('type', '')])
if oversized_components / total_components > 0.3:
return 'poor'
elif oversized_components / total_components > 0.1:
return 'medium'
else:
return 'good'
def _generate_recommendations(self) -> List[str]:
"""生成综合建议"""
recommendations = []
# 基于安全分析的建议
if self.security_issues:
critical_count = len([i for i in self.security_issues if i.severity == 'critical'])
if critical_count > 0:
recommendations.append(f"发现{critical_count}个严重安全问题，需要立即处理")
# 基于性能分析的建议
recommendations.append("定期监控基础设施性能指标")
# 基于成本分析的建议
recommendations.append("定期审查基础设施成本，优化资源配置")
# 基于可用性分析的建议
recommendations.append("实施高可用架构以减少服务中断")
return recommendations
# 使用示例
def main():
analyzer = InfrastructureAnalyzer(InfrastructureType.AWS)
# 分析基础设施
analysis = analyzer.analyze_infrastructure()
print("=== 基础设施分析报告 ===")
print(f"基础设施类型: {analysis['infrastructure_type']}")
print(f"组件数量: {analysis['components_count']}")
print("\n--- 依赖关系分析 ---")
dep_analysis = analysis['dependency_analysis']
print(f"单点故障: {len(dep_analysis['single_points_of_failure'])}")
print(f"循环依赖: {len(dep_analysis['circular_dependencies'])}")
print("\n--- 安全分析 ---")
sec_analysis = analysis['security_analysis']
print(f"安全问题总数: {sec_analysis['total_issues']}")
print(f"严重问题: {sec_analysis['critical_issues']}")
print(f"高危问题: {sec_analysis['high_issues']}")
print(f"安全评分: {sec_analysis['security_score']}")
print("\n--- 性能分析 ---")
perf_analysis = analysis['performance_analysis']
print(f"性能问题: {perf_analysis['total_issues']}")
print(f"性能评分: {perf_analysis['performance_score']}")
print("\n--- 成本分析 ---")
cost_analysis = analysis['cost_analysis']
print(f"预估月成本: ${cost_analysis['total_estimated_cost']:.2f}")
print(f"成本效率: {cost_analysis['cost_efficiency_score']}")
print("\n--- 可用性分析 ---")
avail_analysis = analysis['availability_analysis']
print(f"可用性评分: {avail_analysis['availability_score']}")
print(f"冗余级别: {avail_analysis['redundancy_level']}")
print(f"容错能力: {avail_analysis['fault_tolerance']}")
print("\n--- 建议 ---")
for rec in analysis['recommendations']:
print(f"- {rec}")
if __name__ == "__main__":
main()
基础设施监控器
import time
import psutil
import requests
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import statistics
@dataclass
class SystemMetrics:
"""系统指标"""
timestamp: datetime
cpu_usage: float
memory_usage: float
disk_usage: float
network_io: Dict[str, float]
load_average: List[float]
class InfrastructureMonitor:
"""基础设施监控器"""
def __init__(self):
self.metrics_history = []
self.alert_thresholds = {
'cpu_usage': 80.0,
'memory_usage': 85.0,
'disk_usage': 90.0,
'load_average': 2.0
}
self.alerts = []
def collect_system_metrics(self) -> SystemMetrics:
"""收集系统指标"""
metrics = SystemMetrics(
timestamp=datetime.now(),
cpu_usage=psutil.cpu_percent(interval=1),
memory_usage=psutil.virtual_memory().percent,
disk_usage=psutil.disk_usage('/').percent,
network_io=self._get_network_io(),
load_average=list(psutil.getloadavg())
)
self.metrics_history.append(metrics)
# 检查告警
self._check_alerts(metrics)
return metrics
def _get_network_io(self) -> Dict[str, float]:
"""获取网络IO"""
net_io = psutil.net_io_counters()
return {
'bytes_sent': net_io.bytes_sent,
'bytes_recv': net_io.bytes_recv,
'packets_sent': net_io.packets_sent,
'packets_recv': net_io.packets_recv
}
def _check_alerts(self, metrics: SystemMetrics):
"""检查告警条件"""
if metrics.cpu_usage > self.alert_thresholds['cpu_usage']:
self.alerts.append({
'timestamp': metrics.timestamp,
'type': 'cpu_high',
'message': f"CPU使用率过高: {metrics.cpu_usage:.1f}%",
'severity': 'high'
})
if metrics.memory_usage > self.alert_thresholds['memory_usage']:
self.alerts.append({
'timestamp': metrics.timestamp,
'type': 'memory_high',
'message': f"内存使用率过高: {metrics.memory_usage:.1f}%",
'severity': 'high'
})
if metrics.disk_usage > self.alert_thresholds['disk_usage']:
self.alerts.append({
'timestamp': metrics.timestamp,
'type': 'disk_high',
'message': f"磁盘使用率过高: {metrics.disk_usage:.1f}%",
'severity': 'critical'
})
if metrics.load_average[0] > self.alert_thresholds['load_average']:
self.alerts.append({
'timestamp': metrics.timestamp,
'type': 'load_high',
'message': f"系统负载过高: {metrics.load_average[0]:.2f}",
'severity': 'medium'
})
def analyze_trends(self, hours: int = 24) -> Dict[str, Any]:
"""分析趋势"""
cutoff_time = datetime.now() - timedelta(hours=hours)
recent_metrics = [
m for m in self.metrics_history
if m.timestamp >= cutoff_time
]
if len(recent_metrics) < 2:
return {'message': '数据不足，无法分析趋势'}
cpu_trend = self._calculate_trend([m.cpu_usage for m in recent_metrics])
memory_trend = self._calculate_trend([m.memory_usage for m in recent_metrics])
disk_trend = self._calculate_trend([m.disk_usage for m in recent_metrics])
return {
'period_hours': hours,
'data_points': len(recent_metrics),
'cpu_trend': cpu_trend,
'memory_trend': memory_trend,
'disk_trend': disk_trend,
'avg_cpu_usage': statistics.mean([m.cpu_usage for m in recent_metrics]),
'avg_memory_usage': statistics.mean([m.memory_usage for m in recent_metrics]),
'avg_disk_usage': statistics.mean([m.disk_usage for m in recent_metrics]),
'peak_cpu_usage': max([m.cpu_usage for m in recent_metrics]),
'peak_memory_usage': max([m.memory_usage for m in recent_metrics])
}
def _calculate_trend(self, values: List[float]) -> str:
"""计算趋势"""
if len(values) < 2:
return 'stable'
# 简单的线性趋势
x = list(range(len(values)))
n = len(values)
sum_x = sum(x)
sum_y = sum(values)
sum_xy = sum(x[i] * values[i] for i in range(n))
sum_x2 = sum(x[i] ** 2 for i in range(n))
slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x ** 2)
if slope > 0.5:
return 'increasing'
elif slope < -0.5:
return 'decreasing'
else:
return 'stable'
def generate_health_report(self) -> Dict[str, Any]:
"""生成健康报告"""
if not self.metrics_history:
return {'message': '没有监控数据'}
latest_metrics = self.metrics_history[-1]
# 计算健康评分
health_score = 100
if latest_metrics.cpu_usage > 80:
health_score -= 20
if latest_metrics.memory_usage > 85:
health_score -= 20
if latest_metrics.disk_usage > 90:
health_score -= 30
if latest_metrics.load_average[0] > 2.0:
health_score -= 15
health_score = max(0, health_score)
# 获取最近告警
recent_alerts = [
alert for alert in self.alerts
if alert['timestamp'] > datetime.now() - timedelta(hours=1)
]
return {
'health_score': health_score,
'health_status': self._get_health_status(health_score),
'current_metrics': {
'cpu_usage': latest_metrics.cpu_usage,
'memory_usage': latest_metrics.memory_usage,
'disk_usage': latest_metrics.disk_usage,
'load_average': latest_metrics.load_average[0]
},
'recent_alerts': recent_alerts,
'trend_analysis': self.analyze_trends(),
'recommendations': self._generate_health_recommendations(latest_metrics)
}
def _get_health_status(self, score: int) -> str:
"""获取健康状态"""
if score >= 90:
return 'excellent'
elif score >= 75:
return 'good'
elif score >= 60:
return 'fair'
else:
return 'poor'
def _generate_health_recommendations(self, metrics: SystemMetrics) -> List[str]:
"""生成健康建议"""
recommendations = []
if metrics.cpu_usage > 80:
recommendations.append("CPU使用率较高，建议检查高CPU进程或扩展资源")
if metrics.memory_usage > 85:
recommendations.append("内存使用率较高，建议检查内存泄漏或增加内存")
if metrics.disk_usage > 90:
recommendations.append("磁盘空间不足，建议清理磁盘或扩展存储")
if metrics.load_average[0] > 2.0:
recommendations.append("系统负载较高，建议检查并发进程或优化应用")
return recommendations
# 使用示例
def main():
monitor = InfrastructureMonitor()
# 模拟监控
print("开始监控基础设施...")
for i in range(10):
metrics = monitor.collect_system_metrics()
print(f"第{i+1}次监控 - CPU: {metrics.cpu_usage:.1f}%, 内存: {metrics.memory_usage:.1f}%, 磁盘: {metrics.disk_usage:.1f}%")
time.sleep(2)
# 生成报告
report = monitor.generate_health_report()
print("\n=== 基础设施健康报告 ===")
print(f"健康评分: {report['health_score']}")
print(f"健康状态: {report['health_status']}")
print("\n当前指标:")
current = report['current_metrics']
print(f"CPU使用率: {current['cpu_usage']:.1f}%")
print(f"内存使用率: {current['memory_usage']:.1f}%")
print(f"磁盘使用率: {current['disk_usage']:.1f}%")
print(f"系统负载: {current['load_average']:.2f}")
if report['recent_alerts']:
print("\n最近告警:")
for alert in report['recent_alerts']:
print(f"- {alert['message']}")
if report['recommendations']:
print("\n建议:")
for rec in report['recommendations']:
print(f"- {rec}")
if __name__ == "__main__":
main()
基础设施最佳实践
架构设计
- 高可用性: 避免单点故障
- 可扩展性: 支持水平和垂直扩展
- 安全性: 多层安全防护
- 成本优化: 合理控制成本
监控运维
- 全面监控: 覆盖所有关键指标
- 告警机制: 及时发现问题
- 自动化: 减少人工干预
- 文档化: 完善的运维文档
灾难恢复
- 备份策略: 定期备份和恢复测试
- 故障转移: 自动故障转移机制
- 应急预案: 详细的应急响应流程
- 演练测试: 定期进行灾难演练
相关技能
- kubernetes-analyzer - Kubernetes配置分析
- security-scanner - 安全漏洞扫描
- performance-profiler - 性能分析和监控
- cost-optimizer - 成本优化分析