build.main.Main.on_topology_updated() - Code Metrics - Inspection of "chore: `@rest` endpoints are now run by `starlette..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#45)

by Vinicius

created 2023-05-03 17:50 UTC

build.main.Main.on_topology_updated() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1.037

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
dl	0
loc	8
rs	10
c	0
b	0
f	0
ccs	2
cts	3
cp	0.6667
crap	1.037

"""Main module of kytos/pathfinder Kytos Network Application."""

from threading import Lock
from typing import Generator

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 get_json_or_400)
from napps.kytos.pathfinder.graph import KytosGraph


class Main(KytosNApp):
    """
    Main class of kytos/pathfinder NApp.

    This class is the entry point for this napp.
    """

    def setup(self):
        """Create a graph to handle the nodes and edges."""
        self.graph = KytosGraph()
        self._topology = None
        self._lock = Lock()
        self._topology_updated_at = None
        self._links_updated_at = {}

    def execute(self):
        """Do nothing."""

    def shutdown(self):
        """Shutdown the napp."""

    def _filter_paths_le_cost(self, paths, max_cost):
        """Filter by paths where the cost is le <= max_cost."""
        if not max_cost:
            return paths
        return [path for path in paths if path["cost"] <= max_cost]

    def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
        """Map endpoints from link ids."""
        endpoints = {}
        for link_id in link_ids:
            try:
                link = self._topology.links[link_id]
                endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
                endpoints[(endpoint_a.id, endpoint_b.id)] = link
            except KeyError:
                pass
        return endpoints

    def _find_all_link_ids(
        self, paths: list[dict], link_ids: list[str]
    ) -> Generator[int, None, None]:
        """Find indexes of the paths that contain all link ids."""
        endpoints_links = self._map_endpoints_from_link_ids(link_ids)
        if not endpoints_links:
            return None
        endpoint_keys = set(endpoints_links.keys())
        for idx, path in enumerate(paths):
            head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
            for endpoint_a, endpoint_b in zip(head, tail):
                if (endpoint_a, endpoint_b) in endpoints_links:
                    found_endpoints.add((endpoint_a, endpoint_b))
                if (endpoint_b, endpoint_a) in endpoints_links:
                    found_endpoints.add((endpoint_b, endpoint_a))
            if found_endpoints == endpoint_keys:
                yield idx
        return None

    def _find_any_link_ids(
        self, paths: list[dict], link_ids: list[str]
    ) -> Generator[int, None, None]:
        """Find indexes of the paths that contain any of the link ids."""
        endpoints_links = self._map_endpoints_from_link_ids(link_ids)
        if not endpoints_links:
            return None
        for idx, path in enumerate(paths):
            head, tail, found = path["hops"][:-1], path["hops"][1:], False
            for endpoint_a, endpoint_b in zip(head, tail):
                if any(
                    (
                        (endpoint_a, endpoint_b) in endpoints_links,
                        (endpoint_b, endpoint_a) in endpoints_links,
                    )
                ):
                    found = True
                    break
            if found:
                yield idx
        return None

    def _filter_paths_undesired_links(
        self, paths: list[dict], undesired: list[str]
    ) -> list[dict]:
        """Filter by undesired_links, it performs a logical OR."""
        if not undesired:
            return paths
        excluded_indexes = set(self._find_any_link_ids(paths, undesired))
        return [path for idx, path in enumerate(paths) if idx not in excluded_indexes]

    def _filter_paths_desired_links(
        self, paths: list[dict], desired: list[str]
    ) -> list[dict]:
        """Filter by desired_links, it performs a logical AND."""
        if not desired:
            return paths
        included_indexes = set(self._find_all_link_ids(paths, desired))
        return [path for idx, path in enumerate(paths) if idx in included_indexes]

    def _validate_payload(self, data):
        """Validate shortest_path v2/ POST endpoint."""
        if "source" not in data or not isinstance(data["source"], str):
            raise HTTPException(
                400,
                detail=f"'source' is mandatory, got {data.get('source')}"
            )
        if "destination" not in data or not isinstance(data["destination"], str):
            raise HTTPException(
                400,
                detail=f"'destination' is mandatory, got {data.get('destination')}"
            )
        if data.get("desired_links"):
            if not isinstance(data["desired_links"], list):
                raise HTTPException(
                    400,
                    f"TypeError: desired_links is supposed to be a list."
                    f" type: {type(data['desired_links'])}"
                )

        if data.get("undesired_links"):
            if not isinstance(data["undesired_links"], list):
                raise HTTPException(
                    400,
                    f"TypeError: undesired_links is supposed to be a list."
                    f" type: {type(data['undesired_links'])}"
                )

        parameter = data.get("parameter")
        spf_attr = data.get("spf_attribute")
        if not spf_attr:
            spf_attr = parameter or "hop"
        data["spf_attribute"] = spf_attr

        if spf_attr not in self.graph.spf_edge_data_cbs:
            raise HTTPException(
                400,
                "Invalid 'spf_attribute'. Valid values: "
                f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
            )

        try:
            data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
        except (TypeError, ValueError):
            raise HTTPException(
                400,
                f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
            )

        spf_max_path_cost = data.get("spf_max_path_cost")
        if spf_max_path_cost:
            try:
                spf_max_path_cost = max(int(spf_max_path_cost), 1)
                data["spf_max_path_cost"] = spf_max_path_cost
            except (TypeError, ValueError):
                raise HTTPException(
                    400,
                    f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
                    " be an int"
                )

        data["mandatory_metrics"] = data.get("mandatory_metrics", {})
        data["flexible_metrics"] = data.get("flexible_metrics", {})

        try:
            minimum_hits = data.get("minimum_flexible_hits")
            if minimum_hits:
                minimum_hits = min(
                    len(data["flexible_metrics"]), max(0, int(minimum_hits))
                )
            data["minimum_flexible_hits"] = minimum_hits
        except (TypeError, ValueError):
            raise HTTPException(
                400,
                f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
            )

        return data

    @rest("v2/", methods=["POST"])
    def shortest_path(self, request: Request) -> JSONResponse:
        """Calculate the best path between the source and destination."""
        data = get_json_or_400(request, self.controller.loop)
        if not isinstance(data, dict):
            raise HTTPException(400, detail=f"Invalid body value: {data}")
        data = self._validate_payload(data)

        desired = data.get("desired_links")
        undesired = data.get("undesired_links")

        spf_attr = data.get("spf_attribute")
        spf_max_paths = data.get("spf_max_paths")
        spf_max_path_cost = data.get("spf_max_path_cost")
        mandatory_metrics = data.get("mandatory_metrics")
        flexible_metrics = data.get("flexible_metrics")
        minimum_hits = data.get("minimum_flexible_hits")
        log.debug(f"POST v2/ payload data: {data}")

        try:
            with self._lock:
                if any([mandatory_metrics, flexible_metrics]):
                    paths = self.graph.constrained_k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                        minimum_hits=minimum_hits,
                        mandatory_metrics=mandatory_metrics,
                        flexible_metrics=flexible_metrics,
                    )
                else:
                    paths = self.graph.k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                    )

                paths = self.graph.path_cost_builder(
                    paths,
                    weight=spf_attr,
                )
            log.debug(f"Found paths: {paths}")
        except TypeError as err:
            raise HTTPException(400, str(err))

        paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
        paths = self._filter_paths_undesired_links(paths, undesired)
        paths = self._filter_paths_desired_links(paths, desired)
        log.debug(f"Filtered paths: {paths}")
        return JSONResponse({"paths": paths})

    @listen_to(
        "kytos.topology.updated",
        "kytos/topology.current",
        "kytos/topology.topology_loaded",
    )
    def on_topology_updated(self, event):
        """Update the graph when the network topology is updated."""
        self.update_topology(event)

    def update_topology(self, event):
        """Update the graph when the network topology is updated."""
        if "topology" not in event.content:
            return
        topology = event.content["topology"]
        with self._lock:
            if (
                self._topology_updated_at
                and self._topology_updated_at > event.timestamp
            ):
                return
            self._topology = topology
            self._topology_updated_at = event.timestamp
            self.graph.update_topology(topology)
        switches = list(topology.switches.keys())
        links = list(topology.links.keys())
        log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")

    def update_links_metadata_changed(self, event) -> None:
        """Update the graph when links' metadata are added or removed."""
        link = event.content["link"]
        try:
            with self._lock:
                if (
                    link.id in self._links_updated_at
                    and self._links_updated_at[link.id] > event.timestamp
                ):
                    return
                self.graph.update_link_metadata(link)
                self._links_updated_at[link.id] = event.timestamp
            metadata = event.content["metadata"]
            log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
        except KeyError as exc:
            log.warning(
                f"Unexpected KeyError {str(exc)} on event {event}."
                " pathfinder will reconciliate the topology"
            )
            self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))

    @listen_to("kytos/topology.links.metadata.(added|removed)")
    def on_links_metadata_changed(self, event):
        """Update the graph when links' metadata are added or removed."""
        self.update_links_metadata_changed(event)


1		"""Main module of kytos/pathfinder Kytos Network Application."""
2
3	1	from threading import Lock
4	1	from typing import Generator
5
6	1	from kytos.core import KytosEvent, KytosNApp, log, rest
7	1	from kytos.core.helpers import listen_to
8	1	from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
9		get_json_or_400)
10	1	from napps.kytos.pathfinder.graph import KytosGraph
11
12
13	1	class Main(KytosNApp):
14		"""
15		Main class of kytos/pathfinder NApp.
16
17		This class is the entry point for this napp.
18		"""
19
20	1	def setup(self):
21		"""Create a graph to handle the nodes and edges."""
22	1	self.graph = KytosGraph()
23	1	self._topology = None
24	1	self._lock = Lock()
25	1	self._topology_updated_at = None
26	1	self._links_updated_at = {}
27
28	1	def execute(self):
29		"""Do nothing."""
30
31	1	def shutdown(self):
32		"""Shutdown the napp."""
33
34	1	def _filter_paths_le_cost(self, paths, max_cost):
35		"""Filter by paths where the cost is le <= max_cost."""
36	1	if not max_cost:
37	1	return paths
38	1	return [path for path in paths if path["cost"] <= max_cost]
39
40	1	def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
41		"""Map endpoints from link ids."""
42	1	endpoints = {}
43	1	for link_id in link_ids:
44	1	try:
45	1	link = self._topology.links[link_id]
46	1	endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
47	1	endpoints[(endpoint_a.id, endpoint_b.id)] = link
48	1	except KeyError:
49	1	pass
50	1	return endpoints
51
52	1	def _find_all_link_ids(
53		self, paths: list[dict], link_ids: list[str]
54		) -> Generator[int, None, None]:
55		"""Find indexes of the paths that contain all link ids."""
56	1	endpoints_links = self._map_endpoints_from_link_ids(link_ids)
57	1	if not endpoints_links:
58	1	return None
59	1	endpoint_keys = set(endpoints_links.keys())
60	1	for idx, path in enumerate(paths):
61	1	head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
62	1	for endpoint_a, endpoint_b in zip(head, tail):
63	1	if (endpoint_a, endpoint_b) in endpoints_links:
64	1	found_endpoints.add((endpoint_a, endpoint_b))
65	1	if (endpoint_b, endpoint_a) in endpoints_links:
66		found_endpoints.add((endpoint_b, endpoint_a))
67	1	if found_endpoints == endpoint_keys:
68	1	yield idx
69	1	return None
70
71	1	def _find_any_link_ids(
72		self, paths: list[dict], link_ids: list[str]
73		) -> Generator[int, None, None]:
74		"""Find indexes of the paths that contain any of the link ids."""
75	1	endpoints_links = self._map_endpoints_from_link_ids(link_ids)
76	1	if not endpoints_links:
77	1	return None
78	1	for idx, path in enumerate(paths):
79	1	head, tail, found = path["hops"][:-1], path["hops"][1:], False
80	1	for endpoint_a, endpoint_b in zip(head, tail):
81	1	if any(
82		(
83		(endpoint_a, endpoint_b) in endpoints_links,
84		(endpoint_b, endpoint_a) in endpoints_links,
85		)
86		):
87	1	found = True
88	1	break
89	1	if found:
90	1	yield idx
91	1	return None
92
93	1	def _filter_paths_undesired_links(
94		self, paths: list[dict], undesired: list[str]
95		) -> list[dict]:
96		"""Filter by undesired_links, it performs a logical OR."""
97	1	if not undesired:
98	1	return paths
99	1	excluded_indexes = set(self._find_any_link_ids(paths, undesired))
100	1	return [path for idx, path in enumerate(paths) if idx not in excluded_indexes]
101
102	1	def _filter_paths_desired_links(
103		self, paths: list[dict], desired: list[str]
104		) -> list[dict]:
105		"""Filter by desired_links, it performs a logical AND."""
106	1	if not desired:
107	1	return paths
108	1	included_indexes = set(self._find_all_link_ids(paths, desired))
109	1	return [path for idx, path in enumerate(paths) if idx in included_indexes]
110
111	1	def _validate_payload(self, data):
112		"""Validate shortest_path v2/ POST endpoint."""
113	1	if "source" not in data or not isinstance(data["source"], str):
114		raise HTTPException(
115		400,
116		detail=f"'source' is mandatory, got {data.get('source')}"
117		)
118	1	if "destination" not in data or not isinstance(data["destination"], str):
119		raise HTTPException(
120		400,
121		detail=f"'destination' is mandatory, got {data.get('destination')}"
122		)
123	1	if data.get("desired_links"):
124	1	if not isinstance(data["desired_links"], list):
125		raise HTTPException(
126		400,
127		f"TypeError: desired_links is supposed to be a list."
128		f" type: {type(data['desired_links'])}"
129		)
130
131	1	if data.get("undesired_links"):
132		if not isinstance(data["undesired_links"], list):
133		raise HTTPException(
134		400,
135		f"TypeError: undesired_links is supposed to be a list."
136		f" type: {type(data['undesired_links'])}"
137		)
138
139	1	parameter = data.get("parameter")
140	1	spf_attr = data.get("spf_attribute")
141	1	if not spf_attr:
142	1	spf_attr = parameter or "hop"
143	1	data["spf_attribute"] = spf_attr
144
145	1	if spf_attr not in self.graph.spf_edge_data_cbs:
146		raise HTTPException(
147		400,
148		"Invalid 'spf_attribute'. Valid values: "
149		f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
150		)
151
152	1	try:
153	1	data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
154		except (TypeError, ValueError):
155		raise HTTPException(
156		400,
157		f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
158		)
159
160	1	spf_max_path_cost = data.get("spf_max_path_cost")
161	1	if spf_max_path_cost:
162		try:
163		spf_max_path_cost = max(int(spf_max_path_cost), 1)
164		data["spf_max_path_cost"] = spf_max_path_cost
165		except (TypeError, ValueError):
166		raise HTTPException(
167		400,
168		f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
169		" be an int"
170		)
171
172	1	data["mandatory_metrics"] = data.get("mandatory_metrics", {})
173	1	data["flexible_metrics"] = data.get("flexible_metrics", {})
174
175	1	try:
176	1	minimum_hits = data.get("minimum_flexible_hits")
177	1	if minimum_hits:
178	1	minimum_hits = min(
179		len(data["flexible_metrics"]), max(0, int(minimum_hits))
180		)
181	1	data["minimum_flexible_hits"] = minimum_hits
182		except (TypeError, ValueError):
183		raise HTTPException(
184		400,
185		f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
186		)
187
188	1	return data
189
190	1	@rest("v2/", methods=["POST"])
191	1	def shortest_path(self, request: Request) -> JSONResponse:
192		"""Calculate the best path between the source and destination."""
193	1	data = get_json_or_400(request, self.controller.loop)
194	1	if not isinstance(data, dict):
195		raise HTTPException(400, detail=f"Invalid body value: {data}")
196	1	data = self._validate_payload(data)
197
198	1	desired = data.get("desired_links")
199	1	undesired = data.get("undesired_links")
200
201	1	spf_attr = data.get("spf_attribute")
202	1	spf_max_paths = data.get("spf_max_paths")
203	1	spf_max_path_cost = data.get("spf_max_path_cost")
204	1	mandatory_metrics = data.get("mandatory_metrics")
205	1	flexible_metrics = data.get("flexible_metrics")
206	1	minimum_hits = data.get("minimum_flexible_hits")
207	1	log.debug(f"POST v2/ payload data: {data}")
208
209	1	try:
210	1	with self._lock:
211	1	if any([mandatory_metrics, flexible_metrics]):
212	1	paths = self.graph.constrained_k_shortest_paths(
213		data["source"],
214		data["destination"],
215		weight=self.graph.spf_edge_data_cbs[spf_attr],
216		k=spf_max_paths,
217		minimum_hits=minimum_hits,
218		mandatory_metrics=mandatory_metrics,
219		flexible_metrics=flexible_metrics,
220		)
221		else:
222	1	paths = self.graph.k_shortest_paths(
223		data["source"],
224		data["destination"],
225		weight=self.graph.spf_edge_data_cbs[spf_attr],
226		k=spf_max_paths,
227		)
228
229	1	paths = self.graph.path_cost_builder(
230		paths,
231		weight=spf_attr,
232		)
233	1	log.debug(f"Found paths: {paths}")
234	1	except TypeError as err:
235	1	raise HTTPException(400, str(err))
236
237	1	paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
238	1	paths = self._filter_paths_undesired_links(paths, undesired)
239	1	paths = self._filter_paths_desired_links(paths, desired)
240	1	log.debug(f"Filtered paths: {paths}")
241	1	return JSONResponse({"paths": paths})
242
243	1	@listen_to(
244		"kytos.topology.updated",
245		"kytos/topology.current",
246		"kytos/topology.topology_loaded",
247		)
248	1	def on_topology_updated(self, event):
249		"""Update the graph when the network topology is updated."""
250		self.update_topology(event)
251
252	1	def update_topology(self, event):
253		"""Update the graph when the network topology is updated."""
254	1	if "topology" not in event.content:
255	1	return
256	1	topology = event.content["topology"]
257	1	with self._lock:
258	1	if (
259		self._topology_updated_at
260		and self._topology_updated_at > event.timestamp
261		):
262	1	return
263	1	self._topology = topology
264	1	self._topology_updated_at = event.timestamp
265	1	self.graph.update_topology(topology)
266	1	switches = list(topology.switches.keys())
267	1	links = list(topology.links.keys())
268	1	log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")
269
270	1	def update_links_metadata_changed(self, event) -> None:
271		"""Update the graph when links' metadata are added or removed."""
272	1	link = event.content["link"]
273	1	try:
274	1	with self._lock:
275	1	if (
276		link.id in self._links_updated_at
277		and self._links_updated_at[link.id] > event.timestamp
278		):
279	1	return
280	1	self.graph.update_link_metadata(link)
281	1	self._links_updated_at[link.id] = event.timestamp
282	1	metadata = event.content["metadata"]
283	1	log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
284	1	except KeyError as exc:
285	1	log.warning(
286		f"Unexpected KeyError {str(exc)} on event {event}."
287		" pathfinder will reconciliate the topology"
288		)
289	1	self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))
290
291	1	@listen_to("kytos/topology.links.metadata.(added\|removed)")
292	1	def on_links_metadata_changed(self, event):
293		"""Update the graph when links' metadata are added or removed."""
294		self.update_links_metadata_changed(event)
295

kytos-ng / pathfinder

Pull Request — master (#45)

build.main.Main.on_topology_updated() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like