fix!: Terminating all VFS processes when cleaning up session

src/deadline_worker_agent/sessions/session.py

@@ -52,6 +52,8 @@
 
 from deadline.job_attachments.asset_sync import AssetSync
 from deadline.job_attachments.asset_sync import logger as ASSET_SYNC_LOGGER
+from deadline.job_attachments.exceptions import Fus3ExecutableMissingError
+from deadline.job_attachments.fus3 import Fus3ProcessManager
 from deadline.job_attachments.models import (
     Attachments,
     JobAttachmentS3Settings,
@@ -373,6 +375,12 @@ def _cleanup(self) -> None:
                     logger.info("%s successful", desc)
                 cur_time = monotonic()
         finally:
+            try:
+                Fus3ProcessManager.find_fus3()
+                # Shutdown all running Deadline VFS processes since session is complete
+                Fus3ProcessManager.kill_all_processes(session_dir=self._session.working_directory)
+            except Fus3ExecutableMissingError:
+                logger.error("Virtual File System not found, no processes to kill.")
             # Clean-up the Open Job Description session
             self._session.cleanup()
 

test/unit/sessions/test_session.py

@@ -106,6 +106,13 @@ def mock_openjd_session(mock_openjd_session_cls: MagicMock) -> MagicMock:
     return mock_openjd_session_cls.return_value
 
 
+@pytest.fixture
+def mock_fus3_process_manager_cls() -> Generator[MagicMock, None, None]:
+    """Mocks the Worker Agent Session module's import of the Fus3 Process Manager class"""
+    with patch.object(session_mod, "Fus3ProcessManager") as mock_fus3_process_manager:
+        yield mock_fus3_process_manager
+
+
 @pytest.fixture
 def action_update_callback() -> MagicMock:
     """MagicMock action as the action update callback"""
@@ -1662,6 +1669,20 @@ def test_calls_openjd_cleanup(
         # THEN
         openjd_session_cleanup.assert_called_once_with()
 
+    def test_calls_fus3_kill(
+        self,
+        session: Session,
+        mock_fus3_process_manager_cls,
+    ) -> None:
+        # Mock Session._monitor_action which is used to poll the Open Job Description session status
+        with patch.object(session, "_monitor_action", return_value=[]):
+            # WHEN
+            session._cleanup()
+
+        # THEN
+        mock_fus3_process_manager_cls.find_fus3.assert_called_once()
+        mock_fus3_process_manager_cls.kill_all_processes.assert_called()
+
 
 class TestSessionStartAction:
     """Tests for Session._start_action()"""