def _migrate(self, server, skip_host_check=False):
        """Run migration of the given server.

        :param server: Server object
        :param skip_host_check: Specifies whether to verify the targeted host
                                availability
        """
        server_admin = self.admin_clients("nova").servers.get(server.id)
        host_pre_migrate = getattr(server_admin, "OS-EXT-SRV-ATTR:host")
        server_admin.migrate()
        bench_utils.wait_for(
            server,
            is_ready=bench_utils.resource_is("VERIFY_RESIZE"),
            update_resource=bench_utils.get_from_manager(),
            timeout=CONF.benchmark.nova_server_migrate_timeout,
            check_interval=(
                CONF.benchmark.nova_server_migrate_poll_interval)
        )
        if not skip_host_check:
            server_admin = self.admin_clients("nova").servers.get(server.id)
            host_after_migrate = getattr(server_admin, "OS-EXT-SRV-ATTR:host")
            if host_pre_migrate == host_after_migrate:
                raise exceptions.MigrateException(
                    "Migration complete but instance did not change host: %s" %
                    host_pre_migrate)

    def _live_migrate(self, server, target_host, block_migration=False,
                      disk_over_commit=False, skip_host_check=False):
        """Run live migration of the given server.

        :param server: Server object
        :param target_host: Specifies the target compute node to migrate
        :param block_migration: Specifies the migration type
        :param disk_over_commit: Specifies whether to overcommit migrated
                                 instance or not
        :param skip_host_check: Specifies whether to verify the targeted host
                                availability
        """
        server_admin = self.admin_clients("nova").servers.get(server.id)
        host_pre_migrate = getattr(server_admin, "OS-EXT-SRV-ATTR:host")
        server_admin.live_migrate(target_host,
                                  block_migration=block_migration,
                                  disk_over_commit=disk_over_commit)
        bench_utils.wait_for(
            server,
            is_ready=bench_utils.resource_is("ACTIVE"),
            update_resource=bench_utils.get_from_manager(),
            timeout=CONF.benchmark.nova_server_live_migrate_timeout,
            check_interval=(
                CONF.benchmark.nova_server_live_migrate_poll_interval)
        )
        server_admin = self.admin_clients("nova").servers.get(server.id)
        if (host_pre_migrate == getattr(server_admin, "OS-EXT-SRV-ATTR:host")
                and not skip_host_check):
            raise exceptions.LiveMigrateException(
                "Migration complete but instance did not change host: %s" %
                host_pre_migrate)

    def test_wait_for(self):

        def get_fake_checker_delayed(**delay):
            deadline = datetime.datetime.now() + datetime.timedelta(**delay)
            return lambda obj: datetime.datetime.now() > deadline

        def fake_checker_false(obj):
            return False

        def fake_updater(obj):
            return obj

        resource = object()
        fake_checker_delayed = get_fake_checker_delayed(seconds=0.3)

        loaded_resource = utils.wait_for(resource, fake_checker_delayed,
                                         fake_updater, 1, 0.2)
        self.assertEqual(loaded_resource, resource)

        loaded_resource = utils.wait_for(resource, fake_checker_delayed,
                                         None, 1, 0.2)
        self.assertEqual(loaded_resource, resource)

        self.assertRaises(exceptions.TimeoutException, utils.wait_for,
                          object(), fake_checker_false, fake_updater, 0.3, 0.1)

    def _wait_finish_of_deploy(self, environment):

        bench_utils.wait_for(
            environment, is_ready=bench_utils.resource_is("READY"),
            update_resource=bench_utils.get_from_manager(),
            timeout=1200.0,
            check_interval=5.0
        )

def _wait_for_list_size(mgr, sizes=[0], timeout=10, check_interval=1):
    bench_utils.wait_for(
        mgr,
        is_ready=bench_utils.manager_list_size(sizes),
        update_resource=None,
        timeout=timeout,
        check_interval=check_interval,
    )

 def _wait_for_ping(self, server_ip):
     server_ip = netaddr.IPAddress(server_ip)
     bench_utils.wait_for(
         server_ip,
         is_ready=bench_utils.resource_is(ICMP_UP_STATUS,
                                          self._ping_ip_address),
         timeout=120
     )

 def _resize(self, server, flavor):
     server.resize(flavor)
     bench_utils.wait_for(
         server,
         is_ready=bench_utils.resource_is("VERIFY_RESIZE"),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_server_resize_timeout,
         check_interval=CONF.benchmark.nova_server_resize_poll_interval
     )

 def _do_server_reboot(self, server, reboottype):
     server.reboot(reboot_type=reboottype)
     time.sleep(CONF.benchmark.nova_server_reboot_prepoll_delay)
     bench_utils.wait_for(
         server, is_ready=bench_utils.resource_is("ACTIVE"),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_server_reboot_timeout,
         check_interval=CONF.benchmark.nova_server_reboot_poll_interval
     )

    def setup(self):
        ready_tenants = set()
        wait_dict = dict()

        for user in self.context.get("users", []):
            tenant_id = user["tenant_id"]
            if tenant_id not in ready_tenants:
                ready_tenants.add(tenant_id)

                clients = osclients.Clients(user["endpoint"])

                image_id = self.context["sahara_images"][tenant_id]

                neutron_net = self.config.get("neutron_net")
                if not neutron_net:
                    # Skipping fixed network config
                    neutron_net_id = None
                else:
                    network_cfg = {"name": neutron_net}
                    neutron_net_id = (types.NeutronNetworkResourceType
                                      .transform(clients, network_cfg))

                floating_ip_pool = self.config.get("floating_ip_pool")
                if not floating_ip_pool:
                    # Skipping floating network config
                    floating_ip_pool_id = None
                else:
                    network_cfg = {"name": floating_ip_pool}
                    floating_ip_pool_id = (types.NeutronNetworkResourceType
                                           .transform(clients, network_cfg))

                cluster = utils.SaharaScenario(
                    context=self.context, clients=clients)._launch_cluster(
                        plugin_name=self.config["plugin_name"],
                        hadoop_version=self.config["hadoop_version"],
                        flavor_id=self.config["flavor_id"],
                        node_count=self.config["node_count"],
                        image_id=image_id,
                        floating_ip_pool=floating_ip_pool_id,
                        neutron_net_id=neutron_net_id,
                        volumes_per_node=self.config.get("volumes_per_node"),
                        volumes_size=self.config.get("volumes_size", 1),
                        node_configs=self.config.get("node_configs"),
                        cluster_configs=self.config.get("cluster_configs"),
                        wait_active=False)

                self.context["sahara_clusters"][tenant_id] = cluster.id

                # Need to save the client instance to poll for active status
                wait_dict[cluster] = clients.sahara()

        bench_utils.wait_for(
            resource=wait_dict,
            update_resource=self.update_clusters_dict,
            is_ready=self.all_clusters_active,
            timeout=CONF.benchmark.cluster_create_timeout,
            check_interval=CONF.benchmark.cluster_check_interval)

 def _resize_confirm(self, server):
     server.confirm_resize()
     bench_utils.wait_for(
         server,
         is_ready=bench_utils.resource_is("ACTIVE"),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_server_resize_confirm_timeout,
         check_interval=(
             CONF.benchmark.nova_server_resize_confirm_poll_interval)
     )

 def _resize_revert(self, server, status="ACTIVE"):
     server.revert_resize()
     bench_utils.wait_for(
         server,
         is_ready=bench_utils.resource_is(status),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_server_resize_revert_timeout,
         check_interval=(
             CONF.benchmark.nova_server_resize_revert_poll_interval)
     )

    def _delete_cluster(self, cluster):
        """Calls a Cluster delete by id and waits for complete deletion.

        :param cluster: The Cluster to be deleted
        :return:
        """

        self.clients("sahara").clusters.delete(cluster.id)

        bench_utils.wait_for(resource=cluster.id,
                             is_ready=self._is_cluster_deleted)

 def _detach_volume(self, server, volume):
     server_id = server.id
     volume_id = volume.id
     self.clients("nova").volumes.delete_server_volume(server_id,
                                                       volume_id)
     bench_utils.wait_for(
         volume,
         is_ready=bench_utils.resource_is("available"),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_detach_volume_timeout,
         check_interval=CONF.benchmark.nova_detach_volume_poll_interval
     )

    def _delete_cluster(self, cluster):
        """Delete cluster.

        :param cluster: cluster to delete
        """
        self.clients("sahara").clusters.delete(cluster.id)

        bench_utils.wait_for(
            resource=cluster.id,
            timeout=CONF.benchmark.cluster_delete_timeout,
            check_interval=CONF.benchmark.cluster_check_interval,
            is_ready=self._is_cluster_deleted)

def _wait_for_list_statuses(mgr, statuses, list_query=None, timeout=10, check_interval=1):
    list_query = list_query or {}

    def _list_statuses(mgr):
        for resource in mgr.list(**list_query):
            if resource.status not in statuses:
                return False
        return True

    bench_utils.wait_for(
        mgr, is_ready=_list_statuses, update_resource=None, timeout=timeout, check_interval=check_interval
    )

        def run(self):
            job_execution = self.clients("sahara").job_executions.create(
                job_id=job_id,
                cluster_id=cluster_id,
                input_id=input_id,
                output_id=output_id,
                configs=configs)

            bench_utils.wait_for(
                resource=job_execution.id,
                is_ready=self._job_execution_is_finished,
                timeout=CONF.benchmark.job_execution_timeout,
                check_interval=CONF.benchmark.job_check_interval)

    def _suspend_stack(self, stack):
        """Suspend given stack.

        :param stack: stack that needs to be suspended
        """

        self.clients("heat").actions.suspend(stack.id)
        bench_utils.wait_for(
            stack,
            is_ready=bench_utils.resource_is("SUSPEND_COMPLETE"),
            update_resource=bench_utils.get_from_manager(
                ["SUSPEND_FAILED"]),
            timeout=CONF.benchmark.heat_stack_suspend_timeout,
            check_interval=CONF.benchmark.heat_stack_suspend_poll_interval)

    def _resume_stack(self, stack):
        """Resume given stack.

        :param stack: stack that needs to be resumed
        """

        self.clients("heat").actions.resume(stack.id)
        bench_utils.wait_for(
            stack,
            is_ready=bench_utils.resource_is("RESUME_COMPLETE"),
            update_resource=bench_utils.get_from_manager(
                ["RESUME_FAILED"]),
            timeout=CONF.benchmark.heat_stack_resume_timeout,
            check_interval=CONF.benchmark.heat_stack_resume_poll_interval)

    def _deploy_environment(self, environment, session):
        """Deploy environment.

        :param environment: Environment instance
        :param session: Session instance
        """
        self.clients("murano").sessions.deploy(environment.id,
                                               session.id)
        bench_utils.wait_for(
            environment, is_ready=bench_utils.resource_is("READY"),
            update_resource=bench_utils.get_from_manager(["DEPLOY FAILURE"]),
            timeout=CONF.benchmark.deploy_environment_timeout,
            check_interval=CONF.benchmark.deploy_environment_check_interval
        )

 def _attach_volume(self, server, volume, device=None):
     server_id = server.id
     volume_id = volume.id
     self.clients("nova").volumes.create_server_volume(server_id,
                                                       volume_id,
                                                       device)
     bench_utils.wait_for(
         volume,
         is_ready=bench_utils.resource_is("in-use"),
         update_resource=bench_utils.get_from_manager(),
         timeout=CONF.benchmark.nova_server_resize_revert_timeout,
         check_interval=(
             CONF.benchmark.nova_server_resize_revert_poll_interval)
     )