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

src/analytics/database/Analyzer_DB.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 sqlalchemy_utils
+
+from sqlalchemy     import inspect, or_
+from sqlalchemy.orm import sessionmaker
+
+from analytics.database.AnalyzerModel         import Analyzer as AnalyzerModel
+from analytics.database.AnalyzerEngine        import AnalyzerEngine
+from common.method_wrappers.ServiceExceptions import (OperationFailedException, AlreadyExistsException)
+
+LOGGER = logging.getLogger(__name__)
+DB_NAME = "tfs_analyzer"        # TODO: export name from enviornment variable
+
+class AnalyzerDB:
+    def __init__(self):
+        self.db_engine = AnalyzerEngine.get_engine()
+        if self.db_engine is None:
+            LOGGER.error('Unable to get SQLAlchemy DB Engine...')
+            return False
+        self.db_name = DB_NAME
+        self.Session = sessionmaker(bind=self.db_engine)
+
+    def create_database(self):
+        if not sqlalchemy_utils.database_exists(self.db_engine.url):
+            LOGGER.debug("Database created. {:}".format(self.db_engine.url))
+            sqlalchemy_utils.create_database(self.db_engine.url)
+
+    def drop_database(self) -> None:
+        if sqlalchemy_utils.database_exists(self.db_engine.url):
+            sqlalchemy_utils.drop_database(self.db_engine.url)
+
+    def create_tables(self):
+        try:
+            AnalyzerModel.metadata.create_all(self.db_engine)     # type: ignore
+            LOGGER.debug("Tables created in the database: {:}".format(self.db_name))
+        except Exception as e:
+            LOGGER.debug("Tables cannot be created in the database. {:s}".format(str(e)))
+            raise OperationFailedException ("Tables can't be created", extra_details=["unable to create table {:}".format(e)])
+
+    def verify_tables(self):
+        try:
+            inspect_object = inspect(self.db_engine)
+            if(inspect_object.has_table('analyzer', None)):
+                LOGGER.info("Table exists in DB: {:}".format(self.db_name))
+        except Exception as e:
+            LOGGER.info("Unable to fetch Table names. {:s}".format(str(e)))
+
+# ----------------- CURD OPERATIONS ---------------------
+
+    def add_row_to_db(self, row):
+        session = self.Session()
+        try:
+            session.add(row)
+            session.commit()
+            LOGGER.debug(f"Row inserted into {row.__class__.__name__} table.")
+            return True
+        except Exception as e:
+            session.rollback()
+            if "psycopg2.errors.UniqueViolation" in str(e):
+                LOGGER.error(f"Unique key voilation: {row.__class__.__name__} table. {str(e)}")
+                raise AlreadyExistsException(row.__class__.__name__, row,
+                                             extra_details=["Unique key voilation: {:}".format(e)] )
+            else:
+                LOGGER.error(f"Failed to insert new row into {row.__class__.__name__} table. {str(e)}")
+                raise OperationFailedException ("Deletion by column id", extra_details=["unable to delete row {:}".format(e)])
+        finally:
+            session.close()
+    
+    def search_db_row_by_id(self, model, col_name, id_to_search):
+        session = self.Session()
+        try:
+            entity = session.query(model).filter_by(**{col_name: id_to_search}).first()
+            if entity:
+                # LOGGER.debug(f"{model.__name__} ID found: {str(entity)}")
+                return entity
+            else:
+                LOGGER.debug(f"{model.__name__} ID not found, No matching row: {str(id_to_search)}")
+                print("{:} ID not found, No matching row: {:}".format(model.__name__, id_to_search))
+                return None
+        except Exception as e:
+            session.rollback()
+            LOGGER.debug(f"Failed to retrieve {model.__name__} ID. {str(e)}")
+            raise OperationFailedException ("search by column id", extra_details=["unable to search row {:}".format(e)])
+        finally:
+            session.close()
+    
+    def delete_db_row_by_id(self, model, col_name, id_to_search):
+        session = self.Session()
+        try:
+            record = session.query(model).filter_by(**{col_name: id_to_search}).first()
+            if record:
+                session.delete(record)
+                session.commit()
+                LOGGER.debug("Deleted %s with %s: %s", model.__name__, col_name, id_to_search)
+            else:
+                LOGGER.debug("%s with %s %s not found", model.__name__, col_name, id_to_search)
+                return None
+        except Exception as e:
+            session.rollback()
+            LOGGER.error("Error deleting %s with %s %s: %s", model.__name__, col_name, id_to_search, e)
+            raise OperationFailedException ("Deletion by column id", extra_details=["unable to delete row {:}".format(e)])
+        finally:
+            session.close()
+    
+    def select_with_filter(self, model, filter_object):
+        session = self.Session()
+        try:
+            query = session.query(AnalyzerModel)
+            
+            # Apply filters based on the filter_object
+            if filter_object.analyzer_id:
+                query = query.filter(AnalyzerModel.analyzer_id.in_([a.analyzer_id.uuid for a in filter_object.analyzer_id]))
+
+            if filter_object.algorithm_names:
+                query = query.filter(AnalyzerModel.algorithm_name.in_(filter_object.algorithm_names))
+
+            if filter_object.input_kpi_ids:
+                input_kpi_uuids = [k.kpi_id.uuid for k in filter_object.input_kpi_ids]
+                query = query.filter(AnalyzerModel.input_kpi_ids.op('&&')(input_kpi_uuids))
+
+            if filter_object.output_kpi_ids:
+                output_kpi_uuids = [k.kpi_id.uuid for k in filter_object.output_kpi_ids]
+                query = query.filter(AnalyzerModel.output_kpi_ids.op('&&')(output_kpi_uuids))
+
+            result = query.all()
+            # query should be added to return all rows
+            if result:
+                LOGGER.debug(f"Fetched filtered rows from {model.__name__} table with filters: {filter_object}") #  - Results: {result}
+            else:
+                LOGGER.warning(f"No matching row found in {model.__name__} table with filters: {filter_object}")
+            return result
+        except Exception as e:
+            LOGGER.error(f"Error fetching filtered rows from {model.__name__} table with filters {filter_object} ::: {e}")
+            raise OperationFailedException ("Select by filter", extra_details=["unable to apply the filter {:}".format(e)])
+        finally:
+            session.close()

src/analytics/database/__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/frontend/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/frontend
+WORKDIR /var/teraflow/analytics/frontend
+COPY src/analytics/frontend/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/frontend/. analytics/frontend/
+COPY src/analytics/database/. analytics/database/
+
+# Start the service
+ENTRYPOINT ["python", "-m", "analytics.frontend.service"]

src/analytics/frontend/__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/frontend/client/AnalyticsFrontendClient.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 grpc, logging
+from common.Constants                         import ServiceNameEnum
+from common.proto.context_pb2                 import Empty
+from common.proto.analytics_frontend_pb2_grpc import AnalyticsFrontendServiceStub
+from common.proto.analytics_frontend_pb2      import AnalyzerId, Analyzer, AnalyzerFilter, AnalyzerList
+from common.Settings                          import get_service_host, get_service_port_grpc
+from common.tools.grpc.Tools                  import grpc_message_to_json_string
+from common.tools.client.RetryDecorator       import retry, delay_exponential
+
+LOGGER = logging.getLogger(__name__)
+MAX_RETRIES = 10
+DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
+RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
+
+class AnalyticsFrontendClient:
+    def __init__(self, host=None, port=None):
+        if not host: host = get_service_host(ServiceNameEnum.ANALYTICSFRONTEND)
+        if not port: port = get_service_port_grpc(ServiceNameEnum.ANALYTICSFRONTEND)
+        self.endpoint     = '{:s}:{:s}'.format(str(host), str(port))
+        LOGGER.debug('Creating channel to {:s}...'.format(str(self.endpoint)))
+        self.channel      = None
+        self.stub         = None
+        self.connect()
+        LOGGER.debug('Channel created')
+
+    def connect(self):
+        self.channel      = grpc.insecure_channel(self.endpoint)
+        self.stub         = AnalyticsFrontendServiceStub(self.channel)
+
+    def close(self):
+        if self.channel is not None: self.channel.close()
+        self.channel      = None
+        self.stub         = None
+
+    @RETRY_DECORATOR
+    def StartAnalyzer (self, request: Analyzer) -> AnalyzerId: #type: ignore
+        LOGGER.debug('StartAnalyzer: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.StartAnalyzer(request)
+        LOGGER.debug('StartAnalyzer result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response
+
+    @RETRY_DECORATOR
+    def StopAnalyzer(self, request : AnalyzerId) -> Empty: # type: ignore
+        LOGGER.debug('StopAnalyzer: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.StopAnalyzer(request)
+        LOGGER.debug('StopAnalyzer result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response
+    
+    @RETRY_DECORATOR
+    def SelectAnalyzers(self, request : AnalyzerFilter) -> AnalyzerList: # type: ignore
+        LOGGER.debug('SelectAnalyzers: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.SelectAnalyzers(request)
+        LOGGER.debug('SelectAnalyzers result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response

src/analytics/frontend/client/__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/frontend/requirements.in

+# 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.
+
+apscheduler==3.10.4
+confluent-kafka==2.3.*
+psycopg2-binary==2.9.*
+SQLAlchemy==1.4.*
+sqlalchemy-cockroachdb==1.4.*
+SQLAlchemy-Utils==0.38.*

src/analytics/frontend/service/AnalyticsFrontendService.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.
+
+from common.Constants import ServiceNameEnum
+from common.Settings import get_service_port_grpc
+from common.tools.service.GenericGrpcService import GenericGrpcService
+from common.proto.analytics_frontend_pb2_grpc import add_AnalyticsFrontendServiceServicer_to_server
+from analytics.frontend.service.AnalyticsFrontendServiceServicerImpl import AnalyticsFrontendServiceServicerImpl
+
+class AnalyticsFrontendService(GenericGrpcService):
+    def __init__(self, cls_name: str = __name__):
+        port = get_service_port_grpc(ServiceNameEnum.ANALYTICSFRONTEND)
+        super().__init__(port, cls_name=cls_name)
+        self.analytics_frontend_servicer = AnalyticsFrontendServiceServicerImpl()
+    
+    def install_servicers(self):
+        add_AnalyticsFrontendServiceServicer_to_server(self.analytics_frontend_servicer, self.server)

src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.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, grpc, json, queue
+
+from typing          import Dict
+from confluent_kafka import Consumer as KafkaConsumer
+from confluent_kafka import Producer as KafkaProducer
+from confluent_kafka import KafkaError
+
+from common.tools.kafka.Variables             import KafkaConfig, KafkaTopic
+from common.proto.context_pb2                 import Empty
+from common.method_wrappers.Decorator         import MetricsPool, safe_and_metered_rpc_method
+from common.proto.analytics_frontend_pb2      import Analyzer, AnalyzerId, AnalyzerFilter, AnalyzerList
+from common.proto.analytics_frontend_pb2_grpc import AnalyticsFrontendServiceServicer
+from analytics.database.Analyzer_DB           import AnalyzerDB
+from analytics.database.AnalyzerModel         import Analyzer as AnalyzerModel
+from apscheduler.schedulers.background        import BackgroundScheduler
+from apscheduler.triggers.interval            import IntervalTrigger
+
+LOGGER           = logging.getLogger(__name__)
+METRICS_POOL     = MetricsPool('AnalyticsFrontend', 'NBIgRPC')
+
+class AnalyticsFrontendServiceServicerImpl(AnalyticsFrontendServiceServicer):
+    def __init__(self):
+        LOGGER.info('Init AnalyticsFrontendService')
+        self.listener_topic = KafkaTopic.ANALYTICS_RESPONSE.value
+        self.db_obj         = AnalyzerDB()
+        self.result_queue   = queue.Queue()
+        self.scheduler      = BackgroundScheduler()
+        self.kafka_producer = KafkaProducer({'bootstrap.servers' : KafkaConfig.get_kafka_address()})
+        self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),
+                                            'group.id'           : 'analytics-frontend',
+                                            'auto.offset.reset'  : 'latest'})
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def StartAnalyzer(self, 
+                       request : Analyzer, grpc_context: grpc.ServicerContext # type: ignore
+                      ) -> AnalyzerId: # type: ignore
+        LOGGER.info ("At Service gRPC message: {:}".format(request))
+        response = AnalyzerId()
+
+        self.db_obj.add_row_to_db(
+            AnalyzerModel.ConvertAnalyzerToRow(request)
+        )
+        self.PublishStartRequestOnKafka(request)
+        
+        response.analyzer_id.uuid = request.analyzer_id.analyzer_id.uuid
+        return response
+
+    def PublishStartRequestOnKafka(self, analyzer_obj):
+        """
+        Method to generate analyzer request on Kafka.
+        """
+        analyzer_uuid = analyzer_obj.analyzer_id.analyzer_id.uuid
+        analyzer_to_generate : Dict = {
+            "algo_name"       : analyzer_obj.algorithm_name,
+            "input_kpis"      : [k.kpi_id.uuid for k in analyzer_obj.input_kpi_ids],
+            "output_kpis"     : [k.kpi_id.uuid for k in analyzer_obj.output_kpi_ids],
+            "oper_mode"       : analyzer_obj.operation_mode,
+            "thresholds"      : json.loads(analyzer_obj.parameters["thresholds"]),
+            "window_size"     : analyzer_obj.parameters["window_size"],
+            "window_slider"   : analyzer_obj.parameters["window_slider"],
+            # "store_aggregate" : analyzer_obj.parameters["store_aggregate"] 
+        }
+        self.kafka_producer.produce(
+            KafkaTopic.ANALYTICS_REQUEST.value,
+            key      = analyzer_uuid,
+            value    = json.dumps(analyzer_to_generate),
+            callback = self.delivery_callback
+        )
+        LOGGER.info("Analyzer Start Request Generated: Analyzer Id: {:}, Value: {:}".format(analyzer_uuid, analyzer_to_generate))
+        self.kafka_producer.flush()
+        
+        # self.StartResponseListener(analyzer_uuid)
+
+    def StartResponseListener(self, filter_key=None):
+        """
+        Start the Kafka response listener with APScheduler and return key-value pairs periodically.
+        """
+        LOGGER.info("Starting StartResponseListener")
+        # Schedule the ResponseListener at fixed intervals
+        self.scheduler.add_job(
+            self.response_listener,
+            trigger=IntervalTrigger(seconds=5),
+            args=[filter_key], 
+            id=f"response_listener_{self.listener_topic}",
+            replace_existing=True
+        )
+        self.scheduler.start()
+        LOGGER.info(f"Started Kafka listener for topic {self.listener_topic}...")
+        try:
+            while True:
+                LOGGER.info("entering while...")
+                key, value = self.result_queue.get()  # Wait until a result is available
+                LOGGER.info("In while true ...")
+                yield key, value  # Yield the result to the calling function
+        except KeyboardInterrupt:
+            LOGGER.warning("Listener stopped manually.")
+        finally:
+            self.StopListener()
+
+    def response_listener(self, filter_key=None):
+        """
+        Poll Kafka messages and put key-value pairs into the queue.
+        """
+        LOGGER.info(f"Polling Kafka topic {self.listener_topic}...")
+
+        consumer = self.kafka_consumer
+        consumer.subscribe([self.listener_topic])
+        msg = consumer.poll(2.0)
+        if msg is None:
+            return
+        elif msg.error():
+            if msg.error().code() != KafkaError._PARTITION_EOF:
+                LOGGER.error(f"Kafka error: {msg.error()}")
+            return
+
+        try:
+            key = msg.key().decode('utf-8') if msg.key() else None
+            if filter_key is not None and key == filter_key:
+                value = json.loads(msg.value().decode('utf-8'))
+                LOGGER.info(f"Received key: {key}, value: {value}")
+                self.result_queue.put((key, value))
+            else:
+                LOGGER.info(f"Skipping message with unmatched key: {key}")
+                # value = json.loads(msg.value().decode('utf-8')) # Added for debugging
+                # self.result_queue.put((filter_key, value))             # Added for debugging
+        except Exception as e:
+            LOGGER.error(f"Error processing Kafka message: {e}")
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def StopAnalyzer(self, 
+                      request : AnalyzerId, grpc_context: grpc.ServicerContext # type: ignore
+                     ) -> Empty:  # type: ignore
+        LOGGER.info ("At Service gRPC message: {:}".format(request))
+        try:
+            analyzer_id_to_delete = request.analyzer_id.uuid
+            self.db_obj.delete_db_row_by_id(
+                AnalyzerModel, "analyzer_id", analyzer_id_to_delete
+            )
+            self.PublishStopRequestOnKafka(analyzer_id_to_delete)
+        except Exception as e:
+            LOGGER.error('Unable to delete analyzer. Error: {:}'.format(e))
+        return Empty()
+
+    def PublishStopRequestOnKafka(self, analyzer_uuid):
+        """
+        Method to generate stop analyzer request on Kafka.
+        """
+        # analyzer_uuid = analyzer_id.analyzer_id.uuid
+        analyzer_to_stop :  Dict = {
+            "algo_name"   : None,
+            "input_kpis"  : [],
+            "output_kpis" : [],
+            "oper_mode"   : None
+        }
+        self.kafka_producer.produce(
+            KafkaTopic.ANALYTICS_REQUEST.value,
+            key      = analyzer_uuid,
+            value    = json.dumps(analyzer_to_stop),
+            callback = self.delivery_callback
+        )
+        LOGGER.info("Analyzer Stop Request Generated: Analyzer Id: {:}".format(analyzer_uuid))
+        self.kafka_producer.flush()
+        self.StopListener()
+
+    def StopListener(self):
+        """
+        Gracefully stop the Kafka listener and the scheduler.
+        """
+        LOGGER.info("Stopping Kafka listener...")
+        self.scheduler.shutdown()
+        LOGGER.info("Kafka listener stopped.")
+
+    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
+    def SelectAnalyzers(self, 
+                         filter : AnalyzerFilter, contextgrpc_context: grpc.ServicerContext # type: ignore
+                        ) -> AnalyzerList:  # type: ignore
+        LOGGER.info("At Service gRPC message: {:}".format(filter))
+        response = AnalyzerList()
+        try:
+            rows = self.db_obj.select_with_filter(AnalyzerModel, filter)
+            try:
+                for row in rows:
+                    response.analyzer_list.append(
+                        AnalyzerModel.ConvertRowToAnalyzer(row)
+                    )
+                return response
+            except Exception as e:
+                LOGGER.info('Unable to process filter response {:}'.format(e))
+        except Exception as e:
+            LOGGER.error('Unable to apply filter on table {:}. ERROR: {:}'.format(AnalyzerModel.__name__, e))
+       
+
+    def delivery_callback(self, err, msg):
+        if err:
+            LOGGER.debug('Message delivery failed: {:}'.format(err))
+            print       ('Message delivery failed: {:}'.format(err))
+        # else:
+        #     LOGGER.debug('Message delivered to topic {:}'.format(msg.topic()))
+        #     print('Message delivered to topic {:}'.format(msg.topic()))

src/analytics/frontend/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/frontend/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 .AnalyticsFrontendService import AnalyticsFrontendService
+
+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 = AnalyticsFrontendService()
+    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/frontend/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/frontend/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.
+
+import uuid
+import json
+from common.proto.kpi_manager_pb2        import KpiId
+from common.proto.analytics_frontend_pb2 import ( AnalyzerOperationMode, AnalyzerId,
+                                                Analyzer, AnalyzerFilter )
+
+def create_analyzer_id():
+    _create_analyzer_id                  = AnalyzerId()
+    # _create_analyzer_id.analyzer_id.uuid = str(uuid.uuid4())
+    _create_analyzer_id.analyzer_id.uuid = "efef4d95-1cf1-43c4-9742-95c283ddd7a6"
+    return _create_analyzer_id
+
+def create_analyzer():
+    _create_analyzer                              = Analyzer()
+    # _create_analyzer.analyzer_id.analyzer_id.uuid = str(uuid.uuid4())
+    _create_analyzer.analyzer_id.analyzer_id.uuid = "efef4d95-1cf1-43c4-9742-95c283ddd7a6"
+    _create_analyzer.algorithm_name               = "Test_Aggergate_and_Threshold"
+    _create_analyzer.operation_mode               = AnalyzerOperationMode.ANALYZEROPERATIONMODE_STREAMING
+    
+    _kpi_id = KpiId()
+    # input IDs to analyze
+    _kpi_id.kpi_id.uuid              = str(uuid.uuid4())
+    _kpi_id.kpi_id.uuid              = "6e22f180-ba28-4641-b190-2287bf448888"
+    _create_analyzer.input_kpi_ids.append(_kpi_id)
+    _kpi_id.kpi_id.uuid              = str(uuid.uuid4())
+    _kpi_id.kpi_id.uuid              = "1e22f180-ba28-4641-b190-2287bf446666"
+    _create_analyzer.input_kpi_ids.append(_kpi_id)
+    _kpi_id.kpi_id.uuid              = str(uuid.uuid4())
+    _create_analyzer.input_kpi_ids.append(_kpi_id)
+    # output IDs after analysis
+    _kpi_id.kpi_id.uuid              = str(uuid.uuid4())
+    _create_analyzer.output_kpi_ids.append(_kpi_id)
+    _kpi_id.kpi_id.uuid              = str(uuid.uuid4())
+    _create_analyzer.output_kpi_ids.append(_kpi_id)
+    # parameter
+    _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)}
+    _create_analyzer.parameters['thresholds']      = json.dumps(_threshold_dict)
+    _create_analyzer.parameters['window_size']     = "60 seconds"     # Such as "10 seconds", "2 minutes", "3 hours", "4 days" or "5 weeks" 
+    _create_analyzer.parameters['window_slider']   = "30 seconds"     # should be less than window size
+    _create_analyzer.parameters['store_aggregate'] = str(False)       # TRUE to store. No implemented yet
+
+    return _create_analyzer
+
+def create_analyzer_filter():
+    _create_analyzer_filter           = AnalyzerFilter()
+
+    _analyzer_id_obj                  = AnalyzerId()
+    # _analyzer_id_obj.analyzer_id.uuid = str(uuid.uuid4())
+    _analyzer_id_obj.analyzer_id.uuid = "efef4d95-1cf1-43c4-9742-95c283ddd7a6"
+    _create_analyzer_filter.analyzer_id.append(_analyzer_id_obj)
+
+    _create_analyzer_filter.algorithm_names.append('Test_Aggergate_and_Threshold')
+
+    # _input_kpi_id_obj                 = KpiId()
+    # _input_kpi_id_obj.kpi_id.uuid     = str(uuid.uuid4())
+    # _create_analyzer_filter.input_kpi_ids.append(_input_kpi_id_obj)
+    # another input kpi Id
+    # _input_kpi_id_obj.kpi_id.uuid     = str(uuid.uuid4())
+    # _create_analyzer_filter.input_kpi_ids.append(_input_kpi_id_obj)
+
+    # _output_kpi_id_obj                = KpiId()
+    # _output_kpi_id_obj.kpi_id.uuid    = str(uuid.uuid4())
+    # _create_analyzer_filter.output_kpi_ids.append(_output_kpi_id_obj)
+    # # another output kpi Id
+    # _output_kpi_id_obj.kpi_id.uuid     = str(uuid.uuid4())
+    # _create_analyzer_filter.input_kpi_ids.append(_output_kpi_id_obj)
+
+    return _create_analyzer_filter

src/analytics/frontend/tests/test_frontend.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 os
+import time
+import json
+import pytest
+import logging
+import threading
+
+from common.Constants         import ServiceNameEnum
+from common.proto.context_pb2 import Empty
+from common.Settings          import ( get_service_port_grpc, get_env_var_name, 
+                                      ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC )
+
+from common.tools.kafka.Variables                        import KafkaTopic
+from common.proto.analytics_frontend_pb2                 import AnalyzerId, AnalyzerList
+from analytics.frontend.client.AnalyticsFrontendClient   import AnalyticsFrontendClient
+from analytics.frontend.service.AnalyticsFrontendService import AnalyticsFrontendService
+from analytics.frontend.tests.messages                   import ( create_analyzer_id, create_analyzer,
+                                                                 create_analyzer_filter )
+from analytics.frontend.service.AnalyticsFrontendServiceServicerImpl import AnalyticsFrontendServiceServicerImpl
+from apscheduler.schedulers.background                   import BackgroundScheduler
+from apscheduler.triggers.interval                       import IntervalTrigger
+
+
+###########################
+# Tests Setup
+###########################
+
+LOCAL_HOST = '127.0.0.1'
+
+ANALYTICS_FRONTEND_PORT = str(get_service_port_grpc(ServiceNameEnum.ANALYTICSFRONTEND))
+os.environ[get_env_var_name(ServiceNameEnum.ANALYTICSFRONTEND, ENVVAR_SUFIX_SERVICE_HOST     )] = str(LOCAL_HOST)
+os.environ[get_env_var_name(ServiceNameEnum.ANALYTICSFRONTEND, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(ANALYTICS_FRONTEND_PORT)
+
+LOGGER = logging.getLogger(__name__)
+
+@pytest.fixture(scope='session')
+def analyticsFrontend_service():
+    LOGGER.info('Initializing AnalyticsFrontendService...')
+
+    _service = AnalyticsFrontendService()
+    _service.start()
+
+    # yield the server, when test finishes, execution will resume to stop it
+    LOGGER.info('Yielding AnalyticsFrontendService...')
+    yield _service
+
+    LOGGER.info('Terminating AnalyticsFrontendService...')
+    _service.stop()
+
+    LOGGER.info('Terminated AnalyticsFrontendService...')
+
+@pytest.fixture(scope='session')
+def analyticsFrontend_client(analyticsFrontend_service : AnalyticsFrontendService):
+    LOGGER.info('Initializing AnalyticsFrontendClient...')
+
+    _client = AnalyticsFrontendClient()
+
+    # yield the server, when test finishes, execution will resume to stop it
+    LOGGER.info('Yielding AnalyticsFrontendClient...')
+    yield _client
+
+    LOGGER.info('Closing AnalyticsFrontendClient...')
+    _client.close()
+
+    LOGGER.info('Closed AnalyticsFrontendClient...')
+
+
+###########################
+# Tests Implementation of Analytics Frontend
+###########################
+
+# --- "test_validate_kafka_topics" should be executed 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)
+
+# ----- core funtionality test -----
+# def test_StartAnalytics(analyticsFrontend_client):
+#     LOGGER.info(' >>> test_StartAnalytic START: <<< ')
+#     response = analyticsFrontend_client.StartAnalyzer(create_analyzer())
+#     LOGGER.debug(str(response))
+#     assert isinstance(response, AnalyzerId)
+
+# To test start and stop listener together
+def test_StartStopAnalyzers(analyticsFrontend_client):
+    LOGGER.info(' >>> test_StartStopAnalyzers START: <<< ')
+    LOGGER.info('--> StartAnalyzer')
+    added_analyzer_id = analyticsFrontend_client.StartAnalyzer(create_analyzer())
+    LOGGER.debug(str(added_analyzer_id))
+    LOGGER.info(' --> Calling StartResponseListener... ')
+    class_obj = AnalyticsFrontendServiceServicerImpl()
+    response =  class_obj.StartResponseListener(added_analyzer_id.analyzer_id._uuid)
+    LOGGER.debug(response)
+    LOGGER.info("waiting for timer to comlete ...")
+    time.sleep(3)
+    LOGGER.info('--> StopAnalyzer')
+    response = analyticsFrontend_client.StopAnalyzer(added_analyzer_id)
+    LOGGER.debug(str(response))
+
+# def test_SelectAnalytics(analyticsFrontend_client):
+#     LOGGER.info(' >>> test_SelectAnalytics START: <<< ')
+#     response = analyticsFrontend_client.SelectAnalyzers(create_analyzer_filter())
+#     LOGGER.debug(str(response))
+#     assert isinstance(response, AnalyzerList)
+
+# def test_StopAnalytic(analyticsFrontend_client):
+#     LOGGER.info(' >>> test_StopAnalytic START: <<< ')
+#     response = analyticsFrontend_client.StopAnalyzer(create_analyzer_id())
+#     LOGGER.debug(str(response))
+#     assert isinstance(response, Empty)
+
+# def test_ResponseListener():
+#         LOGGER.info(' >>> test_ResponseListener START <<< ')
+#         analyzer_id = create_analyzer_id()
+#         LOGGER.debug("Starting Response Listener for Analyzer ID: {:}".format(analyzer_id.analyzer_id.uuid))
+#         class_obj = AnalyticsFrontendServiceServicerImpl()
+#         for response in class_obj.StartResponseListener(analyzer_id.analyzer_id.uuid):
+#             LOGGER.debug(response)
+#             assert isinstance(response, tuple)
\ No newline at end of file

src/analytics/requirements.in

+# 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.
+
+java==11.0.*
+pyspark==3.5.2
+confluent-kafka==2.3.*
+psycopg2-binary==2.9.*
+SQLAlchemy==1.4.*
+sqlalchemy-cockroachdb==1.4.*
+SQLAlchemy-Utils==0.38.*