diff --git a/src/imgtests/database/database.py b/src/imgtests/database/database.py index 63c54645..bc1d6146 100644 --- a/src/imgtests/database/database.py +++ b/src/imgtests/database/database.py @@ -1,5 +1,8 @@ import logging +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass, is_dataclass from datetime import datetime +from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, get_args from zoneinfo import ZoneInfo @@ -21,6 +24,29 @@ logger = logging.getLogger(__name__) Table = Literal["configurations", "experiments", "util_run_result"] ExperimentType = Literal["performance", "endurance", "all"] +CommandValue = str | Sequence[str] + + +@dataclass(frozen=True) +class UtilityMetricRecord: + metric_name: str + value: float + context: dict[str, Any] | None = None + description: str | None = None + command: CommandValue | None = None + + +@dataclass(frozen=True) +class UtilityResultRecord: + experiment_id: int + utility: str + command: CommandValue + result: Any + started_at: datetime + ended_at: datetime + description: str | None = None + context: dict[str, Any] | None = None + metrics: tuple[UtilityMetricRecord, ...] = () class PostgresCreds(BaseSettings): @@ -41,7 +67,10 @@ def __init__(self, database: str = "postgres") -> None: def initialize_postgres(self, creds: PostgresCreds) -> None: self.engine = create_engine( - f"postgresql+psycopg://{creds.user}:{creds.password}@{creds.host}:{creds.port}/{creds.database_name}", + ( + f"postgresql+psycopg://{creds.user}:{creds.password}" + f"@{creds.host}:{creds.port}/{creds.database_name}" + ), ) self.session = sessionmaker(self.engine) Base.metadata.create_all(self.engine) @@ -125,8 +154,8 @@ def insert_experiment( experiment_object = ExperimentBase( config_id=config_id, - description=description, - type=experiment_type, + description=_validate_db_str(description), + type=_validate_db_str(experiment_type), started_at=started_at, ended_at=ended_at, ) @@ -157,9 +186,9 @@ def insert_util_run_result( # noqa: PLR0913 util_run_result = UtilRunResult( experiment_id=experiment_id, util_type=util_type, - command=command, - result=result, - description=description, + command=_validate_db_str(command), + result=_coerce_db_payload(result), + description=_validate_db_str(description), started_at=started_at, ended_at=ended_at, ) @@ -171,6 +200,75 @@ def insert_util_run_result( # noqa: PLR0913 session.refresh(util_run_result) return util_run_result + def insert_metric_observation( + self, + experiment_id: int, + utility: str, + metric: UtilityMetricRecord, + started_at: datetime, + ended_at: datetime, + ) -> UtilRunResult: + payload: dict[str, Any] = { + "utility": utility, + "metric_name": metric.metric_name, + "value": float(metric.value), + } + if metric.context: + payload.update(_normalize_db_mapping(metric.context)) + + if metric.command is not None: + payload["command"] = _normalize_command_json(metric.command) + + command_label = ( + _command_db_label(metric.command, fallback=utility) + if metric.command is not None + else f"{utility}:{metric.metric_name}" + ) + + return self.insert_util_run_result( + experiment_id=experiment_id, + util_type="observer", + command=command_label, + result=payload, + description=metric.description or "Observed numeric metric", + started_at=started_at, + ended_at=ended_at, + ) + + def insert_utility_result( + self, + record: UtilityResultRecord, + ) -> tuple[UtilRunResult, tuple[UtilRunResult, ...]]: + result_payload = _coerce_db_payload(record.result) + result_payload.setdefault("utility", record.utility) + result_payload.setdefault("command", _normalize_command_json(record.command)) + + if record.context: + for key, value in _normalize_db_mapping(record.context).items(): + result_payload.setdefault(key, value) + + loader = self.insert_util_run_result( + experiment_id=record.experiment_id, + util_type="loader", + command=_command_db_label(record.command, fallback=record.utility), + result=result_payload, + description=record.description or f"{record.utility} result", + started_at=record.started_at, + ended_at=record.ended_at, + ) + + observers = tuple( + self.insert_metric_observation( + experiment_id=record.experiment_id, + utility=record.utility, + metric=metric, + started_at=record.started_at, + ended_at=record.ended_at, + ) + for metric in record.metrics + ) + return loader, observers + def update_experiment_ended_at( self, experiment_id: int, @@ -221,3 +319,64 @@ def _check_session(self) -> None: if not hasattr(self, "session") or self.session is None: error_message = "Database session not initialized." raise RuntimeError(error_message) + + +def _validate_db_str(value: str, limit: int = 200) -> str: + s = str(value) + if len(s) > limit: + err_msg = f"Value is too long for DB field: {len(s)} > {limit}. Text: {s}" + raise ValueError(err_msg) + return s + + +def _normalize_command_json(command: CommandValue) -> str | list[str]: + if isinstance(command, str): + return command + return [str(part) for part in command] + + +def _command_db_label(command: CommandValue, fallback: str) -> str: + if isinstance(command, str): + parts = command.strip().split() + return parts[0] if parts else fallback + + for part in command: + text = str(part).strip() + if text: + return text + + return fallback + + +def _normalize_db_mapping(mapping: Mapping[str, Any]) -> dict[str, Any]: + return {str(key): _normalize_db_value(value) for key, value in mapping.items()} + + +def _coerce_db_payload(value: Any) -> dict[str, Any]: + normalized = _normalize_db_value(value) + if isinstance(normalized, dict): + return normalized + if isinstance(normalized, list): + return {"items": normalized} + return {"value": normalized} + + +def _normalize_db_value(value: Any) -> Any: + normalized: Any + + if value is None or isinstance(value, (str, int, float, bool)): + normalized = value + elif isinstance(value, Path): + normalized = str(value) + elif isinstance(value, dict): + normalized = {str(key): _normalize_db_value(val) for key, val in value.items()} + elif is_dataclass(value): + normalized = _normalize_db_value(asdict(value)) + elif hasattr(value, "_asdict"): + normalized = _normalize_db_value(value._asdict()) + elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + normalized = [_normalize_db_value(item) for item in value] + else: + normalized = str(value) + + return normalized diff --git a/src/imgtests/exec/user_commands.py b/src/imgtests/exec/user_commands.py index 9434150c..437bce36 100644 --- a/src/imgtests/exec/user_commands.py +++ b/src/imgtests/exec/user_commands.py @@ -1,11 +1,11 @@ from typing import TYPE_CHECKING, NamedTuple +from imgtests.exec.base_util import GenericUtil from imgtests.exec.exec import ExecResult from imgtests.exec.pkgmgrs.mixin import PkgMgrMixin if TYPE_CHECKING: from imgtests.exec.exec import SSHClient -from imgtests.exec.base_util import GenericUtil class MkDir(GenericUtil): diff --git a/src/imgtests/planning/executor.py b/src/imgtests/planning/executor.py index 9b4ea386..f416636e 100644 --- a/src/imgtests/planning/executor.py +++ b/src/imgtests/planning/executor.py @@ -1,15 +1,15 @@ from __future__ import annotations import json -import logging import math import time from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import suppress -from dataclasses import asdict, dataclass +from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING, Any +from imgtests.database.database import UtilityMetricRecord, UtilityResultRecord from imgtests.exec.exec import common_run_command from imgtests.exec.loaders.fio import Fio, fio_metrics_to_samples, get_available_bytes from imgtests.exec.loaders.stress_ng import StressNg, stress_metrics_to_samples @@ -27,8 +27,6 @@ from imgtests.exec.exec import SSHClient from imgtests.planning.models import LoadTask, PlanStage, TestPlan -logger = logging.getLogger(__name__) - _MIN_FIO_SIZE_BYTES = 64 * 1024**2 _MAX_FIO_SIZE_RATIO = 0.25 @@ -70,6 +68,7 @@ def __init__( client: SSHClient | None, db: ImgtestsDatabase, ) -> None: + super().__init__("plan_executor", client, db) self.client = client self.db = db self._cpu_count_cache: int | None = None @@ -132,38 +131,12 @@ def execute( stage_ended_at = datetime.now(UTC) for task_run in task_runs: - full_cmd = " ".join(task_run.command) if task_run.command else "" - subsystem_value = getattr( - task_run.task.subsystem, - "value", - str(task_run.task.subsystem), - ) + self._insert_task_run(experiment_id, stage.name, task_run) collected_metrics.extend(task_run.metrics) counts[task_run.status] += 1 total_count += 1 - self.db.insert_util_run_result( - experiment_id=experiment_id, - util_type="loader", - command=full_cmd, - result={ - "stage_name": stage.name, - "subsystem": subsystem_value, - "tool": task_run.task.tool, - "utility": task_run.task.tool, - "command": list(task_run.command), - "returncode": task_run.returncode, - "stdout": task_run.stdout, - "stderr": task_run.stderr, - "summary": task_run.summary, - "metrics": [asdict(sample) for sample in task_run.metrics], - }, - description=f"Task result for stage={stage.name}", - started_at=task_run.started_at, - ended_at=task_run.ended_at, - ) - stage_runs.append( StageRunResult( stage_name=stage.name, @@ -227,7 +200,7 @@ def _run_stage(self, stage: PlanStage) -> list[TaskRunResult]: results_by_idx[idx] = future.result() except Exception as exc: now = datetime.now(UTC) - logger.exception("Task failed with exception.") + self._logger.exception("Task failed with exception.") results_by_idx[idx] = TaskRunResult( task=task, started_at=now, @@ -243,12 +216,64 @@ def _run_stage(self, stage: PlanStage) -> list[TaskRunResult]: return [results_by_idx[i] for i in range(len(stage.tasks))] + def _insert_task_run( + self, + experiment_id: int, + stage_name: str, + task_run: TaskRunResult, + ) -> None: + subsystem_value = getattr(task_run.task.subsystem, "value", str(task_run.task.subsystem)) + command = task_run.command or (task_run.task.tool,) + + metrics = tuple( + UtilityMetricRecord( + metric_name=sample.metric_name, + value=sample.value, + context={ + "stage_name": sample.stage_name, + "subsystem": sample.subsystem, + "tool": task_run.task.tool, + "label": sample.label, + }, + description="Observed numeric metric", + command=command, + ) + for sample in task_run.metrics + ) + + self.db.insert_utility_result( + UtilityResultRecord( + experiment_id=experiment_id, + utility=task_run.task.tool, + command=command, + result={ + "stage_name": stage_name, + "subsystem": subsystem_value, + "tool": task_run.task.tool, + "returncode": task_run.returncode, + "stdout": _truncate(task_run.stdout), + "stderr": _truncate(task_run.stderr), + "summary": task_run.summary, + "status": task_run.status.value, + }, + description=f"Task result for stage={stage_name}", + started_at=task_run.started_at, + ended_at=task_run.ended_at, + context={ + "stage_name": stage_name, + "subsystem": subsystem_value, + "tool": task_run.task.tool, + }, + metrics=metrics, + ), + ) + def _run_task(self, stage: PlanStage, task: LoadTask) -> TaskRunResult: tool = (task.tool or "").strip().lower().replace("_", "-") started_at = datetime.now(UTC) subsystem_value = getattr(task.subsystem, "value", str(task.subsystem)) - logger.info( + self._logger.info( "[PLAN] run stage=%s tool=%s subsystem=%s dur=%ss", stage.name, tool, @@ -291,7 +316,7 @@ def _retry_stress_run_if_needed( retry_timeout = max(5, int(timeout_sec * 0.5)) - logger.info( + self._logger.info( "[PLAN] stress-ng retry stage=%s rc=%s timeout=%s->%s", stage_name, exec_res.returncode, @@ -337,7 +362,7 @@ def _run_stress_ng( samples = stress_metrics_to_samples(stage.name, subsystem, metrics) summary_dict = summary._asdict() if summary else None - logger.info( + self._logger.info( "[PLAN] done stage=%s tool=stress-ng rc=%s", stage.name, exec_res.returncode, @@ -422,7 +447,7 @@ def _run_fio( if payload: summary = {"jobs_count": len(payload.get("jobs", []))} - logger.info("[PLAN] done stage=%s tool=fio rc=%s", stage.name, result.returncode) + self._logger.info("[PLAN] done stage=%s tool=fio rc=%s", stage.name, result.returncode) return TaskRunResult( task=task, started_at=started_at, @@ -601,6 +626,13 @@ def _parse_dynamic_float(raw_value: str, prefix: str) -> float: return value +def _truncate(text: str, limit: int = 4096) -> str: + raw = str(text or "") + if len(raw) <= limit: + return raw + return f"{raw[:limit]}... [truncated {len(raw) - limit} chars]" + + def _resolve_experiment_type(plan: TestPlan) -> ExperimentType: test_kind = getattr(plan.test_kind, "value", str(plan.test_kind)) diff --git a/src/imgtests/runner.py b/src/imgtests/runner.py index db586f04..1a8c890a 100644 --- a/src/imgtests/runner.py +++ b/src/imgtests/runner.py @@ -223,6 +223,10 @@ class ProfiledPlanRunnerSettings(BaseSettings): duration_volume: int | None = Field(default=None, validation_alias="PLAN_DURATION_VOLUME") duration_isolated: int | None = Field(default=None, validation_alias="PLAN_DURATION_ISOLATED") duration_spike: int | None = Field(default=None, validation_alias="PLAN_DURATION_SPIKE") + duration_diagnostic: int | None = Field( + default=None, + validation_alias="PLAN_DURATION_DIAGNOSTIC", + ) def duration_for(self, profile: TestKind) -> int: durations = { @@ -233,6 +237,7 @@ def duration_for(self, profile: TestKind) -> int: "volume": self.duration_volume, "isolated": self.duration_isolated, "spike": self.duration_spike, + "diagnostic": self.duration_diagnostic, } duration = durations[profile.value] return self.duration_sec if duration is None else duration diff --git a/tests/image/performance/fio_disks.py b/tests/image/performance/fio_disks.py index e7e82f89..35db97b1 100644 --- a/tests/image/performance/fio_disks.py +++ b/tests/image/performance/fio_disks.py @@ -1,13 +1,16 @@ import logging import queue +from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING +from imgtests.database.database import ImgtestsDatabase, UtilityResultRecord from imgtests.exec.loaders.dmsetup import DeviceMapperSetup, setup_block_device from imgtests.exec.observers.resource import get_available_ram_size from imgtests.exec.osinfo import get_os_release from imgtests.runner import AbstractRunnableTimeLimitedTest, TestResult, TestStatus from imgtests.suites.drive.fio import FioSuite, FioSuiteConfig, FioWorkload +from imgtests.sysrep import get_system_info from imgtests.types import Distro, Subsystem if TYPE_CHECKING: @@ -19,6 +22,8 @@ logger = logging.getLogger(__name__) +FIO_RESULTS_DIR = Path.home() / "fio" + SCALING_WORKLOADS: tuple[FioWorkload, ...] = ( FioWorkload("seq_write_1M", "write", "1M", 1.0), @@ -63,14 +68,109 @@ ) -class FioDisksScalingTest(AbstractRunnableTimeLimitedTest): +class _FioDisksBaseTest(AbstractRunnableTimeLimitedTest): + def __init__( + self, + description: str, + timeout: int, + *, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: + super().__init__(description, frozenset({Subsystem.FILE}), timeout) + self.db = db or ImgtestsDatabase() + self.config_id = config_id + self.experiment_description = experiment_description + + def run_fio_suite( + self, + client: SSHClient | None, + timeout: int, + cfg: FioSuiteConfig, + *, + success_msg: str, + default_experiment_description: str, + ) -> Iterable[TestResult]: + started_at = datetime.now(UTC) + suite_results: list[TestResult] = [] + + for result in FioSuite(client, cfg).run(): + suite_results.append(result) + yield result + + ended_at = datetime.now(UTC) + self._write_suite_result_to_db( + client=client, + timeout=timeout, + cfg=cfg, + suite_results=tuple(suite_results), + started_at=started_at, + ended_at=ended_at, + default_experiment_description=default_experiment_description, + ) + self.logger.info("%s cases=%d", success_msg, len(suite_results)) + + def _write_suite_result_to_db( # noqa: PLR0913 + self, + *, + client: SSHClient | None, + timeout: int, + cfg: FioSuiteConfig, + suite_results: tuple[TestResult, ...], + started_at: datetime, + ended_at: datetime, + default_experiment_description: str, + ) -> None: + if self.config_id is None: + cfg_record = self.db.insert_from_system_info(get_system_info(client)) + self.config_id = int(cfg_record.config_id) + + experiment = self.db.insert_experiment( + config_id=int(self.config_id), + description=self.experiment_description or default_experiment_description, + experiment_type="performance", + started_at=started_at, + ended_at=ended_at, + ) + self.db.insert_utility_result( + UtilityResultRecord( + experiment_id=int(experiment.experiment_id), + utility="fio", + command=("fio-suite", cfg.suite), + result={ + "suite": cfg.suite, + "timeout_sec": timeout, + "results_dir": cfg.results_dir, + "workloads": cfg.workloads, + "filename": cfg.filename, + "results_count": len(suite_results), + "results": suite_results, + }, + description=f"fio suite result: {cfg.suite}", + started_at=started_at, + ended_at=ended_at, + context={"test_name": type(self).__name__}, + ), + ) + + +class FioDisksScalingTest(_FioDisksBaseTest): """Test that runs fio on a disk with scaling workloads.""" - def __init__(self, timeout: int) -> None: + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: super().__init__( "Scaling load drives with fio.", - frozenset({Subsystem.FILE}), - timeout=timeout, + timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, ) def _run( @@ -82,17 +182,35 @@ def _run( cfg = FioSuiteConfig( suite="scaling", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=SCALING_WORKLOADS, ) - yield from _handle_fio_suite(client, cfg, "FIO scaling PASSED.") + yield from self.run_fio_suite( + client=client, + timeout=timeout, + cfg=cfg, + success_msg="FIO scaling PASSED.", + default_experiment_description="fio suite 'scaling' run", + ) -class FioDisksNightly(AbstractRunnableTimeLimitedTest): +class FioDisksNightly(_FioDisksBaseTest): """Tests that run fio on a disk with nightly workloads.""" - def __init__(self, timeout: int) -> None: - super().__init__("Nightly load drives with fio.", frozenset({Subsystem.FILE}), timeout) + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: + super().__init__( + "Nightly load drives with fio.", + timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, + ) def _run( self, @@ -103,17 +221,35 @@ def _run( cfg = FioSuiteConfig( suite="nightly", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=NIGHTLY_WORKLOADS, ) - yield from _handle_fio_suite(client, cfg, "FIO nightly PASSED.") + yield from self.run_fio_suite( + client=client, + timeout=timeout, + cfg=cfg, + success_msg="FIO nightly PASSED.", + default_experiment_description="fio suite 'nightly' run", + ) -class FioDisksDMDelay(AbstractRunnableTimeLimitedTest): +class FioDisksDMDelay(_FioDisksBaseTest): """Tests that run fio on a disk with dm-delay.""" - def __init__(self, timeout: int) -> None: - super().__init__("Dm-delay test with fio.", frozenset({Subsystem.FILE}), timeout) + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: + super().__init__( + "Dm-delay test with fio.", + timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, + ) def _run( self, @@ -124,39 +260,58 @@ def _run( os_id = get_os_release(client).id if os_id and os_id != Distro.POKY.value: self.logger.warning("Skipping test due dm-delay test is only supported on poky.") - return TestResult(status=TestStatus.SKIPPED) + yield TestResult(status=TestStatus.SKIPPED) + return result = setup_block_device(client=client) if result is not None and result.returncode: logger.error("Error in block device setup.") - return TestResult(status=TestStatus.BROKEN) + yield TestResult(status=TestStatus.BROKEN) + return dm = DeviceMapperSetup(client) result = dm.create_dm_delay_device() if result.returncode: logger.error("Error in creating dm-delay device.") - return TestResult(status=TestStatus.BROKEN) + yield TestResult(status=TestStatus.BROKEN) + return cfg = FioSuiteConfig( suite="dm-delay", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=SCALING_WORKLOADS, filename=Path("/dev/mapper/delay1"), ) - yield from _handle_fio_suite(client, cfg, "FIO dm-delay PASSED.") - dm.remove_dm_device(device_name="delay1") + try: + yield from self.run_fio_suite( + client=client, + timeout=timeout, + cfg=cfg, + success_msg="FIO dm-delay PASSED.", + default_experiment_description="fio suite 'dm-delay' run", + ) + finally: + dm.remove_dm_device(device_name="delay1") -class FioDisksDMDust(AbstractRunnableTimeLimitedTest): +class FioDisksDMDust(_FioDisksBaseTest): """Tests that run fio on a disk with dm-dust.""" - def __init__(self, timeout: int) -> None: + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: super().__init__( "Dm-dust fio test with errors on read.", - frozenset({Subsystem.FILE}), timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, ) def _run( @@ -168,54 +323,77 @@ def _run( os_id = get_os_release(client).id if os_id and os_id != Distro.POKY.value: self.logger.warning("Skipping test due dm-dust test is only supported on poky.") - return TestResult(status=TestStatus.SKIPPED) + yield TestResult(status=TestStatus.SKIPPED) + return result = setup_block_device(client=client) if result is not None and result.returncode: logger.error("Error in block device setup.") - return TestResult(status=TestStatus.BROKEN) + yield TestResult(status=TestStatus.BROKEN) + return dm = DeviceMapperSetup(client) result = dm.create_dm_dust_device() if result.returncode: - logger.error("Error in creating dm-delay device.") - return TestResult(status=TestStatus.BROKEN) - result = dm.add_bad_blocks(device_name="dust1", block_numbers=list(range(50, 100))) - if result.returncode: - return TestResult(status=TestStatus.BROKEN) - - read_cfg = FioSuiteConfig( - suite="dm-dust", - duration_sec=timeout, - results_dir=Path().home() / "fio", - workloads=DMDUST_READ_WORKLOAD, - filename=Path("/dev/mapper/dust1"), - ) - write_cfg = FioSuiteConfig( - suite="dm-dust", - duration_sec=timeout, - results_dir=Path().home() / "fio", - workloads=DMDUST_WRITE_WORKLOAD, - filename=Path("/dev/mapper/dust1"), - ) + logger.error("Error in creating dm-dust device.") + yield TestResult(status=TestStatus.BROKEN) + return try: - FioSuite(client, read_cfg).run() - except RuntimeError: - logger.info("Error above is intended, dm-dust works.") + result = dm.add_bad_blocks(device_name="dust1", block_numbers=list(range(50, 100))) + if result.returncode: + logger.error("Error in adding bad blocks to dm-dust device.") + yield TestResult(status=TestStatus.BROKEN) + return + + read_cfg = FioSuiteConfig( + suite="dm-dust", + duration_sec=timeout, + results_dir=FIO_RESULTS_DIR, + workloads=DMDUST_READ_WORKLOAD, + filename=Path("/dev/mapper/dust1"), + ) + write_cfg = FioSuiteConfig( + suite="dm-dust", + duration_sec=timeout, + results_dir=FIO_RESULTS_DIR, + workloads=DMDUST_WRITE_WORKLOAD, + filename=Path("/dev/mapper/dust1"), + ) - yield from _handle_fio_suite(client, write_cfg, "FIO dm-dust PASSED.") - dm.remove_dm_device(device_name="dust1") + read_results = tuple(FioSuite(client, read_cfg).run()) + if any(result.status is TestStatus.FAILED for result in read_results): + logger.info("dm-dust read workload failed as expected.") + else: + logger.warning("dm-dust read workload completed without the expected error.") + + yield from self.run_fio_suite( + client=client, + timeout=timeout, + cfg=write_cfg, + success_msg="FIO dm-dust PASSED.", + default_experiment_description="fio suite 'dm-dust' run", + ) + finally: + dm.remove_dm_device(device_name="dust1") -class FioDisksVariationTest(AbstractRunnableTimeLimitedTest): +class FioDisksVariationTest(_FioDisksBaseTest): """Test that runs fio on a disk with variations of bs, rw and offset.""" - def __init__(self, timeout: int) -> None: + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: super().__init__( "Fio parameter variation test.", - frozenset({Subsystem.FILE}), - timeout=timeout, + timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, ) def _run( @@ -242,25 +420,41 @@ def _run( cfg = FioSuiteConfig( suite=f"variation-offset-{offset}-{offset_incr or 'none'}", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=workloads, offset=offset, offset_increment=offset_incr, size=size, - filename=f"variation_offset_{offset}_{offset_incr or 'none'}_testfile", + filename=Path(f"variation_offset_{offset}_{offset_incr or 'none'}_testfile"), ) - yield from _handle_fio_suite( - client, - cfg, - f"FIO variation offset={offset}, incr={offset_incr} PASSED.", + yield from self.run_fio_suite( + client=client, + timeout=timeout, + cfg=cfg, + success_msg=f"FIO variation offset={offset}, incr={offset_incr} PASSED.", + default_experiment_description=( + f"fio suite 'variation' offset={offset} incr={offset_incr or 'none'} run" + ), ) -class FioDisksParallelLoadTest(AbstractRunnableTimeLimitedTest): +class FioDisksParallelLoadTest(_FioDisksBaseTest): """Test that runs fio parallel mixed workloads.""" - def __init__(self, timeout: int) -> None: - super().__init__("Fio parallel load test.", frozenset({Subsystem.FILE}), timeout=timeout) + def __init__( + self, + timeout: int, + db: ImgtestsDatabase | None = None, + config_id: int | None = None, + experiment_description: str | None = None, + ) -> None: + super().__init__( + "Fio parallel load test.", + timeout, + db=db, + config_id=config_id, + experiment_description=experiment_description, + ) def _run( self, @@ -273,73 +467,71 @@ def _run( FioSuiteConfig( suite="small", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=SMALL_BLOCK_WORKLOAD, size=size, - filename="small_testfile", + filename=Path("small_testfile"), ), FioSuiteConfig( suite="large", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=LARGE_BLOCK_WORKLOAD, size=size, - filename="large_testfile", + filename=Path("large_testfile"), ), FioSuiteConfig( suite="large-with-offset", duration_sec=timeout, - results_dir=Path().home() / "fio", + results_dir=FIO_RESULTS_DIR, workloads=LARGE_BLOCK_WORKLOAD, offset_increment="3k", size=size, - filename="large_with_offset_testfile", + filename=Path("large_with_offset_testfile"), ), ] - q = queue.Queue() + q: queue.Queue[TestResult] = queue.Queue() futures = [ executor.submit( _enqueue_fio_results, + self, client, + timeout, cfg, "FIO parallel load test PASSED.", q, ) for cfg in configs ] - while any(not f.done() for f in futures) or not q.empty(): + while any(not future.done() for future in futures) or not q.empty(): try: - r = q.get(timeout=0.5) + yield q.get(timeout=0.5) except queue.Empty: continue - yield r + for future in futures: + future.result() -def _enqueue_fio_results( +def _enqueue_fio_results( # noqa: PLR0913 + runner: _FioDisksBaseTest, client: SSHClient | None, + timeout: int, cfg: FioSuiteConfig, msg: str, q: queue.Queue[TestResult], ) -> None: - for result in _handle_fio_suite(client, cfg, msg): + for result in runner.run_fio_suite( + client=client, + timeout=timeout, + cfg=cfg, + success_msg=msg, + default_experiment_description=f"fio suite '{cfg.suite}' run", + ): q.put(result) -def _handle_fio_suite( - client: SSHClient | None, - cfg: FioSuiteConfig, - msg: str, -) -> Iterable[TestResult]: - fio_gen = FioSuite(client, cfg).run() - yield from fio_gen - try: - next(fio_gen) - except StopIteration: - logger.info(msg) - - def _calculate_fio_ram_percent(percent: int, client: SSHClient | None = None) -> str: - if percent <= 0 or percent > 100: # noqa: PLR2004 + if percent <= 0 or percent > 100: # noqa: PLR2004 return "100MB" ram_size = get_available_ram_size(client) if ram_size is not None: