def delete_proxy_user_for_job_execution(job_execution):
    '''Delete a proxy user based on a JobExecution

    :param job_execution: The job execution with proxy user information
    :returns: An updated job_configs dictionary or None

    '''
    proxy_configs = job_execution.job_configs.get('proxy_configs')
    if proxy_configs is not None:
        proxy_username = proxy_configs.get('proxy_username')
        key = key_manager.API().get(
            context.current(), proxy_configs.get('proxy_password'))
        proxy_password = key.get_encoded()
        proxy_trust_id = proxy_configs.get('proxy_trust_id')
        proxy_user = k.auth_for_proxy(proxy_username,
                                      proxy_password,
                                      proxy_trust_id)
        t.delete_trust(proxy_user, proxy_trust_id)
        proxy_user_delete(proxy_username)
        key_manager.API().delete(context.current(),
                                 proxy_configs.get('proxy_password'))
        update = job_execution.job_configs.to_dict()
        del update['proxy_configs']
        return update
    return None

def url_for(service_catalog=None, service_type="identity", endpoint_type="publicURL"):
    if not service_catalog:
        service_catalog = context.current().service_catalog
    try:
        return keystone_service_catalog.ServiceCatalogV2({"serviceCatalog": json.loads(service_catalog)}).url_for(
            service_type=service_type, endpoint_type=endpoint_type, region_name=CONF.os_region_name
        )
    except keystone_ex.EndpointNotFound:
        ctx = context.current()
        return keystone_service_catalog.ServiceCatalogV3(
            ctx.auth_token, {"catalog": json.loads(service_catalog)}
        ).url_for(service_type=service_type, endpoint_type=endpoint_type, region_name=CONF.os_region_name)

def execute_job(job_id, data):

    # Elements common to all job types
    cluster_id = data['cluster_id']
    configs = data.get('job_configs', {})

    ctx = context.current()
    cluster = conductor.cluster_get(ctx, cluster_id)
    plugin = plugin_base.PLUGINS.get_plugin(cluster.plugin_name)
    instance = plugin.get_oozie_server(cluster)

    extra = {}
    info = None
    if CONF.use_namespaces and not CONF.use_floating_ips:
        info = instance.remote().get_neutron_info()
        extra['neutron'] = info

    # Not in Java job types but present for all others
    input_id = data.get('input_id', None)
    output_id = data.get('output_id', None)

    # Since we will use a unified class in the database, we pass
    # a superset for all job types
    job_ex_dict = {'input_id': input_id, 'output_id': output_id,
                   'job_id': job_id, 'cluster_id': cluster_id,
                   'info': {'status': 'Pending'}, 'job_configs': configs,
                   'extra': extra}
    job_execution = conductor.job_execution_create(context.ctx(), job_ex_dict)

    context.spawn("Starting Job Execution %s" % job_execution.id,
                  manager.run_job, job_execution)
    return job_execution

    def _build_proxy_command(self, command, instance=None, port=None,
                             info=None, rootwrap_command=None):
        # Accepted keywords in the proxy command template:
        # {host}, {port}, {tenant_id}, {network_id}, {router_id}
        keywords = {}

        if not info:
            info = self.get_neutron_info(instance)
        keywords['tenant_id'] = context.current().tenant_id
        keywords['network_id'] = info['network']

        # Query Neutron only if needed
        if '{router_id}' in command:
            client = neutron.NeutronClient(info['network'], info['token'],
                                           info['tenant'])
            keywords['router_id'] = client.get_router()

        keywords['host'] = instance.management_ip
        keywords['port'] = port

        try:
            command = command.format(**keywords)
        except KeyError as e:
            LOG.error(_LE('Invalid keyword in proxy_command: {result}').format(
                result=e))
            # Do not give more details to the end-user
            raise ex.SystemError('Misconfiguration')
        if rootwrap_command:
            command = '{0} {1}'.format(rootwrap_command, command)
        return command

def retrieve_auth_url():
    """This function return auth url v2.0 api. Hadoop Swift library doesn't
    support keystone v3 api.
    """
    info = urlparse.urlparse(context.current().auth_uri)

    return "%s://%s:%s/%s/" % (info.scheme, info.hostname, info.port, 'v2.0')

def mount_to_instances(instances):
    if len(instances) == 0:
        return

    use_xfs = _can_use_xfs(instances)

    for instance in instances:
        with context.set_current_instance_id(instance.instance_id):
            devices = _find_instance_devices(instance)

            if devices:
                cpo.add_provisioning_step(
                    instance.cluster_id,
                    _("Mount volumes to {inst_name} instance").format(
                        inst_name=instance.instance_name), len(devices))

                formatted_devices = []
                lock = threading.Lock()
                with context.ThreadGroup() as tg:
                    # Since formating can take several minutes (for large
                    # disks) and can be done in parallel, launch one thread
                    # per disk.
                    for device in devices:
                        tg.spawn('format-device-%s' % device, _format_device,
                                 instance, device, use_xfs, formatted_devices,
                                 lock)

                conductor.instance_update(
                    context.current(), instance,
                    {"storage_devices_number": len(formatted_devices)})
                for idx, dev in enumerate(formatted_devices):
                    _mount_volume_to_node(instance, idx+1, dev, use_xfs)

def check_cluster_scaling(data, cluster_id, **kwargs):
    ctx = context.current()
    cluster = api.get_cluster(id=cluster_id)

    if cluster is None:
        raise ex.NotFoundException(
            {'id': cluster_id}, _('Object with %s not found'))

    b.check_plugin_labels(
        cluster.plugin_name, cluster.hadoop_version)

    acl.check_tenant_for_update(ctx, cluster)
    acl.check_protected_from_update(cluster, data)

    cluster_engine = cluster.sahara_info.get(
        'infrastructure_engine') if cluster.sahara_info else None

    engine_type_and_version = service_api.OPS.get_engine_type_and_version()
    if (not cluster_engine and
            not engine_type_and_version.startswith('direct')):
        raise ex.InvalidReferenceException(
            _("Cluster created before Juno release "
              "can't be scaled with %(engine)s engine") %
            {"engine": engine_type_and_version})

    if (cluster.sahara_info and
            cluster_engine != engine_type_and_version):
        raise ex.InvalidReferenceException(
            _("Cluster created with %(old_engine)s infrastructure engine "
              "can't be scaled with %(new_engine)s engine") %
            {"old_engine": cluster.sahara_info.get('infrastructure_engine'),
             "new_engine": engine_type_and_version})

    if not (plugin_base.PLUGINS.is_plugin_implements(cluster.plugin_name,
                                                     'scale_cluster') and (
            plugin_base.PLUGINS.is_plugin_implements(cluster.plugin_name,
                                                     'decommission_nodes'))):
        raise ex.InvalidReferenceException(
            _("Requested plugin '%s' doesn't support cluster scaling feature")
            % cluster.plugin_name)

    if cluster.status != c_u.CLUSTER_STATUS_ACTIVE:
        raise ex.InvalidReferenceException(
            _("Cluster cannot be scaled not in 'Active' status. "
              "Cluster status: %s") % cluster.status)

    if cluster.user_keypair_id:
        b.check_keypair_exists(cluster.user_keypair_id)

    if cluster.default_image_id:
        b.check_image_registered(cluster.default_image_id)

    if data.get("resize_node_groups"):
        b.check_resize(cluster, data['resize_node_groups'])

    if data.get("add_node_groups"):
        b.check_add_node_groups(cluster, data['add_node_groups'])
        b.check_cluster_hostnames_lengths(cluster.name,
                                          data['add_node_groups'])

def use_os_admin_auth_token(cluster):
    if cluster.trust_id:
        ctx = context.current()
        ctx.username = CONF.keystone_authtoken.admin_user
        ctx.tenant_id = cluster.tenant_id
        client = keystone.client_for_trusts(cluster.trust_id)
        ctx.token = client.auth_token
        ctx.service_catalog = json.dumps(client.service_catalog.catalog["catalog"])

def retrieve_auth_url():
    version = "v3" if CONF.use_identity_api_v3 else "v2.0"
    ctx = context.current()
    if ctx.service_catalog:
        info = urlparse.urlparse(url_for(ctx.service_catalog, "identity"))
    else:
        info = urlparse.urlparse(CONF.keystone_authtoken.auth_uri)
    return "%s://%s/%s" % (info[:2] + (version,))

def check_cinder_exists():
    service_type = 'volumev2'
    try:
        base.url_for(context.current().service_catalog, service_type,
                     endpoint_type=CONF.cinder.endpoint_type)
        return True
    except ex.SystemError:
        return False

def _retrieve_tenant():
    try:
        return context.current().tenant_name
    except RuntimeError:
        LOG.exception("Cannot retrieve tenant for swift integration. "
                      "Stopping cluster creation")
        #todo(slukjanov?) raise special error here
        raise RuntimeError("Cannot retrieve tenant for swift integration")

def retrieve_auth_url(endpoint_type="internalURL"):
    version = 'v3' if CONF.use_identity_api_v3 else 'v2.0'
    ctx = context.current()
    if ctx.service_catalog:
        info = urlparse.urlparse(url_for(ctx.service_catalog, 'identity',
                                         endpoint_type))
    else:
        info = urlparse.urlparse(CONF.keystone_authtoken.auth_uri)
    return "%s://%s/%s" % (info[:2] + (version,))

def client():
    ctx = context.current()
    heat_url = base.url_for(ctx.service_catalog, 'orchestration',
                            endpoint_type=CONF.heat.endpoint_type)
    return heat_client.Client('1', heat_url, token=ctx.auth_token,
                              cert_file=CONF.heat.ca_file,
                              insecure=CONF.heat.api_insecure,
                              username=ctx.username,
                              include_pass=True)

def retrieve_auth_url(endpoint_type="internalURL", version=None):
    if not version:
        version = 'v3' if CONF.use_identity_api_v3 else 'v2.0'
    ctx = context.current()
    if ctx.service_catalog:
        auth_url = url_for(ctx.service_catalog, 'identity', endpoint_type)
    else:
        auth_url = CONF.keystone_authtoken.auth_uri
    return prepare_auth_url(auth_url, version)

def get_raw_data(job_binary, proxy_configs=None):
    conn_kwargs = {}
    if proxy_configs:
        key = key_manager.API().get(context.current(),
                                    proxy_configs.get('proxy_password'))
        password = key.get_encoded()
        conn_kwargs.update(username=proxy_configs.get('proxy_username'),
                           password=password,
                           trust_id=proxy_configs.get('proxy_trust_id'))
    else:
        key = key_manager.API().get(context.current(),
                                    job_binary.extra.get('password'))
        password = key.get_encoded()
        conn_kwargs.update(username=job_binary.extra.get('user'),
                           password=password)

    conn = sw.client(**conn_kwargs)
    return _get_raw_data(job_binary, conn)

def delete_secret(id, ctx=None):
    """delete a secret from the external key manager

    :param id: The identifier of the secret to delete
    :param ctx: The context, and associated authentication, to use with
                this operation (defaults to the current context)
    """
    if ctx is None:
        ctx = context.current()
    key_manager.API().delete(ctx, id)

def check_cinder_exists():
    if CONF.cinder.api_version == 1:
        service_type = 'volume'
    else:
        service_type = 'volumev2'
    try:
        base.url_for(context.current().service_catalog, service_type)
        return True
    except ex.SystemError:
        return False

def get_secret(id, ctx=None):
    """get a secret associated with an id

    :param id: The identifier of the secret to retrieve
    :param ctx: The context, and associated authentication, to use with
                this operation (defaults to the current context)
    """
    if ctx is None:
        ctx = context.current()
    key = key_manager.API().get(ctx, id)
    return key.get_encoded()

def store_secret(secret, ctx=None):
    """store a secret and return its identifier

    :param secret: The secret to store, this should be a string
    :param ctx: The context, and associated authentication, to use with
                this operation (defaults to the current context)
    """
    if ctx is None:
        ctx = context.current()
    key = passphrase.Passphrase(secret)
    return key_manager.API().store(ctx, key)

def check_cinder_exists():
    if CONF.cinder.api_version == 2:
        service_type = 'volumev2'
    else:
        service_type = 'volumev3'
    try:
        base.url_for(context.current().service_catalog, service_type,
                     endpoint_type=CONF.cinder.endpoint_type)
        return True
    except keystone_exceptions.EndpointNotFound:
        return False