[cleaner] Make cleaner concurrent

Allow running cleaner concurrently via child processes. They synchronize
on the ordering of items added to dataset of each mapper by creating
numbered files in a directory specific for each mapper. Together with
deterministic generation of obfuscated values, this ensures the
individual processes end up with identical mappings.

Resolves: sosreport#3097
Closes: sosreport#3988

Signed-off-by: Pavel Moravec <pmoravec@redhat.com>
Signed-off-by: Pablo Fernández Rodríguez <pafernan@redhat.com>

Author: pafernanr

sos/cleaner/__init__.py

@@ -34,16 +34,14 @@
                                       SoSCollectorDirectory)
 from sos.cleaner.archives.generic import DataDirArchive, TarballArchive
 from sos.cleaner.archives.insights import InsightsArchive
-from sos.utilities import (get_human_readable, import_module, ImporterHelper,
-                           file_is_certificate, is_executable)
+from sos.utilities import (get_human_readable, import_module, ImporterHelper)
 
 
 # an auxiliary method to kick off child processes over its instances
 def obfuscate_arc_files(arc, flist):
     return arc.obfuscate_arc_files(flist)
 
 
-
 class SoSCleaner(SoSComponent):
     """
     This function is designed to obfuscate potentially sensitive information
@@ -588,34 +586,9 @@ def obfuscate_report_paths(self):
                     "WARNING: binary files that potentially contain sensitive "
                     "information will NOT be removed from the final archive\n"
                 )
-            if (self.opts.treat_certificates == "obfuscate"
-                    and not is_executable("openssl")):
-                self.opts.treat_certificates = "remove"
-                self.ui_log.warning(
-                    "WARNING: No `openssl` command available. Replacing "
-                    "`--treat-certificates` from `obfuscate` to `remove`."
-                )
-            if self.opts.treat_certificates == "obfuscate":
-                self.ui_log.warning(
-                    "WARNING: certificate files that potentially contain "
-                    "sensitive information will be CONVERTED to text and "
-                    "OBFUSCATED in the final archive.\n"
-                )
-            elif self.opts.treat_certificates == "keep":
-                self.ui_log.warning(
-                    "WARNING: certificate files that potentially contain "
-                    "sensitive information will be KEPT in the final "
-                    "archive as is.\n"
-                )
-            elif self.opts.treat_certificates == "remove":
-                self.ui_log.warning(
-                    "WARNING: certificate files that potentially contain "
-                    "sensitive information will be REMOVED in the final "
-                    "archive.\n"
-                )
-            for report_path in self.report_paths:
-                self.ui_log.info(f"Obfuscating {report_path.archive_path}")
-                self.obfuscate_report(report_path)
+            pool = ThreadPoolExecutor(self.opts.jobs)
+            pool.map(self.obfuscate_report, self.report_paths, chunksize=1)
+            pool.shutdown(wait=True)
             # finally, obfuscate the nested archive if one exists
             if self.nested_archive:
                 self._replace_obfuscated_archives()
@@ -732,63 +705,31 @@ def obfuscate_report(self, archive):  # pylint: disable=too-many-branches
             if not archive.is_extracted:
                 archive.extract()
             archive.report_msg("Beginning obfuscation...")
+
             for fname in archive.get_file_list():
                 short_name = fname.split(archive.archive_name + '/')[1]
-                is_certificate = file_is_certificate(fname)
-                if is_certificate:
-                    if is_certificate == "certificatekey":
-                        # Always remove certificate Key files
-                        archive.remove_file(short_name)
-                        continue
-                    if self.opts.treat_certificates == "keep":
-                        continue
-                    if self.opts.treat_certificates == "remove":
-                        archive.remove_file(short_name)
-                        continue
-                    if self.opts.treat_certificates == "obfuscate":
-                        archive.certificate_to_text(fname)
-
-            file_list = list(archive.get_files())
-            # we can't call simple
-            # executor.map(archive.obfuscate_arc_files,archive.get_files())
-            # because a child process does not carry forward internal changes
-            # (e.g. mappings' datasets) from one call of obfuscate_arc_files
-            # method to another. Each obfuscate_arc_files method starts with
-            # vanilla parent archive, that is initialised *once* at its
-            # beginning via initializer=archive.load_parser_entries
-            # - but not afterwards..
-            #
-            # So we must pass list of all files for each worker at the
-            # beginning. This means less granularity of the child processes
-            # work (one worker can finish much sooner than the other), but
-            # it is the best we can have (or have found)
-            #
-            # At least, the "file_list[i::self.opts.jobs]" means subsequent
-            # files (speculativelly of similar size and content) are
-            # distributed to different processes, which attempts to split the
-            # load evenly. Yet better approach might be reorderig file_list
-            # based on files' sizes.
-
-            files_obfuscated_count = total_sub_count = removed_file_count = 0
-            archive_list = [archive for i in range(self.opts.jobs)]
-            with ProcessPoolExecutor(
-                    max_workers=self.opts.jobs,
-                    initializer=archive.load_parser_entries) as executor:
-                futures = executor.map(obfuscate_arc_files, archive_list,
-                                       [file_list[i::self.opts.jobs] for i in
-                                        range(self.opts.jobs)])
-                for (foc, tsc, rfc) in futures:
-                    files_obfuscated_count += foc
-                    total_sub_count += tsc
-                    removed_file_count += rfc
-
-            # As there is no easy way to get dataset dicts from child
-            # processes' mappings, we can reload our own parent-process
-            # archive from the disk files. The trick is that sequence of
-            # files/entries is the source of truth of *sequence* of calling
-            # *all* mapping.all(item) methods - so replaying this will
-            # generate the right datasets!
-            archive.load_parser_entries()
+                if archive.should_skip_file(short_name):
+                    continue
+                if (not self.opts.keep_binary_files and
+                        archive.should_remove_file(short_name)):
+                    # We reach this case if the option --keep-binary-files
+                    # was not used, and the file is in a list to be removed
+                    archive.remove_file(short_name)
+                    continue
+                if (self.opts.keep_binary_files and
+                        (file_is_binary(fname) or
+                         archive.should_remove_file(short_name))):
+                    # We reach this case if the option --keep-binary-files
+                    # is used. In this case we want to make sure
+                    # the cleaner doesn't try to clean a binary file
+                    continue
+                try:
+                    count = self.obfuscate_file(fname, short_name,
+                                                archive.archive_name)
+                    if count:
+                        archive.update_sub_count(short_name, count)
+                except Exception as err:
+                    self.log_debug(f"Unable to parse file {short_name}: {err}")
 
             try:
                 self.obfuscate_directory_names(archive)

sos/cleaner/archives/__init__.py

@@ -17,7 +17,8 @@
 import tempfile
 
 from concurrent.futures import ProcessPoolExecutor
-from sos.utilities import file_is_binary, sos_get_command_output
+from sos.utilities import (file_is_binary, sos_get_command_output,
+                           file_is_certificate)
 
 
 # python older than 3.8 will hit a pickling error when we go to spawn a new
@@ -63,7 +64,8 @@ class SoSObfuscationArchive():
     is_nested = False
     prep_files = {}
 
-    def __init__(self, archive_path, tmpdir, keep_binary_files):
+    def __init__(self, archive_path, tmpdir, keep_binary_files,
+                 treat_certificates):
         self.archive_path = archive_path
         self.final_archive_path = self.archive_path
         self.tmpdir = tmpdir
@@ -76,6 +78,7 @@ def __init__(self, archive_path, tmpdir, keep_binary_files):
         self._load_self()
         self.archive_root = ''
         self.keep_binary_files = keep_binary_files
+        self.treat_certificates = treat_certificates
         self.parsers = ()
         self.log_info(
             f"Loaded {self.archive_path} as type {self.description}"
@@ -173,6 +176,19 @@ def obfuscate_arc_files(self, flist):
                     # don't run the obfuscation on the link, but on the actual
                     # file at some other point.
                     continue
+                is_certificate = file_is_certificate(filename)
+                if is_certificate:
+                    if is_certificate == "certificatekey":
+                        # Always remove certificate Key files
+                        self.remove_file(short_name)
+                        continue
+                    if self.treat_certificates == "keep":
+                        continue
+                    if self.treat_certificates == "remove":
+                        self.remove_file(short_name)
+                        continue
+                    if self.treat_certificates == "obfuscate":
+                        self.certificate_to_text(filename)
                 _parsers = [
                     _p for _p in self.parsers if not
                     any(