build.utils.get_evc_unis() - Code Metrics - Inspection of "Merge pull request #27 from kytos-ng/chore/topo" - kytos-ng/telemetry_int - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7f62ad...2ec8a9 )

by Vinicius

created 2023-08-08 19:10 UTC

build.utils.get_evc_unis() A

↳ Parent: build.utils

Complexity

Conditions

Size

Total Lines	14
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	1
dl	0
loc	14
ccs	4
cts	4
cp	1
crap	1
rs	9.9
c	0
b	0
f	0

""" Support function for main.py """

import json

from napps.kytos.telemetry_int import settings

from kytos.core import Controller, log
from kytos.core.interface import Interface

from .kytos_api_helper import (
    get_evcs,
    kytos_push_flows,
    set_telemetry_metadata_false,
    set_telemetry_metadata_true,
)
from .proxy_port import ProxyPort

# mef_eline support functions


def get_evc_with_telemetry() -> dict:
    """Retrieve the list of EVC IDs and list those with
    metadata {"telemetry": {"enabled": true}}"""

    evc_ids = {"evcs_with_telemetry": []}
    for evc in get_evcs().values():
        if has_int_enabled(evc):
            evc_ids["evcs_with_telemetry"].append(evc["id"])
    return evc_ids


def has_int_enabled(evc: dict) -> bool:
    """Check if evc has telemetry."""
    return (
        "telemetry" in evc["metadata"]
        and isinstance(evc["metadata"]["telemetry"], dict)
        and "enabled" in evc["metadata"]["telemetry"]
        and evc["metadata"]["telemetry"]["enabled"]
    )


def get_evc_unis(evc: dict) -> tuple[dict, dict]:
    """Parse evc for unis."""
    uni_a_split = evc["uni_a"]["interface_id"].split(":")
    uni_z_split = evc["uni_z"]["interface_id"].split(":")
    return (
        {
            "interface_id": evc["uni_a"]["interface_id"],
            "port_number": int(uni_a_split[-1]),
            "switch": ":".join(uni_a_split[:-1]),
        },
        {
            "interface_id": evc["uni_z"]["interface_id"],
            "port_number": int(uni_z_split[-1]),
            "switch": ":".join(uni_z_split[:-1]),
        },
    )


def set_telemetry_true_for_evc(evc_id, direction):
    """Change the telemetry's enabled metadata field to true"""
    return set_telemetry_metadata_true(evc_id, direction)


def set_telemetry_false_for_evc(evc_id):
    """Change the telemetry's enabled metadata field to false"""
    return set_telemetry_metadata_false(evc_id)


def create_proxy_port(controller: Controller, interface: Interface):
    """Return the ProxyPort class to support single and multi-home loops"""
    pp = ProxyPort(controller, interface)
    return pp if pp.is_ready() else None


def get_proxy_port(controller: Controller, intf_id: str):
    """Return the proxy port assigned to a UNI"""
    interface = controller.get_interface_by_id(intf_id)
    if not interface or "proxy_port" not in interface.metadata:
        return None
    source_intf = interface.switch.get_interface_by_port_no(
        interface.metadata.get("proxy_port")
    )
    if not source_intf:
        return None
    return create_proxy_port(controller, source_intf)


def add_to_apply_actions(instructions, new_instruction, position):
    """Create the actions list"""
    for instruction in instructions:
        if instruction["instruction_type"] == "apply_actions":
            instruction["actions"].insert(position, new_instruction)
    return instructions


def get_cookie(evc_id: str, mef_cookie_prefix=settings.MEF_COOKIE_PREFIX) -> int:
    """Return the cookie integer from evc id."""
    return int(evc_id, 16) + (mef_cookie_prefix << 56)


def get_cookie_telemetry(evc_id: str, cookie_prefix=settings.COOKIE_PREFIX) -> int:
    """Return telemetry cookie given an evc_id."""
    return int(evc_id, 16) + (cookie_prefix << 56)


# pylint: disable=fixme
def get_path_hop_interface_ids(evc, source, destination):
    """
    source: {'interface_id': x, 'port_number': int, 'switch': 'x'}
    destination: {'interface_id': x, 'port_number': int, 'switch': 'x'}
    """
    # TODO double check convergence deployment optimizations later
    # TODO double check static backup path

    source_id = source["interface_id"]
    destination_id = destination["interface_id"]

    interface_ids = []

    endpoint = (
        "endpoint_b" if evc["uni_a"]["interface_id"] == source_id else "endpoint_a"
    )

    for link in evc["current_path"]:
        if (
            not link[endpoint]["switch"] in destination_id
            and not link[endpoint]["switch"] in source_id
        ):
            interface_ids.append(link[endpoint]["id"])

    for link in evc["failover_path"]:
        if (
            not link[endpoint]["switch"] in destination_id
            and not link[endpoint]["switch"] in source_id
        ):
            interface_ids.append(link[endpoint]["id"])

    return interface_ids


def get_id_from_cookie(cookie: int) -> str:
    """Return the evc id given a cookie value."""
    evc_id = cookie & 0xFFFFFFFFFFFFFF
    return f"{evc_id:x}"


def is_intra_switch_evc(evc):
    """Returns if EVC is intra-switch (two UNIs on the same switch)"""
    uni_a, uni_z = get_evc_unis(evc)
    if uni_a["switch"] == uni_z["switch"]:
        return True
    return False


def modify_actions(actions, actions_to_change, remove=True):
    """Change the current actions
    If remove == True, remove actions_to_change from actions.
    If remove == False, keep actions_to_change, remove everything else
    Args:
        actions = current list of actions on a flow
        actions_to_change = list of actions as strings
        remove = boolean
    Return
        actions
    """
    indexes = []
    count = 0

    for action in actions:
        if remove:
            if action["action_type"] in actions_to_change:
                indexes.append(count)
        else:
            if action["action_type"] not in actions_to_change:
                indexes.append(count)
        count += 1

    for index in sorted(indexes, reverse=True):
        del actions[index]
    return actions


def print_flows(flows: list[dict]) -> None:
    """For debugging purposes"""
    log.info("===================================")
    for flow in sorted(flows, key=lambda x: x["switch"]):
        log.info(json.dumps(flow, indent=4))
        log.info("===================================")


def push_flows(flows):
    """Push INT flows to the Flow_Manager via REST.
    Args:
        flows: list of flows
    Returns:
        True if successful
        False otherwise
    """

    # Debug:
    print_flows(flows)

    for flow in flows:
        flow_to_push = {"flows": [flow]}
        if not kytos_push_flows(flow["switch"], flow_to_push):
            return False
    return True


def set_priority(flow: dict) -> dict:
    """Find a suitable priority number. EP031 describes 100 as the addition."""
    if flow["flow"]["priority"] + 100 < (2**16 - 2):
        flow["flow"]["priority"] += 100
    elif flow["flow"]["priority"] + 1 < (2**16 - 2):
        flow["flow"]["priority"] += 1
    else:
        raise ValueError(f"Flow {flow} would overflow max priority")
    return flow


def get_new_cookie(cookie: int, cookie_prefix=settings.COOKIE_PREFIX) -> int:
    """Convert from mef-eline cookie by replacing the most significant byte."""
    return (cookie & 0xFFFFFFFFFFFFFF) + (cookie_prefix << 56)


def set_new_cookie(flow: dict) -> dict:
    """Set new cookie."""
    flow["flow"]["cookie"] = get_new_cookie(flow["flow"]["cookie"])
    return flow


def set_instructions_from_actions(flow: dict) -> dict:
    """Get intructions or convert from actions."""
    if "instructions" in flow["flow"]:
        return flow

    instructions = [
        {
            "instruction_type": "apply_actions",
            "actions": flow["flow"].get("actions", []),
        }
    ]
    flow["flow"].pop("actions", None)
    flow["flow"]["instructions"] = instructions
    return flow


1		""" Support function for main.py """
2
3	1	import json
4
5	1	from napps.kytos.telemetry_int import settings
6
7	1	from kytos.core import Controller, log
8	1	from kytos.core.interface import Interface
9
10	1	from .kytos_api_helper import (
11		get_evcs,
12		kytos_push_flows,
13		set_telemetry_metadata_false,
14		set_telemetry_metadata_true,
15		)
16	1	from .proxy_port import ProxyPort
17
18		# mef_eline support functions
19
20
21	1	def get_evc_with_telemetry() -> dict:
22		"""Retrieve the list of EVC IDs and list those with
23		metadata {"telemetry": {"enabled": true}}"""
24
25		evc_ids = {"evcs_with_telemetry": []}
26		for evc in get_evcs().values():
27		if has_int_enabled(evc):
28		evc_ids["evcs_with_telemetry"].append(evc["id"])
29		return evc_ids
30
31
32	1	def has_int_enabled(evc: dict) -> bool:
33		"""Check if evc has telemetry."""
34	1	return (
35		"telemetry" in evc["metadata"]
36		and isinstance(evc["metadata"]["telemetry"], dict)
37		and "enabled" in evc["metadata"]["telemetry"]
38		and evc["metadata"]["telemetry"]["enabled"]
39		)
40
41
42	1	def get_evc_unis(evc: dict) -> tuple[dict, dict]:
43		"""Parse evc for unis."""
44	1	uni_a_split = evc["uni_a"]["interface_id"].split(":")
45	1	uni_z_split = evc["uni_z"]["interface_id"].split(":")
46	1	return (
47		{
48		"interface_id": evc["uni_a"]["interface_id"],
49		"port_number": int(uni_a_split[-1]),
50		"switch": ":".join(uni_a_split[:-1]),
51		},
52		{
53		"interface_id": evc["uni_z"]["interface_id"],
54		"port_number": int(uni_z_split[-1]),
55		"switch": ":".join(uni_z_split[:-1]),
56		},
57		)
58
59
60	1	def set_telemetry_true_for_evc(evc_id, direction):
61		"""Change the telemetry's enabled metadata field to true"""
62		return set_telemetry_metadata_true(evc_id, direction)
63
64
65	1	def set_telemetry_false_for_evc(evc_id):
66		"""Change the telemetry's enabled metadata field to false"""
67		return set_telemetry_metadata_false(evc_id)
68
69
70	1	def create_proxy_port(controller: Controller, interface: Interface):
71		"""Return the ProxyPort class to support single and multi-home loops"""
72		pp = ProxyPort(controller, interface)
73		return pp if pp.is_ready() else None
74
75
76	1	def get_proxy_port(controller: Controller, intf_id: str):
77		"""Return the proxy port assigned to a UNI"""
78		interface = controller.get_interface_by_id(intf_id)
79		if not interface or "proxy_port" not in interface.metadata:
80		return None
81		source_intf = interface.switch.get_interface_by_port_no(
82		interface.metadata.get("proxy_port")
83		)
84		if not source_intf:
85		return None
86		return create_proxy_port(controller, source_intf)
87
88
89	1	def add_to_apply_actions(instructions, new_instruction, position):
90		"""Create the actions list"""
91		for instruction in instructions:
92		if instruction["instruction_type"] == "apply_actions":
93		instruction["actions"].insert(position, new_instruction)
94		return instructions
95
96
97	1	def get_cookie(evc_id: str, mef_cookie_prefix=settings.MEF_COOKIE_PREFIX) -> int:
98		"""Return the cookie integer from evc id."""
99		return int(evc_id, 16) + (mef_cookie_prefix << 56)
100
101
102	1	def get_cookie_telemetry(evc_id: str, cookie_prefix=settings.COOKIE_PREFIX) -> int:
103		"""Return telemetry cookie given an evc_id."""
104		return int(evc_id, 16) + (cookie_prefix << 56)
105
106
107		# pylint: disable=fixme
108	1	def get_path_hop_interface_ids(evc, source, destination):
109		"""
110		source: {'interface_id': x, 'port_number': int, 'switch': 'x'}
111		destination: {'interface_id': x, 'port_number': int, 'switch': 'x'}
112		"""
113		# TODO double check convergence deployment optimizations later
114		# TODO double check static backup path
115
116		source_id = source["interface_id"]
117		destination_id = destination["interface_id"]
118
119		interface_ids = []
120
121		endpoint = (
122		"endpoint_b" if evc["uni_a"]["interface_id"] == source_id else "endpoint_a"
123		)
124
125		for link in evc["current_path"]:
126		if (
127		not link[endpoint]["switch"] in destination_id
128		and not link[endpoint]["switch"] in source_id
129		):
130		interface_ids.append(link[endpoint]["id"])
131
132		for link in evc["failover_path"]:
133		if (
134		not link[endpoint]["switch"] in destination_id
135		and not link[endpoint]["switch"] in source_id
136		):
137		interface_ids.append(link[endpoint]["id"])
138
139		return interface_ids
140
141
142	1	def get_id_from_cookie(cookie: int) -> str:
143		"""Return the evc id given a cookie value."""
144	1	evc_id = cookie & 0xFFFFFFFFFFFFFF
145	1	return f"{evc_id:x}"
146
147
148	1	def is_intra_switch_evc(evc):
149		"""Returns if EVC is intra-switch (two UNIs on the same switch)"""
150	1	uni_a, uni_z = get_evc_unis(evc)
151	1	if uni_a["switch"] == uni_z["switch"]:
152	1	return True
153	1	return False
154
155
156	1	def modify_actions(actions, actions_to_change, remove=True):
157		"""Change the current actions
158		If remove == True, remove actions_to_change from actions.
159		If remove == False, keep actions_to_change, remove everything else
160		Args:
161		actions = current list of actions on a flow
162		actions_to_change = list of actions as strings
163		remove = boolean
164		Return
165		actions
166		"""
167		indexes = []
168		count = 0
169
170		for action in actions:
171		if remove:
172		if action["action_type"] in actions_to_change:
173		indexes.append(count)
174		else:
175		if action["action_type"] not in actions_to_change:
176		indexes.append(count)
177		count += 1
178
179		for index in sorted(indexes, reverse=True):
180		del actions[index]
181		return actions
182
183
184	1	def print_flows(flows: list[dict]) -> None:
185		"""For debugging purposes"""
186		log.info("===================================")
187		for flow in sorted(flows, key=lambda x: x["switch"]):
188		log.info(json.dumps(flow, indent=4))
189		log.info("===================================")
190
191
192	1	def push_flows(flows):
193		"""Push INT flows to the Flow_Manager via REST.
194		Args:
195		flows: list of flows
196		Returns:
197		True if successful
198		False otherwise
199		"""
200
201		# Debug:
202		print_flows(flows)
203
204		for flow in flows:
205		flow_to_push = {"flows": [flow]}
206		if not kytos_push_flows(flow["switch"], flow_to_push):
207		return False
208		return True
209
210
211	1	def set_priority(flow: dict) -> dict:
212		"""Find a suitable priority number. EP031 describes 100 as the addition."""
213		if flow["flow"]["priority"] + 100 < (2**16 - 2):
214		flow["flow"]["priority"] += 100
215		elif flow["flow"]["priority"] + 1 < (2**16 - 2):
216		flow["flow"]["priority"] += 1
217		else:
218		raise ValueError(f"Flow {flow} would overflow max priority")
219		return flow
220
221
222	1	def get_new_cookie(cookie: int, cookie_prefix=settings.COOKIE_PREFIX) -> int:
223		"""Convert from mef-eline cookie by replacing the most significant byte."""
224	1	return (cookie & 0xFFFFFFFFFFFFFF) + (cookie_prefix << 56)
225
226
227	1	def set_new_cookie(flow: dict) -> dict:
228		"""Set new cookie."""
229	1	flow["flow"]["cookie"] = get_new_cookie(flow["flow"]["cookie"])
230	1	return flow
231
232
233	1	def set_instructions_from_actions(flow: dict) -> dict:
234		"""Get intructions or convert from actions."""
235	1	if "instructions" in flow["flow"]:
236	1	return flow
237
238	1	instructions = [
239		{
240		"instruction_type": "apply_actions",
241		"actions": flow["flow"].get("actions", []),
242		}
243		]
244	1	flow["flow"].pop("actions", None)
245	1	flow["flow"]["instructions"] = instructions
246		return flow
247

kytos-ng / telemetry_int

Push — master ( 7f62ad...2ec8a9 )

build.utils.get_evc_unis() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like