Many fixes for handling tasks in jobs and executors

airflow/bin/cli.py

@@ -33,11 +33,6 @@ def process_subdir(subdir):
         if "DAGS_FOLDER" in subdir:
             subdir = subdir.replace("DAGS_FOLDER", dags_folder)
         subdir = os.path.abspath(os.path.expanduser(subdir))
-        if dags_folder.rstrip('/') not in subdir.rstrip('/'):
-            raise AirflowException(
-                "subdir has to be part of your DAGS_FOLDER as defined in your "
-                "airflow.cfg. DAGS_FOLDER is {df} and subdir is {sd}".format(
-                    df=dags_folder, sd=subdir))
         return subdir
 
 
@@ -84,6 +79,7 @@ def backfill(args, dag=None):
             donot_pickle=(args.donot_pickle or
                           conf.getboolean('core', 'donot_pickle')),
             ignore_dependencies=args.ignore_dependencies,
+            ignore_first_depends_on_past=args.ignore_first_depends_on_past,
             pool=args.pool)
 
 
@@ -176,15 +172,16 @@ def run(args, dag=None):
             mark_success=args.mark_success,
             force=args.force,
             pickle_id=args.pickle,
-            task_start_date=args.task_start_date,
             ignore_dependencies=args.ignore_dependencies,
+            ignore_depends_on_past=args.ignore_depends_on_past,
             pool=args.pool)
         run_job.run()
     elif args.raw:
         ti.run(
             mark_success=args.mark_success,
             force=args.force,
             ignore_dependencies=args.ignore_dependencies,
+            ignore_depends_on_past=args.ignore_depends_on_past,
             job_id=args.job_id,
             pool=args.pool,
         )
@@ -214,6 +211,7 @@ def run(args, dag=None):
             mark_success=args.mark_success,
             pickle_id=pickle_id,
             ignore_dependencies=args.ignore_dependencies,
+            ignore_depends_on_past=args.ignore_depends_on_past,
             force=args.force,
             pool=args.pool)
         executor.heartbeat()
@@ -500,6 +498,13 @@ class CLIFactory(object):
                 "matching the regexp. Only works in conjunction "
                 "with task_regex"),
             "store_true"),
+        'bf_ignore_first_depends_on_past': Arg(
+            ("-I", "--ignore_first_depends_on_past"),
+            (
+                "Ignores depends_on_past dependencies for the first "
+                "set of tasks only (subsequent executions in the backfill "
+                "DO respect depends_on_past)."),
+            "store_true"),
         'pool': Arg(("--pool",), "Resource pool to use"),
         # list_dags
         'tree': Arg(("-t", "--tree"), "Tree view", "store_true"),
@@ -528,17 +533,18 @@ class CLIFactory(object):
             ("-kt", "--keytab"), "keytab",
             nargs='?', default=conf.get('kerberos', 'keytab')),
         # run
-        'task_start_date': Arg(
-            ("-s", "--task_start_date"),
-            "Override the tasks's start_date (used internally)",
-            type=parsedate),
         'force': Arg(
             ("-f", "--force"),
             "Force a run regardless or previous success", "store_true"),
         'raw': Arg(("-r", "--raw"), argparse.SUPPRESS, "store_true"),
         'ignore_dependencies': Arg(
             ("-i", "--ignore_dependencies"),
             "Ignore upstream and depends_on_past dependencies", "store_true"),
+        'ignore_depends_on_past': Arg(
+            ("-I", "--ignore_depends_on_past"),
+            "Ignore depends_on_past dependencies (but respect "
+            "upstream dependencies)",
+            "store_true"),
         'ship_dag': Arg(
             ("--ship_dag",),
             "Pickles (serializes) the DAG and ships it to the worker",
@@ -624,7 +630,8 @@ class CLIFactory(object):
             'args': (
                 'dag_id', 'task_regex', 'start_date', 'end_date',
                 'mark_success', 'local', 'donot_pickle', 'include_adhoc',
-                'bf_ignore_dependencies', 'subdir', 'pool', 'dry_run')
+                'bf_ignore_dependencies', 'bf_ignore_first_depends_on_past',
+                'subdir', 'pool', 'dry_run')
         }, {
             'func': list_tasks,
             'help': "List the tasks within a DAG",
@@ -661,8 +668,8 @@ class CLIFactory(object):
             'args': (
                 'dag_id', 'task_id', 'execution_date', 'subdir',
                 'mark_success', 'force', 'pool',
-                'task_start_date', 'local', 'raw', 'ignore_dependencies',
-                'ship_dag', 'pickle', 'job_id'),
+                'local', 'raw', 'ignore_dependencies',
+                'ignore_depends_on_past', 'ship_dag', 'pickle', 'job_id'),
         }, {
             'func': initdb,
             'help': "Initialize the metadata database",

airflow/contrib/operators/ssh_execute_operator.py

@@ -8,7 +8,7 @@
 from airflow.exceptions import AirflowException
 
 
-class SSHTempFileContent():
+class SSHTempFileContent(object):
     """This class prvides a functionality that creates tempfile
     with given content at remote host.
     Use like::

airflow/executors/base_executor.py

@@ -36,16 +36,21 @@ def queue_command(self, key, command, priority=1, queue=None):
             self.queued_tasks[key] = (command, priority, queue)
 
     def queue_task_instance(
-            self, task_instance, mark_success=False, pickle_id=None,
-            force=False, ignore_dependencies=False, task_start_date=None,
+            self,
+            task_instance,
+            mark_success=False,
+            pickle_id=None,
+            force=False,
+            ignore_dependencies=False,
+            ignore_depends_on_past=False,
             pool=None):
         pool = pool or task_instance.pool
         command = task_instance.command(
             local=True,
             mark_success=mark_success,
             force=force,
             ignore_dependencies=ignore_dependencies,
-            task_start_date=task_start_date,
+            ignore_depends_on_past=ignore_depends_on_past,
             pool=pool,
             pickle_id=pickle_id)
         self.queue_command(

airflow/executors/celery_executor.py

@@ -38,9 +38,10 @@ class CeleryConfig(object):
 @app.task
 def execute_command(command):
     logging.info("Executing command in Celery " + command)
-    rc = subprocess.Popen(command, shell=True).wait()
-    if rc:
-        logging.error(rc)
+    try:
+        subprocess.check_call(command, shell=True)
+    except subprocess.CalledProcessError as e:
+        logging.error(e)
         raise AirflowException('Celery command failed')
 
 
@@ -66,6 +67,7 @@ def execute_async(self, key, command, queue=DEFAULT_QUEUE):
         self.last_state[key] = celery_states.PENDING
 
     def sync(self):
+
         self.logger.debug(
             "Inquiring about {} celery task(s)".format(len(self.tasks)))
         for key, async in list(self.tasks.items()):