Skip to content

FIO

fio (Flexible I/O Tester) — the standard Linux I/O benchmarking and stress-testing tool. Wraps job file generation, execution, and result parsing.

sts.fio.fio

FIO test execution.

FIO pydantic-model

Bases: StsBaseModel

FIO test execution.

Example 
fio = FIO()  # Uses default parameters
fio = FIO('/dev/sda')  # Uses device with default parameters
fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters
Show JSON schema: 
{
  "$defs": {
    "FIOParameters": {
      "additionalProperties": false,
      "description": "Base FIO parameters.",
      "properties": {
        "name": {
          "default": "sts-fio",
          "title": "Name",
          "type": "string"
        },
        "ioengine": {
          "default": "libaio",
          "enum": [
            "libaio",
            "sync",
            "posixaio",
            "mmap",
            "splice"
          ],
          "title": "Ioengine",
          "type": "string"
        },
        "direct": {
          "default": true,
          "title": "Direct",
          "type": "boolean"
        },
        "rw": {
          "default": "randrw",
          "enum": [
            "read",
            "write",
            "randread",
            "randwrite",
            "randrw",
            "trim"
          ],
          "title": "Rw",
          "type": "string"
        },
        "bs": {
          "default": "4k",
          "minLength": 1,
          "title": "Bs",
          "type": "string"
        },
        "numjobs": {
          "default": 1,
          "title": "Numjobs",
          "type": "integer"
        },
        "group_reporting": {
          "default": false,
          "title": "Group Reporting",
          "type": "boolean"
        },
        "verify_fatal": {
          "default": false,
          "title": "Verify Fatal",
          "type": "boolean"
        },
        "end_fsync": {
          "default": false,
          "title": "End Fsync",
          "type": "boolean"
        },
        "filename": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Filename"
        },
        "size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "runtime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Runtime"
        },
        "iodepth": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Iodepth"
        },
        "verify": {
          "anyOf": [
            {
              "enum": [
                "crc32",
                "crc32c",
                "crc32c-intel",
                "md5",
                "sha1",
                "sha256",
                "sha512",
                "xxhash",
                "meta"
              ],
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Verify"
        },
        "verify_backlog": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Verify Backlog"
        }
      },
      "title": "FIOParameters",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "FIO test execution.\n\nExample:\n    ```python\n    fio = FIO()  # Uses default parameters\n    fio = FIO('/dev/sda')  # Uses device with default parameters\n    fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters\n    ```",
  "properties": {
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "parameters": {
      "anyOf": [
        {
          "$ref": "#/$defs/FIOParameters"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "options": {
      "items": {
        "type": "string"
      },
      "title": "Options",
      "type": "array"
    },
    "config_file": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Config File"
    }
  },
  "title": "FIO",
  "type": "object"
}

Fields:

  • filename (PathOrStr | None)
  • parameters (FIOParameters | None)
  • options (list[str])
  • config_file (PathOrStr | None)

Validators:

  • _init_parameters
Source code in sts_libs/src/sts/fio/fio.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class FIO(StsBaseModel):
    """FIO test execution.

    Example:
        ```python
        fio = FIO()  # Uses default parameters
        fio = FIO('/dev/sda')  # Uses device with default parameters
        fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters
        ```
    """

    # Optional parameters
    filename: PathOrStr | None = None
    parameters: FIOParameters | None = None
    options: list[str] = Field(default_factory=list)
    config_file: PathOrStr | None = None

    @model_validator(mode='after')
    def _init_parameters(self) -> Self:
        """Initialize FIO parameters and config."""
        if self.config_file:
            self.load_config_file(self.config_file)
            if not self.filename and self.parameters:
                self.filename = self.parameters.filename
        elif not self.parameters:
            self.parameters = DefaultParameters(name='sts-fio-default')
            if not self.filename:
                self.filename = self.parameters.filename
        return self

    def load_config_file(self, config_file: str | Path) -> None:
        """Load parameters from FIO config file.

        Raises:
            FIOConfigError: If config file is invalid or cannot be read
        """
        config_path = Path(config_file)
        if not config_path.exists():
            raise FIOConfigError(f'Config file not found: {config_path}')

        try:
            config = configparser.ConfigParser()
            config.read(config_path)

            # Get the first job section
            job_section = next(s for s in config.sections() if s != 'global')
            params = dict(config[job_section])

            # Convert parameters
            self.parameters = FIOParameters(
                name=job_section,
                **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
            )
        except (configparser.Error, StopIteration) as e:
            raise FIOConfigError(f'Invalid config file: {e}') from e

    def save_config_file(self, config_file: str | Path) -> None:
        """Save parameters to FIO config file.

        Raises:
            FIOConfigError: If parameters are not set or file cannot be written
        """
        if not self.parameters:
            raise FIOConfigError('No parameters to save')

        config = configparser.ConfigParser()
        config[self.parameters.name] = self.parameters.to_dict()

        try:
            with Path(config_file).open('w') as f:
                config.write(f)
        except OSError as e:
            raise FIOConfigError(f'Failed to write config file: {e}') from e

    def update_parameters(self, params: dict[str, Any]) -> None:
        """Update parameters."""
        if not self.parameters:
            self.parameters = DefaultParameters(name='sts-fio-default')
        for key, value in params.items():
            setattr(self.parameters, key, value)

    def update_options(self, options: list[str]) -> None:
        """Add options, deduplicating against existing ones."""
        self.options = list(set(self.options + options))

    def load_default_params(self) -> None:
        """Load default parameters."""
        self.parameters = DefaultParameters(name='sts-fio-default')

    def load_fs_params(self) -> None:
        """Load filesystem parameters."""
        self.parameters = FileSystemParameters(name='sts-fio-fs')

    def load_block_params(self) -> None:
        """Load block device parameters."""
        self.parameters = BlockDeviceParameters(name='sts-fio-block')

    def load_stress_params(self) -> None:
        """Load stress test parameters."""
        self.parameters = StressParameters(name='sts-fio-stress')

    def _create_command(self) -> str:
        """Create FIO command string.

        Raises:
            FIOConfigError: If parameters are not set or filename is missing
        """
        if not self.parameters:
            raise FIOConfigError('No parameters set')

        # Build command
        if self.config_file:
            command = f'fio {self.config_file}'
        else:
            if not self.filename:
                raise FIOConfigError('No filename set')
            params = ' '.join(f'--{k}="{v}"' for k, v in self.parameters.to_dict().items() if v)
            command = f'fio --filename "{self.filename}" {params}'

        # Add options
        if self.options:
            opts = ' '.join(f'--{opt}' for opt in self.options)
            command = f'{command} {opts}'

        return command

    def run(self) -> CommandResult:
        """Run FIO test.

        Raises:
            FIOExecutionError: If the fio package cannot be installed or the command cannot be built.
        """
        if not ensure_installed('fio'):
            raise FIOExecutionError('Failed to install fio package')

        try:
            command = self._create_command()
        except FIOConfigError as e:
            raise FIOExecutionError(f'Failed to create command: {e}') from e

        result = run(command)
        if result.failed:
            logger.error(f'FIO run failed:\n{result.stderr}')
        else:
            logger.debug('FIO executed successfully')

        return result

load_block_params()

Load block device parameters.

Source code in sts_libs/src/sts/fio/fio.py 
126
127
128
def load_block_params(self) -> None:
    """Load block device parameters."""
    self.parameters = BlockDeviceParameters(name='sts-fio-block')

load_config_file(config_file)

Load parameters from FIO config file.

Raises:

Type Description
FIOConfigError

If config file is invalid or cannot be read
Source code in sts_libs/src/sts/fio/fio.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def load_config_file(self, config_file: str | Path) -> None:
    """Load parameters from FIO config file.

    Raises:
        FIOConfigError: If config file is invalid or cannot be read
    """
    config_path = Path(config_file)
    if not config_path.exists():
        raise FIOConfigError(f'Config file not found: {config_path}')

    try:
        config = configparser.ConfigParser()
        config.read(config_path)

        # Get the first job section
        job_section = next(s for s in config.sections() if s != 'global')
        params = dict(config[job_section])

        # Convert parameters
        self.parameters = FIOParameters(
            name=job_section,
            **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
        )
    except (configparser.Error, StopIteration) as e:
        raise FIOConfigError(f'Invalid config file: {e}') from e

load_default_params()

Load default parameters.

Source code in sts_libs/src/sts/fio/fio.py 
118
119
120
def load_default_params(self) -> None:
    """Load default parameters."""
    self.parameters = DefaultParameters(name='sts-fio-default')

load_fs_params()

Load filesystem parameters.

Source code in sts_libs/src/sts/fio/fio.py 
122
123
124
def load_fs_params(self) -> None:
    """Load filesystem parameters."""
    self.parameters = FileSystemParameters(name='sts-fio-fs')

load_stress_params()

Load stress test parameters.

Source code in sts_libs/src/sts/fio/fio.py 
130
131
132
def load_stress_params(self) -> None:
    """Load stress test parameters."""
    self.parameters = StressParameters(name='sts-fio-stress')

run()

Run FIO test.

Raises:

Type Description
FIOExecutionError

If the fio package cannot be installed or the command cannot be built.
Source code in sts_libs/src/sts/fio/fio.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def run(self) -> CommandResult:
    """Run FIO test.

    Raises:
        FIOExecutionError: If the fio package cannot be installed or the command cannot be built.
    """
    if not ensure_installed('fio'):
        raise FIOExecutionError('Failed to install fio package')

    try:
        command = self._create_command()
    except FIOConfigError as e:
        raise FIOExecutionError(f'Failed to create command: {e}') from e

    result = run(command)
    if result.failed:
        logger.error(f'FIO run failed:\n{result.stderr}')
    else:
        logger.debug('FIO executed successfully')

    return result

save_config_file(config_file)

Save parameters to FIO config file.

Raises:

Type Description
FIOConfigError

If parameters are not set or file cannot be written
Source code in sts_libs/src/sts/fio/fio.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def save_config_file(self, config_file: str | Path) -> None:
    """Save parameters to FIO config file.

    Raises:
        FIOConfigError: If parameters are not set or file cannot be written
    """
    if not self.parameters:
        raise FIOConfigError('No parameters to save')

    config = configparser.ConfigParser()
    config[self.parameters.name] = self.parameters.to_dict()

    try:
        with Path(config_file).open('w') as f:
            config.write(f)
    except OSError as e:
        raise FIOConfigError(f'Failed to write config file: {e}') from e

update_options(options)

Add options, deduplicating against existing ones.

Source code in sts_libs/src/sts/fio/fio.py 
114
115
116
def update_options(self, options: list[str]) -> None:
    """Add options, deduplicating against existing ones."""
    self.options = list(set(self.options + options))

update_parameters(params)

Update parameters.

Source code in sts_libs/src/sts/fio/fio.py 
107
108
109
110
111
112
def update_parameters(self, params: dict[str, Any]) -> None:
    """Update parameters."""
    if not self.parameters:
        self.parameters = DefaultParameters(name='sts-fio-default')
    for key, value in params.items():
        setattr(self.parameters, key, value)

sts.fio.parameters

FIO parameter configurations.

BlockDeviceParameters pydantic-model

Bases: FIOParameters

Parameters optimized for block device IOPS testing: random reads with high concurrency.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters optimized for block device IOPS testing: random reads with high concurrency.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randread",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "512",
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 4,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 1800,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 32,
      "title": "Iodepth"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "BlockDeviceParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • verify_fatal (bool)
  • end_fsync (bool)
  • filename (str | None)
  • size (str | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • group_reporting (bool)
  • iodepth (int | None)
  • runtime (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
106
107
108
109
110
111
112
113
114
class BlockDeviceParameters(FIOParameters):
    """Parameters optimized for block device IOPS testing: random reads with high concurrency."""

    rw: RWType = 'randread'
    bs: str = '512'
    numjobs: int = 4
    group_reporting: bool = True
    iodepth: int | None = 32
    runtime: int | None = Field(default=1800, gt=0)

DefaultParameters pydantic-model

Bases: FIOParameters

Default parameters for general FIO testing: random read/write with verification.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Default parameters for general FIO testing: random read/write with verification.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": true,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 60,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 32,
      "title": "Iodepth"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": "crc32",
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 1024,
      "title": "Verify Backlog"
    }
  },
  "title": "DefaultParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • end_fsync (bool)
  • filename (str | None)
  • size (str | None)
  • group_reporting (bool)
  • verify_fatal (bool)
  • iodepth (int | None)
  • runtime (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
85
86
87
88
89
90
91
92
93
class DefaultParameters(FIOParameters):
    """Default parameters for general FIO testing: random read/write with verification."""

    group_reporting: bool = True
    verify_fatal: bool = True
    iodepth: int | None = 32
    runtime: int | None = Field(default=60, gt=0)
    verify: VerifyType | None = 'crc32'
    verify_backlog: int | None = 1024

FIOParameters pydantic-model

Bases: StsBaseModel

Base FIO parameters.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Base FIO parameters.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": false,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "FIOParameters",
  "type": "object"
}

Config:

  • validate_assignment: True

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • group_reporting (bool)
  • verify_fatal (bool)
  • end_fsync (bool)
  • filename (str | None)
  • size (str | None)
  • runtime (int | None)
  • iodepth (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class FIOParameters(StsBaseModel):
    """Base FIO parameters."""

    model_config = ConfigDict(validate_assignment=True)

    # Optional parameters with defaults
    name: str = 'sts-fio'
    ioengine: IOEngine = 'libaio'
    direct: bool = True
    rw: RWType = 'randrw'
    bs: str = Field(default='4k', min_length=1)
    numjobs: int = 1
    group_reporting: bool = False
    verify_fatal: bool = False
    end_fsync: bool = False

    # Optional parameters without defaults
    filename: str | None = None
    size: str | None = None
    runtime: int | None = Field(default=None, gt=0)
    iodepth: int | None = None
    verify: VerifyType | None = None
    verify_backlog: int | None = None

    def to_dict(self) -> dict[str, str]:
        """Convert parameters to FIO command format (bools become '1', None fields omitted)."""
        result: dict[str, str] = {}
        for key, value in self.model_dump().items():
            if value is None:
                continue
            if isinstance(value, bool):
                if value:
                    result[key] = '1'
            else:
                result[key] = str(value)
        return result

    @classmethod
    def from_file(cls, path: str | Path) -> FIOParameters | None:
        """Create parameters from FIO config file, or None on parse failure."""
        try:
            config = configparser.ConfigParser()
            config.read(path)

            # Get the first job section
            job_section = next(s for s in config.sections() if s != 'global')
            params = dict(config[job_section])

            # Convert parameters
            return cls(
                name=job_section,
                **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
            )
        except (configparser.Error, StopIteration) as e:
            logger.warning(f'Invalid config file: {e}')
            return None

from_file(path) classmethod

Create parameters from FIO config file, or None on parse failure.

Source code in sts_libs/src/sts/fio/parameters.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@classmethod
def from_file(cls, path: str | Path) -> FIOParameters | None:
    """Create parameters from FIO config file, or None on parse failure."""
    try:
        config = configparser.ConfigParser()
        config.read(path)

        # Get the first job section
        job_section = next(s for s in config.sections() if s != 'global')
        params = dict(config[job_section])

        # Convert parameters
        return cls(
            name=job_section,
            **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
        )
    except (configparser.Error, StopIteration) as e:
        logger.warning(f'Invalid config file: {e}')
        return None

to_dict()

Convert parameters to FIO command format (bools become '1', None fields omitted).

Source code in sts_libs/src/sts/fio/parameters.py 
51
52
53
54
55
56
57
58
59
60
61
62
def to_dict(self) -> dict[str, str]:
    """Convert parameters to FIO command format (bools become '1', None fields omitted)."""
    result: dict[str, str] = {}
    for key, value in self.model_dump().items():
        if value is None:
            continue
        if isinstance(value, bool):
            if value:
                result[key] = '1'
        else:
            result[key] = str(value)
    return result

FileSystemParameters pydantic-model

Bases: FIOParameters

Parameters optimized for filesystem testing: sequential writes with large block size.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters optimized for filesystem testing: sequential writes with large block size.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "sync",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "write",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "1M",
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": false,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": true,
      "title": "End Fsync",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": "10G",
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "FileSystemParameters",
  "type": "object"
}

Fields:

  • name (str)
  • direct (bool)
  • numjobs (int)
  • group_reporting (bool)
  • verify_fatal (bool)
  • filename (str | None)
  • runtime (int | None)
  • iodepth (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • ioengine (IOEngine)
  • rw (RWType)
  • bs (str)
  • end_fsync (bool)
  • size (str | None)
Source code in sts_libs/src/sts/fio/parameters.py 
 96
 97
 98
 99
100
101
102
103
class FileSystemParameters(FIOParameters):
    """Parameters optimized for filesystem testing: sequential writes with large block size."""

    ioengine: IOEngine = 'sync'
    rw: RWType = 'write'
    bs: str = '1M'
    end_fsync: bool = True
    size: str | None = '10G'

StressParameters pydantic-model

Bases: FIOParameters

Parameters for stress/endurance testing: high concurrency random read/write.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters for stress/endurance testing: high concurrency random read/write.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 64,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 3600,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 64,
      "title": "Iodepth"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "StressParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • verify_fatal (bool)
  • end_fsync (bool)
  • filename (str | None)
  • size (str | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • numjobs (int)
  • group_reporting (bool)
  • iodepth (int | None)
  • runtime (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
117
118
119
120
121
122
123
class StressParameters(FIOParameters):
    """Parameters for stress/endurance testing: high concurrency random read/write."""

    numjobs: int = 64
    group_reporting: bool = True
    iodepth: int | None = 64
    runtime: int | None = Field(default=3600, gt=0)