AC-1015 Changed where save_job_events task is started so that job events don't appear to be missing when they're stuck in the queue.

awx/main/tasks.py

@@ -26,7 +26,8 @@ import pexpect
 from kombu import Connection, Exchange, Queue
 
 # Celery
-from celery import Task, task
+from celery import Celery, Task, task
+from celery.execute import send_task
 
 # Django
 from django.conf import settings
@@ -476,6 +477,12 @@ class RunJob(BaseTask):
         '''
         if job.status in ('pending', 'waiting'):
             job = self.update_model(job.pk, status='pending')
+            # Start another task to process job events.
+            if settings.BROKER_URL.startswith('amqp://'):
+                app = Celery('tasks', broker=settings.BROKER_URL)
+                send_task('awx.main.tasks.save_job_events', kwargs={
+                    'job_id': job.id,
+                }, serializer='json')
             return True
         elif job.cancel_flag:
             job = self.update_model(job.pk, status='canceled')

awx/plugins/callback/job_event_callback.py

@@ -51,10 +51,6 @@ except ImportError:
 # Kombu
 from kombu import Connection, Exchange, Queue
 
-# Celery
-from celery import Celery
-from celery.execute import send_task
-
 
 class TokenAuth(requests.auth.AuthBase):
 
@@ -96,12 +92,6 @@ class CallbackModule(object):
     def __del__(self):
         self._cleanup_connection()
 
-    def _start_save_job_events_task(self):
-        app = Celery('tasks', broker=self.broker_url)
-        send_task('awx.main.tasks.save_job_events', kwargs={
-            'job_id': self.job_id,
-        }, serializer='json')
-
     def _publish_errback(self, exc, interval):
         if self.job_callback_debug:
             print 'Publish Error: %r, retry in %s seconds, pid=%s' % (exc, interval, os.getpid())
@@ -194,8 +184,6 @@ class CallbackModule(object):
         if task and event not in self.EVENTS_WITHOUT_TASK:
             event_data['task'] = task
         if self.broker_url:
-            if event == 'playbook_on_start':
-                self._start_save_job_events_task()
             self._post_job_event_queue_msg(event, event_data)
         else:
             self._post_rest_api_event(event, event_data)