def _get_port_infos(self, context, port, agent_host):
        if not agent_host:
            return

        session = db_api.get_session()
        agent = self.get_agent_by_host(session, agent_host)
        if not agent:
            return

        agent_ip = self.get_agent_ip(agent)
        if not agent_ip:
            LOG.warning(_LW("Unable to retrieve the agent ip, check the agent "
                            "configuration."))
            return

        segment = context.bound_segment
        if not segment:
            LOG.warning(_LW("Port %(port)s updated by agent %(agent)s "
                            "isn't bound to any segment"),
                        {'port': port['id'], 'agent': agent})
            return

        network_types = self.get_agent_l2pop_network_types(agent)
        if network_types is None:
            network_types = self.get_agent_tunnel_types(agent)
        if segment['network_type'] not in network_types:
            return

        fdb_entries = self._get_port_fdb_entries(port)

        return agent, agent_host, agent_ip, segment, fdb_entries

    def _sync_base(self):
        ctx = context.get_admin_context()
        # Sync Networks
        for network in self.core_plugin.get_networks(ctx):
            mech_context = driver_context.NetworkContext(self.core_plugin, ctx,
                                                         network)
            try:
                self.driver.create_network_postcommit(mech_context)
            except Exception:
                LOG.warn(_LW("Create network postcommit failed for "
                             "network %s"), network['id'])

        # Sync Subnets
        for subnet in self.core_plugin.get_subnets(ctx):
            mech_context = driver_context.SubnetContext(self.core_plugin, ctx,
                                                        subnet)
            try:
                self.driver.create_subnet_postcommit(mech_context)
            except Exception:
                LOG.warn(_LW("Create subnet postcommit failed for"
                             " subnet %s"), subnet['id'])

        # Sync Ports (compute/gateway/dhcp)
        for port in self.core_plugin.get_ports(ctx):
            _, binding = l2_db.get_locked_port_and_binding(ctx.session,
                                                           port['id'])
            network = self.core_plugin.get_network(ctx, port['network_id'])
            mech_context = driver_context.PortContext(self.core_plugin, ctx,
                                                      port, network, binding)
            try:
                self.driver.create_port_postcommit(mech_context)
            except Exception:
                LOG.warn(_LW("Create port postcommit failed for"
                             " port %s"), port['id'])

    def call_driver(self, action, network, **action_kwargs):
        """Invoke an action on a DHCP driver instance."""
        LOG.debug('Calling driver for network: %(net)s action: %(action)s',
                  {'net': network.id, 'action': action})
        try:
            # the Driver expects something that is duck typed similar to
            # the base models.
            driver = self.dhcp_driver_cls(self.conf,
                                          network,
                                          self.root_helper,
                                          self.dhcp_version,
                                          self.plugin_rpc)

            getattr(driver, action)(**action_kwargs)
            return True
        except exceptions.Conflict:
            # No need to resync here, the agent will receive the event related
            # to a status update for the network
            LOG.warning(_LW('Unable to %(action)s dhcp for %(net_id)s: there '
                            'is a conflict with its current state; please '
                            'check that the network and/or its subnet(s) '
                            'still exist.'),
                        {'net_id': network.id, 'action': action})
        except Exception as e:
            self.schedule_resync(e, network.id)
            if (isinstance(e, messaging.RemoteError)
                and e.exc_type == 'NetworkNotFound'
                or isinstance(e, exceptions.NetworkNotFound)):
                LOG.warning(_LW("Network %s has been deleted."), network.id)
            else:
                LOG.exception(_LE('Unable to %(action)s dhcp for %(net_id)s.')
                              % {'net_id': network.id, 'action': action})

 def _load_all_extensions_from_path(self, path):
     # Sorting the extension list makes the order in which they
     # are loaded predictable across a cluster of load-balanced
     # Neutron Servers
     for f in sorted(os.listdir(path)):
         try:
             LOG.debug('Loading extension file: %s', f)
             mod_name, file_ext = os.path.splitext(os.path.split(f)[-1])
             ext_path = os.path.join(path, f)
             if file_ext.lower() == '.py' and not mod_name.startswith('_'):
                 mod = imp.load_source(mod_name, ext_path)
                 ext_name = mod_name[0].upper() + mod_name[1:]
                 new_ext_class = getattr(mod, ext_name, None)
                 if not new_ext_class:
                     LOG.warn(_LW('Did not find expected name '
                                  '"%(ext_name)s" in %(file)s'),
                              {'ext_name': ext_name,
                               'file': ext_path})
                     continue
                 new_ext = new_ext_class()
                 self.add_extension(new_ext)
         except Exception as exception:
             LOG.warn(_LW("Extension file %(f)s wasn't loaded due to "
                          "%(exception)s"),
                      {'f': f, 'exception': exception})

    def _router_removed(self, router_id, deconfigure=True):
        """Operations when a router is removed.

        Get the RouterInfo object corresponding to the router in the service
        helpers's router_info dict. If deconfigure is set to True,
        remove this router's configuration from the hosting device.
        :param router_id: id of the router
        :param deconfigure: if True, the router's configuration is deleted from
        the hosting device.
        :return: None
        """
        ri = self.router_info.get(router_id)
        if ri is None:
            LOG.warning(_LW("Info for router %s was not found. "
                       "Skipping router removal"), router_id)
            return
        ri.router['gw_port'] = None
        ri.router[l3_constants.INTERFACE_KEY] = []
        ri.router[l3_constants.FLOATINGIP_KEY] = []
        try:
            if deconfigure:
                self._process_router(ri)
                driver = self._drivermgr.get_driver(router_id)
                driver.router_removed(ri, deconfigure)
                self._drivermgr.remove_driver(router_id)
            del self.router_info[router_id]
            self.removed_routers.discard(router_id)
        except cfg_exceptions.DriverException:
            LOG.warning(_LW("Router remove for router_id: %s was incomplete. "
                       "Adding the router to removed_routers list"), router_id)
            self.removed_routers.add(router_id)
            # remove this router from updated_routers if it is there. It might
            # end up there too if exception was thrown earlier inside
            # `_process_router()`
            self.updated_routers.discard(router_id)

    def get_candidates(self, plugin, context, sync_router):
        """Return L3 agents where a router could be scheduled."""
        with context.session.begin(subtransactions=True):
            # allow one router is hosted by just
            # one enabled l3 agent hosting since active is just a
            # timing problem. Non-active l3 agent can return to
            # active any time
            l3_agents = plugin.get_l3_agents_hosting_routers(
                context, [sync_router['id']], admin_state_up=True)
            if l3_agents and not sync_router.get('distributed', False):
                LOG.debug('Router %(router_id)s has already been hosted'
                          ' by L3 agent %(agent_id)s',
                          {'router_id': sync_router['id'],
                           'agent_id': l3_agents[0]['id']})
                return

            active_l3_agents = plugin.get_l3_agents(context, active=True)
            if not active_l3_agents:
                LOG.warn(_LW('No active L3 agents'))
                return
            new_l3agents = plugin.get_l3_agent_candidates(context,
                                                          sync_router,
                                                          active_l3_agents)
            old_l3agentset = set(l3_agents)
            if sync_router.get('distributed', False):
                new_l3agentset = set(new_l3agents)
                candidates = list(new_l3agentset - old_l3agentset)
            else:
                candidates = new_l3agents
                if not candidates:
                    LOG.warn(_LW('No L3 agents can host the router %s'),
                             sync_router['id'])

            return candidates

    def reschedule_routers_from_down_agents(self):
        """Reschedule routers from down l3 agents if admin state is up."""

        # give agents extra time to handle transient failures
        agent_dead_limit = cfg.CONF.agent_down_time * 2

        # check for an abrupt clock change since last check. if a change is
        # detected, sleep for a while to let the agents check in.
        tdelta = timeutils.utcnow() - getattr(self, '_clock_jump_canary',
                                              timeutils.utcnow())
        if timeutils.total_seconds(tdelta) > cfg.CONF.agent_down_time:
            LOG.warn(_LW("Time since last L3 agent reschedule check has "
                         "exceeded the interval between checks. Waiting "
                         "before check to allow agents to send a heartbeat "
                         "in case there was a clock adjustment."))
            time.sleep(agent_dead_limit)
        self._clock_jump_canary = timeutils.utcnow()

        context = n_ctx.get_admin_context()
        cutoff = timeutils.utcnow() - datetime.timedelta(
            seconds=agent_dead_limit)
        down_bindings = (
            context.session.query(RouterL3AgentBinding).
            join(agents_db.Agent).
            filter(agents_db.Agent.heartbeat_timestamp < cutoff,
                   agents_db.Agent.admin_state_up))
        for binding in down_bindings:
            LOG.warn(_LW("Rescheduling router %(router)s from agent %(agent)s "
                         "because the agent did not report to the server in "
                         "the last %(dead_time)s seconds."),
                     {'router': binding.router_id,
                      'agent': binding.l3_agent_id,
                      'dead_time': agent_dead_limit})
            self.reschedule_router(context, binding.router_id)

    def reschedule_routers_from_down_agents(self):
        """Reschedule routers from down l3 agents if admin state is up."""

        # give agents extra time to handle transient failures
        agent_dead_limit = cfg.CONF.agent_down_time * 2

        # check for an abrupt clock change since last check. if a change is
        # detected, sleep for a while to let the agents check in.
        tdelta = timeutils.utcnow() - getattr(self, "_clock_jump_canary", timeutils.utcnow())
        if timeutils.total_seconds(tdelta) > cfg.CONF.agent_down_time:
            LOG.warn(
                _LW(
                    "Time since last L3 agent reschedule check has "
                    "exceeded the interval between checks. Waiting "
                    "before check to allow agents to send a heartbeat "
                    "in case there was a clock adjustment."
                )
            )
            time.sleep(agent_dead_limit)
        self._clock_jump_canary = timeutils.utcnow()

        context = n_ctx.get_admin_context()
        cutoff = timeutils.utcnow() - datetime.timedelta(seconds=agent_dead_limit)
        down_bindings = (
            context.session.query(RouterL3AgentBinding)
            .join(agents_db.Agent)
            .filter(agents_db.Agent.heartbeat_timestamp < cutoff, agents_db.Agent.admin_state_up)
            .outerjoin(
                l3_attrs_db.RouterExtraAttributes,
                l3_attrs_db.RouterExtraAttributes.router_id == RouterL3AgentBinding.router_id,
            )
            .filter(
                sa.or_(
                    l3_attrs_db.RouterExtraAttributes.ha == sql.false(),
                    l3_attrs_db.RouterExtraAttributes.ha == sql.null(),
                )
            )
        )
        try:
            for binding in down_bindings:
                LOG.warn(
                    _LW(
                        "Rescheduling router %(router)s from agent %(agent)s "
                        "because the agent did not report to the server in "
                        "the last %(dead_time)s seconds."
                    ),
                    {"router": binding.router_id, "agent": binding.l3_agent_id, "dead_time": agent_dead_limit},
                )
                try:
                    self.reschedule_router(context, binding.router_id)
                except (l3agentscheduler.RouterReschedulingFailed, n_rpc.RemoteError):
                    # Catch individual router rescheduling errors here
                    # so one broken one doesn't stop the iteration.
                    LOG.exception(_LE("Failed to reschedule router %s"), binding.router_id)
        except db_exc.DBError:
            # Catch DB errors here so a transient DB connectivity issue
            # doesn't stop the loopingcall.
            LOG.exception(_LE("Exception encountered during router " "rescheduling."))

    def get_device_details(self, rpc_context, **kwargs):
        """Agent requests device details."""
        agent_id = kwargs.get('agent_id')
        device = kwargs.get('device')
        host = kwargs.get('host')
        LOG.debug("Device %(device)s details requested by agent "
                  "%(agent_id)s with host %(host)s",
                  {'device': device, 'agent_id': agent_id, 'host': host})

        plugin = manager.NeutronManager.get_plugin()
        port_id = plugin._device_to_port_id(device)
        port_context = plugin.get_bound_port_context(rpc_context,
                                                     port_id,
                                                     host)
        if not port_context:
            LOG.warning(_LW("Device %(device)s requested by agent "
                            "%(agent_id)s not found in database"),
                        {'device': device, 'agent_id': agent_id})
            return {'device': device}

        segment = port_context.bound_segment
        port = port_context.current

        if not segment:
            LOG.warning(_LW("Device %(device)s requested by agent "
                            "%(agent_id)s on network %(network_id)s not "
                            "bound, vif_type: %(vif_type)s"),
                        {'device': device,
                         'agent_id': agent_id,
                         'network_id': port['network_id'],
                         'vif_type': port[portbindings.VIF_TYPE]})
            return {'device': device}

        new_status = (q_const.PORT_STATUS_BUILD if port['admin_state_up']
                      else q_const.PORT_STATUS_DOWN)
        if port['status'] != new_status:
            plugin.update_port_status(rpc_context,
                                      port_id,
                                      new_status,
                                      host)

        entry = {'device': device,
                 'network_id': port['network_id'],
                 'port_id': port_id,
                 'mac_address': port['mac_address'],
                 'admin_state_up': port['admin_state_up'],
                 'network_type': segment[api.NETWORK_TYPE],
                 'segmentation_id': segment[api.SEGMENTATION_ID],
                 'physical_network': segment[api.PHYSICAL_NETWORK],
                 'fixed_ips': port['fixed_ips'],
                 'device_owner': port['device_owner'],
                 'profile': port[portbindings.PROFILE]}
        LOG.debug("Returning: %s", entry)
        return entry

 def call(self, action, resource, data, headers, binary=False):
     resp = self._call(action, resource, data, headers, binary)
     if resp[RESP_STATUS] == -1:
         LOG.warning(_LW('vDirect server is not responding (%s).'),
                     self.server)
         return self._recover(action, resource, data, headers, binary)
     elif resp[RESP_STATUS] in (301, 307):
         LOG.warning(_LW('vDirect server is not active (%s).'),
                     self.server)
         return self._recover(action, resource, data, headers, binary)
     else:
         return resp

 def clear_lock(self):
     LOG.debug("Clearing hash record lock of id %s" % self.random_lock_id)
     with self.session.begin(subtransactions=True):
         res = (self.session.query(ConsistencyHash).
                filter_by(hash_id=self.hash_id).first())
         if not res:
             LOG.warning(_LW("Hash record already gone, no lock to clear."))
             return
         if not res.hash.startswith(self.lock_marker):
             # if these are frequent the server is too slow
             LOG.warning(_LW("Another server already removed the lock. %s"),
                         res.hash)
             return
         res.hash = res.hash.replace(self.lock_marker, '')

    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 _create_backend_port(self, context, db_pool):
        try:
            subnet = self.plugin._core_plugin.get_subnet(context,
                                                         db_pool["subnet_id"])
        except n_exc.SubnetNotFound:
            LOG.warning(_LW("Subnet assigned to pool %s doesn't exist, "
                            "backend port can't be created"), db_pool['id'])
            return

        fixed_ip = {'subnet_id': subnet['id'],
                    'fixed_ips': attributes.ATTR_NOT_SPECIFIED}

        port_data = {
            'tenant_id': db_pool['tenant_id'],
            'name': 'pool-' + db_pool['id'],
            'network_id': subnet['network_id'],
            'mac_address': attributes.ATTR_NOT_SPECIFIED,
            'admin_state_up': False,
            'device_id': '',
            'device_owner': '',
            'fixed_ips': [fixed_ip]
        }

        port = self.plugin._core_plugin.create_port(context,
                                                    {'port': port_data})
        return edb.add_pool_port(context, db_pool['id'], port['id'])

    def _agent_registration(self):
        """Register this agent with the server.

        This method registers the cfg agent with the neutron server so hosting
        devices can be assigned to it. In case the server is not ready to
        accept registration (it sends a False) then we retry registration
        for `MAX_REGISTRATION_ATTEMPTS` with a delay of
        `REGISTRATION_RETRY_DELAY`. If there is no server response or a
        failure to register after the required number of attempts,
        the agent stops itself.
        """
        for attempts in xrange(MAX_REGISTRATION_ATTEMPTS):
            context = n_context.get_admin_context_without_session()
            self.send_agent_report(self.agent_state, context)
            res = self.devmgr_rpc.register_for_duty(context)
            if res is True:
                LOG.info(_LI("[Agent registration] Agent successfully "
                           "registered"))
                return
            elif res is False:
                LOG.warning(_LW("[Agent registration] Neutron server said "
                                "that device manager was not ready. Retrying "
                                "in %0.2f seconds "), REGISTRATION_RETRY_DELAY)
                time.sleep(REGISTRATION_RETRY_DELAY)
            elif res is None:
                LOG.error(_LE("[Agent registration] Neutron server said that "
                              "no device manager was found. Cannot continue. "
                              "Exiting!"))
                raise SystemExit("Cfg Agent exiting")
        LOG.error(_LE("[Agent registration] %d unsuccessful registration "
                    "attempts. Exiting!"), MAX_REGISTRATION_ATTEMPTS)
        raise SystemExit("Cfg Agent exiting")

    def _wait_child(self):
        try:
            # Don't block if no child processes have exited
            pid, status = os.waitpid(0, os.WNOHANG)
            if not pid:
                return None
        except OSError as exc:
            if exc.errno not in (errno.EINTR, errno.ECHILD):
                raise
            return None

        if os.WIFSIGNALED(status):
            sig = os.WTERMSIG(status)
            LOG.info(_LI('Child %(pid)d killed by signal %(sig)d'),
                     dict(pid=pid, sig=sig))
        else:
            code = os.WEXITSTATUS(status)
            LOG.info(_LI('Child %(pid)s exited with status %(code)d'),
                     dict(pid=pid, code=code))

        if pid not in self.children:
            LOG.warning(_LW('pid %d not in child list'), pid)
            return None

        wrap = self.children.pop(pid)
        wrap.children.remove(pid)
        return wrap

def _set_rules(data):
    default_rule = 'default'
    LOG.debug(_("Loading policies from file: %s"), _POLICY_PATH)
    # Ensure backward compatibility with folsom/grizzly convention
    # for extension rules
    policies = policy.Rules.load_json(data, default_rule)
    for pol in policies.keys():
        if any([pol.startswith(depr_pol) for depr_pol in
                DEPRECATED_POLICY_MAP.keys()]):
            LOG.warn(_LW("Found deprecated policy rule:%s. Please consider "
                         "upgrading your policy configuration file"), pol)
            pol_name, action = pol.rsplit(':', 1)
            try:
                new_actions = DEPRECATED_ACTION_MAP[action]
                new_policies = DEPRECATED_POLICY_MAP[pol_name]
                # bind new actions and policies together
                for actual_policy in ['_'.join(item) for item in
                                      itertools.product(new_actions,
                                                        new_policies)]:
                    if actual_policy not in policies:
                        # New policy, same rule
                        LOG.info(_LI("Inserting policy:%(new_policy)s in "
                                     "place of deprecated "
                                     "policy:%(old_policy)s"),
                                 {'new_policy': actual_policy,
                                  'old_policy': pol})
                        policies[actual_policy] = policies[pol]
                # Remove old-style policy
                del policies[pol]
            except KeyError:
                LOG.error(_LE("Backward compatibility unavailable for "
                              "deprecated policy %s. The policy will "
                              "not be enforced"), pol)
    policy.set_rules(policies)

 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)
         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
             net_id = port['port']['network_id']
             LOG.warn(_LW("Action %(action)s for network %(net_id)s "
                          "could not complete successfully: %(reason)s"),
                      {"action": action, "net_id": net_id, 'reason': e})

 def _create_tunnel_port_name(self, tunnel_type, ip_address):
     try:
         ip_hex = '%08x' % netaddr.IPAddress(ip_address, version=4)
         return '%s-%s' % (tunnel_type, ip_hex)
     except Exception:
         LOG.warn(_LW("Unable to create tunnel port. "
                      "Invalid remote IP: %s"), ip_address)

    def get_traffic_counters(self, chain, wrap=True, zero=False):
        """Return the sum of the traffic counters of all rules of a chain."""
        cmd_tables = self._get_traffic_counters_cmd_tables(chain, wrap)
        if not cmd_tables:
            LOG.warn(_LW('Attempted to get traffic counters of chain %s which '
                         'does not exist'), chain)
            return

        name = get_chain_name(chain, wrap)
        acc = {'pkts': 0, 'bytes': 0}

        for cmd, table in cmd_tables:
            args = [cmd, '-t', table, '-L', name, '-n', '-v', '-x']
            if zero:
                args.append('-Z')
            if self.namespace:
                args = ['ip', 'netns', 'exec', self.namespace] + args
            current_table = (self.execute(args,
                             root_helper=self.root_helper))
            current_lines = current_table.split('\n')

            for line in current_lines[2:]:
                if not line:
                    break
                data = line.split()
                if (len(data) < 2 or
                        not data[0].isdigit() or
                        not data[1].isdigit()):
                    break

                acc['pkts'] += int(data[0])
                acc['bytes'] += int(data[1])

        return acc

    def remove_rule(self, chain, rule, wrap=True, top=False, comment=None):
        """Remove a rule from a chain.

        Note: The rule must be exactly identical to the one that was added.
        You cannot switch arguments around like you can with the iptables
        CLI tool.

        """
        chain = get_chain_name(chain, wrap)
        try:
            if '$' in rule:
                rule = ' '.join(
                    self._wrap_target_chain(e, wrap) for e in rule.split(' '))

            self.rules.remove(IptablesRule(chain, rule, wrap, top,
                                           self.wrap_name,
                                           comment=comment))
            if not wrap:
                self.remove_rules.append(IptablesRule(chain, rule, wrap, top,
                                                      self.wrap_name,
                                                      comment=comment))
        except ValueError:
            LOG.warn(_LW('Tried to remove rule that was not there:'
                         ' %(chain)r %(rule)r %(wrap)r %(top)r'),
                     {'chain': chain, 'rule': rule,
                      'top': top, 'wrap': wrap})