ft-orchestration use cgroups to controll memory usage of probe processes

Author: pascal260303

tools/ft-orchestration/src/probe/cento.py

@@ -16,9 +16,9 @@
 from lbr_testsuite.executable import (
     Executor,
     Tool,
-    Daemon,
     ExecutableProcessError,
 )
+from src.probe.process_group import Daemon
 from src.stats.merged import MergedStats
 from src.probe.probe_target import ProbeTarget
 
@@ -109,6 +109,8 @@ def __init__(
         mtu: int = 2048,
         sudo: bool = False,
         cache_size=None,
+        mem_limit=None,
+        cpu_limit=None,
         **kwargs: dict,
     ):
         interfaces_names = [ifc.name for ifc in interfaces]
@@ -136,6 +138,8 @@ def __init__(
         self._settings: CentoSettings = settings
         self._mtu = mtu
         self._target = target
+        self._cpu_limit = cpu_limit
+        self._mem_limit = mem_limit
 
         assert_tool_is_installed("cento", executor)
         if self._zero_copy:
@@ -287,7 +291,13 @@ def start(self):
 
         self._before_start()
 
-        self._process = Daemon(self._cmd, executor=self._executor, sudo=self._sudo)
+        self._process = Daemon(
+            self._cmd,
+            cpu_limit=self._cpu_limit,
+            mem_limit=self._mem_limit,
+            executor=self._executor,
+            sudo=self._sudo,
+        )
         # stderr is implicitly redirected to stdout
         self._process.set_outputs(self._log_file)
         self._process.start()

tools/ft-orchestration/src/probe/ipfixprobe.py

@@ -18,10 +18,10 @@
 from typing import Dict, List, Optional, Tuple
 
 from lbr_testsuite.executable import (
-    Daemon,
     Executor,
     Tool,
 )
+from src.probe.process_group import Daemon
 from src.common.required_field import required_field
 from src.common.tool_is_installed import assert_tool_is_installed
 from src.common.typed_dataclass import bool_convertor, typed_dataclass
@@ -353,6 +353,8 @@ def __init__(
         verbose: bool = False,
         settings: IpfixprobeSettings = None,
         sudo: bool = False,
+        mem_limit=None,
+        cpu_limit=None,
     ):
         """Init ipfixprobe connector.
 
@@ -389,6 +391,8 @@ def __init__(
         self._enabled_plugins = []
         self._last_run_stats = None
         self._timeouts = (settings.active_timeout, settings.inactive_timeout)
+        self._cpu_limit = cpu_limit
+        self._mem_limit = mem_limit
 
         assert_tool_is_installed("ipfixprobe", executor)
         self._cmd = self._prepare_cmd(target, protocols, settings)
@@ -445,6 +449,8 @@ def start(self) -> None:
             executor=self._executor,
             sudo=self._sudo,
             failure_verbosity="no-exception",
+            cpu_limit=self._cpu_limit,
+            mem_limit=self._mem_limit,
         )
         # stderr is implicitly redirected to stdout
         self._process.set_outputs(self._log_file)

tools/ft-orchestration/src/probe/nprobe.py

@@ -19,9 +19,9 @@
 from lbr_testsuite.executable import (
     Executor,
     Tool,
-    Daemon,
     ExecutableProcessError,
 )
+from src.probe.process_group import Daemon
 from src.stats.merged import MergedStats
 from src.probe.probe_target import ProbeTarget
 
@@ -118,6 +118,8 @@ def __init__(
         mtu: int = 2048,
         sudo: bool = False,
         cache_size=None,
+        mem_limit=None,
+        cpu_limit=None,
         **kwargs: dict,
     ):
         interfaces_names = [ifc.name for ifc in interfaces]
@@ -145,6 +147,8 @@ def __init__(
         self._mtu = mtu
         self._target = target
         self._protocols = protocols
+        self._cpu_limit = cpu_limit
+        self._mem_limit = mem_limit
 
         assert_tool_is_installed("nprobe", executor)
         if self._zero_copy:
@@ -304,7 +308,13 @@ def start(self):
             if self._settings.rss_queues > 1:
                 settings.interface = f"{settings.interface}@{i}"
             cmd = self._prepare_cmd(self._target, self._protocols, settings)
-            process = Daemon(cmd, executor=executors[i], sudo=self._sudo)
+            process = Daemon(
+                cmd,
+                executor=executors[i],
+                sudo=self._sudo,
+                cpu_limit=self._cpu_limit,
+                mem_limit=self._mem_limit,
+            )
             self._processes.append(process)
             log_file = Path(self._local_workdir, f"nprobe_{i}.log")
             self._log_files.append(log_file)

tools/ft-orchestration/src/probe/pmacct.py

@@ -9,12 +9,12 @@
 from typing import List, Optional
 
 from lbr_testsuite.executable import (
-    Daemon,
     ExecutableProcessError,
     Executor,
     Rsync,
     Tool,
 )
+from src.probe.process_group import Daemon
 from src.common.tool_is_installed import assert_tool_is_installed
 from src.common.typed_dataclass import typed_dataclass
 from src.common.utils import duplicate_executor
@@ -83,6 +83,8 @@ def __init__(
         cache_size=None,
         rss_queues: int = 1,
         binary: str = "pmacctd",
+        mem_limit=None,
+        cpu_limit=None,
         **kwargs: dict,
     ):
         # initialize PmacctSettings and map FlowTest values to pmacct values
@@ -143,6 +145,8 @@ def __init__(
         self._mtu = mtu
         self._sudo = sudo
         self._binary = binary
+        self._cpu_limit = cpu_limit
+        self._mem_limit = mem_limit
 
         assert_tool_is_installed(self._binary, executor)
 
@@ -249,7 +253,13 @@ def start(self):
                 time.sleep(2)
         self._before_start()
 
-        self._process = Daemon(self._cmd, executor=self._executor, sudo=self._sudo)
+        self._process = Daemon(
+            self._cmd,
+            executor=self._executor,
+            sudo=self._sudo,
+            cpu_limit=self._cpu_limit,
+            mem_limit=self._mem_limit,
+        )
         # stderr is implicitly redirected to stdout
         self._process.set_outputs(self._log_file)
         self._process.start()