Pull conn status from oxp

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:

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

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]

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

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

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