9530886246

Committed 15 Jun 2024 09:07PM UTC coverage: 52.322% (-0.1%) from 52.459%

Build # 9530886246

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #293 from atlanticwave-sdx/update-failure-handling

Update failure handling

Run Details

301 of 551 branches covered (54.63%)

Branch coverage included in aggregate %.

8 of 17 new or added lines in 4 files covered. (47.06%)

1 existing line in 1 file now uncovered.

758 of 1473 relevant lines covered (51.46%)

2.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.81

/sdx_controller/handlers/connection_handler.py

import json
import logging
from typing import Tuple

from sdx_pce.load_balancing.te_solver import TESolver
from sdx_pce.topology.temanager import TEManager

from sdx_controller.messaging.topic_queue_producer import TopicQueueProducer
from sdx_controller.models.simple_link import SimpleLink
from sdx_controller.utils.parse_helper import ParseHelper

logger = logging.getLogger(__name__)
logging.getLogger("pika").setLevel(logging.WARNING)


class ConnectionHandler:
    def __init__(self, db_instance):
        self.db_instance = db_instance
        self.parse_helper = ParseHelper()

    def _send_breakdown_to_lc(self, breakdown, operation, connection_request):
        logger.debug(f"-- BREAKDOWN: {json.dumps(breakdown)}")

        if breakdown is None:
            return "Could not break down the solution", 400

        link_connections_dict_json = (
            self.db_instance.read_from_db("links", "link_connections_dict")[
                "link_connections_dict"
            ]
            if self.db_instance.read_from_db("links", "link_connections_dict")
            else None
        )

        if link_connections_dict_json:
            link_connections_dict = json.loads(link_connections_dict_json)
        else:
            link_connections_dict = {}

        for domain, link in breakdown.items():
            port_list = []
            for key in link.keys():
                if "uni_" in key and "port_id" in link[key]:
                    port_list.append(link[key]["port_id"])

            if port_list:
                simple_link = SimpleLink(port_list).to_string()

                if simple_link not in link_connections_dict:
                    link_connections_dict[simple_link] = []

                if connection_request not in link_connections_dict[simple_link]:
                    link_connections_dict[simple_link].append(connection_request)

                self.db_instance.add_key_value_pair_to_db(
                    "links", "link_connections_dict", json.dumps(link_connections_dict)
                )

            logger.debug(f"Attempting to publish domain: {domain}, link: {link}")

            # 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}"
            exchange_name = "connection"

            logger.debug(
                f"Doing '{operation}' operation for '{link}' with exchange_name: {exchange_name}, "
                f"routing_key: {domain_name}"
            )
            mq_link = {"operation": operation, "link": link}
            producer = TopicQueueProducer(
                timeout=5, exchange_name=exchange_name, routing_key=domain_name
            )
            producer.call(json.dumps(mq_link))
            producer.stop_keep_alive()

        # We will get to this point only if all the previous steps
        # leading up to this point were successful.
        return "Connection published", 200

    def place_connection(
        self, te_manager: TEManager, connection_request: dict
    ) -> Tuple[str, int]:
        """
        Do the actual work of creating a connection.

        This method will call pce library to generate a breakdown
        across relevant domains, and then send individual connection
        requests to each of those domains.

        Note that we can return early if things fail.  Return value is
        a tuple of the form (reason, HTTP code).
        """
        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 "Could not generate a graph", 424

        traffic_matrix = te_manager.generate_traffic_matrix(
            connection_request=connection_request
        )
        if traffic_matrix is None:
            return "Could not generate a traffic matrix", 400

        logger.info(f"Generated graph: '{graph}', traffic matrix: '{traffic_matrix}'")

        solver = TESolver(graph, traffic_matrix)
        solution = solver.solve()
        logger.debug(f"TESolver result: {solution}")

        if solution is None or solution.connection_map is None:
            return "Could not solve the request", 400

        try:
            breakdown = te_manager.generate_connection_breakdown(
                solution, connection_request
            )
            self.db_instance.add_key_value_pair_to_db(
                "breakdowns", connection_request["id"], breakdown
            )
            status, code = self._send_breakdown_to_lc(
                breakdown, "post", connection_request
            )
            logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}")
            return status, code
        except Exception as e:
            logger.debug(f"Error when generating/publishing breakdown: {e}")
            return f"Error: {e}", 400

    def remove_connection(self, te_manager, connection_id) -> Tuple[str, int]:
        te_manager.unreserve_vlan(connection_id)
        breakdown = self.db_instance.read_from_db("breakdowns", connection_id)[
            connection_id
        ]
        connection_request = self.db_instance.read_from_db(
            "connections", connection_id
        )[connection_id]

        try:
            status, code = self._send_breakdown_to_lc(
                breakdown, "delete", connection_request
            )
            logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}")
            return status, code
        except Exception as e:
            logger.debug(f"Error when removing breakdown: {e}")
            return f"Error: {e}", 400

    def handle_link_failure(self, te_manager, msg_json):
        logger.debug("---Handling connections that contain failed link.---")
        link_connections_dict_str = self.db_instance.read_from_db(
            "links", "link_connections_dict"
        )

        if (
            not link_connections_dict_str
            or not link_connections_dict_str["link_connections_dict"]
        ):
            logger.debug("No connection has been placed yet.")
            return

        link_connections_dict = json.loads(
            link_connections_dict_str["link_connections_dict"]
        )

        for link in msg_json["link_failure"]:
            port_list = []
            if "ports" not in link:
                continue
            for port in link["ports"]:
                if "id" not in port:
                    continue
                port_list.append(port["id"])

            simple_link = SimpleLink(port_list).to_string()

            if simple_link in link_connections_dict:
                logger.debug("Found failed link record!")
                connections = link_connections_dict[simple_link]
                for index, connection in enumerate(connections):
                    if "id" not in connection:
                        continue
                    self.remove_connection(te_manager, connection["id"])
                    del link_connections_dict[simple_link][index]
                    logger.debug("Removed connection:")
                    logger.debug(connection)
                    self.place_connection(te_manager, connection)
                    link_connections_dict[simple_link].append(connection)

        self.db_instance.delete_one_entry("links", "link_connections_dict")
        self.db_instance.add_key_value_pair_to_db(
            "links", "link_connections_dict", json.dumps(link_connections_dict)
        )

1	import json	4✔
2	import logging	4✔
3	from typing import Tuple	4✔
4
5	from sdx_pce.load_balancing.te_solver import TESolver	4✔
6	from sdx_pce.topology.temanager import TEManager	4✔
7
8	from sdx_controller.messaging.topic_queue_producer import TopicQueueProducer	4✔
9	from sdx_controller.models.simple_link import SimpleLink	4✔
10	from sdx_controller.utils.parse_helper import ParseHelper	4✔
11
12	logger = logging.getLogger(__name__)	4✔
13	logging.getLogger("pika").setLevel(logging.WARNING)	4✔
14
15
16	class ConnectionHandler:	4✔
17	def __init__(self, db_instance):	4✔
18	self.db_instance = db_instance	4✔
19	self.parse_helper = ParseHelper()	4✔
20
21	def _send_breakdown_to_lc(self, breakdown, operation, connection_request):	4✔
22	logger.debug(f"-- BREAKDOWN: {json.dumps(breakdown)}")	4✔
23
24	if breakdown is None:	4✔
25	return "Could not break down the solution", 400	×
26
27	link_connections_dict_json = (	4✔
28	self.db_instance.read_from_db("links", "link_connections_dict")[
29	"link_connections_dict"
30	]
31	if self.db_instance.read_from_db("links", "link_connections_dict")
32	else None
33	)
34
35	if link_connections_dict_json:	4✔
36	link_connections_dict = json.loads(link_connections_dict_json)	4✔
37	else:
38	link_connections_dict = {}	4✔
39
40	for domain, link in breakdown.items():	4✔
41	port_list = []	4✔
42	for key in link.keys():	4✔
43	if "uni_" in key and "port_id" in link[key]:	4✔
44	port_list.append(link[key]["port_id"])	4✔
45
46	if port_list:	4✔
47	simple_link = SimpleLink(port_list).to_string()	4✔
48
49	if simple_link not in link_connections_dict:	4✔
50	link_connections_dict[simple_link] = []	4✔
51
52	if connection_request not in link_connections_dict[simple_link]:	4✔
53	link_connections_dict[simple_link].append(connection_request)	4✔
54
55	self.db_instance.add_key_value_pair_to_db(	4✔
56	"links", "link_connections_dict", json.dumps(link_connections_dict)
57	)
58
59	logger.debug(f"Attempting to publish domain: {domain}, link: {link}")	4!
60
61	# From "urn:ogf:network:sdx:topology:amlight.net", attempt to
62	# extract a string like "amlight".
63	domain_name = self.parse_helper.find_domain_name(domain, ":") or f"{domain}"	4✔
64	exchange_name = "connection"	4✔
65
66	logger.debug(	4✔
67	f"Doing '{operation}' operation for '{link}' with exchange_name: {exchange_name}, "
68	f"routing_key: {domain_name}"
69	)
70	mq_link = {"operation": operation, "link": link}	4✔
71	producer = TopicQueueProducer(	4✔
72	timeout=5, exchange_name=exchange_name, routing_key=domain_name
73	)
74	producer.call(json.dumps(mq_link))	4✔
75	producer.stop_keep_alive()	4✔
76
77	# We will get to this point only if all the previous steps
78	# leading up to this point were successful.
79	return "Connection published", 200	4✔
80
81	def place_connection(	4✔
82	self, te_manager: TEManager, connection_request: dict
83	) -> Tuple[str, int]:
84	"""
85	Do the actual work of creating a connection.
86
87	This method will call pce library to generate a breakdown
88	across relevant domains, and then send individual connection
89	requests to each of those domains.
90
91	Note that we can return early if things fail. Return value is
92	a tuple of the form (reason, HTTP code).
93	"""
94	for num, val in enumerate(te_manager.get_topology_map().values()):	4✔
95	logger.debug(f"TE topology #{num}: {val}")	4✔
96
97	graph = te_manager.generate_graph_te()	4✔
98	if graph is None:	4✔
99	return "Could not generate a graph", 424	×
100
101	traffic_matrix = te_manager.generate_traffic_matrix(	4✔
102	connection_request=connection_request
103	)
104	if traffic_matrix is None:	4✔
105	return "Could not generate a traffic matrix", 400	4✔
106
107	logger.info(f"Generated graph: '{graph}', traffic matrix: '{traffic_matrix}'")	4✔
108
109	solver = TESolver(graph, traffic_matrix)	4✔
110	solution = solver.solve()	4✔
111	logger.debug(f"TESolver result: {solution}")	4✔
112
113	if solution is None or solution.connection_map is None:	4✔
114	return "Could not solve the request", 400	×
115
116	try:	4✔
117	breakdown = te_manager.generate_connection_breakdown(	4✔
118	solution, connection_request
119	)
120	self.db_instance.add_key_value_pair_to_db(	4✔
121	"breakdowns", connection_request["id"], breakdown
122	)
123	status, code = self._send_breakdown_to_lc(	4✔
124	breakdown, "post", connection_request
125	)
126	logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}")	4✔
127	return status, code	4✔
128	except Exception as e:	×
129	logger.debug(f"Error when generating/publishing breakdown: {e}")	×
130	return f"Error: {e}", 400	×
131
132	def remove_connection(self, te_manager, connection_id) -> Tuple[str, int]:	4✔
133	te_manager.unreserve_vlan(connection_id)	4✔
134	breakdown = self.db_instance.read_from_db("breakdowns", connection_id)[	4✔
135	connection_id
136	]
137	connection_request = self.db_instance.read_from_db(	4✔
138	"connections", connection_id
139	)[connection_id]
140
141	try:	4✔
142	status, code = self._send_breakdown_to_lc(	4✔
143	breakdown, "delete", connection_request
144	)
145	logger.debug(f"Breakdown sent to LC, status: {status}, code: {code}")	4✔
146	return status, code	4✔
147	except Exception as e:	×
148	logger.debug(f"Error when removing breakdown: {e}")	×
149	return f"Error: {e}", 400	×
150
151	def handle_link_failure(self, te_manager, msg_json):	4✔
152	logger.debug("---Handling connections that contain failed link.---")	×
153	link_connections_dict_str = self.db_instance.read_from_db(	×
154	"links", "link_connections_dict"
155	)
156
157	if (	×
158	not link_connections_dict_str
159	or not link_connections_dict_str["link_connections_dict"]
160	):
161	logger.debug("No connection has been placed yet.")	×
162	return	×
163
164	link_connections_dict = json.loads(	×
165	link_connections_dict_str["link_connections_dict"]
166	)
167
168	for link in msg_json["link_failure"]:	×
169	port_list = []	×
170	if "ports" not in link:	×
171	continue	×
172	for port in link["ports"]:	×
173	if "id" not in port:	×
174	continue	×
175	port_list.append(port["id"])	×
176
177	simple_link = SimpleLink(port_list).to_string()	×
178
179	if simple_link in link_connections_dict:	×
180	logger.debug("Found failed link record!")	×
181	connections = link_connections_dict[simple_link]	×
182	for index, connection in enumerate(connections):	×
NEW 183	if "id" not in connection:	×
NEW 184	continue	×
NEW 185	self.remove_connection(te_manager, connection["id"])	×
186	del link_connections_dict[simple_link][index]	×
187	logger.debug("Removed connection:")	×
188	logger.debug(connection)	×
NEW 189	self.place_connection(te_manager, connection)	×
190	link_connections_dict[simple_link].append(connection)	×
191
NEW 192	self.db_instance.delete_one_entry("links", "link_connections_dict")	×
UNCOV 193	self.db_instance.add_key_value_pair_to_db(	×
194	"links", "link_connections_dict", json.dumps(link_connections_dict)
195	)

atlanticwave-sdx / sdx-controller / 9530886246

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous