Merge pull request #291 from powerapi-ng/refactor/k8s-monitor-process

refactor(processor/k8s): Rework Kubernetes pre-processor

Author: gfieni

src/powerapi/actor/actor.py

@@ -150,10 +150,11 @@ def _signal_handler_setup(self):
         Define how to handle signal interrupts
         """
 
-        def term_handler(_, __):
-            self.logger.debug("Term handler")
-            pp_handler = self.state.get_corresponding_handler(PoisonPillMessage())
-            pp_handler.handle(PoisonPillMessage(soft=False, sender_name='Term handler'))
+        def term_handler(signum, _):
+            signame = signal.Signals(signum).name
+            self.logger.debug("Received signal %s (%s), terminating actor...", signame, signum)
+
+            self.state.alive = False
             self._kill_process()
             sys.exit(0)
 

src/powerapi/cli/common_cli_parsing_manager.py

@@ -406,19 +406,7 @@ def __init__(self):
 
         subparser_k8s_pre_processor = SubgroupConfigParsingManager("k8s")
         subparser_k8s_pre_processor.add_argument(
-            "a", "k8s-api-mode", help_text="k8s api mode (local, manual or cluster)"
-        )
-        subparser_k8s_pre_processor.add_argument(
-            "t",
-            "time-interval",
-            help_text="time interval for the k8s monitoring",
-            argument_type=int
-        )
-        subparser_k8s_pre_processor.add_argument(
-            "o",
-            "timeout-query",
-            help_text="timeout for k8s queries",
-            argument_type=int
+            "a", "api-mode", help_text="k8s api mode (local, manual or cluster)"
         )
 
         subparser_k8s_pre_processor.add_argument(
@@ -429,7 +417,7 @@ def __init__(self):
 
         subparser_k8s_pre_processor.add_argument(
             "h",
-            "host",
+            "api-host",
             help_text="host required for k8s manual configuration",
         )
 

src/powerapi/cli/generator.py

@@ -33,20 +33,17 @@
 from typing import Dict, Type, Callable
 
 from powerapi.actor import Actor
+from powerapi.database import MongoDB, CsvDB, OpenTSDB, SocketDB, PrometheusDB, VirtioFSDB, FileDB
 from powerapi.database.influxdb2 import InfluxDB2
 from powerapi.exception import PowerAPIException, ModelNameAlreadyUsed, DatabaseNameDoesNotExist, ModelNameDoesNotExist, \
-    DatabaseNameAlreadyUsed, ProcessorTypeDoesNotExist, ProcessorTypeAlreadyUsed, MonitorTypeDoesNotExist
+    DatabaseNameAlreadyUsed, ProcessorTypeDoesNotExist, ProcessorTypeAlreadyUsed
 from powerapi.filter import Filter
-from powerapi.processor.pre.k8s.k8s_monitor import K8sMonitorAgent
-from powerapi.processor.pre.k8s.k8s_pre_processor_actor import K8sPreProcessorActor, TIME_INTERVAL_DEFAULT_VALUE, \
-    TIMEOUT_QUERY_DEFAULT_VALUE
+from powerapi.processor.pre.k8s import K8sPreProcessorActor
 from powerapi.processor.pre.libvirt.libvirt_pre_processor_actor import LibvirtPreProcessorActor
 from powerapi.processor.processor_actor import ProcessorActor
-from powerapi.report import HWPCReport, PowerReport, ControlReport, ProcfsReport, Report, FormulaReport
-from powerapi.database import MongoDB, CsvDB, OpenTSDB, SocketDB, PrometheusDB, \
-    VirtioFSDB, FileDB
 from powerapi.puller import PullerActor
 from powerapi.pusher import PusherActor
+from powerapi.report import HWPCReport, PowerReport, ControlReport, ProcfsReport, Report, FormulaReport
 
 COMPONENT_TYPE_KEY = 'type'
 COMPONENT_MODEL_KEY = 'model'
@@ -59,23 +56,19 @@
 ACTOR_NAME_KEY = 'actor_name'
 TARGET_ACTORS_KEY = 'target_actors'
 REGEXP_KEY = 'regexp'
-K8S_API_MODE_KEY = 'k8s-api-mode'
-TIME_INTERVAL_KEY = 'time-interval'
-TIMEOUT_QUERY_KEY = 'timeout-query'
+
 PULLER_NAME_KEY = 'puller'
 PUSHER_NAME_KEY = 'pusher'
-API_KEY_KEY = 'api-key'
-HOST_KEY = 'host'
+
+K8S_API_MODE_KEY = 'api-mode'
+K8S_API_KEY_KEY = 'api-key'
+K8S_API_HOST_KEY = 'api-host'
 
 LISTENER_ACTOR_KEY = 'listener_actor'
 
 GENERAL_CONF_STREAM_MODE_KEY = 'stream'
 GENERAL_CONF_VERBOSE_KEY = 'verbose'
 
-MONITOR_NAME_SUFFIX = '_monitor'
-MONITOR_KEY = 'monitor'
-K8S_COMPONENT_TYPE_VALUE = 'k8s'
-
 
 class Generator:
     """
@@ -332,39 +325,48 @@ def _gen_actor(self, component_config: dict, main_config: dict, component_name:
 
 class PreProcessorGenerator(ProcessorGenerator):
     """
-    Generator that initialises the pre-processor from config
+    Generator that initialises the pre-processor from config.
     """
 
     def __init__(self):
-        ProcessorGenerator.__init__(self, 'pre-processor', self._get_default_processor_factories())
+        super().__init__('pre-processor', self._get_default_processor_factories())
 
     @staticmethod
-    def _get_default_processor_factories() -> Dict[str, Callable[[Dict], ProcessorActor]]:
+    def _libvirt_pre_processor_factory(processor_config: dict) -> LibvirtPreProcessorActor:
+        """
+        Libvirt pre-processor actor factory.
+        :param processor_config: Pre-Processor configuration
+        :return: Configured Libvirt pre-processor actor
+        """
+        name = processor_config[ACTOR_NAME_KEY]
+        uri = processor_config[COMPONENT_URI_KEY]
+        regexp = processor_config[REGEXP_KEY]
+        target_actors_name = [processor_config[PULLER_NAME_KEY]]
+        level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
+        return LibvirtPreProcessorActor(name, uri, regexp, [], target_actors_name, level_logger)
+
+    @staticmethod
+    def _k8s_pre_processor_factory(processor_config: Dict) -> K8sPreProcessorActor:
+        """
+        Kubernetes pre-processor actor factory.
+        :param processor_config: Pre-Processor configuration
+        :return: Configured Kubernetes pre-processor actor
+        """
+        name = processor_config[ACTOR_NAME_KEY]
+        api_mode = processor_config.get(K8S_API_MODE_KEY, 'manual')  # use manual mode by default
+        api_host = processor_config.get(K8S_API_HOST_KEY, None)
+        api_key = processor_config.get(K8S_API_KEY_KEY, None)
+        target_actors_name = [processor_config[PULLER_NAME_KEY]]
+        level_logger = logging.DEBUG if processor_config[GENERAL_CONF_VERBOSE_KEY] else logging.INFO
+        return K8sPreProcessorActor(name, [], target_actors_name, api_mode, api_host, api_key, level_logger)
+
+    def _get_default_processor_factories(self) -> Dict[str, Callable[[Dict], ProcessorActor]]:
+        """
+        Return the default pre-processors factory.
+        """
         return {
-            'libvirt': lambda processor_config: LibvirtPreProcessorActor(name=processor_config[ACTOR_NAME_KEY],
-                                                                         uri=processor_config[COMPONENT_URI_KEY],
-                                                                         regexp=processor_config[REGEXP_KEY],
-                                                                         target_actors_names=[processor_config
-                                                                                              [PULLER_NAME_KEY]]),
-            'k8s': lambda processor_config: K8sPreProcessorActor(name=processor_config[ACTOR_NAME_KEY],
-                                                                 ks8_api_mode=None if
-                                                                 K8S_API_MODE_KEY not in processor_config else
-                                                                 processor_config[K8S_API_MODE_KEY],
-                                                                 time_interval=TIME_INTERVAL_DEFAULT_VALUE if
-                                                                 TIME_INTERVAL_KEY not in processor_config else
-                                                                 processor_config[TIME_INTERVAL_KEY],
-                                                                 timeout_query=TIMEOUT_QUERY_DEFAULT_VALUE if
-                                                                 TIMEOUT_QUERY_KEY not in processor_config
-                                                                 else processor_config[TIMEOUT_QUERY_KEY],
-                                                                 api_key=None if API_KEY_KEY not in processor_config
-                                                                 else processor_config[API_KEY_KEY],
-                                                                 host=None if HOST_KEY not in processor_config
-                                                                 else processor_config[HOST_KEY],
-                                                                 level_logger=logging.DEBUG if
-                                                                 processor_config[GENERAL_CONF_VERBOSE_KEY] else
-                                                                 logging.INFO,
-                                                                 target_actors_names=[processor_config[PULLER_NAME_KEY]]
-                                                                 )
+            'libvirt': self._libvirt_pre_processor_factory,
+            'k8s': self._k8s_pre_processor_factory,
         }
 
 
@@ -379,48 +381,3 @@ def __init__(self):
     @staticmethod
     def _get_default_processor_factories() -> Dict[str, Callable[[Dict], ProcessorActor]]:
         return {}
-
-
-class MonitorGenerator(Generator):
-    """
-    Generator that initialises the monitor by using a K8sPreProcessorActor
-    """
-
-    def __init__(self):
-        Generator.__init__(self, component_group_name=MONITOR_KEY)
-
-        self.monitor_factory = {
-            K8S_COMPONENT_TYPE_VALUE: lambda monitor_config: K8sMonitorAgent(
-                name=monitor_config[ACTOR_NAME_KEY],
-                concerned_actor_state=monitor_config[LISTENER_ACTOR_KEY].state,
-                level_logger=monitor_config[LISTENER_ACTOR_KEY].logger.getEffectiveLevel()
-            )
-
-        }
-
-    def _gen_actor(self, component_config: dict, main_config: dict, component_name: str):
-
-        monitor_actor_type = component_config[COMPONENT_TYPE_KEY]
-
-        if monitor_actor_type not in self.monitor_factory:
-            raise MonitorTypeDoesNotExist(monitor_type=monitor_actor_type)
-
-        component_config[ACTOR_NAME_KEY] = component_name + MONITOR_NAME_SUFFIX
-        return self.monitor_factory[monitor_actor_type](component_config)
-
-    def generate_from_processors(self, processors: dict) -> dict:
-        """
-        Generates monitors associated with the given processors
-        :param dict processors: Dictionary with the processors for the generation
-        """
-
-        monitors_config = {MONITOR_KEY: {}}
-
-        for processor_name, processor in processors.items():
-
-            if isinstance(processor, K8sPreProcessorActor):
-                monitors_config[MONITOR_KEY][processor_name + MONITOR_NAME_SUFFIX] = {
-                    COMPONENT_TYPE_KEY: K8S_COMPONENT_TYPE_VALUE,
-                    LISTENER_ACTOR_KEY: processor}
-
-        return self.generate(main_config=monitors_config)

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

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

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

@@ -0,0 +1,51 @@
+# 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.
+
+
+def is_target_a_valid_k8s_cgroups_path(target: str) -> bool:
+    """
+    Checks if the provided target is a valid Kubernetes cgroups path.
+    """
+    return target.startswith('/kubepods')
+
+
+def extract_container_id_from_k8s_cgroups_path(cgroups_path: str) -> str:
+    """
+    Extract the container id from a Kubernetes cgroups path.
+    :param cgroups_path: Cgroups path of the target
+    :return: The container id
+    """
+    container_id = cgroups_path.rsplit('/', 1)[1]
+
+    # handle cgroups v2 (systemd) paths:
+    if container_id.endswith('.scope'):
+        container_id = container_id.replace('.scope', '')
+        container_id = container_id.rsplit('-', 1)[1]  # remove CRI prefix (i.e. 'containerd-<id>')
+
+    return container_id

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

@@ -0,0 +1,88 @@
+# Copyright (c) 2023, INRIA
+# Copyright (c) 2023, University of Lille
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright notice, this
+#   list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright notice,
+#   this list of conditions and the following disclaimer in the documentation
+#   and/or other materials provided with the distribution.
+#
+# * Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import logging
+from multiprocessing import Manager
+
+from powerapi.actor import Actor
+from powerapi.message import StartMessage, PoisonPillMessage
+from powerapi.processor.processor_actor import ProcessorState, 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
+
+
+class K8sPreProcessorState(ProcessorState):
+    """
+    State of the Kubernetes pre-processor actor.
+    """
+
+    def __init__(self, actor: Actor, target_actors: list, target_actors_names: list, api_mode: str, api_host: str, api_key: str):
+        super().__init__(actor, target_actors, target_actors_names)
+
+        self.api_mode = api_mode
+        self.api_host = api_host
+        self.api_key = api_key
+
+        self.manager = None
+        self.metadata_cache_manager = None
+        self.monitor_agent = None
+
+    def initialize_metadata_cache_manager(self):
+        """
+        Initialize the metadata cache manager.
+        This method should **ONLY** be called from the pre-processor actor process.
+        """
+        self.manager = Manager()
+        self.metadata_cache_manager = K8sMetadataCacheManager(self.manager)
+        self.monitor_agent = K8sMonitorAgent(self.metadata_cache_manager, self.api_mode, self.api_host, self.api_key)
+
+
+class K8sPreProcessorActor(ProcessorActor):
+    """
+    Pre-Processor Actor that adds Kubernetes related metadata to reports.
+    """
+
+    def __init__(self, name: str, target_actors: list, target_actors_names: list, api_mode: str = None, api_host: str = None,
+                 api_key: str = None, level_logger: int = logging.WARNING, timeout: int = 5000):
+        super().__init__(name, level_logger, timeout)
+
+        self.state = K8sPreProcessorState(self, target_actors, target_actors_names, api_mode, api_key, api_host)
+
+    def setup(self):
+        """
+        Set up the Kubernetes pre-processor actor.
+        """
+        self.state.initialize_metadata_cache_manager()
+
+        self.add_handler(StartMessage, K8sPreProcessorActorStartMessageHandler(self.state))
+        self.add_handler(HWPCReport, K8sPreProcessorActorHWPCReportHandler(self.state))
+        self.add_handler(PoisonPillMessage, K8sPreProcessorActorPoisonPillMessageHandler(self.state))