Several related fixes and improvements in logging, output capturing, and command execution.

borgmatic/actions/check.py

@@ -448,24 +448,22 @@ def collect_spot_check_archive_paths(
     borgmatic_source_directory = borgmatic.config.paths.get_borgmatic_source_directory(config)
 
     return tuple(
-        path
-        for line in borgmatic.borg.list.capture_archive_listing(
+        entry['path']
+        for entry in borgmatic.borg.list.capture_archive_listing(
             repository['path'],
             archive,
             config,
             local_borg_version,
             global_arguments,
-            path_format='{type} {path}{NUL}',
             local_path=local_path,
             remote_path=remote_path,
         )
-        for (file_type, path) in (line.split(' ', 1),)
-        if file_type not in {BORG_DIRECTORY_FILE_TYPE, BORG_PIPE_FILE_TYPE}
-        if pathlib.Path('borgmatic') not in pathlib.Path(path).parents
+        if entry['type'] not in {BORG_DIRECTORY_FILE_TYPE, BORG_PIPE_FILE_TYPE}
+        if pathlib.Path('borgmatic') not in pathlib.Path(entry['path']).parents
         if pathlib.Path(borgmatic_source_directory.lstrip(os.path.sep))
-        not in pathlib.Path(path).parents
+        not in pathlib.Path(entry['path']).parents
         if pathlib.Path(borgmatic_runtime_directory.lstrip(os.path.sep))
-        not in pathlib.Path(path).parents
+        not in pathlib.Path(entry['path']).parents
     )
 
 
@@ -523,20 +521,33 @@ def compare_spot_check_hashes(
         if not source_sample_paths_subset:
             break
 
+        hash_paths = tuple(
+            path for path in source_sample_paths_subset if path in hashable_source_sample_path
+        )
         hash_lines = borgmatic.execute.execute_command_and_capture_output(
             tuple(
                 shlex.quote(part)
                 for part in shlex.split(spot_check_config.get('xxh64sum_command', 'xxh64sum'))
-            )
-            + tuple(
-                path for path in source_sample_paths_subset if path in hashable_source_sample_path
-            ),
+            ) + hash_paths,
             working_directory=working_directory,
         )
 
         source_hashes.update(
             **dict(
-                (reversed(line.split('  ', 1)) for line in hash_lines),
+                zip(
+                    # xxh64sum rewrites/escapes the paths that it returns alongside its hashes, for
+                    # instance if they contain special characters. When that happens, they don't
+                    # match the original source paths and therefore hash lookups fail. So when
+                    # building this lookup dict, use the original unaltered paths we provided as
+                    # input to xxh64sum.
+                    hash_paths,
+                    (
+                        # For some reason, xxh64sum prefixes the hash with a backslash if the path
+                        # contains a newline. Work around that.
+                        line.split('  ', 1)[0].lstrip('\\')
+                        for line in hash_lines
+                    ),
+                ),
                 # Represent non-existent files as having empty hashes so the comparison below still
                 # works. Same thing for filesystem links, since Borg produces empty archive hashes
                 # for them.
@@ -550,21 +561,21 @@ def compare_spot_check_hashes(
 
         # Get the hash for each file in the archive.
         archive_hashes.update(
-            **dict(
-                reversed(line.split(' ', 1))
-                for line in borgmatic.borg.list.capture_archive_listing(
+            **{
+                entry['path']: entry['xxh64']
+                for entry in borgmatic.borg.list.capture_archive_listing(
                     repository['path'],
                     archive,
                     config,
                     local_borg_version,
                     global_arguments,
                     list_paths=source_sample_paths_subset,
-                    path_format='{xxh64} {path}{NUL}',
+                    path_format='{xxh64}{path}',
                     local_path=local_path,
                     remote_path=remote_path,
                 )
-                if line
-            ),
+                if entry
+            },
         )
 
     # Compare the source hashes with the archive hashes to see how many match.
@@ -602,8 +613,6 @@ def spot_check(
     disk to those stored in the latest archive. If any differences are beyond configured tolerances,
     then the check fails.
     '''
-    logger.debug('Running spot check')
-
     try:
         spot_check_config = next(
             check for check in config.get('checks', ()) if check.get('name') == 'spot'
@@ -788,6 +797,7 @@ def run_check(
             write_check_time(make_check_time_path(config, repository_id, check, archives_check_id))
 
     if 'extract' in checks:
+        logger.info('Running extract check')
         borgmatic.borg.extract.extract_last_archive_dry_run(
             config,
             local_borg_version,
@@ -800,6 +810,7 @@ def run_check(
         write_check_time(make_check_time_path(config, repository_id, 'extract'))
 
     if 'spot' in checks:
+        logger.info('Running spot check')
         with borgmatic.config.paths.Runtime_directory(config) as borgmatic_runtime_directory:
             spot_check(
                 repository,

borgmatic/borg/list.py

@@ -1,5 +1,6 @@
 import argparse
 import copy
+import json
 import logging
 import re
 import shlex
@@ -19,6 +20,7 @@ MAKE_FLAGS_EXCLUDES = (
     'paths',
     'find_paths',
     'format',
+    'json',
     *ARCHIVE_FILTER_FLAGS_MOVED_TO_REPO_LIST,
 )
 
@@ -54,6 +56,7 @@ def make_list_command(
         + flags.make_flags('remote-path', remote_path)
         + flags.make_flags('umask', config.get('umask'))
         + ('--log-json',)
+        + (('--json-lines',) if list_arguments.json else ())
         + flags.make_flags('lock-wait', config.get('lock_wait'))
         + flags.make_flags('format', list_arguments.format or config.get('file_list_format'))
         + flags.make_flags_from_arguments(list_arguments, excludes=MAKE_FLAGS_EXCLUDES)
@@ -109,39 +112,36 @@ def capture_archive_listing(
     remote_path=None,
 ):
     '''
-    Given a local or remote repository path, an archive name, a configuration
-    dict, the local Borg version, global arguments as an argparse.Namespace,
-    the archive paths (or Borg patterns) in which to list files, the Borg path
-    format to use for the output, and local and remote Borg paths, capture the
-    output of listing that archive and return it as a list of file paths.
+    Given a local or remote repository path, an archive name, a configuration dict, the local Borg
+    version, global arguments as an argparse.Namespace, the archive paths (or Borg patterns) in
+    which to list files, the Borg path format indicating keys to include in the output, and local
+    and remote Borg paths, capture the output of listing that archive and return it as a sequence of
+    dicts, one per path.
     '''
     return tuple(
-        ''.join(
-            execute_command_and_capture_output(
-                make_list_command(
-                    repository_path,
-                    config,
-                    local_borg_version,
-                    argparse.Namespace(
-                        repository=repository_path,
-                        archive=archive,
-                        paths=list(list_paths) if list_paths else None,
-                        find_paths=None,
-                        json=None,
-                        format=path_format or '{path}{NUL}',
-                    ),
-                    global_arguments,
-                    local_path,
-                    remote_path,
+        json.loads(entry)
+        for entry in execute_command_and_capture_output(
+            make_list_command(
+                repository_path,
+                config,
+                local_borg_version,
+                argparse.Namespace(
+                    repository=repository_path,
+                    archive=archive,
+                    paths=list(list_paths) if list_paths else None,
+                    find_paths=None,
+                    json=True,
+                    format=path_format or None,
                 ),
-                environment=environment.make_environment(config),
-                working_directory=borgmatic.config.paths.get_working_directory(config),
-                borg_local_path=local_path,
-                borg_exit_codes=config.get('borg_exit_codes'),
-            )
+                global_arguments,
+                local_path,
+                remote_path,
+            ),
+            environment=environment.make_environment(config),
+            working_directory=borgmatic.config.paths.get_working_directory(config),
+            borg_local_path=local_path,
+            borg_exit_codes=config.get('borg_exit_codes'),
         )
-        .strip('\0')
-        .split('\0'),
     )
 
 

borgmatic/execute.py

@@ -6,6 +6,7 @@ import logging
 import select
 import subprocess
 import textwrap
+import time
 
 import borgmatic.logger
 
@@ -101,28 +102,38 @@ def output_buffers_for_process(process, exclude_stdouts):
     )
 
 
-def borg_log_line_to_record(line, log_level):
+def borg_json_log_line_to_record(line, log_level):
     '''
-    Given a single Borg log entry and a log level, return the line converted to a logging.LogRecord
-    instance.
-
-    If it can't be parsed, then fall back to making a record out of the raw line and the given log
-    level.
+    Given a single Borg "--log-json"-style log line and a log level, return the line converted to a
+    logging.LogRecord instance. Return None if the line can't be parsed as JSON.
     '''
     with contextlib.suppress(json.JSONDecodeError, TypeError, KeyError, AttributeError):
         log_data = json.loads(line)
+        log_type = log_data.get('type')
 
-        return logging.makeLogRecord(
-            dict(
-                levelno=logging._nameToLevel.get(log_data.get('levelname')),
-                created=log_data.get('time'),
-                msg=log_data.get('message'),
-                levelname=log_data.get('levelname'),
-                name=log_data.get('name'),
+        if log_type == 'log_message':
+            return logging.makeLogRecord(
+                dict(
+                    levelno=logging._nameToLevel.get(log_data.get('levelname')),
+                    created=log_data.get('time'),
+                    msg=log_data.get('message'),
+                    levelname=log_data.get('levelname'),
+                    name=log_data.get('name'),
+                )
             )
-        )
 
-    return log_line_to_record(line, log_level)
+        if log_type == 'file_status':
+            return logging.makeLogRecord(
+                dict(
+                    levelno=log_level,
+                    created=time.time(),
+                    msg=f'{log_data.get("status")} {log_data.get("path")}',
+                    levelname=logging.getLevelName(log_level),
+                    name='borg.file_status',
+                )
+            )
+
+    return None
 
 
 def log_line_to_record(line, log_level):
@@ -139,11 +150,11 @@ def log_line_to_record(line, log_level):
     )
 
 
-def parse_log_line(line, log_level, came_from_stderr, borg_local_path, command):
+def parse_log_line(line, log_level, elevate_stderr, borg_local_path, command):
     '''
-    Given a raw output line from an external program, whether this line came from stderr, the Borg
-    local path, and the command as a sequence, return a logging.LogRecord instance containing its
-    parsed data.
+    Given a raw output line from an external program, whether this line came from stderr and should
+    be elevated to error/warning, the Borg local path, and the command as a sequence, return a
+    logging.LogRecord instance containing its parsed data.
 
     If the command being run is Borg, and the log line is JSON-formatted log data, then grab the log
     level from it and log the parsed JSON to be consumed later by a Python logging.Formatter.
@@ -153,9 +164,12 @@ def parse_log_line(line, log_level, came_from_stderr, borg_local_path, command):
     just elevate the log level to a WARN.
     '''
     if borg_local_path and command[0] == borg_local_path:
-        return borg_log_line_to_record(line, log_level)
+        log_record = borg_json_log_line_to_record(line, log_level)
 
-    if came_from_stderr:
+        if log_record:
+            return log_record
+
+    if elevate_stderr:
         return log_line_to_record(
             line, logging.WARNING if line.lower().startswith('warning:') else logging.ERROR
         )
@@ -182,7 +196,56 @@ def handle_log_record(log_record, last_lines):
     return log_record
 
 
-def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, borg_exit_codes):  # noqa: PLR0912
+READ_CHUNK_SIZE = 4096
+
+
+def read_lines(buffer, process, line_separator='\n'):
+    '''
+    Given a Python buffer (like stdout) ready for reading, its process, and a line separator, yield
+    each (decoded) line from the buffer until the process has exited.
+
+    It is assumed that this function's generator is used in conjunction with an external select()
+    call to know when to read more lines. Otherwise, the generator will busywait if it's called in a
+    tight loop.
+    '''
+    data = ''
+
+    while True:
+        chunk = buffer.read(READ_CHUNK_SIZE).decode()
+
+        if not chunk:  # EOF
+            # The process is still running, so we keep running too.
+            if process.poll() is None:
+                continue
+
+            break
+
+        data += chunk
+
+        while True:
+            separator_position = data.find(line_separator)
+
+            if separator_position == -1:
+                break
+
+            yield data[:separator_position].rstrip()
+
+            data = data[separator_position + 1:]
+
+    # Yield any leftover data from the end of the buffer.
+    if data:
+        yield data.rstrip()
+
+
+
+def log_outputs(
+    processes,
+    exclude_stdouts,
+    output_log_level,
+    borg_local_path,
+    borg_exit_codes,
+    capture_stderr=False,
+):  # noqa: PLR0912
     '''
     Given a sequence of subprocess.Popen() instances for multiple processes, log the outputs (stderr
     and stdout). Use the requested output log level for stdout, but always log stderr to the ERROR
@@ -190,8 +253,8 @@ def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, b
     warning for exit code 1, if that process does not match the Borg local path).
 
     If the output log level is None, then instead of logging, capture the output for the last
-    process given and yield it one line at a time. But if the output log level is not None, don't
-    yield anything.
+    process given and yield it one line at a time. This includes stderr if capture stderr is set.
+    But if the output log level is not None, don't yield anything.
 
     This yielding means that this function is a generator, and must be consumed in order to execute.
 
@@ -208,6 +271,12 @@ def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, b
         if process.stdout or process.stderr
         for buffer in output_buffers_for_process(process, exclude_stdouts)
     }
+    reader_for_output_buffer = {
+        buffer: read_lines(buffer, process)
+        for process in processes
+        if process.stdout or process.stderr
+        for buffer in output_buffers_for_process(process, exclude_stdouts)
+    }
     output_buffers = list(process_for_output_buffer.keys())
     process_to_capture = processes[-1]
     still_running = True
@@ -233,8 +302,7 @@ def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, b
                             # Add the process's output to output_buffers to ensure it'll get read.
                             output_buffers.append(other_process.stdout)
 
-                while True:
-                    line = ready_buffer.readline().rstrip().decode()
+                for line in reader_for_output_buffer[ready_buffer]:
                     if not line or not ready_process:
                         break
 
@@ -250,7 +318,9 @@ def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, b
                         parse_log_line(
                             line=line,
                             log_level=output_log_level,
-                            came_from_stderr=(ready_buffer == ready_process.stderr),
+                            elevate_stderr=(
+                                ready_buffer == ready_process.stderr and not capture_stderr
+                            ),
                             borg_local_path=borg_local_path,
                             command=command,
                         ),
@@ -284,23 +354,25 @@ def log_outputs(processes, exclude_stdouts, output_log_level, borg_local_path, b
                 for output_buffer in output_buffers_for_process(process, exclude_stdouts):
                     # Collect any straggling output lines that came in since we last gathered output.
                     while output_buffer:  # pragma: no cover
-                        line = output_buffer.readline().rstrip().decode()
-                        if not line:
-                            break
+                        for line in reader_for_output_buffer[output_buffer]:
+                            if not line:
+                                break
 
-                        log_record = handle_log_record(
-                            parse_log_line(
-                                line=line,
-                                log_level=output_log_level,
-                                came_from_stderr=(output_buffer == process.stderr),
-                                borg_local_path=borg_local_path,
-                                command=command,
-                            ),
-                            last_lines=last_lines,
-                        )
+                            log_record = handle_log_record(
+                                parse_log_line(
+                                    line=line,
+                                    log_level=output_log_level,
+                                    elevate_stderr=(
+                                        output_buffer == process.stderr and not capture_stderr
+                                    ),
+                                    borg_local_path=borg_local_path,
+                                    command=command,
+                                ),
+                                last_lines=last_lines,
+                            )
 
-                        if log_record.levelno is None and process == process_to_capture:
-                            yield log_record.getMessage()
+                            if log_record.levelno is None and process == process_to_capture:
+                                yield log_record.getMessage()
 
                 if len(last_lines) == ERROR_OUTPUT_MAX_LINE_COUNT:
                     last_lines.insert(0, '...')
@@ -496,6 +568,7 @@ def execute_command_and_capture_output(
             None,
             borg_local_path,
             borg_exit_codes,
+            capture_stderr=capture_stderr,
         )
 
     yield from captured_lines