10152038944

Committed 29 Jul 2024 09:20PM UTC coverage: 52.567% (-0.003%) from 52.57%

Build # 10152038944

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #306 from atlanticwave-sdx/fix/issue_305

Add traceback information when exception on breakdown

Run Details

305 of 557 branches covered (54.76%)

Branch coverage included in aggregate %.

1 of 3 new or added lines in 1 file covered. (33.33%)

1 existing line in 1 file now uncovered.

770 of 1488 relevant lines covered (51.75%)

2.07 hits per line

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

60.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 = {}

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

atlanticwave-sdx / sdx-controller / 10152038944

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