Merge pull request #521 from atlanticwave-sdx/recover-error-connection

Recover error connection when SDX domain becomes UP

Author: congwang09

sdx_controller/handlers/connection_handler.py

@@ -55,6 +55,229 @@ def __init__(self, db_instance):
         self.db_instance = db_instance
         self.parse_helper = ParseHelper()
 
+    def save_error_connection(self, connection, reason, details=None) -> None:
+        if not connection:
+            return
+
+        service_id = connection.get("id")
+        if not service_id:
+            return
+
+        connection_to_save = dict(connection)
+        connection_to_save["status"] = str(ConnectionStateMachine.State.ERROR)
+
+        error_record = {
+            "connection": connection_to_save,
+            "reason": reason,
+            "updated_at": str(int(time.time())),
+        }
+        if details is not None:
+            error_record["details"] = details
+
+        self.db_instance.add_key_value_pair_to_db(
+            MongoCollections.ERROR_CONNECTIONS,
+            service_id,
+            error_record,
+        )
+        logger.info(f"Saved connection {service_id} to error collection")
+
+    def delete_error_connection(self, service_id) -> None:
+        self.db_instance.delete_one_entry(
+            MongoCollections.ERROR_CONNECTIONS, service_id
+        )
+
+    def get_latest_archived_connection(self, service_id):
+        historical_connections = self.db_instance.get_value_from_db(
+            MongoCollections.HISTORICAL_CONNECTIONS, service_id
+        )
+        if not historical_connections:
+            return None, None
+
+        latest_record = historical_connections[-1]
+        if not latest_record:
+            return None, None
+
+        latest_timestamp = next(reversed(latest_record))
+        latest_entry = latest_record.get(latest_timestamp, {})
+        return latest_entry.get("connection"), latest_entry.get("reason")
+
+    def save_error_connection_from_archive(
+        self, service_id, reason, details=None, expected_archive_reason=None
+    ) -> bool:
+        connection, archive_reason = self.get_latest_archived_connection(service_id)
+        if not connection:
+            return False
+
+        if (
+            expected_archive_reason is not None
+            and archive_reason != expected_archive_reason
+        ):
+            return False
+
+        self.save_error_connection(connection, reason=reason, details=details)
+        return True
+
+    def retry_error_connections(self, te_manager) -> None:
+        error_connections = self.db_instance.get_all_entries_in_collection(
+            MongoCollections.ERROR_CONNECTIONS
+        )
+
+        for error_connection in error_connections:
+            service_id = next(iter(error_connection))
+            error_record = error_connection.get(service_id, {})
+            connection = error_record.get("connection", error_record)
+
+            if not connection:
+                continue
+
+            active_connection = self.db_instance.get_value_from_db(
+                MongoCollections.CONNECTIONS, service_id
+            )
+            if active_connection:
+                active_status = active_connection.get("status")
+                if active_status in {
+                    str(ConnectionStateMachine.State.RECOVERING),
+                    str(ConnectionStateMachine.State.UNDER_PROVISIONING),
+                    str(ConnectionStateMachine.State.UP),
+                    str(ConnectionStateMachine.State.MODIFYING),
+                }:
+                    logger.info(
+                        f"Skipping retry for {service_id}; active status is {active_status}"
+                    )
+                    continue
+                connection = active_connection
+
+            connection_to_retry = dict(connection)
+            connection_to_retry["status"] = str(ConnectionStateMachine.State.RECOVERING)
+            connection_to_retry["oxp_success_count"] = 0
+            connection_to_retry.pop("oxp_response", None)
+
+            self.db_instance.add_key_value_pair_to_db(
+                MongoCollections.CONNECTIONS, service_id, connection_to_retry
+            )
+
+            reason, code = self.place_connection(te_manager, connection_to_retry)
+            logger.info(
+                f"Retry error connection result: ID: {service_id} reason='{reason}', code={code}"
+            )
+
+            if code // 100 != 2:
+                connection_to_retry["status"] = str(ConnectionStateMachine.State.ERROR)
+                self.db_instance.add_key_value_pair_to_db(
+                    MongoCollections.CONNECTIONS, service_id, connection_to_retry
+                )
+                self.save_error_connection(
+                    connection_to_retry,
+                    reason="Retry failed",
+                    details={"place_connection_reason": reason, "code": code},
+                )
+
+    def _response_is_success(self, oxp_response, domain):
+        if not isinstance(oxp_response, dict):
+            return False
+
+        domain_name = self._get_domain_name(domain)
+        for response_key in (domain, domain_name):
+            response = oxp_response.get(response_key)
+            if response is None:
+                continue
+            if isinstance(response, (list, tuple)) and response:
+                status_code = response[0]
+            elif isinstance(response, dict):
+                status_code = response.get("status_code") or response.get("code")
+            else:
+                continue
+            try:
+                return int(status_code) // 100 == 2
+            except (TypeError, ValueError):
+                return False
+
+        return False
+
+    def _count_successful_oxp_responses(self, oxp_response, breakdown):
+        return sum(
+            1 for domain in breakdown if self._response_is_success(oxp_response, domain)
+        )
+
+    def recover_domain_connections(self, recovered_domain):
+        """
+        Re-publish missing OXP breakdowns when an SDX-LC comes back online.
+        """
+        connections = self.db_instance.get_all_entries_in_collection(
+            MongoCollections.CONNECTIONS
+        )
+        recovered = 0
+
+        for connection_entry in connections:
+            service_id = next(iter(connection_entry))
+            connection = self.db_instance.get_value_from_db(
+                MongoCollections.CONNECTIONS, service_id
+            )
+            if not connection:
+                continue
+
+            breakdown = self.db_instance.get_value_from_db(
+                MongoCollections.BREAKDOWNS, service_id
+            )
+            if not breakdown:
+                continue
+
+            if recovered_domain not in {
+                self._get_domain_name(domain) for domain in breakdown
+            }:
+                continue
+
+            connection_status = connection.get("status")
+            if connection_status == str(ConnectionStateMachine.State.UP):
+                continue
+
+            domain_breakdown = {}
+            oxp_response = connection.get("oxp_response") or {}
+            if connection.get("partial_cleanup_requested"):
+                domain_breakdown = breakdown
+                connection["oxp_response"] = {}
+                connection["oxp_success_count"] = 0
+                connection["partial_cleanup_requested"] = False
+                connection["late_cleanup_domains"] = []
+            else:
+                for domain, segment in breakdown.items():
+                    if self._response_is_success(oxp_response, domain):
+                        continue
+                    domain_breakdown[domain] = segment
+                connection["oxp_success_count"] = self._count_successful_oxp_responses(
+                    oxp_response, breakdown
+                )
+
+            if not domain_breakdown:
+                continue
+
+            logger.info(
+                f"Recovering service {service_id} for recovered LC domain {recovered_domain}"
+            )
+            connection["provisioning_timeout_handled"] = False
+            connection["provisioning_started_at"] = time.time()
+            connection.pop("timeout_reason", None)
+            if connection.get("status") in {
+                str(ConnectionStateMachine.State.DOWN),
+                str(ConnectionStateMachine.State.ERROR),
+            }:
+                connection["status"] = str(ConnectionStateMachine.State.RECOVERING)
+
+            self.db_instance.add_key_value_pair_to_db(
+                MongoCollections.CONNECTIONS, service_id, connection
+            )
+
+            _reason, code = self._send_breakdown_to_lc(
+                domain_breakdown, "post", connection
+            )
+            logger.info(
+                f"LC recovery publish result: ID: {service_id} reason='{_reason}', code={code}"
+            )
+            if code // 100 == 2:
+                recovered += 1
+
+        return recovered
+
     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(
@@ -770,11 +993,21 @@ def handle_link_failure(self, te_manager, failed_links):
                             archive=False,
                         )
                         if code // 100 != 2:
+                            self.save_error_connection(
+                                connection,
+                                reason="Failure handling delete failed",
+                                details={"code": code},
+                            )
                             logger.info(
                                 f"Do not remove connection, may be already removed: {connection['id']}, code: {code}"
                             )
                             continue
                     except Exception as err:
+                        self.save_error_connection(
+                            connection,
+                            reason="Failure handling delete raised exception",
+                            details={"error": str(err)},
+                        )
                         logger.info(
                             f"Encountered error when deleting connection: {err}"
                         )
@@ -822,7 +1055,11 @@ def handle_link_failure(self, te_manager, failed_links):
                             "place_connection failed during link failure rerouting"
                         )
                         code = 400
-
+                        self.save_error_connection(
+                            connection,
+                            reason="Failure handling recovery failed",
+                            details={"place_connection_reason": _reason, "code": code},
+                        )
                     logger.info(
                         f"place_connection result: ID: {service_id} reason='{_reason}', code={code}"
                     )

sdx_controller/handlers/lc_message_handler.py

@@ -220,6 +220,9 @@ def process_lc_json_msg(
         elif msg_json.get("msg_type") and msg_json["msg_type"] == "oxp_conn_response":
             logger.info("Received OXP connection response.")
             service_id = msg_json.get("service_id")
+            operation = msg_json.get("operation")
+            oxp_response_code = msg_json.get("oxp_response_code")
+            oxp_response_msg = msg_json.get("oxp_response")
 
             if not service_id:
                 return
@@ -229,6 +232,20 @@ def process_lc_json_msg(
             )
 
             if not connection:
+                if operation == "delete" and oxp_response_code // 100 != 2:
+                    saved = self.connection_handler.save_error_connection_from_archive(
+                        service_id,
+                        reason="Delete failed during failure handling",
+                        details={
+                            "oxp_response_code": oxp_response_code,
+                            "oxp_response": oxp_response_msg,
+                        },
+                        expected_archive_reason="Failure",
+                    )
+                    if saved:
+                        logger.info(
+                            f"Saved archived connection {service_id} to error collection after delete failure"
+                        )
                 return
 
             breakdown = self.db_instance.get_value_from_db(
@@ -243,9 +260,6 @@ def process_lc_json_msg(
             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 = {}
@@ -302,7 +316,17 @@ def process_lc_json_msg(
                             connection, _ = connection_state_machine(
                                 connection, ConnectionStateMachine.State.UP
                             )
+                            self.connection_handler.delete_error_connection(service_id)
             else:
+                if operation == "delete":
+                    self.connection_handler.save_error_connection(
+                        connection,
+                        reason="Delete failed during failure handling",
+                        details={
+                            "oxp_response_code": oxp_response_code,
+                            "oxp_response": oxp_response_msg,
+                        },
+                    )
                 if connection.get("status") and (
                     connection.get("status")
                     == str(ConnectionStateMachine.State.MODIFYING)
@@ -364,8 +388,24 @@ def process_lc_json_msg(
         logger.info("Save to database complete.")
         logger.info("message ID:" + str(db_msg_id))
 
+        previous_domain_status = domain_dict.get(domain_name)
+        is_new_domain = domain_name not in domain_dict
+        is_recovered_domain = (
+            previous_domain_status is not None
+            and previous_domain_status != DomainStatus.UP
+        )
+
         # Update existing topology
         if domain_name in domain_dict:
+            if is_recovered_domain:
+                logger.info(
+                    f"Domain {domain_name} status changed from "
+                    f"{previous_domain_status} to {DomainStatus.UP}"
+                )
+                domain_dict[domain_name] = DomainStatus.UP
+                self.db_instance.add_key_value_pair_to_db(
+                    MongoCollections.DOMAINS, Constants.DOMAIN_DICT, domain_dict
+                )
             logger.info("Updating topo")
             logger.debug(msg_json)
             (
@@ -420,4 +460,6 @@ def process_lc_json_msg(
         self.db_instance.add_key_value_pair_to_db(
             MongoCollections.TOPOLOGIES, Constants.LATEST_TOPOLOGY, latest_topo
         )
+        if is_new_domain or is_recovered_domain:
+            self.connection_handler.retry_error_connections(self.te_manager)
         logger.info("Save to database complete.")