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

Test Failed

Pull Request — master (#34)

unknown

created 2022-09-15 01:38 UTC

build.main.Main.handle_stats_reply_received() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	7
ccs	0
cts	0
cp	0
rs	10
c	0
b	0
f	0
cc	2
nop	3
crap	6

"""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
from pyof.v0x04.common.flow_instructions import InstructionType


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

kytos-ng / flow_stats

Pull Request — master (#34)

build.main.Main.handle_stats_reply_received() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like