build.main.Main.tracepath() - Code Metrics - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

build.main.Main.tracepath() F
last analyzed 2024-03-28 14:34 UTC

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	54
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	37
CRAP Score	15.0949

Importance

Changes

Metric	Value
eloc	46
dl	0
loc	54
ccs	37
cts	40
cp	0.925
rs	2.9998
c	0
b	0
f	0
cc	15
nop	3
crap	15.0949

How to fix Long Method Complexity

"""Main module of amlight/sdntrace_cp Kytos Network Application.

Run tracepaths on OpenFlow in the Control Plane
"""

import pathlib
from datetime import datetime

import tenacity
from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import load_spec, validate_openapi
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 get_json_or_400)
from napps.amlight.sdntrace_cp.utils import (convert_entries,
                                             convert_list_entries,
                                             find_endpoint, get_stored_flows,
                                             match_field_dl_vlan,
                                             match_field_ip, prepare_json)


class Main(KytosNApp):
    """Main class of amlight/sdntrace_cp NApp.

    This application gets the list of flows from the switches
    and uses it to trace paths without using the data plane.
    """

    spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")

    def setup(self):
        """Replace the '__init__' method for the KytosNApp subclass.

        The setup method is automatically called by the controller when your
        application is loaded.

        """
        log.info("Starting Kytos SDNTrace CP App!")

    def execute(self):
        """This method is executed right after the setup method execution.

        You can also use this method in loop mode if you add to the above setup
        method a line like the following example:

            self.execute_as_loop(30)  # 30-second interval.
        """

    def shutdown(self):
        """This method is executed when your napp is unloaded.

        If you have some cleanup procedure, insert it here.
        """

    @rest('/v1/trace', methods=['PUT'])

    @validate_openapi(spec)
    def trace(self, request: Request) -> JSONResponse:
        """Trace a path."""
        result = []
        data = get_json_or_400(request, self.controller.loop)
        entries = convert_entries(data)
        if not entries:
            raise HTTPException(400, "Empty entries")
        try:
            stored_flows = get_stored_flows()
        except tenacity.RetryError as exc:
            raise HTTPException(424, "It couldn't get stored_flows") from exc
        try:
            result = self.tracepath(entries, stored_flows)
        except ValueError as exc:
            raise HTTPException(409, str(exc)) from exc
        return JSONResponse(prepare_json(result))

    @rest('/v1/traces', methods=['PUT'])

    @validate_openapi(spec)
    def get_traces(self, request: Request) -> JSONResponse:
        """For bulk requests."""
        data = get_json_or_400(request, self.controller.loop)
        entries = convert_list_entries(data)
        results = []
        try:
            stored_flows = get_stored_flows()
        except tenacity.RetryError as exc:
            raise HTTPException(424, "It couldn't get stored_flows") from exc
        for entry in entries:
            try:
                results.append(self.tracepath(entry, stored_flows))
            except ValueError as exc:
                raise HTTPException(409, str(exc)) from exc
        return JSONResponse(prepare_json(results))

    def tracepath(self, entries, stored_flows):
        """Trace a path for a packet represented by entries."""
        # pylint: disable=too-many-branches
        trace_result = []
        trace_type = 'starting'
        do_trace = True
        while do_trace:
            if 'dpid' not in entries or 'in_port' not in entries:
                break
            trace_step = {'in': {'dpid': entries['dpid'],
                                 'port': entries['in_port'],
                                 'time': str(datetime.now()),
                                 'type': trace_type}}
            if 'dl_vlan' in entries:
                trace_step['in'].update({'vlan': entries['dl_vlan'][-1]})

            switch = self.controller.get_switch_by_dpid(entries['dpid'])
            if not switch:
                trace_step['in']['type'] = 'last'
                trace_result.append(trace_step)
                break
            result = self.trace_step(switch, entries, stored_flows)
            if result:
                out = {'port': result['out_port']}
                if 'dl_vlan' in result['entries']:
                    out.update({'vlan': result['entries']['dl_vlan'][-1]})
                trace_step.update({
                    'out': out
                })
                if 'dpid' in result:
                    next_step = {'dpid': result['dpid'],
                                 'port': result['in_port']}
                    entries = result['entries']
                    entries['dpid'] = result['dpid']
                    entries['in_port'] = result['in_port']
                    if self.has_loop(next_step, trace_result):
                        trace_step['in']['type'] = 'loop'
                        do_trace = False
                    else:
                        trace_type = 'intermediary'
                else:
                    trace_step['in']['type'] = 'last'
                    do_trace = False
            else:
                # No match
                break
            if 'out' in trace_step and trace_step['out']:
                if self.check_loop_trace_step(trace_step, trace_result):
                    do_trace = False
            trace_result.append(trace_step)
        if len(trace_result) == 1 and \
                trace_result[0]['in']['type'] == 'starting':
            trace_result[0]['in']['type'] = 'last'
        return trace_result

    @staticmethod
    def check_loop_trace_step(trace_step, trace_result):
        """Check if there is a loop in the trace and add the step."""
        # outgoing interface is the same as the input interface
        if not trace_result and \
                trace_step['in']['type'] == 'last' and \
                trace_step['in']['port'] == trace_step['out']['port']:
            trace_step['in']['type'] = 'loop'
            return True
        if trace_result and \
                trace_result[0]['in']['dpid'] == trace_step['in']['dpid'] and \
                trace_result[0]['in']['port'] == trace_step['out']['port']:
            trace_step['in']['type'] = 'loop'
            return True
        return False

    @staticmethod
    def has_loop(trace_step, trace_result):
        """Check if there is a loop in the trace result."""
        for trace in trace_result:
            if trace['in']['dpid'] == trace_step['dpid'] and \
                            trace['in']['port'] == trace_step['port']:
                return True
        return False

    def trace_step(self, switch, entries, stored_flows):
        """Perform a trace step.

        Match the given fields against the switch's list of flows."""
        flow, entries, port = self.match_and_apply(
                                                    switch,
                                                    entries,
                                                    stored_flows
                                                )

        if not flow or not port:
            return None

        endpoint = find_endpoint(switch, port)
        if endpoint is None:
            log.warning(f"Port {port} not found on switch {switch}")
            return None
        endpoint = endpoint['endpoint']
        if endpoint is None:
            return {'out_port': port,
                    'entries': entries}

        return {'dpid': endpoint.switch.dpid,
                'in_port': endpoint.port_number,
                'out_port': port,
                'entries': entries}

    @classmethod
    def do_match(cls, flow, args, table_id):
        """Match a packet against this flow (OF1.3)."""
        # pylint: disable=consider-using-dict-items
        # pylint: disable=too-many-return-statements
        if ('match' not in flow['flow']) or (len(flow['flow']['match']) == 0):
            return False
        table_id_ = flow['flow'].get('table_id', 0)
        if table_id != table_id_:
            return False
        for name in flow['flow']['match']:
            field_flow = flow['flow']['match'][name]
            field = args.get(name)
            if name == 'dl_vlan':
                if not match_field_dl_vlan(field, field_flow):
                    return False
                continue
            # In the case of dl_vlan field, the match must be checked
            # even if this field is not in the packet args.
            if not field:
                return False
            if name in ('nw_src', 'nw_dst', 'ipv6_src', 'ipv6_dst'):
                if not match_field_ip(field, field_flow):
                    return False
                continue
            if field_flow != field:
                return False
        return flow

    # pylint: disable=too-many-arguments
    def match_flows(self, switch, table_id, args, stored_flows, many=True):
        """
        Match the packet in request against the stored flows from flow_manager.
        Try the match with each flow, in other. If many is True, tries the
        match with all flows, if False, tries until the first match.
        :param args: packet data
        :param many: Boolean, indicating whether to continue after matching the
                first flow or not
        :return: If many, the list of matched flows, or the matched flow
        """
        if switch.dpid not in stored_flows:
            return None
        response = []
        if switch.dpid not in stored_flows:
            return None
        try:
            for flow in stored_flows[switch.dpid]:
                match = Main.do_match(flow, args, table_id)
                if match:
                    if many:
                        response.append(match)
                    else:
                        response = match
                        break
        except AttributeError:
            return None
        if not many and isinstance(response, list):
            return None
        return response

    def process_tables(self, switch, table_id, args, stored_flows, actions):
        """Resolve the table context and get actions in the matched flow"""
        goto_table = False
        actions_ = []
        flow = self.match_flows(switch, table_id, args, stored_flows, False)
        if flow and 'actions' in flow['flow']:
            actions_ = flow['flow']['actions']
        elif flow and 'instructions' in flow['flow']:
            for instruction in flow['flow']['instructions']:
                if instruction['instruction_type'] == 'apply_actions':
                    actions_ = instruction['actions']
                elif instruction['instruction_type'] == 'goto_table':
                    table_id_ = instruction['table_id']
                    if table_id < table_id_:
                        table_id = table_id_
                        goto_table = True
                    else:
                        msg = f"Wrong table_id in {flow['flow']}: \
                            The packet can only been directed to a \
                                flow table number greather than {table_id}"
                        raise ValueError(msg) from ValueError
        actions.extend(actions_)
        return flow, actions, goto_table, table_id

    def match_and_apply(self, switch, args, stored_flows):
        """Match flows and apply actions.
        Match given packet (in args) against
        the stored flows (from flow_manager) and,
        if a match flow is found, apply its actions."""
        table_id = 0
        goto_table = True
        port = None
        actions = []
        while goto_table:
            try:
                flow, actions, goto_table, table_id = self.process_tables(
                    switch, table_id, args, stored_flows, actions)
            except ValueError as exception:
                raise exception
        if not flow or switch.ofp_version != '0x04':

            return flow, args, port

        for action in actions:
            action_type = action['action_type']
            if action_type == 'output':
                port = action['port']
            if action_type == 'push_vlan':
                if 'dl_vlan' not in args:
                    args['dl_vlan'] = []
                args['dl_vlan'].append(0)
            if action_type == 'pop_vlan':
                if 'dl_vlan' in args:
                    args['dl_vlan'].pop()
                    if len(args['dl_vlan']) == 0:
                        del args['dl_vlan']
            if action_type == 'set_vlan':
                args['dl_vlan'][-1] = action['vlan_id']
        return flow, args, port


1			"""Main module of amlight/sdntrace_cp Kytos Network Application.
2
3			Run tracepaths on OpenFlow in the Control Plane
4			"""
5
6	1		import pathlib
7	1		from datetime import datetime
8
9	1		import tenacity
10	1		from kytos.core import KytosNApp, log, rest
11	1		from kytos.core.helpers import load_spec, validate_openapi
12	1		from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
13			get_json_or_400)
14	1		from napps.amlight.sdntrace_cp.utils import (convert_entries,
15			convert_list_entries,
16			find_endpoint, get_stored_flows,
17			match_field_dl_vlan,
18			match_field_ip, prepare_json)
19
20
21	1		class Main(KytosNApp):
22			"""Main class of amlight/sdntrace_cp NApp.
23
24			This application gets the list of flows from the switches
25			and uses it to trace paths without using the data plane.
26			"""
27
28	1		spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")
29
30	1		def setup(self):
31			"""Replace the '__init__' method for the KytosNApp subclass.
32
33			The setup method is automatically called by the controller when your
34			application is loaded.
35
36			"""
37	1		log.info("Starting Kytos SDNTrace CP App!")
38
39	1		def execute(self):
40			"""This method is executed right after the setup method execution.
41
42			You can also use this method in loop mode if you add to the above setup
43			method a line like the following example:
44
45			self.execute_as_loop(30) # 30-second interval.
46			"""
47
48	1		def shutdown(self):
49			"""This method is executed when your napp is unloaded.
50
51			If you have some cleanup procedure, insert it here.
52			"""
53
54	1	View Code Duplication	@rest('/v1/trace', methods=['PUT'])
			0 ignored issues – show Duplication introduced 2023-10-11 20:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
55	1		@validate_openapi(spec)
56	1		def trace(self, request: Request) -> JSONResponse:
57			"""Trace a path."""
58	1		result = []
59	1		data = get_json_or_400(request, self.controller.loop)
60	1		entries = convert_entries(data)
61	1		if not entries:
62			raise HTTPException(400, "Empty entries")
63	1		try:
64	1		stored_flows = get_stored_flows()
65			except tenacity.RetryError as exc:
66			raise HTTPException(424, "It couldn't get stored_flows") from exc
67	1		try:
68	1		result = self.tracepath(entries, stored_flows)
69			except ValueError as exc:
70			raise HTTPException(409, str(exc)) from exc
71	1		return JSONResponse(prepare_json(result))
72
73	1	View Code Duplication	@rest('/v1/traces', methods=['PUT'])
			0 ignored issues – show Duplication introduced 2023-10-11 20:56 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
74	1		@validate_openapi(spec)
75	1		def get_traces(self, request: Request) -> JSONResponse:
76			"""For bulk requests."""
77	1		data = get_json_or_400(request, self.controller.loop)
78	1		entries = convert_list_entries(data)
79	1		results = []
80	1		try:
81	1		stored_flows = get_stored_flows()
82	1		except tenacity.RetryError as exc:
83			raise HTTPException(424, "It couldn't get stored_flows") from exc
84	1		for entry in entries:
85	1		try:
86	1		results.append(self.tracepath(entry, stored_flows))
87	1		except ValueError as exc:
88	1		raise HTTPException(409, str(exc)) from exc
89	1		return JSONResponse(prepare_json(results))
90
91	1		def tracepath(self, entries, stored_flows):
92			"""Trace a path for a packet represented by entries."""
93			# pylint: disable=too-many-branches
94	1		trace_result = []
95	1		trace_type = 'starting'
96	1		do_trace = True
97	1		while do_trace:
98	1		if 'dpid' not in entries or 'in_port' not in entries:
99			break
100	1		trace_step = {'in': {'dpid': entries['dpid'],
101			'port': entries['in_port'],
102			'time': str(datetime.now()),
103			'type': trace_type}}
104	1		if 'dl_vlan' in entries:
105	1		trace_step['in'].update({'vlan': entries['dl_vlan'][-1]})
106
107	1		switch = self.controller.get_switch_by_dpid(entries['dpid'])
108	1		if not switch:
109	1		trace_step['in']['type'] = 'last'
110	1		trace_result.append(trace_step)
111	1		break
112	1		result = self.trace_step(switch, entries, stored_flows)
113	1		if result:
114	1		out = {'port': result['out_port']}
115	1		if 'dl_vlan' in result['entries']:
116	1		out.update({'vlan': result['entries']['dl_vlan'][-1]})
117	1		trace_step.update({
118			'out': out
119			})
120	1		if 'dpid' in result:
121	1		next_step = {'dpid': result['dpid'],
122			'port': result['in_port']}
123	1		entries = result['entries']
124	1		entries['dpid'] = result['dpid']
125	1		entries['in_port'] = result['in_port']
126	1		if self.has_loop(next_step, trace_result):
127	1		trace_step['in']['type'] = 'loop'
128	1		do_trace = False
129			else:
130	1		trace_type = 'intermediary'
131			else:
132	1		trace_step['in']['type'] = 'last'
133	1		do_trace = False
134			else:
135			# No match
136			break
137	1		if 'out' in trace_step and trace_step['out']:
138	1		if self.check_loop_trace_step(trace_step, trace_result):
139	1		do_trace = False
140	1		trace_result.append(trace_step)
141	1		if len(trace_result) == 1 and \
142			trace_result[0]['in']['type'] == 'starting':
143			trace_result[0]['in']['type'] = 'last'
144	1		return trace_result
145
146	1		@staticmethod
147	1		def check_loop_trace_step(trace_step, trace_result):
148			"""Check if there is a loop in the trace and add the step."""
149			# outgoing interface is the same as the input interface
150	1		if not trace_result and \
151			trace_step['in']['type'] == 'last' and \
152			trace_step['in']['port'] == trace_step['out']['port']:
153	1		trace_step['in']['type'] = 'loop'
154	1		return True
155	1		if trace_result and \
156			trace_result[0]['in']['dpid'] == trace_step['in']['dpid'] and \
157			trace_result[0]['in']['port'] == trace_step['out']['port']:
158	1		trace_step['in']['type'] = 'loop'
159	1		return True
160	1		return False
161
162	1		@staticmethod
163	1		def has_loop(trace_step, trace_result):
164			"""Check if there is a loop in the trace result."""
165	1		for trace in trace_result:
166	1		if trace['in']['dpid'] == trace_step['dpid'] and \
167			trace['in']['port'] == trace_step['port']:
168	1		return True
169	1		return False
170
171	1		def trace_step(self, switch, entries, stored_flows):
172			"""Perform a trace step.
173
174			Match the given fields against the switch's list of flows."""
175	1		flow, entries, port = self.match_and_apply(
176			switch,
177			entries,
178			stored_flows
179			)
180
181	1		if not flow or not port:
182	1		return None
183
184	1		endpoint = find_endpoint(switch, port)
185	1		if endpoint is None:
186			log.warning(f"Port {port} not found on switch {switch}")
187			return None
188	1		endpoint = endpoint['endpoint']
189	1		if endpoint is None:
190	1		return {'out_port': port,
191			'entries': entries}
192
193	1		return {'dpid': endpoint.switch.dpid,
194			'in_port': endpoint.port_number,
195			'out_port': port,
196			'entries': entries}
197
198	1		@classmethod
199	1		def do_match(cls, flow, args, table_id):
200			"""Match a packet against this flow (OF1.3)."""
201			# pylint: disable=consider-using-dict-items
202			# pylint: disable=too-many-return-statements
203	1		if ('match' not in flow['flow']) or (len(flow['flow']['match']) == 0):
204	1		return False
205	1		table_id_ = flow['flow'].get('table_id', 0)
206	1		if table_id != table_id_:
207	1		return False
208	1		for name in flow['flow']['match']:
209	1		field_flow = flow['flow']['match'][name]
210	1		field = args.get(name)
211	1		if name == 'dl_vlan':
212	1		if not match_field_dl_vlan(field, field_flow):
213	1		return False
214			continue
215			# In the case of dl_vlan field, the match must be checked
216			# even if this field is not in the packet args.
217	1		if not field:
218	1		return False
219	1		if name in ('nw_src', 'nw_dst', 'ipv6_src', 'ipv6_dst'):
220	1		if not match_field_ip(field, field_flow):
221	1		return False
222			continue
223	1		if field_flow != field:
224	1		return False
225	1		return flow
226
227			# pylint: disable=too-many-arguments
228	1		def match_flows(self, switch, table_id, args, stored_flows, many=True):
229			"""
230			Match the packet in request against the stored flows from flow_manager.
231			Try the match with each flow, in other. If many is True, tries the
232			match with all flows, if False, tries until the first match.
233			:param args: packet data
234			:param many: Boolean, indicating whether to continue after matching the
235			first flow or not
236			:return: If many, the list of matched flows, or the matched flow
237			"""
238	1		if switch.dpid not in stored_flows:
239	1		return None
240	1		response = []
241	1		if switch.dpid not in stored_flows:
242			return None
243	1		try:
244	1		for flow in stored_flows[switch.dpid]:
245	1		match = Main.do_match(flow, args, table_id)
246	1		if match:
247	1		if many:
248	1		response.append(match)
249			else:
250	1		response = match
251	1		break
252			except AttributeError:
253			return None
254	1		if not many and isinstance(response, list):
255	1		return None
256	1		return response
257
258	1		def process_tables(self, switch, table_id, args, stored_flows, actions):
259			"""Resolve the table context and get actions in the matched flow"""
260	1		goto_table = False
261	1		actions_ = []
262	1		flow = self.match_flows(switch, table_id, args, stored_flows, False)
263	1		if flow and 'actions' in flow['flow']:
264	1		actions_ = flow['flow']['actions']
265	1		elif flow and 'instructions' in flow['flow']:
266	1		for instruction in flow['flow']['instructions']:
267	1		if instruction['instruction_type'] == 'apply_actions':
268	1		actions_ = instruction['actions']
269	1		elif instruction['instruction_type'] == 'goto_table':
270	1		table_id_ = instruction['table_id']
271	1		if table_id < table_id_:
272	1		table_id = table_id_
273	1		goto_table = True
274			else:
275	1		msg = f"Wrong table_id in {flow['flow']}: \
276			The packet can only been directed to a \
277			flow table number greather than {table_id}"
278	1		raise ValueError(msg) from ValueError
279	1		actions.extend(actions_)
280	1		return flow, actions, goto_table, table_id
281
282	1		def match_and_apply(self, switch, args, stored_flows):
283			"""Match flows and apply actions.
284			Match given packet (in args) against
285			the stored flows (from flow_manager) and,
286			if a match flow is found, apply its actions."""
287	1		table_id = 0
288	1		goto_table = True
289	1		port = None
290	1		actions = []
291	1		while goto_table:
292	1		try:
293	1		flow, actions, goto_table, table_id = self.process_tables(
294			switch, table_id, args, stored_flows, actions)
295	1		except ValueError as exception:
296	1		raise exception
297	1		if not flow or switch.ofp_version != '0x04':
			0 ignored issues – show introduced 2023-09-11 17:46 UTC by Report Bug Copy Issue Report The variable `flow` does not seem to be defined in case the `while` loop on line `291` is not entered. Are you sure this can never be the case? Loading history...
298	1		return flow, args, port
299
300	1		for action in actions:
301	1		action_type = action['action_type']
302	1		if action_type == 'output':
303	1		port = action['port']
304	1		if action_type == 'push_vlan':
305	1		if 'dl_vlan' not in args:
306			args['dl_vlan'] = []
307	1		args['dl_vlan'].append(0)
308	1		if action_type == 'pop_vlan':
309	1		if 'dl_vlan' in args:
310	1		args['dl_vlan'].pop()
311	1		if len(args['dl_vlan']) == 0:
312	1		del args['dl_vlan']
313	1		if action_type == 'set_vlan':
314	1		args['dl_vlan'][-1] = action['vlan_id']
315			return flow, args, port
316

kytos-ng / sdntrace_cp

build.main.Main.tracepath() F last analyzed 2024-03-28 14:34 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

build.main.Main.tracepath() F
last analyzed 2024-03-28 14:34 UTC