diff --git a/compose.yml b/compose.yml index 4e80e76e..a487bfd9 100644 --- a/compose.yml +++ b/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: diff --git a/env.template b/env.template index 1f2c8804..a467f8ff 100644 --- a/env.template +++ b/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 diff --git a/sdx_controller/__init__.py b/sdx_controller/__init__.py index 249efad5..6e6cd30d 100644 --- a/sdx_controller/__init__.py +++ b/sdx_controller/__init__.py @@ -1,12 +1,19 @@ 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 @@ -14,6 +21,8 @@ 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: diff --git a/sdx_controller/controllers/l2vpn_controller.py b/sdx_controller/controllers/l2vpn_controller.py index 5f0a020c..e405d2a1 100644 --- a/sdx_controller/controllers/l2vpn_controller.py +++ b/sdx_controller/controllers/l2vpn_controller.py @@ -26,6 +26,19 @@ logger = logging.getLogger(__name__) logging.getLogger("pika").setLevel(logging.WARNING) logger.setLevel(logging.getLevelName(os.getenv("LOG_LEVEL", "DEBUG"))) +ROLLBACK_SETTLE_TIMEOUT_SECONDS = float( + os.getenv("ROLLBACK_SETTLE_TIMEOUT_SECONDS", "5") +) +ROLLBACK_SETTLE_POLL_SECONDS = float(os.getenv("ROLLBACK_SETTLE_POLL_SECONDS", "0.2")) + +# PATCH must wait for async OXP provisioning responses before deciding whether +# the new service is really up or needs rollback to the previous request. +PATCH_PROVISIONING_SETTLE_TIMEOUT_SECONDS = int( + os.getenv("PATCH_PROVISIONING_SETTLE_TIMEOUT_SECONDS", "10") +) +PATCH_PROVISIONING_SETTLE_POLL_SECONDS = int( + os.getenv("PATCH_PROVISIONING_SETTLE_POLL_SECONDS", "1") +) # Get DB connection and tables set up. db_instance = DbUtils() @@ -33,6 +46,36 @@ connection_handler = ConnectionHandler(db_instance) +def _wait_for_patch_provisioning_to_settle(service_id): + deadline = time.time() + PATCH_PROVISIONING_SETTLE_TIMEOUT_SECONDS + connection = db_instance.get_value_from_db(MongoCollections.CONNECTIONS, service_id) + + while time.time() < deadline: + connection = db_instance.get_value_from_db( + MongoCollections.CONNECTIONS, service_id + ) + if not connection: + return None + + status = connection.get("status") + if status != str(ConnectionStateMachine.State.UNDER_PROVISIONING): + oxp_response = connection.get("oxp_response") or {} + breakdown = db_instance.get_value_from_db( + MongoCollections.BREAKDOWNS, service_id + ) + expected_oxp_responses = len(breakdown) if breakdown else 0 + if connection.get("partial_cleanup_requested") and ( + not expected_oxp_responses or len(oxp_response) < expected_oxp_responses + ): + time.sleep(PATCH_PROVISIONING_SETTLE_POLL_SECONDS) + continue + return connection + + time.sleep(PATCH_PROVISIONING_SETTLE_POLL_SECONDS) + + return connection + + def delete_connection(service_id): """ Delete connection order by ID. @@ -69,27 +112,6 @@ def delete_connection(service_id): return "Did not find connection", 404 logger.info(f"connection: {connection} {type(connection)}") - if connection.get("status") is None: - logger.error("Missing field: status is not in connection.") - connection["status"] = str(ConnectionStateMachine.State.DELETED) - elif connection["status"] == str(ConnectionStateMachine.State.UP): - connection, _ = connection_state_machine( - connection, ConnectionStateMachine.State.DELETED - ) - elif connection["status"] == str( - ConnectionStateMachine.State.UNDER_PROVISIONING - ): - connection, _ = connection_state_machine( - connection, ConnectionStateMachine.State.DOWN - ) - connection, _ = connection_state_machine( - connection, ConnectionStateMachine.State.DELETED - ) - else: - connection, _ = connection_state_machine( - connection, ConnectionStateMachine.State.DELETED - ) - logger.info(f"Removing connection: {service_id} {connection.get('status')}") remove_reason, remove_code = connection_handler.remove_connection( @@ -203,6 +225,10 @@ def place_connection(body): # used in lc_message_handler to count the oxp success response body["oxp_success_count"] = 0 + body["partial_cleanup_requested"] = False + body["provisioning_timeout_handled"] = False + body["provisioning_started_at"] = time.time() + body.pop("timeout_reason", None) db_instance.add_key_value_pair_to_db(MongoCollections.CONNECTIONS, service_id, body) @@ -294,7 +320,7 @@ def patch_connection(service_id, body=None): # noqa: E501 te_manager = current_app.te_manager # Assuming te_manager is accessible like this try: # Validate the new request body - te_manager.generate_traffic_matrix(connection_request=new_body) + traffic_matrix = te_manager.generate_traffic_matrix(connection_request=new_body) except Exception as request_err: logger.error("ERROR: invalid patch request: " + str(request_err)) error_code = getattr(request_err, "request_code", None) @@ -311,10 +337,17 @@ def patch_connection(service_id, body=None): # noqa: E501 f"Could not parse error code from patch validation error: {err_text}" ) return f"Error: patch request is not valid: {request_err}", error_code + if traffic_matrix is None: + return ( + "Error: patch request is not valid: " + "Request does not have a valid JSON or body is incomplete/incorrect", + 400, + ) logger.info("Modifying connection") - # Get roll back connection before removing connection + # Preserve the last successful request so rollback can recreate it cleanly. rollback_conn_body = copy.deepcopy(body) + rollback_conn_body.pop("rollback_performed_for_failed_patch", None) body.update(new_body) conn_status = ConnectionStateMachine.State.MODIFYING @@ -370,24 +403,49 @@ def patch_connection(service_id, body=None): # noqa: E501 body["status"] = str(conn_status) body["oxp_success_count"] = 0 body["oxp_response"] = {} + body["rollback_on_failure"] = True + body["rollback_request"] = rollback_conn_body + body["rollback_in_progress"] = False + body.pop("rollback_performed_for_failed_patch", None) db_instance.add_key_value_pair_to_db(MongoCollections.CONNECTIONS, service_id, body) reason, code = connection_handler.place_connection(current_app.te_manager, body) if code // 100 == 2: - # Service created successfully - # conn_status = ConnectionStateMachine.State.UNDER_PROVISIONING - # body, _ = connection_state_machine(body, conn_status) - # db_instance.add_key_value_pair_to_db( - # MongoCollections.CONNECTIONS, service_id, body - # ) - code = 201 - logger.info(f"Placed: ID: {service_id} reason='{reason}', code={code}") - response = { - "service_id": service_id, - "status": parse_conn_status(body["status"]), - "reason": reason, - } - return response, code + patched_conn = _wait_for_patch_provisioning_to_settle(service_id) + patched_status = patched_conn.get("status") if patched_conn else None + if patched_status == str(ConnectionStateMachine.State.UP): + if patched_conn.get("rollback_performed_for_failed_patch"): + response = { + "service_id": service_id, + "status": parse_conn_status(patched_status), + "reason": "Failure, rolled back to last successful L2VPN: " + "Patched connection provisioning failed", + } + return response, 400 + else: + code = 201 + logger.info(f"Placed: ID: {service_id} reason='{reason}', code={code}") + response = { + "service_id": service_id, + "status": parse_conn_status(patched_status), + "reason": reason, + } + return response, code + if patched_status == str(ConnectionStateMachine.State.UNDER_PROVISIONING): + code = 201 + logger.info( + f"Patch placement for {service_id} is still under provisioning; " + "returning current state." + ) + response = { + "service_id": service_id, + "status": parse_conn_status(patched_status), + "reason": reason, + } + return response, code + + reason = f"Patched connection provisioning failed: {patched_status}" + code = 400 logger.info( f"Modifying: Failed to place new connection. ID: {service_id} reason='{reason}', code={code}" @@ -403,6 +461,18 @@ def patch_connection(service_id, body=None): # noqa: E501 conn_request = rollback_conn_body conn_request["id"] = service_id + conn_request["status"] = str(ConnectionStateMachine.State.REQUESTED) + conn_request["oxp_success_count"] = 0 + conn_request["oxp_response"] = {} + conn_request["late_cleanup_domains"] = [] + conn_request["partial_cleanup_requested"] = False + conn_request["rollback_on_failure"] = False + conn_request["rollback_performed_for_failed_patch"] = True + conn_request.pop("rollback_request", None) + conn_request.pop("rollback_in_progress", None) + conn_request["provisioning_timeout_handled"] = False + conn_request["provisioning_started_at"] = time.time() + conn_request.pop("timeout_reason", None) db_instance.add_key_value_pair_to_db( MongoCollections.CONNECTIONS, service_id, conn_request ) @@ -413,18 +483,17 @@ def patch_connection(service_id, body=None): # noqa: E501 current_app.te_manager, conn_request ) if rollback_conn_code // 100 == 2: - # conn_status = ConnectionStateMachine.State.UNDER_PROVISIONING - # rollback_conn_body, _ = connection_state_machine( - # rollback_conn_body, conn_status - # ) - # db_instance.update_field_in_json( - # MongoCollections.CONNECTIONS, - # service_id, - # "status", - # str(conn_status), - # ) - # still return 400 to indicate the patch request is not successful, since we have already rolled back to original connection, which is under provisioning state, so the connection is not down and not failed. - rollback_conn_code = code + rollback_conn = _wait_for_patch_provisioning_to_settle(service_id) + rollback_status = rollback_conn.get("status") if rollback_conn else None + if rollback_status == str(ConnectionStateMachine.State.UP): + # Still return the patch failure code to indicate the requested + # modification failed, even though rollback succeeded. + rollback_conn_code = code + else: + rollback_conn_reason = ( + f"Rollback provisioning did not recover service: {rollback_status}" + ) + rollback_conn_code = 500 else: conn_status = ConnectionStateMachine.State.REJECTED body, _ = connection_state_machine(body, conn_status) @@ -434,7 +503,17 @@ def patch_connection(service_id, body=None): # noqa: E501 "status", str(conn_status), ) - rollback_conn_code = 500 + deadline = time.time() + ROLLBACK_SETTLE_TIMEOUT_SECONDS + while time.time() < deadline: + current_conn = db_instance.get_value_from_db( + MongoCollections.CONNECTIONS, service_id + ) + current_status = current_conn.get("status") if current_conn else None + if current_status != str( + ConnectionStateMachine.State.UNDER_PROVISIONING + ): + break + time.sleep(ROLLBACK_SETTLE_POLL_SECONDS) logger.info( f"Roll back connection result: ID: {service_id} reason='{rollback_conn_reason}', code={rollback_conn_code}" ) @@ -450,6 +529,7 @@ def patch_connection(service_id, body=None): # noqa: E501 rollback_conn_reason = f"Rollback failed: {e}" rollback_conn_code = 500 + response_code = code if rollback_conn_code // 100 == 2 else rollback_conn_code current_conn = db_instance.get_value_from_db( MongoCollections.CONNECTIONS, f"{service_id}" ) @@ -460,7 +540,7 @@ def patch_connection(service_id, body=None): # noqa: E501 current_conn.get("status", "") if current_conn else "" ), } - return response, rollback_conn_code + return response, response_code def get_archived_connections_by_id(service_id): diff --git a/sdx_controller/handlers/connection_handler.py b/sdx_controller/handlers/connection_handler.py index 7c33921b..5ed8e42b 100644 --- a/sdx_controller/handlers/connection_handler.py +++ b/sdx_controller/handlers/connection_handler.py @@ -1,5 +1,6 @@ import json import logging +import os import sys import time import traceback @@ -28,12 +29,105 @@ MongoCollections.SOLUTIONS = "solutions" +# Wait after publishing OXP delete requests so asynchronous delete callbacks +# have a chance to reach the controller before local cleanup continues. +DELETE_PROPAGATION_WAIT_SECONDS = int(os.getenv("DELETE_PROPAGATION_WAIT_SECONDS", "2")) + +# When deleting a connection that is still under provisioning, wait briefly for +# OXP create responses so cleanup can target only provisioned EVCs. +UNDER_PROVISIONING_DELETE_SETTLE_TIMEOUT_SECONDS = int( + os.getenv("UNDER_PROVISIONING_DELETE_SETTLE_TIMEOUT_SECONDS", "5") +) +UNDER_PROVISIONING_DELETE_SETTLE_POLL_SECONDS = int( + os.getenv("UNDER_PROVISIONING_DELETE_SETTLE_POLL_SECONDS", "1") +) + +# Retry endpoint validation when the latest topology has arrived but the TE +# graph has not caught up yet, avoiding transient "node not found" failures. +TOPOLOGY_SETTLE_RETRY_SECONDS = int(os.getenv("TOPOLOGY_SETTLE_RETRY_SECONDS", "5")) +TOPOLOGY_SETTLE_RETRY_POLL_SECONDS = int( + os.getenv("TOPOLOGY_SETTLE_RETRY_POLL_SECONDS", "1") +) + class ConnectionHandler: def __init__(self, db_instance): self.db_instance = db_instance self.parse_helper = ParseHelper() + def _wait_for_provisioning_to_settle(self, service_id, expected_domains): + deadline = time.time() + UNDER_PROVISIONING_DELETE_SETTLE_TIMEOUT_SECONDS + latest_connection = self.db_instance.get_value_from_db( + MongoCollections.CONNECTIONS, service_id + ) + + while time.time() < deadline: + latest_connection = self.db_instance.get_value_from_db( + MongoCollections.CONNECTIONS, service_id + ) + if not latest_connection: + return None + + latest_status = latest_connection.get("status") + oxp_response = latest_connection.get("oxp_response") or {} + + if latest_status != str(ConnectionStateMachine.State.UNDER_PROVISIONING): + return latest_connection + + if expected_domains and len(oxp_response) >= expected_domains: + return latest_connection + + time.sleep(UNDER_PROVISIONING_DELETE_SETTLE_POLL_SECONDS) + + return latest_connection + + def _get_oxp_service_id(self, oxp_response, domain_name): + if not isinstance(oxp_response, dict): + return None + + domain_response = oxp_response.get(domain_name) + if domain_response is None: + domain_response = oxp_response.get(self._get_domain_name(domain_name)) + if isinstance(domain_response, dict): + return domain_response.get("service_id") or domain_response.get("evc_id") + + if isinstance(domain_response, (list, tuple)) and len(domain_response) > 1: + response_payload = domain_response[1] + if isinstance(response_payload, dict): + return response_payload.get("service_id") or response_payload.get( + "evc_id" + ) + + return None + + def _get_domain_name(self, domain): + domain_name = self.parse_helper.find_domain_name(domain, ":") or f"{domain}" + return domain_name.split("__", 1)[0] + + def _generate_graph_and_traffic_matrix(self, te_manager, connection_request): + deadline = time.time() + TOPOLOGY_SETTLE_RETRY_SECONDS + + while True: + graph = te_manager.generate_graph_te() + if graph is None: + return None, None + + try: + traffic_matrix = te_manager.generate_traffic_matrix( + connection_request=connection_request + ) + return graph, traffic_matrix + except RequestValidationError as request_err: + if "not found in the graph" not in str(request_err): + raise + if time.time() >= deadline: + raise + logger.info( + f"Topology graph does not yet contain request endpoint for " + f"{connection_request.get('id')}; retrying." + ) + time.sleep(TOPOLOGY_SETTLE_RETRY_POLL_SECONDS) + def _process_port(self, connection_service_id, port_id, operation): port_connections_dict_json = self.db_instance.get_value_from_db( MongoCollections.PORTS, Constants.PORT_CONNECTIONS_DICT @@ -128,14 +222,19 @@ def _process_path_to_db(self, temanager, operation, connection_request): else: temanager._logger.warning(f"No endpoints: {connection_request}") - for ports in links: + for ports in links or []: s_port = ports["source"] d_port = ports["destination"] link = temanager.topology_manager._topology.get_link_by_port_id( s_port, d_port ) - temanager._logger.info(f"Links on path: {link.id} {s_port} {d_port}") simple_link = SimpleLink([s_port, d_port]).to_string() + if link is None: + temanager._logger.warning( + f"Link object not found for path ports: {simple_link}" + ) + else: + temanager._logger.info(f"Links on path: {link.id} {s_port} {d_port}") self._process_link_connection_dict( temanager, link_connections_dict, @@ -152,15 +251,26 @@ def _send_breakdown_to_lc(self, breakdown, operation, connection_request): connection_service_id = connection_request.get("id") + sent_domains = 0 + for domain, link in breakdown.items(): port_list = [] link_with_new_format = {} - for key in link.keys(): - if "uni_" in key and "port_id" in link[key]: + if not isinstance(link, dict): + logger.warning( + f"Skipping malformed breakdown segment for {domain}: {link}" + ) + continue + for key, link_endpoint in link.items(): + if ( + "uni_" in key + and isinstance(link_endpoint, dict) + and "port_id" in link_endpoint + ): port_list.append( { - "port_id": link[key]["port_id"], - "vlan_value": link[key].get("tag", {}).get("value"), + "port_id": link_endpoint["port_id"], + "vlan_value": link_endpoint.get("tag", {}).get("value"), } ) @@ -186,7 +296,7 @@ def _send_breakdown_to_lc(self, breakdown, operation, connection_request): ) # From "urn:ogf:network:sdx:topology:amlight.net", attempt to # extract a string like "amlight". - domain_name = self.parse_helper.find_domain_name(domain, ":") or f"{domain}" + domain_name = self._get_domain_name(domain) exchange_name = MessageQueueNames.CONNECTIONS logger.debug( @@ -196,6 +306,7 @@ def _send_breakdown_to_lc(self, breakdown, operation, connection_request): mq_link = { "operation": operation, "service_id": connection_service_id, + "breakdown_domain": domain, "link": link_with_new_format, } @@ -203,32 +314,38 @@ def _send_breakdown_to_lc(self, breakdown, operation, connection_request): logger.debug( f"Handling delete operation for connection {connection_request}" ) - oxp_response = None - evc_id = None - try: - oxp_response = connection_request.get("oxp_response") - # evc_id is the service_id in the OXP response, it differs from the service_id in the connection. - evc_id = ( - oxp_response.get(domain_name, [None, {}])[1].get("service_id") - if oxp_response - else None - ) - if not oxp_response or not evc_id: - return ( - "Connection does not have OXP response, cannot remove connection", - 404, - ) - mq_link["evc_id"] = evc_id - except Exception as e: - logger.error( - f"Error occurred while processing OXP response in delete: {e}" + oxp_response = connection_request.get("oxp_response") + + # evc_id is the service_id in the OXP response, it differs from the service_id in the connection. + evc_id = self._get_oxp_service_id( + oxp_response, domain + ) or self._get_oxp_service_id(oxp_response, domain_name) + + if not oxp_response or not evc_id: + logger.info( + f"Skipping delete for service {connection_service_id} in domain {domain_name}: missing evc_id" ) + continue + mq_link["evc_id"] = evc_id producer = TopicQueueProducer( timeout=5, exchange_name=exchange_name, routing_key=domain_name ) producer.call(json.dumps(mq_link)) producer.stop_keep_alive() + sent_domains += 1 + + if operation == "delete" and sent_domains == 0: + return ( + "No provisioned OXP breakdowns found; connection removed locally", + 200, + ) + + if operation == "delete" and DELETE_PROPAGATION_WAIT_SECONDS > 0: + logger.info( + f"Waiting {DELETE_PROPAGATION_WAIT_SECONDS}s for delete propagation." + ) + time.sleep(DELETE_PROPAGATION_WAIT_SECONDS) # We will get to this point only if all the previous steps # leading up to this point were successful. @@ -236,6 +353,60 @@ def _send_breakdown_to_lc(self, breakdown, operation, connection_request): "Connection deleted" if operation == "delete" else "Connection published" ), 201 + def cleanup_partial_connection( + self, te_manager, service_id, connection_request + ) -> Tuple[str, int]: + try: + te_manager.delete_connection(service_id) + except Exception as e: + logger.info( + f"Failed to release local connection resources for {service_id}: {e}" + ) + + breakdown = self.db_instance.get_value_from_db( + MongoCollections.BREAKDOWNS, service_id + ) + if not breakdown: + return "Did not find breakdown, cannot clean up connection", 404 + + status, code = self._send_breakdown_to_lc( + breakdown, "delete", connection_request + ) + try: + self._process_path_to_db( + te_manager, operation="delete", connection_request=connection_request + ) + topology_db_update(self.db_instance, te_manager) + except Exception as e: + logger.info( + f"Failed to update local state while cleaning up {service_id}: {e}" + ) + + return status, code + + def cleanup_partial_connection_domain( + self, service_id, connection_request, domain_name + ) -> Tuple[str, int]: + breakdown = self.db_instance.get_value_from_db( + MongoCollections.BREAKDOWNS, service_id + ) + if not breakdown: + return "Did not find breakdown, cannot clean up connection", 404 + + domain_breakdown = None + for domain, segment in breakdown.items(): + parsed_domain = self._get_domain_name(domain) + if parsed_domain == domain_name: + domain_breakdown = {domain: segment} + break + + if not domain_breakdown: + return f"Did not find breakdown for domain {domain_name}", 404 + + return self._send_breakdown_to_lc( + domain_breakdown, "delete", connection_request + ) + def place_connection( self, te_manager: TEManager, connection_request: dict ) -> Tuple[str, int]: @@ -252,13 +423,12 @@ def place_connection( # for num, val in enumerate(te_manager.get_topology_map().values()): # logger.debug(f"TE topology #{num}: {val}") - graph = te_manager.generate_graph_te() - if graph is None: - return "No SDX topology found", 424 try: - traffic_matrix = te_manager.generate_traffic_matrix( - connection_request=connection_request + graph, traffic_matrix = self._generate_graph_and_traffic_matrix( + te_manager, connection_request ) + if graph is None: + return "No SDX topology found", 424 except RequestValidationError as request_err: err = traceback.format_exc().replace("\n", ", ") logger.error( @@ -422,7 +592,7 @@ def archive_connection(self, service_id, reason) -> None: logger.debug(f"Archived connection: {service_id}") def remove_connection( - self, te_manager, service_id, archive_reason + self, te_manager, service_id, archive_reason, archive=True ) -> Tuple[str, int]: connection_request = self.db_instance.get_value_from_db( MongoCollections.CONNECTIONS, service_id @@ -431,45 +601,68 @@ def remove_connection( if not connection_request: return "Did not find connection request, cannot remove connection", 404 connection_status = connection_request.get("status") - if ( - (connection_status != str(ConnectionStateMachine.State.UP)) - and (connection_status != str(ConnectionStateMachine.State.MODIFYING)) - and ( - connection_status - != str(ConnectionStateMachine.State.UNDER_PROVISIONING) + removable_statuses = { + str(ConnectionStateMachine.State.UP), + str(ConnectionStateMachine.State.MODIFYING), + str(ConnectionStateMachine.State.UNDER_PROVISIONING), + str(ConnectionStateMachine.State.DOWN), + str(ConnectionStateMachine.State.ERROR), + str(ConnectionStateMachine.State.RECOVERING), + str(ConnectionStateMachine.State.REQUESTED), + str(ConnectionStateMachine.State.REJECTED), + } + normalized_status = ( + connection_status.upper() if isinstance(connection_status, str) else None + ) + normalized_removable_statuses = { + status.upper() for status in removable_statuses if isinstance(status, str) + } + if normalized_status not in normalized_removable_statuses: + logger.info( + f"Connection {service_id} {connection_status} is not in a removable state." ) + return f"Connection {connection_status} is not removable", 409 + + breakdown = self.db_instance.get_value_from_db( + MongoCollections.BREAKDOWNS, service_id + ) + + if ( + connection_status == str(ConnectionStateMachine.State.UNDER_PROVISIONING) + and breakdown ): logger.info( - f"Connection {service_id} {connection_status} is not {str(ConnectionStateMachine.State.UP)}, cannot remove connection." + f"Waiting for in-flight provisioning to settle before removing {service_id}." ) - return "Connection is not UP, Archive", 404 + connection_request = self._wait_for_provisioning_to_settle( + service_id, len(breakdown) + ) + if not connection_request: + return "Did not find connection request, cannot remove connection", 404 + connection_status = connection_request.get("status") try: te_manager.delete_connection(service_id) - breakdown = self.db_instance.get_value_from_db( - MongoCollections.BREAKDOWNS, service_id + except Exception as e: + logger.info( + f"Failed to release local connection resources for {service_id}: {e}" ) - if not breakdown: - return "Did not find breakdown, cannot remove connection", 404 + if not breakdown: + if archive: + self.archive_connection(service_id, archive_reason) + try: + topology_db_update(self.db_instance, te_manager) + except Exception as e: + logger.info( + f"Failed to update local topology state for {service_id}: {e}" + ) + return "Connection removed locally", 200 + + try: status, code = self._send_breakdown_to_lc( breakdown, "delete", connection_request ) - if code // 100 != 2: - logger.error( - f"Could not publish delete breakdown for {service_id}: " - f"reason='{status}', code={code}" - ) - return status, code - self._process_path_to_db( - te_manager, operation="delete", connection_request=connection_request - ) - self.db_instance.delete_one_entry(MongoCollections.BREAKDOWNS, service_id) - self.archive_connection(service_id, archive_reason) - logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}") - # update topology in DB with updated states (bandwidth and available vlan pool) - topology_db_update(self.db_instance, te_manager) - return status, code except Exception as e: exc_type, exc_value, exc_traceback = sys.exc_info() logging.error( @@ -478,6 +671,31 @@ def remove_connection( logger.error(f"Error when removing breakdown: {e}") return f"Error when removing breakdown: {e}", 400 + if code // 100 != 2: + logger.error( + f"Could not publish delete breakdown for {service_id}: " + f"reason='{status}', code={code}" + ) + return status, code + + try: + self._process_path_to_db( + te_manager, operation="delete", connection_request=connection_request + ) + except Exception as e: + logger.info(f"Failed to release path state for {service_id}: {e}") + + self.db_instance.delete_one_entry(MongoCollections.BREAKDOWNS, service_id) + if archive: + self.archive_connection(service_id, archive_reason) + logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}") + try: + # update topology in DB with updated states (bandwidth and available vlan pool) + topology_db_update(self.db_instance, te_manager) + except Exception as e: + logger.info(f"Failed to update local topology state for {service_id}: {e}") + return status, code + def handle_link_removal(self, te_manager, removed_links): logger.debug("Handling connections that contain removed links.") failed_links = [] @@ -497,6 +715,7 @@ def handle_link_failure(self, te_manager, failed_links): return link_connections_dict = json.loads(link_connections_dict) + processed_service_ids = set() for link in failed_links: logger.info(f"Handling link failure on {link['id']}") @@ -513,8 +732,14 @@ def handle_link_failure(self, te_manager, failed_links): if simple_link in link_connections_dict: logger.debug("Found failed link record!") - service_ids = link_connections_dict[simple_link] - for index, service_id in enumerate(service_ids): + service_ids = list(link_connections_dict[simple_link]) + for service_id in service_ids: + if service_id in processed_service_ids: + logger.debug( + f"Skipping already-processed failed service {service_id}" + ) + continue + processed_service_ids.add(service_id) logger.info( f"Connection {service_id} affected by link {link['id']}" ) @@ -539,7 +764,10 @@ def handle_link_failure(self, te_manager, failed_links): f"Removing connection: {service_id} {connection.get('status')}" ) _, code = self.remove_connection( - te_manager, connection["id"], archive_reason="Failure" + te_manager, + connection["id"], + archive_reason="Failure", + archive=False, ) if code // 100 != 2: logger.info( @@ -561,6 +789,10 @@ def handle_link_failure(self, te_manager, failed_links): ) connection["oxp_success_count"] = 0 connection["oxp_response"] = {} + connection["partial_cleanup_requested"] = False + connection["provisioning_timeout_handled"] = False + connection["provisioning_started_at"] = time.time() + connection.pop("timeout_reason", None) self.db_instance.add_key_value_pair_to_db( MongoCollections.CONNECTIONS, service_id, connection ) @@ -568,30 +800,23 @@ def handle_link_failure(self, te_manager, failed_links): if code // 100 == 2: # Service created successfully - # conn_status = ConnectionStateMachine.State.UNDER_PROVISIONING - # connection, _ = connection_state_machine( - # connection, conn_status - # ) - # self.db_instance.update_field_in_json( - # MongoCollections.CONNECTIONS, - # service_id, - # "status", - # str(conn_status), - # ) logger.info( f"link failure rerouting: place_connection succeeds: ID: {service_id} connection='{connection}'" ) code = 201 else: + logger.info( + f"Recovery placement failed for {service_id}; keeping failed recovery state active." + ) + self.db_instance.delete_one_entry( + MongoCollections.BREAKDOWNS, service_id + ) conn_status = ConnectionStateMachine.State.ERROR connection, _ = connection_state_machine( connection, conn_status ) - self.db_instance.update_field_in_json( - MongoCollections.CONNECTIONS, - service_id, - "status", - str(conn_status), + self.db_instance.add_key_value_pair_to_db( + MongoCollections.CONNECTIONS, service_id, connection ) _reason = ( "place_connection failed during link failure rerouting" diff --git a/sdx_controller/handlers/lc_message_handler.py b/sdx_controller/handlers/lc_message_handler.py index 54821d71..c0fccd4c 100644 --- a/sdx_controller/handlers/lc_message_handler.py +++ b/sdx_controller/handlers/lc_message_handler.py @@ -1,5 +1,7 @@ import json import logging +import time +from copy import deepcopy from sdx_datamodel.connection_sm import ConnectionStateMachine from sdx_datamodel.constants import Constants, DomainStatus, MongoCollections @@ -20,6 +22,149 @@ def __init__(self, db_instance, te_manager): self.parse_helper = ParseHelper() self.connection_handler = ConnectionHandler(db_instance) + def _failed_patch_cleanup_is_complete(self, connection, breakdown): + if not connection.get("rollback_on_failure"): + return False + if connection.get("rollback_in_progress"): + return False + if connection.get("status") != str(ConnectionStateMachine.State.DOWN): + return False + if not connection.get("partial_cleanup_requested"): + return False + if not isinstance(connection.get("rollback_request"), dict): + return False + + oxp_response = connection.get("oxp_response") or {} + return bool(breakdown) and len(oxp_response) >= len(breakdown) + + def _rollback_failed_patch(self, service_id, connection): + rollback_request = deepcopy(connection.get("rollback_request") or {}) + if not rollback_request: + return + + logger.info(f"Rolling back failed PATCH for {service_id}") + self.db_instance.update_field_in_json( + MongoCollections.CONNECTIONS, + service_id, + "rollback_in_progress", + True, + ) + + rollback_request["id"] = service_id + rollback_request["status"] = str(ConnectionStateMachine.State.REQUESTED) + rollback_request["oxp_success_count"] = 0 + rollback_request["oxp_response"] = {} + rollback_request["late_cleanup_domains"] = [] + rollback_request["partial_cleanup_requested"] = False + rollback_request["rollback_on_failure"] = False + rollback_request["rollback_performed_for_failed_patch"] = True + rollback_request.pop("rollback_request", None) + rollback_request.pop("rollback_in_progress", None) + rollback_request["provisioning_timeout_handled"] = False + rollback_request["provisioning_started_at"] = time.time() + rollback_request.pop("timeout_reason", None) + + self.db_instance.add_key_value_pair_to_db( + MongoCollections.CONNECTIONS, service_id, rollback_request + ) + rollback_reason, rollback_code = self.connection_handler.place_connection( + self.te_manager, rollback_request + ) + logger.info( + f"Async PATCH rollback result for {service_id}: " + f"reason='{rollback_reason}', code={rollback_code}" + ) + + def _previous_vlan_ranges_by_port(self, topology): + vlan_ranges = {} + if not topology: + return vlan_ranges + + for node in topology.get("nodes", []): + for port in node.get("ports", []): + port_id = port.get("id") + if not port_id: + continue + + services = port.get("services") or {} + for service_name in ("l2vpn-ptp", "l2vpn_ptp"): + service = services.get(service_name) + if service and service.get("vlan_range"): + vlan_ranges[port_id] = deepcopy(service["vlan_range"]) + break + + return vlan_ranges + + def _is_valid_vlan_range(self, vlan_range): + if not vlan_range or not isinstance(vlan_range, list): + return False + + for item in vlan_range: + parsed_item = item + if isinstance(parsed_item, str): + parsed_item = [ + int(vlan) for vlan in parsed_item.split("-") if vlan.isdigit() + ] + if len(parsed_item) == 1: + parsed_item = parsed_item[0] + + if isinstance(parsed_item, int): + if parsed_item < 0 or parsed_item > 4095: + return False + continue + + if not isinstance(parsed_item, list) or len(parsed_item) != 2: + return False + + if not all(isinstance(vlan, int) for vlan in parsed_item): + return False + + if ( + parsed_item[0] > parsed_item[1] + or parsed_item[0] < 0 + or parsed_item[1] < 0 + or parsed_item[0] > 4095 + or parsed_item[1] > 4095 + ): + return False + + return True + + def _sanitize_vlan_ranges(self, topology_update, latest_topo): + previous_vlan_ranges = self._previous_vlan_ranges_by_port(latest_topo) + + for node in topology_update.get("nodes", []): + for port in node.get("ports", []): + port_id = port.get("id") + services = port.get("services") or {} + previous_vlan_range = previous_vlan_ranges.get(port_id) + + for service_name in ("l2vpn-ptp", "l2vpn_ptp"): + service = services.get(service_name) + if not service or "vlan_range" not in service: + continue + + vlan_range = service.get("vlan_range") + if self._is_valid_vlan_range(vlan_range): + continue + + if previous_vlan_range: + logger.warning( + "Ignoring invalid VLAN range %s on port %s; keeping %s", + vlan_range, + port_id, + previous_vlan_range, + ) + service["vlan_range"] = deepcopy(previous_vlan_range) + else: + logger.warning( + "Ignoring invalid VLAN range %s on port %s; using the " + "default valid range", + vlan_range, + port_id, + ) + service["vlan_range"] = [[1, 4095]] + def process_lc_json_msg( self, msg, @@ -97,26 +242,66 @@ def process_lc_json_msg( oxp_number = len(breakdown) oxp_success_count = connection.get("oxp_success_count", 0) lc_domain = msg_json.get("lc_domain") + response_domain = msg_json.get("breakdown_domain") or lc_domain oxp_response_code = msg_json.get("oxp_response_code") oxp_response_msg = msg_json.get("oxp_response") + operation = msg_json.get("operation") oxp_response = connection.get("oxp_response") if not oxp_response: oxp_response = {} - oxp_response[lc_domain] = (oxp_response_code, oxp_response_msg) + + existing_domain_response = oxp_response.get(response_domain) + if ( + operation == "delete" + and isinstance(existing_domain_response, (list, tuple)) + and len(existing_domain_response) > 1 + and isinstance(existing_domain_response[1], dict) + and existing_domain_response[1].get("service_id") + ): + preserved_payload = dict(existing_domain_response[1]) + if isinstance(oxp_response_msg, dict): + preserved_payload.update(oxp_response_msg) + oxp_response[response_domain] = [oxp_response_code, preserved_payload] + else: + oxp_response[response_domain] = [oxp_response_code, oxp_response_msg] connection["oxp_response"] = oxp_response + partial_cleanup_requested = connection.get( + "partial_cleanup_requested", False + ) + late_cleanup_domains = connection.get("late_cleanup_domains", []) if oxp_response_code // 100 == 2: - if msg_json.get("operation") != "delete": - oxp_success_count += 1 - connection["oxp_success_count"] = oxp_success_count - logger.info( - f"Update oxp_success_count: {oxp_success_count}; oxp_number: {oxp_number}" - ) - if oxp_success_count == oxp_number: - conn_status = ConnectionStateMachine.State.UP - connection, _ = connection_state_machine( - connection, conn_status + if operation != "delete": + if partial_cleanup_requested: + if lc_domain not in late_cleanup_domains: + cleanup_status, cleanup_code = ( + self.connection_handler.cleanup_partial_connection_domain( + service_id, connection, lc_domain + ) + ) + logger.info( + f"Late partial cleanup result for {service_id} in {lc_domain}: {cleanup_status}, code={cleanup_code}" + ) + late_cleanup_domains.append(lc_domain) + connection["late_cleanup_domains"] = late_cleanup_domains + else: + oxp_success_count += 1 + connection["oxp_success_count"] = oxp_success_count + logger.info( + f"Update oxp_success_count: {oxp_success_count}; oxp_number: {oxp_number}" ) + if oxp_success_count == oxp_number: + if connection.get("status") and ( + connection.get("status") + == str(ConnectionStateMachine.State.RECOVERING) + ): + connection, _ = connection_state_machine( + connection, + ConnectionStateMachine.State.UNDER_PROVISIONING, + ) + connection, _ = connection_state_machine( + connection, ConnectionStateMachine.State.UP + ) else: if connection.get("status") and ( connection.get("status") @@ -124,31 +309,43 @@ def process_lc_json_msg( or connection.get("status") == str(ConnectionStateMachine.State.UNDER_PROVISIONING) ): - conn_status = ConnectionStateMachine.State.DOWN - connection, _ = connection_state_machine(connection, conn_status) + connection, _ = connection_state_machine( + connection, ConnectionStateMachine.State.DOWN + ) + if operation == "post" and not partial_cleanup_requested: + connection["partial_cleanup_requested"] = True + cleanup_status, cleanup_code = ( + self.connection_handler.cleanup_partial_connection( + self.te_manager, service_id, connection + ) + ) + logger.info( + f"Partial cleanup result for {service_id}: {cleanup_status}, code={cleanup_code}" + ) # ToDo: eg: if 3 oxps in the breakdowns: (1) all up: up (2) parital down: remove_connection() # release successful oxp circuits if some are down: remove_connection() (3) count the responses # to finalize the status of the connection. - self.db_instance.update_field_in_json( - MongoCollections.CONNECTIONS, - service_id, + for field_name in ( "status", - str(conn_status), - ) - self.db_instance.update_field_in_json( - MongoCollections.CONNECTIONS, - service_id, "oxp_response", - oxp_response, - ) - self.db_instance.update_field_in_json( - MongoCollections.CONNECTIONS, - service_id, "oxp_success_count", - oxp_success_count, - ) + "partial_cleanup_requested", + "late_cleanup_domains", + "rollback_on_failure", + "rollback_request", + "rollback_in_progress", + ): + if field_name in connection: + self.db_instance.update_field_in_json( + MongoCollections.CONNECTIONS, + service_id, + field_name, + connection.get(field_name), + ) logger.info("Connection updated: " + str(connection)) + if self._failed_patch_cleanup_is_complete(connection, breakdown): + self._rollback_failed_patch(service_id, connection) return # topology message RPC from OXP: no exchange name is defined. @@ -157,6 +354,7 @@ def process_lc_json_msg( domain_name = self.parse_helper.find_domain_name(msg_id, ":") msg_json["domain_name"] = domain_name + self._sanitize_vlan_ranges(msg_json, latest_topo) db_msg_id = str(msg_id) + "-" + str(msg_version) # add message to db diff --git a/sdx_controller/messaging/rpc_queue_consumer.py b/sdx_controller/messaging/rpc_queue_consumer.py index a564d710..c00c99e5 100644 --- a/sdx_controller/messaging/rpc_queue_consumer.py +++ b/sdx_controller/messaging/rpc_queue_consumer.py @@ -261,7 +261,7 @@ def start_sdx_consumer(self, thread_queue, db_instance): vlan_tags_table = self.te_manager.vlan_tags_table for domain, segment in domain_breakdown.items(): logger.debug(f"domain:{domain};segment:{segment}") - domain_table = vlan_tags_table.get(domain) + domain_table = vlan_tags_table.get(domain.split("__", 1)[0]) uni_a = segment.get("uni_a") vlan_table = domain_table.get(uni_a.get("port_id")) vlan_table[uni_a.get("tag").get("value")] = service_id