KpiManagerServiceServicerImpl.py

# 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_management_pb2_grpc import KpiManagerServiceServicer
from common.proto.kpi_management_pb2 import KpiId, KpiDescriptor, KpiDescriptorFilter, KpiDescriptorList
from monitoring.service.NameMapping import NameMapping
# from monitoring.service import ManagementDBTools

from kpi_management.kpi_manager.database.service.Kpi_DB import Kpi_DB
from kpi_management.kpi_manager.database.service.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.convert_KpiDescriptor_to_row(request)
            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 = KpiModel.convert_row_to_KpiDescriptor(row)
            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 = KpiModel.convert_row_to_KpiDescriptor(row)
                    response.kpi_descriptor_list.append(kpiDescriptor_obj)
            return response
        except Exception as e:
            LOGGER.info('Unable to process response {:}'.format(e))