def set_driver(self, resource):
        """Set the driver for a neutron resource.

        :param resource: Neutron resource in dict format. Expected keys:
                        { 'id': <value>
                          'hosting_device': { 'id': <value>, }
                          'router_type': {'cfg_agent_driver': <value>,  }
                        }
        :return driver : driver object
        """
        try:
            resource_id = resource['id']
            hosting_device = resource['hosting_device']
            hd_id = hosting_device['id']
            if hd_id in self._hosting_device_routing_drivers_binding:
                driver = self._hosting_device_routing_drivers_binding[hd_id]
                self._drivers[resource_id] = driver
            else:
                driver_class = resource['router_type']['cfg_agent_driver']
                driver = importutils.import_object(driver_class,
                                                   **hosting_device)
                self._hosting_device_routing_drivers_binding[hd_id] = driver
                self._drivers[resource_id] = driver
            return driver
        except ImportError:
            LOG.exception(_("Error loading cfg agent driver %(driver)s for "
                            "hosting device template %(t_name)s(%(t_id)s)"),
                          {'driver': driver_class, 't_id': hd_id,
                           't_name': hosting_device['name']})
            with excutils.save_and_reraise_exception(reraise=False):
                raise cfg_exceptions.DriverNotExist(driver=driver_class)
        except KeyError as e:
            with excutils.save_and_reraise_exception(reraise=False):
                raise cfg_exceptions.DriverNotSetForMissingParameter(e)

    def create_vip(self, context, edge_id, vip):
        app_profile = self._convert_app_profile(
            vip['name'], vip.get('session_persistence'))
        try:
            header, response = self.vcns.create_app_profile(
                edge_id, app_profile)
        except vcns_exc.VcnsApiException:
            with excutils.save_and_reraise_exception():
                LOG.exception(_("Failed to create app profile on edge: %s"),
                              edge_id)
        objuri = header['location']
        app_profileid = objuri[objuri.rfind("/") + 1:]

        vip_new = self._convert_lb_vip(context, edge_id, vip, app_profileid)
        try:
            header, response = self.vcns.create_vip(
                edge_id, vip_new)
        except vcns_exc.VcnsApiException:
            with excutils.save_and_reraise_exception():
                LOG.exception(_("Failed to create vip on vshield edge: %s"),
                              edge_id)
        objuri = header['location']
        vip_vseid = objuri[objuri.rfind("/") + 1:]

        # Add the vip mapping
        map_info = {
            "vip_id": vip['id'],
            "vip_vseid": vip_vseid,
            "edge_id": edge_id,
            "app_profileid": app_profileid
        }
        vcns_db.add_vcns_edge_vip_binding(context.session, map_info)

    def _create_ha_network(self, context, tenant_id):
        admin_ctx = context.elevated()

        args = {'network':
                {'name': constants.HA_NETWORK_NAME % tenant_id,
                 'tenant_id': '',
                 'shared': False,
                 'admin_state_up': True,
                 'status': constants.NET_STATUS_ACTIVE}}
        network = self._core_plugin.create_network(context, args)
        try:
            ha_network = self._create_ha_network_tenant_binding(admin_ctx,
                                                                tenant_id,
                                                                network['id'])
        except Exception:
            with excutils.save_and_reraise_exception():
                self._core_plugin.delete_network(admin_ctx, network['id'])

        try:
            self._create_ha_subnet(admin_ctx, network['id'], tenant_id)
        except Exception:
            with excutils.save_and_reraise_exception():
                self._core_plugin.delete_network(admin_ctx, network['id'])

        return ha_network

def add_neutron_nsx_port_mapping(session, neutron_id,
                                 nsx_switch_id, nsx_port_id):
    session.begin(subtransactions=True)
    try:
        mapping = models.NeutronNsxPortMapping(
            neutron_id, nsx_switch_id, nsx_port_id)
        session.add(mapping)
        session.commit()
    except db_exc.DBDuplicateEntry:
        with excutils.save_and_reraise_exception() as ctxt:
            session.rollback()
            # do not complain if the same exact mapping is being added,
            # otherwise re-raise because even though it is possible for the
            # same neutron port to map to different back-end ports over time,
            # this should not occur whilst a mapping already exists
            current = get_nsx_switch_and_port_id(session, neutron_id)
            if current[1] == nsx_port_id:
                LOG.debug(_("Port mapping for %s already available"),
                          neutron_id)
                ctxt.reraise = False
    except db_exc.DBError:
        with excutils.save_and_reraise_exception():
            # rollback for any other db error
            session.rollback()
    return mapping

    def consume_in_thread(self):
        """Runs the ZmqProxy service."""
        ipc_dir = CONF.rpc_zmq_ipc_dir
        consume_in = "tcp://%s:%s" % \
            (CONF.rpc_zmq_bind_address,
             CONF.rpc_zmq_port)
        consumption_proxy = InternalContext(None)

        try:
            os.makedirs(ipc_dir)
        except os.error:
            if not os.path.isdir(ipc_dir):
                with excutils.save_and_reraise_exception():
                    LOG.error(_("Required IPC directory does not exist at"
                                " %s") % (ipc_dir, ))
        try:
            self.register(consumption_proxy,
                          consume_in,
                          zmq.PULL)
        except zmq.ZMQError:
            if os.access(ipc_dir, os.X_OK):
                with excutils.save_and_reraise_exception():
                    LOG.error(_("Permission denied to IPC directory at"
                                " %s") % (ipc_dir, ))
            with excutils.save_and_reraise_exception():
                LOG.error(_("Could not create ZeroMQ receiver daemon. "
                            "Socket may already be in use."))

        super(ZmqProxy, self).consume_in_thread()

    def create_loadbalancer(self, context, loadbalancer):
        with context.session.begin(subtransactions=True):
            self._load_id_and_tenant_id(context, loadbalancer)
            vip_address = loadbalancer.pop('vip_address')
            securitygroup_id = loadbalancer.get('securitygroup_id')
            loadbalancer['status'] = constants.PENDING_CREATE
            loadbalancer['created_at'] = timeutils.utcnow()
            lb_db = models.LoadBalancer(**loadbalancer)
            context.session.add(lb_db)
            context.session.flush()
            lb_db.stats = self._create_loadbalancer_stats(
                context, lb_db.id)
            context.session.add(lb_db)

        # create port outside of lb create transaction since it can sometimes
        # cause lock wait timeouts
        try:
            self._create_port_for_load_balancer(context, lb_db,
                                        vip_address, securitygroup_id)
        except ext_sg.SecurityGroupNotFound:
            LOG.error('_create_port_for_load_balancer %s securitygroup',lb_db.id)
            with excutils.save_and_reraise_exception():
                context.session.delete(lb_db)
                context.session.flush()
                raise loadbalancerv2.SecurityGroupNotFound(id=lb_db.id) 
        except Exception:
            LOG.error('_create_port_for_load_balancer %s',lb_db.id)
            with excutils.save_and_reraise_exception():
                context.session.delete(lb_db)
                context.session.flush()
        return data_models.LoadBalancer.from_sqlalchemy_model(lb_db)

    def _add_rule_below(self, context, ref_rule_id, edge_id, firewall_rule):
        rule_map = vcns_db.get_vcns_edge_firewallrule_binding(context.session, ref_rule_id, edge_id)
        ref_vcns_rule_id = rule_map.rule_vseid
        fwr_vse_next = self._get_firewall_rule_next(context, edge_id, ref_vcns_rule_id)
        fwr_req = self._convert_firewall_rule(context, firewall_rule)
        if fwr_vse_next:
            ref_vcns_rule_id = fwr_vse_next["ruleId"]
            try:
                header = self.vcns.add_firewall_rule_above(edge_id, int(ref_vcns_rule_id), fwr_req)[0]
            except vcns_exc.VcnsApiException:
                with excutils.save_and_reraise_exception():
                    LOG.exception(
                        _("Failed to add firewall rule above: " "%(rule_id)s with edge_id: %(edge_id)s"),
                        {"rule_id": ref_vcns_rule_id, "edge_id": edge_id},
                    )
        else:
            # append the rule at the bottom
            try:
                header = self.vcns.add_firewall_rule(edge_id, fwr_req)[0]
            except vcns_exc.VcnsApiException:
                with excutils.save_and_reraise_exception():
                    LOG.exception(_("Failed to append a firewall rule" "with edge_id: %s"), edge_id)

        objuri = header["location"]
        fwr_vseid = objuri[objuri.rfind("/") + 1 :]
        map_info = {"rule_id": firewall_rule["id"], "rule_vseid": fwr_vseid, "edge_id": edge_id}
        vcns_db.add_vcns_edge_firewallrule_binding(context.session, map_info)

    def create_router(self, host, username, password, rbridge_id, router_id):
        """create vrf and associate vrf."""
        router_id = router_id[0:11]
        vrf_name = template.OS_VRF_NAME.format(id=router_id)
        rd = router_id + ":" + router_id
        try:
            mgr = self.connect(host, username, password)
            self.create_vrf(mgr, rbridge_id, vrf_name)
        except Exception:
            with excutils.save_and_reraise_exception():
                LOG.exception(_("NETCONF error"))
                self.close_session()
        try:
            # For Nos5.0.0
            self.configure_rd_for_vrf(mgr, rbridge_id, vrf_name, rd)
            self.configure_address_family_for_vrf(mgr, rbridge_id, vrf_name)
        except Exception:
            with excutils.save_and_reraise_exception() as ctxt:
                try:
                    # This is done because on 4.0.0 rd doesnt accept alpha
                    # character nor hyphen
                    rd = "".join(i for i in router_id if i in "0123456789")
                    rd = rd[:4] + ":" + rd[:4]
                    self.configure_rd_for_vrf(mgr, rbridge_id, vrf_name, rd)
                    self.configure_address_family_for_vrf_v1(mgr,
                                                             rbridge_id,
                                                             vrf_name)
                except Exception:
                    with excutils.save_and_reraise_exception():
                        LOG.exception(_("NETCONF error"))
                        self.close_session()

                ctxt.reraise = False

    def create_router(self, context, router):
        is_ha = self._is_ha(router['router'])

        if is_ha and l3_dvr_db.is_distributed_router(router['router']):
            raise l3_ha.DistributedHARouterNotSupported()

        router['router']['ha'] = is_ha
        router_dict = super(L3_HA_NAT_db_mixin,
                            self).create_router(context, router)

        if is_ha:
            if cfg.CONF.mulity_ha_network_one_tenant:
                router_db = self._get_router(context, router_dict['id'])
                ha_network_list = self.get_ha_network_list(context, router_db.tenant_id)
                if not ha_network_list:
                    ha_network_list = [self._create_ha_network(context,
                                                         router_db.tenant_id)]
                def get_vr_id():
                    flag = 1
                    for ha_network in ha_network_list:
                        try:
                            self._set_vr_id(context, router_db, ha_network)
                            flag = 0
                            LOG.debug("get the vr_id from  ha network %s" %ha_network.network['id'])
                            break
                        except l3_ha.NoVRIDAvailable:
                            LOG.debug("the ha network %s vr_id is over " %ha_network.network['id'])

                    if flag:
                        LOG.debug("create new ha network")
                        ha_network = self._create_ha_network(context,
                                                         router_db.tenant_id)
                        self._set_vr_id(context, router_db, ha_network)
                    self._create_router_ha_network_info(context, router_db.id, ha_network.network['id'])
                    self._create_ha_interfaces(context, router_db, ha_network)
                    self._notify_ha_interfaces_updated(context, router_db.id)
                    router_dict['ha_vr_id'] = router_db.extra_attributes.ha_vr_id
                try:
                    get_vr_id()
                except Exception:
                     with excutils.save_and_reraise_exception():
                        self.delete_router(context, router_dict['id'])
            else:
                try:
                    router_db = self._get_router(context, router_dict['id'])
                    ha_network = self.get_ha_network(context,
                                                 router_db.tenant_id)
                    if not ha_network:
                        ha_network = self._create_ha_network(context,
                                                         router_db.tenant_id)

                    self._set_vr_id(context, router_db, ha_network)
                    self._create_ha_interfaces(context, router_db, ha_network)
                    self._notify_ha_interfaces_updated(context, router_db.id)
                except Exception:
                    with excutils.save_and_reraise_exception():
                        self.delete_router(context, router_dict['id'])
                router_dict['ha_vr_id'] = router_db.extra_attributes.ha_vr_id
        return router_dict

 def delete_router(self, context, router_id):
     """ delete a vrf on NOS device."""
     router = super(BrocadeSVIPlugin, self).get_router(context, router_id)
     super(BrocadeSVIPlugin, self).delete_router(context, router_id)
     try:
         switch = self._switch
         self._driver.delete_router(switch['address'],
                                    switch['username'],
                                    switch['password'],
                                    switch['rbridge_id'],
                                    str(router['id']))
     except Exception:
         excutils.save_and_reraise_exception()

    def __init__(self, host, conf=None):
        super(L3NATAgent, self).__init__()
        if conf:
            self.conf = conf
        else:
            self.conf = cfg.CONF
        self.router_info = {}
        self.virtual_ip = ''
        self.virtual_mac = ''
        self.context = context.get_admin_context_without_session()
        self.plugin_rpc = L3PluginApi(topics.L3PLUGIN, host)
        self.dvr_base_mac = self.plugin_rpc.get_dvr_base_mac(self.context)
        self._setup_dataenigine_rpc()
        self.dataengine_rpc.delete_all_flows()
        self.fullsync = True

        # Get the list of service plugins from Neutron Server
        # This is the first place where we contact neutron-server on startup
        # so retry in case its not ready to respond.
        retry_count = 5
        while True:
            retry_count = retry_count - 1
            try:
                self.neutron_service_plugins = (
                    self.plugin_rpc.get_service_plugin_list(self.context))
            except n_rpc.RemoteError as e:
                with excutils.save_and_reraise_exception() as ctx:
                    ctx.reraise = False
                    LOG.warning(_LW('l3-agent cannot check service plugins '
                                    'enabled at the neutron server when '
                                    'startup due to RPC error. It happens '
                                    'when the server does not support this '
                                    'RPC API. If the error is '
                                    'UnsupportedVersion you can ignore this '
                                    'warning. Detail message: %s'), e)
                self.neutron_service_plugins = None
            except messaging.MessagingTimeout as e:
                with excutils.save_and_reraise_exception() as ctx:
                    if retry_count > 0:
                        ctx.reraise = False
                        LOG.warning(_LW('l3-agent cannot check service '
                                        'plugins enabled on the neutron '
                                        'server. Retrying. '
                                        'Detail message: %s'), e)
                        continue
            break

        self._queue = RouterProcessingQueue()

 def _port_action(self, plugin, context, port, action):
     """Perform port operations taking care of concurrency issues."""
     try:
         if action == 'create_port':
             return plugin.create_port(context, port)
         elif action == 'update_port':
             return plugin.update_port(context, port['id'], port['port'])
         else:
             msg = _('Unrecognized action')
             raise n_exc.Invalid(message=msg)
     except (db_exc.DBError, n_exc.NetworkNotFound,
             n_exc.SubnetNotFound, n_exc.IpAddressGenerationFailure) as e:
         with excutils.save_and_reraise_exception(reraise=False) as ctxt:
             if isinstance(e, n_exc.IpAddressGenerationFailure):
                 # Check if the subnet still exists and if it does not,
                 # this is the reason why the ip address generation failed.
                 # In any other unlikely event re-raise
                 try:
                     subnet_id = port['port']['fixed_ips'][0]['subnet_id']
                     plugin.get_subnet(context, subnet_id)
                 except n_exc.SubnetNotFound:
                     pass
                 else:
                     ctxt.reraise = True
             network_id = port['port']['network_id']
             LOG.warn(_("Port for network %(net_id)s could not be created: "
                        "%(reason)s") % {"net_id": network_id, 'reason': e})

def serve_rpc():
    plugin = manager.NeutronManager.get_plugin()

    # If 0 < rpc_workers then start_rpc_listeners would be called in a
    # subprocess and we cannot simply catch the NotImplementedError.  It is
    # simpler to check this up front by testing whether the plugin supports
    # multiple RPC workers.
    if not plugin.rpc_workers_supported():
        LOG.debug(_("Active plugin doesn't implement start_rpc_listeners"))
        if 0 < cfg.CONF.rpc_workers:
            msg = _("'rpc_workers = %d' ignored because start_rpc_listeners "
                    "is not implemented.")
            LOG.error(msg, cfg.CONF.rpc_workers)
        raise NotImplementedError()

    try:
        rpc = RpcWorker(plugin)

        if cfg.CONF.rpc_workers < 1:
            rpc.start()
            return rpc
        else:
            launcher = common_service.ProcessLauncher(wait_interval=1.0)
            launcher.launch_service(rpc, workers=cfg.CONF.rpc_workers)
            return launcher
    except Exception:
        with excutils.save_and_reraise_exception():
            LOG.exception(_('Unrecoverable error: please check log '
                            'for details.'))

    def sync_single_resource(self, operation, object_type, obj_id,
                             context, attr_filter_create, attr_filter_update):
        """Sync over a single resource from Neutron to OpenDaylight.

        Handle syncing a single operation over to OpenDaylight, and correctly
        filter attributes out which are not required for the requisite
        operation (create or update) being handled.
        """
        dbcontext = context._plugin_context
        if operation == 'create':
            urlpath = object_type
            method = 'post'
        else:
            urlpath = object_type + '/' + obj_id
            method = 'put'

        try:
            obj_getter = getattr(context._plugin, 'get_%s' % object_type[:-1])
            resource = obj_getter(dbcontext, obj_id)
        except not_found_exception_map[object_type]:
            LOG.debug(_('%(object_type)s not found (%(obj_id)s)'),
                      {'object_type': object_type.capitalize(),
                      'obj_id': obj_id})
        else:
            if operation == 'create':
                attr_filter_create(self, resource, context, dbcontext)
            elif operation == 'update':
                attr_filter_update(self, resource, context, dbcontext)
            try:
                # 400 errors are returned if an object exists, which we ignore.
                self.sendjson(method, urlpath, {object_type[:-1]: resource},
                              [400])
            except Exception:
                with excutils.save_and_reraise_exception():
                    self.out_of_sync = True

    def create_router(self, context, router):
        is_ha = self._is_ha(router['router'])

        if is_ha and l3_dvr_db.is_distributed_router(router['router']):
            raise l3_ha.DistributedHARouterNotSupported()

        router['router']['ha'] = is_ha
        router_dict = super(L3_HA_NAT_db_mixin,
                            self).create_router(context, router)

        if is_ha:
            try:
                router_db = self._get_router(context, router_dict['id'])
                ha_network = self.get_ha_network(context,
                                                 router_db.tenant_id)
                if not ha_network:
                    ha_network = self._create_ha_network(context,
                                                         router_db.tenant_id)

                self._set_vr_id(context, router_db, ha_network)
                self._create_ha_interfaces(context, router_db, ha_network)
                self._notify_ha_interfaces_updated(context, router_db.id)
            except Exception:
                with excutils.save_and_reraise_exception():
                    self.delete_router(context, router_dict['id'])
            router_dict['ha_vr_id'] = router_db.extra_attributes.ha_vr_id
        return router_dict

    def create_network(self, context, network):
        net_data = network['network']
        segments = self._process_provider_create(net_data)
        tenant_id = self._get_tenant_id_for_create(context, net_data)

        session = context.session
        with session.begin(subtransactions=True):
            self._ensure_default_security_group(context, tenant_id)
            result = super(Ml2Plugin, self).create_network(context, network)
            network_id = result['id']
            self._process_l3_create(context, result, net_data)
            # REVISIT(rkukura): Consider moving all segment management
            # to TypeManager.
            if segments:
                for segment in segments:
                    self.type_manager.reserve_provider_segment(session,
                                                               segment)
                    db.add_network_segment(session, network_id, segment)
            else:
                segment = self.type_manager.allocate_tenant_segment(session)
                db.add_network_segment(session, network_id, segment)
            self._extend_network_dict_provider(context, result)
            mech_context = driver_context.NetworkContext(self, context,
                                                         result)
            self.mechanism_manager.create_network_precommit(mech_context)

        try:
            self.mechanism_manager.create_network_postcommit(mech_context)
        except ml2_exc.MechanismDriverError:
            with excutils.save_and_reraise_exception():
                LOG.error(_("mechanism_manager.create_network_postcommit "
                            "failed, deleting network '%s'"), result['id'])
                self.delete_network(context, result['id'])
        return result

    def connect(self, host, username, password):
        """Connect via SSH and initialize the NETCONF session."""

        # Use the persisted NETCONF connection
        if self.mgr and self.mgr.connected:
            return self.mgr

        # check if someone forgot to edit the conf file with real values
        if host == '':
            raise Exception(_("Brocade Switch IP address is not set, "
                              "check config ml2_conf_brocade.ini file"))

        # Open new NETCONF connection
        try:
            self.mgr = manager.connect(host=host, port=SSH_PORT,
                                       username=username, password=password,
                                       unknown_host_cb=nos_unknown_host_cb)

        except Exception:
            with excutils.save_and_reraise_exception():
                LOG.exception(_("Connect failed to switch"))

        LOG.debug(_("Connect success to host %(host)s:%(ssh_port)d"),
                  dict(host=host, ssh_port=SSH_PORT))
        return self.mgr

 def get_nat_config(self, edge_id):
     try:
         return self.vcns.get_nat_config(edge_id)[1]
     except exceptions.VcnsApiException as e:
         with excutils.save_and_reraise_exception():
             LOG.exception(_("VCNS: Failed to get nat config:\n%s"),
                           e.response)

 def _from_xml(self, datastring):
     if datastring is None:
         return None
     plurals = set(self.metadata.get('plurals', {}))
     try:
         node = self._parseXML(datastring)
         root_tag = self._get_key(node.tag)
         # Deserialize link node was needed by unit test for verifying
         # the request's response
         links = self._get_links(root_tag, node)
         result = self._from_xml_node(node, plurals)
         # root_tag = constants.VIRTUAL_ROOT_KEY and links is not None
         # is not possible because of the way data are serialized.
         if root_tag == constants.VIRTUAL_ROOT_KEY:
             return result
         return dict({root_tag: result}, **links)
     except Exception as e:
         with excutils.save_and_reraise_exception():
             parseError = False
             # Python2.7
             if (hasattr(etree, 'ParseError') and
                 isinstance(e, getattr(etree, 'ParseError'))):
                 parseError = True
             # Python2.6
             elif isinstance(e, expat.ExpatError):
                 parseError = True
             if parseError:
                 msg = _("Cannot understand XML")
                 raise exception.MalformedRequestBody(reason=msg)

    def add_router_interface(self, context, router_id, interface_info):
        """Attach a subnet to a router."""
        tenant_id = context.tenant_id
        subnet_id = interface_info["subnet_id"]
        LOG.debug(
            "Attaching subnet %(subnet_id)s to "
            "router %(router_id)s" % {"subnet_id": subnet_id, "router_id": router_id}
        )

        # Get network for this subnet
        subnet = self.get_subnet(context, subnet_id)
        network_id = subnet["network_id"]
        net_name = self.get_network(context, network_id)["name"]

        # Setup tenant filters and contracts
        contract = self.manager.create_tenant_contract(tenant_id)

        # Check for a provider EPG
        epg = self.manager.ensure_epg_created_for_network(tenant_id, network_id, net_name)
        self._add_epg_to_contract(tenant_id, epg, contract)

        # Create DB port
        try:
            return super(ApicL3ServicePlugin, self).add_router_interface(context, router_id, interface_info)
        except Exception:
            LOG.error(
                _("Error attaching subnet %(subnet_id)s to " "router %(router_id)s")
                % {"subnet_id": subnet_id, "router_id": router_id}
            )
            with excutils.save_and_reraise_exception():
                self.manager.delete_contract_for_epg(tenant_id, epg.epg_id, contract.contract_id, provider=epg.provider)