Merge pull request #2527 from enkore/issue/2241.2

Fix remote logging and progress

src/borg/archiver.py

@@ -3815,15 +3815,15 @@ def _setup_implied_logging(self, args):
         """ turn on INFO level logging for args that imply that they will produce output """
         # map of option name to name of logger for that option
         option_logger = {
-                'output_list': 'borg.output.list',
-                'show_version': 'borg.output.show-version',
-                'show_rc': 'borg.output.show-rc',
-                'stats': 'borg.output.stats',
-                'progress': 'borg.output.progress',
-                }
+            'output_list': 'borg.output.list',
+            'show_version': 'borg.output.show-version',
+            'show_rc': 'borg.output.show-rc',
+            'stats': 'borg.output.stats',
+            'progress': 'borg.output.progress',
+        }
         for option, logger_name in option_logger.items():
-            if args.get(option, False):
-                logging.getLogger(logger_name).setLevel('INFO')
+            option_set = args.get(option, False)
+            logging.getLogger(logger_name).setLevel('INFO' if option_set else 'WARN')
 
     def _setup_topic_debugging(self, args):
         """Turn on DEBUG level logging for specified --debug-topics."""
@@ -3839,8 +3839,10 @@ def run(self, args):
         # This works around http://bugs.python.org/issue9351
         func = getattr(args, 'func', None) or getattr(args, 'fallback_func')
         # do not use loggers before this!
-        setup_logging(level=args.log_level, is_serve=func == self.do_serve, json=args.log_json)
+        is_serve = func == self.do_serve
+        setup_logging(level=args.log_level, is_serve=is_serve, json=args.log_json)
         self.log_json = args.log_json
+        args.progress |= is_serve
         self._setup_implied_logging(vars(args))
         self._setup_topic_debugging(args)
         if args.show_version:

src/borg/helpers.py

@@ -1226,6 +1226,14 @@ def __init__(self, msgid=None):
             if self.logger.level == logging.NOTSET:
                 self.logger.setLevel(logging.WARN)
             self.logger.propagate = False
+
+        # If --progress is not set then the progress logger level will be WARN
+        # due to setup_implied_logging (it may be NOTSET with a logging config file,
+        # but the interactions there are generally unclear), so self.emit becomes
+        # False, which is correct.
+        # If --progress is set then the level will be INFO as per setup_implied_logging;
+        # note that this is always the case for serve processes due to a "args.progress |= is_serve".
+        # In this case self.emit is True.
         self.emit = self.logger.getEffectiveLevel() == logging.INFO
 
     def __del__(self):

src/borg/logger.py

@@ -88,15 +88,21 @@ def setup_logging(stream=None, conf_fname=None, env_var='BORG_LOGGING_CONF', lev
     # if we did not / not successfully load a logging configuration, fallback to this:
     logger = logging.getLogger('')
     handler = logging.StreamHandler(stream)
-    if is_serve:
+    if is_serve and not json:
         fmt = '$LOG %(levelname)s %(name)s Remote: %(message)s'
     else:
         fmt = '%(message)s'
-    formatter = JsonFormatter(fmt) if json and not is_serve else logging.Formatter(fmt)
+    formatter = JsonFormatter(fmt) if json else logging.Formatter(fmt)
     handler.setFormatter(formatter)
     borg_logger = logging.getLogger('borg')
     borg_logger.formatter = formatter
     borg_logger.json = json
+    if configured and logger.handlers:
+        # The RepositoryServer can call setup_logging a second time to adjust the output
+        # mode from text-ish is_serve to json is_serve.
+        # Thus, remove the previously installed handler, if any.
+        logger.handlers[0].close()
+        logger.handlers.clear()
     logger.addHandler(handler)
     logger.setLevel(level.upper())
     configured = True
@@ -224,6 +230,8 @@ def format(self, record):
         data = {
             'type': 'log_message',
             'time': record.created,
+            'message': '',
+            'levelname': 'CRITICAL',
         }
         for attr in self.RECORD_ATTRIBUTES:
             value = getattr(record, attr, None)

src/borg/remote.py

@@ -2,29 +2,30 @@
 import fcntl
 import functools
 import inspect
+import json
 import logging
 import os
 import select
 import shlex
 import sys
 import tempfile
-import traceback
 import textwrap
 import time
+import traceback
 from subprocess import Popen, PIPE
 
 import msgpack
 
 from . import __version__
 from .helpers import Error, IntegrityError
-from .helpers import get_home_dir
-from .helpers import sysinfo
 from .helpers import bin_to_hex
-from .helpers import replace_placeholders
+from .helpers import get_home_dir
 from .helpers import hostname_is_unique
+from .helpers import replace_placeholders
+from .helpers import sysinfo
+from .logger import create_logger, setup_logging
 from .repository import Repository, MAX_OBJECT_SIZE, LIST_SCAN_LIMIT
 from .version import parse_version, format_version
-from .logger import create_logger
 
 logger = create_logger(__name__)
 
@@ -312,6 +313,10 @@ def negotiate(self, client_data):
         # clients since 1.1.0b3 use a dict as client_data
         if isinstance(client_data, dict):
             self.client_version = client_data[b'client_version']
+            if client_data.get(b'client_supports_log_v3', False):
+                level = logging.getLevelName(logging.getLogger('').level)
+                setup_logging(is_serve=True, json=True, level=level)
+                logger.debug('Initialized logging system for new (v3) protocol')
         else:
             self.client_version = BORG_VERSION  # seems to be newer than current version (no known old format)
 
@@ -555,7 +560,10 @@ def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock
 
         try:
             try:
-                version = self.call('negotiate', {'client_data': {b'client_version': BORG_VERSION}})
+                version = self.call('negotiate', {'client_data': {
+                    b'client_version': BORG_VERSION,
+                    b'client_supports_log_v3': True,
+                }})
             except ConnectionClosed:
                 raise ConnectionClosedWithHint('Is borg working on the server?') from None
             if version == RPC_PROTOCOL_VERSION:
@@ -646,12 +654,23 @@ def borg_cmd(self, args, testing):
                 opts.append('--critical')
             else:
                 raise ValueError('log level missing, fix this code')
-            try:
-                borg_logger = logging.getLogger('borg')
-                if borg_logger.json:
-                    opts.append('--log-json')
-            except AttributeError:
-                pass
+
+            # Tell the remote server about debug topics it may need to consider.
+            # Note that debug topics are usable for "spew" or "trace" logs which would
+            # be too plentiful to transfer for normal use, so the server doesn't send
+            # them unless explicitly enabled.
+            #
+            # Needless to say, if you do --debug-topic=repository.compaction, for example,
+            # with a 1.0.x server it won't work, because the server does not recognize the
+            # option.
+            #
+            # This is not considered a problem, since this is a debugging feature that
+            # should not be used for regular use.
+            for topic in args.debug_topics:
+                if '.' not in topic:
+                    topic = 'borg.debug.' + topic
+                if 'repository' in topic:
+                    opts.append('--debug-topic=%s' % topic)
         env_vars = []
         if not hostname_is_unique():
             env_vars.append('BORG_HOSTNAME_IS_UNIQUE=no')
@@ -930,7 +949,63 @@ def preload(self, ids):
 
 
 def handle_remote_line(line):
-    if line.startswith('$LOG '):
+    """
+    Handle a remote log line.
+
+    This function is remarkably complex because it handles multiple wire formats.
+    """
+    if line.startswith('{'):
+        # This format is used by Borg since 1.1.0b6 for new-protocol clients.
+        # It is the same format that is exposed by --log-json.
+        msg = json.loads(line)
+
+        if msg['type'] not in ('progress_message', 'progress_percent', 'log_message'):
+            logger.warning('Dropped remote log message with unknown type %r: %s', msg['type'], line)
+            return
+
+        if msg['type'] == 'log_message':
+            # Re-emit log messages on the same level as the remote to get correct log suppression and verbosity.
+            level = getattr(logging, msg['levelname'], logging.CRITICAL)
+            assert isinstance(level, int)
+            target_logger = logging.getLogger(msg['name'])
+            msg['message'] = 'Remote: ' + msg['message']
+            # In JSON mode, we manually check whether the log message should be propagated.
+            if logging.getLogger('borg').json and level >= target_logger.getEffectiveLevel():
+                sys.stderr.write(json.dumps(msg) + '\n')
+            else:
+                target_logger.log(level, '%s', msg['message'])
+        elif msg['type'].startswith('progress_'):
+            # Progress messages are a bit more complex.
+            # First of all, we check whether progress output is enabled. This is signalled
+            # through the effective level of the borg.output.progress logger
+            # (also see ProgressIndicatorBase in borg.helpers).
+            progress_logger = logging.getLogger('borg.output.progress')
+            if progress_logger.getEffectiveLevel() == logging.INFO:
+                # When progress output is enabled, we check whether the client is in
+                # --log-json mode, as signalled by the "json" attribute on the "borg" logger.
+                if logging.getLogger('borg').json:
+                    # In --log-json mode we re-emit the progress JSON line as sent by the server,
+                    # with the message, if any, prefixed with "Remote: ".
+                    if 'message' in msg:
+                        msg['message'] = 'Remote: ' + msg['message']
+                    sys.stderr.write(json.dumps(msg) + '\n')
+                elif 'message' in msg:
+                    # In text log mode we write only the message to stderr and terminate with \r
+                    # (carriage return, i.e. move the write cursor back to the beginning of the line)
+                    # so that the next message, progress or not, overwrites it. This mirrors the behaviour
+                    # of local progress displays.
+                    sys.stderr.write('Remote: ' + msg['message'] + '\r')
+    elif line.startswith('$LOG '):
+        # This format is used by Borg since 1.1.0b1.
+        # It prefixed log lines with $LOG as a marker, followed by the log level
+        # and optionally a logger name, then "Remote:" as a separator followed by the original
+        # message.
+        #
+        # It is the oldest format supported by these servers, so it was important to make
+        # it readable with older (1.0.x) clients.
+        #
+        # TODO: Remove this block (so it'll be handled by the "else:" below) with a Borg 1.1 RC.
+        #       Also check whether client_supports_log_v3 should be removed.
         _, level, msg = line.split(' ', 2)
         level = getattr(logging, level, logging.CRITICAL)  # str -> int
         if msg.startswith('Remote:'):
@@ -941,7 +1016,15 @@ def handle_remote_line(line):
             logname, msg = msg.split(' ', 1)
             logging.getLogger(logname).log(level, msg.rstrip())
     else:
-        sys.stderr.write('Remote: ' + line)
+        # Plain 1.0.x and older format - re-emit to stderr (mirroring what the 1.0.x
+        # client did) or as a generic log message.
+        # We don't know what priority the line had.
+        if logging.getLogger('borg').json:
+            logging.getLogger('').warning('Remote: ' + line.strip())
+        else:
+            # In non-JSON mode we circumvent logging to preserve carriage returns (\r)
+            # which are generated by remote progress displays.
+            sys.stderr.write('Remote: ' + line)
 
 
 class RepositoryNoCache:

src/borg/testsuite/repository.py

@@ -714,6 +714,7 @@ def test_borg_cmd(self):
         class MockArgs:
             remote_path = 'borg'
             umask = 0o077
+            debug_topics = []
 
         assert self.repository.borg_cmd(None, testing=True) == [sys.executable, '-m', 'borg.archiver', 'serve']
         args = MockArgs()
@@ -723,6 +724,9 @@ class MockArgs:
         assert self.repository.borg_cmd(args, testing=False) == ['borg', 'serve', '--umask=077', '--info']
         args.remote_path = 'borg-0.28.2'
         assert self.repository.borg_cmd(args, testing=False) == ['borg-0.28.2', 'serve', '--umask=077', '--info']
+        args.debug_topics = ['something_client_side', 'repository_compaction']
+        assert self.repository.borg_cmd(args, testing=False) == ['borg-0.28.2', 'serve', '--umask=077', '--info',
+                                                                 '--debug-topic=borg.debug.repository_compaction']
 
 
 class RemoteLegacyFree(RepositoryTestCaseBase):