build.models.path.DynamicPathManager.create_path() - Code Metrics - Inspection of "Adding switches to disjointedness algorithm" - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#427)

unknown

created 2024-02-06 17:55 UTC

build.models.path.DynamicPathManager.create_path() A

↳ Parent: build.models.path

Complexity

Conditions

Size

Total Lines	17
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	5

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	17
ccs	13
cts	13
cp	1
rs	9.2833
c	0
b	0
f	0
cc	5
nop	2
crap	5

"""Classes related to paths"""
import requests

from kytos.core import log
from kytos.core.common import EntityStatus, GenericEntity
from kytos.core.interface import TAG
from kytos.core.link import Link
from napps.kytos.mef_eline import settings
from napps.kytos.mef_eline.exceptions import InvalidPath


class Path(list, GenericEntity):
    """Class to represent a Path."""

    def __eq__(self, other=None):
        """Compare paths."""
        if not other or not isinstance(other, Path):
            return False
        return super().__eq__(other)

    def is_affected_by_link(self, link=None):
        """Verify if the current path is affected by link."""
        if not link:
            return False
        return link in self

    def link_affected_by_interface(self, interface=None):
        """Return the link using this interface, if any, or None otherwise."""
        if not interface:
            return None
        for link in self:
            if interface in (link.endpoint_a, link.endpoint_b):
                return link
        return None

    def choose_vlans(self, controller):
        """Choose the VLANs to be used for the circuit."""
        for link in self:
            tag_value = link.get_next_available_tag(controller, link.id)
            tag = TAG('vlan', tag_value)
            link.add_metadata("s_vlan", tag)

    def make_vlans_available(self, controller):
        """Make the VLANs used in a path available when undeployed."""
        for link in self:
            tag = link.get_metadata("s_vlan")
            conflict_a, conflict_b = link.make_tags_available(
                controller, tag.value, link.id, tag.tag_type,
                check_order=False
            )
            if conflict_a:
                log.error(f"Tags {conflict_a} was already available in"
                          f"{link.endpoint_a.id}")
            if conflict_b:
                log.error(f"Tags {conflict_b} was already available in"
                          f"{link.endpoint_b.id}")
            link.remove_metadata("s_vlan")

    def is_valid(self, switch_a, switch_z, is_scheduled=False):
        """Check if this is a valid path."""
        if not self:
            return True
        previous = visited = {switch_a}
        for link in self:
            current = {link.endpoint_a.switch, link.endpoint_b.switch} \
                      - previous
            if len(current) != 1:
                raise InvalidPath(
                    f"Previous switch {previous} is not connected to "
                    f"current link with switches {current}."
                )
            if current & visited:
                raise InvalidPath(
                    f"Loop detected in path, switch {current} was visited"
                    f" more than once."
                )
            if is_scheduled is False and (
                link.endpoint_a.link is None
                or link.endpoint_a.link != link
                or link.endpoint_b.link is None
                or link.endpoint_b.link != link
            ):
                raise InvalidPath(f"Link {link} is not available.")
            previous = current
            visited |= current
        if previous & {switch_z}:
            return True
        raise InvalidPath("Last link does not contain uni_z switch")

    @property
    def status(self):
        """Check for the  status of a path.

        If any link in this path is down, the path is considered down.
        """
        if not self:
            return EntityStatus.DISABLED

        endpoint = f"{settings.TOPOLOGY_URL}/links"
        api_reply = requests.get(endpoint)
        if api_reply.status_code != getattr(requests.codes, "ok"):
            log.error(
                "Failed to get links at %s. Returned %s",
                endpoint,
                api_reply.status_code,
            )
            return None
        links = api_reply.json()["links"]
        return_status = EntityStatus.UP
        for path_link in self:
            try:
                link = links[path_link.id]
            except KeyError:
                return EntityStatus.DISABLED
            if link["enabled"] is False:
                return EntityStatus.DISABLED
            if link["active"] is False:
                return_status = EntityStatus.DOWN
        return return_status

    def as_dict(self):
        """Return list comprehension of links as_dict."""
        return [link.as_dict() for link in self if link]


class DynamicPathManager:
    """Class to handle and create paths."""

    controller = None

    @classmethod
    def set_controller(cls, controller=None):
        """Set the controller to discovery news paths."""
        cls.controller = controller

    @staticmethod
    def get_paths(circuit, max_paths=2, **kwargs) -> list[dict]:
        """Get a valid path for the circuit from the Pathfinder."""
        endpoint = settings.PATHFINDER_URL
        spf_attribute = kwargs.get("spf_attribute") or settings.SPF_ATTRIBUTE
        request_data = {
            "source": circuit.uni_a.interface.id,
            "destination": circuit.uni_z.interface.id,
            "spf_max_paths": max_paths,
            "spf_attribute": spf_attribute
        }
        request_data.update(kwargs)
        api_reply = requests.post(endpoint, json=request_data)

        if api_reply.status_code != getattr(requests.codes, "ok"):
            log.error(
                "Failed to get paths at %s. Returned %s. Payload %s. EVC %s",
                endpoint,
                api_reply.text,
                request_data,
                circuit,
            )
            return []
        reply_data = api_reply.json()
        return reply_data.get("paths", [])

    @staticmethod
    def _clear_path(path):
        """Remove switches from a path, returning only interfaces."""
        return [endpoint for endpoint in path if len(endpoint) > 23]

    @classmethod
    def get_best_path(cls, circuit):
        """Return the best path available for a circuit, if exists."""
        paths = cls.get_paths(circuit)
        if paths:
            return cls.create_path(cls.get_paths(circuit)[0]["hops"])
        return None

    @classmethod
    def get_best_paths(cls, circuit, **kwargs):
        """Return the best paths available for a circuit, if they exist."""
        for path in cls.get_paths(circuit, **kwargs):
            yield cls.create_path(path["hops"])

    @classmethod
    def get_disjoint_paths(
        cls, circuit, unwanted_path, cutoff=settings.DISJOINT_PATH_CUTOFF
    ):
        """Computes the maximum disjoint paths from the unwanted_path for a EVC

        Maximum disjoint paths from the unwanted_path are the paths from the
        source node to the target node that share the minimum number of links
        and switches contained in unwanted_path. In other words, unwanted_path
        is the path we want to avoid: we want the maximum possible disjoint
        path from it. The disjointness of a path in regards to unwanted_path
        is calculated by the complementary percentage of shared links and
        switches between them. As an example, if the unwanted_path has 3
        links and 2 switches, a given path P1 has 1 link shared with
        unwanted_path, and a given path P2 has 2 links and 1 switch shared
        with unwanted_path, then the disjointness of P1 is 0.8 and the
        disjointness of P2 is 0.4. In this example, P1 is preferable over P2
        because it offers a better disjoint path. When two paths have the same
        disjointness they are ordered by 'cost' attributed as returned from
        Pathfinder. When the disjointness of a path is equal to 0 (i.e., it
        shares all the links with unwanted_path), that particular path is not
        considered a candidate.

        Parameters:
        -----------

        circuit : EVC
            The EVC providing source node (uni_a) and target node (uni_z)

        unwanted_path : Path
            The Path which we want to avoid.

        cutoff: int
            Maximum number of paths to consider when calculating the disjoint
            paths (number of paths to request from pathfinder)

        Returns:
        --------
        paths : generator
            Generator of unwanted_path disjoint paths. If unwanted_path is
            not provided or empty, we return an empty list.
        """
        unwanted_links = [
            (link.endpoint_a.id, link.endpoint_b.id) for link in unwanted_path
        ]
        unwanted_switches = set()
        for link in unwanted_path:
            unwanted_switches.add(link.endpoint_a.switch.id)
            unwanted_switches.add(link.endpoint_b.switch.id)
        unwanted_switches.discard(circuit.uni_a.interface.switch.id)
        unwanted_switches.discard(circuit.uni_z.interface.switch.id)

        length_unwanted = (len(unwanted_links) + len(unwanted_switches))
        if not unwanted_links or not unwanted_switches:
            return None

        paths = cls.get_paths(circuit, max_paths=cutoff,
                              **circuit.secondary_constraints)
        for path in paths:
            links_n, switches_n = cls.get_shared_components(
                path, unwanted_links, unwanted_switches
            )
            shared_components = links_n + switches_n
            path["disjointness"] = 1 - shared_components / length_unwanted
        paths = sorted(paths, key=lambda x: (-x['disjointness'], x['cost']))
        for path in paths:
            if path["disjointness"] == 0:
                continue
            yield cls.create_path(path["hops"])
        return None

    @staticmethod
    def get_shared_components(
        path: Path,
        unwanted_links: list[tuple[str, str]],
        unwanted_switches: set[str]
    ) -> tuple[int, int]:
        """Return the number of shared links
        and switches found in path."""
        head = path["hops"][:-1]
        tail = path["hops"][1:]
        shared_links = 0
        for (endpoint_a, endpoint_b) in unwanted_links:
            if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
                (endpoint_b, endpoint_a) in zip(head, tail)
            ):
                shared_links += 1
        copy_switches = unwanted_switches.copy()
        shared_switches = 0
        for component in path["hops"]:
            if component in copy_switches:
                shared_switches += 1
                copy_switches.remove(component)
        return shared_links, shared_switches

    @classmethod
    def create_path(cls, path):
        """Return the path containing only the interfaces."""
        new_path = Path()
        clean_path = cls._clear_path(path)

        if len(clean_path) % 2:
            return None

        for link in zip(clean_path[1:-1:2], clean_path[2::2]):
            interface_a = cls.controller.get_interface_by_id(link[0])
            interface_b = cls.controller.get_interface_by_id(link[1])
            if interface_a is None or interface_b is None:
                return None
            new_path.append(Link(interface_a, interface_b))

        return new_path


1		"""Classes related to paths"""
2	1	import requests
3
4	1	from kytos.core import log
5	1	from kytos.core.common import EntityStatus, GenericEntity
6	1	from kytos.core.interface import TAG
7	1	from kytos.core.link import Link
8	1	from napps.kytos.mef_eline import settings
9	1	from napps.kytos.mef_eline.exceptions import InvalidPath
10
11
12	1	class Path(list, GenericEntity):
13		"""Class to represent a Path."""
14
15	1	def __eq__(self, other=None):
16		"""Compare paths."""
17	1	if not other or not isinstance(other, Path):
18	1	return False
19	1	return super().__eq__(other)
20
21	1	def is_affected_by_link(self, link=None):
22		"""Verify if the current path is affected by link."""
23	1	if not link:
24	1	return False
25	1	return link in self
26
27	1	def link_affected_by_interface(self, interface=None):
28		"""Return the link using this interface, if any, or None otherwise."""
29	1	if not interface:
30	1	return None
31	1	for link in self:
32	1	if interface in (link.endpoint_a, link.endpoint_b):
33	1	return link
34		return None
35
36	1	def choose_vlans(self, controller):
37		"""Choose the VLANs to be used for the circuit."""
38	1	for link in self:
39	1	tag_value = link.get_next_available_tag(controller, link.id)
40	1	tag = TAG('vlan', tag_value)
41	1	link.add_metadata("s_vlan", tag)
42
43	1	def make_vlans_available(self, controller):
44		"""Make the VLANs used in a path available when undeployed."""
45	1	for link in self:
46	1	tag = link.get_metadata("s_vlan")
47	1	conflict_a, conflict_b = link.make_tags_available(
48		controller, tag.value, link.id, tag.tag_type,
49		check_order=False
50		)
51	1	if conflict_a:
52	1	log.error(f"Tags {conflict_a} was already available in"
53		f"{link.endpoint_a.id}")
54	1	if conflict_b:
55	1	log.error(f"Tags {conflict_b} was already available in"
56		f"{link.endpoint_b.id}")
57	1	link.remove_metadata("s_vlan")
58
59	1	def is_valid(self, switch_a, switch_z, is_scheduled=False):
60		"""Check if this is a valid path."""
61	1	if not self:
62	1	return True
63	1	previous = visited = {switch_a}
64	1	for link in self:
65	1	current = {link.endpoint_a.switch, link.endpoint_b.switch} \
66		- previous
67	1	if len(current) != 1:
68	1	raise InvalidPath(
69		f"Previous switch {previous} is not connected to "
70		f"current link with switches {current}."
71		)
72	1	if current & visited:
73	1	raise InvalidPath(
74		f"Loop detected in path, switch {current} was visited"
75		f" more than once."
76		)
77	1	if is_scheduled is False and (
78		link.endpoint_a.link is None
79		or link.endpoint_a.link != link
80		or link.endpoint_b.link is None
81		or link.endpoint_b.link != link
82		):
83		raise InvalidPath(f"Link {link} is not available.")
84	1	previous = current
85	1	visited \|= current
86	1	if previous & {switch_z}:
87	1	return True
88		raise InvalidPath("Last link does not contain uni_z switch")
89
90	1	@property
91	1	def status(self):
92		"""Check for the status of a path.
93
94		If any link in this path is down, the path is considered down.
95		"""
96	1	if not self:
97	1	return EntityStatus.DISABLED
98
99	1	endpoint = f"{settings.TOPOLOGY_URL}/links"
100	1	api_reply = requests.get(endpoint)
101	1	if api_reply.status_code != getattr(requests.codes, "ok"):
102		log.error(
103		"Failed to get links at %s. Returned %s",
104		endpoint,
105		api_reply.status_code,
106		)
107		return None
108	1	links = api_reply.json()["links"]
109	1	return_status = EntityStatus.UP
110	1	for path_link in self:
111	1	try:
112	1	link = links[path_link.id]
113		except KeyError:
114		return EntityStatus.DISABLED
115	1	if link["enabled"] is False:
116	1	return EntityStatus.DISABLED
117	1	if link["active"] is False:
118	1	return_status = EntityStatus.DOWN
119	1	return return_status
120
121	1	def as_dict(self):
122		"""Return list comprehension of links as_dict."""
123	1	return [link.as_dict() for link in self if link]
124
125
126	1	class DynamicPathManager:
127		"""Class to handle and create paths."""
128
129	1	controller = None
130
131	1	@classmethod
132	1	def set_controller(cls, controller=None):
133		"""Set the controller to discovery news paths."""
134	1	cls.controller = controller
135
136	1	@staticmethod
137	1	def get_paths(circuit, max_paths=2, **kwargs) -> list[dict]:
138		"""Get a valid path for the circuit from the Pathfinder."""
139	1	endpoint = settings.PATHFINDER_URL
140	1	spf_attribute = kwargs.get("spf_attribute") or settings.SPF_ATTRIBUTE
141	1	request_data = {
142		"source": circuit.uni_a.interface.id,
143		"destination": circuit.uni_z.interface.id,
144		"spf_max_paths": max_paths,
145		"spf_attribute": spf_attribute
146		}
147	1	request_data.update(kwargs)
148	1	api_reply = requests.post(endpoint, json=request_data)
149
150	1	if api_reply.status_code != getattr(requests.codes, "ok"):
151	1	log.error(
152		"Failed to get paths at %s. Returned %s. Payload %s. EVC %s",
153		endpoint,
154		api_reply.text,
155		request_data,
156		circuit,
157		)
158	1	return []
159	1	reply_data = api_reply.json()
160	1	return reply_data.get("paths", [])
161
162	1	@staticmethod
163	1	def _clear_path(path):
164		"""Remove switches from a path, returning only interfaces."""
165	1	return [endpoint for endpoint in path if len(endpoint) > 23]
166
167	1	@classmethod
168	1	def get_best_path(cls, circuit):
169		"""Return the best path available for a circuit, if exists."""
170	1	paths = cls.get_paths(circuit)
171	1	if paths:
172	1	return cls.create_path(cls.get_paths(circuit)[0]["hops"])
173	1	return None
174
175	1	@classmethod
176	1	def get_best_paths(cls, circuit, **kwargs):
177		"""Return the best paths available for a circuit, if they exist."""
178	1	for path in cls.get_paths(circuit, **kwargs):
179	1	yield cls.create_path(path["hops"])
180
181	1	@classmethod
182	1	def get_disjoint_paths(
183		cls, circuit, unwanted_path, cutoff=settings.DISJOINT_PATH_CUTOFF
184		):
185		"""Computes the maximum disjoint paths from the unwanted_path for a EVC
186
187		Maximum disjoint paths from the unwanted_path are the paths from the
188		source node to the target node that share the minimum number of links
189		and switches contained in unwanted_path. In other words, unwanted_path
190		is the path we want to avoid: we want the maximum possible disjoint
191		path from it. The disjointness of a path in regards to unwanted_path
192		is calculated by the complementary percentage of shared links and
193		switches between them. As an example, if the unwanted_path has 3
194		links and 2 switches, a given path P1 has 1 link shared with
195		unwanted_path, and a given path P2 has 2 links and 1 switch shared
196		with unwanted_path, then the disjointness of P1 is 0.8 and the
197		disjointness of P2 is 0.4. In this example, P1 is preferable over P2
198		because it offers a better disjoint path. When two paths have the same
199		disjointness they are ordered by 'cost' attributed as returned from
200		Pathfinder. When the disjointness of a path is equal to 0 (i.e., it
201		shares all the links with unwanted_path), that particular path is not
202		considered a candidate.
203
204		Parameters:
205		-----------
206
207		circuit : EVC
208		The EVC providing source node (uni_a) and target node (uni_z)
209
210		unwanted_path : Path
211		The Path which we want to avoid.
212
213		cutoff: int
214		Maximum number of paths to consider when calculating the disjoint
215		paths (number of paths to request from pathfinder)
216
217		Returns:
218		--------
219		paths : generator
220		Generator of unwanted_path disjoint paths. If unwanted_path is
221		not provided or empty, we return an empty list.
222		"""
223	1	unwanted_links = [
224		(link.endpoint_a.id, link.endpoint_b.id) for link in unwanted_path
225		]
226	1	unwanted_switches = set()
227	1	for link in unwanted_path:
228	1	unwanted_switches.add(link.endpoint_a.switch.id)
229	1	unwanted_switches.add(link.endpoint_b.switch.id)
230	1	unwanted_switches.discard(circuit.uni_a.interface.switch.id)
231	1	unwanted_switches.discard(circuit.uni_z.interface.switch.id)
232
233	1	length_unwanted = (len(unwanted_links) + len(unwanted_switches))
234	1	if not unwanted_links or not unwanted_switches:
235	1	return None
236
237	1	paths = cls.get_paths(circuit, max_paths=cutoff,
238		**circuit.secondary_constraints)
239	1	for path in paths:
240	1	links_n, switches_n = cls.get_shared_components(
241		path, unwanted_links, unwanted_switches
242		)
243	1	shared_components = links_n + switches_n
244	1	path["disjointness"] = 1 - shared_components / length_unwanted
245	1	paths = sorted(paths, key=lambda x: (-x['disjointness'], x['cost']))
246	1	for path in paths:
247	1	if path["disjointness"] == 0:
248	1	continue
249	1	yield cls.create_path(path["hops"])
250	1	return None
251
252	1	@staticmethod
253	1	def get_shared_components(
254		path: Path,
255		unwanted_links: list[tuple[str, str]],
256		unwanted_switches: set[str]
257		) -> tuple[int, int]:
258		"""Return the number of shared links
259		and switches found in path."""
260	1	head = path["hops"][:-1]
261	1	tail = path["hops"][1:]
262	1	shared_links = 0
263	1	for (endpoint_a, endpoint_b) in unwanted_links:
264	1	if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
265		(endpoint_b, endpoint_a) in zip(head, tail)
266		):
267	1	shared_links += 1
268	1	copy_switches = unwanted_switches.copy()
269	1	shared_switches = 0
270	1	for component in path["hops"]:
271	1	if component in copy_switches:
272	1	shared_switches += 1
273	1	copy_switches.remove(component)
274	1	return shared_links, shared_switches
275
276	1	@classmethod
277	1	def create_path(cls, path):
278		"""Return the path containing only the interfaces."""
279	1	new_path = Path()
280	1	clean_path = cls._clear_path(path)
281
282	1	if len(clean_path) % 2:
283	1	return None
284
285	1	for link in zip(clean_path[1:-1:2], clean_path[2::2]):
286	1	interface_a = cls.controller.get_interface_by_id(link[0])
287	1	interface_b = cls.controller.get_interface_by_id(link[1])
288	1	if interface_a is None or interface_b is None:
289	1	return None
290	1	new_path.append(Link(interface_a, interface_b))
291
292		return new_path
293

kytos-ng / mef_eline

Pull Request — master (#427)

build.models.path.DynamicPathManager.create_path() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like