Skip to content
Snippets Groups Projects
AnalyticsBackendService.py 5.83 KiB
Newer Older
# Copyright 2022-2024 ETSI 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.

Waleed Akbar's avatar
Waleed Akbar committed
import json
import logging
import threading
from common.tools.service.GenericGrpcService import GenericGrpcService
Waleed Akbar's avatar
Waleed Akbar committed
from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
from confluent_kafka import Consumer as KafkaConsumer
from confluent_kafka import KafkaError
from common.Constants import ServiceNameEnum
from common.Settings import get_service_port_grpc
from threading import Thread, Event
from analytics.backend.service.Streamer import DaskStreamer
from common.proto.analytics_frontend_pb2 import Analyzer
from datetime import datetime, timedelta

LOGGER = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format=' %(levelname)s - %(message)s')

class AnalyticsBackendService(GenericGrpcService):
    """
    Class listens for ...
    """
    def __init__(self, cls_name : str = __name__) -> None:
        LOGGER.info('Init AnalyticsBackendService')
        port = get_service_port_grpc(ServiceNameEnum.ANALYTICSBACKEND)
        super().__init__(port, cls_name=cls_name)
        self.active_streamers = {}
        self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),
Waleed Akbar's avatar
Waleed Akbar committed
                                            'group.id'           : 'analytics-frontend',
                                            'auto.offset.reset'  : 'latest'})

    def install_servicers(self):
        threading.Thread(
            target=self.RequestListener,
            args=()
        ).start()
    def RequestListener(self):
Waleed Akbar's avatar
Waleed Akbar committed
        """
        listener for requests on Kafka topic.
        """
        LOGGER.info("Request Listener is initiated ...")
        # print      ("Request Listener is initiated ...")
Waleed Akbar's avatar
Waleed Akbar committed
        consumer = self.kafka_consumer
        consumer.subscribe([KafkaTopic.ANALYTICS_REQUEST.value])
        while True:
Waleed Akbar's avatar
Waleed Akbar committed
            receive_msg = consumer.poll(2.0)
            if receive_msg is None:
                continue
            elif receive_msg.error():
                if receive_msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    LOGGER.error("Consumer error: {:}".format(receive_msg.error()))
                    # print       ("Consumer error: {:}".format(receive_msg.error()))
            try:
                analyzer      = json.loads(receive_msg.value().decode('utf-8'))
                analyzer_uuid = receive_msg.key().decode('utf-8')
                LOGGER.info('Recevied Analyzer: {:} - {:}'.format(analyzer_uuid, analyzer))
                # print       ('Recevied Analyzer: {:} - {:}'.format(analyzer_uuid, analyzer))
Waleed Akbar's avatar
Waleed Akbar committed

                if analyzer["algo_name"] is None and analyzer["oper_mode"] is None:
                    self.StopStreamer(analyzer_uuid)
                    self.StartStreamer(analyzer_uuid, analyzer)
            except Exception as e:
                LOGGER.warning("Unable to consume message from topic: {:}. ERROR: {:}".format(KafkaTopic.ANALYTICS_REQUEST.value, e))
Waleed Akbar's avatar
Waleed Akbar committed

    def StartStreamer(self, analyzer_uuid : str, analyzer : json):
        """
        Start the DaskStreamer with the given parameters.
        """
        if analyzer_uuid in self.active_streamers:
            LOGGER.warning("Dask Streamer already running with the given analyzer_uuid: {:}".format(analyzer_uuid))
            return False
            streamer = DaskStreamer(
                analyzer_uuid,
                analyzer['input_kpis' ],
                analyzer['output_kpis'],
                analyzer['thresholds' ],
                analyzer['batch_size' ],
                analyzer['window_size'],
            streamer.start()
            logging.info(f"Streamer started with analyzer Id: {analyzer_uuid}")

            # Stop the streamer after the given duration
            if analyzer['duration'] is not None:
                def stop_after_duration():
                    time.sleep(analyzer['duration'])
                    logging.info(f"Stopping streamer with analyzer: {analyzer_uuid}")
                    streamer.stop()

                duration_thread = threading.Thread(target=stop_after_duration, daemon=True)
                duration_thread.start()

            self.active_streamers[analyzer_uuid] = streamer
            LOGGER.info("Dask Streamer started.")
            return True
        except Exception as e:
            LOGGER.error("Failed to start Dask Streamer. ERROR: {:}".format(e))
    def StopStreamer(self, analyzer_uuid : str):
        """
        Stop the DaskStreamer with the given analyzer_uuid.
        """
        try:
            if analyzer_uuid not in self.active_streamers:
                LOGGER.warning("Dask Streamer not found with the given analyzer_uuid: {:}".format(analyzer_uuid))
Waleed Akbar's avatar
Waleed Akbar committed
                return False
            LOGGER.info(f"Stopping streamer with key: {analyzer_uuid}")
            streamer = self.active_streamers[analyzer_uuid]
            streamer.stop()
            streamer.join()
            del self.active_streamers[analyzer_uuid]
            LOGGER.info(f"Streamer with analyzer_uuid '{analyzer_uuid}' has been stopped.")
            return True
        except Exception as e:
            LOGGER.error("Failed to stop Dask Streamer. ERROR: {:}".format(e))
            return False