Merge branch 'log-server-new-min-py-ver' into 'master'

Log server to capture nested logging output

See merge request ungleich-public/cdist!891

cdist/__init__.py

@@ -26,6 +26,7 @@ import hashlib
 import cdist.log
 import cdist.version
 
+
 VERSION = cdist.version.VERSION
 
 BANNER = """
@@ -48,6 +49,9 @@ REMOTE_EXEC = "ssh -o User=root"
 REMOTE_CMDS_CLEANUP_PATTERN = "ssh -o User=root -O exit -S {}"
 
 
+MIN_SUPPORTED_PYTHON_VERSION = '3.5'
+
+
 class Error(Exception):
     """Base exception class for this project"""
     pass

cdist/argparse.py

@@ -206,6 +206,13 @@ def get_parsers():
                  'supported. Without argument CPU count is used by default. '),
            action='store', dest='jobs',
            const=multiprocessing.cpu_count())
+    parser['config_main'].add_argument(
+           '--log-server',
+           action='store_true',
+           help=('Start a log server for sub processes to use. '
+                 'This is mainly useful when running cdist nested '
+                 'from a code-local script. Log server is alwasy '
+                 'implicitly started for \'install\' command.'))
     parser['config_main'].add_argument(
            '-n', '--dry-run',
            help='Do not execute code.', action='store_true')

cdist/config.py

@@ -29,18 +29,20 @@ import time
 import itertools
 import tempfile
 import multiprocessing
-from cdist.mputil import mp_pool_run, mp_sig_handler
 import atexit
 import shutil
 import socket
+
+from cdist.mputil import mp_pool_run, mp_sig_handler
+from cdist import core, inventory
+from cdist.util.remoteutil import inspect_ssh_mux_opts
+
 import cdist
 import cdist.hostsource
 import cdist.exec.local
 import cdist.exec.remote
 import cdist.util.ipaddr as ipaddr
 import cdist.configuration
-from cdist import core, inventory
-from cdist.util.remoteutil import inspect_ssh_mux_opts
 
 
 def graph_check_cycle(graph):
@@ -195,7 +197,6 @@ class Config(object):
     @classmethod
     def commandline(cls, args):
         """Configure remote system"""
-
         if (args.parallel and args.parallel != 1) or args.jobs:
             if args.timestamp:
                 cdist.log.setupTimestampingParallelLogging()
@@ -382,10 +383,16 @@ class Config(object):
            If operating in parallel then return tuple (host, True|False, )
            so that main process knows for which host function was successful.
         """
-
         log = logging.getLogger(host)
 
         try:
+            if args.log_server:
+                # Start a log server so that nested `cdist config` runs
+                # have a place to send their logs to.
+                log_server_socket_dir = tempfile.mkdtemp()
+                cls._register_path_for_removal(log_server_socket_dir)
+                cdist.log.setupLogServer(log_server_socket_dir, log)
+
             remote_exec, remote_copy, cleanup_cmd = cls._resolve_remote_cmds(
                 args)
             log.debug("remote_exec for host \"{}\": {}".format(

cdist/core/code.py

@@ -116,6 +116,10 @@ class Code(object):
         if dry_run:
             self.env['__cdist_dry_run'] = '1'
 
+        if '__cdist_log_server_socket_export' in os.environ:
+            self.env['__cdist_log_server_socket'] = os.environ[
+                '__cdist_log_server_socket_export']
+
     def _run_gencode(self, cdist_object, which):
         cdist_type = cdist_object.cdist_type
         script = os.path.join(self.local.type_path,

cdist/install.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
-# 2013 Steven Armstrong (steven-cdist at armstrong.cc)
+# 2013-2019 Steven Armstrong (steven-cdist at armstrong.cc)
 #
 # This file is part of cdist.
 #
@@ -22,9 +22,21 @@
 
 import cdist.config
 import cdist.core
+import cdist.log
 
 
 class Install(cdist.config.Config):
+
+    @classmethod
+    def onehost(cls, host, host_tags, host_base_path, host_dir_name, args,
+                parallel, configuration, remove_remote_files_dirs=False):
+        # Always start log server during cdist install so that nested
+        # `cdist config` runs have a place to send their logs to.
+        args.log_server = True
+
+        super().onehost(host, host_tags, host_base_path, host_dir_name, args,
+                        parallel, configuration, remove_remote_files_dirs)
+
     def object_list(self):
         """Short name for object list retrieval.
         In install mode, we only care about install objects.

cdist/log.py

@@ -2,6 +2,7 @@
 # -*- coding: utf-8 -*-
 #
 # 2010-2013 Nico Schottelius (nico-cdist at schottelius.org)
+# 2019-2020 Steven Armstrong
 #
 # This file is part of cdist.
 #
@@ -20,9 +21,16 @@
 #
 #
 
-import logging
-import sys
 import datetime
+import logging
+import logging.handlers
+import sys
+import os
+import asyncio
+import contextlib
+import pickle
+import struct
+import threading
 
 
 # Define additional cdist logging levels.
@@ -89,6 +97,12 @@ class DefaultLog(logging.Logger):
         super().__init__(name)
         self.propagate = False
 
+        if '__cdist_log_server_socket' in os.environ:
+            log_server_socket = os.environ['__cdist_log_server_socket']
+            socket_handler = logging.handlers.SocketHandler(log_server_socket,
+                                                            None)
+            self.addHandler(socket_handler)
+        else:
             formatter = CdistFormatter(self.FORMAT)
 
             stdout_handler = logging.StreamHandler(sys.stdout)
@@ -152,4 +166,42 @@ def setupParallelLogging():
     logging.setLoggerClass(ParallelLog)
 
 
+async def handle_log_client(reader, writer):
+    while True:
+        chunk = await reader.read(4)
+        if len(chunk) < 4:
+            return
+
+        data_size = struct.unpack('>L', chunk)[0]
+        data = await reader.read(data_size)
+
+        obj = pickle.loads(data)
+        record = logging.makeLogRecord(obj)
+        logger = logging.getLogger(record.name)
+        logger.handle(record)
+
+
+def run_log_server(server_address):
+    # Get a new loop inside the current thread to run the log server.
+    loop = asyncio.new_event_loop()
+    loop.create_task(asyncio.start_unix_server(handle_log_client,
+                                               server_address))
+    loop.run_forever()
+
+
+def setupLogServer(socket_dir, log=logging.getLogger(__name__)):
+    """Run a asyncio based unix socket log server in a background thread.
+    """
+    log_server_socket = os.path.join(socket_dir, 'log-server')
+    log.debug('Starting logging server on: %s', log_server_socket)
+    os.environ['__cdist_log_server_socket_export'] = log_server_socket
+    with contextlib.suppress(FileNotFoundError):
+        os.remove(log_server_socket)
+    t = threading.Thread(target=run_log_server, args=(log_server_socket,))
+    # Deamonizing the thread means we don't have to care about stoping it.
+    # It will die together with the main process.
+    t.daemon = True
+    t.start()
+
+
 setupDefaultLogging()

docs/src/man1/cdist.rst

@@ -17,20 +17,20 @@ SYNOPSIS
 
     cdist config [-h] [-l LOGLEVEL] [-q] [-v] [-b] [--colors WHEN]
                  [-g CONFIG_FILE] [-4] [-6] [-C CACHE_PATH_PATTERN]
-                 [-c CONF_DIR] [-i MANIFEST] [-j [JOBS]] [-n] [-o OUT_PATH] [-P]
+                 [-c CONF_DIR] [-i MANIFEST] [-j [JOBS]] [--log-server]
-                 [-R [{tar,tgz,tbz2,txz}]] [-r REMOTE_OUT_PATH]
+                 [-n] [-o OUT_PATH] [-P] [-R [{tar,tgz,tbz2,txz}]]
-                 [--remote-copy REMOTE_COPY] [--remote-exec REMOTE_EXEC]
+                 [-r REMOTE_OUT_PATH] [--remote-copy REMOTE_COPY]
-                 [-S] [-I INVENTORY_DIR] [-A] [-a] [-f HOSTFILE]
+                 [--remote-exec REMOTE_EXEC] [-S] [-I INVENTORY_DIR] [-A]
-                 [-p [HOST_MAX]] [-s] [-t]
+                 [-a] [-f HOSTFILE] [-p [HOST_MAX]] [-s] [-t]
                  [host [host ...]]
 
     cdist install [-h] [-l LOGLEVEL] [-q] [-v] [-b] [--colors WHEN]
                   [-g CONFIG_FILE] [-4] [-6] [-C CACHE_PATH_PATTERN]
-                  [-c CONF_DIR] [-i MANIFEST] [-j [JOBS]] [-n] [-o OUT_PATH]
+                  [-c CONF_DIR] [-i MANIFEST] [-j [JOBS]] [--log-server]
-                  [-P] [-R [{tar,tgz,tbz2,txz}]] [-r REMOTE_OUT_PATH]
+                  [-n] [-o OUT_PATH] [-P] [-R [{tar,tgz,tbz2,txz}]]
-                  [--remote-copy REMOTE_COPY] [--remote-exec REMOTE_EXEC]
+                  [-r REMOTE_OUT_PATH] [--remote-copy REMOTE_COPY]
-                  [-S] [-I INVENTORY_DIR] [-A] [-a] [-f HOSTFILE]
+                  [--remote-exec REMOTE_EXEC] [-S] [-I INVENTORY_DIR] [-A]
-                  [-p [HOST_MAX]] [-s] [-t]
+                  [-a] [-f HOSTFILE] [-p [HOST_MAX]] [-s] [-t]
                   [host [host ...]]
 
     cdist inventory [-h] {add-host,add-tag,del-host,del-tag,list} ...
@@ -202,6 +202,12 @@ Install command is currently in beta.
     are supported. Without argument CPU count is used by
     default.
 
+**--log-server**
+    Start a log server for sub processes to use. This is
+    mainly useful when running cdist nested from a code-
+    local script. Log server is always implicitly started
+    for 'install' command.
+
 **-n, --dry-run**
     Do not execute code.
 

scripts/cdist

@@ -60,10 +60,9 @@ def commandline():
 
 
 if __name__ == "__main__":
-    cdistpythonversion = '3.5'
+    if sys.version < cdist.MIN_SUPPORTED_PYTHON_VERSION:
-    if sys.version < cdistpythonversion:
         print('Python >= {} is required on the source host.'.format(
-                cdistpythonversion), file=sys.stderr)
+                cdist.MIN_SUPPORTED_PYTHON_VERSIO), file=sys.stderr)
         sys.exit(1)
 
     exit_code = 0