refactor(processor/openstack): Decouple monitor agent config from actor

src/powerapi/cli/generator.py

@@ -464,7 +464,11 @@ def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor:
         :return: Configured OpenStack pre-processor actor
         """
         from powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor
-        name = processor_config[ACTOR_NAME_KEY]
+        from powerapi.processor.pre.openstack.monitor_agent import OpenStackMonitorConfig
+
         api_polling_interval = processor_config['polling-interval']
+        monitor_config = OpenStackMonitorConfig(api_polling_interval)
+
+        name = processor_config[ACTOR_NAME_KEY]
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
-        return OpenStackPreProcessorActor(name, api_polling_interval, level_logger)
+        return OpenStackPreProcessorActor(name, monitor_config, level_logger)

src/powerapi/processor/pre/openstack/actor.py

@@ -36,48 +36,48 @@
 from powerapi.processor.processor_actor import ProcessorActor
 from powerapi.report import HWPCReport
 from .metadata_cache_manager import OpenStackMetadataCacheManager
-from .monitor_agent import OpenStackMonitorAgent
+from .monitor_agent import OpenStackMonitorAgent, OpenStackMonitorConfig
 
 
 class OpenStackProcessorState(State):
     """
     State of the OpenStack processor actor.
     """
 
-    def __init__(self, actor: Actor, polling_interval: float):
+    def __init__(self, actor: Actor, monitor_config: OpenStackMonitorConfig):
         """
         Initializes an OpenStack processor state.
         :param actor: Actor instance
-        :param polling_interval: OpenStack API polling interval (in seconds)
+        :param monitor_config: Monitoring agent configuration
         """
         super().__init__(actor)
 
         self.manager = Manager()
         self.metadata_cache_manager = OpenStackMetadataCacheManager(self.manager)
-        self.monitor_agent = OpenStackMonitorAgent(self.metadata_cache_manager, polling_interval)
+        self.monitor_agent = OpenStackMonitorAgent(self.metadata_cache_manager, monitor_config)
 
 
 class OpenStackPreProcessorActor(ProcessorActor):
     """
     Pre-Processor Actor that adds OpenStack related metadata to reports.
     """
 
-    def __init__(self, name: str, polling_interval: float, level_logger: int = logging.WARNING):
+    def __init__(self, name: str, monitor_config: OpenStackMonitorConfig, level_logger: int = logging.WARNING):
         """
         Initializes an OpenStack pre-processor actor.
         :param name: Name of the actor
-        :param polling_interval: OpenStack API polling interval (in seconds)
+        :param monitor_config: Monitoring agent configuration
         :param level_logger: Logging level of the actor
         """
         super().__init__(name, level_logger, 5000)
 
-        self.polling_interval = polling_interval
+        self.monitor_config = monitor_config
 
     def setup(self):
         """
         Set up the OpenStack pre-processor actor.
         """
-        self.state = OpenStackProcessorState(self, self.polling_interval)
+        self.state = OpenStackProcessorState(self, self.monitor_config)
 
         self.add_handler(StartMessage, StartMessageHandler(self.state))
         self.add_handler(PoisonPillMessage, PoisonPillMessageHandler(self.state))

src/powerapi/processor/pre/openstack/monitor_agent.py

@@ -28,7 +28,8 @@
 
 import logging
 import sys
-from multiprocessing import Process
+from dataclasses import dataclass
+from multiprocessing import Process, Event
 from signal import signal, SIGINT, SIGTERM
 from time import sleep
 
@@ -39,17 +40,26 @@
 from .metadata_cache_manager import OpenStackMetadataCacheManager, ServerMetadata
 
 
+@dataclass
+class OpenStackMonitorConfig:
+    """
+    OpenStack monitoring agent configuration.
+    :param polling_interval: Interval in seconds between OpenStack API synchronizations.
+    """
+    polling_interval: float
+
+
 class OpenStackMonitorAgent(Process):
     """
     Background monitoring agent that updates the shared metadata cache from the OpenStack API.
     It requires credentials with sufficient permissions to access server metadata across all projects.
     Permission to read Nova Extended Server Attributes (OS-EXT-SRV-ATTR) is **mandatory** in order to map cgroups to servers.
     """
 
-    def __init__(self, cache_manager: OpenStackMetadataCacheManager, poll_interval: float, level_logger: int = logging.WARNING):
+    def __init__(self, cache_manager: OpenStackMetadataCacheManager, config: OpenStackMonitorConfig, level_logger: int = logging.WARNING):
         """
         :param cache_manager: Metadata cache manager
-        :param poll_interval: Interval in seconds between OpenStack API synchronizations
+        :param config: Configuration of the monitor agent
         :param level_logger: Logger level
         """
         super().__init__(name='openstack-processor-monitor-agent')
@@ -61,8 +71,9 @@ def __init__(self, cache_manager: OpenStackMetadataCacheManager, poll_interval:
         handler.setFormatter(formatter)
 
         self.metadata_cache_manager = cache_manager
-        self.poll_interval = poll_interval
-        self.stop_monitoring = False
+        self.config = config
+
+        self._stop_monitoring = Event()
 
     @staticmethod
     def _setup_openstack_api_client() -> Connection:
@@ -77,7 +88,7 @@ def _setup_signal_handlers(self):
         Setup signal handlers for the current Process.
         """
         def stop_monitor(_, __):
-            self.stop_monitoring = True
+            self._stop_monitoring = True
             sys.exit(0)
 
         signal(SIGTERM, stop_monitor)
@@ -93,11 +104,11 @@ def run(self):
         # Prevents orphaned entries that no longer exist in the OpenStack API.
         self.metadata_cache_manager.clear_metadata_cache()
 
-        while not self.stop_monitoring:
+        while not self._stop_monitoring:
             for server in self.fetch_servers_metadata(openstack_api):
                 self.metadata_cache_manager.update_server_metadata(server)
 
-            sleep(self.poll_interval)
+            sleep(self.config.polling_interval)
 
     @staticmethod
     def build_metadata_cache_entry_from_server(server: Server) -> ServerMetadata:

tests/unit/cli/test_generator_openstack.py

@@ -64,3 +64,6 @@ def test_preprocessor_generator_with_valid_openstack_config(openstack_config):
 
     preprocessor = preprocessors['pytest-openstack-preprocessor']
     assert isinstance(preprocessor, OpenStackPreProcessorActor)
+
+    expected_preprocessor_attributes = openstack_config['pre-processor']['pytest-openstack-preprocessor']
+    assert preprocessor.monitor_config.polling_interval == expected_preprocessor_attributes['polling-interval']

tests/unit/processor/pre/openstack/test_monitor_agent.py

@@ -36,15 +36,16 @@
 from openstack.compute.v2.server import Server
 
 from powerapi.processor.pre.openstack.metadata_cache_manager import ServerMetadata
-from powerapi.processor.pre.openstack.monitor_agent import OpenStackMonitorAgent
+from powerapi.processor.pre.openstack.monitor_agent import OpenStackMonitorAgent, OpenStackMonitorConfig
 
 
 @pytest.fixture
 def initialized_monitor_agent(initialized_metadata_cache_manager):
     """
     Returns an initialized OpenStack monitor agent.
     """
-    return OpenStackMonitorAgent(initialized_metadata_cache_manager, poll_interval=0.01)
+    monitor_config = OpenStackMonitorConfig(polling_interval=0.01)
+    return OpenStackMonitorAgent(initialized_metadata_cache_manager, monitor_config)
 
 
 def make_server(server_id, server_name, host, instance_name, metadata) -> Server: