From e8d5bd43768d1f69cc33eab04a69969932d1c615 Mon Sep 17 00:00:00 2001 From: Mikhail Golubev Date: Fri, 15 Nov 2019 22:47:29 +0300 Subject: [PATCH] Properly re-configure logging in a subprocess during skeleton generation On platforms where "fork" is not available or not used by multiprocessing root logger configuration is not automatically inherited and, thus, we loose messages sent by worker processes. So as not to depend on particular implementation of the module, e.g. by checking sys.platform or multiprocessing.get_start_method(), we now explicitly reset and setup logging anew in every created process. GitOrigin-RevId: 572b5a1a77346f46421a6dd1e0ed0b1ed9c4ca6f --- python/helpers/generator3/__main__.py | 18 +------- python/helpers/generator3/util_methods.py | 46 +++++++++++++++++-- .../tests/generator3_tests/test_generation.py | 7 +++ 3 files changed, 52 insertions(+), 19 deletions(-) diff --git a/python/helpers/generator3/__main__.py b/python/helpers/generator3/__main__.py index 5ed467bc84e5..d047732f9706 100644 --- a/python/helpers/generator3/__main__.py +++ b/python/helpers/generator3/__main__.py @@ -18,22 +18,8 @@ def _bootstrap_sys_path(): def _setup_logging(): - logging.addLevelName(logging.DEBUG - 1, 'TRACE') - - class JsonFormatter(logging.Formatter): - def format(self, record): - s = super(JsonFormatter, self).format(record) - return json.dumps({ - 'type': 'log', - 'level': record.levelname.lower(), - 'message': s - }) - - root = logging.getLogger() - root.setLevel(logging.DEBUG) - handler = logging.StreamHandler(sys.stdout) - handler.setFormatter(JsonFormatter()) - root.addHandler(handler) + from generator3.util_methods import configure_logging + configure_logging(logging.DEBUG) def _enable_segfault_tracebacks(): diff --git a/python/helpers/generator3/util_methods.py b/python/helpers/generator3/util_methods.py index 612c093531ba..0c1e6d8f198a 100644 --- a/python/helpers/generator3/util_methods.py +++ b/python/helpers/generator3/util_methods.py @@ -1,8 +1,12 @@ import ast +import collections import errno import functools import hashlib +import json import keyword +import logging +import multiprocessing import shutil from contextlib import contextmanager @@ -854,8 +858,12 @@ _bytes_that_never_appears_in_text = set(range(7)) | {11} | set(range(14, 27)) | # This wrapper is intentionally made top-level: local functions can't be pickled. -def _multiprocessing_wrapper(result_conn, func, *args, **kwargs): - result_conn.send(func(*args, **kwargs)) +def _multiprocessing_wrapper(data, func, *args, **kwargs): + configure_logging(data.root_logger_level) + data.result_conn.send(func(*args, **kwargs)) + + +_MainProcessData = collections.namedtuple('_MainProcessData', ['result_conn', 'root_logger_level']) def execute_in_subprocess_synchronously(name, func, args, kwargs, failure_result=None): @@ -870,9 +878,11 @@ def execute_in_subprocess_synchronously(name, func, args, kwargs, failure_result # TODO experiment with a shared queue maintained by multiprocessing.Manager # (it will require an additional service process) recv_conn, send_conn = mp.Pipe(duplex=False) + data = _MainProcessData(result_conn=send_conn, + root_logger_level=logging.getLogger().level) p = mp.Process(name=name, target=_multiprocessing_wrapper, - args=(send_conn, func) + args, + args=(data, func) + args, kwargs=kwargs, **extra_process_kwargs) p.start() @@ -886,3 +896,33 @@ def execute_in_subprocess_synchronously(name, func, args, kwargs, failure_result return recv_conn.recv() else: return failure_result + + +def configure_logging(root_level): + logging.addLevelName(logging.DEBUG - 1, 'TRACE') + + root = logging.getLogger() + root.setLevel(root_level) + + # In environments where fork is implemented entire logging configuration is already inherited by child processes. + # Configuring it twice will lead to duplicated records. + + # Reset logger similarly to how it's done in logging.config + for h in root.handlers[:]: + root.removeHandler(h) + + for f in root.filters[:]: + root.removeFilter(f) + + class JsonFormatter(logging.Formatter): + def format(self, record): + s = super(JsonFormatter, self).format(record) + return json.dumps({ + 'type': 'log', + 'level': record.levelname.lower(), + 'message': s + }) + + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(JsonFormatter()) + root.addHandler(handler) diff --git a/python/helpers/tests/generator3_tests/test_generation.py b/python/helpers/tests/generator3_tests/test_generation.py index 0930e1ef20af..1352a5bed798 100644 --- a/python/helpers/tests/generator3_tests/test_generation.py +++ b/python/helpers/tests/generator3_tests/test_generation.py @@ -438,6 +438,13 @@ class MultiModuleGenerationTest(FunctionalGeneratorTestCase): def test_general_results_and_layout(self): self.check_generator_output() + @test_data_dir('simple') + def test_logging_configured_and_propagates_from_worker_subprocess(self): + result = self.run_generator() + log_messages = [m['message'] for m in result.control_messages if m['type'] == 'log'] + subprocess_messages = [m for m in log_messages if m.startswith('Updating cache for mod')] + self.assertEquals(2, len(subprocess_messages)) + class StatePassingGenerationTest(FunctionalGeneratorTestCase): default_generator_extra_args = ['--state-file-policy', 'readwrite', '--name-pattern', 'mod?']