voltha/core/device_agent.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.
#

"""
A device agent is instantiated for each Device and plays an important role
between the Device object and its adapter.
"""
import structlog
from copy import deepcopy
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks, returnValue

from voltha.core.config.config_proxy import CallbackType
from voltha.protos.common_pb2 import AdminState, OperStatus, ConnectStatus, \
    OperationResp
from voltha.protos.device_pb2 import ImageDownload
from voltha.registry import registry
from voltha.protos.openflow_13_pb2 import Flows, FlowGroups, FlowChanges, \
    FlowGroupChanges


class InvalidStateTransition(Exception): pass


class DeviceAgent(object):

    def __init__(self, core, initial_data):

        self.core = core
        self._tmp_initial_data = initial_data
        self.last_data = None
        self.calback_data = None

        self.proxy = core.get_proxy('/devices/{}'.format(initial_data.id))
        self.flows_proxy = core.get_proxy(
            '/devices/{}/flows'.format(initial_data.id))
        self.groups_proxy = core.get_proxy(
            '/devices/{}/flow_groups'.format(initial_data.id))

        self.pm_config_proxy = core.get_proxy(
            '/devices/{}/pm_configs'.format(initial_data.id))

        self.img_dnld_proxies = {}

        self.proxy.register_callback(
            CallbackType.PRE_UPDATE, self._validate_update)
        self.proxy.register_callback(
            CallbackType.POST_UPDATE, self._process_update)

        self.flows_proxy.register_callback(
            CallbackType.PRE_UPDATE, self._pre_process_flows)

        self.flows_proxy.register_callback(
            CallbackType.POST_UPDATE, self._flow_table_updated)
        self.groups_proxy.register_callback(
            CallbackType.POST_UPDATE, self._group_table_updated)

        self.pm_config_proxy.register_callback(
            CallbackType.POST_UPDATE, self._pm_config_updated)

        # to know device capabilities
        self.device_type = core.get_proxy(
            '/device_types/{}'.format(initial_data.type)).get()

        self.adapter_agent = None
        self.flow_changes = None
        self.log = structlog.get_logger(device_id=initial_data.id)

    @inlineCallbacks
    def start(self, device=None, reconcile=False):
        self.log.debug('starting', device=device)
        self._set_adapter_agent()
        if device:
            # Starting from an existing data, so set the last_data
            self.last_data = device
            if reconcile:
                self.reconcile_existing_device(device)
        else:
            yield self._process_update(self._tmp_initial_data)

        del self._tmp_initial_data

        self.log.info('started')
        returnValue(self)

    @inlineCallbacks
    def stop(self, device):
        self.log.debug('stopping', device=device)

        # First, propagate this request to the device agents
        yield self._delete_device(device)

        self.proxy.unregister_callback(
            CallbackType.PRE_UPDATE, self._validate_update)
        self.proxy.unregister_callback(
            CallbackType.POST_UPDATE, self._process_update)
        self.log.info('stopped')

    @inlineCallbacks
    def reboot_device(self, device, dry_run=False):
        self.log.debug('reboot-device', device=device, dry_run=dry_run)
        if not dry_run:
            yield self.adapter_agent.reboot_device(device)

    def register_image_download(self, request):
        try:
            self.log.debug('register-image-download', request=request)
            path = '/devices/{}/image_downloads/{}'.format(request.id,
                                                           request.name)
            self.img_dnld_proxies[request.name] = self.core.get_proxy(path)
            self.img_dnld_proxies[request.name].register_callback(
                CallbackType.POST_UPDATE, self._update_image)
            # trigger update callback
            request.state = ImageDownload.DOWNLOAD_REQUESTED
            self.img_dnld_proxies[request.name].update('/', request)
        except Exception as e:
            self.log.exception(e.message)

    def activate_image_update(self, request):
        try:
            self.log.debug('activate-image-download', request=request)
            request.image_state = ImageDownload.IMAGE_ACTIVATE
            self.img_dnld_proxies[request.name].update('/', request)
        except Exception as e:
            self.log.exception(e.message)

    def revert_image_update(self, request):
        try:
            self.log.debug('revert-image-download', request=request)
            request.image_state = ImageDownload.IMAGE_REVERT
            self.img_dnld_proxies[request.name].update('/', request)
        except Exception as e:
            self.log.exception(e.message)

    @inlineCallbacks
    def _download_image(self, device, img_dnld):
        try:
            self.log.debug('download-image', img_dnld=img_dnld)
            yield self.adapter_agent.download_image(device, img_dnld)
        except Exception as e:
            self.log.exception(e.message)

    def get_image_download_status(self, request):
        try:
            self.log.debug('get-image-download-status',
                           request=request)
            device = self.proxy.get('/')
            return self.adapter_agent.get_image_download_status(device, request)
        except Exception as e:
            self.log.exception(e.message)
            return None

    def cancel_image_download(self, img_dnld):
        try:
            self.log.debug('cancel-image-download',
                           img_dnld=img_dnld)
            device = self.proxy.get('/')
            self.adapter_agent.cancel_image_download(device, img_dnld)
            if device.admin_state == AdminState.DOWNLOADING_IMAGE:
                device.admin_state = AdminState.ENABLED
                self.proxy.update('/', device)
                self.proxy.remove('/image_downloads/{}' \
                              .format(img_dnld.name), img_dnld)
        except Exception as e:
            self.log.exception(e.message)

    def update_device_image_download(self, img_dnld):
        try:
            self.log.debug('update-device-image-download',
                           img_dnld=img_dnld)
            self.proxy.update('/image_downloads/{}' \
                              .format(img_dnld.name), img_dnld)
        except Exception as e:
            self.log.exception(e.message)

    def unregister_device_image_download(self, name):
        try:
            self.log.debug('unregister-device-image-download',
                           name=name)
            self.self_proxies[name].unregister_callback(
                CallbackType.POST_ADD, self._download_image)
            self.self_proxies[name].unregister_callback(
                CallbackType.POST_UPDATE, self._process_image)
        except Exception as e:
            self.log.exception(e.message)

    @inlineCallbacks
    def _update_image(self, img_dnld):
        try:
            self.log.debug('update-image', img_dnld=img_dnld)
            # handle download
            if img_dnld.state == ImageDownload.DOWNLOAD_REQUESTED:
                device = self.proxy.get('/')
                yield self._download_image(device, img_dnld)
                # set back to ENABLE to be allowed to activate
                device.admin_state = AdminState.ENABLED
                self.proxy.update('/', device)
            if img_dnld.image_state == ImageDownload.IMAGE_ACTIVATE:
                device = self.proxy.get('/')
                device.admin_state = AdminState.DOWNLOADING_IMAGE
                self.proxy.update('/', device)
                yield self.adapter_agent.activate_image_update(device, img_dnld)
                device.admin_state = AdminState.ENABLED
                self.proxy.update('/', device)
            elif img_dnld.image_state == ImageDownload.IMAGE_REVERT:
                device = self.proxy.get('/')
                yield self.adapter_agent.revert_image_update(device, img_dnld)
        except Exception as e:
            self.log.exception(e.message)

    @inlineCallbacks
    def self_test(self, device, dry_run=False):
        self.log.debug('self-test-device', device=device, dry_run=dry_run)
        if not dry_run:
            result = yield self.adapter_agent.self_test(device)
            returnValue(result)

    @inlineCallbacks
    def get_device_details(self, device, dry_run=False):
        self.log.debug('get-device-details', device=device, dry_run=dry_run)
        if not dry_run:
            yield self.adapter_agent.get_device_details(device)

    @inlineCallbacks
    def suppress_alarm(self, filter):
        self.log.debug('suppress-alarms')
        try:
            yield self.adapter_agent.suppress_alarm(filter)
        except Exception as e:
            self.log.exception(e.message)

    @inlineCallbacks
    def unsuppress_alarm(self, filter):
        self.log.debug('unsuppress-alarms')
        try:
            yield self.adapter_agent.unsuppress_alarm(filter)
        except Exception as e:
            self.log.exception(e.message)

    @inlineCallbacks
    def reconcile_existing_device(self, device, dry_run=False):
        self.log.debug('reconcile-existing-device',
                       device=device,
                       dry_run=False)
        if not dry_run:
            yield self.adapter_agent.reconcile_device(device)

    def _set_adapter_agent(self):
        adapter_name = self._tmp_initial_data.adapter
        if adapter_name == '':
            proxy = self.core.get_proxy('/')
            known_device_types = dict(
                (dt.id, dt) for dt in proxy.get('/device_types'))
            device_type = known_device_types[self._tmp_initial_data.type]
            adapter_name = device_type.adapter
        assert adapter_name != ''
        self.adapter_agent = registry('adapter_loader').get_agent(adapter_name)

    @inlineCallbacks
    def _validate_update(self, device):
        """
        Called before each update, it allows the blocking of the update
        (by raising an exception), or even the augmentation of the incoming
        data.
        """
        self.log.debug('device-pre-update', device=device)
        yield self._process_state_transitions(device, dry_run=True)
        returnValue(device)

    @inlineCallbacks
    def _process_update(self, device):
        """
        Called after the device object was updated (individually or part of
        a transaction), and it is used to propagate the change down to the
        adapter
        """
        self.log.debug('device-post-update', device=device)

        # first, process any potential state transition
        yield self._process_state_transitions(device)

        # finally, store this data as last data so we can see what changed
        self.last_data = device

    @inlineCallbacks
    def _process_state_transitions(self, device, dry_run=False):

        old_admin_state = getattr(self.last_data, 'admin_state',
                                  AdminState.UNKNOWN)
        new_admin_state = device.admin_state
        self.log.debug('device-admin-states', old_state=old_admin_state,
                       new_state=new_admin_state, dry_run=dry_run)
        transition_handler = self.admin_state_fsm.get(
            (old_admin_state, new_admin_state), None)
        if transition_handler is None:
            self.log.debug('No Operation', old_state=old_admin_state,
                           new_state=new_admin_state, dry_run=dry_run)
            pass  # no-op
        elif transition_handler is False:
            raise InvalidStateTransition('{} -> {}'.format(
                old_admin_state, new_admin_state))
        else:
            assert callable(transition_handler)
            yield transition_handler(self, device, dry_run)

    @inlineCallbacks
    def _activate_device(self, device, dry_run=False):
        self.log.debug('activate-device', device=device, dry_run=dry_run)
        if not dry_run:
            device.oper_status = OperStatus.ACTIVATING
            self.update_device(device)
            yield self.adapter_agent.adopt_device(device)

    @inlineCallbacks
    def update_device(self, device):
        self.log.debug('updating-device', device=device.id)
        self.last_data = device  # so that we don't propagate back
        self.proxy.update('/', device)
        if device.admin_state == AdminState.ENABLED and \
                device.oper_status == OperStatus.ACTIVE and \
                device.connect_status == ConnectStatus.REACHABLE:
            self.log.info('replay-create-interfaces ', device=device.id)
            self.core.xpon_agent.replay_interface(device.id)
            # if device accepts bulk flow update, lets just call that
            if self.device_type.accepts_bulk_flow_update:
                flows = self.flows_proxy.get('/')  # gather flows
                groups = self.groups_proxy.get('/')  # gather flow groups
                self.log.info('replay-flows ', device=device.id)
                yield self.adapter_agent.update_flows_bulk(
                    device=device,
                    flows=flows,
                    groups=groups)

    def update_device_pm_config(self, device_pm_config, init=False):
        self.callback_data = init  # so that we don't push init data
        self.pm_config_proxy.update('/', device_pm_config)

    def _propagate_change(self, device, dry_run=False):
        self.log.debug('propagate-change', device=device, dry_run=dry_run)
        if device != self.last_data:
            self.log.warn('Not-implemented-default-to-noop')
            # raise NotImplementedError()
        else:
            self.log.debug('no-op')

    def _abandon_device(self, device, dry_run=False):
        self.log.debug('abandon-device', device=device, dry_run=dry_run)
        raise NotImplementedError()

    def _delete_all_flows(self):
        """ Delete all flows on the device """
        try:
            self.flows_proxy.update('/', Flows(items=[]))
            self.groups_proxy.update('/', FlowGroups(items=[]))
        except Exception, e:
            self.exception('flow-delete-exception', e=e)

    @inlineCallbacks
    def _disable_device(self, device, dry_run=False):
        try:
            self.log.debug('disable-device', device=device, dry_run=dry_run)
            if not dry_run:
                # Remove all flows before disabling device
                # Only do this for ONUs. ponsim seems to want the flows
                # to be removed on onu disable and it does not seem to
                # hurt the real devices. Flows on OLT are not removed when
                # OLT is disabled - each adapter may do its own thing
                # for disable -e.g. OpenOLT disables the NNI interface.
                if not device.root:
                    self._delete_all_flows()
                yield self.adapter_agent.disable_device(device)
        except Exception, e:
            self.log.exception('error', e=e)

    @inlineCallbacks
    def _reenable_device(self, device, dry_run=False):
        self.log.debug('reenable-device', device=device, dry_run=dry_run)
        if not dry_run:
            yield self.adapter_agent.reenable_device(device)

    @inlineCallbacks
    def _delete_device(self, device, dry_run=False):
        self.log.debug('delete-device', device=device, dry_run=dry_run)
        if not dry_run:
            yield self.adapter_agent.delete_device(device)

    def simulate_alarm(self, device, alarm):
        try:
            self.log.debug('simulate_alarm', alarm=alarm)
            return self.adapter_agent.simulate_alarm(device, alarm)
        except Exception as e:
            self.log.exception(e.message)
            return OperationResp(code=OperationResp.OPERATION_FAILURE)

    admin_state_fsm = {

        # Missing entries yield no-op
        # False means invalid state change

        (AdminState.UNKNOWN, AdminState.ENABLED): _activate_device,

        (AdminState.PREPROVISIONED, AdminState.UNKNOWN): False,
        (AdminState.PREPROVISIONED, AdminState.ENABLED): _activate_device,
        (AdminState.PREPROVISIONED, AdminState.DOWNLOADING_IMAGE): False,

        (AdminState.ENABLED, AdminState.UNKNOWN): False,
        (AdminState.ENABLED, AdminState.ENABLED): _propagate_change,
        (AdminState.ENABLED, AdminState.DISABLED): _disable_device,
        (AdminState.ENABLED, AdminState.PREPROVISIONED): _abandon_device,

        (AdminState.DISABLED, AdminState.UNKNOWN): False,
        (AdminState.DISABLED, AdminState.PREPROVISIONED): _abandon_device,
        (AdminState.DISABLED, AdminState.ENABLED): _reenable_device,
        (AdminState.DISABLED, AdminState.DOWNLOADING_IMAGE): False,

        (AdminState.DOWNLOADING_IMAGE, AdminState.DISABLED): False

    }

    ## <======================= PM CONFIG UPDATE HANDLING ====================

    # @inlineCallbacks
    def _pm_config_updated(self, pm_configs):
        self.log.debug('pm-config-updated', pm_configs=pm_configs,
                       callback_data=self.callback_data)
        device_id = self.proxy.get('/').id
        if not self.callback_data:
            self.adapter_agent.update_adapter_pm_config(device_id, pm_configs)
        self.callback_data = None

    ## <======================= FLOW TABLE UPDATE HANDLING ====================

    def _pre_process_flows(self, flows):
        """
        This method is invoked before a device flow table data model is
        updated. If the device supports accepts_add_remove_flow_updates then it
        pre-processes the desired flows against what currently
        exist on the device to figure out which flows to delete and which
        ones to add. The resulting data is stored locally and the flow table is
        updated during the post-processing phase, i.e. via the POST_UPDATE
        callback
        :param flows: Desired flows
        :return: None
        """
        self.current_flows = self.flows_proxy.get('/')
        # self.log.debug('pre-processing-flows',
        #                logical_device_id=self.last_data.id,
        #                desired_flows=flows,
        #                existing_flows=self.current_flows)

        if self.flow_changes is None:
            self.flow_changes = FlowChanges()
        else:
            del self.flow_changes.to_add.items[:]
            del self.flow_changes.to_remove.items[:]

        current_flow_ids = set(f.id for f in self.current_flows.items)
        desired_flow_ids = set(f.id for f in flows.items)

        ids_to_add = desired_flow_ids.difference(current_flow_ids)
        ids_to_del = current_flow_ids.difference(desired_flow_ids)

        for f in flows.items:
            if f.id in ids_to_add:
                self.flow_changes.to_add.items.extend([f])

        for f in self.current_flows.items:
            if f.id in ids_to_del:
                self.flow_changes.to_remove.items.extend([f])

        self.log.debug('pre-processed-flows',
                       logical_device_id=self.last_data.id,
                       flows_to_add=len(self.flow_changes.to_add.items),
                       flows_to_remove=len(self.flow_changes.to_remove.items))


    @inlineCallbacks
    def _flow_table_updated(self, flows):
        try:
            self.log.debug('flow-table-updated',
                        logical_device_id=self.last_data.id)
            if self.flow_changes is not None and (len(self.flow_changes.to_remove.items) == 0) and (len(
                    self.flow_changes.to_add.items) == 0):
                self.log.debug('no-flow-update-required',
                            logical_device_id=self.last_data.id)
            else:

                # if device accepts non-bulk flow update, lets just call that first
                if self.device_type.accepts_add_remove_flow_updates:

                    try:
                        flow_changes = deepcopy(self.flow_changes)
                        yield self.adapter_agent.update_flows_incrementally(
                            device=self.last_data,
                            flow_changes=flow_changes,
                            group_changes=FlowGroupChanges()
                        )
                    except Exception as e:
                        self.log.exception("Failure-updating-flows", e=e)

                # if device accepts bulk flow update, lets just call that
                elif self.device_type.accepts_bulk_flow_update:
                    self.log.debug('invoking bulk')
                    groups = self.groups_proxy.get('/')  # gather flow groups
                    yield self.adapter_agent.update_flows_bulk(
                        device=self.last_data,
                        flows=flows,
                        groups=groups)
                    # add ability to notify called when an flow update completes
                    # see https://jira.opencord.org/browse/CORD-839
                else:
                    raise NotImplementedError()
        except Exception as e:
            self.log.error('error', e=e)

    ## <======================= GROUP TABLE UPDATE HANDLING ===================

    @inlineCallbacks
    def _group_table_updated(self, groups):
        self.log.debug('group-table-updated',
                       logical_device_id=self.last_data.id,
                       flow_groups=groups)

        # if device accepts bulk flow update, lets just call that
        if self.device_type.accepts_bulk_flow_update:
            flows = self.flows_proxy.get('/')  # gather flows
            yield self.adapter_agent.update_flows_bulk(
                device=self.last_data,
                flows=flows,
                groups=groups)
            # add ability to notify called when an group update completes
            # see https://jira.opencord.org/browse/CORD-839

        elif self.device_type.accepts_add_remove_flow_updates:
            raise NotImplementedError()

        else:
            raise NotImplementedError()