feat(processor): Add OpenStack Pre-Processor

Author: gfieni

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

@@ -0,0 +1,30 @@
+# Copyright (c) 2025, Inria
+# Copyright (c) 2025, University of Lille
+# 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 powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor

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

@@ -0,0 +1,51 @@
+# Copyright (c) 2025, Inria
+# Copyright (c) 2025, University of Lille
+# 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.
+
+import re
+
+LIBVIRT_INSTANCE_NAME_REGEX = re.compile(r"(instance-\d+)")
+
+
+def get_instance_name_from_libvirt_cgroup(target: str) -> str | None:
+    """
+    Returns the instance name of the target.
+    :param target: Cgroup path
+    :return: Instance name (``instance-XXXXXXXX``)
+    """
+    if "\\x" in target:
+        # Some systems (cgroups v2 managed by systemd) escape special characters in the cgroup path.
+        # Decoding the path is required in order to reliably extract the instance name.
+        # For example: /sys/fs/cgroup/machine.slice/machine-qemu\\x2d3\\x2dinstance\\x2d00000003.scope/libvirt/emulator
+        target = target.encode("utf-8").decode("unicode_escape")
+
+    match = LIBVIRT_INSTANCE_NAME_REGEX.match(target)
+    if match:
+        return match.group(1)
+
+    return None

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

@@ -0,0 +1,79 @@
+# Copyright (c) 2023, INRIA
+# Copyright (c) 2023, University of Lille
+# 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.
+
+import logging
+
+from powerapi.actor import Actor
+from powerapi.message import StartMessage, PoisonPillMessage
+from powerapi.processor.pre.openstack.handlers import StartMessageHandler, PoisonPillMessageHandler, HWPCReportHandler
+from powerapi.processor.processor_actor import ProcessorState, ProcessorActor
+from powerapi.report import HWPCReport
+from .metadata_cache_manager import OpenStackMetadataCacheManager
+
+
+class OpenStackPreProcessorState(ProcessorState):
+    """
+    State of the OpenStack pre-processor actor.
+    """
+
+    def __init__(self, actor: Actor, target_actors: list, target_actors_names: list):
+        """
+        :param actor: Actor instance
+        :param target_actors: list of target actors
+        :param target_actors_names: list of target actor names
+        """
+        super().__init__(actor, target_actors, target_actors_names)
+
+        self.metadata_cache_manager = OpenStackMetadataCacheManager()
+
+
+class OpenStackPreProcessorActor(ProcessorActor):
+    """
+    Pre-Processor Actor that adds OpenStack related metadata to reports.
+    """
+
+    def __init__(self, name: str, target_actors: list, target_actors_names: list, level_logger: int = logging.WARNING):
+        """
+        :param name: Name of the actor
+        :param target_actors: List of target actors
+        :param target_actors_names: List of target actor names
+        :param level_logger: Logging level of the actor
+        """
+        super().__init__(name, level_logger, 5000)
+
+        self.state = OpenStackPreProcessorState(self, target_actors, target_actors_names)
+
+    def setup(self):
+        """
+        Set up the OpenStack pre-processor actor.
+        """
+        self.add_handler(StartMessage, StartMessageHandler(self.state))
+        self.add_handler(PoisonPillMessage, PoisonPillMessageHandler(self.state))
+
+        self.add_handler(HWPCReport, HWPCReportHandler(self.state))

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

@@ -0,0 +1,105 @@
+# Copyright (c) 2023, INRIA
+# Copyright (c) 2023, University of Lille
+# 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 powerapi.handler import StartHandler, PoisonPillMessageHandler as PoisonPillHandler
+from powerapi.processor.handlers import ProcessorReportHandler
+from powerapi.report import HWPCReport
+from ._utils import get_instance_name_from_libvirt_cgroup
+from .metadata_cache_manager import ServerMetadata, MetadataSyncFailed
+
+
+class StartMessageHandler(StartHandler):
+    """
+    Start message handler for the OpenStack processor actor.
+    """
+
+    def initialization(self):
+        """
+        Initialize the OpenStack processor.
+        """
+        for actor in self.state.target_actors:
+            actor.connect_data()
+
+        self.state.monitor_agent.start()
+
+
+class PoisonPillMessageHandler(PoisonPillHandler):
+    """
+    PoisonPill message handler for the OpenStack processor actor.
+    """
+
+    def teardown(self, soft: bool = False):
+        """
+        Teardown the OpenStack processor.
+        """
+        for actor in self.state.target_actors:
+            actor.socket_interface.close()
+
+
+class HWPCReportHandler(ProcessorReportHandler):
+    """
+    Generic report handler for the OpenStack processor actor.
+    Used to add the server metadata (from the OpenStack API) to the processed report.
+    """
+
+    def try_get_server_metadata(self, sensor_name: str, instance_name: str) -> ServerMetadata | None:
+        """
+        Try to get the server metadata from the cache.
+        :param sensor_name: Name of the sensor
+        :param instance_name: Name of the instance to fetch metadata for
+        :return: Server metadata entry or None if not found
+        """
+        server_metadata = None
+        try:
+            server_metadata = self.state.metadata_cache_manager.get_server_metadata(sensor_name, instance_name)
+            if server_metadata is None:
+                # Retry once after syncing the metadata cache.
+                self.state.metadata_cache_manager.sync_metadata_cache_from_api()
+                server_metadata = self.state.metadata_cache_manager.get_server_metadata(sensor_name, instance_name)
+        except MetadataSyncFailed as exn:
+            self.state.logger.warning(exn)
+
+        return server_metadata
+
+    def handle(self, msg: HWPCReport):
+        """
+        Process an HWPCReport to add the Kubernetes metadata.
+        :param msg: The HWPCReport to process
+        """
+        instance_name = get_instance_name_from_libvirt_cgroup(msg.target)
+        if instance_name is not None:
+            server_metadata = self.try_get_server_metadata(msg.sensor, instance_name)
+            if server_metadata is None:
+                # Drop the report if the server metadata is not present in the cache.
+                return
+
+            msg.target = server_metadata.server_name
+            msg.metadata['openstack'] = vars(server_metadata)
+
+        self._send_report(msg)

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

@@ -0,0 +1,91 @@
+# Copyright (c) 2025, Inria
+# Copyright (c) 2025, University of Lille
+# 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 dataclasses import dataclass
+
+from openstack.connection import Connection
+from openstack.exceptions import SDKException
+
+
+class MetadataSyncFailed(Exception):
+    """
+    Exception raised when the metadata cache sync operation fails.
+    """
+
+
+@dataclass(frozen=True, slots=True)
+class ServerMetadata:
+    """
+    Represents an OpenStack server metadata cache entry.
+    """
+    server_id: str
+    server_name: str
+    host: str
+    metadata: dict[str, str]
+
+
+class OpenStackMetadataCacheManager:
+    """
+    OpenStack metadata cache manager.
+    Use the OpenStack API to fetch details about the servers hosted on the infrastructure.
+    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):
+        self._openstack_api = Connection(app_name='PowerAPI')  # Configuration is taken from OS_* environment variables
+        self._metadata_cache: dict[tuple, ServerMetadata] = {}
+
+    def get_server_metadata(self, host: str, instance_name: str) -> ServerMetadata | None:
+        """
+        Get metadata for the server of the specified host from the cache.
+        :param host: Name of the host (hypervisor) where the server is located
+        :param instance_name: Name of the instance (libvirt instance name)
+        :return: Server metadata cache entry
+        """
+        return self._metadata_cache.get((host, instance_name), None)
+
+    def sync_metadata_cache_from_api(self) -> None:
+        """
+        Sync the running servers metadata cache from the OpenStack API.
+        """
+        try:
+            for server in self._openstack_api.compute.servers(details=True, all_projects=True):
+                cache_entry = ServerMetadata(server.id, server.name, server.host, server.metadata)
+                self._metadata_cache[(server.host, server.instance_name)] = cache_entry
+        except SDKException as exn:
+            raise MetadataSyncFailed('Failed to retrieve servers metadata from OpenStack API') from exn
+        except ValueError as exn:
+            raise MetadataSyncFailed('Required server attribute is missing from the OpenStack API response') from exn
+
+    def clear_metadata_cache(self) -> None:
+        """
+        Clears all server metadata entries from the cache.
+        """
+        self._metadata_cache.clear()