<s> from six . moves import queue <EOL> import time <EOL> from oslo_serialization import jsonutils <EOL> from networking_cisco . _i18n import _LE , _LI <EOL> from networking_cisco . apps . saf . agent import detect_uplink as uplink_det <EOL> from networking_cisco . apps . saf . agent . vdp import ovs_vdp <EOL> from networking_cisco . apps . saf . common import constants <EOL> from networking_cisco . apps . saf . common import dfa_logger as logging <EOL> from networking_cisco . apps . saf . common import rpc <EOL> from networking_cisco . apps . saf . common import utils <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class VdpMsgPriQue ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self ) : <EOL> self . _queue = queue . PriorityQueue ( ) <EOL> def enqueue ( self , priority , msg ) : <EOL> msg_tupl = ( priority , msg ) <EOL> self . _queue . put ( msg_tupl ) <EOL> def dequeue ( self ) : <EOL> msg_tupl = self . _queue . get ( ) <EOL> return msg_tupl [ <NUM_LIT:0> ] , msg_tupl [ <NUM_LIT:1> ] <EOL> def dequeue_nonblock ( self ) : <EOL> msg_tupl = self . _queue . get_nowait ( ) <EOL> return msg_tupl [ <NUM_LIT:0> ] , msg_tupl [ <NUM_LIT:1> ] <EOL> def is_not_empty ( self ) : <EOL> return not self . _queue . empty ( ) <EOL> class VdpQueMsg ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , msg_type , port_uuid = None , vm_mac = None , oui = None , <EOL> net_uuid = None , segmentation_id = None , status = None , <EOL> phy_uplink = None , br_int = None , br_ex = None , root_helper = None ) : <EOL> self . msg_dict = { } <EOL> self . msg_type = msg_type <EOL> if msg_type == constants . VM_MSG_TYPE : <EOL> self . construct_vm_msg ( port_uuid , vm_mac , net_uuid , <EOL> segmentation_id , status , oui , phy_uplink ) <EOL> elif msg_type == constants . UPLINK_MSG_TYPE : <EOL> self . construct_uplink_msg ( status , phy_uplink , br_int , br_ex , <EOL> root_helper ) <EOL> def construct_vm_msg ( self , port_uuid , vm_mac , net_uuid , <EOL> segmentation_id , status , oui , phy_uplink ) : <EOL> self . msg_dict [ '<STR_LIT>' ] = port_uuid <EOL> self . msg_dict [ '<STR_LIT>' ] = vm_mac <EOL> self . msg_dict [ '<STR_LIT>' ] = net_uuid <EOL> self . msg_dict [ '<STR_LIT>' ] = segmentation_id <EOL> self . msg_dict [ '<STR_LIT:status>' ] = status <EOL> self . msg_dict [ '<STR_LIT>' ] = oui <EOL> self . msg_dict [ '<STR_LIT>' ] = phy_uplink <EOL> def construct_uplink_msg ( self , status , phy_uplink , br_int , br_ex , <EOL> root_helper ) : <EOL> self . msg_dict [ '<STR_LIT:status>' ] = status <EOL> self . msg_dict [ '<STR_LIT>' ] = phy_uplink <EOL> self . msg_dict [ '<STR_LIT>' ] = br_int <EOL> self . msg_dict [ '<STR_LIT>' ] = br_ex <EOL> self . msg_dict [ '<STR_LIT>' ] = root_helper <EOL> def get_oui ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_uplink ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_mac ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_status ( self ) : <EOL> return self . msg_dict [ '<STR_LIT:status>' ] <EOL> def get_segmentation_id ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_net_uuid ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_port_uuid ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_integ_br ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_ext_br ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def get_root_helper ( self ) : <EOL> return self . msg_dict [ '<STR_LIT>' ] <EOL> def set_uplink ( self , uplink ) : <EOL> self . msg_dict [ '<STR_LIT>' ] = uplink <EOL> class VdpMgr ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , config_dict , rpc_client , hostname ) : <EOL> self . br_integ = config_dict . get ( '<STR_LIT>' ) <EOL> self . br_ex = config_dict . get ( '<STR_LIT>' ) <EOL> self . root_helper = config_dict . get ( '<STR_LIT>' ) <EOL> self . host_id = config_dict . get ( '<STR_LIT>' ) <EOL> self . node_list = config_dict [ '<STR_LIT>' ] <EOL> self . node_uplink_list = config_dict [ '<STR_LIT>' ] <EOL> self . que = VdpMsgPriQue ( ) <EOL> self . err_que = VdpMsgPriQue ( ) <EOL> self . phy_uplink = None <EOL> self . veth_intf = None <EOL> self . restart_uplink_called = False <EOL> self . ovs_vdp_obj_dict = { } <EOL> self . rpc_clnt = rpc_client <EOL> self . host_name = hostname <EOL> self . uplink_det_compl = False <EOL> self . process_uplink_ongoing = False <EOL> self . uplink_down_cnt = <NUM_LIT:0> <EOL> self . is_os_run = False <EOL> self . static_uplink = False <EOL> self . static_uplink_port = None <EOL> self . static_uplink_first = True <EOL> self . read_static_uplink ( ) <EOL> self . start ( ) <EOL> def read_static_uplink ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . node_list is None or self . node_uplink_list is None : <EOL> return <EOL> for node , port in zip ( self . node_list . split ( '<STR_LIT:U+002C>' ) , <EOL> self . node_uplink_list . split ( '<STR_LIT:U+002C>' ) ) : <EOL> if node . strip ( ) == self . host_name : <EOL> self . static_uplink = True <EOL> self . static_uplink_port = port . strip ( ) <EOL> return <EOL> def update_vm_result ( self , port_uuid , result ) : <EOL> context = { '<STR_LIT>' : self . host_id } <EOL> args = jsonutils . dumps ( dict ( port_uuid = port_uuid , result = result ) ) <EOL> msg = self . rpc_clnt . make_msg ( '<STR_LIT>' , context , msg = args ) <EOL> try : <EOL> resp = self . rpc_clnt . call ( msg ) <EOL> return resp <EOL> except rpc . MessagingTimeout : <EOL> LOG . error ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def process_vm_event ( self , msg , phy_uplink ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT:status>' : msg . get_status ( ) , '<STR_LIT>' : msg . get_mac ( ) , <EOL> '<STR_LIT>' : msg . get_port_uuid ( ) , '<STR_LIT>' : msg . get_oui ( ) } ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> if msg . get_status ( ) == '<STR_LIT>' : <EOL> res_fail = constants . CREATE_FAIL <EOL> else : <EOL> res_fail = constants . DELETE_FAIL <EOL> if ( not self . uplink_det_compl or <EOL> phy_uplink not in self . ovs_vdp_obj_dict ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> self . update_vm_result ( msg . get_port_uuid ( ) , res_fail ) <EOL> return <EOL> ovs_vdp_obj = self . ovs_vdp_obj_dict [ phy_uplink ] <EOL> ret = ovs_vdp_obj . send_vdp_port_event ( msg . get_port_uuid ( ) , <EOL> msg . get_mac ( ) , <EOL> msg . get_net_uuid ( ) , <EOL> msg . get_segmentation_id ( ) , <EOL> msg . get_status ( ) , <EOL> msg . get_oui ( ) ) <EOL> if not ret : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> self . update_vm_result ( msg . get_port_uuid ( ) , res_fail ) <EOL> else : <EOL> self . update_vm_result ( msg . get_port_uuid ( ) , <EOL> constants . RESULT_SUCCESS ) <EOL> def process_uplink_event ( self , msg , phy_uplink ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , <EOL> { '<STR_LIT>' : msg . get_status ( ) , '<STR_LIT>' : phy_uplink } ) <EOL> if msg . get_status ( ) == '<STR_LIT>' : <EOL> ovs_exc_raised = False <EOL> try : <EOL> self . ovs_vdp_obj_dict [ phy_uplink ] = ovs_vdp . OVSNeutronVdp ( <EOL> phy_uplink , msg . get_integ_br ( ) , msg . get_ext_br ( ) , <EOL> msg . get_root_helper ( ) ) <EOL> except Exception as exc : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , str ( exc ) ) <EOL> ovs_exc_raised = True <EOL> if ( ovs_exc_raised or not self . ovs_vdp_obj_dict [ phy_uplink ] . <EOL> is_lldpad_setup_done ( ) ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> self . err_que . enqueue ( constants . Q_UPL_PRIO , msg ) <EOL> else : <EOL> self . uplink_det_compl = True <EOL> veth_intf = ( self . ovs_vdp_obj_dict [ self . phy_uplink ] . <EOL> get_lldp_bridge_port ( ) ) <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT>' : self . phy_uplink , '<STR_LIT>' : veth_intf } ) <EOL> self . save_uplink ( uplink = self . phy_uplink , veth_intf = veth_intf ) <EOL> elif msg . get_status ( ) == '<STR_LIT>' : <EOL> if phy_uplink in self . ovs_vdp_obj_dict : <EOL> self . ovs_vdp_obj_dict [ phy_uplink ] . clear_obj_params ( ) <EOL> else : <EOL> ovs_vdp . delete_uplink_and_flows ( self . root_helper , self . br_ex , <EOL> phy_uplink ) <EOL> self . save_uplink ( ) <EOL> def process_queue ( self ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> while True : <EOL> prio , msg = self . que . dequeue ( ) <EOL> msg_type = msg . msg_type <EOL> phy_uplink = msg . get_uplink ( ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , msg_type ) <EOL> try : <EOL> if msg_type == constants . VM_MSG_TYPE : <EOL> self . process_vm_event ( msg , phy_uplink ) <EOL> elif msg_type == constants . UPLINK_MSG_TYPE : <EOL> try : <EOL> self . process_uplink_event ( msg , phy_uplink ) <EOL> except Exception as eu : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , str ( eu ) ) <EOL> self . process_uplink_ongoing = False <EOL> except Exception as e : <EOL> LOG . exceptin ( _LE ( "<STR_LIT>" ) , str ( e ) ) <EOL> def process_err_queue ( self ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> try : <EOL> while self . err_que . is_not_empty ( ) : <EOL> prio , msg = self . err_que . dequeue_nonblock ( ) <EOL> msg_type = msg . msg_type <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , <EOL> msg_type ) <EOL> if msg_type == constants . UPLINK_MSG_TYPE : <EOL> self . que . enqueue ( constants . Q_UPL_PRIO , msg ) <EOL> except Exception as e : <EOL> LOG . exceptin ( _LE ( "<STR_LIT>" ) , str ( e ) ) <EOL> def start ( self ) : <EOL> thr_q = utils . EventProcessingThread ( "<STR_LIT>" , self , '<STR_LIT>' ) <EOL> thr_q . start ( ) <EOL> task_err_proc = utils . PeriodicTask ( constants . ERR_PROC_INTERVAL , <EOL> self . process_err_queue ) <EOL> task_err_proc . run ( ) <EOL> task_uplink = utils . PeriodicTask ( constants . UPLINK_DET_INTERVAL , <EOL> self . vdp_uplink_proc_top ) <EOL> task_uplink . run ( ) <EOL> def is_openstack_running ( self ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> if ( ovs_vdp . is_bridge_present ( self . br_ex , self . root_helper ) and <EOL> ovs_vdp . is_bridge_present ( self . br_integ , <EOL> self . root_helper ) ) : <EOL> return True <EOL> else : <EOL> return False <EOL> except Exception as e : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , str ( e ) ) <EOL> return False <EOL> def vdp_uplink_proc_top ( self ) : <EOL> try : <EOL> self . vdp_uplink_proc ( ) <EOL> except Exception as e : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def save_uplink ( self , uplink = "<STR_LIT>" , veth_intf = "<STR_LIT>" ) : <EOL> context = { } <EOL> args = jsonutils . dumps ( dict ( agent = self . host_id , uplink = uplink , <EOL> veth_intf = veth_intf ) ) <EOL> msg = self . rpc_clnt . make_msg ( '<STR_LIT>' , context , msg = args ) <EOL> try : <EOL> resp = self . rpc_clnt . call ( msg ) <EOL> return resp <EOL> except rpc . MessagingTimeout : <EOL> LOG . error ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def static_uplink_detect ( self , veth ) : <EOL> """<STR_LIT>""" <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , { '<STR_LIT>' : veth } ) <EOL> if self . static_uplink_first : <EOL> self . static_uplink_first = False <EOL> if self . phy_uplink is not None and ( <EOL> self . phy_uplink != self . static_uplink_port ) : <EOL> return '<STR_LIT>' <EOL> if veth is None : <EOL> return self . static_uplink_port <EOL> else : <EOL> return '<STR_LIT>' <EOL> def vdp_uplink_proc ( self ) : <EOL> '''<STR_LIT>''' <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> if not self . is_os_run : <EOL> if not self . is_openstack_running ( ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> return <EOL> else : <EOL> self . is_os_run = True <EOL> if not self . restart_uplink_called or self . process_uplink_ongoing : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT>' : self . restart_uplink_called , <EOL> '<STR_LIT>' : self . process_uplink_ongoing } ) <EOL> return <EOL> if self . phy_uplink is not None : <EOL> if ( self . uplink_det_compl and ( <EOL> self . phy_uplink not in self . ovs_vdp_obj_dict ) ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , self . phy_uplink ) <EOL> return <EOL> if self . phy_uplink in self . ovs_vdp_obj_dict : <EOL> self . veth_intf = ( self . ovs_vdp_obj_dict [ self . phy_uplink ] . <EOL> get_lldp_bridge_port ( ) ) <EOL> else : <EOL> if self . veth_intf is None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> return <EOL> if self . static_uplink : <EOL> ret = self . static_uplink_detect ( self . veth_intf ) <EOL> else : <EOL> ret = uplink_det . detect_uplink ( self . veth_intf ) <EOL> if ret is '<STR_LIT>' : <EOL> if self . phy_uplink is None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> return <EOL> self . uplink_down_cnt = self . uplink_down_cnt + <NUM_LIT:1> <EOL> if not self . static_uplink and ( <EOL> self . uplink_down_cnt < constants . UPLINK_DOWN_THRES ) : <EOL> return <EOL> self . process_uplink_ongoing = True <EOL> upl_msg = VdpQueMsg ( constants . UPLINK_MSG_TYPE , <EOL> status = '<STR_LIT>' , <EOL> phy_uplink = self . phy_uplink , <EOL> br_int = self . br_integ , br_ex = self . br_ex , <EOL> root_helper = self . root_helper ) <EOL> self . que . enqueue ( constants . Q_UPL_PRIO , upl_msg ) <EOL> self . phy_uplink = None <EOL> self . veth_intf = None <EOL> self . uplink_det_compl = False <EOL> self . uplink_down_cnt = <NUM_LIT:0> <EOL> elif ret is None : <EOL> if self . veth_intf is not None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> return <EOL> self . save_uplink ( ) <EOL> elif ret is '<STR_LIT>' : <EOL> if self . veth_intf is None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> return <EOL> self . uplink_down_cnt = <NUM_LIT:0> <EOL> if not self . uplink_det_compl : <EOL> if self . phy_uplink is None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> return <EOL> self . process_uplink_ongoing = True <EOL> upl_msg = VdpQueMsg ( constants . UPLINK_MSG_TYPE , <EOL> status = '<STR_LIT>' , <EOL> phy_uplink = self . phy_uplink , <EOL> br_int = self . br_integ , br_ex = self . br_ex , <EOL> root_helper = self . root_helper ) <EOL> self . que . enqueue ( constants . Q_UPL_PRIO , upl_msg ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> else : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , ret ) <EOL> self . save_uplink ( uplink = ret , veth_intf = self . veth_intf ) <EOL> self . phy_uplink = ret <EOL> self . process_uplink_ongoing = True <EOL> upl_msg = VdpQueMsg ( constants . UPLINK_MSG_TYPE , <EOL> status = '<STR_LIT>' , <EOL> phy_uplink = self . phy_uplink , <EOL> br_int = self . br_integ , br_ex = self . br_ex , <EOL> root_helper = self . root_helper ) <EOL> self . que . enqueue ( constants . Q_UPL_PRIO , upl_msg ) <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> def vdp_vm_event ( self , vm_dict ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT:status>' : vm_dict [ '<STR_LIT:status>' ] , '<STR_LIT>' : vm_dict [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : vm_dict [ '<STR_LIT>' ] , '<STR_LIT>' : vm_dict [ '<STR_LIT>' ] } ) <EOL> vm_msg = VdpQueMsg ( constants . VM_MSG_TYPE , <EOL> port_uuid = vm_dict [ '<STR_LIT>' ] , <EOL> vm_mac = vm_dict [ '<STR_LIT>' ] , <EOL> net_uuid = vm_dict [ '<STR_LIT>' ] , <EOL> segmentation_id = vm_dict [ '<STR_LIT>' ] , <EOL> status = vm_dict [ '<STR_LIT:status>' ] , <EOL> oui = vm_dict [ '<STR_LIT>' ] , <EOL> phy_uplink = self . phy_uplink ) <EOL> self . que . enqueue ( constants . Q_VM_PRIO , vm_msg ) <EOL> def dfa_uplink_restart ( self , uplink_dict ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , uplink_dict ) <EOL> if self . phy_uplink is not None : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , self . phy_uplink ) <EOL> return <EOL> uplink = uplink_dict . get ( '<STR_LIT>' ) <EOL> veth_intf = uplink_dict . get ( '<STR_LIT>' ) <EOL> if uplink is None or len ( uplink ) == <NUM_LIT:0> : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> self . restart_uplink_called = True <EOL> return <EOL> if veth_intf is not None and len ( veth_intf ) != <NUM_LIT:0> : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , <EOL> { '<STR_LIT>' : uplink , '<STR_LIT>' : veth_intf } ) <EOL> self . phy_uplink = uplink <EOL> self . veth_intf = veth_intf <EOL> self . restart_uplink_called = True <EOL> return <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , uplink ) <EOL> ovs_vdp . delete_uplink_and_flows ( self . root_helper , self . br_ex , uplink ) <EOL> self . restart_uplink_called = True </s>
