diff --git a/compose.yml b/compose.yml index f64e830..b9b77f9 100644 --- a/compose.yml +++ b/compose.yml @@ -57,8 +57,8 @@ services: - MQ_PASS=${MQ_PASS:-guest} # OXP settings. - OXP_PROVISION_URL=${OXP_PROVISION_URL} - - OXP_PULL_URL=${OXP_PULL_URL} - - OXP_PULL_INTERVAL=${OXP_PULL_INTERVAL} + - OXP_TOPOLOGY_URL=${OXP_TOPOLOGY_URL} + - OXP_PULL_TOPOLOGY_INTERVAL=${OXP_PULL_TOPOLOGY_INTERVAL} - OXP_CONNECTION_URL=${OXP_CONNECTION_URL} volumes: diff --git a/env.template b/env.template index 8658ec8..ea5ece3 100644 --- a/env.template +++ b/env.template @@ -37,6 +37,8 @@ MQ_PASS=guest # Kytos/OESS API address OXP_PROVISION_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/provision -OXP_PULL_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/topology -OXP_PULL_INTERVAL=180 +OXP_TOPOLOGY_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/topology +OXP_PULL_TOPOLOGY_INTERVAL=180 OXP_CONNECTION_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/connection +OXP_LIST_CONNECTIONS_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/connections +OXP_PULL_CONNECTIONS_INTERVAL=180 diff --git a/pyproject.toml b/pyproject.toml index 676a01f..81ed1b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ dependencies = [ "connexion[swagger-ui] == 2.14.2", "asgiref >= 3.7.2", "pymongo > 3.0", - "sdx-datamodel @ git+https://github.com/atlanticwave-sdx/datamodel@v3.2.1", + "sdx-datamodel @ git+https://github.com/atlanticwave-sdx/datamodel@v3.2.0", ] [project.optional-dependencies] diff --git a/sdx_lc/handlers/sdx_controller_msg_handler.py b/sdx_lc/handlers/sdx_controller_msg_handler.py index ea6157c..29ca4a4 100644 --- a/sdx_lc/handlers/sdx_controller_msg_handler.py +++ b/sdx_lc/handlers/sdx_controller_msg_handler.py @@ -91,8 +91,12 @@ def process_sdx_controller_json_msg(self, msg): if "link" in msg_json and ("endpoints" in msg_json["link"]): service_id = msg_json.get("service_id") + if not service_id: + self.logger.info(f"Connection did not include service_id. Ignored.") + return + connection = msg_json.get("link") - self.db_instance.add_key_value_pair_to_db(self.message_id, connection) + self.db_instance.add_key_value_pair_to_db(service_id, connection) self.logger.info("Save to database complete.") self.logger.info("Message ID:" + str(self.message_id)) self.message_id += 1 diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py new file mode 100644 index 0000000..b6b798a --- /dev/null +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -0,0 +1,112 @@ +import json +import logging +import os.path +import sys +import time + +import requests +from sdx_datamodel.constants import Constants, MessageQueueNames + +# append abspath, so this file can import other modules from parent directory +sys.path.append( + os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)) +) + +from messaging.rpc_queue_producer import RpcProducer +from utils.db_utils import DbUtils + +SDXLC_DOMAIN = os.environ.get("SDXLC_DOMAIN") +OXP_LIST_CONNECTIONS_URL = os.environ.get("OXP_LIST_CONNECTIONS_URL") +OXP_PULL_CONNECTIONS_INTERVAL = os.environ.get("OXP_PULL_CONNECTIONS_INTERVAL") +PUB_QUEUE = MessageQueueNames.OXP_UPDATE +logger = logging.getLogger(__name__) + + +def main(): + db_instance = DbUtils() + db_instance.initialize_db() + process_oxp_connections(db_instance) + + +# Periodically pull l2vpn (connection) status from OXP, and handle status change. +# Possible l2vpn status are: +# “up” if the L2VPN is operational, +# “down” if the L2VPN is not operational due to topology issues/lack of path, or endpoints being down, +# “error” when there is an error with the L2VPN, +# “under provisioning” when the L2VPN is still being provisioned by the OXPs, +# “maintenance” when the L2VPN is being affected by a network maintenance. +def process_oxp_connections(db_instance): + while True: + time.sleep(int(OXP_PULL_CONNECTIONS_INTERVAL)) + try: + try: + response = requests.get(OXP_LIST_CONNECTIONS_URL, timeout=10) + connections = response.content + assert response.ok, response.text + except (requests.ConnectionError, requests.HTTPError) as err: + logger.error(f"Error connecting to OXP: {err}") + continue + + logger.debug("Received connections from OXP.") + + try: + connections_json = response.json() + except ValueError: + logger.debug("Cannot parse connections, invalid JSON.") + continue + + if not connections_json: + logger.debug("No connections yet.") + continue + + for service_id, connection in connections_json.items(): + # Fetch existing connection from DB + existing_connection = db_instance.get_value_by_key(service_id) + + if not existing_connection: + logger.debug(f"New connection {service_id}, ignored") + continue + + try: + existing_connection_json = json.loads(existing_connection) + except ValueError: + logger.debug(f"Invalid JSON in DB for {service_id}") + continue + + existing_connection_status = ( + existing_connection_json.get("status") + if existing_connection_json + else None + ) + new_status = connection.get("status") + + if existing_connection_status == new_status: + logger.debug(f"Status unchanged for {service_id}") + continue + + existing_connection_json["status"] = new_status + logger.info( + f"Status change for {service_id}: " + f"{existing_connection_status} changed to {new_status}" + ) + db_instance.add_key_value_pair_to_db( + service_id, existing_connection_json + ) + rpc_msg = { + "lc_domain": SDXLC_DOMAIN, + "msg_type": "oxp_conn_status_change", + "service_id": service_id, + "existing_status": existing_connection_status, + "new_status": new_status, + } + rpc_producer = RpcProducer(5, "", PUB_QUEUE) + rpc_producer.call(json.dumps(rpc_msg)) + rpc_producer.stop() + except Exception: + logger.exception( + "Unexpected error while processing OXP connections; Retrying." + ) + + +if __name__ == "__main__": + main() diff --git a/sdx_lc/jobs/pull_topo_changes.py b/sdx_lc/jobs/pull_topo_changes.py index 7adce92..ad2a75d 100644 --- a/sdx_lc/jobs/pull_topo_changes.py +++ b/sdx_lc/jobs/pull_topo_changes.py @@ -17,8 +17,8 @@ OXPO_USER = os.environ.get("OXPO_USER", None) OXPO_PASS = os.environ.get("OXPO_PASS", None) -OXP_PULL_URL = os.environ.get("OXP_PULL_URL") -OXP_PULL_INTERVAL = os.environ.get("OXP_PULL_INTERVAL") +OXP_TOPOLOGY_URL = os.environ.get("OXP_TOPOLOGY_URL") +OXP_PULL_TOPOLOGY_INTERVAL = os.environ.get("OXP_PULL_TOPOLOGY_INTERVAL") PUB_QUEUE = MessageQueueNames.OXP_UPDATE logger = logging.getLogger(__name__) @@ -32,7 +32,7 @@ def main(): def process_domain_controller_topo(db_instance): while True: - time.sleep(int(OXP_PULL_INTERVAL)) + time.sleep(int(OXP_PULL_TOPOLOGY_INTERVAL)) latest_topology_exists = False latest_topology = db_instance.read_from_db(Constants.LATEST_TOPOLOGY) @@ -55,7 +55,7 @@ def process_domain_controller_topo(db_instance): logger.debug("Latest topology does not exist") try: - response = requests.get(OXP_PULL_URL, auth=(OXPO_USER, OXPO_PASS)) + response = requests.get(OXP_TOPOLOGY_URL, auth=(OXPO_USER, OXPO_PASS)) pulled_topology = response.content except (requests.ConnectionError, requests.HTTPError): logger.debug("Error connecting to domain controller...")