Improve logging for starting jobs and test cleanup

Author: AlanCoding

awx/main/tasks/jobs.py

@@ -116,11 +116,23 @@ def _wrapped(self, *args, **kwargs):
 
 @task(on_duplicate='queue_one', bind=True, queue=get_task_queuename)
 def dispatch_waiting_jobs(binder):
-    for uj in UnifiedJob.objects.filter(status='waiting', controller_node=settings.CLUSTER_HOST_ID).only('id', 'status', 'polymorphic_ctype', 'celery_task_id'):
+    start_time = time.monotonic()
+    job_id_list = []
+    for uj in UnifiedJob.objects.filter(status='waiting', controller_node=settings.CLUSTER_HOST_ID).only('id', 'status', 'polymorphic_ctype', 'celery_task_id')[
+        :25
+    ]:
         kwargs = uj.get_start_kwargs()
         if not kwargs:
             kwargs = {}
+        job_id_list.append(uj.pk)
         binder.control('run', data={'task': serialize_task(uj._get_task_class()), 'args': [uj.id], 'kwargs': kwargs, 'uuid': uj.celery_task_id})
+    if job_id_list:
+        logger.info(f'Dispatching off waiting jobs {job_id_list}, in time {time.monotonic() - start_time}')
+        # If this is a burst, the task manager may still be producing more jobs so reschedule
+        if len(job_id_list) > 1:
+            binder.control('run', data={'task': serialize_task(dispatch_waiting_jobs), 'args': [], 'kwargs': {}})
+    else:
+        logger.debug('No more waiting jobs to dispatch on this node')
 
 
 class BaseTask(object):

awx/main/tests/functional/test_bulk.py

@@ -2,6 +2,8 @@
 
 from uuid import uuid4
 
+from django.test import override_settings
+
 from awx.api.versioning import reverse
 
 from awx.main.models.jobs import JobTemplate
@@ -98,10 +100,10 @@ def test_bulk_job_launch_queries(job_template, organization, inventory, project,
     inventory.save()
     jobs = [{'unified_job_template': jt.id, 'inventory': inventory.id} for _ in range(num_jobs)]
 
-    # This is not working, we need to figure that out if we want to include tests for more jobs
-    # with mock.patch('awx.api.serializers.settings.BULK_JOB_MAX_LAUNCH', num_jobs + 1):
-    with django_assert_max_num_queries(num_queries):
-        bulk_job_launch_response = post(reverse('api:bulk_job_launch'), {'name': 'Bulk Job Launch', 'jobs': jobs}, normal_user, expect=201).data
+    # Assure settings allow given number of jobs
+    with override_settings(BULK_JOB_MAX_LAUNCH=num_jobs + 1):
+        with django_assert_max_num_queries(num_queries):
+            bulk_job_launch_response = post(reverse('api:bulk_job_launch'), {'name': 'Bulk Job Launch', 'jobs': jobs}, normal_user, expect=201).data
 
     # Run task manager so the workflow job nodes actually spawn
     TaskManager().schedule()