forked from gioenn/dynaSpark-launcher
-
Notifications
You must be signed in to change notification settings - Fork 0
/
metrics.py
228 lines (197 loc) · 8.54 KB
/
metrics.py
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
"""
"""
import glob
import json
import math
from datetime import timedelta
from pathlib import Path
import numpy as np
from log import load_app_data, load_worker_data
from util.utils import timing
PLOT_SID_STAGE = 0
def load_config(folder):
"""
:param folder:
:return:
"""
config_file = Path(folder + "config.json")
print(config_file)
if config_file.exists():
config = json.load(open(folder + "config.json"))
if len(config) == 0:
from config import CONFIG_DICT
return CONFIG_DICT
else:
return config
else:
from config import CONFIG_DICT
return CONFIG_DICT
def compute_cpu_time(app_id, app_info, workers_dict, config, folder):
"""
:param app_id:
:param app_info:
:param workers_dict:
:param config:
:param folder:
:return:
"""
cpu_time = 0
cpu_time_max = 0
cpus = 0.0
for worker_log in workers_dict:
worker_dict = workers_dict[worker_log]
try:
for sid in worker_dict[app_id]:
cpus += sum(worker_dict[app_id][sid]["cpu"])
cpu_time += (config["Control"]["TSample"] / 1000) * sum(
worker_dict[app_id][sid]["cpu"])
time_cpu = worker_dict["time_cpu"]
for cpu, time in zip(worker_dict[app_id][sid]["cpu"],
worker_dict[app_id][sid]["time"]):
try:
index = time_cpu.index(time)
except ValueError:
index = min(range(len(time_cpu)), key=lambda i: abs(time_cpu[i] - time))
# print(index)
cpu_time_max += (config["Control"]["TSample"] / 1000) * max(cpu, worker_dict[
"cpu_real"][index + int(config["Control"]["TSample"] / 1000)])
except KeyError:
print(app_id + " not found")
duration_s = app_info[app_id][max(list(app_info[app_id].keys()))]["end"].timestamp() - \
app_info[app_id][PLOT_SID_STAGE]["start"].timestamp()
if cpus == 0.0:
speed = config["Control"]["MaxExecutor"] * config["Control"]["CoreVM"]
speed_20 = math.ceil(config["Control"]["CoreVM"] * duration_s / 525.8934) * \
config["Control"]["MaxExecutor"]
speed_40 = math.ceil(config["Control"]["CoreVM"] * duration_s / 613.5423) * \
config["Control"]["MaxExecutor"]
print(duration_s)
print("SPEED NATIVE 0%", speed)
print("SPEED NATIVE 20% ", speed_20)
print("SPEED NATIVE 40% ", speed_40)
else:
speed = (float(cpus) * (config["Control"]["TSample"] / 1000)) / duration_s
num_task = 0.0
for sid in app_info[app_id]:
num_task += len(app_info[app_id][sid]["tasktimestamps"])
throughput = float(num_task) / duration_s
if cpu_time == 0:
cpu_time = ((app_info[app_id][max(list(app_info[app_id].keys()))]["end"].timestamp() -
app_info[app_id][PLOT_SID_STAGE]["start"].timestamp())) * \
config["Control"]["MaxExecutor"] * config["Control"]["CoreVM"]
cpu_time_max = cpu_time
cpu_time_max = math.floor(cpu_time_max)
print("CPU_TIME: " + str(cpu_time))
print("CPU TIME MAX: " + str(cpu_time_max))
print("SID " + str(app_info[app_id].keys()))
print("CHECK NON CONTROLLED STAGE FOR CPU_TIME")
with open(folder + "CPU_TIME.txt", "w") as cpu_time_f:
cpu_time_f.write("CPU_TIME " + str(cpu_time) + "\n")
cpu_time_f.write("CPU_TIME_MAX " + str(cpu_time_max) + "\n")
cpu_time_f.write("SPEED " + str(speed) + "\n")
cpu_time_f.write("THROUGHPUT " + str(throughput) + "\n")
def save_deadline_errors(folder, deadline_error, stage_errors):
"""Save the error of the application's stages in the folder
with some statistics (mean, stddev, median, max, min)
:param folder: the output folder
:param deadline_error: the application's deadline error
:param stage_errors: the list of the stages errors
:return: Nothing
"""
with open(folder + "ERROR.txt", "w") as error_f:
error_f.write("DEADLINE_ERROR " + str(abs(deadline_error)) + "\n")
if len(stage_errors) > 0:
error_f.write("MEAN_ERROR " + str(np.mean(stage_errors)) + "\n")
error_f.write("DEVSTD_ERROR: " + str(np.std(stage_errors)) + "\n")
error_f.write("MEDIAN_ERROR: " + str(np.median(stage_errors)) + "\n")
error_f.write("MAX_ERROR: " + str(max(stage_errors)) + "\n")
error_f.write("MIN_ERROR: " + str(min(stage_errors)) + "\n")
def compute_errors(app_id, app_dict, folder, config):
if len(app_dict) > 0:
timestamps = []
times = []
app_deadline = 0
first_ts = app_dict[PLOT_SID_STAGE]["start"].timestamp()
for sid in sorted(app_dict):
try:
app_deadline = app_dict[PLOT_SID_STAGE]["start"] + timedelta(
milliseconds=config["Deadline"])
app_deadline = app_deadline.replace(microsecond=0)
for timestamp in app_dict[sid]["tasktimestamps"]:
if first_ts == 0:
timestamps.append(0.0)
first_ts = timestamp.timestamp()
else:
timestamps.append(timestamp.timestamp() - first_ts)
if len(times) == 0:
times.append(1)
else:
times.append(times[-1] + 1)
except KeyError:
None
app_alpha_deadline = app_deadline - timedelta(
milliseconds=((1 - float(config["Control"]["Alpha"])) * float(config["Deadline"])))
app_alpha_deadline_ts = app_alpha_deadline.timestamp() - first_ts
# COMPUTE ERRORS
errors = []
sorted_sid = sorted(app_dict)
total_duration = app_alpha_deadline.timestamp() - app_dict[PLOT_SID_STAGE][
"start"].timestamp()
for sid in sorted_sid:
try:
start_ts = app_dict[sid]["start"].timestamp() - first_ts
end = app_dict[sid]["end"].timestamp()
end_ts = end - first_ts
int_dead = app_dict[sid]["deadline"].timestamp()
dead_ts = int_dead - first_ts
if sid == sorted_sid[-1] and start_ts < app_alpha_deadline_ts:
dead_ts = app_alpha_deadline_ts
deadline_error = round(round(((abs(dead_ts - end_ts)) / total_duration), 4) * 100,
3)
print(sid, abs(int_dead - end), total_duration, end, deadline_error)
errors.append(deadline_error)
except KeyError:
None
end = app_dict[sorted_sid[-1]]["end"].timestamp()
print(abs(app_alpha_deadline.timestamp() - end), total_duration, end)
app_deadline_error = round(
round(((abs(app_alpha_deadline.timestamp() - end)) / total_duration), 4) * 100, 3)
stage_errors = np.array(errors)
print("DEADLINE_ERROR " + str(app_deadline_error))
if len(stage_errors) > 0:
print("MEAN ERROR: " + str(np.mean(stage_errors)))
print("DEVSTD ERROR: " + str(np.std(stage_errors)))
print("MEDIAN ERROR: " + str(np.median(stage_errors)))
print("MAX ERROR: " + str(max(stage_errors)))
print("MIN ERROR: " + str(min(stage_errors)))
save_deadline_errors(folder, app_deadline_error, stage_errors)
@timing
def compute_metrics(folder):
"""
:param folder:
:return:
"""
print(folder)
if folder[-1] != "/":
folder += "/"
config = load_config(folder)
print(config)
global PLOT_SID_STAGE
PLOT_SID_STAGE = 1 if config["HDFS"] else 0
app_logs = glob.glob(folder + "*.err") + glob.glob(folder + "*.dat")
app_info = {}
for app_log in sorted(app_logs):
app_info = load_app_data(app_log)
for app_id in app_info:
compute_errors(app_id, app_info[app_id], folder, config)
worker_logs = glob.glob(folder + "*worker*.out")
cpu_logs = glob.glob(folder + "sar*.log")
if len(worker_logs) == len(cpu_logs):
workers_dict = {}
for worker_log, cpu_log in zip(sorted(worker_logs), sorted(cpu_logs)):
worker_dict = load_worker_data(worker_log, cpu_log, config)
workers_dict[worker_log] = worker_dict
for app_id in app_info:
compute_cpu_time(app_id, app_info, workers_dict, config, folder)
else:
print("ERROR: SAR != WORKER LOGS")