<s> from neutron_lib import constants <EOL> from oslo_config import cfg <EOL> from oslo_log import log <EOL> from neutron . _i18n import _ , _LE , _LW <EOL> from neutron . extensions import portbindings <EOL> from neutron . plugins . common import constants as p_const <EOL> from neutron . plugins . ml2 import driver_api as api <EOL> from neutron . plugins . ml2 . drivers . mech_sriov . mech_driver import exceptions as exc <EOL> from neutron . services . qos import qos_consts <EOL> LOG = log . getLogger ( __name__ ) <EOL> VIF_TYPE_HW_VEB = '<STR_LIT>' <EOL> VIF_TYPE_HOSTDEV_PHY = '<STR_LIT>' <EOL> FLAT_VLAN = <NUM_LIT:0> <EOL> sriov_opts = [ <EOL> cfg . ListOpt ( '<STR_LIT>' , <EOL> default = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> help = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) , <EOL> ] <EOL> cfg . CONF . register_opts ( sriov_opts , "<STR_LIT>" ) <EOL> class SriovNicSwitchMechanismDriver ( api . MechanismDriver ) : <EOL> """<STR_LIT>""" <EOL> supported_qos_rule_types = [ qos_consts . RULE_TYPE_BANDWIDTH_LIMIT ] <EOL> def __init__ ( self , <EOL> agent_type = constants . AGENT_TYPE_NIC_SWITCH , <EOL> vif_details = { portbindings . CAP_PORT_FILTER : False } , <EOL> supported_vnic_types = [ portbindings . VNIC_DIRECT , <EOL> portbindings . VNIC_MACVTAP , <EOL> portbindings . VNIC_DIRECT_PHYSICAL ] , <EOL> supported_pci_vendor_info = None ) : <EOL> """<STR_LIT>""" <EOL> self . agent_type = agent_type <EOL> self . supported_vnic_types = supported_vnic_types <EOL> self . vnic_type_for_vif_type = ( <EOL> { vtype : VIF_TYPE_HOSTDEV_PHY <EOL> if vtype == portbindings . VNIC_DIRECT_PHYSICAL <EOL> else VIF_TYPE_HW_VEB <EOL> for vtype in self . supported_vnic_types } ) <EOL> self . vif_details = vif_details <EOL> self . supported_network_types = ( p_const . TYPE_VLAN , p_const . TYPE_FLAT ) <EOL> def initialize ( self ) : <EOL> try : <EOL> self . pci_vendor_info = cfg . CONF . ml2_sriov . supported_pci_vendor_devs <EOL> self . _check_pci_vendor_config ( self . pci_vendor_info ) <EOL> except ValueError : <EOL> LOG . exception ( _LE ( "<STR_LIT>" ) ) <EOL> raise cfg . Error ( _ ( "<STR_LIT>" ) ) <EOL> def bind_port ( self , context ) : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:port>' : context . current [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : context . network . current [ '<STR_LIT:id>' ] } ) <EOL> vnic_type = context . current . get ( portbindings . VNIC_TYPE , <EOL> portbindings . VNIC_NORMAL ) <EOL> if vnic_type not in self . supported_vnic_types : <EOL> LOG . debug ( "<STR_LIT>" , <EOL> vnic_type ) <EOL> return <EOL> vif_type = self . vnic_type_for_vif_type . get ( vnic_type , <EOL> VIF_TYPE_HW_VEB ) <EOL> if not self . _check_supported_pci_vendor_device ( context ) : <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> return <EOL> if vnic_type == portbindings . VNIC_DIRECT_PHYSICAL : <EOL> self . try_to_bind ( context , None , vif_type ) <EOL> return <EOL> for agent in context . host_agents ( self . agent_type ) : <EOL> LOG . debug ( "<STR_LIT>" , agent ) <EOL> if agent [ '<STR_LIT>' ] : <EOL> if self . try_to_bind ( context , agent , vif_type ) : <EOL> return <EOL> else : <EOL> LOG . warning ( _LW ( "<STR_LIT>" ) , <EOL> agent ) <EOL> def try_to_bind ( self , context , agent , vif_type ) : <EOL> for segment in context . segments_to_bind : <EOL> if self . check_segment ( segment , agent ) : <EOL> port_status = ( constants . PORT_STATUS_ACTIVE if agent is None <EOL> else constants . PORT_STATUS_DOWN ) <EOL> context . set_binding ( segment [ api . ID ] , <EOL> vif_type , <EOL> self . _get_vif_details ( segment ) , <EOL> port_status ) <EOL> LOG . debug ( "<STR_LIT>" , segment ) <EOL> return True <EOL> return False <EOL> def filter_hosts_with_segment_access ( <EOL> self , context , segments , candidate_hosts , agent_getter ) : <EOL> hosts = set ( ) <EOL> filters = { '<STR_LIT:host>' : candidate_hosts , '<STR_LIT>' : [ self . agent_type ] } <EOL> for agent in agent_getter ( context , filters = filters ) : <EOL> if any ( self . check_segment ( s , agent ) for s in segments ) : <EOL> hosts . add ( agent [ '<STR_LIT:host>' ] ) <EOL> return hosts <EOL> def check_segment ( self , segment , agent = None ) : <EOL> """<STR_LIT>""" <EOL> network_type = segment [ api . NETWORK_TYPE ] <EOL> if network_type in self . supported_network_types : <EOL> if agent : <EOL> mappings = agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , { } ) <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : segment , '<STR_LIT>' : mappings } ) <EOL> return segment [ api . PHYSICAL_NETWORK ] in mappings <EOL> return True <EOL> return False <EOL> def _check_supported_pci_vendor_device ( self , context ) : <EOL> if self . pci_vendor_info : <EOL> profile = context . current . get ( portbindings . PROFILE , { } ) <EOL> if not profile : <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> return False <EOL> pci_vendor_info = profile . get ( '<STR_LIT>' ) <EOL> if not pci_vendor_info : <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> return False <EOL> if pci_vendor_info not in self . pci_vendor_info : <EOL> LOG . debug ( "<STR_LIT>" , pci_vendor_info ) <EOL> return False <EOL> return True <EOL> return False <EOL> def _get_vif_details ( self , segment ) : <EOL> network_type = segment [ api . NETWORK_TYPE ] <EOL> if network_type == p_const . TYPE_FLAT : <EOL> vlan_id = FLAT_VLAN <EOL> elif network_type == p_const . TYPE_VLAN : <EOL> vlan_id = segment [ api . SEGMENTATION_ID ] <EOL> else : <EOL> raise exc . SriovUnsupportedNetworkType ( net_type = network_type ) <EOL> vif_details = self . vif_details . copy ( ) <EOL> vif_details [ portbindings . VIF_DETAILS_VLAN ] = str ( vlan_id ) <EOL> return vif_details <EOL> @ staticmethod <EOL> def _check_pci_vendor_config ( pci_vendor_list ) : <EOL> for pci_vendor_info in pci_vendor_list : <EOL> try : <EOL> vendor_id , product_id = [ <EOL> item . strip ( ) for item in pci_vendor_info . split ( '<STR_LIT::>' ) <EOL> if item . strip ( ) ] <EOL> except ValueError : <EOL> raise ValueError ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % <EOL> pci_vendor_info ) </s>
