EdgeVPNio
/
evio
mirror of https://github.com/EdgeVPNio/evio.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Update for multiprocess Tincan dataplane

master
Ken Subratie 2023-10-02 16:11:27 -04:00
parent 7fe946c7d3
commit 641db953ef
16 changed files with 542 additions and 681 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
evio/broker/__init__.py
View File

@ -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": "",
},
}

33
evio/broker/broker.py
View File

@ -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()

13
evio/broker/controller_module.py
View File

@ -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):

22
evio/broker/nexus.py
View File

@ -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):

14
evio/broker/process_proxy.py
View File

@ -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)

4
evio/broker/remote_action.py
View File

@ -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)

15
evio/controllers/bounded_flood.py
View File

@ -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,

127
evio/controllers/bridge_controller.py
View File

@ -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")

32
evio/controllers/geneve_tunnel.py
View File

@ -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])

1
evio/controllers/graph_builder.py
View File

@ -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
)

251
evio/controllers/link_manager.py
View File

@ -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.

5
evio/controllers/network_graph.py
View File

@ -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):

8
evio/controllers/signal.py
View File

@ -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

518
evio/controllers/tincan_tunnel.py
View File

@ -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
)

157
evio/controllers/topology.py
View File

@ -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):

3
evio/controllers/tunnel.py
View File

@ -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()