# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/) # # 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. import grpc, json, logging, queue, threading from typing import Dict from common.method_wrappers.ServiceExceptions import ServiceException from common.proto import monitoring_pb2 from common.proto.context_pb2 import ConfigActionEnum, DeviceOperationalStatusEnum, Empty, EventTypeEnum from common.proto.kpi_sample_types_pb2 import KpiSampleType from common.tools.grpc.Tools import grpc_message_to_json_string from context.client.ContextClient import ContextClient from monitoring.client.MonitoringClient import MonitoringClient from monitoring.service.MonitoringServiceServicerImpl import LOGGER from monitoring.service.NameMapping import NameMapping LOGGER = logging.getLogger(__name__) DEVICE_OP_STATUS_UNDEFINED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_UNDEFINED DEVICE_OP_STATUS_DISABLED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED DEVICE_OP_STATUS_ENABLED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED DEVICE_OP_STATUS_NOT_ENABLED = {DEVICE_OP_STATUS_UNDEFINED, DEVICE_OP_STATUS_DISABLED} KPISAMPLETYPE_UNKNOWN = KpiSampleType.KPISAMPLETYPE_UNKNOWN class EventsDeviceCollector: def __init__(self, name_mapping : NameMapping) -> None: # pylint: disable=redefined-outer-name self._events_queue = queue.Queue() self._context_client_grpc = ContextClient() self._device_stream = self._context_client_grpc.GetDeviceEvents(Empty()) self._context_client = self._context_client_grpc self._channel = self._context_client_grpc.channel self._monitoring_client = MonitoringClient(host='127.0.0.1') self._device_thread = threading.Thread(target=self._collect, args=(self._device_stream,), daemon=False) self._device_to_state : Dict[str, DeviceOperationalStatusEnum] = dict() self._name_mapping = name_mapping def grpc_server_on(self): try: grpc.channel_ready_future(self._channel).result(timeout=15) return True except grpc.FutureTimeoutError: return False def _collect(self, events_stream): 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): try: self._device_thread.start() except RuntimeError: LOGGER.exception('Start EventTools exception') 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): try: kpi_id_list = [] while True: # LOGGER.info('getting Kpi by KpiID') try: event = self.get_event(block=True, timeout=0.5) event_type = event.event.event_type device_uuid = event.device_id.device_uuid.uuid if event_type in {EventTypeEnum.EVENTTYPE_REMOVE}: LOGGER.info('removing device') self._device_to_state.pop(device_uuid, None) continue if event_type not in {EventTypeEnum.EVENTTYPE_CREATE, EventTypeEnum.EVENTTYPE_UPDATE}: LOGGER.info('unexpected event') # Unknown event type continue device = self._context_client.GetDevice(event.device_id) self._name_mapping.set_device_name(device_uuid, device.name) old_operational_status = self._device_to_state.get(device_uuid, DEVICE_OP_STATUS_UNDEFINED) LOGGER.info('old_operational_status={:s}'.format(str(old_operational_status))) device_was_not_enabled = (old_operational_status in DEVICE_OP_STATUS_NOT_ENABLED) LOGGER.info('device_was_not_enabled={:s}'.format(str(device_was_not_enabled))) new_operational_status = device.device_operational_status LOGGER.info('new_operational_status={:s}'.format(str(new_operational_status))) device_is_enabled = (new_operational_status == DEVICE_OP_STATUS_ENABLED) LOGGER.info('device_is_enabled={:s}'.format(str(device_is_enabled))) self._device_to_state[device_uuid] = new_operational_status activate_monitoring = device_was_not_enabled and device_is_enabled LOGGER.info('activate_monitoring={:s}'.format(str(activate_monitoring))) if not activate_monitoring: LOGGER.info('device not ready for activation') # device is not ready for monitoring continue LOGGER.info('checking endpoints') enabled_endpoint_names = set() for config_rule in device.device_config.config_rules: LOGGER.info(grpc_message_to_json_string(config_rule)) if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: LOGGER.info(' not set config rule') continue if config_rule.WhichOneof('config_rule') != 'custom': LOGGER.info(' not custom') continue str_resource_key = str(config_rule.custom.resource_key) LOGGER.info(' str_resource_key={:s}'.format(str(str_resource_key))) if not str_resource_key.startswith('/interface['): LOGGER.info(' not /interface') continue json_resource_value = json.loads(config_rule.custom.resource_value) LOGGER.info(' json_resource_value={:s}'.format(str(json_resource_value))) if 'name' not in json_resource_value: LOGGER.info(' missing name') continue if 'enabled' not in json_resource_value: LOGGER.info(' missing enabled') continue if not json_resource_value['enabled']: LOGGER.info(' not enabled') continue LOGGER.info(' enabled') enabled_endpoint_names.add(json_resource_value['name']) LOGGER.info('enabled_endpoint_names={:s}'.format(str(enabled_endpoint_names))) for endpoint in device.device_endpoints: LOGGER.info(' endpoint: {:s}'.format(str(endpoint.name))) if endpoint.name not in enabled_endpoint_names: LOGGER.info(' skipping not enabled') continue endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid self._name_mapping.set_endpoint_name(endpoint_uuid, endpoint.name) for value in endpoint.kpi_sample_types: if value == KPISAMPLETYPE_UNKNOWN: LOGGER.info(' unknown kpi sample type') continue kpi_descriptor = monitoring_pb2.KpiDescriptor() kpi_descriptor.kpi_description = device.device_type kpi_descriptor.kpi_sample_type = value kpi_descriptor.device_id.CopyFrom(device.device_id) # pylint: disable=no-member kpi_descriptor.endpoint_id.CopyFrom(endpoint.endpoint_id) # pylint: disable=no-member kpi_id = self._monitoring_client.SetKpi(kpi_descriptor) kpi_id_list.append(kpi_id) except queue.Empty: break return kpi_id_list except ServiceException: LOGGER.exception('ListenEvents exception') except Exception: # pragma: no cover # pylint: disable=broad-except LOGGER.exception('ListenEvents exception')