[AIRFLOW-245] Add access to task instance to executors

airflow/contrib/executors/mesos_executor.py

@@ -258,7 +258,7 @@ def start(self):
         self.mesos_driver = driver
         self.mesos_driver.start()
 
-    def execute_async(self, key, command, queue=None):
+    def execute_async(self, key, command, queue=None, task=None):
         self.task_queue.put((key, command))
 
     def sync(self):

airflow/executors/base_executor.py

@@ -96,7 +96,7 @@ def heartbeat(self):
             ti.refresh_from_db()
             if ti.state != State.RUNNING:
                 self.running[key] = command
-                self.execute_async(key, command=command, queue=queue)
+                self.execute_async(key, command=command, queue=queue, task=ti)
             else:
                 self.logger.debug(
                     'Task is already running, not sending to '
@@ -124,7 +124,7 @@ def get_event_buffer(self):
         self.event_buffer = {}
         return d
 
-    def execute_async(self, key, command, queue=None):  # pragma: no cover
+    def execute_async(self, key, command, queue=None, task=None):  # pragma: no cover
         """
         This method will execute the command asynchronously.
         """

airflow/executors/celery_executor.py

@@ -59,7 +59,7 @@ def start(self):
         self.tasks = {}
         self.last_state = {}
 
-    def execute_async(self, key, command, queue=DEFAULT_QUEUE):
+    def execute_async(self, key, command, queue=DEFAULT_QUEUE, task=None):
         self.logger.info( "[celery] queuing {key} through celery, "
                        "queue={queue}".format(**locals()))
         self.tasks[key] = execute_command.apply_async(

airflow/executors/local_executor.py

@@ -60,7 +60,7 @@ def start(self):
         for w in self.workers:
             w.start()
 
-    def execute_async(self, key, command, queue=None):
+    def execute_async(self, key, command, queue=None, task=None):
         self.queue.put((key, command))
 
     def sync(self):

airflow/executors/sequential_executor.py

@@ -18,7 +18,7 @@ def __init__(self):
         super(SequentialExecutor, self).__init__()
         self.commands_to_run = []
 
-    def execute_async(self, key, command, queue=None):
+    def execute_async(self, key, command, queue=None, task=None):
         self.commands_to_run.append((key, command,))
 
     def sync(self):