Commit 7dba376e authored by Carlos Natalino's avatar Carlos Natalino
Browse files

Merge remote-tracking branch 'origin/feat/monitoring-service' into... 

Merge remote-tracking branch 'origin/feat/monitoring-service' into feat/opticalcentralizedattackdetector-service
parents 98cd56a1 04247fe8

src/common/message_broker/Factory.py

+2 −2
Original line number Diff line number Diff line
@@ -3,13 +3,13 @@ from typing import Optional, Union 
from .backend._Backend import _Backend

from .backend.BackendEnum import BackendEnum

from .backend.inmemory.InMemoryBackend import InMemoryBackend

from .backend.redis.RedisBackend import RedisBackend

# from .backend.redis.RedisBackend import RedisBackend



LOGGER = logging.getLogger(__name__)



BACKENDS = {

    BackendEnum.INMEMORY.value: InMemoryBackend,

    BackendEnum.REDIS.value: RedisBackend,

    # BackendEnum.REDIS.value: RedisBackend,

    #BackendEnum.KAFKA.value: KafkaBackend,

    #BackendEnum.RABBITMQ.value: RabbitMQBackend,

    #BackendEnum.ZEROMQ.value: ZeroMQBackend,

src/common/orm/Factory.py

+2 −2
Original line number Diff line number Diff line
@@ -3,13 +3,13 @@ from typing import Optional, Union 
from .backend._Backend import _Backend

from .backend.BackendEnum import BackendEnum

from .backend.inmemory.InMemoryBackend import InMemoryBackend

from .backend.redis.RedisBackend import RedisBackend

# from .backend.redis.RedisBackend import RedisBackend



LOGGER = logging.getLogger(__name__)



BACKENDS = {

    BackendEnum.INMEMORY.value: InMemoryBackend,

    BackendEnum.REDIS.value: RedisBackend,

    # BackendEnum.REDIS.value: RedisBackend,

    #BackendEnum.MONGODB.value: MongoDBBackend,

    #BackendEnum.RETHINKDB.value: RethinkDBBackend,

    #BackendEnum.ETCD.value: EtcdBackend,

src/context/tests/example_objects.py

+22 −20
Original line number Diff line number Diff line
@@ -2,12 +2,13 @@ from copy import deepcopy 
from common.Constants import DEFAULT_CONTEXT_UUID, DEFAULT_TOPOLOGY_UUID

from context.proto.context_pb2 import (

    ConfigActionEnum, DeviceDriverEnum, DeviceOperationalStatusEnum, ServiceStatusEnum, ServiceTypeEnum)

from context.proto.kpi_sample_types_pb2 import KpiSampleType



# Some example objects to be used by the tests



# Helper methods

def config_rule(action, resource_key, resource_value):

    return {'action': action, 'resource_key': resource_key, 'resource_value': resource_value}

def config_rule(action, resource_key, resource_value, kpi_sample_type):

    return {'action': action, 'resource_key': resource_key, 'resource_value': resource_value, 'kpi_sample_type': kpi_sample_type}



def endpoint_id(topology_id, device_id, endpoint_uuid):

    return {'topology_id': deepcopy(topology_id), 'device_id': deepcopy(device_id),
@@ -40,9 +41,10 @@ DEVICE1 = { 
    'device_id': deepcopy(DEVICE1_ID),

    'device_type': 'packet-router',

    'device_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value1'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value2'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value3'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value1', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value2', KpiSampleType.PACKETS_RECEIVED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value3', KpiSampleType.BYTES_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc4/value', 'value4', KpiSampleType.BYTES_RECEIVED),

    ]},

    'device_operational_status': DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED,

    'device_drivers': [DeviceDriverEnum.DEVICEDRIVER_OPENCONFIG, DeviceDriverEnum.DEVICEDRIVER_P4],
@@ -59,9 +61,9 @@ DEVICE2 = { 
    'device_id': deepcopy(DEVICE2_ID),

    'device_type': 'packet-router',

    'device_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value4'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value5'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value6'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value4', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value5', KpiSampleType.PACKETS_RECEIVED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value6', KpiSampleType.BYTES_TRANSMITTED),

    ]},

    'device_operational_status': DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED,

    'device_drivers': [DeviceDriverEnum.DEVICEDRIVER_OPENCONFIG, DeviceDriverEnum.DEVICEDRIVER_P4],
@@ -78,9 +80,9 @@ DEVICE3 = { 
    'device_id': deepcopy(DEVICE3_ID),

    'device_type': 'packet-router',

    'device_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value4'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value5'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value6'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc1/value', 'value4', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc2/value', 'value5', KpiSampleType.PACKETS_RECEIVED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'dev/rsrc3/value', 'value6', KpiSampleType.BYTES_TRANSMITTED),

    ]},

    'device_operational_status': DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED,

    'device_drivers': [DeviceDriverEnum.DEVICEDRIVER_OPENCONFIG, DeviceDriverEnum.DEVICEDRIVER_P4],
@@ -139,9 +141,9 @@ SERVICE_DEV1_DEV2 = { 
    ],

    'service_status': {'service_status': ServiceStatusEnum.SERVICESTATUS_ACTIVE},

    'service_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9', KpiSampleType.PACKETS_TRANSMITTED),

    ]},

}
@@ -163,9 +165,9 @@ SERVICE_DEV1_DEV3 = { 
    ],

    'service_status': {'service_status': ServiceStatusEnum.SERVICESTATUS_ACTIVE},

    'service_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9', KpiSampleType.PACKETS_TRANSMITTED),

    ]},

}
@@ -187,8 +189,8 @@ SERVICE_DEV2_DEV3 = { 
    ],

    'service_status': {'service_status': ServiceStatusEnum.SERVICESTATUS_ACTIVE},

    'service_config': {'config_rules': [

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9'),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc1/value', 'value7', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc2/value', 'value8', KpiSampleType.PACKETS_TRANSMITTED),

        config_rule(ConfigActionEnum.CONFIGACTION_SET, 'svc/rsrc3/value', 'value9', KpiSampleType.PACKETS_TRANSMITTED),

    ]},

}

src/monitoring/client/monitoring_client.py

+6 −7
Original line number Diff line number Diff line
@@ -53,16 +53,15 @@ class MonitoringClient: 
        LOGGER.info('GetKpiDescriptor result: {}'.format(response))

        return monitoring_pb2.KpiDescriptor()



    def ListenEvents(self, ):

        LOGGER.info('ListenEvents: {}'.format())

        response = self.server.ListenEvents()

        LOGGER.info('ListenEvents result: {}'.format(response))

        return monitoring_pb2.KpiDescriptor()



if __name__ == '__main__':

    # get port

    port = sys.argv[1] if len(sys.argv) > 1 else '7070'



    # form request

    kpi_request = monitoring_pb2.KpiRequest()

    kpi_request.device_id.device_id = 'KPIID0000' # pylint: disable=maybe-no-member

    kpi_request.kpiDescription = 'KPI Description'

    kpi_request.kpi_sample_type = monitoring_pb2.KpiSampleType.PACKETS_TRANSMITTED



    # make call to server

    client = MonitoringClient(port=port)
 
    response=client.IncludeKpi(kpi_request)

src/monitoring/service/EventTools.py

0 → 100644
+78 −0
Original line number Diff line number Diff line 
import threading

from queue import Queue



import grpc



from common.rpc_method_wrapper.ServiceExceptions import ServiceException

from context.client.ContextClient import ContextClient

from context.proto import kpi_sample_types_pb2

from context.proto.context_pb2 import Empty, EventTypeEnum



from common.logger import getJSONLogger

from monitoring.client.monitoring_client import MonitoringClient

from monitoring.proto import monitoring_pb2



LOGGER = getJSONLogger('monitoringservice-server')

LOGGER.setLevel('DEBUG')



class EventsDeviceCollector:

    def __init__(self, context_client_grpc : ContextClient, monitoring_client_grpc : MonitoringClient) -> None: # pylint: disable=redefined-outer-name

        self._events_queue = Queue()



        self._device_stream   = context_client_grpc.GetDeviceEvents(Empty())

        self._context_client  = context_client_grpc

        self._monitoring_client = monitoring_client_grpc



        self._device_thread   = threading.Thread(target=self._collect, args=(self._device_stream  ,), daemon=False)



    def _collect(self, events_stream) -> None:

        try:

            for event in events_stream:

                self._events_queue.put_nowait(event)

        except grpc.RpcError as e:

            if e.code() != grpc.StatusCode.CANCELLED: # pylint: disable=no-member

                raise # pragma: no cover



    def start(self):

        self._device_thread.start()



    def get_event(self, block : bool = True, timeout : float = 0.1):

        return self._events_queue.get(block=block, timeout=timeout)



    def stop(self):



        self._device_stream.cancel()



        self._device_thread.join()



    def listen_events(self):

        LOGGER.info('getting Kpi by KpiID')



        try:

            kpi_id_list = []

            for i in range(self._events_queue.qsize()):

                event = self.get_event(block=True)



                if event.event.event_type == EventTypeEnum.EVENTTYPE_CREATE:

                    device = self._context_client.GetDevice(event.device_id)



                    for j,end_point in enumerate(device.device_endpoints):

                        # for k,rule in enumerate(device.device_config.config_rules):

                        kpi_descriptor = monitoring_pb2.KpiDescriptor()



                        kpi_descriptor.kpi_description                      = device.device_type

                        kpi_descriptor.kpi_sample_type                      = kpi_sample_types_pb2.KpiSampleType.PACKETS_TRANSMITTED

                        kpi_descriptor.device_id.CopyFrom(device.device_id)

                        kpi_descriptor.endpoint_id.CopyFrom(end_point.endpoint_id)

                        kpi_descriptor.service_id.service_uuid.uuid         = "SERV"+str(i+1)



                        kpi_id = self._monitoring_client.CreateKpi(kpi_descriptor)

                        kpi_id_list.append(kpi_id)



            return kpi_id_list



        except ServiceException as e:

            LOGGER.exception('ListenEvents exception')



        except Exception as e:  # pragma: no cover

            LOGGER.exception('ListenEvents exception')