From 7a7e14ef855dc4018609ebb6a95db1a5ba5c2fa6 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Thu, 2 Feb 2017 15:53:56 -0800 Subject: [PATCH] Visualize recent tasks in timeline. (#240) --- webui/README.md | 3 +- webui/backend/ray_ui.py | 83 +++++++++++++++++++++++++++------ webui/index.html | 2 + webui/src/ray-app.html | 2 + webui/src/ray-recent-tasks.html | 76 ++++++++++++++++++++++++++++++ webui/src/recent-tasks.js | 74 +++++++++++++++++++++++++++++ 6 files changed, 223 insertions(+), 17 deletions(-) create mode 100644 webui/src/ray-recent-tasks.html create mode 100644 webui/src/recent-tasks.js diff --git a/webui/README.md b/webui/README.md index 6bd97d807..07cc2a76c 100644 --- a/webui/README.md +++ b/webui/README.md @@ -3,8 +3,7 @@ This is Ray's Web UI. It consists of two components: * The **frontend** is a [Polymer](https://www.polymer-project.org/1.0/) app that - uses [google-charts](https://elements.polymer-project.org/elements/google-chart) - for visualization. + uses [D3](https://d3js.org/) for visualization. * The **backend** is a Python 3 websocket server (see `backend/ray_ui.py`) that connects to Redis and potentially Ray. diff --git a/webui/backend/ray_ui.py b/webui/backend/ray_ui.py index 93b6cbaab..d0dbb5d96 100644 --- a/webui/backend/ray_ui.py +++ b/webui/backend/ray_ui.py @@ -4,6 +4,8 @@ import asyncio import binascii from collections import defaultdict import json +import numpy as np +import redis import websockets parser = argparse.ArgumentParser(description="parse information for the web ui") @@ -31,36 +33,87 @@ def key_to_hex_identifiers(key): task_id = hex_identifier(key[offset:(offset + IDENTIFIER_LENGTH)]) return worker_id, task_id +worker_ids = [] -async def hello(websocket, path): - conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) +async def handle_get_recent_tasks(websocket, redis_conn, num_tasks): + keys = await redis_conn.execute("keys", "event_log:*") + if len(keys) == 0: + # There are no tasks, so send a message to the client saying so. + await websocket.send(json.dumps({"num_tasks": 0})) + else: + timestamps = [] + contents = [] + for key in keys: + content = await redis_conn.execute("lrange", key, "0", "-1") + contents.append(json.loads(content[0].decode())) + timestamps += [timestamp for (timestamp, task, kind, info) in contents[-1] if task == "ray:task"] + timestamps.sort() + time_cutoff = timestamps[(-2 * num_tasks):][0] + + max_time = timestamps[-1] + min_time = time_cutoff - (max_time - time_cutoff) * 0.1 + max_time = max_time + (max_time - time_cutoff) * 0.1 + + task_data = [] + for i in range(len(keys)): + worker_id, task_id = key_to_hex_identifiers(keys[i]) + data = contents[i] + if worker_id not in worker_ids: + worker_ids.append(worker_id) + worker_index = worker_ids.index(worker_id) + + task_times = [timestamp for (timestamp, task, kind, info) in data if task == "ray:task"] + if task_times[1] <= time_cutoff: + continue + + task_get_arguments_times = [timestamp for (timestamp, task, kind, info) in data if task == "ray:task:get_arguments"] + task_execute_times = [timestamp for (timestamp, task, kind, info) in data if task == "ray:task:execute"] + task_store_outputs_times = [timestamp for (timestamp, task, kind, info) in data if task == "ray:task:store_outputs"] + task_data.append({"task": task_times, + "get_arguments": task_get_arguments_times, + "execute": task_execute_times, + "store_outputs": task_store_outputs_times, + "worker_index": worker_index}) + reply = {"min_time": min_time, + "max_time": max_time, + "num_tasks": len(task_data), + "task_data": task_data} + await websocket.send(json.dumps(reply)) + +async def serve_requests(websocket, path): + redis_conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) + + # We loop infinitely because otherwise the websocket will be closed. while True: command = json.loads(await websocket.recv()) print("received command {}".format(command)) + if command["command"] == "get-recent-tasks": + await handle_get_recent_tasks(websocket, redis_conn, command["num"]) + if command["command"] == "get-workers": result = [] - workers = await conn.execute("keys", "WorkerInfo:*") + workers = await redis_conn.execute("keys", "WorkerInfo:*") for key in workers: - content = await conn.execute("hgetall", key) + content = await redis_conn.execute("hgetall", key) worker_id = key_to_hex_identifier(key) result.append({"worker": worker_id, "export_counter": int(content[1])}) await websocket.send(json.dumps(result)) elif command["command"] == "get-clients": result = [] - clients = await conn.execute("keys", "CL:*") + clients = await redis_conn.execute("keys", "CL:*") for key in clients: - content = await conn.execute("hgetall", key) + content = await redis_conn.execute("hgetall", key) result.append({"client": hex_identifier(content[1]), "node_ip_address": content[3].decode(), "client_type": content[5].decode()}) await websocket.send(json.dumps(result)) elif command["command"] == "get-objects": result = [] - objects = await conn.execute("keys", "OI:*") + objects = await redis_conn.execute("keys", "OI:*") for key in objects: - content = await conn.execute("hgetall", key) + content = await redis_conn.execute("hgetall", key) result.append({"object_id": hex_identifier(content[1]), "hash": hex_identifier(content[3]), "data_size": content[5].decode()}) @@ -73,18 +126,18 @@ async def hello(websocket, path): await websocket.send(json.dumps({"object_id": "none"})) elif command["command"] == "get-tasks": result = [] - tasks = await conn.execute("keys", "TT:*") + tasks = await redis_conn.execute("keys", "TT:*") for key in tasks: - content = await conn.execute("hgetall", key) + content = await redis_conn.execute("hgetall", key) result.append({"task_id": key_to_hex_identifier(key), "state": int(content[1]), "node_id": hex_identifier(content[3])}) await websocket.send(json.dumps(result)) elif command["command"] == "get-timeline": tasks = defaultdict(list) - for key in await conn.execute("keys", "event_log:*"): + for key in await redis_conn.execute("keys", "event_log:*"): worker_id, task_id = key_to_hex_identifiers(key) - content = await conn.execute("lrange", key, "0", "-1") + content = await redis_conn.execute("lrange", key, "0", "-1") data = json.loads(content[0].decode()) begin_and_end_time = [timestamp for (timestamp, task, kind, info) in data if task == "ray:task"] tasks[worker_id].append({"task_id": task_id, @@ -93,9 +146,9 @@ async def hello(websocket, path): await websocket.send(json.dumps(tasks)) elif command["command"] == "get-events": result = [] - for key in await conn.execute("keys", "event_log:*"): + for key in await redis_conn.execute("keys", "event_log:*"): worker_id, task_id = key_to_hex_identifiers(key) - answer = await conn.execute("lrange", key, "0", "-1") + answer = await redis_conn.execute("lrange", key, "0", "-1") assert len(answer) == 1 events = json.loads(answer[0].decode()) result.extend([{"worker_id": worker_id, @@ -109,7 +162,7 @@ if __name__ == "__main__": redis_address = args.redis_address.split(":") redis_ip_address, redis_port = redis_address[0], int(redis_address[1]) - start_server = websockets.serve(hello, "localhost", args.port) + start_server = websockets.serve(serve_requests, "localhost", args.port) loop.run_until_complete(start_server) loop.run_forever() diff --git a/webui/index.html b/webui/index.html index d91861d7c..97d1ed5fe 100644 --- a/webui/index.html +++ b/webui/index.html @@ -15,6 +15,8 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN + + Ray UI diff --git a/webui/src/ray-app.html b/webui/src/ray-app.html index 218873458..beecbcfbb 100644 --- a/webui/src/ray-app.html +++ b/webui/src/ray-app.html @@ -75,6 +75,7 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN Tasks Events Timeline + Recent Tasks @@ -98,6 +99,7 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN + diff --git a/webui/src/ray-recent-tasks.html b/webui/src/ray-recent-tasks.html new file mode 100644 index 000000000..f8abdf0e4 --- /dev/null +++ b/webui/src/ray-recent-tasks.html @@ -0,0 +1,76 @@ + + + + + + + + + + diff --git a/webui/src/recent-tasks.js b/webui/src/recent-tasks.js new file mode 100644 index 000000000..2f5e87d8d --- /dev/null +++ b/webui/src/recent-tasks.js @@ -0,0 +1,74 @@ +RecentTasks = function(elem, options) { + var self = this; + + this.options = options; + var barHeight = 25; + + var svg = d3.select(elem) + .attr("width", this.options.width) + .attr("height", this.options.height); + + this.draw_new_tasks = function(task_info) { + this.task_info = task_info; + var x = d3.scaleLinear() + .domain([task_info.min_time, task_info.max_time]) + .range([-1, width + 1]); + + var xAxis = d3.axisBottom(x) + .tickSize(-height); + + var gx = svg.append("g") + .attr("class", "axis axis--x") + .attr("transform", "translate(0," + (height - 10) + ")") + .call(xAxis); + + var task_rects = svg.append("g").attr("class", "task_rects"); + var get_arguments_rects = svg.append("g").attr("class", "get_arguments_rects"); + var execute_rects = svg.append("g").attr("class", "execute_rects"); + var store_outputs_rects = svg.append("g").attr("class", "store_outputs_rects"); + + task_rects.selectAll("rect") + .data(this.task_info.task_data) + .enter() + .append("rect") + .attr("x", function (d) { return x(d.task[0]); }) + .attr("y", function (d) { return (d.worker_index + 1) * barHeight; }) + .attr("width", function (d) { return x(d.task[1]) - x(d.task[0]); }) + .attr("height", function (d) { return barHeight - 1; }) + .attr("fill", "orange") + + get_arguments_rects.selectAll("rect") + .data(this.task_info.task_data) + .enter() + .append("rect") + .attr("x", function (d) { return x(d.get_arguments[0]); }) + .attr("y", function (d) { return (d.worker_index + 1) * barHeight + 1; }) + .attr("width", function (d) { return x(d.get_arguments[1]) - x(d.get_arguments[0]); }) + .attr("height", function (d) { return barHeight - 3; }) + .attr("fill", "black") + + execute_rects.selectAll("rect") + .data(this.task_info.task_data) + .enter() + .append("rect") + .attr("x", function (d) { return x(d.execute[0]); }) + .attr("y", function (d) { return (d.worker_index + 1) * barHeight + 1; }) + .attr("width", function (d) { return x(d.execute[1]) - x(d.execute[0]); }) + .attr("height", function (d) { return barHeight - 3; }) + .attr("fill", "blue") + + store_outputs_rects.selectAll("rect") + .data(this.task_info.task_data) + .enter() + .append("rect") + .attr("x", function (d) { return x(d.store_outputs[0]); }) + .attr("y", function (d) { return (d.worker_index + 1) * barHeight + 1; }) + .attr("width", function (d) { return x(d.store_outputs[1]) - x(d.store_outputs[0]); }) + .attr("height", function (d) { return barHeight - 3; }) + .attr("fill", "green") + } + + this.erase = function() { + svg.selectAll("g").remove() + } +}