From bc27a0744dbe7d2fca28be8eb2bf14a432c7051c Mon Sep 17 00:00:00 2001 From: Guillaume Fieni Date: Thu, 13 Aug 2026 16:34:47 +0200 Subject: [PATCH 1/3] feat(utils/metadata): Add metadata name normalization utilities Add utilities functions to normalize platform metadata names into lowercase ASCII identifiers and build source-to-normalized mappings. --- src/powerapi/utils/metadata.py | 75 ++++++++++++++++++++ tests/unit/utils/__init__.py | 27 ++++++++ tests/unit/utils/test_metadata.py | 110 ++++++++++++++++++++++++++++++ 3 files changed, 212 insertions(+) create mode 100644 src/powerapi/utils/metadata.py create mode 100644 tests/unit/utils/__init__.py create mode 100644 tests/unit/utils/test_metadata.py diff --git a/src/powerapi/utils/metadata.py b/src/powerapi/utils/metadata.py new file mode 100644 index 00000000..63e0ef9c --- /dev/null +++ b/src/powerapi/utils/metadata.py @@ -0,0 +1,75 @@ +# 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. + +import re +import unicodedata +from collections.abc import Iterable, Mapping + +_INVALID_METADATA_NAME_CHARACTER = re.compile(r'[^A-Za-z0-9_]') + + +def normalize_metadata_name(name: str, prefix: str) -> str: + """ + Build a canonical metadata name from a prefix and a source name. + + The combined name is case-folded and converted to ASCII. Characters without ASCII representation are discarded. + Characters outside ``[A-Za-z0-9_]`` are replaced with underscores. + + :param name: Source metadata name provided by the platform + :param prefix: Namespace prefix prepended to the source name + :return: Lowercase, ASCII-only canonical metadata name + """ + prefixed_name = f'{prefix}{name}'.casefold() + ascii_prefixed_name = unicodedata.normalize('NFKD', prefixed_name).encode('ascii', 'ignore').decode('ascii') + return _INVALID_METADATA_NAME_CHARACTER.sub('_', ascii_prefixed_name) + + +def build_metadata_mapping(names: Iterable[str], prefix: str) -> Mapping[str, str]: + """ + Map source metadata names to their canonical metadata names. + + Duplicate source names are included once in first-occurrence order. + Distinct source names that produce the same canonical name are rejected. + + :param names: Source metadata names provided by the platform + :param prefix: Namespace prefix prepended to every source name + :return: Mapping from source metadata names to canonical metadata names + :raises ValueError: If distinct source names produce the same canonical name + """ + metadata_mapping = {} + source_by_normalized_name = {} + for source_name in dict.fromkeys(names): + normalized_name = normalize_metadata_name(source_name, prefix) + if normalized_name in source_by_normalized_name: + previous_source_name = source_by_normalized_name[normalized_name] + raise ValueError(f'Metadata names {previous_source_name} and {source_name} both normalize to {normalized_name}') + + source_by_normalized_name[normalized_name] = source_name + metadata_mapping[source_name] = normalized_name + + return metadata_mapping diff --git a/tests/unit/utils/__init__.py b/tests/unit/utils/__init__.py new file mode 100644 index 00000000..20687782 --- /dev/null +++ b/tests/unit/utils/__init__.py @@ -0,0 +1,27 @@ +# 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. diff --git a/tests/unit/utils/test_metadata.py b/tests/unit/utils/test_metadata.py new file mode 100644 index 00000000..4c56a98b --- /dev/null +++ b/tests/unit/utils/test_metadata.py @@ -0,0 +1,110 @@ +# 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. + +import pytest + +from powerapi.utils.metadata import build_metadata_mapping, normalize_metadata_name + + +@pytest.mark.parametrize( + ('name', 'prefix', 'expected'), + [ + ('namespace', 'k8s_label_', 'k8s_label_namespace'), + ('app.kubernetes.io/name', 'k8s_label_', 'k8s_label_app_kubernetes_io_name'), + ('application-type/version', 'openstack_metadata_', 'openstack_metadata_application_type_version'), + ('Consommation Énergie', 'PowerAPI_', 'powerapi_consommation_energie'), + ('container/name', '', 'container_name'), + ('test_metadata', 'powerapi_', 'powerapi_test_metadata'), + ('test_metadata', '', 'test_metadata'), + ], +) +def test_normalize_metadata_name(name: str, prefix: str, expected: str) -> None: + """ + Metadata names should be converted to prefixed ASCII identifiers. + """ + assert normalize_metadata_name(name, prefix) == expected + + +def test_build_metadata_mapping() -> None: + """ + Metadata mapping should associate source names with normalized names. + """ + names = ['app.kubernetes.io/name', 'pytest-example'] + + mapping = build_metadata_mapping(names, 'k8s_label_') + + assert mapping == { + 'app.kubernetes.io/name': 'k8s_label_app_kubernetes_io_name', + 'pytest-example': 'k8s_label_pytest_example', + } + + +def test_build_metadata_mapping_with_generator() -> None: + """ + Metadata mapping should accept any iterable of source names. + """ + names = (name for name in ['application.type', 'environment']) + + mapping = build_metadata_mapping(names, 'openstack_metadata_') + + assert mapping == { + 'application.type': 'openstack_metadata_application_type', + 'environment': 'openstack_metadata_environment', + } + + +def test_build_metadata_mapping_ignores_repeated_source_names() -> None: + """ + Repeated source names should appear only once in the mapping. + """ + mapping = build_metadata_mapping(['pytest', 'environment', 'pytest'], 'k8s_label_') + + assert list(mapping) == ['pytest', 'environment'] + assert mapping == { + 'pytest': 'k8s_label_pytest', + 'environment': 'k8s_label_environment', + } + + +def test_build_metadata_mapping_with_empty_iterable() -> None: + """ + An empty collection of source names should produce an empty mapping. + """ + assert build_metadata_mapping([], 'k8s_label_') == {} + + +def test_build_metadata_mapping_rejects_normalization_collision() -> None: + """ + Distinct names producing the same normalized name should be rejected. + """ + a = 'app.kubernetes.io/name' + b = 'app/kubernetes/io/name' + normalized = 'test_app_kubernetes_io_name' + + with pytest.raises(ValueError, match=f'Metadata names {a} and {b} both normalize to {normalized}'): + build_metadata_mapping(['app.kubernetes.io/name', 'app/kubernetes/io/name'], 'test_') From dcf6f42841a0127d6f006446e1ce67b83940937f Mon Sep 17 00:00:00 2001 From: Guillaume Fieni Date: Wed, 19 Aug 2026 17:10:18 +0200 Subject: [PATCH 2/3] feat(processor/k8s): Support selecting report metadata from pod labels Allow users to expose selected Kubernetes pod labels as normalized report metadata instead of propagating every available label. --- .../cli/common_cli_parsing_manager.py | 6 + src/powerapi/cli/generator.py | 10 +- src/powerapi/processor/pre/k8s/actor.py | 37 +-- src/powerapi/processor/pre/k8s/handlers.py | 20 +- ..._cache_manager.py => metadata_registry.py} | 52 +--- .../processor/pre/k8s/monitor_agent.py | 137 ++++----- .../processor/pre/k8s/pod_event_handler.py | 116 ++++++++ tests/unit/cli/test_generator_k8s.py | 6 +- tests/unit/processor/pre/k8s/conftest.py | 15 +- tests/unit/processor/pre/k8s/test_actor.py | 88 ++++++ tests/unit/processor/pre/k8s/test_handlers.py | 112 ++++++-- .../pre/k8s/test_metadata_cache_manager.py | 127 --------- .../pre/k8s/test_metadata_registry.py | 60 ++++ .../processor/pre/k8s/test_monitor_agent.py | 263 +++++++++++++----- .../pre/k8s/test_pod_event_handler.py | 192 +++++++++++++ 15 files changed, 853 insertions(+), 388 deletions(-) rename src/powerapi/processor/pre/k8s/{metadata_cache_manager.py => metadata_registry.py} (56%) create mode 100644 src/powerapi/processor/pre/k8s/pod_event_handler.py create mode 100644 tests/unit/processor/pre/k8s/test_actor.py delete mode 100644 tests/unit/processor/pre/k8s/test_metadata_cache_manager.py create mode 100644 tests/unit/processor/pre/k8s/test_metadata_registry.py create mode 100644 tests/unit/processor/pre/k8s/test_pod_event_handler.py diff --git a/src/powerapi/cli/common_cli_parsing_manager.py b/src/powerapi/cli/common_cli_parsing_manager.py index 929821a7..9b9609e8 100644 --- a/src/powerapi/cli/common_cli_parsing_manager.py +++ b/src/powerapi/cli/common_cli_parsing_manager.py @@ -460,6 +460,12 @@ def _register_k8s_pre_processor_parser(self): help_text='Kubernetes API host for manual API mode', ) + subparser_k8s_pre_processor.add_argument( + 'l', 'labels', + help_text='Comma-separated list of Kubernetes pod labels added to reports as metadata', + argument_type=list + ) + self.add_subgroup_parser('pre-processor', subparser_k8s_pre_processor) def _register_openstack_pre_processor_parser(self): diff --git a/src/powerapi/cli/generator.py b/src/powerapi/cli/generator.py index 4da9e999..0bff3585 100644 --- a/src/powerapi/cli/generator.py +++ b/src/powerapi/cli/generator.py @@ -39,6 +39,7 @@ from powerapi.puller import PullerActor from powerapi.pusher import PusherActor from powerapi.report import HWPCReport, PowerReport, Report, FormulaReport +from powerapi.utils.metadata import build_metadata_mapping COMPONENT_TYPE_KEY = 'type' COMPONENT_MODEL_KEY = 'model' @@ -435,17 +436,18 @@ def _k8s_pre_processor_factory(processor_config: dict) -> ProcessorActor: :param processor_config: Pre-Processor configuration :return: Configured Kubernetes pre-processor actor """ - from powerapi.processor.pre.k8s.actor import K8sPreProcessorActor - from powerapi.processor.pre.k8s.monitor_agent import K8sMonitorConfig + from powerapi.processor.pre.k8s.actor import KubernetesPreProcessorActor + from powerapi.processor.pre.k8s.monitor_agent import KubernetesMonitorConfig 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) - monitor_config = K8sMonitorConfig(api_mode, api_host, api_key) + label_mapping = build_metadata_mapping(processor_config.get('labels', []), prefix='k8s_pod_label_') + monitor_config = KubernetesMonitorConfig(api_mode, api_host, api_key, label_mapping) name = processor_config[ACTOR_NAME_KEY] level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO - return K8sPreProcessorActor(name, monitor_config, level_logger) + return KubernetesPreProcessorActor(name, monitor_config, level_logger) @staticmethod def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor: diff --git a/src/powerapi/processor/pre/k8s/actor.py b/src/powerapi/processor/pre/k8s/actor.py index 4a866496..16c021b7 100644 --- a/src/powerapi/processor/pre/k8s/actor.py +++ b/src/powerapi/processor/pre/k8s/actor.py @@ -31,45 +31,48 @@ from multiprocessing import Manager from powerapi.actor import Actor, State -from powerapi.actor.message import StartMessage, PoisonPillMessage +from powerapi.actor.message import PoisonPillMessage, StartMessage from powerapi.processor.processor_actor import ProcessorActor from powerapi.report import HWPCReport -from .handlers import K8sPreProcessorActorHWPCReportHandler -from .handlers import K8sPreProcessorActorStartMessageHandler, K8sPreProcessorActorPoisonPillMessageHandler -from .metadata_cache_manager import K8sMetadataCacheManager -from .monitor_agent import K8sMonitorAgent, K8sMonitorConfig +from .handlers import ( + ActorPoisonPillMessageHandler, + ActorStartMessageHandler, + HWPCReportHandler, +) +from .metadata_registry import KubernetesMetadataRegistry +from .monitor_agent import KubernetesMonitorAgent, KubernetesMonitorConfig -class K8sProcessorState(State): + +class KubernetesProcessorState(State): """ State of the Kubernetes processor actor. """ - def __init__(self, actor: Actor, monitor_config: K8sMonitorConfig): + def __init__(self, actor: Actor, monitor_config: KubernetesMonitorConfig): """ Initializes a Kubernetes pre-processor state. """ super().__init__(actor) self.manager = Manager() - self.metadata_cache_manager = K8sMetadataCacheManager(self.manager) - self.monitor_agent = K8sMonitorAgent(self.metadata_cache_manager, monitor_config) + self.metadata_registry = KubernetesMetadataRegistry(self.manager) + self.monitor_agent = KubernetesMonitorAgent(self.metadata_registry, monitor_config) -class K8sPreProcessorActor(ProcessorActor): +class KubernetesPreProcessorActor(ProcessorActor): """ Pre-Processor Actor that adds Kubernetes related metadata to reports. """ - def __init__(self, name: str, monitor_config: K8sMonitorConfig, level_logger: int = logging.WARNING, timeout: int = 5000): + def __init__(self, name: str, monitor_config: KubernetesMonitorConfig, level_logger: int = logging.WARNING): """ Initializes a Kubernetes pre-processor actor. :param name: The name of the actor :param monitor_config: Configuration of the monitoring agent :param level_logger: logging level of the actor - :param timeout: timeout in seconds """ - super().__init__(name, level_logger, timeout) + super().__init__(name, level_logger, 5000) self.monitor_config = monitor_config @@ -77,8 +80,8 @@ def setup(self): """ Set up the Kubernetes pre-processor actor. """ - self.state = K8sProcessorState(self, self.monitor_config) + self.state = KubernetesProcessorState(self, self.monitor_config) - self.add_handler(StartMessage, K8sPreProcessorActorStartMessageHandler(self.state)) - self.add_handler(HWPCReport, K8sPreProcessorActorHWPCReportHandler(self.state)) - self.add_handler(PoisonPillMessage, K8sPreProcessorActorPoisonPillMessageHandler(self.state)) + self.add_handler(StartMessage, ActorStartMessageHandler(self.state)) + self.add_handler(PoisonPillMessage, ActorPoisonPillMessageHandler(self.state)) + self.add_handler(HWPCReport, HWPCReportHandler(self.state)) diff --git a/src/powerapi/processor/pre/k8s/handlers.py b/src/powerapi/processor/pre/k8s/handlers.py index 4b5a51bd..d215e988 100644 --- a/src/powerapi/processor/pre/k8s/handlers.py +++ b/src/powerapi/processor/pre/k8s/handlers.py @@ -27,13 +27,17 @@ # 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 +from powerapi.handler import PoisonPillMessageHandler, StartHandler from powerapi.processor.handlers import ProcessorReportHandler from powerapi.report import HWPCReport -from ._utils import extract_container_id_from_k8s_cgroups_path, is_target_a_valid_k8s_cgroups_path +from ._utils import ( + extract_container_id_from_k8s_cgroups_path, + is_target_a_valid_k8s_cgroups_path, +) -class K8sPreProcessorActorStartMessageHandler(StartHandler): + +class ActorStartMessageHandler(StartHandler): """ Start message handler for the Kubernetes processor actor. """ @@ -48,7 +52,7 @@ def initialization(self): self.state.monitor_agent.start() -class K8sPreProcessorActorPoisonPillMessageHandler(PoisonPillMessageHandler): +class ActorPoisonPillMessageHandler(PoisonPillMessageHandler): """ Poison Pill message handler for the Kubernetes processor actor. """ @@ -66,7 +70,7 @@ def teardown(self, soft: bool = False): actor.disconnect() -class K8sPreProcessorActorHWPCReportHandler(ProcessorReportHandler): +class HWPCReportHandler(ProcessorReportHandler): """ HWPCReport message handler for the Kubernetes processor actor. """ @@ -78,14 +82,12 @@ def handle(self, msg: HWPCReport): """ if is_target_a_valid_k8s_cgroups_path(msg.target): container_id = extract_container_id_from_k8s_cgroups_path(msg.target) - container_metadata = self.state.metadata_cache_manager.get_container_metadata(container_id) - + container_metadata = self.state.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 - msg.target = container_metadata.container_name - msg.metadata['k8s'] = vars(container_metadata) + msg.metadata.update(container_metadata) self._send_report(msg) diff --git a/src/powerapi/processor/pre/k8s/metadata_cache_manager.py b/src/powerapi/processor/pre/k8s/metadata_registry.py similarity index 56% rename from src/powerapi/processor/pre/k8s/metadata_cache_manager.py rename to src/powerapi/processor/pre/k8s/metadata_registry.py index 600b1c6a..ece09d36 100644 --- a/src/powerapi/processor/pre/k8s/metadata_cache_manager.py +++ b/src/powerapi/processor/pre/k8s/metadata_registry.py @@ -28,58 +28,32 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. from collections.abc import MutableMapping -from dataclasses import dataclass from multiprocessing.managers import SyncManager -ADDED_EVENT = 'ADDED' -DELETED_EVENT = 'DELETED' -MODIFIED_EVENT = 'MODIFIED' - -@dataclass -class K8sContainerMetadata: - """ - Represents a metadata cache entry for Kubernetes containers. - """ - container_id: str - container_name: str - namespace: str - pod_name: str - pod_labels: dict - - -class K8sMetadataCacheManager: +class KubernetesMetadataRegistry: """ - Kubernetes container metadata cache manager. + Kubernetes metadata registry. """ def __init__(self, manager: SyncManager): """ - :param manager: Manager of the shared metadata cache + :param manager: Manager of the shared metadata registry """ - self.metadata_cache: MutableMapping[str, K8sContainerMetadata] = manager.dict() + self._container_metadata: MutableMapping[str, dict[str, str]] = manager.dict() - def update_container_metadata(self, event: str, container_metadata: K8sContainerMetadata): + def set_metadata(self, container_id: str, metadata: dict[str, str]) -> None: """ - Updates the metadata cache according to an event. - :param event: Event of the metadata cache - :param container_metadata: Container metadata entry + Set the metadata for the given container ID. + :param container_id: Container ID + :param metadata: Metadata entry """ - if event in {ADDED_EVENT, MODIFIED_EVENT}: - self.metadata_cache[container_metadata.container_id] = container_metadata - if event == DELETED_EVENT: - self.metadata_cache.pop(container_metadata.container_id, None) + self._container_metadata[container_id] = metadata - def get_container_metadata(self, container_id: str) -> K8sContainerMetadata | None: + def get_metadata(self, container_id: str) -> dict[str, str] | None: """ - Get metadata for a specific container from the cache. + Get metadata for a specific container. :param container_id: Container ID (hexadecimal string of 64 characters, short format is not supported) - :return: Container metadata entry - """ - return self.metadata_cache.get(container_id) - - def clear_metadata_cache(self): - """ - Clears all container metadata entries from the cache. + :return: Metadata entry or None if not found """ - self.metadata_cache.clear() + return self._container_metadata.get(container_id) diff --git a/src/powerapi/processor/pre/k8s/monitor_agent.py b/src/powerapi/processor/pre/k8s/monitor_agent.py index 53fd4f3d..2feaef9a 100644 --- a/src/powerapi/processor/pre/k8s/monitor_agent.py +++ b/src/powerapi/processor/pre/k8s/monitor_agent.py @@ -29,34 +29,38 @@ import logging import sys +from collections.abc import Mapping from dataclasses import dataclass -from multiprocessing import Process, Event -from signal import signal, SIGTERM, SIGINT +from multiprocessing import Event, Process +from signal import SIGINT, SIGTERM, signal -from kubernetes import client, config, watch -from kubernetes.client import V1Pod, V1PodList, V1ContainerStatus -from kubernetes.client.rest import ApiException +from kubernetes.client import ApiClient, ApiException, Configuration, CoreV1Api +from kubernetes.config import load_incluster_config, load_kube_config +from kubernetes.watch import Watch from urllib3.exceptions import ProtocolError -from .metadata_cache_manager import K8sMetadataCacheManager, K8sContainerMetadata, ADDED_EVENT, MODIFIED_EVENT, DELETED_EVENT +from .metadata_registry import KubernetesMetadataRegistry +from .pod_event_handler import KubernetesPodEventHandler -K8S_MONITOR_RETRY_DELAY_SECONDS = 1.0 +K8S_MONITOR_RETRY_DELAY_SECONDS = 5.0 @dataclass(frozen=True) -class K8sMonitorConfig: +class KubernetesMonitorConfig: """ Kubernetes monitoring agent configuration. :param api_mode: Kubernetes API mode (manual, local, cluster) :param api_host: Kubernetes API host to connect to :param api_key: Kubernetes API key (Bearer Token) to authenticate with + :param label_mapping: Mapping from Kubernetes pod label names to canonical report metadata names """ api_mode: str - api_host: str | None = None - api_key: str | None = None + api_host: str | None + api_key: str | None + label_mapping: Mapping[str, str] -def load_manual_k8s_config(configuration: client.Configuration, api_host: str | None, api_key: str | None) -> None: +def load_manual_k8s_config(configuration: Configuration, api_host: str | None, api_key: str | None) -> None: """ Setup Kubernetes API client configuration manually. This method only supports authentication by Bearer Token. @@ -75,7 +79,7 @@ def load_manual_k8s_config(configuration: client.Configuration, api_host: str | configuration.api_key_prefix['authorization'] = 'Bearer' -def build_k8s_api_client_configuration(api_mode: str, api_host: str | None, api_key: str | None) -> client.Configuration: +def build_k8s_api_client_configuration(api_mode: str, api_host: str | None, api_key: str | None) -> Configuration: """ Build a Kubernetes API client configuration. :param api_mode: The Kubernetes API mode (manual, local, cluster) @@ -83,14 +87,14 @@ def build_k8s_api_client_configuration(api_mode: str, api_host: str | None, api_ :param api_key: The Kubernetes API key (Bearer Token) :return: Kubernetes API client configuration """ - configuration = client.Configuration() + configuration = Configuration() match api_mode.casefold(): case 'local': # Setup Kubernetes API client with a kube-config file. (from KUBECONFIG environment variable, or ~/.kube/config) - config.load_kube_config(client_configuration=configuration) + load_kube_config(client_configuration=configuration) case 'cluster': # Setup Kubernetes API client with the pod service account. (requires PowerAPI to be running in a pod) - config.load_incluster_config(client_configuration=configuration) + load_incluster_config(client_configuration=configuration) case 'manual': load_manual_k8s_config(configuration, api_host, api_key) case _: @@ -99,16 +103,16 @@ def build_k8s_api_client_configuration(api_mode: str, api_host: str | None, api_ return configuration -class K8sMonitorAgent(Process): +class KubernetesMonitorAgent(Process): """ Background monitoring agent that update the shared metadata cache from Kubernetes API events. """ - def __init__(self, cache_manager: K8sMetadataCacheManager, conf: K8sMonitorConfig, level_logger: int = logging.WARNING): + def __init__(self, registry: KubernetesMetadataRegistry, config: KubernetesMonitorConfig, level_logger: int = logging.WARNING): """ - :param K8sMetadataCacheManager cache_manager: Metadata cache manager - :param conf: Configuration of the k8s processor actor - :param int level_logger: The logger level + :param registry: Metadata cache registry + :param config: Configuration of the monitoring agent + :param level_logger: The logger level """ super().__init__(name='k8s-processor-monitor-agent') @@ -118,22 +122,22 @@ def __init__(self, cache_manager: K8sMetadataCacheManager, conf: K8sMonitorConfi handler = logging.StreamHandler() handler.setFormatter(formatter) - self.metadata_cache_manager = cache_manager - self.config = conf + self.config = config + self.pod_event_handler = KubernetesPodEventHandler(registry, config.label_mapping) self._stop_monitoring = Event() @staticmethod - def build_k8s_api_client(api_config: client.Configuration) -> client.CoreV1Api: + def build_k8s_api_client(api_config: Configuration) -> CoreV1Api: """ Build a Kubernetes API client with the given configuration. :param api_config: Kubernetes API configuration :return: Kubernetes API client """ - api_client = client.ApiClient(configuration=api_config) - return client.CoreV1Api(api_client) + api_client = ApiClient(configuration=api_config) + return CoreV1Api(api_client) - def _setup_signal_handlers(self): + def _setup_signal_handlers(self) -> None: """ Setup signal handlers for the current Process. """ @@ -144,91 +148,50 @@ def stop_monitor(_, __): signal(SIGTERM, stop_monitor) signal(SIGINT, stop_monitor) - def run(self): + def run(self) -> None: """ Main code executed by the Kubernetes monitor agent. """ self._setup_signal_handlers() - self.metadata_cache_manager.clear_metadata_cache() # Prevents orphaned cache entries. - api_config = build_k8s_api_client_configuration(self.config.api_mode, self.config.api_host, self.config.api_key) api_client = self.build_k8s_api_client(api_config) + while not self._stop_monitoring.is_set(): - resource_id = self.fetch_list_all_pod_for_all_namespaces(api_client) - self.watch_list_pod_for_all_namespaces(api_client, resource_id) + try: + resource_id = self.fetch_list_all_pod_for_all_namespaces(api_client) + self.watch_list_pod_for_all_namespaces(api_client, resource_id) + except ApiException as e: + logging.error("Kubernetes API request failed: %s %s", e.status, e.reason) + except ProtocolError as e: + logging.error("Failed to connect to Kubernetes API: %s", e) if self._stop_monitoring.wait(K8S_MONITOR_RETRY_DELAY_SECONDS): break - @staticmethod - def get_containers_id_name_from_statuses(container_statuses: list[V1ContainerStatus]) -> dict[str, str]: - """ - Extract containers ID and name from the statuses. - :param container_statuses: List of container statuses - :return: Dictionary mapping the containers ID to their name - """ - return { - container_status.container_id.split('://')[1]: container_status.name - for container_status in container_statuses or [] if container_status.container_id is not None - } - - def build_metadata_cache_entries_from_pod(self, pod: V1Pod) -> list[K8sContainerMetadata]: - """ - Build and return metadata cache entries from a Kubernetes pod object. - :param pod: Kubernetes pod - :return: List of metadata cache entries - """ - pod_name = pod.metadata.name - pod_labels = pod.metadata.labels - namespace = pod.metadata.namespace - container_statuses = pod.status.container_statuses - return [ - K8sContainerMetadata(container_id, container_name, namespace, pod_name, pod_labels) - for container_id, container_name in self.get_containers_id_name_from_statuses(container_statuses).items() - ] - - def fetch_list_all_pod_for_all_namespaces(self, api_client: client.CoreV1Api) -> int | None: + def fetch_list_all_pod_for_all_namespaces(self, api_client: CoreV1Api) -> str | None: """ Fetch all pod for all namespaces and populate the metadata cache. :param api_client: Kubernetes api client :return: Resource version of the last fetched entry """ - resource_version = None - try: - pods: V1PodList = api_client.list_pod_for_all_namespaces(watch=False) - resource_version = pods.metadata.resource_version - for pod in pods.items: - for entry in self.build_metadata_cache_entries_from_pod(pod): - self.metadata_cache_manager.update_container_metadata(ADDED_EVENT, entry) - - except ApiException as e: - logging.warning('API exception caught in fetch: %s %s', e.status, e.reason) - except ProtocolError as e: - logging.warning('Protocol error caught in fetch: %s', e) + pods = api_client.list_pod_for_all_namespaces(watch=False) + for pod in pods.items: + self.pod_event_handler.handle("ADDED", pod) - return resource_version + return pods.metadata.resource_version - def watch_list_pod_for_all_namespaces(self, api_client: client.CoreV1Api, resource_version: int | None = None): + def watch_list_pod_for_all_namespaces(self, api_client: CoreV1Api, resource_version: str | None = None) -> None: """ Watch k8s pods events for all namespaces and update the local metadata cache accordingly. :param api_client: Kubernetes API client :param resource_version: Resource version from where the watcher begin """ + w = Watch() try: - w = watch.Watch() for event in w.stream(api_client.list_pod_for_all_namespaces, resource_version=resource_version): - event_type = event["type"] - if event_type not in {ADDED_EVENT, MODIFIED_EVENT, DELETED_EVENT}: - logging.warning('Unexpected pod event: %s', event_type) - continue - - for entry in self.build_metadata_cache_entries_from_pod(event["object"]): - self.metadata_cache_manager.update_container_metadata(event_type, entry) - + self.pod_event_handler.handle(event["type"], event["object"]) + except ValueError as e: + logging.warning("Failed to process event: %s", e) + finally: w.stop() - - except ApiException as e: - logging.warning('API exception caught in watcher: %s %s', e.status, e.reason) - except ProtocolError as e: - logging.warning('Protocol error caught in watcher: %s', e) diff --git a/src/powerapi/processor/pre/k8s/pod_event_handler.py b/src/powerapi/processor/pre/k8s/pod_event_handler.py new file mode 100644 index 00000000..7c854ff0 --- /dev/null +++ b/src/powerapi/processor/pre/k8s/pod_event_handler.py @@ -0,0 +1,116 @@ +# 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 __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING + +from .metadata_registry import KubernetesMetadataRegistry + +if TYPE_CHECKING: + from kubernetes.client import V1ContainerStatus, V1Pod + + +class KubernetesPodEventHandler: + """ + Translates Kubernetes pod events into metadata registry operations. + """ + + def __init__(self, metadata_registry: KubernetesMetadataRegistry, labels_mapping: Mapping[str, str]): + """ + Initialize the Kubernetes pod event handler. + :param metadata_registry: Registry containing metadata + :param labels_mapping: Mapping from Kubernetes pod label names to canonical report metadata names + """ + self.metadata_registry = metadata_registry + self.label_mapping = labels_mapping + + def handle(self, event_type: str, pod: V1Pod) -> None: + """ + Apply a Kubernetes pod event to the metadata registry. + :param event_type: Kubernetes pod event type + :param pod: Pod associated with the event + :raises ValueError: If the event type is unsupported + """ + match event_type: + case 'ADDED' | 'MODIFIED': + self._set_pod_metadata(pod) + case 'DELETED': + # Retain metadata so reports already in the pipeline can still be enriched. + return + case _: + raise ValueError(f'Unexpected Kubernetes pod event: {event_type}') + + @staticmethod + def _extract_container_id(container_status: V1ContainerStatus) -> str: + """ + Extract the runtime-specific ID from a Kubernetes container status. + :param container_status: Container status from which to extract the ID + :return: Container ID without its runtime prefix + """ + return container_status.container_id.partition('://')[2] + + def _build_pod_metadata(self, pod: V1Pod) -> dict[str, str]: + """ + Build the report metadata shared by all containers of a pod. + :param pod: Pod from which to build metadata + :return: Mapping of the metadata shared by the pod containers + """ + metadata = { + 'k8s_pod_name': pod.metadata.name, + 'k8s_pod_namespace': pod.metadata.namespace, + } + + if pod.metadata.labels: + for source_name, canonical_name in self.label_mapping.items(): + if source_name in pod.metadata.labels: + metadata[canonical_name] = pod.metadata.labels[source_name] + + return metadata + + def _set_pod_metadata(self, pod: V1Pod) -> None: + """ + Register metadata for every started container of a pod. + Container statuses without a valid runtime ID are ignored. + :param pod: Pod whose container metadata should be registered + """ + pod_metadata = self._build_pod_metadata(pod) + + if pod.status.container_statuses: + for container_status in pod.status.container_statuses: + if container_status.container_id is None: + continue + + container_id = self._extract_container_id(container_status) + + container_metadata = pod_metadata.copy() + container_metadata['k8s_container_name'] = container_status.name + container_metadata['k8s_container_image'] = container_status.image + + self.metadata_registry.set_metadata(container_id, container_metadata) diff --git a/tests/unit/cli/test_generator_k8s.py b/tests/unit/cli/test_generator_k8s.py index d4694e99..8c5eb5c3 100644 --- a/tests/unit/cli/test_generator_k8s.py +++ b/tests/unit/cli/test_generator_k8s.py @@ -28,11 +28,11 @@ import pytest -pytest.importorskip('powerapi.processor.pre.k8s.actor') # The Kubernetes processor requires external dependencies to work. +pytest.importorskip('kubernetes') from powerapi.cli.generator import PreProcessorGenerator from powerapi.exception import PowerAPIException -from powerapi.processor.pre.k8s.actor import K8sPreProcessorActor +from powerapi.processor.pre.k8s.actor import KubernetesPreProcessorActor @pytest.fixture @@ -66,7 +66,7 @@ def test_preprocessor_generator_with_valid_k8s_config(k8s_processor_config): assert 'pytest-k8s-preprocessor' in preprocessors preprocessor = preprocessors['pytest-k8s-preprocessor'] - assert isinstance(preprocessor, K8sPreProcessorActor) + assert isinstance(preprocessor, KubernetesPreProcessorActor) expected_preprocessor_attributes = k8s_processor_config['pre-processor']['pytest-k8s-preprocessor'] assert preprocessor.monitor_config.api_mode == expected_preprocessor_attributes['api-mode'] diff --git a/tests/unit/processor/pre/k8s/conftest.py b/tests/unit/processor/pre/k8s/conftest.py index 0d27119e..ab97bcc9 100644 --- a/tests/unit/processor/pre/k8s/conftest.py +++ b/tests/unit/processor/pre/k8s/conftest.py @@ -27,19 +27,18 @@ # 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 multiprocessing import Manager +from unittest.mock import Mock import pytest -from powerapi.processor.pre.k8s.metadata_cache_manager import K8sMetadataCacheManager +from powerapi.processor.pre.k8s.metadata_registry import KubernetesMetadataRegistry @pytest.fixture -def initialized_metadata_cache_manager(): +def metadata_registry(): """ - Returns an initialized metadata cache manager. + Return a metadata registry. """ - manager = Manager() - yield K8sMetadataCacheManager(manager) - - manager.shutdown() + manager = Mock() + manager.dict.return_value = {} + return KubernetesMetadataRegistry(manager) diff --git a/tests/unit/processor/pre/k8s/test_actor.py b/tests/unit/processor/pre/k8s/test_actor.py new file mode 100644 index 00000000..ab5a789d --- /dev/null +++ b/tests/unit/processor/pre/k8s/test_actor.py @@ -0,0 +1,88 @@ +# 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 datetime import datetime +from unittest.mock import Mock, patch + +import pytest + +pytest.importorskip("kubernetes") + +from powerapi.actor import State +from powerapi.actor.message import PoisonPillMessage, StartMessage +from powerapi.processor.pre.k8s.actor import ( + KubernetesPreProcessorActor, + KubernetesProcessorState, +) +from powerapi.processor.pre.k8s.handlers import ( + ActorPoisonPillMessageHandler, + ActorStartMessageHandler, + HWPCReportHandler, +) +from powerapi.report import HWPCReport + + +def test_processor_state_builds_metadata_components(): + """ + The processor state should initialize its manager, registry, and monitor agent. + """ + monitor_config = Mock() + + with ( + patch("powerapi.processor.pre.k8s.actor.Manager"), + patch("powerapi.processor.pre.k8s.actor.KubernetesMetadataRegistry") as metadata_registry_class, + patch("powerapi.processor.pre.k8s.actor.KubernetesMonitorAgent") as monitor_agent_class, + ): + state = KubernetesProcessorState(Mock(), monitor_config) + + metadata_registry_class.assert_called_once_with(state.manager) + monitor_agent_class.assert_called_once_with(state.metadata_registry, monitor_config) + + +@pytest.mark.parametrize( + ("message", "expected_handler_type"), + [ + (StartMessage(), ActorStartMessageHandler), + (PoisonPillMessage(), ActorPoisonPillMessageHandler), + (HWPCReport(datetime.now(), "pytest", "pytest", {}), HWPCReportHandler) + ], + ids=["start_message", "poison_pill_message", "hwpc_report"], +) +def test_actor_setup_registers_kubernetes_handlers(message, expected_handler_type): + """ + Actor setup should create its state and register Kubernetes handlers. + """ + monitor_config = Mock() + actor = KubernetesPreProcessorActor("pytest", monitor_config) + state = State(actor) + + with patch("powerapi.processor.pre.k8s.actor.KubernetesProcessorState", return_value=state): + actor.setup() + + assert actor.state is state + assert isinstance(state.get_corresponding_handler(message), expected_handler_type) diff --git a/tests/unit/processor/pre/k8s/test_handlers.py b/tests/unit/processor/pre/k8s/test_handlers.py index c5e60d97..3aef636e 100644 --- a/tests/unit/processor/pre/k8s/test_handlers.py +++ b/tests/unit/processor/pre/k8s/test_handlers.py @@ -32,30 +32,63 @@ import pytest -pytest.importorskip('powerapi.processor.pre.k8s.actor') # The actor module requires external dependencies. - -from powerapi.processor.pre.k8s.actor import K8sProcessorState -from powerapi.processor.pre.k8s.monitor_agent import K8sMonitorConfig -from powerapi.processor.pre.k8s.handlers import K8sPreProcessorActorHWPCReportHandler -from powerapi.processor.pre.k8s.metadata_cache_manager import K8sContainerMetadata +from powerapi.processor.pre.k8s.handlers import ( + ActorPoisonPillMessageHandler, + ActorStartMessageHandler, + HWPCReportHandler, +) from powerapi.report import HWPCReport @pytest.fixture -def hwpc_report_handler(): +def start_message_handler(): """ - Factory fixture creating an HwPC report handler. + Factory fixture creating a start message handler. """ - def _create_handler() -> tuple[K8sPreProcessorActorHWPCReportHandler, list[HWPCReport]]: + def _create_handler() -> ActorStartMessageHandler: actor = Mock(name='processor-actor') actor.target_actors = [Mock(name='target_actor_a'), Mock(name='target_actor_b')] - monitor_config = K8sMonitorConfig('manual', 'https://localhost:6443', 'pytest-token') - state = K8sProcessorState(actor, monitor_config) - state.metadata_cache_manager = Mock(name='metadata_cache_manager') + state = Mock(name='state') + state.actor = actor + state.monitor_agent = Mock(name='monitor_agent') + + return ActorStartMessageHandler(state) + + return _create_handler + + +@pytest.fixture +def poison_pill_message_handler(): + """ + Factory fixture creating a Poison-Pill message handler. + """ + + def _create_handler() -> ActorPoisonPillMessageHandler: + actor = Mock(name='processor-actor') + actor.target_actors = [Mock(name='target_actor_a'), Mock(name='target_actor_b')] + + state = Mock(name='state') + state.actor = actor + state.manager = Mock(name='manager') + state.monitor_agent = Mock(name='monitor_agent') + + return ActorPoisonPillMessageHandler(state) + + return _create_handler + - handler = K8sPreProcessorActorHWPCReportHandler(state) +@pytest.fixture +def hwpc_report_handler(): + """ + Factory fixture creating an HwPC report handler. + """ + + def _create_handler() -> tuple[HWPCReportHandler, list[HWPCReport]]: + state = Mock(name='state') + state.metadata_registry = Mock(name='metadata_registry') + handler = HWPCReportHandler(state) reports_sent = [] handler._send_report = Mock(side_effect=lambda msg: reports_sent.append(msg)) @@ -84,6 +117,34 @@ def make_pod_hwpc_report() -> tuple[HWPCReport, str]: return HWPCReport(timestamp, sensor, str(target), {}, metadata), container_id +def test_start_handler_connects_targets_and_starts_monitor(start_message_handler): + """ + The start handler should connect target actors and start monitoring. + """ + handler = start_message_handler() + + handler.initialization() + + handler.state.monitor_agent.start.assert_called_once() + for actor in handler.state.actor.target_actors: + actor.connect_data.assert_called_once() + + +def test_poison_pill_handler_stops_resources_and_disconnects_targets(poison_pill_message_handler): + """ + The Poison-Pill handler should stop resources and disconnect targets. + """ + handler = poison_pill_message_handler() + + handler.teardown() + + handler.state.monitor_agent.terminate.assert_called_once() + handler.state.monitor_agent.join.assert_called_once() + handler.state.manager.shutdown.assert_called_once() + for actor in handler.state.actor.target_actors: + actor.disconnect.assert_called_once() + + def test_hwpc_report_handler_adds_k8s_metadata_and_forwards_report(hwpc_report_handler): """ Test that the HwPC report handler forwards a report for a valid k8s target when it is in the metadata cache. @@ -91,23 +152,22 @@ def test_hwpc_report_handler_adds_k8s_metadata_and_forwards_report(hwpc_report_h handler, reports_sent = hwpc_report_handler() report, container_id = make_pod_hwpc_report() - container_metadata = K8sContainerMetadata( - container_id=container_id, - container_name='powerapi-test-container', - namespace='powerapi-test-namespace', - pod_name='powerapi-test-pod', - pod_labels={'app': 'powerapi'} - ) - handler.state.metadata_cache_manager.get_container_metadata.return_value = container_metadata + k8s_metadata = { + 'k8s_container_name': 'powerapi-test-container', + 'k8s_pod_namespace': 'powerapi-test-namespace', + 'k8s_pod_name': 'powerapi-test-pod', + 'k8s_label_app': 'powerapi', + } + handler.state.metadata_registry.get_metadata.return_value = k8s_metadata handler.handle(report) + handler.state.metadata_registry.get_metadata.assert_called_once_with(container_id) assert reports_sent == [report] (processed_report,) = reports_sent - assert processed_report.target == container_metadata.container_name - assert processed_report.metadata['k8s'] == vars(container_metadata) - assert processed_report.metadata['scope'] == 'pytest' + assert processed_report.target == report.target + assert processed_report.metadata == report.metadata | k8s_metadata def test_hwpc_report_handler_drops_report_when_container_metadata_missing(hwpc_report_handler): @@ -117,10 +177,11 @@ def test_hwpc_report_handler_drops_report_when_container_metadata_missing(hwpc_r handler, reports_sent = hwpc_report_handler() report, _ = make_pod_hwpc_report() - handler.state.metadata_cache_manager.get_container_metadata.return_value = None + handler.state.metadata_registry.get_metadata.return_value = None handler.handle(report) + handler.state.metadata_registry.get_metadata.assert_called_once() assert reports_sent == [] @@ -138,3 +199,4 @@ def test_hwpc_report_handler_forwards_non_k8s_targets(hwpc_report_handler): assert processed_report.target == 'test-container' assert processed_report.metadata == {'scope': 'pytest'} + handler.state.metadata_registry.get_metadata.assert_not_called() diff --git a/tests/unit/processor/pre/k8s/test_metadata_cache_manager.py b/tests/unit/processor/pre/k8s/test_metadata_cache_manager.py deleted file mode 100644 index 884067d2..00000000 --- a/tests/unit/processor/pre/k8s/test_metadata_cache_manager.py +++ /dev/null @@ -1,127 +0,0 @@ -# Copyright (c) 2024, Inria -# Copyright (c) 2024, 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.k8s.metadata_cache_manager import ADDED_EVENT, MODIFIED_EVENT, DELETED_EVENT -from powerapi.processor.pre.k8s.metadata_cache_manager import K8sContainerMetadata - - -def _generate_metadata_cache_entry(container_id: str, counter: int = 0) -> K8sContainerMetadata: - """ - Generate a K8sContainerMetadata object from the given parameters. - """ - container_name = 'powerapi-test-container' - namespace = 'powerapi-test-namespace' - pod_name = 'powerapi-test-pod' - pod_labels = { - 'app.kubernetes.io/name': 'powerapi-test-app', - 'app.kubernetes.io/instance': 'powerapi-test-app-abcxzy', - 'app.kubernetes.io/version': 'v1.0.0', - 'app.kubernetes.io/component': 'test', - 'app.kubernetes.io/part-of': 'powerapi-test', - 'helm.sh/chart': 'powerapi-test-1.0.0', - 'powerapi.org/test-counter': counter - } - return K8sContainerMetadata(container_id, container_name, namespace, pod_name, pod_labels) - - -def test_container_metadata_cache_with_added_event(initialized_metadata_cache_manager): - """ - Test that an 'ADDED' event add the entry to the metadata cache. - """ - container_id = 'e6eb9dd88e7189933861634cc9626b3a85a1f6425989caa51094df34c34c2787' - entry = _generate_metadata_cache_entry(container_id) - initialized_metadata_cache_manager.update_container_metadata(ADDED_EVENT, entry) - - assert initialized_metadata_cache_manager.get_container_metadata(container_id) == entry - - -def test_container_metadata_cache_with_updated_event(initialized_metadata_cache_manager): - """ - Test that an 'MODIFIED' event update the entry in the metadata cache. - """ - container_id = '35d31dfb0b83cf9d6c689711d9ab6f4667f6784107df3783a3333492cdbcbce2' - - first_entry = _generate_metadata_cache_entry(container_id, 0) - initialized_metadata_cache_manager.update_container_metadata(ADDED_EVENT, first_entry) - - second_entry = _generate_metadata_cache_entry(container_id, 1) - initialized_metadata_cache_manager.update_container_metadata(MODIFIED_EVENT, second_entry) - - assert initialized_metadata_cache_manager.get_container_metadata(container_id) == second_entry - - -def test_container_metadata_cache_with_deleted_event(initialized_metadata_cache_manager): - """ - Test that an 'DELETED' event remove the entry from the metadata cache. - """ - container_id = '885754eae01c1e4d5389677d6bab564381c12500324c8da9731ac74452740faa' - entry = _generate_metadata_cache_entry(container_id) - initialized_metadata_cache_manager.update_container_metadata(DELETED_EVENT, entry) - - assert initialized_metadata_cache_manager.get_container_metadata(container_id) is None - - -def test_container_metadata_cache_with_valid_container_id(initialized_metadata_cache_manager): - """ - Test fetching a metadata cache entry with a valid container id. - """ - container_id = '5375cd086e95601123cedbd0d1dc77058fb7d9890ae67aed0f551e032e1ebf87' - - entry = _generate_metadata_cache_entry(container_id) - initialized_metadata_cache_manager.update_container_metadata(ADDED_EVENT, entry) - - fetched_entry = initialized_metadata_cache_manager.get_container_metadata(container_id) - - assert fetched_entry is not None - assert fetched_entry == entry - - -def test_container_metadata_cache_with_unknown_container_id(initialized_metadata_cache_manager): - """ - Test fetching a metadata cache entry with an unknown container id. - """ - container_id = '0000000000000000000000000000000000000000000000000000000000000000' - entry = initialized_metadata_cache_manager.get_container_metadata(container_id) - - assert entry is None - - -def test_container_metadata_cache_clear(initialized_metadata_cache_manager): - """ - Test clearing the metadata cache. - """ - container_id = '9c4b8e6491219cf5112bdc8c6aab02ff19ccc8870cda70f264a41add2dc57fbb' - entry = _generate_metadata_cache_entry(container_id) - initialized_metadata_cache_manager.update_container_metadata(ADDED_EVENT, entry) - - assert len(initialized_metadata_cache_manager.metadata_cache) == 1 - - initialized_metadata_cache_manager.clear_metadata_cache() - - assert len(initialized_metadata_cache_manager.metadata_cache) == 0 diff --git a/tests/unit/processor/pre/k8s/test_metadata_registry.py b/tests/unit/processor/pre/k8s/test_metadata_registry.py new file mode 100644 index 00000000..1e1ea1f5 --- /dev/null +++ b/tests/unit/processor/pre/k8s/test_metadata_registry.py @@ -0,0 +1,60 @@ +# 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. + + +def test_set_and_get_metadata(metadata_registry): + """ + Metadata can be retrieved using its container ID. + """ + container_id = 'e6eb9dd88e7189933861634cc9626b3a85a1f6425989caa51094df34c34c2787' + metadata = {'k8s_pod_name': 'powerapi-pytest'} + + metadata_registry.set_metadata(container_id, metadata) + + assert metadata_registry.get_metadata(container_id) == metadata + + +def test_set_metadata_replaces_existing_entry(metadata_registry): + """ + Setting metadata twice replaces the previous entry. + """ + container_id = '35d31dfb0b83cf9d6c689711d9ab6f4667f6784107df3783a3333492cdbcbce2' + metadata_registry.set_metadata(container_id, {'version': 'old'}) + + metadata_registry.set_metadata(container_id, {'version': 'new'}) + + assert metadata_registry.get_metadata(container_id) == {'version': 'new'} + + +def test_get_metadata_returns_none_for_unknown_container(metadata_registry): + """ + An unknown container ID has no associated metadata. + """ + container_id = '0000000000000000000000000000000000000000000000000000000000000000' + + assert metadata_registry.get_metadata(container_id) is None diff --git a/tests/unit/processor/pre/k8s/test_monitor_agent.py b/tests/unit/processor/pre/k8s/test_monitor_agent.py index 8eb5c732..e30a2d2c 100644 --- a/tests/unit/processor/pre/k8s/test_monitor_agent.py +++ b/tests/unit/processor/pre/k8s/test_monitor_agent.py @@ -27,121 +27,246 @@ # 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 unittest.mock import Mock, call, patch, sentinel + import pytest -pytest.importorskip('powerapi.processor.pre.k8s.monitor_agent') # The monitor agent requires external dependencies. +pytest.importorskip('kubernetes') + +from kubernetes.client import ApiException, CoreV1Api +from urllib3.exceptions import ProtocolError + +from powerapi.processor.pre.k8s.metadata_registry import KubernetesMetadataRegistry +from powerapi.processor.pre.k8s.monitor_agent import ( + KubernetesMonitorAgent, + KubernetesMonitorConfig, + build_k8s_api_client_configuration, +) +from powerapi.processor.pre.k8s.pod_event_handler import KubernetesPodEventHandler -from kubernetes.client import V1Pod, V1ContainerStatus, Configuration, V1ObjectMeta, V1PodStatus -from powerapi.processor.pre.k8s.monitor_agent import K8sMonitorAgent, K8sMonitorConfig +@pytest.fixture +def pod_event_handler(): + """ + Return a mocked Pod event handler. + """ + return Mock(spec=KubernetesPodEventHandler) @pytest.fixture -def initialized_monitor_agent(initialized_metadata_cache_manager): +def monitor_agent(pod_event_handler): """ - Returns an initialized monitor agent. + Return a monitor agent with a mocked Pod event handler. """ - monitor_config = K8sMonitorConfig('manual', 'https://localhost:6443', 'pytest-token') - agent = K8sMonitorAgent(initialized_metadata_cache_manager, monitor_config) + config = KubernetesMonitorConfig( + api_mode='manual', + api_host='https://localhost:6443', + api_key='pytest-token', + label_mapping={}, + ) + agent = KubernetesMonitorAgent(Mock(spec=KubernetesMetadataRegistry), config) + agent.pod_event_handler = pod_event_handler return agent -def generate_k8s_config_for_tests() -> Configuration: +@pytest.fixture +def watcher(): + """ + Return the watcher created by the monitor agent. + """ + with patch('powerapi.processor.pre.k8s.monitor_agent.Watch', autospec=True) as watch_class: + yield watch_class.return_value + + +@pytest.fixture +def configure_monitor_agent(monitor_agent): + """ + Configure the monitor agent for a finite run without external calls. + """ + + def _configure( + *, + fetch_pods: Mock | None = None, + watch_pods: Mock | None = None, + stop_event: Mock | None = None, + wait_results: Iterable[bool] = (True,), + ) -> KubernetesMonitorAgent: + if fetch_pods is None: + fetch_pods = Mock(return_value='pytest') + if watch_pods is None: + watch_pods = Mock() + if stop_event is None: + stop_event = Mock(spec_set=['is_set', 'wait']) + + stop_event.is_set.return_value = False + stop_event.wait.side_effect = wait_results + + monitor_agent._setup_signal_handlers = Mock() + monitor_agent._stop_monitoring = stop_event + monitor_agent.fetch_list_all_pod_for_all_namespaces = fetch_pods + monitor_agent.watch_list_pod_for_all_namespaces = watch_pods + return monitor_agent + + return _configure + + +def test_build_manual_configuration_sets_bearer_authentication(): """ - Generate a Kubernetes configuration for tests. + Manual configuration should set the API endpoint and bearer token. """ - config = Configuration() - config.client_side_validation = False # needs to be disabled for tests - return config + configuration = build_k8s_api_client_configuration( + 'manual', + 'https://powerapi:6443', + 'pytest-token', + ) + assert configuration.host == 'https://powerapi:6443' + assert configuration.api_key['authorization'] == 'pytest-token' + assert configuration.api_key_prefix['authorization'] == 'Bearer' -def generate_container_status(container_id, container_name) -> V1ContainerStatus: + +def test_build_manual_configuration_requires_api_host(): + """ + Manual configuration should reject a missing API host. + """ + with pytest.raises(ValueError, match='Kubernetes API host is not defined'): + build_k8s_api_client_configuration('manual', None, 'pytest-token') + + +def test_build_manual_configuration_requires_api_key(): """ - Generate an initialized container status object. + Manual configuration should reject a missing API key. """ - config = generate_k8s_config_for_tests() - status = V1ContainerStatus(container_id=container_id, name=container_name, local_vars_configuration=config) - return status + with pytest.raises(ValueError, match='Kubernetes API key is not defined'): + build_k8s_api_client_configuration('manual', 'https://localhost:6443', None) -def generate_pod(pod_name, pod_namespace, pod_labels, container_statuses: list[V1ContainerStatus]) -> V1Pod: +def test_build_configuration_rejects_unknown_mode(): """ - Generate an initialized POD object. + An unsupported Kubernetes API mode should be rejected. """ - config = generate_k8s_config_for_tests() - metadata = V1ObjectMeta(name=pod_name, labels=pod_labels, namespace=pod_namespace, local_vars_configuration=config) - status = V1PodStatus(container_statuses=container_statuses, local_vars_configuration=config) - pod = V1Pod(metadata=metadata, status=status, local_vars_configuration=config) - return pod + with pytest.raises(ValueError, match='Invalid Kubernetes API mode'): + build_k8s_api_client_configuration('pytest', None, None) -def test_extract_containers_id_name_from_statuses(initialized_monitor_agent): +def test_fetch_forwards_pods_and_returns_resource_version(monitor_agent, pod_event_handler): """ - Test extract the containers id and name from the statuses. + The initial list should be forwarded as added events. """ - cri = 'containerd' - cid = '0000000000000000000000000000000000000000000000000000000000000000' - container_id = f'{cri}://{cid}' - container_name = 'test-container-name' + api_client = Mock(spec=CoreV1Api) + api_client.list_pod_for_all_namespaces.return_value = Mock( + metadata=Mock(resource_version='42'), + items=[sentinel.first_pod, sentinel.second_pod], + ) - status = generate_container_status(container_id, container_name) + resource_version = monitor_agent.fetch_list_all_pod_for_all_namespaces(api_client) + + assert resource_version == '42' + api_client.list_pod_for_all_namespaces.assert_called_once_with(watch=False) + assert pod_event_handler.handle.call_args_list == [ + call('ADDED', sentinel.first_pod), + call('ADDED', sentinel.second_pod), + ] + + +def test_watch_forwards_events(monitor_agent, pod_event_handler, watcher): + """ + Watch events should be delegated in stream order. + """ + api_client = Mock(spec=CoreV1Api) + watcher.stream.return_value = iter([ + {'type': 'ADDED', 'object': sentinel.added_pod}, + {'type': 'MODIFIED', 'object': sentinel.modified_pod}, + ]) - res = initialized_monitor_agent.get_containers_id_name_from_statuses([status]) + monitor_agent.watch_list_pod_for_all_namespaces(api_client, '42') - assert res == {cid: container_name} + watcher.stream.assert_called_once() + assert watcher.stream.call_args.kwargs['resource_version'] == '42' + assert pod_event_handler.handle.call_args_list == [ + call('ADDED', sentinel.added_pod), + call('MODIFIED', sentinel.modified_pod), + ] + watcher.stop.assert_called_once() -def test_extract_containers_id_name_from_statuses_with_none_container_id(initialized_monitor_agent): +def test_watch_stops_before_propagating_transport_failure(monitor_agent, watcher): """ - Test extract the containers id and name from the statuses with None container id. - This happens when processing an event where the container is created but has not yet been started. + A transport failure should propagate after watcher cleanup. """ - container_id = None - container_name = 'test-container-name' + api_client = Mock(spec=CoreV1Api) + watcher.stream.side_effect = ProtocolError('pytest') - status = generate_container_status(container_id, container_name) + with pytest.raises(ProtocolError): + monitor_agent.watch_list_pod_for_all_namespaces(api_client, '42') - res = initialized_monitor_agent.get_containers_id_name_from_statuses([status]) + watcher.stop.assert_called_once() - assert res == {} + +def test_watch_handles_invalid_event(monitor_agent, pod_event_handler, watcher): + """ + An invalid event should end the watch without escaping the monitor. + """ + api_client = Mock(spec=CoreV1Api) + watcher.stream.return_value = iter([ + {'type': 'PYTEST', 'object': sentinel.invalid_pod}, + ]) + pod_event_handler.handle.side_effect = ValueError('unexpected event') + + monitor_agent.watch_list_pod_for_all_namespaces(api_client, '42') + + pod_event_handler.handle.assert_called_once_with('PYTEST', sentinel.invalid_pod) + watcher.stop.assert_called_once() -def test_building_metadata_cache_entry_from_pod(initialized_monitor_agent): +def test_run_waits_after_fetch_failure_without_starting_watch(configure_monitor_agent): """ - Test building metadata cache entries from a Kubernetes POD object. + A failed initial list should delay the retry without starting a watch. """ - pod_name = 'test-pod' - pod_namespace = 'powerapi' - pod_labels = {'executor': 'pytest'} + stop_event = Mock() + fetch_pods = Mock(side_effect=ApiException(status=500, reason='pytest')) + watch_pods = Mock() + monitor_agent = configure_monitor_agent( + fetch_pods=fetch_pods, + watch_pods=watch_pods, + stop_event=stop_event, + ) - cri = 'containerd' - cid = '0000000000000000000000000000000000000000000000000000000000000000' - container_id = f'{cri}://{cid}' - container_name = 'test-container' - container_statuses = [generate_container_status(container_id, container_name)] + monitor_agent.run() + + fetch_pods.assert_called_once() + watch_pods.assert_not_called() + stop_event.wait.assert_called_once() + + +def test_run_waits_after_watch_failure(configure_monitor_agent): + """ + A failed watch should delay the retry after the initial list. + """ + stop_event = Mock() + watch_pods = Mock(side_effect=ProtocolError('pytest')) + monitor_agent = configure_monitor_agent(watch_pods=watch_pods, stop_event=stop_event) - pod = generate_pod(pod_name, pod_namespace, pod_labels, container_statuses) - cache_entries = initialized_monitor_agent.build_metadata_cache_entries_from_pod(pod) - assert len(cache_entries) == 1 + monitor_agent.run() - cache_entry = cache_entries[0] - assert cache_entry.pod_name == pod_name - assert cache_entry.namespace == pod_namespace - assert cache_entry.pod_labels == pod_labels - assert cache_entry.container_id == cid - assert cache_entry.container_name == container_name + watch_pods.assert_called_once() + stop_event.wait.assert_called_once() -def test_building_metadata_cache_entry_from_pod_without_containers(initialized_monitor_agent): +def test_run_retries_after_delay(configure_monitor_agent): """ - Test building metadata cache entries from a Kubernetes POD object without containers. + The monitor should start another list-watch cycle after the retry delay. """ - pod_name = 'test-pod' - pod_namespace = 'powerapi' - pod_labels = {'executor': 'pytest'} + fetch_pods = Mock(side_effect=[ProtocolError('pytest'), '42']) + watch_pods = Mock() + monitor_agent = configure_monitor_agent( + fetch_pods=fetch_pods, + watch_pods=watch_pods, + wait_results=[False, True], + ) - container_statuses = [] + monitor_agent.run() - pod = generate_pod(pod_name, pod_namespace, pod_labels, container_statuses) - cache_entries = initialized_monitor_agent.build_metadata_cache_entries_from_pod(pod) - assert len(cache_entries) == 0 + assert fetch_pods.call_count == 2 + watch_pods.assert_called_once() diff --git a/tests/unit/processor/pre/k8s/test_pod_event_handler.py b/tests/unit/processor/pre/k8s/test_pod_event_handler.py new file mode 100644 index 00000000..c5593628 --- /dev/null +++ b/tests/unit/processor/pre/k8s/test_pod_event_handler.py @@ -0,0 +1,192 @@ +# 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 Mapping +from unittest.mock import Mock + +import pytest + +pytest.importorskip("kubernetes") + +from kubernetes.client import V1ContainerStatus, V1ObjectMeta, V1Pod, V1PodStatus + +from powerapi.processor.pre.k8s.pod_event_handler import KubernetesPodEventHandler +from powerapi.utils.metadata import build_metadata_mapping + + +def make_container_status( + container_id: str | None = None, + name: str = "powerapi", + image: str = "powerapi:pytest" +) -> V1ContainerStatus: + """ + Build the subset of a container status used by the event handler. + """ + status = Mock(spec=V1ContainerStatus) + status.name = name + status.container_id = container_id + status.image = image + return status + + +def make_pod( + name: str = "pytest", + namespace: str = "powerapi", + labels: dict[str, str] | None = None, + container_statuses: list[V1ContainerStatus] | None = None +) -> V1Pod: + """ + Build the subset of a Pod used by the event handler. + """ + metadata = Mock(spec=V1ObjectMeta) + metadata.name = name + metadata.namespace = namespace + metadata.labels = labels + + status = Mock(spec=V1PodStatus) + status.container_statuses = container_statuses + + return Mock(spec=V1Pod, metadata=metadata, status=status) + + +@pytest.fixture +def pod_event_handler(metadata_registry): + """ + Factory fixture for creating a pod event handler. + """ + + def _create_event_handler(labels_mapping: Mapping[str, str]) -> KubernetesPodEventHandler: + event_handler = KubernetesPodEventHandler(metadata_registry, labels_mapping) + return event_handler + + return _create_event_handler + + +@pytest.mark.parametrize("event_type", ["ADDED", "MODIFIED"]) +def test_event_register_containers_metadata(event_type, metadata_registry, pod_event_handler): + """ + Valid events should populate the registry for each started container in the Pod. + """ + containers = [ + ("1111111111111111111111111111111111111111111111111111111111111111", "first", "powerapi:first"), + ("2222222222222222222222222222222222222222222222222222222222222222", "second", "powerapi:second"), + ] + + event_handler = pod_event_handler(build_metadata_mapping(["app.kubernetes.io/name"], "k8s_pod_label_")) + containers_statuses = [ + make_container_status(f'pytest://{container_id}', name, image) + for container_id, name, image in containers + ] + pod = make_pod( + labels={"app.kubernetes.io/name": "pytest", "ignored/label": "ignored"}, + container_statuses=containers_statuses + ) + + event_handler.handle(event_type, pod) + + for container_id, name, image in containers: + assert metadata_registry.get_metadata(container_id) == { + "k8s_pod_name": pod.metadata.name, + "k8s_pod_namespace": pod.metadata.namespace, + "k8s_container_name": name, + "k8s_container_image": image, + "k8s_pod_label_app_kubernetes_io_name": "pytest", + } + + +@pytest.mark.parametrize("labels", [None, {}]) +def test_event_registers_container_when_pod_has_no_labels(labels, metadata_registry, pod_event_handler): + """ + Containers should be registered without label metadata when Pod labels are absent. + """ + event_handler = pod_event_handler(build_metadata_mapping(["app.kubernetes.io/name"], "k8s_pod_label_")) + + container_id = "1111111111111111111111111111111111111111111111111111111111111111" + containers_statuses = [ + make_container_status(f'pytest://{container_id}', "test", "powerapi:test"), + ] + pod = make_pod(labels=labels, container_statuses=containers_statuses) + + event_handler.handle("ADDED", pod) + + assert metadata_registry.get_metadata(container_id) == { + "k8s_pod_name": "pytest", + "k8s_pod_namespace": "powerapi", + "k8s_container_name": "test", + "k8s_container_image": "powerapi:test", + } + + +@pytest.mark.parametrize("container_statuses", [None, []]) +def test_event_accepts_pod_without_container_statuses(container_statuses, metadata_registry, pod_event_handler): + """ + A Pod without available container statuses should not create registry entries. + """ + metadata_registry.set_metadata = Mock(wraps=metadata_registry.set_metadata) + event_handler = pod_event_handler({}) + pod = make_pod(container_statuses=container_statuses) + + event_handler.handle("ADDED", pod) + + metadata_registry.set_metadata.assert_not_called() + + +@pytest.mark.parametrize("event_type", ["ADDED", "MODIFIED"]) +def test_event_ignores_containers_without_runtime_id(event_type, metadata_registry, pod_event_handler): + """ + Containers without ID should not be added to the registry. + """ + metadata_registry.set_metadata = Mock(wraps=metadata_registry.set_metadata) + event_handler = pod_event_handler({}) + pod = make_pod(container_statuses=[make_container_status(None)]) + + event_handler.handle(event_type, pod) + + metadata_registry.set_metadata.assert_not_called() + + +def test_deleted_event_retains_metadata(metadata_registry, pod_event_handler): + """ + Deleted Pod events deliberately leave existing registry entries intact. + """ + metadata_registry.set_metadata = Mock(wraps=metadata_registry.set_metadata) + event_handler = pod_event_handler({}) + + event_handler.handle("DELETED", make_pod()) + + metadata_registry.set_metadata.assert_not_called() + + +def test_unexpected_event_raises_value_error(metadata_registry, pod_event_handler): + """ + Unsupported Kubernetes event types should be rejected. + """ + event_handler = pod_event_handler({}) + + with pytest.raises(ValueError, match="Unexpected Kubernetes pod event: PYTEST"): + event_handler.handle("PYTEST", make_pod()) From 8fe74e993a1fc5836f2b538180b29a5a47103d0e Mon Sep 17 00:00:00 2001 From: Guillaume Fieni Date: Thu, 20 Aug 2026 14:35:32 +0200 Subject: [PATCH 3/3] feat(processor/openstack): Support selecting report metadata from server metadata Only propagate explicitly selected server metadata instead of adding every available entry to reports. --- .../cli/common_cli_parsing_manager.py | 6 + src/powerapi/cli/generator.py | 5 +- src/powerapi/processor/pre/openstack/actor.py | 15 +- .../processor/pre/openstack/handlers.py | 17 +- ..._cache_manager.py => metadata_registry.py} | 52 ++---- .../processor/pre/openstack/monitor_agent.py | 72 ++++----- .../pre/openstack/server_change_handler.py | 80 +++++++++ tests/unit/cli/test_generator_openstack.py | 2 +- .../unit/processor/pre/openstack/conftest.py | 21 +-- .../processor/pre/openstack/test_actor.py | 88 ++++++++++ .../processor/pre/openstack/test_handlers.py | 116 +++++++++---- .../openstack/test_metadata_cache_manager.py | 66 -------- .../pre/openstack/test_metadata_registry.py | 56 +++++++ .../pre/openstack/test_monitor_agent.py | 153 +++++++++++++----- .../openstack/test_server_change_handler.py | 90 +++++++++++ 15 files changed, 592 insertions(+), 247 deletions(-) rename src/powerapi/processor/pre/openstack/{metadata_cache_manager.py => metadata_registry.py} (54%) create mode 100644 src/powerapi/processor/pre/openstack/server_change_handler.py create mode 100644 tests/unit/processor/pre/openstack/test_actor.py delete mode 100644 tests/unit/processor/pre/openstack/test_metadata_cache_manager.py create mode 100644 tests/unit/processor/pre/openstack/test_metadata_registry.py create mode 100644 tests/unit/processor/pre/openstack/test_server_change_handler.py diff --git a/src/powerapi/cli/common_cli_parsing_manager.py b/src/powerapi/cli/common_cli_parsing_manager.py index 9b9609e8..86df9594 100644 --- a/src/powerapi/cli/common_cli_parsing_manager.py +++ b/src/powerapi/cli/common_cli_parsing_manager.py @@ -481,4 +481,10 @@ def _register_openstack_pre_processor_parser(self): default_value=10.0 ) + subparser_openstack_pre_processor.add_argument( + 'm', 'metadata', + help_text='Comma-separated list of OpenStack server metadata fields added to reports', + argument_type=list + ) + self.add_subgroup_parser('pre-processor', subparser_openstack_pre_processor) diff --git a/src/powerapi/cli/generator.py b/src/powerapi/cli/generator.py index 0bff3585..b9d91b8e 100644 --- a/src/powerapi/cli/generator.py +++ b/src/powerapi/cli/generator.py @@ -452,7 +452,7 @@ def _k8s_pre_processor_factory(processor_config: dict) -> ProcessorActor: @staticmethod def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor: """ - Openstack pre-processor actor factory. + OpenStack pre-processor actor factory. :param processor_config: Pre-Processor configuration :return: Configured OpenStack pre-processor actor """ @@ -460,7 +460,8 @@ def _openstack_pre_processor_factory(processor_config: dict) -> ProcessorActor: from powerapi.processor.pre.openstack.monitor_agent import OpenStackMonitorConfig api_polling_interval = processor_config['polling-interval'] - monitor_config = OpenStackMonitorConfig(api_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 diff --git a/src/powerapi/processor/pre/openstack/actor.py b/src/powerapi/processor/pre/openstack/actor.py index 9fe20c7b..2110ae82 100644 --- a/src/powerapi/processor/pre/openstack/actor.py +++ b/src/powerapi/processor/pre/openstack/actor.py @@ -31,11 +31,16 @@ from multiprocessing import Manager from powerapi.actor import Actor, State -from powerapi.actor.message import StartMessage, PoisonPillMessage -from powerapi.processor.pre.openstack.handlers import StartMessageHandler, PoisonPillMessageHandler, HWPCReportHandler +from powerapi.actor.message import PoisonPillMessage, StartMessage +from powerapi.processor.pre.openstack.handlers import ( + HWPCReportHandler, + PoisonPillMessageHandler, + StartMessageHandler, +) from powerapi.processor.processor_actor import ProcessorActor from powerapi.report import HWPCReport -from .metadata_cache_manager import OpenStackMetadataCacheManager + +from .metadata_registry import OpenStackMetadataRegistry from .monitor_agent import OpenStackMonitorAgent, OpenStackMonitorConfig @@ -53,8 +58,8 @@ def __init__(self, actor: Actor, monitor_config: OpenStackMonitorConfig): super().__init__(actor) self.manager = Manager() - self.metadata_cache_manager = OpenStackMetadataCacheManager(self.manager) - self.monitor_agent = OpenStackMonitorAgent(self.metadata_cache_manager, monitor_config) + self.metadata_registry = OpenStackMetadataRegistry(self.manager) + self.monitor_agent = OpenStackMonitorAgent(self.metadata_registry, monitor_config) class OpenStackPreProcessorActor(ProcessorActor): diff --git a/src/powerapi/processor/pre/openstack/handlers.py b/src/powerapi/processor/pre/openstack/handlers.py index d9de9b76..30c51fef 100644 --- a/src/powerapi/processor/pre/openstack/handlers.py +++ b/src/powerapi/processor/pre/openstack/handlers.py @@ -31,7 +31,6 @@ 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 class StartMessageHandler(StartHandler): @@ -73,15 +72,6 @@ class HWPCReportHandler(ProcessorReportHandler): 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 - """ - return self.state.metadata_cache_manager.get_server_metadata(sensor_name, instance_name) - def handle(self, msg: HWPCReport): """ Process an HWPCReport to add the OpenStack metadata. @@ -89,12 +79,11 @@ def handle(self, msg: HWPCReport): """ 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) + server_metadata = self.state.metadata_registry.get_metadata(msg.sensor, instance_name) if server_metadata is None: - # Drop the report if the server metadata is not present in the cache. + # Drop the report if the server metadata is not present in the registry. return - msg.target = server_metadata.server_name - msg.metadata['openstack'] = vars(server_metadata) + msg.metadata.update(server_metadata) self._send_report(msg) diff --git a/src/powerapi/processor/pre/openstack/metadata_cache_manager.py b/src/powerapi/processor/pre/openstack/metadata_registry.py similarity index 54% rename from src/powerapi/processor/pre/openstack/metadata_cache_manager.py rename to src/powerapi/processor/pre/openstack/metadata_registry.py index b7481d6c..77cc8e52 100644 --- a/src/powerapi/processor/pre/openstack/metadata_cache_manager.py +++ b/src/powerapi/processor/pre/openstack/metadata_registry.py @@ -1,5 +1,4 @@ -# Copyright (c) 2025, Inria -# Copyright (c) 2025, University of Lille +# Copyright (c) 2026, Inria # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -28,51 +27,34 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. from collections.abc import MutableMapping -from dataclasses import dataclass from multiprocessing.managers import SyncManager -@dataclass(frozen=True) -class ServerMetadata: +class OpenStackMetadataRegistry: """ - Represents an OpenStack server metadata cache entry. - """ - server_id: str - server_name: str - host: str - instance_name: str - metadata: dict[str, str] - - -class OpenStackMetadataCacheManager: - """ - OpenStack metadata cache manager. + OpenStack metadata registry. """ def __init__(self, manager: SyncManager): """ - :param manager: Manager of the shared metadata cache - """ - self._metadata_cache: MutableMapping[tuple[str, str], ServerMetadata] = manager.dict() - - 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 or None if not found + :param manager: Manager of the shared metadata registry """ - return self._metadata_cache.get((host, instance_name), None) + self._server_metadata: MutableMapping[tuple[str, str], dict[str, str]] = manager.dict() - def update_server_metadata(self, server_metadata: ServerMetadata) -> None: + def set_metadata(self, host: str, instance_name: str, metadata: dict[str, str]) -> None: """ - Add or update metadata for a server. - :param server_metadata: Server metadata cache entry + Set metadata for the given OpenStack server. + :param host: Name of the host running the server + :param instance_name: Internal instance name of the server + :param metadata: Metadata entry """ - self._metadata_cache[(server_metadata.host, server_metadata.instance_name)] = server_metadata + self._server_metadata[(host, instance_name)] = metadata - def clear_metadata_cache(self) -> None: + def get_metadata(self, host: str, instance_name: str) -> dict[str, str] | None: """ - Clears all server metadata entries from the cache. + Get metadata for the given OpenStack server. + :param host: Name of the host running the server + :param instance_name: Internal instance name of the server + :return: Metadata entry or None if not found """ - self._metadata_cache.clear() + return self._server_metadata.get((host, instance_name)) diff --git a/src/powerapi/processor/pre/openstack/monitor_agent.py b/src/powerapi/processor/pre/openstack/monitor_agent.py index 07dd09d4..20410968 100644 --- a/src/powerapi/processor/pre/openstack/monitor_agent.py +++ b/src/powerapi/processor/pre/openstack/monitor_agent.py @@ -28,24 +28,28 @@ import logging import sys +from collections.abc import Mapping from dataclasses import dataclass -from multiprocessing import Process, Event -from signal import signal, SIGINT, SIGTERM +from datetime import UTC, datetime +from multiprocessing import Event, Process +from signal import SIGINT, SIGTERM, signal -from openstack.compute.v2.server import Server from openstack.connection import Connection from openstack.exceptions import SDKException -from .metadata_cache_manager import OpenStackMetadataCacheManager, ServerMetadata +from .metadata_registry import OpenStackMetadataRegistry +from .server_change_handler import OpenStackServerChangeHandler -@dataclass +@dataclass(frozen=True) class OpenStackMonitorConfig: """ OpenStack monitoring agent configuration. :param polling_interval: Interval in seconds between OpenStack API synchronizations. + :param metadata_mapping: Mapping from OpenStack server metadata names to canonical report metadata names. """ polling_interval: float + metadata_mapping: Mapping[str, str] class OpenStackMonitorAgent(Process): @@ -55,9 +59,9 @@ class OpenStackMonitorAgent(Process): 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, config: OpenStackMonitorConfig, level_logger: int = logging.WARNING): + def __init__(self, registry: OpenStackMetadataRegistry, config: OpenStackMonitorConfig, level_logger: int = logging.WARNING): """ - :param cache_manager: Metadata cache manager + :param registry: OpenStack metadata registry :param config: Configuration of the monitor agent :param level_logger: Logger level """ @@ -69,8 +73,8 @@ def __init__(self, cache_manager: OpenStackMetadataCacheManager, config: OpenSta handler = logging.StreamHandler() handler.setFormatter(formatter) - self.metadata_cache_manager = cache_manager self.config = config + self.server_change_handler = OpenStackServerChangeHandler(registry, config.metadata_mapping) self._stop_monitoring = Event() @@ -82,7 +86,7 @@ def _setup_openstack_api_client() -> Connection: """ return Connection(app_name='PowerAPI') - def _setup_signal_handlers(self): + def _setup_signal_handlers(self) -> None: """ Setup signal handlers for the current Process. """ @@ -93,46 +97,38 @@ def stop_monitor(_, __): signal(SIGTERM, stop_monitor) signal(SIGINT, stop_monitor) - def run(self): + def run(self) -> None: """ Main code executed by the OpenStack monitor agent. """ self._setup_signal_handlers() - openstack_api = self._setup_openstack_api_client() - # Prevents orphaned entries that no longer exist in the OpenStack API. - self.metadata_cache_manager.clear_metadata_cache() + api_client = self._setup_openstack_api_client() + changes_since = None while not self._stop_monitoring.is_set(): - for server in self.fetch_servers_metadata(openstack_api): - self.metadata_cache_manager.update_server_metadata(server) + try: + changes_since = self.fetch_server_changes(api_client, changes_since) + except SDKException as exn: + logging.warning('Failed to retrieve server changes from OpenStack API: %s', exn) + except (AttributeError, ValueError) as exn: + logging.error('Required server attribute is missing from the OpenStack API response: %s', exn) if self._stop_monitoring.wait(self.config.polling_interval): break - @staticmethod - def build_metadata_cache_entry_from_server(server: Server) -> ServerMetadata: - """ - Build and return a metadata cache entry from an OpenStack server object. - :param server: OpenStack server object - :return: Cache key and server metadata entry - """ - return ServerMetadata(server.id, server.name, server.host, server.instance_name, server.metadata) - - def fetch_servers_metadata(self, openstack_api: Connection) -> list[ServerMetadata]: + def fetch_server_changes(self, openstack_api: Connection, changes_since: str | None = None) -> str: """ - Fetch servers metadata from the OpenStack API. + Fetch and handle OpenStack server changes. + When no synchronization timestamp is provided, all servers are fetched. :param openstack_api: OpenStack API client - :return: List of servers metadata + :param changes_since: ISO 8601 timestamp of the previous synchronization + :return: ISO 8601 timestamp to use for the next synchronization """ - try: - return [ - self.build_metadata_cache_entry_from_server(server) - for server in openstack_api.compute.servers(details=True, all_projects=True) - ] - except SDKException as exn: - logging.warning('Failed to retrieve server metadata from OpenStack API: %s', exn.message) - except (AttributeError, ValueError) as exn: - logging.error('Required server attribute is missing from the OpenStack API response: %s', exn) - - return [] + next_changes_since = datetime.now(UTC).isoformat(timespec='seconds') + query = {} if changes_since is None else {'changes_since': changes_since} + + for server in openstack_api.compute.servers(details=True, all_projects=True, **query): + self.server_change_handler.handle(server) + + return next_changes_since diff --git a/src/powerapi/processor/pre/openstack/server_change_handler.py b/src/powerapi/processor/pre/openstack/server_change_handler.py new file mode 100644 index 00000000..8ca8fa7b --- /dev/null +++ b/src/powerapi/processor/pre/openstack/server_change_handler.py @@ -0,0 +1,80 @@ +# 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 __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING + +from .metadata_registry import OpenStackMetadataRegistry + +if TYPE_CHECKING: + from openstack.compute.v2.server import Server + + +class OpenStackServerChangeHandler: + """ + Translates OpenStack server changes into metadata registry operations. + """ + + def __init__(self, metadata_registry: OpenStackMetadataRegistry, metadata_mapping: Mapping[str, str]): + """ + Initialize the OpenStack server change handler. + :param metadata_registry: Registry containing server metadata + :param metadata_mapping: Mapping from OpenStack metadata names to canonical report metadata names + """ + self.metadata_registry = metadata_registry + self.metadata_mapping = metadata_mapping + + def handle(self, server: Server) -> None: + """ + Apply an OpenStack server change to the metadata registry. + :param server: Changed OpenStack server + """ + if server.status == "DELETED": + # Retain metadata so reports already in the pipeline can still be enriched. + return + + self._set_server_metadata(server) + + def _set_server_metadata(self, server: Server) -> None: + """ + Register metadata for an OpenStack server. + :param server: Server whose metadata should be registered + """ + metadata = { + "openstack_server_name": server.name, + "openstack_project_id": server.project_id, + "openstack_availability_zone": server.availability_zone + } + + for source_name, canonical_name in self.metadata_mapping.items(): + if source_name in server.metadata: + metadata[canonical_name] = server.metadata[source_name] + + self.metadata_registry.set_metadata(server.host, server.instance_name, metadata) diff --git a/tests/unit/cli/test_generator_openstack.py b/tests/unit/cli/test_generator_openstack.py index ff59f783..53dc6848 100644 --- a/tests/unit/cli/test_generator_openstack.py +++ b/tests/unit/cli/test_generator_openstack.py @@ -28,7 +28,7 @@ import pytest -pytest.importorskip('powerapi.processor.pre.openstack.actor') # The OpenStack processor requires external dependencies to work. +pytest.importorskip('openstack') from powerapi.cli.generator import PreProcessorGenerator from powerapi.processor.pre.openstack.actor import OpenStackPreProcessorActor diff --git a/tests/unit/processor/pre/openstack/conftest.py b/tests/unit/processor/pre/openstack/conftest.py index b5080d31..09c1f704 100644 --- a/tests/unit/processor/pre/openstack/conftest.py +++ b/tests/unit/processor/pre/openstack/conftest.py @@ -26,25 +26,18 @@ # 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 multiprocessing.managers import SyncManager +from unittest.mock import Mock import pytest -from powerapi.processor.pre.openstack.metadata_cache_manager import OpenStackMetadataCacheManager - - -class InProcessManager(SyncManager): - """ - Minimal manager stub for unit tests that do not need cross-process sharing. - """ - - def dict(self): - return {} +from powerapi.processor.pre.openstack.metadata_registry import OpenStackMetadataRegistry @pytest.fixture -def initialized_metadata_cache_manager(): +def metadata_registry(): """ - Returns an initialized metadata cache manager. + Return an OpenStack metadata registry. """ - return OpenStackMetadataCacheManager(InProcessManager()) + manager = Mock() + manager.dict.return_value = {} + return OpenStackMetadataRegistry(manager) diff --git a/tests/unit/processor/pre/openstack/test_actor.py b/tests/unit/processor/pre/openstack/test_actor.py new file mode 100644 index 00000000..3ed7a267 --- /dev/null +++ b/tests/unit/processor/pre/openstack/test_actor.py @@ -0,0 +1,88 @@ +# 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 datetime import datetime +from unittest.mock import Mock, patch + +import pytest + +pytest.importorskip('openstack') + +from powerapi.actor import State +from powerapi.actor.message import PoisonPillMessage, StartMessage +from powerapi.processor.pre.openstack.actor import ( + OpenStackPreProcessorActor, + OpenStackProcessorState, +) +from powerapi.processor.pre.openstack.handlers import ( + HWPCReportHandler, + PoisonPillMessageHandler, + StartMessageHandler, +) +from powerapi.report import HWPCReport + + +def test_processor_state_builds_metadata_components(): + """ + The processor state should initialize its manager, registry, and monitor agent. + """ + monitor_config = Mock() + + with ( + patch('powerapi.processor.pre.openstack.actor.Manager'), + patch('powerapi.processor.pre.openstack.actor.OpenStackMetadataRegistry') as metadata_registry_class, + patch('powerapi.processor.pre.openstack.actor.OpenStackMonitorAgent') as monitor_agent_class, + ): + state = OpenStackProcessorState(Mock(), monitor_config) + + metadata_registry_class.assert_called_once_with(state.manager) + monitor_agent_class.assert_called_once_with(state.metadata_registry, monitor_config) + + +@pytest.mark.parametrize( + ('message', 'expected_handler_type'), + [ + (StartMessage(), StartMessageHandler), + (PoisonPillMessage(), PoisonPillMessageHandler), + (HWPCReport(datetime.now(), 'pytest', 'pytest', {}), HWPCReportHandler), + ], + ids=['start_message', 'poison_pill_message', 'hwpc_report'], +) +def test_actor_setup_registers_openstack_handlers(message, expected_handler_type): + """ + Actor setup should create its state and register OpenStack handlers. + """ + monitor_config = Mock() + actor = OpenStackPreProcessorActor('pytest', monitor_config) + state = State(actor) + + with patch('powerapi.processor.pre.openstack.actor.OpenStackProcessorState', return_value=state): + actor.setup() + + assert actor.state is state + assert isinstance(state.get_corresponding_handler(message), expected_handler_type) diff --git a/tests/unit/processor/pre/openstack/test_handlers.py b/tests/unit/processor/pre/openstack/test_handlers.py index 3f8b5ca4..d1b82e2b 100644 --- a/tests/unit/processor/pre/openstack/test_handlers.py +++ b/tests/unit/processor/pre/openstack/test_handlers.py @@ -31,24 +31,62 @@ import pytest -from powerapi.processor.pre.openstack.handlers import HWPCReportHandler -from powerapi.processor.pre.openstack.metadata_cache_manager import ServerMetadata +from powerapi.processor.pre.openstack.handlers import ( + HWPCReportHandler, + PoisonPillMessageHandler, + StartMessageHandler, +) from powerapi.report import HWPCReport @pytest.fixture -def hwpc_report_handler(): +def start_message_handler(): """ - Factory fixture creating an HwPC report handler. + Factory fixture creating a start message handler. """ - def _create_handler() -> tuple[HWPCReportHandler, list[HWPCReport]]: + def _create_handler() -> StartMessageHandler: actor = Mock(name='processor-actor') actor.target_actors = [Mock(name='target_actor_a'), Mock(name='target_actor_b')] - state = Mock() + state = Mock(name='state') state.actor = actor - state.metadata_cache_manager = Mock(name='metadata_cache_manager') + state.monitor_agent = Mock(name='monitor_agent') + + return StartMessageHandler(state) + + return _create_handler + + +@pytest.fixture +def poison_pill_message_handler(): + """ + Factory fixture creating a Poison-Pill message handler. + """ + + def _create_handler() -> PoisonPillMessageHandler: + actor = Mock(name='processor-actor') + actor.target_actors = [Mock(name='target_actor_a'), Mock(name='target_actor_b')] + + state = Mock(name='state') + state.actor = actor + state.manager = Mock(name='manager') + state.monitor_agent = Mock(name='monitor_agent') + + return PoisonPillMessageHandler(state) + + return _create_handler + + +@pytest.fixture +def hwpc_report_handler(): + """ + Factory fixture creating an HwPC report handler. + """ + + def _create_handler() -> tuple[HWPCReportHandler, list[HWPCReport]]: + state = Mock(name='state') + state.metadata_registry = Mock(name='metadata_registry') handler = HWPCReportHandler(state) @@ -68,31 +106,55 @@ def make_libvirt_hwpc_report() -> HWPCReport: return HWPCReport(datetime.now(), 'compute-1', target, {}, {'scope': 'pytest'}) +def test_start_handler_connects_targets_and_starts_monitor(start_message_handler): + """ + The start handler should connect target actors and start monitoring. + """ + handler = start_message_handler() + + handler.initialization() + + handler.state.monitor_agent.start.assert_called_once() + for actor in handler.state.actor.target_actors: + actor.connect_data.assert_called_once() + + +def test_poison_pill_handler_stops_resources_and_disconnects_targets(poison_pill_message_handler): + """ + The Poison-Pill handler should stop resources and disconnect targets. + """ + handler = poison_pill_message_handler() + + handler.teardown() + + handler.state.monitor_agent.terminate.assert_called_once() + handler.state.monitor_agent.join.assert_called_once() + handler.state.manager.shutdown.assert_called_once() + for actor in handler.state.actor.target_actors: + actor.disconnect.assert_called_once() + + def test_hwpc_report_handler_adds_openstack_metadata_and_forwards_report(hwpc_report_handler): """ - Test that the OpenStack report handler forwards a report when metadata is in the cache. + Test that the OpenStack report handler forwards a report when metadata is in the registry. """ handler, reports_sent = hwpc_report_handler() report = make_libvirt_hwpc_report() + target = report.target - server_metadata = ServerMetadata( - 'server-id', - 'server-name', - 'compute-1', - 'instance-00000003', - {'app': 'powerapi'} - ) - handler.state.metadata_cache_manager.get_server_metadata.return_value = server_metadata + server_metadata = {'openstack_server_name': 'server-name', 'app': 'powerapi'} + handler.state.metadata_registry.get_metadata.return_value = server_metadata handler.handle(report) assert reports_sent == [report] - (processed_report,) = reports_sent - - assert processed_report.target == server_metadata.server_name - assert processed_report.metadata['openstack'] == vars(server_metadata) - assert processed_report.metadata['scope'] == 'pytest' - handler.state.metadata_cache_manager.get_server_metadata.assert_called_once_with('compute-1', 'instance-00000003') + assert report.target == target + assert report.metadata == { + 'scope': 'pytest', + 'openstack_server_name': 'server-name', + 'app': 'powerapi', + } + handler.state.metadata_registry.get_metadata.assert_called_once_with('compute-1', 'instance-00000003') def test_hwpc_report_handler_drops_report_when_server_metadata_missing(hwpc_report_handler): @@ -102,7 +164,7 @@ def test_hwpc_report_handler_drops_report_when_server_metadata_missing(hwpc_repo handler, reports_sent = hwpc_report_handler() report = make_libvirt_hwpc_report() - handler.state.metadata_cache_manager.get_server_metadata.return_value = None + handler.state.metadata_registry.get_metadata.return_value = None handler.handle(report) @@ -119,8 +181,6 @@ def test_hwpc_report_handler_forwards_non_openstack_targets(hwpc_report_handler) handler.handle(report) assert reports_sent == [report] - (processed_report,) = reports_sent - - assert processed_report.target == 'plain-container' - assert processed_report.metadata == {'scope': 'pytest'} - handler.state.metadata_cache_manager.get_server_metadata.assert_not_called() + assert report.target == 'plain-container' + assert report.metadata == {'scope': 'pytest'} + handler.state.metadata_registry.get_metadata.assert_not_called() diff --git a/tests/unit/processor/pre/openstack/test_metadata_cache_manager.py b/tests/unit/processor/pre/openstack/test_metadata_cache_manager.py deleted file mode 100644 index d2b153ad..00000000 --- a/tests/unit/processor/pre/openstack/test_metadata_cache_manager.py +++ /dev/null @@ -1,66 +0,0 @@ -# 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 powerapi.processor.pre.openstack.metadata_cache_manager import ServerMetadata - - -def test_update_and_get_server_metadata(initialized_metadata_cache_manager): - """ - Test storing and retrieving OpenStack server metadata. - """ - metadata = ServerMetadata('server-id', 'server-name', 'compute-1', 'instance-00000001', {'app': 'pytest'}) - - initialized_metadata_cache_manager.update_server_metadata(metadata) - - assert initialized_metadata_cache_manager.get_server_metadata('compute-1', 'instance-00000001') == metadata - - -def test_update_server_metadata_uses_host_and_instance_name(initialized_metadata_cache_manager): - """ - Test that OpenStack server metadata is indexed by host and instance name. - """ - first_metadata = ServerMetadata('server-id-a', 'server-name-a', 'compute-1', 'instance-00000001', {}) - second_metadata = ServerMetadata('server-id-b', 'server-name-b', 'compute-1', 'instance-00000002', {'app': 'pytest'}) - - initialized_metadata_cache_manager.update_server_metadata(first_metadata) - initialized_metadata_cache_manager.update_server_metadata(second_metadata) - - assert initialized_metadata_cache_manager.get_server_metadata('compute-1', 'instance-00000001') == first_metadata - assert initialized_metadata_cache_manager.get_server_metadata('compute-1', 'instance-00000002') == second_metadata - - -def test_clear_metadata_cache(initialized_metadata_cache_manager): - """ - Test clearing the OpenStack metadata cache. - """ - metadata = ServerMetadata('server-id', 'server-name', 'compute-1', 'instance-00000001', {}) - initialized_metadata_cache_manager.update_server_metadata(metadata) - - initialized_metadata_cache_manager.clear_metadata_cache() - - assert initialized_metadata_cache_manager.get_server_metadata('compute-1', 'instance-00000001') is None diff --git a/tests/unit/processor/pre/openstack/test_metadata_registry.py b/tests/unit/processor/pre/openstack/test_metadata_registry.py new file mode 100644 index 00000000..4b728c07 --- /dev/null +++ b/tests/unit/processor/pre/openstack/test_metadata_registry.py @@ -0,0 +1,56 @@ +# 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. + + +def test_set_and_get_metadata(metadata_registry): + """ + Metadata can be retrieved using its host and instance name. + """ + metadata = {'openstack_server_name': 'powerapi-pytest'} + + metadata_registry.set_metadata('compute-1', 'instance-00000001', metadata) + + assert metadata_registry.get_metadata('compute-1', 'instance-00000001') == metadata + + +def test_set_metadata_replaces_existing_entry(metadata_registry): + """ + Setting metadata twice replaces the previous entry. + """ + metadata_registry.set_metadata('compute-1', 'instance-00000001', {'version': 'old'}) + + metadata_registry.set_metadata('compute-1', 'instance-00000001', {'version': 'new'}) + + assert metadata_registry.get_metadata('compute-1', 'instance-00000001') == {'version': 'new'} + + +def test_get_metadata_returns_none_for_unknown_server(metadata_registry): + """ + An unknown host and instance name have no associated metadata. + """ + assert metadata_registry.get_metadata('compute-1', 'instance-00000001') is None diff --git a/tests/unit/processor/pre/openstack/test_monitor_agent.py b/tests/unit/processor/pre/openstack/test_monitor_agent.py index c7b5f77a..b012a4ab 100644 --- a/tests/unit/processor/pre/openstack/test_monitor_agent.py +++ b/tests/unit/processor/pre/openstack/test_monitor_agent.py @@ -26,84 +26,149 @@ # 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 unittest.mock import Mock +from collections.abc import Iterable +from datetime import datetime +from unittest.mock import Mock, call import pytest -pytest.importorskip('powerapi.processor.pre.openstack.monitor_agent') +pytest.importorskip('openstack') from openstack.exceptions import SDKException -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, OpenStackMonitorConfig +from powerapi.processor.pre.openstack.metadata_registry import OpenStackMetadataRegistry +from powerapi.processor.pre.openstack.monitor_agent import ( + OpenStackMonitorAgent, + OpenStackMonitorConfig, +) @pytest.fixture -def initialized_monitor_agent(initialized_metadata_cache_manager): +def monitor_agent(): """ - Returns an initialized OpenStack monitor agent. + Return an OpenStack monitor agent using mocked collaborators. """ - monitor_config = OpenStackMonitorConfig(polling_interval=0.01) - return OpenStackMonitorAgent(initialized_metadata_cache_manager, monitor_config) + registry = Mock(spec=OpenStackMetadataRegistry) + config = OpenStackMonitorConfig(polling_interval=1.0, metadata_mapping={}) + agent = OpenStackMonitorAgent(registry, config) + agent.server_change_handler = Mock() + return agent -def make_server(server_id, server_name, host, instance_name, metadata) -> Server: +@pytest.fixture +def configure_monitor_agent(monitor_agent): """ - Generate a fake OpenStack server object. + Configure the monitor agent for a finite run without external calls. """ - server = Mock() - server.id = server_id - server.name = server_name - server.host = host - server.instance_name = instance_name - server.metadata = metadata - return server + def _configure( + *, + fetch_server_changes: Mock | None = None, + stop_event: Mock | None = None, + wait_results: Iterable[bool] = (True,), + ) -> OpenStackMonitorAgent: + if fetch_server_changes is None: + fetch_server_changes = Mock(return_value='pytest') + if stop_event is None: + stop_event = Mock(spec_set=['is_set', 'wait']) -def test_build_metadata_cache_entry_from_server(initialized_monitor_agent): - """ - Test building a metadata cache entry from an OpenStack server object. - """ - server = make_server('server-id', 'server-name', 'compute-1', 'instance-00000001', {'app': 'pytest'}) + stop_event.is_set.return_value = False + stop_event.wait.side_effect = wait_results - metadata = initialized_monitor_agent.build_metadata_cache_entry_from_server(server) + monitor_agent._setup_signal_handlers = Mock() + monitor_agent._setup_openstack_api_client = Mock() + monitor_agent._stop_monitoring = stop_event + monitor_agent.fetch_server_changes = fetch_server_changes + return monitor_agent - assert metadata == ServerMetadata('server-id', 'server-name', 'compute-1', 'instance-00000001', {'app': 'pytest'}) + return _configure -def test_fetch_servers_metadata(initialized_monitor_agent): +def test_initial_fetch_forwards_all_servers(monitor_agent): """ - Test fetching OpenStack servers metadata. + The initial synchronization should fetch and handle every server. """ - server = make_server('server-id', 'server-name', 'compute-1', 'instance-00000001', {'app': 'pytest'}) + servers = [Mock(name='first_server'), Mock(name='second_server')] openstack_api = Mock() - openstack_api.compute.servers.return_value = [server] + openstack_api.compute.servers.return_value = servers - metadata_entries = initialized_monitor_agent.fetch_servers_metadata(openstack_api) + next_changes_since = monitor_agent.fetch_server_changes(openstack_api) - assert metadata_entries == [ - ServerMetadata('server-id', 'server-name', 'compute-1', 'instance-00000001', {'app': 'pytest'}), - ] + openstack_api.compute.servers.assert_called_once_with(details=True, all_projects=True) + assert monitor_agent.server_change_handler.handle.call_args_list == [call(servers[0]), call(servers[1])] + assert datetime.fromisoformat(next_changes_since).tzinfo is not None -def test_fetch_servers_metadata_returns_empty_list_on_sdk_exception(initialized_monitor_agent): +def test_incremental_fetch_uses_previous_synchronization_timestamp(monitor_agent): """ - Test that OpenStack SDK errors return an empty server metadata list. + Later synchronizations should request only servers changed since the previous one. """ openstack_api = Mock() - openstack_api.compute.servers.side_effect = SDKException('pytest') + openstack_api.compute.servers.return_value = [] + + monitor_agent.fetch_server_changes(openstack_api, '2026-08-19T12:00:00+00:00') + + openstack_api.compute.servers.assert_called_once_with( + details=True, + all_projects=True, + changes_since='2026-08-19T12:00:00+00:00', + ) + + +@pytest.mark.parametrize( + 'exception', + [ + pytest.param(SDKException('pytest'), id='sdk-error'), + pytest.param(AttributeError('pytest'), id='missing-attribute'), + pytest.param(ValueError('pytest'), id='invalid-attribute'), + ], +) +def test_run_waits_after_fetch_failure(configure_monitor_agent, exception): + """ + A failed synchronization should wait before leaving or retrying the loop. + """ + fetch_server_changes = Mock(side_effect=exception) + stop_event = Mock() + monitor_agent = configure_monitor_agent( + fetch_server_changes=fetch_server_changes, + stop_event=stop_event, + ) + + monitor_agent.run() - assert initialized_monitor_agent.fetch_servers_metadata(openstack_api) == [] + fetch_server_changes.assert_called_once() + stop_event.wait.assert_called_once() -def test_fetch_servers_metadata_returns_empty_list_on_missing_attribute(initialized_monitor_agent): +def test_run_uses_successful_synchronization_timestamp(configure_monitor_agent): """ - Test that missing OpenStack server attributes return an empty server metadata list. + A successful synchronization should provide its timestamp to the next one. """ - server = make_server('server-id', 'server-name', 'compute-1', 'instance-00000001', {}) - del server.instance_name - openstack_api = Mock() - openstack_api.compute.servers.return_value = [server] + fetch_server_changes = Mock(side_effect=['first-sync', 'second-sync']) + monitor_agent = configure_monitor_agent( + fetch_server_changes=fetch_server_changes, + wait_results=[False, True], + ) + + monitor_agent.run() + + assert [change.args[1] for change in fetch_server_changes.call_args_list] == [None, 'first-sync'] + - assert initialized_monitor_agent.fetch_servers_metadata(openstack_api) == [] +def test_run_preserves_synchronization_timestamp_after_failure(configure_monitor_agent): + """ + A failed synchronization should retry from the last successful timestamp. + """ + fetch_server_changes = Mock(side_effect=['first-sync', SDKException('pytest'), 'second-sync']) + monitor_agent = configure_monitor_agent( + fetch_server_changes=fetch_server_changes, + wait_results=[False, False, True], + ) + + monitor_agent.run() + + assert [change.args[1] for change in fetch_server_changes.call_args_list] == [ + None, + 'first-sync', + 'first-sync', + ] diff --git a/tests/unit/processor/pre/openstack/test_server_change_handler.py b/tests/unit/processor/pre/openstack/test_server_change_handler.py new file mode 100644 index 00000000..38502352 --- /dev/null +++ b/tests/unit/processor/pre/openstack/test_server_change_handler.py @@ -0,0 +1,90 @@ +# 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 unittest.mock import Mock + +import pytest + +pytest.importorskip('openstack') + +from openstack.compute.v2.server import Server + +from powerapi.processor.pre.openstack.server_change_handler import ( + OpenStackServerChangeHandler, +) + + +def make_server(status: str = 'ACTIVE') -> Server: + """ + Build an OpenStack server containing the attributes used by the change handler. + """ + server = Mock(spec=Server) + server.name = 'server-name' + server.host = 'compute-1' + server.instance_name = 'instance-00000001' + server.status = status + server.project_id = 'project-id' + server.availability_zone = 'nova' + server.metadata = { + 'environment': 'pytest', + 'ignored': 'ignored', + } + return server + + +def test_server_change_registers_selected_metadata(metadata_registry): + """ + A server change should register only selected metadata. + """ + handler = OpenStackServerChangeHandler( + metadata_registry, + {'environment': 'openstack_metadata_environment'}, + ) + server = make_server() + + handler.handle(server) + + assert metadata_registry.get_metadata(server.host, server.instance_name) == { + 'openstack_server_name': server.name, + 'openstack_project_id': server.project_id, + 'openstack_availability_zone': server.availability_zone, + 'openstack_metadata_environment': 'pytest', + } + + +def test_deleted_server_change_retains_metadata(metadata_registry): + """ + A deleted server change should leave existing metadata intact. + """ + metadata = {'openstack_server_id': 'server-id'} + metadata_registry.set_metadata('compute-1', 'instance-00000001', metadata) + handler = OpenStackServerChangeHandler(metadata_registry, {}) + + handler.handle(make_server(status='DELETED')) + + assert metadata_registry.get_metadata('compute-1', 'instance-00000001') == metadata