# 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 ast import threading from typing import Tuple, Any import grpc import logging from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method from common.tools.kafka.Variables import KafkaConfig, KafkaTopic from confluent_kafka import Consumer as KafkaConsumer from confluent_kafka import Producer as KafkaProducer from confluent_kafka import KafkaError from common.proto.context_pb2 import Empty from common.proto.telemetry_frontend_pb2 import CollectorId, Collector, CollectorFilter, CollectorList from common.proto.telemetry_frontend_pb2_grpc import TelemetryFrontendServiceServicer from telemetry.database.TelemetryModel import Collector as CollectorModel from telemetry.database.Telemetry_DB import TelemetryDB LOGGER = logging.getLogger(__name__) METRICS_POOL = MetricsPool('TelemetryFrontend', 'NBIgRPC') ACTIVE_COLLECTORS = [] class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer): def __init__(self): LOGGER.info('Init TelemetryFrontendService') self.DBobj = TelemetryDB() self.kafka_producer = KafkaProducer({'bootstrap.servers' : KafkaConfig.SERVER_ADDRESS.value}) self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.SERVER_ADDRESS.value, 'group.id' : 'frontend', 'auto.offset.reset' : 'latest'}) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def StartCollector(self, request : Collector, grpc_context: grpc.ServicerContext # type: ignore ) -> CollectorId: # type: ignore LOGGER.info ("gRPC message: {:}".format(request)) response = CollectorId() # TODO: Verify the presence of Kpi ID in KpiDB or assume that KPI ID already exists. self.DBobj.add_row_to_db( CollectorModel.ConvertCollectorToRow(request) ) self.PublishRequestOnKafka(request) response.collector_id.uuid = request.collector_id.collector_id.uuid # type: ignore return response def PublishRequestOnKafka(self, collector_obj): """ Method to generate collector request on Kafka. """ collector_id = collector_obj.collector_id.collector_id.uuid collector_to_generate : Tuple [str, int, int] = ( collector_obj.kpi_id.kpi_id.uuid, collector_obj.duration_s, collector_obj.interval_s ) self.kafka_producer.produce( KafkaTopic.REQUEST.value, key = collector_id, value = str(collector_to_generate), callback = self.delivery_callback ) LOGGER.info("Collector Request Generated: Collector Id: {:}, Value: {:}".format(collector_id, collector_to_generate)) ACTIVE_COLLECTORS.append(collector_id) self.kafka_producer.flush() def run_kafka_listener(self): # print ("--- STARTED: run_kafka_listener ---") threading.Thread(target=self.kafka_listener).start() return True def kafka_listener(self): """ listener for response on Kafka topic. """ # # print ("--- STARTED: kafka_listener ---") # conusmer_configs = { # 'bootstrap.servers' : KAFKA_SERVER_IP, # 'group.id' : 'frontend', # 'auto.offset.reset' : 'latest' # } # # topic_response = "topic_response" # consumerObj = KafkaConsumer(conusmer_configs) self.kafka_consumer.subscribe([KAFKA_TOPICS['response']]) # print (time.time()) while True: receive_msg = self.kafka_consumer.poll(2.0) if receive_msg is None: # print (" - Telemetry frontend listening on Kafka Topic: ", KAFKA_TOPICS['response']) # added for debugging purposes continue elif receive_msg.error(): if receive_msg.error().code() == KafkaError._PARTITION_EOF: continue else: print("Consumer error: {}".format(receive_msg.error())) break try: collector_id = receive_msg.key().decode('utf-8') if collector_id in ACTIVE_COLLECTORS: (kpi_id, kpi_value) = ast.literal_eval(receive_msg.value().decode('utf-8')) self.process_response(collector_id, kpi_id, kpi_value) else: print(f"collector id does not match.\nRespone ID: '{collector_id}' --- Active IDs: '{ACTIVE_COLLECTORS}' ") except Exception as e: print(f"No message key found: {str(e)}") continue # return None def process_response(self, collector_id: str, kpi_id: str, kpi_value: Any): if kpi_id == "-1" and kpi_value == -1: # LOGGER.info("Sucessfully terminated Collector: {:}".format(collector_id)) print ("Sucessfully terminated Collector: ", collector_id) else: print ("Frontend-Received values Collector Id:", collector_id, "-KPI:", kpi_id, "-VALUE:", kpi_value) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def delivery_callback(self, err, msg): """ Callback function to handle message delivery status. Args: err (KafkaError): Kafka error object. msg (Message): Kafka message object. """ if err: print(f'Message delivery failed: {err}') else: print(f'Message delivered to topic {msg.topic()}') @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def StopCollector(self, request : CollectorId, grpc_context: grpc.ServicerContext # type: ignore ) -> Empty: # type: ignore LOGGER.info ("gRPC message: {:}".format(request)) _collector_id = request.collector_id.uuid self.publish_to_kafka_request_topic(_collector_id, "", -1, -1) return Empty() @safe_and_metered_rpc_method(METRICS_POOL, LOGGER) def SelectCollectors(self, request : CollectorFilter, contextgrpc_context: grpc.ServicerContext # type: ignore ) -> CollectorList: # type: ignore LOGGER.info("gRPC message: {:}".format(request)) response = CollectorList() filter_to_apply = dict() filter_to_apply['kpi_id'] = request.kpi_id[0].kpi_id.uuid # filter_to_apply['duration_s'] = request.duration_s[0] try: rows = self.managementDBobj.select_with_filter(CollectorModel, **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: collector_obj = Collector() collector_obj.collector_id.collector_id.uuid = row.collector_id response.collector_list.append(collector_obj) return response except Exception as e: LOGGER.info('Unable to process response {:}'.format(e))