Introduce optional file locking for appending to perflogs

Also:

- Introduce an `osext.flock()` utility function to handle the underlying
  locking for the different Python versions.
- Use Python syntax for ``true`` and ``false`` in the configuration
  reference to be consistent with the use of ``None`` instead of the
  JSON's ``null``.

Author: vkarak

docs/config_reference.rst

@@ -2,6 +2,10 @@ archspec==0.2.5
 ClusterShell==1.9.3
 docutils==0.18.1; python_version < '3.9'
 docutils==0.21.2; python_version >= '3.9'
+filelock==3.4.1; python_version == '3.6'
+filelock==3.12.2; python_version == '3.7'
+filelock==3.16.1; python_version == '3.8'
+filelock==3.18.0; python_version > '3.8'
 jinja2==3.0.3; python_version == '3.6'
 jinja2==3.1.6; python_version >= '3.7'
 jsonschema==3.2.0

docs/requirements.txt

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
 import abc
+import atexit
 import itertools
 import logging
 import logging.handlers
@@ -149,7 +150,8 @@ class MultiFileHandler(logging.FileHandler):
     '''
 
     def __init__(self, prefix, mode='a', encoding=None, fmt=None,
-                 perffmt=None, ignore_keys=None):
+                 perffmt=None, ignore_keys=None, use_locking=False,
+                 lockfile_mode=None):
         super().__init__(prefix, mode, encoding, delay=True)
 
         # Reset FileHandler's filename
@@ -164,6 +166,8 @@ def __init__(self, prefix, mode='a', encoding=None, fmt=None,
         self.__perffmt = perffmt
         self.__attr_patt = re.compile(r'\%\((.*?)\)s(.*?(?=%|$))?')
         self.__ignore_keys = set(ignore_keys) if ignore_keys else set()
+        self.__use_locking = use_locking
+        self.__lockfile_mode = lockfile_mode
 
     def __generate_header(self, record):
         # Generate the header from the record and fmt
@@ -234,13 +238,25 @@ def _emit_header(self, record):
 
             os.rename(self.baseFilename, self.baseFilename + f'.h{hcnt}')
         finally:
-            # Open the file for writing and write the header
-            fp = open(self.baseFilename,
-                      mode=self.mode, encoding=self.encoding)
-            if record_header != header:
-                fp.write(f'{record_header}\n')
+            if self.__use_locking:
+                # When using locking, we need to  open, append and write to
+                # the file at once
+                lockfile = os.path.join(f'{self.baseFilename}.lock')
+                with osext.flock(lockfile, self.__lockfile_mode):
+                    with open(self.baseFilename, mode=self.mode,
+                              encoding=self.encoding) as fp:
+                        if record_header != header:
+                            fp.write(f'{record_header}\n')
+
+                self.__streams[self.baseFilename] = None
+            else:
+                # Open the file for writing and write the header
+                fp = open(self.baseFilename,
+                          mode=self.mode, encoding=self.encoding)
+                if record_header != header:
+                    fp.write(f'{record_header}\n')
 
-            self.__streams[self.baseFilename] = fp
+                self.__streams[self.baseFilename] = fp
 
     def emit(self, record):
         try:
@@ -255,14 +271,23 @@ def emit(self, record):
         check_basename = type(record.__rfm_check__).variant_name()
         self.baseFilename = os.path.join(dirname, f'{check_basename}.log')
         self._emit_header(record)
-        self.stream = self.__streams[self.baseFilename]
-        super().emit(record)
+        if self.__use_locking:
+            lockfile = os.path.join(dirname, f'.{check_basename}.lock')
+            with osext.flock(lockfile, self.__lockfile_mode):
+                with open(self.baseFilename, mode=self.mode,
+                          encoding=self.encoding) as fp:
+                    self.stream = fp
+                    super().emit(record)
+        else:
+            self.stream = self.__streams[self.baseFilename]
+            super().emit(record)
 
     def close(self):
         # Close all open streams
-        for s in self.__streams.values():
-            self.stream = s
-            super().close()
+        for stream in self.__streams.values():
+            if stream:
+                self.stream = stream
+                super().close()
 
 
 def _format_time_rfc3339(timestamp, datefmt):
@@ -459,9 +484,13 @@ def _create_filelog_handler(site_config, config_prefix):
     format = site_config.get(f'{config_prefix}/format')
     format_perf = site_config.get(f'{config_prefix}/format_perfvars')
     ignore_keys = site_config.get(f'{config_prefix}/ignore_keys')
+    use_locking = site_config.get(f'{config_prefix}/locking_enable')
+    lockfile_mode = site_config.get(f'{config_prefix}/locking_file_mode')
     return MultiFileHandler(filename_patt, mode='a+' if append else 'w+',
                             fmt=format, perffmt=format_perf,
-                            ignore_keys=ignore_keys)
+                            ignore_keys=ignore_keys,
+                            use_locking=use_locking,
+                            lockfile_mode=lockfile_mode)
 
 
 @register_log_handler('syslog')
@@ -806,6 +835,13 @@ def debug(self, message, *args, **kwargs):
     def debug2(self, message, *args, **kwargs):
         self.log(DEBUG2, message, *args, **kwargs)
 
+    def shutdown(self):
+        '''Shutdown logger by removing all handlers and closing any files.'''
+
+        for h in list(self.handlers):
+            h.close()
+            self.removeHandler(h)
+
 
 # This is a cache for warnings that we don't want to repeat
 _WARN_ONCE = set()
@@ -1014,6 +1050,8 @@ def configure_logging(site_config):
         _context_logger = null_logger
         return
 
+    # Shutdown the previously setup loggers and close any files
+    shutdown()
     _logger = _create_logger(site_config, 'handlers$', 'handlers')
     _perf_logger = _create_logger(site_config, 'handlers_perflog')
     _context_logger = LoggerAdapter(_logger)
@@ -1066,6 +1104,17 @@ def _fn(*args, **kwargs):
     return _fn
 
 
+@atexit.register
+def shutdown():
+    '''Shutdown logging'''
+
+    if _logger:
+        _logger.shutdown()
+
+    if _perf_logger:
+        _perf_logger.shutdown()
+
+
 # The following is meant to be used only by the unit tests
 
 class logging_sandbox:
@@ -1085,6 +1134,7 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_value, traceback):
         global _logger, _perf_logger, _context_logger
 
+        shutdown()
         _logger = self._logger
         _perf_logger = self._perf_logger
         _context_logger = self._context_logger

reframe/core/logging.py

@@ -9,8 +9,6 @@
 import os
 import re
 import sqlite3
-import sys
-from filelock import FileLock
 
 import reframe.utility.jsonext as jsonext
 import reframe.utility.osext as osext
@@ -124,22 +122,8 @@ def _db_connect(self, *args, **kwargs):
             return sqlite3.connect(*args, **kwargs)
 
     def _db_lock(self):
-        prefix = os.path.dirname(self.__db_file)
-        if sys.version_info >= (3, 7):
-            kwargs = {'mode': self.__db_file_mode}
-        else:
-            # Python 3.6 forces us to use an older filelock version that does
-            # not support file modes. File modes where introduced in
-            # filelock 3.10
-            kwargs = {}
-
-        # Create parent directories of the lock file
-        #
-        # NOTE: This is not necessary for filelock >= 3.12.3 and Python >= 3.8
-        # However, we do create it here, in order to support the older Python
-        # versions.
-        os.makedirs(prefix, exist_ok=True)
-        return FileLock(os.path.join(prefix, '.db.lock'), **kwargs)
+        lockfile = os.path.join(os.path.dirname(self.__db_file), '.db.lock')
+        return osext.flock(lockfile, self.__db_file_mode)
 
     def _db_create(self):
         clsname = type(self).__name__

reframe/frontend/reporting/storage.py

@@ -85,7 +85,9 @@
                         "ignore_keys": {
                             "type": "array",
                             "items": {"type": "string"}
-                        }
+                        },
+                        "locking_enable": {"type": "boolean"},
+                        "locking_file_mode": {"type": "string"}
                     },
                     "required": ["prefix"]
                 }
@@ -633,6 +635,8 @@
         "logging/handlers_perflog/filelog_append": true,
         "logging/handlers_perflog/filelog_basedir": "./perflogs",
         "logging/handlers_perflog/filelog_ignore_keys": [],
+        "logging/handlers_perflog/filelog_locking_enable": false,
+        "logging/handlers_perflog/filelog_locking_file_mode": "644",
         "logging/handlers_perflog/graylog_extras": {},
         "logging/handlers_perflog/httpjson_extras": {},
         "logging/handlers_perflog/httpjson_ignore_keys": [],

reframe/schemas/config.json

@@ -9,6 +9,7 @@
 
 import collections.abc
 import errno
+import filelock
 import getpass
 import grp
 import os
@@ -863,6 +864,44 @@ def unique_abs_paths(paths, prune_children=True):
     return list(unique_paths - children)
 
 
+def flock(name, mode=None):
+    '''Obtain a file lock
+
+    :arg name: the name of the lock file
+    :arg mode: the file mode of the lock file
+    :returns: a context manager that holds the lock.
+        Example usage:
+
+        .. code-block:: python
+
+            with flock('file.lock'):
+                # do something
+
+    .. versionadded:: 4.8.3
+
+    '''
+
+    if not isinstance(mode, int):
+        mode = int(mode, base=8)
+
+    if mode and sys.version_info >= (3, 7):
+        kwargs = {'mode': mode}
+    else:
+        # Python 3.6 forces us to use an older filelock version that does
+        # not support file modes. File modes where introduced in
+        # filelock 3.10
+        kwargs = {}
+
+    # Create parent directories of the lock file
+    #
+    # NOTE: This is not necessary for filelock >= 3.12.3 and Python >= 3.8
+    # However, we do create it here, in order to support the older Python
+    # versions.
+    prefix = os.path.dirname(name)
+    os.makedirs(prefix, exist_ok=True)
+    return filelock.FileLock(name, **kwargs)
+
+
 def cray_cdt_version():
     '''Return either the Cray Development Toolkit (CDT) version, the Cray
     Programming Environment (CPE) version or :class:`None` if the version

reframe/utility/osext.py

@@ -69,7 +69,8 @@ def handler(logfile, rfc3339formatter):
 def logger(handler):
     logger = rlog.Logger('reframe')
     logger.addHandler(handler)
-    return logger
+    yield logger
+    logger.shutdown()
 
 
 @pytest.fixture
@@ -545,7 +546,7 @@ def url_scheme(request):
 
 def test_httpjson_handler_no_port(make_exec_ctx, config_file,
                                   url_scheme, logging_sandbox):
-    ctx = make_exec_ctx(
+    make_exec_ctx(
         config_file({
             'level': 'info',
             'handlers_perflog': [{

unittests/test_logging.py

@@ -465,3 +465,31 @@ def validate(self):
 
     assert len(lines) == 2
     assert lines[1] == 'fail|sanity error: no way|None|None\n'
+
+
+def test_perf_logging_locking(make_runner, make_exec_ctx,
+                              config_perflog, perf_test, tmp_path):
+    make_exec_ctx(config_perflog(
+        fmt='',
+        logging_opts={
+            'handlers_perflog': [{
+                'type': 'filelog',
+                'use_locking': True,
+                'prefix': '%(check_system)s/%(check_partition)s',
+                'level': 'info',
+                'format': (
+                    '%(check_job_completion_time)s,%(version)s,'
+                    '%(check_display_name)s,%(check_system)s,'
+                    '%(check_partition)s,%(check_environ)s,'
+                    '%(check_jobid)s,%(check_result)s,%(check_perfvalues)s'
+                )
+            }]
+        }
+    ))
+    logging.configure_logging(rt.runtime().site_config)
+    runner = make_runner()
+    testcases = executors.generate_testcases([perf_test])
+    _assert_no_logging_error(runner.runall, testcases)
+    logfile = tmp_path / 'perflogs' / 'generic' / 'default' / '_MyPerfTest.log'
+    assert os.path.exists(logfile)
+    assert _count_lines(logfile) == 2