# 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 logging, grpc import sqlalchemy, sqlalchemy_utils from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method from common.proto.context_pb2 import Empty from common.proto.kpi_manager_pb2_grpc import KpiManagerServiceServicer from common.proto.kpi_manager_pb2 import KpiId, KpiDescriptor, KpiDescriptorFilter, KpiDescriptorList from monitoring.service.NameMapping import NameMapping # from monitoring.service import ManagementDBTools from kpi_manager.service.database.Kpi_DB import Kpi_DB from kpi_manager.service.database.KpiModel import Kpi as KpiModel # from telemetry.database.TelemetryModel import Kpi as KpiModel from common.proto.context_pb2 import DeviceId, LinkId, ServiceId, SliceId,\ ConnectionId, EndPointId LOGGER = logging.getLogger(__name__) METRICS_POOL = MetricsPool('Monitoring', 'KpiManager') class KpiManagerServiceServicerImpl(KpiManagerServiceServicer): def __init__(self, name_mapping : NameMapping): LOGGER.info('Init KpiManagerService') self.Kpi_DBobj = Kpi_DB() @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def SetKpiDescriptor(self, request: KpiDescriptor, grpc_context: grpc.ServicerContext # type: ignore ) -> KpiId: # type: ignore response = KpiId() LOGGER.info("Received gRPC message object: {:}".format(request)) try: kpi_to_insert = KpiModel() kpi_to_insert.kpi_id = request.kpi_id.kpi_id.uuid kpi_to_insert.kpi_description = request.kpi_description kpi_to_insert.kpi_sample_type = request.kpi_sample_type kpi_to_insert.device_id = request.device_id.device_uuid.uuid kpi_to_insert.endpoint_id = request.endpoint_id.endpoint_uuid.uuid kpi_to_insert.service_id = request.service_id.service_uuid.uuid kpi_to_insert.slice_id = request.slice_id.slice_uuid.uuid kpi_to_insert.connection_id = request.connection_id.connection_uuid.uuid kpi_to_insert.link_id = request.link_id.link_uuid.uuid if(self.Kpi_DBobj.add_row_to_db(kpi_to_insert)): response.kpi_id.uuid = request.kpi_id.kpi_id.uuid # LOGGER.info("Added Row: {:}".format(response)) return response except Exception as e: LOGGER.info("Unable to create KpiModel class object. {:}".format(e)) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def GetKpiDescriptor(self, request: KpiId, grpc_context: grpc.ServicerContext # type: ignore ) -> KpiDescriptor: # type: ignore response = KpiDescriptor() LOGGER.info("Received gRPC message object: {:}".format(request)) try: kpi_id_to_search = request.kpi_id.uuid row = self.Kpi_DBobj.search_db_row_by_id(KpiModel, 'kpi_id', kpi_id_to_search) if row is not None: response.kpi_id.kpi_id.uuid = row.kpi_id response.kpi_description = row.kpi_description response.kpi_sample_type = row.kpi_sample_type response.service_id.service_uuid.uuid = row.service_id response.device_id.device_uuid.uuid = row.device_id response.slice_id.slice_uuid.uuid = row.slice_id response.endpoint_id.endpoint_uuid.uuid = row.endpoint_id response.connection_id.connection_uuid.uuid = row.connection_id response.link_id.link_uuid.uuid = row.link_id return response except Exception as e: LOGGER.info('Unable to search kpi id. {:}'.format(e)) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def DeleteKpiDescriptor(self, request: KpiId, grpc_context: grpc.ServicerContext # type: ignore ) -> Empty: # type: ignore LOGGER.info("Received gRPC message object: {:}".format(request)) try: kpi_id_to_search = request.kpi_id.uuid self.Kpi_DBobj.delete_db_row_by_id(KpiModel, 'kpi_id', kpi_id_to_search) except Exception as e: LOGGER.info('Unable to search kpi id. {:}'.format(e)) finally: return Empty() @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def SelectKpiDescriptor(self, request: KpiDescriptorFilter, grpc_context: grpc.ServicerContext # type: ignore ) -> KpiDescriptorList: # type: ignore LOGGER.info("Received gRPC message object: {:}".format(request)) response = KpiDescriptorList() # LOGGER.info("Recevied requested Object: {:}".format(request)) # re-structre the filter. create dynamic filter filter_to_apply = dict() filter_to_apply['kpi_sample_type'] = request.kpi_sample_type[0] filter_to_apply['device_id'] = request.device_id[0].device_uuid.uuid filter_to_apply['endpoint_id'] = request.endpoint_id[0].endpoint_uuid.uuid filter_to_apply['service_id'] = request.service_id[0].service_uuid.uuid filter_to_apply['slice_id'] = request.slice_id[0].slice_uuid.uuid filter_to_apply['connection_id'] = request.connection_id[0].connection_uuid.uuid filter_to_apply['link_id'] = request.link_id[0].link_uuid.uuid try: rows = self.Kpi_DBobj.select_with_filter(KpiModel, **filter_to_apply) except Exception as e: LOGGER.info('Unable to apply filter on kpi descriptor. {:}'.format(e)) try: if len(rows) != 0: for row in rows: kpiDescriptor_obj = KpiDescriptor() kpiDescriptor_obj.kpi_id.kpi_id.uuid = row.kpi_id kpiDescriptor_obj.kpi_description = row.kpi_description kpiDescriptor_obj.kpi_sample_type = row.kpi_sample_type kpiDescriptor_obj.service_id.service_uuid.uuid = row.service_id kpiDescriptor_obj.device_id.device_uuid.uuid = row.device_id kpiDescriptor_obj.slice_id.slice_uuid.uuid = row.slice_id kpiDescriptor_obj.endpoint_id.endpoint_uuid.uuid = row.endpoint_id kpiDescriptor_obj.connection_id.connection_uuid.uuid = row.connection_id response.kpi_descriptor_list.append(kpiDescriptor_obj) return response except Exception as e: LOGGER.info('Unable to process response {:}'.format(e))