build.main.Main.match_and_apply() - Code Metrics - Inspection of "Merge pull request #107 from kytos-ng/feat/instruc..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0bafa5...8e49b2 )

unknown

created 2023-08-23 16:29 UTC

build.main.Main.match_and_apply() F

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	34
Code Lines	28

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	27
CRAP Score	15.0102

Importance

Changes

Metric	Value
eloc	28
dl	0
loc	34
ccs	27
cts	28
cp	0.9643
rs	2.9998
c	0
b	0
f	0
cc	15
nop	4
crap	15.0102

How to fix 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
        result = self.tracepath(entries, stored_flows)
        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:
            results.append(self.tracepath(entry, stored_flows))
        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):
        """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
        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

    def match_flows(self, switch, args, stored_flows, many=True):
        # pylint: disable=bad-staticmethod-argument
        """
        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)
                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

    # pylint: disable=redefined-outer-name
    # pylint: disable=too-many-branches
    def match_and_apply(self, switch, args, stored_flows):
        # pylint: disable=bad-staticmethod-argument
        """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."""
        flow = self.match_flows(switch, args, stored_flows, False)
        port = None
        actions = []
        # pylint: disable=too-many-nested-blocks
        if not flow or switch.ofp_version != '0x04':
            return flow, args, port
        if 'actions' in flow['flow']:
            actions = flow['flow']['actions']
        elif 'instructions' in flow['flow']:
            for instruction in flow['flow']['instructions']:
                if instruction['instruction_type'] == 'apply_actions':
                    actions = instruction['actions']
        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	@rest('/v1/trace', methods=['PUT'])
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	result = self.tracepath(entries, stored_flows)
68	1	return JSONResponse(prepare_json(result))
69
70	1	@rest('/v1/traces', methods=['PUT'])
71	1	@validate_openapi(spec)
72	1	def get_traces(self, request: Request) -> JSONResponse:
73		"""For bulk requests."""
74	1	data = get_json_or_400(request, self.controller.loop)
75	1	entries = convert_list_entries(data)
76	1	results = []
77	1	try:
78	1	stored_flows = get_stored_flows()
79	1	except tenacity.RetryError as exc:
80		raise HTTPException(424, "It couldn't get stored_flows") from exc
81	1	for entry in entries:
82	1	results.append(self.tracepath(entry, stored_flows))
83	1	return JSONResponse(prepare_json(results))
84
85	1	def tracepath(self, entries, stored_flows):
86		"""Trace a path for a packet represented by entries."""
87		# pylint: disable=too-many-branches
88	1	trace_result = []
89	1	trace_type = 'starting'
90	1	do_trace = True
91	1	while do_trace:
92	1	if 'dpid' not in entries or 'in_port' not in entries:
93		break
94	1	trace_step = {'in': {'dpid': entries['dpid'],
95		'port': entries['in_port'],
96		'time': str(datetime.now()),
97		'type': trace_type}}
98	1	if 'dl_vlan' in entries:
99	1	trace_step['in'].update({'vlan': entries['dl_vlan'][-1]})
100
101	1	switch = self.controller.get_switch_by_dpid(entries['dpid'])
102	1	if not switch:
103	1	trace_step['in']['type'] = 'last'
104	1	trace_result.append(trace_step)
105	1	break
106	1	result = self.trace_step(switch, entries, stored_flows)
107	1	if result:
108	1	out = {'port': result['out_port']}
109	1	if 'dl_vlan' in result['entries']:
110	1	out.update({'vlan': result['entries']['dl_vlan'][-1]})
111	1	trace_step.update({
112		'out': out
113		})
114	1	if 'dpid' in result:
115	1	next_step = {'dpid': result['dpid'],
116		'port': result['in_port']}
117	1	entries = result['entries']
118	1	entries['dpid'] = result['dpid']
119	1	entries['in_port'] = result['in_port']
120	1	if self.has_loop(next_step, trace_result):
121	1	trace_step['in']['type'] = 'loop'
122	1	do_trace = False
123		else:
124	1	trace_type = 'intermediary'
125		else:
126	1	trace_step['in']['type'] = 'last'
127	1	do_trace = False
128		else:
129		# No match
130		break
131	1	if 'out' in trace_step and trace_step['out']:
132	1	if self.check_loop_trace_step(trace_step, trace_result):
133	1	do_trace = False
134	1	trace_result.append(trace_step)
135	1	if len(trace_result) == 1 and \
136		trace_result[0]['in']['type'] == 'starting':
137		trace_result[0]['in']['type'] = 'last'
138	1	return trace_result
139
140	1	@staticmethod
141	1	def check_loop_trace_step(trace_step, trace_result):
142		"""Check if there is a loop in the trace and add the step."""
143		# outgoing interface is the same as the input interface
144	1	if not trace_result and \
145		trace_step['in']['type'] == 'last' and \
146		trace_step['in']['port'] == trace_step['out']['port']:
147	1	trace_step['in']['type'] = 'loop'
148	1	return True
149	1	if trace_result and \
150		trace_result[0]['in']['dpid'] == trace_step['in']['dpid'] and \
151		trace_result[0]['in']['port'] == trace_step['out']['port']:
152	1	trace_step['in']['type'] = 'loop'
153	1	return True
154	1	return False
155
156	1	@staticmethod
157	1	def has_loop(trace_step, trace_result):
158		"""Check if there is a loop in the trace result."""
159	1	for trace in trace_result:
160	1	if trace['in']['dpid'] == trace_step['dpid'] and \
161		trace['in']['port'] == trace_step['port']:
162	1	return True
163	1	return False
164
165	1	def trace_step(self, switch, entries, stored_flows):
166		"""Perform a trace step.
167
168		Match the given fields against the switch's list of flows."""
169	1	flow, entries, port = self.match_and_apply(
170		switch,
171		entries,
172		stored_flows
173		)
174
175	1	if not flow or not port:
176	1	return None
177
178	1	endpoint = find_endpoint(switch, port)
179	1	if endpoint is None:
180		log.warning(f"Port {port} not found on switch {switch}")
181		return None
182	1	endpoint = endpoint['endpoint']
183	1	if endpoint is None:
184	1	return {'out_port': port,
185		'entries': entries}
186
187	1	return {'dpid': endpoint.switch.dpid,
188		'in_port': endpoint.port_number,
189		'out_port': port,
190		'entries': entries}
191
192	1	@classmethod
193	1	def do_match(cls, flow, args):
194		"""Match a packet against this flow (OF1.3)."""
195		# pylint: disable=consider-using-dict-items
196		# pylint: disable=too-many-return-statements
197	1	if ('match' not in flow['flow']) or (len(flow['flow']['match']) == 0):
198	1	return False
199	1	for name in flow['flow']['match']:
200	1	field_flow = flow['flow']['match'][name]
201	1	field = args.get(name)
202	1	if name == 'dl_vlan':
203	1	if not match_field_dl_vlan(field, field_flow):
204	1	return False
205		continue
206		# In the case of dl_vlan field, the match must be checked
207		# even if this field is not in the packet args.
208	1	if not field:
209	1	return False
210	1	if name in ('nw_src', 'nw_dst', 'ipv6_src', 'ipv6_dst'):
211	1	if not match_field_ip(field, field_flow):
212	1	return False
213		continue
214	1	if field_flow != field:
215	1	return False
216	1	return flow
217
218	1	def match_flows(self, switch, args, stored_flows, many=True):
219		# pylint: disable=bad-staticmethod-argument
220		"""
221		Match the packet in request against the stored flows from flow_manager.
222		Try the match with each flow, in other. If many is True, tries the
223		match with all flows, if False, tries until the first match.
224		:param args: packet data
225		:param many: Boolean, indicating whether to continue after matching the
226		first flow or not
227		:return: If many, the list of matched flows, or the matched flow
228		"""
229	1	if switch.dpid not in stored_flows:
230	1	return None
231	1	response = []
232	1	if switch.dpid not in stored_flows:
233		return None
234	1	try:
235	1	for flow in stored_flows[switch.dpid]:
236	1	match = Main.do_match(flow, args)
237	1	if match:
238	1	if many:
239	1	response.append(match)
240		else:
241	1	response = match
242	1	break
243		except AttributeError:
244		return None
245	1	if not many and isinstance(response, list):
246	1	return None
247	1	return response
248
249		# pylint: disable=redefined-outer-name
250		# pylint: disable=too-many-branches
251	1	def match_and_apply(self, switch, args, stored_flows):
252		# pylint: disable=bad-staticmethod-argument
253		"""Match flows and apply actions.
254		Match given packet (in args) against
255		the stored flows (from flow_manager) and,
256		if a match flow is found, apply its actions."""
257	1	flow = self.match_flows(switch, args, stored_flows, False)
258	1	port = None
259	1	actions = []
260		# pylint: disable=too-many-nested-blocks
261	1	if not flow or switch.ofp_version != '0x04':
262	1	return flow, args, port
263	1	if 'actions' in flow['flow']:
264	1	actions = flow['flow']['actions']
265	1	elif '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	for action in actions:
270	1	action_type = action['action_type']
271	1	if action_type == 'output':
272	1	port = action['port']
273	1	if action_type == 'push_vlan':
274	1	if 'dl_vlan' not in args:
275		args['dl_vlan'] = []
276	1	args['dl_vlan'].append(0)
277	1	if action_type == 'pop_vlan':
278	1	if 'dl_vlan' in args:
279	1	args['dl_vlan'].pop()
280	1	if len(args['dl_vlan']) == 0:
281	1	del args['dl_vlan']
282	1	if action_type == 'set_vlan':
283	1	args['dl_vlan'][-1] = action['vlan_id']
284		return flow, args, port
285

kytos-ng / sdntrace_cp

Push — master ( 0bafa5...8e49b2 )

build.main.Main.match_and_apply() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like