Using locks for VM dynamic scenarios

This patch introduces various functions related to locks in dynamic_utils.py,
which are used in VM dynamic scenarios to prevent the same server from being used
in multiple scenarios. It also introduces a function to release a lock.

Change-Id: I36b5a0f8c0db2c897ba926327570f7bbdf1591c6

ansible/install/roles/rally/files/browbeat-rally/browbeat_rally/db/api.py

@@ -30,3 +30,10 @@ def lock_list(session):
     for lock in query.all():
         locks.append(lock.as_dict())
     return locks
+
+
+@db.with_session
+def release_lock(session, lock_uuid):
+    session.query(models.RallyLock).filter_by(
+        lock_uuid=lock_uuid).delete(
+        synchronize_session=False)

rally/rally-plugins/dynamic-workloads/README.rst

@@ -37,6 +37,10 @@ Functions:
 - _get_servers_by_tag: Retrieve list of servers based on tag
 - dissociate_and_delete_floating_ip: Dissociate and delete floating IP of port
 - create_floating_ip_and_associate_to_port: Create and associate floating IP for port
+- acquire_lock: Acquire lock on object
+- list_locks: List all locks in database
+- release_lock: Release lock on object
+- cleanup_locks: Release all locks in database
 - provider_netcreate_nova_boot_ping: Creates a provider Network and Boots VM and ping
 - provider_net_nova_boot_ping: Boots a VM and ping on random existing provider network
 - provider_net_nova_delete: Delete all VM's and provider network

rally/rally-plugins/dynamic-workloads/dynamic_utils.py

@@ -19,6 +19,9 @@ from rally_openstack.scenarios.neutron import utils as neutron_utils
 from rally.task import atomic
 from rally.task import utils
 
+from browbeat_rally.db import api as db_api
+from oslo_db import exception as db_exc
+
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
@@ -190,3 +193,35 @@ class NeutronUtils(neutron_utils.NeutronScenario):
             port_fip["id"], {"floatingip": fip_update_dict}
         )
         return port_fip
+
+class LockUtils:
+
+    def acquire_lock(self, object_id):
+        """Acquire lock on object
+        :param object_id: id of object to lock
+        :returns: bool, whether the lock was acquired successfully or not
+        """
+        try:
+            db_api.get_lock(object_id)
+            return True
+        except db_exc.DBDuplicateEntry:
+            return False
+
+    def list_locks(self):
+        """List all locks in database
+        :returns: list, list of lock dictionaries
+        """
+        return db_api.lock_list()
+
+    def release_lock(self, object_id):
+        """Release lock on object
+        :param object_id: id of object to release lock from
+        """
+        db_api.release_lock(object_id)
+
+    def cleanup_locks(self):
+        """Release all locks in database
+        """
+        locks_list = self.list_locks()
+        for lock in locks_list:
+            self.release_lock(lock["lock_uuid"])

rally/rally-plugins/dynamic-workloads/vm.py

@@ -19,7 +19,8 @@ LOG = logging.getLogger(__name__)
 
 
 class VMDynamicScenario(dynamic_utils.NovaUtils,
-                        dynamic_utils.NeutronUtils):
+                        dynamic_utils.NeutronUtils,
+                        dynamic_utils.LockUtils):
 
     def boot_servers(self, image, flavor, num_vms=2,
                      network_create_args=None, subnet_create_args=None):
@@ -49,8 +50,11 @@ class VMDynamicScenario(dynamic_utils.NovaUtils,
 
         servers_to_delete = random.sample(eligible_servers, num_vms)
         for server in servers_to_delete:
+            server_id = server.id
+            self.acquire_lock(server_id)
             LOG.info("Deleting server {}".format(server))
             self._delete_server(server, force=True)
+            self.release_lock(server_id)
 
     def boot_servers_with_fip(self, image, flavor, ext_net_id, num_vms=1, router_create_args=None,
                               network_create_args=None, subnet_create_args=None, **kwargs):
@@ -83,7 +87,7 @@ class VMDynamicScenario(dynamic_utils.NovaUtils,
         for i in range(num_vms):
             kwargs["nics"] = [{"net-id": network["network"]["id"]}]
             guest = self._boot_server_with_fip_and_tag(
-                image, flavor, "migrate",
+                image, flavor, "migrate_or_swap",
                 True, ext_net_name, **kwargs
             )
             self._wait_for_ping(guest[1]["ip"])
@@ -94,9 +98,9 @@ class VMDynamicScenario(dynamic_utils.NovaUtils,
         :param num_migrate_vms: int, number of servers to migrate between computes
         :returns: list of server objects to migrate between computes
         """
-        eligible_servers = self._get_servers_by_tag("migrate")
+        eligible_servers = self._get_servers_by_tag("migrate_or_swap")
 
-        num_servers_to_migrate = min(num_migrate_vms, len(eligible_servers))
+        num_servers_to_migrate = min(2*num_migrate_vms, len(eligible_servers))
         list_of_servers_to_migrate = random.sample(eligible_servers, num_servers_to_migrate)
 
         return list_of_servers_to_migrate
@@ -106,11 +110,29 @@ class VMDynamicScenario(dynamic_utils.NovaUtils,
 
         :param num_migrate_vms: int, number of servers to migrate between computes
         """
-        for server in self.get_servers_migration_list(num_migrate_vms):
-            fip = list(server.addresses.values())[0][1]['addr']
+        server_migration_list = self.get_servers_migration_list(num_migrate_vms)
+
+        loop_counter = 0
+        num_migrated = 0
+        length_server_migration_list = len(server_migration_list)
+
+        while loop_counter < length_server_migration_list and num_migrated < num_migrate_vms:
+            server_to_migrate = server_migration_list[loop_counter]
+
+            loop_counter += 1
+            if not self.acquire_lock(server_to_migrate.id):
+                continue
+
+            fip = list(server_to_migrate.addresses.values())[0][1]['addr']
             LOG.info("ping {} before server migration".format(fip))
             self._wait_for_ping(fip)
-            self._migrate(server)
-            self._resize_confirm(server, status="ACTIVE")
+            self._migrate(server_to_migrate)
+            self._resize_confirm(server_to_migrate, status="ACTIVE")
             LOG.info("ping {} after server migration".format(fip))
             self._wait_for_ping(fip)
+            self.release_lock(server_to_migrate.id)
+            num_migrated += 1
+
+        if num_migrated == 0:
+            LOG.info("""No servers which are not under lock, so
+                      cannot migrate any servers.""")