def _run_operation(self, operation_id, param):

        self._update_operation_state(
            operation_id,
            {'state': constants.OPERATION_STATE_RUNNING})

        try:
            check_item = [self._CHECK_ITEMS['is_canceled']]
            if self._check_operation(operation_id, check_item):
                return

            try:
                operation = objects.ScheduledOperation.get_by_id(
                    context.get_admin_context(), operation_id)
            except Exception:
                LOG.exception(_LE("Run operation(%s), get operation failed"),
                              operation_id)
                return

            try:
                self._operation_manager.run_operation(
                    operation.operation_type, operation.project_id,
                    operation.operation_definition,
                    param=param)
            except Exception:
                LOG.exception(_LE("Run operation(%s) failed"), operation_id)

        finally:
            self._update_operation_state(
                operation_id,
                {'state': constants.OPERATION_STATE_REGISTERED})

    def report_state(self):
        """Update the state of this service in the datastore."""
        if not self.manager.is_working():
            # NOTE(dulek): If manager reports a problem we're not sending
            # heartbeats - to indicate that service is actually down.
            LOG.error(_LE('Manager for service %(binary)s %(host)s is '
                          'reporting problems, not sending heartbeat. '
                          'Service will appear "down".'),
                      {'binary': self.binary,
                       'host': self.host})
            return

        ctxt = context.get_admin_context()
        state_catalog = {}
        try:
            try:
                service_ref = db.service_get(ctxt, self.service_id)
            except exception.NotFound:
                LOG.debug('The service database object disappeared, '
                          'recreating it.')
                self._create_service_ref(ctxt)
                service_ref = db.service_get(ctxt, self.service_id)

            state_catalog['report_count'] = service_ref['report_count'] + 1

            db.service_update(ctxt,
                              self.service_id, state_catalog)

            # TODO(termie): make this pattern be more elegant.
            if getattr(self, 'model_disconnected', False):
                self.model_disconnected = False
                LOG.error(_LE('Recovered model server connection!'))

        except db_exc.DBConnectionError:
            if not getattr(self, 'model_disconnected', False):
                self.model_disconnected = True
                LOG.exception(_LE('model server went away'))

        # NOTE(jsbryant) Other DB errors can happen in HA configurations.
        # such errors shouldn't kill this thread, so we handle them here.
        except db_exc.DBError:
            if not getattr(self, 'model_disconnected', False):
                self.model_disconnected = True
                LOG.exception(_LE('DBError encountered: '))

        except Exception:
            if not getattr(self, 'model_disconnected', False):
                self.model_disconnected = True
                LOG.exception(_LE('Exception encountered: '))

def load_plugin(namespace, plugin_name, *args, **kwargs):
    try:
        LOG.debug('Start load plugin %s. ', plugin_name)
        # Try to resolve plugin by name
        mgr = driver.DriverManager(namespace, plugin_name)
        plugin_class = mgr.driver
    except RuntimeError as e1:
        # fallback to class name
        try:
            plugin_class = importutils.import_class(plugin_name)
        except ImportError as e2:
            LOG.exception(_LE("Error loading plugin by name, %s"), e1)
            LOG.exception(_LE("Error loading plugin by class, %s"), e2)
            raise ImportError(_("Class not found."))
    return plugin_class(*args, **kwargs)

    def __init__(self, provider_config):
        super(PluggableProtectionProvider, self).__init__()
        self._config = provider_config
        self._id = self._config.provider.id
        self._name = self._config.provider.name
        self._description = self._config.provider.description
        self._extended_info_schema = {'options_schema': {},
                                      'restore_schema': {},
                                      'saved_info_schema': {}}
        self.checkpoint_collection = None
        self._bank_plugin = None
        self._plugin_map = {}

        if hasattr(self._config.provider, 'bank') \
                and not self._config.provider.bank:
            raise ImportError("Empty bank")
        self._load_bank(self._config.provider.bank)
        if hasattr(self._config.provider, 'plugin'):
            for plugin_name in self._config.provider.plugin:
                if not plugin_name:
                    raise ImportError("Empty protection plugin")
                self._load_plugin(plugin_name)

        if self._bank_plugin:
            self.checkpoint_collection = checkpoint.CheckpointCollection(
                self._bank_plugin)
        else:
            LOG.error(_LE('Bank plugin not exist, check your configuration'))

 def output(self, flow_engine, target=None):
     if flow_engine is None:
         LOG.error(_LE("Flow engine is None,return nothing"))
         return
     if target:
         return flow_engine.storage.fetch(target)
     return flow_engine.storage.fetch_all()

    def create_backup(self, cntxt, checkpoint, **kwargs):
        resource_node = kwargs.get("node")
        image_id = resource_node.value.id
        bank_section = checkpoint.get_resource_bank_section(image_id)

        resource_definition = {"resource_id": image_id}
        glance_client = self._glance_client(cntxt)

        LOG.info(_("creating image backup, image_id: %s."), image_id)
        try:
            bank_section.create_object("status",
                                       constants.RESOURCE_STATUS_PROTECTING)
            image_info = glance_client.images.get(image_id)
            image_metadata = {
                "disk_format": image_info.disk_format,
                "container_format": image_info.container_format
            }
            resource_definition["image_metadata"] = image_metadata
            resource_definition["backup_id"] = image_id

            bank_section.create_object("metadata", resource_definition)
        except Exception as err:
            LOG.error(_LE("create image backup failed, image_id: %s."),
                      image_id)
            bank_section.update_object("status",
                                       constants.RESOURCE_STATUS_ERROR)
            raise exception.CreateBackupFailed(
                reason=err,
                resource_id=image_id,
                resource_type=constants.IMAGE_RESOURCE_TYPE)

        self._add_to_threadpool(self._create_backup, glance_client,
                                bank_section, image_id)

    def list_protectable_instances(self, context,
                                   protectable_type=None,
                                   marker=None,
                                   limit=None,
                                   sort_keys=None,
                                   sort_dirs=None,
                                   filters=None):

        LOG.info(_LI("Start to list protectable instances of type: %s"),
                 protectable_type)

        try:
            resource_instances = self.protectable_registry.list_resources(
                context, protectable_type)
        except exception.ListProtectableResourceFailed as err:
            LOG.error(_LE("List resources of type %(type)s failed: %(err)s"),
                      {'type': protectable_type,
                       'err': six.text_type(err)})
            raise

        result = []
        for resource in resource_instances:
            result.append(dict(id=resource.id, name=resource.name))

        return result

    def list_protectable_dependents(self, context,
                                    protectable_id,
                                    protectable_type):
        LOG.info(_LI("Start to list dependents of resource "
                     "(type:%(type)s, id:%(id)s)"),
                 {'type': protectable_type,
                  'id': protectable_id})

        parent_resource = Resource(type=protectable_type, id=protectable_id,
                                   name="")

        try:
            dependent_resources = \
                self.protectable_registry.fetch_dependent_resources(
                    context, parent_resource)
        except exception.ListProtectableResourceFailed as err:
            LOG.error(_LE("List dependent resources of (%(res)s) "
                          "failed: %(err)s"),
                      {'res': parent_resource,
                       'err': six.text_type(err)})
            raise

        result = []
        for resource in dependent_resources:
            result.append(dict(type=resource.type, id=resource.id,
                               name=resource.name))

        return result

    def delete_backup(self, cntxt, checkpoint, **kwargs):
        resource_node = kwargs.get("node")
        resource_id = resource_node.value.id

        bank_section = checkpoint.get_resource_bank_section(resource_id)
        cinder_client = self._cinder_client(cntxt)

        LOG.info(_("deleting volume backup, volume_id: %s."), resource_id)
        try:
            bank_section.update_object("status",
                                       constants.RESOURCE_STATUS_DELETING)
            resource_definition = bank_section.get_object("metadata")
            backup_id = resource_definition["backup_id"]
            cinder_client.backups.delete(backup_id)
            bank_section.delete_object("metadata", resource_definition)
            self.protection_resource_map[resource_id] = {
                "bank_section": bank_section,
                "backup_id": backup_id,
                "cinder_client": cinder_client,
                "operation": "delete"
            }
        except Exception as e:
            LOG.error(_LE("delete volume backup failed, volume_id: %s."),
                      resource_id)
            bank_section.update_object("status",
                                       constants.CHECKPOINT_STATUS_ERROR)

            raise exception.DeleteBackupFailed(
                reason=six.text_type(e),
                resource_id=resource_id,
                resource_type=constants.VOLUME_RESOURCE_TYPE
            )

def create(context, conf):
    register_opts(conf)

    if hasattr(conf.swift_client, 'swift_auth_url') and \
            conf.swift_client.swift_auth_url:
        connection = swift.Connection(
            authurl=conf.swift_client.swift_auth_url,
            auth_version=conf.swift_client.swift_auth_version,
            tenant_name=conf.swift_client.swift_tenant_name,
            user=conf.swift_client.swift_user,
            key=conf.swift_client.swift_key,
            retries=conf.swift_client.swift_retry_attempts,
            starting_backoff=conf.swift_client.swift_retry_backoff,
            insecure=conf.swift_client.swift_auth_insecure,
            cacert=conf.swift_client.swift_ca_cert_file)
    else:
        try:
            url = utils.get_url(SERVICE, context, conf,
                                append_project_fmt='%(url)s/AUTH_%(project)s')
        except Exception:
            LOG.error(_LE("Get swift service endpoint url failed"))
            raise
        LOG.info(_LI("Creating swift client with url %s."), url)
        connection = swift.Connection(
            preauthurl=url,
            preauthtoken=context.auth_token,
            retries=conf.swift_client.swift_retry_attempts,
            starting_backoff=conf.swift_client.swift_retry_backoff,
            insecure=conf.swift_client.swift_auth_insecure,
            cacert=conf.swift_client.swift_ca_cert_file)
    return connection

 def build_flow(self, flow_name, flow_type='graph'):
     if flow_type == 'linear':
         return linear_flow.Flow(flow_name)
     elif flow_type == 'graph':
         return graph_flow.Flow(flow_name)
     else:
         raise ValueError(_LE("unsupported flow type:%s") % flow_type)

def create(context, conf, **kwargs):
    auth_url = kwargs["auth_url"]
    username = kwargs["username"]
    password = kwargs["password"]
    tenant_name = context.project_name
    LOG.info(_LI('Creating heat client with url %s.'), auth_url)
    try:
        keystone = kc.Client(version=KEYSTONECLIENT_VERSION,
                             username=username,
                             tenant_name=tenant_name,
                             password=password,
                             auth_url=auth_url)

        auth_token = keystone.auth_ref['token']['id']
        heat_endpoint = ''
        services = keystone.auth_ref['serviceCatalog']
        for service in services:
            if service['name'] == 'heat':
                heat_endpoint = service['endpoints'][0]['publicURL']
        heat = hc.Client(HEATCLIENT_VERSION, endpoint=heat_endpoint,
                         token=auth_token)
        return heat
    except Exception:
        LOG.error(_LE('Creating heat client with url %s.'), auth_url)
        raise

    def restore_backup(self, cntxt, checkpoint, **kwargs):
        resource_node = kwargs.get("node")
        resource_id = resource_node.value.id
        heat_template = kwargs.get("heat_template")

        name = kwargs.get("restore_name",
                          "%[email protected]%s" % (checkpoint.id, resource_id))
        description = kwargs.get("restore_description")

        heat_resource_id = str(uuid4())
        heat_resource = HeatResource(heat_resource_id,
                                     constants.VOLUME_RESOURCE_TYPE)

        bank_section = checkpoint.get_resource_bank_section(resource_id)
        try:
            resource_definition = bank_section.get_object("metadata")
            backup_id = resource_definition["backup_id"]
            properties = {"backup_id": backup_id,
                          "name": name}

            if description is not None:
                properties["description"] = description

            for key, value in properties.items():
                heat_resource.set_property(key, value)

            heat_template.put_resource(resource_id, heat_resource)
        except Exception as e:
            LOG.error(_LE("restore volume backup failed, volume_id: %s."),
                      resource_id)
            raise exception.RestoreBackupFailed(
                reason=six.text_type(e),
                resource_id=resource_id,
                resource_type=constants.VOLUME_RESOURCE_TYPE
            )

 def sync_status(self):
     for resource_id, resource_info in self.protection_resource_map.items():
         backup_id = resource_info["backup_id"]
         bank_section = resource_info["bank_section"]
         cinder_client = resource_info["cinder_client"]
         operation = resource_info["operation"]
         try:
             backup = cinder_client.backups.get(backup_id)
             if backup.status == "available":
                 bank_section.update_object(
                     "status", constants.RESOURCE_STATUS_AVAILABLE)
                 self.protection_resource_map.pop(resource_id)
             elif backup.status in ["error", "error-deleting"]:
                 bank_section.update_object(
                     "status", constants.RESOURCE_STATUS_ERROR)
                 self.protection_resource_map.pop(resource_id)
             else:
                 continue
         except Exception as exc:
             if operation == "delete" and type(exc) == NotFound:
                 bank_section.update_object(
                     "status",
                     constants.RESOURCE_STATUS_DELETED)
                 LOG.info(_("deleting volume backup finished."
                            "backup id: %s"), backup_id)
             else:
                 LOG.error(_LE("deleting volume backup error.exc:%s."),
                           six.text_type(exc))
             self.protection_resource_map.pop(resource_id)

    def get_method(self, request, action, content_type, body):
        """Look up the action-specific method and its extensions."""

        # Look up the method
        try:
            if not self.controller:
                meth = getattr(self, action)
            else:
                meth = getattr(self.controller, action)
        except AttributeError as e:
            with excutils.save_and_reraise_exception(e) as ctxt:
                if (not self.wsgi_actions or action not in ['action',
                                                            'create',
                                                            'delete',
                                                            'update']):
                    LOG.exception(_LE('Get method error.'))
                else:
                    ctxt.reraise = False
        else:
            return meth, self.wsgi_extensions.get(action, [])

        if action == 'action':
            # OK, it's an action; figure out which action...
            mtype = _MEDIA_TYPE_MAP.get(content_type)
            action_name = self.action_peek[mtype](body)
            LOG.debug("Action body: %s", body)
        else:
            action_name = action

        # Look up the action method
        return (self.wsgi_actions[action_name],
                self.wsgi_action_extensions.get(action_name, []))

 def _get_protection_plugin(self, resource_type):
     for plugin in self.plugin_map.values():
         if hasattr(plugin, "get_supported_resources_types"):
             if resource_type in plugin.get_supported_resources_types():
                 return plugin
     LOG.error(_LE("no plugin support this resource_type:%s"),
               resource_type)
     raise Exception("no plugin support this resource_type")

 def get_object(self, key):
     try:
         return self._get_object(container=self.bank_object_container,
                                 obj=key)
     except SwiftConnectionFailed as err:
         LOG.error(_LE("get object failed, err: %s."), err)
         raise exception.BankGetObjectFailed(reasone=err,
                                             key=key)

    def __init__(self, config, context, object_container):
        super(SwiftBankPlugin, self).__init__(config)
        self.context = context
        self._config.register_opts(swift_client_opts, "swift_client")
        self.swift_retry_attempts = \
            self._config.swift_client.bank_swift_retry_attempts
        self.swift_retry_backoff = \
            self._config.swift_client.bank_swift_retry_backoff
        self.swift_auth_insecure = \
            self._config.swift_client.bank_swift_auth_insecure
        self.swift_ca_cert_file = \
            self._config.swift_client.bank_swift_ca_cert_file
        self.lease_expire_window = self._config.lease_expire_window
        self.lease_renew_window = self._config.lease_renew_window
        # TODO(luobin):
        # init lease_validity_window
        # according to lease_renew_window if not configured
        self.lease_validity_window = self._config.lease_validity_window

        # TODO(luobin): create a uuid of this bank_plugin
        self.owner_id = str(uuid.uuid4())
        self.lease_expire_time = 0
        self.bank_leases_container = "leases"
        self.bank_object_container = object_container
        self.connection = self._setup_connection()

        # create container
        try:
            self._put_container(self.bank_object_container)
            self._put_container(self.bank_leases_container)
        except SwiftConnectionFailed as err:
            LOG.error(_LE("bank plugin create container failed."))
            raise exception.CreateContainerFailed(reason=err)

        # acquire lease
        try:
            self.acquire_lease()
        except exception.AcquireLeaseFailed as err:
            LOG.error(_LE("bank plugin acquire lease failed."))
            raise err

        # start renew lease
        renew_lease_loop = loopingcall.FixedIntervalLoopingCall(
            self.renew_lease)
        renew_lease_loop.start(interval=self.lease_renew_window,
                               initial_delay=self.lease_renew_window)

 def output(self, flow_engine, target=None):
     if flow_engine is None:
         LOG.error(_LE("Flow engine is None,return nothing"))
         raise exception.InvalidTaskFlowObject(reason="The flow_engine"
                                                      " is None")
     if target:
         return flow_engine.storage.fetch(target)
     return flow_engine.storage.fetch_all()

 def delete_object(self, key):
     try:
         self._delete_object(container=self.bank_object_container,
                             obj=key)
     except SwiftConnectionFailed as err:
         LOG.error(_LE("delete object failed, err: %s."), err)
         raise exception.BankDeleteObjectFailed(reasone=err,
                                                key=key)