We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[dashboard] When autoscaler adds/removes worker, dashboard crashes
<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->

### What is the problem?
This is originally reported by Pieterjan for 0.8.7 but also appears to be in issue in the latest nightly. The issue would appear to occur across 

### Reproduction (REQUIRED)
This is not a script, but given I will author the fix, I think this is sufficient:
- Start a cluster with autoscaling enabled
- Increase workload so that an additional worker or workers are spawned
- When one of those workers is removed, their log entries still exist on the backend
- The following stack trace occurs and crashes the dashboard: `/ray/dashboard/node_stats.py", line 63, in _insert_log_counts
    self._node_stats[hostname]["log_count"] = logs_by_pid` 

- [ ] I have verified my script runs in a clean environment and reproduces the issue.
- [x] I have verified the issue also occurs with the [latest wheels](https://docs.ray.io/en/latest/installation.html).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/dashboard/node_stats.py`
Content:
```
1 from collections import defaultdict
2 from ray.dashboard.util import to_unix_time, format_reply_id
3 from base64 import b64decode
4 import ray
5 import threading
6 import json
7 import traceback
8 import copy
9 import logging
10 import datetime
11 import time
12 from typing import Dict
13 import re
14 
15 from operator import itemgetter
16 
17 logger = logging.getLogger(__name__)
18 
19 
20 class NodeStats(threading.Thread):
21     def __init__(self, redis_address, redis_password=None):
22         self.redis_key = "{}.*".format(ray.gcs_utils.REPORTER_CHANNEL)
23         self.redis_client = ray.services.create_redis_client(
24             redis_address, password=redis_password)
25 
26         self._node_stats = {}
27         self._ip_to_hostname = {}
28         self._addr_to_owner_addr = {}
29         self._addr_to_actor_id = {}
30         self._addr_to_extra_info_dict = {}
31         self._node_stats_lock = threading.Lock()
32 
33         self._default_info = {
34             "actorId": "",
35             "children": {},
36             "currentTaskFuncDesc": [],
37             "ipAddress": "",
38             "jobId": "",
39             "numExecutedTasks": 0,
40             "numLocalObjects": 0,
41             "numObjectRefsInScope": 0,
42             "port": 0,
43             "state": 0,
44             "taskQueueLength": 0,
45             "usedObjectStoreMemory": 0,
46             "usedResources": {},
47         }
48 
49         # Mapping from IP address to PID to list of log lines
50         self._logs = defaultdict(lambda: defaultdict(list))
51 
52         # Mapping from IP address to PID to list of error messages
53         self._errors = defaultdict(lambda: defaultdict(list))
54 
55         ray.state.state._initialize_global_state(
56             redis_address=redis_address, redis_password=redis_password)
57 
58         super().__init__()
59 
60     def _insert_log_counts(self):
61         for ip, logs_by_pid in self._logs.items():
62             hostname = self._ip_to_hostname[ip]
63             logs_by_pid = {pid: len(logs) for pid, logs in logs_by_pid.items()}
64             self._node_stats[hostname]["log_count"] = logs_by_pid
65 
66     def _insert_error_counts(self):
67         for ip, errs_by_pid in self._errors.items():
68             hostname = self._ip_to_hostname[ip]
69             errs_by_pid = {pid: len(errs) for pid, errs in errs_by_pid.items()}
70             self._node_stats[hostname]["error_count"] = errs_by_pid
71 
72     def _purge_outdated_stats(self):
73         def current(then, now):
74             if (now - then) > 5:
75                 return False
76 
77             return True
78 
79         now = to_unix_time(datetime.datetime.utcnow())
80         self._node_stats = {
81             k: v
82             for k, v in self._node_stats.items() if current(v["now"], now)
83         }
84 
85     def get_node_stats(self):
86         with self._node_stats_lock:
87             self._purge_outdated_stats()
88             self._insert_error_counts()
89             self._insert_log_counts()
90             node_stats = sorted(
91                 (v for v in self._node_stats.values()),
92                 key=itemgetter("boot_time"))
93             return {"clients": node_stats}
94 
95     # Gets actors in a flat way to allow for grouping by actor type.
96     def get_actors(self, workers_info_by_node, infeasible_tasks, ready_tasks):
97         now = time.time()
98         actors: Dict[str, Dict[str, any]] = {}
99         # construct flattened actor tree
100         with self._node_stats_lock:
101             for addr, actor_id in self._addr_to_actor_id.items():
102                 actors[actor_id] = copy.deepcopy(self._default_info)
103                 actors[actor_id].update(self._addr_to_extra_info_dict[addr])
104 
105             for node_id, workers_info in workers_info_by_node.items():
106                 for worker_info in workers_info:
107                     if "coreWorkerStats" in worker_info:
108                         core_worker_stats = worker_info["coreWorkerStats"]
109                         addr = (core_worker_stats["ipAddress"],
110                                 str(core_worker_stats["port"]))
111                         if addr in self._addr_to_actor_id:
112                             actor_info = actors[self._addr_to_actor_id[addr]]
113                             format_reply_id(core_worker_stats)
114                             actor_info.update(core_worker_stats)
115                             actor_info["averageTaskExecutionSpeed"] = round(
116                                 actor_info["numExecutedTasks"] /
117                                 (now - actor_info["timestamp"] / 1000), 2)
118                             actor_info["nodeId"] = node_id
119                             actor_info["pid"] = worker_info["pid"]
120 
121             def _update_from_actor_tasks(task, task_spec_type,
122                                          invalid_state_type):
123                 actor_id = ray.utils.binary_to_hex(
124                     b64decode(task[task_spec_type]["actorId"]))
125                 task["state"] = -1
126                 task["invalidStateType"] = invalid_state_type
127                 task["actorTitle"] = task["functionDescriptor"][
128                     "pythonFunctionDescriptor"]["className"]
129                 format_reply_id(task)
130                 actors[actor_id] = task
131 
132             for infeasible_task in infeasible_tasks:
133                 _update_from_actor_tasks(infeasible_task,
134                                          "actorCreationTaskSpec",
135                                          "infeasibleActor")
136 
137             for ready_task in ready_tasks:
138                 _update_from_actor_tasks(ready_task, "actorCreationTaskSpec",
139                                          "pendingActor")
140 
141         return actors
142 
143     # Gets actors in a nested structure showing parent child relationships
144     def get_actor_tree(self, workers_info_by_node, infeasible_tasks,
145                        ready_tasks):
146         now = time.time()
147         # construct flattened actor tree
148         flattened_tree = {"root": {"children": {}}}
149         child_to_parent = {}
150         with self._node_stats_lock:
151             for addr, actor_id in self._addr_to_actor_id.items():
152                 flattened_tree[actor_id] = copy.deepcopy(self._default_info)
153                 flattened_tree[actor_id].update(
154                     self._addr_to_extra_info_dict[addr])
155                 parent_id = self._addr_to_actor_id.get(
156                     self._addr_to_owner_addr[addr], "root")
157                 child_to_parent[actor_id] = parent_id
158 
159             for node_id, workers_info in workers_info_by_node.items():
160                 for worker_info in workers_info:
161                     if "coreWorkerStats" in worker_info:
162                         core_worker_stats = worker_info["coreWorkerStats"]
163                         addr = (core_worker_stats["ipAddress"],
164                                 str(core_worker_stats["port"]))
165                         if addr in self._addr_to_actor_id:
166                             actor_info = flattened_tree[self._addr_to_actor_id[
167                                 addr]]
168                             format_reply_id(core_worker_stats)
169                             actor_info.update(core_worker_stats)
170                             actor_info["averageTaskExecutionSpeed"] = round(
171                                 actor_info["numExecutedTasks"] /
172                                 (now - actor_info["timestamp"] / 1000), 2)
173                             actor_info["nodeId"] = node_id
174                             actor_info["pid"] = worker_info["pid"]
175 
176             def _update_flatten_tree(task, task_spec_type, invalid_state_type):
177                 actor_id = ray.utils.binary_to_hex(
178                     b64decode(task[task_spec_type]["actorId"]))
179                 caller_addr = (task["callerAddress"]["ipAddress"],
180                                str(task["callerAddress"]["port"]))
181                 caller_id = self._addr_to_actor_id.get(caller_addr, "root")
182                 child_to_parent[actor_id] = caller_id
183                 task["state"] = -1
184                 task["invalidStateType"] = invalid_state_type
185                 task["actorTitle"] = task["functionDescriptor"][
186                     "pythonFunctionDescriptor"]["className"]
187                 format_reply_id(task)
188                 flattened_tree[actor_id] = task
189 
190             for infeasible_task in infeasible_tasks:
191                 _update_flatten_tree(infeasible_task, "actorCreationTaskSpec",
192                                      "infeasibleActor")
193 
194             for ready_task in ready_tasks:
195                 _update_flatten_tree(ready_task, "actorCreationTaskSpec",
196                                      "pendingActor")
197 
198         # construct actor tree
199         actor_tree = flattened_tree
200         for actor_id, parent_id in child_to_parent.items():
201             actor_tree[parent_id]["children"][actor_id] = actor_tree[actor_id]
202         return actor_tree["root"]["children"]
203 
204     def get_logs(self, hostname, pid):
205         ip = self._node_stats.get(hostname, {"ip": None})["ip"]
206         logs = self._logs.get(ip, {})
207         if pid:
208             logs = {pid: logs.get(pid, [])}
209         return logs
210 
211     def get_errors(self, hostname, pid):
212         ip = self._node_stats.get(hostname, {"ip": None})["ip"]
213         errors = self._errors.get(ip, {})
214         if pid:
215             errors = {pid: errors.get(pid, [])}
216         return errors
217 
218     def run(self):
219         p = self.redis_client.pubsub(ignore_subscribe_messages=True)
220 
221         p.psubscribe(self.redis_key)
222         logger.info("NodeStats: subscribed to {}".format(self.redis_key))
223 
224         log_channel = ray.gcs_utils.LOG_FILE_CHANNEL
225         p.subscribe(log_channel)
226         logger.info("NodeStats: subscribed to {}".format(log_channel))
227 
228         error_channel = ray.gcs_utils.RAY_ERROR_PUBSUB_PATTERN
229         p.psubscribe(error_channel)
230         logger.info("NodeStats: subscribed to {}".format(error_channel))
231 
232         actor_channel = ray.gcs_utils.RAY_ACTOR_PUBSUB_PATTERN
233         p.psubscribe(actor_channel)
234         logger.info("NodeStats: subscribed to {}".format(actor_channel))
235 
236         current_actor_table = ray.actors()
237         with self._node_stats_lock:
238             for actor_data in current_actor_table.values():
239                 addr = (actor_data["Address"]["IPAddress"],
240                         str(actor_data["Address"]["Port"]))
241                 owner_addr = (actor_data["OwnerAddress"]["IPAddress"],
242                               str(actor_data["OwnerAddress"]["Port"]))
243                 self._addr_to_owner_addr[addr] = owner_addr
244                 self._addr_to_actor_id[addr] = actor_data["ActorID"]
245                 self._addr_to_extra_info_dict[addr] = {
246                     "jobId": actor_data["JobID"],
247                     "state": actor_data["State"],
248                     "timestamp": actor_data["Timestamp"]
249                 }
250 
251         for x in p.listen():
252             try:
253                 with self._node_stats_lock:
254                     channel = ray.utils.decode(x["channel"])\
255                                 if "pattern" not in x or x["pattern"] is None\
256                                 else x["pattern"]
257                     data = x["data"]
258                     if channel == log_channel:
259                         data = json.loads(ray.utils.decode(data))
260                         ip = data["ip"]
261                         pid = str(data["pid"])
262                         self._logs[ip][pid].extend(data["lines"])
263                     elif channel == str(error_channel):
264                         pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(
265                             data)
266                         error_data = ray.gcs_utils.ErrorTableData.FromString(
267                             pubsub_msg.data)
268                         message = error_data.error_message
269                         message = re.sub(r"\x1b\[\d+m", "", message)
270                         match = re.search(r"\(pid=(\d+), ip=(.*?)\)", message)
271                         if match:
272                             pid = match.group(1)
273                             ip = match.group(2)
274                             self._errors[ip][pid].append({
275                                 "message": message,
276                                 "timestamp": error_data.timestamp,
277                                 "type": error_data.type
278                             })
279                     elif channel == actor_channel:
280                         pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(
281                             data)
282                         actor_data = ray.gcs_utils.ActorTableData.FromString(
283                             pubsub_msg.data)
284                         addr = (actor_data.address.ip_address,
285                                 str(actor_data.address.port))
286                         owner_addr = (actor_data.owner_address.ip_address,
287                                       str(actor_data.owner_address.port))
288                         self._addr_to_owner_addr[addr] = owner_addr
289                         self._addr_to_actor_id[addr] = ray.utils.binary_to_hex(
290                             actor_data.actor_id)
291                         self._addr_to_extra_info_dict[addr] = {
292                             "jobId": ray.utils.binary_to_hex(
293                                 actor_data.job_id),
294                             "state": actor_data.state,
295                             "timestamp": actor_data.timestamp
296                         }
297                     elif channel == ray.gcs_utils.RAY_REPORTER_PUBSUB_PATTERN:
298                         data = json.loads(ray.utils.decode(data))
299                         self._ip_to_hostname[data["ip"]] = data["hostname"]
300                         self._node_stats[data["hostname"]] = data
301                     else:
302                         logger.warning("Unexpected channel data received, "
303                                        "channel: {}, data: {}".format(
304                                            channel,
305                                            json.loads(ray.utils.decode(data))))
306 
307             except Exception:
308                 logger.exception(traceback.format_exc())
309                 continue
310
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/dashboard/node_stats.py b/python/ray/dashboard/node_stats.py
--- a/python/ray/dashboard/node_stats.py
+++ b/python/ray/dashboard/node_stats.py
@@ -60,14 +60,22 @@
     def _insert_log_counts(self):
         for ip, logs_by_pid in self._logs.items():
             hostname = self._ip_to_hostname[ip]
-            logs_by_pid = {pid: len(logs) for pid, logs in logs_by_pid.items()}
-            self._node_stats[hostname]["log_count"] = logs_by_pid
+            if hostname in self._node_stats:
+                logs_by_pid = {
+                    pid: len(logs)
+                    for pid, logs in logs_by_pid.items()
+                }
+                self._node_stats[hostname]["log_count"] = logs_by_pid
 
     def _insert_error_counts(self):
         for ip, errs_by_pid in self._errors.items():
             hostname = self._ip_to_hostname[ip]
-            errs_by_pid = {pid: len(errs) for pid, errs in errs_by_pid.items()}
-            self._node_stats[hostname]["error_count"] = errs_by_pid
+            if hostname in self._node_stats:
+                errs_by_pid = {
+                    pid: len(errs)
+                    for pid, errs in errs_by_pid.items()
+                }
+                self._node_stats[hostname]["error_count"] = errs_by_pid
 
     def _purge_outdated_stats(self):
         def current(then, now):

{"golden_diff": "diff --git a/python/ray/dashboard/node_stats.py b/python/ray/dashboard/node_stats.py\n--- a/python/ray/dashboard/node_stats.py\n+++ b/python/ray/dashboard/node_stats.py\n@@ -60,14 +60,22 @@\n     def _insert_log_counts(self):\n         for ip, logs_by_pid in self._logs.items():\n             hostname = self._ip_to_hostname[ip]\n-            logs_by_pid = {pid: len(logs) for pid, logs in logs_by_pid.items()}\n-            self._node_stats[hostname][\"log_count\"] = logs_by_pid\n+            if hostname in self._node_stats:\n+                logs_by_pid = {\n+                    pid: len(logs)\n+                    for pid, logs in logs_by_pid.items()\n+                }\n+                self._node_stats[hostname][\"log_count\"] = logs_by_pid\n \n     def _insert_error_counts(self):\n         for ip, errs_by_pid in self._errors.items():\n             hostname = self._ip_to_hostname[ip]\n-            errs_by_pid = {pid: len(errs) for pid, errs in errs_by_pid.items()}\n-            self._node_stats[hostname][\"error_count\"] = errs_by_pid\n+            if hostname in self._node_stats:\n+                errs_by_pid = {\n+                    pid: len(errs)\n+                    for pid, errs in errs_by_pid.items()\n+                }\n+                self._node_stats[hostname][\"error_count\"] = errs_by_pid\n \n     def _purge_outdated_stats(self):\n         def current(then, now):\n", "issue": "[dashboard] When autoscaler adds/removes worker, dashboard crashes\n<!--Please include [tune], [rllib], [autoscaler] etc. in the issue title if relevant-->\r\n\r\n### What is the problem?\r\nThis is originally reported by Pieterjan for 0.8.7 but also appears to be in issue in the latest nightly. The issue would appear to occur across \r\n\r\n### Reproduction (REQUIRED)\r\nThis is not a script, but given I will author the fix, I think this is sufficient:\r\n- Start a cluster with autoscaling enabled\r\n- Increase workload so that an additional worker or workers are spawned\r\n- When one of those workers is removed, their log entries still exist on the backend\r\n- The following stack trace occurs and crashes the dashboard: `/ray/dashboard/node_stats.py\", line 63, in _insert_log_counts\r\n    self._node_stats[hostname][\"log_count\"] = logs_by_pid` \r\n\r\n- [ ] I have verified my script runs in a clean environment and reproduces the issue.\r\n- [x] I have verified the issue also occurs with the [latest wheels](https://docs.ray.io/en/latest/installation.html).\r\n\n", "before_files": [{"content": "from collections import defaultdict\nfrom ray.dashboard.util import to_unix_time, format_reply_id\nfrom base64 import b64decode\nimport ray\nimport threading\nimport json\nimport traceback\nimport copy\nimport logging\nimport datetime\nimport time\nfrom typing import Dict\nimport re\n\nfrom operator import itemgetter\n\nlogger = logging.getLogger(__name__)\n\n\nclass NodeStats(threading.Thread):\n    def __init__(self, redis_address, redis_password=None):\n        self.redis_key = \"{}.*\".format(ray.gcs_utils.REPORTER_CHANNEL)\n        self.redis_client = ray.services.create_redis_client(\n            redis_address, password=redis_password)\n\n        self._node_stats = {}\n        self._ip_to_hostname = {}\n        self._addr_to_owner_addr = {}\n        self._addr_to_actor_id = {}\n        self._addr_to_extra_info_dict = {}\n        self._node_stats_lock = threading.Lock()\n\n        self._default_info = {\n            \"actorId\": \"\",\n            \"children\": {},\n            \"currentTaskFuncDesc\": [],\n            \"ipAddress\": \"\",\n            \"jobId\": \"\",\n            \"numExecutedTasks\": 0,\n            \"numLocalObjects\": 0,\n            \"numObjectRefsInScope\": 0,\n            \"port\": 0,\n            \"state\": 0,\n            \"taskQueueLength\": 0,\n            \"usedObjectStoreMemory\": 0,\n            \"usedResources\": {},\n        }\n\n        # Mapping from IP address to PID to list of log lines\n        self._logs = defaultdict(lambda: defaultdict(list))\n\n        # Mapping from IP address to PID to list of error messages\n        self._errors = defaultdict(lambda: defaultdict(list))\n\n        ray.state.state._initialize_global_state(\n            redis_address=redis_address, redis_password=redis_password)\n\n        super().__init__()\n\n    def _insert_log_counts(self):\n        for ip, logs_by_pid in self._logs.items():\n            hostname = self._ip_to_hostname[ip]\n            logs_by_pid = {pid: len(logs) for pid, logs in logs_by_pid.items()}\n            self._node_stats[hostname][\"log_count\"] = logs_by_pid\n\n    def _insert_error_counts(self):\n        for ip, errs_by_pid in self._errors.items():\n            hostname = self._ip_to_hostname[ip]\n            errs_by_pid = {pid: len(errs) for pid, errs in errs_by_pid.items()}\n            self._node_stats[hostname][\"error_count\"] = errs_by_pid\n\n    def _purge_outdated_stats(self):\n        def current(then, now):\n            if (now - then) > 5:\n                return False\n\n            return True\n\n        now = to_unix_time(datetime.datetime.utcnow())\n        self._node_stats = {\n            k: v\n            for k, v in self._node_stats.items() if current(v[\"now\"], now)\n        }\n\n    def get_node_stats(self):\n        with self._node_stats_lock:\n            self._purge_outdated_stats()\n            self._insert_error_counts()\n            self._insert_log_counts()\n            node_stats = sorted(\n                (v for v in self._node_stats.values()),\n                key=itemgetter(\"boot_time\"))\n            return {\"clients\": node_stats}\n\n    # Gets actors in a flat way to allow for grouping by actor type.\n    def get_actors(self, workers_info_by_node, infeasible_tasks, ready_tasks):\n        now = time.time()\n        actors: Dict[str, Dict[str, any]] = {}\n        # construct flattened actor tree\n        with self._node_stats_lock:\n            for addr, actor_id in self._addr_to_actor_id.items():\n                actors[actor_id] = copy.deepcopy(self._default_info)\n                actors[actor_id].update(self._addr_to_extra_info_dict[addr])\n\n            for node_id, workers_info in workers_info_by_node.items():\n                for worker_info in workers_info:\n                    if \"coreWorkerStats\" in worker_info:\n                        core_worker_stats = worker_info[\"coreWorkerStats\"]\n                        addr = (core_worker_stats[\"ipAddress\"],\n                                str(core_worker_stats[\"port\"]))\n                        if addr in self._addr_to_actor_id:\n                            actor_info = actors[self._addr_to_actor_id[addr]]\n                            format_reply_id(core_worker_stats)\n                            actor_info.update(core_worker_stats)\n                            actor_info[\"averageTaskExecutionSpeed\"] = round(\n                                actor_info[\"numExecutedTasks\"] /\n                                (now - actor_info[\"timestamp\"] / 1000), 2)\n                            actor_info[\"nodeId\"] = node_id\n                            actor_info[\"pid\"] = worker_info[\"pid\"]\n\n            def _update_from_actor_tasks(task, task_spec_type,\n                                         invalid_state_type):\n                actor_id = ray.utils.binary_to_hex(\n                    b64decode(task[task_spec_type][\"actorId\"]))\n                task[\"state\"] = -1\n                task[\"invalidStateType\"] = invalid_state_type\n                task[\"actorTitle\"] = task[\"functionDescriptor\"][\n                    \"pythonFunctionDescriptor\"][\"className\"]\n                format_reply_id(task)\n                actors[actor_id] = task\n\n            for infeasible_task in infeasible_tasks:\n                _update_from_actor_tasks(infeasible_task,\n                                         \"actorCreationTaskSpec\",\n                                         \"infeasibleActor\")\n\n            for ready_task in ready_tasks:\n                _update_from_actor_tasks(ready_task, \"actorCreationTaskSpec\",\n                                         \"pendingActor\")\n\n        return actors\n\n    # Gets actors in a nested structure showing parent child relationships\n    def get_actor_tree(self, workers_info_by_node, infeasible_tasks,\n                       ready_tasks):\n        now = time.time()\n        # construct flattened actor tree\n        flattened_tree = {\"root\": {\"children\": {}}}\n        child_to_parent = {}\n        with self._node_stats_lock:\n            for addr, actor_id in self._addr_to_actor_id.items():\n                flattened_tree[actor_id] = copy.deepcopy(self._default_info)\n                flattened_tree[actor_id].update(\n                    self._addr_to_extra_info_dict[addr])\n                parent_id = self._addr_to_actor_id.get(\n                    self._addr_to_owner_addr[addr], \"root\")\n                child_to_parent[actor_id] = parent_id\n\n            for node_id, workers_info in workers_info_by_node.items():\n                for worker_info in workers_info:\n                    if \"coreWorkerStats\" in worker_info:\n                        core_worker_stats = worker_info[\"coreWorkerStats\"]\n                        addr = (core_worker_stats[\"ipAddress\"],\n                                str(core_worker_stats[\"port\"]))\n                        if addr in self._addr_to_actor_id:\n                            actor_info = flattened_tree[self._addr_to_actor_id[\n                                addr]]\n                            format_reply_id(core_worker_stats)\n                            actor_info.update(core_worker_stats)\n                            actor_info[\"averageTaskExecutionSpeed\"] = round(\n                                actor_info[\"numExecutedTasks\"] /\n                                (now - actor_info[\"timestamp\"] / 1000), 2)\n                            actor_info[\"nodeId\"] = node_id\n                            actor_info[\"pid\"] = worker_info[\"pid\"]\n\n            def _update_flatten_tree(task, task_spec_type, invalid_state_type):\n                actor_id = ray.utils.binary_to_hex(\n                    b64decode(task[task_spec_type][\"actorId\"]))\n                caller_addr = (task[\"callerAddress\"][\"ipAddress\"],\n                               str(task[\"callerAddress\"][\"port\"]))\n                caller_id = self._addr_to_actor_id.get(caller_addr, \"root\")\n                child_to_parent[actor_id] = caller_id\n                task[\"state\"] = -1\n                task[\"invalidStateType\"] = invalid_state_type\n                task[\"actorTitle\"] = task[\"functionDescriptor\"][\n                    \"pythonFunctionDescriptor\"][\"className\"]\n                format_reply_id(task)\n                flattened_tree[actor_id] = task\n\n            for infeasible_task in infeasible_tasks:\n                _update_flatten_tree(infeasible_task, \"actorCreationTaskSpec\",\n                                     \"infeasibleActor\")\n\n            for ready_task in ready_tasks:\n                _update_flatten_tree(ready_task, \"actorCreationTaskSpec\",\n                                     \"pendingActor\")\n\n        # construct actor tree\n        actor_tree = flattened_tree\n        for actor_id, parent_id in child_to_parent.items():\n            actor_tree[parent_id][\"children\"][actor_id] = actor_tree[actor_id]\n        return actor_tree[\"root\"][\"children\"]\n\n    def get_logs(self, hostname, pid):\n        ip = self._node_stats.get(hostname, {\"ip\": None})[\"ip\"]\n        logs = self._logs.get(ip, {})\n        if pid:\n            logs = {pid: logs.get(pid, [])}\n        return logs\n\n    def get_errors(self, hostname, pid):\n        ip = self._node_stats.get(hostname, {\"ip\": None})[\"ip\"]\n        errors = self._errors.get(ip, {})\n        if pid:\n            errors = {pid: errors.get(pid, [])}\n        return errors\n\n    def run(self):\n        p = self.redis_client.pubsub(ignore_subscribe_messages=True)\n\n        p.psubscribe(self.redis_key)\n        logger.info(\"NodeStats: subscribed to {}\".format(self.redis_key))\n\n        log_channel = ray.gcs_utils.LOG_FILE_CHANNEL\n        p.subscribe(log_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(log_channel))\n\n        error_channel = ray.gcs_utils.RAY_ERROR_PUBSUB_PATTERN\n        p.psubscribe(error_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(error_channel))\n\n        actor_channel = ray.gcs_utils.RAY_ACTOR_PUBSUB_PATTERN\n        p.psubscribe(actor_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(actor_channel))\n\n        current_actor_table = ray.actors()\n        with self._node_stats_lock:\n            for actor_data in current_actor_table.values():\n                addr = (actor_data[\"Address\"][\"IPAddress\"],\n                        str(actor_data[\"Address\"][\"Port\"]))\n                owner_addr = (actor_data[\"OwnerAddress\"][\"IPAddress\"],\n                              str(actor_data[\"OwnerAddress\"][\"Port\"]))\n                self._addr_to_owner_addr[addr] = owner_addr\n                self._addr_to_actor_id[addr] = actor_data[\"ActorID\"]\n                self._addr_to_extra_info_dict[addr] = {\n                    \"jobId\": actor_data[\"JobID\"],\n                    \"state\": actor_data[\"State\"],\n                    \"timestamp\": actor_data[\"Timestamp\"]\n                }\n\n        for x in p.listen():\n            try:\n                with self._node_stats_lock:\n                    channel = ray.utils.decode(x[\"channel\"])\\\n                                if \"pattern\" not in x or x[\"pattern\"] is None\\\n                                else x[\"pattern\"]\n                    data = x[\"data\"]\n                    if channel == log_channel:\n                        data = json.loads(ray.utils.decode(data))\n                        ip = data[\"ip\"]\n                        pid = str(data[\"pid\"])\n                        self._logs[ip][pid].extend(data[\"lines\"])\n                    elif channel == str(error_channel):\n                        pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(\n                            data)\n                        error_data = ray.gcs_utils.ErrorTableData.FromString(\n                            pubsub_msg.data)\n                        message = error_data.error_message\n                        message = re.sub(r\"\\x1b\\[\\d+m\", \"\", message)\n                        match = re.search(r\"\\(pid=(\\d+), ip=(.*?)\\)\", message)\n                        if match:\n                            pid = match.group(1)\n                            ip = match.group(2)\n                            self._errors[ip][pid].append({\n                                \"message\": message,\n                                \"timestamp\": error_data.timestamp,\n                                \"type\": error_data.type\n                            })\n                    elif channel == actor_channel:\n                        pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(\n                            data)\n                        actor_data = ray.gcs_utils.ActorTableData.FromString(\n                            pubsub_msg.data)\n                        addr = (actor_data.address.ip_address,\n                                str(actor_data.address.port))\n                        owner_addr = (actor_data.owner_address.ip_address,\n                                      str(actor_data.owner_address.port))\n                        self._addr_to_owner_addr[addr] = owner_addr\n                        self._addr_to_actor_id[addr] = ray.utils.binary_to_hex(\n                            actor_data.actor_id)\n                        self._addr_to_extra_info_dict[addr] = {\n                            \"jobId\": ray.utils.binary_to_hex(\n                                actor_data.job_id),\n                            \"state\": actor_data.state,\n                            \"timestamp\": actor_data.timestamp\n                        }\n                    elif channel == ray.gcs_utils.RAY_REPORTER_PUBSUB_PATTERN:\n                        data = json.loads(ray.utils.decode(data))\n                        self._ip_to_hostname[data[\"ip\"]] = data[\"hostname\"]\n                        self._node_stats[data[\"hostname\"]] = data\n                    else:\n                        logger.warning(\"Unexpected channel data received, \"\n                                       \"channel: {}, data: {}\".format(\n                                           channel,\n                                           json.loads(ray.utils.decode(data))))\n\n            except Exception:\n                logger.exception(traceback.format_exc())\n                continue\n", "path": "python/ray/dashboard/node_stats.py"}], "after_files": [{"content": "from collections import defaultdict\nfrom ray.dashboard.util import to_unix_time, format_reply_id\nfrom base64 import b64decode\nimport ray\nimport threading\nimport json\nimport traceback\nimport copy\nimport logging\nimport datetime\nimport time\nfrom typing import Dict\nimport re\n\nfrom operator import itemgetter\n\nlogger = logging.getLogger(__name__)\n\n\nclass NodeStats(threading.Thread):\n    def __init__(self, redis_address, redis_password=None):\n        self.redis_key = \"{}.*\".format(ray.gcs_utils.REPORTER_CHANNEL)\n        self.redis_client = ray.services.create_redis_client(\n            redis_address, password=redis_password)\n\n        self._node_stats = {}\n        self._ip_to_hostname = {}\n        self._addr_to_owner_addr = {}\n        self._addr_to_actor_id = {}\n        self._addr_to_extra_info_dict = {}\n        self._node_stats_lock = threading.Lock()\n\n        self._default_info = {\n            \"actorId\": \"\",\n            \"children\": {},\n            \"currentTaskFuncDesc\": [],\n            \"ipAddress\": \"\",\n            \"jobId\": \"\",\n            \"numExecutedTasks\": 0,\n            \"numLocalObjects\": 0,\n            \"numObjectRefsInScope\": 0,\n            \"port\": 0,\n            \"state\": 0,\n            \"taskQueueLength\": 0,\n            \"usedObjectStoreMemory\": 0,\n            \"usedResources\": {},\n        }\n\n        # Mapping from IP address to PID to list of log lines\n        self._logs = defaultdict(lambda: defaultdict(list))\n\n        # Mapping from IP address to PID to list of error messages\n        self._errors = defaultdict(lambda: defaultdict(list))\n\n        ray.state.state._initialize_global_state(\n            redis_address=redis_address, redis_password=redis_password)\n\n        super().__init__()\n\n    def _insert_log_counts(self):\n        for ip, logs_by_pid in self._logs.items():\n            hostname = self._ip_to_hostname[ip]\n            if hostname in self._node_stats:\n                logs_by_pid = {\n                    pid: len(logs)\n                    for pid, logs in logs_by_pid.items()\n                }\n                self._node_stats[hostname][\"log_count\"] = logs_by_pid\n\n    def _insert_error_counts(self):\n        for ip, errs_by_pid in self._errors.items():\n            hostname = self._ip_to_hostname[ip]\n            if hostname in self._node_stats:\n                errs_by_pid = {\n                    pid: len(errs)\n                    for pid, errs in errs_by_pid.items()\n                }\n                self._node_stats[hostname][\"error_count\"] = errs_by_pid\n\n    def _purge_outdated_stats(self):\n        def current(then, now):\n            if (now - then) > 5:\n                return False\n\n            return True\n\n        now = to_unix_time(datetime.datetime.utcnow())\n        self._node_stats = {\n            k: v\n            for k, v in self._node_stats.items() if current(v[\"now\"], now)\n        }\n\n    def get_node_stats(self):\n        with self._node_stats_lock:\n            self._purge_outdated_stats()\n            self._insert_error_counts()\n            self._insert_log_counts()\n            node_stats = sorted(\n                (v for v in self._node_stats.values()),\n                key=itemgetter(\"boot_time\"))\n            return {\"clients\": node_stats}\n\n    # Gets actors in a flat way to allow for grouping by actor type.\n    def get_actors(self, workers_info_by_node, infeasible_tasks, ready_tasks):\n        now = time.time()\n        actors: Dict[str, Dict[str, any]] = {}\n        # construct flattened actor tree\n        with self._node_stats_lock:\n            for addr, actor_id in self._addr_to_actor_id.items():\n                actors[actor_id] = copy.deepcopy(self._default_info)\n                actors[actor_id].update(self._addr_to_extra_info_dict[addr])\n\n            for node_id, workers_info in workers_info_by_node.items():\n                for worker_info in workers_info:\n                    if \"coreWorkerStats\" in worker_info:\n                        core_worker_stats = worker_info[\"coreWorkerStats\"]\n                        addr = (core_worker_stats[\"ipAddress\"],\n                                str(core_worker_stats[\"port\"]))\n                        if addr in self._addr_to_actor_id:\n                            actor_info = actors[self._addr_to_actor_id[addr]]\n                            format_reply_id(core_worker_stats)\n                            actor_info.update(core_worker_stats)\n                            actor_info[\"averageTaskExecutionSpeed\"] = round(\n                                actor_info[\"numExecutedTasks\"] /\n                                (now - actor_info[\"timestamp\"] / 1000), 2)\n                            actor_info[\"nodeId\"] = node_id\n                            actor_info[\"pid\"] = worker_info[\"pid\"]\n\n            def _update_from_actor_tasks(task, task_spec_type,\n                                         invalid_state_type):\n                actor_id = ray.utils.binary_to_hex(\n                    b64decode(task[task_spec_type][\"actorId\"]))\n                task[\"state\"] = -1\n                task[\"invalidStateType\"] = invalid_state_type\n                task[\"actorTitle\"] = task[\"functionDescriptor\"][\n                    \"pythonFunctionDescriptor\"][\"className\"]\n                format_reply_id(task)\n                actors[actor_id] = task\n\n            for infeasible_task in infeasible_tasks:\n                _update_from_actor_tasks(infeasible_task,\n                                         \"actorCreationTaskSpec\",\n                                         \"infeasibleActor\")\n\n            for ready_task in ready_tasks:\n                _update_from_actor_tasks(ready_task, \"actorCreationTaskSpec\",\n                                         \"pendingActor\")\n\n        return actors\n\n    # Gets actors in a nested structure showing parent child relationships\n    def get_actor_tree(self, workers_info_by_node, infeasible_tasks,\n                       ready_tasks):\n        now = time.time()\n        # construct flattened actor tree\n        flattened_tree = {\"root\": {\"children\": {}}}\n        child_to_parent = {}\n        with self._node_stats_lock:\n            for addr, actor_id in self._addr_to_actor_id.items():\n                flattened_tree[actor_id] = copy.deepcopy(self._default_info)\n                flattened_tree[actor_id].update(\n                    self._addr_to_extra_info_dict[addr])\n                parent_id = self._addr_to_actor_id.get(\n                    self._addr_to_owner_addr[addr], \"root\")\n                child_to_parent[actor_id] = parent_id\n\n            for node_id, workers_info in workers_info_by_node.items():\n                for worker_info in workers_info:\n                    if \"coreWorkerStats\" in worker_info:\n                        core_worker_stats = worker_info[\"coreWorkerStats\"]\n                        addr = (core_worker_stats[\"ipAddress\"],\n                                str(core_worker_stats[\"port\"]))\n                        if addr in self._addr_to_actor_id:\n                            actor_info = flattened_tree[self._addr_to_actor_id[\n                                addr]]\n                            format_reply_id(core_worker_stats)\n                            actor_info.update(core_worker_stats)\n                            actor_info[\"averageTaskExecutionSpeed\"] = round(\n                                actor_info[\"numExecutedTasks\"] /\n                                (now - actor_info[\"timestamp\"] / 1000), 2)\n                            actor_info[\"nodeId\"] = node_id\n                            actor_info[\"pid\"] = worker_info[\"pid\"]\n\n            def _update_flatten_tree(task, task_spec_type, invalid_state_type):\n                actor_id = ray.utils.binary_to_hex(\n                    b64decode(task[task_spec_type][\"actorId\"]))\n                caller_addr = (task[\"callerAddress\"][\"ipAddress\"],\n                               str(task[\"callerAddress\"][\"port\"]))\n                caller_id = self._addr_to_actor_id.get(caller_addr, \"root\")\n                child_to_parent[actor_id] = caller_id\n                task[\"state\"] = -1\n                task[\"invalidStateType\"] = invalid_state_type\n                task[\"actorTitle\"] = task[\"functionDescriptor\"][\n                    \"pythonFunctionDescriptor\"][\"className\"]\n                format_reply_id(task)\n                flattened_tree[actor_id] = task\n\n            for infeasible_task in infeasible_tasks:\n                _update_flatten_tree(infeasible_task, \"actorCreationTaskSpec\",\n                                     \"infeasibleActor\")\n\n            for ready_task in ready_tasks:\n                _update_flatten_tree(ready_task, \"actorCreationTaskSpec\",\n                                     \"pendingActor\")\n\n        # construct actor tree\n        actor_tree = flattened_tree\n        for actor_id, parent_id in child_to_parent.items():\n            actor_tree[parent_id][\"children\"][actor_id] = actor_tree[actor_id]\n        return actor_tree[\"root\"][\"children\"]\n\n    def get_logs(self, hostname, pid):\n        ip = self._node_stats.get(hostname, {\"ip\": None})[\"ip\"]\n        logs = self._logs.get(ip, {})\n        if pid:\n            logs = {pid: logs.get(pid, [])}\n        return logs\n\n    def get_errors(self, hostname, pid):\n        ip = self._node_stats.get(hostname, {\"ip\": None})[\"ip\"]\n        errors = self._errors.get(ip, {})\n        if pid:\n            errors = {pid: errors.get(pid, [])}\n        return errors\n\n    def run(self):\n        p = self.redis_client.pubsub(ignore_subscribe_messages=True)\n\n        p.psubscribe(self.redis_key)\n        logger.info(\"NodeStats: subscribed to {}\".format(self.redis_key))\n\n        log_channel = ray.gcs_utils.LOG_FILE_CHANNEL\n        p.subscribe(log_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(log_channel))\n\n        error_channel = ray.gcs_utils.RAY_ERROR_PUBSUB_PATTERN\n        p.psubscribe(error_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(error_channel))\n\n        actor_channel = ray.gcs_utils.RAY_ACTOR_PUBSUB_PATTERN\n        p.psubscribe(actor_channel)\n        logger.info(\"NodeStats: subscribed to {}\".format(actor_channel))\n\n        current_actor_table = ray.actors()\n        with self._node_stats_lock:\n            for actor_data in current_actor_table.values():\n                addr = (actor_data[\"Address\"][\"IPAddress\"],\n                        str(actor_data[\"Address\"][\"Port\"]))\n                owner_addr = (actor_data[\"OwnerAddress\"][\"IPAddress\"],\n                              str(actor_data[\"OwnerAddress\"][\"Port\"]))\n                self._addr_to_owner_addr[addr] = owner_addr\n                self._addr_to_actor_id[addr] = actor_data[\"ActorID\"]\n                self._addr_to_extra_info_dict[addr] = {\n                    \"jobId\": actor_data[\"JobID\"],\n                    \"state\": actor_data[\"State\"],\n                    \"timestamp\": actor_data[\"Timestamp\"]\n                }\n\n        for x in p.listen():\n            try:\n                with self._node_stats_lock:\n                    channel = ray.utils.decode(x[\"channel\"])\\\n                                if \"pattern\" not in x or x[\"pattern\"] is None\\\n                                else x[\"pattern\"]\n                    data = x[\"data\"]\n                    if channel == log_channel:\n                        data = json.loads(ray.utils.decode(data))\n                        ip = data[\"ip\"]\n                        pid = str(data[\"pid\"])\n                        self._logs[ip][pid].extend(data[\"lines\"])\n                    elif channel == str(error_channel):\n                        pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(\n                            data)\n                        error_data = ray.gcs_utils.ErrorTableData.FromString(\n                            pubsub_msg.data)\n                        message = error_data.error_message\n                        message = re.sub(r\"\\x1b\\[\\d+m\", \"\", message)\n                        match = re.search(r\"\\(pid=(\\d+), ip=(.*?)\\)\", message)\n                        if match:\n                            pid = match.group(1)\n                            ip = match.group(2)\n                            self._errors[ip][pid].append({\n                                \"message\": message,\n                                \"timestamp\": error_data.timestamp,\n                                \"type\": error_data.type\n                            })\n                    elif channel == actor_channel:\n                        pubsub_msg = ray.gcs_utils.PubSubMessage.FromString(\n                            data)\n                        actor_data = ray.gcs_utils.ActorTableData.FromString(\n                            pubsub_msg.data)\n                        addr = (actor_data.address.ip_address,\n                                str(actor_data.address.port))\n                        owner_addr = (actor_data.owner_address.ip_address,\n                                      str(actor_data.owner_address.port))\n                        self._addr_to_owner_addr[addr] = owner_addr\n                        self._addr_to_actor_id[addr] = ray.utils.binary_to_hex(\n                            actor_data.actor_id)\n                        self._addr_to_extra_info_dict[addr] = {\n                            \"jobId\": ray.utils.binary_to_hex(\n                                actor_data.job_id),\n                            \"state\": actor_data.state,\n                            \"timestamp\": actor_data.timestamp\n                        }\n                    elif channel == ray.gcs_utils.RAY_REPORTER_PUBSUB_PATTERN:\n                        data = json.loads(ray.utils.decode(data))\n                        self._ip_to_hostname[data[\"ip\"]] = data[\"hostname\"]\n                        self._node_stats[data[\"hostname\"]] = data\n                    else:\n                        logger.warning(\"Unexpected channel data received, \"\n                                       \"channel: {}, data: {}\".format(\n                                           channel,\n                                           json.loads(ray.utils.decode(data))))\n\n            except Exception:\n                logger.exception(traceback.format_exc())\n                continue\n", "path": "python/ray/dashboard/node_stats.py"}]}