pyvoltha/adapters/kafka/core_proxy.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.
#

"""
Agent to play gateway between CORE and an adapter.
"""
from __future__ import absolute_import
import structlog
import arrow
from google.protobuf.message import Message
from twisted.internet.defer import inlineCallbacks, returnValue

from .container_proxy import ContainerProxy

from voltha_protos.common_pb2 import ID, ConnectStatus, OperStatus
from voltha_protos.inter_container_pb2 import StrType, BoolType, IntType, Packet
from voltha_protos.device_pb2 import Device, Ports, Devices
from voltha_protos.voltha_pb2 import CoreInstance, EventFilterRuleKey
from voltha_protos.events_pb2 import Event
from voltha_protos.events_pb2 import KpiEvent2, KpiEventType, MetricInformation, MetricMetaData
import six

log = structlog.get_logger()


def createSubTopic(*args):
    return '_'.join(args)

class CoreProxy(ContainerProxy):

    def __init__(self, kafka_proxy, default_core_topic, default_event_topic, my_listening_topic):
        super(CoreProxy, self).__init__(kafka_proxy, default_core_topic,
                                        my_listening_topic)
        self.core_default_topic = default_core_topic
        self.event_default_topic = default_event_topic
        self.deviceId_to_core_map = dict()

    def update_device_core_reference(self, device_id, core_topic):
        log.debug("update_device_core_reference")
        self.deviceId_to_core_map[device_id] = core_topic

    def delete_device_core_reference(self, device_id, core_topic):
        log.debug("delete_device_core_reference")
        del self.deviceId_to_core_map[device_id]

    def get_adapter_topic(self, **kwargs):
        return self.listening_topic

    def get_core_topic(self, device_id):
        if device_id in self.deviceId_to_core_map:
            return self.deviceId_to_core_map[device_id]
        return self.core_default_topic

    @ContainerProxy.wrap_request(CoreInstance)
    @inlineCallbacks
    def register(self, adapter, deviceTypes):
        log.debug("register")

        if adapter.totalReplicas == 0 and adapter.currentReplica != 0:
            raise Exception("totalReplicas can't be 0, since you're here you have at least one")

        if adapter.currentReplica == 0 and adapter.totalReplicas != 0:
            raise Exception("currentReplica can't be 0, it has to start from 1")

        if adapter.currentReplica == 0 and adapter.totalReplicas == 0:
            # if the adapter is not setting these fields they default to 0,
            # in that case it means the adapter is not ready to be scaled
            # and thus it defaults to a single instance
            adapter.currentReplica = 1
            adapter.totalReplicas = 1

        if adapter.currentReplica > adapter.totalReplicas:
            raise Exception("currentReplica (%d) can't be greater than totalReplicas (%d)"
                          % (adapter.currentReplica, adapter.totalReplicas))

        try:
            res = yield self.invoke(rpc="Register",
                                    adapter=adapter,
                                    deviceTypes=deviceTypes)
            log.info("registration-returned", res=res)
            returnValue(res)
        except Exception as e:
            log.exception("registration-exception", e=e)
            raise

    @ContainerProxy.wrap_request(Device)
    @inlineCallbacks
    def get_device(self, device_id):
        log.debug("get-device")
        id = ID()
        id.id = device_id
        # Once we have a device being managed, all communications between the
        # the adapter and the core occurs over a topic associated with that
        # device
        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="GetDevice",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id)
        returnValue(res)

    @ContainerProxy.wrap_request(Device)
    @inlineCallbacks
    def get_child_device(self, parent_device_id, **kwargs):
        log.debug("get-child-device")
        id = ID()
        id.id = parent_device_id
        to_topic = self.get_core_topic(parent_device_id)
        reply_topic = self.get_adapter_topic()
        args = self._to_proto(**kwargs)
        res = yield self.invoke(rpc="GetChildDevice",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                **args)
        returnValue(res)

    @ContainerProxy.wrap_request(Ports)
    @inlineCallbacks
    def get_ports(self, device_id, port_type):
        id = ID()
        id.id = device_id
        p_type = IntType()
        p_type.val = port_type
        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="GetPorts",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                port_type=p_type)
        returnValue(res)

    @ContainerProxy.wrap_request(Devices)
    @inlineCallbacks
    def get_child_devices(self, parent_device_id):
        log.debug("get-child-devices")
        id = ID()
        id.id = parent_device_id
        to_topic = self.get_core_topic(parent_device_id)
        reply_topic = self.get_adapter_topic()
        res = yield self.invoke(rpc="GetChildDevices",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id)
        returnValue(res)

    @ContainerProxy.wrap_request(Device)
    @inlineCallbacks
    def get_child_device_with_proxy_address(self, proxy_address):
        log.debug("get-child-device-with-proxy-address")
        id = ID()
        id.id = proxy_address.device_id
        to_topic = self.get_core_topic(proxy_address.device_id)
        reply_topic = self.get_adapter_topic()
        res = yield self.invoke(rpc="GetChildDeviceWithProxyAddress",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                proxy_address=proxy_address)
        returnValue(res)

    def _to_proto(self, **kwargs):
        encoded = {}
        for k, v in six.iteritems(kwargs):
            if isinstance(v, Message):
                encoded[k] = v
            elif type(v) == int:
                i_proto = IntType()
                i_proto.val = v
                encoded[k] = i_proto
            elif type(v) == str:
                s_proto = StrType()
                s_proto.val = v
                encoded[k] = s_proto
            elif type(v) == bool:
                b_proto = BoolType()
                b_proto.val = v
                encoded[k] = b_proto
        return encoded

    @ContainerProxy.wrap_request(Device)
    @inlineCallbacks
    def child_device_detected(self,
                              parent_device_id,
                              parent_port_no,
                              child_device_type,
                              channel_id,
                              **kw):
        id = ID()
        id.id = parent_device_id
        ppn = IntType()
        ppn.val = parent_port_no
        cdt = StrType()
        cdt.val = child_device_type
        channel = IntType()
        channel.val = channel_id
        to_topic = self.get_core_topic(parent_device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, parent_device_id)
        # reply_topic = createSubTopic(self.listening_topic, parent_device_id)
        args = self._to_proto(**kw)
        res = yield self.invoke(rpc="ChildDeviceDetected",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                parent_device_id=id,
                                parent_port_no=ppn,
                                child_device_type=cdt,
                                channel_id=channel,
                                **args)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def device_update(self, device):
        log.debug("device_update")
        to_topic = self.get_core_topic(device.id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device.id)
        # reply_topic = createSubTopic(self.listening_topic, device.id)
        res = yield self.invoke(rpc="DeviceUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device=device)
        returnValue(res)

    def child_device_removed(parent_device_id, child_device_id):
        raise NotImplementedError()

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def device_state_update(self, device_id,
                            oper_status=None,
                            connect_status=None):
        id = ID()
        id.id = device_id
        o_status = IntType()
        if oper_status or oper_status == OperStatus.UNKNOWN:
            o_status.val = oper_status
        else:
            o_status.val = -1
        c_status = IntType()
        if connect_status or connect_status == ConnectStatus.UNKNOWN:
            c_status.val = connect_status
        else:
            c_status.val = -1

        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        #     reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="DeviceStateUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                oper_status=o_status,
                                connect_status=c_status)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def children_state_update(self, device_id,
                              oper_status=None,
                              connect_status=None):
        id = ID()
        id.id = device_id
        o_status = IntType()
        if oper_status or oper_status == OperStatus.UNKNOWN:
            o_status.val = oper_status
        else:
            o_status.val = -1
        c_status = IntType()
        if connect_status or connect_status == ConnectStatus.UNKNOWN:
            c_status.val = connect_status
        else:
            c_status.val = -1

        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="ChildrenStateUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                oper_status=o_status,
                                connect_status=c_status)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def port_state_update(self,
                          device_id,
                          port_type,
                          port_no,
                          oper_status):
        id = ID()
        id.id = device_id
        pt = IntType()
        pt.val = port_type
        pNo = IntType()
        pNo.val = port_no
        o_status = IntType()
        o_status.val = oper_status

        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="PortStateUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                port_type=pt,
                                port_no=pNo,
                                oper_status=o_status)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def child_devices_state_update(self, parent_device_id,
                                   oper_status=None,
                                   connect_status=None):

        id = ID()
        id.id = parent_device_id
        o_status = IntType()
        if oper_status or oper_status == OperStatus.UNKNOWN:
            o_status.val = oper_status
        else:
            o_status.val = -1
        c_status = IntType()
        if connect_status or connect_status == ConnectStatus.UNKNOWN:
            c_status.val = connect_status
        else:
            c_status.val = -1

        to_topic = self.get_core_topic(parent_device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, parent_device_id)
        # reply_topic = createSubTopic(self.listening_topic, parent_device_id)
        res = yield self.invoke(rpc="child_devices_state_update",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                parent_device_id=id,
                                oper_status=o_status,
                                connect_status=c_status)
        returnValue(res)

    def child_devices_removed(parent_device_id):
        raise NotImplementedError()

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def device_pm_config_update(self, device_pm_config, init=False):
        log.debug("device_pm_config_update")
        b = BoolType()
        b.val = init
        to_topic = self.get_core_topic(device_pm_config.id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_pm_config.id)
        # reply_topic = createSubTopic(self.listening_topic, device_pm_config.id)
        res = yield self.invoke(rpc="DevicePMConfigUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_pm_config=device_pm_config,
                                init=b)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def port_created(self, device_id, port):
        log.debug("port_created")
        proto_id = ID()
        proto_id.id = device_id
        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="PortCreated",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=proto_id,
                                port=port)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def ports_state_update(self,
                          device_id,
                          port_type_filter,
                          oper_status):
        log.debug("ports_state_update", device_id=device_id, oper_status=oper_status)
        id = ID()
        id.id = device_id
        t_filter = IntType()
        t_filter.val = port_type_filter
        o_status = IntType()
        o_status.val = oper_status

        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="PortsStateUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                port_type_filter=t_filter,
                                oper_status=o_status)
        log.debug("ports_state_update_response", device_id=device_id, port_type_filter=port_type_filter, oper_status=oper_status, response=res)
        returnValue(res)

    def port_removed(device_id, port):
        raise NotImplementedError()

    def ports_enabled(device_id):
        raise NotImplementedError()

    def ports_disabled(device_id):
        raise NotImplementedError()

    def ports_oper_status_update(device_id, oper_status):
        raise NotImplementedError()

    def image_download_update(img_dnld):
        raise NotImplementedError()

    def image_download_deleted(img_dnld):
        raise NotImplementedError()

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def send_packet_in(self, device_id, port, packet):
        log.debug("send_packet_in", device_id=device_id)
        proto_id = ID()
        proto_id.id = device_id
        p = IntType()
        p.val = port
        pac = Packet()
        pac.payload = packet
        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()
        # to_topic = createSubTopic(self.core_topic, device_id)
        # reply_topic = createSubTopic(self.listening_topic, device_id)
        res = yield self.invoke(rpc="PacketIn",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=proto_id,
                                port=p,
                                packet=pac)
        returnValue(res)

    @ContainerProxy.wrap_request(None)
    @inlineCallbacks
    def device_reason_update(self, device_id, reason):
        id = ID()
        id.id = device_id
        rsn = StrType()
        rsn.val = reason
        to_topic = self.get_core_topic(device_id)
        reply_topic = self.get_adapter_topic()

        res = yield self.invoke(rpc="DeviceReasonUpdate",
                                to_topic=to_topic,
                                reply_topic=reply_topic,
                                device_id=id,
                                device_reason=rsn)

        returnValue(res)

    # ~~~~~~~~~~~~~~~~~~~ Handle event submissions ~~~~~~~~~~~~~~~~~~~~~

    def filter_alarm(self, device_id, alarm_event):
        '''
        TODO
        alarm filtering functionality is not implemented
        in Voltha 1.x
        '''
        log.warn('filter_alarm is not implemented')
        return
        #alarm_filters = self.root_proxy.get('/alarm_filters')

        rule_values = {
            'id': alarm_event.id,
            'type': AlarmEventType.AlarmEventType.Name(alarm_event.type),
            'category': AlarmEventCategory.AlarmEventCategory.Name(
                alarm_event.category),
            'severity': AlarmEventSeverity.AlarmEventSeverity.Name(
                alarm_event.severity),
            'resource_id': alarm_event.resource_id,
            'device_id': device_id
        }

        for alarm_filter in alarm_filters:
            if alarm_filter.rules:
                exclude = True
                for rule in alarm_filter.rules:
                    log.debug("compare-alarm-event",
                                   key=EventFilterRuleKey.EventFilterRuleKey.Name(
                                       rule.key),
                                   actual=rule_values[
                                       EventFilterRuleKey.EventFilterRuleKey.Name(
                                           rule.key)].lower(),
                                   expected=rule.value.lower())
                    exclude = exclude and \
                              (rule_values[
                                   EventFilterRuleKey.EventFilterRuleKey.Name(
                                       rule.key)].lower() == rule.value.lower())
                    if not exclude:
                        break

                if exclude:
                    log.info("filtered-alarm-event", alarm=alarm_event)
                    return True

        return False

    @inlineCallbacks
    def submit_event(self, event_msg):
        try:
            assert isinstance(event_msg, Event)
            res = yield self.kafka_proxy._send_kafka_message(self.event_default_topic, event_msg)
            returnValue(res)
        except Exception as e:
            log.exception('failed-event-submission',
                        type=type(event_msg), e=e)