Merge pull request #524 from atlanticwave-sdx/rollback-on-failed-oxp-post

Rollback on failed OXP POST

Author: congwang09

compose.yml

@@ -54,6 +54,8 @@ services:
       # MQ settings
       - MQ_HOST=${MQ_HOST}
       - MQ_PORT=${MQ_PORT:-5672}
+      - OXP_RESPONSE_TIMEOUT=${OXP_RESPONSE_TIMEOUT:-60}
+      - PROVISIONING_MONITOR_INTERVAL=${PROVISIONING_MONITOR_INTERVAL:-5}
 
 volumes:
   mongodb:

env.template

@@ -5,6 +5,8 @@ SDX_PORT=8080
 SDX_NAME=sdx-controller-test
 HEARTBEAT_INTERVAL=30
 HEARTBEAT_TOLERANCE=3
+OXP_RESPONSE_TIMEOUT=60
+PROVISIONING_MONITOR_INTERVAL=5
 
 # Message queue settings for SDX Controller.
 MQ_HOST=aw-sdx-monitor.renci.org

sdx_controller/__init__.py

@@ -1,19 +1,28 @@
 import logging
 import os
 import threading
+import time
 from queue import Queue
 
 import connexion
+from sdx_datamodel.connection_sm import ConnectionStateMachine
+from sdx_datamodel.constants import MongoCollections
 from sdx_pce.topology.temanager import TEManager
 
 from sdx_controller import encoder
+from sdx_controller.handlers.connection_handler import (
+    ConnectionHandler,
+    connection_state_machine,
+)
 from sdx_controller.messaging.rpc_queue_consumer import RpcConsumer
 from sdx_controller.utils.db_utils import DbUtils
 
 logger = logging.getLogger(__name__)
 logging.getLogger("pika").setLevel(logging.WARNING)
 LOG_FILE = os.environ.get("LOG_FILE")
 LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG")
+OXP_RESPONSE_TIMEOUT = int(os.getenv("OXP_RESPONSE_TIMEOUT", 60))
+PROVISIONING_MONITOR_INTERVAL = int(os.getenv("PROVISIONING_MONITOR_INTERVAL", 5))
 
 
 def create_rpc_thread(app):
@@ -32,6 +41,82 @@ def create_rpc_thread(app):
     rpc_thread.start()
 
 
+def create_provisioning_timeout_thread(app):
+    """
+    Start a background monitor for connections stuck in
+    UNDER_PROVISIONING longer than the configured timeout.
+    """
+    if OXP_RESPONSE_TIMEOUT <= 0:
+        logger.info("[ProvisioningTimeout] Disabled.")
+        app.provisioning_timeout_thread = None
+        return
+
+    connection_handler = ConnectionHandler(app.db_instance)
+
+    def monitor_loop():
+        logger.info(
+            f"[ProvisioningTimeout] Started monitoring with timeout={OXP_RESPONSE_TIMEOUT}s interval={PROVISIONING_MONITOR_INTERVAL}s."
+        )
+        while True:
+            try:
+                now = time.time()
+                connections = app.db_instance.get_all_entries_in_collection(
+                    MongoCollections.CONNECTIONS
+                )
+                for connection_entry in connections:
+                    service_id = next(iter(connection_entry), None)
+                    connection = (
+                        connection_entry.get(service_id) if service_id else None
+                    )
+                    if not isinstance(connection, dict):
+                        continue
+                    if connection.get("status") != str(
+                        ConnectionStateMachine.State.UNDER_PROVISIONING
+                    ):
+                        continue
+
+                    started_at = connection.get("provisioning_started_at")
+                    if not isinstance(started_at, (int, float)):
+                        continue
+                    if connection.get("provisioning_timeout_handled"):
+                        continue
+                    if now - started_at < OXP_RESPONSE_TIMEOUT:
+                        continue
+
+                    logger.warning(
+                        f"[ProvisioningTimeout] Connection {service_id} timed out after {int(now - started_at)}s waiting for OXP responses."
+                    )
+
+                    connection["provisioning_timeout_handled"] = True
+                    connection["partial_cleanup_requested"] = True
+                    connection["timeout_reason"] = (
+                        f"OXP response timeout after {OXP_RESPONSE_TIMEOUT} seconds"
+                    )
+                    connection, _ = connection_state_machine(
+                        connection, ConnectionStateMachine.State.DOWN
+                    )
+                    app.db_instance.add_key_value_pair_to_db(
+                        MongoCollections.CONNECTIONS, service_id, connection
+                    )
+                    cleanup_status, cleanup_code = (
+                        connection_handler.cleanup_partial_connection(
+                            app.te_manager, service_id, connection
+                        )
+                    )
+                    logger.info(
+                        f"[ProvisioningTimeout] Cleanup result for {service_id}: {cleanup_status}, code={cleanup_code}"
+                    )
+            except Exception as e:
+                logger.exception(
+                    f"[ProvisioningTimeout] Error while monitoring connections: {e}"
+                )
+            time.sleep(PROVISIONING_MONITOR_INTERVAL)
+
+    provisioning_thread = threading.Thread(target=monitor_loop, daemon=True)
+    provisioning_thread.start()
+    app.provisioning_timeout_thread = provisioning_thread
+
+
 def create_app(run_listener: bool = True):
     """
     Create a connexion app.
@@ -74,6 +159,8 @@ def create_app(run_listener: bool = True):
     # pass this around.
     app.app.te_manager = app.te_manager
 
+    create_provisioning_timeout_thread(app)
+
     if run_listener:
         create_rpc_thread(app)
     else: