From 641db953ef625c93bbbbf1216550e5b8b2af9c2b Mon Sep 17 00:00:00 2001 From: Ken Subratie Date: Mon, 2 Oct 2023 16:11:27 -0400 Subject: [PATCH] Update for multiprocess Tincan dataplane --- evio/broker/__init__.py | 20 +- evio/broker/broker.py | 33 +- evio/broker/controller_module.py | 13 +- evio/broker/nexus.py | 22 +- evio/broker/process_proxy.py | 14 +- evio/broker/remote_action.py | 4 +- evio/controllers/bounded_flood.py | 15 +- evio/controllers/bridge_controller.py | 127 ++----- evio/controllers/geneve_tunnel.py | 32 +- evio/controllers/graph_builder.py | 1 + evio/controllers/link_manager.py | 251 +++++-------- evio/controllers/network_graph.py | 5 + evio/controllers/signal.py | 8 +- evio/controllers/tincan_tunnel.py | 518 ++++++++++++++------------ evio/controllers/topology.py | 157 ++------ evio/controllers/tunnel.py | 3 +- 16 files changed, 542 insertions(+), 681 deletions(-) diff --git a/evio/broker/__init__.py b/evio/broker/__init__.py index 9aae849..a457d73 100644 --- a/evio/broker/__init__.py +++ b/evio/broker/__init__.py @@ -68,6 +68,7 @@ __all__ = [ "SUCCESSIVE_FAIL_INCR", "SUCCESSIVE_FAIL_DECR", "STALE_INTERVAL", + "MAX_HEARTBEATS", "perfd", "CONFIG", "CTL_CREATE_CTRL_LINK", @@ -123,6 +124,8 @@ MAX_CONCURRENT_OPS: Literal[1] = 1 SUCCESSIVE_FAIL_INCR: Literal[1] = 1 SUCCESSIVE_FAIL_DECR: Literal[2] = 2 STALE_INTERVAL = float(2 * 3600) # 2 hrs +MAX_HEARTBEATS: Literal[5] = 3 + perfd = PerformanceData(LogFile=os.path.join(LOG_DIRECTORY, PERFDATA_LOG_NAME)) CONFIG = { @@ -174,25 +177,18 @@ CTL_ECHO = { "ControlType": "Request", "Request": {"Command": "Echo", "Message": "ECHO TEST"}, } -CTL_QUERY_TUNNEL_INFO = { - "ProtocolVersion": EVIO_VER_CTL, - "TransactionId": 0, - "ControlType": "Request", - "Request": {"Command": "QueryOverlayInfo", "OverlayId": "", "TunnelId": ""}, -} + CTL_CREATE_TUNNEL = { "ProtocolVersion": EVIO_VER_CTL, "ControlType": "Request", "TransactionId": 0, "Request": { "Command": "CreateTunnel", - "OverlayId": "", "NodeId": "", "TunnelId": "", "TapName": "", "StunServers": [], "TurnServers": [], - "Type": "", }, } CTL_CREATE_LINK = { @@ -201,7 +197,6 @@ CTL_CREATE_LINK = { "ControlType": "Request", "Request": { "Command": "CreateLink", - "OverlayId": "", "TunnelId": "", "LinkId": "", "PeerInfo": {"UID": "", "MAC": "", "FPR": ""}, @@ -211,13 +206,13 @@ CTL_REMOVE_TUNNEL = { "ProtocolVersion": EVIO_VER_CTL, "TransactionId": 0, "ControlType": "Request", - "Request": {"Command": "RemoveTunnel", "OverlayId": "", "TunnelId": ""}, + "Request": {"Command": "RemoveTunnel", "TunnelId": ""}, } CTL_REMOVE_LINK = { "ProtocolVersion": EVIO_VER_CTL, "TransactionId": 0, "ControlType": "Request", - "Request": {"Command": "RemoveLink", "OverlayId": "", "LinkId": ""}, + "Request": {"Command": "RemoveLink", "TunnelId": "", "LinkId": ""}, } RESP = { "ProtocolVersion": EVIO_VER_CTL, @@ -230,7 +225,7 @@ CTL_QUERY_LINK_STATS = { "ProtocolVersion": EVIO_VER_CTL, "TransactionId": 0, "ControlType": "Request", - "Request": {"Command": "QueryLinkStats", "TunnelIds": []}, + "Request": {"Command": "QueryLinkStats", "TunnelId": ""}, } CTL_QUERY_CAS = { "ProtocolVersion": EVIO_VER_CTL, @@ -238,7 +233,6 @@ CTL_QUERY_CAS = { "ControlType": "Request", "Request": { "Command": "QueryCandidateAddressSet", - "OverlayId": "", "LinkId": "", }, } diff --git a/evio/broker/broker.py b/evio/broker/broker.py index 6cb447a..2018dbc 100644 --- a/evio/broker/broker.py +++ b/evio/broker/broker.py @@ -60,6 +60,8 @@ from .process_proxy import ProcessProxy, ProxyMsg from .subscription import Subscription from .timed_transactions import TimedTransactions, Transaction +# import faulthandler + class Broker: @staticmethod @@ -103,7 +105,7 @@ class Broker: def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: - print(exc_type, exc_val, exc_tb) + print("__exit__: ", exc_type, exc_val, exc_tb) return self.terminate() def parse_config(self): @@ -172,13 +174,18 @@ class Broker: "[%(asctime)s.%(msecs)03d] %(levelname)s:%(name)s: %(message)s", datefmt="%Y%m%d %H:%M:%S", ) - file_handler = TimedRotatingFileHandler( - filename=bkr_logname, when="midnight", backupCount=7, utc=True + file_handler = RotatingFileHandler( + filename=bkr_logname, + maxBytes=self._config["Broker"].get("MaxFileSize", MAX_FILE_SIZE), + backupCount=self._config["Broker"].get("MaxArchives", MAX_ARCHIVES), ) broker_log_level = self._config["Broker"].get( "BrokerLogLevel", BROKER_LOG_LEVEL ) - file_handler.setLevel(broker_log_level) + file_handler.setLevel( + "DEBUG" + ) # the root file handler has the broadest capture level + # file_handler.setLevel(broker_log_level) file_handler.setFormatter(formatter) handlers.append(file_handler) # console logging @@ -203,10 +210,8 @@ class Broker: # if os.path.isfile(logname): # os.remove(logname) level = self._config["Broker"].get("LogLevel", def_log_level) - file_handler = RotatingFileHandler( - filename=logname, - maxBytes=self._config["Broker"].get("MaxFileSize", MAX_FILE_SIZE), - backupCount=self._config["Broker"].get("MaxArchives", MAX_ARCHIVES), + file_handler = TimedRotatingFileHandler( + filename=logname, when="midnight", backupCount=7, utc=True ) file_handler.setFormatter(formatter) file_handler.setLevel(level) @@ -348,12 +353,13 @@ class Broker: def run(self): for sig in [signal.SIGINT, signal.SIGTERM]: signal.signal(sig, Broker.__handler) - # sleeps until exit signal is received - signal.pause() + signo = signal.sigwait([signal.SIGINT, signal.SIGTERM]) + self.logger.debug("Received Signal: %s", signal.Signals(signo).name) def terminate(self): - self._timers.terminate() with self._nexus_lock: + self._timers.terminate() + self._ipc.terminate() for ctrl_name in reversed(self._load_order): wn = self._nexus_map[ctrl_name]._cm_thread.name self._nexus_map[ctrl_name].work_queue.put(None) @@ -361,7 +367,6 @@ class Broker: wn = self._nexus_map[ctrl_name]._cm_thread.name self._nexus_map[ctrl_name]._cm_thread.join() self.logger.info("%s exited", wn) - self._ipc.terminate() for ql in self._cm_qlisteners: ql.stop() @@ -499,7 +504,7 @@ class Broker: else: tgt = task["Request"].get("Recipient") if tgt is None: - self.logger.warning("No recipient specified in IPC message") + self.logger.warning("No recipient specified in IPC message %s", msg) return with self._nexus_lock: nexus = self._nexus_map[tgt] @@ -512,3 +517,5 @@ class Broker: if __name__ == "__main__": cf = Broker() cf.initialize() + cf.run() + cf.terminate() diff --git a/evio/broker/controller_module.py b/evio/broker/controller_module.py index 043e585..075506b 100644 --- a/evio/broker/controller_module.py +++ b/evio/broker/controller_module.py @@ -168,14 +168,14 @@ class ControllerModule: self.logger.info("controller state: %s", state) def register_cbt( - self, _recipient, _action, _params=None, parent_cbt=None, **kwargs + self, _recipient, _action, _params=None, _parent_cbt=None, **kwargs ): cbt = self._nexus.create_cbt( initiator=self.name, recipient=_recipient, action=_action, params=_params, - parent_cbt=parent_cbt, + parent_cbt=_parent_cbt, **kwargs, ) self._nexus.submit_req_cbt(cbt) @@ -192,10 +192,15 @@ class ControllerModule: self._nexus.submit_req_cbt(cbt) def create_cbt( - self, initiator, recipient, action, params=None, parent_cbt=None, **kwargs + self, _recipient, _action, _params=None, _parent_cbt=None, **kwargs ) -> CBT: return self._nexus.create_cbt( - initiator, recipient, action, params, parent_cbt, **kwargs + initiator=self.name, + recipient=_recipient, + action=_action, + params=_params, + parent_cbt=_parent_cbt, + **kwargs, ) def complete_cbt(self, cbt: CBT): diff --git a/evio/broker/nexus.py b/evio/broker/nexus.py index 93d8e39..6fbd907 100644 --- a/evio/broker/nexus.py +++ b/evio/broker/nexus.py @@ -60,6 +60,10 @@ class Nexus: return self._cm_queue def submit_req_cbt(self, cbt): + if cbt is None: + self._controller.logger.warning( + "None is not a permissible CBT value for submit_req_cbt" + ) if cbt.is_request and not cbt.is_submited: cbt.time_submited = time.time() self._broker.submit_cbt(cbt) @@ -93,6 +97,10 @@ class Nexus: cbt.time_freed = time.time() def complete_cbt(self, cbt): + if cbt is None: + self._controller.logger.warning( + "None is not a permissible CBT value for complete_cbt" + ) self._pending_cbts.pop(cbt.tag, None) cbt.time_completed = time.time() self._broker.submit_cbt(cbt) @@ -141,7 +149,17 @@ class Nexus: self._controller.handle_ipc(cbt) except RuntimeError as err: self._controller.logger.warning( - "Process CBT exception: %s\nCBT: %s", err, cbt, exc_info=True + "Process CBT RuntimeError exception: %s\nCBT: %s", + err, + cbt, + exc_info=True, + ) + except KeyError as kerr: + self._controller.logger.warning( + "Process CBT KeyError exception: %s\nCBT: %s", + kerr, + cbt, + exc_info=True, ) finally: self._cm_queue.task_done() @@ -159,7 +177,7 @@ class Nexus: self.work_queue.put(cbt) else: self._controller.logger.info( - f"Unexpected CBT state when expired event. {cbt}" + f"Unexpected CBT state for expired event. {cbt}" ) def _schedule_ctlr_update(self): diff --git a/evio/broker/process_proxy.py b/evio/broker/process_proxy.py index 51ff845..fc2bc1a 100644 --- a/evio/broker/process_proxy.py +++ b/evio/broker/process_proxy.py @@ -84,17 +84,15 @@ class ProcessProxy: the controller modules and external local processes. """ - def __init__( - self, - dispatch_msg_cb, - logger: logging.Logger, - ): + def __init__(self, dispatch_msg_cb, logger: logging.Logger, create_svc_thread=True): self.logger = logger self.tx_que = queue.Queue() self.dispatch_msg = dispatch_msg_cb - self._svr_thread = threading.Thread( - target=self.serve, name="ProcessProxyServer", daemon=False - ) + self._svr_thread = None + if create_svc_thread: + self._svr_thread = threading.Thread( + target=self.serve, name="ProcessProxyServer", daemon=False + ) self._exit_ev = threading.Event() self._server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) self._server_sock.setblocking(0) diff --git a/evio/broker/remote_action.py b/evio/broker/remote_action.py index 7083c92..bb71546 100644 --- a/evio/broker/remote_action.py +++ b/evio/broker/remote_action.py @@ -66,8 +66,6 @@ class RemoteAction: def submit_remote_act(self, cm, parent_cbt=None, **kwargs): self.initiator_id = cm.node_id self.initiator_cm = cm.name - cbt = cm.create_cbt( - cm.name, "Signal", "SIG_REMOTE_ACTION", self, parent_cbt, **kwargs - ) + cbt = cm.create_cbt("Signal", "SIG_REMOTE_ACTION", self, parent_cbt, **kwargs) self.action_tag = cbt.tag cm.submit_cbt(cbt) diff --git a/evio/controllers/bounded_flood.py b/evio/controllers/bounded_flood.py index 17a3df2..fd2b327 100644 --- a/evio/controllers/bounded_flood.py +++ b/evio/controllers/bounded_flood.py @@ -666,6 +666,7 @@ class EvioSwitch: ) elif in_port in self._leaf_prts: self._leaf_macs.add(src_mac) + else: self.logger.debug( f"learn sw:{self.name}, leaf_mac:{src_mac}, ingress:{in_port}" ) @@ -870,10 +871,8 @@ class BoundedFlood(app_manager.RyuApp): self.evio_portal.terminate() hub.joinall(self._monitors) self.logger.info("BoundedFlood terminated") - print("BoundedFlood terminated") - os.makedirs("/var/log/evio/bfterm", exist_ok=True) - # self._que_listener.stop() - # logging.shutdown() + self._que_listener.stop() + logging.shutdown() @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER) def switch_features_handler(self, ev): @@ -1012,9 +1011,9 @@ class BoundedFlood(app_manager.RyuApp): else: """Vanilla Ethernet frame but the destination MAC is not in our LT. Currently, only broadcast addresses originating from local leaf ports are broadcasted using FRB. - Multiricepient frames that ingress on a link port is a protocol logic error, and + Multiricepient frames that ingress on a link port is a protocol violation, and flooding unicast frames which have no LT info, prevents accumulating enough port - data to ever create a flow rule""" + data to create a flow rule""" if in_port in sw.leaf_ports and is_multiricepient(eth.dst): self._broadcast_frame(msg.datapath, pkt, in_port, msg) elif in_port not in sw.leaf_ports and is_multiricepient(eth.dst): @@ -1294,7 +1293,9 @@ class BoundedFlood(app_manager.RyuApp): ofproto = datapath.ofproto parser = datapath.ofproto_parser inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS, actions)] - self.logger.debug("Adding flow rule %s: %s", datapath.id, match) + self.logger.debug( + "Adding flow rule %s: %s", self._lt[datapath.id].name, match + ) mod = parser.OFPFlowMod( datapath=datapath, priority=priority, diff --git a/evio/controllers/bridge_controller.py b/evio/controllers/bridge_controller.py index 1bc96c4..9b0e71d 100644 --- a/evio/controllers/bridge_controller.py +++ b/evio/controllers/bridge_controller.py @@ -27,12 +27,7 @@ except ImportError: import copy import logging -import signal - -# import socket -# import select -# import time -# import socketserver +import subprocess import threading from abc import ABCMeta, abstractmethod from collections.abc import MutableMapping @@ -55,18 +50,7 @@ from broker.controller_module import ControllerModule from broker.process_proxy import ProxyMsg from pyroute2 import IPRoute -from .tunnel import DATAPLANE_TYPES, TUNNEL_EVENTS - -# BR_NAME_MAX_LENGTH: Literal[15] = 15 -# NAME_PREFIX_EVI: Literal["evi"] = "evi" -# NAME_PREFIX_APP_BR: Literal["app"] = "app" -# MTU: Literal[1410] = 1410 -# BRIDGE_AUTO_DELETE: bool = True -# DEFAULT_BRIDGE_PROVIDER: Literal["OVS"] = "OVS" -# DEFAULT_SWITCH_PROTOCOL: Literal["BF"] = "BF" -# PROXY_LISTEN_ADDRESS: Literal["127.0.0.1"] = "127.0.0.1" -# PROXY_LISTEN_PORT: Literal[5802] = 5802 -# SDN_CONTROLLER_PORT: Literal[6633] = 6633 +from .tunnel import TUNNEL_EVENTS class BridgeABC: @@ -186,6 +170,7 @@ class OvsBridge(BridgeABC): broker.run_proc([OvsBridge.brctl, "--if-exists", "del-br", self.name]) def add_port(self, port_name, port_descr): + self.del_port(port_name) self.flush_ip_addresses(self.name) with IPRoute() as ipr: idx = ipr.link_lookup(ifname=port_name)[0] @@ -448,14 +433,12 @@ class BridgeController(ControllerModule): def __init__(self, nexus, module_config): super().__init__(nexus, module_config) - # self._bfproxy = None - # self._bfproxy_thread = None self._ovl_net: dict[str, Union[VNIC, LinuxBridge, OvsBridge]] = {} self._appbr: dict[str, Union[LinuxBridge, OvsBridge]] = {} - # self._lock = threading.Lock() self._tunnels: dict[str, TunnelsLog] = {} def initialize(self): + self._bf_proc = None self._register_abort_handlers() self._register_req_handlers() self._register_resp_handlers() @@ -468,17 +451,6 @@ class BridgeController(ControllerModule): # create and configure the bridge for each overlay _ = self._create_overlay_bridges() publishers = self.get_registered_publishers() - if ( - "TincanTunnel" not in publishers - or "TCI_TINCAN_MSG_NOTIFY" - not in self.get_available_subscriptions("TincanTunnel") - ): - raise RuntimeError( - "The TincanTunnel MESSAGE NOTIFY subscription is not available." - "Link Manager cannot continue." - ) - self.start_subscription("TincanTunnel", "TCI_TINCAN_MSG_NOTIFY") - if ( "LinkManager" not in publishers or "LNK_TUNNEL_EVENTS" @@ -510,7 +482,6 @@ class BridgeController(ControllerModule): "GNV_TUNNEL_EVENTS": self.req_handler_manage_bridge, "LNK_TUNNEL_EVENTS": self.req_handler_manage_bridge, "VIS_DATA_REQ": self.req_handler_vis_data, - "TCI_TINCAN_MSG_NOTIFY": self.req_handler_tincan_notify, } def _register_resp_handlers(self): @@ -531,26 +502,6 @@ class BridgeController(ControllerModule): ) bf_config[br_name] = bf_ovls[olid] bf_config[br_name]["OverlayId"] = olid - # while True: - # try: - # self._bfproxy = BoundedFloodProxy( - # bf_config, - # self, - # ) - # self._bfproxy_thread = threading.Thread( - # target=self._bfproxy.serve_forever, name="BFProxyServer" - # ) - # break - # except socket.error as err: - # self.logger.warning( - # "Failed to start the BoundedFlood Proxy, will retry. Error msg= %s", - # err, - # ) - # time.sleep(10) - # self._server_thread.setDaemon(True) - # self._bfproxy_thread.start() - # start the BF RYU module - # self._bfproxy.start_bf_client_module() self.start_bf_client_module(bf_config) def _create_overlay_bridges(self) -> dict: @@ -615,53 +566,12 @@ class BridgeController(ControllerModule): cbt.set_response(None, True) self.complete_cbt(cbt) - def req_handler_tincan_notify(self, cbt: CBT): - if cbt.request.params["Command"] == "ResetTincanTunnels": - sid = cbt.request.params["SessionId"] - for olid, br in self._ovl_net.items(): - self.logger.info("Clearing Tincan TAPs from %s for session %s", br, sid) - for port_name in [*br.ports]: - if ( - br.port_descriptors[port_name]["Dataplane"] - == DATAPLANE_TYPES.Tincan - ): - br.del_port(port_name) - self._tunnels[olid].pop(port_name, None) - self.logger.info( - "Port %s removed from bridge %s", port_name, br - ) - cbt.set_response(data=None, status=True) - self.complete_cbt(cbt) - def on_timer_event(self): for tnl in self._tunnels.values(): tnl.trim() - # def abort_handler(self, cbt: CBT): - # """Additional resouce clean here, eg., fail edge negotiate or create""" - # self.free_cbt(cbt) - - def process_cbt(self, cbt): - if cbt.is_expired: - self.abort_handler(cbt) - elif cbt.is_pending: - if cbt.request.action in ("LNK_TUNNEL_EVENTS", "GNV_TUNNEL_EVENTS"): - self.req_handler_manage_bridge(cbt) - elif cbt.request.action == "VIS_DATA_REQ": - self.req_handler_vis_data(cbt) - elif cbt.request.action == "TCI_TINCAN_MSG_NOTIFY": - self.req_handler_tincan_notify(cbt) - else: - self.req_handler_default(cbt) - elif cbt.is_completed: - self.resp_handler_default(cbt) - def terminate(self): try: - # if self._bfproxy: - # self._bfproxy.stop_bf_module() - # self._bfproxy.server_close() - # self._bfproxy_thread.join() self.stop_bf_module() for olid, bridge in self._ovl_net.items(): if self.overlays[olid]["NetDevice"].get( @@ -759,7 +669,6 @@ class BridgeController(ControllerModule): Status=False, Data=dict(ErrorMsg="Unsupported request") ) msg.data = json.dumps(task).encode("utf-8") - # resp = ProxyMsg(msg.fileno, json.dumps(task).encode("utf-8")) self.send_ipc(msg) def start_bf_client_module(self, bf_config): @@ -776,10 +685,26 @@ class BridgeController(ControllerModule): json.dumps(bf_config), "controllers/bounded_flood.py", ] - self._bf_proc = broker.create_process(cmd) + self._bf_proc = subprocess.Popen(cmd) - def stop_bf_module(self): - if hasattr(self, "_bf_proc"): - if self._bf_proc: - self._bf_proc.send_signal(signal.SIGINT) - self._bf_proc.wait() + def stop_bf_module(self, wt: int = 1.15): + if self._bf_proc is not None: + try: + exit_code = self._bf_proc.poll() + if exit_code is None: + self._bf_proc.terminate() + self._bf_proc.wait() + else: + self.logger.debug( + "BoundedFlood process %s already exited with %s", + self._bf_proc.pid, + exit_code, + ) + except subprocess.TimeoutExpired: + exit_code = self._bf_proc.poll() + if exit_code is None: + self.logger.info( + "Killing unresponsive BoundedFlood: %s", self._bf_proc.pid + ) + self._bf_proc.kill() + self.logger.info("BoundedFlood terminated") diff --git a/evio/controllers/geneve_tunnel.py b/evio/controllers/geneve_tunnel.py index 54312dc..22353fb 100644 --- a/evio/controllers/geneve_tunnel.py +++ b/evio/controllers/geneve_tunnel.py @@ -22,6 +22,7 @@ import time from broker import GENEVE_SETUP_TIMEOUT +from broker.cbt import CBT from broker.controller_module import ControllerModule from broker.remote_action import RemoteAction from pyroute2 import IPRoute @@ -94,7 +95,7 @@ class GeneveTunnel(ControllerModule): } self._gnv_updates_publisher.post_update(param) - def _create_tunnel(self, tap_name, vnid, remote_addr): + def _create_tunnel(self, tap_name: str, vnid: int, remote_addr: str): self.logger.info( "Creating Geneve tunnel %s vnid=%s, remote addr=%s", tap_name, @@ -112,9 +113,9 @@ class GeneveTunnel(ControllerModule): idx = ipr.link_lookup(ifname=tap_name)[0] ipr.link("set", index=idx, state="up") - def _remove_tunnel(self, tap_name): + def _remove_tunnel(self, tap_name: str): try: - self.logger.info("Removing Geneve tunnel %s", tap_name) + self.logger.info("Removing Geneve TAP %s", tap_name) with IPRoute() as ipr: idx = ipr.link_lookup(ifname=tap_name) if len(idx) > 0: @@ -123,23 +124,23 @@ class GeneveTunnel(ControllerModule): ipr.link("del", index=idx) except Exception as e: self.logger.warning( - "Failed to remove geneve tunnel %s, error code: %s", tap_name, e + "Failed to remove Geneve tunnel %s, error code: %s", tap_name, e ) - def _is_tap_exist(self, tap_name): + def _is_tap_exist(self, tap_name: str) -> bool: with IPRoute() as ipr: idx = ipr.link_lookup(ifname=tap_name) if len(idx) == 1: return True return False - def _is_tunnel_authorized(self, tunnel_id): + def _is_tunnel_authorized(self, tunnel_id: str) -> bool: tnl = self._tunnels.get(tunnel_id) if tnl and tnl.state == TUNNEL_STATES.AUTHORIZED: return True return False - def req_handler_auth_tunnel(self, cbt): + def req_handler_auth_tunnel(self, cbt: CBT): """Node B""" olid = cbt.request.params["OverlayId"] peer_id = cbt.request.params["PeerId"] @@ -182,7 +183,7 @@ class GeneveTunnel(ControllerModule): self._gnv_updates_publisher.post_update(event_param) self.complete_cbt(cbt) - def req_handler_create_tunnel(self, cbt): + def req_handler_create_tunnel(self, cbt: CBT): """Role A. Issued from local Topology.""" olid = cbt.request.params["OverlayId"] tnlid = cbt.request.params["TunnelId"] @@ -193,6 +194,7 @@ class GeneveTunnel(ControllerModule): if tnlid in self._tunnels: cbt.set_response({"Message": "Tunnel already exists"}, False) self.complete_cbt(cbt) + return if self._is_tap_exist(tap_name): # delete remenants self._remove_tunnel(tap_name) @@ -221,7 +223,7 @@ class GeneveTunnel(ControllerModule): ) rem_act.submit_remote_act(self, cbt) - def req_handler_exchnge_endpt(self, cbt): + def req_handler_exchnge_endpt(self, cbt: CBT): """ Role B """ @@ -262,7 +264,7 @@ class GeneveTunnel(ControllerModule): cbt.set_response(msg, False) self.complete_cbt(cbt) - def req_handler_update_peer_mac(self, cbt): + def req_handler_update_peer_mac(self, cbt: CBT): """Role B""" params = cbt.request.params olid = params["OverlayId"] @@ -294,7 +296,7 @@ class GeneveTunnel(ControllerModule): cbt.set_response({"Message": "Invalid request for tunnel"}, False) self.complete_cbt(cbt) - def req_handler_cancel_tunnel(self, cbt): + def req_handler_cancel_tunnel(self, cbt: CBT): """ Role B Operation should always succeed. @@ -311,7 +313,7 @@ class GeneveTunnel(ControllerModule): cbt.set_response({"Message": "Tunnel cancelled"}, True) self.complete_cbt(cbt) - def req_handler_remove_tunnel(self, cbt): + def req_handler_remove_tunnel(self, cbt: CBT): """ Issued from local Topology. Operation always succeed. """ @@ -334,7 +336,7 @@ class GeneveTunnel(ControllerModule): } self._gnv_updates_publisher.post_update(gnv_param) - def resp_handler_remote_action(self, cbt): + def resp_handler_remote_action(self, cbt: CBT): """Role A""" parent_cbt = cbt.parent rem_act = cbt.request.params @@ -412,7 +414,7 @@ class GeneveTunnel(ControllerModule): elif rem_act.action == "GNV_CANCEL_TUNNEL": self.free_cbt(cbt) - def abort_handler_remote_action(self, cbt): + def abort_handler_remote_action(self, cbt: CBT): parent_cbt = cbt.parent rem_act = cbt.request.params if rem_act.action == "GNV_EXCHANGE_ENDPT": @@ -429,7 +431,7 @@ class GeneveTunnel(ControllerModule): parent_cbt.set_response(cbt.response.data, False) self.complete_cbt(parent_cbt) - def get_tap_name(self, olid, peer_id) -> str: + def get_tap_name(self, olid, peer_id: str) -> str: tap_name_prefix = self.config["Overlays"][olid].get("TapNamePrefix", olid[:5]) end_i = self.TAPNAME_MAXLEN - len(tap_name_prefix) tap_name = tap_name_prefix + str(peer_id[:end_i]) diff --git a/evio/controllers/graph_builder.py b/evio/controllers/graph_builder.py index 827b3e2..1fe89b0 100644 --- a/evio/controllers/graph_builder.py +++ b/evio/controllers/graph_builder.py @@ -273,6 +273,7 @@ class GraphBuilder: request_list: Optional[list[dict]], relink: bool = False, ) -> GraphTransformation: + self.logger.debug("Using peer list: %s", peers) new_adj_list = self.build_adj_list( peers, initial_adj_list, request_list, relink ) diff --git a/evio/controllers/link_manager.py b/evio/controllers/link_manager.py index 1616d18..76718d4 100644 --- a/evio/controllers/link_manager.py +++ b/evio/controllers/link_manager.py @@ -60,7 +60,6 @@ class Tunnel: peer_id: str, tnl_state, dataplane, - dp_instance_id: int, ): self.tnlid = tnlid self.overlay_id = overlay_id @@ -72,7 +71,6 @@ class Tunnel: self.peer_mac = None self._tunnel_state = tnl_state self.dataplane = dataplane - self.dp_instance_id = dp_instance_id def __repr__(self): return broker.introspect(self) @@ -97,7 +95,6 @@ class LinkManager(ControllerModule): self._lock = threading.Lock() # serializes access to _overlays, _links self._link_updates_publisher = None self._ignored_net_interfaces = dict() - self._tc_session_id: int = 0 def initialize(self): self._register_abort_handlers() @@ -107,14 +104,14 @@ class LinkManager(ControllerModule): publishers = self.get_registered_publishers() if ( "TincanTunnel" not in publishers - or "TCI_TINCAN_MSG_NOTIFY" + or "TCI_TUNNEL_EVENT" not in self.get_available_subscriptions("TincanTunnel") ): raise RuntimeError( "The TincanTunnel MESSAGE NOTIFY subscription is not available." "Link Manager cannot continue." ) - self.start_subscription("TincanTunnel", "TCI_TINCAN_MSG_NOTIFY") + self.start_subscription("TincanTunnel", "TCI_TUNNEL_EVENT") if ( "OverlayVisualizer" in publishers and "VIS_DATA_REQ" in self.get_available_subscriptions("OverlayVisualizer") @@ -132,15 +129,6 @@ class LinkManager(ControllerModule): self.logger.info("Controller module loaded") - @property - def tc_session_id(self): - return self._tc_session_id - - @tc_session_id.setter - def tc_session_id(self, val: int): - self.logger.info("Updating Tincan session ID %s->%s", self.tc_session_id, val) - self._tc_session_id = val - def terminate(self): self.logger.info("Controller module terminating") @@ -172,7 +160,6 @@ class LinkManager(ControllerModule): peer_id, tnl_state=TUNNEL_STATES.AUTHORIZED, dataplane=DATAPLANE_TYPES.Tincan, - dp_instance_id=self.tc_session_id, ) self._tunnels[tnlid] = tnl self.register_timed_transaction( @@ -260,7 +247,6 @@ class LinkManager(ControllerModule): peer_id, tnl_state=TUNNEL_STATES.CREATING, dataplane=DATAPLANE_TYPES.Tincan, - dp_instance_id=self.tc_session_id, ) self._assign_link_to_tunnel(tnlid, lnkid, 0xA1) @@ -304,7 +290,6 @@ class LinkManager(ControllerModule): return lnkid = tnlid self._tunnels[tnlid].tunnel_state = TUNNEL_STATES.CREATING - self._tunnels[tnlid].dp_instance_id = self.tc_session_id self._assign_link_to_tunnel(tnlid, lnkid, 0xB1) self.logger.debug( "Creating link %s to peer %s (1/4 Target)", lnkid[:7], peer_id[:7] @@ -330,7 +315,6 @@ class LinkManager(ControllerModule): "MAC": node_data["MAC"], "UID": node_data["UID"], }, - "TincanId": self.tc_session_id, } if self.config.get("Turn"): create_link_params["TurnServers"] = self.config["Turn"] @@ -356,61 +340,69 @@ class LinkManager(ControllerModule): self.logger.debug( "Creating link %s to peer %s (3/4 Target)", lnkid[:7], peer_id[:7] ) - params["TincanId"] = self._tunnels[tnlid].dp_instance_id self.register_cbt("TincanTunnel", "TCI_CREATE_LINK", params, cbt) def req_handler_tincan_msg(self, cbt: CBT): lts = time.time() - if cbt.request.params["Command"] == "LinkStateChange": + tnlid = cbt.request.params["TunnelId"] + if tnlid not in self._tunnels: + cbt.set_response(data=None, status=True) + self.complete_cbt(cbt) + return + elif cbt.request.params["Command"] == "LinkConnected": lnkid = cbt.request.params["LinkId"] - tnlid = cbt.request.params["TunnelId"] - if tnlid not in self._tunnels: - return - if (cbt.request.params["Data"] == "LINK_STATE_DOWN") and ( - self._tunnels[tnlid].tunnel_state != TUNNEL_STATES.QUERYING - ): - self.logger.debug("Link %s is DOWN cbt=%s", tnlid, cbt) + self.logger.debug("Link %s is connected", lnkid) + olid = self._tunnels[tnlid].overlay_id + peer_id = self._tunnels[tnlid].peer_id + lnk_status = self._tunnels[tnlid].tunnel_state + self._tunnels[tnlid].tunnel_state = TUNNEL_STATES.ONLINE + if lnk_status != TUNNEL_STATES.QUERYING: + param = { + "UpdateType": TUNNEL_EVENTS.Connected, + "OverlayId": olid, + "PeerId": peer_id, + "TunnelId": tnlid, + "LinkId": lnkid, + "ConnectedTimestamp": lts, + "TapName": self._tunnels[tnlid].tap_name, + "MAC": self._tunnels[tnlid].mac, + "PeerMac": self._tunnels[tnlid].peer_mac, + "Dataplane": self._tunnels[tnlid].dataplane, + } + self._link_updates_publisher.post_update(param) + elif lnk_status == TUNNEL_STATES.QUERYING: + # Do not post a notification if the the connection state was being queried + self._tunnels[tnlid].link.status_retry = 0 + elif cbt.request.params["Command"] == "LinkDisconnected": + if self._tunnels[tnlid].tunnel_state != TUNNEL_STATES.QUERYING: + self.logger.debug("Link %s is disconnected", tnlid) # issue a link state check only if it not already being done self._tunnels[tnlid].tunnel_state = TUNNEL_STATES.QUERYING cbt.set_response(data=None, status=True) - self.register_cbt("TincanTunnel", "TCI_QUERY_LINK_STATS", [tnlid]) - elif cbt.request.params["Data"] == "LINK_STATE_UP": - tnlid = self.tunnel_id(lnkid) - olid = self._tunnels[tnlid].overlay_id - peer_id = self._tunnels[tnlid].peer_id - lnk_status = self._tunnels[tnlid].tunnel_state - self._tunnels[tnlid].tunnel_state = TUNNEL_STATES.ONLINE - if lnk_status != TUNNEL_STATES.QUERYING: - param = { - "UpdateType": TUNNEL_EVENTS.Connected, - "OverlayId": olid, - "PeerId": peer_id, - "TunnelId": tnlid, - "LinkId": lnkid, - "ConnectedTimestamp": lts, - "TapName": self._tunnels[tnlid].tap_name, - "MAC": self._tunnels[tnlid].mac, - "PeerMac": self._tunnels[tnlid].peer_mac, - "Dataplane": self._tunnels[tnlid].dataplane, - } - self._link_updates_publisher.post_update(param) - elif lnk_status == TUNNEL_STATES.QUERYING: - # Do not post a notification if the the connection state was being queried - self._tunnels[tnlid].link.status_retry = 0 - cbt.set_response(data=None, status=True) - elif cbt.request.params["Command"] == "TincanReady": - self.tc_session_id = cbt.request.params.get("SessionId", self.tc_session_id) - cbt.set_response(data=None, status=True) - elif cbt.request.params["Command"] == "ResetTincanTunnels": - self.logger.info( - "Clearing Tincan tunnels for session %s", self.tc_session_id - ) - self._tunnels.clear() - self._links.clear() - self.tc_session_id = 0 - cbt.set_response(data=None, status=True) + self.register_cbt( + "TincanTunnel", "TCI_QUERY_LINK_INFO", {"TunnelId": tnlid} + ) + elif cbt.request.params["Command"] == "TincanTunnelFailed": + lnkid = self.link_id(tnlid) + if lnkid: + self._links.pop(lnkid, None) + tnl = self._tunnels.pop(tnlid) + tnl.tunnel_state = TUNNEL_STATES.FAILED + param = { + "UpdateType": TUNNEL_EVENTS.Removed, + "OverlayId": tnl.overlay_id, + "PeerId": tnl.peer_id, + "TunnelId": tnlid, + "LinkId": lnkid, + "TapName": tnl.tap_name, + } + self._link_updates_publisher.post_update(param) else: - cbt.set_response(data=None, status=True) + self.logger.warning( + "Unexpected Tincan event command received %s", + cbt.request.params["Command"], + ) + cbt.set_response(data=None, status=True) self.complete_cbt(cbt) def req_handler_query_tunnels_info(self, cbt: CBT): @@ -449,7 +441,6 @@ class LinkManager(ControllerModule): "TunnelId": tnlid, "PeerId": peer_id, "TapName": tn, - "TincanId": self.tc_session_id, } self.register_cbt("TincanTunnel", "TCI_REMOVE_TUNNEL", params, cbt) else: @@ -526,8 +517,6 @@ class LinkManager(ControllerModule): if parent_cbt: parent_cbt.set_response(resp_data, False) self.complete_cbt(parent_cbt) - if resp_data and "CurrentId" in resp_data: - self.tc_session_id = resp_data["CurrentId"] return if parent_cbt.request.action == "LNK_REQ_LINK_ENDPT": @@ -595,8 +584,6 @@ class LinkManager(ControllerModule): "The create tunnel operation failed: %s or the parent expired", resp_data, ) - if resp_data and "CurrentId" in resp_data: - self.tc_session_id = resp_data["CurrentId"] return # transistion connection connection state self._tunnels[tnlid].link.creation_state = 0xA2 @@ -622,9 +609,6 @@ class LinkManager(ControllerModule): peer_id = rmv_tnl_cbt.request.params["PeerId"] olid = rmv_tnl_cbt.request.params["OverlayId"] tap_name = rmv_tnl_cbt.request.params["TapName"] - resp_data = rmv_tnl_cbt.response.data - if resp_data and "CurrentId" in resp_data: - self.tc_session_id = resp_data["CurrentId"] self._tunnels.pop(tnlid, None) self._links.pop(lnkid, None) self.free_cbt(rmv_tnl_cbt) @@ -654,67 +638,42 @@ class LinkManager(ControllerModule): if not cbt.response.status: self.logger.warning("Link stats update error: %s", cbt.response.data) self.free_cbt(cbt) - if resp_data and "CurrentId" in resp_data: - self.tc_session_id = resp_data["CurrentId"] return # Handle any connection failures and update tracking data - for tnlid in resp_data: - for lnkid in resp_data[tnlid]: - if resp_data[tnlid][lnkid]["Status"] == "UNKNOWN": - self._tunnels.pop(tnlid, None) - elif tnlid in self._tunnels: - tnl = self._tunnels[tnlid] - if resp_data[tnlid][lnkid]["Status"] == "OFFLINE": - # tincan indicates offline so recheck the link status - retry = tnl.link.status_retry - if retry >= 2 and tnl.tunnel_state == TUNNEL_STATES.CREATING: - # link is stuck creating so destroy it - olid = tnl.overlay_id - peer_id = tnl.peer_id - params = { - "OverlayId": olid, - "TunnelId": tnlid, - "LinkId": lnkid, - "PeerId": peer_id, - "TapName": tnl.tap_name, - "TincanId": self.tc_session_id, - } - self.register_cbt( - "TincanTunnel", "TCI_REMOVE_TUNNEL", params - ) - elif (tnl.tunnel_state == TUNNEL_STATES.QUERYING) or ( - retry >= 1 and tnl.tunnel_state == TUNNEL_STATES.ONLINE - ): - # LINK_STATE_DOWN event or QUERY_LNK_STATUS response - post notify - tnl.tunnel_state = TUNNEL_STATES.OFFLINE - olid = tnl.overlay_id - peer_id = tnl.peer_id - param = { - "UpdateType": TUNNEL_EVENTS.Disconnected, - "OverlayId": olid, - "PeerId": peer_id, - "TunnelId": tnlid, - "LinkId": lnkid, - "TapName": tnl.tap_name, - } - self._link_updates_publisher.post_update(param) - else: - self.logger.warning( - "Link %s is offline, no further attempts to to query its stats will" - "be made.", - tnlid, - ) - elif resp_data[tnlid][lnkid]["Status"] == "ONLINE": - tnl.tunnel_state = TUNNEL_STATES.ONLINE - tnl.link.stats = resp_data[tnlid][lnkid]["Stats"] - tnl.link.status_retry = 0 - else: - self.logger.warning( - "Unrecognized tunnel state ", - "%s:%s", - lnkid, - resp_data[tnlid][lnkid]["Status"], - ) + tnlid = resp_data["TunnelId"] + lnkid = resp_data["LinkId"] + if tnlid in self._tunnels: + tnl = self._tunnels[tnlid] + if resp_data["Status"] == "OFFLINE": + # tincan indicates offline so recheck the link status + retry = tnl.link.status_retry + if (tnl.tunnel_state == TUNNEL_STATES.QUERYING) or ( + retry >= 1 and tnl.tunnel_state == TUNNEL_STATES.ONLINE + ): + # LINK_STATE_DOWN event or QUERY_LNK_STATUS response - post notify + tnl.tunnel_state = TUNNEL_STATES.OFFLINE + olid = tnl.overlay_id + peer_id = tnl.peer_id + param = { + "UpdateType": TUNNEL_EVENTS.Disconnected, + "OverlayId": olid, + "PeerId": peer_id, + "TunnelId": tnlid, + "LinkId": lnkid, + "TapName": tnl.tap_name, + } + self._link_updates_publisher.post_update(param) + elif resp_data["Status"] == "ONLINE": + tnl.tunnel_state = TUNNEL_STATES.ONLINE + tnl.link.stats = resp_data["Stats"] + tnl.link.status_retry = 0 + else: + self.logger.warning( + "Unrecognized tunnel state ", + "%s:%s", + lnkid, + resp_data["Status"], + ) self.free_cbt(cbt) def on_tnl_timeout(self, tnl: Tunnel, timeout: float): @@ -726,7 +685,7 @@ class LinkManager(ControllerModule): "TCI_CREATE_LINK": self.abort_handler_tunnel, "TCI_CREATE_TUNNEL": self.abort_handler_tunnel, "TCI_REMOVE_TUNNEL": self.abort_handler_tunnel, - "TCI_QUERY_LINK_STATS": self.abort_handler_default, + "TCI_QUERY_LINK_INFO": self.abort_handler_default, "TCI_REMOVE_LINK": self.abort_handler_default, "LNK_TUNNEL_EVENTS": self.abort_handler_default, } @@ -739,7 +698,7 @@ class LinkManager(ControllerModule): "LNK_REMOVE_TUNNEL": self.req_handler_remove_tnl, "LNK_QUERY_TUNNEL_INFO": self.req_handler_query_tunnels_info, "VIS_DATA_REQ": self.req_handler_query_viz_data, - "TCI_TINCAN_MSG_NOTIFY": self.req_handler_tincan_msg, + "TCI_TUNNEL_EVENT": self.req_handler_tincan_msg, "LNK_ADD_IGN_INF": self.req_handler_add_ign_inf, "LNK_AUTH_TUNNEL": self.req_handler_auth_tunnel, } @@ -749,7 +708,7 @@ class LinkManager(ControllerModule): "SIG_REMOTE_ACTION": self.resp_handler_remote_action, "TCI_CREATE_LINK": self.resp_handler_create_link_endpt, "TCI_CREATE_TUNNEL": self.resp_handler_create_tunnel, - "TCI_QUERY_LINK_STATS": self.resp_handler_query_link_stats, + "TCI_QUERY_LINK_INFO": self.resp_handler_query_link_stats, "TCI_REMOVE_TUNNEL": self.resp_handler_remove_tunnel, } @@ -782,16 +741,6 @@ class LinkManager(ControllerModule): def is_link_completed(self, tnl: Tunnel) -> bool: return bool(tnl.link and tnl.link.creation_state == 0xC0) - def _query_link_stats(self): - """Query the status of links that have completed creation process""" - params = [] - for tnlid in self._tunnels: - link = self._tunnels[tnlid].link - if link and link.creation_state == 0xC0: - params.append(tnlid) - if params: - self.register_cbt("TincanTunnel", "TCI_QUERY_LINK_STATS", params) - def _remove_link_from_tunnel(self, tnlid): tnl = self._tunnels.get(tnlid) if tnl: @@ -834,7 +783,6 @@ class LinkManager(ControllerModule): "IgnoredNetInterfaces": list( self._get_ignored_tap_names(overlay_id, tap_name) ), - "TincanId": self.tc_session_id, } if self.config.get("Turn"): create_tnl_params["TurnServers"] = self.config["Turn"] @@ -894,7 +842,6 @@ class LinkManager(ControllerModule): "CAS": node_data["CAS"], "FPR": node_data["FPR"], }, - "TincanId": self._tunnels[tnlid].dp_instance_id, } self.register_cbt("TincanTunnel", "TCI_CREATE_LINK", cbt_params, parent_cbt) @@ -997,9 +944,8 @@ class LinkManager(ControllerModule): self.free_cbt(cbt) self.complete_cbt(parent_cbt) self.logger.info( - "Tunnel %s Link %s accepted: %s:%s<-%s", + "Tunnel %s accepted: %s:%s<-%s", tnlid[:7], - lnkid[:7], olid[:7], self.node_id[:7], peer_id[:7], @@ -1073,7 +1019,6 @@ class LinkManager(ControllerModule): "TunnelId": tnlid, "LinkId": lnkid, "TapName": tnl.tap_name, - "TincanId": self.tc_session_id, } self.logger.info( "Initiating removal of incomplete tunnnel: " @@ -1093,7 +1038,7 @@ class LinkManager(ControllerModule): self._links.pop(lnkid, None) -""" +""" TODO: OUTDATED, NEED TO BE UPDATED ################################################################################################### Link Manager state and event specifications ################################################################################################### @@ -1120,10 +1065,10 @@ tunnel descriptor is removed. Tunnel must be in TUNNEL_STATES.ONLINE or TUNNEL_S (1) TUNNEL_STATES.AUTHORIZED - After a successful completion of CBT LNK_AUTH_TUNNEL, the tunnel descriptor exists. (2) TUNNEL_STATES.CREATING - entered on reception of CBT LNK_CREATE_TUNNEL. -(3) TUNNEL_STATES.QUERYING - entered before issuing CBT TCI_QUERY_LINK_STATS. Happens when +(3) TUNNEL_STATES.QUERYING - entered before issuing CBT TCI_QUERY_LINK_INFO. Happens when LinkStateChange is LINK_STATE_DOWN and state is not already TUNNEL_STATES.QUERYING; OR -TCI_QUERY_LINK_STATS is OFFLINE and state is not already TUNNEL_STATES.QUERYING. -(4) TUNNEL_STATES.ONLINE - entered when CBT TCI_QUERY_LINK_STATS is ONLINE or LinkStateChange is +TCI_QUERY_LINK_INFO is OFFLINE and state is not already TUNNEL_STATES.QUERYING. +(4) TUNNEL_STATES.ONLINE - entered when CBT TCI_QUERY_LINK_INFO is ONLINE or LinkStateChange is LINK_STATE_UP. (5) TUNNEL_STATES.OFFLINE - entered when QUERY_LNK_STATUS is OFFLINE or LinkStateChange is LINK_STATE_DOWN event. diff --git a/evio/controllers/network_graph.py b/evio/controllers/network_graph.py index 9e315b0..1608a08 100644 --- a/evio/controllers/network_graph.py +++ b/evio/controllers/network_graph.py @@ -370,6 +370,11 @@ class ConnEdgeAdjacenctList(MutableMapping): for peer_id in rml: self.remove_conn_edge(peer_id) + def remove_edge_by_id(self, edge_id: str): + for ce in self._conn_edges.values(): + if ce.edge_id == edge_id: + self.remove_conn_edge(ce.peer_id) + class GraphEdit: def __init__(self, conn_edge: ConnectionEdge, op_type: str, priority: int): diff --git a/evio/controllers/signal.py b/evio/controllers/signal.py index 045007a..1a2b285 100644 --- a/evio/controllers/signal.py +++ b/evio/controllers/signal.py @@ -112,7 +112,6 @@ class XmppTransport(slixmpp.ClientXMPP): ): # param for coressponding XmppCircle slixmpp.ClientXMPP.__init__(self, jid, password, sasl_mech=sasl_mech) self._overlay_id = None - # self._sig: Signal = None self._node_id = None self.logger = None self.on_presence = None @@ -421,10 +420,10 @@ class XmppTransport(slixmpp.ClientXMPP): ): self.logger.debug("Initiating shutdown of XMPP overlay=%s", self._overlay_id) self.loop.call_soon_threadsafe(self.disconnect(reason="controller shutdown")) + self.logger.debug("Disconnect of XMPP overlay=%s", self._overlay_id) class XmppCircle: - # _REFLECT: list[str] = [ ] def __init__( self, node_id: str, overlay_id: str, ovl_config: dict, **kwargs ) -> None: @@ -442,7 +441,6 @@ class XmppCircle: self.xport: XmppTransport = None self._xport_thread = threading.Thread( target=self._setup_transport_instance, - # kwargs={"overlay_id": overlay_id}, daemon=True, name="XMPP.Client", ) @@ -571,9 +569,7 @@ class Signal(ControllerModule): "A mis-delivered remote action was discarded: %s", rem_act ) return - n_cbt = self.create_cbt( - self.name, rem_act.recipient_cm, rem_act.action, rem_act.params - ) + n_cbt = self.create_cbt(rem_act.recipient_cm, rem_act.action, rem_act.params) # store the remote action for completion with self._lck: self._recv_remote_acts_invk_locally[n_cbt.tag] = rem_act diff --git a/evio/controllers/tincan_tunnel.py b/evio/controllers/tincan_tunnel.py index 07a72f4..3df5eaf 100644 --- a/evio/controllers/tincan_tunnel.py +++ b/evio/controllers/tincan_tunnel.py @@ -24,8 +24,8 @@ try: except ImportError: import json -import signal import subprocess +import time from threading import Event import broker @@ -34,85 +34,130 @@ from broker.cbt import CBT from broker.controller_module import ControllerModule from broker.process_proxy import ProxyMsg from broker.version import EVIO_VER_CTL +from pyroute2 import IPRoute -# from pyroute2 import IPRoute + +class TincanProcess: + def __init__( + self, + olid: str = "", + tnlid: str = "", + ipc_id: int = -1, + proc: subprocess = None, + do_chk: bool = False, + echo_replies: int = broker.MAX_HEARTBEATS, + ): + self.ovlid = olid + self.tnlid = tnlid + self.ipc_id = ipc_id + self.echo_replies = echo_replies + self.do_chk = do_chk + self.proc = proc + self.tap_name: str = "" + + def __repr__(self) -> str: + broker.introspect() class TincanTunnel(ControllerModule): def __init__(self, nexus, module_config): super().__init__(nexus, module_config) - self._echo_replies: int = 2 self.exit_ev = Event() self._tci_publisher = None - self._tc_pid = -1 - self._tc_proc = None - self._tunnel_pid: dict[str, int] = {} - self._ipc_id: int = -1 - self._use_log_defaults = False + self._tc_proc_tbl: dict[str, TincanProcess] = {} + self._pids: dict[int, str] = {} + self._kill_times: list[float] = [ + 0.0, + ] + self._tnl_cbts: dict[str, CBT] = {} def initialize(self): self._register_abort_handlers() self._register_req_handlers() self._register_resp_handlers() - self._tci_publisher = self.publish_subscription("TCI_TINCAN_MSG_NOTIFY") - self._start_tincan() + self._tci_publisher = self.publish_subscription("TCI_TUNNEL_EVENT") + self.on_expire_chk_tincan() self.logger.info("Controller module loaded") def _register_abort_handlers(self): self._abort_handler_tbl = { - "TCI_CONFIGURE_LOGGING": self.abort_handler_configure_tincan_logging, "_TCI_SEND_ECHO": self.abort_handler_send_echo, - "TCI_TINCAN_MSG_NOTIFY": self.abort_handler_default, + "TCI_TUNNEL_EVENT": self.abort_handler_default, } def _register_req_handlers(self): self._req_handler_tbl = { - "TCI_CREATE_LINK": self.req_handler_create_link, - "TCI_REMOVE_LINK": self.req_handler_remove_link, "TCI_CREATE_TUNNEL": self.req_handler_create_tunnel, - "TCI_QUERY_CAS": self.req_handler_query_candidate_address_set, - "TCI_QUERY_LINK_STATS": self.req_handler_query_link_stats, - "TCI_QUERY_TUNNEL_INFO": self.req_handler_query_tunnel_info, + "TCI_CREATE_LINK": self.req_handler_create_link, + "TCI_QUERY_LINK_INFO": self.req_handler_query_link_stats, + "TCI_REMOVE_LINK": self.req_handler_remove_link, "TCI_REMOVE_TUNNEL": self.req_handler_remove_tunnel, - "_TCI_SEND_ECHO": self.req_handler_send_send_echo, - "TCI_CONFIGURE_LOGGING": self.req_handler_configure_tincan_logging, + "_TCI_SEND_ECHO": self.req_handler_send_echo, } def _register_resp_handlers(self): self._resp_handler_tbl = { - "TCI_CONFIGURE_LOGGING": self.resp_handler_configure_tincan_logging, "_TCI_SEND_ECHO": self.resp_handler_send_echo, } - def req_handler_configure_tincan_logging(self, cbt: CBT): - ctl = broker.CTL_CONFIGURE_LOGGING - ctl["TransactionId"] = cbt.tag - if cbt.request.params and not self._use_log_defaults: - ctl["Request"].update(cbt.request.params) - self.send_control(json.dumps(ctl)) + def req_handler_create_tunnel(self, cbt: CBT): + try: + msg = cbt.request.params + olid = msg["OverlayId"] + tnlid = msg["TunnelId"] + if tnlid in self._tc_proc_tbl: + cbt.set_response({"Message": "Tunnel already exists"}, False) + self.complete_cbt(cbt) + return + if self._is_tap_exist(msg["TapName"]): + self._remove_tap(msg["TapName"]) + cbt.next_op = self._create_tunnel + self._tnl_cbts[tnlid] = cbt + self._start_tincan(tnlid) + self._tc_proc_tbl[tnlid].ovlid = olid + self._tc_proc_tbl[tnlid].tap_name = msg["TapName"] + except Exception: + self._tnl_cbts.pop(tnlid) + cbt.set_response("Failed to create Tincan tunnel process", False) + self.complete_cbt(cbt) - def resp_handler_configure_tincan_logging(self, cbt: CBT): - status = cbt.response.status - self.free_cbt(cbt) - if status == "False": - self.logger.warning("Failed to configure Tincan logging: CBT=%s", cbt) - self._use_log_defaults = True - self._restart_tincan() - return - self._notify_tincan_ready() - self.on_exp_chk_tincan() + def _create_tunnel(self, cbt: CBT): + msg = cbt.request.params + tnlid = msg["TunnelId"] + ctl = broker.CTL_CREATE_TUNNEL + ctl["TransactionId"] = cbt.tag + req = ctl["Request"] + req["StunServers"] = msg["StunServers"] + req["TurnServers"] = msg.get("TurnServers") + req["TapName"] = msg["TapName"] + req["TunnelId"] = tnlid + req["NodeId"] = msg.get("NodeId") + req["IgnoredNetInterfaces"] = msg.get("IgnoredNetInterfaces") + tc_proc = self._tc_proc_tbl[tnlid] + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) def req_handler_create_link(self, cbt: CBT): - if not self._is_request_current(cbt): # also sets the response to failed + try: + msg = cbt.request.params + tnlid = msg["TunnelId"] + if tnlid not in self._tc_proc_tbl: + cbt.next_op = self._create_link + self._tnl_cbts[tnlid] = cbt + self._start_tincan(tnlid) + else: + self._create_link(cbt) + except Exception: + self._tnl_cbts.pop(tnlid) + cbt.set_response("Failed to create Tincan tunnel process", False) self.complete_cbt(cbt) - return + + def _create_link(self, cbt: CBT): msg = cbt.request.params - self._tunnel_pid[msg["TunnelId"]] = self._tc_pid + tnlid = msg["TunnelId"] ctl = broker.CTL_CREATE_LINK ctl["TransactionId"] = cbt.tag req = ctl["Request"] - req["OverlayId"] = msg["OverlayId"] - req["TunnelId"] = msg["TunnelId"] + req["TunnelId"] = tnlid req["NodeId"] = msg.get("NodeId") req["LinkId"] = msg["LinkId"] req["PeerInfo"]["UID"] = msg["NodeData"].get("UID") @@ -124,262 +169,230 @@ class TincanTunnel(ControllerModule): req["TurnServers"] = msg.get("TurnServers") req["TapName"] = msg.get("TapName") req["IgnoredNetInterfaces"] = msg.get("IgnoredNetInterfaces") - self.send_control(json.dumps(ctl)) - - def req_handler_create_tunnel(self, cbt: CBT): - if not self._is_request_current(cbt): - self.complete_cbt(cbt) - return - msg = cbt.request.params - self._tunnel_pid[msg["TunnelId"]] = self._tc_pid - ctl = broker.CTL_CREATE_TUNNEL - ctl["TransactionId"] = cbt.tag - req = ctl["Request"] - req["StunServers"] = msg["StunServers"] - req["TurnServers"] = msg.get("TurnServers") - req["TapName"] = msg["TapName"] - req["OverlayId"] = msg["OverlayId"] - req["TunnelId"] = msg["TunnelId"] - req["NodeId"] = msg.get("NodeId") - req["IgnoredNetInterfaces"] = msg.get("IgnoredNetInterfaces") - self.send_control(json.dumps(ctl)) + tc_proc = self._tc_proc_tbl[tnlid] + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) def req_handler_query_candidate_address_set(self, cbt: CBT): - if not self._is_request_current(cbt): - self.complete_cbt(cbt) - return msg = cbt.request.params + tnlid = msg["TunnelId"] + if tnlid not in self._tc_proc_tbl: + err_msg = f"No tunnel exists for tunnel ID: {tnlid[:7]}" + cbt.set_response({"ErrorMsg": err_msg, "Status": False}) + return ctl = broker.CTL_QUERY_CAS ctl["TransactionId"] = cbt.tag - req = ctl["Request"] - req["OverlayId"] = msg["OverlayId"] - req["LinkId"] = msg["LinkId"] - self.send_control(json.dumps(ctl)) + ctl["Request"]["TunnelId"] = tnlid + tc_proc = self._tc_proc_tbl[tnlid] + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) def req_handler_query_link_stats(self, cbt: CBT): - # if not self._is_request_current(cbt): - # Todo: TypeError - list indices must be integers or slices, not str - # self.complete_cbt(cbt) - # return msg = cbt.request.params + tnlid = msg["TunnelId"] + if tnlid not in self._tc_proc_tbl: + err_msg = f"No tunnel exists for tunnel ID: {tnlid[:7]}" + cbt.set_response({"ErrorMsg": err_msg, "Status": False}) + self.complete_cbt(cbt) + return ctl = broker.CTL_QUERY_LINK_STATS ctl["TransactionId"] = cbt.tag - req = ctl["Request"] - req["TunnelIds"] = msg - self.send_control(json.dumps(ctl)) - - def req_handler_query_tunnel_info(self, cbt: CBT): - if not self._is_request_current(cbt): - self.complete_cbt(cbt) - return - msg = cbt.request.params - ctl = broker.CTL_QUERY_TUNNEL_INFO - ctl["TransactionId"] = cbt.tag - req = ctl["Request"] - req["OverlayId"] = msg["OverlayId"] - self.send_control(json.dumps(ctl)) + ctl["Request"]["TunnelId"] = tnlid + tc_proc = self._tc_proc_tbl[tnlid] + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) def req_handler_remove_tunnel(self, cbt: CBT): - if not self._is_request_current(cbt): + msg = cbt.request.params + tnlid = msg["TunnelId"] + if tnlid not in self._tc_proc_tbl: + err_msg = f"No tunnel exists for tunnel ID: {tnlid[:7]}" + cbt.set_response(err_msg, True) self.complete_cbt(cbt) return - msg = cbt.request.params - ctl = broker.CTL_REMOVE_TUNNEL - ctl["TransactionId"] = cbt.tag - req = ctl["Request"] - req["OverlayId"] = msg["OverlayId"] - req["TunnelId"] = msg["TunnelId"] - self.send_control(json.dumps(ctl)) - # if "TapName" in msg and msg["TapName"]: - # try: - # with IPRoute() as ipr: - # idx = ipr.link_lookup(ifname=msg["TapName"]) - # if len(idx) > 0: - # idx = idx[0] - # ipr.link("set", index=idx, state="down") - # ipr.link("del", index=idx) - # except Exception: - # pass + self.logger.debug("Removing tunnel %s", tnlid) + tc_proc = self._tc_proc_tbl.pop(tnlid, None) + self._stop_tincan(tc_proc) + cbt.set_response("Tunnel removed", True) + self.complete_cbt(cbt) def req_handler_remove_link(self, cbt: CBT): - if not self._is_request_current(cbt): + msg = cbt.request.params + tnlid = msg["TunnelId"] + if tnlid not in self._tc_proc_tbl: + err_msg = f"No tunnel exists for tunnel ID: {tnlid[:7]}" + cbt.set_response({"ErrorMsg": err_msg, "Status": False}) self.complete_cbt(cbt) return - msg = cbt.request.params ctl = broker.CTL_REMOVE_LINK ctl["TransactionId"] = cbt.tag req = ctl["Request"] - req["OverlayId"] = msg["OverlayId"] - req["TunnelId"] = msg["TunnelId"] + req["TunnelId"] = tnlid req["LinkId"] = msg["LinkId"] - self.send_control(json.dumps(ctl)) + tc_proc = self._tc_proc_tbl[tnlid] + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) - def req_handler_send_send_echo(self, cbt: CBT): + def req_handler_send_echo(self, cbt: CBT): ctl = broker.CTL_ECHO ctl["TransactionId"] = cbt.tag - if cbt.request.params: - ctl["Request"]["Message"] = cbt.request.params - self._echo_replies -= 1 - self.send_control(json.dumps(ctl)) + tnlid = cbt.request.params + tc_proc = self._tc_proc_tbl.get(tnlid) + if tc_proc and tc_proc.do_chk and tc_proc.echo_replies > 0: + tc_proc.echo_replies -= 1 + ctl["Request"]["Message"] = tc_proc.tnlid + self.send_control(tc_proc.ipc_id, json.dumps(ctl)) def resp_handler_send_echo(self, cbt: CBT): - self._echo_replies += 1 + tnlid = cbt.response.data + if tnlid in self._tc_proc_tbl: + self._tc_proc_tbl[tnlid].echo_replies = broker.MAX_HEARTBEATS self.free_cbt(cbt) - def abort_handler_configure_tincan_logging(self, cbt): + def abort_handler_send_echo(self, cbt: CBT): + tnlid = cbt.request.params self.free_cbt(cbt) - self.logger.warning("Configure Tincan logging timeout: CBT=%s", cbt) - self._restart_tincan() - - def abort_handler_send_echo(self, cbt): - self.free_cbt(cbt) - self.logger.warning("Echo test timeout") + if tnlid in self._tc_proc_tbl: + tc_proc = self._tc_proc_tbl[tnlid] + tc_proc.echo_replies -= 1 + if tc_proc.echo_replies > 0: + self.logger.debug( + "Tunnel: %s health check timeout, countdown: %s", + tnlid, + tc_proc.echo_replies, + ) + else: + # tincan process unresponsive + self.logger.warning( + "unnel: %s health check failed, terminating process: %s", + tnlid, + tc_proc, + ) + self._stop_tincan(tc_proc) + self._notify_tincan_terminated(tnlid) + self._tc_proc_tbl.pop(tnlid, None) def on_timer_event(self): - if self._echo_replies > 0: - self.register_internal_cbt("_TCI_SEND_ECHO", "Tincan liveliness check") + if self.exit_ev.is_set(): + return + # send an echo health check every timer interval, eg., 30s + for tnlid, tc_proc in self._tc_proc_tbl.items(): + if tc_proc.do_chk: + self.register_internal_cbt("_TCI_SEND_ECHO", tnlid) - def on_exp_chk_tincan(self, *_): + def on_expire_chk_tincan(self, *_): + # runs 5 secs after posting if self.exit_ev.is_set(): return exit_code = None - if self._tc_proc: - exit_code = self._tc_proc.poll() - if exit_code: - if self._tc_pid == -1: - self.logger.error( - "Tincan process has not called back to register the communication endpoint" - ) - self._restart_tincan() - elif self._tc_pid > 0: + rmv = [] + for tnlid, tc_proc in self._tc_proc_tbl.items(): + exit_code = tc_proc.proc.poll() + if exit_code: # tincan process crashed - self.logger.warning("Tincan process exited with code, %s", exit_code) - self._notify_tincan_terminated(self._tc_pid) - self._start_tincan() - elif self._echo_replies <= 0: - # tincan process unresponsive - self.logger.warning("No replies from Tincan echo check, resetting ...") - self._restart_tincan() + rmv.append(tnlid) + for tnlid in rmv: + self.logger.warning( + "Tincan process %s exited unexpectedly with code, %s", + tc_proc.proc.pid, + exit_code, + ) + self._notify_tincan_terminated(tnlid) + self._tc_proc_tbl.pop(tnlid, None) self.register_timed_transaction( self, statement_false, - self.on_exp_chk_tincan, + self.on_expire_chk_tincan, TINCAN_CHK_INTERVAL, ) def terminate(self): self.exit_ev.set() - self._stop_tincan() + for tc_proc in self._tc_proc_tbl.values(): + self._stop_tincan(tc_proc, wt=1.5) + self.logger.debug("avg tok = %s", self._kill_times[-1] / len(self._kill_times)) self.logger.info("Controller module terminating") - def send_control(self, ctl: str): - msg: ProxyMsg = ProxyMsg(self._ipc_id, payload=ctl.encode("utf-8")) + def send_control(self, ipc_id: int, ctl: str): + msg: ProxyMsg = ProxyMsg(ipc_id, payload=ctl.encode("utf-8")) + # self.logger.debug("Sending dataplane control %s", msg) self.send_ipc(msg) - def _start_tincan(self): + def _start_tincan(self, tnlid: str): if self.exit_ev.is_set(): return - # self.logger.info( - # "start Tincan with ./tincan -s %s", - # self.process_proxy_address[1:].decode("utf-8"), - # ) - self._tc_proc = subprocess.Popen( - ["./tincan", "-s", self.process_proxy_address[1:]] + if not tnlid: + raise ValueError("Tunnel ID cannot be None") + if tnlid in self._tc_proc_tbl: + raise ValueError( + "Tunnel ID %s is already assigned to active Tincan process %s", + tnlid, + self._tc_proc_tbl[tnlid], + ) + sub_proc = subprocess.Popen( + [ + "./tincan", + "-s", + self.process_proxy_address[1:], + "-t", + tnlid, + "-l", + json.dumps(self.log_config), + ] + ) + self._pids[sub_proc.pid] = tnlid + self._tc_proc_tbl[tnlid] = TincanProcess(tnlid=tnlid, proc=sub_proc) + self.logger.info( + "New Tincan session %s started for tunnel %s", sub_proc.pid, tnlid ) - self._echo_replies = 2 # reset the echo counter - self.logger.info("New Tincan session started %s", self._tc_proc.pid) - def _stop_tincan(self): - if self._tc_proc is None: + def _stop_tincan(self, tc_proc: TincanProcess, wt: int = 5.15): + if tc_proc is None: return try: - if self._tc_proc.poll() is not None: - self._tc_proc.send_signal(signal.SIGTERM) - self._tc_proc.wait(10.15) - except subprocess.TimeoutExpired: - if self._tc_proc is not None and self._tc_proc.poll() is not None: - self.logger.info("Killing unresponsive Tincan: %s", self._tc_proc.pid) - self._tc_proc.kill() - finally: - if not self.exit_ev.is_set(): - self._notify_tincan_terminated(self._tc_pid) - self._tc_pid = 0 - self._tc_proc = None - - def _restart_tincan(self): - self._stop_tincan() - self._start_tincan() - - def _notify_tincan_ready(self): - self._tc_pid = self._tc_proc.pid - self._tci_publisher.post_update( - { - "Command": "TincanReady", - "SessionId": self._tc_pid, - } - ) - - def _notify_tincan_terminated(self, old_pid): - self._tci_publisher.post_update( - { - "Command": "ResetTincanTunnels", - "Reason": "Tincan process terminated", - "SessionId": old_pid, - } - ) - - def _is_request_current(self, cbt) -> bool: - """There are 3 failure scenarios: - 1. No Tincan process currently exists - 2. The Tincan ID in the request does not match the one associated - with the tunnel ID. - 3. The Tincan ID in the request does not match the current one. - """ - is_current: bool = False - tnlid = cbt.request.params["TunnelId"] - tracked_sid = self._tunnel_pid.get(tnlid) - try: - msg = cbt.request.params - if self._tc_pid <= 0: - cbt.set_response( - { - "Message": "Tincan session not ready for request. Try again later.", - "NodeId": self.node_id, - "CurrentId": self._tc_pid, - }, - False, - ) - elif (tracked_sid and tracked_sid != msg["TincanId"]) or ( - msg["TincanId"] != self._tc_pid - ): - cbt.set_response( - { - "Message": "The requested Tincan session is invalid.", - "CurrentId": self._tc_pid, - }, - False, + exit_code = tc_proc.proc.poll() + if exit_code is None: + self.logger.debug( + "Terminating process %s - Tincan %s", + tc_proc.proc.pid, + tc_proc.tnlid, ) + ts = time.time() + tc_proc.proc.terminate() + tc_proc.proc.wait(wt) + self._kill_times.append(self._kill_times[-1] + time.time() - ts) else: - is_current = True - except Exception as exc: - self.logger.exception(exc) - if cbt: - cbt.set_response( - { - "Message": "The requested failed.", - "CurrentId": self._tc_pid, - }, - False, + self.logger.debug( + "Process %s for tunnel %s has already exited with code %s", + tc_proc.proc.pid, + tc_proc.tnlid[:7], + exit_code, ) - return is_current + except subprocess.TimeoutExpired: + exit_code = tc_proc.proc.poll() + if exit_code is None: + self._remove_tap() + tc_proc.proc.kill() + self._kill_times.append(self._kill_times[-1] + time.time() - ts) + self.logger.debug("Killed unresponsive Tincan: %s", tc_proc.proc.pid) + self.logger.info( + "Process %s for tunnel %s terminated", tc_proc.proc.pid, tc_proc.tnlid + ) + + def _notify_tincan_terminated(self, tnlid: str): + self._tci_publisher.post_update( + { + "Command": "TincanTunnelFailed", + "Reason": "Tincan process terminated", + "OverlayId": self._tc_proc_tbl[tnlid].ovlid, + "TunnelId": tnlid, + "TapName": self._tc_proc_tbl[tnlid].tap_name, + } + ) def handle_ipc(self, msg: ProxyMsg): try: ctl = msg.json - self._ipc_id = msg.fileno if ctl["ProtocolVersion"] != EVIO_VER_CTL: raise ValueError("Invalid control version detected") + # self.logger.debug("Received dataplane control - %s", ctl) # Get the original CBT if this is the response if ctl["ControlType"] == "Response": cbt = self.get_pending_cbt(ctl["TransactionId"]) @@ -392,13 +405,42 @@ class TincanTunnel(ControllerModule): else: req = ctl["Request"] if req["Command"] == "RegisterDataplane": - self.logger.debug("Received Tincan dataplane registration") - self.register_internal_cbt("TCI_CONFIGURE_LOGGING", self.log_config) - elif req["Command"] == "LinkStateChange": + pid = req["SessionId"] + self.logger.info( + "Received Tincan dataplane registration for session: %s", pid + ) + tnlid = self._pids[pid] + self._tc_proc_tbl[tnlid].ipc_id = msg.fileno + cbt = self._tnl_cbts.pop(tnlid, None) + if cbt: + cbt.next_op(cbt) + self._tc_proc_tbl[tnlid].do_chk = True + elif req["Command"] in ("LinkConnected", "LinkDisconnected"): self._tci_publisher.post_update(req) else: - self.loggger.warning( + self.logger.warning( "Invalid Tincan control command: %s", req["Command"] ) - except ValueError as vr: - self.logger.exception(str(vr)) + except Exception as err: + self.logger.exception(str(err)) + + def _is_tap_exist(self, tap_name: str) -> bool: + with IPRoute() as ipr: + idx = ipr.link_lookup(ifname=tap_name) + if len(idx) == 1: + return True + return False + + def _remove_tap(self, tap_name: str): + try: + self.logger.info("Removing Tincan TAP device %s", tap_name) + with IPRoute() as ipr: + idx = ipr.link_lookup(ifname=tap_name) + if len(idx) > 0: + idx = idx[0] + ipr.link("set", index=idx, state="down") + ipr.link("del", index=idx) + except Exception as e: + self.logger.warning( + "Failed to remove Tincan TAP device %s, error code: %s", tap_name, e + ) diff --git a/evio/controllers/topology.py b/evio/controllers/topology.py index 33c7628..7fc82a7 100644 --- a/evio/controllers/topology.py +++ b/evio/controllers/topology.py @@ -140,7 +140,6 @@ class NetworkOverlay: "_max_concurrent_edits", "num_active_edits", "new_peer_count", - "known_peers", "ond_peers", "_graph_transformation", "adjacency_list", @@ -150,16 +149,12 @@ class NetworkOverlay: # used to limit number of concurrent operations initiated self._max_concurrent_edits = kwargs.get("MaxConcurrentOps", MAX_CONCURRENT_OPS) self._bsemp = threading.BoundedSemaphore(self._max_concurrent_edits) - # self._refc: int = self._max_concurrent_edits - # self._reflk = threading.Lock() self.node_id: str = node_id self.overlay_id: str = overlay_id self.logger: logging.Logger = kwargs["Logger"] self.new_peer_count: int = 0 self._graph_transformation: GraphTransformation = None - # self.transformation: GraphTransformation = None self.known_peers: dict[str, DiscoveredPeer] = {} - # self.pending_auth: dict[str, EdgeResponse] = {} self.ond_peers: list[dict] = [] self.adjacency_list = ConnEdgeAdjacenctList(overlay_id, node_id) self._loc_id: int = kwargs.get("LocationId") @@ -247,16 +242,6 @@ class Topology(ControllerModule): "The Signal PEER PRESENCE subscription is not available. Topology cannot continue." ) self.start_subscription("Signal", "SIG_PEER_PRESENCE_NOTIFY") - if ( - "TincanTunnel" not in publishers - or "TCI_TINCAN_MSG_NOTIFY" - not in self.get_available_subscriptions("TincanTunnel") - ): - raise RuntimeError( - "The TincanTunnel MESSAGE NOTIFY subscription is not available." - "Link Manager cannot continue." - ) - self.start_subscription("TincanTunnel", "TCI_TINCAN_MSG_NOTIFY") if ( "LinkManager" not in publishers or "LNK_TUNNEL_EVENTS" @@ -318,7 +303,6 @@ class Topology(ControllerModule): "TOP_NEGOTIATE_EDGE": self.req_handler_negotiate_edge, "TOP_QUERY_KNOWN_PEERS": self.req_handler_query_known_peers, "_TOPOLOGY_UPDATE_": self._req_handler_manage_topology, - "TCI_TINCAN_MSG_NOTIFY": self.req_handler_tincan_notify, } def _register_resp_handlers(self): @@ -375,23 +359,13 @@ class Topology(ControllerModule): if self._net_ovls[olid].new_peer_count >= self.config.get( "PeerDiscoveryCoalesce", PEER_DISCOVERY_COALESCE ): - self.logger.debug( - "%s/%s discovered - coalesced %s of %s, " - "attempting overlay update", - olid, - peer_id, + self.logger.info( + "Coalesced %d of %d discovered peers, attempting update on overlay %s", self._net_ovls[olid].new_peer_count, self.config.get("PeerDiscoveryCoalesce", PEER_DISCOVERY_COALESCE), + olid, ) self._update_overlay(olid) - elif self.logger.isEnabledFor(logging.INFO): - self.logger.info( - "%s/%s discovered - coalesced %s of %s", - olid, - peer_id, - self._net_ovls[olid].new_peer_count, - self.config.get("PeerDiscoveryCoalesce", PEER_DISCOVERY_COALESCE), - ) cbt.set_response(None, True) self.complete_cbt(cbt) @@ -456,9 +430,6 @@ class Topology(ControllerModule): "Data": ce, } ) - # if ce.edge_type in EDGE_TYPE_OUT: - # ovl.release() - # self._process_next_transition(ovl) elif event == TUNNEL_EVENTS.Disconnected: ce = ovl.adjacency_list[peer_id] if ce.edge_state != EDGE_STATES.Connected: @@ -484,57 +455,10 @@ class Topology(ControllerModule): ) self._remove_tunnel(ovl, ce.dataplane, peer_id, ce.edge_id) elif event == TUNNEL_EVENTS.Removed: - """Role B""" - ce = ovl.adjacency_list.get(peer_id, None) - if ( - ce - and ce.role == CONNECTION_ROLE.Target - and ce.edge_state == EDGE_STATES.Authorized - ): - # Tunnel/Link cm handles creating the tunnel. Only need to remove CE from adj list. - ovl.adjacency_list.pop(peer_id, None) - # if ( - # ce - # and ce.role == CONNECTION_ROLE.Initiator - # and ce.edge_state == EDGE_STATES.Authorized - # ): - # # ce will be none as the resp handler for the failed create tunnel - # # would have removed the CE - # # raise RuntimeError( - # # f"Tunnel removed event is invalid for authorized initiator CE= {ce}" - # # ) - # pass - # elif ( - # ce - # and ce.role == CONNECTION_ROLE.Initiator - # and ce.edge_state == EDGE_STATES.Connected - # ): # topo initiated the removal - # raise RuntimeError( - # f"Expected ce as None, resp handler remove tnl should clean up but got {ce}" - # ) - # elif ce and ce.edge_state == EDGE_STATES.Disconnected: - # ce.edge_state = EDGE_STATES.Deleting - # elif ( - # ce - # and ce.role == CONNECTION_ROLE.Target - # and ce.edge_state == EDGE_STATES.Authorized - # ): - # # Tunnel/Link cm handles creating the tunnel. Only need to remove CE from adj list. - # ovl.adjacency_list.pop(peer_id, None) - # # raise RuntimeError(f"Tunnel removed event is invalid for auth tgt since tnl cm - # # handles creating the tunnel {ce}") - # elif ( - # ce - # and ce.role == CONNECTION_ROLE.Target - # and ce.edge_state == EDGE_STATES.Connected - # ): # the peer disconnected - # raise RuntimeError( - # f"The target node should not initiate removal on connect edges {ce}" - # ) - # elif ce: - # self.logger.error( - # "Tunnel event remove is unexpected for conn edge %s", ce - # ) + """The removed event is also generated for tincan process failure""" + ce = ovl.adjacency_list.pop(peer_id, None) + if ce: + self.logger.info("Edge %s removed from adjacency list", ce.edge_id) else: self.logger.warning("Invalid UpdateType specified for event %s", event) @@ -605,9 +529,6 @@ class Topology(ControllerModule): "leaf".casefold(), ROLES, ): - self.logger.info( - "The edge request was refused as this is a pendant device." - ) edge_cbt.set_response( "E6 - Not accepting incoming connections, leaf device", False ) @@ -615,7 +536,13 @@ class Topology(ControllerModule): return net_ovl = self._net_ovls[olid] edge_resp: EdgeResponse = None - self.logger.debug("Rcvd EdgeRequest=%s", str(edge_req)) + self.logger.info( + "Received %s EdgeRequest=%s/%s from %s", + edge_req.edge_type, + edge_req.overlay_id, + edge_req.edge_id, + edge_req.initiator_id, + ) peer_id = edge_req.initiator_id if peer_id in net_ovl.adjacency_list: edge_resp = self._resolve_request_collision( @@ -625,8 +552,6 @@ class Topology(ControllerModule): edge_resp = self._negotiate_response(net_ovl, edge_req) if edge_resp and edge_resp.is_accepted: - # net_ovl.pending_auth[peer_id] = edge_resp - # edge_cbt.add_context("pending_auth", edge_resp) if edge_resp.message[:2] == "E0": net_ovl.adjacency_list.pop(peer_id) et = transpose_edge_type(edge_req.edge_type) @@ -675,19 +600,6 @@ class Topology(ControllerModule): cbt.set_response(ovl_peers, True) self.complete_cbt(cbt) - def req_handler_tincan_notify(self, cbt: CBT): - if cbt.request.params["Command"] == "ResetTincanTunnels": - sid = cbt.request.params["SessionId"] - for olid, ovl in self._net_ovls.items(): - self.logger.info( - "Clearing Tincan CE's from %s for session %s", olid, sid - ) - ovl.adjacency_list.clear_tincan_ces() - if ovl.transformation: - ovl.transformation.clear() - cbt.set_response(data=None, status=True) - self.complete_cbt(cbt) - def resp_handler_auth_tunnel(self, cbt: CBT): """Role B2 LNK auth completed, add the CE to Netbuilder and send response to initiator ie., Role A @@ -916,7 +828,13 @@ class Topology(ControllerModule): capability=dp_types, ) edge_params = er._asdict() - self.logger.info("Initiating %s", er) + self.logger.info( + "Initiating %s EdgeRequest to %s/%s to %s", + er.edge_type, + er.overlay_id, + er.edge_id, + er.recipient_id, + ) rem_act = RemoteAction( net_ovl.overlay_id, er.recipient_id, @@ -930,7 +848,13 @@ class Topology(ControllerModule): self, net_ovl: NetworkOverlay, edge_nego: EdgeNegotiate ): """Role A2""" - self.logger.debug("Completing %s", str(edge_nego)) + self.logger.info( + "Completing %s EdgeNegotiate of %s/%s to %s", + edge_nego.edge_type, + edge_nego.overlay_id, + edge_nego.edge_id[:7], + edge_nego.recipient_id[:7], + ) if edge_nego.recipient_id not in net_ovl.adjacency_list: self.logger.warning( "The peer specified in edge negotiation %s is not in current " @@ -949,7 +873,6 @@ class Topology(ControllerModule): ce.edge_state = EDGE_STATES.Deleting del net_ovl.adjacency_list[ce.peer_id] net_ovl.known_peers[peer_id].exclude() - # net_ovl.release() # release on explicit negotiate fail self._process_next_transition(net_ovl) else: if ce.edge_state != EDGE_STATES.PreAuth: @@ -974,7 +897,6 @@ class Topology(ControllerModule): ce.edge_state = EDGE_STATES.Deleting del net_ovl.adjacency_list[ce.peer_id] net_ovl.known_peers[peer_id].exclude() - # net_ovl.release() # release on explicit negotiate fail self._process_next_transition(net_ovl) return # record tunnel start on node A after successful edge negotiation @@ -999,9 +921,9 @@ class Topology(ControllerModule): neg_edge_cbt: CBT, edge_resp: EdgeResponse, ): - """->Role B1""" + """Role B1""" self.logger.info( - "Authorizing peer edge %s from %s:%s->%s", + "Authorizing incoming peer edge %s from %s:%s->%s", edge_id, net_ovl.overlay_id, peer_id[:7], @@ -1163,7 +1085,7 @@ class Topology(ControllerModule): ) # succ threshold -> at/below the min required ): raise ValueError("Successor threshold not met") - self.logger.info("Removing edge %s", ce) + self.logger.debug("Removing edge %s", ce) self._remove_tunnel(net_ovl, ce.dataplane, ce.peer_id, ce.edge_id) return True return False @@ -1180,7 +1102,12 @@ class Topology(ControllerModule): "PeerId": peer_id, "TunnelId": tunnel_id, } - self.logger.info("Removing tunnel %s to %s", tunnel_id[:7], peer_id[:7]) + self.logger.info( + "Removing tunnel %s/%s to %s", + net_ovl.overlay_id, + tunnel_id[:7], + peer_id[:7], + ) if dataplane == DATAPLANE_TYPES.Geneve: self.register_cbt("GeneveTunnel", "GNV_REMOVE_TUNNEL", params) elif dataplane == DATAPLANE_TYPES.Tincan: @@ -1201,15 +1128,11 @@ class Topology(ControllerModule): def _cleanup_expired_incomplete_edge(self, cbt: CBT): self.logger.debug("Abort CBT %s", cbt) - olid = cbt.request.params.get("OverlayId", None) - if not olid: - self.logger.warning("No overlay ID found in expired CBT") - return + olid = cbt.request.params["OverlayId"] net_ovl = self._net_ovls[olid] cbt.pop_context("pending_auth") - peer_id = cbt.request.params.get("PeerId", None) - if peer_id: - net_ovl.adjacency_list.pop(peer_id, None) + peer_id = cbt.request.params["PeerId"] + net_ovl.adjacency_list.pop(peer_id, None) self.free_cbt(cbt) def _abort_handler_remote_action(self, cbt: CBT): diff --git a/evio/controllers/tunnel.py b/evio/controllers/tunnel.py index 02fe40a..0dbf404 100644 --- a/evio/controllers/tunnel.py +++ b/evio/controllers/tunnel.py @@ -42,13 +42,14 @@ TUNNEL_EVENTS = TunnelEvents() TunnelStates = namedtuple( "TUNNEL_STATES", - ["AUTHORIZED", "CREATING", "QUERYING", "ONLINE", "OFFLINE"], + ["AUTHORIZED", "CREATING", "QUERYING", "ONLINE", "OFFLINE", "FAILED"], defaults=[ "TNL_AUTHORIZED", "TNL_CREATING", "TNL_QUERYING", "TNL_ONLINE", "TNL_OFFLINE", + "TNL_FAILED", ], ) TUNNEL_STATES = TunnelStates()