# 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 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() @staticmethod def create_database_if_not_exist(engine: sqlalchemy.engine.Engine) -> None: if not sqlalchemy_utils.database_exists(engine.url): sqlalchemy_utils.create_database(engine.url) @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_id.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 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 ) -> 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['device_id'] = request.device_id[0].device_uuid.uuid filter_to_apply['kpi_sample_type'] = request.kpi_sample_type[0] 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 response.kpi_descriptor_list.append(kpiDescriptor_obj) return response except Exception as e: LOGGER.info('Unable to process response {:}'.format(e)) # @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) # def DeleteKpiDescriptor(self, request: KpiId, grpc_context: grpc.ServicerContext) -> Empty: # type: ignore # kpi_id = int(request.kpi_id.uuid) # kpi = self.management_db.get_KPI(kpi_id) # if kpi: # self.management_db.delete_KPI(kpi_id) # else: # LOGGER.info('DeleteKpi error: KpiID({:s}): not found in database'.format(str(kpi_id))) # return Empty() # @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) # def GetKpiDescriptor(self, request: KpiId, grpc_context: grpc.ServicerContext) -> KpiDescriptor: # type: ignore # kpi_id = request.kpi_id.uuid # kpi_db = self.management_db.get_KPI(int(kpi_id)) # kpiDescriptor = KpiDescriptor() # if kpi_db is None: # LOGGER.info('GetKpiDescriptor error: KpiID({:s}): not found in database'.format(str(kpi_id))) # else: # kpiDescriptor.kpi_description = kpi_db[1] # kpiDescriptor.kpi_sample_type = kpi_db[2] # kpiDescriptor.device_id.device_uuid.uuid = str(kpi_db[3]) # kpiDescriptor.endpoint_id.endpoint_uuid.uuid = str(kpi_db[4]) # kpiDescriptor.service_id.service_uuid.uuid = str(kpi_db[5]) # kpiDescriptor.slice_id.slice_uuid.uuid = str(kpi_db[6]) # kpiDescriptor.connection_id.connection_uuid.uuid = str(kpi_db[7]) # kpiDescriptor.link_id.link_uuid.uuid = str(kpi_db[8]) # return kpiDescriptor # @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) # def SelectKpiDescriptor(self, request: KpiDescriptorFilter, grpc_context: grpc.ServicerContext) -> KpiDescriptorList: # type: ignore # kpi_descriptor_list = KpiDescriptorList() # data = self.management_db.get_KPIS() # LOGGER.debug(f"data: {data}") # for item in data: # kpi_descriptor = KpiDescriptor() # kpi_descriptor.kpi_id.kpi_id.uuid = str(item[0]) # kpi_descriptor.kpi_description = item[1] # kpi_descriptor.kpi_sample_type = item[2] # kpi_descriptor.device_id.device_uuid.uuid = str(item[3]) # kpi_descriptor.endpoint_id.endpoint_uuid.uuid = str(item[4]) # kpi_descriptor.service_id.service_uuid.uuid = str(item[5]) # kpi_descriptor.slice_id.slice_uuid.uuid = str(item[6]) # kpi_descriptor.connection_id.connection_uuid.uuid = str(item[7]) # kpi_descriptor.link_id.link_uuid.uuid = str(item[8]) # kpi_descriptor_list.kpi_descriptor_list.append(kpi_descriptor) # return kpi_descriptor_list