python/adapters/openolt/openolt.py

#
# Copyright 2018 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.
#

"""
Openolt adapter.
"""
import arrow
import grpc
import structlog
from google.protobuf.empty_pb2 import Empty
from google.protobuf.json_format import MessageToDict
from scapy.layers.inet import Raw
import json
from google.protobuf.message import Message
from grpc._channel import _Rendezvous
from scapy.layers.l2 import Ether, Dot1Q
from simplejson import dumps
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks, returnValue
from twisted.internet.task import LoopingCall

from pyvoltha.adapters.common.frameio.frameio import BpfProgramFilter, hexify
from pyvoltha.adapters.iadapter import OltAdapter
from pyvoltha.common.utils.asleep import asleep
from pyvoltha.common.utils.registry import registry
from pyvoltha.adapters.kafka.kafka_proxy import get_kafka_proxy
from pyvoltha.protos import openolt_pb2
from pyvoltha.protos import third_party
from pyvoltha.protos.common_pb2 import OperStatus, ConnectStatus
from pyvoltha.protos.common_pb2 import LogLevel
from pyvoltha.protos.common_pb2 import OperationResp
from pyvoltha.protos.inter_container_pb2 import SwitchCapability, PortCapability, \
    InterAdapterMessageType, InterAdapterResponseBody
from pyvoltha.protos.device_pb2 import Port, PmConfig, PmConfigs, \
    DeviceType, DeviceTypes
from pyvoltha.protos.adapter_pb2 import Adapter
from pyvoltha.protos.adapter_pb2 import AdapterConfig
from openolt_flow_mgr import OpenOltFlowMgr
from openolt_alarms import OpenOltAlarmMgr
from openolt_statistics import OpenOltStatisticsMgr
from openolt_bw import OpenOltBW
from openolt_platform import OpenOltPlatform
from openolt_resource_manager import OpenOltResourceMgr
from openolt_device import OpenoltDevice
 
from pyvoltha.protos.events_pb2 import KpiEvent, KpiEventType, MetricValuePairs
from pyvoltha.protos.logical_device_pb2 import LogicalPort
from pyvoltha.protos.openflow_13_pb2 import OFPPS_LIVE, OFPPF_FIBER, \
    OFPPF_1GB_FD, \
    OFPC_GROUP_STATS, OFPC_PORT_STATS, OFPC_TABLE_STATS, OFPC_FLOW_STATS, \
    ofp_switch_features, ofp_desc
from pyvoltha.protos.openflow_13_pb2 import ofp_port

_ = third_party
log = structlog.get_logger()
OpenOltDefaults = {
    'support_classes': {
        'platform': OpenOltPlatform,
        'resource_mgr': OpenOltResourceMgr,
        'flow_mgr': OpenOltFlowMgr,
        'alarm_mgr': OpenOltAlarmMgr,
        'stats_mgr': OpenOltStatisticsMgr,
        'bw_mgr': OpenOltBW
    }
}

class AdapterPmMetrics:
    def __init__(self, device):
        self.pm_names = {'tx_64_pkts', 'tx_65_127_pkts', 'tx_128_255_pkts',
                         'tx_256_511_pkts', 'tx_512_1023_pkts',
                         'tx_1024_1518_pkts', 'tx_1519_9k_pkts',
                         'rx_64_pkts', 'rx_65_127_pkts',
                         'rx_128_255_pkts', 'rx_256_511_pkts',
                         'rx_512_1023_pkts', 'rx_1024_1518_pkts',
                         'rx_1519_9k_pkts'}
        self.device = device
        self.id = device.id
        self.name = 'ponsim_olt'
        self.default_freq = 150
        self.grouped = False
        self.freq_override = False
        self.pon_metrics_config = dict()
        self.nni_metrics_config = dict()
        self.lc = None
        for m in self.pm_names:
            self.pon_metrics_config[m] = PmConfig(name=m,
                                                  type=PmConfig.COUNTER,
                                                  enabled=True)
            self.nni_metrics_config[m] = PmConfig(name=m,
                                                  type=PmConfig.COUNTER,
                                                  enabled=True)

    def update(self, pm_config):
        if self.default_freq != pm_config.default_freq:
            # Update the callback to the new frequency.
            self.default_freq = pm_config.default_freq
            self.lc.stop()
            self.lc.start(interval=self.default_freq / 10)
        for m in pm_config.metrics:
            self.pon_metrics_config[m.name].enabled = m.enabled
            self.nni_metrics_config[m.name].enabled = m.enabled

    def make_proto(self):
        pm_config = PmConfigs(
            id=self.id,
            default_freq=self.default_freq,
            grouped=False,
            freq_override=False)
        for m in sorted(self.pon_metrics_config):
            pm = self.pon_metrics_config[m]  # Either will do they're the same
            pm_config.metrics.extend([PmConfig(name=pm.name,
                                               type=pm.type,
                                               enabled=pm.enabled)])
        return pm_config

    def collect_port_metrics(self, channel):
        rtrn_port_metrics = dict()
        stub = ponsim_pb2.PonSimStub(channel)
        stats = stub.GetStats(Empty())
        rtrn_port_metrics['pon'] = self.extract_pon_metrics(stats)
        rtrn_port_metrics['nni'] = self.extract_nni_metrics(stats)
        return rtrn_port_metrics

    def extract_pon_metrics(self, stats):
        rtrn_pon_metrics = dict()
        for m in stats.metrics:
            if m.port_name == "pon":
                for p in m.packets:
                    if self.pon_metrics_config[p.name].enabled:
                        rtrn_pon_metrics[p.name] = p.value
                return rtrn_pon_metrics

    def extract_nni_metrics(self, stats):
        rtrn_pon_metrics = dict()
        for m in stats.metrics:
            if m.port_name == "nni":
                for p in m.packets:
                    if self.pon_metrics_config[p.name].enabled:
                        rtrn_pon_metrics[p.name] = p.value
                return rtrn_pon_metrics

    def start_collector(self, callback):
        log.info("starting-pm-collection", device_name=self.name,
                 device_id=self.device.id)
        prefix = 'voltha.{}.{}'.format(self.name, self.device.id)
        self.lc = LoopingCall(callback, self.device.id, prefix)
        self.lc.start(interval=self.default_freq / 10)

    def stop_collector(self):
        log.info("stopping-pm-collection", device_name=self.name,
                 device_id=self.device.id)
        self.lc.stop()


class AdapterAlarms:
    def __init__(self, adapter, device):
        self.adapter = adapter
        self.device = device
        self.lc = None

    # TODO: Implement code to send to kafka cluster directly instead of
    # going through the voltha core.
    def send_alarm(self, context_data, alarm_data):
        log.debug("send-alarm-not-implemented")
        return



class OpenoltAdapter(OltAdapter):
    name = 'openolt'

    supported_device_types = [
        DeviceType(
            id=name,
            adapter=name,
            accepts_bulk_flow_update=True,
            accepts_add_remove_flow_updates=True
        )
    ]

    # System Init Methods #
    def __init__(self, core_proxy, adapter_proxy, config):
	super(OpenoltAdapter, self).__init__(core_proxy=core_proxy,
                                               adapter_proxy=adapter_proxy,
                                               config=config,
                                               device_handler_class=OpenoltHandler,
                                               name='openolt',
                                               vendor='Voltha project',
                                               version='0.4',
                                               device_type='openolt',
                                               accepts_bulk_flow_update=True,
                                               accepts_add_remove_flow_updates=False)
        self.adapter_proxy = adapter_proxy
        self.core_proxy = core_proxy
        self.config = config
        self.descriptor = Adapter(
            id=self.name,
            vendor='OLT white box vendor',
            version='0.1',
            config=config
        )
        log.debug('openolt.__init__', adapter_proxy=adapter_proxy)
        self.devices = dict()  # device_id -> OpenoltDevice()
        self.interface = registry('main').get_args().interface
        self.logical_device_id_to_root_device_id = dict()
        self.num_devices = 0

    def start(self):
        log.info('started', interface=self.interface)

    def stop(self):
        log.info('stopped', interface=self.interface)


    # Info Methods #
    def adapter_descriptor(self):
        log.debug('get descriptor', interface=self.interface)
        return self.descriptor

    def device_types(self):
        log.debug('get device_types', interface=self.interface,
                  items=self.supported_device_types)
        return DeviceTypes(items=self.supported_device_types)

    def health(self):
        log.debug('get health', interface=self.interface)
        raise NotImplementedError()

    def get_device_details(self, device):
        log.debug('get_device_details', device=device)
        raise NotImplementedError()


    # Device Operation Methods #
    def change_master_state(self, master):
        log.debug('change_master_state', interface=self.interface,
                  master=master)
        raise NotImplementedError()

    def abandon_device(self, device):
        log.info('abandon-device', device=device)
        raise NotImplementedError()


    # Configuration Methods #
    def update_flows_incrementally(self, device, flow_changes, group_changes):
        log.debug('update_flows_incrementally', device=device,
                  flow_changes=flow_changes, group_changes=group_changes)
        log.info('This device does not allow this, therefore it is Not '
                 'implemented')
        raise NotImplementedError()

    def update_pm_config(self, device, pm_configs):
        log.info('update_pm_config - Not implemented yet', device=device,
                 pm_configs=pm_configs)
        raise NotImplementedError()

    def receive_proxied_message(self, proxy_address, msg):
        log.debug('receive_proxied_message - Not implemented',
                  proxy_address=proxy_address,
                  proxied_msg=msg)
        raise NotImplementedError()

    def receive_inter_adapter_message(self, msg):
        log.info('rx_inter_adapter_msg - Not implemented')
        raise NotImplementedError()


    # Image Operations Methods #
    def download_image(self, device, request):
        log.info('image_download - Not implemented yet', device=device,
                 request=request)
        raise NotImplementedError()

    def get_image_download_status(self, device, request):
        log.info('get_image_download - Not implemented yet', device=device,
                 request=request)
        raise NotImplementedError()

    def cancel_image_download(self, device, request):
        log.info('cancel_image_download - Not implemented yet', device=device)
        raise NotImplementedError()

    def activate_image_update(self, device, request):
        log.info('activate_image_update - Not implemented yet',
                 device=device, request=request)
        raise NotImplementedError()

    def revert_image_update(self, device, request):
        log.info('revert_image_update - Not implemented yet',
                 device=device, request=request)
        raise NotImplementedError()

    def self_test_device(self, device):
        log.info('Not implemented yet')
        raise NotImplementedError()


    # PON Operations Methods #
    def create_interface(self, device, data):
        log.debug('create-interface - Not implemented - We do not use this',
                  data=data)
        raise NotImplementedError()

    def update_interface(self, device, data):
        log.debug('update-interface - Not implemented - We do not use this',
                  data=data)
        raise NotImplementedError()

    def remove_interface(self, device, data):
        log.debug('remove-interface - Not implemented - We do not use this',
                  data=data)
        raise NotImplementedError()

    def receive_onu_detect_state(self, proxy_address, state):
        log.debug('receive-onu-detect-state - Not implemented - We do not '
                  'use this', proxy_address=proxy_address,
                  state=state)
        raise NotImplementedError()

    def create_tcont(self, device, tcont_data, traffic_descriptor_data):
        log.info('create-tcont - Not implemented - We do not use this',
                 tcont_data=tcont_data,
                 traffic_descriptor_data=traffic_descriptor_data)
        raise NotImplementedError()

    def update_tcont(self, device, tcont_data, traffic_descriptor_data):
        log.info('update-tcont - Not implemented - We do not use this',
                 tcont_data=tcont_data,
                 traffic_descriptor_data=traffic_descriptor_data)
        raise NotImplementedError()

    def remove_tcont(self, device, tcont_data, traffic_descriptor_data):
        log.info('remove-tcont - Not implemented - We do not use this',
                 tcont_data=tcont_data,
                 traffic_descriptor_data=traffic_descriptor_data)
        raise NotImplementedError()

    def create_gemport(self, device, data):
        log.info('create-gemport - Not implemented - We do not use this',
                 data=data)
        raise NotImplementedError()

    def update_gemport(self, device, data):
        log.info('update-gemport - Not implemented - We do not use this',
                 data=data)
        raise NotImplementedError()

    def remove_gemport(self, device, data):
        log.info('remove-gemport - Not implemented - We do not use this',
                 data=data)
        raise NotImplementedError()

    def create_multicast_gemport(self, device, data):
        log.info('create-mcast-gemport  - Not implemented - We do not use '
                 'this', data=data)
        raise NotImplementedError()

    def update_multicast_gemport(self, device, data):
        log.info('update-mcast-gemport - Not implemented - We do not use '
                 'this', data=data)
        raise NotImplementedError()

    def remove_multicast_gemport(self, device, data):
        log.info('remove-mcast-gemport - Not implemented - We do not use '
                 'this', data=data)
        raise NotImplementedError()

    def create_multicast_distribution_set(self, device, data):
        log.info('create-mcast-distribution-set - Not implemented - We do '
                 'not use this', data=data)
        raise NotImplementedError()

    def update_multicast_distribution_set(self, device, data):
        log.info('update-mcast-distribution-set - Not implemented - We do '
                 'not use this', data=data)
        raise NotImplementedError()

    def remove_multicast_distribution_set(self, device, data):
        log.info('remove-mcast-distribution-set - Not implemented - We do '
                 'not use this', data=data)
        raise NotImplementedError()


    # Alarm Methods #
    def suppress_alarm(self, filter):
        log.info('suppress_alarm - Not implemented yet', filter=filter)
        raise NotImplementedError()

    def unsuppress_alarm(self, filter):
        log.info('unsuppress_alarm - Not implemented yet', filter=filter)
        raise NotImplementedError()

class OpenoltHandler(object):
    def __init__(self, adapter, device_id):
        self.adapter = adapter
        self.core_proxy = adapter.core_proxy
        self.adapter_proxy = adapter.adapter_proxy
        self.device_id = device_id
        self.log = structlog.get_logger(device_id=device_id)
        self.channel = None
        self.io_port = None
        self.logical_device_id = None
        self.nni_port = None
        self.ofp_port_no = None
        self.interface = registry('main').get_args().interface
        self.pm_metrics = None
        self.alarms = None
        self.frames = None
	self.num_devices = 0

    @inlineCallbacks
    def get_channel(self):
        if self.channel is None:
            try:
                device = yield self.core_proxy.get_device(self.device_id)
                self.log.info('device-info', device=device,
                              host_port=device.host_and_port)
                self.channel = grpc.insecure_channel(device.host_and_port)
            except Exception as e:
                log.exception("ponsim-connection-failure", e=e)

        # returnValue(self.channel)

    def close_channel(self):
        if self.channel is None:
            self.log.info('grpc-channel-already-closed')
            return
        else:
            if self.frames is not None:
                self.frames.cancel()
                self.frames = None
                self.log.info('cancelled-grpc-frame-stream')

            self.channel.unsubscribe(lambda *args: None)
            self.channel = None

            self.log.info('grpc-channel-closed')

    @inlineCallbacks
    def _get_nni_port(self):
        ports = yield self.core_proxy.get_ports(self.device_id,
                                                Port.ETHERNET_NNI)
        returnValue(ports)
    
    def init_device(self, kwargs):
	self.device = OpenoltDevice(**kwargs)

    @inlineCallbacks
    def activate(self, device):
        try:
            self.log.info('activating')
            if not device.host_and_port:
                device.oper_status = OperStatus.FAILED
                device.reason = 'No host_and_port field provided'
                self.core_proxy.device_update(device)
                return
	          
	    kwargs = {
            	'support_classes': OpenOltDefaults['support_classes'],
                'adapter_agent': self.core_proxy,
                'device': device,
                'device_num': self.num_devices + 1
            }
            try:
		yield self.init_device(kwargs)
            except Exception as e:
                log.error('Failed to adopt OpenOLT device', error=e)
                # TODO set status to ERROR so that is clear something went wrong
                #del self.devices[device.id]
                raise
            else:
                self.num_devices += 1

	    """
            yield self.get_channel()
            stub = PonSimStub(self.channel)
            info = stub.GetDeviceInfo(Empty())
            log.info('got-info', info=info, device_id=device.id)
            self.ofp_port_no = info.nni_port

            device.root = True
            device.vendor = 'ponsim'
            device.model = 'n/a'
            device.serial_number = device.host_and_port
            device.mac_address = "AA:BB:CC:DD:EE:FF"
            yield self.core_proxy.device_update(device)

            # Now set the initial PM configuration for this device
            self.pm_metrics = AdapterPmMetrics(device)
            pm_config = self.pm_metrics.make_proto()
            log.info("initial-pm-config", pm_config=pm_config)
            self.core_proxy.device_pm_config_update(pm_config, init=True)

            # Setup alarm handler
            self.alarms = AdapterAlarms(self.adapter, device)

            nni_port = Port(
                port_no=info.nni_port,
                label='NNI facing Ethernet port',
                type=Port.ETHERNET_NNI,
                oper_status=OperStatus.ACTIVE
            )
            self.nni_port = nni_port
            yield self.core_proxy.port_created(device.id, nni_port)
            yield self.core_proxy.port_created(device.id, Port(
                port_no=1,
                label='PON port',
                type=Port.PON_OLT,
                oper_status=OperStatus.ACTIVE
            ))

            yield self.core_proxy.device_state_update(device.id,
                                                      connect_status=ConnectStatus.REACHABLE,
                                                      oper_status=OperStatus.ACTIVE)

            # register ONUS
            self.log.info('onu-found', onus=info.onus, len=len(info.onus))
            for onu in info.onus:
                vlan_id = onu.uni_port
                yield self.core_proxy.child_device_detected(
                    parent_device_id=device.id,
                    parent_port_no=1,
                    child_device_type='ponsim_onu',
                    channel_id=vlan_id,
                )

            self.log.info('starting-frame-grpc-stream')
            reactor.callInThread(self.rcv_grpc)
            self.log.info('started-frame-grpc-stream')

            # Start collecting stats from the device after a brief pause
            self.start_kpi_collection(device.id)
	    """
        except Exception as e:
            log.exception("Exception-activating", e=e)

    def get_ofp_device_info(self, device):
        return SwitchCapability(
            desc=ofp_desc(
                hw_desc='ponsim pon',
                sw_desc='ponsim pon',
                serial_num=device.serial_number,
                dp_desc='n/a'
            ),
            switch_features=ofp_switch_features(
                n_buffers=256,  # TODO fake for now
                n_tables=2,  # TODO ditto
                capabilities=(  # TODO and ditto
                        OFPC_FLOW_STATS
                        | OFPC_TABLE_STATS
                        | OFPC_PORT_STATS
                        | OFPC_GROUP_STATS
                )
            )
        )

    def get_ofp_port_info(self, device, port_no):
        # Since the adapter created the device port then it has the reference of the port to
        # return the capability.   TODO:  Do a lookup on the NNI port number and return the
        # appropriate attributes
        self.log.info('get_ofp_port_info', port_no=port_no,
                      info=self.ofp_port_no, device_id=device.id)
        cap = OFPPF_1GB_FD | OFPPF_FIBER
        return PortCapability(
            port=LogicalPort(
                ofp_port=ofp_port(
                    hw_addr=mac_str_to_tuple(
                        'AA:BB:CC:DD:EE:%02x' % port_no),
                    config=0,
                    state=OFPPS_LIVE,
                    curr=cap,
                    advertised=cap,
                    peer=cap,
                    curr_speed=OFPPF_1GB_FD,
                    max_speed=OFPPF_1GB_FD
                ),
                device_id=device.id,
                device_port_no=port_no
            )
        )

    # TODO - change for core 2.0
    def reconcile(self, device):
        self.log.info('reconciling-OLT-device')

    @inlineCallbacks
    def _rcv_frame(self, frame):
        pkt = Ether(frame)

        if pkt.haslayer(Dot1Q):
            outer_shim = pkt.getlayer(Dot1Q)

            if isinstance(outer_shim.payload, Dot1Q):
                inner_shim = outer_shim.payload
                cvid = inner_shim.vlan
                popped_frame = (
                        Ether(src=pkt.src, dst=pkt.dst, type=inner_shim.type) /
                        inner_shim.payload
                )
                self.log.info('sending-packet-in',device_id=self.device_id, port=cvid)
                yield self.core_proxy.send_packet_in(device_id=self.device_id,
                                               port=cvid,
                                               packet=str(popped_frame))
            elif pkt.haslayer(Raw):
                raw_data = json.loads(pkt.getlayer(Raw).load)
                self.alarms.send_alarm(self, raw_data)

    @inlineCallbacks
    def rcv_grpc(self):
        """
        This call establishes a GRPC stream to receive frames.
        """
        yield self.get_channel()
        stub = PonSimStub(self.channel)
        # stub = PonSimStub(self.get_channel())

        # Attempt to establish a grpc stream with the remote ponsim service
        self.frames = stub.ReceiveFrames(Empty())

        self.log.info('start-receiving-grpc-frames')

        try:
            for frame in self.frames:
                self.log.info('received-grpc-frame',
                              frame_len=len(frame.payload))
                yield self._rcv_frame(frame.payload)

        except _Rendezvous, e:
            log.warn('grpc-connection-lost', message=e.message)

        self.log.info('stopped-receiving-grpc-frames')

    @inlineCallbacks
    def update_flow_table(self, flows):
        yield self.get_channel()
        stub = PonSimStub(self.channel)

        self.log.info('pushing-olt-flow-table')
        stub.UpdateFlowTable(FlowTable(
            port=0,
            flows=flows
        ))
        self.log.info('success')

    def remove_from_flow_table(self, flows):
        self.log.debug('remove-from-flow-table', flows=flows)
        # TODO: Update PONSIM code to accept incremental flow changes
        # Once completed, the accepts_add_remove_flow_updates for this
        # device type can be set to True

    def add_to_flow_table(self, flows):
        self.log.debug('add-to-flow-table', flows=flows)
        # TODO: Update PONSIM code to accept incremental flow changes
        # Once completed, the accepts_add_remove_flow_updates for this
        # device type can be set to True

    def update_pm_config(self, device, pm_config):
        log.info("handler-update-pm-config", device=device,
                 pm_config=pm_config)
        self.pm_metrics.update(pm_config)

    def send_proxied_message(self, proxy_address, msg):
        self.log.info('sending-proxied-message')
        if isinstance(msg, FlowTable):
            stub = PonSimStub(self.get_channel())
            self.log.info('pushing-onu-flow-table', port=msg.port)
            res = stub.UpdateFlowTable(msg)
            self.core_proxy.receive_proxied_message(proxy_address, res)

    @inlineCallbacks
    def process_inter_adapter_message(self, request):
        self.log.info('process-inter-adapter-message', msg=request)
        try:
            if request.header.type == InterAdapterMessageType.FLOW_REQUEST:
                f = FlowTable()
                if request.body:
                    request.body.Unpack(f)
                    stub = PonSimStub(self.channel)
                    self.log.info('pushing-onu-flow-table')
                    res = stub.UpdateFlowTable(f)
                    # Send response back
                    reply = InterAdapterResponseBody()
                    reply.status = True
                    self.log.info('sending-response-back', reply=reply)
                    yield self.adapter_proxy.send_inter_adapter_message(
                        msg=reply,
                        type=InterAdapterMessageType.FLOW_RESPONSE,
                        from_adapter=self.adapter.name,
                        to_adapter=request.header.from_topic,
                        to_device_id=request.header.to_device_id,
                        message_id=request.header.id
                    )
            elif request.header.type == InterAdapterMessageType.METRICS_REQUEST:
                m = PonSimMetricsRequest()
                if request.body:
                    request.body.Unpack(m)
                    stub = PonSimStub(self.channel)
                    self.log.info('proxying onu stats request', port=m.port)
                    res = stub.GetStats(m)
                    # Send response back
                    reply = InterAdapterResponseBody()
                    reply.status = True
                    reply.body.Pack(res)
                    self.log.info('sending-response-back', reply=reply)
                    yield self.adapter_proxy.send_inter_adapter_message(
                        msg=reply,
                        type=InterAdapterMessageType.METRICS_RESPONSE,
                        from_adapter=self.adapter.name,
                        to_adapter=request.header.from_topic,
                        to_device_id=request.header.to_device_id,
                        message_id=request.header.id
                    )
        except Exception as e:
            self.log.exception("error-processing-inter-adapter-message", e=e)

    def packet_out(self, egress_port, msg):
        self.log.info('sending-packet-out', egress_port=egress_port,
                      msg=hexify(msg))
        try:
            pkt = Ether(msg)
            out_pkt = pkt
            if egress_port != self.nni_port.port_no:
                # don't do the vlan manipulation for the NNI port, vlans are already correct
                out_pkt = (
                        Ether(src=pkt.src, dst=pkt.dst) /
                        Dot1Q(vlan=egress_port, type=pkt.type) /
                        pkt.payload
                )

            # TODO need better way of mapping logical ports to PON ports
            out_port = self.nni_port.port_no if egress_port == self.nni_port.port_no else 1

            # send over grpc stream
            stub = PonSimStub(self.channel)
            frame = PonSimFrame(id=self.device_id, payload=str(out_pkt),
                                out_port=out_port)
            stub.SendFrame(frame)
        except Exception as e:
            self.log.exception("error-processing-packet-out", e=e)


    @inlineCallbacks
    def reboot(self):
        self.log.info('rebooting', device_id=self.device_id)

        yield self.core_proxy.device_state_update(self.device_id,
                                                  connect_status=ConnectStatus.UNREACHABLE)

        # Update the child devices connect state to UNREACHABLE
        yield self.core_proxy.children_state_update(self.device_id,
                                                    connect_status=ConnectStatus.UNREACHABLE)

        # Sleep 10 secs, simulating a reboot
        # TODO: send alert and clear alert after the reboot
        yield asleep(10)

        # Change the connection status back to REACHABLE.  With a
        # real OLT the connection state must be the actual state
        yield self.core_proxy.device_state_update(self.device_id,
                                                  connect_status=ConnectStatus.REACHABLE)

        # Update the child devices connect state to REACHABLE
        yield self.core_proxy.children_state_update(self.device_id,
                                                    connect_status=ConnectStatus.REACHABLE)

        self.log.info('rebooted', device_id=self.device_id)

    def self_test_device(self, device):
        """
        This is called to Self a device based on a NBI call.
        :param device: A Voltha.Device object.
        :return: Will return result of self test
        """
        log.info('self-test-device', device=device.id)
        raise NotImplementedError()

    @inlineCallbacks
    def disable(self):
        self.log.info('disabling', device_id=self.device_id)

        self.stop_kpi_collection()

        # Update the operational status to UNKNOWN and connection status to UNREACHABLE
        yield self.core_proxy.device_state_update(self.device_id,
                                                  oper_status=OperStatus.UNKNOWN,
                                                  connect_status=ConnectStatus.UNREACHABLE)

        self.close_channel()
        self.log.info('disabled-grpc-channel')

        self.stop_kpi_collection()

        # TODO:
        # 1) Remove all flows from the device
        # 2) Remove the device from ponsim

        self.log.info('disabled', device_id=self.device_id)

    @inlineCallbacks
    def reenable(self):
        self.log.info('re-enabling', device_id=self.device_id)

        # Set the ofp_port_no and nni_port in case we bypassed the reconcile
        # process if the device was in DISABLED state on voltha restart
        if not self.ofp_port_no and not self.nni_port:
            yield self.get_channel()
            stub = PonSimStub(self.channel)
            info = stub.GetDeviceInfo(Empty())
            log.info('got-info', info=info)
            self.ofp_port_no = info.nni_port
            ports = yield self._get_nni_port()
            # For ponsim, we are using only 1 NNI port
            if ports.items:
                self.nni_port = ports.items[0]

        # Update the state of the NNI port
        yield self.core_proxy.port_state_update(self.device_id,
                                                port_type=Port.ETHERNET_NNI,
                                                port_no=self.ofp_port_no,
                                                oper_status=OperStatus.ACTIVE)

        # Update the state of the PON port
        yield self.core_proxy.port_state_update(self.device_id,
                                                port_type=Port.PON_OLT,
                                                port_no=1,
                                                oper_status=OperStatus.ACTIVE)

        # Set the operational state of the device to ACTIVE and connect status to REACHABLE
        yield self.core_proxy.device_state_update(self.device_id,
                                                  connect_status=ConnectStatus.REACHABLE,
                                                  oper_status=OperStatus.ACTIVE)

        # TODO: establish frame grpc-stream
        # yield reactor.callInThread(self.rcv_grpc)

        self.start_kpi_collection(self.device_id)

        self.log.info('re-enabled', device_id=self.device_id)

    def delete(self):
        self.log.info('deleting', device_id=self.device_id)

        self.close_channel()
        self.log.info('disabled-grpc-channel')

        # TODO:
        # 1) Remove all flows from the device
        # 2) Remove the device from ponsim

        self.log.info('deleted', device_id=self.device_id)

    def start_kpi_collection(self, device_id):

        kafka_cluster_proxy = get_kafka_proxy()

        def _collect(device_id, prefix):

            try:
                # Step 1: gather metrics from device
                port_metrics = \
                    self.pm_metrics.collect_port_metrics(self.channel)

                # Step 2: prepare the KpiEvent for submission
                # we can time-stamp them here (or could use time derived from OLT
                ts = arrow.utcnow().timestamp
                kpi_event = KpiEvent(
                    type=KpiEventType.slice,
                    ts=ts,
                    prefixes={
                        # OLT NNI port
                        prefix + '.nni': MetricValuePairs(
                            metrics=port_metrics['nni']),
                        # OLT PON port
                        prefix + '.pon': MetricValuePairs(
                            metrics=port_metrics['pon'])
                    }
                )

                # Step 3: submit directly to the kafka bus
                if kafka_cluster_proxy:
                    if isinstance(kpi_event, Message):
                        kpi_event = dumps(MessageToDict(kpi_event, True, True))
                    kafka_cluster_proxy.send_message("voltha.kpis", kpi_event)

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

        self.pm_metrics.start_collector(_collect)

    def stop_kpi_collection(self):
        self.pm_metrics.stop_collector()