[Plugin] Allow writing command output directly to disk

This commit addresses a long standing ask in sos, regarding resource
consumption when sos is run with `--all-logs`.

For executions where `--all-logs` is used, or for specific commands
where `sizelimit` has been set to 0, `add_cmd_output()` and
`add_journal()` will now instruct `sos_get_command_output()` to write
output directly to a file rather than saving the output in memory.

When this occurs, the `output` key in the returned dict will be empty.

Note that this does extend to `collect_cmd_output()` or `exec_cmd()`.

Resolves: #1506

Signed-off-by: Jake Hunsaker <jhunsake@redhat.com>

4 files changed, 123 insertions, 40 deletions

diff --git a/sos/archive.py b/sos/archive.py
index e3c68b77..e92d5d60 100644
--- a/sos/archive.py
+++ b/sos/archive.py
@@ -251,7 +251,7 @@ class FileCacheArchive(Archive):
 
         return dest
 
-    def _check_path(self, src, path_type, dest=None, force=False):
+    def check_path(self, src, path_type, dest=None, force=False):
         """Check a new destination path in the archive.
 
             Since it is possible for multiple plugins to collect the same
@@ -345,7 +345,7 @@ class FileCacheArchive(Archive):
             if not dest:
                 dest = src
 
-            dest = self._check_path(dest, P_FILE)
+            dest = self.check_path(dest, P_FILE)
             if not dest:
                 return
 
@@ -384,7 +384,7 @@ class FileCacheArchive(Archive):
             # over any exixting content in the archive, since it is used by
             # the Plugin postprocessing hooks to perform regex substitution
             # on file content.
-            dest = self._check_path(dest, P_FILE, force=True)
+            dest = self.check_path(dest, P_FILE, force=True)
 
             f = codecs.open(dest, mode, encoding='utf-8')
             if isinstance(content, bytes):
@@ -397,7 +397,7 @@ class FileCacheArchive(Archive):
 
     def add_binary(self, content, dest):
         with self._path_lock:
-            dest = self._check_path(dest, P_FILE)
+            dest = self.check_path(dest, P_FILE)
             if not dest:
                 return
 
@@ -409,7 +409,7 @@ class FileCacheArchive(Archive):
     def add_link(self, source, link_name):
         self.log_debug("adding symlink at '%s' -> '%s'" % (link_name, source))
         with self._path_lock:
-            dest = self._check_path(link_name, P_LINK)
+            dest = self.check_path(link_name, P_LINK)
             if not dest:
                 return
 
@@ -484,10 +484,10 @@ class FileCacheArchive(Archive):
         """
         # Establish path structure
         with self._path_lock:
-            self._check_path(path, P_DIR)
+            self.check_path(path, P_DIR)
 
     def add_node(self, path, mode, device):
-        dest = self._check_path(path, P_NODE)
+        dest = self.check_path(path, P_NODE)
         if not dest:
             return
 
diff --git a/sos/report/__init__.py b/sos/report/__init__.py
index 249ff119..46c7f219 100644
--- a/sos/report/__init__.py
+++ b/sos/report/__init__.py
@@ -476,7 +476,8 @@ class SoSReport(SoSComponent):
             'verbosity': self.opts.verbosity,
             'cmdlineopts': self.opts,
             'devices': self.devices,
-            'namespaces': self.namespaces
+            'namespaces': self.namespaces,
+            'tempfile_util': self.tempfile_util
         }
 
     def get_temp_file(self):
@@ -1075,7 +1076,12 @@ class SoSReport(SoSComponent):
                 _plug.manifest.add_field('end_time', end)
                 _plug.manifest.add_field('run_time', end - start)
             except TimeoutError:
-                self.ui_log.error("\n Plugin %s timed out\n" % plugin[1])
+                msg = "Plugin %s timed out" % plugin[1]
+                # log to ui_log.error to show the user, log to soslog.info
+                # so that someone investigating the sos execution has it all
+                # in one place, but without double notifying the user.
+                self.ui_log.error("\n %s\n" % msg)
+                self.soslog.info(msg)
                 self.running_plugs.remove(plugin[1])
                 self.loaded_plugins[plugin[0]-1][1].set_timeout_hit()
                 pool.shutdown(wait=True)
diff --git a/sos/report/plugins/__init__.py b/sos/report/plugins/__init__.py
index 3ff7c191..38529a9d 100644
--- a/sos/report/plugins/__init__.py
+++ b/sos/report/plugins/__init__.py
@@ -15,6 +15,7 @@ from sos.utilities import (sos_get_command_output, import_module, grep,
                            path_exists, path_isdir, path_isfile, path_islink,
                            listdir, path_join)
 
+from sos.archive import P_FILE
 import os
 import glob
 import re
@@ -1686,6 +1687,8 @@ class Plugin():
             kwargs['priority'] = 10
         if 'changes' not in kwargs:
             kwargs['changes'] = False
+        if self.get_option('all_logs') or kwargs['sizelimit'] == 0:
+            kwargs['to_file'] = True
         soscmd = SoSCommand(**kwargs)
         self._log_debug("packed command: " + soscmd.__str__())
         for _skip_cmd in self.skip_commands:
@@ -1707,7 +1710,8 @@ class Plugin():
                        chroot=True, runat=None, env=None, binary=False,
                        sizelimit=None, pred=None, subdir=None,
                        changes=False, foreground=False, tags=[],
-                       priority=10, cmd_as_tag=False, container=None):
+                       priority=10, cmd_as_tag=False, container=None,
+                       to_file=False):
         """Run a program or a list of programs and collect the output
 
         Output will be limited to `sizelimit`, collecting the last X amount
@@ -1776,6 +1780,10 @@ class Plugin():
         :param container: Run the specified `cmds` inside a container with this
                           ID or name
         :type container:  ``str``
+
+        :param to_file: Should command output be written directly to a new
+                        file rather than stored in memory?
+        :type to_file:  ``bool``
         """
         if isinstance(cmds, str):
             cmds = [cmds]
@@ -1800,7 +1808,8 @@ class Plugin():
                                  env=env, binary=binary, sizelimit=sizelimit,
                                  pred=pred, subdir=subdir, tags=tags,
                                  changes=changes, foreground=foreground,
-                                 priority=priority, cmd_as_tag=cmd_as_tag)
+                                 priority=priority, cmd_as_tag=cmd_as_tag,
+                                 to_file=to_file)
 
     def add_cmd_tags(self, tagdict):
         """Retroactively add tags to any commands that have been run by this
@@ -1966,7 +1975,7 @@ class Plugin():
                             stderr=True, chroot=True, runat=None, env=None,
                             binary=False, sizelimit=None, subdir=None,
                             changes=False, foreground=False, tags=[],
-                            priority=10, cmd_as_tag=False):
+                            priority=10, cmd_as_tag=False, to_file=False):
         """Execute a command and save the output to a file for inclusion in the
         report.
 
@@ -1990,6 +1999,8 @@ class Plugin():
                                         on the system?
             :param tags:                Add tags in the archive manifest
             :param cmd_as_tag:          Format command string to tag
+            :param to_file:             Write output directly to file instead
+                                        of saving in memory
 
         :returns:       dict containing status, output, and filename in the
                         archive for the executed cmd
@@ -2019,27 +2030,46 @@ class Plugin():
         else:
             root = None
 
+        if suggest_filename:
+            outfn = self._make_command_filename(suggest_filename, subdir)
+        else:
+            outfn = self._make_command_filename(cmd, subdir)
+
+        outfn_strip = outfn[len(self.commons['cmddir'])+1:]
+
+        if to_file:
+            self._log_debug("collecting '%s' output directly to disk"
+                            % cmd)
+            self.archive.check_path(outfn, P_FILE)
+            out_file = os.path.join(self.archive.get_archive_path(), outfn)
+        else:
+            out_file = False
+
         start = time()
 
         result = sos_get_command_output(
             cmd, timeout=timeout, stderr=stderr, chroot=root,
             chdir=runat, env=env, binary=binary, sizelimit=sizelimit,
-            poller=self.check_timeout, foreground=foreground
+            poller=self.check_timeout, foreground=foreground,
+            to_file=out_file
         )
 
         end = time()
         run_time = end - start
 
         if result['status'] == 124:
-            self._log_warn(
-                "command '%s' timed out after %ds" % (cmd, timeout)
-            )
+            warn = "command '%s' timed out after %ds" % (cmd, timeout)
+            self._log_warn(warn)
+            if to_file:
+                msg = (" - output up until the timeout may be available at "
+                       "%s" % outfn)
+                self._log_debug("%s%s" % (warn, msg))
 
         manifest_cmd = {
             'command': cmd.split(' ')[0],
             'parameters': cmd.split(' ')[1:],
             'exec': cmd,
-            'filepath': None,
+            'filepath': outfn if to_file else None,
             'truncated': result['truncated'],
             'return_code': result['status'],
             'priority': priority,
@@ -2060,7 +2090,7 @@ class Plugin():
                     result = sos_get_command_output(
                         cmd, timeout=timeout, chroot=False, chdir=runat,
                         env=env, binary=binary, sizelimit=sizelimit,
-                        poller=self.check_timeout
+                        poller=self.check_timeout, to_file=out_file
                     )
                     run_time = time() - start
             self._log_debug("could not run '%s': command not found" % cmd)
@@ -2077,22 +2107,15 @@ class Plugin():
         if result['truncated']:
             self._log_info("collected output of '%s' was truncated"
                            % cmd.split()[0])
-
-        if suggest_filename:
-            outfn = self._make_command_filename(suggest_filename, subdir)
-        else:
-            outfn = self._make_command_filename(cmd, subdir)
-
-        outfn_strip = outfn[len(self.commons['cmddir'])+1:]
-
-        if result['truncated']:
             linkfn = outfn
             outfn = outfn.replace('sos_commands', 'sos_strings') + '.tailed'
 
-        if binary:
-            self.archive.add_binary(result['output'], outfn)
-        else:
-            self.archive.add_string(result['output'], outfn)
+        if not to_file:
+            if binary:
+                self.archive.add_binary(result['output'], outfn)
+            else:
+                self.archive.add_string(result['output'], outfn)
+
         if result['truncated']:
             # we need to manually build the relative path from the paths that
             # exist within the build dir to properly drop these symlinks
@@ -2543,6 +2566,9 @@ class Plugin():
         all_logs = self.get_option("all_logs")
         log_size = sizelimit or self.get_option("log_size")
         log_size = max(log_size, journal_size) if not all_logs else 0
+        if sizelimit == 0:
+            # allow for specific sizelimit overrides in plugins
+            log_size = 0
 
         if isinstance(units, str):
             units = [units]
diff --git a/sos/utilities.py b/sos/utilities.py
index 70d5c0ab..f422d7a0 100644
--- a/sos/utilities.py
+++ b/sos/utilities.py
@@ -110,7 +110,8 @@ def is_executable(command, sysroot=None):
 
 def sos_get_command_output(command, timeout=TIMEOUT_DEFAULT, stderr=False,
                            chroot=None, chdir=None, env=None, foreground=False,
-                           binary=False, sizelimit=None, poller=None):
+                           binary=False, sizelimit=None, poller=None,
+                           to_file=False):
     """Execute a command and return a dictionary of status and output,
     optionally changing root or current working directory before
     executing command.
@@ -124,6 +125,12 @@ def sos_get_command_output(command, timeout=TIMEOUT_DEFAULT, stderr=False,
         if (chdir):
             os.chdir(chdir)
 
+    def _check_poller(proc):
+        if poller():
+            proc.terminate()
+            raise SoSTimeoutError
+        time.sleep(0.01)
+
     cmd_env = os.environ.copy()
     # ensure consistent locale for collected command output
     cmd_env['LC_ALL'] = 'C'
@@ -154,24 +161,46 @@ def sos_get_command_output(command, timeout=TIMEOUT_DEFAULT, stderr=False,
                 expanded_args.append(arg)
         else:
             expanded_args.append(arg)
+    if to_file:
+        _output = open(to_file, 'w')
+    else:
+        _output = PIPE
     try:
-        p = Popen(expanded_args, shell=False, stdout=PIPE,
+        p = Popen(expanded_args, shell=False, stdout=_output,
                   stderr=STDOUT if stderr else PIPE,
                   bufsize=-1, env=cmd_env, close_fds=True,
                   preexec_fn=_child_prep_fn)
 
-        reader = AsyncReader(p.stdout, sizelimit, binary)
+        if not to_file:
+            reader = AsyncReader(p.stdout, sizelimit, binary)
+        else:
+            reader = FakeReader(p, binary)
+
         if poller:
             while reader.running:
-                if poller():
-                    p.terminate()
-                    raise SoSTimeoutError
-                time.sleep(0.01)
-        stdout = reader.get_contents()
-        truncated = reader.is_full
+                _check_poller(p)
+        else:
+            try:
+                # override timeout=0 to timeout=None, as Popen will treat the
+                # former as a literal 0-second timeout
+                p.wait(timeout if timeout else None)
+            except Exception:
+                p.terminate()
+                _output.close()
+                # until we separate timeouts from the `timeout` command
+                # handle per-cmd timeouts via Plugin status checks
+                return {'status': 124, 'output': reader.get_contents(),
+                        'truncated': reader.is_full}
+        if to_file:
+            _output.close()
+
+        # wait for Popen to set the returncode
         while p.poll() is None:
             pass
 
+        stdout = reader.get_contents()
+        truncated = reader.is_full
+
     except OSError as e:
         if e.errno == errno.ENOENT:
             return {'status': 127, 'output': "", 'truncated': ''}
@@ -256,6 +285,28 @@ def path_join(path, *p, sysroot=os.sep):
     return os.path.join(path, *p)
 
 
+class FakeReader():
+    """Used as a replacement AsyncReader for when we are writing directly to
+    disk, and allows us to keep more simplified flows for executing,
+    monitoring, and collecting command output.
+    """
+
+    def __init__(self, process, binary):
+        self.process = process
+        self.binary = binary
+
+    @property
+    def is_full(self):
+        return False
+
+    def get_contents(self):
+        return '' if not self.binary else b''
+
+    @property
+    def running(self):
+        return self.process.poll() is None
+
+
 class AsyncReader(threading.Thread):
     """Used to limit command output to a given size without deadlocking
     sos.