Added new Event Automatation functionalities (tested locally, unit tests not working in the repo) (04247fe8) · Commits · TFS / controller

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')