[cleaner] Make cleaner concurrent

Draft version adding to the traditional sequential backend also sqlite3
and file based concurrent ones.

Closes: #3097

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,145 @@ 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 initialization / opening the sql connection - or in __init__?
+
+    def load_parser_entries(self):
+        for parser in self.parsers:
+            parser.load_map_entries()
+
+    def unload_parser_entries(self):
+        for parser in self.parsers:
+            parser.unload_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(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
+                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} due to "
+                        f"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, count = self.obfuscate_line(line, _parsers)
+                                subs += count
+                                tfile.write(line)
+                            except Exception as err:
+                                self.log_debug(f"Unable to obfuscate {short_name}:"
+                                               f"{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 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 +256,17 @@ 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[{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
@@ -345,11 +484,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

@@ -8,9 +8,9 @@
 #
 # See the LICENSE file in the source distribution for further information.
 
-import re
-
-from threading import Lock
+import re, os, tempfile
+import sqlite3, time
+from pathlib import Path
 
 
 class SoSMap():
@@ -28,11 +28,41 @@ class SoSMap():
     ignore_short_items = False
     match_full_words_only = False
 
-    def __init__(self):
-        self.dataset = {}
+    def __init__(self, concur_backend, workdir):
+        # TODO: ALL mappings must be deterministic; NO randomization in hostname or ip(6) mapper!
+        self.dataset = {}  # TODO: mention the child classes can update it as well as add_sanitised_item_to_dataset method, so sqlite or files dont neet to have all counter values stored; e.g. when adding an IP address - sqlite and files approach guarantees same input is added in all processes at the same ordering - it can add a new network to the dataset. So dataset bumps by 2
         self._regexes_made = set()
         self.compiled_regexes = []
-        self.lock = Lock()
+        self.cname = self.__class__.__name__.lower()
+        # TODO: ensure the directory <workdir> does exist - but only when concur_backend='files'
+        # TODO: deal with permissions..? IMHO default ones are OK
+        self.workdir = workdir
+        self.link_dir = os.path.join(self.workdir, 'cleaner_links', self.cname) #TODO: some better dir name?
+        self.concur_backend = concur_backend
+        self.load_entries()
+
+    def unload_entries(self):
+        self.sqlconn = self.sqlcursor = None
+
+    def load_entries(self): # originally this was in __init__, but we need to trigger after cloning processes, and also reload content in files or sqlite DB in parent archive class, once children are done
+        # TODO: call load_entries at beginning of obfuscation (list of) *files*
+        if self.concur_backend == 'files':
+            Path(self.link_dir).mkdir(parents=True, exist_ok=True)
+            self.load_new_entries_from_dir(1) # if there are already stored items, load them first
+        elif self.concur_backend == 'sql':
+            Path(os.path.join(self.workdir, 'sqlite_cache')).mkdir(parents=True, exist_ok=True)
+            self.sqlconn = sqlite3.connect(os.path.join(self.workdir, 'sqlite_cache' , f"{self.cname}.db"), check_same_thread=False) # TODO: better name?
+            self.sqlconn.execute('pragma journal_mode = wal') # to improve DB locking on frequent concurrent writes
+            self.sqlconn.execute('pragma temp_store = memory')  # 3 lines added from jcastillo's comment
+            self.sqlconn.execute('pragma synchronous = normal')
+            self.sqlconn.execute('pragma busy_timeout = 5000')
+            self.sqlconn.execute('pragma mmap_size = 30000000000') # TODO: cant this be a problem on low performant systems? some kernel limitation behind..?
+            self.sqlcursor = self.sqlconn.cursor()
+            self.sqlcursor.execute(f"CREATE TABLE IF NOT EXISTS {self.cname} (counter, item)")
+            self.sqlcursor.execute(f"CREATE UNIQUE INDEX IF NOT EXISTS ind_{self.cname}_counter_uniq ON {self.cname} (counter)")
+            self.sqlcursor.execute(f"CREATE UNIQUE INDEX IF NOT EXISTS ind_{self.cname}_item_uniq ON {self.cname} (item)")
+            self.sqlconn.commit()
+            self.load_new_entries_from_db(1) # if there are already stored items, load them first
 
     def ignore_item(self, item):
         """Some items need to be completely ignored, for example link-local or
@@ -46,6 +76,42 @@ 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 as err:
+            # TODO: add seomthing like self.log_debug(f"Unable to obfuscate {item}: {err}", caller=arc_name)
+            self.dataset[item] = item
+        if self.compile_regexes:
+            self.add_regex_item(item)
+
+    def load_new_entries_from_dir(self, counter):
+        # TODO: this is a performance hack; there can be gaps in counter values as e.g. #14 is an IP address (in file)
+        # while #15 is its network (in dataset but not in files). So the next file number is 16. The diffs should be
+        # at most 2, but let be conservative and test next 5 numbers "only".
+        no_files_cnt = 5
+        while no_files_cnt > 0:
+            fname = os.path.join(self.link_dir, f"{counter}")
+            while os.path.isfile(fname):
+                no_files_cnt = 5
+                with open(fname, 'r') 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.link_dir, f"{counter}")
+            # no next file, but try a new next ones until no_files_cnt==0
+            no_files_cnt -= 1
+            counter += 1
+
+    def load_new_entries_from_db(self, counter):
+        if not self.sqlconn:
+            self.load_entries()
+        resp = self.sqlcursor.execute(f"SELECT * FROM {self.cname} WHERE counter>={counter} ORDER BY counter ASC")
+        for _, item in resp.fetchall():
+            if not self.dataset.get(item, False):
+                self.add_sanitised_item_to_dataset(item)
+
     def add(self, item):
         """Add a particular item to the map, generating an obfuscated pair
         for it.
@@ -56,11 +122,45 @@ 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]
+
+        if self.concur_backend == 'files':
+            tmpfile = None
+            while not self.dataset.get(item, False):
+                if not tmpfile:
+                    tmpfile = tempfile.NamedTemporaryFile(dir=self.link_dir)
+                    with open(tmpfile.name, 'w') as f:
+                        f.write(item)
+                try:
+                    counter = len(self.dataset) + 1
+                    os.link(tmpfile.name, os.path.join(self.link_dir, f"{counter}"))
+                    self.add_sanitised_item_to_dataset(item)
+                except FileExistsError:
+                    self.load_new_entries_from_dir(counter)
+
+        elif self.concur_backend == 'sql':
+            self.load_new_entries_from_db(1)
+            while not self.dataset.get(item, False):
+                try:
+                    counter = len(self.dataset) + 1
+                    self.sqlcursor.execute(f"INSERT INTO {self.cname} VALUES ({counter}, '{item}')")
+                    self.sqlconn.commit()
+                    self.add_sanitised_item_to_dataset(item)
+                except sqlite3.IntegrityError as e:
+                    print(f"    {os.getpid()}-{self.cname}: sql conflict for counter={counter} when inserting {item}")  # TODO: remove this
+                    self.load_new_entries_from_db(counter)
+                except sqlite3.OperationalError as e:
+                    print(f"    {os.getpid()}-{self.cname}: sql ERROR for counter={counter} when inserting {item}") # TODO: remove this (here we can get a live-lock..)
+                    self.unload_entries()
+                    time.sleep(1)
+                    self.load_entries()
+                    self.load_new_entries_from_db(counter)
+                except Exception as e:
+                    print(f"    {os.getpid()}-{self.cname}: sql generic ERROR {e} for counter={counter} when inserting {item}") # TODO: remove this
+                    raise e
+        else:
+            self.add_sanitised_item_to_dataset(item)
+
+        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