10729515210

Committed 05 Sep 2024 11:09PM UTC coverage: 57.757% (+0.1%) from 57.645%

Build # 10729515210

Build Type

Pull #326

github

Committed by

web-flow

Commit Message

Merge 9ab49848f 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 %.

15 of 21 new or added lines in 4 files covered. (71.43%)

4 existing lines in 1 file now uncovered.

1052 of 1885 relevant lines covered (55.81%)

2.23 hits per line

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

66.99

/sdx_controller/handlers/connection_handler.py

import json
import logging
import time
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
        )
        # Current timestamp in seconds
        timestamp = int(time.time())

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

atlanticwave-sdx / sdx-controller / 10729515210

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