build.main.Main.on_table_stats_received() - Code Metrics - Inspection of "feature: Add table stats" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#56)

unknown

created 2023-05-13 06:00 UTC

build.main.Main.on_table_stats_received() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1.037

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	4
ccs	2
cts	3
cp	0.6667
rs	10
c	0
b	0
f	0
cc	1
nop	2
crap	1.037

"""Main module of amlight/kytos_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

from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import listen_to
from kytos.core.rest_api import HTTPException, JSONResponse, Request


# pylint: disable=too-many-public-methods
class Main(KytosNApp):
    """Main class of amlight/kytos_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')
        self.flows_stats_dict = {}
        self.tables_stats_dict = {}

    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."""
        return self.flows_stats_dict.get(flow_id)

    def flow_stats_by_dpid_flow_id(self, dpids):
        """ Auxiliar funcion for v1/flow/stats endpoint implementation.
        """
        flow_stats_by_id = {}
        flows_stats_dict_copy = self.flows_stats_dict.copy()
        for flow_id, flow in flows_stats_dict_copy.items():
            dpid = flow.switch.dpid
            if dpid in dpids:
                if dpid not in flow_stats_by_id:
                    flow_stats_by_id[dpid] = {}
                info_flow_as_dict = flow.stats.as_dict()
                info_flow_as_dict.update({"cookie": flow.cookie})
                info_flow_as_dict.update({"priority": flow.priority})
                info_flow_as_dict.update({"match": flow.match.as_dict()})
                flow_stats_by_id[dpid].update({flow_id: info_flow_as_dict})
        return flow_stats_by_id

    def table_stats_by_dpid_table_id(self, dpids, table_ids):
        """ Auxiliar funcion for v1/table/stats endpoint implementation.
        """
        table_stats_by_id = {}
        tables_stats_dict_copy = self.tables_stats_dict.copy()
        for dpid_dict, tables in tables_stats_dict_copy.items():
            if dpid_dict not in dpids:
                continue
            table_stats_by_id[dpid_dict] = {}
            if len(table_ids) == 0:
                table_ids = list(tables.keys())
            for _id, table in tables.items():
                if _id in table_ids:
                    table_dict = table.as_dict()
                    del table_dict['switch']
                    table_stats_by_id[dpid_dict].update({_id: table_dict})
        return table_stats_by_id

    @rest('v1/flow/stats')
    def flow_stats(self, request: Request) -> JSONResponse:
        """Return the flows stats by dpid.
        Return the stats of all flows if dpid is None
        """
        dpids = request.query_params.getlist("dpid")
        if len(dpids) == 0:
            dpids = [sw.dpid for sw in self.controller.switches.values()]
        flow_stats_by_id = self.flow_stats_by_dpid_flow_id(dpids)
        return JSONResponse(flow_stats_by_id)

    @rest('v1/table/stats')
    def table_stats(self, request: Request) -> JSONResponse:
        """Return the table stats by dpid,
        and optionally by table_id.
        """
        dpids = request.query_params.getlist("dpid")
        if len(dpids) == 0:
            dpids = [sw.dpid for sw in self.controller.switches.values()]
        table_ids = request.query_params.getlist("table")
        table_ids = list(map(int, table_ids))
        table_stats_dpid = self.table_stats_by_dpid_table_id(dpids, table_ids)
        return JSONResponse(table_stats_dpid)

    @rest('v1/packet_count/{flow_id}')
    def packet_count(self, request: Request) -> JSONResponse:
        """Packet count of an specific flow."""
        flow_id = request.path_params["flow_id"]
        flow = self.flow_from_id(flow_id)
        if flow is None:
            raise HTTPException(404, detail="Flow does not exist")
        packet_stats = {
            'flow_id': flow_id,
            'packet_counter': flow.stats.packet_count,
            'packet_per_second':
                flow.stats.packet_count / flow.stats.duration_sec
            }
        return JSONResponse(packet_stats)

    @rest('v1/bytes_count/{flow_id}')
    def bytes_count(self, request: Request) -> JSONResponse:
        """Bytes count of an specific flow."""
        flow_id = request.path_params["flow_id"]
        flow = self.flow_from_id(flow_id)
        if flow is None:
            raise HTTPException(404, detail="Flow does not exist")
        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 JSONResponse(bytes_stats)

    @rest('v1/packet_count/per_flow/{dpid}')
    def packet_count_per_flow(self, request: Request) -> JSONResponse:
        """Per flow packet count."""
        dpid = request.path_params["dpid"]
        return self.flows_counters('packet_count',
                                   dpid,
                                   counter='packet_counter',
                                   rate='packet_per_second')

    @rest('v1/bytes_count/per_flow/{dpid}')
    def bytes_count_per_flow(self, request: Request) -> JSONResponse:
        """Per flow bytes count."""
        dpid = request.path_params["dpid"]
        return self.flows_counters('byte_count',
                                   dpid,
                                   counter='bytes_counter',
                                   rate='bits_per_second')

    def flows_counters(self, field, dpid, counter=None, rate=None,
                       total=False) -> JSONResponse:
        """Calculate flows statistics.
        The returned statistics are both per flow and for the sum of flows
        """

        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.flow_stats_by_dpid_flow_id([dpid])
        flows = flows.get(dpid)

        if flows is None:
            return JSONResponse(count_flows)
        for flow_id, stats in flows.items():
            count = stats[field]
            if total:
                count_flows += count
            else:
                per_second = count / stats['duration_sec']
                if rate.startswith('bits'):
                    per_second *= 8
                count_flows.append({'flow_id': flow_id,
                                    counter: count,
                                    rate: per_second})
        return JSONResponse(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."""
        if 'replies_flows' in event.content:
            replies_flows = event.content['replies_flows']
            self.handle_stats_reply_received(replies_flows)

    def handle_stats_reply_received(self, replies_flows):
        """Update the set of flows stats"""
        self.flows_stats_dict.update({flow.id: flow for flow in replies_flows})

    @listen_to('kytos/of_core.table_stats.received')
    def on_table_stats_received(self, event):
        """Capture table stats messages for OpenFlow 1.3."""
        self.handle_table_stats_received(event)

    def handle_table_stats_received(self, event):
        """Handle table stats messages for OpenFlow 1.3."""
        if 'replies_tables' in event.content:
            replies_tables = event.content['replies_tables']
            self.handle_table_stats_reply_received(replies_tables)

    def handle_table_stats_reply_received(self, replies_tables):
        """Update the set of tables stats"""
        for table in replies_tables:
            switch_id = table.switch.id
            if switch_id not in self.tables_stats_dict:
                self.tables_stats_dict[switch_id] = {}
            self.tables_stats_dict[switch_id][table.table_id] = table


1		"""Main module of amlight/kytos_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	from kytos.core import KytosNApp, log, rest
9	1	from kytos.core.helpers import listen_to
10	1	from kytos.core.rest_api import HTTPException, JSONResponse, Request
11
12
13		# pylint: disable=too-many-public-methods
14	1	class Main(KytosNApp):
15		"""Main class of amlight/kytos_stats NApp.
16		This class is the entry point for this napp.
17		"""
18
19	1	def setup(self):
20		"""Replace the '__init__' method for the KytosNApp subclass.
21		The setup method is automatically called by the controller when your
22		application is loaded.
23		So, if you have any setup routine, insert it here.
24		"""
25	1	log.info('Starting Kytos/Amlight flow manager')
26	1	self.flows_stats_dict = {}
27	1	self.tables_stats_dict = {}
28
29	1	def execute(self):
30		"""This method is executed right after the setup method execution.
31		You can also use this method in loop mode if you add to the above setup
32		method a line like the following example:
33		self.execute_as_loop(30) # 30-second interval.
34		"""
35
36	1	def shutdown(self):
37		"""This method is executed when your napp is unloaded.
38		If you have some cleanup procedure, insert it here.
39		"""
40
41	1	def flow_from_id(self, flow_id):
42		"""Flow from given flow_id."""
43	1	return self.flows_stats_dict.get(flow_id)
44
45	1	def flow_stats_by_dpid_flow_id(self, dpids):
46		""" Auxiliar funcion for v1/flow/stats endpoint implementation.
47		"""
48	1	flow_stats_by_id = {}
49	1	flows_stats_dict_copy = self.flows_stats_dict.copy()
50	1	for flow_id, flow in flows_stats_dict_copy.items():
51		dpid = flow.switch.dpid
52		if dpid in dpids:
53		if dpid not in flow_stats_by_id:
54		flow_stats_by_id[dpid] = {}
55		info_flow_as_dict = flow.stats.as_dict()
56		info_flow_as_dict.update({"cookie": flow.cookie})
57		info_flow_as_dict.update({"priority": flow.priority})
58		info_flow_as_dict.update({"match": flow.match.as_dict()})
59		flow_stats_by_id[dpid].update({flow_id: info_flow_as_dict})
60	1	return flow_stats_by_id
61
62	1	def table_stats_by_dpid_table_id(self, dpids, table_ids):
63		""" Auxiliar funcion for v1/table/stats endpoint implementation.
64		"""
65		table_stats_by_id = {}
66		tables_stats_dict_copy = self.tables_stats_dict.copy()
67		for dpid_dict, tables in tables_stats_dict_copy.items():
68		if dpid_dict not in dpids:
69		continue
70		table_stats_by_id[dpid_dict] = {}
71		if len(table_ids) == 0:
72		table_ids = list(tables.keys())
73		for _id, table in tables.items():
74		if _id in table_ids:
75		table_dict = table.as_dict()
76		del table_dict['switch']
77		table_stats_by_id[dpid_dict].update({_id: table_dict})
78		return table_stats_by_id
79
80	1	@rest('v1/flow/stats')
81	1	def flow_stats(self, request: Request) -> JSONResponse:
82		"""Return the flows stats by dpid.
83		Return the stats of all flows if dpid is None
84		"""
85	1	dpids = request.query_params.getlist("dpid")
86	1	if len(dpids) == 0:
87	1	dpids = [sw.dpid for sw in self.controller.switches.values()]
88	1	flow_stats_by_id = self.flow_stats_by_dpid_flow_id(dpids)
89	1	return JSONResponse(flow_stats_by_id)
90
91	1	@rest('v1/table/stats')
92	1	def table_stats(self, request: Request) -> JSONResponse:
93		"""Return the table stats by dpid,
94		and optionally by table_id.
95		"""
96	1	dpids = request.query_params.getlist("dpid")
97	1	if len(dpids) == 0:
98	1	dpids = [sw.dpid for sw in self.controller.switches.values()]
99	1	table_ids = request.query_params.getlist("table")
100	1	table_ids = list(map(int, table_ids))
101	1	table_stats_dpid = self.table_stats_by_dpid_table_id(dpids, table_ids)
102	1	return JSONResponse(table_stats_dpid)
103
104	1	@rest('v1/packet_count/{flow_id}')
105	1	def packet_count(self, request: Request) -> JSONResponse:
106		"""Packet count of an specific flow."""
107	1	flow_id = request.path_params["flow_id"]
108	1	flow = self.flow_from_id(flow_id)
109	1	if flow is None:
110	1	raise HTTPException(404, detail="Flow does not exist")
111	1	packet_stats = {
112		'flow_id': flow_id,
113		'packet_counter': flow.stats.packet_count,
114		'packet_per_second':
115		flow.stats.packet_count / flow.stats.duration_sec
116		}
117	1	return JSONResponse(packet_stats)
118
119	1	@rest('v1/bytes_count/{flow_id}')
120	1	def bytes_count(self, request: Request) -> JSONResponse:
121		"""Bytes count of an specific flow."""
122	1	flow_id = request.path_params["flow_id"]
123	1	flow = self.flow_from_id(flow_id)
124	1	if flow is None:
125	1	raise HTTPException(404, detail="Flow does not exist")
126	1	bytes_stats = {
127		'flow_id': flow_id,
128		'bytes_counter': flow.stats.byte_count,
129		'bits_per_second':
130		flow.stats.byte_count * 8 / flow.stats.duration_sec
131		}
132	1	return JSONResponse(bytes_stats)
133
134	1	@rest('v1/packet_count/per_flow/{dpid}')
135	1	def packet_count_per_flow(self, request: Request) -> JSONResponse:
136		"""Per flow packet count."""
137	1	dpid = request.path_params["dpid"]
138	1	return self.flows_counters('packet_count',
139		dpid,
140		counter='packet_counter',
141		rate='packet_per_second')
142
143	1	@rest('v1/bytes_count/per_flow/{dpid}')
144	1	def bytes_count_per_flow(self, request: Request) -> JSONResponse:
145		"""Per flow bytes count."""
146	1	dpid = request.path_params["dpid"]
147	1	return self.flows_counters('byte_count',
148		dpid,
149		counter='bytes_counter',
150		rate='bits_per_second')
151
152	1	def flows_counters(self, field, dpid, counter=None, rate=None,
153		total=False) -> JSONResponse:
154		"""Calculate flows statistics.
155		The returned statistics are both per flow and for the sum of flows
156		"""
157
158	1	if total:
159		count_flows = 0
160		else:
161	1	count_flows = []
162	1	if not counter:
163		counter = field
164	1	if not rate:
165		rate = field
166
167		# We don't have statistics persistence yet, so for now this only works
168		# for start and end equals to zero
169	1	flows = self.flow_stats_by_dpid_flow_id([dpid])
170	1	flows = flows.get(dpid)
171
172	1	if flows is None:
173	1	return JSONResponse(count_flows)
174	1	for flow_id, stats in flows.items():
175	1	count = stats[field]
176	1	if total:
177		count_flows += count
178		else:
179	1	per_second = count / stats['duration_sec']
180	1	if rate.startswith('bits'):
181	1	per_second *= 8
182	1	count_flows.append({'flow_id': flow_id,
183		counter: count,
184		rate: per_second})
185	1	return JSONResponse(count_flows)
186
187	1	@listen_to('kytos/of_core.flow_stats.received')
188	1	def on_stats_received(self, event):
189		"""Capture flow stats messages for OpenFlow 1.3."""
190		self.handle_stats_received(event)
191
192	1	def handle_stats_received(self, event):
193		"""Handle flow stats messages for OpenFlow 1.3."""
194	1	if 'replies_flows' in event.content:
195	1	replies_flows = event.content['replies_flows']
196	1	self.handle_stats_reply_received(replies_flows)
197
198	1	def handle_stats_reply_received(self, replies_flows):
199		"""Update the set of flows stats"""
200	1	self.flows_stats_dict.update({flow.id: flow for flow in replies_flows})
201
202	1	@listen_to('kytos/of_core.table_stats.received')
203	1	def on_table_stats_received(self, event):
204		"""Capture table stats messages for OpenFlow 1.3."""
205		self.handle_table_stats_received(event)
206
207	1	def handle_table_stats_received(self, event):
208		"""Handle table stats messages for OpenFlow 1.3."""
209		if 'replies_tables' in event.content:
210		replies_tables = event.content['replies_tables']
211		self.handle_table_stats_reply_received(replies_tables)
212
213	1	def handle_table_stats_reply_received(self, replies_tables):
214		"""Update the set of tables stats"""
215	1	for table in replies_tables:
216	1	switch_id = table.switch.id
217	1	if switch_id not in self.tables_stats_dict:
218	1	self.tables_stats_dict[switch_id] = {}
219		self.tables_stats_dict[switch_id][table.table_id] = table
220

kytos-ng / flow_stats

Pull Request — master (#56)

build.main.Main.on_table_stats_received() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like