Updated Telemetry Backend.

- Refactor Telemetry backend service methods
- Enhance test logging for better clarity.

src/telemetry/backend/service/TelemetryBackendService.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import queue
 import json
 import time
 import logging
@@ -27,6 +26,7 @@ from common.Settings  import get_service_port_grpc
 from common.method_wrappers.Decorator        import MetricsPool
 from common.tools.kafka.Variables            import KafkaConfig, KafkaTopic
 from common.tools.service.GenericGrpcService import GenericGrpcService
+from telemetry.backend.collectors.emulated.EmulatedCollector import EmulatedCollector
 
 LOGGER             = logging.getLogger(__name__)
 METRICS_POOL       = MetricsPool('TelemetryBackend', 'backendService')
@@ -44,9 +44,7 @@ class TelemetryBackendService(GenericGrpcService):
         self.kafka_consumer = KafkaConsumer({'bootstrap.servers' : KafkaConfig.get_kafka_address(),
                                             'group.id'           : 'backend',
                                             'auto.offset.reset'  : 'latest'})
-        self.running_threads   = {}
-        self.emulatorCollector = None
-        self.metric_queue      = queue.Queue()
+        self.active_jobs = {}
 
     def install_servicers(self):
         threading.Thread(target=self.RequestListener).start()
@@ -60,49 +58,88 @@ class TelemetryBackendService(GenericGrpcService):
         consumer = self.kafka_consumer
         consumer.subscribe([KafkaTopic.TELEMETRY_REQUEST.value])
         while True:
-            receive_msg = consumer.poll(2.0)
+            receive_msg = consumer.poll(1.0)
             if receive_msg is None:
                 continue
             elif receive_msg.error():
                 if receive_msg.error().code() == KafkaError._PARTITION_EOF:
                     continue
+                elif receive_msg.error().code() == KafkaError.UNKNOWN_TOPIC_OR_PART:
+                    LOGGER.warning(f"Subscribed topic {receive_msg.topic()} does not exist. May be topic does not have any messages.")
+                    continue
                 else:
                     LOGGER.error("Consumer error: {}".format(receive_msg.error()))
                     break
             try: 
-                collector = json.loads(receive_msg.value().decode('utf-8'))
+                collector = json.loads(
+                    receive_msg.value().decode('utf-8')
+                )
                 collector_id = receive_msg.key().decode('utf-8')
                 LOGGER.debug('Recevied Collector: {:} - {:}'.format(collector_id, collector))
 
-                if collector['duration'] == -1 and collector['interval'] == -1:
-                    self.TerminateCollectorBackend(collector_id)
+                duration = collector.get('duration', -1)
+                if duration == -1 and collector['interval'] == -1:
+                    self.TerminateCollector(collector_id)
                 else:
-                    threading.Thread(target=self.InitiateCollectorBackend, 
-                                  args=(collector_id, collector)).start()
+                    LOGGER.info("Collector ID: {:} - Scheduling...".format(collector_id))
+                    if collector_id not in self.active_jobs:
+                        stop_event = threading.Event()
+                        self.active_jobs[collector_id] = stop_event
+                        threading.Thread(target = self.CollectorHandler, 
+                                    args=(
+                                        collector_id, 
+                                        collector['kpi_id'],
+                                        duration, 
+                                        collector['interval'],
+                                        stop_event
+                                    )).start()
+                        # Stop the Collector after the given duration
+                        if duration > 0:
+                            def stop_after_duration():
+                                time.sleep(duration)
+                                LOGGER.warning(f"Execution duration ({duration}) completed of Collector: {collector_id}")
+                                self.TerminateCollector(collector_id)
+                                
+                            duration_thread = threading.Thread(
+                                target=stop_after_duration, daemon=True, name=f"stop_after_duration_{collector_id}"
+                                )
+                            duration_thread.start()
+                    else:
+                        LOGGER.warning("Collector ID: {:} - Already scheduled or running".format(collector_id))
             except Exception as e:
                 LOGGER.warning("Unable to consumer message from topic: {:}. ERROR: {:}".format(KafkaTopic.TELEMETRY_REQUEST.value, e))
 
-    def InitiateCollectorBackend(self, collector_id, collector):
+    def CollectorHandler(self, collector_id, kpi_id, duration, interval, stop_event):
         """
-        Method receives collector request and initiates collecter backend.
+        Method to handle collector request.
         """
-        LOGGER.info("Initiating backend for collector: (Not Implemented... In progress ) {:s}".format(str(collector_id)))
-        # start_time = time.time()
-        # self.emulatorCollector = NetworkMetricsEmulator(
-        #     duration           = collector['duration'],
-        #     interval           = collector['interval'],
-        #     metric_queue       = self.metric_queue
-        # )
-        # self.emulatorCollector.start()
-        # self.running_threads[collector_id] = self.emulatorCollector
-
-        # while self.emulatorCollector.is_alive():
-        #     if not self.metric_queue.empty():
-        #         metric_value = self.metric_queue.get()
-        #         LOGGER.debug("Metric: {:} - Value : {:}".format(collector['kpi_id'], metric_value))
-        #         self.GenerateKpiValue(collector_id, collector['kpi_id'] , metric_value)
-        #     time.sleep(1)
-        # self.TerminateCollectorBackend(collector_id)
+        end_points : list = self.get_endpoints_from_kpi_id(kpi_id)
+        if not end_points:
+            LOGGER.warning("KPI ID: {:} - Endpoints not found. Skipping...".format(kpi_id))
+        
+        device_type : str = self.get_device_type_from_kpi_id(kpi_id)
+        
+        if device_type == "Unknown":
+            LOGGER.warning("KPI ID: {:} - Device Type not found. Skipping...".format(kpi_id))
+
+        if device_type == "EMU-Device":
+            LOGGER.info("KPI ID: {:} - Device Type: {:} - Endpoints: {:}".format(kpi_id, device_type, end_points))
+            subscription = [collector_id, end_points, duration, interval]
+            self.EmulatedCollectorHandler(subscription, kpi_id, stop_event)
+        else:
+            LOGGER.warning("KPI ID: {:} - Device Type: {:} - Not Supported".format(kpi_id, device_type))
+
+
+    def EmulatedCollectorHandler(self, subscription, kpi_id, stop_event):
+            # EmulatedCollector
+            collector = EmulatedCollector(address="127.0.0.1", port=8000)
+            collector.Connect()
+            while not stop_event.is_set():
+                # samples = collector.SubscribeState(subscription)
+                # LOGGER.debug("KPI: {:} - Value: {:}".format(kpi_id, samples))
+                # self.GenerateKpiValue(job_id, kpi_id, samples)
+                LOGGER.info("Generating KPI Values ...")
+                time.sleep(1)
 
     def GenerateKpiValue(self, collector_id: str, kpi_id: str, measured_kpi_value: Any):
         """
@@ -122,38 +159,74 @@ class TelemetryBackendService(GenericGrpcService):
         )
         producer.flush()
 
-    def TerminateCollectorBackend(self, collector_id):
+    def TerminateCollector(self, job_id):
         LOGGER.debug("Terminating collector backend...")
-        if collector_id in self.running_threads:
-            thread = self.running_threads[collector_id]
-            thread.stop()
-            del self.running_threads[collector_id]
-            LOGGER.debug("Collector backend terminated. Collector ID: {:}".format(collector_id))
-            self.GenerateCollectorTerminationSignal(collector_id, "-1", -1)          # Termination confirmation to frontend.
-        else:
-            LOGGER.warning('Backend collector {:} not found'.format(collector_id))
+        try:
+            if job_id not in self.active_jobs:  # not job_ids:
+                # self.logger.warning(f"Active jobs: {self.active_jobs}") 
+                self.logger.warning(f"No active jobs found for {job_id}. It might have already terminated.")
+            else:
+                LOGGER.info(f"Terminating job: {job_id}")
+                stop_event = self.active_jobs.pop(job_id, None)
+                if stop_event:
+                    stop_event.set()
+                    LOGGER.info(f"Job {job_id} terminated.")
+                else:
+                    LOGGER.warning(f"Job {job_id} not found in active jobs.")
+        except:
+            LOGGER.exception("Error terminating job: {:}".format(job_id))
 
-    def GenerateCollectorTerminationSignal(self, collector_id: str, kpi_id: str, measured_kpi_value: Any):
+    def get_endpoints_from_kpi_id(self, kpi_id: str) -> list:
         """
-        Method to write kpi Termination signat on TELEMETRY_RESPONSE Kafka topic
+        Method to get endpoints based on kpi_id.
         """
-        producer = self.kafka_producer
-        kpi_value : Dict = {
-            "kpi_id"    : kpi_id,
-            "kpi_value" : measured_kpi_value,
+        kpi_endpoints = {
+            '6e22f180-ba28-4641-b190-2287bf448888': {"uuid": "123e4567-e89b-12d3-a456-42661417ed06", "name": "eth0",   "type": "ethernet", "sample_types": [101, 102]},
+            '123e4567-e89b-12d3-a456-426614174001': {"uuid": "123e4567-e89b-12d3-a456-42661417ed07", "name": "eth1",   "type": "ethernet", "sample_types": []},
+            '123e4567-e89b-12d3-a456-426614174002': {"uuid": "123e4567-e89b-12d3-a456-42661417ed08", "name": "13/1/2", "type": "copper",   "sample_types": [101, 102, 201, 202]},
         }
-        producer.produce(
-            KafkaTopic.TELEMETRY_RESPONSE.value,
-            key      = collector_id,
-            value    = json.dumps(kpi_value),
-            callback = self.delivery_callback
-        )
-        producer.flush()
+        return [kpi_endpoints.get(kpi_id, {})] if kpi_id in kpi_endpoints else []
+
+    def get_device_type_from_kpi_id(self, kpi_id: str) -> str:
+        """
+        Method to get device type based on kpi_id.
+        """
+        kpi_device_types = {
+            "123e4567-e89b-12d3-a456-42661type003" : {'device_type': "PKT-Device"},
+            "123e4567-e89b-12d3-a456-42661type004" : {'device_type': "OPT-Device"},
+            "6e22f180-ba28-4641-b190-2287bf448888" : {'device_type': "EMU-Device"},
+        }
+        return kpi_device_types.get(kpi_id, {}).get('device_type', "Unknown")
+
+
+    # def TerminateCollectorBackend(self, collector_id):
+    #     LOGGER.debug("Terminating collector backend...")
+    #     if collector_id in self.running_threads:
+    #         thread = self.running_threads[collector_id]
+    #         thread.stop()
+    #         del self.running_threads[collector_id]
+    #         LOGGER.debug("Collector backend terminated. Collector ID: {:}".format(collector_id))
+    #         self.GenerateCollectorTerminationSignal(collector_id, "-1", -1)          # Termination confirmation to frontend.
+    #     else:
+    #         LOGGER.warning('Backend collector {:} not found'.format(collector_id))
+
+    # def GenerateCollectorTerminationSignal(self, collector_id: str, kpi_id: str, measured_kpi_value: Any):
+    #     """
+    #     Method to write kpi Termination signat on TELEMETRY_RESPONSE Kafka topic
+    #     """
+    #     producer = self.kafka_producer
+    #     kpi_value : Dict = {
+    #         "kpi_id"    : kpi_id,
+    #         "kpi_value" : measured_kpi_value,
+    #     }
+    #     producer.produce(
+    #         KafkaTopic.TELEMETRY_RESPONSE.value,
+    #         key      = collector_id,
+    #         value    = json.dumps(kpi_value),
+    #         callback = self.delivery_callback
+    #     )
+    #     producer.flush()
 
     def delivery_callback(self, err, msg):
         if err: 
             LOGGER.error('Message delivery failed: {:s}'.format(str(err)))
-            # print(f'Message delivery failed: {err}')
-        # else:
-        #     LOGGER.info('Message delivered to topic {:}'.format(msg.topic()))
-            # print(f'Message delivered to topic {msg.topic()}')

src/telemetry/backend/tests/test_backend.py

@@ -12,12 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import pytest
 import logging
 import time
-from typing import Dict
-from common.tools.kafka.Variables import KafkaTopic
 from telemetry.backend.service.TelemetryBackendService import TelemetryBackendService
 from .messages import create_collector_request
+from .Fixtures import context_client, device_client
+from .add_devices import load_topology
 
 LOGGER = logging.getLogger(__name__)
 
@@ -26,28 +27,42 @@ LOGGER = logging.getLogger(__name__)
 # Tests Implementation of Telemetry Backend
 ###########################
 
+@pytest.fixture(autouse=True)
+def log_all_methods(request):
+    '''
+    This fixture logs messages before and after each test function runs, indicating the start and end of the test.
+    The autouse=True parameter ensures that this logging happens automatically for all tests in the module.
+    '''
+    LOGGER.info(f" >>>>> Starting test: {request.node.name} ")
+    yield
+    LOGGER.info(f" <<<<< Finished test: {request.node.name} ")
+
+@pytest.fixture
+def telemetryBackend_service():
+    LOGGER.info('Initializing TelemetryBackendService...')
+
+    _service = TelemetryBackendService()
+    _service.start()
+
+    LOGGER.info('Yielding TelemetryBackendService...')
+    yield _service
+
+    LOGGER.info('Terminating TelemetryBackendService...')
+    _service.stop()
+    LOGGER.info('Terminated TelemetryBackendService...')
+
+
+def test_InitiateCollectorBackend(telemetryBackend_service):
+    LOGGER.info(" Backend Initiated Successfully. Waiting for timer to finish ...")
+    time.sleep(300)
+    LOGGER.info(" Backend Timer Finished Successfully. ")
+
 # --- "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_RunRequestListener():
-#     LOGGER.info('test_RunRequestListener')
-#     TelemetryBackendServiceObj = TelemetryBackendService()
-#     threading.Thread(target=TelemetryBackendServiceObj.RequestListener).start()
-
-def test_RunInitiateCollectorBackend():
-    LOGGER.debug(">>> RunInitiateCollectorBackend <<<")
-    collector_obj = create_collector_request()
-    collector_id = collector_obj.collector_id.collector_id.uuid
-    collector_dict :  Dict = {
-        "kpi_id"  : collector_obj.kpi_id.kpi_id.uuid,
-        "duration": collector_obj.duration_s,
-        "interval": collector_obj.interval_s
-    }
-    TeleObj = TelemetryBackendService()
-    TeleObj.InitiateCollectorBackend(collector_id, collector_dict)
-    time.sleep(20)
-
-    LOGGER.debug("--- Execution Finished Sucessfully---")
+# def test_validate_kafka_topics():
+#     LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+#     response = KafkaTopic.create_all_topics()
+#     assert isinstance(response, bool)
+
+# # Call load_topology from the add_devices.py file
+# def test_load_topology(context_client, device_client):
+#     load_topology(context_client, device_client)

src/telemetry/frontend/service/TelemetryFrontendServiceServicerImpl.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import json
-import threading
 from typing import Any, Dict
 import grpc
 import logging
@@ -29,7 +28,6 @@ from telemetry.database.Telemetry_DB import TelemetryDB
 
 from confluent_kafka import Consumer as KafkaConsumer
 from confluent_kafka import Producer as KafkaProducer
-from confluent_kafka import KafkaError
 
 
 LOGGER            = logging.getLogger(__name__)
@@ -49,7 +47,7 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
    
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def StartCollector(self, 
-                       request : Collector, grpc_context: grpc.ServicerContext # type: ignore
+                       request : Collector, context: grpc.ServicerContext # type: ignore
                       ) -> CollectorId: # type: ignore
         LOGGER.info ("gRPC message: {:}".format(request))
         response = CollectorId()
@@ -86,7 +84,7 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def StopCollector(self, 
-                      request : CollectorId, grpc_context: grpc.ServicerContext # type: ignore
+                      request : CollectorId, context: grpc.ServicerContext # type: ignore
                      ) -> Empty:  # type: ignore
         LOGGER.info ("gRPC message: {:}".format(request))
         try:
@@ -125,7 +123,7 @@ class TelemetryFrontendServiceServicerImpl(TelemetryFrontendServiceServicer):
 
     @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
     def SelectCollectors(self, 
-                         request : CollectorFilter, contextgrpc_context: grpc.ServicerContext # type: ignore
+                         request : CollectorFilter, context: grpc.ServicerContext # type: ignore
                         ) -> CollectorList:  # type: ignore
         LOGGER.info("gRPC message: {:}".format(request))
         response = CollectorList()