10727295815

Committed 05 Sep 2024 07:59PM UTC coverage: 57.725% (+0.08%) from 57.645%

Build # 10727295815

Build Type

Pull #326

github

Committed by

web-flow

Commit Message

Merge d96bf1512 into def24445a

Pull Request Pull Request #326: Reuse service id for patch

Run Details

467 of 745 branches covered (62.68%)

Branch coverage included in aggregate %.

13 of 19 new or added lines in 4 files covered. (68.42%)

4 existing lines in 1 file now uncovered.

1050 of 1883 relevant lines covered (55.76%)

2.23 hits per line

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

66.67

/sdx_controller/handlers/connection_handler.py

import json
import logging
import traceback
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 = {}

        interdomain_a, interdomain_b = None, None
        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 (
                    operation == "post"
                    and connection_request not in link_connections_dict[simple_link]
                ):
                    link_connections_dict[simple_link].append(connection_request)

                if (
                    operation == "delete"
                    and connection_request in link_connections_dict[simple_link]
                ):
                    link_connections_dict[simple_link].remove(connection_request)

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

            if interdomain_a:
                interdomain_b = link.get("uni_a", {}).get("port_id")
            else:
                interdomain_a = link.get("uni_z", {}).get("port_id")

            if interdomain_a and interdomain_b:
                simple_link = SimpleLink([interdomain_a, interdomain_b]).to_string()

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

                if (
                    operation == "post"
                    and connection_request not in link_connections_dict[simple_link]
                ):
                    link_connections_dict[simple_link].append(connection_request)

                if (
                    operation == "delete"
                    and connection_request in link_connections_dict[simple_link]
                ):
                    link_connections_dict[simple_link].remove(connection_request)

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

                interdomain_a = link.get("uni_z", {}).get("port_id")

            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 "No SDX topology found", 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:
            err = traceback.format_exc().replace("\n", ", ")
            logger.error(f"Error when generating/publishing breakdown: {e} - {err}")
            return f"Error: {e}", 400

    def archive_connection(self, service_id) -> None:
        connection_request = self.db_instance.read_from_db("connections", service_id)
        if not connection_request:
            return

        connection_request_str = connection_request[service_id]
        self.db_instance.delete_one_entry("connections", service_id)

        historical_connections = self.db_instance.read_from_db(
            "historical_connections", service_id
        )

        if historical_connections:
            historical_connections_list = historical_connections[service_id]
            historical_connections_list.append(connection_request_str)
            self.db_instance.add_key_value_pair_to_db(
                "historical_connections", service_id, historical_connections_list
            )
        else:
            self.db_instance.add_key_value_pair_to_db(
                "historical_connections", service_id, [connection_request_str]
            )
        logger.debug(f"Archived connection: {service_id}")

    def remove_connection(self, te_manager, service_id) -> Tuple[str, int]:
        te_manager.unreserve_vlan(service_id)
        connection_request = self.db_instance.read_from_db("connections", service_id)
        if not connection_request:
            return "Did not find connection request, cannot remove connection", 404

        connection_request = connection_request[service_id]

        breakdown = self.db_instance.read_from_db("breakdowns", service_id)
        if not breakdown:
            return "Did not find breakdown, cannot remove connection", 404
        breakdown = breakdown[service_id]

        try:
            status, code = self._send_breakdown_to_lc(
                breakdown, "delete", json.loads(connection_request)
            )
            self.db_instance.delete_one_entry("breakdowns", service_id)
            self.archive_connection(service_id)
            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, failed_links):
        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 failed_links:
            logger.info(f"Handling link failure on {link['id']} ({link['ports']})")
            port_list = []
            if "ports" not in link:
                continue
            for port in link["ports"]:
                port_id = port if isinstance(port, str) else port.get("id")
                if not port_id:
                    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):
                    logger.info(
                        f"Connection {connection['id']} affected by link {link['id']}"
                    )
                    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)

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

atlanticwave-sdx / sdx-controller / 10727295815

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