<s> import os <EOL> from neutron_lib import constants <EOL> from oslo_config import cfg <EOL> from neutron . agent import securitygroups_rpc <EOL> from neutron . extensions import portbindings <EOL> from neutron . plugins . common import constants as p_constants <EOL> from neutron . plugins . ml2 import driver_api as api <EOL> from neutron . plugins . ml2 . drivers import mech_agent <EOL> from neutron . plugins . ml2 . drivers . openvswitch . agent . common import constants as a_const <EOL> from neutron . services . qos import qos_consts <EOL> IPTABLES_FW_DRIVER_FULL = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class OpenvswitchMechanismDriver ( mech_agent . SimpleAgentMechanismDriverBase ) : <EOL> """<STR_LIT>""" <EOL> supported_qos_rule_types = [ qos_consts . RULE_TYPE_BANDWIDTH_LIMIT , <EOL> qos_consts . RULE_TYPE_DSCP_MARK ] <EOL> def __init__ ( self ) : <EOL> sg_enabled = securitygroups_rpc . is_firewall_enabled ( ) <EOL> hybrid_plug_required = ( cfg . CONF . SECURITYGROUP . firewall_driver in ( <EOL> IPTABLES_FW_DRIVER_FULL , '<STR_LIT>' ) ) and sg_enabled <EOL> vif_details = { portbindings . CAP_PORT_FILTER : sg_enabled , <EOL> portbindings . OVS_HYBRID_PLUG : hybrid_plug_required } <EOL> super ( OpenvswitchMechanismDriver , self ) . __init__ ( <EOL> constants . AGENT_TYPE_OVS , <EOL> portbindings . VIF_TYPE_OVS , <EOL> vif_details ) <EOL> def get_allowed_network_types ( self , agent ) : <EOL> return ( agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , [ ] ) + <EOL> [ p_constants . TYPE_LOCAL , p_constants . TYPE_FLAT , <EOL> p_constants . TYPE_VLAN ] ) <EOL> def get_mappings ( self , agent ) : <EOL> return agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , { } ) <EOL> def check_vlan_transparency ( self , context ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> def try_to_bind_segment_for_agent ( self , context , segment , agent ) : <EOL> if self . check_segment_for_agent ( segment , agent ) : <EOL> context . set_binding ( segment [ api . ID ] , <EOL> self . get_vif_type ( agent , context ) , <EOL> self . get_vif_details ( agent , context ) ) <EOL> return True <EOL> else : <EOL> return False <EOL> def get_vif_type ( self , agent , context ) : <EOL> caps = agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , { } ) <EOL> if ( a_const . OVS_DPDK_VHOST_USER in caps . get ( '<STR_LIT>' , [ ] ) and <EOL> agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) == <EOL> a_const . OVS_DATAPATH_NETDEV ) : <EOL> return portbindings . VIF_TYPE_VHOST_USER <EOL> return self . vif_type <EOL> def get_vif_details ( self , agent , context ) : <EOL> if ( agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) != <EOL> a_const . OVS_DATAPATH_NETDEV ) : <EOL> return self . vif_details <EOL> caps = agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , { } ) <EOL> if a_const . OVS_DPDK_VHOST_USER in caps . get ( '<STR_LIT>' , [ ] ) : <EOL> sock_path = self . agent_vhu_sockpath ( agent , context . current [ '<STR_LIT:id>' ] ) <EOL> return { <EOL> portbindings . CAP_PORT_FILTER : False , <EOL> portbindings . VHOST_USER_MODE : <EOL> portbindings . VHOST_USER_MODE_CLIENT , <EOL> portbindings . VHOST_USER_OVS_PLUG : True , <EOL> portbindings . VHOST_USER_SOCKET : sock_path <EOL> } <EOL> return self . vif_details <EOL> @ staticmethod <EOL> def agent_vhu_sockpath ( agent , port_id ) : <EOL> """<STR_LIT>""" <EOL> sockdir = agent [ '<STR_LIT>' ] . get ( '<STR_LIT>' , <EOL> a_const . VHOST_USER_SOCKET_DIR ) <EOL> sock_name = ( constants . VHOST_USER_DEVICE_PREFIX + port_id ) [ : <NUM_LIT> ] <EOL> return os . path . join ( sockdir , sock_name ) </s>
