diff --git a/webui/backend/ray_ui.py b/webui/backend/ray_ui.py index e28815256..b8c107551 100644 --- a/webui/backend/ray_ui.py +++ b/webui/backend/ray_ui.py @@ -50,6 +50,7 @@ async def hgetall_as_dict(redis_conn, key): # Cache information about the local schedulers. local_schedulers = {} +errors = [] def duration_to_string(duration): """Format a duration in seconds as a string. @@ -125,6 +126,35 @@ async def handle_get_drivers(websocket, redis_conn): reply = sorted(drivers, key=(lambda driver: driver["start time"]))[::-1] await websocket.send(json.dumps(reply)) +async def listen_for_errors(redis_ip_address, redis_port): + pubsub_conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) + data_conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) + + error_pattern = "__keyspace@0__:ErrorKeys" + psub = await pubsub_conn.execute_pubsub("psubscribe", error_pattern) + channel = pubsub_conn.pubsub_patterns[error_pattern] + print("Listening for error messages...") + index = 0 + while (await channel.wait_message()): + msg = await channel.get() + info = await data_conn.execute("lrange", "ErrorKeys", index, -1) + + for error_key in info: + worker, task = key_to_hex_identifiers(error_key) + # TODO(richard): Filter out workers so that only relevant task errors are + # necessary. + result = await data_conn.execute("hget", error_key, "message") + result = result.decode("ascii") + # TODO(richard): Maybe also get rid of the coloring. + errors.append({"driver_id": worker, + "task_id": task, + "error": result}) + index += 1 + +async def handle_get_errors(websocket): + """Send error messages to the frontend.""" + await websocket.send(json.dumps(errors)) + node_info = collections.OrderedDict() worker_info = collections.OrderedDict() @@ -245,12 +275,18 @@ async def send_heartbeats(websocket, redis_conn): continue local_schedulers[local_scheduler_id]["last_heartbeat"] = time.time() -async def serve_requests(websocket, path): - redis_conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) +async def cache_data_from_redis(redis_ip_address, redis_port): + """Open up ports to listen for new updates from Redis.""" + # TODO(richard): A lot of code needs to be ported in order to open new + # websockets. + asyncio.ensure_future(listen_for_errors(redis_ip_address, redis_port)) + +async def serve_requests(websocket, path): # We loop infinitely because otherwise the websocket will be closed. # TODO(rkn): Maybe we should open a new web sockets for every request instead # of looping here. + redis_conn = await aioredis.create_connection((redis_ip_address, redis_port), loop=loop) while True: command = json.loads(await websocket.recv()) print("received command {}".format(command)) @@ -261,6 +297,8 @@ async def serve_requests(websocket, path): await handle_get_drivers(websocket, redis_conn) elif command["command"] == "get-recent-tasks": await handle_get_recent_tasks(websocket, redis_conn, command["num"]) + elif command["command"] == "get-errors": + await handle_get_errors(websocket) elif command["command"] == "get-heartbeats": await send_heartbeats(websocket, redis_conn) @@ -335,9 +373,12 @@ if __name__ == "__main__": redis_ip_address, redis_port = redis_address[0], int(redis_address[1]) # The port here must match the value used by the frontend to connect over - # websockets. + # websockets. TODO(richard): Automatically increment the port if it is already + # taken. port = 8888 - start_server = websockets.serve(serve_requests, "localhost", port) + loop.run_until_complete(cache_data_from_redis(redis_ip_address, redis_port)) + + start_server = websockets.serve(serve_requests, "localhost", port) loop.run_until_complete(start_server) loop.run_forever() diff --git a/webui/index.html b/webui/index.html index 97d1ed5fe..5ff9cdc8c 100644 --- a/webui/index.html +++ b/webui/index.html @@ -1,13 +1,3 @@ - - diff --git a/webui/src/ray-app.html b/webui/src/ray-app.html index 60a748b86..83e94e75e 100644 --- a/webui/src/ray-app.html +++ b/webui/src/ray-app.html @@ -1,13 +1,3 @@ - - @@ -75,6 +65,7 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN Objects Tasks Events + Errors Timeline Recent Tasks @@ -100,6 +91,7 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN + diff --git a/webui/src/ray-errors.html b/webui/src/ray-errors.html new file mode 100644 index 000000000..c671b4beb --- /dev/null +++ b/webui/src/ray-errors.html @@ -0,0 +1,46 @@ + + + + + + + + diff --git a/webui/src/ray-events.html b/webui/src/ray-events.html index 454ca9241..17ebece1e 100644 --- a/webui/src/ray-events.html +++ b/webui/src/ray-events.html @@ -1,13 +1,3 @@ - - diff --git a/webui/src/ray-icons.html b/webui/src/ray-icons.html index fcd758c80..8b93b1bc2 100644 --- a/webui/src/ray-icons.html +++ b/webui/src/ray-icons.html @@ -1,13 +1,3 @@ - - @@ -28,4 +18,4 @@ subject to an additional IP rights grant found at http://polymer.github.io/PATEN - \ No newline at end of file + diff --git a/webui/src/ray-objects.html b/webui/src/ray-objects.html index d7581a13d..47ee968ea 100644 --- a/webui/src/ray-objects.html +++ b/webui/src/ray-objects.html @@ -1,13 +1,3 @@ - - diff --git a/webui/src/ray-overview.html b/webui/src/ray-overview.html index 4362afb3a..5df612c70 100644 --- a/webui/src/ray-overview.html +++ b/webui/src/ray-overview.html @@ -1,13 +1,3 @@ - - diff --git a/webui/src/ray-tasks.html b/webui/src/ray-tasks.html index 0703f7060..4c8c9a0a0 100644 --- a/webui/src/ray-tasks.html +++ b/webui/src/ray-tasks.html @@ -1,13 +1,3 @@ - - diff --git a/webui/src/ray-timeline.html b/webui/src/ray-timeline.html index 5ad8b7b98..49375a108 100644 --- a/webui/src/ray-timeline.html +++ b/webui/src/ray-timeline.html @@ -1,13 +1,3 @@ - -