build.main.Main.on_stats_reply_0x04() - 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-13 19:12 UTC

build.main.Main.on_stats_reply_0x04() A

↳ Parent: build.main.Main.flows_counters()

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1.037

Importance

Changes

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

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

kytos-ng / flow_stats

Pull Request — master (#34)

build.main.Main.on_stats_reply_0x04() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like