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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
| # ========== DORA指标 ==========
class DORAMetrics:
"""DORA (DevOps Research and Assessment) 核心指标"""
@staticmethod
def deployment_frequency(deployments: int, days: int) -> float:
"""
部署频率
- 精英级：按需部署 (每天多次)
- 高绩效：每周1-6个月
- 中等绩效：每月1-6个月
- 低绩效：少于每6个月1次
"""
return deployments / days
@staticmethod
def lead_time_for_changes(commit_time: str, deploy_time: str) -> float:
"""
变更前置时间
从代码提交到成功部署的时间
- 精英级：小于1小时
- 高绩效：小于1天
- 中等绩效：1周-1个月
- 低绩效：超过1个月
"""
from datetime import datetime
commit = datetime.fromisoformat(commit_time)
deploy = datetime.fromisoformat(deploy_time)
return (deploy - commit).total_seconds() / 3600 # 小时
@staticmethod
def time_to_restore_service(incident_time: str, restore_time: str) -> float:
"""
服务恢复时间
- 精英级：小于1小时
- 高绩效：小于1天
- 中等绩效：1天-1周
- 低绩效：超过1周
"""
from datetime import datetime
incident = datetime.fromisoformat(incident_time)
restore = datetime.fromisoformat(restore_time)
return (restore - incident).total_seconds() / 3600 # 小时
@staticmethod
def change_failure_rate(total_deployments: int, failed_deployments: int) -> float:
"""
变更失败率
- 精英级：0-15%
- 高绩效：15-30%
- 中等绩效：30-60%
- 低绩效：超过60%
"""
return (failed_deployments / total_deployments) * 100
@classmethod
def evaluate_performance(cls, metrics: dict) -> str:
"""评估团队DevOps绩效等级"""
score = 0
if metrics['deployment_frequency'] >= 1: # 每天至少1次
score += 1
if metrics['lead_time'] <= 1: # 小于1小时
score += 1
if metrics['restore_time'] <= 1: # 小于1小时
score += 1
if metrics['failure_rate'] <= 15: # 小于15%
score += 1
levels = {
4: "精英级",
3: "高绩效",
2: "中等绩效",
1: "低绩效",
0: "低绩效"
}
return levels[score]
# ========== 自定义DevOps指标 ==========
class DevOpsMetricsCollector:
"""DevOps指标收集器"""
def __init__(self):
self.metrics = {
'builds': [],
'deployments': [],
'incidents': [],
'tests': []
}
def record_build(self, build_info: dict):
"""记录构建信息"""
self.metrics['builds'].append({
'timestamp': build_info['timestamp'],
'branch': build_info['branch'],
'commit': build_info['commit'],
'status': build_info['status'],
'duration': build_info['duration'],
'triggered_by': build_info['triggered_by']
})
def record_deployment(self, deployment_info: dict):
"""记录部署信息"""
self.metrics['deployments'].append({
'timestamp': deployment_info['timestamp'],
'environment': deployment_info['environment'],
'version': deployment_info['version'],
'status': deployment_info['status'],
'duration': deployment_info['duration'],
'deployed_by': deployment_info['deployed_by']
})
def record_incident(self, incident_info: dict):
"""记录故障信息"""
self.metrics['incidents'].append({
'detected_at': incident_info['detected_at'],
'resolved_at': incident_info.get('resolved_at'),
'severity': incident_info['severity'],
'affected_services': incident_info['affected_services'],
'root_cause': incident_info.get('root_cause')
})
def calculate_metrics(self, days: int = 30) -> dict:
"""计算DevOps指标"""
from datetime import datetime, timedelta
cutoff_time = datetime.now() - timedelta(days=days)
# 筛选时间范围内的数据
recent_builds = [
b for b in self.metrics['builds']
if datetime.fromisoformat(b['timestamp']) > cutoff_time
]
recent_deployments = [
d for d in self.metrics['deployments']
if datetime.fromisoformat(d['timestamp']) > cutoff_time
]
recent_incidents = [
i for i in self.metrics['incidents']
if datetime.fromisoformat(i['detected_at']) > cutoff_time
]
# 计算指标
total_builds = len(recent_builds)
successful_builds = len([b for b in recent_builds if b['status'] == 'success'])
build_success_rate = (successful_builds / total_builds * 100) if total_builds > 0 else 0
total_deployments = len(recent_deployments)
successful_deployments = len([d for d in recent_deployments if d['status'] == 'success'])
deployment_success_rate = (successful_deployments / total_deployments * 100) if total_deployments > 0 else 0
deployment_frequency = total_deployments / days
total_incidents = len(recent_incidents)
resolved_incidents = len([i for i in recent_incidents if i.get('resolved_at')])
avg_resolution_time = 0
if resolved_incidents > 0:
resolution_times = []
for incident in recent_incidents:
if incident.get('resolved_at'):
detected = datetime.fromisoformat(incident['detected_at'])
resolved = datetime.fromisoformat(incident['resolved_at'])
resolution_times.append((resolved - detected).total_seconds() / 3600)
avg_resolution_time = sum(resolution_times) / len(resolution_times)
return {
'build_metrics': {
'total_builds': total_builds,
'success_rate': round(build_success_rate, 2),
'avg_duration': round(
sum(b['duration'] for b in recent_builds) / total_builds if total_builds > 0 else 0,
2
)
},
'deployment_metrics': {
'total_deployments': total_deployments,
'success_rate': round(deployment_success_rate, 2),
'frequency_per_day': round(deployment_frequency, 2)
},
'incident_metrics': {
'total_incidents': total_incidents,
'resolved_incidents': resolved_incidents,
'avg_resolution_time_hours': round(avg_resolution_time, 2)
}
}
|