local_scheduler: ignore bad utf-8 data instead of crashing (#718)

Author: d4l3k

torchx/schedulers/local_scheduler.py

@@ -438,7 +438,7 @@ def _fmt_io_filename(std_io: Optional[BinaryIO]) -> str:
 
     def __repr__(self) -> str:
         role_to_pid = {}
-        for (role_name, replicas) in self.role_replicas.items():
+        for role_name, replicas in self.role_replicas.items():
             pids = role_to_pid.setdefault(role_name, [])
             for r in replicas:
                 pids.append(r.proc.pid)
@@ -612,7 +612,7 @@ def _evict_lru(self) -> bool:
         """
         lru_time = sys.maxsize
         lru_app_id = None
-        for (app_id, app) in self._apps.items():
+        for app_id, app in self._apps.items():
             if is_terminal(app.state):
                 if app.last_updated <= lru_time:
                     lru_app_id = app_id
@@ -988,7 +988,7 @@ def _cancel_existing(self, app_id: str) -> None:
 
     def close(self) -> None:
         # terminate all apps
-        for (app_id, app) in self._apps.items():
+        for app_id, app in self._apps.items():
             log.debug(f"Terminating app: {app_id}")
             app.kill()
         # delete logdir if torchx created a log dir
@@ -1037,7 +1037,12 @@ def __iter__(self) -> "LogIterator":
             self._check_finished()  # check to see if app has finished running
 
             if os.path.isfile(self._log_file):
-                self._log_fp = open(self._log_file, "rt", newline="\n")  # noqa: P201
+                self._log_fp = open(
+                    self._log_file,
+                    mode="rt",
+                    newline="\n",
+                    errors="replace",  # replace bad utf-8 with \uFFFD
+                )  # noqa: P201
                 break
 
             if self._app_finished:

torchx/schedulers/test/local_scheduler_test.py

@@ -223,6 +223,9 @@ def setUp(self) -> None:
             session_name="test_session",
             image_provider_class=LocalDirectoryImageProvider,
         )
+        write_shell_script(
+            self.test_dir, "print_invalid_utf8.sh", ["printf '\\x02\\xc5\\xd8'"]
+        )
 
     def tearDown(self) -> None:
         self.scheduler.close()
@@ -672,6 +675,20 @@ def test_log_iterator_no_log_dir(self) -> None:
         logs = list(self.scheduler.log_iter(app_id, "role1", k=0))
         self.assertEqual(len(logs), 11)
 
+    def test_log_iterator_invalid_utf8(self) -> None:
+        role = Role(
+            "role1",
+            image=self.test_dir,
+            entrypoint="print_invalid_utf8.sh",
+            num_replicas=1,
+        )
+
+        app = AppDef(name="test_app", roles=[role])
+
+        app_id = self.scheduler.submit(app, cfg={})
+        logs = list(self.scheduler.log_iter(app_id, "role1", k=0))
+        self.assertEqual(logs, ["\x02\ufffd\ufffd"])
+
     def test_submit_multiple_roles(self) -> None:
         test_file1 = join(self.test_dir, "test_file_1")
         test_file2 = join(self.test_dir, "test_file_2")