Mailing-List: contact commits-help@airflow.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@airflow.incubator.apache.org
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: davydov@apache.org
To: commits@airflow.incubator.apache.org
Message-Id: <bc9df9a2264a4ce99445afc7ef8a42ed@git.apache.org>
Subject: incubator-airflow git commit: [AIRFLOW-780] Fix dag import errors no
 longer working
Date: Fri, 27 Jan 2017 09:30:50 +0000 (UTC)
archived-at: Fri, 27 Jan 2017 09:30:58 -0000

Repository: incubator-airflow
Updated Branches:
  refs/heads/master dc97bcd3b -> 67cbb9664


[AIRFLOW-780] Fix dag import errors no longer working

The import errors were no longer working after the
multiprocessor update
(since they are cleared after each DAG directory
is parsed). This change
fixes them, and adds tests to prevent future
regressions.

Also fix a couple of linter errors.

Note that there are a few inefficiencies (e.g.
sometimes we delete then add import errors in the
same place instead of just doing an update), but
this is equivalent to the old behavior.

Testing Done:
- Added missing unit tests for dag imports. Note
that some of them strangely fail for python 3 and
it became too time consuming to debug since I
don't have a copy of the travis environment, I
even ran with the same version of python locally
and couldn't reproduce. I have skipped those 3
tests in python 3 for now.

Closes #2018 from aoen/fix_parse_errors_not_displa


Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/67cbb966
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/67cbb966
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/67cbb966

Branch: refs/heads/master
Commit: 67cbb966410226c1489bb730af3af45330fc51b9
Parents: dc97bcd
Author: Dan Davydov <dan.davydov@airbnb.com>
Authored: Fri Jan 27 01:29:00 2017 -0800
Committer: Dan Davydov <dan.davydov@airbnb.com>
Committed: Fri Jan 27 01:29:04 2017 -0800

----------------------------------------------------------------------
 airflow/configuration.py        |   1 +
 airflow/jobs.py                 |  48 +++++----
 airflow/models.py               |   3 +
 airflow/settings.py             |   3 +-
 airflow/utils/dag_processing.py |   4 +-
 tests/jobs.py                   | 193 +++++++++++++++++++++++++++++++++--
 6 files changed, 219 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/airflow/configuration.py
----------------------------------------------------------------------
diff --git a/airflow/configuration.py b/airflow/configuration.py
index 979b071..011f764 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -503,6 +503,7 @@ authenticate = true
 max_threads = 2
 catchup_by_default = True
 scheduler_zombie_task_threshold = 300
+dag_dir_list_interval = 0
 """
 
 
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/airflow/jobs.py
----------------------------------------------------------------------
diff --git a/airflow/jobs.py b/airflow/jobs.py
index 201d87f..1a581e9 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -657,34 +657,44 @@ class SchedulerJob(BaseJob):
             session.close()
 
     @staticmethod
-    def record_import_errors(session, dagbag):
+    @provide_session
+    def clear_nonexistent_import_errors(session, known_file_paths):
         """
-        For the DAGs in the given DagBag, record any associated import errors.
-        These are usually displayed through the Airflow UI so that users know
-        that there are issues parsing DAGs.
+        Clears import errors for files that no longer exist.
 
         :param session: session for ORM operations
         :type session: sqlalchemy.orm.session.Session
-        :param dagbag: DagBag containing DAGs with import errors
-        :type dagbag: models.Dagbag
+        :param known_file_paths: The list of existing files that are parsed for DAGs
+        :type known_file_paths: list[unicode]
         """
-        for filename, stacktrace in list(dagbag.import_errors.items()):
-            session.query(models.ImportError).filter(
-                models.ImportError.filename == filename
-            ).delete()
-            session.add(models.ImportError(
-                filename=filename, stacktrace=stacktrace))
+        session.query(models.ImportError).filter(
+            ~models.ImportError.filename.in_(known_file_paths)
+        ).delete(synchronize_session='fetch')
         session.commit()
 
     @staticmethod
-    def clear_import_errors(session):
+    def update_import_errors(session, dagbag):
         """
-        Remove all the known import errors from the DB.
+        For the DAGs in the given DagBag, record any associated import errors and clears
+        errors for files that no longer have them. These are usually displayed through the
+        Airflow UI so that users know that there are issues parsing DAGs.
 
         :param session: session for ORM operations
         :type session: sqlalchemy.orm.session.Session
+        :param dagbag: DagBag containing DAGs with import errors
+        :type dagbag: models.Dagbag
         """
-        session.query(models.ImportError).delete()
+        # Clear the errors of the processed files
+        for dagbag_file in dagbag.file_last_changed:
+            session.query(models.ImportError).filter(
+                models.ImportError.filename == dagbag_file
+            ).delete()
+
+        # Add the errors of the processed files
+        for filename, stacktrace in dagbag.import_errors.iteritems():
+            session.add(models.ImportError(
+                filename=filename,
+                stacktrace=stacktrace))
         session.commit()
 
     @provide_session
@@ -1352,8 +1362,6 @@ class SchedulerJob(BaseJob):
                                                       dr.execution_date))
             self._reset_state_for_orphaned_tasks(dr, session=session)
 
-        self.logger.info("Removing old import errors")
-        self.clear_import_errors(session)
         session.close()
 
         execute_start_time = datetime.now()
@@ -1388,6 +1396,9 @@ class SchedulerJob(BaseJob):
                                  .format(len(known_file_paths), self.subdir))
                 processor_manager.set_file_paths(known_file_paths)
 
+                self.logger.debug("Removing old import errors")
+                self.clear_nonexistent_import_errors(known_file_paths=known_file_paths)
+
             # Kick of new processes and collect results from finished ones
             self.logger.info("Heartbeating the process manager")
             simple_dags = processor_manager.heartbeat()
@@ -1523,6 +1534,7 @@ class SchedulerJob(BaseJob):
                                      file_path))
         else:
             self.logger.warn("No viable dags retrieved from {}".format(file_path))
+            self.update_import_errors(session, dagbag)
             return []
 
         # Save individual DAGs in the ORM and update DagModel.last_scheduled_time
@@ -1598,7 +1610,7 @@ class SchedulerJob(BaseJob):
 
         # Record import errors into the ORM
         try:
-            self.record_import_errors(session, dagbag)
+            self.update_import_errors(session, dagbag)
         except Exception:
             self.logger.exception("Error logging import errors!")
         try:

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 95c40a9..b9015eb 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -247,6 +247,7 @@ class DagBag(BaseDagBag, LoggingMixin):
                 with open(filepath, 'rb') as f:
                     content = f.read()
                     if not all([s in content for s in (b'DAG', b'airflow')]):
+                        self.file_last_changed[filepath] = file_last_changed_on_disk
                         return found_dags
 
             self.logger.debug("Importing {}".format(filepath))
@@ -283,6 +284,8 @@ class DagBag(BaseDagBag, LoggingMixin):
                                               format(mod.filename, filepath))
                             content = zf.read()
                             if not all([s in content for s in (b'DAG', b'airflow')]):
+                                self.file_last_changed[filepath] = (
+                                    file_last_changed_on_disk)
                                 # todo: create ignore list
                                 return found_dags
 

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/airflow/settings.py
----------------------------------------------------------------------
diff --git a/airflow/settings.py b/airflow/settings.py
index 4882875..45f7fba 100644
--- a/airflow/settings.py
+++ b/airflow/settings.py
@@ -143,7 +143,6 @@ def configure_orm(disable_connection_pool=False):
         engine_args['pool_size'] = conf.getint('core', 'SQL_ALCHEMY_POOL_SIZE')
         engine_args['pool_recycle'] = conf.getint('core',
                                                   'SQL_ALCHEMY_POOL_RECYCLE')
-        # engine_args['echo'] = True
 
     engine = create_engine(SQL_ALCHEMY_CONN, **engine_args)
     Session = scoped_session(
@@ -163,4 +162,4 @@ configure_orm()
 
 KILOBYTE = 1024
 MEGABYTE = KILOBYTE * KILOBYTE
-WEB_COLORS = {'LIGHTBLUE': '#4d9de0'}
\ No newline at end of file
+WEB_COLORS = {'LIGHTBLUE': '#4d9de0'}

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/airflow/utils/dag_processing.py
----------------------------------------------------------------------
diff --git a/airflow/utils/dag_processing.py b/airflow/utils/dag_processing.py
index aa502fb..6209946 100644
--- a/airflow/utils/dag_processing.py
+++ b/airflow/utils/dag_processing.py
@@ -198,8 +198,8 @@ def list_py_file_paths(directory, safe_mode=True):
                     if safe_mode:
                         with open(file_path, 'rb') as f:
                             content = f.read()
-                            might_contain_dag = all([s in content
-                                                     for s in (b'DAG', b'airflow')])
+                            might_contain_dag = all(
+                                [s in content for s in (b'DAG', b'airflow')])
 
                     if not might_contain_dag:
                         continue

http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/67cbb966/tests/jobs.py
----------------------------------------------------------------------
diff --git a/tests/jobs.py b/tests/jobs.py
index 53626ee..b674bcd 100644
--- a/tests/jobs.py
+++ b/tests/jobs.py
@@ -20,13 +20,15 @@ from __future__ import unicode_literals
 import datetime
 import logging
 import os
+import shutil
 import unittest
 import six
+import sys
+from tempfile import mkdtemp
 
 from airflow import AirflowException, settings
 from airflow import models
 from airflow.bin import cli
-from airflow.executors import DEFAULT_EXECUTOR
 from airflow.jobs import BackfillJob, SchedulerJob
 from airflow.models import DAG, DagModel, DagBag, DagRun, Pool, TaskInstance as TI
 from airflow.operators.dummy_operator import DummyOperator
@@ -49,9 +51,20 @@ except ImportError:
     except ImportError:
         mock = None
 
+IS_PYTHON_3_TRAVIS = sys.version_info >= (3, 0) and "TRAVIS" in os.environ
+
 DEV_NULL = '/dev/null'
 DEFAULT_DATE = datetime.datetime(2016, 1, 1)
 
+# Include the words "airflow" and "dag" in the file contents, tricking airflow into thinking these
+# files contain a DAG (otherwise Airflow will skip them)
+PARSEABLE_DAG_FILE_CONTENTS = '"airflow DAG"'
+UNPARSEABLE_DAG_FILE_CONTENTS = 'airflow DAG'
+
+# Filename to be used for dags that are created in an ad-hoc manner and can be removed/
+# created at runtime
+TEMP_DAG_FILENAME = "temp_dag.py"
+
 
 class BackfillJobTest(unittest.TestCase):
 
@@ -97,7 +110,7 @@ class BackfillJobTest(unittest.TestCase):
         job = BackfillJob(
             dag=dag,
             start_date=DEFAULT_DATE,
-            end_date=DEFAULT_DATE+datetime.timedelta(days=1),
+            end_date=DEFAULT_DATE + datetime.timedelta(days=1),
             ignore_first_depends_on_past=True
         )
         job.run()
@@ -109,7 +122,8 @@ class BackfillJobTest(unittest.TestCase):
 
         self.assertTrue(drs[0].execution_date == DEFAULT_DATE)
         self.assertTrue(drs[0].state == State.SUCCESS)
-        self.assertTrue(drs[1].execution_date == DEFAULT_DATE+datetime.timedelta(days=1))
+        self.assertTrue(drs[1].execution_date ==
+                        DEFAULT_DATE + datetime.timedelta(days=1))
         self.assertTrue(drs[1].state == State.SUCCESS)
 
         dag.clear()
@@ -134,9 +148,8 @@ class BackfillJobTest(unittest.TestCase):
         logger = logging.getLogger('BackfillJobTest.test_backfill_examples')
         dags = [
             dag for dag in self.dagbag.dags.values()
-            if 'example_dags' in dag.full_filepath
-            and dag.dag_id not in skip_dags
-            ]
+            if 'example_dags' in dag.full_filepath and dag.dag_id not in skip_dags
+        ]
 
         for dag in dags:
             dag.clear(
@@ -244,6 +257,27 @@ class SchedulerJobTest(unittest.TestCase):
 
     def setUp(self):
         self.dagbag = DagBag()
+        session = settings.Session()
+        session.query(models.ImportError).delete()
+        session.commit()
+
+    @staticmethod
+    def run_single_scheduler_loop_with_no_dags(dags_folder):
+        """
+        Utility function that runs a single scheduler loop without actually
+        changing/scheduling any dags. This is useful to simulate the other side effects of
+        running a scheduler loop, e.g. to see what parse errors there are in the
+        dags_folder.
+
+        :param dags_folder: the directory to traverse
+        :type directory: str
+        """
+        scheduler = SchedulerJob(
+            dag_id='this_dag_doesnt_exist',  # We don't want to actually run anything
+            num_runs=1,
+            subdir=os.path.join(dags_folder))
+        scheduler.heartrate = 0
+        scheduler.run()
 
     @provide_session
     def evaluate_dagrun(
@@ -1177,11 +1211,11 @@ class SchedulerJobTest(unittest.TestCase):
         self.assertLess(dr.execution_date, datetime.datetime.now())
 
         dag3 = DAG(DAG_NAME3,
-                  schedule_interval='@hourly',
-                  max_active_runs=1,
-                  catchup=False,
-                  default_args=default_args
-              )
+                   schedule_interval='@hourly',
+                   max_active_runs=1,
+                   catchup=False,
+                   default_args=default_args
+                   )
 
         run_this_1 = DummyOperator(task_id='run_this_1', dag=dag3)
         run_this_2 = DummyOperator(task_id='run_this_2', dag=dag3)
@@ -1209,3 +1243,140 @@ class SchedulerJobTest(unittest.TestCase):
 
         # The DR should be scheduled BEFORE now
         self.assertLess(dr.execution_date, datetime.datetime.now())
+
+    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
+                     "Fails in Python 3 on Travis but not reproducible locally")
+    def test_add_unparseable_file_before_sched_start_creates_import_error(self):
+        try:
+            dags_folder = mkdtemp()
+            unparseable_filename = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+            with open(unparseable_filename, 'w') as unparseable_file:
+                unparseable_file.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 1)
+        import_error = import_errors[0]
+        self.assertEqual(import_error.filename,
+                         unparseable_filename)
+        self.assertEqual(import_error.stacktrace,
+                         "invalid syntax ({}, line 1)".format(TEMP_DAG_FILENAME))
+
+    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
+                     "Fails in Python 3 on Travis but not reproducible locally")
+    def test_add_unparseable_file_after_sched_start_creates_import_error(self):
+        try:
+            dags_folder = mkdtemp()
+            unparseable_filename = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+
+            with open(unparseable_filename, 'w') as unparseable_file:
+                unparseable_file.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 1)
+        import_error = import_errors[0]
+        self.assertEqual(import_error.filename,
+                         unparseable_filename)
+        self.assertEqual(import_error.stacktrace,
+                         "invalid syntax ({}, line 1)".format(TEMP_DAG_FILENAME))
+
+    def test_no_import_errors_with_parseable_dag(self):
+        try:
+            dags_folder = mkdtemp()
+            parseable_filename = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+
+            with open(parseable_filename, 'w') as parseable_file:
+                parseable_file.writelines(PARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 0)
+
+    @unittest.skipIf(IS_PYTHON_3_TRAVIS,
+                     "Fails in Python 3 on Travis but not reproducible locally")
+    def test_new_import_error_replaces_old(self):
+        try:
+            dags_folder = mkdtemp()
+            unparseable_filename = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+
+            # Generate original import error
+            with open(unparseable_filename, 'w') as unparseable_file:
+                unparseable_file.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+
+            # Generate replacement import error (the error will be on the second line now)
+            with open(unparseable_filename, 'w') as unparseable_file:
+                unparseable_file.writelines(
+                    PARSEABLE_DAG_FILE_CONTENTS +
+                    os.linesep +
+                    UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 1)
+        import_error = import_errors[0]
+        self.assertEqual(import_error.filename,
+                         unparseable_filename)
+        self.assertEqual(import_error.stacktrace,
+                         "invalid syntax ({}, line 2)".format(TEMP_DAG_FILENAME))
+
+    def test_remove_error_clears_import_error(self):
+        try:
+            dags_folder = mkdtemp()
+            filename_to_parse = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+
+            # Generate original import error
+            with open(filename_to_parse, 'w') as file_to_parse:
+                file_to_parse.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+
+            # Remove the import error from the file
+            with open(filename_to_parse, 'w') as file_to_parse:
+                file_to_parse.writelines(
+                    PARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 0)
+
+    def test_remove_file_clears_import_error(self):
+        try:
+            dags_folder = mkdtemp()
+            filename_to_parse = os.path.join(dags_folder, TEMP_DAG_FILENAME)
+
+            # Generate original import error
+            with open(filename_to_parse, 'w') as file_to_parse:
+                file_to_parse.writelines(UNPARSEABLE_DAG_FILE_CONTENTS)
+            self.run_single_scheduler_loop_with_no_dags(dags_folder)
+        finally:
+            shutil.rmtree(dags_folder)
+
+        # Rerun the scheduler once the dag file has been removed
+        self.run_single_scheduler_loop_with_no_dags(dags_folder)
+
+        session = settings.Session()
+        import_errors = session.query(models.ImportError).all()
+
+        self.assertEqual(len(import_errors), 0)