[Dashboard] Dashboard basic modules (#10303)

* Improve reporter module

* Add test_node_physical_stats to test_reporter.py

* Add test_class_method_route_table to test_dashboard.py

* Add stats_collector module for dashboard

* Subscribe actor table data

* Add log module for dashboard

* Only enable test module in some test cases

* CI run all dashboard tests

* Reduce test timeout to 10s

* Use fstring

* Remove unused code

* Remove blank line

* Fix dashboard tests

* Fix asyncio.create_task not available in py36; Fix lint

* Add format_web_url to ray.test_utils

* Update dashboard/modules/reporter/reporter_head.py

Co-authored-by: Max Fitton <mfitton@berkeley.edu>

* Add DictChangeItem type for Dict change

* Refine logger.exception

* Refine GET /api/launch_profiling

* Remove disable_test_module fixture

* Fix test_basic may fail

Co-authored-by: 刘宝 <po.lb@antfin.com>
Co-authored-by: Max Fitton <mfitton@berkeley.edu>

bazel/python.bzl

@@ -1,12 +1,22 @@
-# py_test_module_list creates a py_test target for each 
+# py_test_module_list creates a py_test target for each
 # Python file in `files`
 def py_test_module_list(files, size, deps, extra_srcs, **kwargs):
-  for file in files:
-    # remove .py
-    name = file[:-3]
-    native.py_test(
-      name=name,
-      size=size,
-      srcs=extra_srcs+[file],
-      **kwargs
-    )
+    for file in files:
+        # remove .py
+        name = file[:-3]
+        native.py_test(
+            name = name,
+            size = size,
+            srcs = extra_srcs + [file],
+            **kwargs
+        )
+
+def py_test_run_all_subdirectory(include, exclude, extra_srcs, **kwargs):
+    for file in native.glob(include = include, exclude = exclude):
+        print(file)
+        basename = file.rpartition("/")[-1]
+        native.py_test(
+            name = basename[:-3],
+            srcs = extra_srcs + [file],
+            **kwargs
+        )

dashboard/BUILD

@@ -1,13 +1,19 @@
+load("//bazel:python.bzl", "py_test_run_all_subdirectory")
+
 # This is a dummy test dependency that causes the above tests to be
 # re-run if any of these files changes.
 py_library(
     name = "dashboard_lib",
-    srcs = glob(["**/*.py"],exclude=["tests/*"]),
+    srcs = glob(
+        ["**/*.py"],
+        exclude = ["tests/*"],
+    ),
 )
 
-py_test(
-    name = "test_dashboard",
+py_test_run_all_subdirectory(
     size = "small",
-    srcs = glob(["tests/*.py"]),
-    deps = [":dashboard_lib"]
+    include = ["**/test*.py"],
+    exclude = ["modules/test/**"],
+    extra_srcs = [],
+    tags = ["exclusive"],
 )

dashboard/agent.py

@@ -24,6 +24,11 @@ from ray.core.generated import agent_manager_pb2
 from ray.core.generated import agent_manager_pb2_grpc
 import psutil
 
+try:
+    create_task = asyncio.create_task
+except AttributeError:
+    create_task = asyncio.ensure_future
+
 logger = logging.getLogger(__name__)
 routes = dashboard_utils.ClassMethodRouteTable
 
@@ -36,6 +41,7 @@ class DashboardAgent(object):
                  redis_password=None,
                  temp_dir=None,
                  log_dir=None,
+                 metrics_export_port=None,
                  node_manager_port=None,
                  object_store_name=None,
                  raylet_name=None):
@@ -45,6 +51,7 @@ class DashboardAgent(object):
         self.redis_password = redis_password
         self.temp_dir = temp_dir
         self.log_dir = log_dir
+        self.metrics_export_port = metrics_export_port
         self.node_manager_port = node_manager_port
         self.object_store_name = object_store_name
         self.raylet_name = raylet_name
@@ -69,7 +76,7 @@ class DashboardAgent(object):
             c = cls(self)
             dashboard_utils.ClassMethodRouteTable.bind(c)
             modules.append(c)
-        logger.info("Loaded {} modules.".format(len(modules)))
+        logger.info("Loaded %d modules.", len(modules))
         return modules
 
     async def run(self):
@@ -86,7 +93,7 @@ class DashboardAgent(object):
                     dashboard_consts.
                     DASHBOARD_AGENT_CHECK_PARENT_INTERVAL_SECONDS)
 
-        check_parent_task = asyncio.create_task(_check_parent())
+        check_parent_task = create_task(_check_parent())
 
         # Create an aioredis client for all modules.
         try:
@@ -172,6 +179,11 @@ if __name__ == "__main__":
         required=True,
         type=str,
         help="The address to use for Redis.")
+    parser.add_argument(
+        "--metrics-export-port",
+        required=True,
+        type=int,
+        help="The port to expose metrics through Prometheus.")
     parser.add_argument(
         "--node-manager-port",
         required=True,
@@ -277,6 +289,7 @@ if __name__ == "__main__":
             redis_password=args.redis_password,
             temp_dir=temp_dir,
             log_dir=log_dir,
+            metrics_export_port=args.metrics_export_port,
             node_manager_port=args.node_manager_port,
             object_store_name=args.object_store_name,
             raylet_name=args.raylet_name)

dashboard/dashboard.py

@@ -37,10 +37,10 @@ def setup_static_dir():
             errno.ENOENT, "Dashboard build directory not found. If installing "
             "from source, please follow the additional steps "
             "required to build the dashboard"
-            "(cd python/ray/{}/client "
+            f"(cd python/ray/{module_name}/client "
             "&& npm install "
             "&& npm ci "
-            "&& npm run build)".format(module_name), build_dir)
+            "&& npm run build)", build_dir)
 
     static_dir = os.path.join(build_dir, "static")
     routes.static("/static", static_dir, follow_symlinks=True)
@@ -59,14 +59,21 @@ class Dashboard:
         port(int): Port number of dashboard aiohttp server.
         redis_address(str): GCS address of a Ray cluster
         redis_password(str): Redis password to access GCS
+        log_dir(str): Log directory of dashboard.
     """
 
-    def __init__(self, host, port, redis_address, redis_password=None):
+    def __init__(self,
+                 host,
+                 port,
+                 redis_address,
+                 redis_password=None,
+                 log_dir=None):
         self.dashboard_head = dashboard_head.DashboardHead(
             http_host=host,
             http_port=port,
             redis_address=redis_address,
-            redis_password=redis_password)
+            redis_password=redis_password,
+            log_dir=log_dir)
 
         # Setup Dashboard Routes
         build_dir = setup_static_dir()
@@ -197,7 +204,8 @@ if __name__ == "__main__":
             args.host,
             args.port,
             args.redis_address,
-            redis_password=args.redis_password)
+            redis_password=args.redis_password,
+            log_dir=log_dir)
         loop = asyncio.get_event_loop()
         loop.run_until_complete(dashboard.run())
     except Exception as e:

dashboard/datacenter.py

@@ -19,7 +19,7 @@ class DataSource:
     # {actor id hex(str): actor table data(dict of ActorTableData
     # in gcs.proto)}
     actors = Dict()
-    # {ip address(str): dashboard agent grpc server port(int)}
+    # {ip address(str): dashboard agent [http port(int), grpc port(int)]}
     agents = Dict()
     # {ip address(str): gcs node info(dict of GcsNodeInfo in gcs.proto)}
     nodes = Dict()
@@ -56,7 +56,7 @@ class DataOrganizer:
             node_worker_id_set.add(worker_stats["workerId"])
         node_actors = {}
         for actor_id, actor_table_data in DataSource.actors.items():
-            if actor_table_data["workerId"] in node_worker_id_set:
+            if actor_table_data["address"]["workerId"] in node_worker_id_set:
                 node_actors[actor_id] = actor_table_data
         return node_actors
 
@@ -102,6 +102,7 @@ class DataOrganizer:
         for hostname in DataSource.hostname_to_ip.keys():
             node_info = await cls.get_node_info(hostname)
             node_info.pop("workers", None)
+            node_info.pop("actors", None)
             node_info["raylet"].pop("workersStats", None)
             node_info["raylet"].pop("viewData", None)
             all_nodes_summary.append(node_info)

dashboard/head.py

@@ -28,7 +28,8 @@ def gcs_node_info_to_dict(message):
 
 
 class DashboardHead:
-    def __init__(self, http_host, http_port, redis_address, redis_password):
+    def __init__(self, http_host, http_port, redis_address, redis_password,
+                 log_dir):
         # NodeInfoGcsService
         self._gcs_node_info_stub = None
         self._gcs_rpc_error_counter = 0
@@ -37,6 +38,7 @@ class DashboardHead:
         self.http_port = http_port
         self.redis_address = dashboard_utils.address_tuple(redis_address)
         self.redis_password = redis_password
+        self.log_dir = log_dir
         self.aioredis_client = None
         self.aiogrpc_gcs_channel = None
         self.http_session = None
@@ -74,6 +76,22 @@ class DashboardHead:
         while True:
             try:
                 nodes = await self._get_nodes()
+
+                # Get correct node info by state,
+                #   1. The node is ALIVE if any ALIVE node info
+                #      of the hostname exists.
+                #   2. The node is DEAD if all node info of the
+                #      hostname are DEAD.
+                hostname_to_node_info = {}
+                for node in nodes:
+                    hostname = node["nodeManagerAddress"]
+                    assert node["state"] in ["ALIVE", "DEAD"]
+                    choose = hostname_to_node_info.get(hostname)
+                    if choose is not None and choose["state"] == "ALIVE":
+                        continue
+                    hostname_to_node_info[hostname] = node
+                nodes = hostname_to_node_info.values()
+
                 self._gcs_rpc_error_counter = 0
                 node_ips = [node["nodeManagerAddress"] for node in nodes]
                 node_hostnames = [
@@ -83,13 +101,11 @@ class DashboardHead:
                 agents = dict(DataSource.agents)
                 for node in nodes:
                     node_ip = node["nodeManagerAddress"]
-                    if node_ip not in agents:
-                        key = "{}{}".format(
-                            dashboard_consts.DASHBOARD_AGENT_PORT_PREFIX,
-                            node_ip)
-                        agent_port = await self.aioredis_client.get(key)
-                        if agent_port:
-                            agents[node_ip] = json.loads(agent_port)
+                    key = "{}{}".format(
+                        dashboard_consts.DASHBOARD_AGENT_PORT_PREFIX, node_ip)
+                    agent_port = await self.aioredis_client.get(key)
+                    if agent_port:
+                        agents[node_ip] = json.loads(agent_port)
                 for ip in agents.keys() - set(node_ips):
                     agents.pop(ip, None)
 
@@ -99,8 +115,8 @@ class DashboardHead:
                     dict(zip(node_hostnames, node_ips)))
                 DataSource.ip_to_hostname.reset(
                     dict(zip(node_ips, node_hostnames)))
-            except aiogrpc.AioRpcError as ex:
-                logger.exception(ex)
+            except aiogrpc.AioRpcError:
+                logger.exception("Got AioRpcError when updating nodes.")
                 self._gcs_rpc_error_counter += 1
                 if self._gcs_rpc_error_counter > \
                         dashboard_consts.MAX_COUNT_OF_GCS_RPC_ERROR:
@@ -109,8 +125,8 @@ class DashboardHead:
                         self._gcs_rpc_error_counter,
                         dashboard_consts.MAX_COUNT_OF_GCS_RPC_ERROR)
                     sys.exit(-1)
-            except Exception as ex:
-                logger.exception(ex)
+            except Exception:
+                logger.exception("Error updating nodes.")
             finally:
                 await asyncio.sleep(
                     dashboard_consts.UPDATE_NODES_INTERVAL_SECONDS)
@@ -126,7 +142,7 @@ class DashboardHead:
             c = cls(self)
             dashboard_utils.ClassMethodRouteTable.bind(c)
             modules.append(c)
-        logger.info("Loaded {} modules.".format(len(modules)))
+        logger.info("Loaded %d modules.", len(modules))
         return modules
 
     async def run(self):
@@ -183,8 +199,8 @@ class DashboardHead:
                 co = await dashboard_utils.NotifyQueue.get()
                 try:
                     await co
-                except Exception as e:
-                    logger.exception(e)
+                except Exception:
+                    logger.exception(f"Error notifying coroutine {co}")
 
         async def _purge_data():
             """Purge data in datacenter."""
@@ -193,8 +209,8 @@ class DashboardHead:
                     dashboard_consts.PURGE_DATA_INTERVAL_SECONDS)
                 try:
                     await DataOrganizer.purge()
-                except Exception as e:
-                    logger.exception(e)
+                except Exception:
+                    logger.exception("Error purging data.")
 
         modules = self._load_modules()
 

dashboard/modules/log/log_agent.py

@@ -0,0 +1,19 @@
+import logging
+
+import mimetypes
+
+import ray.new_dashboard.utils as dashboard_utils
+
+logger = logging.getLogger(__name__)
+routes = dashboard_utils.ClassMethodRouteTable
+
+
+class LogAgent(dashboard_utils.DashboardAgentModule):
+    def __init__(self, dashboard_agent):
+        super().__init__(dashboard_agent)
+        mimetypes.add_type("text/plain", ".err")
+        mimetypes.add_type("text/plain", ".out")
+        routes.static("/logs", self._dashboard_agent.log_dir, show_index=True)
+
+    async def run(self, server):
+        pass

dashboard/modules/log/log_head.py

@@ -0,0 +1,69 @@
+import logging
+
+import mimetypes
+
+import aiohttp.web
+import ray.new_dashboard.utils as dashboard_utils
+from ray.new_dashboard.datacenter import DataSource, GlobalSignals
+
+logger = logging.getLogger(__name__)
+routes = dashboard_utils.ClassMethodRouteTable
+
+
+class LogHead(dashboard_utils.DashboardHeadModule):
+    LOG_URL_TEMPLATE = "http://{ip}:{port}/logs"
+
+    def __init__(self, dashboard_head):
+        super().__init__(dashboard_head)
+        mimetypes.add_type("text/plain", ".err")
+        mimetypes.add_type("text/plain", ".out")
+        routes.static("/logs", self._dashboard_head.log_dir, show_index=True)
+        GlobalSignals.node_info_fetched.append(
+            self.insert_log_url_to_node_info)
+
+    async def insert_log_url_to_node_info(self, node_info):
+        ip = node_info.get("ip")
+        if ip is None:
+            return
+        agent_port = DataSource.agents.get(ip)
+        if agent_port is None:
+            return
+        agent_http_port, _ = agent_port
+        log_url = self.LOG_URL_TEMPLATE.format(ip=ip, port=agent_http_port)
+        node_info["logUrl"] = log_url
+
+    @routes.get("/log_index")
+    async def get_log_index(self, req) -> aiohttp.web.Response:
+        url_list = []
+        for ip, ports in DataSource.agents.items():
+            url_list.append(
+                self.LOG_URL_TEMPLATE.format(ip=ip, port=str(ports[0])))
+        if self._dashboard_head.ip not in DataSource.agents:
+            url_list.append(
+                self.LOG_URL_TEMPLATE.format(
+                    ip=self._dashboard_head.ip,
+                    port=self._dashboard_head.http_port))
+        return aiohttp.web.Response(
+            text=self._directory_as_html(url_list), content_type="text/html")
+
+    @staticmethod
+    def _directory_as_html(url_list) -> str:
+        # returns directory's index as html
+
+        index_of = "Index of logs"
+        h1 = f"<h1>{index_of}</h1>"
+
+        index_list = []
+        for url in sorted(url_list):
+            index_list.append(f'<li><a href="{url}">{url}</a></li>')
+        index_list = "\n".join(index_list)
+        ul = f"<ul>\n{index_list}\n</ul>"
+        body = f"<body>\n{h1}\n{ul}\n</body>"
+
+        head_str = f"<head>\n<title>{index_of}</title>\n</head>"
+        html = f"<html>\n{head_str}\n{body}\n</html>"
+
+        return html
+
+    async def run(self, server):
+        pass

dashboard/modules/log/test_log.py

@@ -0,0 +1,115 @@
+import os
+import sys
+import logging
+import requests
+import socket
+import time
+import traceback
+import html.parser
+import urllib.parse
+
+import pytest
+import ray
+from ray.new_dashboard.tests.conftest import *  # noqa
+from ray.test_utils import (
+    format_web_url,
+    wait_until_server_available,
+)
+
+os.environ["RAY_USE_NEW_DASHBOARD"] = "1"
+
+logger = logging.getLogger(__name__)
+
+
+class LogUrlParser(html.parser.HTMLParser):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._urls = []
+
+    def handle_starttag(self, tag, attrs):
+        if tag == "a":
+            self._urls.append(dict(attrs)["href"])
+
+    def error(self, message):
+        logger.error(message)
+
+    def get_urls(self):
+        return self._urls
+
+
+def test_log(ray_start_with_dashboard):
+    @ray.remote
+    def write_log(s):
+        print(s)
+
+    test_log_text = "test_log_text"
+    ray.get(write_log.remote(test_log_text))
+    assert (wait_until_server_available(ray_start_with_dashboard["webui_url"])
+            is True)
+    webui_url = ray_start_with_dashboard["webui_url"]
+    webui_url = format_web_url(webui_url)
+
+    timeout_seconds = 10
+    start_time = time.time()
+    last_ex = None
+    while True:
+        time.sleep(1)
+        try:
+            response = requests.get(webui_url + "/log_index")
+            response.raise_for_status()
+            parser = LogUrlParser()
+            parser.feed(response.text)
+            all_nodes_log_urls = parser.get_urls()
+            assert len(all_nodes_log_urls) == 1
+
+            response = requests.get(all_nodes_log_urls[0])
+            response.raise_for_status()
+            parser = LogUrlParser()
+            parser.feed(response.text)
+
+            # Search test_log_text from all worker logs.
+            parsed_url = urllib.parse.urlparse(all_nodes_log_urls[0])
+            paths = parser.get_urls()
+            urls = []
+            for p in paths:
+                if "worker" in p:
+                    urls.append(parsed_url._replace(path=p).geturl())
+
+            for u in urls:
+                response = requests.get(u)
+                response.raise_for_status()
+                if test_log_text in response.text:
+                    break
+            else:
+                raise Exception(f"Can't find {test_log_text} from {urls}")
+
+            # Test range request.
+            response = requests.get(
+                webui_url + "/logs/dashboard.log",
+                headers={"Range": "bytes=43-51"})
+            response.raise_for_status()
+            assert response.text == "Dashboard"
+
+            # Test logUrl in node info.
+            response = requests.get(webui_url +
+                                    f"/nodes/{socket.gethostname()}")
+            response.raise_for_status()
+            node_info = response.json()
+            assert node_info["result"] is True
+            node_info = node_info["data"]["detail"]
+            assert "logUrl" in node_info
+            assert node_info["logUrl"] in all_nodes_log_urls
+            break
+        except Exception as ex:
+            last_ex = ex
+        finally:
+            if time.time() > start_time + timeout_seconds:
+                ex_stack = traceback.format_exception(
+                    type(last_ex), last_ex,
+                    last_ex.__traceback__) if last_ex else []
+                ex_stack = "".join(ex_stack)
+                raise Exception(f"Timed out while testing, {ex_stack}")
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

dashboard/modules/reporter/reporter_agent.py

@@ -6,6 +6,7 @@ import os
 import socket
 import subprocess
 import sys
+import traceback
 
 import aioredis
 
@@ -17,6 +18,7 @@ import ray.services
 import ray.utils
 from ray.core.generated import reporter_pb2
 from ray.core.generated import reporter_pb2_grpc
+from ray.metrics_agent import MetricsAgent
 import psutil
 
 logger = logging.getLogger(__name__)
@@ -67,31 +69,46 @@ class ReporterAgent(dashboard_utils.DashboardAgentModule,
         self._hostname = socket.gethostname()
         self._workers = set()
         self._network_stats_hist = [(0, (0.0, 0.0))]  # time, (sent, recv)
+        self._metrics_agent = MetricsAgent(dashboard_agent.metrics_export_port)
 
     async def GetProfilingStats(self, request, context):
         pid = request.pid
         duration = request.duration
         profiling_file_path = os.path.join(ray.utils.get_ray_temp_dir(),
-                                           "{}_profiling.txt".format(pid))
+                                           f"{pid}_profiling.txt")
         sudo = "sudo" if ray.utils.get_user() != "root" else ""
-        process = subprocess.Popen(
-            (f"{sudo} $(which py-spy) record -o {profiling_file_path} -p {pid}"
-             f" -d {duration} -f speedscope"),
+        process = await asyncio.create_subprocess_shell(
+            f"{sudo} $(which py-spy) record "
+            f"-o {profiling_file_path} -p {pid} -d {duration} -f speedscope",
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             shell=True)
-        stdout, stderr = process.communicate()
+        stdout, stderr = await process.communicate()
         if process.returncode != 0:
             profiling_stats = ""
         else:
             with open(profiling_file_path, "r") as f:
                 profiling_stats = f.read()
         return reporter_pb2.GetProfilingStatsReply(
-            profiling_stats=profiling_stats, stdout=stdout, stderr=stderr)
+            profiling_stats=profiling_stats, std_out=stdout, std_err=stderr)
 
     async def ReportMetrics(self, request, context):
-        # TODO(sang): Process metrics here.
-        return reporter_pb2.ReportMetricsReply()
+        # NOTE: Exceptions are not propagated properly
+        # when we don't catch them here.
+        try:
+            metrcs_description_required = (
+                self._metrics_agent.record_metrics_points(
+                    request.metrics_points))
+        except Exception as e:
+            logger.error(e)
+            logger.error(traceback.format_exc())
+
+        # If metrics description is missing, we should notify cpp processes
+        # that we need them. Cpp processes will then report them to here.
+        # We need it when (1) a new metric is reported (application metric)
+        # (2) a reporter goes down and restarted (currently not implemented).
+        return reporter_pb2.ReportMetricsReply(
+            metrcs_description_required=metrcs_description_required)
 
     @staticmethod
     def _get_cpu_percent():
@@ -223,8 +240,8 @@ class ReporterAgent(dashboard_utils.DashboardAgentModule,
                 await aioredis_client.publish(
                     "{}{}".format(reporter_consts.REPORTER_PREFIX,
                                   self._hostname), jsonify_asdict(stats))
-            except Exception as ex:
-                logger.exception(ex)
+            except Exception:
+                logger.exception("Error publishing node physical stats.")
             await asyncio.sleep(
                 reporter_consts.REPORTER_UPDATE_INTERVAL_MS / 1000)
 

dashboard/modules/reporter/reporter_head.py

@@ -1,6 +1,5 @@
 import json
 import logging
-import uuid
 
 import aiohttp.web
 from aioredis.pubsub import Receiver
@@ -24,58 +23,32 @@ class ReportHead(dashboard_utils.DashboardHeadModule):
     def __init__(self, dashboard_head):
         super().__init__(dashboard_head)
         self._stubs = {}
-        self._profiling_stats = {}
         DataSource.agents.signal.append(self._update_stubs)
 
     async def _update_stubs(self, change):
+        if change.old:
+            ip, port = change.old
+            self._stubs.pop(ip)
         if change.new:
-            ip, ports = next(iter(change.new.items()))
-            channel = aiogrpc.insecure_channel("{}:{}".format(ip, ports[1]))
+            ip, ports = change.new
+            channel = aiogrpc.insecure_channel(f"{ip}:{ports[1]}")
             stub = reporter_pb2_grpc.ReporterServiceStub(channel)
             self._stubs[ip] = stub
-        if change.old:
-            ip, port = next(iter(change.old.items()))
-            self._stubs.pop(ip)
 
     @routes.get("/api/launch_profiling")
     async def launch_profiling(self, req) -> aiohttp.web.Response:
-        node_id = req.query.get("node_id")
-        pid = int(req.query.get("pid"))
-        duration = int(req.query.get("duration"))
-        profiling_id = str(uuid.uuid4())
-        reporter_stub = self._stubs[node_id]
+        ip = req.query["ip"]
+        pid = int(req.query["pid"])
+        duration = int(req.query["duration"])
+        reporter_stub = self._stubs[ip]
         reply = await reporter_stub.GetProfilingStats(
             reporter_pb2.GetProfilingStatsRequest(pid=pid, duration=duration))
-        self._profiling_stats[profiling_id] = reply
+        profiling_info = (json.loads(reply.profiling_stats)
+                          if reply.profiling_stats else reply.std_out)
         return await dashboard_utils.rest_response(
             success=True,
-            message="Profiling launched.",
-            profiling_id=profiling_id)
-
-    @routes.get("/api/check_profiling_status")
-    async def check_profiling_status(self, req) -> aiohttp.web.Response:
-        profiling_id = req.query.get("profiling_id")
-        is_present = profiling_id in self._profiling_stats
-        if not is_present:
-            status = {"status": "pending"}
-        else:
-            reply = self._profiling_stats[profiling_id]
-            if reply.stderr:
-                status = {"status": "error", "error": reply.stderr}
-            else:
-                status = {"status": "finished"}
-        return await dashboard_utils.rest_response(
-            success=True, message="Profiling status fetched.", status=status)
-
-    @routes.get("/api/get_profiling_info")
-    async def get_profiling_info(self, req) -> aiohttp.web.Response:
-        profiling_id = req.query.get("profiling_id")
-        profiling_stats = self._profiling_stats.get(profiling_id)
-        assert profiling_stats, "profiling not finished"
-        return await dashboard_utils.rest_response(
-            success=True,
-            message="Profiling info fetched.",
-            profiling_info=json.loads(profiling_stats.profiling_stats))
+            message="Profiling success.",
+            profiling_info=profiling_info)
 
     async def run(self, server):
         aioredis_client = self._dashboard_head.aioredis_client
@@ -83,12 +56,13 @@ class ReportHead(dashboard_utils.DashboardHeadModule):
 
         reporter_key = "{}*".format(reporter_consts.REPORTER_PREFIX)
         await aioredis_client.psubscribe(receiver.pattern(reporter_key))
-        logger.info("Subscribed to {}".format(reporter_key))
+        logger.info(f"Subscribed to {reporter_key}")
 
         async for sender, msg in receiver.iter():
             try:
                 _, data = msg
                 data = json.loads(ray.utils.decode(data))
                 DataSource.node_physical_stats[data["ip"]] = data
-            except Exception as ex:
-                logger.exception(ex)
+            except Exception:
+                logger.exception(
+                    "Error receiving node physical stats from reporter agent.")

dashboard/modules/reporter/test_reporter.py

@@ -0,0 +1,102 @@
+import os
+import sys
+import logging
+import requests
+import time
+
+import pytest
+import ray
+from ray.new_dashboard.tests.conftest import *  # noqa
+from ray.test_utils import (
+    format_web_url,
+    RayTestTimeoutException,
+    wait_until_server_available,
+    wait_for_condition,
+)
+
+os.environ["RAY_USE_NEW_DASHBOARD"] = "1"
+
+logger = logging.getLogger(__name__)
+
+
+def test_profiling(shutdown_only):
+    addresses = ray.init(include_dashboard=True, num_cpus=6)
+
+    @ray.remote(num_cpus=2)
+    class Actor:
+        def getpid(self):
+            return os.getpid()
+
+    c = Actor.remote()
+    actor_pid = ray.get(c.getpid.remote())
+
+    webui_url = addresses["webui_url"]
+    assert (wait_until_server_available(webui_url) is True)
+    webui_url = format_web_url(webui_url)
+
+    start_time = time.time()
+    launch_profiling = None
+    while True:
+        # Sometimes some startup time is required
+        if time.time() - start_time > 10:
+            raise RayTestTimeoutException(
+                "Timed out while collecting profiling stats, "
+                f"launch_profiling: {launch_profiling}")
+        launch_profiling = requests.get(
+            webui_url + "/api/launch_profiling",
+            params={
+                "ip": ray.nodes()[0]["NodeManagerAddress"],
+                "pid": actor_pid,
+                "duration": 5
+            }).json()
+        if launch_profiling["result"]:
+            profiling_info = launch_profiling["data"]["profilingInfo"]
+            break
+        time.sleep(1)
+    logger.info(profiling_info)
+
+
+def test_node_physical_stats(enable_test_module, shutdown_only):
+    addresses = ray.init(include_dashboard=True, num_cpus=6)
+
+    @ray.remote(num_cpus=1)
+    class Actor:
+        def getpid(self):
+            return os.getpid()
+
+    actors = [Actor.remote() for _ in range(6)]
+    actor_pids = ray.get([actor.getpid.remote() for actor in actors])
+    actor_pids = set(actor_pids)
+
+    webui_url = addresses["webui_url"]
+    assert (wait_until_server_available(webui_url) is True)
+    webui_url = format_web_url(webui_url)
+
+    def _check_workers():
+        try:
+            resp = requests.get(webui_url +
+                                "/test/dump?key=node_physical_stats")
+            resp.raise_for_status()
+            result = resp.json()
+            assert result["result"] is True
+            node_physical_stats = result["data"]["nodePhysicalStats"]
+            assert len(node_physical_stats) == 1
+            current_stats = node_physical_stats[addresses["raylet_ip_address"]]
+            # Check Actor workers
+            current_actor_pids = set()
+            for worker in current_stats["workers"]:
+                if "ray::Actor" in worker["cmdline"][0]:
+                    current_actor_pids.add(worker["pid"])
+            assert current_actor_pids == actor_pids
+            # Check raylet cmdline
+            assert "raylet" in current_stats["cmdline"][0]
+            return True
+        except Exception as ex:
+            logger.info(ex)
+            return False
+
+    wait_for_condition(_check_workers, timeout=10)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

dashboard/modules/stats_collector/stats_collector_consts.py

@@ -0,0 +1,3 @@
+NODE_STATS_UPDATE_INTERVAL_SECONDS = 1
+RETRY_GET_ALL_ACTOR_INFO_INTERVAL_SECONDS = 1
+ACTOR_CHANNEL = "ACTOR"

dashboard/modules/stats_collector/stats_collector_head.py

@@ -0,0 +1,155 @@
+import asyncio
+import logging
+
+import aiohttp.web
+from aioredis.pubsub import Receiver
+from grpc.experimental import aio as aiogrpc
+
+import ray.gcs_utils
+import ray.new_dashboard.modules.stats_collector.stats_collector_consts \
+    as stats_collector_consts
+import ray.new_dashboard.utils as dashboard_utils
+from ray.new_dashboard.utils import async_loop_forever
+from ray.core.generated import node_manager_pb2
+from ray.core.generated import node_manager_pb2_grpc
+from ray.core.generated import gcs_service_pb2
+from ray.core.generated import gcs_service_pb2_grpc
+from ray.new_dashboard.datacenter import DataSource, DataOrganizer
+from ray.utils import binary_to_hex
+
+logger = logging.getLogger(__name__)
+routes = dashboard_utils.ClassMethodRouteTable
+
+
+def node_stats_to_dict(message):
+    return dashboard_utils.message_to_dict(
+        message, {
+            "actorId", "jobId", "taskId", "parentTaskId", "sourceActorId",
+            "callerId", "rayletId", "workerId"
+        })
+
+
+def actor_table_data_to_dict(message):
+    return dashboard_utils.message_to_dict(
+        message, {
+            "actorId", "parentId", "jobId", "workerId", "rayletId",
+            "actorCreationDummyObjectId"
+        },
+        including_default_value_fields=True)
+
+
+class StatsCollector(dashboard_utils.DashboardHeadModule):
+    def __init__(self, dashboard_head):
+        super().__init__(dashboard_head)
+        self._stubs = {}
+        # JobInfoGcsServiceStub
+        self._gcs_job_info_stub = None
+        # ActorInfoGcsService
+        self._gcs_actor_info_stub = None
+        DataSource.nodes.signal.append(self._update_stubs)
+
+    async def _update_stubs(self, change):
+        if change.old:
+            ip, port = change.old
+            self._stubs.pop(ip)
+        if change.new:
+            ip, node_info = change.new
+            address = "{}:{}".format(ip, int(node_info["nodeManagerPort"]))
+            channel = aiogrpc.insecure_channel(address)
+            stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
+            self._stubs[ip] = stub
+
+    @routes.get("/nodes")
+    async def get_all_nodes(self, req) -> aiohttp.web.Response:
+        view = req.query.get("view")
+        if view == "summary":
+            all_node_summary = await DataOrganizer.get_all_node_summary()
+            return await dashboard_utils.rest_response(
+                success=True,
+                message="Node summary fetched.",
+                summary=all_node_summary)
+        elif view is not None and view.lower() == "hostNameList".lower():
+            return await dashboard_utils.rest_response(
+                success=True,
+                message="Node hostname list fetched.",
+                host_name_list=list(DataSource.hostname_to_ip.keys()))
+        else:
+            return await dashboard_utils.rest_response(
+                success=False, message=f"Unknown view {view}")
+
+    @routes.get("/nodes/{hostname}")
+    async def get_node(self, req) -> aiohttp.web.Response:
+        hostname = req.match_info.get("hostname")
+        node_info = await DataOrganizer.get_node_info(hostname)
+        return await dashboard_utils.rest_response(
+            success=True, message="Node detail fetched.", detail=node_info)
+
+    async def _update_actors(self):
+        # Subscribe actor channel.
+        aioredis_client = self._dashboard_head.aioredis_client
+        receiver = Receiver()
+
+        key = "{}:*".format(stats_collector_consts.ACTOR_CHANNEL)
+        pattern = receiver.pattern(key)
+        await aioredis_client.psubscribe(pattern)
+        logger.info("Subscribed to %s", key)
+
+        # Get all actor info.
+        while True:
+            try:
+                logger.info("Getting all actor info from GCS.")
+                request = gcs_service_pb2.GetAllActorInfoRequest()
+                reply = await self._gcs_actor_info_stub.GetAllActorInfo(
+                    request, timeout=2)
+                if reply.status.code == 0:
+                    result = {}
+                    for actor_info in reply.actor_table_data:
+                        result[binary_to_hex(actor_info.actor_id)] = \
+                            actor_table_data_to_dict(actor_info)
+                    DataSource.actors.reset(result)
+                    logger.info("Received %d actor info from GCS.",
+                                len(result))
+                    break
+                else:
+                    raise Exception(
+                        f"Failed to GetAllActorInfo: {reply.status.message}")
+            except Exception:
+                logger.exception("Error Getting all actor info from GCS.")
+                await asyncio.sleep(stats_collector_consts.
+                                    RETRY_GET_ALL_ACTOR_INFO_INTERVAL_SECONDS)
+
+        # Receive actors from channel.
+        async for sender, msg in receiver.iter():
+            try:
+                _, data = msg
+                pubsub_message = ray.gcs_utils.PubSubMessage.FromString(data)
+                actor_info = ray.gcs_utils.ActorTableData.FromString(
+                    pubsub_message.data)
+                DataSource.actors[binary_to_hex(actor_info.actor_id)] = \
+                    actor_table_data_to_dict(actor_info)
+            except Exception:
+                logger.exception("Error receiving actor info.")
+
+    @async_loop_forever(
+        stats_collector_consts.NODE_STATS_UPDATE_INTERVAL_SECONDS)
+    async def _update_node_stats(self):
+        for ip, stub in self._stubs.items():
+            node_info = DataSource.nodes.get(ip)
+            if node_info["state"] != "ALIVE":
+                continue
+            try:
+                reply = await stub.GetNodeStats(
+                    node_manager_pb2.GetNodeStatsRequest(), timeout=2)
+                reply_dict = node_stats_to_dict(reply)
+                DataSource.node_stats[ip] = reply_dict
+            except Exception:
+                logger.exception(f"Error updating node stats of {ip}.")
+
+    async def run(self, server):
+        gcs_channel = self._dashboard_head.aiogrpc_gcs_channel
+        self._gcs_job_info_stub = \
+            gcs_service_pb2_grpc.JobInfoGcsServiceStub(gcs_channel)
+        self._gcs_actor_info_stub = \
+            gcs_service_pb2_grpc.ActorInfoGcsServiceStub(gcs_channel)
+
+        await asyncio.gather(self._update_node_stats(), self._update_actors())

dashboard/modules/stats_collector/test_stats_collector.py

@@ -0,0 +1,93 @@
+import os
+import sys
+import logging
+import requests
+import time
+import traceback
+
+import pytest
+import ray
+from ray.new_dashboard.tests.conftest import *  # noqa
+from ray.test_utils import (
+    format_web_url,
+    wait_until_server_available,
+)
+
+os.environ["RAY_USE_NEW_DASHBOARD"] = "1"
+
+logger = logging.getLogger(__name__)
+
+
+def test_node_info(ray_start_with_dashboard):
+    @ray.remote
+    class Actor:
+        def getpid(self):
+            return os.getpid()
+
+    actors = [Actor.remote(), Actor.remote()]
+    actor_pids = [actor.getpid.remote() for actor in actors]
+    actor_pids = set(ray.get(actor_pids))
+
+    assert (wait_until_server_available(ray_start_with_dashboard["webui_url"])
+            is True)
+    webui_url = ray_start_with_dashboard["webui_url"]
+    webui_url = format_web_url(webui_url)
+
+    timeout_seconds = 10
+    start_time = time.time()
+    last_ex = None
+    while True:
+        time.sleep(1)
+        try:
+            response = requests.get(webui_url + "/nodes?view=hostnamelist")
+            response.raise_for_status()
+            hostname_list = response.json()
+            assert hostname_list["result"] is True, hostname_list["msg"]
+            hostname_list = hostname_list["data"]["hostNameList"]
+            assert len(hostname_list) == 1
+
+            hostname = hostname_list[0]
+            response = requests.get(webui_url + f"/nodes/{hostname}")
+            response.raise_for_status()
+            detail = response.json()
+            assert detail["result"] is True, detail["msg"]
+            detail = detail["data"]["detail"]
+            assert detail["hostname"] == hostname
+            assert detail["state"] == "ALIVE"
+            assert "raylet" in detail["cmdline"][0]
+            assert len(detail["workers"]) >= 2
+            assert len(detail["actors"]) == 2, detail["actors"]
+            assert len(detail["raylet"]["viewData"]) > 0
+
+            actor_worker_pids = set()
+            for worker in detail["workers"]:
+                if "ray::Actor" in worker["cmdline"][0]:
+                    actor_worker_pids.add(worker["pid"])
+            assert actor_worker_pids == actor_pids
+
+            response = requests.get(webui_url + "/nodes?view=summary")
+            response.raise_for_status()
+            summary = response.json()
+            assert summary["result"] is True, summary["msg"]
+            assert len(summary["data"]["summary"]) == 1
+            summary = summary["data"]["summary"][0]
+            assert summary["hostname"] == hostname
+            assert summary["state"] == "ALIVE"
+            assert "raylet" in summary["cmdline"][0]
+            assert "workers" not in summary
+            assert "actors" not in summary
+            assert "viewData" not in summary["raylet"]
+            break
+        except Exception as ex:
+            last_ex = ex
+        finally:
+            if time.time() > start_time + timeout_seconds:
+                ex_stack = traceback.format_exception(
+                    type(last_ex), last_ex,
+                    last_ex.__traceback__) if last_ex else []
+                ex_stack = "".join(ex_stack)
+                raise Exception(f"Timed out while testing, {ex_stack}")
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

dashboard/modules/test/test_agent.py

@@ -4,12 +4,16 @@ import aiohttp.web
 
 import ray.new_dashboard.utils as dashboard_utils
 import ray.new_dashboard.modules.test.test_utils as test_utils
+import ray.new_dashboard.modules.test.test_consts as test_consts
+from ray.ray_constants import env_bool
 
 logger = logging.getLogger(__name__)
 routes = dashboard_utils.ClassMethodRouteTable
 
 
-class HeadAgent(dashboard_utils.DashboardAgentModule):
+@dashboard_utils.dashboard_module(
+    enable=env_bool(test_consts.TEST_MODULE_ENVIRONMENT_KEY, False))
+class TestAgent(dashboard_utils.DashboardAgentModule):
     def __init__(self, dashboard_agent):
         super().__init__(dashboard_agent)
 
@@ -20,5 +24,17 @@ class HeadAgent(dashboard_utils.DashboardAgentModule):
                                            url)
         return aiohttp.web.json_response(result)
 
+    @routes.head("/test/route_head")
+    async def route_head(self, req) -> aiohttp.web.Response:
+        pass
+
+    @routes.post("/test/route_post")
+    async def route_post(self, req) -> aiohttp.web.Response:
+        pass
+
+    @routes.patch("/test/route_patch")
+    async def route_patch(self, req) -> aiohttp.web.Response:
+        pass
+
     async def run(self, server):
         pass

dashboard/modules/test/test_consts.py

@@ -0,0 +1 @@
+TEST_MODULE_ENVIRONMENT_KEY = "RAY_DASHBOARD_MODULE_TEST"

dashboard/modules/test/test_head.py

@@ -4,12 +4,16 @@ import aiohttp.web
 
 import ray.new_dashboard.utils as dashboard_utils
 import ray.new_dashboard.modules.test.test_utils as test_utils
+import ray.new_dashboard.modules.test.test_consts as test_consts
 from ray.new_dashboard.datacenter import DataSource
+from ray.ray_constants import env_bool
 
 logger = logging.getLogger(__name__)
 routes = dashboard_utils.ClassMethodRouteTable
 
 
+@dashboard_utils.dashboard_module(
+    enable=env_bool(test_consts.TEST_MODULE_ENVIRONMENT_KEY, False))
 class TestHead(dashboard_utils.DashboardHeadModule):
     def __init__(self, dashboard_head):
         super().__init__(dashboard_head)
@@ -17,12 +21,28 @@ class TestHead(dashboard_utils.DashboardHeadModule):
         DataSource.agents.signal.append(self._update_notified_agents)
 
     async def _update_notified_agents(self, change):
-        if change.new:
-            ip, ports = next(iter(change.new.items()))
-            self._notified_agents[ip] = ports
         if change.old:
-            ip, port = next(iter(change.old.items()))
+            ip, port = change.old
             self._notified_agents.pop(ip)
+        if change.new:
+            ip, ports = change.new
+            self._notified_agents[ip] = ports
+
+    @routes.get("/test/route_get")
+    async def route_get(self, req) -> aiohttp.web.Response:
+        pass
+
+    @routes.put("/test/route_put")
+    async def route_put(self, req) -> aiohttp.web.Response:
+        pass
+
+    @routes.delete("/test/route_delete")
+    async def route_delete(self, req) -> aiohttp.web.Response:
+        pass
+
+    @routes.view("/test/route_view")
+    async def route_view(self, req) -> aiohttp.web.Response:
+        pass
 
     @routes.get("/test/dump")
     async def dump(self, req) -> aiohttp.web.Response:
@@ -41,7 +61,7 @@ class TestHead(dashboard_utils.DashboardHeadModule):
             data = dict(DataSource.__dict__.get(key))
             return await dashboard_utils.rest_response(
                 success=True,
-                message="Fetch {} from datacenter success.".format(key),
+                message=f"Fetch {key} from datacenter success.",
                 **{key: data})
 
     @routes.get("/test/notified_agents")

dashboard/tests/conftest.py

@@ -1 +1,10 @@
+import os
+import pytest
 from ray.tests.conftest import *  # noqa
+
+
+@pytest.fixture
+def enable_test_module():
+    os.environ["RAY_DASHBOARD_MODULE_TEST"] = "true"
+    yield
+    os.environ.pop("RAY_DASHBOARD_MODULE_TEST", None)

dashboard/tests/test_dashboard.py

@@ -1,7 +1,11 @@
 import os
+import sys
+import json
 import time
 import logging
+import asyncio
 
+import aiohttp.web
 import ray
 import psutil
 import pytest
@@ -9,13 +13,20 @@ import redis
 import requests
 
 from ray import ray_constants
-from ray.test_utils import wait_for_condition, wait_until_server_available
+from ray.test_utils import (
+    format_web_url,
+    wait_for_condition,
+    wait_until_server_available,
+    run_string_as_driver,
+)
 import ray.new_dashboard.consts as dashboard_consts
+import ray.new_dashboard.utils as dashboard_utils
 import ray.new_dashboard.modules
 
 os.environ["RAY_USE_NEW_DASHBOARD"] = "1"
 
 logger = logging.getLogger(__name__)
+routes = dashboard_utils.ClassMethodRouteTable
 
 
 def cleanup_test_files():
@@ -106,7 +117,9 @@ def test_basic(ray_start_with_dashboard):
 
     logger.info("Test agent register is OK.")
     wait_for_condition(lambda: _search_agent(raylet_proc.children()))
-    assert dashboard_proc.status() == psutil.STATUS_RUNNING
+    assert dashboard_proc.status() in [
+        psutil.STATUS_RUNNING, psutil.STATUS_SLEEPING
+    ]
     agent_proc = _search_agent(raylet_proc.children())
     agent_pid = agent_proc.pid
 
@@ -130,13 +143,13 @@ def test_basic(ray_start_with_dashboard):
     assert agent_ports is not None
 
 
-def test_nodes_update(ray_start_with_dashboard):
+def test_nodes_update(enable_test_module, ray_start_with_dashboard):
     assert (wait_until_server_available(ray_start_with_dashboard["webui_url"])
             is True)
     webui_url = ray_start_with_dashboard["webui_url"]
-    webui_url = webui_url.replace("localhost", "http://127.0.0.1")
+    webui_url = format_web_url(webui_url)
 
-    timeout_seconds = 20
+    timeout_seconds = 10
     start_time = time.time()
     while True:
         time.sleep(1)
@@ -146,7 +159,7 @@ def test_nodes_update(ray_start_with_dashboard):
             try:
                 dump_info = response.json()
             except Exception as ex:
-                logger.info("failed response: {}".format(response.text))
+                logger.info("failed response: %s", response.text)
                 raise ex
             assert dump_info["result"] is True
             dump_data = dump_info["data"]
@@ -162,7 +175,7 @@ def test_nodes_update(ray_start_with_dashboard):
             try:
                 notified_agents = response.json()
             except Exception as ex:
-                logger.info("failed response: {}".format(response.text))
+                logger.info("failed response: %s", response.text)
                 raise ex
             assert notified_agents["result"] is True
             notified_agents = notified_agents["data"]
@@ -173,19 +186,18 @@ def test_nodes_update(ray_start_with_dashboard):
             logger.info("Retry because of %s", e)
         finally:
             if time.time() > start_time + timeout_seconds:
-                raise Exception(
-                    "Timed out while waiting for dashboard to start.")
+                raise Exception("Timed out while testing.")
 
 
-def test_http_get(ray_start_with_dashboard):
+def test_http_get(enable_test_module, ray_start_with_dashboard):
     assert (wait_until_server_available(ray_start_with_dashboard["webui_url"])
             is True)
     webui_url = ray_start_with_dashboard["webui_url"]
-    webui_url = webui_url.replace("localhost", "http://127.0.0.1")
+    webui_url = format_web_url(webui_url)
 
     target_url = webui_url + "/test/dump"
 
-    timeout_seconds = 20
+    timeout_seconds = 10
     start_time = time.time()
     while True:
         time.sleep(1)
@@ -196,7 +208,7 @@ def test_http_get(ray_start_with_dashboard):
             try:
                 dump_info = response.json()
             except Exception as ex:
-                logger.info("failed response: {}".format(response.text))
+                logger.info("failed response: %s", response.text)
                 raise ex
             assert dump_info["result"] is True
             dump_data = dump_info["data"]
@@ -205,13 +217,13 @@ def test_http_get(ray_start_with_dashboard):
             http_port, grpc_port = ports
 
             response = requests.get(
-                "http://{}:{}/test/http_get_from_agent?url={}".format(
-                    ip, http_port, target_url))
+                f"http://{ip}:{http_port}"
+                f"/test/http_get_from_agent?url={target_url}")
             response.raise_for_status()
             try:
                 dump_info = response.json()
             except Exception as ex:
-                logger.info("failed response: {}".format(response.text))
+                logger.info("failed response: %s", response.text)
                 raise ex
             assert dump_info["result"] is True
             break
@@ -219,5 +231,135 @@ def test_http_get(ray_start_with_dashboard):
             logger.info("Retry because of %s", e)
         finally:
             if time.time() > start_time + timeout_seconds:
-                raise Exception(
-                    "Timed out while waiting for dashboard to start.")
+                raise Exception("Timed out while testing.")
+
+
+def test_class_method_route_table(enable_test_module):
+    head_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardHeadModule)
+    agent_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardAgentModule)
+    test_head_cls = None
+    for cls in head_cls_list:
+        if cls.__name__ == "TestHead":
+            test_head_cls = cls
+            break
+    assert test_head_cls is not None
+    test_agent_cls = None
+    for cls in agent_cls_list:
+        if cls.__name__ == "TestAgent":
+            test_agent_cls = cls
+            break
+    assert test_agent_cls is not None
+
+    def _has_route(route, method, path):
+        if isinstance(route, aiohttp.web.RouteDef):
+            if route.method == method and route.path == path:
+                return True
+        return False
+
+    def _has_static(route, path, prefix):
+        if isinstance(route, aiohttp.web.StaticDef):
+            if route.path == path and route.prefix == prefix:
+                return True
+        return False
+
+    all_routes = dashboard_utils.ClassMethodRouteTable.routes()
+    assert any(_has_route(r, "HEAD", "/test/route_head") for r in all_routes)
+    assert any(_has_route(r, "GET", "/test/route_get") for r in all_routes)
+    assert any(_has_route(r, "POST", "/test/route_post") for r in all_routes)
+    assert any(_has_route(r, "PUT", "/test/route_put") for r in all_routes)
+    assert any(_has_route(r, "PATCH", "/test/route_patch") for r in all_routes)
+    assert any(
+        _has_route(r, "DELETE", "/test/route_delete") for r in all_routes)
+    assert any(_has_route(r, "*", "/test/route_view") for r in all_routes)
+
+    # Test bind()
+    bound_routes = dashboard_utils.ClassMethodRouteTable.bound_routes()
+    assert len(bound_routes) == 0
+    dashboard_utils.ClassMethodRouteTable.bind(
+        test_agent_cls.__new__(test_agent_cls))
+    bound_routes = dashboard_utils.ClassMethodRouteTable.bound_routes()
+    assert any(_has_route(r, "POST", "/test/route_post") for r in bound_routes)
+    assert all(
+        not _has_route(r, "PUT", "/test/route_put") for r in bound_routes)
+
+    # Static def should be in bound routes.
+    routes.static("/test/route_static", "/path")
+    bound_routes = dashboard_utils.ClassMethodRouteTable.bound_routes()
+    assert any(
+        _has_static(r, "/path", "/test/route_static") for r in bound_routes)
+
+    # Test duplicated routes should raise exception.
+    try:
+
+        @routes.get("/test/route_get")
+        def _duplicated_route(req):
+            pass
+
+        raise Exception("Duplicated routes should raise exception.")
+    except Exception as ex:
+        message = str(ex)
+        assert "/test/route_get" in message
+        assert "test_head.py" in message
+
+    # Test exception in handler
+    post_handler = None
+    for r in bound_routes:
+        if _has_route(r, "POST", "/test/route_post"):
+            post_handler = r.handler
+            break
+    assert post_handler is not None
+
+    loop = asyncio.get_event_loop()
+    r = loop.run_until_complete(post_handler())
+    assert r.status == 200
+    resp = json.loads(r.body)
+    assert resp["result"] is False
+    assert "Traceback" in resp["msg"]
+
+
+def test_async_loop_forever():
+    counter = [0]
+
+    @dashboard_utils.async_loop_forever(interval_seconds=0.1)
+    async def foo():
+        counter[0] += 1
+        raise Exception("Test exception")
+
+    loop = asyncio.get_event_loop()
+    loop.create_task(foo())
+    loop.call_later(1, loop.stop)
+    loop.run_forever()
+    assert counter[0] > 2
+
+
+def test_dashboard_module_decorator(enable_test_module):
+    head_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardHeadModule)
+    agent_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardAgentModule)
+
+    assert any(cls.__name__ == "TestHead" for cls in head_cls_list)
+    assert any(cls.__name__ == "TestAgent" for cls in agent_cls_list)
+
+    test_code = """
+import os
+import ray.new_dashboard.utils as dashboard_utils
+
+os.environ.pop("RAY_DASHBOARD_MODULE_TEST")
+head_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardHeadModule)
+agent_cls_list = dashboard_utils.get_all_modules(
+        dashboard_utils.DashboardAgentModule)
+print(head_cls_list)
+print(agent_cls_list)
+assert all(cls.__name__ != "TestHead" for cls in head_cls_list)
+assert all(cls.__name__ != "TestAgent" for cls in agent_cls_list)
+print("success")
+"""
+    run_string_as_driver(test_code)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

dashboard/utils.py

@@ -13,6 +13,7 @@ import pkgutil
 import traceback
 from base64 import b64decode
 from collections.abc import MutableMapping, Mapping
+from collections import namedtuple
 from typing import Any
 
 import aioredis
@@ -103,10 +104,9 @@ class ClassMethodRouteTable:
         def _wrapper(handler):
             if path in cls._bind_map[method]:
                 bind_info = cls._bind_map[method][path]
-                raise Exception("Duplicated route path: {}, "
-                                "previous one registered at {}:{}".format(
-                                    path, bind_info.filename,
-                                    bind_info.lineno))
+                raise Exception(f"Duplicated route path: {path}, "
+                                f"previous one registered at "
+                                f"{bind_info.filename}:{bind_info.lineno}")
 
             bind_info = cls._BindInfo(handler.__code__.co_filename,
                                       handler.__code__.co_firstlineno, None)
@@ -174,15 +174,28 @@ class ClassMethodRouteTable:
                 h.__func__.__route_path__].instance = instance
 
 
+def dashboard_module(enable):
+    """A decorator for dashboard module."""
+
+    def _cls_wrapper(cls):
+        cls.__ray_dashboard_module_enable__ = enable
+        return cls
+
+    return _cls_wrapper
+
+
 def get_all_modules(module_type):
-    logger.info("Get all modules by type: {}".format(module_type.__name__))
+    logger.info(f"Get all modules by type: {module_type.__name__}")
     import ray.new_dashboard.modules
 
     for module_loader, name, ispkg in pkgutil.walk_packages(
             ray.new_dashboard.modules.__path__,
             ray.new_dashboard.modules.__name__ + "."):
         importlib.import_module(name)
-    return module_type.__subclasses__()
+    return [
+        m for m in module_type.__subclasses__()
+        if getattr(m, "__ray_dashboard_module_enable__", True)
+    ]
 
 
 def to_posix_time(dt):
@@ -314,8 +327,7 @@ class Change:
         self.new = new
 
     def __str__(self):
-        return "Change(owner: {}, old: {}, new: {}".format(
-            self.owner, self.old, self.new)
+        return f"Change(owner: {self.owner}, old: {self.old}, new: {self.new}"
 
 
 class NotifyQueue:
@@ -338,6 +350,8 @@ class Dict(MutableMapping):
     :note: Only the first level data report change.
     """
 
+    ChangeItem = namedtuple("DictChangeItem", ["key", "value"])
+
     def __init__(self, *args, **kwargs):
         self._data = dict(*args, **kwargs)
         self.signal = Signal(self)
@@ -347,10 +361,14 @@ class Dict(MutableMapping):
         self._data[key] = value
         if len(self.signal) and old != value:
             if old is None:
-                co = self.signal.send(Change(owner=self, new={key: value}))
+                co = self.signal.send(
+                    Change(owner=self, new=Dict.ChangeItem(key, value)))
             else:
                 co = self.signal.send(
-                    Change(owner=self, old={key: old}, new={key: value}))
+                    Change(
+                        owner=self,
+                        old=Dict.ChangeItem(key, old),
+                        new=Dict.ChangeItem(key, value)))
             NotifyQueue.put(co)
 
     def __getitem__(self, item):
@@ -359,7 +377,8 @@ class Dict(MutableMapping):
     def __delitem__(self, key):
         old = self._data.pop(key, None)
         if len(self.signal) and old is not None:
-            co = self.signal.send(Change(owner=self, old={key: old}))
+            co = self.signal.send(
+                Change(owner=self, old=Dict.ChangeItem(key, old)))
             NotifyQueue.put(co)
 
     def __len__(self):
@@ -387,3 +406,19 @@ async def get_aioredis_client(redis_address, redis_password,
     # Raise exception from create_redis_pool
     return await aioredis.create_redis_pool(
         address=redis_address, password=redis_password)
+
+
+def async_loop_forever(interval_seconds):
+    def _wrapper(coro):
+        @functools.wraps(coro)
+        async def _looper(*args, **kwargs):
+            while True:
+                try:
+                    await coro(*args, **kwargs)
+                except Exception:
+                    logger.exception(f"Error looping coroutine {coro}.")
+                await asyncio.sleep(interval_seconds)
+
+        return _looper
+
+    return _wrapper

python/ray/services.py

@@ -1413,6 +1413,7 @@ def start_raylet(redis_address,
             os.path.dirname(os.path.abspath(__file__)),
             "new_dashboard/agent.py"),
         "--redis-address={}".format(redis_address),
+        "--metrics-export-port={}".format(metrics_export_port),
         "--node-manager-port={}".format(node_manager_port),
         "--object-store-name={}".format(plasma_store_name),
         "--raylet-name={}".format(raylet_name),

python/ray/test_utils.py

@@ -428,3 +428,11 @@ def get_error_message(pub_sub, num, error_type=None, timeout=5):
             time.sleep(0.01)
 
     return msgs
+
+
+def format_web_url(url):
+    """Format web url."""
+    url = url.replace("localhost", "http://127.0.0.1")
+    if not url.startswith("http://"):
+        return "http://" + url
+    return url

python/ray/utils.py

@@ -122,7 +122,7 @@ def push_error_to_driver_through_redis(redis_client,
     error_data = ray.gcs_utils.construct_error_message(job_id, error_type,
                                                        message, time.time())
     pubsub_msg = ray.gcs_utils.PubSubMessage()
-    pubsub_msg.id = job_id.hex()
+    pubsub_msg.id = job_id.binary()
     pubsub_msg.data = error_data
     redis_client.publish("ERROR_INFO:" + job_id.hex(),
                          pubsub_msg.SerializeAsString())

src/ray/core_worker/core_worker.cc

@@ -2146,6 +2146,7 @@ void CoreWorker::HandleGetCoreWorkerStats(const rpc::GetCoreWorkerStatsRequest &
   stats->set_current_task_func_desc(current_task_.FunctionDescriptor()->ToString());
   stats->set_ip_address(rpc_address_.ip_address());
   stats->set_port(rpc_address_.port());
+  stats->set_job_id(worker_context_.GetCurrentJobID().Binary());
   stats->set_actor_id(actor_id_.Binary());
   auto used_resources_map = stats->mutable_used_resources();
   for (auto const &it : *resource_ids_) {

src/ray/protobuf/common.proto

@@ -384,6 +384,8 @@ message CoreWorkerStats {
   string actor_title = 16;
   // Local reference table.
   repeated ObjectRefInfo object_refs = 17;
+  // Job ID.
+  bytes job_id = 18;
 }
 
 message MetricPoint {

src/ray/protobuf/node_manager.proto

@@ -114,6 +114,10 @@ message WorkerStats {
   CoreWorkerStats core_worker_stats = 3;
   // Error string if fetching core worker stats failed.
   string fetch_error = 4;
+  // Worker id of core worker.
+  bytes worker_id = 5;
+  // Worker language.
+  Language language = 6;
 }
 
 message GetNodeStatsReply {
@@ -122,6 +126,7 @@ message GetNodeStatsReply {
   uint32 num_workers = 3;
   repeated TaskSpec infeasible_tasks = 4;
   repeated TaskSpec ready_tasks = 5;
+  int32 pid = 6;
 }
 
 message GlobalGCRequest {

src/ray/raylet/node_manager.cc

@@ -3147,6 +3147,7 @@ void NodeManager::FlushObjectsToFree() {
 void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_request,
                                      rpc::GetNodeStatsReply *reply,
                                      rpc::SendReplyCallback send_reply_callback) {
+  reply->set_pid(getpid());
   for (const auto &task : local_queues_.GetTasks(TaskState::INFEASIBLE)) {
     if (task.GetTaskSpecification().IsActorCreationTask()) {
       auto infeasible_task = reply->add_infeasible_tasks();
@@ -3211,6 +3212,10 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_
     all_workers.push_back(driver);
     driver_ids.insert(driver->WorkerId());
   }
+  if (all_workers.empty()) {
+    send_reply_callback(Status::OK(), nullptr, nullptr);
+    return;
+  }
   for (const auto &worker : all_workers) {
     rpc::GetCoreWorkerStatsRequest request;
     request.set_intended_worker_id(worker->WorkerId().Binary());
@@ -3220,7 +3225,9 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &node_stats_
                      const ray::Status &status, const rpc::GetCoreWorkerStatsReply &r) {
           auto worker_stats = reply->add_workers_stats();
           worker_stats->set_pid(worker->GetProcess().GetId());
+          worker_stats->set_worker_id(worker->WorkerId().Binary());
           worker_stats->set_is_driver(driver_ids.contains(worker->WorkerId()));
+          worker_stats->set_language(worker->GetLanguage());
           reply->set_num_workers(reply->num_workers() + 1);
           if (status.ok()) {
             worker_stats->mutable_core_worker_stats()->MergeFrom(r.core_worker_stats());