Compare commits

...

7 commits

Author SHA1 Message Date
9ab4def44f Store k6 output and add socket info collection
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
2023-11-20 17:39:15 +03:00
ed8f90dfc0 Change output time format to unix
Signed-off-by: Dmitriy Zayakin <d.zayakin@yadro.com>
2023-11-20 15:53:30 +03:00
ed70dada96 Add support test maintenance
Signed-off-by: Dmitriy Zayakin <d.zayakin@yadro.com>
2023-11-20 11:07:24 +00:00
22647c6d59 [#119] Renamed Github to Gitea in links
Some links changed to git.frostfs from github

Signed-off-by: Mikhail Kadilov <m.kadilov@yadro.com>
2023-11-15 13:08:58 +03:00
61a1b28652 s3local.js scenario
Signed-off-by: m.malygina <m.malygina@yadro.com>
2023-11-14 14:00:08 +03:00
6519cfafc9 [#116] Updates for local scenario teardown
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
2023-11-13 13:34:49 +03:00
72bd467c53 [#114] Add yaml configuration controllers
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
2023-11-10 19:46:18 +00:00
31 changed files with 795 additions and 275 deletions

View file

@ -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
[pull requests](https://github.com/TrueCloudLab/frostfs-testlib/pulls) for existing
- Check the open [issues](https://git.frostfs.info/TrueCloudLab/frostfs-testlib/issues) and
[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
Fork [FrostFS testlib upstream](https://github.com/TrueCloudLab/frostfs-testlib/fork) source
### Set up your Git Repository
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
document](https://help.github.com/articles/creating-a-pull-request/) for
Pull requests can be created via Git. Refer to [this
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.

View file

@ -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).

View file

@ -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"]

View file

@ -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"]
},
)
)

View file

@ -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.

View file

@ -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)

View file

@ -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"]},
)

View file

@ -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]

View file

@ -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

View file

@ -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"""

View file

@ -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"""

View file

@ -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.load_config import (
K6ProcessAllocationStrategy,
LoadParams,
LoadScenario,
LoadType,
)
from frostfs_testlib.load.interfaces.loader import Loader
from frostfs_testlib.load.load_config import 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 (
result.return_code == EXIT_RESULT_CODE
), f"Return code of preset is not zero: {result.stdout}"
return result.stdout.strip("\n")
assert result.return_code == EXIT_RESULT_CODE, f"Return code of preset is not zero: {result.stdout}"
self.preset_output = 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(
"\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]
)
reporter.attach("\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])
def start(self) -> None:
with reporter.step(
f"Start load from loader {self.loader.ip} on endpoints {self.endpoints}"
):
with reporter.step(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(
command, self.shell, self.load_params.working_dir, user
)
self._k6_process = RemoteProcess.create(command, self.shell, self.load_params.working_dir, user)
def wait_until_finished(self, soft_timeout: int = 0) -> None:
with reporter.step(
f"Wait until load is finished from loader {self.loader.ip} on endpoints {self.endpoints}"
):
with reporter.step(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_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...")
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 ""
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(
f"Get load results from loader {self.loader.ip} on endpoints {self.endpoints}"
):
with reporter.step(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(
K6_STOP_SIGNAL_TIMEOUT, 15, False, False, "Can not stop K6 process within timeout"
)
@wait_for_success(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()

View file

@ -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

View file

@ -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,
}

View file

@ -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]

View file

@ -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,

View file

@ -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 (
BACKGROUND_LOAD_VUS_COUNT_DIVISOR,
LOAD_NODE_SSH_USER,
LOAD_NODES,
)
from frostfs_testlib.resources.load_params import 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 = [
node.service(S3Gate).get_wallet_public_key() for node in cluster_nodes
]
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_loader, self.loaders, load_params, grpc_peer, s3_public_keys, k6_dir
)
parallel(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(
r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output
).group("aws_access_key_id")
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(
@ -125,9 +121,7 @@ class DefaultRunner(RunnerBase):
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"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(
[[endpoint] for endpoint in endpoints]
),
K6ProcessAllocationStrategy.PER_ENDPOINT: itertools.cycle([[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(
load_params, k6_processes_count
)
distributed_load_params_list = self._get_distributed_load_params_list(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(
self, load_params_for_loader: LoadParams, loader: Loader, endpoints: list[str], k6_dir: str
):
def _init_k6_instance(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(
math.ceil(original_value / divisor), workers_count
)
distribution = self._get_distribution(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 = [
clients_per_worker + 1 if i < remainder else clients_per_worker
for i in range(workers_count)
]
distribution = [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(
f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"
):
with reporter.step(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 = (
f"{STORAGE_USER_NAME} hard nofile 65536\n{STORAGE_USER_NAME} soft nofile 65536\n"
)
content = 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(
f"{k6_dir}/scenarios/files/wallet.json", "", "/tmp/empty_passwd.yml"
)
self.wallet = WalletInfo(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_all_storage_services()
self.cluster_state_controller.stop_services_of_type(S3Gate)
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(
f"Wait for start timeout + couple more seconds ({wait_after_start_time}) before moving on"
):
with reporter.step(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_stopped_s3_gates()
self.cluster_state_controller.start_all_stopped_services()
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))

View file

@ -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]

View file

@ -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"return code: {exc.returncode}\n"
f"output: {exc.output}"
f"Command: {command}\nError:\n" 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

View file

@ -6,27 +6,11 @@ from functools import lru_cache, wraps
from time import sleep
from typing import ClassVar, Optional, Tuple
from paramiko import (
AutoAddPolicy,
Channel,
ECDSAKey,
Ed25519Key,
PKey,
RSAKey,
SSHClient,
SSHException,
ssh_exception,
)
from paramiko import 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 (
CommandInspector,
CommandOptions,
CommandResult,
Shell,
SshCredentials,
)
from frostfs_testlib.shell.interfaces import 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"(attempt {attempt})"
f"Trying to connect to host {host} as {creds.ssh_login} using password " f"(attempt {attempt})"
)
connection.connect(
hostname=host,
@ -141,9 +124,7 @@ class HostIsNotAvailable(Exception):
def log_command(func):
@wraps(func)
def wrapper(
shell: "SSHShell", command: str, options: CommandOptions, *args, **kwargs
) -> CommandResult:
def wrapper(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(
command, timeout=options.timeout, get_pty=True
)
stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout, get_pty=True)
for interactive_input in options.interactive_inputs:
input = interactive_input.input
if not input.endswith("\n"):

View file

@ -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)

View file

@ -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 (
HTTPGate,
InnerRing,
MorphChain,
S3Gate,
StorageNode,
)
from frostfs_testlib.storage.dataclasses.frostfs_services import 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 [
config.attributes[ConfigAttributes.SERVICE_NAME] for config in self.host.config.services
]
return [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
for name_interface, ip_address in self.host.config.interfaces.items()
if "data" in name_interface
ip_address 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(
[node for node in self.cluster_nodes if node.service(type(service)) == service]
)
cluster_nodes.update([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 = [
node for node in self.cluster_nodes if URL(node.morph_chain.get_endpoint()).host in ips
]
cluster_nodes = [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

View file

@ -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
"""

View file

@ -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}")

View file

@ -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"

View file

@ -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.load_config import (
EndpointSelectionStrategy,
LoadParams,
LoadScenario,
LoadType,
)
from frostfs_testlib.load.interfaces.scenario_runner import ScenarioRunner
from frostfs_testlib.load.load_config import 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(
self, load_type: LoadType, endpoint_selection_strategy: EndpointSelectionStrategy
):
def _get_endpoints(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(
node_under_load.service(S3Gate).get_endpoint()
for node_under_load in self.nodes_under_load
)
set(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.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.endpoints = self._get_endpoints(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.init_k6_instances(self.load_params, self.endpoints, self.k6_dir)
@run_optionally(optionals.OPTIONAL_BACKGROUND_LOAD_ENABLED)

View file

@ -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("hwclock --hctosys")
shell.exec(f"date -s @{time.mktime(in_date.timetuple())}")
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("hwclock --hctosys")
shell.exec(f"date -s @{time.mktime(now_time.timetuple())}")
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")

View file

@ -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()

View file

@ -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(
self, attribute_name: str, default_attribute_name: Optional[str] = None
) -> str:
def _get_attribute(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:

View file

@ -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

View file

@ -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):
with reporter.step(f"Run health check for storage at '{node}'"):
wait_for_host_online(shell, node)
wait_for_node_online(node)
nodes = cluster.services(StorageNode)
parallel(_wait_for_storage_node, nodes, shell=shell)
@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(
f"Restart {service_systemctl_name}; sleep {sleep_interval} seconds and continue"
)
logger.info(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(
service: str, shell: Shell, expected_status: Optional[str] = "active"
):
def wait_service_in_desired_state(service: str, shell: Shell, expected_status: Optional[str] = "active"):
real_status = service_status(service=service, shell=shell)
assert (
expected_status == real_status