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

Test Failed

Pull Request — master (#427)

unknown

created 2024-02-06 14:10 UTC

DynamicPathManager.get_disjoint_paths() D

↳ Parent: build.models.path

Complexity

Conditions

Size

Total Lines	79
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	30
CRAP Score	13

Importance

Changes

Metric	Value
cc	13
eloc	36
nop	4
dl	0
loc	79
ccs	30
cts	30
cp	1
crap	13
rs	4.2
c	0
b	0
f	0

How to fix Long Method Complexity

"""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)

        unwanted_components_n = (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:
            head = path["hops"][:-1]
            tail = path["hops"][1:]
            copy_switches = unwanted_switches.copy()
            shared_components = 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_components += 1
            for component in path["hops"]:
                if component in copy_switches:
                    shared_components += 1
                    copy_switches.remove(component)
            path["disjointness"] = 1 - shared_components / unwanted_components_n
        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

    @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	1	"""
223		unwanted_links = [
224		(link.endpoint_a.id, link.endpoint_b.id) for link in unwanted_path
225	1	]
226	1	unwanted_switches = set()
227		for link in unwanted_path:
228	1	unwanted_switches.add(link.endpoint_a.switch.id)
229		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	1
233	1	unwanted_components_n = (len(unwanted_links) + len(unwanted_switches))
234	1	if not unwanted_links or not unwanted_switches:
235	1	return None
236
237		paths = cls.get_paths(circuit, max_paths=cutoff,
238	1	**circuit.secondary_constraints)
239	1	for path in paths:
240	1	head = path["hops"][:-1]
241	1	tail = path["hops"][1:]
242	1	copy_switches = unwanted_switches.copy()
243	1	shared_components = 0
244	1	for (endpoint_a, endpoint_b) in unwanted_links:
245	1	if ((endpoint_a, endpoint_b) in zip(head, tail)) or (
246		(endpoint_b, endpoint_a) in zip(head, tail)
247	1	):
248	1	shared_components += 1
249		for component in path["hops"]:
250	1	if component in copy_switches:
251	1	shared_components += 1
252		copy_switches.remove(component)
253	1	path["disjointness"] = 1 - shared_components / unwanted_components_n
254	1	paths = sorted(paths, key=lambda x: (-x['disjointness'], x['cost']))
255		for path in paths:
256	1	if path["disjointness"] == 0:
257	1	continue
258	1	yield cls.create_path(path["hops"])
259	1	return None
260	1
261	1	@classmethod
262		def create_path(cls, path):
263	1	"""Return the path containing only the interfaces."""
264		new_path = Path()
265		clean_path = cls._clear_path(path)
266
267		if len(clean_path) % 2:
268		return None
269
270		for link in zip(clean_path[1:-1:2], clean_path[2::2]):
271		interface_a = cls.controller.get_interface_by_id(link[0])
272		interface_b = cls.controller.get_interface_by_id(link[1])
273		if interface_a is None or interface_b is None:
274		return None
275		new_path.append(Link(interface_a, interface_b))
276
277		return new_path
278

kytos-ng / mef_eline

Pull Request — master (#427)

DynamicPathManager.get_disjoint_paths() D

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like