Adjust inventory computed field calculations to happen in the celery context

awx/api/serializers.py

@@ -755,6 +755,7 @@ class HostSerializer(BaseSerializerWithVariables):
             d['last_job']['job_template_name'] = obj.last_job.job_template.name
         except (KeyError, AttributeError):
             pass
+        # TODO: This is slow
         d['all_groups'] = [{'id': g.id, 'name': g.name} for g in obj.all_groups.all()]
         d['groups'] = [{'id': g.id, 'name': g.name} for g in obj.groups.all()]
         d['recent_jobs'] = [{'id': j.job.id, 'name': j.job.job_template.name, 'status': j.job.status, 'finished': j.job.finished} \

awx/api/views.py

@@ -38,7 +38,6 @@ from awx.main.task_engine import TaskSerializer
 from awx.main.models import *
 from awx.main.utils import *
 from awx.main.access import get_user_queryset
-from awx.main.signals import ignore_inventory_computed_fields, ignore_inventory_group_removal
 from awx.api.authentication import JobTaskAuthentication
 from awx.api.permissions import *
 from awx.api.renderers import *

awx/main/management/commands/inventory_import.py

@@ -27,7 +27,8 @@ from django.contrib.auth.models import User
 
 # AWX
 from awx.main.models import *
-from awx.main.signals import ignore_inventory_computed_fields, disable_activity_stream
+from awx.main.utils import ignore_inventory_computed_fields
+from awx.main.signals import disable_activity_stream
 from awx.main.task_engine import TaskSerializer as LicenseReader
 
 logger = logging.getLogger('awx.main.commands.inventory_import')

awx/main/models/inventory.py

@@ -33,7 +33,7 @@ from awx.main.fields import AutoOneToOneField
 from awx.main.models.base import *
 from awx.main.models.jobs import Job
 from awx.main.models.unified_jobs import *
-from awx.main.utils import encrypt_field
+from awx.main.utils import encrypt_field, ignore_inventory_computed_fields
 
 __all__ = ['Inventory', 'Host', 'Group', 'InventorySource', 'InventoryUpdate']
 
@@ -108,7 +108,6 @@ class Inventory(CommonModel):
         '''
         When marking inventory inactive, also mark hosts and groups inactive.
         '''
-        from awx.main.signals import ignore_inventory_computed_fields
         with ignore_inventory_computed_fields():
             for host in self.hosts.filter(active=True):
                 host.mark_inactive()
@@ -374,7 +373,6 @@ class Group(CommonModelNameNotUnique):
             self.parents.clear()
             self.children.clear()
             self.hosts.clear()
-        from awx.main.signals import ignore_inventory_computed_fields
         i = self.inventory
 
         if recompute:

awx/main/models/jobs.py

@@ -38,7 +38,7 @@ from polymorphic import PolymorphicModel
 # AWX
 from awx.main.models.base import *
 from awx.main.models.unified_jobs import *
-from awx.main.utils import encrypt_field, decrypt_field
+from awx.main.utils import encrypt_field, decrypt_field, ignore_inventory_computed_fields
 
 # Celery
 from celery import chain
@@ -706,7 +706,6 @@ class JobEvent(CreatedModifiedModel):
 
     def update_host_summary_from_stats(self):
         from awx.main.models.inventory import Host
-        from awx.main.signals import ignore_inventory_computed_fields
         if self.event != 'playbook_on_stats':
             return
         hostnames = set()

awx/main/models/schedules.py

@@ -13,6 +13,7 @@ from django.utils.timezone import now, make_aware, get_default_timezone
 
 # AWX
 from awx.main.models.base import *
+from awx.main.utils import ignore_inventory_computed_fields
 from django.core.urlresolvers import reverse
 
 logger = logging.getLogger('awx.main.models.schedule')
@@ -104,7 +105,6 @@ class Schedule(CommonModel):
             self.dtend = make_aware(datetime.datetime.strptime(until_date, "%Y%m%dT%H%M%SZ"), get_default_timezone())
         if 'count' in self.rrule.lower():
             self.dtend = future_rs[-1]
-        from awx.main.signals import ignore_inventory_computed_fields
         with ignore_inventory_computed_fields():
             self.unified_job_template.update_computed_fields()
 

awx/main/signals.py

@@ -20,6 +20,8 @@ from crum.signals import current_user_getter
 from awx.main.models import *
 from awx.api.serializers import *
 from awx.main.utils import model_instance_diff, model_to_dict, camelcase_to_underscore, emit_websocket_notification
+from awx.main.utils import ignore_inventory_computed_fields, ignore_inventory_group_removal, _inventory_updates
+from awx.main.tasks import update_inventory_computed_fields
 
 __all__ = []
 
@@ -29,37 +31,8 @@ logger = logging.getLogger('awx.main.signals')
 # or marked inactive, when a Host-Group or Group-Group relationship is updated,
 # or when a Job is deleted or marked inactive.
 
-_inventory_updates = threading.local()
 
-@contextlib.contextmanager
+def emit_update_inventory_computed_fields(sender, **kwargs):
-def ignore_inventory_computed_fields():
-    '''
-    Context manager to ignore updating inventory computed fields.
-    '''
-    try:
-        previous_value = getattr(_inventory_updates, 'is_updating', False)
-        _inventory_updates.is_updating = True
-        yield
-    finally:
-        _inventory_updates.is_updating = previous_value
-
-@contextlib.contextmanager
-def ignore_inventory_group_removal():
-    '''
-    Context manager to ignore moving groups/hosts when group is deleted.
-    '''
-    try:
-        previous_value = getattr(_inventory_updates, 'is_removing', False)
-        _inventory_updates.is_removing = True
-        yield
-    finally:
-        _inventory_updates.is_removing = previous_value
-
-def update_inventory_computed_fields(sender, **kwargs):
-    '''
-    Signal handler and wrapper around inventory.update_computed_fields to
-    prevent unnecessary recursive calls.
-    '''
     logger.debug("In update inventory computed fields")
     if getattr(_inventory_updates, 'is_updating', False):
         return
@@ -86,46 +59,25 @@ def update_inventory_computed_fields(sender, **kwargs):
         return
     logger.debug('%s %s, updating inventory computed fields: %r %r',
                  sender_name, sender_action, sender, kwargs)
-    with ignore_inventory_computed_fields():
+    try:
-        try:
+        inventory = instance.inventory
-            inventory = instance.inventory
+    except Inventory.DoesNotExist:
-        except Inventory.DoesNotExist:
+        pass
-            pass
+    else:
-        else:
+        update_inventory_computed_fields.delay(inventory.id, issubclass(sender, Job))
-            update_hosts = issubclass(sender, Job)
-            inventory.update_computed_fields(update_hosts=update_hosts)
 
-def emit_job_event_detail(sender, **kwargs):
+post_save.connect(emit_update_inventory_computed_fields, sender=Host)
-    instance = kwargs['instance']
+post_delete.connect(emit_update_inventory_computed_fields, sender=Host)
-    created = kwargs['created']
+post_save.connect(emit_update_inventory_computed_fields, sender=Group)
-    if created:
+post_delete.connect(emit_update_inventory_computed_fields, sender=Group)
-        if instance.host is not None:
+m2m_changed.connect(emit_update_inventory_computed_fields, sender=Group.hosts.through)
-            host_id = instance.host.id
+m2m_changed.connect(emit_update_inventory_computed_fields, sender=Group.parents.through)
-        else:
+m2m_changed.connect(emit_update_inventory_computed_fields, sender=Host.inventory_sources.through)
-            host_id = None
+m2m_changed.connect(emit_update_inventory_computed_fields, sender=Group.inventory_sources.through)
-        if instance.parent is not None:
+post_save.connect(emit_update_inventory_computed_fields, sender=Job)
-            parent_id = instance.parent.id
+post_delete.connect(emit_update_inventory_computed_fields, sender=Job)
-        else:
+post_save.connect(emit_update_inventory_computed_fields, sender=InventorySource)
-            parent_id = None
+post_delete.connect(emit_update_inventory_computed_fields, sender=InventorySource)
-        event_serialized = JobEventSerializer(instance).data
-        event_serialized['id'] = instance.id
-        event_serialized["created"] = event_serialized["created"].isoformat()
-        event_serialized["modified"] = event_serialized["modified"].isoformat()
-        event_serialized["event_name"] = instance.event
-        emit_websocket_notification('/socket.io/job_events', 'job_events-' + str(instance.job.id), event_serialized)
-
-post_save.connect(update_inventory_computed_fields, sender=Host)
-post_delete.connect(update_inventory_computed_fields, sender=Host)
-post_save.connect(update_inventory_computed_fields, sender=Group)
-post_delete.connect(update_inventory_computed_fields, sender=Group)
-m2m_changed.connect(update_inventory_computed_fields, sender=Group.hosts.through)
-m2m_changed.connect(update_inventory_computed_fields, sender=Group.parents.through)
-m2m_changed.connect(update_inventory_computed_fields, sender=Host.inventory_sources.through)
-m2m_changed.connect(update_inventory_computed_fields, sender=Group.inventory_sources.through)
-post_save.connect(update_inventory_computed_fields, sender=Job)
-post_delete.connect(update_inventory_computed_fields, sender=Job)
-post_save.connect(update_inventory_computed_fields, sender=InventorySource)
-post_delete.connect(update_inventory_computed_fields, sender=InventorySource)
 post_save.connect(emit_job_event_detail, sender=JobEvent)
 
 # Migrate hosts, groups to parent group(s) whenever a group is deleted or

awx/main/tasks.py

@@ -37,9 +37,9 @@ from django.utils.timezone import now
 
 # AWX
 from awx.main.models import * # Job, JobEvent, ProjectUpdate, InventoryUpdate, Schedule, UnifiedJobTemplate
-from awx.main.utils import get_ansible_version, decrypt_field, update_scm_url
+from awx.main.utils import get_ansible_version, decrypt_field, update_scm_url, ignore_inventory_computed_fields
 
-__all__ = ['RunJob', 'RunProjectUpdate', 'RunInventoryUpdate', 'handle_work_error']
+__all__ = ['RunJob', 'RunProjectUpdate', 'RunInventoryUpdate', 'handle_work_error', 'update_inventory_computed_fields']
 
 HIDDEN_PASSWORD = '**********'
 
@@ -126,6 +126,18 @@ def handle_work_error(self, task_id, subtasks=None):
                 instance.save()
                 instance.socketio_emit_status("failed")
 
+@task()
+def update_inventory_computed_fields(inventory_id, should_update_hosts):
+    '''
+    Signal handler and wrapper around inventory.update_computed_fields to
+    prevent unnecessary recursive calls.
+    '''
+    i = Inventory.objects.filter(id=inventory_id)
+    if not i.exists():
+        logger.error("Update Inventory Computed Fields failed due to missing inventory: " + str(inventory_id))
+    i = i[0]
+    i.update_computed_fields(update_hosts=should_update_hosts)
+
 class BaseTask(Task):
 
     name = None

awx/main/utils.py

@@ -9,6 +9,8 @@ import re
 import subprocess
 import sys
 import urlparse
+import threading
+import contextlib
 
 # Django REST Framework
 from rest_framework.exceptions import ParseError, PermissionDenied
@@ -21,7 +23,8 @@ import zmq
 
 __all__ = ['get_object_or_400', 'get_object_or_403', 'camelcase_to_underscore',
            'get_ansible_version', 'get_awx_version', 'update_scm_url',
-           'get_type_for_model', 'get_model_for_type']
+           'get_type_for_model', 'get_model_for_type', 'ignore_inventory_computed_fields',
+           'ignore_inventory_group_removal', '_inventory_updates']
 
 def get_object_or_400(klass, *args, **kwargs):
     '''
@@ -350,3 +353,29 @@ def emit_websocket_notification(endpoint, event, payload):
         payload['event'] = event
         payload['endpoint'] = endpoint
         emit_socket.send_json(payload);
+
+_inventory_updates = threading.local()
+
+@contextlib.contextmanager
+def ignore_inventory_computed_fields():
+    '''
+    Context manager to ignore updating inventory computed fields.
+    '''
+    try:
+        previous_value = getattr(_inventory_updates, 'is_updating', False)
+        _inventory_updates.is_updating = True
+        yield
+    finally:
+        _inventory_updates.is_updating = previous_value
+
+@contextlib.contextmanager
+def ignore_inventory_group_removal():
+    '''
+    Context manager to ignore moving groups/hosts when group is deleted.
+    '''
+    try:
+        previous_value = getattr(_inventory_updates, 'is_removing', False)
+        _inventory_updates.is_removing = True
+        yield
+    finally:
+        _inventory_updates.is_removing = previous_value