Debug monitoring component (05c66d71) · Commits · TFS / controller

proto/monitoring.proto

+1 −1

Original line number	Diff line number	Diff line
		@@ -94,7 +94,7 @@ message KpiId {

		message Kpi {
		KpiId kpi_id = 1;
		context.Timestamp timestamp = 2;
		string timestamp = 2;
		KpiValue kpi_value = 3;
		}

src/common/proto/init.py

deleted100644 → 0

+0 −0

Empty file deleted.

src/monitoring/Config.py

deleted100644 → 0

+0 −40

Original line number	Diff line number	Diff line
		# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
		#
		# 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

		# General settings
		LOG_LEVEL = logging.WARNING

		# gRPC settings
		GRPC_SERVICE_PORT = 7070
		GRPC_MAX_WORKERS = 10
		GRPC_GRACE_PERIOD = 60
		GRPC_SERVICE_HOST = '127.0.0.1'

		# Prometheus settings
		METRICS_PORT = 9192

		# Dependency micro-service connection settings
		CONTEXT_SERVICE_HOST = '127.0.0.1'
		CONTEXT_GRPC_SERVICE_PORT = 1010

		DEVICE_SERVICE_HOST = '127.0.0.1'
		DEVICE_GRPC_SERVICE_PORT = 2020
		DEVICE_GRPC_MAX_WORKERS = 10
		DEVICE_GRPC_GRACE_PERIOD = 60

src/monitoring/client/MonitoringClient.py

+17 −19

Original line number	Diff line number	Diff line
		@@ -16,16 +16,14 @@ import grpc, logging
		from typing import Iterator
		from common.Constants import ServiceNameEnum
		from common.Settings import get_service_host, get_service_port_grpc
		from common.proto.context_pb2 import Empty
		from common.proto.monitoring_pb2 import Kpi, KpiDescriptor, KpiId, MonitorKpiRequest
		from common.proto.monitoring_pb2_grpc import MonitoringServiceStub

		from common.tools.client.RetryDecorator import retry, delay_exponential
		from common.tools.grpc.Tools import grpc_message_to_json_string
		from monitoring.proto.context_pb2 import Empty
		from monitoring.proto.monitoring_pb2 import Kpi, KpiDescriptor, KpiId, MonitorKpiRequest, EditedKpiDescriptor, \
		from common.proto.context_pb2 import Empty
		from common.proto.monitoring_pb2 import Kpi, KpiDescriptor, KpiId, MonitorKpiRequest, EditedKpiDescriptor, \
		KpiDescriptorList, BundleKpiDescriptor, KpiQuery, KpiList, SubsDescriptor, SubscriptionID, SubsIDList, \
		AlarmDescriptor, AlarmID, AlarmIDList, AlarmResponse
		from monitoring.proto.monitoring_pb2_grpc import MonitoringServiceStub
		from common.proto.monitoring_pb2_grpc import MonitoringServiceStub

		LOGGER = logging.getLogger(__name__)
		MAX_RETRIES = 15
		@@ -176,19 +174,19 @@ class MonitoringClient:
		LOGGER.debug('GetAlarmResponseStream result: {:s}'.format(grpc_message_to_json_string(response)))
		return response

		# @RETRY_DECORATOR
		# def GetStreamKpi(self, request : KpiId) -> Iterator[Kpi]:
		# LOGGER.debug('GetStreamKpi: {:s}'.format(grpc_message_to_json_string(request)))
		# response = self.stub.GetStreamKpi(request)
		# LOGGER.debug('GetStreamKpi result: {:s}'.format(grpc_message_to_json_string(response)))
		# return response
		#
		# @RETRY_DECORATOR
		# def GetInstantKpi(self, request : KpiId) -> Kpi:
		# LOGGER.debug('GetInstantKpi: {:s}'.format(grpc_message_to_json_string(request)))
		# response = self.stub.GetInstantKpi(request)
		# LOGGER.debug('GetInstantKpi result: {:s}'.format(grpc_message_to_json_string(response)))
		# return response
		@RETRY_DECORATOR
		def GetStreamKpi(self, request : KpiId) -> Iterator[Kpi]:
		LOGGER.debug('GetStreamKpi: {:s}'.format(grpc_message_to_json_string(request)))
		response = self.stub.GetStreamKpi(request)
		LOGGER.debug('GetStreamKpi result: {:s}'.format(grpc_message_to_json_string(response)))
		return response

		@RETRY_DECORATOR
		def GetInstantKpi(self, request : KpiId) -> Kpi:
		LOGGER.debug('GetInstantKpi: {:s}'.format(grpc_message_to_json_string(request)))
		response = self.stub.GetInstantKpi(request)
		LOGGER.debug('GetInstantKpi result: {:s}'.format(grpc_message_to_json_string(response)))
		return response


		if __name__ == '__main__':

src/monitoring/service/MonitoringServiceServicerImpl.py

+47 −59

Original line number	Diff line number	Diff line
		@@ -13,36 +13,24 @@
		# limitations under the License.

		import os, grpc, logging
		import socket
		from typing import Iterator

		from prometheus_client import Summary
		from prometheus_client import Counter
		from common.Settings import get_setting
		from context.proto.context_pb2 import Empty

		from monitoring.Config import DEVICE_GRPC_SERVICE_PORT, DEVICE_SERVICE_HOST
		from monitoring.proto.kpi_sample_types_pb2 import KpiSampleType
		from monitoring.proto.monitoring_pb2 import AlarmResponse, AlarmDescriptor, AlarmIDList, SubsIDList, KpiId, \
		KpiDescriptor, KpiList, KpiQuery, SubsDescriptor, SubscriptionID, AlarmID
		from monitoring.service import SqliteTools, InfluxTools
		from monitoring.proto import monitoring_pb2
		from monitoring.proto import monitoring_pb2_grpc
		from typing import Iterator

		import os, grpc, logging
		from prometheus_client import Counter, Summary
		from common.proto import context_pb2
		from common.proto import device_pb2
		from common.proto import monitoring_pb2
		from common.proto import monitoring_pb2_grpc
		from common.Constants import ServiceNameEnum
		from common.Settings import get_setting, get_service_port_grpc, get_service_host
		from common.proto.context_pb2 import Empty
		from common.proto.device_pb2 import MonitoringSettings
		from common.proto.kpi_sample_types_pb2 import KpiSampleType
		from common.proto.monitoring_pb2_grpc import MonitoringServiceServicer
		from common.proto.monitoring_pb2 import AlarmResponse, AlarmDescriptor, AlarmIDList, SubsIDList, KpiId, \
		KpiDescriptor, KpiList, KpiQuery, SubsDescriptor, SubscriptionID, AlarmID, EditedKpiDescriptor, KpiDescriptorList, \
		BundleKpiDescriptor, MonitorKpiRequest, Kpi
		from common.rpc_method_wrapper.ServiceExceptions import ServiceException

		from context.proto import context_pb2

		from monitoring.service import SqliteTools, InfluxTools
		from device.client.DeviceClient import DeviceClient
		from device.proto import device_pb2

		from prometheus_client import Counter, Summary

		LOGGER = logging.getLogger(__name__)

		@@ -55,30 +43,30 @@ INFLUXDB_USER = os.environ.get("INFLUXDB_USER")
		INFLUXDB_PASSWORD = os.environ.get("INFLUXDB_PASSWORD")
		INFLUXDB_DATABASE = os.environ.get("INFLUXDB_DATABASE")

		DEVICESERVICE_SERVICE_HOST = get_setting('DEVICESERVICE_SERVICE_HOST', default=DEVICE_SERVICE_HOST )
		DEVICESERVICE_SERVICE_PORT_GRPC = get_setting('DEVICESERVICE_SERVICE_PORT_GRPC', default=DEVICE_GRPC_SERVICE_PORT)
		DEVICESERVICE_SERVICE_HOST = get_setting('DEVICESERVICE_SERVICE_HOST', default=get_service_host(ServiceNameEnum.DEVICE) )
		DEVICESERVICE_SERVICE_PORT_GRPC = get_setting('DEVICESERVICE_SERVICE_PORT_GRPC', default=get_service_port_grpc(ServiceNameEnum.DEVICE))


		class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceServicer):
		class MonitoringServiceServicerImpl(MonitoringServiceServicer):
		def __init__(self):
		LOGGER.info('Init monitoringService')

		# Init sqlite monitoring db
		self.sql_db = SqliteTools.SQLite('monitoring.db')
		self.deviceClient = DeviceClient(host=DEVICESERVICE_SERVICE_HOST, port=DEVICE_GRPC_SERVICE_PORT) # instantiate the client
		self.deviceClient = DeviceClient(host=DEVICESERVICE_SERVICE_HOST, port=DEVICESERVICE_SERVICE_PORT_GRPC) # instantiate the client

		# Create influx_db client
		self.influx_db = InfluxTools.Influx(INFLUXDB_HOSTNAME,"8086",INFLUXDB_USER,INFLUXDB_PASSWORD,INFLUXDB_DATABASE)

		# CreateKpi (CreateKpiRequest) returns (KpiId) {}
		def CreateKpi(
		self, request : monitoring_pb2.KpiDescriptor, grpc_context : grpc.ServicerContext
		) -> monitoring_pb2.KpiId:
		self, request : KpiDescriptor, grpc_context : grpc.ServicerContext
		) -> KpiId:
		# CREATEKPI_COUNTER_STARTED.inc()
		LOGGER.info('CreateKpi')
		try:
		# Here the code to create a sqlite query to crete a KPI and return a KpiID
		kpi_id = monitoring_pb2.KpiId()
		kpi_id = KpiId()

		kpi_description = request.kpi_description
		kpi_sample_type = request.kpi_sample_type
		@@ -102,48 +90,48 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		# CREATEKPI_COUNTER_FAILED.inc()
		grpc_context.abort(grpc.StatusCode.INTERNAL, str(e))

		def EditKpiDescriptor ( self, request : monitoring_pb2.EditedKpiDescriptor, grpc_context : grpc.ServicerContext) -> context_pb2.Empty:
		def EditKpiDescriptor ( self, request : EditedKpiDescriptor, grpc_context : grpc.ServicerContext) -> Empty:

		LOGGER.info('EditKpiDescriptor')
		try:
		# TBC
		return context_pb2.Empty()
		return Empty()
		except ServiceException as e:
		LOGGER.exception('EditKpiDescriptor exception')
		grpc_context.abort(e.code, e.details)
		except Exception as e: # pragma: no cover
		LOGGER.exception('EditKpiDescriptor exception')

		def DeleteKpi ( self, request : monitoring_pb2.KpiId, grpc_context : grpc.ServicerContext) -> context_pb2.Empty:
		def DeleteKpi ( self, request : KpiId, grpc_context : grpc.ServicerContext) -> Empty:

		LOGGER.info('DeleteKpi')
		try:
		# TBC
		return context_pb2.Empty()
		return Empty()
		except ServiceException as e:
		LOGGER.exception('DeleteKpi exception')
		grpc_context.abort(e.code, e.details)
		except Exception as e: # pragma: no cover
		LOGGER.exception('DeleteKpi exception')

		def GetKpiDescriptorList ( self, request : context_pb2.Empty, grpc_context : grpc.ServicerContext) -> monitoring_pb2.KpiDescriptorList:
		def GetKpiDescriptorList ( self, request : Empty, grpc_context : grpc.ServicerContext) -> KpiDescriptorList:

		LOGGER.info('GetKpiDescriptorList')
		try:
		# TBC
		return monitoring_pb2.KpiDescriptorList()
		return KpiDescriptorList()
		except ServiceException as e:
		LOGGER.exception('GetKpiDescriptorList exception')
		grpc_context.abort(e.code, e.details)
		except Exception as e: # pragma: no cover
		LOGGER.exception('GetKpiDescriptorList exception')

		def CreateBundleKpi ( self, request : monitoring_pb2.BundleKpiDescriptor, grpc_context : grpc.ServicerContext) -> monitoring_pb2.KpiId:
		def CreateBundleKpi ( self, request : BundleKpiDescriptor, grpc_context : grpc.ServicerContext) -> KpiId:

		LOGGER.info('CreateBundleKpi')
		try:
		# TBC
		return monitoring_pb2.KpiId()
		return KpiId()
		except ServiceException as e:
		LOGGER.exception('CreateBundleKpi exception')
		grpc_context.abort(e.code, e.details)
		@@ -152,19 +140,19 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService


		# rpc MonitorKpi (MonitorKpiRequest) returns (context.Empty) {}
		def MonitorKpi ( self, request : monitoring_pb2.MonitorKpiRequest, grpc_context : grpc.ServicerContext) -> context_pb2.Empty:
		def MonitorKpi ( self, request : MonitorKpiRequest, grpc_context : grpc.ServicerContext) -> Empty:

		LOGGER.info('MonitorKpi')
		try:
		# Creates the request to send to the device service
		monitor_device_request = device_pb2.MonitoringSettings()
		monitor_device_request = MonitoringSettings()

		kpiDescriptor = self.GetKpiDescriptor(request.kpi_id, grpc_context)

		monitor_device_request.kpi_descriptor.CopyFrom(kpiDescriptor)
		monitor_device_request.kpi_id.kpi_id.uuid = request.kpi_id.kpi_id.uuid
		monitor_device_request.sampling_duration_s = request.sampling_duration_s
		monitor_device_request.sampling_interval_s = request.sampling_interval_s
		monitor_device_request.sampling_duration_s = request.monitoring_window_s
		monitor_device_request.sampling_interval_s = request.sampling_rate_s

		device_client = DeviceClient()
		device_client.MonitorDeviceKpi(monitor_device_request)
		@@ -178,10 +166,10 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		grpc_context.abort(grpc.StatusCode.INTERNAL, str(e))
		# CREATEKPI_COUNTER_FAILED.inc()

		return context_pb2.Empty()
		return Empty()

		# rpc IncludeKpi(IncludeKpiRequest) returns(context.Empty) {}
		def IncludeKpi(self, request : monitoring_pb2.Kpi, grpc_context : grpc.ServicerContext) -> context_pb2.Empty:
		def IncludeKpi(self, request : Kpi, grpc_context : grpc.ServicerContext) -> Empty:

		LOGGER.info('IncludeKpi')

		@@ -189,7 +177,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		kpiDescriptor = self.GetKpiDescriptor(request.kpi_id, grpc_context)
		if kpiDescriptor is None:
		LOGGER.warning('Ignoring sample with KPIId({:s}): not found in database'.format(str(request.kpi_id)))
		return context_pb2.Empty()
		return Empty()

		kpiSampleType = KpiSampleType.Name(kpiDescriptor.kpi_sample_type).upper().replace('KPISAMPLETYPE_', '')
		kpiId = request.kpi_id.kpi_id.uuid
		@@ -211,7 +199,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		except Exception: # pragma: no cover
		LOGGER.exception('IncludeKpi exception')
		# CREATEKPI_COUNTER_FAILED.inc()
		return context_pb2.Empty()
		return Empty()

		# def GetStreamKpi ( self, request, grpc_context : grpc.ServicerContext):
		#
		@@ -233,7 +221,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		#LOGGER.info('kpi_db={:s}'.format(str(kpi_db)))
		if kpi_db is None: return None

		kpiDescriptor = monitoring_pb2.KpiDescriptor()
		kpiDescriptor = KpiDescriptor()

		kpiDescriptor.kpi_description = kpi_db[1]
		kpiDescriptor.kpi_sample_type = kpi_db[2]
		@@ -254,7 +242,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('QueryKpiData')
		try:
		# TBC
		return monitoring_pb2.KpiQuery()
		return KpiQuery()
		except ServiceException as e:
		LOGGER.exception('QueryKpiData exception')
		grpc_context.abort(e.code, e.details)
		@@ -266,7 +254,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('SubscribeKpi')
		try:
		# TBC
		yield monitoring_pb2.KpiList()
		yield KpiList()
		except ServiceException as e:
		LOGGER.exception('SubscribeKpi exception')
		grpc_context.abort(e.code, e.details)
		@@ -279,7 +267,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('GetSubsDescriptor')
		try:
		# TBC
		return monitoring_pb2.SubsDescriptor()
		return SubsDescriptor()
		except ServiceException as e:
		LOGGER.exception('GetSubsDescriptor exception')
		grpc_context.abort(e.code, e.details)
		@@ -291,7 +279,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('GetSubscriptions')
		try:
		# TBC
		return monitoring_pb2.SubsIDList()
		return SubsIDList()
		except ServiceException as e:
		LOGGER.exception('GetSubscriptions exception')
		grpc_context.abort(e.code, e.details)
		@@ -303,7 +291,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('EditKpiSubscription')
		try:
		# TBC
		return context_pb2.Empty()
		return Empty()
		except ServiceException as e:
		LOGGER.exception('EditKpiSubscription exception')
		grpc_context.abort(e.code, e.details)
		@@ -315,7 +303,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('CreateKpiAlarm')
		try:
		# TBC
		return monitoring_pb2.AlarmResponse()
		return AlarmResponse()
		except ServiceException as e:
		LOGGER.exception('CreateKpiAlarm exception')
		grpc_context.abort(e.code, e.details)
		@@ -327,19 +315,19 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('EditKpiAlarm')
		try:
		# TBC
		return context_pb2.Empty()
		return Empty()
		except ServiceException as e:
		LOGGER.exception('EditKpiAlarm exception')
		grpc_context.abort(e.code, e.details)
		except Exception as e: # pragma: no cover
		LOGGER.exception('EditKpiAlarm exception')

		def GetAlarms ( self, request : context_pb2.Empty, grpc_context : grpc.ServicerContext) -> AlarmIDList:
		def GetAlarms ( self, request : Empty, grpc_context : grpc.ServicerContext) -> AlarmIDList:

		LOGGER.info('GetAlarms')
		try:
		# TBC
		return monitoring_pb2.AlarmIDList()
		return AlarmIDList()
		except ServiceException as e:
		LOGGER.exception('GetAlarms exception')
		grpc_context.abort(e.code, e.details)
		@@ -351,7 +339,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('GetAlarmDescriptor')
		try:
		# TBC
		return monitoring_pb2.AlarmDescriptor()
		return AlarmDescriptor()
		except ServiceException as e:
		LOGGER.exception('GetAlarmDescriptor exception')
		grpc_context.abort(e.code, e.details)
		@@ -363,7 +351,7 @@ class MonitoringServiceServicerImpl(monitoring_pb2_grpc.MonitoringServiceService
		LOGGER.info('GetAlarmResponseStream')
		try:
		# TBC
		yield monitoring_pb2.AlarmResponse()
		yield AlarmResponse()
		except ServiceException as e:
		LOGGER.exception('GetAlarmResponseStream exception')
		grpc_context.abort(e.code, e.details)