forked from ray-project/ray
-
Notifications
You must be signed in to change notification settings - Fork 15
/
datacenter.py
312 lines (276 loc) · 12.2 KB
/
datacenter.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
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
import logging
import ray.new_dashboard.consts as dashboard_consts
import ray.new_dashboard.memory_utils as memory_utils
# TODO(fyrestone): Not import from dashboard module.
from ray.new_dashboard.modules.actor.actor_utils import \
actor_classname_from_task_spec
from ray.new_dashboard.utils import Dict, Signal, async_loop_forever
logger = logging.getLogger(__name__)
class GlobalSignals:
node_info_fetched = Signal(dashboard_consts.SIGNAL_NODE_INFO_FETCHED)
node_summary_fetched = Signal(dashboard_consts.SIGNAL_NODE_SUMMARY_FETCHED)
job_info_fetched = Signal(dashboard_consts.SIGNAL_JOB_INFO_FETCHED)
worker_info_fetched = Signal(dashboard_consts.SIGNAL_WORKER_INFO_FETCHED)
class DataSource:
# {node id hex(str): node stats(dict of GetNodeStatsReply
# in node_manager.proto)}
node_stats = Dict()
# {node id hex(str): node physical stats(dict from reporter_agent.py)}
node_physical_stats = Dict()
# {actor id hex(str): actor table data(dict of ActorTableData
# in gcs.proto)}
actors = Dict()
# {job id hex(str): job table data(dict of JobTableData in gcs.proto)}
jobs = Dict()
# {node id hex(str): dashboard agent [http port(int), grpc port(int)]}
agents = Dict()
# {node id hex(str): gcs node info(dict of GcsNodeInfo in gcs.proto)}
nodes = Dict()
# {node id hex(str): ip address(str)}
node_id_to_ip = Dict()
# {node id hex(str): hostname(str)}
node_id_to_hostname = Dict()
# {node id hex(str): worker list}
node_workers = Dict()
# {node id hex(str): {actor id hex(str): actor table data}}
node_actors = Dict()
# {job id hex(str): worker list}
job_workers = Dict()
# {job id hex(str): {actor id hex(str): actor table data}}
job_actors = Dict()
# {worker id(str): core worker stats}
core_worker_stats = Dict()
# {job id hex(str): {event id(str): event dict}}
events = Dict()
# {node ip (str): log entries by pid
# (dict from pid to list of latest log entries)}
ip_and_pid_to_logs = Dict()
# {node ip (str): error entries by pid
# (dict from pid to list of latest err entries)}
ip_and_pid_to_errors = Dict()
class DataOrganizer:
@staticmethod
@async_loop_forever(dashboard_consts.PURGE_DATA_INTERVAL_SECONDS)
async def purge():
# Purge data that is out of date.
# These data sources are maintained by DashboardHead,
# we do not needs to purge them:
# * agents
# * nodes
# * node_id_to_ip
# * node_id_to_hostname
logger.info("Purge data.")
alive_nodes = {
node_id
for node_id, node_info in DataSource.nodes.items()
if node_info["state"] == "ALIVE"
}
for key in DataSource.node_stats.keys() - alive_nodes:
DataSource.node_stats.pop(key)
for key in DataSource.node_physical_stats.keys() - alive_nodes:
DataSource.node_physical_stats.pop(key)
@classmethod
@async_loop_forever(dashboard_consts.ORGANIZE_DATA_INTERVAL_SECONDS)
async def organize(cls):
job_workers = {}
node_workers = {}
core_worker_stats = {}
# await inside for loop, so we create a copy of keys().
for node_id in list(DataSource.nodes.keys()):
workers = await cls.get_node_workers(node_id)
for worker in workers:
job_id = worker["jobId"]
job_workers.setdefault(job_id, []).append(worker)
for stats in worker.get("coreWorkerStats", []):
worker_id = stats["workerId"]
core_worker_stats[worker_id] = stats
node_workers[node_id] = workers
DataSource.job_workers.reset(job_workers)
DataSource.node_workers.reset(node_workers)
DataSource.core_worker_stats.reset(core_worker_stats)
@classmethod
async def get_node_workers(cls, node_id):
workers = []
node_ip = DataSource.node_id_to_ip[node_id]
node_logs = DataSource.ip_and_pid_to_logs.get(node_ip, {})
node_errs = DataSource.ip_and_pid_to_errors.get(node_ip, {})
node_physical_stats = DataSource.node_physical_stats.get(node_id, {})
node_stats = DataSource.node_stats.get(node_id, {})
# Merge coreWorkerStats (node stats) to workers (node physical stats)
pid_to_worker_stats = {}
pid_to_language = {}
pid_to_job_id = {}
for core_worker_stats in node_stats.get("coreWorkersStats", []):
pid = core_worker_stats["pid"]
pid_to_worker_stats.setdefault(pid, []).append(core_worker_stats)
pid_to_language[pid] = core_worker_stats["language"]
pid_to_job_id[pid] = core_worker_stats["jobId"]
for worker in node_physical_stats.get("workers", []):
worker = dict(worker)
pid = worker["pid"]
worker["logCount"] = len(node_logs.get(str(pid), []))
worker["errorCount"] = len(node_errs.get(str(pid), []))
worker["coreWorkerStats"] = pid_to_worker_stats.get(pid, [])
worker["language"] = pid_to_language.get(
pid, dashboard_consts.DEFAULT_LANGUAGE)
worker["jobId"] = pid_to_job_id.get(
pid, dashboard_consts.DEFAULT_JOB_ID)
await GlobalSignals.worker_info_fetched.send(node_id, worker)
workers.append(worker)
return workers
@classmethod
async def get_node_info(cls, node_id):
node_physical_stats = dict(
DataSource.node_physical_stats.get(node_id, {}))
node_stats = dict(DataSource.node_stats.get(node_id, {}))
node = DataSource.nodes.get(node_id, {})
node_ip = DataSource.node_id_to_ip.get(node_id)
# Merge node log count information into the payload
log_info = DataSource.ip_and_pid_to_logs.get(node_ip, {})
node_log_count = 0
for entries in log_info.values():
node_log_count += len(entries)
error_info = DataSource.ip_and_pid_to_errors.get(node_ip, {})
node_err_count = 0
for entries in error_info.values():
node_err_count += len(entries)
node_stats.pop("coreWorkersStats", None)
view_data = node_stats.get("viewData", [])
ray_stats = cls._extract_view_data(
view_data,
{"object_store_used_memory", "object_store_available_memory"})
node_info = node_physical_stats
# Merge node stats to node physical stats under raylet
node_info["raylet"] = node_stats
node_info["raylet"].update(ray_stats)
# Merge GcsNodeInfo to node physical stats
node_info["raylet"].update(node)
# Merge actors to node physical stats
node_info["actors"] = DataSource.node_actors.get(node_id, {})
# Update workers to node physical stats
node_info["workers"] = DataSource.node_workers.get(node_id, [])
node_info["logCount"] = node_log_count
node_info["errorCount"] = node_err_count
await GlobalSignals.node_info_fetched.send(node_info)
return node_info
@classmethod
async def get_node_summary(cls, node_id):
node_physical_stats = dict(
DataSource.node_physical_stats.get(node_id, {}))
node_stats = dict(DataSource.node_stats.get(node_id, {}))
node = DataSource.nodes.get(node_id, {})
node_physical_stats.pop("workers", None)
node_stats.pop("workersStats", None)
view_data = node_stats.get("viewData", [])
ray_stats = cls._extract_view_data(
view_data,
{"object_store_used_memory", "object_store_available_memory"})
node_stats.pop("viewData", None)
node_summary = node_physical_stats
# Merge node stats to node physical stats
node_summary["raylet"] = node_stats
node_summary["raylet"].update(ray_stats)
# Merge GcsNodeInfo to node physical stats
node_summary["raylet"].update(node)
await GlobalSignals.node_summary_fetched.send(node_summary)
return node_summary
@classmethod
async def get_all_node_summary(cls):
return [
await DataOrganizer.get_node_summary(node_id)
for node_id in DataSource.nodes.keys()
]
@classmethod
async def get_all_node_details(cls):
return [
await DataOrganizer.get_node_info(node_id)
for node_id in DataSource.nodes.keys()
]
@classmethod
async def get_all_actors(cls):
return {
actor_id: await cls._get_actor(actor)
for actor_id, actor in DataSource.actors.items()
}
@staticmethod
async def _get_actor(actor):
actor = dict(actor)
worker_id = actor["address"]["workerId"]
core_worker_stats = DataSource.core_worker_stats.get(worker_id, {})
actor_constructor = core_worker_stats.get("actorTitle",
"Unknown actor constructor")
actor["actorConstructor"] = actor_constructor
actor.update(core_worker_stats)
# TODO(fyrestone): remove this, give a link from actor
# info to worker info in front-end.
node_id = actor["address"]["rayletId"]
pid = core_worker_stats.get("pid")
node_physical_stats = DataSource.node_physical_stats.get(node_id, {})
actor_process_stats = None
actor_process_gpu_stats = []
if pid:
for process_stats in node_physical_stats.get("workers", []):
if process_stats["pid"] == pid:
actor_process_stats = process_stats
break
for gpu_stats in node_physical_stats.get("gpus", []):
for process in gpu_stats.get("processes", []):
if process["pid"] == pid:
actor_process_gpu_stats.append(gpu_stats)
break
actor["gpus"] = actor_process_gpu_stats
actor["processStats"] = actor_process_stats
return actor
@classmethod
async def get_actor_creation_tasks(cls):
infeasible_tasks = sum(
(list(node_stats.get("infeasibleTasks", []))
for node_stats in DataSource.node_stats.values()), [])
new_infeasible_tasks = []
for task in infeasible_tasks:
task = dict(task)
task["actorClass"] = actor_classname_from_task_spec(task)
task["state"] = "INFEASIBLE"
new_infeasible_tasks.append(task)
resource_pending_tasks = sum(
(list(data.get("readyTasks", []))
for data in DataSource.node_stats.values()), [])
new_resource_pending_tasks = []
for task in resource_pending_tasks:
task = dict(task)
task["actorClass"] = actor_classname_from_task_spec(task)
task["state"] = "PENDING_RESOURCES"
new_resource_pending_tasks.append(task)
results = {
task["actorCreationTaskSpec"]["actorId"]: task
for task in new_resource_pending_tasks + new_infeasible_tasks
}
return results
@classmethod
async def get_memory_table(cls,
sort_by=memory_utils.SortingType.OBJECT_SIZE,
group_by=memory_utils.GroupByType.STACK_TRACE):
all_worker_stats = []
for node_stats in DataSource.node_stats.values():
all_worker_stats.extend(node_stats.get("coreWorkersStats", []))
memory_information = memory_utils.construct_memory_table(
all_worker_stats, group_by=group_by, sort_by=sort_by)
return memory_information
@staticmethod
def _extract_view_data(views, data_keys):
view_data = {}
for view in views:
view_name = view["viewName"]
if view_name in data_keys:
if not view.get("measures"):
view_data[view_name] = 0
continue
measure = view["measures"][0]
if "doubleValue" in measure:
measure_value = measure["doubleValue"]
elif "intValue" in measure:
measure_value = measure["intValue"]
else:
measure_value = 0
view_data[view_name] = measure_value
return view_data