Merge branch 'develop' of ssh://gifrerenom_labs.etsi.org/tfs/controller into...

Merge branch 'develop' of ssh://gifrerenom_labs.etsi.org/tfs/controller into feat/247-cttc-analytics-module-enhancements

my_deploy.sh

@@ -26,7 +26,7 @@ export TFS_COMPONENTS="context device pathcomp service slice nbi webui"
 #export TFS_COMPONENTS="${TFS_COMPONENTS} monitoring"
 
 # Uncomment to activate Monitoring Framework (new)
-export TFS_COMPONENTS="${TFS_COMPONENTS} kpi_manager kpi_value_writer kpi_value_api telemetry analytics automation"
+#export TFS_COMPONENTS="${TFS_COMPONENTS} kpi_manager kpi_value_writer kpi_value_api telemetry analytics automation"
 
 # Uncomment to activate QoS Profiles
 #export TFS_COMPONENTS="${TFS_COMPONENTS} qos_profile"

scripts/run_tests_locally-telemetry-emulated.sh

 #!/bin/bash
-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/driver_api/_Driver.py → src/telemetry/backend/collector_api/_Collector.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,7 +15,7 @@
 import threading
 from typing import Any, Iterator, List, Optional, Tuple, Union
 
-# Special resource names to request to the driver to retrieve the specified
+# Special resource names to request to the collector to retrieve the specified
 # configuration/structural resources.
 # These resource names should be used with GetConfig() method.
 RESOURCE_ENDPOINTS = '__endpoints__'
@@ -27,16 +27,16 @@ RESOURCE_ACL = '__acl__'
 RESOURCE_INVENTORY = '__inventory__'
 
 
-class _Driver:
+class _Collector:
     def __init__(self, name : str, address: str, port: int, **settings) -> None:
-        """ Initialize Driver.
+        """ Initialize Collector.
             Parameters:
                 address : str
                     The address of the device
                 port : int
                     The port of the device
                 **settings
-                    Extra settings required by the driver.
+                    Extra settings required by the collector.
         """
         self._name = name
         self._address = address
@@ -139,7 +139,7 @@ class _Driver:
             List[Union[bool, Exception]]:
         """ Subscribe to state information of entire device or
         selected resources. Subscriptions are incremental.
-            Driver should keep track of requested resources.
+            Collector should keep track of requested resources.
             Parameters:
                 subscriptions : List[Tuple[str, float, float]]
                     List of tuples, each containing a resource_key pointing the
@@ -162,7 +162,7 @@ class _Driver:
             -> List[Union[bool, Exception]]:
         """ Unsubscribe from state information of entire device
         or selected resources. Subscriptions are incremental.
-            Driver should keep track of requested resources.
+            Collector should keep track of requested resources.
             Parameters:
                 subscriptions : List[str]
                     List of tuples, each containing a resource_key pointing the
@@ -188,37 +188,37 @@ class _Driver:
         Operates as a generator, so this method should be called once and will
         block until values are available. When values are available,
         it should yield each of them and block again until new values are
-        available. When the driver is destroyed, GetState() can return instead
+        available. When the collector is destroyed, GetState() can return instead
         of yield to terminate the loop.
         Terminate enables to request interruption of the generation.
             Examples:
                 # keep looping waiting for extra samples (generator loop)
                 terminate = threading.Event()
                 i = 0
-                for timestamp,resource_key,resource_value in my_driver.GetState(blocking=True, terminate=terminate):
+                for timestamp,resource_key,resource_value in my_collector.GetState(blocking=True, terminate=terminate):
                     process(timestamp, resource_key, resource_value)
                     i += 1
                     if i == 10: terminate.set()
 
                 # just retrieve accumulated samples
-                samples = my_driver.GetState(blocking=False, terminate=terminate)
+                samples = my_collector.GetState(blocking=False, terminate=terminate)
                 # or (as classical loop)
                 i = 0
-                for timestamp,resource_key,resource_value in my_driver.GetState(blocking=False, terminate=terminate):
+                for timestamp,resource_key,resource_value in my_collector.GetState(blocking=False, terminate=terminate):
                     process(timestamp, resource_key, resource_value)
                     i += 1
                     if i == 10: terminate.set()
             Parameters:
                 blocking : bool
-                    Select the driver behaviour. In both cases, the driver will
+                    Select the collector behaviour. In both cases, the collector will
                     first retrieve the samples accumulated and available in the
-                    internal queue. Then, if blocking, the driver does not
+                    internal queue. Then, if blocking, the collector does not
                     terminate the loop and waits for additional samples to come,
-                    thus behaving as a generator. If non-blocking, the driver
+                    thus behaving as a generator. If non-blocking, the collector
                     terminates the loop and returns. Non-blocking behaviour can
-                    be used for periodically polling the driver, while blocking
+                    be used for periodically polling the collector, while blocking
                     can be used when a separate thread is in charge of
-                    collecting the samples produced by the driver.
+                    collecting the samples produced by the collector.
                 terminate : threading.Event
                     Signals the interruption of the GetState method as soon as
                     possible.

src/telemetry/backend/drivers/emulated/__init__.py → src/telemetry/backend/collector_api/__init__.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/driver_api/__init__.py → src/telemetry/backend/collectors/__init__.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/drivers/emulated/EmulatedDriver.py → src/telemetry/backend/collectors/emulated/EmulatedCollector.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,8 +17,6 @@ import queue
 import logging
 import uuid
 import json
-from telemetry.backend.drivers.emulated.EmulatedHelper import EmulatedDriverHelper
-from telemetry.backend.driver_api._Driver import _Driver
 from anytree import Node, Resolver
 from apscheduler.events import EVENT_JOB_ADDED, EVENT_JOB_REMOVED
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -26,18 +24,18 @@ from apscheduler.jobstores.memory import MemoryJobStore
 from apscheduler.executors.pool import ThreadPoolExecutor
 from datetime import datetime, timedelta
 from typing import Any, Iterator, List, Tuple, Union, Optional
+from telemetry.backend.collector_api._Collector import _Collector
+from .EmulatedHelper import EmulatedCollectorHelper
 from .SyntheticMetricsGenerator import SyntheticMetricsGenerator
 
 
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-
-class EmulatedDriver(_Driver):
+class EmulatedCollector(_Collector):
     """
-    EmulatedDriver is a class that simulates a network driver for testing purposes.
+    EmulatedCollector is a class that simulates a network collector for testing purposes.
     It provides functionalities to manage configurations, state subscriptions, and synthetic data generation.
     """
     def __init__(self, address: str, port: int, **settings):
-        super().__init__('emulated_driver', address, port, **settings)
+        super().__init__('emulated_collector', address, port, **settings)
         self._initial_config = Node('root')                 # Tree structure for initial config
         self._running_config = Node('root')                 # Tree structure for running config
         self._subscriptions  = Node('subscriptions')        # Tree for state subscriptions
@@ -52,11 +50,11 @@ class EmulatedDriver(_Driver):
         )
         self._scheduler.add_listener(self._listener_job_added_to_subscription_tree,     EVENT_JOB_ADDED)
         self._scheduler.add_listener(self._listener_job_removed_from_subscription_tree, EVENT_JOB_REMOVED)
-        self._helper_methods = EmulatedDriverHelper()
+        self._helper_methods = EmulatedCollectorHelper()
 
         self.logger    = logging.getLogger(__name__)
         self.connected = False          # To track connection state
-        self.logger.info("EmulatedDriver initialized")
+        self.logger.info("EmulatedCollector initialized")
 
     def Connect(self) -> bool:
         self.logger.info(f"Connecting to {self.address}:{self.port}")
@@ -68,7 +66,7 @@ class EmulatedDriver(_Driver):
     def Disconnect(self) -> bool:
         self.logger.info(f"Disconnecting from {self.address}:{self.port}")
         if not self.connected:
-            self.logger.warning("Driver is not connected. Nothing to disconnect.")
+            self.logger.warning("Collector is not connected. Nothing to disconnect.")
             return False
         self._scheduler.shutdown()
         self.connected = False
@@ -77,7 +75,7 @@ class EmulatedDriver(_Driver):
 
     def _require_connection(self):
         if not self.connected:
-            raise RuntimeError("Driver is not connected. Please connect before performing operations.")
+            raise RuntimeError("Collector is not connected. Please connect before performing operations.")
 
     def SubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
         self._require_connection()
@@ -219,7 +217,7 @@ class EmulatedDriver(_Driver):
 # ------- The below methods are kept for debugging purposes (test-case) only ---------
 #-------------------------------------------------------------------------------------
 
-#  This method can be commented but this will arise an error in the test-case (@pytest.fixture --> connected_configured_driver()).
+#  This method can be commented but this will arise an error in the test-case (@pytest.fixture --> connected_configured_collector()).
     def SetConfig(self, resources: dict) -> List[Union[bool, Exception]]:  # For debugging purposes.
         self._require_connection()
         results = []

src/telemetry/backend/drivers/emulated/EmulatedHelper.py → src/telemetry/backend/collectors/emulated/EmulatedHelper.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,9 +17,9 @@ import json
 from typing import Any, List
 
 
-class EmulatedDriverHelper:
+class EmulatedCollectorHelper:
     """
-    Helper class for the emulated driver.
+    Helper class for the emulated collector.
     """
     def __init__(self):
         pass
@@ -48,7 +48,7 @@ class EmulatedDriverHelper:
 # ------- Below function is kept for debugging purposes (test-cases) only -------------
 #--------------------------------------------------------------------------------------
 
-#  This below methods can be commented but are called by the SetConfig method in EmulatedDriver.py
+#  This below methods can be commented but are called by the SetConfig method in EmulatedCollector.py
 
     def _find_or_create_node(self, name: str, parent: Node) -> Node:
         """

src/telemetry/backend/drivers/emulated/SyntheticMetricsGenerator.py → src/telemetry/backend/collectors/emulated/SyntheticMetricsGenerator.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/drivers/__init__.py → src/telemetry/backend/collectors/emulated/__init__.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/tests/messages_emulated.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/telemetry/backend/tests/test_emulated.py

-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,7 +15,7 @@
 import logging
 import time
 import pytest
-from telemetry.backend.drivers.emulated.EmulatedDriver import EmulatedDriver
+from telemetry.backend.collectors.emulated.EmulatedCollector import EmulatedCollector
 from telemetry.backend.tests.messages_emulated import (
     create_test_configuration,
     create_specific_config_keys,
@@ -27,82 +27,82 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(
 logger = logging.getLogger(__name__)
 
 @pytest.fixture
-def setup_driver():
-    """Sets up an EmulatedDriver instance for testing."""
-    yield EmulatedDriver(address="127.0.0.1", port=8080)
+def setup_collector():
+    """Sets up an EmulatedCollector instance for testing."""
+    yield EmulatedCollector(address="127.0.0.1", port=8080)
 
 @pytest.fixture
-def connected_configured_driver(setup_driver):
-    driver = setup_driver # EmulatedDriver(address="127.0.0.1", port=8080)
-    driver.Connect()
-    driver.SetConfig(create_test_configuration())
-    yield driver
-    driver.Disconnect()
-
-def test_connect(setup_driver):
+def connected_configured_collector(setup_collector):
+    collector = setup_collector # EmulatedCollector(address="127.0.0.1", port=8080)
+    collector.Connect()
+    collector.SetConfig(create_test_configuration())
+    yield collector
+    collector.Disconnect()
+
+def test_connect(setup_collector):
     logger.info(">>> test_connect <<<")
-    driver = setup_driver
-    assert driver.Connect() is True
-    assert driver.connected is True
+    collector = setup_collector
+    assert collector.Connect() is True
+    assert collector.connected is True
 
-def test_disconnect(setup_driver):
+def test_disconnect(setup_collector):
     logger.info(">>> test_disconnect <<<")
-    driver = setup_driver
-    driver.Connect()
-    assert driver.Disconnect() is True
-    assert driver.connected is False
+    collector = setup_collector
+    collector.Connect()
+    assert collector.Disconnect() is True
+    assert collector.connected is False
 
-# def test_set_config(setup_driver):
+# def test_set_config(setup_collector):
 #     logger.info(">>> test_set_config <<<")
-#     driver = setup_driver
-#     driver.Connect()
+#     collector = setup_collector
+#     collector.Connect()
 
 #     config = create_test_configuration()
 
-#     results = driver.SetConfig(config)
+#     results = collector.SetConfig(config)
 #     assert all(result is True for result in results)
 
-# def test_get_config(connected_configured_driver):
+# def test_get_config(connected_configured_collector):
 #     logger.info(">>> test_get_config <<<")
 #     resource_keys = create_specific_config_keys() 
-#     results = connected_configured_driver.GetConfig(resource_keys)
+#     results = connected_configured_collector.GetConfig(resource_keys)
 
 #     for key, value in results:
 #         assert key in create_specific_config_keys()
 #         assert value is not None
 
-# def test_delete_config(connected_configured_driver):
+# def test_delete_config(connected_configured_collector):
 #     logger.info(">>> test_delete_config <<<")
 #     resource_keys = create_config_for_delete() 
 
-#     results = connected_configured_driver.DeleteConfig(resource_keys)
+#     results = connected_configured_collector.DeleteConfig(resource_keys)
 #     assert all(result is True for result in results)
 
-def test_subscribe_state(connected_configured_driver):
+def test_subscribe_state(connected_configured_collector):
     logger.info(">>> test_subscribe_state <<<")
     subscriptions = create_test_subscriptions() 
 
-    results = connected_configured_driver.SubscribeState(subscriptions)
+    results = connected_configured_collector.SubscribeState(subscriptions)
     # logger.info(f"Subscribed result: {results}.")
     assert results ==  [False, True, True] # all(result is True for result in results)
 
-def test_unsubscribe_state(connected_configured_driver):
+def test_unsubscribe_state(connected_configured_collector):
     logger.info(">>> test_unsubscribe_state <<<")
     subscriptions = create_test_subscriptions()
 
-    connected_configured_driver.SubscribeState(subscriptions)
-    results = connected_configured_driver.UnsubscribeState(subscriptions)
+    connected_configured_collector.SubscribeState(subscriptions)
+    results = connected_configured_collector.UnsubscribeState(subscriptions)
     assert results ==  [False, True, True] # all(result is True for result in results)
 
-def test_get_state(connected_configured_driver):
+def test_get_state(connected_configured_collector):
     logger.info(">>> test_get_state <<<")
     subscriptions = create_test_subscriptions()
 
-    connected_configured_driver.SubscribeState(subscriptions)
+    connected_configured_collector.SubscribeState(subscriptions)
     logger.info(f"Subscribed to state: {subscriptions}. waiting for 12 seconds ...")
     time.sleep(12)
 
-    state_iterator = connected_configured_driver.GetState(blocking=False)
+    state_iterator = connected_configured_collector.GetState(blocking=False)
     states = list(state_iterator)
 
     assert len(states) > 0