[AIRFLOW-1018] Make processor use logging framework

Until now, the dga processor had its own logging
implementation,
making it hard to adjust for certain use cases
like working
in a container.

This patch moves everything to the standard
logging framework.

Closes apache#2728 from bolkedebruin/AIRFLOW-1018

Author: bolkedebruin

airflow/config_templates/airflow_local_settings.py

@@ -25,8 +25,10 @@
 LOG_FORMAT = conf.get('core', 'log_format')
 
 BASE_LOG_FOLDER = conf.get('core', 'BASE_LOG_FOLDER')
+PROCESSOR_LOG_FOLDER = conf.get('scheduler', 'child_process_log_directory')
 
 FILENAME_TEMPLATE = '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log'
+PROCESSOR_FILENAME_TEMPLATE = '{{ filename }}.log'
 
 DEFAULT_LOGGING_CONFIG = {
     'version': 1,
@@ -35,6 +37,9 @@
         'airflow.task': {
             'format': LOG_FORMAT,
         },
+        'airflow.processor': {
+            'format': LOG_FORMAT,
+        },
     },
     'handlers': {
         'console': {
@@ -47,6 +52,12 @@
             'formatter': 'airflow.task',
             'base_log_folder': os.path.expanduser(BASE_LOG_FOLDER),
             'filename_template': FILENAME_TEMPLATE,
+        },
+        'file.processor': {
+            'class': 'airflow.utils.log.file_processor_handler.FileProcessorHandler',
+            'formatter': 'airflow.processor',
+            'base_log_folder': os.path.expanduser(PROCESSOR_LOG_FOLDER),
+            'filename_template': PROCESSOR_FILENAME_TEMPLATE,
         }
         # When using s3 or gcs, provide a customized LOGGING_CONFIG
         # in airflow_local_settings within your PYTHONPATH, see UPDATING.md
@@ -67,6 +78,11 @@
         # },
     },
     'loggers': {
+        'airflow.processor' : {
+            'handlers': ['file.processor'],
+            'level': LOG_LEVEL,
+            'propagate': False,
+        },
         'airflow.task': {
             'handlers': ['file.task'],
             'level': LOG_LEVEL,

airflow/jobs.py

@@ -18,6 +18,7 @@
 from __future__ import unicode_literals
 
 import getpass
+import logging
 import multiprocessing
 import os
 import psutil
@@ -285,19 +286,16 @@ class DagFileProcessor(AbstractDagFileProcessor, LoggingMixin):
     # Counter that increments everytime an instance of this class is created
     class_creation_counter = 0
 
-    def __init__(self, file_path, pickle_dags, dag_id_white_list, log_file):
+    def __init__(self, file_path, pickle_dags, dag_id_white_list):
         """
         :param file_path: a Python file containing Airflow DAG definitions
         :type file_path: unicode
         :param pickle_dags: whether to serialize the DAG objects to the DB
         :type pickle_dags: bool
         :param dag_id_whitelist: If specified, only look at these DAG ID's
         :type dag_id_whitelist: list[unicode]
-        :param log_file: the path to the file where log lines should be output
-        :type log_file: unicode
         """
         self._file_path = file_path
-        self._log_file = log_file
         # Queue that's used to pass results from the child process.
         self._result_queue = multiprocessing.Queue()
         # The process that was launched to process the given .
@@ -319,17 +317,12 @@ def __init__(self, file_path, pickle_dags, dag_id_white_list, log_file):
     def file_path(self):
         return self._file_path
 
-    @property
-    def log_file(self):
-        return self._log_file
-
     @staticmethod
     def _launch_process(result_queue,
                         file_path,
                         pickle_dags,
                         dag_id_white_list,
-                        thread_name,
-                        log_file):
+                        thread_name):
         """
         Launch a process to process the given file.
 
@@ -345,35 +338,21 @@ def _launch_process(result_queue,
         :type dag_id_white_list: list[unicode]
         :param thread_name: the name to use for the process that is launched
         :type thread_name: unicode
-        :param log_file: the logging output for the process should be directed
-        to this file
-        :type log_file: unicode
         :return: the process that was launched
         :rtype: multiprocessing.Process
         """
         def helper():
             # This helper runs in the newly created process
-
-            # Re-direct stdout and stderr to a separate log file. Otherwise,
-            # the main log becomes too hard to read. No buffering to enable
-            # responsive file tailing
-            parent_dir, _ = os.path.split(log_file)
-
-            _log = LoggingMixin().log
-
-            # Create the parent directory for the log file if necessary.
-            if not os.path.isdir(parent_dir):
-                os.makedirs(parent_dir)
-
-            f = open(log_file, "a")
-            original_stdout = sys.stdout
-            original_stderr = sys.stderr
-
-            sys.stdout = f
-            sys.stderr = f
+            log = logging.getLogger("airflow.processor")
+            for handler in log.handlers:
+                try:
+                    handler.set_context(file_path)
+                except AttributeError:
+                    # Not all handlers need to have context passed in so we ignore
+                    # the error when handlers do not have set_context defined.
+                    pass
 
             try:
-                configure_logging()
                 # Re-configure the ORM engine as there are issues with multiple processes
                 settings.configure_orm()
 
@@ -383,26 +362,20 @@ def helper():
                 threading.current_thread().name = thread_name
                 start_time = time.time()
 
-                _log.info("Started process (PID=%s) to work on %s",
-                             os.getpid(),
-                             file_path)
-                scheduler_job = SchedulerJob(dag_ids=dag_id_white_list)
+                log.info("Started process (PID=%s) to work on %s",
+                         os.getpid(), file_path)
+                scheduler_job = SchedulerJob(dag_ids=dag_id_white_list, log=log)
                 result = scheduler_job.process_file(file_path,
                                                     pickle_dags)
                 result_queue.put(result)
                 end_time = time.time()
-                _log.info(
-                    "Processing %s took %.3f seconds",
-                    file_path, end_time - start_time
+                log.info(
+                    "Processing %s took %.3f seconds", file_path, end_time - start_time
                 )
             except:
                 # Log exceptions through the logging framework.
-                _log.exception("Got an exception! Propagating...")
+                log.exception("Got an exception! Propagating...")
                 raise
-            finally:
-                sys.stdout = original_stdout
-                sys.stderr = original_stderr
-                f.close()
 
         p = multiprocessing.Process(target=helper,
                                     args=(),
@@ -419,8 +392,7 @@ def start(self):
             self.file_path,
             self._pickle_dags,
             self._dag_id_white_list,
-            "DagFileProcessor{}".format(self._instance_id),
-            self.log_file)
+            "DagFileProcessor{}".format(self._instance_id))
         self._start_time = datetime.utcnow()
 
     def terminate(self, sigkill=False):
@@ -538,6 +510,7 @@ def __init__(
             processor_poll_interval=1.0,
             run_duration=None,
             do_pickle=False,
+            log=None,
             *args, **kwargs):
         """
         :param dag_id: if specified, only schedule tasks with this DAG ID
@@ -574,6 +547,10 @@ def __init__(
 
         self.heartrate = conf.getint('scheduler', 'SCHEDULER_HEARTBEAT_SEC')
         self.max_threads = conf.getint('scheduler', 'max_threads')
+
+        if log:
+            self._log = log
+
         self.using_sqlite = False
         if 'sqlite' in conf.get('core', 'sql_alchemy_conn'):
             if self.max_threads > 1:
@@ -591,9 +568,7 @@ def __init__(
         # Parse and schedule each file no faster than this interval. Default
         # to 3 minutes.
         self.file_process_interval = file_process_interval
-        # Directory where log files for the processes that scheduled the DAGs reside
-        self.child_process_log_directory = conf.get('scheduler',
-                                                    'child_process_log_directory')
+
         self.max_tis_per_query = conf.getint('scheduler', 'max_tis_per_query')
         if run_duration is None:
             self.run_duration = conf.getint('scheduler',
@@ -1548,17 +1523,15 @@ def _execute(self):
         known_file_paths = list_py_file_paths(self.subdir)
         self.log.info("There are %s files in %s", len(known_file_paths), self.subdir)
 
-        def processor_factory(file_path, log_file_path):
+        def processor_factory(file_path):
             return DagFileProcessor(file_path,
                                     pickle_dags,
-                                    self.dag_ids,
-                                    log_file_path)
+                                    self.dag_ids)
 
         processor_manager = DagFileProcessorManager(self.subdir,
                                                     known_file_paths,
                                                     self.max_threads,
                                                     self.file_process_interval,
-                                                    self.child_process_log_directory,
                                                     self.num_runs,
                                                     processor_factory)
 

airflow/utils/dag_processing.py

@@ -276,15 +276,6 @@ def start_time(self):
         """
         raise NotImplementedError()
 
-    @property
-    @abstractmethod
-    def log_file(self):
-        """
-        :return: the log file associated with this processor
-        :rtype: unicode
-        """
-        raise NotImplementedError()
-
     @property
     @abstractmethod
     def file_path(self):
@@ -313,7 +304,6 @@ def __init__(self,
                  file_paths,
                  parallelism,
                  process_file_interval,
-                 child_process_log_directory,
                  max_runs,
                  processor_factory):
         """
@@ -330,12 +320,9 @@ def __init__(self,
         :param max_runs: The number of times to parse and schedule each file. -1
         for unlimited.
         :type max_runs: int
-        :param child_process_log_directory: Store logs for child processes in
-        this directory
-        :type child_process_log_directory: unicode
         :type process_file_interval: float
         :param processor_factory: function that creates processors for DAG
-        definition files. Arguments are (dag_definition_path, log_file_path)
+        definition files. Arguments are (dag_definition_path)
         :type processor_factory: (unicode, unicode) -> (AbstractDagFileProcessor)
 
         """
@@ -345,7 +332,6 @@ def __init__(self,
         self._dag_directory = dag_directory
         self._max_runs = max_runs
         self._process_file_interval = process_file_interval
-        self._child_process_log_directory = child_process_log_directory
         self._processor_factory = processor_factory
         # Map from file path to the processor
         self._processors = {}
@@ -447,85 +433,6 @@ def set_file_paths(self, new_file_paths):
                 processor.stop()
         self._processors = filtered_processors
 
-    @staticmethod
-    def _split_path(file_path):
-        """
-        Return the path elements of a path as an array. E.g. /a/b/c ->
-        ['a', 'b', 'c']
-
-        :param file_path: the file path to split
-        :return: a list of the elements of the file path
-        :rtype: list[unicode]
-        """
-        results = []
-        while True:
-            head, tail = os.path.split(file_path)
-            if len(tail) != 0:
-                results.append(tail)
-            if file_path == head:
-                break
-            file_path = head
-        results.reverse()
-        return results
-
-    def _get_log_directory(self):
-        """
-        Log output from processing DAGs for the current day should go into
-        this directory.
-
-        :return: the path to the corresponding log directory
-        :rtype: unicode
-        """
-        now = datetime.utcnow()
-        return os.path.join(self._child_process_log_directory,
-                            now.strftime("%Y-%m-%d"))
-
-    def _get_log_file_path(self, dag_file_path):
-        """
-        Log output from processing the specified file should go to this
-        location.
-
-        :param dag_file_path: file containing a DAG
-        :type dag_file_path: unicode
-        :return: the path to the corresponding log file
-        :rtype: unicode
-        """
-        log_directory = self._get_log_directory()
-        # General approach is to put the log file under the same relative path
-        # under the log directory as the DAG file in the DAG directory
-        relative_dag_file_path = os.path.relpath(dag_file_path, start=self._dag_directory)
-        path_elements = self._split_path(relative_dag_file_path)
-
-        # Add a .log suffix for the log file
-        path_elements[-1] += ".log"
-
-        return os.path.join(log_directory, *path_elements)
-
-    def symlink_latest_log_directory(self):
-        """
-        Create symbolic link to the current day's log directory to
-        allow easy access to the latest scheduler log files.
-
-        :return: None
-        """
-        log_directory = self._get_log_directory()
-        latest_log_directory_path = os.path.join(
-            self._child_process_log_directory, "latest")
-        if os.path.isdir(log_directory):
-            # if symlink exists but is stale, update it
-            if os.path.islink(latest_log_directory_path):
-                if os.readlink(latest_log_directory_path) != log_directory:
-                    os.unlink(latest_log_directory_path)
-                    os.symlink(log_directory, latest_log_directory_path)
-            elif (os.path.isdir(latest_log_directory_path) or
-                    os.path.isfile(latest_log_directory_path)):
-                self.log.warning(
-                    "%s already exists as a dir/file. Skip creating symlink.",
-                    latest_log_directory_path
-                )
-            else:
-                os.symlink(log_directory, latest_log_directory_path)
-
     def processing_count(self):
         """
         :return: the number of files currently being processed
@@ -574,8 +481,8 @@ def heartbeat(self):
         for file_path, processor in finished_processors.items():
             if processor.result is None:
                 self.log.warning(
-                    "Processor for %s exited with return code %s. See %s for details.",
-                    processor.file_path, processor.exit_code, processor.log_file
+                    "Processor for %s exited with return code %s.",
+                    processor.file_path, processor.exit_code
                 )
             else:
                 for simple_dag in processor.result:
@@ -622,19 +529,15 @@ def heartbeat(self):
         while (self._parallelism - len(self._processors) > 0 and
                len(self._file_path_queue) > 0):
             file_path = self._file_path_queue.pop(0)
-            log_file_path = self._get_log_file_path(file_path)
-            processor = self._processor_factory(file_path, log_file_path)
+            processor = self._processor_factory(file_path)
 
             processor.start()
             self.log.info(
-                "Started a process (PID: %s) to generate tasks for %s - logging into %s",
-                processor.pid, file_path, log_file_path
+                "Started a process (PID: %s) to generate tasks for %s",
+                processor.pid, file_path
             )
-
             self._processors[file_path] = processor
 
-        self.symlink_latest_log_directory()
-
         # Update scheduler heartbeat count.
         self._run_count[self._heart_beat_key] += 1