We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
log monitor report UnicodeDecodeError
<!--
General questions should be asked on the mailing list ray-dev@googlegroups.com.
Questions about how to use Ray should be asked on
[StackOverflow](https://stackoverflow.com/questions/tagged/ray).

Before submitting an issue, please fill out the following form.
-->

### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: centos 7
- **Ray installed from (source or binary)**: binay
- **Ray version**: 0.7.2 0.8
- **Python version**: python 3.7
- **Exact command to reproduce**:

<!--
You can obtain the Ray version with

python -c "import ray; print(ray.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->
Now the `log_monitor` monitor all the files under logs. This could be causing the following errors when we read those log file with `VIM` because of the `.*.swap` file will be created. 
```Traceback (most recent call last):
  File "/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py", line 278, in <module>
    raise e
  File "/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py", line 268, in <module>
    log_monitor.run()
  File "/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py", line 219, in run
    anything_published = self.check_log_files_and_publish_updates()
  File "/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py", line 175, in check_log_files_and_publish_updates
    next_line = file_info.file_handle.readline()
  File "/home/xianyang/anaconda3/lib/python3.7/codecs.py", line 322, in decode
    (result, consumed) = self._buffer_decode(data, self.errors, final)
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xca in position 21: invalid continuation byte

```
### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->


--- 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/log_monitor.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 import argparse
6 import errno
7 import json
8 import logging
9 import os
10 import time
11 import traceback
12 
13 import ray.ray_constants as ray_constants
14 import ray.services as services
15 import ray.utils
16 
17 # Logger for this module. It should be configured at the entry point
18 # into the program using Ray. Ray provides a default configuration at
19 # entry/init points.
20 logger = logging.getLogger(__name__)
21 
22 
23 class LogFileInfo(object):
24     def __init__(self,
25                  filename=None,
26                  size_when_last_opened=None,
27                  file_position=None,
28                  file_handle=None):
29         assert (filename is not None and size_when_last_opened is not None
30                 and file_position is not None)
31         self.filename = filename
32         self.size_when_last_opened = size_when_last_opened
33         self.file_position = file_position
34         self.file_handle = file_handle
35         self.worker_pid = None
36 
37 
38 class LogMonitor(object):
39     """A monitor process for monitoring Ray log files.
40 
41     This class mantains a list of open files and a list of closed log files. We
42     can't simply leave all files open because we'll run out of file
43     descriptors.
44 
45     The "run" method of this class will cycle between doing several things:
46     1. First, it will check if any new files have appeared in the log
47        directory. If so, they will be added to the list of closed files.
48     2. Then, if we are unable to open any new files, we will close all of the
49        files.
50     3. Then, we will open as many closed files as we can that may have new
51        lines (judged by an increase in file size since the last time the file
52        was opened).
53     4. Then we will loop through the open files and see if there are any new
54        lines in the file. If so, we will publish them to Redis.
55 
56     Attributes:
57         host (str): The hostname of this machine. Used to improve the log
58             messages published to Redis.
59         logs_dir (str): The directory that the log files are in.
60         redis_client: A client used to communicate with the Redis server.
61         log_filenames (set): This is the set of filenames of all files in
62             open_file_infos and closed_file_infos.
63         open_file_infos (list[LogFileInfo]): Info for all of the open files.
64         closed_file_infos (list[LogFileInfo]): Info for all of the closed
65             files.
66         can_open_more_files (bool): True if we can still open more files and
67             false otherwise.
68     """
69 
70     def __init__(self, logs_dir, redis_address, redis_password=None):
71         """Initialize the log monitor object."""
72         self.ip = services.get_node_ip_address()
73         self.logs_dir = logs_dir
74         self.redis_client = ray.services.create_redis_client(
75             redis_address, password=redis_password)
76         self.log_filenames = set()
77         self.open_file_infos = []
78         self.closed_file_infos = []
79         self.can_open_more_files = True
80 
81     def close_all_files(self):
82         """Close all open files (so that we can open more)."""
83         while len(self.open_file_infos) > 0:
84             file_info = self.open_file_infos.pop(0)
85             file_info.file_handle.close()
86             file_info.file_handle = None
87             self.closed_file_infos.append(file_info)
88         self.can_open_more_files = True
89 
90     def update_log_filenames(self):
91         """Update the list of log files to monitor."""
92         log_filenames = os.listdir(self.logs_dir)
93 
94         for log_filename in log_filenames:
95             full_path = os.path.join(self.logs_dir, log_filename)
96             if full_path not in self.log_filenames:
97                 self.log_filenames.add(full_path)
98                 self.closed_file_infos.append(
99                     LogFileInfo(
100                         filename=full_path,
101                         size_when_last_opened=0,
102                         file_position=0,
103                         file_handle=None))
104                 logger.info("Beginning to track file {}".format(log_filename))
105 
106     def open_closed_files(self):
107         """Open some closed files if they may have new lines.
108 
109         Opening more files may require us to close some of the already open
110         files.
111         """
112         if not self.can_open_more_files:
113             # If we can't open any more files. Close all of the files.
114             self.close_all_files()
115 
116         files_with_no_updates = []
117         while len(self.closed_file_infos) > 0:
118             if (len(self.open_file_infos) >=
119                     ray_constants.LOG_MONITOR_MAX_OPEN_FILES):
120                 self.can_open_more_files = False
121                 break
122 
123             file_info = self.closed_file_infos.pop(0)
124             assert file_info.file_handle is None
125             # Get the file size to see if it has gotten bigger since we last
126             # opened it.
127             try:
128                 file_size = os.path.getsize(file_info.filename)
129             except (IOError, OSError) as e:
130                 # Catch "file not found" errors.
131                 if e.errno == errno.ENOENT:
132                     logger.warning("Warning: The file {} was not "
133                                    "found.".format(file_info.filename))
134                     self.log_filenames.remove(file_info.filename)
135                     continue
136                 raise e
137 
138             # If some new lines have been added to this file, try to reopen the
139             # file.
140             if file_size > file_info.size_when_last_opened:
141                 try:
142                     f = open(file_info.filename, "r")
143                 except (IOError, OSError) as e:
144                     if e.errno == errno.ENOENT:
145                         logger.warning("Warning: The file {} was not "
146                                        "found.".format(file_info.filename))
147                         self.log_filenames.remove(file_info.filename)
148                         continue
149                     else:
150                         raise e
151 
152                 f.seek(file_info.file_position)
153                 file_info.filesize_when_last_opened = file_size
154                 file_info.file_handle = f
155                 self.open_file_infos.append(file_info)
156             else:
157                 files_with_no_updates.append(file_info)
158 
159         # Add the files with no changes back to the list of closed files.
160         self.closed_file_infos += files_with_no_updates
161 
162     def check_log_files_and_publish_updates(self):
163         """Get any changes to the log files and push updates to Redis.
164 
165         Returns:
166             True if anything was published and false otherwise.
167         """
168         anything_published = False
169         for file_info in self.open_file_infos:
170             assert not file_info.file_handle.closed
171 
172             lines_to_publish = []
173             max_num_lines_to_read = 100
174             for _ in range(max_num_lines_to_read):
175                 next_line = file_info.file_handle.readline()
176                 if next_line == "":
177                     break
178                 if next_line[-1] == "\n":
179                     next_line = next_line[:-1]
180                 lines_to_publish.append(next_line)
181 
182             # Publish the lines if this is a worker process.
183             filename = file_info.filename.split("/")[-1]
184             is_worker = (filename.startswith("worker")
185                          and (filename.endswith("out")
186                               or filename.endswith("err")))
187 
188             if is_worker and file_info.file_position == 0:
189                 if (len(lines_to_publish) > 0 and
190                         lines_to_publish[0].startswith("Ray worker pid: ")):
191                     file_info.worker_pid = int(
192                         lines_to_publish[0].split(" ")[-1])
193                     lines_to_publish = lines_to_publish[1:]
194 
195             # Record the current position in the file.
196             file_info.file_position = file_info.file_handle.tell()
197 
198             if len(lines_to_publish) > 0 and is_worker:
199                 self.redis_client.publish(
200                     ray.gcs_utils.LOG_FILE_CHANNEL,
201                     json.dumps({
202                         "ip": self.ip,
203                         "pid": file_info.worker_pid,
204                         "lines": lines_to_publish
205                     }))
206                 anything_published = True
207 
208         return anything_published
209 
210     def run(self):
211         """Run the log monitor.
212 
213         This will query Redis once every second to check if there are new log
214         files to monitor. It will also store those log files in Redis.
215         """
216         while True:
217             self.update_log_filenames()
218             self.open_closed_files()
219             anything_published = self.check_log_files_and_publish_updates()
220             # If nothing was published, then wait a little bit before checking
221             # for logs to avoid using too much CPU.
222             if not anything_published:
223                 time.sleep(0.05)
224 
225 
226 if __name__ == "__main__":
227     parser = argparse.ArgumentParser(
228         description=("Parse Redis server for the "
229                      "log monitor to connect "
230                      "to."))
231     parser.add_argument(
232         "--redis-address",
233         required=True,
234         type=str,
235         help="The address to use for Redis.")
236     parser.add_argument(
237         "--redis-password",
238         required=False,
239         type=str,
240         default=None,
241         help="the password to use for Redis")
242     parser.add_argument(
243         "--logging-level",
244         required=False,
245         type=str,
246         default=ray_constants.LOGGER_LEVEL,
247         choices=ray_constants.LOGGER_LEVEL_CHOICES,
248         help=ray_constants.LOGGER_LEVEL_HELP)
249     parser.add_argument(
250         "--logging-format",
251         required=False,
252         type=str,
253         default=ray_constants.LOGGER_FORMAT,
254         help=ray_constants.LOGGER_FORMAT_HELP)
255     parser.add_argument(
256         "--logs-dir",
257         required=True,
258         type=str,
259         help="Specify the path of the temporary directory used by Ray "
260         "processes.")
261     args = parser.parse_args()
262     ray.utils.setup_logger(args.logging_level, args.logging_format)
263 
264     log_monitor = LogMonitor(
265         args.logs_dir, args.redis_address, redis_password=args.redis_password)
266 
267     try:
268         log_monitor.run()
269     except Exception as e:
270         # Something went wrong, so push an error to all drivers.
271         redis_client = ray.services.create_redis_client(
272             args.redis_address, password=args.redis_password)
273         traceback_str = ray.utils.format_error_message(traceback.format_exc())
274         message = ("The log monitor on node {} failed with the following "
275                    "error:\n{}".format(os.uname()[1], traceback_str))
276         ray.utils.push_error_to_driver_through_redis(
277             redis_client, ray_constants.LOG_MONITOR_DIED_ERROR, message)
278         raise e
279
```
--- 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/log_monitor.py b/python/ray/log_monitor.py
--- a/python/ray/log_monitor.py
+++ b/python/ray/log_monitor.py
@@ -4,6 +4,7 @@
 
 import argparse
 import errno
+import glob
 import json
 import logging
 import os
@@ -89,18 +90,20 @@
 
     def update_log_filenames(self):
         """Update the list of log files to monitor."""
-        log_filenames = os.listdir(self.logs_dir)
-
-        for log_filename in log_filenames:
-            full_path = os.path.join(self.logs_dir, log_filename)
-            if full_path not in self.log_filenames:
-                self.log_filenames.add(full_path)
+        # we only monior worker log files
+        log_file_paths = glob.glob("{}/worker*[.out|.err]".format(
+            self.logs_dir))
+        for file_path in log_file_paths:
+            if os.path.isfile(
+                    file_path) and file_path not in self.log_filenames:
+                self.log_filenames.add(file_path)
                 self.closed_file_infos.append(
                     LogFileInfo(
-                        filename=full_path,
+                        filename=file_path,
                         size_when_last_opened=0,
                         file_position=0,
                         file_handle=None))
+                log_filename = os.path.basename(file_path)
                 logger.info("Beginning to track file {}".format(log_filename))
 
     def open_closed_files(self):
@@ -172,20 +175,21 @@
             lines_to_publish = []
             max_num_lines_to_read = 100
             for _ in range(max_num_lines_to_read):
-                next_line = file_info.file_handle.readline()
-                if next_line == "":
-                    break
-                if next_line[-1] == "\n":
-                    next_line = next_line[:-1]
-                lines_to_publish.append(next_line)
-
-            # Publish the lines if this is a worker process.
-            filename = file_info.filename.split("/")[-1]
-            is_worker = (filename.startswith("worker")
-                         and (filename.endswith("out")
-                              or filename.endswith("err")))
-
-            if is_worker and file_info.file_position == 0:
+                try:
+                    next_line = file_info.file_handle.readline()
+                    if next_line == "":
+                        break
+                    if next_line[-1] == "\n":
+                        next_line = next_line[:-1]
+                    lines_to_publish.append(next_line)
+                except Exception:
+                    logger.error("Error: Reading file: {}, position: {} "
+                                 "failed.".format(
+                                     file_info.full_path,
+                                     file_info.file_info.file_handle.tell()))
+                    raise
+
+            if file_info.file_position == 0:
                 if (len(lines_to_publish) > 0 and
                         lines_to_publish[0].startswith("Ray worker pid: ")):
                     file_info.worker_pid = int(
@@ -195,7 +199,7 @@
             # Record the current position in the file.
             file_info.file_position = file_info.file_handle.tell()
 
-            if len(lines_to_publish) > 0 and is_worker:
+            if len(lines_to_publish) > 0:
                 self.redis_client.publish(
                     ray.gcs_utils.LOG_FILE_CHANNEL,
                     json.dumps({

{"golden_diff": "diff --git a/python/ray/log_monitor.py b/python/ray/log_monitor.py\n--- a/python/ray/log_monitor.py\n+++ b/python/ray/log_monitor.py\n@@ -4,6 +4,7 @@\n \n import argparse\n import errno\n+import glob\n import json\n import logging\n import os\n@@ -89,18 +90,20 @@\n \n     def update_log_filenames(self):\n         \"\"\"Update the list of log files to monitor.\"\"\"\n-        log_filenames = os.listdir(self.logs_dir)\n-\n-        for log_filename in log_filenames:\n-            full_path = os.path.join(self.logs_dir, log_filename)\n-            if full_path not in self.log_filenames:\n-                self.log_filenames.add(full_path)\n+        # we only monior worker log files\n+        log_file_paths = glob.glob(\"{}/worker*[.out|.err]\".format(\n+            self.logs_dir))\n+        for file_path in log_file_paths:\n+            if os.path.isfile(\n+                    file_path) and file_path not in self.log_filenames:\n+                self.log_filenames.add(file_path)\n                 self.closed_file_infos.append(\n                     LogFileInfo(\n-                        filename=full_path,\n+                        filename=file_path,\n                         size_when_last_opened=0,\n                         file_position=0,\n                         file_handle=None))\n+                log_filename = os.path.basename(file_path)\n                 logger.info(\"Beginning to track file {}\".format(log_filename))\n \n     def open_closed_files(self):\n@@ -172,20 +175,21 @@\n             lines_to_publish = []\n             max_num_lines_to_read = 100\n             for _ in range(max_num_lines_to_read):\n-                next_line = file_info.file_handle.readline()\n-                if next_line == \"\":\n-                    break\n-                if next_line[-1] == \"\\n\":\n-                    next_line = next_line[:-1]\n-                lines_to_publish.append(next_line)\n-\n-            # Publish the lines if this is a worker process.\n-            filename = file_info.filename.split(\"/\")[-1]\n-            is_worker = (filename.startswith(\"worker\")\n-                         and (filename.endswith(\"out\")\n-                              or filename.endswith(\"err\")))\n-\n-            if is_worker and file_info.file_position == 0:\n+                try:\n+                    next_line = file_info.file_handle.readline()\n+                    if next_line == \"\":\n+                        break\n+                    if next_line[-1] == \"\\n\":\n+                        next_line = next_line[:-1]\n+                    lines_to_publish.append(next_line)\n+                except Exception:\n+                    logger.error(\"Error: Reading file: {}, position: {} \"\n+                                 \"failed.\".format(\n+                                     file_info.full_path,\n+                                     file_info.file_info.file_handle.tell()))\n+                    raise\n+\n+            if file_info.file_position == 0:\n                 if (len(lines_to_publish) > 0 and\n                         lines_to_publish[0].startswith(\"Ray worker pid: \")):\n                     file_info.worker_pid = int(\n@@ -195,7 +199,7 @@\n             # Record the current position in the file.\n             file_info.file_position = file_info.file_handle.tell()\n \n-            if len(lines_to_publish) > 0 and is_worker:\n+            if len(lines_to_publish) > 0:\n                 self.redis_client.publish(\n                     ray.gcs_utils.LOG_FILE_CHANNEL,\n                     json.dumps({\n", "issue": "log monitor report UnicodeDecodeError\n<!--\r\nGeneral questions should be asked on the mailing list ray-dev@googlegroups.com.\r\nQuestions about how to use Ray should be asked on\r\n[StackOverflow](https://stackoverflow.com/questions/tagged/ray).\r\n\r\nBefore submitting an issue, please fill out the following form.\r\n-->\r\n\r\n### System information\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: centos 7\r\n- **Ray installed from (source or binary)**: binay\r\n- **Ray version**: 0.7.2 0.8\r\n- **Python version**: python 3.7\r\n- **Exact command to reproduce**:\r\n\r\n<!--\r\nYou can obtain the Ray version with\r\n\r\npython -c \"import ray; print(ray.__version__)\"\r\n-->\r\n\r\n### Describe the problem\r\n<!-- Describe the problem clearly here. -->\r\nNow the `log_monitor` monitor all the files under logs. This could be causing the following errors when we read those log file with `VIM` because of the `.*.swap` file will be created. \r\n```Traceback (most recent call last):\r\n  File \"/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py\", line 278, in <module>\r\n    raise e\r\n  File \"/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py\", line 268, in <module>\r\n    log_monitor.run()\r\n  File \"/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py\", line 219, in run\r\n    anything_published = self.check_log_files_and_publish_updates()\r\n  File \"/home/xianyang/anaconda3/lib/python3.7/site-packages/ray/log_monitor.py\", line 175, in check_log_files_and_publish_updates\r\n    next_line = file_info.file_handle.readline()\r\n  File \"/home/xianyang/anaconda3/lib/python3.7/codecs.py\", line 322, in decode\r\n    (result, consumed) = self._buffer_decode(data, self.errors, final)\r\nUnicodeDecodeError: 'utf-8' codec can't decode byte 0xca in position 21: invalid continuation byte\r\n\r\n```\r\n### Source code / logs\r\n<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport argparse\nimport errno\nimport json\nimport logging\nimport os\nimport time\nimport traceback\n\nimport ray.ray_constants as ray_constants\nimport ray.services as services\nimport ray.utils\n\n# Logger for this module. It should be configured at the entry point\n# into the program using Ray. Ray provides a default configuration at\n# entry/init points.\nlogger = logging.getLogger(__name__)\n\n\nclass LogFileInfo(object):\n    def __init__(self,\n                 filename=None,\n                 size_when_last_opened=None,\n                 file_position=None,\n                 file_handle=None):\n        assert (filename is not None and size_when_last_opened is not None\n                and file_position is not None)\n        self.filename = filename\n        self.size_when_last_opened = size_when_last_opened\n        self.file_position = file_position\n        self.file_handle = file_handle\n        self.worker_pid = None\n\n\nclass LogMonitor(object):\n    \"\"\"A monitor process for monitoring Ray log files.\n\n    This class mantains a list of open files and a list of closed log files. We\n    can't simply leave all files open because we'll run out of file\n    descriptors.\n\n    The \"run\" method of this class will cycle between doing several things:\n    1. First, it will check if any new files have appeared in the log\n       directory. If so, they will be added to the list of closed files.\n    2. Then, if we are unable to open any new files, we will close all of the\n       files.\n    3. Then, we will open as many closed files as we can that may have new\n       lines (judged by an increase in file size since the last time the file\n       was opened).\n    4. Then we will loop through the open files and see if there are any new\n       lines in the file. If so, we will publish them to Redis.\n\n    Attributes:\n        host (str): The hostname of this machine. Used to improve the log\n            messages published to Redis.\n        logs_dir (str): The directory that the log files are in.\n        redis_client: A client used to communicate with the Redis server.\n        log_filenames (set): This is the set of filenames of all files in\n            open_file_infos and closed_file_infos.\n        open_file_infos (list[LogFileInfo]): Info for all of the open files.\n        closed_file_infos (list[LogFileInfo]): Info for all of the closed\n            files.\n        can_open_more_files (bool): True if we can still open more files and\n            false otherwise.\n    \"\"\"\n\n    def __init__(self, logs_dir, redis_address, redis_password=None):\n        \"\"\"Initialize the log monitor object.\"\"\"\n        self.ip = services.get_node_ip_address()\n        self.logs_dir = logs_dir\n        self.redis_client = ray.services.create_redis_client(\n            redis_address, password=redis_password)\n        self.log_filenames = set()\n        self.open_file_infos = []\n        self.closed_file_infos = []\n        self.can_open_more_files = True\n\n    def close_all_files(self):\n        \"\"\"Close all open files (so that we can open more).\"\"\"\n        while len(self.open_file_infos) > 0:\n            file_info = self.open_file_infos.pop(0)\n            file_info.file_handle.close()\n            file_info.file_handle = None\n            self.closed_file_infos.append(file_info)\n        self.can_open_more_files = True\n\n    def update_log_filenames(self):\n        \"\"\"Update the list of log files to monitor.\"\"\"\n        log_filenames = os.listdir(self.logs_dir)\n\n        for log_filename in log_filenames:\n            full_path = os.path.join(self.logs_dir, log_filename)\n            if full_path not in self.log_filenames:\n                self.log_filenames.add(full_path)\n                self.closed_file_infos.append(\n                    LogFileInfo(\n                        filename=full_path,\n                        size_when_last_opened=0,\n                        file_position=0,\n                        file_handle=None))\n                logger.info(\"Beginning to track file {}\".format(log_filename))\n\n    def open_closed_files(self):\n        \"\"\"Open some closed files if they may have new lines.\n\n        Opening more files may require us to close some of the already open\n        files.\n        \"\"\"\n        if not self.can_open_more_files:\n            # If we can't open any more files. Close all of the files.\n            self.close_all_files()\n\n        files_with_no_updates = []\n        while len(self.closed_file_infos) > 0:\n            if (len(self.open_file_infos) >=\n                    ray_constants.LOG_MONITOR_MAX_OPEN_FILES):\n                self.can_open_more_files = False\n                break\n\n            file_info = self.closed_file_infos.pop(0)\n            assert file_info.file_handle is None\n            # Get the file size to see if it has gotten bigger since we last\n            # opened it.\n            try:\n                file_size = os.path.getsize(file_info.filename)\n            except (IOError, OSError) as e:\n                # Catch \"file not found\" errors.\n                if e.errno == errno.ENOENT:\n                    logger.warning(\"Warning: The file {} was not \"\n                                   \"found.\".format(file_info.filename))\n                    self.log_filenames.remove(file_info.filename)\n                    continue\n                raise e\n\n            # If some new lines have been added to this file, try to reopen the\n            # file.\n            if file_size > file_info.size_when_last_opened:\n                try:\n                    f = open(file_info.filename, \"r\")\n                except (IOError, OSError) as e:\n                    if e.errno == errno.ENOENT:\n                        logger.warning(\"Warning: The file {} was not \"\n                                       \"found.\".format(file_info.filename))\n                        self.log_filenames.remove(file_info.filename)\n                        continue\n                    else:\n                        raise e\n\n                f.seek(file_info.file_position)\n                file_info.filesize_when_last_opened = file_size\n                file_info.file_handle = f\n                self.open_file_infos.append(file_info)\n            else:\n                files_with_no_updates.append(file_info)\n\n        # Add the files with no changes back to the list of closed files.\n        self.closed_file_infos += files_with_no_updates\n\n    def check_log_files_and_publish_updates(self):\n        \"\"\"Get any changes to the log files and push updates to Redis.\n\n        Returns:\n            True if anything was published and false otherwise.\n        \"\"\"\n        anything_published = False\n        for file_info in self.open_file_infos:\n            assert not file_info.file_handle.closed\n\n            lines_to_publish = []\n            max_num_lines_to_read = 100\n            for _ in range(max_num_lines_to_read):\n                next_line = file_info.file_handle.readline()\n                if next_line == \"\":\n                    break\n                if next_line[-1] == \"\\n\":\n                    next_line = next_line[:-1]\n                lines_to_publish.append(next_line)\n\n            # Publish the lines if this is a worker process.\n            filename = file_info.filename.split(\"/\")[-1]\n            is_worker = (filename.startswith(\"worker\")\n                         and (filename.endswith(\"out\")\n                              or filename.endswith(\"err\")))\n\n            if is_worker and file_info.file_position == 0:\n                if (len(lines_to_publish) > 0 and\n                        lines_to_publish[0].startswith(\"Ray worker pid: \")):\n                    file_info.worker_pid = int(\n                        lines_to_publish[0].split(\" \")[-1])\n                    lines_to_publish = lines_to_publish[1:]\n\n            # Record the current position in the file.\n            file_info.file_position = file_info.file_handle.tell()\n\n            if len(lines_to_publish) > 0 and is_worker:\n                self.redis_client.publish(\n                    ray.gcs_utils.LOG_FILE_CHANNEL,\n                    json.dumps({\n                        \"ip\": self.ip,\n                        \"pid\": file_info.worker_pid,\n                        \"lines\": lines_to_publish\n                    }))\n                anything_published = True\n\n        return anything_published\n\n    def run(self):\n        \"\"\"Run the log monitor.\n\n        This will query Redis once every second to check if there are new log\n        files to monitor. It will also store those log files in Redis.\n        \"\"\"\n        while True:\n            self.update_log_filenames()\n            self.open_closed_files()\n            anything_published = self.check_log_files_and_publish_updates()\n            # If nothing was published, then wait a little bit before checking\n            # for logs to avoid using too much CPU.\n            if not anything_published:\n                time.sleep(0.05)\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(\n        description=(\"Parse Redis server for the \"\n                     \"log monitor to connect \"\n                     \"to.\"))\n    parser.add_argument(\n        \"--redis-address\",\n        required=True,\n        type=str,\n        help=\"The address to use for Redis.\")\n    parser.add_argument(\n        \"--redis-password\",\n        required=False,\n        type=str,\n        default=None,\n        help=\"the password to use for Redis\")\n    parser.add_argument(\n        \"--logging-level\",\n        required=False,\n        type=str,\n        default=ray_constants.LOGGER_LEVEL,\n        choices=ray_constants.LOGGER_LEVEL_CHOICES,\n        help=ray_constants.LOGGER_LEVEL_HELP)\n    parser.add_argument(\n        \"--logging-format\",\n        required=False,\n        type=str,\n        default=ray_constants.LOGGER_FORMAT,\n        help=ray_constants.LOGGER_FORMAT_HELP)\n    parser.add_argument(\n        \"--logs-dir\",\n        required=True,\n        type=str,\n        help=\"Specify the path of the temporary directory used by Ray \"\n        \"processes.\")\n    args = parser.parse_args()\n    ray.utils.setup_logger(args.logging_level, args.logging_format)\n\n    log_monitor = LogMonitor(\n        args.logs_dir, args.redis_address, redis_password=args.redis_password)\n\n    try:\n        log_monitor.run()\n    except Exception as e:\n        # Something went wrong, so push an error to all drivers.\n        redis_client = ray.services.create_redis_client(\n            args.redis_address, password=args.redis_password)\n        traceback_str = ray.utils.format_error_message(traceback.format_exc())\n        message = (\"The log monitor on node {} failed with the following \"\n                   \"error:\\n{}\".format(os.uname()[1], traceback_str))\n        ray.utils.push_error_to_driver_through_redis(\n            redis_client, ray_constants.LOG_MONITOR_DIED_ERROR, message)\n        raise e\n", "path": "python/ray/log_monitor.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport argparse\nimport errno\nimport glob\nimport json\nimport logging\nimport os\nimport time\nimport traceback\n\nimport ray.ray_constants as ray_constants\nimport ray.services as services\nimport ray.utils\n\n# Logger for this module. It should be configured at the entry point\n# into the program using Ray. Ray provides a default configuration at\n# entry/init points.\nlogger = logging.getLogger(__name__)\n\n\nclass LogFileInfo(object):\n    def __init__(self,\n                 filename=None,\n                 size_when_last_opened=None,\n                 file_position=None,\n                 file_handle=None):\n        assert (filename is not None and size_when_last_opened is not None\n                and file_position is not None)\n        self.filename = filename\n        self.size_when_last_opened = size_when_last_opened\n        self.file_position = file_position\n        self.file_handle = file_handle\n        self.worker_pid = None\n\n\nclass LogMonitor(object):\n    \"\"\"A monitor process for monitoring Ray log files.\n\n    This class mantains a list of open files and a list of closed log files. We\n    can't simply leave all files open because we'll run out of file\n    descriptors.\n\n    The \"run\" method of this class will cycle between doing several things:\n    1. First, it will check if any new files have appeared in the log\n       directory. If so, they will be added to the list of closed files.\n    2. Then, if we are unable to open any new files, we will close all of the\n       files.\n    3. Then, we will open as many closed files as we can that may have new\n       lines (judged by an increase in file size since the last time the file\n       was opened).\n    4. Then we will loop through the open files and see if there are any new\n       lines in the file. If so, we will publish them to Redis.\n\n    Attributes:\n        host (str): The hostname of this machine. Used to improve the log\n            messages published to Redis.\n        logs_dir (str): The directory that the log files are in.\n        redis_client: A client used to communicate with the Redis server.\n        log_filenames (set): This is the set of filenames of all files in\n            open_file_infos and closed_file_infos.\n        open_file_infos (list[LogFileInfo]): Info for all of the open files.\n        closed_file_infos (list[LogFileInfo]): Info for all of the closed\n            files.\n        can_open_more_files (bool): True if we can still open more files and\n            false otherwise.\n    \"\"\"\n\n    def __init__(self, logs_dir, redis_address, redis_password=None):\n        \"\"\"Initialize the log monitor object.\"\"\"\n        self.ip = services.get_node_ip_address()\n        self.logs_dir = logs_dir\n        self.redis_client = ray.services.create_redis_client(\n            redis_address, password=redis_password)\n        self.log_filenames = set()\n        self.open_file_infos = []\n        self.closed_file_infos = []\n        self.can_open_more_files = True\n\n    def close_all_files(self):\n        \"\"\"Close all open files (so that we can open more).\"\"\"\n        while len(self.open_file_infos) > 0:\n            file_info = self.open_file_infos.pop(0)\n            file_info.file_handle.close()\n            file_info.file_handle = None\n            self.closed_file_infos.append(file_info)\n        self.can_open_more_files = True\n\n    def update_log_filenames(self):\n        \"\"\"Update the list of log files to monitor.\"\"\"\n        # we only monior worker log files\n        log_file_paths = glob.glob(\"{}/worker*[.out|.err]\".format(\n            self.logs_dir))\n        for file_path in log_file_paths:\n            if os.path.isfile(\n                    file_path) and file_path not in self.log_filenames:\n                self.log_filenames.add(file_path)\n                self.closed_file_infos.append(\n                    LogFileInfo(\n                        filename=file_path,\n                        size_when_last_opened=0,\n                        file_position=0,\n                        file_handle=None))\n                log_filename = os.path.basename(file_path)\n                logger.info(\"Beginning to track file {}\".format(log_filename))\n\n    def open_closed_files(self):\n        \"\"\"Open some closed files if they may have new lines.\n\n        Opening more files may require us to close some of the already open\n        files.\n        \"\"\"\n        if not self.can_open_more_files:\n            # If we can't open any more files. Close all of the files.\n            self.close_all_files()\n\n        files_with_no_updates = []\n        while len(self.closed_file_infos) > 0:\n            if (len(self.open_file_infos) >=\n                    ray_constants.LOG_MONITOR_MAX_OPEN_FILES):\n                self.can_open_more_files = False\n                break\n\n            file_info = self.closed_file_infos.pop(0)\n            assert file_info.file_handle is None\n            # Get the file size to see if it has gotten bigger since we last\n            # opened it.\n            try:\n                file_size = os.path.getsize(file_info.filename)\n            except (IOError, OSError) as e:\n                # Catch \"file not found\" errors.\n                if e.errno == errno.ENOENT:\n                    logger.warning(\"Warning: The file {} was not \"\n                                   \"found.\".format(file_info.filename))\n                    self.log_filenames.remove(file_info.filename)\n                    continue\n                raise e\n\n            # If some new lines have been added to this file, try to reopen the\n            # file.\n            if file_size > file_info.size_when_last_opened:\n                try:\n                    f = open(file_info.filename, \"r\")\n                except (IOError, OSError) as e:\n                    if e.errno == errno.ENOENT:\n                        logger.warning(\"Warning: The file {} was not \"\n                                       \"found.\".format(file_info.filename))\n                        self.log_filenames.remove(file_info.filename)\n                        continue\n                    else:\n                        raise e\n\n                f.seek(file_info.file_position)\n                file_info.filesize_when_last_opened = file_size\n                file_info.file_handle = f\n                self.open_file_infos.append(file_info)\n            else:\n                files_with_no_updates.append(file_info)\n\n        # Add the files with no changes back to the list of closed files.\n        self.closed_file_infos += files_with_no_updates\n\n    def check_log_files_and_publish_updates(self):\n        \"\"\"Get any changes to the log files and push updates to Redis.\n\n        Returns:\n            True if anything was published and false otherwise.\n        \"\"\"\n        anything_published = False\n        for file_info in self.open_file_infos:\n            assert not file_info.file_handle.closed\n\n            lines_to_publish = []\n            max_num_lines_to_read = 100\n            for _ in range(max_num_lines_to_read):\n                try:\n                    next_line = file_info.file_handle.readline()\n                    if next_line == \"\":\n                        break\n                    if next_line[-1] == \"\\n\":\n                        next_line = next_line[:-1]\n                    lines_to_publish.append(next_line)\n                except Exception:\n                    logger.error(\"Error: Reading file: {}, position: {} \"\n                                 \"failed.\".format(\n                                     file_info.full_path,\n                                     file_info.file_info.file_handle.tell()))\n                    raise\n\n            if file_info.file_position == 0:\n                if (len(lines_to_publish) > 0 and\n                        lines_to_publish[0].startswith(\"Ray worker pid: \")):\n                    file_info.worker_pid = int(\n                        lines_to_publish[0].split(\" \")[-1])\n                    lines_to_publish = lines_to_publish[1:]\n\n            # Record the current position in the file.\n            file_info.file_position = file_info.file_handle.tell()\n\n            if len(lines_to_publish) > 0:\n                self.redis_client.publish(\n                    ray.gcs_utils.LOG_FILE_CHANNEL,\n                    json.dumps({\n                        \"ip\": self.ip,\n                        \"pid\": file_info.worker_pid,\n                        \"lines\": lines_to_publish\n                    }))\n                anything_published = True\n\n        return anything_published\n\n    def run(self):\n        \"\"\"Run the log monitor.\n\n        This will query Redis once every second to check if there are new log\n        files to monitor. It will also store those log files in Redis.\n        \"\"\"\n        while True:\n            self.update_log_filenames()\n            self.open_closed_files()\n            anything_published = self.check_log_files_and_publish_updates()\n            # If nothing was published, then wait a little bit before checking\n            # for logs to avoid using too much CPU.\n            if not anything_published:\n                time.sleep(0.05)\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(\n        description=(\"Parse Redis server for the \"\n                     \"log monitor to connect \"\n                     \"to.\"))\n    parser.add_argument(\n        \"--redis-address\",\n        required=True,\n        type=str,\n        help=\"The address to use for Redis.\")\n    parser.add_argument(\n        \"--redis-password\",\n        required=False,\n        type=str,\n        default=None,\n        help=\"the password to use for Redis\")\n    parser.add_argument(\n        \"--logging-level\",\n        required=False,\n        type=str,\n        default=ray_constants.LOGGER_LEVEL,\n        choices=ray_constants.LOGGER_LEVEL_CHOICES,\n        help=ray_constants.LOGGER_LEVEL_HELP)\n    parser.add_argument(\n        \"--logging-format\",\n        required=False,\n        type=str,\n        default=ray_constants.LOGGER_FORMAT,\n        help=ray_constants.LOGGER_FORMAT_HELP)\n    parser.add_argument(\n        \"--logs-dir\",\n        required=True,\n        type=str,\n        help=\"Specify the path of the temporary directory used by Ray \"\n        \"processes.\")\n    args = parser.parse_args()\n    ray.utils.setup_logger(args.logging_level, args.logging_format)\n\n    log_monitor = LogMonitor(\n        args.logs_dir, args.redis_address, redis_password=args.redis_password)\n\n    try:\n        log_monitor.run()\n    except Exception as e:\n        # Something went wrong, so push an error to all drivers.\n        redis_client = ray.services.create_redis_client(\n            args.redis_address, password=args.redis_password)\n        traceback_str = ray.utils.format_error_message(traceback.format_exc())\n        message = (\"The log monitor on node {} failed with the following \"\n                   \"error:\\n{}\".format(os.uname()[1], traceback_str))\n        ray.utils.push_error_to_driver_through_redis(\n            redis_client, ray_constants.LOG_MONITOR_DIED_ERROR, message)\n        raise e\n", "path": "python/ray/log_monitor.py"}]}