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
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
"""Utility to support parsing a TestStat."""
from collections import defaultdict
from dataclasses import dataclass
from itertools import chain
from typing import NamedTuple, List, Callable, Optional
import requests
from requests.adapters import HTTPAdapter, Retry
from buildscripts.util.testname import split_test_hook_name, is_resmoke_hook, get_short_name_from_test_file
TASK_LEVEL_HOOKS = {"CleanEveryN"}
TESTS_STATS_S3_LOCATION = "https://mongo-test-stats.s3.amazonaws.com"
class HistoricalTestInformation(NamedTuple):
"""
Container for information about the historical runtime of a test.
test_name: Name of test.
avg_duration_pass: Average of runtime of test that passed.
num_pass: Number of times the test has passed.
num_fail: Number of times the test has failed.
"""
test_name: str
num_pass: int
num_fail: int
avg_duration_pass: float
class TestRuntime(NamedTuple):
"""
Container for the runtime of a test.
test_name: Name of test.
runtime: Average of runtime of test.
"""
test_name: str
runtime: float
@dataclass
class _RuntimeHistory:
"""
History of runtime results.
duration: Average duration of test runtime.
num_runs: Number of test runs seen.
"""
duration: float
num_runs: int
@classmethod
def empty(cls) -> "_RuntimeHistory":
"""Create an empty runtime entry."""
return cls(duration=0.0, num_runs=0)
def add_runtimes(self, duration: float, num_runs: int) -> None:
"""
Add the given duration number this history.
:param duration: Average duration to include.
:param num_runs: Number of runs to include.
"""
self.duration = _average(self.duration, self.num_runs, duration, num_runs)
self.num_runs += num_runs
def normalize_test_name(test_name: str) -> str:
"""Normalize test names that may have been run on windows or unix."""
return test_name.replace("\\", "/")
def _average(value_a: float, num_a: int, value_b: float, num_b: int) -> float:
"""Compute a weighted average of 2 values with associated numbers."""
divisor = num_a + num_b
if divisor == 0:
return 0
else:
return float(value_a * num_a + value_b * num_b) / divisor
class HistoricHookInfo(NamedTuple):
"""Historic information about a test hook."""
hook_id: str
num_pass: int
avg_duration: float
@classmethod
def from_test_stats(cls, test_stats: HistoricalTestInformation) -> "HistoricHookInfo":
"""Create an instance from a test_stats object."""
return cls(hook_id=test_stats.test_name, num_pass=test_stats.num_pass,
avg_duration=test_stats.avg_duration_pass)
def test_name(self) -> str:
"""Get the name of the test associated with this hook."""
return split_test_hook_name(self.hook_id)[0]
def hook_name(self) -> str:
"""Get the name of this hook."""
return split_test_hook_name(self.hook_id)[-1]
def is_task_level_hook(self) -> bool:
"""Determine if this hook should be counted against the task not the test."""
return self.hook_name() in TASK_LEVEL_HOOKS
class HistoricTestInfo(NamedTuple):
"""Historic information about a test."""
test_name: str
num_pass: int
avg_duration: float
hooks: List[HistoricHookInfo]
@classmethod
def from_test_stats(cls, test_stats: HistoricalTestInformation,
hooks: List[HistoricHookInfo]) -> "HistoricTestInfo":
"""Create an instance from a test_stats object."""
return cls(test_name=test_stats.test_name, num_pass=test_stats.num_pass,
avg_duration=test_stats.avg_duration_pass, hooks=hooks)
def normalized_test_name(self) -> str:
"""Get the normalized version of the test name."""
return normalize_test_name(self.test_name)
def total_hook_runtime(self,
predicate: Optional[Callable[[HistoricHookInfo], bool]] = None) -> float:
"""Get the average runtime of all the hooks associated with this test."""
if not predicate:
predicate = lambda _: True
return sum([
hook.avg_duration * (hook.num_pass // self.num_pass if self.num_pass else 1)
for hook in self.hooks if predicate(hook)
])
def total_test_runtime(self) -> float:
"""Get the average runtime of this test and it's non-task level hooks."""
return self.avg_duration + self.total_hook_runtime(lambda h: not h.is_task_level_hook())
def get_hook_overhead(self) -> float:
"""Get the average runtime of this test and it's non-task level hooks."""
return self.total_hook_runtime(lambda h: h.is_task_level_hook())
class HistoricTaskData(object):
"""Represent the test statistics for the task that is being analyzed."""
def __init__(self, historic_test_results: List[HistoricTestInfo]) -> None:
"""Initialize the TestStats with raw results from the Evergreen API."""
self.historic_test_results = historic_test_results
@staticmethod
def get_stats_from_s3(project: str, task: str, variant: str) -> List[HistoricalTestInformation]:
"""
Retrieve test stats from s3 for a given task.
:param project: Project to query.
:param task: Task to query.
:param variant: Build variant to query.
:return: A list of the Test stats for the specified task.
"""
session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
response = session.get(f"{TESTS_STATS_S3_LOCATION}/{project}/{variant}/{task}")
data = response.json()
return [HistoricalTestInformation(**item) for item in data]
@classmethod
def from_s3(cls, project: str, task: str, variant: str) -> "HistoricTaskData":
"""
Retrieve test stats from s3 for a given task.
:param project: Project to query.
:param task: Task to query.
:param variant: Build variant to query.
:return: Test stats for the specified task.
"""
historical_test_data = cls.get_stats_from_s3(project, task, variant)
return cls.from_stats_list(historical_test_data)
@classmethod
def from_stats_list(
cls, historical_test_data: List[HistoricalTestInformation]) -> "HistoricTaskData":
"""
Build historic task data from a list of historic stats.
:param historical_test_data: A list of information about the runtime of a test.
:return: Historic task data from the list of stats.
"""
hooks = defaultdict(list)
for hook in [stat for stat in historical_test_data if is_resmoke_hook(stat.test_name)]:
historical_hook = HistoricHookInfo.from_test_stats(hook)
hooks[historical_hook.test_name()].append(historical_hook)
return cls([
HistoricTestInfo.from_test_stats(stat,
hooks[get_short_name_from_test_file(stat.test_name)])
for stat in historical_test_data if not is_resmoke_hook(stat.test_name)
])
def get_tests_runtimes(self) -> List[TestRuntime]:
"""Return the list of (test_file, runtime_in_secs) tuples ordered by decreasing runtime."""
tests = [
TestRuntime(test_name=test_stats.normalized_test_name(),
runtime=test_stats.total_test_runtime())
for test_stats in self.historic_test_results
]
return sorted(tests, key=lambda x: x.runtime, reverse=True)
def get_avg_hook_runtime(self, hook_name: str) -> float:
"""Get the average runtime for the specified hook."""
hook_instances = list(
chain.from_iterable([[hook for hook in test.hooks if hook.hook_name() == hook_name]
for test in self.historic_test_results]))
if not hook_instances:
return 0
return sum([hook.avg_duration for hook in hook_instances]) / len(hook_instances)
def __len__(self) -> int:
"""Get the number of historical entries."""
return len(self.historic_test_results)
|
