Skip to content
Snippets Groups Projects
Commit 68e713f6 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Merge branch 'feat/188-cttc-new-analytics-component' into 'develop'

(CTTC) New Analytics Component

See merge request !261
parents f3ea42c1 a94fe1fd
No related branches found
No related tags found
2 merge requests!294Release TeraFlowSDN 4.0,!261(CTTC) New Analytics Component
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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.
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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
from analytics.database.Analyzer_DB import AnalyzerDB
LOGGER = logging.getLogger(__name__)
def test_verify_databases_and_tables():
LOGGER.info('>>> test_verify_databases_and_tables : START <<< ')
AnalyzerDBobj = AnalyzerDB()
# AnalyzerDBobj.drop_database()
# AnalyzerDBobj.verify_tables()
AnalyzerDBobj.create_database()
AnalyzerDBobj.create_tables()
AnalyzerDBobj.verify_tables()
......@@ -66,6 +66,8 @@ class ServiceNameEnum(Enum):
KPIVALUEWRITER = 'kpi-value-writer'
TELEMETRYFRONTEND = 'telemetry-frontend'
TELEMETRYBACKEND = 'telemetry-backend'
ANALYTICSFRONTEND = 'analytics-frontend'
ANALYTICSBACKEND = 'analytics-backend'
# Used for test and debugging only
DLT_GATEWAY = 'dltgateway'
......@@ -100,6 +102,8 @@ DEFAULT_SERVICE_GRPC_PORTS = {
ServiceNameEnum.KPIVALUEWRITER .value : 30030,
ServiceNameEnum.TELEMETRYFRONTEND .value : 30050,
ServiceNameEnum.TELEMETRYBACKEND .value : 30060,
ServiceNameEnum.ANALYTICSFRONTEND .value : 30080,
ServiceNameEnum.ANALYTICSBACKEND .value : 30090,
# Used for test and debugging only
ServiceNameEnum.DLT_GATEWAY .value : 50051,
......
......@@ -14,7 +14,6 @@
import logging
from enum import Enum
from confluent_kafka import KafkaException
from confluent_kafka.admin import AdminClient, NewTopic
from common.Settings import get_setting
......@@ -26,11 +25,11 @@ class KafkaConfig(Enum):
@staticmethod
def get_kafka_address() -> str:
kafka_server_address = get_setting('KFK_SERVER_ADDRESS', default=None)
if kafka_server_address is None:
KFK_NAMESPACE = get_setting('KFK_NAMESPACE')
KFK_PORT = get_setting('KFK_SERVER_PORT')
kafka_server_address = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)
# kafka_server_address = get_setting('KFK_SERVER_ADDRESS', default=None)
# if kafka_server_address is None:
KFK_NAMESPACE = get_setting('KFK_NAMESPACE')
KFK_PORT = get_setting('KFK_SERVER_PORT')
kafka_server_address = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)
return kafka_server_address
@staticmethod
......@@ -41,11 +40,14 @@ class KafkaConfig(Enum):
class KafkaTopic(Enum):
REQUEST = 'topic_request'
RESPONSE = 'topic_response'
RAW = 'topic_raw'
LABELED = 'topic_labeled'
VALUE = 'topic_value'
# TODO: Later to be populated from ENV variable.
REQUEST = 'topic_request'
RESPONSE = 'topic_response'
RAW = 'topic_raw'
LABELED = 'topic_labeled'
VALUE = 'topic_value'
ANALYTICS_REQUEST = 'topic_request_analytics'
ANALYTICS_RESPONSE = 'topic_response_analytics'
@staticmethod
def create_all_topics() -> bool:
......
......@@ -16,8 +16,6 @@ import logging, sqlalchemy
from common.Settings import get_setting
LOGGER = logging.getLogger(__name__)
# CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@127.0.0.1:{:s}/{:s}?sslmode={:s}'
CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@cockroachdb-public.{:s}.svc.cluster.local:{:s}/{:s}?sslmode={:s}'
class KpiEngine:
......@@ -33,12 +31,10 @@ class KpiEngine:
CRDB_SSLMODE = get_setting('CRDB_SSLMODE')
crdb_uri = CRDB_URI_TEMPLATE.format(
CRDB_USERNAME, CRDB_PASSWORD, CRDB_NAMESPACE, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
# crdb_uri = CRDB_URI_TEMPLATE.format(
# CRDB_USERNAME, CRDB_PASSWORD, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
try:
engine = sqlalchemy.create_engine(crdb_uri, echo=False)
LOGGER.info(' KpiDBmanager initalized with DB URL: {:}'.format(crdb_uri))
except: # pylint: disable=bare-except # pragma: no cover
LOGGER.exception('Failed to connect to database: {:s}'.format(str(crdb_uri)))
return None # type: ignore
return engine
return engine
......@@ -28,12 +28,8 @@ from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
from common.method_wrappers.Decorator import MetricsPool
from common.tools.service.GenericGrpcService import GenericGrpcService
LOGGER = logging.getLogger(__name__)
METRICS_POOL = MetricsPool('TelemetryBackend', 'backendService')
# EXPORTER_ENDPOINT = "http://10.152.183.2:9100/metrics"
class TelemetryBackendService(GenericGrpcService):
"""
......
......@@ -13,7 +13,8 @@
# limitations under the License.
import logging, signal, sys, threading
from common.Settings import get_log_level
from prometheus_client import start_http_server
from common.Settings import get_log_level, get_metrics_port
from .TelemetryBackendService import TelemetryBackendService
terminate = threading.Event()
......@@ -27,13 +28,17 @@ def main():
global LOGGER # pylint: disable=global-statement
log_level = get_log_level()
logging.basicConfig(level=log_level)
logging.basicConfig(level=log_level, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s")
LOGGER = logging.getLogger(__name__)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
LOGGER.debug('Starting...')
LOGGER.info('Starting...')
# Start metrics server
metrics_port = get_metrics_port()
start_http_server(metrics_port)
grpc_service = TelemetryBackendService()
grpc_service.start()
......@@ -41,10 +46,10 @@ def main():
# Wait for Ctrl+C or termination signal
while not terminate.wait(timeout=1.0): pass
LOGGER.debug('Terminating...')
LOGGER.info('Terminating...')
grpc_service.stop()
LOGGER.debug('Bye')
LOGGER.info('Bye')
return 0
if __name__ == '__main__':
......
......@@ -16,27 +16,25 @@ import logging, sqlalchemy
from common.Settings import get_setting
LOGGER = logging.getLogger(__name__)
# CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@127.0.0.1:{:s}/{:s}?sslmode={:s}'
CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@cockroachdb-public.{:s}.svc.cluster.local:{:s}/{:s}?sslmode={:s}'
class TelemetryEngine:
@staticmethod
def get_engine() -> sqlalchemy.engine.Engine:
crdb_uri = get_setting('CRDB_URI', default=None)
if crdb_uri is None:
CRDB_NAMESPACE = "crdb"
CRDB_SQL_PORT = "26257"
CRDB_DATABASE = "tfs-telemetry"
CRDB_USERNAME = "tfs"
CRDB_PASSWORD = "tfs123"
CRDB_SSLMODE = "require"
if crdb_uri is None:
CRDB_NAMESPACE = get_setting('CRDB_NAMESPACE')
CRDB_SQL_PORT = get_setting('CRDB_SQL_PORT')
CRDB_DATABASE = "tfs-telemetry" # TODO: define variable get_setting('CRDB_DATABASE_KPI_MGMT')
CRDB_USERNAME = get_setting('CRDB_USERNAME')
CRDB_PASSWORD = get_setting('CRDB_PASSWORD')
CRDB_SSLMODE = get_setting('CRDB_SSLMODE')
crdb_uri = CRDB_URI_TEMPLATE.format(
CRDB_USERNAME, CRDB_PASSWORD, CRDB_NAMESPACE, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
CRDB_USERNAME, CRDB_PASSWORD, CRDB_NAMESPACE, CRDB_SQL_PORT, CRDB_DATABASE, CRDB_SSLMODE)
try:
engine = sqlalchemy.create_engine(crdb_uri, echo=False)
LOGGER.info(' TelemetryDB initalized with DB URL: {:}'.format(crdb_uri))
except: # pylint: disable=bare-except # pragma: no cover
LOGGER.exception('Failed to connect to database: {:s}'.format(str(crdb_uri)))
return None # type: ignore
return engine # type: ignore
return engine
......@@ -89,7 +89,14 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
request : CollectorId, grpc_context: grpc.ServicerContext # type: ignore
) -> Empty: # type: ignore
LOGGER.info ("gRPC message: {:}".format(request))
self.PublishStopRequestOnKafka(request)
try:
collector_to_delete = request.collector_id.uuid
self.tele_db_obj.delete_db_row_by_id(
CollectorModel, "collector_id", collector_to_delete
)
self.PublishStopRequestOnKafka(request)
except Exception as e:
LOGGER.error('Unable to delete collector. Error: {:}'.format(e))
return Empty()
def PublishStopRequestOnKafka(self, collector_id):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment