Put all log files in redis and visualize them in UI. (#350)

* Start process for monitoring log files and push changes to redis.

* Display log files in UI.

* Bug fix for recent tasks.

* Use flatbuffers to parse local scheduler heartbeats.

webui/backend/ray_ui.py

@@ -12,6 +12,9 @@ import sys
 import time
 import websockets
 
+# Import flatbuffer bindings.
+from ray.core.generated.LocalSchedulerInfoMessage import LocalSchedulerInfoMessage
+
 parser = argparse.ArgumentParser(description="parse information for the web ui")
 parser.add_argument("--redis-address", required=True, type=str, help="the address to use for redis")
 
@@ -212,17 +215,19 @@ async def handle_get_recent_tasks(websocket, redis_conn, num_tasks):
       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[node_index]["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,
-           "node_ip_address": node_ip_address,
-           "task_formatted_time": duration_to_string(task_times[1] - task_times[0]),
-           "get_arguments_formatted_time": duration_to_string(task_get_arguments_times[1] - task_get_arguments_times[0]),
-           "execute_formatted_time": duration_to_string(task_execute_times[1] - task_execute_times[0]),
-           "store_outputs_formatted_time": duration_to_string(task_store_outputs_times[1] - task_store_outputs_times[0])})
+      task_info = {"task": task_times,
+                   "get_arguments": task_get_arguments_times,
+                   "execute": task_execute_times,
+                   "store_outputs": task_store_outputs_times,
+                   "worker_index": worker_index,
+                   "node_ip_address": node_ip_address,
+                   "task_formatted_time": duration_to_string(task_times[1] - task_times[0]),
+                   "get_arguments_formatted_time": duration_to_string(task_get_arguments_times[1] - task_get_arguments_times[0])}
+      if len(task_execute_times) == 2:
+        task_info["execute_formatted_time"] = duration_to_string(task_execute_times[1] - task_execute_times[0])
+      if len(task_store_outputs_times) == 2:
+        task_info["store_outputs_formatted_time"] = duration_to_string(task_store_outputs_times[1] - task_store_outputs_times[0])
+      task_data[node_index]["task_data"].append(task_info)
       num_tasks += 1
     reply = {"min_time": min_time,
              "max_time": max_time,
@@ -267,7 +272,8 @@ async def send_heartbeats(websocket, redis_conn):
 
   while True:
     msg = await redis_conn.pubsub_channels["local_schedulers"].get()
-    local_scheduler_id_bytes = msg[:IDENTIFIER_LENGTH]
+    heartbeat = LocalSchedulerInfoMessage.GetRootAsLocalSchedulerInfoMessage(msg, 0)
+    local_scheduler_id_bytes = heartbeat.DbClientId()
     local_scheduler_id = hex_identifier(local_scheduler_id_bytes)
     if local_scheduler_id not in local_schedulers:
       # A new local scheduler has joined the cluster. Ignore it. This won't be
@@ -282,10 +288,25 @@ async def cache_data_from_redis(redis_ip_address, redis_port):
 
   asyncio.ensure_future(listen_for_errors(redis_ip_address, redis_port))
 
+async def handle_get_log_files(websocket, redis_conn):
+  reply = {}
+  # First get all keys for the log file lists.
+  log_file_list_keys = await redis_conn.execute("keys", "LOG_FILENAMES:*")
+  for log_file_list_key in log_file_list_keys:
+    node_ip_address = log_file_list_key.decode("ascii").split(":")[1]
+    reply[node_ip_address] = {}
+    # Get all of the log filenames for this node IP address.
+    log_filenames = await redis_conn.execute("lrange", log_file_list_key, 0, -1)
+    for log_filename in log_filenames:
+      log_filename_key = "LOGFILE:{}:{}".format(node_ip_address, log_filename.decode("ascii"))
+      logfile = await redis_conn.execute("lrange", log_filename_key, 0, -1)
+      logfile = [line.decode("ascii") for line in logfile]
+      reply[node_ip_address][log_filename.decode("ascii")] = logfile
+
+  # Send the reply back to the front end.
+  await websocket.send(json.dumps(reply))
+
 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())
@@ -301,6 +322,8 @@ async def serve_requests(websocket, path):
       await handle_get_errors(websocket)
     elif command["command"] == "get-heartbeats":
       await send_heartbeats(websocket, redis_conn)
+    elif command["command"] == "get-log-files":
+      await handle_get_log_files(websocket, redis_conn)
 
     if command["command"] == "get-workers":
       result = []

webui/src/ray-app.html

@@ -68,6 +68,7 @@
           <a name="errors" href="/errors">Errors</a>
           <a name="timeline" href="/timeline">Timeline</a>
           <a name="recent-tasks" href="/recent-tasks">Recent Tasks</a>
+          <a name="log-files" href="/log-files">Log Files</a>
         </iron-selector>
       </app-drawer>
 
@@ -94,6 +95,7 @@
           <ray-errors name="errors"></ray-errors>
           <ray-timeline name="timeline"></ray-timeline>
           <ray-recent-tasks name="recent-tasks"></ray-recent-tasks>
+          <ray-log-files name="log-files"></ray-log-files>
           <ray-view404 name="view404"></ray-view404>
         </iron-pages>
       </app-header-layout>

webui/src/ray-log-files.html

@@ -0,0 +1,90 @@
+<link rel="import" href="../bower_components/polymer/polymer.html">
+<link rel="import" href="shared-styles.html">
+
+<dom-module id="ray-log-files">
+  <template>
+    <style include="shared-styles">
+      :host {
+        display: block;
+
+        padding: 10px;
+      }
+      rect:hover
+      {
+        opacity: 0.5;
+      }
+
+    </style>
+
+    <div class="card">
+      <h1>Ray Log Files</h1>
+      <div id="all_log_files"></div>
+
+    </div>
+  </template>
+
+  <script>
+    var backend_address = "ws://127.0.0.1:8888";
+
+    Polymer({
+      is: 'ray-log-files',
+      ready: function() {
+
+        var self = this;
+        var socket = new WebSocket(backend_address);
+
+        socket.onopen = function() {
+          socket.send(JSON.stringify({"command": "get-log-files"}));
+        }
+
+        socket.onmessage = function(messageEvent) {
+          var reply = JSON.parse(messageEvent.data);
+          console.log(reply);
+          for (i = 0; i < Object.keys(reply).length; i++) {
+            key = Object.keys(reply)[i];
+            dict = reply[key];
+
+            var node = document.createElement("LI");
+            var bold_node = document.createElement("B");
+            var textnode = document.createTextNode(key);
+            bold_node.appendChild(textnode);
+            node.appendChild(bold_node);
+
+            var node_log_files_node = document.createElement("UL");
+
+            for (j = 0; j < Object.keys(dict).length; j++) {
+              file_key = Object.keys(dict)[j];
+              line_list = dict[file_key];
+              console.log(file_key);
+              console.log(line_list);
+
+              var logfile_node = document.createElement("LI");
+              var log_filename_node = document.createTextNode(file_key);
+              var bold_node = document.createElement("B");
+              bold_node.appendChild(log_filename_node);
+              logfile_node.appendChild(bold_node);
+
+              for (k = 0; k < line_list.length; k++) {
+                var paragraph = document.createElement("BR");
+                var logfile_line_node = document.createTextNode(line_list[k]);
+                logfile_node.append(paragraph);
+                logfile_node.append(logfile_line_node);
+              }
+              node_log_files_node.append(logfile_node);
+
+            }
+            node.appendChild(node_log_files_node);
+
+            self.$.all_log_files.appendChild(node);
+          }
+
+        }
+
+        socket.onclose = function(closeEvent) {
+          console.log(closeEvent)
+        }
+      },
+
+    });
+  </script>
+</dom-module>