Repository: exasol/integration-test-docker-environment (language: python)

Issue:
Using run_task with log_level duplicates log entries
# Background

- After n calls to run_task with log_level, the log entries get printed n times
- The reason is that each call to run_task with log_level calls luigi.build with log_level and because we disable each time the initialized flag of luigi's config setup, it add a handler to the luigi logger


Discussion / hints:

Fix patch:
diff --git a/doc/changes/changes_1.6.0.md b/doc/changes/changes_1.6.0.md
index e179a80b5..fd1abde54 100644
--- a/doc/changes/changes_1.6.0.md
+++ b/doc/changes/changes_1.6.0.md
@@ -20,4 +20,5 @@ If you need further versions, please open an issue.
  - #294: Fixed that the client logging config gets overwritten
  - #295: Fixed that the test_container_content parameter pollutes the logs
  - #300: Fixed BucketFS uploads during reusing of an environment
- - #291: Fixed that test container gets rebuild when force_rebuild is true
\ No newline at end of file
+ - #310: Fixed that using run_task with log_level duplicates log entries
+ - #291: Fixed that test container gets rebuild when force_rebuild is true


Test patch:
diff --git a/exasol_integration_test_docker_environment/lib/api/common.py b/exasol_integration_test_docker_environment/lib/api/common.py
index 4e123fcf5..6ab9c9385 100644
--- a/exasol_integration_test_docker_environment/lib/api/common.py
+++ b/exasol_integration_test_docker_environment/lib/api/common.py
@@ -173,7 +173,7 @@ def _configure_logging(
         log_level: Optional[str],
         use_job_specific_log_file: bool) -> Iterator[Dict[str, str]]:
     with get_luigi_log_config(log_file_target=log_file_path,
-                              console_log_level=log_level,
+                              log_level=log_level,
                               use_job_specific_log_file=use_job_specific_log_file) as luigi_config:
         no_configure_logging, run_kwargs = _configure_logging_parameter(
             log_level=log_level,
@@ -194,24 +194,13 @@ def _configure_logging(
 
 def _configure_logging_parameter(log_level: str, luigi_config: Path, use_job_specific_log_file: bool) \
         -> Tuple[bool, Dict[str, str]]:
-    if log_level is None and not use_job_specific_log_file:
-        run_kwargs = {}
-        no_configure_logging = True
-    else:
-        no_configure_logging = False
-        run_kwargs = _get_run_kwargs_log_configuration(
-            log_level, luigi_config, use_job_specific_log_file)
-    return no_configure_logging, run_kwargs
-
-
-def _get_run_kwargs_log_configuration(log_level: str, luigi_config: Path, use_job_specific_log_file: bool) \
-        -> Dict[str, str]:
     if use_job_specific_log_file:
+        no_configure_logging = False
         run_kwargs = {"logging_conf_file": f'{luigi_config}'}
     else:
-        run_kwargs = {"log_level": log_level}
-    return run_kwargs
-
+        no_configure_logging = True
+        run_kwargs = {}
+    return no_configure_logging, run_kwargs
 
 def generate_graph_from_task_dependencies(task: DependencyLoggerBaseTask, task_dependencies_dot_file: Optional[str]):
     if task_dependencies_dot_file is not None:
diff --git a/exasol_integration_test_docker_environment/lib/base/luigi_log_config.py b/exasol_integration_test_docker_environment/lib/base/luigi_log_config.py
index cf7f8a23f..962c61434 100644
--- a/exasol_integration_test_docker_environment/lib/base/luigi_log_config.py
+++ b/exasol_integration_test_docker_environment/lib/base/luigi_log_config.py
@@ -2,7 +2,7 @@
 import os
 import tempfile
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Callable
 
 import jinja2
 import logging
@@ -19,6 +19,10 @@
 
 LOG_LEVEL = "LOG_LEVEL"
 
+LUIGI_INTERFACE_LOGGER = "luigi-interface"
+
+LUIGI_LOGGER = "luigi"
+
 
 def get_log_path(job_id: str) -> Path:
     """
@@ -37,43 +41,48 @@ def get_log_path(job_id: str) -> Path:
 
 
 @contextlib.contextmanager
-def restore_root_logger(use_job_specific_log_file: bool):
-    root_logger_info = None
-    if use_job_specific_log_file:
-        root_logger_info = {
-            LOG_LEVEL: logging.root.level,
-            HANDLERS: list(logging.root.handlers),
-            FILTERS: list(logging.root.filters),
-            PROPAGATE: logging.root.propagate
-        }
+def restore_logger(logger_creator: Callable[[], logging.Logger]):
+    before_logger = logger_creator()
+    logger_info = {
+        LOG_LEVEL: before_logger.level,
+        HANDLERS: list(before_logger.handlers),
+        FILTERS: list(before_logger.filters),
+        PROPAGATE: before_logger.propagate
+    }
     yield
-    if use_job_specific_log_file and root_logger_info is not None:
-        logging.root.level = root_logger_info[LOG_LEVEL]
-        logging.root.handlers = root_logger_info[HANDLERS]
-        logging.root.filters = root_logger_info[FILTERS]
-        logging.root.propagate = root_logger_info[PROPAGATE]
+    after_logger = logger_creator()
+    after_logger.level = logger_info[LOG_LEVEL]
+    after_logger.handlers = logger_info[HANDLERS]
+    after_logger.filters = logger_info[FILTERS]
+    after_logger.propagate = logger_info[PROPAGATE]
 
 
 @contextlib.contextmanager
 def get_luigi_log_config(log_file_target: Path,
                          use_job_specific_log_file: bool,
-                         console_log_level: Optional[str] = None) -> Path:
+                         log_level: Optional[str] = None) -> Path:
     """
     Yields a context manager containing the path of the log-config file.
     log_file_target contains the location of the log-file.
     console_log_level indicates the log_level (@see https://docs.python.org/3/library/logging.html#logging-levels)
     The log-level for the log-file is always logging.DEBUG!
     """
-    if console_log_level is None:
-        console_log_level = logging.getLevelName(logging.WARNING)
+    if log_level is None and use_job_specific_log_file:
+        log_level = logging.getLevelName(logging.WARNING)
     env = jinja2.Environment(loader=jinja2.PackageLoader(PACKAGE_NAME),
                              autoescape=jinja2.select_autoescape())
     template = env.get_template("luigi_log.conf")
-    rendered_template = template.render(console_log_level=console_log_level,
+    rendered_template = template.render(console_log_level=log_level,
                                         log_file_target=str(log_file_target))
     with tempfile.TemporaryDirectory() as temp_dir:
         temp_luigi_conf_path = Path(temp_dir) / "luigi_log.conf"
         with open(temp_luigi_conf_path, "w") as f:
             f.write(rendered_template)
-        with restore_root_logger(use_job_specific_log_file=use_job_specific_log_file):
+        with restore_logger(logger_creator=lambda: logging.root), \
+                restore_logger(logger_creator=lambda: logging.getLogger(LUIGI_INTERFACE_LOGGER)), \
+                restore_logger(logger_creator=lambda: logging.getLogger(LUIGI_LOGGER)):
+            if log_level is not None and not use_job_specific_log_file:
+                logging.getLogger(LUIGI_INTERFACE_LOGGER).level = logging.getLevelName(log_level)
+                logging.getLogger(LUIGI_LOGGER).level = logging.getLevelName(log_level)
+
             yield temp_luigi_conf_path
diff --git a/exasol_integration_test_docker_environment/lib/base/task_logger_wrapper.py b/exasol_integration_test_docker_environment/lib/base/task_logger_wrapper.py
index 757ca4522..c354fe738 100644
--- a/exasol_integration_test_docker_environment/lib/base/task_logger_wrapper.py
+++ b/exasol_integration_test_docker_environment/lib/base/task_logger_wrapper.py
@@ -1,5 +1,8 @@
+import logging
+
+
 class TaskLoggerWrapper():
-    def __init__(self, logger, task_id):
+    def __init__(self, logger: logging.Logger, task_id):
         self.task_id = task_id
         self.logger = logger
 
@@ -27,4 +30,4 @@ def exception(self, msg, *args, exc_info=True, **kwargs):
     def critical(self, msg, *args, **kwargs):
         self.logger.critical(self.get_message(msg), *args, **kwargs)
 
-    fatal = critical
\ No newline at end of file
+    fatal = critical
diff --git a/exasol_integration_test_docker_environment/test/test_api_logging.py b/exasol_integration_test_docker_environment/test/test_api_logging.py
index fd26b230f..8e8ce6927 100644
--- a/exasol_integration_test_docker_environment/test/test_api_logging.py
+++ b/exasol_integration_test_docker_environment/test/test_api_logging.py
@@ -1,12 +1,11 @@
 import contextlib
-import glob
 import logging
+import sys
+import tempfile
 import unittest
 import warnings
 from contextlib import redirect_stderr
-from io import StringIO
 from pathlib import Path
-from sys import stderr
 from tempfile import TemporaryDirectory
 from typing import Any, Dict, Optional
 
@@ -38,6 +37,8 @@
 
 ROOT_LOGGER = "root"
 
+TEST_FORMAT = "TEST_FORMAT"
+
 from exasol_integration_test_docker_environment.lib.api.common import generate_root_task, run_task, \
     set_build_config
 from exasol_integration_test_docker_environment.lib.base.dependency_logger_base_task import \
@@ -51,29 +52,44 @@ def ignore_resource_warning():
     """
     with warnings.catch_warnings():
         warnings.filterwarnings(action="ignore",
-                                category=ResourceWarning,
-                                message=".*unclosed <socket.socket.*")
+                                category=ResourceWarning)
         yield
 
 
 class DummyTask(DependencyLoggerBaseTask):
 
     def run_task(self):
-        self.logger.info("DUMMY LOGGER")
+        self.logger.info("DUMMY LOGGER INFO")
+        self.logger.error("DUMMY LOGGER ERROR")
         self.return_object("DUMMY SUCCES")
 
 
+@contextlib.contextmanager
+def catch_stderr():
+    with tempfile.TemporaryFile("w+t") as temp_file:
+        try:
+            with ignore_resource_warning():
+                with redirect_stderr(temp_file) as catched_stderr:
+                    yield catched_stderr
+        finally:
+            temp_file.seek(0)
+            print(temp_file.read(), file=sys.stderr)
+
+
 class APIClientLoggingTest(unittest.TestCase):
 
     def setUp(self):
-        self.maxDiff = 30000
-        print(f"SetUp {self.__class__.__name__}", file=stderr)
+        self.old_max_length = unittest.util._MAX_LENGTH
+        unittest.util._MAX_LENGTH = 20000
+        print(f"SetUp {self.__class__.__name__}", file=sys.stderr)
         self._build_output_temp_dir = TemporaryDirectory()
         self._build_output_temp_dir.__enter__()
         self.reset_logging()
 
     def tearDown(self):
         self._build_output_temp_dir.__exit__(None, None, None)
+        self.reset_logging()
+        unittest.util._MAX_LENGTH = self.old_max_length
 
     @ignore_resource_warning()
     def dummy_api_command(self, log_level: Optional[str], use_job_specific_log_file: bool):
@@ -87,108 +103,120 @@ def dummy_api_command(self, log_level: Optional[str], use_job_specific_log_file:
                          "test")
         task_creator = lambda: generate_root_task(task_class=DummyTask)
         result = run_task(task_creator=task_creator,
-                          workers=5,
+                          workers=2,
                           task_dependencies_dot_file=None,
                           log_level=log_level,
                           use_job_specific_log_file=use_job_specific_log_file)
         return result
 
-    def test_luigi_log_level_info_and_basic_logging_error(self):
-        with redirect_stderr(StringIO()) as f:
-            logging.basicConfig(
-                format='%(asctime)s %(levelname)-8s %(message)s',
-                level=logging.ERROR,
-                datefmt='%Y-%m-%d %H:%M:%S',
-                force=True)
-            logger_infos_before = self.create_logger_infos()
-            result = self.dummy_api_command(log_level="INFO", use_job_specific_log_file=False)
-            logger_infos_after = self.create_logger_infos()
+    def configure_logging(self, log_level: int):
+        logging.basicConfig(
+            format=f'{TEST_FORMAT} %(levelname)s %(message)s',
+            level=log_level,
+            force=True)
+
+    def assert_loggers_are_equal(self, logger_infos_after, logger_infos_before):
         self.assertEqual(logger_infos_before[ROOT_LOGGER], logger_infos_after[ROOT_LOGGER])
         self.assertEqual(logger_infos_before[API_CLIENT_LOGGING_TEST_LOGGER],
                          logger_infos_after[API_CLIENT_LOGGING_TEST_LOGGER])
         self.assertEqual(logger_infos_before[LUIGI_LOGGER], logger_infos_after[LUIGI_LOGGER])
-        self.assertNotEqual(logger_infos_before[LUIGI_INTERFACE_LOGGER], logger_infos_after[LUIGI_INTERFACE_LOGGER])
-        self.assertEqual(logger_infos_after[LUIGI_INTERFACE_LOGGER][LEVEL_NAME], "INFO")
-        stderr = f.getvalue()
-        self.assertNotEqual(stderr, "")
-        self.assertRegex(stderr, ".*===== Luigi Execution Summary =====.*")
-        main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
-        self.assertEqual(main_log_glob, [])
+        self.assertEqual(logger_infos_before[LUIGI_INTERFACE_LOGGER], logger_infos_after[LUIGI_INTERFACE_LOGGER])
 
-    def test_luigi_use_job_specific_log_file_and_basic_logging_error(self):
-        with redirect_stderr(StringIO()) as f:
-            logging.basicConfig(
-                format='%(asctime)s %(levelname)-8s %(message)s',
-                level=logging.ERROR,
-                datefmt='%Y-%m-%d %H:%M:%S',
-                force=True)
+    def create_test_regex(self, log_level: int):
+        level_name = logging.getLevelName(log_level)
+        return f".*{TEST_FORMAT} {level_name} DummyTask_.* DUMMY LOGGER {level_name}.*"
+
+    def test_luigi_log_level_info_and_basic_logging_error(self):
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.ERROR)
             logger_infos_before = self.create_logger_infos()
-            result = self.dummy_api_command(log_level=None, use_job_specific_log_file=True)
+            result = self.dummy_api_command(log_level="INFO", use_job_specific_log_file=False)
             logger_infos_after = self.create_logger_infos()
-        self.assertEqual(logger_infos_before[ROOT_LOGGER], logger_infos_after[ROOT_LOGGER])
-        self.assertEqual(logger_infos_before[API_CLIENT_LOGGING_TEST_LOGGER],
-                         logger_infos_after[API_CLIENT_LOGGING_TEST_LOGGER])
-        self.assertNotEqual(logger_infos_before[LUIGI_LOGGER], logger_infos_after[LUIGI_LOGGER])
-        self.assertNotEqual(logger_infos_before[LUIGI_INTERFACE_LOGGER], logger_infos_after[LUIGI_INTERFACE_LOGGER])
-        stderr = f.getvalue()
-        self.assertEqual(stderr, "")
-        main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
-        self.assertNotEqual(main_log_glob, [])
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assertNotEqual(catched_stderr, "")
+            self.assertRegex(stderr_output, self.create_test_regex(logging.ERROR))
+            self.assertRegex(stderr_output, self.create_test_regex(logging.INFO))
+            self.assertRegex(stderr_output, ".*===== Luigi Execution Summary =====.*")
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertEqual(main_log_glob, [])
 
     def test_luigi_log_level_error_and_basic_logging_info(self):
-        with redirect_stderr(StringIO()) as f:
-            logging.basicConfig(
-                format='%(asctime)s %(levelname)-8s %(message)s',
-                level=logging.INFO,
-                datefmt='%Y-%m-%d %H:%M:%S',
-                force=True)
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.INFO)
             logger_infos_before = self.create_logger_infos()
             result = self.dummy_api_command(log_level="ERROR", use_job_specific_log_file=False)
             logger_infos_after = self.create_logger_infos()
-        self.assertEqual(logger_infos_before[ROOT_LOGGER], logger_infos_after[ROOT_LOGGER])
-        self.assertEqual(logger_infos_before[API_CLIENT_LOGGING_TEST_LOGGER],
-                         logger_infos_after[API_CLIENT_LOGGING_TEST_LOGGER])
-        self.assertEqual(logger_infos_before[LUIGI_LOGGER], logger_infos_after[LUIGI_LOGGER])
-        self.assertNotEqual(logger_infos_before[LUIGI_INTERFACE_LOGGER], logger_infos_after[LUIGI_INTERFACE_LOGGER])
-        self.assertEqual(logger_infos_after[LUIGI_INTERFACE_LOGGER][LEVEL_NAME], "ERROR")
-        stderr = f.getvalue()
-        self.assertNotEqual(stderr, "")
-        self.assertRegex(stderr, ".*logging configured by default settings")
-        main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
-        self.assertEqual(main_log_glob, [])
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assertNotEqual(catched_stderr, "")
+            self.assertRegex(stderr_output, self.create_test_regex(logging.ERROR))
+            self.assertNotRegex(stderr_output, self.create_test_regex(logging.INFO))
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertEqual(main_log_glob, [])
+
+    def test_luigi_log_level_error_multiple_calls_and_basic_logging_info(self):
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.INFO)
+            logger_infos_before = self.create_logger_infos()
+            result = self.dummy_api_command(log_level="ERROR", use_job_specific_log_file=False)
+            result = self.dummy_api_command(log_level="ERROR", use_job_specific_log_file=False)
+            logger_infos_after = self.create_logger_infos()
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assertNotEqual(catched_stderr, "")
+            self.assertRegex(stderr_output, self.create_test_regex(logging.ERROR))
+            self.assertNotRegex(stderr_output, self.create_test_regex(logging.INFO))
+            self.assertEqual(2, stderr_output.count("DUMMY LOGGER ERROR"))
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertEqual(main_log_glob, [])
+
+    def test_luigi_use_job_specific_log_file_and_basic_logging_error(self):
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.ERROR)
+            logger_infos_before = self.create_logger_infos()
+            result = self.dummy_api_command(log_level=None, use_job_specific_log_file=True)
+            logger_infos_after = self.create_logger_infos()
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            self.assertEqual(stderr_output, "")
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertNotEqual(main_log_glob, [])
 
     def test_luigi_no_log_config_and_basic_logging_info(self):
-        with redirect_stderr(StringIO()) as f:
-            logging.basicConfig(
-                format='%(asctime)s %(levelname)-8s %(message)s',
-                level=logging.INFO,
-                datefmt='%Y-%m-%d %H:%M:%S',
-                force=True)
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.INFO)
             logger_infos_before = self.create_logger_infos()
             result = self.dummy_api_command(log_level=None, use_job_specific_log_file=False)
             logger_infos_after = self.create_logger_infos()
-        self.assertEqual(logger_infos_before, logger_infos_after)
-        stderr = f.getvalue()
-        self.assertRegex(stderr, ".*INFO     logging disabled in settings.*")
-        self.assertRegex(stderr, ".*===== Luigi Execution Summary =====.*")
-        main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
-        self.assertEqual(main_log_glob, [])
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assertRegex(stderr_output, self.create_test_regex(logging.ERROR))
+            self.assertRegex(stderr_output, self.create_test_regex(logging.INFO))
+            self.assertRegex(stderr_output, ".*===== Luigi Execution Summary =====.*")
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertEqual(main_log_glob, [])
 
     def test_luigi_no_log_config_and_basic_logging_error(self):
-        with redirect_stderr(StringIO()) as f:
-            logging.basicConfig(
-                format='%(asctime)s %(levelname)-8s %(message)s',
-                level=logging.ERROR,
-                datefmt='%Y-%m-%d %H:%M:%S',
-                force=True)
+        with catch_stderr() as catched_stderr:
+            self.configure_logging(log_level=logging.ERROR)
             logger_infos_before = self.create_logger_infos()
             result = self.dummy_api_command(log_level=None, use_job_specific_log_file=False)
             logger_infos_after = self.create_logger_infos()
-        self.assertEqual(logger_infos_before, logger_infos_after)
-        stderr = f.getvalue()
-        self.assertEqual(stderr, "")
-        main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
-        self.assertEqual(main_log_glob, [])
+            self.assert_loggers_are_equal(logger_infos_after, logger_infos_before)
+            catched_stderr.seek(0)
+            stderr_output = catched_stderr.read()
+            self.assertNotEqual(stderr_output, "")
+            self.assertRegex(stderr_output, self.create_test_regex(logging.ERROR))
+            self.assertNotRegex(stderr_output, self.create_test_regex(logging.INFO))
+            main_log_glob = list(Path(self._build_output_temp_dir.name).glob("**/main.log"))
+            self.assertEqual(main_log_glob, [])
 
     def reset_logging(self):
         logging.basicConfig(