python/adapters/brcm_openomci_onu/brcm_openomci_onu_handler.py

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

"""
Broadcom OpenOMCI OLT/ONU adapter handler.
"""

from __future__ import absolute_import
import six
import arrow
import structlog
import json

from collections import OrderedDict

from twisted.internet import reactor
from twisted.internet.defer import DeferredQueue, inlineCallbacks

from heartbeat import HeartBeat
from pyvoltha.adapters.extensions.events.device_events.onu.onu_active_event import OnuActiveEvent
from pyvoltha.adapters.extensions.events.kpi.onu.onu_pm_metrics import OnuPmMetrics
from pyvoltha.adapters.extensions.events.kpi.onu.onu_omci_pm import OnuOmciPmMetrics
from pyvoltha.adapters.extensions.events.adapter_events import AdapterEvents

import pyvoltha.common.openflow.utils as fd
from pyvoltha.common.utils.registry import registry
from pyvoltha.common.utils.nethelpers import mac_str_to_tuple
from pyvoltha.common.config.config_backend import ConsulStore
from pyvoltha.common.config.config_backend import EtcdStore
from voltha_protos.logical_device_pb2 import LogicalPort
from voltha_protos.common_pb2 import OperStatus, ConnectStatus, AdminState
from voltha_protos.openflow_13_pb2 import OFPXMC_OPENFLOW_BASIC, ofp_port, OFPPS_LIVE, OFPPF_FIBER, OFPPF_1GB_FD
from voltha_protos.inter_container_pb2 import InterAdapterMessageType, \
    InterAdapterOmciMessage, PortCapability, InterAdapterTechProfileDownloadMessage
from voltha_protos.openolt_pb2 import OnuIndication
from pyvoltha.adapters.extensions.omci.onu_configuration import OMCCVersion
from pyvoltha.adapters.extensions.omci.onu_device_entry import OnuDeviceEvents, \
    OnuDeviceEntry, IN_SYNC_KEY
from omci.brcm_mib_download_task import BrcmMibDownloadTask
from omci.brcm_tp_service_specific_task import BrcmTpServiceSpecificTask
from omci.brcm_uni_lock_task import BrcmUniLockTask
from omci.brcm_vlan_filter_task import BrcmVlanFilterTask
from onu_gem_port import OnuGemPort
from onu_tcont import OnuTCont
from pon_port import PonPort
from uni_port import UniPort, UniType
from onu_traffic_descriptor import OnuTrafficDescriptor
from pyvoltha.common.tech_profile.tech_profile import TechProfile
from pyvoltha.adapters.extensions.omci.tasks.omci_test_request import OmciTestRequest
from pyvoltha.adapters.extensions.omci.omci_entities import AniG
from pyvoltha.adapters.extensions.omci.omci_defs import EntityOperations, ReasonCodes

OP = EntityOperations
RC = ReasonCodes

log = structlog.get_logger()

_STARTUP_RETRY_WAIT = 20


class BrcmOpenomciOnuHandler(object):

    def __init__(self, adapter, device_id):
        self.log = structlog.get_logger(device_id=device_id)
        self.log.debug('function-entry')
        self.adapter = adapter
        self.core_proxy = adapter.core_proxy
        self.adapter_proxy = adapter.adapter_proxy
        self.parent_adapter = None
        self.parent_id = None
        self.device_id = device_id
        self.incoming_messages = DeferredQueue()
        self.event_messages = DeferredQueue()
        self.proxy_address = None
        self.tx_id = 0
        self._enabled = False
        self.events = None
        self.pm_metrics = None
        self._omcc_version = OMCCVersion.Unknown
        self._total_tcont_count = 0  # From ANI-G ME
        self._qos_flexibility = 0  # From ONT2_G ME

        self._onu_indication = None
        self._unis = dict()  # Port # -> UniPort

        self._pon = None
        # TODO: probably shouldnt be hardcoded, determine from olt maybe?
        self._pon_port_number = 100
        self.logical_device_id = None

        self._heartbeat = HeartBeat.create(self, device_id)

        # Set up OpenOMCI environment
        self._onu_omci_device = None
        self._dev_info_loaded = False
        self._deferred = None

        self._in_sync_subscription = None
        self._connectivity_subscription = None
        self._capabilities_subscription = None

        self.mac_bridge_service_profile_entity_id = 0x201
        self.gal_enet_profile_entity_id = 0x1

        self._tp_service_specific_task = dict()
        self._tech_profile_download_done = dict()
        # Stores information related to queued vlan filter tasks
        # Dictionary with key being uni_id and value being device,uni port ,uni id and vlan id

        self._queued_vlan_filter_task = dict()

        # Initialize KV store client
        self.args = registry('main').get_args()
        if self.args.backend == 'etcd':
            host, port = self.args.etcd.split(':', 1)
            self.kv_client = EtcdStore(host, port,
                                       TechProfile.KV_STORE_TECH_PROFILE_PATH_PREFIX)
        elif self.args.backend == 'consul':
            host, port = self.args.consul.split(':', 1)
            self.kv_client = ConsulStore(host, port,
                                         TechProfile.KV_STORE_TECH_PROFILE_PATH_PREFIX)
        else:
            self.log.error('Invalid-backend')
            raise Exception("Invalid-backend-for-kv-store")

    @property
    def enabled(self):
        return self._enabled

    @enabled.setter
    def enabled(self, value):
        if self._enabled != value:
            self._enabled = value

    @property
    def omci_agent(self):
        return self.adapter.omci_agent

    @property
    def omci_cc(self):
        return self._onu_omci_device.omci_cc if self._onu_omci_device is not None else None

    @property
    def heartbeat(self):
        return self._heartbeat

    @property
    def uni_ports(self):
        return list(self._unis.values())

    def uni_port(self, port_no_or_name):
        if isinstance(port_no_or_name, six.string_types):
            return next((uni for uni in self.uni_ports
                         if uni.name == port_no_or_name), None)

        assert isinstance(port_no_or_name, int), 'Invalid parameter type'
        return next((uni for uni in self.uni_ports
                    if uni.port_number == port_no_or_name), None)

    @property
    def pon_port(self):
        return self._pon

    def receive_message(self, msg):
        if self.omci_cc is not None:
            self.omci_cc.receive_message(msg)

    def get_ofp_port_info(self, device, port_no):
        self.log.info('get_ofp_port_info', port_no=port_no, device_id=device.id)
        cap = OFPPF_1GB_FD | OFPPF_FIBER

        hw_addr=mac_str_to_tuple('08:%02x:%02x:%02x:%02x:%02x' %
                                ((device.parent_port_no >> 8 & 0xff),
                                  device.parent_port_no & 0xff,
                                  (port_no >> 16) & 0xff,
                                  (port_no >> 8) & 0xff,
                                   port_no & 0xff))

        uni_port = self.uni_port(int(port_no))
        name = device.serial_number + '-' + str(uni_port.mac_bridge_port_num)
        self.log.debug('ofp_port_name', port_no=port_no, name=name)

        return PortCapability(
            port=LogicalPort(
                ofp_port=ofp_port(
                    name=name,
                    hw_addr=hw_addr,
                    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
            )
        )

    # Called once when the adapter creates the device/onu instance
    @inlineCallbacks
    def activate(self, device):
        self.log.debug('function-entry', device=device)

        assert device.parent_id
        assert device.parent_port_no
        assert device.proxy_address.device_id

        self.proxy_address = device.proxy_address
        self.parent_id = device.parent_id
        self._pon_port_number = device.parent_port_no
        if self.enabled is not True:
            self.log.info('activating-new-onu')
            # populate what we know.  rest comes later after mib sync
            device.root = False
            device.vendor = 'OpenONU'
            device.reason = 'activating-onu'

            # TODO NEW CORE:  Need to either get logical device id from core or use regular device id
            # pm_metrics requires a logical device id.  For now set to just device_id
            self.logical_device_id = self.device_id

            yield self.core_proxy.device_update(device)
            self.log.debug('device updated', device=device)

            yield self._init_pon_state()

            self.log.debug('pon state initialized', device=device)
            ############################################################################
            # Setup Alarm handler
            self.events = AdapterEvents(self.core_proxy, device.id, self.logical_device_id,
                                        device.serial_number)
            ############################################################################
            # Setup PM configuration for this device
            # Pass in ONU specific options
            kwargs = {
                OnuPmMetrics.DEFAULT_FREQUENCY_KEY: OnuPmMetrics.DEFAULT_ONU_COLLECTION_FREQUENCY,
                'heartbeat': self.heartbeat,
                OnuOmciPmMetrics.OMCI_DEV_KEY: self._onu_omci_device
            }
            self.log.debug('create-OnuPmMetrics', serial_number=device.serial_number)
            self.pm_metrics = OnuPmMetrics(self.events, self.core_proxy, self.device_id,
                                           self.logical_device_id, device.serial_number,
                                           grouped=True, freq_override=False, **kwargs)
            pm_config = self.pm_metrics.make_proto()
            self._onu_omci_device.set_pm_config(self.pm_metrics.omci_pm.openomci_interval_pm)
            self.log.info("initial-pm-config", pm_config=pm_config)
            yield self.core_proxy.device_pm_config_update(pm_config, init=True)

            # Note, ONU ID and UNI intf set in add_uni_port method
            self._onu_omci_device.alarm_synchronizer.set_alarm_params(mgr=self.events,
                                                                      ani_ports=[self._pon])

            #Start collecting stats from the device after a brief pause
            reactor.callLater(10, self.pm_metrics.start_collector)

            # Code to Run OMCI Test Action
            kwargs_omci_test_action = {
                OmciTestRequest.DEFAULT_FREQUENCY_KEY:
                    OmciTestRequest.DEFAULT_COLLECTION_FREQUENCY
            }
            serial_number = device.serial_number
            test_request = OmciTestRequest(self.core_proxy,
                                           self.omci_agent, self.device_id,
                                           AniG, serial_number,
                                           self.logical_device_id,
                                           exclusive=False,
                                           **kwargs_omci_test_action)
            reactor.callLater(60, test_request.start_collector)
            self.enabled = True
        else:
            self.log.info('onu-already-activated')

    # Called once when the adapter needs to re-create device.  usually on vcore restart
    @inlineCallbacks
    def reconcile(self, device):
        self.log.debug('function-entry', device=device)

        # first we verify that we got parent reference and proxy info
        assert device.parent_id
        assert device.proxy_address.device_id

        self.proxy_address = device.proxy_address
        self.parent_id = device.parent_id
        self._pon_port_number = device.parent_port_no

        if self.enabled is not True:
            self.log.info('reconciling-broadcom-onu-device')
            self.logical_device_id = self.device_id
            self._init_pon_state()

            # need to restart state machines on vcore restart.  there is no indication to do it for us.
            self._onu_omci_device.start()
            yield self.core_proxy.device_reason_update(self.device_id, "restarting-openomci")

            # TODO: this is probably a bit heavy handed
            # Force a reboot for now.  We need indications to reflow to reassign tconts and gems given vcore went away
            # This may not be necessary when mib resync actually works
            reactor.callLater(1, self.reboot)

            self.enabled = True
        else:
            self.log.info('onu-already-activated')

    @inlineCallbacks
    def _init_pon_state(self):
        self.log.debug('function-entry', deviceId=self.device_id)

        self._pon = PonPort.create(self, self._pon_port_number)
        self._pon.add_peer(self.parent_id, self._pon_port_number)
        self.log.debug('adding-pon-port-to-agent', pon=self._pon.get_port())

        yield self.core_proxy.port_created(self.device_id, self._pon.get_port())

        self.log.debug('added-pon-port-to-agent', pon=self._pon.get_port())

        # Create and start the OpenOMCI ONU Device Entry for this ONU
        self._onu_omci_device = self.omci_agent.add_device(self.device_id,
                                                           self.core_proxy,
                                                           self.adapter_proxy,
                                                           support_classes=self.adapter.broadcom_omci,
                                                           custom_me_map=self.adapter.custom_me_entities())
        # Port startup
        if self._pon is not None:
            self._pon.enabled = True

    def delete(self, device):
        self.log.info('delete-onu', device=device)
        if self.parent_adapter:
            try:
                self.parent_adapter.delete_child_device(self.parent_id, device)
            except AttributeError:
                self.log.debug('parent-device-delete-child-not-implemented')
        else:
            self.log.debug("parent-adapter-not-available")

    def _create_tconts(self, uni_id, us_scheduler):
        alloc_id = us_scheduler['alloc_id']
        q_sched_policy = us_scheduler['q_sched_policy']
        self.log.debug('create-tcont', us_scheduler=us_scheduler)

        tcontdict = dict()
        tcontdict['alloc-id'] = alloc_id
        tcontdict['q_sched_policy'] = q_sched_policy
        tcontdict['uni_id'] = uni_id

        # TODO: Not sure what to do with any of this...
        tddata = dict()
        tddata['name'] = 'not-sure-td-profile'
        tddata['fixed-bandwidth'] = "not-sure-fixed"
        tddata['assured-bandwidth'] = "not-sure-assured"
        tddata['maximum-bandwidth'] = "not-sure-max"
        tddata['additional-bw-eligibility-indicator'] = "not-sure-additional"

        td = OnuTrafficDescriptor.create(tddata)
        tcont = OnuTCont.create(self, tcont=tcontdict, td=td)

        self._pon.add_tcont(tcont)

        self.log.debug('pon-add-tcont', tcont=tcont)

    # Called when there is an olt up indication, providing the gem port id chosen by the olt handler
    def _create_gemports(self, uni_id, gem_ports, alloc_id_ref, direction):
        self.log.debug('create-gemport',
                       gem_ports=gem_ports, direction=direction)

        for gem_port in gem_ports:
            gemdict = dict()
            gemdict['gemport_id'] = gem_port['gemport_id']
            gemdict['direction'] = direction
            gemdict['alloc_id_ref'] = alloc_id_ref
            gemdict['encryption'] = gem_port['aes_encryption']
            gemdict['discard_config'] = dict()
            gemdict['discard_config']['max_probability'] = \
                gem_port['discard_config']['max_probability']
            gemdict['discard_config']['max_threshold'] = \
                gem_port['discard_config']['max_threshold']
            gemdict['discard_config']['min_threshold'] = \
                gem_port['discard_config']['min_threshold']
            gemdict['discard_policy'] = gem_port['discard_policy']
            gemdict['max_q_size'] = gem_port['max_q_size']
            gemdict['pbit_map'] = gem_port['pbit_map']
            gemdict['priority_q'] = gem_port['priority_q']
            gemdict['scheduling_policy'] = gem_port['scheduling_policy']
            gemdict['weight'] = gem_port['weight']
            gemdict['uni_id'] = uni_id

            gem_port = OnuGemPort.create(self, gem_port=gemdict)

            self._pon.add_gem_port(gem_port)

            self.log.debug('pon-add-gemport', gem_port=gem_port)

    def _execute_queued_vlan_filter_tasks(self, uni_id):
        # During OLT Reboots, ONU Reboots, ONU Disable/Enable, it is seen that vlan_filter
        # task is scheduled even before tp task. So we queue vlan-filter task if tp_task
        # or initial-mib-download is not done. Once the tp_task is completed, we execute
        # such queued vlan-filter tasks
        try:
            if uni_id in self._queued_vlan_filter_task:
                self.log.info("executing-queued-vlan-filter-task",
                              uni_id=uni_id)
                filter_info = self._queued_vlan_filter_task[uni_id]
                reactor.callLater(0, self._add_vlan_filter_task, filter_info.get("device"),
                                  uni_id, filter_info.get("uni_port"), filter_info.get("set_vlan_vid"))
                # Now remove the entry from the dictionary
                self._queued_vlan_filter_task[uni_id].clear()
                self.log.debug("executed-queued-vlan-filter-task",
                               uni_id=uni_id)
        except Exception as e:
            self.log.error("vlan-filter-configuration-failed", uni_id=uni_id, error=e)


    def _do_tech_profile_configuration(self, uni_id, tp):
        us_scheduler = tp['us_scheduler']
        alloc_id = us_scheduler['alloc_id']
        self._create_tconts(uni_id, us_scheduler)
        upstream_gem_port_attribute_list = tp['upstream_gem_port_attribute_list']
        self._create_gemports(uni_id, upstream_gem_port_attribute_list, alloc_id, "UPSTREAM")
        downstream_gem_port_attribute_list = tp['downstream_gem_port_attribute_list']
        self._create_gemports(uni_id, downstream_gem_port_attribute_list, alloc_id, "DOWNSTREAM")

    def load_and_configure_tech_profile(self, uni_id, tp_path):
        self.log.debug("loading-tech-profile-configuration", uni_id=uni_id, tp_path=tp_path)

        if uni_id not in self._tp_service_specific_task:
            self._tp_service_specific_task[uni_id] = dict()

        if uni_id not in self._tech_profile_download_done:
            self._tech_profile_download_done[uni_id] = dict()

        if tp_path not in self._tech_profile_download_done[uni_id]:
            self._tech_profile_download_done[uni_id][tp_path] = False

        if not self._tech_profile_download_done[uni_id][tp_path]:
            try:
                if tp_path in self._tp_service_specific_task[uni_id]:
                    self.log.info("tech-profile-config-already-in-progress",
                                   tp_path=tp_path)
                    return

                tpstored = self.kv_client[tp_path]
                tpstring = tpstored.decode('ascii')
                tp = json.loads(tpstring)
                self.log.debug("tp-instance", tp=tp)
                self._do_tech_profile_configuration(uni_id, tp)

                @inlineCallbacks
                def success(_results):
                    self.log.info("tech-profile-config-done-successfully")
                    yield self.core_proxy.device_reason_update(self.device_id, 'tech-profile-config-download-success')
                    if tp_path in self._tp_service_specific_task[uni_id]:
                        del self._tp_service_specific_task[uni_id][tp_path]
                    self._tech_profile_download_done[uni_id][tp_path] = True
                    # Now execute any vlan filter tasks that were queued for later
                    self._execute_queued_vlan_filter_tasks(uni_id)
                @inlineCallbacks
                def failure(_reason):
                    self.log.warn('tech-profile-config-failure-retrying',
                                   _reason=_reason)
                    yield self.core_proxy.device_reason_update(self.device_id, 'tech-profile-config-download-failure-retrying')
                    if tp_path in self._tp_service_specific_task[uni_id]:
                        del self._tp_service_specific_task[uni_id][tp_path]
                    self._deferred = reactor.callLater(_STARTUP_RETRY_WAIT, self.load_and_configure_tech_profile,
                                                       uni_id, tp_path)

                self.log.info('downloading-tech-profile-configuration')
                self._tp_service_specific_task[uni_id][tp_path] = \
                       BrcmTpServiceSpecificTask(self.omci_agent, self, uni_id)
                self._deferred = \
                       self._onu_omci_device.task_runner.queue_task(self._tp_service_specific_task[uni_id][tp_path])
                self._deferred.addCallbacks(success, failure)

            except Exception as e:
                self.log.exception("error-loading-tech-profile", e=e)
        else:
            self.log.info("tech-profile-config-already-done")

    def update_pm_config(self, device, pm_config):
        # TODO: This has not been tested
        self.log.info('update_pm_config', pm_config=pm_config)
        self.pm_metrics.update(pm_config)

    # Calling this assumes the onu is active/ready and had at least an initial mib downloaded.   This gets called from
    # flow decomposition that ultimately comes from onos
    def update_flow_table(self, device, flows):
        self.log.debug('function-entry', device=device, flows=flows)

        #
        # We need to proxy through the OLT to get to the ONU
        # Configuration from here should be using OMCI
        #
        # self.log.info('bulk-flow-update', device_id=device.id, flows=flows)

        # no point in pushing omci flows if the device isnt reachable
        if device.connect_status != ConnectStatus.REACHABLE or \
           device.admin_state != AdminState.ENABLED:
            self.log.warn("device-disabled-or-offline-skipping-flow-update",
                          admin=device.admin_state, connect=device.connect_status)
            return

        def is_downstream(port):
            return port == self._pon_port_number

        def is_upstream(port):
            return not is_downstream(port)

        for flow in flows:
            _type = None
            _port = None
            _vlan_vid = None
            _udp_dst = None
            _udp_src = None
            _ipv4_dst = None
            _ipv4_src = None
            _metadata = None
            _output = None
            _push_tpid = None
            _field = None
            _set_vlan_vid = None
            _tunnel_id = None

            self.log.debug('bulk-flow-update', device_id=device.id, flow=flow)
            try:
                _in_port = fd.get_in_port(flow)
                assert _in_port is not None

                _out_port = fd.get_out_port(flow)  # may be None

                if is_downstream(_in_port):
                    self.log.debug('downstream-flow', in_port=_in_port, out_port=_out_port)
                    uni_port = self.uni_port(_out_port)
                    uni_id = _out_port  & 0xF
                elif is_upstream(_in_port):
                    self.log.debug('upstream-flow', in_port=_in_port, out_port=_out_port)
                    uni_port = self.uni_port(_in_port)
                    uni_id = _in_port & 0xF
                else:
                    raise Exception('port should be 1 or 2 by our convention')

                self.log.debug('flow-ports', in_port=_in_port, out_port=_out_port, uni_port=str(uni_port))

                for field in fd.get_ofb_fields(flow):
                    if field.type == fd.ETH_TYPE:
                        _type = field.eth_type
                        self.log.debug('field-type-eth-type',
                                       eth_type=_type)

                    elif field.type == fd.IP_PROTO:
                        _proto = field.ip_proto
                        self.log.debug('field-type-ip-proto',
                                       ip_proto=_proto)

                    elif field.type == fd.IN_PORT:
                        _port = field.port
                        self.log.debug('field-type-in-port',
                                       in_port=_port)

                    elif field.type == fd.VLAN_VID:
                        _vlan_vid = field.vlan_vid & 0xfff
                        self.log.debug('field-type-vlan-vid',
                                       vlan=_vlan_vid)

                    elif field.type == fd.VLAN_PCP:
                        _vlan_pcp = field.vlan_pcp
                        self.log.debug('field-type-vlan-pcp',
                                       pcp=_vlan_pcp)

                    elif field.type == fd.UDP_DST:
                        _udp_dst = field.udp_dst
                        self.log.debug('field-type-udp-dst',
                                       udp_dst=_udp_dst)

                    elif field.type == fd.UDP_SRC:
                        _udp_src = field.udp_src
                        self.log.debug('field-type-udp-src',
                                       udp_src=_udp_src)

                    elif field.type == fd.IPV4_DST:
                        _ipv4_dst = field.ipv4_dst
                        self.log.debug('field-type-ipv4-dst',
                                       ipv4_dst=_ipv4_dst)

                    elif field.type == fd.IPV4_SRC:
                        _ipv4_src = field.ipv4_src
                        self.log.debug('field-type-ipv4-src',
                                       ipv4_dst=_ipv4_src)

                    elif field.type == fd.METADATA:
                        _metadata = field.table_metadata
                        self.log.debug('field-type-metadata',
                                       metadata=_metadata)

                    elif field.type == fd.TUNNEL_ID:
                        _tunnel_id = field.tunnel_id
                        self.log.debug('field-type-tunnel-id',
                                       tunnel_id=_tunnel_id)

                    else:
                        raise NotImplementedError('field.type={}'.format(
                            field.type))

                for action in fd.get_actions(flow):

                    if action.type == fd.OUTPUT:
                        _output = action.output.port
                        self.log.debug('action-type-output',
                                       output=_output, in_port=_in_port)

                    elif action.type == fd.POP_VLAN:
                        self.log.debug('action-type-pop-vlan',
                                       in_port=_in_port)

                    elif action.type == fd.PUSH_VLAN:
                        _push_tpid = action.push.ethertype
                        self.log.debug('action-type-push-vlan',
                                       push_tpid=_push_tpid, in_port=_in_port)
                        if action.push.ethertype != 0x8100:
                            self.log.error('unhandled-tpid',
                                           ethertype=action.push.ethertype)

                    elif action.type == fd.SET_FIELD:
                        _field = action.set_field.field.ofb_field
                        assert (action.set_field.field.oxm_class ==
                                OFPXMC_OPENFLOW_BASIC)
                        self.log.debug('action-type-set-field',
                                       field=_field, in_port=_in_port)
                        if _field.type == fd.VLAN_VID:
                            _set_vlan_vid = _field.vlan_vid & 0xfff
                            self.log.debug('set-field-type-vlan-vid',
                                           vlan_vid=_set_vlan_vid)
                        else:
                            self.log.error('unsupported-action-set-field-type',
                                           field_type=_field.type)
                    else:
                        self.log.error('unsupported-action-type',
                                       action_type=action.type, in_port=_in_port)

                # OMCI set vlan task can only filter and set on vlan header attributes.  Any other openflow
                # supported match and action criteria cannot be handled by omci and must be ignored.
                if _set_vlan_vid is None or _set_vlan_vid == 0:
                    self.log.warn('ignoring-flow-that-does-not-set-vlanid')
                else:
                    self.log.info('set-vlanid', uni_id=uni_id, uni_port=uni_port, set_vlan_vid=_set_vlan_vid)
                    self._add_vlan_filter_task(device, uni_id, uni_port, _set_vlan_vid)
            except Exception as e:
                self.log.exception('failed-to-install-flow', e=e, flow=flow)



    def _add_vlan_filter_task(self, device,uni_id, uni_port, _set_vlan_vid):
        assert uni_port is not None
        if uni_id in self._tech_profile_download_done and self._tech_profile_download_done[uni_id] != {}:
            @inlineCallbacks
            def success(_results):
                self.log.info('vlan-tagging-success', uni_port=uni_port, vlan=_set_vlan_vid)
                yield self.core_proxy.device_reason_update(self.device_id, 'omci-flows-pushed')
                self._vlan_filter_task = None

            @inlineCallbacks
            def failure(_reason):
                self.log.warn('vlan-tagging-failure', uni_port=uni_port, vlan=_set_vlan_vid)
                yield self.core_proxy.device_reason_update(self.device_id, 'omci-flows-failed-retrying')
                self._vlan_filter_task = reactor.callLater(_STARTUP_RETRY_WAIT,
                                                       self._add_vlan_filter_task, device,uni_port.port_number, uni_port, _set_vlan_vid)

            self.log.info('setting-vlan-tag')
            self._vlan_filter_task = BrcmVlanFilterTask(self.omci_agent, self, uni_port, _set_vlan_vid)
            self._deferred = self._onu_omci_device.task_runner.queue_task(self._vlan_filter_task)
            self._deferred.addCallbacks(success, failure)
        else:
            self.log.info('tp-service-specific-task-not-done-adding-request-to-local-cache',
                          uni_id=uni_id)
            self._queued_vlan_filter_task[uni_id] = {"device": device,
                                                     "uni_id":uni_id,
                                                     "uni_port": uni_port,
                                                     "set_vlan_vid": _set_vlan_vid}

    def get_tx_id(self):
        self.log.debug('function-entry')
        self.tx_id += 1
        return self.tx_id

    def process_inter_adapter_message(self, request):
        self.log.debug('process-inter-adapter-message', msg=request)
        try:
            if request.header.type == InterAdapterMessageType.OMCI_REQUEST:
                omci_msg = InterAdapterOmciMessage()
                request.body.Unpack(omci_msg)
                self.log.debug('inter-adapter-recv-omci', omci_msg=omci_msg)

                self.receive_message(omci_msg.message)

            elif request.header.type == InterAdapterMessageType.ONU_IND_REQUEST:
                onu_indication = OnuIndication()
                request.body.Unpack(onu_indication)
                self.log.debug('inter-adapter-recv-onu-ind', onu_indication=onu_indication)

                if onu_indication.oper_state == "up":
                    self.create_interface(onu_indication)
                elif onu_indication.oper_state == "down" or onu_indication.oper_state=="unreachable":
                    self.update_interface(onu_indication)
                else:
                    self.log.error("unknown-onu-indication", onu_indication=onu_indication)

            elif request.header.type == InterAdapterMessageType.TECH_PROFILE_DOWNLOAD_REQUEST:
                tech_msg = InterAdapterTechProfileDownloadMessage()
                request.body.Unpack(tech_msg)
                self.log.debug('inter-adapter-recv-tech-profile', tech_msg=tech_msg)

                self.load_and_configure_tech_profile(tech_msg.uni_id, tech_msg.path)

            else:
                self.log.error("inter-adapter-unhandled-type", request=request)

        except Exception as e:
            self.log.exception("error-processing-inter-adapter-message", e=e)

    # Called each time there is an onu "up" indication from the olt handler
    @inlineCallbacks
    def create_interface(self, onu_indication):
        self.log.debug('function-entry', onu_indication=onu_indication)
        self._onu_indication = onu_indication

        yield self.core_proxy.device_state_update(self.device_id, oper_status=OperStatus.ACTIVATING,
                                                  connect_status=ConnectStatus.REACHABLE)

        onu_device = yield self.core_proxy.get_device(self.device_id)

        self.log.debug('starting-openomci-statemachine')
        self._subscribe_to_events()
        onu_device.reason = "starting-openomci"
        reactor.callLater(1, self._onu_omci_device.start,onu_device)
        yield self.core_proxy.device_reason_update(self.device_id, onu_device.reason)
        self._heartbeat.enabled = True

    # Currently called each time there is an onu "down" indication from the olt handler
    # TODO: possibly other reasons to "update" from the olt?
    @inlineCallbacks
    def update_interface(self, onu_indication):
        self.log.debug('function-entry', onu_indication=onu_indication)

        if onu_indication.oper_state == 'down' or onu_indication.oper_state == "unreachable":
            self.log.debug('stopping-openomci-statemachine')
            reactor.callLater(0, self._onu_omci_device.stop)

            # Let TP download happen again
            for uni_id in self._tp_service_specific_task:
                self._tp_service_specific_task[uni_id].clear()
            for uni_id in self._tech_profile_download_done:
                self._tech_profile_download_done[uni_id].clear()

            self.disable_ports()
            yield self.core_proxy.device_reason_update(self.device_id, "stopping-openomci")
            yield self.core_proxy.device_state_update(self.device_id, oper_status=OperStatus.DISCOVERED,
                                                      connect_status=ConnectStatus.UNREACHABLE)
        else:
            self.log.debug('not-changing-openomci-statemachine')

    # Not currently called by olt or anything else
    @inlineCallbacks
    def remove_interface(self, data):
        self.log.debug('function-entry', data=data)

        self.log.debug('stopping-openomci-statemachine')
        reactor.callLater(0, self._onu_omci_device.stop)

        # Let TP download happen again
        for uni_id in self._tp_service_specific_task:
            self._tp_service_specific_task[uni_id].clear()
        for uni_id in self._tech_profile_download_done:
            self._tech_profile_download_done[uni_id].clear()

        self.disable_ports()
        yield self.core_proxy.device_reason_update(self.device_id, "stopping-openomci")

        # TODO: im sure there is more to do here

    # Not currently called.  Would be called presumably from the olt handler
    @inlineCallbacks
    def remove_gemport(self, data):
        self.log.debug('remove-gemport', data=data)
        device = yield self.core_proxy.get_device(self.device_id)
        if device.connect_status != ConnectStatus.REACHABLE:
            self.log.error('device-unreachable')
            return

    # Not currently called.  Would be called presumably from the olt handler
    @inlineCallbacks
    def remove_tcont(self, tcont_data, traffic_descriptor_data):
        self.log.debug('remove-tcont', tcont_data=tcont_data, traffic_descriptor_data=traffic_descriptor_data)
        device = yield self.core_proxy.get_device(self.device_id)
        if device.connect_status != ConnectStatus.REACHABLE:
            self.log.error('device-unreachable')
            return

        # TODO: Create some omci task that encompases this what intended

    # Not currently called.  Would be called presumably from the olt handler
    def create_multicast_gemport(self, data):
        self.log.debug('function-entry', data=data)

        # TODO: create objects and populate for later omci calls

    def disable(self, device):
        self.log.debug('function-entry', device=device)
        try:
            self.log.info('sending-uni-lock-towards-device', device=device)

            @inlineCallbacks
            def stop_anyway(reason):
                # proceed with disable regardless if we could reach the onu. for example onu is unplugged
                self.log.debug('stopping-openomci-statemachine')
                reactor.callLater(0, self._onu_omci_device.stop)

                # Let TP download happen again
                for uni_id in self._tp_service_specific_task:
                    self._tp_service_specific_task[uni_id].clear()
                for uni_id in self._tech_profile_download_done:
                    self._tech_profile_download_done[uni_id].clear()

                self.disable_ports()
                device.oper_status = OperStatus.UNKNOWN
                device.reason = "omci-admin-lock"
                yield self.core_proxy.device_update(device)

            # lock all the unis
            task = BrcmUniLockTask(self.omci_agent, self.device_id, lock=True)
            self._deferred = self._onu_omci_device.task_runner.queue_task(task)
            self._deferred.addCallbacks(stop_anyway, stop_anyway)
        except Exception as e:
            log.exception('exception-in-onu-disable', exception=e)

    @inlineCallbacks
    def reenable(self, device):
        self.log.debug('function-entry', device=device)
        try:
            # Start up OpenOMCI state machines for this device
            # this will ultimately resync mib and unlock unis on successful redownloading the mib
            self.log.debug('restarting-openomci-statemachine')
            self._subscribe_to_events()
            yield self.core_proxy.device_reason_update(self.device_id, "restarting-openomci")
            reactor.callLater(1, self._onu_omci_device.start, device)
            self._heartbeat.enabled = True
        except Exception as e:
            log.exception('exception-in-onu-reenable', exception=e)

    @inlineCallbacks
    def reboot(self):
        self.log.info('reboot-device')
        device = yield self.core_proxy.get_device(self.device_id)
        if device.connect_status != ConnectStatus.REACHABLE:
            self.log.error("device-unreachable")
            return

        @inlineCallbacks
        def success(_results):
            self.log.info('reboot-success', _results=_results)
            self.disable_ports()
            device.connect_status = ConnectStatus.UNREACHABLE
            device.oper_status = OperStatus.DISCOVERED
            device.reason = "rebooting"
            yield self.core_proxy.device_update(device)

        def failure(_reason):
            self.log.info('reboot-failure', _reason=_reason)

        self._deferred = self._onu_omci_device.reboot()
        self._deferred.addCallbacks(success, failure)

    @inlineCallbacks
    def disable_ports(self):
        self.log.info('disable-ports', device_id=self.device_id)

        # Disable all ports on that device
        yield self.core_proxy.ports_state_update(self.device_id, OperStatus.UNKNOWN)

    @inlineCallbacks
    def enable_ports(self):
        self.log.info('enable-ports', device_id=self.device_id)

        # Enable all ports on that device
        yield self.core_proxy.ports_state_update(self.device_id, OperStatus.ACTIVE)

    # Called just before openomci state machine is started.  These listen for events from selected state machines,
    # most importantly, mib in sync.  Which ultimately leads to downloading the mib
    def _subscribe_to_events(self):
        self.log.debug('function-entry')

        # OMCI MIB Database sync status
        bus = self._onu_omci_device.event_bus
        topic = OnuDeviceEntry.event_bus_topic(self.device_id,
                                               OnuDeviceEvents.MibDatabaseSyncEvent)
        self._in_sync_subscription = bus.subscribe(topic, self.in_sync_handler)

        # OMCI Capabilities
        bus = self._onu_omci_device.event_bus
        topic = OnuDeviceEntry.event_bus_topic(self.device_id,
                                               OnuDeviceEvents.OmciCapabilitiesEvent)
        self._capabilities_subscription = bus.subscribe(topic, self.capabilties_handler)

    # Called when the mib is in sync
    def in_sync_handler(self, _topic, msg):
        self.log.debug('function-entry', _topic=_topic, msg=msg)
        if self._in_sync_subscription is not None:
            try:
                in_sync = msg[IN_SYNC_KEY]

                if in_sync:
                    # Only call this once
                    bus = self._onu_omci_device.event_bus
                    bus.unsubscribe(self._in_sync_subscription)
                    self._in_sync_subscription = None

                    # Start up device_info load
                    self.log.debug('running-mib-sync')
                    reactor.callLater(0, self._mib_in_sync)

            except Exception as e:
                self.log.exception('in-sync', e=e)

    def capabilties_handler(self, _topic, _msg):
        self.log.debug('function-entry', _topic=_topic, msg=_msg)
        if self._capabilities_subscription is not None:
            self.log.debug('capabilities-handler-done')

    # Mib is in sync, we can now query what we learned and actually start pushing ME (download) to the ONU.
    # Currently uses a basic mib download task that create a bridge with a single gem port and uni, only allowing EAP
    # Implement your own MibDownloadTask if you wish to setup something different by default
    @inlineCallbacks
    def _mib_in_sync(self):
        self.log.debug('function-entry')

        omci = self._onu_omci_device
        in_sync = omci.mib_db_in_sync

        device = yield self.core_proxy.get_device(self.device_id)
        yield self.core_proxy.device_reason_update(self.device_id, 'discovery-mibsync-complete')

        if not self._dev_info_loaded:
            self.log.info('loading-device-data-from-mib', in_sync=in_sync, already_loaded=self._dev_info_loaded)

            omci_dev = self._onu_omci_device
            config = omci_dev.configuration

            # TODO: run this sooner somehow. shouldnt have to wait for mib sync to push an initial download
            # In Sync, we can register logical ports now. Ideally this could occur on
            # the first time we received a successful (no timeout) OMCI Rx response.
            try:

                # sort the lists so we get consistent port ordering.
                ani_list = sorted(config.ani_g_entities) if config.ani_g_entities else []
                uni_list = sorted(config.uni_g_entities) if config.uni_g_entities else []
                pptp_list = sorted(config.pptp_entities) if config.pptp_entities else []
                veip_list = sorted(config.veip_entities) if config.veip_entities else []

                if ani_list is None or (pptp_list is None and veip_list is None):
                    self.log.warn("no-ani-or-unis")
                    yield self.core_proxy.device_reason_update(self.device_id, 'onu-missing-required-elements')
                    raise Exception("onu-missing-required-elements")

                # Currently logging the ani, pptp, veip, and uni for information purposes.
                # Actually act on the veip/pptp as its ME is the most correct one to use in later tasks.
                # And in some ONU the UNI-G list is incomplete or incorrect...
                for entity_id in ani_list:
                    ani_value = config.ani_g_entities[entity_id]
                    self.log.debug("discovered-ani", entity_id=entity_id, value=ani_value)
                    # TODO: currently only one OLT PON port/ANI, so this works out.  With NGPON there will be 2..?
                    self._total_tcont_count = ani_value.get('total-tcont-count')
                    self.log.debug("set-total-tcont-count", tcont_count=self._total_tcont_count)

                for entity_id in uni_list:
                    uni_value = config.uni_g_entities[entity_id]
                    self.log.debug("discovered-uni", entity_id=entity_id, value=uni_value)

                uni_entities = OrderedDict()
                for entity_id in pptp_list:
                    pptp_value = config.pptp_entities[entity_id]
                    self.log.debug("discovered-pptp", entity_id=entity_id, value=pptp_value)
                    uni_entities[entity_id] = UniType.PPTP

                for entity_id in veip_list:
                    veip_value = config.veip_entities[entity_id]
                    self.log.debug("discovered-veip", entity_id=entity_id, value=veip_value)
                    uni_entities[entity_id] = UniType.VEIP

                uni_id = 0
                for entity_id, uni_type in uni_entities.items():
                    try:
                        yield self._add_uni_port(device, entity_id, uni_id, uni_type)
                        uni_id += 1
                    except AssertionError as e:
                        self.log.warn("could not add UNI", entity_id=entity_id, uni_type=uni_type, e=e)

                self._qos_flexibility = config.qos_configuration_flexibility or 0
                self._omcc_version = config.omcc_version or OMCCVersion.Unknown

                if self._unis:
                    self._dev_info_loaded = True
                else:
                    yield self.core_proxy.device_reason_update(self.device_id, 'no-usable-unis')
                    self.log.warn("no-usable-unis")
                    raise Exception("no-usable-unis")

            except Exception as e:
                self.log.exception('device-info-load', e=e)
                self._deferred = reactor.callLater(_STARTUP_RETRY_WAIT, self._mib_in_sync)

        else:
            self.log.info('device-info-already-loaded', in_sync=in_sync, already_loaded=self._dev_info_loaded)

        if self._dev_info_loaded:
            if device.admin_state == AdminState.PREPROVISIONED or device.admin_state == AdminState.ENABLED:

                @inlineCallbacks
                def success(_results):
                    self.log.info('mib-download-success', _results=_results)
                    yield self.enable_ports()
                    yield self.core_proxy.device_state_update(device.id,
                                            oper_status=OperStatus.ACTIVE, connect_status=ConnectStatus.REACHABLE)
                    yield self.core_proxy.device_reason_update(self.device_id, 'initial-mib-downloaded')
                    self._mib_download_task = None
                    yield self.onu_active_event()

                @inlineCallbacks
                def failure(_reason):
                    self.log.warn('mib-download-failure-retrying', _reason=_reason)
                    yield self.core_proxy.device_reason_update(self.device_id, 'initial-mib-download-failure-retrying')
                    self._deferred = reactor.callLater(_STARTUP_RETRY_WAIT, self._mib_in_sync)

                # Download an initial mib that creates simple bridge that can pass EAP.  On success (above) finally set
                # the device to active/reachable.   This then opens up the handler to openflow pushes from outside
                self.log.info('downloading-initial-mib-configuration')
                self._mib_download_task = BrcmMibDownloadTask(self.omci_agent, self)
                self._deferred = self._onu_omci_device.task_runner.queue_task(self._mib_download_task)
                self._deferred.addCallbacks(success, failure)
            else:
                self.log.info('admin-down-disabling')
                self.disable(device)
        else:
            self.log.info('device-info-not-loaded-skipping-mib-download')

    @inlineCallbacks
    def _add_uni_port(self, device, entity_id, uni_id, uni_type=UniType.PPTP):
        self.log.debug('function-entry')

        uni_no = self.mk_uni_port_num(self._onu_indication.intf_id, self._onu_indication.onu_id, uni_id)

        # TODO: Some or parts of this likely need to move to UniPort. especially the format stuff
        uni_name = "uni-{}".format(uni_no)

        mac_bridge_port_num = uni_id + 1 # TODO +1 is only to test non-zero index

        self.log.debug('uni-port-inputs', uni_no=uni_no, uni_id=uni_id, uni_name=uni_name, uni_type=uni_type,
                       entity_id=entity_id, mac_bridge_port_num=mac_bridge_port_num, serial_number=device.serial_number)

        uni_port = UniPort.create(self, uni_name, uni_id, uni_no, uni_name, uni_type)
        uni_port.entity_id = entity_id
        uni_port.enabled = True
        uni_port.mac_bridge_port_num = mac_bridge_port_num

        self.log.debug("created-uni-port", uni=uni_port)

        yield self.core_proxy.port_created(device.id, uni_port.get_port())

        self._unis[uni_port.port_number] = uni_port

        self._onu_omci_device.alarm_synchronizer.set_alarm_params(onu_id=self._onu_indication.onu_id,
                                                                  uni_ports=self.uni_ports, serial_number=device.serial_number)

    # TODO NEW CORE: Figure out how to gain this knowledge from the olt.  for now cheat terribly.
    def mk_uni_port_num(self, intf_id, onu_id, uni_id):
        MAX_PONS_PER_OLT = 16
        MAX_ONUS_PER_PON = 128
        MAX_UNIS_PER_ONU = 16

        assert intf_id < MAX_PONS_PER_OLT
        assert onu_id < MAX_ONUS_PER_PON
        assert uni_id < MAX_UNIS_PER_ONU
        return intf_id << 11 | onu_id << 4 | uni_id

    @inlineCallbacks
    def onu_active_event(self):
        self.log.debug('function-entry')
        try:
            device = yield self.core_proxy.get_device(self.device_id)
            parent_device = yield self.core_proxy.get_device(self.parent_id)
            olt_serial_number = parent_device.serial_number
            raised_ts = arrow.utcnow().timestamp

            self.log.debug("onu-indication-context-data",
                       pon_id=self._onu_indication.intf_id,
                       onu_id=self._onu_indication.onu_id,
                       registration_id=self.device_id,
                       device_id=self.device_id,
                       onu_serial_number=device.serial_number,
                       olt_serial_number=olt_serial_number,
                       raised_ts=raised_ts)

            self.log.debug("Trying-to-raise-onu-active-event")
            OnuActiveEvent(self.events, self.device_id,
                           self._onu_indication.intf_id,
                           device.serial_number,
                           str(self.device_id),
                           olt_serial_number,raised_ts,
                           onu_id=self._onu_indication.onu_id).send(True)
        except Exception as active_event_error:
            self.log.exception('onu-activated-event-error',
                               errmsg=active_event_error.message)