Drop stdout and stderr from message status

These may be large and are transferred as files anyway.

Author: mvdbeek

pulsar/client/test/check.py

@@ -304,16 +304,8 @@ def run(options):
             client.job_id = job_id
         result_status = waiter.wait()
 
-        expecting_full_metadata = getattr(options, "expecting_full_metadata", True)
-        if expecting_full_metadata:
-            stdout = result_status["stdout"].strip()
-            std_streams_debug = f"actual stdout [{stdout}], actual stderr [{result_status['stderr']}]"
-            assert "stdout output".startswith(stdout), f"Standard output is not an initial substring of [stdout output], {std_streams_debug}"
-
-            if hasattr(options, "maximum_stream_size"):
-                assert len(stdout) == options.maximum_stream_size
-
         assert result_status["complete"] == "true"
+        expecting_full_metadata = getattr(options, "expecting_full_metadata", True)
         if expecting_full_metadata:
             assert result_status["returncode"] == 4, f"Expected exit code of 4, got {result_status['returncode']} - {std_streams_debug}"
         if expecting_full_metadata:

pulsar/manager_endpoint_util.py

@@ -38,8 +38,6 @@ def __job_complete_dict(complete_status, manager, job_id):
     return_code = manager.return_code(job_id)
     if return_code == PULSAR_UNKNOWN_RETURN_CODE:
         return_code = None
-    stdout_contents = unicodify(manager.stdout_contents(job_id))
-    stderr_contents = unicodify(manager.stderr_contents(job_id))
     job_stdout_contents = unicodify(manager.job_stdout_contents(job_id).decode("utf-8"))
     job_stderr_contents = unicodify(manager.job_stderr_contents(job_id).decode("utf-8"))
     job_directory = manager.job_directory(job_id)
@@ -48,8 +46,6 @@ def __job_complete_dict(complete_status, manager, job_id):
         complete="true",  # Is this still used or is it legacy.
         status=complete_status,
         returncode=return_code,
-        stdout=stdout_contents,
-        stderr=stderr_contents,
         job_stdout=job_stdout_contents,
         job_stderr=job_stderr_contents,
         working_directory=job_directory.working_directory(),

pulsar/managers/__init__.py

@@ -50,18 +50,6 @@ def return_code(self, job_id):
         PULSAR_UNKNOWN_RETURN_CODE.
         """
 
-    @abstractmethod
-    def stdout_contents(self, job_id):
-        """
-        After completion, return contents of stdout of the tool script.
-        """
-
-    @abstractmethod
-    def stderr_contents(self, job_id):
-        """
-        After completion, return contents of stderr of the tool script.
-        """
-
     @abstractmethod
     def job_stdout_contents(self, job_id):
         """
@@ -111,12 +99,6 @@ def get_status(self, *args, **kwargs):
     def return_code(self, *args, **kwargs):
         return self._proxied_manager.return_code(*args, **kwargs)
 
-    def stdout_contents(self, *args, **kwargs):
-        return self._proxied_manager.stdout_contents(*args, **kwargs)
-
-    def stderr_contents(self, *args, **kwargs):
-        return self._proxied_manager.stderr_contents(*args, **kwargs)
-
     def job_stdout_contents(self, *args, **kwargs):
         return self._proxied_manager.job_stdout_contents(*args, **kwargs)
 

pulsar/managers/base/directory.py

@@ -15,8 +15,6 @@
 # should be able to replace metadata backing with non-file stuff now that
 # the abstractions are fairly well utilized.
 JOB_FILE_RETURN_CODE = "return_code"
-TOOL_FILE_STANDARD_OUTPUT = os.path.join("metadata", "tool_stdout")
-TOOL_FILE_STANDARD_ERROR = os.path.join("metadata", "tool_stderr")
 JOB_FILE_STANDARD_OUTPUT = os.path.join("metadata", "job_stdout")
 JOB_FILE_STANDARD_ERROR = os.path.join("metadata", "job_stderr")
 JOB_FILE_TOOL_ID = "tool_id"
@@ -43,20 +41,6 @@ def return_code(self, job_id):
         return_code_str = self._read_job_file(job_id, JOB_FILE_RETURN_CODE, default=PULSAR_UNKNOWN_RETURN_CODE)
         return int(return_code_str) if return_code_str and return_code_str != PULSAR_UNKNOWN_RETURN_CODE else return_code_str
 
-    def stdout_contents(self, job_id):
-        try:
-            return self._read_job_file(job_id, TOOL_FILE_STANDARD_OUTPUT, size=self.maximum_stream_size)
-        except FileNotFoundError:
-            # Could be old job finishing up, drop in 2024?
-            return self._read_job_file(job_id, "tool_stdout", size=self.maximum_stream_size, default=b"")
-
-    def stderr_contents(self, job_id):
-        try:
-            return self._read_job_file(job_id, TOOL_FILE_STANDARD_ERROR, size=self.maximum_stream_size)
-        except FileNotFoundError:
-            # Could be old job finishing up, drop in 2024?
-            return self._read_job_file(job_id, "tool_stderr", size=self.maximum_stream_size, default=b"")
-
     def job_stdout_contents(self, job_id):
         return self._read_job_file(job_id, JOB_FILE_STANDARD_OUTPUT, size=self.maximum_stream_size, default=b"")
 
@@ -71,12 +55,6 @@ def read_command_line(self, job_id):
             command_line = json.loads(command_line)
         return command_line
 
-    def _tool_stdout_path(self, job_id):
-        return self._job_file(job_id, TOOL_FILE_STANDARD_OUTPUT)
-
-    def _tool_stderr_path(self, job_id):
-        return self._job_file(job_id, TOOL_FILE_STANDARD_ERROR)
-
     def _job_stdout_path(self, job_id):
         return self._job_file(job_id, JOB_FILE_STANDARD_OUTPUT)
 

test/test_utils.py

@@ -184,8 +184,6 @@ def _test_simple_execution(self, manager, timeout=None):
                 raise Exception("Timeout.")
         self.assertEqual(manager.job_stderr_contents(job_id), b"")
         self.assertEqual(manager.job_stdout_contents(job_id), b"")
-        self.assertEqual(manager.stderr_contents(job_id), b"moo")
-        self.assertEqual(manager.stdout_contents(job_id), b"Hello World!")
         self.assertEqual(manager.return_code(job_id), 0)
         manager.clean(job_id)
         self.assertEqual(len(listdir(self.staging_directory)), 0)