Merge pull request #797 from powerapi-ng/feat/processor-lazy-import

feat: Improve handling of pre-processors with missing dependencies

Author: gfieni

src/powerapi/cli/generator.py

@@ -35,8 +35,6 @@
 from powerapi.exception import PowerAPIException, ModelNameAlreadyUsed, DatabaseNameDoesNotExist, ModelNameDoesNotExist, \
     DatabaseNameAlreadyUsed, ProcessorTypeDoesNotExist, ProcessorTypeAlreadyUsed
 from powerapi.filter import ReportFilter
-from powerapi.processor.pre.k8s import K8sPreProcessorActor, K8sProcessorConfig
-from powerapi.processor.pre.openstack import OpenStackPreProcessorActor
 from powerapi.processor.processor_actor import ProcessorActor
 from powerapi.puller import PullerActor
 from powerapi.pusher import PusherActor
@@ -375,14 +373,13 @@ class ProcessorGenerator(Generator):
     Generator that initializes the processor actor(s) from the configuration.
     """
 
-    def __init__(self, component_group_name: str, processor_factory: dict[str, Callable[[dict], ProcessorActor]]):
+    def __init__(self, component_group_name: str):
         """
         :param component_group_name: Name of the component group
-        :param processor_factory: Dictionary mapping processor type to actor factory
         """
         super().__init__(component_group_name)
 
-        self.processor_factory = processor_factory
+        self.processor_factory: dict[str, Callable[[dict], ProcessorActor]] = {}
 
     def remove_processor_factory(self, processor_type: str) -> None:
         """
@@ -405,6 +402,14 @@ def add_processor_factory(self, processor_type: str, processor_factory_function:
 
         self.processor_factory[processor_type] = processor_factory_function
 
+    def _generate_processor(self, processor_name: str, component_config: dict) -> ProcessorActor:
+        try:
+            return self.processor_factory[processor_name](component_config)
+        except KeyError as exn:
+            raise PowerAPIException('Configuration error: Invalid processor type: %s', processor_name) from exn
+        except ImportError as exn:
+            raise PowerAPIException('Dependencies for %s processor are not installed', processor_name) from exn
+
     def _gen_actor(self, component_config: dict, main_config: dict, component_name: str) -> ProcessorActor:
         """
         Helper method to generate a processor actor from the given configuration.
@@ -414,12 +419,9 @@ def _gen_actor(self, component_config: dict, main_config: dict, component_name:
         :return: Processor actor
         """
         processor_actor_type = component_config[COMPONENT_TYPE_KEY]
-        if processor_actor_type not in self.processor_factory:
-            raise PowerAPIException(f'Configuration error: Unknown processor actor type: {processor_actor_type}')
-
         component_config[ACTOR_NAME_KEY] = component_name
         component_config[GENERAL_CONF_VERBOSE_KEY] = main_config[GENERAL_CONF_VERBOSE_KEY]
-        return self.processor_factory[processor_actor_type](component_config)
+        return self._generate_processor(processor_actor_type, component_config)
 
 
 class PreProcessorGenerator(ProcessorGenerator):
@@ -428,39 +430,35 @@ class PreProcessorGenerator(ProcessorGenerator):
     """
 
     def __init__(self):
-        super().__init__('pre-processor', self._get_default_processor_factories())
+        super().__init__('pre-processor')
+
+        self.add_processor_factory('k8s', self._k8s_pre_processor_factory)
+        self.add_processor_factory('openstack', self._openstack_pre_processor_factory)
 
     @staticmethod
-    def _k8s_pre_processor_factory(processor_config: dict) -> K8sPreProcessorActor:
+    def _k8s_pre_processor_factory(processor_config: dict) -> ProcessorActor:
         """
         Kubernetes pre-processor actor factory.
         :param processor_config: Pre-Processor configuration
         :return: Configured Kubernetes pre-processor actor
         """
+        from powerapi.processor.pre.k8s.actor import K8sPreProcessorActor, K8sProcessorConfig
         name = processor_config[ACTOR_NAME_KEY]
-        api_mode = processor_config.get(K8S_API_MODE_KEY, 'manual')  # use manual mode by default
+        api_mode = processor_config[K8S_API_MODE_KEY]
         api_host = processor_config.get(K8S_API_HOST_KEY, None)
         api_key = processor_config.get(K8S_API_KEY_KEY, None)
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
         config = K8sProcessorConfig(api_mode, api_host, api_key)
         return K8sPreProcessorActor(name, config, level_logger)
 
     @staticmethod
-    def _openstack_pre_processor_factory(processor_config: dict) -> OpenStackPreProcessorActor:
+    def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor:
         """
         Openstack pre-processor actor factory.
         :param processor_config: Pre-Processor configuration
         :return: Configured OpenStack pre-processor actor
         """
+        from powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor
         name = processor_config[ACTOR_NAME_KEY]
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
         return OpenStackPreProcessorActor(name, level_logger)
-
-    def _get_default_processor_factories(self) -> dict[str, Callable[[dict], ProcessorActor]]:
-        """
-        Return the default pre-processors factory.
-        """
-        return {
-            'k8s': self._k8s_pre_processor_factory,
-            'openstack': self._openstack_pre_processor_factory
-        }

src/powerapi/processor/pre/k8s/__init__.py

@@ -26,5 +26,3 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
-from .actor import K8sPreProcessorActor, K8sProcessorConfig

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

@@ -27,4 +27,3 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-from powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor

tests/unit/cli/conftest.py

@@ -80,23 +80,6 @@ def several_inputs_outputs_stream_socket_without_some_arguments_config(several_i
     return several_inputs_outputs_stream_config
 
 
-@pytest.fixture
-def several_k8s_pre_processors_config():
-    """
-    Configuration with several k8s processors
-    """
-    return load_configuration_from_json_file(file_name='several_k8s_pre_processors_configuration.json')
-
-
-@pytest.fixture
-def several_k8s_pre_processors_without_some_arguments_config():
-    """
-    Configuration with several k8s processors
-    """
-    return load_configuration_from_json_file(
-        file_name='several_k8s_pre_processors_without_some_arguments_configuration.json')
-
-
 @pytest.fixture
 def csv_io_postmortem_config(invalid_csv_io_stream_config):
     """
@@ -146,14 +129,6 @@ def config_without_output(csv_io_postmortem_config):
     return csv_io_postmortem_config
 
 
-@pytest.fixture
-def k8s_pre_processor_config():
-    """
-    Configuration with k8s as pre-processor
-    """
-    return load_configuration_from_json_file(file_name='k8s_pre_processor_configuration.json')
-
-
 @pytest.fixture
 def subgroup_parser():
     """
@@ -427,124 +402,12 @@ def empty_pre_processor_config(pre_processor_complete_configuration):
     return pre_processor_complete_configuration
 
 
-@pytest.fixture(params=['k8s_pre_processor_wrong_binding_configuration.json'])
-def pre_processor_wrong_binding_configuration(request):
-    """
-    Return a dictionary containing wrong bindings with a pre-processor
-    """
-    return load_configuration_from_json_file(file_name=request.param)
-
-
 @pytest.fixture(params=['k8s_pre_processor_with_non_existing_puller_configuration.json'])
 def pre_processor_with_unexisting_puller_configuration(request):
     """
     Return a dictionary containing a pre-processor with a puller that doesn't exist
     """
-    return load_configuration_from_json_file(
-        file_name=request.param)
-
-
-@pytest.fixture(params=['k8s_pre_processor_with_reused_puller_in_bindings_configuration.json'])
-def pre_processor_with_reused_puller_in_bindings_configuration(request):
-    """
-    Return a dictionary containing a pre-processor with a puller that doesn't exist
-    """
-    return load_configuration_from_json_file(
-        file_name=request.param)
-
-
-@pytest.fixture
-def pre_processor_pullers_and_processors_dictionaries(pre_processor_complete_configuration):
-    """
-    Return a dictionary which contains puller actors, a dictionary of processors as well as the pushers
-    """
-    return get_pre_processor_pullers_and_processors_dictionaries_from_configuration(
-        configuration=pre_processor_complete_configuration)
-
-
-def get_pre_processor_pullers_and_processors_dictionaries_from_configuration(configuration: dict) -> (dict, dict, dict):
-    """
-    Return a tuple of dictionaries (pullers, processors) created from the given configuration.
-    :param dict configuration : Dictionary containing the configuration
-    """
-    report_filter = BroadcastReportFilter()
-    puller_generator = PullerGenerator(report_filter=report_filter)
-    pullers = puller_generator.generate(main_config=configuration)
-
-    pusher_generator = PusherGenerator()
-    pushers = pusher_generator.generate(main_config=configuration)
-
-    route_table = RouteTable()
-
-    dispatcher = DispatcherActor('dispatcher', None, pushers, route_table)
-
-    report_filter.register(lambda msg: True, dispatcher.get_proxy())
-
-    processor_generator = PreProcessorGenerator()
-    processors = processor_generator.generate(main_config=configuration)
-
-    return pullers, processors, pushers
-
-
-@pytest.fixture
-def pre_processor_binding_manager(pre_processor_complete_configuration, pre_processor_pullers_and_processors_dictionaries):
-    """
-    Return a ProcessorBindingManager with a Processor
-    """
-    pullers = pre_processor_pullers_and_processors_dictionaries[0]
-    processors = pre_processor_pullers_and_processors_dictionaries[1]
-
-    return PreProcessorBindingManager(
-        config=pre_processor_complete_configuration,
-        pullers=pullers,
-        processors=processors
-    )
-
-
-@pytest.fixture
-def pre_processor_binding_manager_with_wrong_binding_types(pre_processor_wrong_binding_configuration):
-    """
-    Return a PreProcessorBindingManager with wrong target for the pre-processor (a pusher instead of a puller)
-    """
-    _, processors, pushers = get_pre_processor_pullers_and_processors_dictionaries_from_configuration(
-        configuration=pre_processor_wrong_binding_configuration)
-
-    return PreProcessorBindingManager(
-        config=pre_processor_wrong_binding_configuration,
-        pullers=pushers,
-        processors=processors
-    )
-
-
-@pytest.fixture
-def pre_processor_binding_manager_with_unexisting_puller(pre_processor_with_unexisting_puller_configuration):
-    """
-    Return a PreProcessorBindingManager with an unexisting target for the pre-processor (a puller that doesn't exist)
-    """
-    pullers, processors, _ = get_pre_processor_pullers_and_processors_dictionaries_from_configuration(
-        configuration=pre_processor_with_unexisting_puller_configuration)
-
-    return PreProcessorBindingManager(
-        config=pre_processor_with_unexisting_puller_configuration,
-        pullers=pullers,
-        processors=processors
-    )
-
-
-@pytest.fixture
-def pre_processor_binding_manager_with_reused_puller_in_bindings(
-        pre_processor_with_reused_puller_in_bindings_configuration):
-    """
-    Return a PreProcessorBindingManager with a puller used by two different pre-processors
-    """
-    pullers, processors, _ = get_pre_processor_pullers_and_processors_dictionaries_from_configuration(
-        configuration=pre_processor_with_reused_puller_in_bindings_configuration)
-
-    return PreProcessorBindingManager(
-        config=pre_processor_with_reused_puller_in_bindings_configuration,
-        pullers=pullers,
-        processors=processors
-    )
+    return load_configuration_from_json_file(request.param)
 
 
 def get_config_with_longest_argument_names(config: dict, arguments: dict):