scripts/run_tests_locally-kpi-value-API.sh

@@ -19,7 +19,8 @@ PROJECTDIR=`pwd`
 cd $PROJECTDIR/src
 
 RCFILE=$PROJECTDIR/coverage/.coveragerc
-
+KAFKA_IP=$(docker inspect kafka --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}")
+KFK_SERVER_ADDRESS=${KAFKA_IP}:9092
 # helpful pytest flags: --log-level=INFO -o log_cli=true --verbose --maxfail=1 --durations=0
 python3 -m pytest --log-level=DEBUG --log-cli-level=DEBUG -o log_cli=true --verbose \
     kpi_value_api/tests/test_kpi_value_api.py

scripts/run_tests_locally-telemetry-DB.sh

@@ -22,5 +22,5 @@ cd $PROJECTDIR/src
 #     kpi_manager/tests/test_unitary.py
 
 RCFILE=$PROJECTDIR/coverage/.coveragerc
-python3 -m pytest --log-cli-level=INFO --verbose \
-    telemetry/database/tests/telemetryDBtests.py
+python3 -m pytest --log-level=DEBUG --log-cli-level=debug --verbose \
+    telemetry/tests/test_telemetryDB.py

scripts/run_tests_locally-telemetry-backend.sh

@@ -24,5 +24,5 @@ cd $PROJECTDIR/src
 # python3 kpi_manager/tests/test_unitary.py
 
 RCFILE=$PROJECTDIR/coverage/.coveragerc
-python3 -m pytest --log-level=INFO --log-cli-level=INFO --verbose \
-    telemetry/backend/tests/testTelemetryBackend.py
+python3 -m pytest --log-level=INFO --log-cli-level=debug --verbose \
+    telemetry/backend/tests/test_TelemetryBackend.py

scripts/run_tests_locally-telemetry-frontend.sh

@@ -24,5 +24,5 @@ cd $PROJECTDIR/src
 # python3 kpi_manager/tests/test_unitary.py
 
 RCFILE=$PROJECTDIR/coverage/.coveragerc
-python3 -m pytest --log-level=INFO --log-cli-level=INFO --verbose \
+python3 -m pytest --log-level=DEBUG --log-cli-level=DEBUG --verbose \
     telemetry/frontend/tests/test_frontend.py

scripts/show_logs_analytics_backend.sh

+#!/bin/bash
+# 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.
+
+########################################################################################################################
+# Define your deployment settings here
+########################################################################################################################
+
+# If not already set, set the name of the Kubernetes namespace to deploy to.
+export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}
+
+########################################################################################################################
+# Automated steps start here
+########################################################################################################################
+
+kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/analyticsservice -c backend

scripts/show_logs_analytics_frontend.sh

+#!/bin/bash
+# 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.
+
+########################################################################################################################
+# Define your deployment settings here
+########################################################################################################################
+
+# If not already set, set the name of the Kubernetes namespace to deploy to.
+export TFS_K8S_NAMESPACE=${TFS_K8S_NAMESPACE:-"tfs"}
+
+########################################################################################################################
+# Automated steps start here
+########################################################################################################################
+
+kubectl --namespace $TFS_K8S_NAMESPACE logs deployment/analyticsservice -c frontend

src/analytics/README.md

+# How to locally run and test Analytic service (To be added soon)
+
+### Pre-requisets 
+The following requirements should be fulfilled before the execuation of Telemetry service.

src/telemetry/database/tests/__init__.py → src/analytics/__init__.py

@@ -10,4 +10,4 @@
 # 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.
\ No newline at end of file
+# limitations under the License.

src/analytics/backend/Dockerfile

+# 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.
+
+FROM python:3.9-slim
+
+# Install dependencies
+RUN apt-get --yes --quiet --quiet update && \
+    apt-get --yes --quiet --quiet install wget g++ git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Set Python to show logs as they occur
+ENV PYTHONUNBUFFERED=0
+
+# Download the gRPC health probe
+RUN GRPC_HEALTH_PROBE_VERSION=v0.2.0 && \
+    wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \
+    chmod +x /bin/grpc_health_probe
+
+# Get generic Python packages
+RUN python3 -m pip install --upgrade pip
+RUN python3 -m pip install --upgrade setuptools wheel
+RUN python3 -m pip install --upgrade pip-tools
+
+# Get common Python packages
+# Note: this step enables sharing the previous Docker build steps among all the Python components
+WORKDIR /var/teraflow
+COPY common_requirements.in common_requirements.in
+RUN pip-compile --quiet --output-file=common_requirements.txt common_requirements.in
+RUN python3 -m pip install -r common_requirements.txt
+
+# Add common files into working directory
+WORKDIR /var/teraflow/common
+COPY src/common/. ./
+RUN rm -rf proto
+
+# Create proto sub-folder, copy .proto files, and generate Python code
+RUN mkdir -p /var/teraflow/common/proto
+WORKDIR /var/teraflow/common/proto
+RUN touch __init__.py
+COPY proto/*.proto ./
+RUN python3 -m grpc_tools.protoc -I=. --python_out=. --grpc_python_out=. *.proto
+RUN rm *.proto
+RUN find . -type f -exec sed -i -E 's/(import\ .*)_pb2/from . \1_pb2/g' {} \;
+
+# Create component sub-folders, get specific Python packages
+RUN mkdir -p /var/teraflow/analytics/backend
+WORKDIR /var/teraflow/analytics/backend
+COPY src/analytics/backend/requirements.in requirements.in
+RUN pip-compile --quiet --output-file=requirements.txt requirements.in
+RUN python3 -m pip install -r requirements.txt
+
+# Add component files into working directory
+WORKDIR /var/teraflow
+COPY src/analytics/__init__.py analytics/__init__.py
+COPY src/analytics/backend/. analytics/backend/
+
+# Start the service
+ENTRYPOINT ["python", "-m", "analytics.backend.service"]

src/analytics/backend/__init__.py

+# 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.

src/telemetry/database/tests/managementDBtests.py → src/analytics/backend/requirements.in

@@ -12,11 +12,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-from telemetry.database.managementDB import managementDB
-from telemetry.database.tests.messages import create_collector_model_object
-
-
-def test_add_row_to_db():
-    managementDBobj = managementDB()
-    managementDBobj.add_row_to_db(create_collector_model_object())
\ No newline at end of file
+pyspark==3.5.2
+confluent-kafka==2.3.*

src/analytics/backend/service/AnalyticsBackendService.py

+# 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 json
+import logging
+import threading
+from common.tools.service.GenericGrpcService import GenericGrpcService
+from analytics.backend.service.SparkStreaming import SparkStreamer
+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
+
+
+LOGGER = logging.getLogger(__name__)
+
+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.running_threads = {}       # To keep track of all running analyzers 
+        self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),
+                                            'group.id'           : 'analytics-frontend',
+                                            'auto.offset.reset'  : 'latest'})
+
+    def StartSparkStreamer(self, analyzer_uuid, analyzer):
+        kpi_list      = analyzer['input_kpis'] 
+        oper_list     = [s.replace('_value', '') for s in list(analyzer["thresholds"].keys())]  # TODO: update this line...
+        thresholds    = analyzer['thresholds']
+        window_size   = analyzer['window_size']
+        window_slider = analyzer['window_slider']
+        print ("Received parameters: {:} - {:} - {:} - {:} - {:}".format(
+            kpi_list, oper_list, thresholds, window_size, window_slider))
+        LOGGER.debug ("Received parameters: {:} - {:} - {:} - {:} - {:}".format(
+            kpi_list, oper_list, thresholds, window_size, window_slider))
+        try:
+            stop_event = threading.Event()
+            thread = threading.Thread(target=SparkStreamer, 
+                            args=(analyzer_uuid, kpi_list, oper_list, thresholds, stop_event,
+                                  window_size, window_slider, None ))
+            self.running_threads[analyzer_uuid] = (thread, stop_event)
+            thread.start()
+            print      ("Initiated Analyzer backend: {:}".format(analyzer_uuid))
+            LOGGER.info("Initiated Analyzer backend: {:}".format(analyzer_uuid))
+            return True
+        except Exception as e:
+            print       ("Failed to initiate Analyzer backend: {:}".format(e))
+            LOGGER.error("Failed to initiate Analyzer backend: {:}".format(e))
+            return False
+
+    def StopRequestListener(self, threadInfo: tuple):
+        try:
+            thread, stop_event = threadInfo
+            stop_event.set()
+            thread.join()
+            print      ("Terminating Analytics backend RequestListener")
+            LOGGER.info("Terminating Analytics backend RequestListener")
+            return True
+        except Exception as e:
+            print       ("Failed to terminate analytics backend {:}".format(e))
+            LOGGER.error("Failed to terminate analytics backend {:}".format(e))
+            return False
+
+    def install_services(self):
+        stop_event = threading.Event()
+        thread = threading.Thread(target=self.RequestListener,
+                                  args=(stop_event,) )
+        thread.start()
+        return (thread, stop_event)
+
+    def RequestListener(self, stop_event):
+        """
+        listener for requests on Kafka topic.
+        """
+        consumer = self.kafka_consumer
+        consumer.subscribe([KafkaTopic.ANALYTICS_REQUEST.value])
+        while not stop_event.is_set():
+            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:
+                    print("Consumer error: {}".format(receive_msg.error()))
+                    break
+            analyzer    = json.loads(receive_msg.value().decode('utf-8'))
+            analyzer_uuid = receive_msg.key().decode('utf-8')
+            LOGGER.debug('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.TerminateAnalyzerBackend(analyzer_uuid)
+            else:
+                self.StartSparkStreamer(analyzer_uuid, analyzer)
+        LOGGER.debug("Stop Event activated. Terminating...")
+        print       ("Stop Event activated. Terminating...")
+
+    def TerminateAnalyzerBackend(self, analyzer_uuid):
+        if analyzer_uuid in self.running_threads:
+            try:
+                thread, stop_event = self.running_threads[analyzer_uuid]
+                stop_event.set()
+                thread.join()
+                del self.running_threads[analyzer_uuid]
+                print      ("Terminating backend (by TerminateBackend): Analyzer Id: {:}".format(analyzer_uuid))
+                LOGGER.info("Terminating backend (by TerminateBackend): Analyzer Id: {:}".format(analyzer_uuid))
+                return True
+            except Exception as e:
+                LOGGER.error("Failed to terminate. Analyzer Id: {:} - ERROR: {:}".format(analyzer_uuid, e))
+                return False
+        else:
+            print         ("Analyzer not found in active collectors. Analyzer Id: {:}".format(analyzer_uuid))
+            LOGGER.warning("Analyzer not found in active collectors: Analyzer Id: {:}".format(analyzer_uuid))           
+            # generate confirmation towards frontend

src/analytics/backend/service/SparkStreaming.py

+# 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, time
+from pyspark.sql                  import SparkSession
+from pyspark.sql.types            import StructType, StructField, StringType, DoubleType, TimestampType
+from pyspark.sql.functions        import from_json, col, window, avg, min, max, first, last, stddev, when, round
+from common.tools.kafka.Variables import KafkaConfig, KafkaTopic
+
+LOGGER = logging.getLogger(__name__)
+
+def DefiningSparkSession():
+    # Create a Spark session with specific spark verions (3.5.0)
+    return SparkSession.builder \
+            .appName("Analytics") \
+            .config("spark.sql.streaming.forceDeleteTempCheckpointLocation", "true") \
+            .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
+            .getOrCreate()
+
+def SettingKafkaConsumerParams():   # TODO:  create get_kafka_consumer() in common with inputs (bootstrap server, subscribe, startingOffset and failOnDataLoss with default values)
+    return {
+            # "kafka.bootstrap.servers": '127.0.0.1:9092',
+            "kafka.bootstrap.servers": KafkaConfig.get_kafka_address(),
+            "subscribe"              : KafkaTopic.VALUE.value,
+            "startingOffsets"        : 'latest',
+            "failOnDataLoss"         : 'false'              # Optional: Set to "true" to fail the query on data loss
+        }
+
+def DefiningRequestSchema():
+    return StructType([
+            StructField("time_stamp" ,  StringType()  , True),
+            StructField("kpi_id"     ,  StringType()  , True),
+            StructField("kpi_value"  ,  DoubleType()  , True)
+        ])
+
+def GetAggregations(oper_list):
+    # Define the possible aggregation functions
+    agg_functions = {
+        'avg'  :  round(avg    ("kpi_value"), 3) .alias("avg_value"),
+        'min'  :  round(min    ("kpi_value"), 3) .alias("min_value"),
+        'max'  :  round(max    ("kpi_value"), 3) .alias("max_value"),
+        'first':  round(first  ("kpi_value"), 3) .alias("first_value"),
+        'last' :  round(last   ("kpi_value"), 3) .alias("last_value"),
+        'stdev':  round(stddev ("kpi_value"), 3) .alias("stdev_value")
+    }
+    return [agg_functions[op] for op in oper_list if op in agg_functions]   # Filter and return only the selected aggregations
+
+def ApplyThresholds(aggregated_df, thresholds):
+    # Apply thresholds (TH-Fail and TH-RAISE) based on the thresholds dictionary on the aggregated DataFrame.
+    
+    # Loop through each column name and its associated thresholds
+    for col_name, (fail_th, raise_th) in thresholds.items():
+        # Apply TH-Fail condition (if column value is less than the fail threshold)
+        aggregated_df = aggregated_df.withColumn(
+            f"{col_name}_THRESHOLD_FAIL", 
+            when(col(col_name) < fail_th, True).otherwise(False)
+        )
+        # Apply TH-RAISE condition (if column value is greater than the raise threshold)
+        aggregated_df = aggregated_df.withColumn(
+            f"{col_name}_THRESHOLD_RAISE", 
+            when(col(col_name) > raise_th, True).otherwise(False)
+        )
+    return aggregated_df
+
+def SparkStreamer(key, kpi_list, oper_list, thresholds, stop_event,
+                  window_size=None, win_slide_duration=None, time_stamp_col=None):
+    """
+    Method to perform Spark operation Kafka stream.
+    NOTE: Kafka topic to be processesd should have atleast one row before initiating the spark session. 
+    """
+    kafka_consumer_params = SettingKafkaConsumerParams()         # Define the Kafka consumer parameters
+    schema                = DefiningRequestSchema()              # Define the schema for the incoming JSON data
+    spark                 = DefiningSparkSession()               # Define the spark session with app name and spark version
+    
+    # extra options default assignment
+    if window_size        is None: window_size        = "60 seconds"    # default
+    if win_slide_duration is None: win_slide_duration = "30 seconds"    # default
+    if time_stamp_col     is None: time_stamp_col     = "time_stamp"    # default
+    
+    try:
+        # Read data from Kafka
+        raw_stream_data = spark \
+            .readStream \
+            .format("kafka") \
+            .options(**kafka_consumer_params) \
+            .load()
+
+        # Convert the value column from Kafka to a string
+        stream_data          = raw_stream_data.selectExpr("CAST(value AS STRING)")
+        # Parse the JSON string into a DataFrame with the defined schema
+        parsed_stream_data   = stream_data.withColumn("parsed_value", from_json(col("value"), schema))
+        # Select the parsed fields
+        final_stream_data    = parsed_stream_data.select("parsed_value.*")
+        # Convert the time_stamp to proper timestamp (assuming it's in ISO format)
+        final_stream_data    = final_stream_data.withColumn(time_stamp_col, col(time_stamp_col).cast(TimestampType()))
+        # Filter the stream to only include rows where the kpi_id is in the kpi_list
+        filtered_stream_data = final_stream_data.filter(col("kpi_id").isin(kpi_list))
+         # Define a window for aggregation
+        windowed_stream_data = filtered_stream_data \
+                                .groupBy(
+                                    window( col(time_stamp_col), 
+                                           window_size, slideDuration=win_slide_duration
+                                           ),
+                                    col("kpi_id")
+                                ) \
+                                .agg(*GetAggregations(oper_list))
+        # Apply thresholds to the aggregated data
+        thresholded_stream_data = ApplyThresholds(windowed_stream_data, thresholds)
+
+        # --- This will write output on console: FOR TESTING PURPOSES
+        # Start the Spark streaming query
+        # query = thresholded_stream_data \
+        #     .writeStream \
+        #     .outputMode("update") \
+        #     .format("console") 
+
+        # --- This will write output to Kafka: ACTUAL IMPLEMENTATION
+        query = thresholded_stream_data \
+            .selectExpr(f"'{key}' AS key", "to_json(struct(*)) AS value") \
+            .writeStream \
+            .format("kafka") \
+            .option("kafka.bootstrap.servers", KafkaConfig.get_kafka_address()) \
+            .option("topic",                   KafkaTopic.ANALYTICS_RESPONSE.value) \
+            .option("checkpointLocation",      "analytics/.spark/checkpoint") \
+            .outputMode("update")
+
+        # Start the query execution
+        queryHandler = query.start()
+
+        # Loop to check for stop event flag. To be set by stop collector method.
+        while True:
+            if stop_event.is_set():
+                LOGGER.debug("Stop Event activated. Terminating in 5 seconds...")
+                print       ("Stop Event activated. Terminating in 5 seconds...")
+                time.sleep(5)
+                queryHandler.stop()
+                break
+            time.sleep(5)
+
+    except Exception as e:
+        print("Error in Spark streaming process: {:}".format(e))
+        LOGGER.debug("Error in Spark streaming process: {:}".format(e))

src/analytics/backend/service/__init__.py

+# 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.

src/analytics/backend/service/__main__.py

+# 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, signal, sys, threading
+from prometheus_client import start_http_server
+from common.Settings import get_log_level, get_metrics_port
+from .AnalyticsBackendService import AnalyticsBackendService
+
+terminate = threading.Event()
+LOGGER = None
+
+def signal_handler(signal, frame): # pylint: disable=redefined-outer-name
+    LOGGER.warning('Terminate signal received')
+    terminate.set()
+
+def main():
+    global LOGGER # pylint: disable=global-statement
+
+    log_level = get_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.info('Starting...')
+
+    # Start metrics server
+    metrics_port = get_metrics_port()
+    start_http_server(metrics_port)
+
+    grpc_service = AnalyticsBackendService()
+    grpc_service.start()
+
+    # Wait for Ctrl+C or termination signal
+    while not terminate.wait(timeout=1.0): pass
+
+    LOGGER.info('Terminating...')
+    grpc_service.stop()
+
+    LOGGER.info('Bye')
+    return 0
+
+if __name__ == '__main__':
+    sys.exit(main())

src/analytics/backend/tests/__init__.py

+# 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.

src/analytics/backend/tests/messages.py

+# 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.
+
+
+def get_kpi_id_list():
+    return ["6e22f180-ba28-4641-b190-2287bf448888", "1e22f180-ba28-4641-b190-2287bf446666"]
+
+def get_operation_list():
+    return [ 'avg', 'max' ]     # possibilities ['avg', 'min', 'max', 'first', 'last', 'stdev']
+
+def get_threshold_dict():
+    threshold_dict = {
+        'avg_value'    : (20, 30),
+        'min_value'    : (00, 10), 
+        'max_value'    : (45, 50),
+        'first_value'  : (00, 10),
+        'last_value'   : (40, 50),
+        'stdev_value'  : (00, 10),
+    }
+    # Filter threshold_dict based on the operation_list
+    return {
+        op + '_value': threshold_dict[op+'_value'] for op in get_operation_list() if op + '_value' in threshold_dict
+    }

src/analytics/backend/tests/test_backend.py

+# 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 time
+import logging
+import threading
+from common.tools.kafka.Variables import KafkaTopic
+from analytics.backend.service.AnalyticsBackendService import AnalyticsBackendService
+from analytics.backend.tests.messages import get_kpi_id_list, get_operation_list, get_threshold_dict
+
+LOGGER = logging.getLogger(__name__)
+
+
+###########################
+# Tests Implementation of Telemetry Backend
+###########################
+
+# --- "test_validate_kafka_topics" should be run before the functionality tests ---
+def test_validate_kafka_topics():
+    LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+    response = KafkaTopic.create_all_topics()
+    assert isinstance(response, bool)
+
+# def test_StartRequestListener():
+#     LOGGER.info('test_RunRequestListener')
+#     AnalyticsBackendServiceObj = AnalyticsBackendService()
+#     response = AnalyticsBackendServiceObj.StartRequestListener() # response is Tuple (thread, stop_event)
+#     LOGGER.debug(str(response)) 
+#     assert isinstance(response, tuple)
+
+# To test START and STOP communication together
+def test_StopRequestListener():
+    LOGGER.info('test_RunRequestListener')
+    LOGGER.info('Initiating StartRequestListener...')
+    AnalyticsBackendServiceObj = AnalyticsBackendService()
+    response_thread = AnalyticsBackendServiceObj.StartRequestListener() # response is Tuple (thread, stop_event)
+    # LOGGER.debug(str(response_thread))
+    time.sleep(10)
+    LOGGER.info('Initiating StopRequestListener...')
+    AnalyticsBackendServiceObj = AnalyticsBackendService()
+    response = AnalyticsBackendServiceObj.StopRequestListener(response_thread)
+    LOGGER.debug(str(response)) 
+    assert isinstance(response, bool)
+
+# To independently tests the SparkListener functionality
+# def test_SparkListener():
+#     LOGGER.info('test_RunRequestListener')
+#     AnalyticsBackendServiceObj = AnalyticsBackendService()
+#     response = AnalyticsBackendServiceObj.RunSparkStreamer(
+#         get_kpi_id_list(), get_operation_list(), get_threshold_dict()
+#         )
+#     LOGGER.debug(str(response))
+#     assert isinstance(response, bool)

src/analytics/database/AnalyzerEngine.py

+# 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, sqlalchemy
+from common.Settings import get_setting
+
+LOGGER = logging.getLogger(__name__)
+CRDB_URI_TEMPLATE = 'cockroachdb://{:s}:{:s}@cockroachdb-public.{:s}.svc.cluster.local:{:s}/{:s}?sslmode={:s}'
+
+class AnalyzerEngine:
+    @staticmethod
+    def get_engine() -> sqlalchemy.engine.Engine:
+        crdb_uri = get_setting('CRDB_URI', default=None)
+        if crdb_uri is None:
+            CRDB_NAMESPACE = get_setting('CRDB_NAMESPACE')
+            CRDB_SQL_PORT  = get_setting('CRDB_SQL_PORT')
+            CRDB_DATABASE  = "tfs-analyzer"             # 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)
+        try:
+            engine = sqlalchemy.create_engine(crdb_uri, echo=False)
+            LOGGER.info(' AnalyzerDB 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

src/analytics/database/AnalyzerModel.py

+# 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
+import enum
+
+from sqlalchemy     import Column, String, Float, Enum, BigInteger, JSON
+from sqlalchemy.orm import registry
+from common.proto   import analytics_frontend_pb2
+from common.proto   import kpi_manager_pb2
+
+from sqlalchemy.dialects.postgresql import UUID, ARRAY
+
+
+logging.basicConfig(level=logging.INFO)
+LOGGER = logging.getLogger(__name__)
+
+# Create a base class for declarative models
+Base = registry().generate_base()
+
+class AnalyzerOperationMode (enum.Enum):
+    BATCH     = analytics_frontend_pb2.AnalyzerOperationMode.ANALYZEROPERATIONMODE_BATCH
+    STREAMING = analytics_frontend_pb2.AnalyzerOperationMode.ANALYZEROPERATIONMODE_STREAMING
+
+class Analyzer(Base):
+    __tablename__ = 'analyzer'
+
+    analyzer_id           = Column( UUID(as_uuid=False)        , primary_key=True)
+    algorithm_name        = Column( String                     , nullable=False  )
+    input_kpi_ids         = Column( ARRAY(UUID(as_uuid=False)) , nullable=False  )
+    output_kpi_ids        = Column( ARRAY(UUID(as_uuid=False)) , nullable=False  )
+    operation_mode        = Column( Enum(AnalyzerOperationMode), nullable=False  )
+    parameters            = Column( JSON                       , nullable=True   )
+    batch_min_duration_s  = Column( Float                      , nullable=False  )
+    batch_max_duration_s  = Column( Float                      , nullable=False  )
+    batch_min_size        = Column( BigInteger                 , nullable=False  )
+    batch_max_size        = Column( BigInteger                 , nullable=False  )
+
+    # helps in logging the information
+    def __repr__(self):
+            return (f"<Analyzer(analyzer_id='{self.analyzer_id}'       , algorithm_name='{self.algorithm_name}', "
+                    f"input_kpi_ids={self.input_kpi_ids}               , output_kpi_ids={self.output_kpi_ids}, "
+                    f"operation_mode='{self.operation_mode}'           , parameters={self.parameters}, "
+                    f"batch_min_duration_s={self.batch_min_duration_s} , batch_max_duration_s={self.batch_max_duration_s}, "
+                    f"batch_min_size={self.batch_min_size}             , batch_max_size={self.batch_max_size})>")
+
+
+    @classmethod
+    def ConvertAnalyzerToRow(cls, request):
+        """
+        Create an instance of Analyzer table rows from a request object.
+        Args:    request: The request object containing analyzer gRPC message.
+        Returns: A row (an instance of Analyzer table) initialized with content of the request.
+        """
+        return cls(
+            analyzer_id          = request.analyzer_id.analyzer_id.uuid,
+            algorithm_name       = request.algorithm_name,
+            input_kpi_ids        = [k.kpi_id.uuid for k in request.input_kpi_ids],
+            output_kpi_ids       = [k.kpi_id.uuid for k in request.output_kpi_ids],
+            operation_mode       = AnalyzerOperationMode(request.operation_mode),   # converts integer to coresponding Enum class member
+            parameters           = dict(request.parameters),
+            batch_min_duration_s = request.batch_min_duration_s,
+            batch_max_duration_s = request.batch_max_duration_s,
+            batch_min_size       = request.batch_min_size,
+            batch_max_size       = request.batch_max_size
+        )
+
+    @classmethod
+    def ConvertRowToAnalyzer(cls, row):
+        """
+        Create and return an Analyzer gRPC message initialized with the content of a row.
+        Args: row: The Analyzer table instance (row) containing the data.
+        Returns:   An Analyzer gRPC message initialized with the content of the row.
+        """
+        # Create an instance of the Analyzer message
+        response                              = analytics_frontend_pb2.Analyzer()
+        response.analyzer_id.analyzer_id.uuid = row.analyzer_id
+        response.algorithm_name               = row.algorithm_name
+        response.operation_mode               = row.operation_mode.value
+        response.parameters.update(row.parameters)
+        
+        for input_kpi_id in row.input_kpi_ids:
+            _kpi_id = kpi_manager_pb2.KpiId()
+            _kpi_id.kpi_id.uuid = input_kpi_id
+            response.input_kpi_ids.append(_kpi_id)
+        for output_kpi_id in row.output_kpi_ids:
+            _kpi_id = kpi_manager_pb2.KpiId()
+            _kpi_id.kpi_id.uuid = output_kpi_id
+            response.output_kpi_ids.append(_kpi_id)
+
+        response.batch_min_duration_s = row.batch_min_duration_s
+        response.batch_max_duration_s = row.batch_max_duration_s
+        response.batch_min_size       = row.batch_min_size
+        response.batch_max_size       = row.batch_max_size
+        return response