Updated Analytics backend Service and streamer class (04e1cf6c) · Commits · TFS / controller

proto/kpi_sample_types.proto

+10 −0

Original line number	Diff line number	Diff line
		@@ -39,4 +39,14 @@ enum KpiSampleType {
		KPISAMPLETYPE_L3_SECURITY_STATUS_CRYPTO = 605;

		KPISAMPLETYPE_SERVICE_LATENCY_MS = 701;

		// output KPIs
		KPISAMPLETYPE_PACKETS_TRANSMITTED_AGG_OUTPUT = 1101;
		KPISAMPLETYPE_PACKETS_RECEIVED_AGG_OUTPUT = 1102;
		KPISAMPLETYPE_PACKETS_DROPPED_AGG_OUTPUT = 1103;
		KPISAMPLETYPE_BYTES_TRANSMITTED_AGG_OUTPUT = 1201;
		KPISAMPLETYPE_BYTES_RECEIVED_AGG_OUTPUT = 1202;
		KPISAMPLETYPE_BYTES_DROPPED_AGG_OUTPUT = 1203;

		KPISAMPLETYPE_SERVICE_LATENCY_MS_AGG_OUTPUT = 1701;
		}

src/analytics/backend/service/AnalyticsBackendService.py

+57 −29

Original line number	Diff line number	Diff line
		@@ -17,17 +17,14 @@ import json
		import logging
		import threading

		import pytz
		from common.tools.service.GenericGrpcService import GenericGrpcService
		from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
		from confluent_kafka import Consumer as KafkaConsumer
		from confluent_kafka import Consumer
		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
		from analytics.backend.service.AnalyzerHelper import AnalyzerHelper


		LOGGER = logging.getLogger(__name__)
		@@ -35,16 +32,25 @@ logging.basicConfig(level=logging.INFO, format=' %(levelname)s - %(message)s')

		class AnalyticsBackendService(GenericGrpcService):
		"""
		Class listens for ...
		AnalyticsBackendService class is responsible for handling the requests from the AnalyticsFrontendService.
		It listens to the Kafka topic for the requests and starts/stops the DaskStreamer accordingly.
		It also initializes the Kafka producer and Dask cluster for the streamer.
		"""
		def __init__(self, cls_name : str = __name__) -> None:
		def __init__(self, cls_name : str = __name__, n_workers=1, threads_per_worker=1
		) -> 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(),
		'group.id' : 'analytics-frontend',
		'auto.offset.reset' : 'latest'})
		self.central_producer = AnalyzerHelper.initialize_kafka_producer() # Multi-threaded producer
		self.cluster = AnalyzerHelper.initialize_dask_cluster(
		n_workers, threads_per_worker) # Local cluster
		self.request_consumer = Consumer({
		'bootstrap.servers' : KafkaConfig.get_kafka_address(),
		'group.id' : 'analytics-backend',
		'auto.offset.reset' : 'latest',
		})


		def install_servicers(self):
		threading.Thread(
		@@ -58,7 +64,7 @@ class AnalyticsBackendService(GenericGrpcService):
		"""
		LOGGER.info("Request Listener is initiated ...")
		# print ("Request Listener is initiated ...")
		consumer = self.kafka_consumer
		consumer = self.request_consumer
		consumer.subscribe([KafkaTopic.ANALYTICS_REQUEST.value])
		while True:
		receive_msg = consumer.poll(2.0)
		@@ -69,23 +75,27 @@ class AnalyticsBackendService(GenericGrpcService):
		continue
		else:
		LOGGER.error("Consumer error: {:}".format(receive_msg.error()))
		# print ("Consumer error: {:}".format(receive_msg.error()))
		break
		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))

		if analyzer["algo_name"] is None and analyzer["oper_mode"] is None:
		self.StopStreamer(analyzer_uuid)
		if self.StopStreamer(analyzer_uuid):
		LOGGER.info("Dask Streamer stopped.")
		else:
		LOGGER.error("Failed to stop Dask Streamer.")
		else:
		if self.StartStreamer(analyzer_uuid, analyzer):
		LOGGER.info("Dask Streamer started.")
		else:
		self.StartStreamer(analyzer_uuid, analyzer)
		LOGGER.error("Failed to start Dask Streamer.")
		except Exception as e:
		LOGGER.warning("Unable to consume message from topic: {:}. ERROR: {:}".format(KafkaTopic.ANALYTICS_REQUEST.value, e))


		def StartStreamer(self, analyzer_uuid : str, analyzer : json):
		def StartStreamer(self, analyzer_uuid : str, analyzer : dict):
		"""
		Start the DaskStreamer with the given parameters.
		"""
		@@ -94,28 +104,30 @@ class AnalyticsBackendService(GenericGrpcService):
		return False
		try:
		streamer = DaskStreamer(
		analyzer_uuid,
		analyzer['input_kpis' ],
		analyzer['output_kpis'],
		analyzer['thresholds' ],
		analyzer['batch_size' ],
		analyzer['window_size'],
		key = analyzer_uuid,
		input_kpis = analyzer['input_kpis' ],
		output_kpis = analyzer['output_kpis'],
		thresholds = analyzer['thresholds' ],
		batch_size = analyzer['batch_size' ],
		window_size = analyzer['window_size'],
		cluster_instance = self.cluster,
		producer_instance = self.central_producer,
		)
		streamer.start()
		logging.info(f"Streamer started with analyzer Id: {analyzer_uuid}")
		LOGGER.info(f"Streamer started with analyzer Id: {analyzer_uuid}")

		# Stop the streamer after the given duration
		if analyzer['duration'] is not None:
		if analyzer['duration'] > 0:
		def stop_after_duration():
		time.sleep(analyzer['duration'])
		logging.info(f"Stopping streamer with analyzer: {analyzer_uuid}")
		streamer.stop()
		LOGGER.warning(f"Execution duration completed of Analyzer: {analyzer_uuid}")
		if not self.StopStreamer(analyzer_uuid):
		LOGGER.warning("Failed to stop Dask Streamer. Streamer may be already terminated.")

		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))
		@@ -129,14 +141,30 @@ class AnalyticsBackendService(GenericGrpcService):
		if analyzer_uuid not in self.active_streamers:
		LOGGER.warning("Dask Streamer not found with the given analyzer_uuid: {:}".format(analyzer_uuid))
		return False
		LOGGER.info(f"Stopping streamer with key: {analyzer_uuid}")
		LOGGER.info(f"Terminating streamer with Analyzer Id: {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.")
		LOGGER.info(f"Streamer with analyzer_uuid '{analyzer_uuid}' has been trerminated sucessfully.")
		return True
		except Exception as e:
		LOGGER.error("Failed to stop Dask Streamer. ERROR: {:}".format(e))
		return False

		def close(self): # TODO: Is this function needed?
		"""
		Close the producer and cluster cleanly.
		"""
		if self.central_producer:
		try:
		self.central_producer.flush()
		LOGGER.info("Kafka producer flushed and closed.")
		except Exception as e:
		LOGGER.error(f"Error closing Kafka producer: {e}")
		if self.cluster:
		try:
		self.cluster.close()
		LOGGER.info("Dask cluster closed.")
		except Exception as e:
		LOGGER.error(f"Error closing Dask cluster: {e}")

src/analytics/backend/service/AnalyzerHandlers.py

+8 −7

Original line number	Diff line number	Diff line
		@@ -16,11 +16,11 @@ import logging
		from enum import Enum
		import pandas as pd


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

		class AnalyzerHandlers(Enum):

		class Handlers(Enum):
		AGGREGATION_HANDLER = "AggregationHandler"
		UNSUPPORTED_HANDLER = "UnsupportedHandler"

		@@ -56,7 +56,7 @@ def threshold_handler(key, aggregated_df, thresholds):
		aggregated_df[f"{metric_name}_TH_RAISE"] = aggregated_df[metric_name] > raise_th
		aggregated_df[f"{metric_name}_TH_FALL"] = aggregated_df[metric_name] < fail_th
		else:
		logger.warning(f"Threshold values for '{metric_name}' ({threshold_values}) are not a tuple of length 2. Skipping threshold application.")
		logger.warning(f"Threshold values for '{metric_name}' ({threshold_values}) are not a list of length 2. Skipping threshold application.")
		return aggregated_df

		def aggregation_handler(
		@@ -79,6 +79,10 @@ def aggregation_handler(
		# Filter the DataFrame to retain rows where kpi_id is in the input list (subscribed endpoints only)
		df = df[df['kpi_id'].isin(input_kpi_list)].copy()

		if df.empty:
		logger.warning(f"No data available for KPIs: {input_kpi_list}. Skipping processing.")
		return []

		# Define all possible aggregation methods
		aggregation_methods = {
		"min" : ('kpi_value', 'min'),
		@@ -108,7 +112,6 @@ def aggregation_handler(
		selected_methods = {method: aggregation_methods[method] for method in valid_task_parameters}

		# logger.info(f"2. Processing KPI: {kpi_id} with task parameters: {kpi_task_parameters}")

		kpi_df = df[df['kpi_id'] == kpi_id]

		# Check if kpi_df is not empty before applying the aggregation methods
		@@ -119,9 +122,6 @@ def aggregation_handler(

		agg_df['kpi_id'] = output_kpi_list[kpi_index]

		# if agg_df.empty:
		# logger.warning(f"No data available for KPI: {kpi_id}. Skipping threshold application.")
		# continue
		# logger.info(f"4. Applying thresholds for df: {agg_df['kpi_id']}")
		result = threshold_handler(key, agg_df, kpi_task_parameters)

		@@ -129,3 +129,4 @@ def aggregation_handler(
		else:
		logger.warning(f"No data available for KPIs: {kpi_id}. Skipping aggregation.")
		continue
		return []

src/analytics/backend/service/AnalyzerHelper.py

+17 −9

Original line number	Diff line number	Diff line
		@@ -15,12 +15,11 @@

		from dask.distributed import Client, LocalCluster
		from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
		from confluent_kafka import Consumer, Producer, KafkaException, KafkaError
		from confluent_kafka import Consumer, Producer

		import logging
		logging.basicConfig(level=logging.INFO)
		logger = logging.getLogger(__name__)
		logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(funcName)s - %(levelname)s - %(message)s')
		logging.basicConfig(level=logging.INFO, format=' %(levelname)s - %(message)s')


		class AnalyzerHelper:
		@@ -28,15 +27,24 @@ class AnalyzerHelper:
		pass

		@staticmethod
		def initialize_dask_client(n_workers=1, threads_per_worker=1):
		"""Initialize a local Dask cluster and client."""
		cluster = LocalCluster(n_workers=n_workers, threads_per_worker=threads_per_worker)
		client = Client(cluster)
		def initialize_dask_client(cluster_instance):
		"""Initialize a local Dask client."""
		if cluster_instance is None:
		logger.error("Dask Cluster is not initialized. Exiting.")
		return None
		client = Client(cluster_instance)
		logger.info(f"Dask Client Initialized: {client}")
		return client, cluster
		return client

		@staticmethod
		def initialize_dask_cluster(n_workers=1, threads_per_worker=2):
		"""Initialize a local Dask cluster"""
		cluster = LocalCluster(n_workers=n_workers, threads_per_worker=threads_per_worker)
		logger.info(f"Dask Cluster Initialized: {cluster}")
		return cluster

		@staticmethod
		def initialize_kafka_consumer():
		def initialize_kafka_consumer(): # TODO: update to receive topic and group_id as parameters
		"""Initialize the Kafka consumer."""
		consumer_conf = {
		'bootstrap.servers': KafkaConfig.get_kafka_address(),

src/analytics/backend/service/Streamer.py

+44 −39

Original line number	Diff line number	Diff line
		@@ -12,22 +12,28 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import logging
		import time
		import json
		import threading
		import logging

		from confluent_kafka import KafkaException, KafkaError
		from common.tools.kafka.Variables import KafkaTopic
		from .AnalyzerHandlers import AnalyzerHandlers, aggregation_handler
		from .AnalyzerHelper import AnalyzerHelper
		import threading
		from analytics.backend.service.AnalyzerHandlers import Handlers, aggregation_handler
		from analytics.backend.service.AnalyzerHelper import AnalyzerHelper


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


		class DaskStreamer(threading.Thread):
		def __init__(self, key, input_kpis, output_kpis, thresholds, batch_size=5,
		window_size=None, n_workers=1, threads_per_worker=1):
		def __init__(self, key, input_kpis, output_kpis, thresholds,
		batch_size = 5,
		window_size = None,
		cluster_instance = None,
		producer_instance = AnalyzerHelper.initialize_kafka_producer()
		):
		super().__init__()
		self.key = key
		self.input_kpis = input_kpis
		@@ -35,15 +41,14 @@ class DaskStreamer(threading.Thread):
		self.thresholds = thresholds
		self.window_size = window_size
		self.batch_size = batch_size
		self.n_workers = n_workers
		self.threads_per_worker = threads_per_worker
		self.running = True
		self.batch = []

		# Initialize Kafka and Dask components
		self.client, self.cluster = AnalyzerHelper.initialize_dask_client(n_workers, threads_per_worker)
		self.consumer = AnalyzerHelper.initialize_kafka_consumer()
		self.producer = AnalyzerHelper.initialize_kafka_producer()
		self.client = AnalyzerHelper.initialize_dask_client(cluster_instance)
		self.consumer = AnalyzerHelper.initialize_kafka_consumer() # Single-threaded consumer
		self.producer = producer_instance

		logger.info("Dask Streamer initialized.")

		def run(self):
		@@ -56,9 +61,12 @@ class DaskStreamer(threading.Thread):
		logger.warning("Kafka consumer is not initialized or stopped. Exiting loop.")
		break
		if not self.running:
		logger.warning("Dask Streamer is not running. Exiting loop.")
		logger.warning("Dask Streamer instance has been terminated. Exiting loop.")
		break
		if not self.client:
		logger.warning("Dask client is not running. Exiting loop.")
		break
		message = self.consumer.poll(timeout=2.0) # Poll for new messages after 2 sceonds
		message = self.consumer.poll(timeout=2.0)
		if message is None:
		# logger.info("No new messages received.")
		continue
		@@ -74,11 +82,10 @@ class DaskStreamer(threading.Thread):
		logger.error(f"Failed to decode message: {message.value()}")
		continue
		self.batch.append(value)
		# logger.info(f"Received message: {value}")

		# Window size has a priority over batch size
		# Window size has a precedence over batch size
		if self.window_size is None:
		if len(self.batch) >= self.batch_size: # If batch size is not provided, process continue with default batch size
		if len(self.batch) >= self.batch_size: # If batch size is not provided, process continue with the default batch size
		logger.info(f"Processing based on batch size {self.batch_size}.")
		self.task_handler_selector()
		self.batch = []
		@@ -94,15 +101,20 @@ class DaskStreamer(threading.Thread):
		except Exception as e:
		logger.exception(f"Error in Dask streaming process: {e}")
		finally:
		self.stop()
		logger.info(">>> Exiting Dask Streamer...")

		def task_handler_selector(self):
		"""Select the task handler based on the task type."""
		if AnalyzerHandlers.is_valid_handler(self.thresholds["task_type"]):
		if self.client.status == 'running':
		logger.info(f"Batch to be processed: {self.batch}")
		if Handlers.is_valid_handler(self.thresholds["task_type"]):
		if self.client is not None and self.client.status == 'running':
		try:
		future = self.client.submit(aggregation_handler, "batch size", self.key,
		self.batch, self.input_kpis, self.output_kpis, self.thresholds)
		future.add_done_callback(lambda fut: self.produce_result(fut.result(), KafkaTopic.ALARMS.value))
		except Exception as e:
		logger.error(f"Failed to submit task to Dask client or unable to process future. See error for detail: {e}")
		else:
		logger.warning("Dask client is not running. Skipping processing.")
		else:
		@@ -110,6 +122,9 @@ class DaskStreamer(threading.Thread):

		def produce_result(self, result, destination_topic):
		"""Produce results to the Kafka topic."""
		if not result:
		logger.warning("Nothing to produce. Skipping.")
		return
		for record in result:
		try:
		self.producer.produce(
		@@ -124,9 +139,13 @@ class DaskStreamer(threading.Thread):
		logger.info(f"Produced {len(result)} aggregated records to '{destination_topic}'.")

		def stop(self):
		"""Clean up Kafka and Dask resources."""
		logger.info("Shutting down resources...")
		"""Clean up Kafka and Dask thread resources."""
		if not self.running:
		logger.info("Dask Streamer is already stopped.")
		return
		self.running = False
		logger.info("Streamer running status is set to False. Waiting 5 seconds before stopping...")
		time.sleep(5) # Waiting time for running tasks to complete
		if self.consumer:
		try:
		self.consumer.close()
		@@ -134,13 +153,6 @@ class DaskStreamer(threading.Thread):
		except Exception as e:
		logger.error(f"Error closing Kafka consumer: {e}")

		if self.producer:
		try:
		self.producer.flush()
		logger.info("Kafka producer flushed and closed.")
		except Exception as e:
		logger.error(f"Error closing Kafka producer: {e}")

		if self.client is not None and hasattr(self.client, 'status') and self.client.status == 'running':
		try:
		self.client.close()
		@@ -148,11 +160,4 @@ class DaskStreamer(threading.Thread):
		except Exception as e:
		logger.error(f"Error closing Dask client: {e}")

		if self.cluster is not None and hasattr(self.cluster, 'close'):
		try:
		self.cluster.close(timeout=5)
		logger.info("Dask cluster closed.")
		except Exception as e:
		logger.error(f"Timeout error while closing Dask cluster: {e}")


		# TODO: May be Single streamer for all analyzers ... ?