wrap instance and queue registration in postgres advisory locks

see: #7040

awx/main/management/commands/deprovision_node.py

@@ -1,10 +1,14 @@
 # Copyright (c) 2016 Ansible, Inc.
 # All Rights Reserved
 
-import subprocess
-from django.core.management.base import BaseCommand, CommandError
 from optparse import make_option
+import subprocess
+
+from django.db import transaction
+from django.core.management.base import BaseCommand, CommandError
+
 from awx.main.models import Instance
+from awx.main.utils.pglock import advisory_lock
 
 
 class Command(BaseCommand):
@@ -17,19 +21,22 @@ class Command(BaseCommand):
                     help='Hostname used during provisioning'),
     )
 
+    @transaction.atomic
     def handle(self, *args, **options):
-        if not options.get('name'):
+        hostname = options.get('name')
+        if not hostname:
             raise CommandError("--name is a required argument")
-        instance = Instance.objects.filter(hostname=options.get('name'))
+        with advisory_lock('instance_registration_%s' % hostname):
-        if instance.exists():
+            instance = Instance.objects.filter(hostname=hostname)
-            instance.delete()
+            if instance.exists():
-            print("Instance Removed")
+                instance.delete()
-            result = subprocess.Popen("rabbitmqctl forget_cluster_node rabbitmq@{}".format(options.get('name')), shell=True).wait()
+                print("Instance Removed")
-            if result != 0:
+                result = subprocess.Popen("rabbitmqctl forget_cluster_node rabbitmq@{}".format(hostname), shell=True).wait()
-                print("Node deprovisioning may have failed when attempting to remove the RabbitMQ instance from the cluster")
+                if result != 0:
+                    print("Node deprovisioning may have failed when attempting to remove the RabbitMQ instance from the cluster")
+                else:
+                    print('Successfully deprovisioned {}'.format(hostname))
+                print('(changed: True)')
             else:
-                print('Successfully deprovisioned {}'.format(options.get('name')))
+                print('No instance found matching name {}'.format(hostname))
-            print('(changed: True)')
-        else:
-            print('No instance found matching name {}'.format(options.get('name')))
 

awx/main/management/commands/register_instance.py

@@ -2,9 +2,11 @@
 # All Rights Reserved
 
 from awx.main.models import Instance
+from awx.main.utils.pglock import advisory_lock
 from django.conf import settings
 
 from optparse import make_option
+from django.db import transaction
 from django.core.management.base import BaseCommand
 
 
@@ -25,15 +27,17 @@ class Command(BaseCommand):
     def _register_hostname(self, hostname):
         if not hostname:
             return
-        instance = Instance.objects.filter(hostname=hostname)
+        with advisory_lock('instance_registration_%s' % hostname):
-        if instance.exists():
+            instance = Instance.objects.filter(hostname=hostname)
-            print("Instance already registered {}".format(instance[0]))
+            if instance.exists():
-            return
+                print("Instance already registered {}".format(instance[0]))
-        instance = Instance(uuid=self.uuid, hostname=hostname)
+                return
-        instance.save()
+            instance = Instance(uuid=self.uuid, hostname=hostname)
+            instance.save()
         print('Successfully registered instance {}'.format(hostname))
         self.changed = True
 
+    @transaction.atomic
     def handle(self, **options):
         self.uuid = settings.SYSTEM_UUID
         self.changed = False

awx/main/management/commands/register_queue.py

@@ -2,9 +2,11 @@
 # All Rights Reserved.
 import sys
 
+from awx.main.utils.pglock import advisory_lock
 from awx.main.models import Instance, InstanceGroup
 
 from optparse import make_option
+from django.db import transaction
 from django.core.management.base import BaseCommand, CommandError
 
 
@@ -19,45 +21,48 @@ class Command(BaseCommand):
                     help='The controlling group (makes this an isolated group)'),
     )
 
+    @transaction.atomic
     def handle(self, **options):
-        if not options.get('queuename'):
+        queuename = options.get('queuename')
+        if not queuename:
             raise CommandError("Specify `--queuename` to use this command.")
         changed = False
-        ig = InstanceGroup.objects.filter(name=options.get('queuename'))
+        with advisory_lock('instance_group_registration_%s' % queuename):
-        control_ig = None
+            ig = InstanceGroup.objects.filter(name=queuename)
-        if options.get('controller'):
+            control_ig = None
-            control_ig = InstanceGroup.objects.filter(name=options.get('controller')).first()
+            if options.get('controller'):
-        if ig.exists():
+                control_ig = InstanceGroup.objects.filter(name=options.get('controller')).first()
-            print("Instance Group already registered {}".format(ig[0].name))
+            if ig.exists():
-            ig = ig[0]
+                print("Instance Group already registered {}".format(ig[0].name))
-            if control_ig and ig.controller_id != control_ig.pk:
+                ig = ig[0]
-                ig.controller = control_ig
+                if control_ig and ig.controller_id != control_ig.pk:
-                ig.save()
+                    ig.controller = control_ig
-                print("Set controller group {} on {}.".format(control_ig.name, ig.name))
+                    ig.save()
-                changed = True
+                    print("Set controller group {} on {}.".format(control_ig.name, ig.name))
-        else:
+                    changed = True
-            print("Creating instance group {}".format(options.get('queuename')))
-            ig = InstanceGroup(name=options.get('queuename'))
-            if control_ig:
-                ig.controller = control_ig
-            ig.save()
-            changed = True
-        hostname_list = []
-        if options.get('hostnames'):
-            hostname_list = options.get('hostnames').split(",")
-        instance_list = [x.strip() for x in hostname_list if x]
-        for inst_name in instance_list:
-            instance = Instance.objects.filter(hostname=inst_name)
-            if instance.exists() and instance[0] not in ig.instances.all():
-                ig.instances.add(instance[0])
-                print("Added instance {} to {}".format(instance[0].hostname, ig.name))
-                changed = True
-            elif not instance.exists():
-                print("Instance does not exist: {}".format(inst_name))
-                if changed:
-                    print('(changed: True)')
-                sys.exit(1)
             else:
-                print("Instance already registered {}".format(instance[0].hostname))
+                print("Creating instance group {}".format(queuename))
-        if changed:
+                ig = InstanceGroup(name=queuename)
-            print('(changed: True)')
+                if control_ig:
+                    ig.controller = control_ig
+                ig.save()
+                changed = True
+            hostname_list = []
+            if options.get('hostnames'):
+                hostname_list = options.get('hostnames').split(",")
+            instance_list = [x.strip() for x in hostname_list if x]
+            for inst_name in instance_list:
+                instance = Instance.objects.filter(hostname=inst_name)
+                if instance.exists() and instance[0] not in ig.instances.all():
+                    ig.instances.add(instance[0])
+                    print("Added instance {} to {}".format(instance[0].hostname, ig.name))
+                    changed = True
+                elif not instance.exists():
+                    print("Instance does not exist: {}".format(inst_name))
+                    if changed:
+                        print('(changed: True)')
+                    sys.exit(1)
+                else:
+                    print("Instance already registered {}".format(instance[0].hostname))
+            if changed:
+                print('(changed: True)')

awx/main/management/commands/unregister_queue.py

@@ -2,9 +2,11 @@
 # All Rights Reserved.
 import sys
 
+from awx.main.utils.pglock import advisory_lock
 from awx.main.models import InstanceGroup
 
 from optparse import make_option
+from django.db import transaction
 from django.core.management.base import BaseCommand, CommandError
 
 
@@ -20,14 +22,17 @@ class Command(BaseCommand):
                     help='Queue to create/update'),
     )
 
+    @transaction.atomic
     def handle(self, **options):
-        if not options.get('queuename'):
+        queuename = options.get('queuename')
+        if not queuename:
             raise CommandError('Must specify `--queuename` in order to use command.')
-        ig = InstanceGroup.objects.filter(name=options.get('queuename'))
+        with advisory_lock('instance_group_registration_%s' % queuename):
-        if not ig.exists():
+            ig = InstanceGroup.objects.filter(name=queuename)
-            print("Instance group doesn't exist")
+            if not ig.exists():
-            sys.exit(1)
+                print("Instance group doesn't exist")
-        ig = ig.first()
+                sys.exit(1)
-        ig.delete()
+            ig = ig.first()
-        print("Instance Group Removed")
+            ig.delete()
-        print('(changed: True)')
+            print("Instance Group Removed")
+            print('(changed: True)')