ponsim/ponsim.py

#
# Copyright 2016 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
Simple PON Simulator which would not be needed if openvswitch could do
802.1ad (QinQ), which it cannot (the reason is beyond me), or if CPQD could
handle 0-tagged packets (no comment).
"""
import structlog
import random
import arrow
import json
from scapy.layers.inet import IP, UDP, TCP, Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.packet import Packet

from voltha.protos import third_party
from voltha.protos.ponsim_pb2 import PonSimMetrics, PonSimPortMetrics, \
    PonSimPacketCounter
from voltha.protos.events_pb2 import AlarmEventType, AlarmEventSeverity, \
    AlarmEventState, AlarmEventCategory
from voltha.core.flow_decomposer import *
from twisted.internet.task import LoopingCall
from twisted.internet import reactor

_ = third_party


def ipv4int2str(ipv4int):
    return '{}.{}.{}.{}'.format(
        (ipv4int >> 24) & 0xff,
        (ipv4int >> 16) & 0xff,
        (ipv4int >> 8) & 0xff,
        ipv4int & 0xff
    )


class _FlowMatchMask(object):
    """
    Enum of mask values based on flow match priority. For instance, a port
    match has higher priority when match that a UDP match.
    """
    UDP_DST = 1
    UDP_SRC = 2
    IPV4_DST = 4
    VLAN_PCP = 8
    VLAN_VID = 16
    IP_PROTO = 34
    ETH_TYPE = 64
    IN_PORT = 128


class FrameIOCounter(object):
    class SingleFrameCounter(object):
        def __init__(self, name, min, max):
            # Currently there are 2 values, one for the PON interface (port 1)
            # and one for the Network Interface (port 2). This can be extended if
            # the virtual devices extend the number of ports. 
            self.value = [0, 0]  # {PON,NI}
            self.name = name
            self.min = min
            self.max = max

    def __init__(self, device):
        self.device = device
        self.tx_counters = dict(
            tx_64_pkts=self.SingleFrameCounter("tx_64_pkts", 1, 64),
            tx_65_127_pkts=self.SingleFrameCounter("tx_65_127_pkts", 65, 127),
            tx_128_255_pkts=self.SingleFrameCounter("tx_128_255_pkts", 128, 255),
            tx_256_511_pkts=self.SingleFrameCounter("tx_256_511_pkts", 256, 511),
            tx_512_1023_pkts=self.SingleFrameCounter("tx_512_1023_pkts", 512, 1024),
            tx_1024_1518_pkts=self.SingleFrameCounter("tx_1024_1518_pkts", 1024, 1518),
            tx_1519_9k_pkts=self.SingleFrameCounter("tx_1519_9k_pkts", 1519, 9216),
        )
        self.rx_counters = dict(
            rx_64_pkts=self.SingleFrameCounter("rx_64_pkts", 1, 64),
            rx_65_127_pkts=self.SingleFrameCounter("rx_65_127_pkts", 65, 127),
            rx_128_255_pkts=self.SingleFrameCounter("rx_128_255_pkts", 128, 255),
            rx_256_511_pkts=self.SingleFrameCounter("rx_256_511_pkts", 256, 511),
            rx_512_1023_pkts=self.SingleFrameCounter("rx_512_1023_pkts", 512, 1024),
            rx_1024_1518_pkts=self.SingleFrameCounter("rx_1024_1518_pkts", 1024, 1518),
            rx_1519_9k_pkts=self.SingleFrameCounter("rx_1519_9k_pkts", 1519, 9216)
        )

    def count_rx_frame(self, port, size):
        log.info("counting-rx-frame", size=size, port=port)
        for k, v in self.rx_counters.iteritems():
            if size >= v.min and size <= v.max:
                self.rx_counters[k].value[port - 1] += 1
                return
        log.warn("unsupported-packet-size", size=size)

    def count_tx_frame(self, port, size):
        for k, v in self.tx_counters.iteritems():
            if size >= v.min and size <= v.max:
                self.tx_counters[k].value[port - 1] += 1
                return
        log.warn("unsupported-packet-size", size=size)

    def log_counts(self):
        rx_ct_list = [(v.name, v.value[0], v.value[1]) for v in
                      self.rx_counters.values()]
        tx_ct_list = [(v.name, v.value[0], v.value[1]) for v in
                      self.tx_counters.values()]
        log.info("rx-counts", rx_ct_list=rx_ct_list)
        log.info("tx-counts", tx_ct_list=tx_ct_list)

    def make_proto(self):
        sim_metrics = PonSimMetrics(
            device=self.device
        )
        pon_port_metrics = PonSimPortMetrics(
            port_name="pon"
        )
        ni_port_metrics = PonSimPortMetrics(
            port_name="nni"
        )
        for c in sorted(self.rx_counters):
            ctr = self.rx_counters[c]
            pon_port_metrics.packets.extend([
                PonSimPacketCounter(name=ctr.name, value=ctr.value[0])
            ])
            # Since they're identical keys, save some time and cheat
            ni_port_metrics.packets.extend([
                PonSimPacketCounter(name=ctr.name, value=ctr.value[1])
            ])

        for c in sorted(self.tx_counters):
            ctr = self.tx_counters[c]
            pon_port_metrics.packets.extend([
                PonSimPacketCounter(name=ctr.name, value=ctr.value[0])
            ])
            # Since they're identical keys, save some time and cheat
            ni_port_metrics.packets.extend([
                PonSimPacketCounter(name=ctr.name, value=ctr.value[1])
            ])
        sim_metrics.metrics.extend([pon_port_metrics])
        sim_metrics.metrics.extend([ni_port_metrics])

        return sim_metrics


class SimAlarms:
    def __init__(self):
        self.lc = None

    @staticmethod
    def _prepare_alarm():
        alarm_event = dict()

        try:
            # Randomly choose values for each enum types
            alm_severity = random.choice(list(
                v for k, v in
                AlarmEventSeverity.DESCRIPTOR.enum_values_by_name.items()))

            alm_type = random.choice(list(
                v for k, v in
                AlarmEventType.DESCRIPTOR.enum_values_by_name.items()))

            alm_category = random.choice(list(
                v for k, v in
                AlarmEventCategory.DESCRIPTOR.enum_values_by_name.items()))

            alarm_event['severity'] = alm_severity.number
            alarm_event['type'] = alm_type.number
            alarm_event['category'] = alm_category.number
            alarm_event['state'] = AlarmEventState.RAISED
            alarm_event['ts'] = arrow.utcnow().timestamp
            alarm_event['description'] = "{}.{} alarm".format(alm_type.name, alm_category.name)

            return alarm_event

        except Exception as e:
            log.exception('failed-to-prepare-alarm', e=e)

    @staticmethod
    def _raise_alarm(alarm_event, olt, egress):
        try:
            frame = Ether() / Dot1Q(vlan=4000) / IP() / TCP() / Raw(load=json.dumps(alarm_event))
            egress(0, frame)

        except Exception as e:
            log.exception('failed-to-raise-alarm', e=e)

    @staticmethod
    def _clear_alarm(alarm_event, olt, egress):
        try:
            alarm_event['state'] = AlarmEventState.CLEARED
            frame = Ether() / Dot1Q(vlan=4000) / IP() / TCP() / Raw(load=json.dumps(alarm_event))
            egress(0, frame)

        except Exception as e:
            log.exception('failed-to-clear-alarm', e=e)

    def _generate_alarm(self, olt, egress):
        try:
            alarm = self._prepare_alarm()
            self._raise_alarm(alarm, olt, egress)
            reactor.callLater(random.randint(20, 60), self._clear_alarm, alarm, olt, egress)
        except Exception as e:
            log.exception(e=e)

    def start_simulation(self, olt, egress, config):
        log.info("starting-alarm-simulation")

        """Simulate periodic device alarms"""
        self.lc = LoopingCall(self._generate_alarm, olt, egress)
        self.lc.start(config['frequency'])

    def stop_simulation(self):
        log.info("stopping-alarm-simulation")
        self.lc.stop()


class SimDevice(object):
    def __init__(self, name, logical_port_no):
        self.name = name
        self.logical_port_no = logical_port_no
        self.links = dict()
        self.flows = list()
        self.log = structlog.get_logger(name=name,
                                        logical_port_no=logical_port_no)
        self.counter = FrameIOCounter(name)

    def link(self, port, egress_fun):
        self.links.setdefault(port, []).append(egress_fun)

    def ingress(self, port, frame):
        self.log.debug('ingress', ingress_port=port, name=self.name)
        self.counter.count_rx_frame(port, len(frame["Ether"].payload))
        outcome = self.process_frame(port, frame)
        if outcome is not None:
            egress_port, egress_frame = outcome
            forwarded = 0
            links = self.links.get(egress_port)
            if links is not None:
                self.counter.count_tx_frame(egress_port,
                                            len(egress_frame["Ether"].payload))
                for fun in links:
                    forwarded += 1
                    self.log.debug('forwarding', egress_port=egress_port)
                    fun(egress_port, egress_frame)
            if not forwarded:
                self.log.debug('no-one-to-forward-to', egress_port=egress_port)
        else:
            self.log.debug('dropped')

    def install_flows(self, flows):
        # store flows in precedence order so we can roll down on frame arrival
        self.flows = sorted(flows, key=lambda fm: fm.priority, reverse=True)

    def process_frame(self, ingress_port, ingress_frame):
        matched_mask = 0
        highest_priority = 0
        matched_flow = None
        for flow in self.flows:
            # flows are sorted by highest priority.
            if matched_flow and flow.priority < highest_priority:
                break

            highest_priority = flow.priority
            current_mask = self.is_match(flow, ingress_port, ingress_frame)
            if current_mask > matched_mask:
                matched_mask = current_mask
                matched_flow = flow

        if matched_flow:
            egress_port, egress_frame = self.process_actions(
                matched_flow, ingress_frame)
            return egress_port, egress_frame
        return None

    @staticmethod
    def is_match(flow, ingress_port, frame):
        matched_mask = 0

        def get_non_shim_ether_type(f):
            if f.haslayer(Dot1Q):
                f = f.getlayer(Dot1Q)
            return f.type

        def get_vlan_pcp(f):
            if f.haslayer(Dot1Q):
                return f.getlayer(Dot1Q).prio

        def get_ip_proto(f):
            if f.haslayer(IP):
                return f.getlayer(IP).proto

        def get_ipv4_dst(f):
            if f.haslayer(IP):
                return f.getlayer(IP).dst

        def get_udp_src(f):
            if f.haslayer(UDP):
                return f.getlayer(UDP).sport

        def get_udp_dst(f):
            if f.haslayer(UDP):
                return f.getlayer(UDP).dport

        for field in get_ofb_fields(flow):

            if field.type == IN_PORT:
                if field.port != ingress_port:
                    return 0
                matched_mask |= _FlowMatchMask.IN_PORT

            elif field.type == ETH_TYPE:
                if field.eth_type != get_non_shim_ether_type(frame):
                    return 0
                matched_mask |= _FlowMatchMask.ETH_TYPE

            elif field.type == IP_PROTO:
                if field.ip_proto != get_ip_proto(frame):
                    return 0
                matched_mask |= _FlowMatchMask.IP_PROTO

            elif field.type == VLAN_VID:
                expected_vlan = field.vlan_vid
                tagged = frame.haslayer(Dot1Q)
                if bool(expected_vlan & 4096) != bool(tagged):
                    return 0
                if tagged:
                    actual_vid = frame.getlayer(Dot1Q).vlan
                    if actual_vid != expected_vlan & 4095:
                        return 0
                matched_mask |= _FlowMatchMask.VLAN_VID

            elif field.type == VLAN_PCP:
                if field.vlan_pcp != get_vlan_pcp(frame):
                    return 0
                matched_mask |= _FlowMatchMask.VLAN_PCP

            elif field.type == IPV4_DST:
                if ipv4int2str(field.ipv4_dst) != get_ipv4_dst(frame):
                    return 0
                matched_mask |= _FlowMatchMask.IPV4_DST

            elif field.type == UDP_SRC:
                if field.udp_src != get_udp_src(frame):
                    return 0
                matched_mask |= _FlowMatchMask.UDP_SRC

            elif field.type == UDP_DST:
                if field.udp_dst != get_udp_dst(frame):
                    return 0
                matched_mask |= _FlowMatchMask.UDP_DST

            elif field.type == METADATA:
                pass  # safe to ignore

            else:
                raise NotImplementedError('field.type=%d' % field.type)

        return matched_mask

    @staticmethod
    def process_actions(flow, frame):
        egress_port = None
        for action in get_actions(flow):

            if action.type == OUTPUT:
                egress_port = action.output.port

            elif action.type == POP_VLAN:
                if frame.haslayer(Dot1Q):
                    shim = frame.getlayer(Dot1Q)
                    frame = Ether(
                        src=frame.src,
                        dst=frame.dst,
                        type=shim.type) / shim.payload

            elif action.type == PUSH_VLAN:
                frame = (
                    Ether(src=frame.src, dst=frame.dst,
                          type=action.push.ethertype) /
                    Dot1Q(type=frame.type) /
                    frame.payload
                )

            elif action.type == SET_FIELD:
                assert (action.set_field.field.oxm_class ==
                        ofp.OFPXMC_OPENFLOW_BASIC)
                field = action.set_field.field.ofb_field

                if field.type == VLAN_VID:
                    shim = frame.getlayer(Dot1Q)
                    shim.vlan = field.vlan_vid & 4095

                elif field.type == VLAN_PCP:
                    shim = frame.getlayer(Dot1Q)
                    shim.prio = field.vlan_pcp

                else:
                    raise NotImplementedError('set_field.field.type=%d'
                                              % field.type)

            else:
                raise NotImplementedError('action.type=%d' % action.type)

        return egress_port, frame


class PonSim(object):
    def __init__(self, onus, egress_fun, alarm_config):
        self.egress_fun = egress_fun

        self.log = structlog.get_logger()
        # Create OLT and hook NNI port up for egress
        self.olt = SimDevice('olt', 0)
        self.olt.link(2, lambda _, frame: self.egress_fun(0, frame))
        self.devices = dict()
        self.devices[0] = self.olt
        # TODO: This can be removed, it's for debugging purposes
        self.lc = LoopingCall(self.olt.counter.log_counts)
        self.lc.start(90)  # To correlate with Kafka

        # Create ONUs of the requested number and hook them up with OLT
        # and with egress fun
        def mk_egress_fun(port_no):
            return lambda _, frame: self.egress_fun(port_no, frame)

        def mk_onu_ingress(onu):
            return lambda _, frame: onu.ingress(1, frame)

        for i in range(onus):
            port_no = 128 + i
            onu = SimDevice('onu%d' % i, port_no)
            onu.link(1, lambda _, frame: self.olt.ingress(1,
                                                          frame))  # Send to the OLT
            onu.link(2,
                     mk_egress_fun(port_no))  # Send from the ONU to the world
            self.olt.link(1, mk_onu_ingress(onu))  # Internal send to the ONU
            self.devices[port_no] = onu
        for d in self.devices:
            self.log.info("pon-sim-init", port=d, name=self.devices[d].name,
                          links=self.devices[d].links)

        if alarm_config['simulation']:
            self.alarms = SimAlarms()
            self.alarms.start_simulation(self.olt, self.egress_fun, alarm_config)

    def get_ports(self):
        return sorted(self.devices.keys())

    def get_stats(self):
        return self.olt.counter.make_proto()

    def olt_install_flows(self, flows):
        self.olt.install_flows(flows)

    def onu_install_flows(self, onu_port, flows):
        self.devices[onu_port].install_flows(flows)

    def ingress(self, port, frame):
        if not isinstance(frame, Packet):
            frame = Ether(frame)
        self.devices[port].ingress(2, frame)

class XPonSim(object):
    def __init__(self):
        self.log = structlog.get_logger()

    def CreateInterface(self, request):
        self.log.info("create-interface-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def UpdateInterface(self, request):
        self.log.info("update-interface-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def RemoveInterface(self, request):
        self.log.info("remove-interface-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def CreateTcont(self, request, request2):
        self.log.info("create-tcont-request",
                      tcont_config_data=request,
                      traffic_descriptor_profile_config_data=request2)
        return

    def UpdateTcont(self, request, request2):
        self.log.info("update-tcont-request",
                      tcont_config_data=request,
                      traffic_descriptor_profile_config_data=request2)
        return

    def RemoveTcont(self, request, request2):
        self.log.info("remove-tcont-request",
                      tcont_config_data=request,
                      traffic_descriptor_profile_config_data=request2)
        return

    def CreateGemport(self, request):
        self.log.info("create-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def UpdateGemport(self, request):
        self.log.info("update-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def RemoveGemport(self, request):
        self.log.info("remove-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def CreateMulticastGemport(self, request):
        self.log.info("create-multicast-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def UpdateMulticastGemport(self, request):
        self.log.info("update-multicast-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def RemoveMulticastGemport(self, request):
        self.log.info("remove-multicast-gemport-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def CreateMulticastDistributionSet(self, request):
        self.log.info("create-multicast-distribution-set-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def UpdateMulticastDistributionSet(self, request):
        self.log.info("update-multicast-distribution-set-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return

    def RemoveMulticastDistributionSet(self, request):
        self.log.info("remove-multicast-distribution-set-request",
                      interface_type=request.WhichOneof("interface_type"),
                      data=request)
        return