build.main.Main.handle_stats_reply() - Code Metrics - Inspection of "Drop GenericFlow abstraction" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#38)

unknown

created 2022-09-22 21:02 UTC

build.main.Main.handle_stats_reply() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	30
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	20
CRAP Score	10.0751

Importance

Changes

Metric	Value
eloc	26
dl	0
loc	30
rs	5.9999
c	0
b	0
f	0
ccs	20
cts	22
cp	0.9091
cc	10
nop	3
crap	10.0751

How to fix Complexity

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

This NApp does operations with flows not covered by Kytos itself.
"""
# pylint: disable=too-many-return-statements,too-many-instance-attributes
# pylint: disable=too-many-arguments,too-many-branches,too-many-statements

import hashlib
import ipaddress
import json
from threading import Lock

from flask import jsonify, request
from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.amlight.flow_stats.utils import format_request
from napps.kytos.of_core.v0x04.flow import Action as Action13
from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory

# pylint: disable=too-many-public-methods
class Main(KytosNApp):
    """Main class of amlight/flow_stats NApp.

    This class is the entry point for this napp.
    """

    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.

        So, if you have any setup routine, insert it here.
        """
        log.info('Starting Kytos/Amlight flow manager')
        for switch in self.controller.switches.copy().values():
            switch.stat_flows = []

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

    def flow_from_id(self, flow_id):
        """Flow from given flow_id."""
        for switch in self.controller.switches.copy().values():
            try:
                for flow in switch.stat_flows: 
                    if flow.id == flow_id:
                        return flow
            except KeyError:
                pass
        return None

    @rest('flow/match/<dpid>')
    def flow_match(self, dpid):
        """Return first flow matching request."""
        switch = self.controller.get_switch_by_dpid(dpid)
        flow = self.match_flows(switch, format_request(request.args), False)
        if flow:
            return jsonify(flow.to_dict())
        return "No match", 404

    @rest('flow/stats/<dpid>')
    def flow_stats(self, dpid):
        """Return all flows matching request."""
        switch = self.controller.get_switch_by_dpid(dpid)
        if not switch:
            return f"switch {dpid} not found", 404
        flows = self.match_flows(switch, format_request(request.args), True)
        flows = [flow.to_dict() for flow in flows]
        return jsonify(flows)

    @staticmethod
    def match_flows(switch, args, many=True):
        # pylint: disable=bad-staticmethod-argument
        """
        Match the packet in request against the flows installed in the switch.

        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
        """
        response = []
        try:
            for flow in switch.stat_flows: 
                match = flow.do_match(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

    @staticmethod
    def match_and_apply(switch, args):
        # pylint: disable=bad-staticmethod-argument
        """Match flows and apply actions.

        Match given packet (in args) against the switch flows and,
        if a match flow is found, apply its actions."""
        flow = Main.match_flows(switch, args, False)
        port = None
        actions = None
        # pylint: disable=too-many-nested-blocks
        if flow:
            actions = flow.actions
            if switch.ofp_version == '0x04':
                for action in actions:
                    action_type = action.action_type
                    if action_type == 'output':
                        port = action.port
                    if action_type == 'push_vlan':
                        if 'vlan_vid' not in args:
                            args['vlan_vid'] = []
                        args['vlan_vid'].append(0)
                    if action_type == 'pop_vlan':
                        if 'vlan_vid' in args:
                            args['vlan_vid'].pop()
                            if len(args['vlan_vid']) == 0:
                                del args['vlan_vid']
                    if action_type == 'set_vlan':
                        args['vlan_vid'][-1] = action.vlan_id
        return flow, args, port

    @rest('packet_count/<flow_id>')
    def packet_count(self, flow_id):
        """Packet count of an specific flow."""
        flow = self.flow_from_id(flow_id)
        if flow is None:
            return "Flow does not exist", 404
        packet_stats = {
            'flow_id': flow_id,
            'packet_counter': flow.stats.packet_count,
            'packet_per_second': flow.stats.packet_count / flow.stats.duration_sec
            }
        return jsonify(packet_stats)

    @rest('bytes_count/<flow_id>')
    def bytes_count(self, flow_id):
        """Bytes count of an specific flow."""
        flow = self.flow_from_id(flow_id)
        if flow is None:
            return "Flow does not exist", 404
        bytes_stats = {
            'flow_id': flow_id,
            'bytes_counter': flow.stats.byte_count,
            'bits_per_second': flow.stats.byte_count * 8 / flow.stats.duration_sec
            }
        return jsonify(bytes_stats)

    @rest('packet_count/per_flow/<dpid>')
    def packet_count_per_flow(self, dpid):
        """Per flow packet count."""
        return self.flows_counters('packet_count',
                                   dpid,
                                   counter='packet_counter',
                                   rate='packet_per_second')

    @rest('packet_count/sum/<dpid>')
    def packet_count_sum(self, dpid):
        """Sum of packet count flow stats."""
        return self.flows_counters('packet_count',
                                   dpid,
                                   total=True)

    @rest('bytes_count/per_flow/<dpid>')
    def bytes_count_per_flow(self, dpid):
        """Per flow bytes count."""
        return self.flows_counters('byte_count',
                                   dpid,
                                   counter='bytes_counter',
                                   rate='bits_per_second')

    @rest('bytes_count/sum/<dpid>')
    def bytes_count_sum(self, dpid):
        """Sum of bytes count flow stats."""
        return self.flows_counters('byte_count',
                                   dpid,
                                   total=True)

    def flows_counters(self, field, dpid, counter=None, rate=None,
                       total=False):
        """Calculate flows statistics.

        The returned statistics are both per flow and for the sum of flows
        """
        # pylint: disable=too-many-arguments
        # pylint: disable=unused-variable
        start_date = request.args.get('start_date', 0)
        end_date = request.args.get('end_date', 0)
        # pylint: enable=unused-variable

        if total:
            count_flows = 0
        else:
            count_flows = []
            if not counter:
                counter = field
            if not rate:
                rate = field

        # We don't have statistics persistence yet, so for now this only works
        # for start and end equals to zero
        flows = self.controller.get_switch_by_dpid(dpid).stat_flows 

        for flow in flows:
            count = getattr(flow, field)
            if total:
                count_flows += count
            else:
                per_second = count / flow.stats.duration_sec
                if rate.startswith('bits'):
                    per_second *= 8
                count_flows.append({'flow_id': flow.id,
                                    counter: count,
                                    rate: per_second})

        return jsonify(count_flows)

    @listen_to('kytos/of_core.flow_stats.received')
    def on_stats_received(self, event):
        """Capture flow stats messages for OpenFlow 1.3."""
        self.handle_stats_received(event)

    def handle_stats_received(self, event):
        """Handle flow stats messages for OpenFlow 1.3."""
        switch = event.content['switch']
        if 'replies_flows' in event.content:
            replies_flows = event.content['replies_flows']
            self.handle_stats_reply_received(switch, replies_flows)

    def handle_stats_reply_received(self, switch, replies_flows):
        """Iterate on the replies and set the list of flows"""
        switch.stat_flows = replies_flows
        switch.stat_flows.sort(
                    key=lambda f: (f.priority, f.stats.duration_sec),
                    reverse=True
                    )


1		"""Main module of amlight/flow_stats Kytos Network Application.
2
3		This NApp does operations with flows not covered by Kytos itself.
4		"""
5		# pylint: disable=too-many-return-statements,too-many-instance-attributes
6		# pylint: disable=too-many-arguments,too-many-branches,too-many-statements
7
8	1	import hashlib
9	1	import ipaddress
10	1	import json
11	1	from threading import Lock
12
13	1	from flask import jsonify, request
14	1	from kytos.core import KytosEvent, KytosNApp, log, rest
15	1	from kytos.core.helpers import listen_to
16	1	from napps.amlight.flow_stats.utils import format_request
17	1	from napps.kytos.of_core.v0x04.flow import Action as Action13
18	1	from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory
19
20		# pylint: disable=too-many-public-methods
21	1	class Main(KytosNApp):
22		"""Main class of amlight/flow_stats NApp.
23
24		This class is the entry point for this napp.
25		"""
26	1
27		def setup(self):
28		"""Replace the '__init__' method for the KytosNApp subclass.
29
30	1	The setup method is automatically called by the controller when your
31	1	application is loaded.
32	1
33	1	So, if you have any setup routine, insert it here.
34	1	"""
35	1	log.info('Starting Kytos/Amlight flow manager')
36	1	for switch in self.controller.switches.copy().values():
37	1	switch.stat_flows = []
38	1
39	1	def execute(self):
40	1	"""This method is executed right after the setup method execution.
41	1
42		You can also use this method in loop mode if you add to the above setup
43	1	method a line like the following example:
44
45		self.execute_as_loop(30) # 30-second interval.
46	1	"""
47	1
48		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		def flow_from_id(self, flow_id):
55	1	"""Flow from given flow_id."""
56	1	for switch in self.controller.switches.copy().values():
57	1	try:
58	1	for flow in switch.stat_flows:
59	1	if flow.id == flow_id:
60	1	return flow
61	1	except KeyError:
62	1	pass
63	1	return None
64	1
65		@rest('flow/match/<dpid>')
66	1	def flow_match(self, dpid):
67		"""Return first flow matching request."""
68	1	switch = self.controller.get_switch_by_dpid(dpid)
69		flow = self.match_flows(switch, format_request(request.args), False)
70	1	if flow:
71	1	return jsonify(flow.to_dict())
72	1	return "No match", 404
73	1
74	1	@rest('flow/stats/<dpid>')
75	1	def flow_stats(self, dpid):
76	1	"""Return all flows matching request."""
77	1	switch = self.controller.get_switch_by_dpid(dpid)
78	1	if not switch:
79	1	return f"switch {dpid} not found", 404
80	1	flows = self.match_flows(switch, format_request(request.args), True)
81	1	flows = [flow.to_dict() for flow in flows]
82		return jsonify(flows)
83	1
84		@staticmethod
85	1	def match_flows(switch, args, many=True):
86		# pylint: disable=bad-staticmethod-argument
87		"""
88	1	Match the packet in request against the flows installed in the switch.
89	1
90	1	Try the match with each flow, in other. If many is True, tries the
91	1	match with all flows, if False, tries until the first match.
92		:param args: packet data
93	1	:param many: Boolean, indicating whether to continue after matching the
94		first flow or not
95		:return: If many, the list of matched flows, or the matched flow
96		"""
97		response = []
98		try:
99		for flow in switch.stat_flows:
100		match = flow.do_match(args)
101		if match:
102		if many:
103		response.append(match)
104		else:
105		response = match
106		break
107		except AttributeError:
108		return None
109		if not many and isinstance(response, list):
110		return None
111		return response
112
113		@staticmethod
114		def match_and_apply(switch, args):
115		# pylint: disable=bad-staticmethod-argument
116		"""Match flows and apply actions.
117	1
118	1	Match given packet (in args) against the switch flows and,
119		if a match flow is found, apply its actions."""
120	1	flow = Main.match_flows(switch, args, False)
121	1	port = None
122	1	actions = None
123	1	# pylint: disable=too-many-nested-blocks
124	1	if flow:
125	1	actions = flow.actions
126	1	if switch.ofp_version == '0x04':
127	1	for action in actions:
128	1	action_type = action.action_type
129	1	if action_type == 'output':
130		port = action.port
131		if action_type == 'push_vlan':
132		if 'vlan_vid' not in args:
133		args['vlan_vid'] = []
134		args['vlan_vid'].append(0)
135	1	if action_type == 'pop_vlan':
136	1	if 'vlan_vid' in args:
137		args['vlan_vid'].pop()
138		if len(args['vlan_vid']) == 0:
139		del args['vlan_vid']
140		if action_type == 'set_vlan':
141	1	args['vlan_vid'][-1] = action.vlan_id
142		return flow, args, port
143	1
144	1	@rest('packet_count/<flow_id>')
145		def packet_count(self, flow_id):
146		"""Packet count of an specific flow."""
147		flow = self.flow_from_id(flow_id)
148	1	if flow is None:
149	1	return "Flow does not exist", 404
150	1	packet_stats = {
151	1	'flow_id': flow_id,
152	1	'packet_counter': flow.stats.packet_count,
153	1	'packet_per_second': flow.stats.packet_count / flow.stats.duration_sec
154	1	}
155	1	return jsonify(packet_stats)
156	1
157		@rest('bytes_count/<flow_id>')
158	1	def bytes_count(self, flow_id):
159	1	"""Bytes count of an specific flow."""
160	1	flow = self.flow_from_id(flow_id)
161	1	if flow is None:
162	1	return "Flow does not exist", 404
163		bytes_stats = {
164	1	'flow_id': flow_id,
165	1	'bytes_counter': flow.stats.byte_count,
166	1	'bits_per_second': flow.stats.byte_count * 8 / flow.stats.duration_sec
167	1	}
168	1	return jsonify(bytes_stats)
169	1
170	1	@rest('packet_count/per_flow/<dpid>')
171		def packet_count_per_flow(self, dpid):
172	1	"""Per flow packet count."""
173		return self.flows_counters('packet_count',
174		dpid,
175		counter='packet_counter',
176		rate='packet_per_second')
177
178	1	@rest('packet_count/sum/<dpid>')
179		def packet_count_sum(self, dpid):
180		"""Sum of packet count flow stats."""
181		return self.flows_counters('packet_count',
182		dpid,
183		total=True)
184
185		@rest('bytes_count/per_flow/<dpid>')
186		def bytes_count_per_flow(self, dpid):
187		"""Per flow bytes count."""
188		return self.flows_counters('byte_count',
189		dpid,
190		counter='bytes_counter',
191		rate='bits_per_second')
192
193		@rest('bytes_count/sum/<dpid>')
194		def bytes_count_sum(self, dpid):
195		"""Sum of bytes count flow stats."""
196		return self.flows_counters('byte_count',
197		dpid,
198		total=True)
199
200	1	def flows_counters(self, field, dpid, counter=None, rate=None,
201		total=False):
202		"""Calculate flows statistics.
203
204		The returned statistics are both per flow and for the sum of flows
205		"""
206	1	# pylint: disable=too-many-arguments
207		# pylint: disable=unused-variable
208		start_date = request.args.get('start_date', 0)
209		end_date = request.args.get('end_date', 0)
210		# pylint: enable=unused-variable
211
212		if total:
213		count_flows = 0
214	1	else:
215	1	count_flows = []
216		if not counter:
217	1	counter = field
218	1	if not rate:
219		rate = field
220	1
221		# We don't have statistics persistence yet, so for now this only works
222		# for start and end equals to zero
223		flows = self.controller.get_switch_by_dpid(dpid).stat_flows
224
225		for flow in flows:
226		count = getattr(flow, field)
227		if total:
228		count_flows += count
229	1	else:
230		per_second = count / flow.stats.duration_sec
231		if rate.startswith('bits'):
232		per_second *= 8
233		count_flows.append({'flow_id': flow.id,
234		counter: count,
235	1	rate: per_second})
236
237	1	return jsonify(count_flows)
238	1
239	1	@listen_to('kytos/of_core.flow_stats.received')
240	1	def on_stats_received(self, event):
241	1	"""Capture flow stats messages for OpenFlow 1.3."""
242		self.handle_stats_received(event)
243
244	1	def handle_stats_received(self, event):
245		"""Handle flow stats messages for OpenFlow 1.3."""
246	1	switch = event.content['switch']
247	1	if 'replies_flows' in event.content:
248		replies_flows = event.content['replies_flows']
249	1	self.handle_stats_reply_received(switch, replies_flows)
250	1
251	1	def handle_stats_reply_received(self, switch, replies_flows):
252	1	"""Iterate on the replies and set the list of flows"""
253	1	switch.stat_flows = replies_flows
254		switch.stat_flows.sort(
255	1	key=lambda f: (f.priority, f.stats.duration_sec),
256	1	reverse=True
257		)
258

kytos-ng / flow_stats

Pull Request — master (#38)

build.main.Main.handle_stats_reply() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like