src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -1,16 +1,15 @@
 import copy
-import itertools
 import time
 
 import frostfs_testlib.resources.optionals as optionals
 from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.shell import CommandOptions, Shell
-from frostfs_testlib.steps import epoch
 from frostfs_testlib.steps.iptables import IpTablesHelper
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, StorageNode
 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.testing.test_control import run_optionally
 from frostfs_testlib.utils.failover_utils import (
     wait_all_storage_nodes_returned,
     wait_for_host_offline,
@@ -28,6 +27,7 @@ class ClusterStateController:
         self.stopped_storage_nodes: list[ClusterNode] = []
         self.stopped_s3_gates: list[ClusterNode] = []
         self.dropped_traffic: list[ClusterNode] = []
+        self.stopped_services: set[NodeBase] = set()
         self.cluster = cluster
         self.shell = shell
         self.suspended_services: dict[str, list[ClusterNode]] = {}
@@ -128,6 +128,51 @@ class ClusterStateController:
         node.storage_node.stop_service()
         self.stopped_storage_nodes.append(node)
 
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Stop all {service_type} services")
+    def stop_services_of_type(self, service_type: type[ServiceClass]):
+        services = self.cluster.services(service_type)
+        self.stopped_services.update(services)
+        parallel([service.stop_service for service in services])
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Start all {service_type} services")
+    def start_services_of_type(self, service_type: type[ServiceClass]):
+        services = self.cluster.services(service_type)
+        parallel([service.start_service for service in services])
+
+        if service_type == StorageNode:
+            wait_all_storage_nodes_returned(self.shell, self.cluster)
+
+        self.stopped_services = self.stopped_services - set(services)
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Start all stopped services")
+    def start_all_stopped_services(self):
+        parallel([service.start_service for service in self.stopped_services])
+
+        for service in self.stopped_services:
+            if isinstance(service, StorageNode):
+                wait_all_storage_nodes_returned(self.shell, self.cluster)
+                break
+
+        self.stopped_services.clear()
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Stop {service_type} service on {node}")
+    def stop_service_of_type(self, node: ClusterNode, service_type: type[ServiceClass]):
+        service = node.service(service_type)
+        service.stop_service()
+        self.stopped_services.add(service)
+
+    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
+    @reporter.step_deco("Start {service_type} service on {node}")
+    def start_service_of_type(self, node: ClusterNode, service_type: type[ServiceClass]):
+        service = node.service(service_type)
+        service.start_service()
+        if service in self.stopped_services:
+            self.stopped_services.remove(service)
+
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step_deco("Start storage service on {node}")
     def start_storage_service(self, node: ClusterNode):

src/frostfs_testlib/storage/dataclasses/node_base.py

@@ -1,4 +1,4 @@
-from abc import ABC, abstractmethod
+from abc import abstractmethod
 from dataclasses import dataclass
 from typing import Optional, Tuple, TypedDict, TypeVar
 
@@ -6,10 +6,13 @@ import yaml
 
 from frostfs_testlib.hosting.config import ServiceConfig
 from frostfs_testlib.hosting.interfaces import Host
+from frostfs_testlib.reporter import get_reporter
 from frostfs_testlib.storage.constants import ConfigAttributes
 from frostfs_testlib.testing.readable import HumanReadableABC
 from frostfs_testlib.utils import wallet_utils
 
+reporter = get_reporter()
+
 
 @dataclass
 class NodeBase(HumanReadableABC):
@@ -54,17 +57,20 @@ class NodeBase(HumanReadableABC):
         return self._process_name
 
     def start_service(self):
-        self.host.start_service(self.name)
+        with reporter.step(f"Start {self.name} service on {self.host.config.address}"):
+            self.host.start_service(self.name)
 
     @abstractmethod
     def service_healthcheck(self) -> bool:
         """Service healthcheck."""
 
     def stop_service(self):
-        self.host.stop_service(self.name)
+        with reporter.step(f"Stop {self.name} service on {self.host.config.address}"):
+            self.host.stop_service(self.name)
 
     def restart_service(self):
-        self.host.restart_service(self.name)
+        with reporter.step(f"Restart {self.name} service on {self.host.config.address}"):
+            self.host.restart_service(self.name)
 
     def get_wallet_password(self) -> str:
         return self._get_attribute(ConfigAttributes.WALLET_PASSWORD)