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
|
"""
自动降配脚本 - 识别并建议低利用率实例的降配方案
"""
import boto3
from datetime import datetime, timedelta
class CostOptimizer:
def __init__(self, region='us-east-1'):
self.ec2 = boto3.client('ec2', region_name=region)
self.cloudwatch = boto3.client('cloudwatch', region_name=region)
def get_underutilized_instances(self, threshold=20, days=7):
"""获取低利用率实例"""
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days)
instances = self.ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
underutilized = []
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
instance_id = instance['InstanceId']
metrics = self.cloudwatch.get_metric_statistics(
Namespace='AWS/EC2',
MetricName='CPUUtilization',
Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
StartTime=start_time,
EndTime=end_time,
Period=3600,
Statistics=['Average']
)
if metrics['Datapoints']:
avg_cpu = sum(d['Average'] for d in metrics['Datapoints']) / len(metrics['Datapoints'])
if avg_cpu < threshold:
underutilized.append({
'InstanceId': instance_id,
'InstanceType': instance['InstanceType'],
'AvgCPU': avg_cpu,
'CurrentCost': self.get_instance_cost(instance['InstanceType'])
})
return underutilized
def get_instance_cost(self, instance_type):
"""获取实例成本（简化版）"""
costs = {
'm5.large': 0.096,
'm5.xlarge': 0.192,
'm5.2xlarge': 0.384,
'c5.large': 0.085,
'c5.xlarge': 0.17,
'r5.large': 0.126,
'r5.xlarge': 0.252,
}
return costs.get(instance_type, 0.1)
def recommend_downgrade(self, instance_type):
"""推荐降配方案"""
downgrade_map = {
'm5.2xlarge': 'm5.xlarge',
'm5.xlarge': 'm5.large',
'c5.2xlarge': 'c5.xlarge',
'c5.xlarge': 'c5.large',
'r5.2xlarge': 'r5.xlarge',
'r5.xlarge': 'r5.large',
}
return downgrade_map.get(instance_type, instance_type)
def generate_report(self):
"""生成优化报告"""
underutilized = self.get_underutilized_instances()
print("=" * 60)
print("成本优化报告")
print("=" * 60)
total_savings = 0
for inst in underutilized:
recommended = self.recommend_downgrade(inst['InstanceType'])
current_cost = inst['CurrentCost']
new_cost = self.get_instance_cost(recommended)
savings = (current_cost - new_cost) * 24 * 30
print(f"\n实例：{inst['InstanceId']}")
print(f" 当前规格：{inst['InstanceType']}")
print(f" 平均 CPU: {inst['AvgCPU']:.1f}%")
print(f" 建议规格：{recommended}")
print(f" 预计月节省：${savings:.2f}")
total_savings += savings
print(f"\n{'=' * 60}")
print(f"预计总月节省：${total_savings:.2f}")
print(f"{'=' * 60}")
if __name__ == '__main__':
optimizer = CostOptimizer()
optimizer.generate_report()
|
