refactor(processor)!: Separate platform processing from actor execution

src/powerapi/config/generator.py

@@ -31,10 +31,10 @@
 from collections.abc import Callable
 
 from powerapi.actor import Actor, ActorProxy
-from powerapi.database.driver import ReadableDatabaseFactory, WritableDatabaseFactory
 from powerapi.config.exceptions import ConfigurationError
+from powerapi.database.driver import ReadableDatabaseFactory, WritableDatabaseFactory
 from powerapi.filter import ReportFilter
-from powerapi.processor.processor_actor import ProcessorActor
+from powerapi.processor.processor_actor import PreProcessorActor, ProcessorActor
 from powerapi.puller import PullerActor
 from powerapi.pusher import PusherActor
 from powerapi.report import FormulaReport, HWPCReport, PowerReport, Report
@@ -472,8 +472,8 @@ def _k8s_pre_processor_factory(processor_config: dict) -> ProcessorActor:
         :param processor_config: Resolved Kubernetes pre-processor configuration.
         :return: Configured Kubernetes pre-processor actor.
         """
-        from powerapi.processor.pre.k8s.actor import KubernetesPreProcessorActor
         from powerapi.processor.pre.k8s.monitor_agent import KubernetesMonitorConfig
+        from powerapi.processor.pre.k8s.processor import KubernetesProcessor
 
         api_mode = processor_config['api-mode']
         api_host = processor_config.get('api-host')
@@ -483,7 +483,7 @@ def _k8s_pre_processor_factory(processor_config: dict) -> ProcessorActor:
 
         name = processor_config[ACTOR_NAME_KEY]
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
-        return KubernetesPreProcessorActor(name, monitor_config, level_logger)
+        return PreProcessorActor(name, KubernetesProcessor(monitor_config), level_logger)
 
     @staticmethod
     def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor:
@@ -492,15 +492,13 @@ def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor:
         :param processor_config: Resolved OpenStack pre-processor configuration.
         :return: Configured OpenStack pre-processor actor.
         """
-        from powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor
-        from powerapi.processor.pre.openstack.monitor_agent import (
-            OpenStackMonitorConfig,
-        )
+        from powerapi.processor.pre.openstack.monitor_agent import OpenStackMonitorConfig
+        from powerapi.processor.pre.openstack.processor import OpenStackProcessor
 
         api_polling_interval = processor_config['polling-interval']
         metadata_mapping = build_metadata_mapping(processor_config.get('metadata', []), prefix='openstack_metadata_')
         monitor_config = OpenStackMonitorConfig(api_polling_interval, metadata_mapping)
 
         name = processor_config[ACTOR_NAME_KEY]
         level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
-        return OpenStackPreProcessorActor(name, monitor_config, level_logger)
+        return PreProcessorActor(name, OpenStackProcessor(monitor_config), level_logger)

src/powerapi/processor/__init__.py

@@ -28,3 +28,4 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 from powerapi.processor.processor_actor import ProcessorActor, PreProcessorActor, PostProcessorActor
+from powerapi.processor.processor import Processor, ReportProcessor

src/powerapi/processor/handlers.py

@@ -27,19 +27,37 @@
 # 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.actor import InitializedStateHandler
+from powerapi.actor import InitializedStateHandler, State
+from powerapi.processor.processor import ReportProcessor
 from powerapi.report import Report
 
 
 class ProcessorReportHandler[ReportT: Report](InitializedStateHandler[ReportT]):
     """
-    Processor report handler class.
-    Used by processor actors to handles incoming reports.
+    Process incoming reports and forward retained results.
     """
 
+    def __init__(self, state: State, processor: ReportProcessor[ReportT]):
+        """
+        Initialize a report handler with its processing implementation.
+        :param state: Actor state used for initialization gating and forwarding
+        :param processor: Processor accepting the registered report type
+        """
+        super().__init__(state)
+        self._report_processor = processor
+
+    def handle(self, msg: ReportT) -> None:
+        """
+        Process a report and forward it unless it was dropped.
+        :param msg: Incoming report
+        """
+        report = self._report_processor.process(msg)
+        if report is not None:
+            self._send_report(report)
+
     def _send_report(self, report: Report) -> None:
         """
-        Helper method to send the processed report to its targets actors.
+        Send the processed report to all target actors.
         :param report: Report to send
         """
         for target in self.state.actor.target_actors:

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

@@ -0,0 +1,134 @@
+# Copyright (c) 2026, Inria
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright notice, this
+#   list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright notice,
+#   this list of conditions and the following disclaimer in the documentation
+#   and/or other materials provided with the distribution.
+#
+# * Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# 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 collections.abc import Iterable
+from multiprocessing import Manager
+from typing import Any
+
+from powerapi.processor.processor import Processor, ReportProcessor
+from powerapi.report import HWPCReport
+
+from ._utils import (
+    extract_container_id_from_k8s_cgroups_path,
+    is_target_a_valid_k8s_cgroups_path,
+)
+from .metadata_registry import KubernetesMetadataRegistry
+from .monitor_agent import KubernetesMonitorAgent, KubernetesMonitorConfig
+
+
+class KubernetesProcessor(Processor):
+    """
+    Manage Kubernetes metadata resources for report processors.
+    """
+
+    def __init__(self, monitor_config: KubernetesMonitorConfig):
+        """
+        Create a new Kubernetes processor.
+        :param monitor_config: Configuration of the monitoring agent
+        """
+        self.monitor_config = monitor_config
+
+        self.manager = None
+        self.metadata_registry = None
+        self.monitor_agent = None
+
+    def initialize(self) -> None:
+        """
+        Initialize shared metadata resources and start Kubernetes API monitoring.
+        :raises RuntimeError: If the processor is already initialized
+        """
+        if self.manager is not None:
+            raise RuntimeError('Processor already initialized')
+
+        try:
+            self.manager = Manager()
+            self.metadata_registry = KubernetesMetadataRegistry(self.manager)
+            self.monitor_agent = KubernetesMonitorAgent(self.metadata_registry, self.monitor_config)
+            self.monitor_agent.start()
+        except BaseException:
+            self.teardown()
+            raise
+
+    def get_report_processors(self) -> Iterable[ReportProcessor[Any]]:
+        """
+        Return initialized report processors for all supported report types.
+        :return: Iterable of initialized report processors
+        """
+        return (
+            HWPCReportProcessor(self.metadata_registry),
+        )
+
+    def teardown(self) -> None:
+        """
+        Clean up resources allocated during initialization.
+        """
+        try:
+            if self.monitor_agent is not None:
+                if self.monitor_agent.pid is not None:
+                    self.monitor_agent.terminate()
+                    self.monitor_agent.join()
+                self.monitor_agent = None
+        finally:
+            if self.manager is not None:
+                self.manager.shutdown()
+                self.manager = None
+
+
+class HWPCReportProcessor(ReportProcessor[HWPCReport]):
+    """
+    Enrich or filter HWPC reports using cached Kubernetes metadata.
+    """
+
+    def __init__(self, metadata_registry: KubernetesMetadataRegistry):
+        """
+        Initialize a report processor for HWPC reports.
+        :param metadata_registry: Shared platform metadata registry
+        """
+        super().__init__(HWPCReport)
+
+        self.metadata_registry = metadata_registry
+
+    def process(self, report: HWPCReport) -> HWPCReport | None:
+        """
+        Enrich matching reports in place with cached Kubernetes metadata.
+        Cached values overwrite existing metadata keys.
+        Non-Kubernetes targets pass through unchanged; matching reports without cached metadata are dropped.
+        :param report: Incoming HWPC report
+        :return: The report, possibly modified, or None if dropped
+        """
+        if is_target_a_valid_k8s_cgroups_path(report.target):
+            container_id = extract_container_id_from_k8s_cgroups_path(report.target)
+            container_metadata = self.metadata_registry.get_metadata(container_id)
+            if container_metadata is None:
+                # Drop the report if the container metadata is not present in the cache.
+                # This is mainly to filter out the empty pause container present for every running POD.
+                return None
+
+            report.metadata.update(container_metadata)
+
+        return report