[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>

Author: pmoravec

sos/cleaner/__init__.py

@@ -13,6 +13,7 @@
 import shutil
 import stat
 import tarfile
+import tempfile
 import re
 
 from concurrent.futures import ProcessPoolExecutor
@@ -54,15 +55,15 @@ class SoSObfuscationArchive():
     class. All report-level operations should be contained within this class.
     """
 
-    file_sub_list = []
+    files_obfuscated_count = 0
     total_sub_count = 0
     removed_file_count = 0
     type_name = 'undetermined'
     description = 'undetermined'
     is_nested = False
     prep_files = {}
 
-    def __init__(self, archive_path, tmpdir):
+    def __init__(self, archive_path, tmpdir, keep_binary_files):
         self.archive_path = archive_path
         self.final_archive_path = self.archive_path
         self.tmpdir = tmpdir
@@ -74,10 +75,144 @@ def __init__(self, archive_path, tmpdir):
         self.is_extracted = False
         self._load_self()
         self.archive_root = ''
+        self.keep_binary_files = keep_binary_files
+        self.parsers = ()
         self.log_info(
             f"Loaded {self.archive_path} as type {self.description}"
         )
 
+    def obfuscate_string(self, string_data):
+        for parser in self.parsers:
+            try:
+                string_data = parser.parse_string_for_keys(string_data)
+            except Exception as err:
+                self.log_info(f"Error obfuscating string data: {err}")
+        return string_data
+
+    # TODO: merge content to obfuscate_arc_files as that is the only place we
+    # call obfuscate_filename ?
+    def obfuscate_filename(self, short_name, filename):
+        _ob_short_name = self.obfuscate_string(short_name.split('/')[-1])
+        _ob_filename = short_name.replace(short_name.split('/')[-1],
+                                          _ob_short_name)
+
+        if _ob_filename != short_name:
+            arc_path = filename.split(short_name)[0]
+            _ob_path = os.path.join(arc_path, _ob_filename)
+            # ensure that any plugin subdirs that contain obfuscated strings
+            # get created with obfuscated counterparts
+            if not os.path.islink(filename):
+                os.rename(filename, _ob_path)
+            else:
+                # generate the obfuscated name of the link target
+                _target_ob = self.obfuscate_string(os.readlink(filename))
+                # remove the unobfuscated original symlink first, in case the
+                # symlink name hasn't changed but the target has
+                os.remove(filename)
+                # create the newly obfuscated symlink, pointing to the
+                # obfuscated target name, which may not exist just yet, but
+                # when the actual file is obfuscated, will be created
+                os.symlink(_target_ob, _ob_path)
+
+    def set_parsers(self, parsers):
+        self.parsers = parsers  # TODO: include this in __init__?
+
+    def load_parser_entries(self):
+        for parser in self.parsers:
+            parser.load_map_entries()
+
+    def obfuscate_line(self, line, parsers=None):
+        """Run a line through each of the obfuscation parsers, keeping a
+        cumulative total of substitutions done on that particular line.
+
+        Positional arguments:
+
+            :param line str:        The raw line as read from the file being
+                                    processed
+            :param parsers:         A list of parser objects to obfuscate
+                                    with. If None, use all.
+
+        Returns the fully obfuscated line and the number of substitutions made
+        """
+        # don't iterate over blank lines, but still write them to the tempfile
+        # to maintain the same structure when we write a scrubbed file back
+        count = 0
+        if not line.strip():
+            return line, count
+        if parsers is None:
+            parsers = self.parsers
+        for parser in parsers:
+            try:
+                line, _count = parser.parse_line(line)
+                count += _count
+            except Exception as err:
+                self.log_debug(f"failed to parse line: {err}", parser.name)
+        return line, count
+
+    def obfuscate_arc_files(self, flist):
+        for filename in flist:
+            self.log_debug(f"    pid={os.getpid()}: obfuscating {filename}")
+            try:
+                short_name = filename.split(self.archive_name + '/')[1]
+                if self.should_skip_file(short_name):
+                    continue
+                if (not self.keep_binary_files and
+                        self.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
+                    self.remove_file(short_name)
+                    continue
+                if (self.keep_binary_files and
+                        (file_is_binary(filename) or
+                         self.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
+                if os.path.islink(filename):
+                    # don't run the obfuscation on the link, but on the actual
+                    # file at some other point.
+                    continue
+                _parsers = [
+                    _p for _p in self.parsers if not
+                    any(
+                        _skip.match(short_name) for _skip in _p.skip_patterns
+                    )
+                ]
+                if not _parsers:
+                    self.log_debug(
+                        f"Skipping obfuscation of {short_name or filename} "
+                        f"due to matching file skip pattern"
+                    )
+                    continue
+                self.log_debug(f"Obfuscating {short_name or filename}")
+                subs = 0
+                with tempfile.NamedTemporaryFile(mode='w', dir=self.tmpdir) \
+                        as tfile:
+                    with open(filename, 'r', encoding='utf-8',
+                              errors='replace') as fname:
+                        for line in fname:
+                            try:
+                                line, cnt = self.obfuscate_line(line, _parsers)
+                                subs += cnt
+                                tfile.write(line)
+                            except Exception as err:
+                                self.log_debug(f"Unable to obfuscate "
+                                               f"{short_name}: {err}")
+                    tfile.seek(0)
+                    if subs:
+                        shutil.copyfile(tfile.name, filename)
+                        self.update_sub_count(subs)
+
+                self.obfuscate_filename(short_name, filename)
+
+            except Exception as err:
+                self.log_debug(f"    pid={os.getpid()}: caught exception on "
+                               f"obfuscating file {filename}: {err}")
+
+        return (self.files_obfuscated_count, self.total_sub_count,
+                self.removed_file_count)
+
     @classmethod
     def check_is_type(cls, arc_path):
         """Check if the archive is a well-known type we directly support"""
@@ -120,14 +255,18 @@ def report_msg(self, msg):
         """Helper to easily format ui messages on a per-report basis"""
         self.ui_log.info(f"{self.ui_name + ' :':<50} {msg}")
 
-    def _fmt_log_msg(self, msg):
-        return f"[cleaner:{self.archive_name}] {msg}"
+    def _fmt_log_msg(self, msg, caller=None):
+        return f"[cleaner{f':{caller}' if caller else ''}" \
+               f"[{self.archive_name}]] {msg}"
+
+    def log_debug(self, msg, caller=None):
+        self.soslog.debug(self._fmt_log_msg(msg, caller))
 
-    def log_debug(self, msg):
-        self.soslog.debug(self._fmt_log_msg(msg))
+    def log_info(self, msg, caller=None):
+        self.soslog.info(self._fmt_log_msg(msg, caller))
 
-    def log_info(self, msg):
-        self.soslog.info(self._fmt_log_msg(msg))
+    def log_error(self, msg, caller=None):
+        self.soslog.error(self._fmt_log_msg(msg, caller))
 
     def _load_skip_list(self):
         """Provide a list of files and file regexes to skip obfuscation on
@@ -201,6 +340,7 @@ def extract(self, quiet=False):
                 self.report_msg("Extracting...")
             self.extracted_path = self.extract_self()
             self.is_extracted = True
+            self.tarobj = None    # we can't pickle this & not further needed
         else:
             self.extracted_path = self.archive_path
         # if we're running as non-root (e.g. collector), then we can have a
@@ -326,7 +466,7 @@ def get_symlinks(self):
                 if os.path.islink(_fname):
                     yield _fname
 
-    def get_file_list(self):
+    def get_files(self):
         """Iterator for a list of files in the archive, to allow clean to
         iterate over.
 
@@ -345,11 +485,11 @@ def get_directory_list(self):
             dir_list.append(dirname)
         return dir_list
 
-    def update_sub_count(self, fname, count):
+    def update_sub_count(self, count):
         """Called when a file has finished being parsed and used to track
         total substitutions made and number of files that had changes made
         """
-        self.file_sub_list.append(fname)
+        self.files_obfuscated_count += 1
         self.total_sub_count += count
 
     def get_file_path(self, fname):

sos/cleaner/archives/__init__.py

@@ -69,7 +69,8 @@ def get_nested_archives(self):
         for fname in os.listdir(_path):
             arc_name = os.path.join(_path, fname)
             if 'sosreport-' in fname and tarfile.is_tarfile(arc_name):
-                archives.append(SoSReportArchive(arc_name, self.tmpdir))
+                archives.append(SoSReportArchive(arc_name, self.tmpdir,
+                                                 self.keep_binary_files))
         return archives
 
 

sos/cleaner/archives/sos.py

@@ -9,8 +9,9 @@
 # See the LICENSE file in the source distribution for further information.
 
 import re
-
-from threading import Lock
+import os
+import tempfile
+from pathlib import Path
 
 
 class SoSMap():
@@ -28,11 +29,33 @@ class SoSMap():
     ignore_short_items = False
     match_full_words_only = False
 
-    def __init__(self):
+    def __init__(self, workdir):
         self.dataset = {}
         self._regexes_made = set()
         self.compiled_regexes = []
-        self.lock = Lock()
+        self.cname = self.__class__.__name__.lower()
+        # workdir's default value '/tmp' is used just by avocado tests,
+        # otherwise we override it to /etc/sos/cleaner (or map_file dir)
+        self.workdir = workdir
+        self.cache_dir = os.path.join(self.workdir, 'cleaner_cache',
+                                      self.cname)
+        self.load_entries()
+
+    def load_entries(self):
+        """ Load cached entries from the disk. This method must be called when
+        we initialize a Map instance and whenever we want to retrieve
+        self.dataset (e.g. to store default_mapping file). The later is
+        essential since a concurrent Map can add more objects to the cache,
+        so we need to update self.dataset up to date.
+
+        Keep in mind that size of self.dataset is usually bigger than number
+        of files in the corresponding cleaner's directory: directory contains
+        just whole items (e.g. IP addresses) while dataset contains more
+        derived objects (e.g. subnets).
+        """
+
+        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
+        self.load_new_entries_from_dir(1)
 
     def ignore_item(self, item):
         """Some items need to be completely ignored, for example link-local or
@@ -46,6 +69,36 @@ def ignore_item(self, item):
                 return True
         return False
 
+    def add_sanitised_item_to_dataset(self, item):
+        try:
+            self.dataset[item] = self.sanitize_item(item)
+        except Exception:
+            self.dataset[item] = item
+        if self.compile_regexes:
+            self.add_regex_item(item)
+
+    def load_new_entries_from_dir(self, counter):
+        # this is a performance hack; there can be gaps in counter values as
+        # e.g. sanitised item #14 is an IP address (in file) while item #15
+        # is its network (in dataset but not in files). So the next file
+        # number is 16. The diffs should be at most 2, the above is so far
+        # the only type of "underneath dataset growth". But let be
+        # conservative and test next 5 numbers "only".
+        no_files_cnt = 5
+        while no_files_cnt > 0:
+            fname = os.path.join(self.cache_dir, f"{counter}")
+            while os.path.isfile(fname):
+                no_files_cnt = 5
+                with open(fname, 'r', encoding='utf-8') as f:
+                    item = f.read()
+                if not self.dataset.get(item, False):
+                    self.add_sanitised_item_to_dataset(item)
+                counter += 1
+                fname = os.path.join(self.cache_dir, f"{counter}")
+            # no next file, but try a new next ones until no_files_cnt==0
+            no_files_cnt -= 1
+            counter += 1
+
     def add(self, item):
         """Add a particular item to the map, generating an obfuscated pair
         for it.
@@ -56,11 +109,23 @@ def add(self, item):
         """
         if self.ignore_item(item):
             return item
-        with self.lock:
-            self.dataset[item] = self.sanitize_item(item)
-            if self.compile_regexes:
-                self.add_regex_item(item)
-            return self.dataset[item]
+
+        tmpfile = None
+        while not self.dataset.get(item, False):
+            if not tmpfile:
+                # pylint: disable=consider-using-with
+                tmpfile = tempfile.NamedTemporaryFile(dir=self.cache_dir)
+                with open(tmpfile.name, 'w', encoding='utf-8') as f:
+                    f.write(item)
+            try:
+                counter = len(self.dataset) + 1
+                os.link(tmpfile.name, os.path.join(self.cache_dir,
+                                                   f"{counter}"))
+                self.add_sanitised_item_to_dataset(item)
+            except FileExistsError:
+                self.load_new_entries_from_dir(counter)
+
+        return self.dataset[item]
 
     def add_regex_item(self, item):
         """Add an item to the regexes dict and then re-sort the list that the

sos/cleaner/mappings/__init__.py

@@ -9,7 +9,6 @@
 # See the LICENSE file in the source distribution for further information.
 
 import ipaddress
-import random
 
 from sos.cleaner.mappings import SoSMap
 
@@ -45,6 +44,11 @@ class SoSIPMap(SoSMap):
     network_first_octet = 100
     skip_network_octets = ['127', '169', '172', '192']
     compile_regexes = False
+    # counter for obfuscating a single IP address; the value stands for
+    # 172.17.0.0; we use a private block of IP addresses and ignore
+    # 172.16.0.0/16 block as those addresses are more often used in real
+    # (an attempt to prevent confusion)
+    _saddr_cnt = 2886795264
 
     def ip_in_dataset(self, ipaddr):
         """There are multiple ways in which an ip address could be handed to us
@@ -162,13 +166,14 @@ def sanitize_ipaddr(self, addr):
         return self._new_obfuscated_single_address()
 
     def _new_obfuscated_single_address(self):
-        def _gen_address():
-            _octets = []
-            for _ in range(0, 4):
-                _octets.append(random.randint(11, 99))
-            return f"{_octets[0]}.{_octets[1]}.{_octets[2]}.{_octets[3]}"
-
-        _addr = _gen_address()
+        # increment the counter and ignore *.0 and *.255 addresses
+        self._saddr_cnt += 1
+        while self._saddr_cnt % 256 in (0, 255):
+            self._saddr_cnt += 1
+        # split the counter value to four octets (i.e. % 256) to get an
+        # obfuscated IP address
+        _addr = f"{self._saddr_cnt >> 24}.{(self._saddr_cnt >> 16) % 256}." \
+            f"{(self._saddr_cnt >> 8) % 256}.{self._saddr_cnt % 256}"
         if _addr in self.dataset.values():
             return self._new_obfuscated_single_address()
         return _addr