build.main.Main.on_stats_received() - Code Metrics - Inspection of "Fix error in from_replies_flows" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#34)

unknown

created 2022-09-15 03:51 UTC

build.main.Main.on_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
rs	10
c	0
b	0
f	0
ccs	2
cts	3
cp	0.6667
cc	1
nop	2
crap	1.037

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

import pyof.v0x01.controller2switch.common as common01
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.amlight.sdntrace import constants
from napps.kytos.of_core.v0x01.flow import Action as Action10
from napps.kytos.of_core.v0x04.flow import Action as Action13
from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory
from pyof.v0x01.common.flow_match import FlowWildCards


class GenericFlow():
    """Class to represent a flow.

        This class represents a flow regardless of the OF version."""

    def __init__(self, version='0x01', match=None, idle_timeout=0,
                 hard_timeout=0, duration_sec=0, packet_count=0, byte_count=0,
                 priority=0, table_id=0xff, cookie=None, buffer_id=None,
                 actions=None):
        self.version = version
        self.match = match if match else {}
        self.idle_timeout = idle_timeout
        self.hard_timeout = hard_timeout
        self.duration_sec = duration_sec
        self.packet_count = packet_count
        self.byte_count = byte_count
        self.priority = priority
        self.table_id = table_id
        self.cookie = cookie
        self.buffer_id = buffer_id
        self.actions = actions if actions else []

    def __eq__(self, other):
        return self.id == other.id

    @property
    def id(self):
        # pylint: disable=invalid-name
        """Return the hash of the object.
        Calculates the hash of the object by using the hashlib we use md5 of
        strings.
        Returns:
            string: Hash of object.
        """
        hash_result = hashlib.md5()
        hash_result.update(str(self.version).encode('utf-8'))
        for value in self.match.copy().values():
            if self.version == '0x01':
                hash_result.update(str(value).encode('utf-8'))
            else:
                hash_result.update(str(value.value).encode('utf-8'))
        hash_result.update(str(self.idle_timeout).encode('utf-8'))
        hash_result.update(str(self.hard_timeout).encode('utf-8'))
        hash_result.update(str(self.priority).encode('utf-8'))
        hash_result.update(str(self.table_id).encode('utf-8'))
        hash_result.update(str(self.cookie).encode('utf-8'))
        hash_result.update(str(self.buffer_id).encode('utf-8'))

        return hash_result.hexdigest()

    def to_dict(self):
        """Convert flow to a dictionary."""
        flow_dict = {}
        flow_dict['version'] = self.version
        if self.version == '0x01':
            flow_dict.update(self.match)
        else:
            flow_dict.update(self.match_to_dict())
        flow_dict['idle_timeout'] = self.idle_timeout
        flow_dict['hard_timeout'] = self.hard_timeout
        flow_dict['priority'] = self.priority
        flow_dict['table_id'] = self.table_id
        flow_dict['cookie'] = self.cookie
        flow_dict['buffer_id'] = self.buffer_id
        flow_dict['actions'] = []
        for action in self.actions:
            flow_dict['actions'].append(action.as_dict())

        return flow_dict

    def match_to_dict(self):
        """Convert a match in OF 1.3 to a dictionary."""
        # pylint: disable=consider-using-dict-items
        match = {}
        for name in self.match.copy():
            match[name] = self.match[name].value
        return match

    def to_json(self):
        """Return a json version of the flow."""
        return json.dumps(self.to_dict())

    # @staticmethod
    # def from_dict(flow_dict):
    #     """Create a flow from a dict."""
    #     flow = GenericFlow()
    #     for attr_name, value in flow_dict.items():
    #         if attr_name == 'actions':
    #             flow.actions = []
    #             for action in value:
    #                 new_action = ACTION_TYPES[int(action['action_type'])]()
    #                 for action_attr_name,
    #                     action_attr_value in action.items():
    #
    #                     setattr(new_action,
    #                             action_attr_name,
    #                             action_attr_value)
    #                 flow.actions.append(new_action)
    #         else:
    #             setattr(flow, attr_name, value)
    #     return flow

    @classmethod
    def from_flow_stats(cls, flow_stats, version='0x01'):
        """Create a flow from OF flow stats."""
        flow = GenericFlow(version=version)
        flow.idle_timeout = flow_stats.idle_timeout.value
        flow.hard_timeout = flow_stats.hard_timeout.value
        flow.priority = flow_stats.priority.value
        flow.table_id = flow_stats.table_id.value
        flow.duration_sec = flow_stats.duration_sec.value
        flow.packet_count = flow_stats.packet_count.value
        flow.byte_count = flow_stats.byte_count.value
        if version == '0x01':
            flow.match['wildcards'] = flow_stats.match.wildcards.value
            flow.match['in_port'] = flow_stats.match.in_port.value
            flow.match['eth_src'] = flow_stats.match.dl_src.value
            flow.match['eth_dst'] = flow_stats.match.dl_dst.value
            flow.match['vlan_vid'] = flow_stats.match.dl_vlan.value
            flow.match['vlan_pcp'] = flow_stats.match.dl_vlan_pcp.value
            flow.match['eth_type'] = flow_stats.match.dl_type.value
            flow.match['ip_tos'] = flow_stats.match.nw_tos.value
            flow.match['ipv4_src'] = flow_stats.match.nw_src.value
            flow.match['ipv4_dst'] = flow_stats.match.nw_dst.value
            flow.match['ip_proto'] = flow_stats.match.nw_proto.value
            flow.match['tcp_src'] = flow_stats.match.tp_src.value
            flow.match['tcp_dst'] = flow_stats.match.tp_dst.value
            flow.actions = []
            for of_action in flow_stats.actions:
                action = Action10.from_of_action(of_action)
                flow.actions.append(action)
        elif version == '0x04':
            for match in flow_stats.match.oxm_match_fields:
                match_field = MatchFieldFactory.from_of_tlv(match)
                field_name = match_field.name
                if field_name == 'dl_vlan':
                    field_name = 'vlan_vid'
                flow.match[field_name] = match_field
            flow.actions = []
            for instruction in flow_stats.instructions:
                if instruction.instruction_type == 'apply_actions':
                    for of_action in instruction.actions:
                        action = Action13.from_of_action(of_action)
                        flow.actions.append(action)
        return flow

    @classmethod
    def from_replies_flows(cls, flow04):
        """Create a flow from a flow passed on
        replies_flows in event kytos/of_core.flow_stats.received."""

        flow = GenericFlow(version='0x04')
        flow.idle_timeout = flow04.idle_timeout
        flow.hard_timeout = flow04.hard_timeout
        flow.priority = flow04.priority
        flow.table_id = flow04.table_id
        flow.cookie = flow04.cookie
        flow.duration_sec = flow04.stats.duration_sec
        flow.packet_count = flow04.stats.packet_count
        flow.byte_count = flow04.stats.byte_count

        as_of_match = flow04.match.as_of_match()
        for match in as_of_match.oxm_match_fields:
            match_field = MatchFieldFactory.from_of_tlv(match)
            field_name = match_field.name
            if field_name == 'dl_vlan':
                field_name = 'vlan_vid'
            flow.match[field_name] = match_field
        flow.actions = []
        for instruction in flow04.instructions:
            if instruction.instruction_type == 'apply_actions':
                for of_action in instruction.actions:
                    flow.actions.append(of_action)
        return flow

    def do_match(self, args):
        """Match a packet against this flow."""
        if self.version == '0x01':
            return self.match10(args)
        if self.version == '0x04':
            return self.match13(args)
        return None

    def match10(self, args):
        """Match a packet against this flow (OF1.0)."""
        log.debug('Matching packet')
        if not self.match['wildcards'] & FlowWildCards.OFPFW_IN_PORT:
            if 'in_port' not in args:
                return False
            if self.match['in_port'] != int(args['in_port']):
                return False
        if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_VLAN_PCP:
            if 'vlan_pcp' not in args:
                return False
            if self.match['vlan_pcp'] != int(args['vlan_pcp']):
                return False
        if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_VLAN:
            if 'vlan_vid' not in args:
                return False
            if self.match['vlan_vid'] != args['vlan_vid'][-1]:
                return False
        if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_SRC:
            if 'eth_src' not in args:
                return False
            if self.match['eth_src'] != args['eth_src']:
                return False
        if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_DST:
            if 'eth_dst' not in args:
                return False
            if self.match['eth_dst'] != args['eth_dst']:
                return False
        if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_TYPE:
            if 'eth_type' not in args:
                return False
            if self.match['eth_type'] != int(args['eth_type']):
                return False
        if self.match['eth_type'] == constants.IPV4:
            flow_ip_int = int(ipaddress.IPv4Address(self.match['ipv4_src']))
            if flow_ip_int != 0:
                mask = ((self.match['wildcards'] &
                         FlowWildCards.OFPFW_NW_SRC_MASK) >>
                        FlowWildCards.OFPFW_NW_SRC_SHIFT)
                mask = min(mask, 32)
                if mask != 32 and 'ipv4_src' not in args:
                    return False
                mask = (0xffffffff << mask) & 0xffffffff
                ip_int = int(ipaddress.IPv4Address(args['ipv4_src']))
                if ip_int & mask != flow_ip_int & mask:
                    return False

            flow_ip_int = int(ipaddress.IPv4Address(self.match['ipv4_dst']))
            if flow_ip_int != 0:
                mask = ((self.match['wildcards'] &
                         FlowWildCards.OFPFW_NW_DST_MASK) >>
                        FlowWildCards.OFPFW_NW_DST_SHIFT)
                mask = min(mask, 32)
                if mask != 32 and 'ipv4_dst' not in args:
                    return False
                mask = (0xffffffff << mask) & 0xffffffff
                ip_int = int(ipaddress.IPv4Address(args['ipv4_dst']))
                if ip_int & mask != flow_ip_int & mask:
                    return False
            if not self.match['wildcards'] & FlowWildCards.OFPFW_NW_TOS:
                if 'ip_tos' not in args:
                    return False
                if self.match['ip_tos'] != int(args['ip_tos']):
                    return False
            if not self.match['wildcards'] & FlowWildCards.OFPFW_NW_PROTO:
                if 'ip_proto' not in args:
                    return False
                if self.match['ip_proto'] != int(args['ip_proto']):
                    return False
            if not self.match['wildcards'] & FlowWildCards.OFPFW_TP_SRC:
                if 'tp_src' not in args:
                    return False
                if self.match['tcp_src'] != int(args['tp_src']):
                    return False
            if not self.match['wildcards'] & FlowWildCards.OFPFW_TP_DST:
                if 'tp_dst' not in args:
                    return False
                if self.match['tcp_dst'] != int(args['tp_dst']):
                    return False
        return self

    def match13(self, args):
        """Match a packet against this flow (OF1.3)."""
        # pylint: disable=consider-using-dict-items
        for name in self.match.copy():
            if name not in args:
                return False
            if name == 'vlan_vid':
                field = args[name][-1]
            else:
                field = args[name]
            if name not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
                if self.match[name].value != field:
                    return False
            else:
                packet_ip = int(ipaddress.ip_address(field))
                ip_addr = self.match[name].value
                if packet_ip & ip_addr.netmask != ip_addr.address:
                    return False
        return self


# 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.generic_flows = []
        self.switch_stats_xid = {}
        self.switch_stats_lock = {}

    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.generic_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.generic_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 == '0x01':
                for action in actions:
                    action_type = action.action_type
                    if action_type == 'output':
                        port = action.port
                    elif action_type == 'set_vlan':
                        if 'vlan_vid' in args:
                            args['vlan_vid'][-1] = action.vlan_id
                        else:
                            args['vlan_vid'] = [action.vlan_id]
            elif 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.packet_count,
            'packet_per_second': flow.packet_count / flow.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.byte_count,
            'bits_per_second': flow.byte_count * 8 / flow.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).generic_flows

        for flow in flows:
            count = getattr(flow, field)
            if total:
                count_flows += count
            else:
                per_second = count / flow.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.v0x01.messages.in.ofpt_stats_reply')
    def on_stats_reply_0x01(self, event):
        """Capture flow stats messages for v0x01 switches."""
        self.handle_stats_reply_0x01(event)

    def handle_stats_reply_0x01(self, event):
        """Handle stats replies for v0x01 switches."""
        msg = event.content['message']
        if msg.body_type == common01.StatsType.OFPST_FLOW:
            switch = event.source.switch
            self.handle_stats_reply(msg, switch)

    def handle_stats_reply(self, msg, switch):
        """Insert flows received in the switch list of flows."""
        try:
            old_flows = switch.generic_flows
        except AttributeError:
            old_flows = []
        is_new_xid = (
            int(msg.header.xid) != self.switch_stats_xid.get(switch.id, 0)
        )
        is_last_part = msg.flags.value % 2 == 0
        self.switch_stats_lock.setdefault(switch.id, Lock())
        with self.switch_stats_lock[switch.id]:
            if is_new_xid:
                switch.generic_new_flows = []
                self.switch_stats_xid[switch.id] = int(msg.header.xid)
            for flow_stats in msg.body:
                flow = GenericFlow.from_flow_stats(flow_stats,
                                                   switch.ofp_version)
                switch.generic_new_flows.append(flow)
            if is_last_part:
                switch.generic_flows = switch.generic_new_flows
                switch.generic_flows.sort(
                    key=lambda f: (f.priority, f.duration_sec),
                    reverse=True
                )
        if is_new_xid and is_last_part and switch.generic_flows != old_flows:
            # Generate an event informing that flows have changed
            event = KytosEvent('amlight/flow_stats.flows_updated')
            event.content['switch'] = switch.dpid
            self.controller.buffers.app.put(event)

    @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)

    # pylint: disable=no-self-use
    def handle_stats_reply_received(self, switch, replies_flows):
        """Iterate on the replies and set the generic flows"""
        switch.generic_flows = [GenericFlow.from_replies_flows(flow)
                                for flow in replies_flows]
        switch.generic_flows.sort(
                    key=lambda f: (f.priority, f.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	import pyof.v0x01.controller2switch.common as common01
14	1	from flask import jsonify, request
15	1	from kytos.core import KytosEvent, KytosNApp, log, rest
16	1	from kytos.core.helpers import listen_to
17	1	from napps.amlight.flow_stats.utils import format_request
18	1	from napps.amlight.sdntrace import constants
19	1	from napps.kytos.of_core.v0x01.flow import Action as Action10
20	1	from napps.kytos.of_core.v0x04.flow import Action as Action13
21	1	from napps.kytos.of_core.v0x04.match_fields import MatchFieldFactory
22	1	from pyof.v0x01.common.flow_match import FlowWildCards
23
24
25	1	class GenericFlow():
26		"""Class to represent a flow.
27
28		This class represents a flow regardless of the OF version."""
29
30	1	def __init__(self, version='0x01', match=None, idle_timeout=0,
31		hard_timeout=0, duration_sec=0, packet_count=0, byte_count=0,
32		priority=0, table_id=0xff, cookie=None, buffer_id=None,
33		actions=None):
34	1	self.version = version
35	1	self.match = match if match else {}
36	1	self.idle_timeout = idle_timeout
37	1	self.hard_timeout = hard_timeout
38	1	self.duration_sec = duration_sec
39	1	self.packet_count = packet_count
40	1	self.byte_count = byte_count
41	1	self.priority = priority
42	1	self.table_id = table_id
43	1	self.cookie = cookie
44	1	self.buffer_id = buffer_id
45	1	self.actions = actions if actions else []
46
47	1	def __eq__(self, other):
48		return self.id == other.id
49
50	1	@property
51	1	def id(self):
52		# pylint: disable=invalid-name
53		"""Return the hash of the object.
54		Calculates the hash of the object by using the hashlib we use md5 of
55		strings.
56		Returns:
57		string: Hash of object.
58		"""
59	1	hash_result = hashlib.md5()
60	1	hash_result.update(str(self.version).encode('utf-8'))
61	1	for value in self.match.copy().values():
62	1	if self.version == '0x01':
63	1	hash_result.update(str(value).encode('utf-8'))
64		else:
65	1	hash_result.update(str(value.value).encode('utf-8'))
66	1	hash_result.update(str(self.idle_timeout).encode('utf-8'))
67	1	hash_result.update(str(self.hard_timeout).encode('utf-8'))
68	1	hash_result.update(str(self.priority).encode('utf-8'))
69	1	hash_result.update(str(self.table_id).encode('utf-8'))
70	1	hash_result.update(str(self.cookie).encode('utf-8'))
71	1	hash_result.update(str(self.buffer_id).encode('utf-8'))
72
73	1	return hash_result.hexdigest()
74
75	1	def to_dict(self):
76		"""Convert flow to a dictionary."""
77	1	flow_dict = {}
78	1	flow_dict['version'] = self.version
79	1	if self.version == '0x01':
80	1	flow_dict.update(self.match)
81		else:
82	1	flow_dict.update(self.match_to_dict())
83	1	flow_dict['idle_timeout'] = self.idle_timeout
84	1	flow_dict['hard_timeout'] = self.hard_timeout
85	1	flow_dict['priority'] = self.priority
86	1	flow_dict['table_id'] = self.table_id
87	1	flow_dict['cookie'] = self.cookie
88	1	flow_dict['buffer_id'] = self.buffer_id
89	1	flow_dict['actions'] = []
90	1	for action in self.actions:
91	1	flow_dict['actions'].append(action.as_dict())
92
93	1	return flow_dict
94
95	1	def match_to_dict(self):
96		"""Convert a match in OF 1.3 to a dictionary."""
97		# pylint: disable=consider-using-dict-items
98	1	match = {}
99	1	for name in self.match.copy():
100	1	match[name] = self.match[name].value
101	1	return match
102
103	1	def to_json(self):
104		"""Return a json version of the flow."""
105		return json.dumps(self.to_dict())
106
107		# @staticmethod
108		# def from_dict(flow_dict):
109		# """Create a flow from a dict."""
110		# flow = GenericFlow()
111		# for attr_name, value in flow_dict.items():
112		# if attr_name == 'actions':
113		# flow.actions = []
114		# for action in value:
115		# new_action = ACTION_TYPES[int(action['action_type'])]()
116		# for action_attr_name,
117		# action_attr_value in action.items():
118		#
119		# setattr(new_action,
120		# action_attr_name,
121		# action_attr_value)
122		# flow.actions.append(new_action)
123		# else:
124		# setattr(flow, attr_name, value)
125		# return flow
126
127	1	@classmethod
128	1	def from_flow_stats(cls, flow_stats, version='0x01'):
129		"""Create a flow from OF flow stats."""
130	1	flow = GenericFlow(version=version)
131	1	flow.idle_timeout = flow_stats.idle_timeout.value
132	1	flow.hard_timeout = flow_stats.hard_timeout.value
133	1	flow.priority = flow_stats.priority.value
134	1	flow.table_id = flow_stats.table_id.value
135	1	flow.duration_sec = flow_stats.duration_sec.value
136	1	flow.packet_count = flow_stats.packet_count.value
137	1	flow.byte_count = flow_stats.byte_count.value
138	1	if version == '0x01':
139	1	flow.match['wildcards'] = flow_stats.match.wildcards.value
140	1	flow.match['in_port'] = flow_stats.match.in_port.value
141	1	flow.match['eth_src'] = flow_stats.match.dl_src.value
142	1	flow.match['eth_dst'] = flow_stats.match.dl_dst.value
143	1	flow.match['vlan_vid'] = flow_stats.match.dl_vlan.value
144	1	flow.match['vlan_pcp'] = flow_stats.match.dl_vlan_pcp.value
145	1	flow.match['eth_type'] = flow_stats.match.dl_type.value
146	1	flow.match['ip_tos'] = flow_stats.match.nw_tos.value
147	1	flow.match['ipv4_src'] = flow_stats.match.nw_src.value
148	1	flow.match['ipv4_dst'] = flow_stats.match.nw_dst.value
149	1	flow.match['ip_proto'] = flow_stats.match.nw_proto.value
150	1	flow.match['tcp_src'] = flow_stats.match.tp_src.value
151	1	flow.match['tcp_dst'] = flow_stats.match.tp_dst.value
152	1	flow.actions = []
153	1	for of_action in flow_stats.actions:
154	1	action = Action10.from_of_action(of_action)
155	1	flow.actions.append(action)
156		elif version == '0x04':
157		for match in flow_stats.match.oxm_match_fields:
158		match_field = MatchFieldFactory.from_of_tlv(match)
159		field_name = match_field.name
160		if field_name == 'dl_vlan':
161		field_name = 'vlan_vid'
162		flow.match[field_name] = match_field
163		flow.actions = []
164		for instruction in flow_stats.instructions:
165		if instruction.instruction_type == 'apply_actions':
166		for of_action in instruction.actions:
167		action = Action13.from_of_action(of_action)
168		flow.actions.append(action)
169	1	return flow
170
171	1	@classmethod
172	1	def from_replies_flows(cls, flow04):
173		"""Create a flow from a flow passed on
174		replies_flows in event kytos/of_core.flow_stats.received."""
175
176	1	flow = GenericFlow(version='0x04')
177	1	flow.idle_timeout = flow04.idle_timeout
178	1	flow.hard_timeout = flow04.hard_timeout
179	1	flow.priority = flow04.priority
180	1	flow.table_id = flow04.table_id
181	1	flow.cookie = flow04.cookie
182	1	flow.duration_sec = flow04.stats.duration_sec
183	1	flow.packet_count = flow04.stats.packet_count
184	1	flow.byte_count = flow04.stats.byte_count
185
186	1	as_of_match = flow04.match.as_of_match()
187	1	for match in as_of_match.oxm_match_fields:
188	1	match_field = MatchFieldFactory.from_of_tlv(match)
189	1	field_name = match_field.name
190	1	if field_name == 'dl_vlan':
191		field_name = 'vlan_vid'
192	1	flow.match[field_name] = match_field
193	1	flow.actions = []
194	1	for instruction in flow04.instructions:
195	1	if instruction.instruction_type == 'apply_actions':
196	1	for of_action in instruction.actions:
197	1	flow.actions.append(of_action)
198	1	return flow
199
200	1	def do_match(self, args):
201		"""Match a packet against this flow."""
202		if self.version == '0x01':
203		return self.match10(args)
204		if self.version == '0x04':
205		return self.match13(args)
206		return None
207
208	1	def match10(self, args):
209		"""Match a packet against this flow (OF1.0)."""
210		log.debug('Matching packet')
211		if not self.match['wildcards'] & FlowWildCards.OFPFW_IN_PORT:
212		if 'in_port' not in args:
213		return False
214		if self.match['in_port'] != int(args['in_port']):
215		return False
216		if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_VLAN_PCP:
217		if 'vlan_pcp' not in args:
218		return False
219		if self.match['vlan_pcp'] != int(args['vlan_pcp']):
220		return False
221		if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_VLAN:
222		if 'vlan_vid' not in args:
223		return False
224		if self.match['vlan_vid'] != args['vlan_vid'][-1]:
225		return False
226		if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_SRC:
227		if 'eth_src' not in args:
228		return False
229		if self.match['eth_src'] != args['eth_src']:
230		return False
231		if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_DST:
232		if 'eth_dst' not in args:
233		return False
234		if self.match['eth_dst'] != args['eth_dst']:
235		return False
236		if not self.match['wildcards'] & FlowWildCards.OFPFW_DL_TYPE:
237		if 'eth_type' not in args:
238		return False
239		if self.match['eth_type'] != int(args['eth_type']):
240		return False
241		if self.match['eth_type'] == constants.IPV4:
242		flow_ip_int = int(ipaddress.IPv4Address(self.match['ipv4_src']))
243		if flow_ip_int != 0:
244		mask = ((self.match['wildcards'] &
245		FlowWildCards.OFPFW_NW_SRC_MASK) >>
246		FlowWildCards.OFPFW_NW_SRC_SHIFT)
247		mask = min(mask, 32)
248		if mask != 32 and 'ipv4_src' not in args:
249		return False
250		mask = (0xffffffff << mask) & 0xffffffff
251		ip_int = int(ipaddress.IPv4Address(args['ipv4_src']))
252		if ip_int & mask != flow_ip_int & mask:
253		return False
254
255		flow_ip_int = int(ipaddress.IPv4Address(self.match['ipv4_dst']))
256		if flow_ip_int != 0:
257		mask = ((self.match['wildcards'] &
258		FlowWildCards.OFPFW_NW_DST_MASK) >>
259		FlowWildCards.OFPFW_NW_DST_SHIFT)
260		mask = min(mask, 32)
261		if mask != 32 and 'ipv4_dst' not in args:
262		return False
263		mask = (0xffffffff << mask) & 0xffffffff
264		ip_int = int(ipaddress.IPv4Address(args['ipv4_dst']))
265		if ip_int & mask != flow_ip_int & mask:
266		return False
267		if not self.match['wildcards'] & FlowWildCards.OFPFW_NW_TOS:
268		if 'ip_tos' not in args:
269		return False
270		if self.match['ip_tos'] != int(args['ip_tos']):
271		return False
272		if not self.match['wildcards'] & FlowWildCards.OFPFW_NW_PROTO:
273		if 'ip_proto' not in args:
274		return False
275		if self.match['ip_proto'] != int(args['ip_proto']):
276		return False
277		if not self.match['wildcards'] & FlowWildCards.OFPFW_TP_SRC:
278		if 'tp_src' not in args:
279		return False
280		if self.match['tcp_src'] != int(args['tp_src']):
281		return False
282		if not self.match['wildcards'] & FlowWildCards.OFPFW_TP_DST:
283		if 'tp_dst' not in args:
284		return False
285		if self.match['tcp_dst'] != int(args['tp_dst']):
286		return False
287		return self
288
289	1	def match13(self, args):
290		"""Match a packet against this flow (OF1.3)."""
291		# pylint: disable=consider-using-dict-items
292		for name in self.match.copy():
293		if name not in args:
294		return False
295		if name == 'vlan_vid':
296		field = args[name][-1]
297		else:
298		field = args[name]
299		if name not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
300		if self.match[name].value != field:
301		return False
302		else:
303		packet_ip = int(ipaddress.ip_address(field))
304		ip_addr = self.match[name].value
305		if packet_ip & ip_addr.netmask != ip_addr.address:
306		return False
307		return self
308
309
310		# pylint: disable=too-many-public-methods
311	1	class Main(KytosNApp):
312		"""Main class of amlight/flow_stats NApp.
313
314		This class is the entry point for this napp.
315		"""
316
317	1	def setup(self):
318		"""Replace the '__init__' method for the KytosNApp subclass.
319
320		The setup method is automatically called by the controller when your
321		application is loaded.
322
323		So, if you have any setup routine, insert it here.
324		"""
325	1	log.info('Starting Kytos/Amlight flow manager')
326	1	for switch in self.controller.switches.copy().values():
327		switch.generic_flows = []
328	1	self.switch_stats_xid = {}
329	1	self.switch_stats_lock = {}
330
331	1	def execute(self):
332		"""This method is executed right after the setup method execution.
333
334		You can also use this method in loop mode if you add to the above setup
335		method a line like the following example:
336
337		self.execute_as_loop(30) # 30-second interval.
338		"""
339
340	1	def shutdown(self):
341		"""This method is executed when your napp is unloaded.
342
343		If you have some cleanup procedure, insert it here.
344		"""
345
346	1	def flow_from_id(self, flow_id):
347		"""Flow from given flow_id."""
348	1	for switch in self.controller.switches.copy().values():
349	1	try:
350	1	for flow in switch.generic_flows:
351	1	if flow.id == flow_id:
352	1	return flow
353		except KeyError:
354		pass
355	1	return None
356
357	1	@rest('flow/match/<dpid>')
358	1	def flow_match(self, dpid):
359		"""Return first flow matching request."""
360	1	switch = self.controller.get_switch_by_dpid(dpid)
361	1	flow = self.match_flows(switch, format_request(request.args), False)
362	1	if flow:
363	1	return jsonify(flow.to_dict())
364	1	return "No match", 404
365
366	1	@rest('flow/stats/<dpid>')
367	1	def flow_stats(self, dpid):
368		"""Return all flows matching request."""
369	1	switch = self.controller.get_switch_by_dpid(dpid)
370	1	if not switch:
371		return f"switch {dpid} not found", 404
372	1	flows = self.match_flows(switch, format_request(request.args), True)
373	1	flows = [flow.to_dict() for flow in flows]
374	1	return jsonify(flows)
375
376	1	@staticmethod
377	1	def match_flows(switch, args, many=True):
378		# pylint: disable=bad-staticmethod-argument
379		"""
380		Match the packet in request against the flows installed in the switch.
381
382		Try the match with each flow, in other. If many is True, tries the
383		match with all flows, if False, tries until the first match.
384		:param args: packet data
385		:param many: Boolean, indicating whether to continue after matching the
386		first flow or not
387		:return: If many, the list of matched flows, or the matched flow
388		"""
389		response = []
390		try:
391		for flow in switch.generic_flows:
392		match = flow.do_match(args)
393		if match:
394		if many:
395		response.append(match)
396		else:
397		response = match
398		break
399		except AttributeError:
400		return None
401		if not many and isinstance(response, list):
402		return None
403		return response
404
405	1	@staticmethod
406	1	def match_and_apply(switch, args):
407		# pylint: disable=bad-staticmethod-argument
408		"""Match flows and apply actions.
409
410		Match given packet (in args) against the switch flows and,
411		if a match flow is found, apply its actions."""
412		flow = Main.match_flows(switch, args, False)
413		port = None
414		actions = None
415		# pylint: disable=too-many-nested-blocks
416		if flow:
417		actions = flow.actions
418		if switch.ofp_version == '0x01':
419		for action in actions:
420		action_type = action.action_type
421		if action_type == 'output':
422		port = action.port
423		elif action_type == 'set_vlan':
424		if 'vlan_vid' in args:
425		args['vlan_vid'][-1] = action.vlan_id
426		else:
427		args['vlan_vid'] = [action.vlan_id]
428		elif switch.ofp_version == '0x04':
429		for action in actions:
430		action_type = action.action_type
431		if action_type == 'output':
432		port = action.port
433		if action_type == 'push_vlan':
434		if 'vlan_vid' not in args:
435		args['vlan_vid'] = []
436		args['vlan_vid'].append(0)
437		if action_type == 'pop_vlan':
438		if 'vlan_vid' in args:
439		args['vlan_vid'].pop()
440		if len(args['vlan_vid']) == 0:
441		del args['vlan_vid']
442		if action_type == 'set_vlan':
443		args['vlan_vid'][-1] = action.vlan_id
444		return flow, args, port
445
446	1	@rest('packet_count/<flow_id>')
447	1	def packet_count(self, flow_id):
448		"""Packet count of an specific flow."""
449	1	flow = self.flow_from_id(flow_id)
450	1	if flow is None:
451	1	return "Flow does not exist", 404
452	1	packet_stats = {
453		'flow_id': flow_id,
454		'packet_counter': flow.packet_count,
455		'packet_per_second': flow.packet_count / flow.duration_sec
456		}
457	1	return jsonify(packet_stats)
458
459	1	@rest('bytes_count/<flow_id>')
460	1	def bytes_count(self, flow_id):
461		"""Bytes count of an specific flow."""
462	1	flow = self.flow_from_id(flow_id)
463	1	if flow is None:
464	1	return "Flow does not exist", 404
465	1	bytes_stats = {
466		'flow_id': flow_id,
467		'bytes_counter': flow.byte_count,
468		'bits_per_second': flow.byte_count * 8 / flow.duration_sec
469		}
470	1	return jsonify(bytes_stats)
471
472	1	@rest('packet_count/per_flow/<dpid>')
473	1	def packet_count_per_flow(self, dpid):
474		"""Per flow packet count."""
475	1	return self.flows_counters('packet_count',
476		dpid,
477		counter='packet_counter',
478		rate='packet_per_second')
479
480	1	@rest('packet_count/sum/<dpid>')
481	1	def packet_count_sum(self, dpid):
482		"""Sum of packet count flow stats."""
483	1	return self.flows_counters('packet_count',
484		dpid,
485		total=True)
486
487	1	@rest('bytes_count/per_flow/<dpid>')
488	1	def bytes_count_per_flow(self, dpid):
489		"""Per flow bytes count."""
490	1	return self.flows_counters('byte_count',
491		dpid,
492		counter='bytes_counter',
493		rate='bits_per_second')
494
495	1	@rest('bytes_count/sum/<dpid>')
496	1	def bytes_count_sum(self, dpid):
497		"""Sum of bytes count flow stats."""
498	1	return self.flows_counters('byte_count',
499		dpid,
500		total=True)
501
502	1	def flows_counters(self, field, dpid, counter=None, rate=None,
503		total=False):
504		"""Calculate flows statistics.
505
506		The returned statistics are both per flow and for the sum of flows
507		"""
508		# pylint: disable=too-many-arguments
509		# pylint: disable=unused-variable
510	1	start_date = request.args.get('start_date', 0)
511	1	end_date = request.args.get('end_date', 0)
512		# pylint: enable=unused-variable
513
514	1	if total:
515	1	count_flows = 0
516		else:
517	1	count_flows = []
518	1	if not counter:
519		counter = field
520	1	if not rate:
521		rate = field
522
523		# We don't have statistics persistence yet, so for now this only works
524		# for start and end equals to zero
525	1	flows = self.controller.get_switch_by_dpid(dpid).generic_flows
526
527	1	for flow in flows:
528	1	count = getattr(flow, field)
529	1	if total:
530	1	count_flows += count
531		else:
532	1	per_second = count / flow.duration_sec
533	1	if rate.startswith('bits'):
534	1	per_second *= 8
535	1	count_flows.append({'flow_id': flow.id,
536		counter: count,
537		rate: per_second})
538
539	1	return jsonify(count_flows)
540
541	1	@listen_to('kytos/of_core.v0x01.messages.in.ofpt_stats_reply')
542	1	def on_stats_reply_0x01(self, event):
543		"""Capture flow stats messages for v0x01 switches."""
544		self.handle_stats_reply_0x01(event)
545
546	1	def handle_stats_reply_0x01(self, event):
547		"""Handle stats replies for v0x01 switches."""
548	1	msg = event.content['message']
549	1	if msg.body_type == common01.StatsType.OFPST_FLOW:
550	1	switch = event.source.switch
551	1	self.handle_stats_reply(msg, switch)
552
553	1	def handle_stats_reply(self, msg, switch):
554		"""Insert flows received in the switch list of flows."""
555	1	try:
556	1	old_flows = switch.generic_flows
557		except AttributeError:
558		old_flows = []
559	1	is_new_xid = (
560		int(msg.header.xid) != self.switch_stats_xid.get(switch.id, 0)
561		)
562	1	is_last_part = msg.flags.value % 2 == 0
563	1	self.switch_stats_lock.setdefault(switch.id, Lock())
564	1	with self.switch_stats_lock[switch.id]:
565	1	if is_new_xid:
566	1	switch.generic_new_flows = []
567	1	self.switch_stats_xid[switch.id] = int(msg.header.xid)
568	1	for flow_stats in msg.body:
569	1	flow = GenericFlow.from_flow_stats(flow_stats,
570		switch.ofp_version)
571	1	switch.generic_new_flows.append(flow)
572	1	if is_last_part:
573	1	switch.generic_flows = switch.generic_new_flows
574	1	switch.generic_flows.sort(
575		key=lambda f: (f.priority, f.duration_sec),
576		reverse=True
577		)
578	1	if is_new_xid and is_last_part and switch.generic_flows != old_flows:
579		# Generate an event informing that flows have changed
580	1	event = KytosEvent('amlight/flow_stats.flows_updated')
581	1	event.content['switch'] = switch.dpid
582	1	self.controller.buffers.app.put(event)
583
584	1	@listen_to('kytos/of_core.flow_stats.received')
585	1	def on_stats_received(self, event):
586		"""Capture flow stats messages for OpenFlow 1.3."""
587		self.handle_stats_received(event)
588
589	1	def handle_stats_received(self, event):
590		"""Handle flow stats messages for OpenFlow 1.3."""
591	1	switch = event.content['switch']
592	1	if 'replies_flows' in event.content:
593	1	replies_flows = event.content['replies_flows']
594	1	self.handle_stats_reply_received(switch, replies_flows)
595
596		# pylint: disable=no-self-use
597	1	def handle_stats_reply_received(self, switch, replies_flows):
598		"""Iterate on the replies and set the generic flows"""
599	1	switch.generic_flows = [GenericFlow.from_replies_flows(flow)
600		for flow in replies_flows]
601	1	switch.generic_flows.sort(
602		key=lambda f: (f.priority, f.duration_sec),
603		reverse=True
604		)
605

kytos-ng / flow_stats

Pull Request — master (#34)

build.main.Main.on_stats_received() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like