Add scheduler log tab to performance reports

distributed/dashboard/components/scheduler.py

@@ -32,6 +32,7 @@
     value,
 )
 from bokeh.models.widgets import DataTable, TableColumn
+from bokeh.models.widgets.markups import Div
 from bokeh.palettes import Viridis11
 from bokeh.plotting import figure
 from bokeh.themes import Theme
@@ -69,7 +70,7 @@
 from distributed.diagnostics.task_stream import color_of as ts_color_of
 from distributed.diagnostics.task_stream import colors as ts_color_lookup
 from distributed.metrics import time
-from distributed.utils import format_time, log_errors, parse_timedelta
+from distributed.utils import Logs, format_time, log_errors, parse_timedelta
 
 if dask.config.get("distributed.dashboard.export-tool"):
     from distributed.dashboard.export_tool import ExportTool
@@ -2162,6 +2163,13 @@ def update(self):
         self.source.data.update(data)
 
 
+class SchedulerLogs:
+    def __init__(self, scheduler):
+        logs = Logs(scheduler.get_logs())._repr_html_()
+
+        self.root = Div(text=logs)
+
+
 def systemmonitor_doc(scheduler, extra, doc):
     with log_errors():
         sysmon = SystemMonitor(scheduler, sizing_mode="stretch_both")

distributed/deploy/cluster.py

@@ -14,8 +14,7 @@
 
 from ..core import Status
 from ..utils import (
-    Log,
-    Logs,
+    MultiLogs,
     format_dashboard_link,
     log_errors,
     parse_timedelta,
@@ -208,21 +207,19 @@ def _log(self, log):
             print(log)
 
     async def _get_logs(self, cluster=True, scheduler=True, workers=True):
-        logs = Logs()
+        logs = MultiLogs()
 
         if cluster:
-            logs["Cluster"] = Log(
-                "\n".join(line[1] for line in self._cluster_manager_logs)
-            )
+            logs["Cluster"] = self._cluster_manager_logs
 
         if scheduler:
             L = await self.scheduler_comm.get_logs()
-            logs["Scheduler"] = Log("\n".join(line for level, line in L))
+            logs["Scheduler"] = L
 
         if workers:
             d = await self.scheduler_comm.worker_logs(workers=workers)
             for k, v in d.items():
-                logs[k] = Log("\n".join(line for level, line in v))
+                logs[k] = v
 
         return logs
 

distributed/scheduler.py

@@ -6905,6 +6905,11 @@ def profile_to_figure(state):
         sysmon = SystemMonitor(self, last_count=last_count, sizing_mode="stretch_both")
         sysmon.update()
 
+        # Scheduler logs
+        from distributed.dashboard.components.scheduler import SchedulerLogs
+
+        logs = SchedulerLogs(self)
+
         from bokeh.models import Div, Panel, Tabs
 
         import distributed
@@ -6963,12 +6968,14 @@ def profile_to_figure(state):
         )
         bandwidth_types = Panel(child=bandwidth_types.root, title="Bandwidth (Types)")
         system = Panel(child=sysmon.root, title="System")
+        logs = Panel(child=logs.root, title="Scheduler Logs")
 
         tabs = Tabs(
             tabs=[
                 html,
                 task_stream,
                 system,
+                logs,
                 compute,
                 workers,
                 scheduler,

distributed/tests/test_client.py

@@ -6308,6 +6308,7 @@ async def f(stacklevel):
     assert "Dask Performance Report" in data
     assert "x = da.random" in data
     assert "Threads: 4" in data
+    assert "distributed.scheduler - INFO - Clear task state" in data
     assert dask.__version__ in data
 
     # Stacklevel two captures code two frames back -- which in this case

distributed/tests/test_utils.py

@@ -19,9 +19,8 @@
 from distributed.utils import (
     LRU,
     All,
-    Log,
-    Logs,
     LoopRunner,
+    MultiLogs,
     TimeoutError,
     _maybe_complex,
     deprecated,
@@ -559,7 +558,7 @@ def test_format_bytes_compat():
 
 
 def test_logs():
-    d = Logs({"123": Log("Hello"), "456": Log("World!")})
+    d = MultiLogs({"123": [("INFO", "Hello")], "456": [("INFO", "World!")]})
     text = d._repr_html_()
     assert is_valid_xml("<div>" + text + "</div>")
     assert "Hello" in text

distributed/utils.py

@@ -1363,25 +1363,44 @@ def parse_ports(port):
 is_coroutine_function = iscoroutinefunction
 
 
-class Log(str):
-    """A container for logs"""
+class Log(tuple):
+    """A container for a single log entry"""
+
+    level_styles = {
+        "WARNING": "font-weight: bold; color: orange;",
+        "CRITICAL": "font-weight: bold; color: orangered;",
+        "ERROR": "font-weight: bold; color: crimson;",
+    }
 
     def _repr_html_(self):
-        return "<pre><code>\n{log}\n</code></pre>".format(
-            log=html.escape(self.rstrip())
+        level, message = self
+
+        style = "font-family: monospace; margin: 0;"
+        style += self.level_styles.get(level, "")
+
+        return '<p style="{style}">{message}</p>'.format(
+            style=html.escape(style),
+            message=html.escape(message),
         )
 
 
-class Logs(dict):
-    """A container for multiple logs"""
+class Logs(list):
+    """A container for a list of log entries"""
+
+    def _repr_html_(self):
+        return "\n".join(Log(entry)._repr_html_() for entry in self)
+
+
+class MultiLogs(dict):
+    """A container for a dict mapping strings to lists of log entries"""
 
     def _repr_html_(self):
         summaries = [
             "<details>\n"
             "<summary style='display:list-item'>{title}</summary>\n"
-            "{log}\n"
-            "</details>".format(title=title, log=log._repr_html_())
-            for title, log in sorted(self.items())
+            "{logs}\n"
+            "</details>".format(title=title, logs=Logs(entries)._repr_html_())
+            for title, entries in sorted(self.items())
         ]
         return "\n".join(summaries)