Store k6 output and add socket info collection

Signed-off-by: Andrey Berezin <a.berezin@yadro.com>

CONTRIBUTING.md

@@ -3,8 +3,8 @@
 First, thank you for contributing! We love and encourage pull requests from
 everyone. Please follow the guidelines:
 
-- Check the open [issues](https://github.com/TrueCloudLab/frostfs-testlib/issues) and
+- Check the open [issues](https://git.frostfs.info/TrueCloudLab/frostfs-testlib/issues) and
-  [pull requests](https://github.com/TrueCloudLab/frostfs-testlib/pulls) for existing
+  [pull requests](https://git.frostfs.info/TrueCloudLab/frostfs-testlib/pulls) for existing
   discussions.
 
 - Open an issue first, to discuss a new feature or enhancement.
@@ -26,8 +26,8 @@ Start by forking the `frostfs-testlib` repository, make changes in a branch and
 send a pull request. We encourage pull requests to discuss code changes. Here
 are the steps in details:
 
-### Set up your GitHub Repository
+### Set up your Git Repository
-Fork [FrostFS testlib upstream](https://github.com/TrueCloudLab/frostfs-testlib/fork) source
+Fork [FrostFS testlib upstream](https://git.frostfs.info/TrueCloudLab/frostfs-testlib/forks) source
 repository to your own personal repository. Copy the URL of your fork and clone it:
 
 ```shell
@@ -37,7 +37,7 @@ $ git clone <url of your fork>
 ### Set up git remote as ``upstream``
 ```shell
 $ cd frostfs-testlib
-$ git remote add upstream https://github.com/TrueCloudLab/frostfs-testlib
+$ git remote add upstream https://git.frostfs.info/TrueCloudLab/frostfs-testlib
 $ git fetch upstream
 ```
 
@@ -99,8 +99,8 @@ $ git push origin feature/123-something_awesome
 ```
 
 ### Create a Pull Request
-Pull requests can be created via GitHub. Refer to [this
+Pull requests can be created via Git. Refer to [this
-document](https://help.github.com/articles/creating-a-pull-request/) for
+document](https://docs.codeberg.org/collaborating/pull-requests-and-git-flow/) for
 detailed steps on how to create a pull request. After a Pull Request gets peer
 reviewed and approved, it will be merged.
 

README.md

@@ -92,4 +92,4 @@ The library provides the following primary components:
  
 
 ## Contributing
-Any contributions to the library should conform to the [contribution guideline](https://github.com/TrueCloudLab/frostfs-testlib/blob/master/CONTRIBUTING.md).
+Any contributions to the library should conform to the [contribution guideline](https://git.frostfs.info/TrueCloudLab/frostfs-testlib/src/branch/master/CONTRIBUTING.md).

pyproject.toml

@@ -36,7 +36,7 @@ requires-python = ">=3.10"
 dev = ["black", "bumpver", "isort", "pre-commit"]
 
 [project.urls]
-Homepage = "https://github.com/TrueCloudLab/frostfs-testlib"
+Homepage = "https://git.frostfs.info/TrueCloudLab/frostfs-testlib"
 
 [project.entry-points."frostfs.testlib.reporter"]
 allure = "frostfs_testlib.reporter.allure_handler:AllureHandler"
@@ -47,6 +47,10 @@ docker = "frostfs_testlib.hosting.docker_host:DockerHost"
 [project.entry-points."frostfs.testlib.healthcheck"]
 basic = "frostfs_testlib.healthcheck.basic_healthcheck:BasicHealthcheck"
 
+[project.entry-points."frostfs.testlib.csc_managers"]
+config = "frostfs_testlib.storage.controllers.state_managers.config_state_manager:ConfigStateManager"
+
+
 [tool.isort]
 profile = "black"
 src_paths = ["src", "tests"]

src/frostfs_testlib/cli/frostfs_adm/morph.py

@@ -27,11 +27,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph deposit-notary",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def dump_balances(
@@ -56,11 +52,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph dump-balances",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def dump_config(self, rpc_endpoint: str) -> CommandResult:
@@ -74,11 +66,25 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph dump-config",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
+        )
-                for param, param_value in locals().items()
+
-                if param not in ["self"]
+    def set_config(
-            },
+        self, set_key_value: str, rpc_endpoint: Optional[str] = None, alphabet_wallets: Optional[str] = None
+    ) -> CommandResult:
+        """Add/update global config value in the FrostFS network.
+
+        Args:
+            set_key_value: key1=val1 [key2=val2 ...]
+            alphabet_wallets: Path to alphabet wallets dir
+            rpc_endpoint: N3 RPC node endpoint
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            f"morph set-config {set_key_value}",
+            **{param: param_value for param, param_value in locals().items() if param not in ["self", "set_key_value"]},
         )
 
     def dump_containers(
@@ -101,11 +107,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph dump-containers",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def dump_hashes(self, rpc_endpoint: str) -> CommandResult:
@@ -119,11 +121,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph dump-hashes",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def force_new_epoch(
@@ -140,11 +138,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph force-new-epoch",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def generate_alphabet(
@@ -165,11 +159,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph generate-alphabet",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def generate_storage_wallet(
@@ -192,11 +182,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph generate-storage-wallet",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def init(
@@ -219,7 +205,7 @@ class FrostfsAdmMorph(CliCommand):
             container_alias_fee: Container alias fee (default 500).
             container_fee: Container registration fee (default 1000).
             contracts: Path to archive with compiled FrostFS contracts
-                (default fetched from latest github release).
+                (default fetched from latest git release).
             epoch_duration: Amount of side chain blocks in one FrostFS epoch (default 240).
             homomorphic_disabled: Disable object homomorphic hashing.
             local_dump: Path to the blocks dump file.
@@ -232,11 +218,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph init",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def refill_gas(
@@ -259,11 +241,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph refill-gas",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def restore_containers(
@@ -286,11 +264,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph restore-containers",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def set_policy(
@@ -340,7 +314,7 @@ class FrostfsAdmMorph(CliCommand):
         Args:
             alphabet_wallets: Path to alphabet wallets dir.
             contracts: Path to archive with compiled FrostFS contracts
-                (default fetched from latest github release).
+                (default fetched from latest git release).
             rpc_endpoint: N3 RPC node endpoint.
 
         Returns:
@@ -348,17 +322,13 @@ class FrostfsAdmMorph(CliCommand):
         """
         return self._execute(
             "morph update-contracts",
-            **{
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
-                param: param_value
-                for param, param_value in locals().items()
-                if param not in ["self"]
-            },
         )
 
     def remove_nodes(
         self, node_netmap_keys: list[str], rpc_endpoint: Optional[str] = None, alphabet_wallets: Optional[str] = None
     ) -> CommandResult:
-        """    Move node to the Offline state in the candidates list 
+        """Move node to the Offline state in the candidates list
         and tick an epoch to update the netmap using frostfs-adm
 
         Args:
@@ -371,7 +341,7 @@ class FrostfsAdmMorph(CliCommand):
         """
         if not len(node_netmap_keys):
             raise AttributeError("Got empty node_netmap_keys list")
-        
+
         return self._execute(
             f"morph remove-nodes {' '.join(node_netmap_keys)}",
             **{
@@ -379,4 +349,4 @@ class FrostfsAdmMorph(CliCommand):
                 for param, param_value in locals().items()
                 if param not in ["self", "node_netmap_keys"]
             },
-        )
+        )

src/frostfs_testlib/cli/frostfs_cli/acl.py

@@ -22,7 +22,7 @@ class FrostfsCliACL(CliCommand):
             Well-known system object headers start with '$Object:' prefix.
             User defined headers start without prefix.
             Read more about filter keys at:
-                http://github.com/TrueCloudLab/frostfs-api/blob/master/proto-docs/acl.md#message-eaclrecordfilter
+                https://git.frostfs.info/TrueCloudLab/frostfs-api/src/branch/master/proto-docs/acl.md#message-eaclrecord-filter
           Match is '=' for matching and '!=' for non-matching filter.
           Value is a valid unicode string corresponding to object or request header value.
 

src/frostfs_testlib/cli/frostfs_cli/cli.py

@@ -3,6 +3,7 @@ from typing import Optional
 from frostfs_testlib.cli.frostfs_cli.accounting import FrostfsCliAccounting
 from frostfs_testlib.cli.frostfs_cli.acl import FrostfsCliACL
 from frostfs_testlib.cli.frostfs_cli.container import FrostfsCliContainer
+from frostfs_testlib.cli.frostfs_cli.control import FrostfsCliControl
 from frostfs_testlib.cli.frostfs_cli.netmap import FrostfsCliNetmap
 from frostfs_testlib.cli.frostfs_cli.object import FrostfsCliObject
 from frostfs_testlib.cli.frostfs_cli.session import FrostfsCliSession
@@ -25,6 +26,7 @@ class FrostfsCli:
     storagegroup: FrostfsCliStorageGroup
     util: FrostfsCliUtil
     version: FrostfsCliVersion
+    control: FrostfsCliControl
 
     def __init__(self, shell: Shell, frostfs_cli_exec_path: str, config_file: Optional[str] = None):
         self.accounting = FrostfsCliAccounting(shell, frostfs_cli_exec_path, config=config_file)
@@ -38,3 +40,4 @@ class FrostfsCli:
         self.util = FrostfsCliUtil(shell, frostfs_cli_exec_path, config=config_file)
         self.version = FrostfsCliVersion(shell, frostfs_cli_exec_path, config=config_file)
         self.tree = FrostfsCliTree(shell, frostfs_cli_exec_path, config=config_file)
+        self.control = FrostfsCliControl(shell, frostfs_cli_exec_path, config=config_file)

src/frostfs_testlib/cli/frostfs_cli/control.py

@@ -0,0 +1,58 @@
+from typing import Optional
+
+from frostfs_testlib.cli.cli_command import CliCommand
+from frostfs_testlib.shell import CommandResult
+
+
+class FrostfsCliControl(CliCommand):
+    def set_status(
+        self,
+        endpoint: str,
+        status: str,
+        wallet: Optional[str] = None,
+        force: Optional[bool] = None,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """Set status of the storage node in FrostFS network map
+
+        Args:
+        wallet: Path to the wallet or binary key
+        address: Address of wallet account
+        endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+        force: Force turning to local maintenance
+        status: New netmap status keyword ('online', 'offline', 'maintenance')
+        timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command`s result.
+        """
+        return self._execute(
+            "control set-status",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def healthcheck(
+        self,
+        endpoint: str,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """Set status of the storage node in FrostFS network map
+
+        Args:
+        wallet: Path to the wallet or binary key
+        address: Address of wallet account
+        endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+        force: Force turning to local maintenance
+        status: New netmap status keyword ('online', 'offline', 'maintenance')
+        timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command`s result.
+        """
+        return self._execute(
+            "control healthcheck",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/cli/netmap_parser.py

@@ -0,0 +1,86 @@
+import re
+
+from frostfs_testlib.storage.cluster import ClusterNode
+from frostfs_testlib.storage.dataclasses.storage_object_info import NodeNetInfo, NodeNetmapInfo
+
+
+class NetmapParser:
+    @staticmethod
+    def netinfo(output: str) -> NodeNetInfo:
+        regexes = {
+            "epoch": r"Epoch: (?P<epoch>\d+)",
+            "network_magic": r"Network magic: (?P<network_magic>.*$)",
+            "time_per_block": r"Time per block: (?P<time_per_block>\d+\w+)",
+            "container_fee": r"Container fee: (?P<container_fee>\d+)",
+            "epoch_duration": r"Epoch duration: (?P<epoch_duration>\d+)",
+            "inner_ring_candidate_fee": r"Inner Ring candidate fee: (?P<inner_ring_candidate_fee>\d+)",
+            "maximum_object_size": r"Maximum object size: (?P<maximum_object_size>\d+)",
+            "withdrawal_fee": r"Withdrawal fee: (?P<withdrawal_fee>\d+)",
+            "homomorphic_hashing_disabled": r"Homomorphic hashing disabled: (?P<homomorphic_hashing_disabled>true|false)",
+            "maintenance_mode_allowed": r"Maintenance mode allowed: (?P<maintenance_mode_allowed>true|false)",
+            "eigen_trust_alpha": r"EigenTrustAlpha: (?P<eigen_trust_alpha>\d+\w+$)",
+            "eigen_trust_iterations": r"EigenTrustIterations: (?P<eigen_trust_iterations>\d+)",
+        }
+        parse_result = {}
+
+        for key, regex in regexes.items():
+            search_result = re.search(regex, output, flags=re.MULTILINE)
+            if search_result == None:
+                parse_result[key] = None
+                continue
+            parse_result[key] = search_result[key].strip()
+
+        node_netinfo = NodeNetInfo(**parse_result)
+
+        return node_netinfo
+
+    @staticmethod
+    def snapshot_all_nodes(output: str) -> list[NodeNetmapInfo]:
+        """The code will parse each line and return each node as dataclass."""
+        netmap_nodes = output.split("Node ")[1:]
+        dataclasses_netmap = []
+        result_netmap = {}
+
+        regexes = {
+            "node_id": r"\d+: (?P<node_id>\w+)",
+            "node_data_ips": r"(?P<node_data_ips>/ip4/.+?)$",
+            "node_status": r"(?P<node_status>ONLINE|OFFLINE)",
+            "cluster_name": r"ClusterName: (?P<cluster_name>\w+)",
+            "continent": r"Continent: (?P<continent>\w+)",
+            "country": r"Country: (?P<country>\w+)",
+            "country_code": r"CountryCode: (?P<country_code>\w+)",
+            "external_address": r"ExternalAddr: (?P<external_address>/ip[4].+?)$",
+            "location": r"Location: (?P<location>\w+.*)",
+            "node": r"Node: (?P<node>\d+\.\d+\.\d+\.\d+)",
+            "price": r"Price: (?P<price>\d+)",
+            "sub_div": r"SubDiv: (?P<sub_div>.*)",
+            "sub_div_code": r"SubDivCode: (?P<sub_div_code>\w+)",
+            "un_locode": r"UN-LOCODE: (?P<un_locode>\w+.*)",
+            "role": r"role: (?P<role>\w+)",
+        }
+
+        for node in netmap_nodes:
+            for key, regex in regexes.items():
+                search_result = re.search(regex, node, flags=re.MULTILINE)
+                if key == "node_data_ips":
+                    result_netmap[key] = search_result[key].strip().split(" ")
+                    continue
+                if key == "external_address":
+                    result_netmap[key] = search_result[key].strip().split(",")
+                    continue
+                if search_result == None:
+                    result_netmap[key] = None
+                    continue
+                result_netmap[key] = search_result[key].strip()
+
+            dataclasses_netmap.append(NodeNetmapInfo(**result_netmap))
+
+        return dataclasses_netmap
+
+    @staticmethod
+    def snapshot_one_node(output: str, cluster_node: ClusterNode) -> NodeNetmapInfo | None:
+        snapshot_nodes = NetmapParser.snapshot_all_nodes(output=output)
+        snapshot_node = [node for node in snapshot_nodes if node.node == cluster_node.host_ip]
+        if not snapshot_node:
+            return None
+        return snapshot_node[0]

src/frostfs_testlib/load/__init__.py

@@ -1,4 +1,5 @@
-from frostfs_testlib.load.interfaces import Loader, ScenarioRunner
+from frostfs_testlib.load.interfaces.loader import Loader
+from frostfs_testlib.load.interfaces.scenario_runner import ScenarioRunner
 from frostfs_testlib.load.load_config import (
     EndpointSelectionStrategy,
     K6ProcessAllocationStrategy,
@@ -11,4 +12,4 @@ from frostfs_testlib.load.load_config import (
 )
 from frostfs_testlib.load.load_report import LoadReport
 from frostfs_testlib.load.loaders import NodeLoader, RemoteLoader
-from frostfs_testlib.load.runners import DefaultRunner, LocalRunner
+from frostfs_testlib.load.runners import DefaultRunner, LocalRunner, S3LocalRunner

src/frostfs_testlib/load/interfaces/loader.py

@@ -0,0 +1,14 @@
+from abc import ABC, abstractmethod
+
+from frostfs_testlib.shell.interfaces import Shell
+
+
+class Loader(ABC):
+    @abstractmethod
+    def get_shell(self) -> Shell:
+        """Get shell for the loader"""
+
+    @property
+    @abstractmethod
+    def ip(self):
+        """Get address of the loader"""

src/frostfs_testlib/load/interfaces/scenario_runner.py

@@ -1,20 +1,8 @@
 from abc import ABC, abstractmethod
 
+from frostfs_testlib.load.k6 import K6
 from frostfs_testlib.load.load_config import LoadParams
-from frostfs_testlib.shell.interfaces import Shell
 from frostfs_testlib.storage.cluster import ClusterNode
-from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
-
-
-class Loader(ABC):
-    @abstractmethod
-    def get_shell(self) -> Shell:
-        """Get shell for the loader"""
-
-    @property
-    @abstractmethod
-    def ip(self):
-        """Get address of the loader"""
 
 
 class ScenarioRunner(ABC):
@@ -32,6 +20,10 @@ class ScenarioRunner(ABC):
     def init_k6_instances(self, load_params: LoadParams, endpoints: list[str], k6_dir: str):
         """Init K6 instances"""
 
+    @abstractmethod
+    def get_k6_instances(self) -> list[K6]:
+        """Get K6 instances"""
+
     @abstractmethod
     def start(self):
         """Start K6 instances"""

src/frostfs_testlib/load/k6.py

@@ -8,13 +8,8 @@ from time import sleep
 from typing import Any
 from urllib.parse import urlparse
 
-from frostfs_testlib.load.interfaces import Loader
+from frostfs_testlib.load.interfaces.loader import Loader
-from frostfs_testlib.load.load_config import (
+from frostfs_testlib.load.load_config import K6ProcessAllocationStrategy, LoadParams, LoadScenario, LoadType
-    K6ProcessAllocationStrategy,
-    LoadParams,
-    LoadScenario,
-    LoadType,
-)
 from frostfs_testlib.processes.remote_process import RemoteProcess
 from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.resources.common import STORAGE_USER_NAME
@@ -59,6 +54,7 @@ class K6:
         self.loader: Loader = loader
         self.shell: Shell = shell
         self.wallet = wallet
+        self.preset_output: str = ""
         self.summary_json: str = os.path.join(
             self.load_params.working_dir,
             f"{self.load_params.load_id}_{self.load_params.scenario.value}_summary.json",
@@ -101,10 +97,10 @@ class K6:
             command = " ".join(command_args)
             result = self.shell.exec(command)
 
-            assert (
+            assert result.return_code == EXIT_RESULT_CODE, f"Return code of preset is not zero: {result.stdout}"
-                result.return_code == EXIT_RESULT_CODE
+
-            ), f"Return code of preset is not zero: {result.stdout}"
+            self.preset_output = result.stdout.strip("\n")
-            return result.stdout.strip("\n")
+            return self.preset_output
 
     @reporter.step_deco("Generate K6 command")
     def _generate_env_variables(self) -> str:
@@ -113,31 +109,21 @@ class K6:
         env_vars[f"{self.load_params.load_type.value.upper()}_ENDPOINTS"] = ",".join(self.endpoints)
         env_vars["SUMMARY_JSON"] = self.summary_json
 
-        reporter.attach(
+        reporter.attach("\n".join(f"{param}: {value}" for param, value in env_vars.items()), "K6 ENV variables")
-            "\n".join(f"{param}: {value}" for param, value in env_vars.items()), "K6 ENV variables"
+        return " ".join([f"-e {param}='{value}'" for param, value in env_vars.items() if value is not None])
-        )
-        return " ".join(
-            [f"-e {param}='{value}'" for param, value in env_vars.items() if value is not None]
-        )
 
     def start(self) -> None:
-        with reporter.step(
+        with reporter.step(f"Start load from loader {self.loader.ip} on endpoints {self.endpoints}"):
-            f"Start load from loader {self.loader.ip} on endpoints {self.endpoints}"
-        ):
             self._start_time = int(datetime.utcnow().timestamp())
             command = (
                 f"{self._k6_dir}/k6 run {self._generate_env_variables()} "
                 f"{self._k6_dir}/scenarios/{self.load_params.scenario.value}.js"
             )
             user = STORAGE_USER_NAME if self.load_params.scenario == LoadScenario.LOCAL else None
-            self._k6_process = RemoteProcess.create(
+            self._k6_process = RemoteProcess.create(command, self.shell, self.load_params.working_dir, user)
-                command, self.shell, self.load_params.working_dir, user
-            )
 
     def wait_until_finished(self, soft_timeout: int = 0) -> None:
-        with reporter.step(
+        with reporter.step(f"Wait until load is finished from loader {self.loader.ip} on endpoints {self.endpoints}"):
-            f"Wait until load is finished from loader {self.loader.ip} on endpoints {self.endpoints}"
-        ):
             if self.load_params.scenario == LoadScenario.VERIFY:
                 timeout = self.load_params.verify_time or 0
             else:
@@ -180,9 +166,11 @@ class K6:
             while timeout > 0:
                 if not self._k6_process.running():
                     return
-                remaining_time_hours = f"{timeout//3600}h" if timeout//3600 != 0 else ""
+                remaining_time_hours = f"{timeout//3600}h" if timeout // 3600 != 0 else ""
-                remaining_time_minutes = f"{timeout//60%60}m" if timeout//60%60 != 0 else ""
+                remaining_time_minutes = f"{timeout//60%60}m" if timeout // 60 % 60 != 0 else ""
-                logger.info(f"K6 is running. Remaining time {remaining_time_hours}{remaining_time_minutes}{timeout%60}s. Next check after {wait_interval} seconds...")
+                logger.info(
+                    f"K6 is running. Remaining time {remaining_time_hours}{remaining_time_minutes}{timeout%60}s. Next check after {wait_interval} seconds..."
+                )
                 sleep(wait_interval)
                 timeout -= min(timeout, wait_interval)
                 wait_interval = max(
@@ -198,9 +186,7 @@ class K6:
                 raise TimeoutError(f"Expected K6 to finish after {original_timeout} sec.")
 
     def get_results(self) -> Any:
-        with reporter.step(
+        with reporter.step(f"Get load results from loader {self.loader.ip} on endpoints {self.endpoints}"):
-            f"Get load results from loader {self.loader.ip} on endpoints {self.endpoints}"
-        ):
             self.__log_output()
 
             if not self.summary_json:
@@ -231,9 +217,7 @@ class K6:
         return False
 
     @reporter.step_deco("Wait until K6 process end")
-    @wait_for_success(
+    @wait_for_success(K6_STOP_SIGNAL_TIMEOUT, 15, False, False, "Can not stop K6 process within timeout")
-        K6_STOP_SIGNAL_TIMEOUT, 15, False, False, "Can not stop K6 process within timeout"
-    )
     def _wait_until_process_end(self):
         return self._k6_process.running()
 

src/frostfs_testlib/load/load_config.py

@@ -20,6 +20,7 @@ class LoadScenario(Enum):
     S3 = "s3"
     S3_CAR = "s3_car"
     S3_MULTIPART = "s3_multipart"
+    S3_LOCAL = "s3local"
     HTTP = "http"
     VERIFY = "verify"
     LOCAL = "local"
@@ -38,11 +39,12 @@ all_load_scenarios = [
     LoadScenario.S3_CAR,
     LoadScenario.gRPC_CAR,
     LoadScenario.LOCAL,
-    LoadScenario.S3_MULTIPART
+    LoadScenario.S3_MULTIPART,
+    LoadScenario.S3_LOCAL
 ]
 all_scenarios = all_load_scenarios.copy() + [LoadScenario.VERIFY]
 
-constant_vus_scenarios = [LoadScenario.gRPC, LoadScenario.S3, LoadScenario.HTTP, LoadScenario.LOCAL, LoadScenario.S3_MULTIPART]
+constant_vus_scenarios = [LoadScenario.gRPC, LoadScenario.S3, LoadScenario.HTTP, LoadScenario.LOCAL, LoadScenario.S3_MULTIPART, LoadScenario.S3_LOCAL]
 constant_arrival_rate_scenarios = [LoadScenario.gRPC_CAR, LoadScenario.S3_CAR]
 
 grpc_preset_scenarios = [
@@ -51,7 +53,7 @@ grpc_preset_scenarios = [
     LoadScenario.gRPC_CAR,
     LoadScenario.LOCAL,
 ]
-s3_preset_scenarios = [LoadScenario.S3, LoadScenario.S3_CAR, LoadScenario.S3_MULTIPART]
+s3_preset_scenarios = [LoadScenario.S3, LoadScenario.S3_CAR, LoadScenario.S3_MULTIPART, LoadScenario.S3_LOCAL]
 
 
 @dataclass
@@ -172,9 +174,13 @@ class LoadParams:
     preset: Optional[Preset] = None
     # K6 download url
     k6_url: Optional[str] = None
+    # Requests module url
+    requests_module_url: Optional[str] = None
+    # aws cli download url
+    awscli_url: Optional[str] = None
     # No ssl verification flag
     no_verify_ssl: Optional[bool] = metadata_field(
-        [LoadScenario.S3, LoadScenario.S3_CAR, LoadScenario.S3_MULTIPART, LoadScenario.VERIFY, LoadScenario.HTTP],
+        [LoadScenario.S3, LoadScenario.S3_CAR, LoadScenario.S3_MULTIPART, LoadScenario.S3_LOCAL, LoadScenario.VERIFY, LoadScenario.HTTP],
         "no-verify-ssl",
         "NO_VERIFY_SSL",
         False,
@@ -283,7 +289,9 @@ class LoadParams:
 
     # ------- LOCAL SCENARIO PARAMS -------
     # Config file location (filled automatically)
-    config_file: Optional[str] = metadata_field([LoadScenario.LOCAL], None, "CONFIG_FILE", False)
+    config_file: Optional[str] = metadata_field([LoadScenario.LOCAL, LoadScenario.S3_LOCAL], None, "CONFIG_FILE", False)
+    # Config directory location (filled automatically)
+    config_dir: Optional[str] = metadata_field([LoadScenario.S3_LOCAL], None, "CONFIG_DIR", False)
 
     def set_id(self, load_id):
         self.load_id = load_id

src/frostfs_testlib/load/load_metrics.py

@@ -165,6 +165,14 @@ class S3Metrics(MetricsBase):
     _DELETE_ERRORS = "aws_obj_delete_fails"
     _DELETE_LATENCY = "aws_obj_delete_duration"
 
+class S3LocalMetrics(MetricsBase):
+    _WRITE_SUCCESS = "s3local_obj_put_total"
+    _WRITE_ERRORS = "s3local_obj_put_fails"
+    _WRITE_LATENCY = "s3local_obj_put_duration"
+
+    _READ_SUCCESS = "s3local_obj_get_total"
+    _READ_ERRORS = "s3local_obj_get_fails"
+    _READ_LATENCY = "s3local_obj_get_duration"
 
 class LocalMetrics(MetricsBase):
     _WRITE_SUCCESS = "local_obj_put_total"
@@ -197,6 +205,7 @@ def get_metrics_object(load_type: LoadScenario, summary: dict[str, Any]) -> Metr
         LoadScenario.S3: S3Metrics,
         LoadScenario.S3_CAR: S3Metrics,
         LoadScenario.S3_MULTIPART: S3Metrics,
+        LoadScenario.S3_LOCAL: S3LocalMetrics,
         LoadScenario.VERIFY: VerifyMetrics,
         LoadScenario.LOCAL: LocalMetrics,
     }

src/frostfs_testlib/load/load_report.py

@@ -97,6 +97,7 @@ class LoadReport:
             LoadScenario.gRPC_CAR: "open model",
             LoadScenario.S3_CAR: "open model",
             LoadScenario.LOCAL: "local fill",
+            LoadScenario.S3_LOCAL: "local fill"
         }
 
         return model_map[self.load_params.scenario]

src/frostfs_testlib/load/loaders.py

@@ -1,4 +1,4 @@
-from frostfs_testlib.load.interfaces import Loader
+from frostfs_testlib.load.interfaces.loader import Loader
 from frostfs_testlib.resources.load_params import (
     LOAD_NODE_SSH_PASSWORD,
     LOAD_NODE_SSH_PRIVATE_KEY_PASSPHRASE,

src/frostfs_testlib/load/runners.py

@@ -10,7 +10,8 @@ from urllib.parse import urlparse
 import yaml
 
 from frostfs_testlib.cli.frostfs_authmate.authmate import FrostfsAuthmate
-from frostfs_testlib.load.interfaces import Loader, ScenarioRunner
+from frostfs_testlib.load.interfaces.loader import Loader
+from frostfs_testlib.load.interfaces.scenario_runner import ScenarioRunner
 from frostfs_testlib.load.k6 import K6
 from frostfs_testlib.load.load_config import K6ProcessAllocationStrategy, LoadParams, LoadType
 from frostfs_testlib.load.loaders import NodeLoader, RemoteLoader
@@ -18,11 +19,7 @@ from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.resources import optionals
 from frostfs_testlib.resources.cli import FROSTFS_AUTHMATE_EXEC
 from frostfs_testlib.resources.common import STORAGE_USER_NAME
-from frostfs_testlib.resources.load_params import (
+from frostfs_testlib.resources.load_params import BACKGROUND_LOAD_VUS_COUNT_DIVISOR, LOAD_NODE_SSH_USER, LOAD_NODES
-    BACKGROUND_LOAD_VUS_COUNT_DIVISOR,
-    LOAD_NODE_SSH_USER,
-    LOAD_NODES,
-)
 from frostfs_testlib.shell.command_inspectors import SuInspector
 from frostfs_testlib.shell.interfaces import CommandOptions, InteractiveInput
 from frostfs_testlib.storage.cluster import ClusterNode
@@ -54,6 +51,9 @@ class RunnerBase(ScenarioRunner):
 
         return any([future.result() for future in futures])
 
+    def get_k6_instances(self):
+        return self.k6_instances
+
 
 class DefaultRunner(RunnerBase):
     loaders: list[Loader]
@@ -83,14 +83,10 @@ class DefaultRunner(RunnerBase):
 
         with reporter.step("Init s3 client on loaders"):
             storage_node = nodes_under_load[0].service(StorageNode)
-            s3_public_keys = [
+            s3_public_keys = [node.service(S3Gate).get_wallet_public_key() for node in cluster_nodes]
-                node.service(S3Gate).get_wallet_public_key() for node in cluster_nodes
-            ]
             grpc_peer = storage_node.get_rpc_endpoint()
 
-            parallel(
+            parallel(self._prepare_loader, self.loaders, load_params, grpc_peer, s3_public_keys, k6_dir)
-                self._prepare_loader, self.loaders, load_params, grpc_peer, s3_public_keys, k6_dir
-            )
 
     def _prepare_loader(
         self,
@@ -112,9 +108,9 @@ class DefaultRunner(RunnerBase):
                 wallet_password=self.loaders_wallet.password,
             ).stdout
             aws_access_key_id = str(
-                re.search(
+                re.search(r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output).group(
-                    r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output
+                    "aws_access_key_id"
-                ).group("aws_access_key_id")
+                )
             )
             aws_secret_access_key = str(
                 re.search(
@@ -125,9 +121,7 @@ class DefaultRunner(RunnerBase):
 
             configure_input = [
                 InteractiveInput(prompt_pattern=r"AWS Access Key ID.*", input=aws_access_key_id),
-                InteractiveInput(
+                InteractiveInput(prompt_pattern=r"AWS Secret Access Key.*", input=aws_secret_access_key),
-                    prompt_pattern=r"AWS Secret Access Key.*", input=aws_secret_access_key
-                ),
                 InteractiveInput(prompt_pattern=r".*", input=""),
                 InteractiveInput(prompt_pattern=r".*", input=""),
             ]
@@ -144,16 +138,12 @@ class DefaultRunner(RunnerBase):
         }
         endpoints_generators = {
             K6ProcessAllocationStrategy.PER_LOAD_NODE: itertools.cycle([endpoints]),
-            K6ProcessAllocationStrategy.PER_ENDPOINT: itertools.cycle(
+            K6ProcessAllocationStrategy.PER_ENDPOINT: itertools.cycle([[endpoint] for endpoint in endpoints]),
-                [[endpoint] for endpoint in endpoints]
-            ),
         }
         k6_processes_count = k6_distribution_count[load_params.k6_process_allocation_strategy]
         endpoints_gen = endpoints_generators[load_params.k6_process_allocation_strategy]
 
-        distributed_load_params_list = self._get_distributed_load_params_list(
+        distributed_load_params_list = self._get_distributed_load_params_list(load_params, k6_processes_count)
-            load_params, k6_processes_count
-        )
 
         futures = parallel(
             self._init_k6_instance,
@@ -164,9 +154,7 @@ class DefaultRunner(RunnerBase):
         )
         self.k6_instances = [future.result() for future in futures]
 
-    def _init_k6_instance(
+    def _init_k6_instance(self, load_params_for_loader: LoadParams, loader: Loader, endpoints: list[str], k6_dir: str):
-        self, load_params_for_loader: LoadParams, loader: Loader, endpoints: list[str], k6_dir: str
-    ):
         shell = loader.get_shell()
         with reporter.step(f"Init K6 instance on {loader.ip} for endpoints {endpoints}"):
             with reporter.step(f"Make working directory"):
@@ -204,9 +192,7 @@ class DefaultRunner(RunnerBase):
                 and getattr(original_load_params, field.name) is not None
             ):
                 original_value = getattr(original_load_params, field.name)
-                distribution = self._get_distribution(
+                distribution = self._get_distribution(math.ceil(original_value / divisor), workers_count)
-                    math.ceil(original_value / divisor), workers_count
-                )
                 for i in range(workers_count):
                     setattr(distributed_load_params[i], field.name, distribution[i])
 
@@ -233,10 +219,7 @@ class DefaultRunner(RunnerBase):
         # Remainder of clients left to be distributed
         remainder = clients_count - clients_per_worker * workers_count
 
-        distribution = [
+        distribution = [clients_per_worker + 1 if i < remainder else clients_per_worker for i in range(workers_count)]
-            clients_per_worker + 1 if i < remainder else clients_per_worker
-            for i in range(workers_count)
-        ]
         return distribution
 
     def start(self):
@@ -245,9 +228,7 @@ class DefaultRunner(RunnerBase):
         parallel([k6.start for k6 in self.k6_instances])
 
         wait_after_start_time = datetime_utils.parse_time(load_params.setup_timeout) + 5
-        with reporter.step(
+        with reporter.step(f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"):
-            f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"
-        ):
             time.sleep(wait_after_start_time)
 
     def stop(self):
@@ -327,9 +308,7 @@ class LocalRunner(RunnerBase):
         with reporter.step("Update limits.conf"):
             limits_path = "/etc/security/limits.conf"
             self.file_keeper.add(cluster_node.storage_node, limits_path)
-            content = (
+            content = f"{STORAGE_USER_NAME} hard nofile 65536\n{STORAGE_USER_NAME} soft nofile 65536\n"
-                f"{STORAGE_USER_NAME} hard nofile 65536\n{STORAGE_USER_NAME} soft nofile 65536\n"
-            )
             shell.exec(f"echo '{content}' | sudo tee {limits_path}")
 
         with reporter.step("Download K6"):
@@ -339,9 +318,7 @@ class LocalRunner(RunnerBase):
             shell.exec(f"sudo chmod -R 777 {k6_dir}")
 
         with reporter.step("Create empty_passwd"):
-            self.wallet = WalletInfo(
+            self.wallet = WalletInfo(f"{k6_dir}/scenarios/files/wallet.json", "", "/tmp/empty_passwd.yml")
-                f"{k6_dir}/scenarios/files/wallet.json", "", "/tmp/empty_passwd.yml"
-            )
             content = yaml.dump({"password": ""})
             shell.exec(f'echo "{content}" | sudo tee {self.wallet.config_path}')
             shell.exec(f"sudo chmod -R 777 {self.wallet.config_path}")
@@ -383,15 +360,13 @@ class LocalRunner(RunnerBase):
     def start(self):
         load_params = self.k6_instances[0].load_params
 
-        self.cluster_state_controller.stop_all_s3_gates()
+        self.cluster_state_controller.stop_services_of_type(S3Gate)
-        self.cluster_state_controller.stop_all_storage_services()
+        self.cluster_state_controller.stop_services_of_type(StorageNode)
 
         parallel([k6.start for k6 in self.k6_instances])
 
         wait_after_start_time = datetime_utils.parse_time(load_params.setup_timeout) + 5
-        with reporter.step(
+        with reporter.step(f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"):
-            f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"
-        ):
             time.sleep(wait_after_start_time)
 
     @reporter.step_deco("Restore passwd on {cluster_node}")
@@ -408,8 +383,7 @@ class LocalRunner(RunnerBase):
         for k6_instance in self.k6_instances:
             k6_instance.stop()
 
-        self.cluster_state_controller.start_stopped_storage_services()
+        self.cluster_state_controller.start_all_stopped_services()
-        self.cluster_state_controller.start_stopped_s3_gates()
 
     def get_results(self) -> dict:
         results = {}
@@ -420,3 +394,124 @@ class LocalRunner(RunnerBase):
         parallel(self.restore_passwd_on_node, self.nodes_under_load)
 
         return results
+
+
+class S3LocalRunner(LocalRunner):
+    endpoints: list[str]
+    k6_dir: str
+
+    @reporter.step_deco("Run preset on loaders")
+    def preset(self):
+        LocalRunner.preset(self)
+        with reporter.step(f"Resolve containers in preset"):
+            parallel(self._resolve_containers_in_preset, self.k6_instances)
+
+    @reporter.step_deco("Resolve containers in preset")
+    def _resolve_containers_in_preset(self, k6_instance: K6):
+        k6_instance.shell.exec(
+            f"sudo {self.k6_dir}/scenarios/preset/resolve_containers_in_preset.py --endpoint {k6_instance.endpoints[0]} --preset_file {k6_instance.load_params.preset.pregen_json}"
+        )
+
+    @reporter.step_deco("Init k6 instances")
+    def init_k6_instances(self, load_params: LoadParams, endpoints: list[str], k6_dir: str):
+        self.k6_instances = []
+        futures = parallel(
+            self._init_k6_instance_,
+            self.loaders,
+            load_params,
+            endpoints,
+            k6_dir,
+        )
+        self.k6_instances = [future.result() for future in futures]
+
+    def _init_k6_instance_(self, loader: Loader, load_params: LoadParams, endpoints: list[str], k6_dir: str):
+        shell = loader.get_shell()
+        with reporter.step(f"Init K6 instance on {loader.ip} for endpoints {endpoints}"):
+            with reporter.step(f"Make working directory"):
+                shell.exec(f"sudo mkdir -p {load_params.working_dir}")
+                # If we chmod /home/<user_name> folder we can no longer ssh to the node
+                # !! IMPORTANT !!
+                if (
+                    load_params.working_dir
+                    and not load_params.working_dir == f"/home/{LOAD_NODE_SSH_USER}"
+                    and not load_params.working_dir == f"/home/{LOAD_NODE_SSH_USER}/"
+                ):
+                    shell.exec(f"sudo chmod -R 777 {load_params.working_dir}")
+
+            return K6(
+                load_params,
+                self.endpoints,
+                k6_dir,
+                shell,
+                loader,
+                self.wallet,
+            )
+
+    @run_optionally(optionals.OPTIONAL_BACKGROUND_LOAD_ENABLED)
+    @reporter.step_deco("Preparation steps")
+    def prepare(
+        self,
+        load_params: LoadParams,
+        cluster_nodes: list[ClusterNode],
+        nodes_under_load: list[ClusterNode],
+        k6_dir: str,
+    ):
+        self.k6_dir = k6_dir
+        with reporter.step("Init s3 client on loaders"):
+            storage_node = nodes_under_load[0].service(StorageNode)
+            s3_public_keys = [node.service(S3Gate).get_wallet_public_key() for node in cluster_nodes]
+            grpc_peer = storage_node.get_rpc_endpoint()
+
+        parallel(self.prepare_node, nodes_under_load, k6_dir, load_params, s3_public_keys, grpc_peer)
+
+    @reporter.step_deco("Prepare node {cluster_node}")
+    def prepare_node(
+        self, cluster_node: ClusterNode, k6_dir: str, load_params: LoadParams, s3_public_keys: list[str], grpc_peer: str
+    ):
+        LocalRunner.prepare_node(self, cluster_node, k6_dir, load_params)
+        self.endpoints = cluster_node.s3_gate.get_all_endpoints()
+        shell = cluster_node.host.get_shell()
+
+        with reporter.step("Uninstall previous installation of aws cli"):
+            shell.exec(f"sudo rm -rf /usr/local/aws-cli")
+            shell.exec(f"sudo rm -rf /usr/local/bin/aws")
+            shell.exec(f"sudo rm -rf /usr/local/bin/aws_completer")
+
+        with reporter.step("Install aws cli"):
+            shell.exec(f"sudo curl {load_params.awscli_url} -o {k6_dir}/awscliv2.zip")
+            shell.exec(f"sudo unzip -q {k6_dir}/awscliv2.zip -d {k6_dir}")
+            shell.exec(f"sudo {k6_dir}/aws/install")
+
+        with reporter.step("Install requests python module"):
+            shell.exec(f"sudo apt-get -y install python3-pip")
+            shell.exec(f"sudo curl -so {k6_dir}/requests.tar.gz {load_params.requests_module_url}")
+            shell.exec(f"sudo python3 -m pip install -I {k6_dir}/requests.tar.gz")
+
+        with reporter.step(f"Init s3 client on {cluster_node.host_ip}"):
+            frostfs_authmate_exec: FrostfsAuthmate = FrostfsAuthmate(shell, FROSTFS_AUTHMATE_EXEC)
+            issue_secret_output = frostfs_authmate_exec.secret.issue(
+                wallet=self.wallet.path,
+                peer=grpc_peer,
+                gate_public_key=s3_public_keys,
+                container_placement_policy=load_params.preset.container_placement_policy,
+                container_policy=f"{k6_dir}/scenarios/files/policy.json",
+                wallet_password=self.wallet.password,
+            ).stdout
+            aws_access_key_id = str(
+                re.search(r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output).group(
+                    "aws_access_key_id"
+                )
+            )
+            aws_secret_access_key = str(
+                re.search(
+                    r"secret_access_key.*:\s.(?P<aws_secret_access_key>\w*)",
+                    issue_secret_output,
+                ).group("aws_secret_access_key")
+            )
+            configure_input = [
+                InteractiveInput(prompt_pattern=r"AWS Access Key ID.*", input=aws_access_key_id),
+                InteractiveInput(prompt_pattern=r"AWS Secret Access Key.*", input=aws_secret_access_key),
+                InteractiveInput(prompt_pattern=r".*", input=""),
+                InteractiveInput(prompt_pattern=r".*", input=""),
+            ]
+            shell.exec("aws configure", CommandOptions(interactive_inputs=configure_input))

src/frostfs_testlib/plugins/__init__.py

@@ -17,3 +17,16 @@ def load_plugin(plugin_group: str, name: str) -> Any:
         return None
     plugin = plugins[name]
     return plugin.load()
+
+
+def load_all(group: str) -> Any:
+    """Loads all plugins using entry point specification.
+
+    Args:
+        plugin_group: Name of plugin group.
+
+    Returns:
+        Classes from specified group.
+    """
+    plugins = entry_points(group=group)
+    return [plugin.load() for plugin in plugins]

src/frostfs_testlib/shell/local_shell.py

@@ -62,7 +62,8 @@ class LocalShell(Shell):
         if options.check and result.return_code != 0:
             raise RuntimeError(
                 f"Command: {command}\nreturn code: {result.return_code}\n"
-                f"Output: {result.stdout}"
+                f"Output: {result.stdout}\n"
+                f"Stderr: {result.stderr}\n"
             )
         return result
 
@@ -94,9 +95,7 @@ class LocalShell(Shell):
                 return_code=exc.returncode,
             )
             raise RuntimeError(
-                f"Command: {command}\nError:\n"
+                f"Command: {command}\nError:\n" f"return code: {exc.returncode}\n" f"output: {exc.output}"
-                f"return code: {exc.returncode}\n"
-                f"output: {exc.output}"
             ) from exc
         except OSError as exc:
             raise RuntimeError(f"Command: {command}\nOutput: {exc.strerror}") from exc

src/frostfs_testlib/shell/ssh_shell.py

@@ -6,27 +6,11 @@ from functools import lru_cache, wraps
 from time import sleep
 from typing import ClassVar, Optional, Tuple
 
-from paramiko import (
+from paramiko import AutoAddPolicy, Channel, ECDSAKey, Ed25519Key, PKey, RSAKey, SSHClient, SSHException, ssh_exception
-    AutoAddPolicy,
-    Channel,
-    ECDSAKey,
-    Ed25519Key,
-    PKey,
-    RSAKey,
-    SSHClient,
-    SSHException,
-    ssh_exception,
-)
 from paramiko.ssh_exception import AuthenticationException
 
 from frostfs_testlib.reporter import get_reporter
-from frostfs_testlib.shell.interfaces import (
+from frostfs_testlib.shell.interfaces import CommandInspector, CommandOptions, CommandResult, Shell, SshCredentials
-    CommandInspector,
-    CommandOptions,
-    CommandResult,
-    Shell,
-    SshCredentials,
-)
 
 logger = logging.getLogger("frostfs.testlib.shell")
 reporter = get_reporter()
@@ -97,8 +81,7 @@ class SshConnectionProvider:
                     )
                 else:
                     logger.info(
-                        f"Trying to connect to host {host} as {creds.ssh_login} using password "
+                        f"Trying to connect to host {host} as {creds.ssh_login} using password " f"(attempt {attempt})"
-                        f"(attempt {attempt})"
                     )
                     connection.connect(
                         hostname=host,
@@ -141,9 +124,7 @@ class HostIsNotAvailable(Exception):
 
 def log_command(func):
     @wraps(func)
-    def wrapper(
+    def wrapper(shell: "SSHShell", command: str, options: CommandOptions, *args, **kwargs) -> CommandResult:
-        shell: "SSHShell", command: str, options: CommandOptions, *args, **kwargs
-    ) -> CommandResult:
         command_info = command.removeprefix("$ProgressPreference='SilentlyContinue'\n")
         with reporter.step(command_info):
             logger.info(f'Execute command "{command}" on "{shell.host}"')
@@ -238,15 +219,13 @@ class SSHShell(Shell):
 
         if options.check and result.return_code != 0:
             raise RuntimeError(
-                f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}"
+                f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}\nStderr: {result.stderr}\n"
             )
         return result
 
     @log_command
     def _exec_interactive(self, command: str, options: CommandOptions) -> CommandResult:
-        stdin, stdout, stderr = self._connection.exec_command(
+        stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout, get_pty=True)
-            command, timeout=options.timeout, get_pty=True
-        )
         for interactive_input in options.interactive_inputs:
             input = interactive_input.input
             if not input.endswith("\n"):

src/frostfs_testlib/steps/node_management.py

@@ -169,7 +169,7 @@ def include_node_to_network_map(
     storage_node_set_status(node_to_include, status="online")
 
     # Per suggestion of @fyrchik we need to wait for 2 blocks after we set status and after tick epoch.
-    # First sleep can be omitted after https://github.com/TrueCloudLab/frostfs-node/issues/60 complete.
+    # First sleep can be omitted after https://git.frostfs.info/TrueCloudLab/frostfs-node/issues/60 complete.
 
     time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * 2)
     tick_epoch(shell, cluster)

src/frostfs_testlib/storage/cluster.py

@@ -8,14 +8,10 @@ from frostfs_testlib.hosting import Host, Hosting
 from frostfs_testlib.hosting.config import ServiceConfig
 from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.storage import get_service_registry
+from frostfs_testlib.storage.configuration.interfaces import ServiceConfigurationYml
+from frostfs_testlib.storage.configuration.service_configuration import ServiceConfiguration
 from frostfs_testlib.storage.constants import ConfigAttributes
-from frostfs_testlib.storage.dataclasses.frostfs_services import (
+from frostfs_testlib.storage.dataclasses.frostfs_services import HTTPGate, InnerRing, MorphChain, S3Gate, StorageNode
-    HTTPGate,
-    InnerRing,
-    MorphChain,
-    S3Gate,
-    StorageNode,
-)
 from frostfs_testlib.storage.dataclasses.node_base import NodeBase, ServiceClass
 from frostfs_testlib.storage.dataclasses.storage_object_info import Interfaces
 from frostfs_testlib.storage.service_registry import ServiceRegistry
@@ -93,6 +89,9 @@ class ClusterNode:
         config_str = yaml.dump(new_config)
         shell.exec(f"echo '{config_str}' | sudo tee {config_file_path}")
 
+    def config(self, service_type: type[ServiceClass]) -> ServiceConfigurationYml:
+        return ServiceConfiguration(self.service(service_type))
+
     def service(self, service_type: type[ServiceClass]) -> ServiceClass:
         """
         Get a service cluster node of specified type.
@@ -118,9 +117,7 @@ class ClusterNode:
         )
 
     def get_list_of_services(self) -> list[str]:
-        return [
+        return [config.attributes[ConfigAttributes.SERVICE_NAME] for config in self.host.config.services]
-            config.attributes[ConfigAttributes.SERVICE_NAME] for config in self.host.config.services
-        ]
 
     def get_all_interfaces(self) -> dict[str, str]:
         return self.host.config.interfaces
@@ -130,9 +127,7 @@ class ClusterNode:
 
     def get_data_interfaces(self) -> list[str]:
         return [
-            ip_address
+            ip_address for name_interface, ip_address in self.host.config.interfaces.items() if "data" in name_interface
-            for name_interface, ip_address in self.host.config.interfaces.items()
-            if "data" in name_interface
         ]
 
     def get_data_interface(self, search_interface: str) -> list[str]:
@@ -221,9 +216,7 @@ class Cluster:
 
         cluster_nodes = set()
         for service in services:
-            cluster_nodes.update(
+            cluster_nodes.update([node for node in self.cluster_nodes if node.service(type(service)) == service])
-                [node for node in self.cluster_nodes if node.service(type(service)) == service]
-            )
 
         return list(cluster_nodes)
 
@@ -331,8 +324,6 @@ class Cluster:
         return [node.get_endpoint() for node in nodes]
 
     def get_nodes_by_ip(self, ips: list[str]) -> list[ClusterNode]:
-        cluster_nodes = [
+        cluster_nodes = [node for node in self.cluster_nodes if URL(node.morph_chain.get_endpoint()).host in ips]
-            node for node in self.cluster_nodes if URL(node.morph_chain.get_endpoint()).host in ips
-        ]
         with reporter.step(f"Return cluster nodes - {cluster_nodes}"):
             return cluster_nodes

src/frostfs_testlib/storage/configuration/interfaces.py

@@ -0,0 +1,65 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class ServiceConfigurationYml(ABC):
+    """
+    Class to manipulate yml configuration for service
+    """
+
+    def _find_option(self, key: str, data: dict):
+        tree = key.split(":")
+        current = data
+        for node in tree:
+            if isinstance(current, list) and len(current) - 1 >= int(node):
+                current = current[int(node)]
+                continue
+
+            if node not in current:
+                return None
+
+            current = current[node]
+
+        return current
+
+    def _set_option(self, key: str, value: Any, data: dict):
+        tree = key.split(":")
+        current = data
+        for node in tree[:-1]:
+            if isinstance(current, list) and len(current) - 1 >= int(node):
+                current = current[int(node)]
+                continue
+
+            if node not in current:
+                current[node] = {}
+
+            current = current[node]
+
+        current[tree[-1]] = value
+
+    @abstractmethod
+    def get(self, key: str) -> str:
+        """
+        Get parameter value from current configuration
+
+        Args:
+            key: key of the parameter in yaml format like 'storage:shard:default:resync_metabase'
+
+        Returns:
+            value of the parameter
+        """
+
+    @abstractmethod
+    def set(self, values: dict[str, Any]):
+        """
+        Sets parameters to configuration
+
+        Args:
+            values: dict where key is the key of the parameter in yaml format like 'storage:shard:default:resync_metabase' and value is the value of the option to set
+        """
+
+    @abstractmethod
+    def revert(self):
+        """
+        Revert changes
+        """

src/frostfs_testlib/storage/configuration/service_configuration.py

@@ -0,0 +1,67 @@
+import os
+import re
+from typing import Any
+
+import yaml
+
+from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.shell.interfaces import CommandOptions
+from frostfs_testlib.storage.configuration.interfaces import ServiceConfigurationYml
+from frostfs_testlib.storage.dataclasses.node_base import ServiceClass
+
+reporter = get_reporter()
+
+
+class ServiceConfiguration(ServiceConfigurationYml):
+    def __init__(self, service: "ServiceClass") -> None:
+        self.service = service
+        self.shell = self.service.host.get_shell()
+        self.confd_path = os.path.join(self.service.config_dir, "conf.d")
+        self.custom_file = os.path.join(self.confd_path, "99_changes.yml")
+
+    def _path_exists(self, path: str) -> bool:
+        return not self.shell.exec(f"test -e {path}", options=CommandOptions(check=False)).return_code
+
+    def _get_data_from_file(self, path: str) -> dict:
+        content = self.shell.exec(f"cat {path}").stdout
+        data = yaml.safe_load(content)
+        return data
+
+    def get(self, key: str) -> str:
+        with reporter.step(f"Get {key} configuration value for {self.service}"):
+            config_files = [self.service.main_config_path]
+
+            if self._path_exists(self.confd_path):
+                files = self.shell.exec(f"find {self.confd_path} -type f").stdout.strip().split()
+                # Sorting files in backwards order from latest to first one
+                config_files.extend(sorted(files, key=lambda x: -int(re.findall("^\d+", os.path.basename(x))[0])))
+
+            result = None
+            for file in files:
+                data = self._get_data_from_file(file)
+                result = self._find_option(key, data)
+                if result is not None:
+                    break
+
+        return result
+
+    def set(self, values: dict[str, Any]):
+        with reporter.step(f"Change configuration for {self.service}"):
+            if not self._path_exists(self.confd_path):
+                self.shell.exec(f"mkdir {self.confd_path}")
+
+            if self._path_exists(self.custom_file):
+                data = self._get_data_from_file(self.custom_file)
+            else:
+                data = {}
+
+            for key, value in values.items():
+                self._set_option(key, value, data)
+
+            content = yaml.dump(data)
+            self.shell.exec(f"echo '{content}' | sudo tee {self.custom_file}")
+            self.shell.exec(f"chmod 777 {self.custom_file}")
+
+    def revert(self):
+        with reporter.step(f"Revert changed options for {self.service}"):
+            self.shell.exec(f"rm -rf {self.custom_file}")

src/frostfs_testlib/storage/constants.py

@@ -3,6 +3,7 @@ class ConfigAttributes:
     WALLET_PASSWORD = "wallet_password"
     WALLET_PATH = "wallet_path"
     WALLET_CONFIG = "wallet_config"
+    CONFIG_DIR = "service_config_dir"
     CONFIG_PATH = "config_path"
     SHARD_CONFIG_PATH = "shard_config_path"
     LOCAL_WALLET_PATH = "local_wallet_path"

src/frostfs_testlib/storage/controllers/background_load_controller.py

@@ -2,13 +2,8 @@ import copy
 from typing import Optional
 
 import frostfs_testlib.resources.optionals as optionals
-from frostfs_testlib.load.interfaces import ScenarioRunner
+from frostfs_testlib.load.interfaces.scenario_runner import ScenarioRunner
-from frostfs_testlib.load.load_config import (
+from frostfs_testlib.load.load_config import EndpointSelectionStrategy, LoadParams, LoadScenario, LoadType
-    EndpointSelectionStrategy,
-    LoadParams,
-    LoadScenario,
-    LoadType,
-)
 from frostfs_testlib.load.load_report import LoadReport
 from frostfs_testlib.load.load_verifiers import LoadVerifier
 from frostfs_testlib.reporter import get_reporter
@@ -56,9 +51,7 @@ class BackgroundLoadController:
             raise RuntimeError("endpoint_selection_strategy should not be None")
 
     @run_optionally(optionals.OPTIONAL_BACKGROUND_LOAD_ENABLED, [])
-    def _get_endpoints(
+    def _get_endpoints(self, load_type: LoadType, endpoint_selection_strategy: EndpointSelectionStrategy):
-        self, load_type: LoadType, endpoint_selection_strategy: EndpointSelectionStrategy
-    ):
         all_endpoints = {
             LoadType.gRPC: {
                 EndpointSelectionStrategy.ALL: list(
@@ -85,10 +78,7 @@ class BackgroundLoadController:
                     )
                 ),
                 EndpointSelectionStrategy.FIRST: list(
-                    set(
+                    set(node_under_load.service(S3Gate).get_endpoint() for node_under_load in self.nodes_under_load)
-                        node_under_load.service(S3Gate).get_endpoint()
-                        for node_under_load in self.nodes_under_load
-                    )
                 ),
             },
         }
@@ -98,12 +88,8 @@ class BackgroundLoadController:
     @run_optionally(optionals.OPTIONAL_BACKGROUND_LOAD_ENABLED)
     @reporter.step_deco("Prepare load instances")
     def prepare(self):
-        self.endpoints = self._get_endpoints(
+        self.endpoints = self._get_endpoints(self.load_params.load_type, self.load_params.endpoint_selection_strategy)
-            self.load_params.load_type, self.load_params.endpoint_selection_strategy
+        self.runner.prepare(self.load_params, self.cluster_nodes, self.nodes_under_load, self.k6_dir)
-        )
-        self.runner.prepare(
-            self.load_params, self.cluster_nodes, self.nodes_under_load, self.k6_dir
-        )
         self.runner.init_k6_instances(self.load_params, self.endpoints, self.k6_dir)
 
     @run_optionally(optionals.OPTIONAL_BACKGROUND_LOAD_ENABLED)

src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -1,9 +1,15 @@
 import datetime
 import time
+from typing import TypeVar
 
 import frostfs_testlib.resources.optionals as optionals
+from frostfs_testlib.cli import FrostfsAdm, FrostfsCli
+from frostfs_testlib.cli.netmap_parser import NetmapParser
 from frostfs_testlib.healthcheck.interfaces import Healthcheck
+from frostfs_testlib.plugins import load_all
 from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.resources.cli import FROSTFS_ADM_CONFIG_PATH, FROSTFS_ADM_EXEC, FROSTFS_CLI_EXEC
+from frostfs_testlib.resources.common import DEFAULT_WALLET_CONFIG, MORPH_BLOCK_TIME
 from frostfs_testlib.shell import CommandOptions, Shell, SshConnectionProvider
 from frostfs_testlib.steps.network import IfUpDownHelper, IpTablesHelper
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, S3Gate, StorageNode
@@ -11,6 +17,7 @@ from frostfs_testlib.storage.controllers.disk_controller import DiskController
 from frostfs_testlib.storage.dataclasses.node_base import NodeBase, ServiceClass
 from frostfs_testlib.testing import parallel
 from frostfs_testlib.testing.test_control import run_optionally, wait_for_success
+from frostfs_testlib.utils.datetime_utils import parse_time
 from frostfs_testlib.utils.failover_utils import (
     wait_all_storage_nodes_returned,
     wait_for_host_offline,
@@ -22,6 +29,14 @@ reporter = get_reporter()
 if_up_down_helper = IfUpDownHelper()
 
 
+class StateManager:
+    def __init__(self, cluster_state_controller: "ClusterStateController") -> None:
+        self.csc = cluster_state_controller
+
+
+StateManagerClass = TypeVar("StateManagerClass", bound=StateManager)
+
+
 class ClusterStateController:
     def __init__(self, shell: Shell, cluster: Cluster, healthcheck: Healthcheck) -> None:
         self.stopped_nodes: list[ClusterNode] = []
@@ -33,6 +48,18 @@ class ClusterStateController:
         self.shell = shell
         self.suspended_services: dict[str, list[ClusterNode]] = {}
         self.nodes_with_modified_interface: list[ClusterNode] = []
+        self.managers: list[StateManagerClass] = []
+
+        # TODO: move all functionality to managers
+        managers = set(load_all(group="frostfs.testlib.csc_managers"))
+        for manager in managers:
+            self.managers.append(manager(self))
+
+    def manager(self, manager_type: type[StateManagerClass]) -> StateManagerClass:
+        for manager in self.managers:
+            # Subclasses here for the future if we have overriding subclasses of base interface
+            if issubclass(type(manager), manager_type):
+                return manager
 
     def _get_stopped_by_node(self, node: ClusterNode) -> set[NodeBase]:
         stopped_by_node = [svc for svc in self.stopped_services if svc.host == node.host]
@@ -82,12 +109,14 @@ class ClusterStateController:
 
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Start host of node {node}")
-    def start_node_host(self, node: ClusterNode):
+    def start_node_host(self, node: ClusterNode, tree_healthcheck: bool = True):
         with reporter.step(f"Start host {node.host.config.address}"):
             node.host.start_host()
             wait_for_host_online(self.shell, node.storage_node)
+            self.stopped_nodes.remove(node)
             wait_for_node_online(node.storage_node)
-        self.stopped_nodes.remove(node)
+            if tree_healthcheck:
+                self.wait_tree_healthcheck()
 
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Start stopped hosts")
@@ -151,7 +180,8 @@ class ClusterStateController:
     @reporter.step_deco("Wait for S3Gates reconnection to local storage")
     def wait_s3gates(self):
         online_s3gates = self._get_online(S3Gate)
-        parallel(self.wait_s3gate, online_s3gates)
+        if online_s3gates:
+            parallel(self.wait_s3gate, online_s3gates)
 
     @wait_for_success(600, 60)
     def wait_tree_healthcheck(self):
@@ -336,7 +366,7 @@ class ClusterStateController:
 
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Hard reboot host {node} via magic SysRq option")
-    def panic_reboot_host(self, node: ClusterNode, wait_for_return: bool = True):
+    def panic_reboot_host(self, node: ClusterNode, wait_for_return: bool = True, tree_healthcheck: bool = True):
         shell = node.host.get_shell()
         shell.exec('sudo sh -c "echo 1 > /proc/sys/kernel/sysrq"')
 
@@ -353,6 +383,8 @@ class ClusterStateController:
             time.sleep(10)
             wait_for_host_online(self.shell, node.storage_node)
             wait_for_node_online(node.storage_node)
+            if tree_healthcheck:
+                self.wait_tree_healthcheck()
 
     @reporter.step_deco("Down {interface} to {nodes}")
     def down_interface(self, nodes: list[ClusterNode], interface: str):
@@ -382,8 +414,8 @@ class ClusterStateController:
     @reporter.step_deco("Set node time to {in_date}")
     def change_node_date(self, node: ClusterNode, in_date: datetime) -> None:
         shell = node.host.get_shell()
-        shell.exec(f"hwclock --set --date='{in_date}'")
+        shell.exec(f"date -s @{time.mktime(in_date.timetuple())}")
-        shell.exec("hwclock --hctosys")
+        shell.exec("hwclock --systohc")
         node_time = self.get_node_date(node)
         with reporter.step(f"Verify difference between {node_time} and {in_date} is less than a minute"):
             assert (self.get_node_date(node) - in_date) < datetime.timedelta(minutes=1)
@@ -393,8 +425,8 @@ class ClusterStateController:
         shell = node.host.get_shell()
         now_time = datetime.datetime.now(datetime.timezone.utc)
         with reporter.step(f"Set {now_time} time"):
-            shell.exec(f"hwclock --set --date='{now_time}'")
+            shell.exec(f"date -s @{time.mktime(now_time.timetuple())}")
-            shell.exec("hwclock --hctosys")
+            shell.exec("hwclock --systohc")
 
     @reporter.step_deco("Change the synchronizer status to {status}")
     def set_sync_date_all_nodes(self, status: str):
@@ -403,6 +435,79 @@ class ClusterStateController:
             return
         parallel(self._disable_date_synchronizer, self.cluster.cluster_nodes)
 
+    @reporter.step_deco("Set MaintenanceModeAllowed - {status}")
+    def set_maintenance_mode_allowed(self, status: str, cluster_node: ClusterNode) -> None:
+        frostfs_adm = FrostfsAdm(
+            shell=cluster_node.host.get_shell(),
+            frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
+            config_file=FROSTFS_ADM_CONFIG_PATH,
+        )
+        frostfs_adm.morph.set_config(set_key_value=f"MaintenanceModeAllowed={status}")
+
+    @reporter.step_deco("Set mode node to {status}")
+    def set_mode_node(self, cluster_node: ClusterNode, wallet: str, status: str, await_tick: bool = True) -> None:
+        rpc_endpoint = cluster_node.storage_node.get_rpc_endpoint()
+        control_endpoint = cluster_node.service(StorageNode).get_control_endpoint()
+
+        frostfs_adm, frostfs_cli, frostfs_cli_remote = self._get_cli(local_shell=self.shell, cluster_node=cluster_node)
+        node_netinfo = NetmapParser.netinfo(frostfs_cli.netmap.netinfo(rpc_endpoint=rpc_endpoint, wallet=wallet).stdout)
+
+        with reporter.step("If status maintenance, then check that the option is enabled"):
+            if node_netinfo.maintenance_mode_allowed == "false":
+                frostfs_adm.morph.set_config(set_key_value="MaintenanceModeAllowed=true")
+
+        with reporter.step(f"Change the status to {status}"):
+            frostfs_cli_remote.control.set_status(endpoint=control_endpoint, status=status)
+
+        if not await_tick:
+            return
+
+        with reporter.step("Tick 1 epoch, and await 2 block"):
+            frostfs_adm.morph.force_new_epoch()
+            time.sleep(parse_time(MORPH_BLOCK_TIME) * 2)
+
+        self.check_node_status(status=status, wallet=wallet, cluster_node=cluster_node)
+
+    @wait_for_success(80, 8)
+    @reporter.step_deco("Check status node, status - {status}")
+    def check_node_status(self, status: str, wallet: str, cluster_node: ClusterNode):
+        frostfs_cli = FrostfsCli(
+            shell=self.shell, frostfs_cli_exec_path=FROSTFS_CLI_EXEC, config_file=DEFAULT_WALLET_CONFIG
+        )
+        netmap = NetmapParser.snapshot_all_nodes(
+            frostfs_cli.netmap.snapshot(rpc_endpoint=cluster_node.storage_node.get_rpc_endpoint(), wallet=wallet).stdout
+        )
+        netmap = [node for node in netmap if cluster_node.host_ip == node.node]
+        if status == "offline":
+            assert cluster_node.host_ip not in netmap, f"{cluster_node.host_ip} not in Offline"
+        else:
+            assert netmap[0].node_status == status.upper(), f"Node state - {netmap[0].node_status} != {status} expect"
+
+    def _get_cli(self, local_shell: Shell, cluster_node: ClusterNode) -> tuple[FrostfsAdm, FrostfsCli, FrostfsCli]:
+        # TODO Move to service config
+        host = cluster_node.host
+        service_config = host.get_service_config(cluster_node.storage_node.name)
+        wallet_path = service_config.attributes["wallet_path"]
+        wallet_password = service_config.attributes["wallet_password"]
+
+        shell = host.get_shell()
+        wallet_config_path = f"/tmp/{cluster_node.storage_node.name}-config.yaml"
+        wallet_config = f'wallet: {wallet_path}\npassword: "{wallet_password}"'
+        shell.exec(f"echo '{wallet_config}' > {wallet_config_path}")
+
+        frostfs_adm = FrostfsAdm(
+            shell=shell, frostfs_adm_exec_path=FROSTFS_ADM_EXEC, config_file=FROSTFS_ADM_CONFIG_PATH
+        )
+        frostfs_cli = FrostfsCli(
+            shell=local_shell, frostfs_cli_exec_path=FROSTFS_CLI_EXEC, config_file=DEFAULT_WALLET_CONFIG
+        )
+        frostfs_cli_remote = FrostfsCli(
+            shell=shell,
+            frostfs_cli_exec_path=FROSTFS_CLI_EXEC,
+            config_file=wallet_config_path,
+        )
+        return frostfs_adm, frostfs_cli, frostfs_cli_remote
+
     def _enable_date_synchronizer(self, cluster_node: ClusterNode):
         shell = cluster_node.host.get_shell()
         shell.exec("timedatectl set-ntp true")

src/frostfs_testlib/storage/controllers/state_managers/config_state_manager.py

@@ -0,0 +1,51 @@
+from typing import Any
+
+from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.storage.cluster import ClusterNode
+from frostfs_testlib.storage.controllers.cluster_state_controller import ClusterStateController, StateManager
+from frostfs_testlib.storage.dataclasses.node_base import ServiceClass
+from frostfs_testlib.testing import parallel
+
+reporter = get_reporter()
+
+
+class ConfigStateManager(StateManager):
+    def __init__(self, cluster_state_controller: ClusterStateController) -> None:
+        super().__init__(cluster_state_controller)
+        self.services_with_changed_config: set[tuple[ClusterNode, ServiceClass]] = set()
+        self.cluster = self.csc.cluster
+
+    @reporter.step_deco("Change configuration for {service_type} on all nodes")
+    def set_on_all_nodes(self, service_type: type[ServiceClass], values: dict[str, Any]):
+        services = self.cluster.services(service_type)
+        nodes = self.cluster.nodes(services)
+        self.services_with_changed_config.update([(node, service_type) for node in nodes])
+
+        self.csc.stop_services_of_type(service_type)
+        parallel([node.config(service_type).set for node in nodes], values=values)
+        self.csc.start_services_of_type(service_type)
+
+    @reporter.step_deco("Change configuration for {service_type} on {node}")
+    def set_on_node(self, node: ClusterNode, service_type: type[ServiceClass], values: dict[str, Any]):
+        self.services_with_changed_config.add((node, service_type))
+
+        self.csc.stop_service_of_type(node, service_type)
+        node.config(service_type).set(values)
+        self.csc.start_service_of_type(node, service_type)
+
+    @reporter.step_deco("Revert all configuration changes")
+    def revert_all(self):
+        if not self.services_with_changed_config:
+            return
+
+        parallel(self._revert_svc, self.services_with_changed_config)
+        self.services_with_changed_config.clear()
+
+        self.csc.start_all_stopped_services()
+
+    # TODO: parallel can't have multiple parallel_items :(
+    @reporter.step_deco("Revert all configuration {node_and_service}")
+    def _revert_svc(self, node_and_service: tuple[ClusterNode, ServiceClass]):
+        node, service_type = node_and_service
+        self.csc.stop_service_of_type(node, service_type)
+        node.config(service_type).revert()

src/frostfs_testlib/storage/dataclasses/node_base.py

@@ -120,6 +120,15 @@ class NodeBase(HumanReadableABC):
             ConfigAttributes.WALLET_CONFIG,
         )
 
+    @property
+    def config_dir(self) -> str:
+        return self._get_attribute(ConfigAttributes.CONFIG_DIR)
+
+    @property
+    def main_config_path(self) -> str:
+        return self._get_attribute(ConfigAttributes.CONFIG_PATH)
+
+    # TODO: Deprecated
     def get_config(self, config_file_path: Optional[str] = None) -> tuple[str, dict]:
         if config_file_path is None:
             config_file_path = self._get_attribute(ConfigAttributes.CONFIG_PATH)
@@ -132,6 +141,7 @@ class NodeBase(HumanReadableABC):
         config = yaml.safe_load(config_text)
         return config_file_path, config
 
+    # TODO: Deprecated
     def save_config(self, new_config: dict, config_file_path: Optional[str] = None) -> None:
         if config_file_path is None:
             config_file_path = self._get_attribute(ConfigAttributes.CONFIG_PATH)
@@ -146,9 +156,7 @@ class NodeBase(HumanReadableABC):
         storage_wallet_pass = self.get_wallet_password()
         return wallet_utils.get_wallet_public_key(storage_wallet_path, storage_wallet_pass)
 
-    def _get_attribute(
+    def _get_attribute(self, attribute_name: str, default_attribute_name: Optional[str] = None) -> str:
-        self, attribute_name: str, default_attribute_name: Optional[str] = None
-    ) -> str:
         config = self.host.get_service_config(self.name)
 
         if attribute_name not in config.attributes:

src/frostfs_testlib/storage/dataclasses/storage_object_info.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass
-from enum import Enum
 from typing import Optional
 
 from frostfs_testlib.testing.readable import HumanReadableEnum
@@ -28,10 +27,16 @@ class StorageObjectInfo(ObjectRef):
     locks: Optional[list[LockObjectInfo]] = None
 
 
+class ModeNode(HumanReadableEnum):
+    MAINTENANCE: str = "maintenance"
+    ONLINE: str = "online"
+    OFFLINE: str = "offline"
+
+
 @dataclass
 class NodeNetmapInfo:
     node_id: str = None
-    node_status: str = None
+    node_status: ModeNode = None
     node_data_ips: list[str] = None
     cluster_name: str = None
     continent: str = None
@@ -53,3 +58,19 @@ class Interfaces(HumanReadableEnum):
     MGMT: str = "mgmt"
     INTERNAL_0: str = "internal0"
     INTERNAL_1: str = "internal1"
+
+
+@dataclass
+class NodeNetInfo:
+    epoch: str = None
+    network_magic: str = None
+    time_per_block: str = None
+    container_fee: str = None
+    epoch_duration: str = None
+    inner_ring_candidate_fee: str = None
+    maximum_object_size: str = None
+    withdrawal_fee: str = None
+    homomorphic_hashing_disabled: str = None
+    maintenance_mode_allowed: str = None
+    eigen_trust_alpha: str = None
+    eigen_trust_iterations: str = None

src/frostfs_testlib/utils/failover_utils.py

@@ -12,6 +12,7 @@ from frostfs_testlib.steps.node_management import storage_node_healthcheck
 from frostfs_testlib.steps.storage_policy import get_nodes_with_object
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, NodeBase, StorageNode
 from frostfs_testlib.storage.dataclasses.frostfs_services import MorphChain
+from frostfs_testlib.testing.parallel import parallel
 from frostfs_testlib.testing.test_control import retry, wait_for_success
 from frostfs_testlib.utils.datetime_utils import parse_time
 
@@ -26,12 +27,17 @@ def ping_host(shell: Shell, host: Host):
     return shell.exec(f"ping {host.config.address} -c 1", options).return_code
 
 
+# TODO: Move to ClusterStateController
 @reporter.step_deco("Wait for storage nodes returned to cluster")
 def wait_all_storage_nodes_returned(shell: Shell, cluster: Cluster) -> None:
-    for node in cluster.services(StorageNode):
+    nodes = cluster.services(StorageNode)
-        with reporter.step(f"Run health check for storage at '{node}'"):
+    parallel(_wait_for_storage_node, nodes, shell=shell)
-            wait_for_host_online(shell, node)
+
-            wait_for_node_online(node)
+
+@reporter.step_deco("Run health check for storage at '{node}'")
+def _wait_for_storage_node(node: StorageNode, shell: Shell) -> None:
+    wait_for_host_online(shell, node)
+    wait_for_node_online(node)
 
 
 @retry(max_attempts=60, sleep_interval=5, expected_result=0)
@@ -64,10 +70,17 @@ def wait_for_node_online(node: StorageNode):
     except Exception as err:
         logger.warning(f"Node healthcheck fails with error {err}")
         return False
+    finally:
+        gather_socket_info(node)
 
     return health_check.health_status == "READY" and health_check.network_status == "ONLINE"
 
 
+@reporter.step_deco("Gather socket info for {node}")
+def gather_socket_info(node: StorageNode):
+    node.host.get_shell().exec("ss -tuln | grep 8080", CommandOptions(check=False))
+
+
 @reporter.step_deco("Check and return  status of given service")
 def service_status(service: str, shell: Shell) -> str:
     return shell.exec(f"sudo systemctl is-active {service}").stdout.rstrip()
@@ -139,9 +152,7 @@ def multiple_restart(
     service_name = node.service(service_type).name
     for _ in range(count):
         node.host.restart_service(service_name)
-        logger.info(
+        logger.info(f"Restart {service_systemctl_name}; sleep {sleep_interval} seconds and continue")
-            f"Restart {service_systemctl_name}; sleep {sleep_interval} seconds and continue"
-        )
         sleep(sleep_interval)
 
 
@@ -164,9 +175,7 @@ def check_services_status(service_list: list[str], expected_status: str, shell:
 
 @reporter.step_deco("Wait for active status of passed service")
 @wait_for_success(60, 5)
-def wait_service_in_desired_state(
+def wait_service_in_desired_state(service: str, shell: Shell, expected_status: Optional[str] = "active"):
-    service: str, shell: Shell, expected_status: Optional[str] = "active"
-):
     real_status = service_status(service=service, shell=shell)
     assert (
         expected_status == real_status