[AIRFLOW-160] Parse DAG files through child processes

airflow/bin/cli.py

@@ -31,6 +31,8 @@
 from daemon.pidfile import TimeoutPIDLockFile
 import signal
 import sys
+import threading
+import traceback
 
 import airflow
 from airflow import jobs, settings
@@ -45,10 +47,28 @@
 DAGS_FOLDER = os.path.expanduser(conf.get('core', 'DAGS_FOLDER'))
 
 
-def sigint_handler(signal, frame):
+def sigint_handler(sig, frame):
     sys.exit(0)
 
 
+def sigquit_handler(sig, frame):
+    """Helps debug deadlocks by printing stacktraces when this gets a SIGQUIT
+    e.g. kill -s QUIT <PID> or CTRL+\
+    """
+    print("Dumping stack traces for all threads in PID {}".format(os.getpid()))
+    id_to_name = dict([(th.ident, th.name) for th in threading.enumerate()])
+    code = []
+    for thread_id, stack in sys._current_frames().items():
+        code.append("\n# Thread: {}({})"
+                    .format(id_to_name.get(thread_id, ""), thread_id))
+        for filename, line_number, name, line in traceback.extract_stack(stack):
+            code.append('File: "{}", line {}, in {}'
+                        .format((filename, line_number, name)))
+            if line:
+                code.append("  {}".format(line.strip()))
+    print("\n".join(code))
+
+
 def setup_logging(filename):
     root = logging.getLogger()
     handler = logging.FileHandler(filename)
@@ -483,6 +503,7 @@ def scheduler(args):
     job = jobs.SchedulerJob(
         dag_id=args.dag_id,
         subdir=process_subdir(args.subdir),
+        run_duration=args.run_duration,
         num_runs=args.num_runs,
         do_pickle=args.do_pickle)
 
@@ -506,6 +527,7 @@ def scheduler(args):
     else:
         signal.signal(signal.SIGINT, sigint_handler)
         signal.signal(signal.SIGTERM, sigint_handler)
+        signal.signal(signal.SIGQUIT, sigquit_handler)
         job.run()
 
 
@@ -851,6 +873,10 @@ class CLIFactory(object):
             default=False),
         # scheduler
         'dag_id_opt': Arg(("-d", "--dag_id"), help="The id of the dag to run"),
+        'run_duration': Arg(
+            ("-r", "--run-duration"),
+            default=None, type=int,
+            help="Set number of seconds to execute before exiting"),
         'num_runs': Arg(
             ("-n", "--num_runs"),
             default=None, type=int,
@@ -989,8 +1015,9 @@ class CLIFactory(object):
         }, {
             'func': scheduler,
             'help': "Start a scheduler instance",
-            'args': ('dag_id_opt', 'subdir', 'num_runs', 'do_pickle',
-                     'pid', 'daemon', 'stdout', 'stderr', 'log_file'),
+            'args': ('dag_id_opt', 'subdir', 'run_duration', 'num_runs',
+                     'do_pickle', 'pid', 'daemon', 'stdout', 'stderr',
+                     'log_file'),
         }, {
             'func': worker,
             'help': "Start a Celery worker node",

airflow/configuration.py

@@ -141,6 +141,11 @@ def run_command(command):
         'scheduler_heartbeat_sec': 60,
         'authenticate': False,
         'max_threads': 2,
+        'run_duration': 30 * 60,
+        'dag_dir_list_interval': 5 * 60,
+        'print_stats_interval': 30,
+        'min_file_process_interval': 180,
+        'child_process_log_directory': '/tmp/airflow/scheduler/logs'
     },
     'celery': {
         'broker_url': 'sqla+mysql://airflow:airflow@localhost:3306/airflow',

airflow/dag/__init__.py

@@ -0,0 +1,14 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#

airflow/dag/base_dag.py

@@ -0,0 +1,96 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+from __future__ import unicode_literals
+
+from abc import ABCMeta, abstractmethod, abstractproperty
+
+
+class BaseDag(object):
+    """
+    Base DAG object that both the SimpleDag and DAG inherit.
+    """
+    __metaclass__ = ABCMeta
+
+    @abstractproperty
+    def dag_id(self):
+        """
+        :return: the DAG ID
+        :rtype: unicode
+        """
+        raise NotImplementedError()
+
+    @abstractproperty
+    def task_ids(self):
+        """
+        :return: A list of task IDs that are in this DAG
+        :rtype: List[unicode]
+        """
+        raise NotImplementedError()
+
+    @abstractproperty
+    def full_filepath(self):
+        """
+        :return: The absolute path to the file that contains this DAG's definition
+        :rtype: unicode
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def concurrency(self):
+        """
+        :return: maximum number of tasks that can run simultaneously from this DAG
+        :rtype: int
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def is_paused(self):
+        """
+        :return: whether this DAG is paused or not
+        :rtype: bool
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def pickle_id(self):
+        """
+        :return: The pickle ID for this DAG, if it has one. Otherwise None.
+        :rtype: unicode
+        """
+        raise NotImplementedError
+
+
+class BaseDagBag(object):
+    """
+    Base object that both the SimpleDagBag and DagBag inherit.
+    """
+    @abstractproperty
+    def dag_ids(self):
+        """
+        :return: a list of DAG IDs in this bag
+        :rtype: List[unicode]
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def get_dag(self, dag_id):
+        """
+        :return: whether the task exists in this bag
+        :rtype: BaseDag
+        """
+        raise NotImplementedError()

airflow/executors/base_executor.py

@@ -102,7 +102,7 @@ def heartbeat(self):
             # TODO(jlowin) without a way to know what Job ran which tasks,
             # there is a danger that another Job started running a task
             # that was also queued to this executor. This is the last chance
-            # to check if that hapened. The most probable way is that a
+            # to check if that happened. The most probable way is that a
             # Scheduler tried to run a task that was originally queued by a
             # Backfill. This fix reduces the probability of a collision but
             # does NOT eliminate it.