From 75302ef8ab017bc8eb2985e38c059a73fb8d8ca7 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 10 Dec 2025 17:11:23 -0500 Subject: [PATCH 01/10] Pull OXP connection status --- compose.yml | 4 +- env.template | 6 +- sdx_lc/handlers/sdx_controller_msg_handler.py | 6 +- sdx_lc/jobs/pull_connection_changes.py | 96 +++++++++++++++++++ sdx_lc/jobs/pull_topo_changes.py | 8 +- 5 files changed, 111 insertions(+), 9 deletions(-) create mode 100644 sdx_lc/jobs/pull_connection_changes.py 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 ffc56cf..dbcd4f7 100644 --- a/env.template +++ b/env.template @@ -36,6 +36,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/sdx_lc/handlers/sdx_controller_msg_handler.py b/sdx_lc/handlers/sdx_controller_msg_handler.py index a08a45d..1017d54 100644 --- a/sdx_lc/handlers/sdx_controller_msg_handler.py +++ b/sdx_lc/handlers/sdx_controller_msg_handler.py @@ -78,8 +78,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..f1ed91c --- /dev/null +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -0,0 +1,96 @@ +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 + +OXPO_USER = os.environ.get("OXPO_USER", None) +OXPO_PASS = os.environ.get("OXPO_PASS", None) +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) + + +def process_oxp_connections(db_instance): + while True: + time.sleep(int(OXP_PULL_CONNECTIONS_INTERVAL)) + + try: + response = requests.get(OXP_LIST_CONNECTIONS_URL) + connections = response.content + except (requests.ConnectionError, requests.HTTPError): + logger.debug("Error connecting to OXP...") + continue + + if not response.ok: + 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.debug( + 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_producer = RpcProducer(5, "", PUB_QUEUE) + rpc_producer.call(json.dumps(existing_connection_json)) + rpc_producer.stop() + + +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...") From 9a770c94507af02fcbe97fdaee3c55ca376f1a81 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Thu, 11 Dec 2025 11:24:20 -0500 Subject: [PATCH 02/10] Update rpc_msg --- sdx_lc/jobs/pull_connection_changes.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index f1ed91c..a23cb00 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -86,9 +86,14 @@ def process_oxp_connections(db_instance): 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_response", + "service_id": service_id, + "oxp_response": existing_connection_json, + } rpc_producer = RpcProducer(5, "", PUB_QUEUE) - rpc_producer.call(json.dumps(existing_connection_json)) + rpc_producer.call(json.dumps(rpc_msg)) rpc_producer.stop() From e9e9851bbae56076322f997d44fc0678d82ce062 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 11 Feb 2026 11:27:37 -0500 Subject: [PATCH 03/10] Get domain from env --- sdx_lc/jobs/pull_connection_changes.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index a23cb00..57406df 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -15,8 +15,7 @@ from messaging.rpc_queue_producer import RpcProducer from utils.db_utils import DbUtils -OXPO_USER = os.environ.get("OXPO_USER", None) -OXPO_PASS = os.environ.get("OXPO_PASS", None) +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 From c37f3c91b1046e9eff331f75934a3bb90eb772c9 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Tue, 17 Feb 2026 21:57:18 -0500 Subject: [PATCH 04/10] Update status change --- pyproject.toml | 2 +- sdx_lc/jobs/pull_connection_changes.py | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) 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/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index 57406df..cbc353d 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -87,9 +87,10 @@ def process_oxp_connections(db_instance): db_instance.add_key_value_pair_to_db(service_id, existing_connection_json) rpc_msg = { "lc_domain": SDXLC_DOMAIN, - "msg_type": "oxp_conn_response", + "msg_type": "oxp_conn_status_change", "service_id": service_id, - "oxp_response": existing_connection_json, + "existing_status": existing_connection_status, + "new_status": new_status } rpc_producer = RpcProducer(5, "", PUB_QUEUE) rpc_producer.call(json.dumps(rpc_msg)) From 5369336b29c767689783ef2d93018be5df5abe98 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 18 Feb 2026 10:57:28 -0500 Subject: [PATCH 05/10] Add some comments --- sdx_lc/jobs/pull_connection_changes.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index cbc353d..972a6e6 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -28,6 +28,13 @@ def main(): 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)) From 1e7bd649019178a4e97b1269757890841300db67 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 18 Feb 2026 11:07:19 -0500 Subject: [PATCH 06/10] Format --- sdx_lc/jobs/pull_connection_changes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index 972a6e6..532e0c2 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -97,7 +97,7 @@ def process_oxp_connections(db_instance): "msg_type": "oxp_conn_status_change", "service_id": service_id, "existing_status": existing_connection_status, - "new_status": new_status + "new_status": new_status, } rpc_producer = RpcProducer(5, "", PUB_QUEUE) rpc_producer.call(json.dumps(rpc_msg)) From 2902fe9efc7fa713cb4827501582a9c20cefe3de Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 25 Feb 2026 16:06:19 -0500 Subject: [PATCH 07/10] Handle exceptions --- sdx_lc/jobs/pull_connection_changes.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index 532e0c2..204c36f 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -42,11 +42,9 @@ def process_oxp_connections(db_instance): try: response = requests.get(OXP_LIST_CONNECTIONS_URL) connections = response.content - except (requests.ConnectionError, requests.HTTPError): - logger.debug("Error connecting to OXP...") - continue - - if not response.ok: + 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.") From f14395de341883b8dbb38ba0d1313027d8c85c50 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 25 Feb 2026 16:09:12 -0500 Subject: [PATCH 08/10] Add timeout --- sdx_lc/jobs/pull_connection_changes.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index 204c36f..fd8a5ab 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -40,7 +40,7 @@ def process_oxp_connections(db_instance): time.sleep(int(OXP_PULL_CONNECTIONS_INTERVAL)) try: - response = requests.get(OXP_LIST_CONNECTIONS_URL) + response = requests.get(OXP_LIST_CONNECTIONS_URL, timeout=10) connections = response.content assert response.ok, response.text except (requests.ConnectionError, requests.HTTPError) as err: @@ -85,7 +85,7 @@ def process_oxp_connections(db_instance): continue existing_connection_json["status"] = new_status - logger.debug( + logger.info( f"Status change for {service_id}: " f"{existing_connection_status} changed to {new_status}" ) From a6a88cf0cf05c87a114d47df2f473d46ba2b54f7 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 25 Feb 2026 16:10:45 -0500 Subject: [PATCH 09/10] Add retry --- sdx_lc/jobs/pull_connection_changes.py | 108 +++++++++++++------------ 1 file changed, 56 insertions(+), 52 deletions(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index fd8a5ab..159c93e 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -38,68 +38,72 @@ def main(): def process_oxp_connections(db_instance): while True: time.sleep(int(OXP_PULL_CONNECTIONS_INTERVAL)) - - 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") + 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: - existing_connection_json = json.loads(existing_connection) + connections_json = response.json() except ValueError: - logger.debug(f"Invalid JSON in DB for {service_id}") + logger.debug("Cannot parse connections, invalid JSON.") 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}") + if not connections_json: + logger.debug("No connections yet.") continue - existing_connection_json["status"] = new_status - logger.info( - f"Status change for {service_id}: " - f"{existing_connection_status} changed to {new_status}" + 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." ) - 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() if __name__ == "__main__": From 8f21d7604e1f8e9fb6d7272cf9db4cf4dce8a0f2 Mon Sep 17 00:00:00 2001 From: Cong Wang Date: Wed, 25 Feb 2026 16:17:41 -0500 Subject: [PATCH 10/10] Format --- sdx_lc/jobs/pull_connection_changes.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sdx_lc/jobs/pull_connection_changes.py b/sdx_lc/jobs/pull_connection_changes.py index 159c93e..b6b798a 100644 --- a/sdx_lc/jobs/pull_connection_changes.py +++ b/sdx_lc/jobs/pull_connection_changes.py @@ -89,7 +89,9 @@ def process_oxp_connections(db_instance): 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) + 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",