diff --git a/benchmark/rawfio.py b/benchmark/rawfio.py
index fad27550..962dae02 100644
--- a/benchmark/rawfio.py
+++ b/benchmark/rawfio.py
@@ -1,8 +1,13 @@
+import os
+import re
+import json
+import hashlib
import common
import settings
import monitoring
import time
import logging
+from pathlib import Path
from .benchmark import Benchmark
@@ -12,15 +17,38 @@
class RawFio(Benchmark):
def __init__(self, archive_dir, cluster, config):
- super(RawFio, self).__init__(archive_dir, cluster, config)
+ cbt_logger = logging.getLogger("cbt")
+ original_level = cbt_logger.level
+ # Suppress the spurious per-iodepth "Results dir" log from base class
+ cbt_logger.setLevel(logging.WARNING)
+ try:
+ super(RawFio, self).__init__(archive_dir, cluster, config)
+ finally:
+ cbt_logger.setLevel(original_level if original_level != logging.NOTSET else logging.DEBUG)
+ # Recompute archive_dir hash excluding iodepth so all iodepth permutations share same archive_dir
+ config_without_iodepth = {k: v for k, v in self.config.items() if k not in ('iodepth', 'total_iodepth')}
+ hashable = json.dumps(sorted(config_without_iodepth.items())).encode()
+ digest = hashlib.sha1(hashable).hexdigest()[:8]
+ # archive_dir mirrors the librbdfio layout: /results//id-
+ self.archive_dir = os.path.join(self._base_archive_directory,
+ 'results',
+ '{:0>8}'.format(self.config.get('iteration')),
+ 'id-{}'.format(digest))
# comma-separated list of block devices to use inside the client host/VM/container
- self.block_device_list = config.get('block_devices', '/dev/vdb')
- self.block_devices = [d.strip() for d in self.block_device_list.split(',')]
+ _block_devices = config.get('block_devices', '/dev/vdb')
+ if isinstance(_block_devices, list):
+ self.block_devices = [d.strip() for d in _block_devices]
+ else:
+ self.block_devices = [d.strip() for d in _block_devices.split(',')]
+ self.block_device_list = ','.join(self.block_devices)
self.concurrent_procs = config.get('concurrent_procs', len(self.block_devices))
self.total_procs = self.concurrent_procs * len(settings.getnodes('clients').split(','))
- self.fio_out_format = "json"
+ # Use json,normal so the output format matches librbdfio and parse() can extract
+ # the JSON block correctly from the mixed output.
+ self.fio_out_format = "json,normal"
self.time = str(config.get('time', '300'))
self.ramp = str(config.get('ramp', '0'))
+ self.invalidate = config.get('invalidate', 0)
self.startdelay = config.get('startdelay', None)
self.rate_iops = config.get('rate_iops', None)
self.iodepth = config.get('iodepth', 16)
@@ -31,48 +59,61 @@ def __init__(self, archive_dir, cluster, config):
self.rwmixwrite = 100 - self.rwmixread
self.ioengine = config.get('ioengine', 'libaio')
self.op_size = config.get('op_size', 4194304)
- self.vol_size = config.get('vol_size', 65536) * 0.9
+ self.vol_size = config.get('vol_size', 65536)
self.fio_cmd = config.get('fio_cmd', 'sudo /usr/bin/fio')
# FIXME there are too many permutations, need to put results in SQLITE3
- self.run_dir = '%sraw_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.run_dir, int(self.osd_ra), int(self.op_size), int(self.total_procs), int(self.iodepth), self.mode)
- self.out_dir = '%s/raw_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.archive_dir, int(self.osd_ra), int(self.op_size), int(self.total_procs), int(self.iodepth), self.mode)
-
- # def exists(self):
- # if os.path.exists(self.out_dir):
- # logger.info('Skipping existing test in %s.', self.out_dir)
- # return True
- # return False
+ self.run_dir += (f'op_size-{int(self.op_size):08d}/'
+ f'concurrent_procs-{int(self.total_procs):03d}/'
+ f'{self.mode}/iodepth-{int(self.iodepth):03d}')
+ # out_dir = archive_dir/mode/iodepth-NNN so that:
+ # 1. Each iodepth run archives to its own subdir (no overwrite).
+ # 2. The formatter receives archive_dir/mode as the io_pattern_dir
+ # and recurses into iodepth-NNN/ finding all json_output files.
+ # 3. FIO._get_iodepth() reads the correct iodepth from the path.
+ self.out_dir = os.path.join(self.archive_dir, self.mode,
+ f'iodepth-{int(self.iodepth):03d}')
+ logger.info("Results dir: %s", self.out_dir)
+
+ def exists(self):
+ marker = os.path.join(self._base_archive_directory, 'rawfio.initialized')
+ if os.path.exists(marker):
+ os.makedirs(self.out_dir, exist_ok=True)
+ return True
+ logger.info('rawfio exists returning False')
+ return False
def initialize(self):
super(RawFio, self).initialize()
- common.pdsh(settings.getnodes('clients'),
- 'sudo rm -rf %s' % self.run_dir,
- continue_if_error=False).communicate()
- common.make_remote_dir(self.run_dir)
- clnts = settings.getnodes('clients')
- logger.info('creating mountpoints...')
+ marker = os.path.join(self._base_archive_directory, 'rawfio.initialized')
+ open(marker, 'w').close()
- logger.info('Attempting to initialize fio files...')
+ def prefill(self):
+ clnts = settings.getnodes('clients')
+ logger.info('Attempting to prefill fio devices...')
initializer_list = []
+
+ logger.info('%s', self.block_devices)
+
for i in range(self.concurrent_procs):
b = self.block_devices[i % len(self.block_devices)]
fiopath = b
- pre_cmd = 'sudo %s --rw=write -ioengine=%s --bs=%s ' % (self.fio_cmd, self.ioengine, self.op_size)
- pre_cmd = '%s --size %dM --name=%s --output-format=%s> /dev/null' % (
- pre_cmd, self.vol_size, fiopath, self.fio_out_format)
+ pre_cmd = 'sudo %s --rw=write -ioengine=%s --numjobs=1 --bs=65536 ' % (self.fio_cmd, self.ioengine)
+ pre_cmd += '--size %dM --invalidate=%s --name=%s --output-format=%s> /dev/null' % (
+ self.vol_size, self.invalidate, fiopath, self.fio_out_format)
+
initializer_list.append(common.pdsh(clnts, pre_cmd,
continue_if_error=False))
+
for p in initializer_list:
p.communicate()
- # Create the run directory
+ # Recreate the run directory after prefill
common.pdsh(clnts, 'rm -rf %s' % self.run_dir,
continue_if_error=False).communicate()
common.make_remote_dir(self.run_dir)
def run(self):
super(RawFio, self).run()
- # Set client readahead
clnts = settings.getnodes('clients')
# We'll always drop caches for rados bench
@@ -96,6 +137,8 @@ def run(self):
fio_cmd += ' --ioengine=%s' % self.ioengine
fio_cmd += ' --runtime=%s' % self.time
fio_cmd += ' --ramp_time=%s' % self.ramp
+ if self.invalidate:
+ fio_cmd += ' --invalidate=%s' % self.invalidate
if self.startdelay:
fio_cmd += ' --startdelay=%s' % self.startdelay
if self.rate_iops:
@@ -104,16 +147,15 @@ def run(self):
fio_cmd += ' --direct=%s' % self.direct
fio_cmd += ' --bs=%dB' % self.op_size
fio_cmd += ' --iodepth=%d' % self.iodepth
- fio_cmd += ' --size=%dM' % self.vol_size
if self.log_iops:
fio_cmd += ' --write_iops_log=%s' % out_file
if self.log_bw:
fio_cmd += ' --write_bw_log=%s' % out_file
if self.log_lat:
fio_cmd += ' --write_lat_log=%s' % out_file
- fio_cmd += ' --output-format=%s' % self.fio_out_format
if 'recovery_test' in self.cluster.config:
fio_cmd += ' --time_based'
+ fio_cmd += ' --output-format=%s' % self.fio_out_format
fio_cmd += ' --name=%s > %s' % (fiopath, out_file)
logger.debug("FIO CMD: %s" % fio_cmd)
fio_process_list.append(common.pdsh(clnts, fio_cmd, continue_if_error=False))
@@ -123,6 +165,48 @@ def run(self):
logger.info('Finished raw fio test')
common.sync_files('%s/*' % self.run_dir, self.out_dir)
+ self.analyze(self.out_dir)
+
+ def parse(self, out_dir):
+ # rpdcp appends the client hostname to every synced file, turning
+ # "output.0" into "output.0.hostname.domain". We match both the
+ # plain form (output.) and the suffixed form
+ # (output..), then always write the output as the
+ # plain "json_output." so fio_common_output_wrapper can find it.
+ archive_path = Path(out_dir)
+ files_to_process = [
+ f for f in archive_path.glob("**/output.*")
+ if re.search(r"output\.\d+", str(f))
+ and not re.search(r"output\.\d+_", str(f)) # exclude _bw/_iops/_lat etc.
+ ]
+ for file in files_to_process:
+ # Extract the numeric index from the filename regardless of any
+ # trailing hostname suffix, e.g. "output.7.soko07.front.sepia.ceph.com" -> "7"
+ m = re.search(r"output\.(\d+)", file.name)
+ if not m:
+ continue
+ index = m.group(1)
+ output_file_name = f"{file.parent}/json_output.{index}"
+ output_path = Path(output_file_name)
+ found = False
+ with file.open("r", encoding="utf-8") as input_file:
+ with output_path.open("w", encoding="utf-8") as output_file:
+ for line in input_file.readlines():
+ if re.search("^{$", line):
+ # Write the opening brace and mark that we are inside the JSON block.
+ output_file.write(line)
+ found = True
+ continue
+ if re.search("^}$", line):
+ output_file.write(line)
+ found = False
+ break
+ if found:
+ output_file.write(line)
+
+ def analyze(self, out_dir):
+ logger.info('Convert results to json format.')
+ self.parse(out_dir)
def cleanup(self):
super(RawFio, self).cleanup()
diff --git a/common.py b/common.py
index a870c968..4c8b2ccf 100644
--- a/common.py
+++ b/common.py
@@ -29,6 +29,8 @@ def all_configs(config):
# the set for permutation
if param == "acceptable":
default[param] = value
+ elif param == "block_devices":
+ default[param] = value
elif isinstance(value, list):
cycle_over_lists.append(value)
cycle_over_names.append(param)
diff --git a/example/nvme-raw.yaml b/example/nvme-raw.yaml
new file mode 100755
index 00000000..175f64c3
--- /dev/null
+++ b/example/nvme-raw.yaml
@@ -0,0 +1,54 @@
+cluster:
+ user: 'root'
+ head: "xxxxx.front.sepia.ceph.com"
+ clients: ["xxxxx.front.sepia.ceph.com"]
+ osds: ["xxxxx.front.sepia.ceph.com"]
+
+ use_existing: True
+ osds_per_node: 6
+ clients: [xxxxx.front.sepia.ceph.com]
+ iterations: 1
+ tmp_dir: "/tmp/cbt"
+ pdsh_ssh_args: "-a -x -l%u %h"
+
+ # Workload support to come - For now this only supports a single workload
+ # 4krandomwrite:
+ # jobname: 'randwrite'
+ # mode: 'randwrite'
+ # op_size: 4096
+ # numjobs: [ 1 ]
+ # total_iodepth: [ 2, 4, 8, 16, 32, 64, 128, 256, 384, 512, 768 ]
+ # 4krandomread:
+ # jobname: 'randread'
+ # mode: 'randread'
+ # op_size: 4096
+ # numjobs: [ 1 ]
+ # total_iodepth: [ 4, 8, 12, 16, 32, 48, 64, 128, 256, 384, 588, 768 ]
+
+monitoring_profiles:
+ collectl:
+ args: '-c 18 -sCD -i 10 -P -oz -F0 --rawtoo --sep ";" -f {collectl_dir}'
+
+benchmarks:
+ rawfio:
+ iterations: 1
+ time: 10
+ ramp: 10
+ iodepth: [1,2,3,4,5,6,7,8,9,10,16,24,32,48]
+ numjobs: 1
+ osd_ra: [4096]
+ mode: [ randwrite ]
+ ioengine: libaio
+ # Block Size
+ op_size: [4096]
+ # size o volume test
+ # vol_size: 100000
+ vol_size: 100
+ direct: 1
+ # Readahead settings
+ client_ra: 128
+ # Number of concurrent processes, if this number is greater than the number of block_devices, multiple instances of FIO will be started for that device
+ concurrent_procs: 16
+ fio_cmd: '/usr/local/bin/fio'
+
+ block_devices: [ /dev/nvme8n1, /dev/nvme8n2, /dev/nvme8n3, /dev/nvme8n4, /dev/nvme8n5, /dev/nvme8n6, /dev/nvme8n7, /dev/nvme8n8, /dev/nvme8n9, /dev/nvme8n10, /dev/nvme8n11, /dev/nvme8n12, /dev/nvme8n13, /dev/nvme8n14, /dev/nvme8n15, /dev/nvme8n16 ]