From 91ea4b43b1a2d9c57ab46c9f3f81664886eb95d9 Mon Sep 17 00:00:00 2001 From: Chris Harris Date: Tue, 15 Sep 2026 11:22:22 +0100 Subject: [PATCH] logging: Add progress bars to the CLI output Add progress bars to the CLI only output from CBT to better show progress to the user. It includes an estimated time for the run as well as percentage through the currnt task and the overall run. IBM Bob 2.0.1 and 2.0.3 were used to help with the change Signed-off-by: Chris Harris --- benchmark/benchmark.py | 9 ++ benchmark/fio.py | 6 + benchmark/getput.py | 120 +++++++++------- benchmark/hsbench.py | 118 ++++++++------- benchmark/kvmrbdfio.py | 184 +++++++++++++----------- benchmark/librbdfio.py | 14 ++ benchmark/radosbench.py | 196 ++++++++++++++----------- benchmark/rawfio.py | 148 +++++++++++-------- benchmark/rbdfio.py | 179 ++++++++++++++--------- cbt.py | 148 ++++++++++++++----- progress.py | 311 ++++++++++++++++++++++++++++++++++++++++ tests/test_progress.py | 298 ++++++++++++++++++++++++++++++++++++++ workloads/workload.py | 30 ++++ workloads/workloads.py | 57 ++++++-- 14 files changed, 1363 insertions(+), 455 deletions(-) create mode 100644 progress.py create mode 100644 tests/test_progress.py diff --git a/benchmark/benchmark.py b/benchmark/benchmark.py index b78bc4d0..5cadc4c0 100644 --- a/benchmark/benchmark.py +++ b/benchmark/benchmark.py @@ -140,6 +140,15 @@ def initialize_endpoints(self): def prefill(self): pass + def estimate_duration(self) -> int: + """Return estimated total wall-clock seconds for all run-phase work. + + The base implementation returns 0. Concrete benchmarks should override + this to sum up their configured ``time``, ``ramp``, and any prefill + durations so that :mod:`progress` can size the overall progress bar. + """ + return 0 + def run(self): if self.osd_ra and self.osd_ra_changed: logger.info("Setting OSD Read Ahead to: %s", self.osd_ra) diff --git a/benchmark/fio.py b/benchmark/fio.py index 42a145e3..e92de4e6 100644 --- a/benchmark/fio.py +++ b/benchmark/fio.py @@ -54,6 +54,12 @@ def exists(self): return True return False + def estimate_duration(self) -> int: + """Estimate run-phase seconds: runtime + ramp time.""" + total = int(self.time) if self.time is not None else 0 + total += int(self.ramp) if self.ramp is not None else 0 + return total + def initialize(self): super(Fio, self).initialize() diff --git a/benchmark/getput.py b/benchmark/getput.py index aed81800..e3d2dfe6 100644 --- a/benchmark/getput.py +++ b/benchmark/getput.py @@ -17,29 +17,47 @@ def __init__(self, archive_dir, cluster, config): super(Getput, self).__init__(archive_dir, cluster, config) self.tmp_conf = self.cluster.tmp_conf - self.runtime = config.get('runtime', None) - self.container_prefix = config.get('container_prefix', 'cbt-getput') - self.object_prefix = config.get('object_prefix', 'cbt-getput') - self.procs = config.get('procs', 1) - self.ops_per_proc = config.get('ops_per_proc', None) - self.test = config.get('test', "p") - self.op_size = config.get('op_size', 4194304) - self.ctype = config.get('ctype', None) - self.debug = config.get('debug', None) - self.logops = config.get('logops', None) - self.grace = config.get('grace', None) - self.run_dir = '%s/osd_ra-%08d/op_size-%08d/procs-%08d/%s/%s' % (self.run_dir, int(self.osd_ra), int(self.op_size), int(self.procs), self.test, self.ctype) - self.out_dir = '%s/osd_ra-%08d/op_size-%08d/procs-%08d/%s/%s' % (self.archive_dir, int(self.osd_ra), int(self.op_size), int(self.procs), self.test, self.ctype) - self.pool_profile = config.get('pool_profile', 'default') - self.cmd_path = config.get('cmd_path', "/usr/bin/getput") - self.user = config.get('user', 'cbt') - self.subuser = '%s:swift' % self.user - self.key = config.get('key', 'vzCEkuryfn060dfee4fgQPqFrncKEIkh3ZcdOANY') # dummy key from ceph radosgw docs - self.auth_urls = config.get('auth', self.cluster.get_auth_urls()) + self.runtime = config.get("runtime", None) + self.container_prefix = config.get("container_prefix", "cbt-getput") + self.object_prefix = config.get("object_prefix", "cbt-getput") + self.procs = config.get("procs", 1) + self.ops_per_proc = config.get("ops_per_proc", None) + self.test = config.get("test", "p") + self.op_size = config.get("op_size", 4194304) + self.ctype = config.get("ctype", None) + self.debug = config.get("debug", None) + self.logops = config.get("logops", None) + self.grace = config.get("grace", None) + self.run_dir = "%s/osd_ra-%08d/op_size-%08d/procs-%08d/%s/%s" % ( + self.run_dir, + int(self.osd_ra), + int(self.op_size), + int(self.procs), + self.test, + self.ctype, + ) + self.out_dir = "%s/osd_ra-%08d/op_size-%08d/procs-%08d/%s/%s" % ( + self.archive_dir, + int(self.osd_ra), + int(self.op_size), + int(self.procs), + self.test, + self.ctype, + ) + self.pool_profile = config.get("pool_profile", "default") + self.cmd_path = config.get("cmd_path", "/usr/bin/getput") + self.user = config.get("user", "cbt") + self.subuser = "%s:swift" % self.user + self.key = config.get("key", "vzCEkuryfn060dfee4fgQPqFrncKEIkh3ZcdOANY") # dummy key from ceph radosgw docs + self.auth_urls = config.get("auth", self.cluster.get_auth_urls()) + + def estimate_duration(self) -> int: + """Estimate run-phase seconds from the configured runtime.""" + return int(self.runtime) if self.runtime is not None else 0 def exists(self): if os.path.exists(self.out_dir): - logger.info('Skipping existing test in %s.', self.out_dir) + logger.info("Skipping existing test in %s.", self.out_dir) return True return False @@ -57,55 +75,61 @@ def initialize(self): common.clean_remote_dir(self.run_dir) common.make_remote_dir(self.run_dir) - logger.info('Pausing for 60s for idle monitoring.') + logger.info("Pausing for 60s for idle monitoring.") MonitoringFactory.start("%s/idle_monitoring" % self.run_dir) time.sleep(60) MonitoringFactory.stop() - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def mkcredfiles(self): for i in range(0, len(self.auth_urls)): - cred = "export ST_AUTH=%s\\nexport ST_USER=%s\\nexport ST_KEY=%s" % (self.auth_urls[i], self.subuser, self.key) - common.pdsh(settings.getnodes('clients'), 'echo -e "%s" > %s/gw%02d.cred' % (cred, self.run_dir, i)).communicate() + cred = "export ST_AUTH=%s\\nexport ST_USER=%s\\nexport ST_KEY=%s" % ( + self.auth_urls[i], + self.subuser, + self.key, + ) + common.pdsh( + settings.getnodes("clients"), 'echo -e "%s" > %s/gw%02d.cred' % (cred, self.run_dir, i) + ).communicate() def mkgetputcmd(self, cred_file, gwnum): # grab the executable to use - getput_cmd = '%s ' % self.cmd_path + getput_cmd = "%s " % self.cmd_path # Set the options if self.container_prefix is not None: - container_prefix_flag = '-c%s' % self.container_prefix - if self.ctype == 'byproc' or self.ctype == 'bynodegw': - container_prefix_flag = '%s-gw%s' % (container_prefix_flag, gwnum) - getput_cmd += '%s ' % container_prefix_flag + container_prefix_flag = "-c%s" % self.container_prefix + if self.ctype == "byproc" or self.ctype == "bynodegw": + container_prefix_flag = "%s-gw%s" % (container_prefix_flag, gwnum) + getput_cmd += "%s " % container_prefix_flag # For now we'll only test distinct objects per client/gw if self.object_prefix is not None: - getput_cmd += '-o%s-`%s`-gw%s ' % (self.object_prefix, common.get_fqdn_cmd(), gwnum) + getput_cmd += "-o%s-`%s`-gw%s " % (self.object_prefix, common.get_fqdn_cmd(), gwnum) else: - getput_cmd += '-o`%s`-gw%s ' % (common.get_fqdn_cmd(), gwnum) + getput_cmd += "-o`%s`-gw%s " % (common.get_fqdn_cmd(), gwnum) - getput_cmd += '-s%s ' % self.op_size - getput_cmd += '-t%s ' % self.test - getput_cmd += '--procs %s ' % self.procs + getput_cmd += "-s%s " % self.op_size + getput_cmd += "-t%s " % self.test + getput_cmd += "--procs %s " % self.procs if self.ops_per_proc is not None: - getput_cmd += '-n%s ' % self.ops_per_proc + getput_cmd += "-n%s " % self.ops_per_proc if self.runtime is not None: - getput_cmd += '--runtime %s ' % self.runtime + getput_cmd += "--runtime %s " % self.runtime if self.ctype is not None: - getput_cmd += '--ctype %s ' % self.ctype + getput_cmd += "--ctype %s " % self.ctype if self.debug is not None: - getput_cmd += '--debug %s ' % self.debug + getput_cmd += "--debug %s " % self.debug if self.logops is not None: - getput_cmd += '--logops %s ' % self.logops + getput_cmd += "--logops %s " % self.logops if self.grace is not None: - getput_cmd += '--grace %s ' % self.grace + getput_cmd += "--grace %s " % self.grace - getput_cmd += '--cred %s ' % cred_file + getput_cmd += "--cred %s " % cred_file # End the getput_cmd - getput_cmd += '> %s/output.gw%s' % (self.run_dir, gwnum) + getput_cmd += "> %s/output.gw%s" % (self.run_dir, gwnum) return getput_cmd @@ -120,37 +144,37 @@ def run(self): self.cluster.dump_config(self.run_dir) # Run the backfill testing thread if requested - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: recovery_callback = self.recovery_callback self.cluster.create_recovery_test(self.run_dir, recovery_callback) # Run getput MonitoringFactory.start(self.run_dir) - logger.info('Running getput %s test.' % self.test) + logger.info("Running getput %s test." % self.test) ps = [] for i in range(0, len(self.auth_urls)): cmd = self.mkgetputcmd("%s/gw%02d.cred" % (self.run_dir, i), i) - p = common.pdsh(settings.getnodes('clients'), cmd) + p = common.pdsh(settings.getnodes("clients"), cmd) ps.append(p) for p in ps: p.wait() MonitoringFactory.stop(self.run_dir) # If we were doing recovery, wait until it's done. - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: self.cluster.wait_recovery_done() # Finally, get the historic ops self.cluster.dump_historic_ops(self.run_dir) - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def recovery_callback(self): self.cleanup() def cleanup(self): cmd_name = pathlib.PurePath(self.cmd_path).name - common.pdsh(settings.getnodes('clients'), 'sudo killall -9 %s' % cmd_name).communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall -9 %s" % cmd_name).communicate() def __str__(self): return "%s\n%s\n%s" % (self.run_dir, self.out_dir, super(Getput, self).__str__()) diff --git a/benchmark/hsbench.py b/benchmark/hsbench.py index efa9bb3b..8ea044f0 100644 --- a/benchmark/hsbench.py +++ b/benchmark/hsbench.py @@ -15,29 +15,33 @@ class Hsbench(Benchmark): def __init__(self, archive_dir, cluster, config): super(Hsbench, self).__init__(archive_dir, cluster, config) - self.cmd_path = config.get('cmd_path', '/usr/local/bin/hsbench') + self.cmd_path = config.get("cmd_path", "/usr/local/bin/hsbench") self.tmp_conf = self.cluster.tmp_conf - self.buckets = config.get('buckets', None) - self.bucket_prefix = config.get('bucket_prefix', None) - self.duration = config.get('duration', None) - self.loop = config.get('loop', None) - self.modes = config.get('modes', None) - self.max_keys = config.get('max_keys', None) - self.objects = config.get('objects', None) - self.object_prefix = config.get('object_prefix', None) - self.per_client_object_prefix = config.get('per_client_object_prefix', True) - self.region = config.get('region', None) - self.report_intervals = config.get('report_intervals', None) - self.threads = config.get('threads', None) - self.size = config.get('size', None) + self.buckets = config.get("buckets", None) + self.bucket_prefix = config.get("bucket_prefix", None) + self.duration = config.get("duration", None) + self.loop = config.get("loop", None) + self.modes = config.get("modes", None) + self.max_keys = config.get("max_keys", None) + self.objects = config.get("objects", None) + self.object_prefix = config.get("object_prefix", None) + self.per_client_object_prefix = config.get("per_client_object_prefix", True) + self.region = config.get("region", None) + self.report_intervals = config.get("report_intervals", None) + self.threads = config.get("threads", None) + self.size = config.get("size", None) self.out_dir = self.archive_dir self.client_endpoints = config.get("client_endpoints", None) - self.prefill_flag = config.get('prefill', False) - self.prefill_modes = config.get('prefill_modes', 'cxip') + self.prefill_flag = config.get("prefill", False) + self.prefill_modes = config.get("prefill_modes", "cxip") + + def estimate_duration(self) -> int: + """Estimate run-phase seconds from the configured duration.""" + return int(self.duration) if self.duration is not None else 0 def exists(self): if os.path.exists(self.out_dir): - logger.info('Skipping existing test in %s.', self.out_dir) + logger.info("Skipping existing test in %s.", self.out_dir) return True return False @@ -52,7 +56,7 @@ def initialize(self): def initialize_endpoints(self): super(Hsbench, self).initialize_endpoints() if self.client_endpoints is None: - raise ValueError('No client_endpoints defined!') + raise ValueError("No client_endpoints defined!") self.client_endpoints_object = client_endpoints_factory.get(self.cluster, self.client_endpoints) if not self.client_endpoints_object.get_initialized(): @@ -64,54 +68,60 @@ def initialize_endpoints(self): def mkcredfiles(self): for i in range(0, len(self.auth_urls)): - cred = "export ST_AUTH=%s\\nexport ST_USER=%s\\nexport ST_KEY=%s" % (self.auth_urls[i], self.subuser, self.key) - common.pdsh(settings.getnodes('clients'), 'echo -e "%s" > %s/gw%02d.cred' % (cred, self.run_dir, i)).communicate() + cred = "export ST_AUTH=%s\\nexport ST_USER=%s\\nexport ST_KEY=%s" % ( + self.auth_urls[i], + self.subuser, + self.key, + ) + common.pdsh( + settings.getnodes("clients"), 'echo -e "%s" > %s/gw%02d.cred' % (cred, self.run_dir, i) + ).communicate() def run_command(self, ep_num, cmd, prefill): - out_csv = '%s/output.%d.csv' % (self.run_dir, ep_num) - out_json = '%s/output.%d.json' % (self.run_dir, ep_num) + out_csv = "%s/output.%d.csv" % (self.run_dir, ep_num) + out_json = "%s/output.%d.json" % (self.run_dir, ep_num) - cmd = 'sudo %s' % cmd + cmd = "sudo %s" % cmd if self.buckets: - cmd += ' -b %d' % self.buckets + cmd += " -b %d" % self.buckets if self.bucket_prefix: - cmd += ' -bp %s' % self.bucket_prefix + cmd += " -bp %s" % self.bucket_prefix if self.duration: - cmd += ' -d %d' % self.duration + cmd += " -d %d" % self.duration if self.loop: - cmd += ' -l %d' % self.loop + cmd += " -l %d" % self.loop if prefill: - cmd += ' -m %s' % self.prefill_modes + cmd += " -m %s" % self.prefill_modes elif self.modes: - cmd += ' -m %s' % self.modes + cmd += " -m %s" % self.modes if self.max_keys: - cmd += ' -mk %d' % self.max_keys + cmd += " -mk %d" % self.max_keys if self.objects: - cmd += ' -n %d' % self.objects + cmd += " -n %d" % self.objects if self.object_prefix: object_prefix = self.object_prefix if self.per_client_object_prefix: - object_prefix += '-`%s`-%s-' % (common.get_fqdn_cmd(), ep_num) - cmd += ' -op %s' % object_prefix + object_prefix += "-`%s`-%s-" % (common.get_fqdn_cmd(), ep_num) + cmd += " -op %s" % object_prefix elif self.per_client_object_prefix: - cmd += ' -op `%s`-%s-' % (common.get_fqdn_cmd(), ep_num) + cmd += " -op `%s`-%s-" % (common.get_fqdn_cmd(), ep_num) if self.region: - cmd += ' -r %s' % self.region + cmd += " -r %s" % self.region if self.report_intervals: - cmd += ' -ri %s' % self.report_intervals + cmd += " -ri %s" % self.report_intervals if self.threads: - cmd += ' -t %d' % self.threads + cmd += " -t %d" % self.threads if self.size: - cmd += ' -z %s' % self.size - cmd += ' -o %s' % out_csv + cmd += " -z %s" % self.size + cmd += " -o %s" % out_csv if prefill: - cmd += '.prefill' - cmd += ' -j %s' % out_json + cmd += ".prefill" + cmd += " -j %s" % out_json if prefill: - cmd += '.prefill' - cmd += ' -s %s' % self.endpoints[ep_num % len(self.endpoints)]["secret_key"] - cmd += ' -a %s' % self.endpoints[ep_num % len(self.endpoints)]["access_key"] - cmd += ' -u %s' % self.endpoints[ep_num % len(self.endpoints)]["url"] + cmd += ".prefill" + cmd += " -s %s" % self.endpoints[ep_num % len(self.endpoints)]["secret_key"] + cmd += " -a %s" % self.endpoints[ep_num % len(self.endpoints)]["access_key"] + cmd += " -u %s" % self.endpoints[ep_num % len(self.endpoints)]["url"] return cmd @@ -119,10 +129,10 @@ def prefill(self): super(Hsbench, self).prefill() if not self.prefill_flag: return - logger.info('Attempting to prefill hsbench objects...') + logger.info("Attempting to prefill hsbench objects...") ps = [] for i in range(self.endpoints_per_client): - p = common.pdsh(settings.getnodes('clients'), self.run_command(i, self.cmd_path, True)) + p = common.pdsh(settings.getnodes("clients"), self.run_command(i, self.cmd_path, True)) ps.append(p) for p in ps: p.wait() @@ -137,38 +147,38 @@ def run(self): self.cluster.dump_config(self.run_dir) # Run the backfill testing thread if requested - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: recovery_callback = self.recovery_callback self.cluster.create_recovery_test(self.run_dir, recovery_callback) MonitoringFactory.start(self.run_dir) - logger.info('Running hsbench %s test.' % self.modes) + logger.info("Running hsbench %s test." % self.modes) ps = [] for i in range(self.endpoints_per_client): - p = common.pdsh(settings.getnodes('clients'), self.run_command(i, self.cmd_path_full, False)) + p = common.pdsh(settings.getnodes("clients"), self.run_command(i, self.cmd_path_full, False)) ps.append(p) for p in ps: p.wait() # If we were doing recovery, wait until it's done. - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: self.cluster.wait_recovery_done() MonitoringFactory.stop(self.run_dir) # If we were doing recovery, wait until it's done. - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: self.cluster.wait_recovery_done() # Finally, get the historic ops self.cluster.dump_historic_ops(self.run_dir) - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def recovery_callback(self): self.cleanup() def cleanup(self): cmd_name = pathlib.PurePath(self.cmd_path).name - common.pdsh(settings.getnodes('clients'), 'sudo killall -9 %s' % cmd_name).communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall -9 %s" % cmd_name).communicate() def __str__(self): return "%s\n%s\n%s" % (self.run_dir, self.out_dir, super(Hsbench, self).__str__()) diff --git a/benchmark/kvmrbdfio.py b/benchmark/kvmrbdfio.py index 225b3770..87e15f9a 100644 --- a/benchmark/kvmrbdfio.py +++ b/benchmark/kvmrbdfio.py @@ -15,81 +15,98 @@ class KvmRbdFio(Benchmark): def __init__(self, archive_dir, cluster, config): super(KvmRbdFio, self).__init__(archive_dir, cluster, config) # 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(',')] - self.concurrent_procs = config.get('concurrent_procs', len(self.block_devices)) - self.total_procs = self.concurrent_procs * len(settings.getnodes('clients').split(',')) - - self.time = str(config.get('time', '300')) - self.ramp = str(config.get('ramp', '0')) - self.startdelay = config.get('startdelay', None) - self.rate_iops = config.get('rate_iops', None) - self.iodepth = config.get('iodepth', 16) - self.numjobs = config.get('numjobs', 1) - self.mode = config.get('mode', 'write') - self.rwmixread = config.get('rwmixread', 50) + self.block_device_list = config.get("block_devices", "/dev/vdb") + self.block_devices = [d.strip() for d in self.block_device_list.split(",")] + self.concurrent_procs = config.get("concurrent_procs", len(self.block_devices)) + self.total_procs = self.concurrent_procs * len(settings.getnodes("clients").split(",")) + + self.time = str(config.get("time", "300")) + self.ramp = str(config.get("ramp", "0")) + self.startdelay = config.get("startdelay", None) + self.rate_iops = config.get("rate_iops", None) + self.iodepth = config.get("iodepth", 16) + self.numjobs = config.get("numjobs", 1) + self.mode = config.get("mode", "write") + self.rwmixread = config.get("rwmixread", 50) self.rwmixwrite = 100 - self.rwmixread - self.ioengine = config.get('ioengine', 'libaio') - self.op_size = config.get('op_size', 4194304) - self.pgs = config.get('pgs', 2048) - self.vol_size = config.get('vol_size', 65536) * 0.9 - self.rep_size = config.get('rep_size', 1) - self.rbdadd_mons = config.get('rbdadd_mons') - self.rbdadd_options = config.get('rbdadd_options') - self.client_ra = config.get('client_ra', '128') - self.fio_cmd = config.get('fio_cmd', '/usr/bin/fio') + self.ioengine = config.get("ioengine", "libaio") + self.op_size = config.get("op_size", 4194304) + self.pgs = config.get("pgs", 2048) + self.vol_size = config.get("vol_size", 65536) * 0.9 + self.rep_size = config.get("rep_size", 1) + self.rbdadd_mons = config.get("rbdadd_mons") + self.rbdadd_options = config.get("rbdadd_options") + self.client_ra = config.get("client_ra", "128") + self.fio_cmd = config.get("fio_cmd", "/usr/bin/fio") # FIXME there are too many permutations, need to put results in SQLITE3 - self.run_dir = '%sclient_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.run_dir, int(self.client_ra), int(self.op_size), int(self.total_procs), int(self.iodepth), self.mode) - self.out_dir = '%s/osd_ra-%08d/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.archive_dir, int(self.osd_ra), int(self.client_ra), int(self.op_size), int(self.total_procs), int(self.iodepth), self.mode) + self.run_dir = "%sclient_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s" % ( + self.run_dir, + int(self.client_ra), + int(self.op_size), + int(self.total_procs), + int(self.iodepth), + self.mode, + ) + self.out_dir = "%s/osd_ra-%08d/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s" % ( + self.archive_dir, + int(self.osd_ra), + int(self.client_ra), + int(self.op_size), + int(self.total_procs), + int(self.iodepth), + self.mode, + ) + + def estimate_duration(self) -> int: + """Estimate run-phase seconds: runtime + ramp time.""" + total = int(self.time) if self.time not in (None, "None") else 0 + total += int(self.ramp) if self.ramp not in (None, "None") else 0 + return total def exists(self): if os.path.exists(self.out_dir): - logger.info('Skipping existing test in %s.', self.out_dir) + logger.info("Skipping existing test in %s.", self.out_dir) return True return False def initialize(self): super(KvmRbdFio, self).initialize() - common.pdsh(settings.getnodes('clients', 'osds', 'mons', 'rgws'), - 'sudo rm -rf %s' % self.run_dir, - continue_if_error=False).communicate() + common.pdsh( + settings.getnodes("clients", "osds", "mons", "rgws"), + "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...') + clnts = settings.getnodes("clients") + logger.info("creating mountpoints...") for b in self.block_devices: bnm = os.path.basename(b) - mtpt = '/srv/rbdfio-`%s`-%s' % (common.get_fqdn_cmd(), bnm) - common.pdsh(clnts, 'sudo mkfs.ext4 %s' % b, - continue_if_error=False).communicate() - common.pdsh(clnts, 'sudo mkdir -p %s' % mtpt, - continue_if_error=False).communicate() - common.pdsh(clnts, 'sudo mount -t ext4 -o noatime %s %s' % (b, mtpt), - continue_if_error=False).communicate() - logger.info('Attempting to initialize fio files...') + mtpt = "/srv/rbdfio-`%s`-%s" % (common.get_fqdn_cmd(), bnm) + common.pdsh(clnts, "sudo mkfs.ext4 %s" % b, continue_if_error=False).communicate() + common.pdsh(clnts, "sudo mkdir -p %s" % mtpt, continue_if_error=False).communicate() + common.pdsh(clnts, "sudo mount -t ext4 -o noatime %s %s" % (b, mtpt), continue_if_error=False).communicate() + logger.info("Attempting to initialize fio files...") initializer_list = [] for i in range(self.concurrent_procs): b = self.block_devices[i % len(self.block_devices)] bnm = os.path.basename(b) - mtpt = '/srv/rbdfio-`hostname -s`-%s' % bnm - fiopath = os.path.join(mtpt, 'fio%d.img' % i) - pre_cmd = 'sudo %s --rw=write -ioengine=sync --bs=4M ' % self.fio_cmd - pre_cmd = '%s --size %dM --name=%s > /dev/null' % ( - pre_cmd, self.vol_size, fiopath) - initializer_list.append(common.pdsh(clnts, pre_cmd, - continue_if_error=False)) + mtpt = "/srv/rbdfio-`hostname -s`-%s" % bnm + fiopath = os.path.join(mtpt, "fio%d.img" % i) + pre_cmd = "sudo %s --rw=write -ioengine=sync --bs=4M " % self.fio_cmd + pre_cmd = "%s --size %dM --name=%s > /dev/null" % (pre_cmd, self.vol_size, fiopath) + initializer_list.append(common.pdsh(clnts, pre_cmd, continue_if_error=False)) for p in initializer_list: p.communicate() # Create the run directory - common.pdsh(clnts, 'rm -rf %s' % self.run_dir, - continue_if_error=False).communicate() + 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(KvmRbdFio, self).run() # Set client readahead - self.set_client_param('read_ahead_kb', self.client_ra) - clnts = settings.getnodes('clients') + self.set_client_param("read_ahead_kb", self.client_ra) + clnts = settings.getnodes("clients") # We'll always drop caches for rados bench self.dropcaches() @@ -98,69 +115,68 @@ def run(self): time.sleep(5) # Run the backfill testing thread if requested - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: recovery_callback = self.recovery_callback self.cluster.create_recovery_test(self.run_dir, recovery_callback) - logger.info('Starting rbd fio %s test.', self.mode) + logger.info("Starting rbd fio %s test.", self.mode) fio_process_list = [] for i in range(self.concurrent_procs): b = self.block_devices[i % len(self.block_devices)] bnm = os.path.basename(b) - mtpt = '/srv/rbdfio-`hostname -s`-%s' % bnm - fiopath = os.path.join(mtpt, 'fio%d.img' % i) - out_file = '%s/output.%d' % (self.run_dir, i) - fio_cmd = 'sudo %s' % self.fio_cmd - fio_cmd += ' --rw=%s' % self.mode - if (self.mode == 'readwrite' or self.mode == 'randrw'): - fio_cmd += ' --rwmixread=%s --rwmixwrite=%s' % (self.rwmixread, self.rwmixwrite) - fio_cmd += ' --ioengine=%s' % self.ioengine - fio_cmd += ' --runtime=%s' % self.time - fio_cmd += ' --ramp_time=%s' % self.ramp + mtpt = "/srv/rbdfio-`hostname -s`-%s" % bnm + fiopath = os.path.join(mtpt, "fio%d.img" % i) + out_file = "%s/output.%d" % (self.run_dir, i) + fio_cmd = "sudo %s" % self.fio_cmd + fio_cmd += " --rw=%s" % self.mode + if self.mode == "readwrite" or self.mode == "randrw": + fio_cmd += " --rwmixread=%s --rwmixwrite=%s" % (self.rwmixread, self.rwmixwrite) + fio_cmd += " --ioengine=%s" % self.ioengine + fio_cmd += " --runtime=%s" % self.time + fio_cmd += " --ramp_time=%s" % self.ramp if self.startdelay: - fio_cmd += ' --startdelay=%s' % self.startdelay + fio_cmd += " --startdelay=%s" % self.startdelay if self.rate_iops: - fio_cmd += ' --rate_iops=%s' % self.rate_iops - fio_cmd += ' --numjobs=%s' % self.numjobs - fio_cmd += ' --direct=1' - fio_cmd += ' --bs=%dB' % self.op_size - fio_cmd += ' --iodepth=%d' % self.iodepth - fio_cmd += ' --size=%dM' % self.vol_size + fio_cmd += " --rate_iops=%s" % self.rate_iops + fio_cmd += " --numjobs=%s" % self.numjobs + fio_cmd += " --direct=1" + 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 + fio_cmd += " --write_iops_log=%s" % out_file if self.log_bw: - fio_cmd += ' --write_bw_log=%s' % out_file + fio_cmd += " --write_bw_log=%s" % out_file if self.log_lat: - fio_cmd += ' --write_lat_log=%s' % out_file - if 'recovery_test' in self.cluster.config: - fio_cmd += ' --time_based' - fio_cmd += ' --name=%s > %s' % (fiopath, out_file) + fio_cmd += " --write_lat_log=%s" % out_file + if "recovery_test" in self.cluster.config: + fio_cmd += " --time_based" + fio_cmd += " --name=%s > %s" % (fiopath, out_file) fio_process_list.append(common.pdsh(clnts, fio_cmd, continue_if_error=False)) for p in fio_process_list: p.communicate() MonitoringFactory.stop(self.run_dir) - logger.info('Finished rbd fio test') + logger.info("Finished rbd fio test") - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def cleanup(self): super(KvmRbdFio, self).cleanup() - clnts = settings.getnodes('clients') - common.pdsh(clnts, 'killall fio').communicate() + clnts = settings.getnodes("clients") + common.pdsh(clnts, "killall fio").communicate() time.sleep(3) - common.pdsh(clnts, 'killall -9 fio').communicate() + common.pdsh(clnts, "killall -9 fio").communicate() time.sleep(3) - common.pdsh(clnts, 'rm -rf /srv/*/*', - continue_if_error=False).communicate() - common.pdsh(clnts, 'sudo umount /srv/* || echo -n').communicate() + common.pdsh(clnts, "rm -rf /srv/*/*", continue_if_error=False).communicate() + common.pdsh(clnts, "sudo umount /srv/* || echo -n").communicate() def set_client_param(self, param, value): cmd = 'find /sys/block/vd* ! -iname vda -exec sudo sh -c "echo %s > {}/queue/%s" \\;' % (value, param) - common.pdsh(settings.getnodes('clients'), cmd).communicate() + common.pdsh(settings.getnodes("clients"), cmd).communicate() def __str__(self): return "%s\n%s\n%s" % (self.run_dir, self.out_dir, super(KvmRbdFio, self).__str__()) def recovery_callback(self): - common.pdsh(settings.getnodes('clients'), 'sudo killall fio').communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall fio").communicate() diff --git a/benchmark/librbdfio.py b/benchmark/librbdfio.py index 9147be75..7e923f67 100644 --- a/benchmark/librbdfio.py +++ b/benchmark/librbdfio.py @@ -92,6 +92,20 @@ def __init__(self, archive_dir, cluster, config): rbd_name = f"cbt-rbdfio-`{common.get_fqdn_cmd()}`-file-{proc_num:d}" self.names += f"--name={rbd_name} " + def estimate_duration(self) -> int: + """Estimate run-phase seconds. + + When workloads are configured, delegates to + :meth:`~workloads.workloads.Workloads.estimate_duration` which accounts + for the number of parameter-set combinations across all workloads. + Falls back to plain ``time + ramp`` for non-workload runs. + """ + if self._workloads.exist(): + return self._workloads.estimate_duration() + total = int(self.time) if self.time is not None else 0 + total += int(self.ramp) if self.ramp is not None else 0 + return total + def exists(self): """ Verify whether the out_dir exists diff --git a/benchmark/radosbench.py b/benchmark/radosbench.py index 26f1582f..12b16ce3 100644 --- a/benchmark/radosbench.py +++ b/benchmark/radosbench.py @@ -19,28 +19,28 @@ def __init__(self, archive_dir, cluster, config): super(Radosbench, self).__init__(archive_dir, cluster, config) self.tmp_conf = self.cluster.tmp_conf - self.time = str(config.get('time', '300')) - self.concurrent_procs = config.get('concurrent_procs', 1) - self.concurrent_ops = config.get('concurrent_ops', 16) - self.pool_per_proc = config.get('pool_per_proc', False) # default behavior used to be True - self.write_only = config.get('write_only', False) - self.write_time = config.get('write_time', self.time) - self.read_only = config.get('read_only', False) - self.read_time = config.get('read_time', self.time) - self.op_size = config.get('op_size', 4194304) - self.object_set_id = config.get('object_set_id', '') - self.run_dir = os.path.join(self.run_dir, - 'op_size-{:0>8}'.format(self.op_size), - 'concurrent_ops-{:0>8}'.format(self.concurrent_ops)) + self.time = str(config.get("time", "300")) + self.concurrent_procs = config.get("concurrent_procs", 1) + self.concurrent_ops = config.get("concurrent_ops", 16) + self.pool_per_proc = config.get("pool_per_proc", False) # default behavior used to be True + self.write_only = config.get("write_only", False) + self.write_time = config.get("write_time", self.time) + self.read_only = config.get("read_only", False) + self.read_time = config.get("read_time", self.time) + self.op_size = config.get("op_size", 4194304) + self.object_set_id = config.get("object_set_id", "") + self.run_dir = os.path.join( + self.run_dir, "op_size-{:0>8}".format(self.op_size), "concurrent_ops-{:0>8}".format(self.concurrent_ops) + ) self.out_dir = self.archive_dir - self.pool_profile = config.get('pool_profile', 'default') - self.cmd_path = config.get('cmd_path', self.cluster.rados_cmd) - self.pool = config.get('target_pool', 'rados-bench-cbt') - self.readmode = config.get('readmode', 'seq') - self.max_objects = config.get('max_objects', None) - self.write_omap = config.get('write_omap', False) - self.prefill_time = config.get('prefill_time', None) - self.prefill_objects = config.get('prefill_objects', None) + self.pool_profile = config.get("pool_profile", "default") + self.cmd_path = config.get("cmd_path", self.cluster.rados_cmd) + self.pool = config.get("target_pool", "rados-bench-cbt") + self.readmode = config.get("readmode", "seq") + self.max_objects = config.get("max_objects", None) + self.write_omap = config.get("write_omap", False) + self.prefill_time = config.get("prefill_time", None) + self.prefill_objects = config.get("prefill_objects", None) def create_data_analyzer(self, run, host, proc): return RadosBenchAnalyzer(self.out_dir, run, host, proc) @@ -48,28 +48,41 @@ def create_data_analyzer(self, run, host, proc): def exists(self, expect_exists=False): if os.path.exists(self.out_dir): if not expect_exists: - logger.info('Skipping existing test in %s.', self.out_dir) + logger.info("Skipping existing test in %s.", self.out_dir) return True else: if expect_exists: - logger.info('test result does not exist in %s.', self.out_dir) + logger.info("test result does not exist in %s.", self.out_dir) return False + def estimate_duration(self) -> int: + """Estimate total run-phase seconds: prefill + write + read phases.""" + total = 0 + if self.prefill_time or self.prefill_objects: + total += int(self.prefill_time or self.time) + if not self.read_only: + total += int(self.write_time) + if not self.write_only: + total += int(self.read_time) + return total + # Initialize may only be called once depending on rebuild_every_test setting def initialize(self): super(Radosbench, self).initialize() - logger.info('Pausing for 60s for idle monitoring.') + logger.info("Pausing for 60s for idle monitoring.") with MonitoringFactory.monitor("%s/idle_monitoring" % self.run_dir): time.sleep(60) - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def get_rados_version(self): - stdout, _ = common.pdsh(settings.getnodes('head'), '%s -c %s -v' % (self.cmd_path, self.tmp_conf)).communicate() - m = (re.findall(r"version (\d+)(?:.\d+)* \([0-9a-f]+\)", stdout) or - re.findall(r"version v(\d+)(?:.\d+)* \([0-9a-f]+\)", stdout) or - (255, 0)) + stdout, _ = common.pdsh(settings.getnodes("head"), "%s -c %s -v" % (self.cmd_path, self.tmp_conf)).communicate() + m = ( + re.findall(r"version (\d+)(?:.\d+)* \([0-9a-f]+\)", stdout) + or re.findall(r"version v(\d+)(?:.\d+)* \([0-9a-f]+\)", stdout) + or (255, 0) + ) return int(m[0]) def run(self): @@ -78,12 +91,13 @@ def run(self): do_prefill = self.prefill_time or self.prefill_objects # sanity tests if self.read_only and self.write_only: - logger.error('Both "read_only" and "write_only" are specified, ' - 'but they are mutually exclusive.') + logger.error('Both "read_only" and "write_only" are specified, ' "but they are mutually exclusive.") return elif self.read_only and not do_prefill: - logger.error('Please prefill the testbench with "prefill_time" and/or ' - '"prefill_objects" option for a "read_only" test') + logger.error( + 'Please prefill the testbench with "prefill_time" and/or ' + '"prefill_objects" option for a "read_only" test' + ) return # Remake the pools @@ -91,52 +105,60 @@ def run(self): # Run prefill if do_prefill: - self._run(mode='prefill', run_dir='prefill', out_dir='prefill', - max_objects=self.prefill_objects, - runtime=self.prefill_time or self.time) + self._run( + mode="prefill", + run_dir="prefill", + out_dir="prefill", + max_objects=self.prefill_objects, + runtime=self.prefill_time or self.time, + ) # Run write test if not self.read_only: - self._run(mode='write', run_dir='write', out_dir='write', - max_objects=self.max_objects, - runtime=self.write_time) + self._run( + mode="write", run_dir="write", out_dir="write", max_objects=self.max_objects, runtime=self.write_time + ) # Run read test unless write_only if not self.write_only: - self._run(mode=self.readmode, run_dir=self.readmode, out_dir=self.readmode, - max_objects=None, - runtime=self.read_time) + self._run( + mode=self.readmode, + run_dir=self.readmode, + out_dir=self.readmode, + max_objects=None, + runtime=self.read_time, + ) def _run(self, mode, run_dir, out_dir, max_objects, runtime): # We'll always drop caches for rados bench self.dropcaches() if self.concurrent_ops: - concurrent_ops_str = '--concurrent-ios %s' % self.concurrent_ops + concurrent_ops_str = "--concurrent-ios %s" % self.concurrent_ops rados_version = self.get_rados_version() # Max Objects - max_objects_str = '' + max_objects_str = "" if max_objects: if rados_version < 10: - raise ValueError('max_objects not supported by rados_version < 10') - max_objects_str = '--max-objects %s' % max_objects + raise ValueError("max_objects not supported by rados_version < 10") + max_objects_str = "--max-objects %s" % max_objects # Operation type op_type = mode - if mode == 'prefill': - op_type = 'write' + if mode == "prefill": + op_type = "write" - if op_type == 'write': - op_size_str = '-b %s' % self.op_size + if op_type == "write": + op_size_str = "-b %s" % self.op_size else: - op_size_str = '' + op_size_str = "" # Write to OMAP - write_omap_str = '' + write_omap_str = "" if self.write_omap: if rados_version < 10: - raise ValueError('write_omap not supported by rados_version < 10') - write_omap_str = '--write-omap' + raise ValueError("write_omap not supported by rados_version < 10") + write_omap_str = "--write-omap" run_dir = os.path.join(self.run_dir, run_dir) common.make_remote_dir(run_dir) @@ -145,35 +167,35 @@ def _run(self, mode, run_dir, out_dir, max_objects, runtime): self.cluster.dump_config(run_dir) # Run the backfill testing thread if requested (but not for prefill) - if mode != 'prefill' and 'recovery_test' in self.cluster.config: + if mode != "prefill" and "recovery_test" in self.cluster.config: recovery_callback = self.recovery_callback self.cluster.create_recovery_test(run_dir, recovery_callback) # Run rados bench with MonitoringFactory.monitor(run_dir) as monitor: - logger.info('Running radosbench %s test.' % mode) + logger.info("Running radosbench %s test." % mode) ps = [] for i in range(self.concurrent_procs): - out_file = '%s/output.%s' % (run_dir, i) - objecter_log = '%s/objecter.%s.log' % (run_dir, i) + out_file = "%s/output.%s" % (run_dir, i) + objecter_log = "%s/objecter.%s.log" % (run_dir, i) if self.pool_per_proc: # support previous behavior of 1 storage pool per rados process - pool_name_cmd = 'rados-bench-`{fqdn_cmd}`-{i}' + pool_name_cmd = "rados-bench-`{fqdn_cmd}`-{i}" pool_name = pool_name_cmd.format(fqdn_cmd=common.get_fqdn_cmd(), i=i) - run_name = '' + run_name = "" else: # default behavior is to use a single storage pool pool_name = self.pool - run_name_fmt = '--run-name {object_set_id} `{fqdn_cmd}`-{i}' + run_name_fmt = "--run-name {object_set_id} `{fqdn_cmd}`-{i}" run_name = run_name_fmt.format( - object_set_id=self.object_set_id, - fqdn_cmd=common.get_fqdn_cmd(), - i=i) - rados_bench_cmd_fmt = \ - '{cmd} -c {conf} -p {pool} bench {op_size_arg} {duration} ' \ - '{op_type} {concurrent_ops_arg} {max_objects_arg} ' \ - '{write_omap_arg} {run_name} --no-cleanup ' \ - '2> {stderr} > {stdout}' + object_set_id=self.object_set_id, fqdn_cmd=common.get_fqdn_cmd(), i=i + ) + rados_bench_cmd_fmt = ( + "{cmd} -c {conf} -p {pool} bench {op_size_arg} {duration} " + "{op_type} {concurrent_ops_arg} {max_objects_arg} " + "{write_omap_arg} {run_name} --no-cleanup " + "2> {stderr} > {stdout}" + ) rados_bench_cmd = rados_bench_cmd_fmt.format( cmd=self.cmd_path_full, conf=self.tmp_conf, @@ -186,51 +208,51 @@ def _run(self, mode, run_dir, out_dir, max_objects, runtime): write_omap_arg=write_omap_str, run_name=run_name, stderr=objecter_log, - stdout=out_file) - p = common.pdsh(settings.getnodes('clients'), rados_bench_cmd) + stdout=out_file, + ) + p = common.pdsh(settings.getnodes("clients"), rados_bench_cmd) ps.append(p) for p in ps: p.wait() # If we were doing recovery, wait until it's done (but not for prefill). - if mode != 'prefill' and 'recovery_test' in self.cluster.config: + if mode != "prefill" and "recovery_test" in self.cluster.config: self.cluster.wait_recovery_done() # Finally, get the historic ops self.cluster.dump_historic_ops(run_dir) out_dir = os.path.join(self.out_dir, out_dir) - common.sync_files('%s/*' % run_dir, out_dir) + common.sync_files("%s/*" % run_dir, out_dir) self.analyze(out_dir) def mkpools(self): with MonitoringFactory.monitor("%s/pool_monitoring" % self.run_dir): if self.pool_per_proc: # allow use of a separate storage pool per process for i in range(self.concurrent_procs): - for node in settings.getnodes('clients').split(','): + for node in settings.getnodes("clients").split(","): node = node.rpartition("@")[2] - self.cluster.rmpool('rados-bench-%s-%s' % (node, i), self.pool_profile) - self.cluster.mkpool('rados-bench-%s-%s' % (node, i), self.pool_profile, 'radosbench') + self.cluster.rmpool("rados-bench-%s-%s" % (node, i), self.pool_profile) + self.cluster.mkpool("rados-bench-%s-%s" % (node, i), self.pool_profile, "radosbench") else: # the default behavior is to use a single Ceph storage pool for all rados bench processes - self.cluster.rmpool('rados-bench-cbt', self.pool_profile) - self.cluster.mkpool('rados-bench-cbt', self.pool_profile, 'radosbench') - + self.cluster.rmpool("rados-bench-cbt", self.pool_profile) + self.cluster.mkpool("rados-bench-cbt", self.pool_profile, "radosbench") def cleanup(self): cmd_name = pathlib.PurePath(self.cmd_path).name - common.pdsh(settings.getnodes('clients'), 'sudo killall -9 %s' % cmd_name).communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall -9 %s" % cmd_name).communicate() def recovery_callback(self): - cleanup(); + cleanup() def parse(self, out_dir): - for client in settings.getnodes('clients').split(','): + for client in settings.getnodes("clients").split(","): host = settings.host_info(client)["host"] for i in range(self.concurrent_procs): result = {} found = 0 - out_file = '%s/output.%s.%s' % (out_dir, i, host) - json_out_file = '%s/json_output.%s.%s' % (out_dir, i, host) + out_file = "%s/output.%s.%s" % (out_dir, i, host) + json_out_file = "%s/json_output.%s.%s" % (out_dir, i, host) with open(out_file) as fd: for line in fd.readlines(): if found == 0: @@ -240,11 +262,11 @@ def parse(self, out_dir): line = line.strip() key, val = line.split(":") result[key.strip()] = val.strip() - with open(json_out_file, 'w') as json_fd: + with open(json_out_file, "w") as json_fd: json.dump(result, json_fd) def analyze(self, out_dir): - logger.info('Convert results to json format.') + logger.info("Convert results to json format.") self.parse(out_dir) def __str__(self): @@ -255,7 +277,7 @@ class RadosBenchAnalyzer(DataAnalyzer): def __init__(self, archive_dir, run, host, proc): super().__init__(archive_dir, run, host, proc) self.out_dir = os.path.join(self.archive_dir, run) - self.radosbench_out_fname = os.path.join(self.out_dir, f'json_output.{proc}.{host}') + self.radosbench_out_fname = os.path.join(self.out_dir, f"json_output.{proc}.{host}") self.radosbench_json_output = json.load(open(self.radosbench_out_fname)) def get_total_ops(self): diff --git a/benchmark/rawfio.py b/benchmark/rawfio.py index 85b818cb..53390e61 100644 --- a/benchmark/rawfio.py +++ b/benchmark/rawfio.py @@ -14,28 +14,42 @@ class RawFio(Benchmark): def __init__(self, archive_dir, cluster, config): super(RawFio, self).__init__(archive_dir, cluster, config) # 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(',')] - self.concurrent_procs = config.get('concurrent_procs', len(self.block_devices)) - self.total_procs = self.concurrent_procs * len(settings.getnodes('clients').split(',')) + self.block_device_list = config.get("block_devices", "/dev/vdb") + self.block_devices = [d.strip() for d in self.block_device_list.split(",")] + 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" - self.time = str(config.get('time', '300')) - self.ramp = str(config.get('ramp', '0')) - self.startdelay = config.get('startdelay', None) - self.rate_iops = config.get('rate_iops', None) - self.iodepth = config.get('iodepth', 16) - self.direct = config.get('direct', 1) - self.numjobs = config.get('numjobs', 1) - self.mode = config.get('mode', 'write') - self.rwmixread = config.get('rwmixread', 50) + self.time = str(config.get("time", "300")) + self.ramp = str(config.get("ramp", "0")) + self.startdelay = config.get("startdelay", None) + self.rate_iops = config.get("rate_iops", None) + self.iodepth = config.get("iodepth", 16) + self.direct = config.get("direct", 1) + self.numjobs = config.get("numjobs", 1) + self.mode = config.get("mode", "write") + self.rwmixread = config.get("rwmixread", 50) 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.fio_cmd = config.get('fio_cmd', 'sudo /usr/bin/fio') + 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.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) + 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): @@ -43,37 +57,45 @@ def __init__(self, archive_dir, cluster, config): # return True # return False + def estimate_duration(self) -> int: + """Estimate run-phase seconds: runtime + ramp time.""" + total = int(self.time) if self.time not in (None, "None") else 0 + total += int(self.ramp) if self.ramp not in (None, "None") else 0 + return total + 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.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...') + clnts = settings.getnodes("clients") + logger.info("creating mountpoints...") - logger.info('Attempting to initialize fio files...') + logger.info("Attempting to initialize fio files...") initializer_list = [] 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) - initializer_list.append(common.pdsh(clnts, pre_cmd, - continue_if_error=False)) + 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, + ) + initializer_list.append(common.pdsh(clnts, pre_cmd, continue_if_error=False)) for p in initializer_list: p.communicate() # Create the run directory - common.pdsh(clnts, 'rm -rf %s' % self.run_dir, - continue_if_error=False).communicate() + 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') + clnts = settings.getnodes("clients") # We'll always drop caches for rados bench self.dropcaches() @@ -82,63 +104,63 @@ def run(self): time.sleep(5) - logger.info('Starting raw fio %s test.', self.mode) + logger.info("Starting raw fio %s test.", self.mode) fio_process_list = [] for i in range(self.concurrent_procs): b = self.block_devices[i % len(self.block_devices)] fiopath = b - out_file = '%s/output.%d' % (self.run_dir, i) - fio_cmd = 'sudo %s' % self.fio_cmd - fio_cmd += ' --rw=%s' % self.mode - if (self.mode == 'readwrite' or self.mode == 'randrw'): - fio_cmd += ' --rwmixread=%s --rwmixwrite=%s' % (self.rwmixread, self.rwmixwrite) - fio_cmd += ' --ioengine=%s' % self.ioengine - fio_cmd += ' --runtime=%s' % self.time - fio_cmd += ' --ramp_time=%s' % self.ramp + out_file = "%s/output.%d" % (self.run_dir, i) + fio_cmd = "sudo %s" % self.fio_cmd + fio_cmd += " --rw=%s" % self.mode + if self.mode == "readwrite" or self.mode == "randrw": + fio_cmd += " --rwmixread=%s --rwmixwrite=%s" % (self.rwmixread, self.rwmixwrite) + fio_cmd += " --ioengine=%s" % self.ioengine + fio_cmd += " --runtime=%s" % self.time + fio_cmd += " --ramp_time=%s" % self.ramp if self.startdelay: - fio_cmd += ' --startdelay=%s' % self.startdelay + fio_cmd += " --startdelay=%s" % self.startdelay if self.rate_iops: - fio_cmd += ' --rate_iops=%s' % self.rate_iops - fio_cmd += ' --numjobs=%s' % self.numjobs - 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 + fio_cmd += " --rate_iops=%s" % self.rate_iops + fio_cmd += " --numjobs=%s" % self.numjobs + 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 + fio_cmd += " --write_iops_log=%s" % out_file if self.log_bw: - fio_cmd += ' --write_bw_log=%s' % out_file + 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 += ' --name=%s > %s' % (fiopath, out_file) + 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 += " --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)) for p in fio_process_list: p.communicate() MonitoringFactory.stop(self.run_dir) - logger.info('Finished raw fio test') + logger.info("Finished raw fio test") - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def cleanup(self): super(RawFio, self).cleanup() - clnts = settings.getnodes('clients') + clnts = settings.getnodes("clients") logger.debug("Kill fio: %s" % clnts) - common.pdsh(clnts, 'killall fio').communicate() + common.pdsh(clnts, "killall fio").communicate() time.sleep(3) - common.pdsh(clnts, 'killall -9 fio').communicate() + common.pdsh(clnts, "killall -9 fio").communicate() def set_client_param(self, param, value): cmd = 'find /sys/block/vd* ! -iname vda -exec sudo sh -c "echo %s > {}/queue/%s" \;' % (value, param) - common.pdsh(settings.getnodes('clients'), cmd).communicate() + common.pdsh(settings.getnodes("clients"), cmd).communicate() def __str__(self): return "%s\n%s\n%s" % (self.run_dir, self.out_dir, super(RawFio, self).__str__()) def recovery_callback(self): - common.pdsh(settings.getnodes('clients'), 'sudo killall fio').communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall fio").communicate() diff --git a/benchmark/rbdfio.py b/benchmark/rbdfio.py index 8321c641..e22752f4 100644 --- a/benchmark/rbdfio.py +++ b/benchmark/rbdfio.py @@ -16,54 +16,75 @@ def __init__(self, archive_dir, cluster, config): super(RbdFio, self).__init__(archive_dir, cluster, config) # FIXME there are too many permutations, need to put results in SQLITE3 - self.cmd_path = config.get('cmd_path', '/usr/bin/fio') - self.pool_profile = config.get('pool_profile', 'default') - - self.concurrent_procs = config.get('concurrent_procs', 1) - self.total_procs = self.concurrent_procs * len(settings.getnodes('clients').split(',')) - self.time = str(config.get('time', None)) - self.ramp = str(config.get('ramp', None)) - self.iodepth = config.get('iodepth', 16) - self.numjobs = config.get('numjobs', 1) - self.end_fsync = str(config.get('end_fsync', 0)) - self.mode = config.get('mode', 'write') - self.rwmixread = config.get('rwmixread', 50) + self.cmd_path = config.get("cmd_path", "/usr/bin/fio") + self.pool_profile = config.get("pool_profile", "default") + + self.concurrent_procs = config.get("concurrent_procs", 1) + self.total_procs = self.concurrent_procs * len(settings.getnodes("clients").split(",")) + self.time = str(config.get("time", None)) + self.ramp = str(config.get("ramp", None)) + self.iodepth = config.get("iodepth", 16) + self.numjobs = config.get("numjobs", 1) + self.end_fsync = str(config.get("end_fsync", 0)) + self.mode = config.get("mode", "write") + self.rwmixread = config.get("rwmixread", 50) self.rwmixwrite = 100 - self.rwmixread - self.log_avg_msec = config.get('log_avg_msec', None) - self.ioengine = config.get('ioengine', 'libaio') - self.op_size = config.get('op_size', 4194304) - self.vol_size = config.get('vol_size', 65536) - self.vol_object_size = config.get('vol_object_size', '4M') - self.random_distribution = config.get('random_distribution', None) - self.rbdadd_mons = config.get('rbdadd_mons') - self.rbdadd_options = config.get('rbdadd_options', 'share') - self.client_ra = config.get('client_ra', 128) - self.direct = config.get('direct', 1) + self.log_avg_msec = config.get("log_avg_msec", None) + self.ioengine = config.get("ioengine", "libaio") + self.op_size = config.get("op_size", 4194304) + self.vol_size = config.get("vol_size", 65536) + self.vol_object_size = config.get("vol_object_size", "4M") + self.random_distribution = config.get("random_distribution", None) + self.rbdadd_mons = config.get("rbdadd_mons") + self.rbdadd_options = config.get("rbdadd_options", "share") + self.client_ra = config.get("client_ra", 128) + self.direct = config.get("direct", 1) self.poolname = "cbt-kernelrbdfio" - self.run_dir = '%srbdfio/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.run_dir, int(self.client_ra), int(self.op_size), int(self.concurrent_procs), int(self.iodepth), self.mode) - self.out_dir = '%s/rbdfio/osd_ra-%08d/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s' % (self.archive_dir, int(self.osd_ra), int(self.client_ra), int(self.op_size), int(self.concurrent_procs), int(self.iodepth), self.mode) + self.run_dir = "%srbdfio/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s" % ( + self.run_dir, + int(self.client_ra), + int(self.op_size), + int(self.concurrent_procs), + int(self.iodepth), + self.mode, + ) + self.out_dir = "%s/rbdfio/osd_ra-%08d/client_ra-%08d/op_size-%08d/concurrent_procs-%03d/iodepth-%03d/%s" % ( + self.archive_dir, + int(self.osd_ra), + int(self.client_ra), + int(self.op_size), + int(self.concurrent_procs), + int(self.iodepth), + self.mode, + ) # Make the file names string - self.names = '' + self.names = "" for i in range(self.concurrent_procs): - self.names += '--name=%s/cbt-kernelrbdfio-`hostname -s`/cbt-kernelrbdfio-%d ' % (self.cluster.mnt_dir, i) + self.names += "--name=%s/cbt-kernelrbdfio-`hostname -s`/cbt-kernelrbdfio-%d " % (self.cluster.mnt_dir, i) + + def estimate_duration(self) -> int: + """Estimate run-phase seconds: runtime + ramp time.""" + total = int(self.time) if self.time not in (None, "None") else 0 + total += int(self.ramp) if self.ramp not in (None, "None") else 0 + return total def exists(self): if os.path.exists(self.out_dir): - logger.info('Skipping existing test in %s.', self.out_dir) + logger.info("Skipping existing test in %s.", self.out_dir) return True return False def initialize(self): super(RbdFio, self).initialize() - logger.info('Pausing for 60s for idle monitoring.') + logger.info("Pausing for 60s for idle monitoring.") MonitoringFactory.start("%s/idle_monitoring" % self.run_dir) time.sleep(60) MonitoringFactory.stop() - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) self.mkimages() @@ -71,16 +92,22 @@ def initialize(self): common.make_remote_dir(self.run_dir) # populate the fio files - logger.info('Attempting to populating fio files...') + logger.info("Attempting to populating fio files...") size = self.vol_size * 0.9 / self.concurrent_procs - pre_cmd = 'sudo %s --ioengine=%s --rw=write --numjobs=%s --bs=4M --size %dM %s > /dev/null' % (self.cmd_path, self.ioengine, self.numjobs, size, self.names) - common.pdsh(settings.getnodes('clients'), pre_cmd).communicate() + pre_cmd = "sudo %s --ioengine=%s --rw=write --numjobs=%s --bs=4M --size %dM %s > /dev/null" % ( + self.cmd_path, + self.ioengine, + self.numjobs, + size, + self.names, + ) + common.pdsh(settings.getnodes("clients"), pre_cmd).communicate() def run(self): super(RbdFio, self).run() # Set client readahead - self.set_client_param('read_ahead_kb', self.client_ra) + self.set_client_param("read_ahead_kb", self.client_ra) # We'll always drop caches for rados bench self.dropcaches() @@ -88,58 +115,61 @@ def run(self): MonitoringFactory.start(self.run_dir) # Run the backfill testing thread if requested - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: recovery_callback = self.recovery_callback self.cluster.create_recovery_test(self.run_dir, recovery_callback) time.sleep(5) - out_file = '%s/output' % self.run_dir - fio_cmd = 'sudo %s' % (self.cmd_path_full) - fio_cmd += ' --rw=%s' % self.mode - if (self.mode == 'readwrite' or self.mode == 'randrw'): - fio_cmd += ' --rwmixread=%s --rwmixwrite=%s' % (self.rwmixread, self.rwmixwrite) - fio_cmd += ' --ioengine=%s' % self.ioengine + out_file = "%s/output" % self.run_dir + fio_cmd = "sudo %s" % (self.cmd_path_full) + fio_cmd += " --rw=%s" % self.mode + if self.mode == "readwrite" or self.mode == "randrw": + fio_cmd += " --rwmixread=%s --rwmixwrite=%s" % (self.rwmixread, self.rwmixwrite) + fio_cmd += " --ioengine=%s" % self.ioengine if self.time is not None: - fio_cmd += ' --runtime=%s' % self.time + fio_cmd += " --runtime=%s" % self.time if self.ramp is not None: - fio_cmd += ' --ramp_time=%s' % self.ramp - fio_cmd += ' --numjobs=%s' % self.numjobs - fio_cmd += ' --direct=%s' % self.direct - fio_cmd += ' --bs=%dB' % self.op_size - fio_cmd += ' --iodepth=%d' % self.iodepth + fio_cmd += " --ramp_time=%s" % self.ramp + fio_cmd += " --numjobs=%s" % self.numjobs + fio_cmd += " --direct=%s" % self.direct + fio_cmd += " --bs=%dB" % self.op_size + fio_cmd += " --iodepth=%d" % self.iodepth if self.vol_size: - fio_cmd += ' --size=%dM' % (int(self.vol_size) * 0.9) + fio_cmd += " --size=%dM" % (int(self.vol_size) * 0.9) if self.log_iops: - fio_cmd += ' --write_iops_log=%s' % out_file + fio_cmd += " --write_iops_log=%s" % out_file if self.log_bw: - fio_cmd += ' --write_bw_log=%s' % out_file + fio_cmd += " --write_bw_log=%s" % out_file if self.log_lat: - fio_cmd += ' --write_lat_log=%s' % out_file - if 'recovery_test' in self.cluster.config: - fio_cmd += ' --time_based' + fio_cmd += " --write_lat_log=%s" % out_file + if "recovery_test" in self.cluster.config: + fio_cmd += " --time_based" if self.random_distribution is not None: - fio_cmd += ' --random_distribution=%s' % self.random_distribution - fio_cmd += ' %s > %s' % (self.names, out_file) + fio_cmd += " --random_distribution=%s" % self.random_distribution + fio_cmd += " %s > %s" % (self.names, out_file) if self.log_avg_msec is not None: - fio_cmd += ' --log_avg_msec=%s' % self.log_avg_msec - logger.info('Running rbd fio %s test.', self.mode) - common.pdsh(settings.getnodes('clients'), fio_cmd).communicate() + fio_cmd += " --log_avg_msec=%s" % self.log_avg_msec + logger.info("Running rbd fio %s test.", self.mode) + common.pdsh(settings.getnodes("clients"), fio_cmd).communicate() # If we were doing recovery, wait until it's done. - if 'recovery_test' in self.cluster.config: + if "recovery_test" in self.cluster.config: self.cluster.wait_recovery_done() MonitoringFactory.stop(self.run_dir) # Finally, get the historic ops self.cluster.dump_historic_ops(self.run_dir) - common.sync_files('%s/*' % self.run_dir, self.out_dir) + common.sync_files("%s/*" % self.run_dir, self.out_dir) def cleanup(self): super(RbdFio, self).cleanup() def set_client_param(self, param, value): - common.pdsh(settings.getnodes('clients'), 'find /sys/block/rbd* -exec sudo sh -c "echo %s > {}/queue/%s" \;' % (value, param)).communicate() + common.pdsh( + settings.getnodes("clients"), + 'find /sys/block/rbd* -exec sudo sh -c "echo %s > {}/queue/%s" \;' % (value, param), + ).communicate() def __str__(self): return "%s\n%s\n%s" % (self.run_dir, self.out_dir, super(RbdFio, self).__str__()) @@ -147,13 +177,28 @@ def __str__(self): def mkimages(self): MonitoringFactory.start("%s/pool_monitoring" % self.run_dir) self.cluster.rmpool(self.poolname, self.pool_profile) - self.cluster.mkpool(self.poolname, self.pool_profile, 'rbd') - common.pdsh(settings.getnodes('clients'), '/usr/bin/rbd create cbt-kernelrbdfio-`hostname -s` --size %s --pool %s' % (self.vol_size, self.poolname)).communicate() - common.pdsh(settings.getnodes('clients'), 'sudo rbd map cbt-kernelrbdfio-`hostname -s` --pool %s --id admin' % self.poolname).communicate() - common.pdsh(settings.getnodes('clients'), 'sudo mkfs.xfs /dev/rbd/cbt-kernelrbdfio/cbt-kernelrbdfio-`hostname -s`').communicate() - common.pdsh(settings.getnodes('clients'), 'sudo mkdir -p -m0755 -- %s/cbt-kernelrbdfio-`hostname -s`' % self.cluster.mnt_dir).communicate() - common.pdsh(settings.getnodes('clients'), 'sudo mount -t xfs -o noatime,inode64 /dev/rbd/cbt-kernelrbdfio/cbt-kernelrbdfio-`hostname -s` %s/cbt-kernelrbdfio-`hostname -s`' % self.cluster.mnt_dir).communicate() + self.cluster.mkpool(self.poolname, self.pool_profile, "rbd") + common.pdsh( + settings.getnodes("clients"), + "/usr/bin/rbd create cbt-kernelrbdfio-`hostname -s` --size %s --pool %s" % (self.vol_size, self.poolname), + ).communicate() + common.pdsh( + settings.getnodes("clients"), + "sudo rbd map cbt-kernelrbdfio-`hostname -s` --pool %s --id admin" % self.poolname, + ).communicate() + common.pdsh( + settings.getnodes("clients"), "sudo mkfs.xfs /dev/rbd/cbt-kernelrbdfio/cbt-kernelrbdfio-`hostname -s`" + ).communicate() + common.pdsh( + settings.getnodes("clients"), + "sudo mkdir -p -m0755 -- %s/cbt-kernelrbdfio-`hostname -s`" % self.cluster.mnt_dir, + ).communicate() + common.pdsh( + settings.getnodes("clients"), + "sudo mount -t xfs -o noatime,inode64 /dev/rbd/cbt-kernelrbdfio/cbt-kernelrbdfio-`hostname -s` %s/cbt-kernelrbdfio-`hostname -s`" + % self.cluster.mnt_dir, + ).communicate() MonitoringFactory.stop() def recovery_callback(self): - common.pdsh(settings.getnodes('clients'), 'sudo killall -9 fio').communicate() + common.pdsh(settings.getnodes("clients"), "sudo killall -9 fio").communicate() diff --git a/cbt.py b/cbt.py index 39983d84..60bfbda9 100755 --- a/cbt.py +++ b/cbt.py @@ -6,7 +6,9 @@ import sys import benchmarkfactory +import progress import settings +from benchmark.benchmark import Benchmark from cluster.ceph import Ceph from logging_configuration import setup_loggers @@ -33,6 +35,13 @@ def parse_args(args): help="The ceph.conf file to use.", ) + parser.add_argument( + "--no-progress", + action="store_true", + default=False, + help="Disable CLI progress bars (implied when stdout is not a TTY or is_teuthology is set).", + ) + parser.add_argument( "config_file", help="YAML config file.", @@ -41,6 +50,31 @@ def parse_args(args): return parser.parse_args(args[1:]) +def _compute_total_duration(benchmarks_list: list[Benchmark], rebuild_every_test: bool) -> int: + """Return the estimated total wall-clock seconds for the whole CBT run. + + Includes: + * cluster.initialize() cost (once, or once per benchmark when rebuild_every_test) + * per-benchmark estimate_duration() for all non-skipped benchmarks + + Args: + benchmarks_list: Pre-materialised list of Benchmark objects. + rebuild_every_test: Whether the cluster is rebuilt for each benchmark. + + Returns: + Estimated run time in seconds. + """ + cluster_init_secs = progress.cluster_init_estimate() + total = cluster_init_secs # at least one cluster init + + for b in benchmarks_list: + if rebuild_every_test: + total += cluster_init_secs + total += b.estimate_duration() + + return max(total, 1) + + def main(argv): # Set up console-only logging early so any startup errors are visible setup_loggers() @@ -51,6 +85,9 @@ def main(argv): archive_dir = settings.cluster.get("archive_dir") setup_loggers(logfile_name=f"{archive_dir}/cbt.log") + # Initialise progress reporting (after settings so is_teuthology is readable) + progress.setup(no_progress=ctx.no_progress) + logger.debug("Settings.cluster:\n %s", pprint.pformat(settings.cluster).replace("\n", "\n ")) global_init = collections.OrderedDict() @@ -59,45 +96,80 @@ def main(argv): # FIXME: Create ClusterFactory and parametrically match benchmarks and clusters. cluster = Ceph(settings.cluster) - # Only initialize and prefill upfront if we aren't rebuilding for each test. - if not rebuild_every_test: - if not cluster.use_existing: - cluster.initialize() - # Why does it need to iterate for the creation of benchmarks? - for iteration in range(settings.cluster.get("iterations", 0)): - benchmarks = benchmarkfactory.get_all(archive_dir, cluster, iteration) - for b in benchmarks: - if b.exists(): - continue - if b.getclass() not in global_init: - b.initialize() - b.initialize_endpoints() - b.prefill() - b.cleanup() - # Only initialize once per class. - global_init[b.getclass()] = b - - # logger.debug("Settings.cluster.is_teuthology:%s",settings.cluster.get('is_teuthology', False)) - # Run the benchmarks - return_code = 0 - try: - for iteration in range(settings.cluster.get("iterations", 0)): - benchmarks = benchmarkfactory.get_all(archive_dir, cluster, iteration) - for b in benchmarks: - if not b.exists() and not settings.cluster.get("is_teuthology", False): - continue - - if rebuild_every_test: + # Materialise the first-iteration benchmarks into a list so we can iterate + # it multiple times (duration estimate + init + run loops). + # benchmarkfactory.get_all() is a generator — wrapping in list() here + # prevents it from being exhausted by _compute_total_duration. + first_iter_benchmarks: list[Benchmark] = list(benchmarkfactory.get_all(archive_dir, cluster, 0)) + iterations = settings.cluster.get("iterations", 0) + # Scale estimated duration across all iterations + total_secs = _compute_total_duration(first_iter_benchmarks, rebuild_every_test) * max(iterations, 1) + + with progress.overall_bar(total_secs) as overall: + # Only initialize and prefill upfront if we aren't rebuilding for each test. + if not rebuild_every_test: + if not cluster.use_existing: + with progress.phase_bar("Cluster initialize", progress.cluster_init_estimate(), overall): cluster.initialize() - b.initialize() - # Always try to initialize endpoints before running the test - b.initialize_endpoints() - logger.info("Running benchmark %s == iteration %d ==", b, iteration) - b.run() - logger.info("Benchmark %s iteration %d complete.", b, iteration) - except: - return_code = 1 # FAIL - logger.exception("During tests") + for iteration in range(iterations): + benchmarks = ( + first_iter_benchmarks + if iteration == 0 + else list(benchmarkfactory.get_all(archive_dir, cluster, iteration)) + ) + for b in benchmarks: + if b.exists(): + continue + if b.getclass() not in global_init: + with progress.phase_bar( + f"Initialize {b.getclass()}", + overall=overall, + ): + b.initialize() + b.initialize_endpoints() + b.prefill() + b.cleanup() + # Only initialize once per class. + global_init[b.getclass()] = b + + # logger.debug("Settings.cluster.is_teuthology:%s",settings.cluster.get('is_teuthology', False)) + # Run the benchmarks + return_code = 0 + try: + for iteration in range(iterations): + benchmarks = ( + first_iter_benchmarks + if iteration == 0 + else list(benchmarkfactory.get_all(archive_dir, cluster, iteration)) + ) + for b in benchmarks: + if not b.exists() and not settings.cluster.get("is_teuthology", False): + continue + + if rebuild_every_test: + with progress.phase_bar("Cluster initialize", progress.cluster_init_estimate(), overall): + cluster.initialize() + b.initialize() + + # Always try to initialize endpoints before running the test + b.initialize_endpoints() + logger.info("Running benchmark %s == iteration %d ==", b.getclass(), iteration) + run_secs = b.estimate_duration() or None + has_workloads = getattr(b, "_workloads", None) and b._workloads.exist() # type: ignore[union-attr] + if has_workloads: + # Workload benchmarks drive their own per-param-set phase bars + # inside Workloads.run(). Wrapping with an outer phase bar here + # would put two bars at position=1 simultaneously, causing them + # to swap/flicker. The inner bars also advance the overall bar + # directly via progress.get_overall_bar(), so no outer bar is needed. + b.run() + else: + with progress.phase_bar(f"Run {b.getclass()} iter {iteration}", run_secs, overall): + b.run() + logger.info("Benchmark %s iteration %d complete.", b.getclass(), iteration) + except: + return_code = 1 # FAIL + logger.exception("During tests") return return_code diff --git a/progress.py b/progress.py new file mode 100644 index 00000000..25ce6f06 --- /dev/null +++ b/progress.py @@ -0,0 +1,311 @@ +""" +progress.py — CLI progress bars for CBT runs. + +All output is written to *stderr* via tqdm. While bars are active, logging is +redirected through ``tqdm.write()`` so log lines appear cleanly above the bars +with no blank-line or mid-line glitches. + +Bars are automatically suppressed when: + - stdout is not a TTY (non-interactive / piped) + - settings.cluster contains ``is_teuthology: true`` + - ``--no-progress`` was passed on the command line + +Call ``setup()`` once after ``settings.initialize()`` and after arg parsing +so the disabled flag is resolved correctly. + +Design notes +------------ +* No background ticker thread — tqdm's own ``{elapsed}`` token in the format + string provides live elapsed time without any threading. +* ``dynamic_ncols=True`` on every bar so they reflow correctly on terminal resize. +* ``_logging_redirect`` is active for the entire run (entered in + ``overall_bar``) — it replaces the console StreamHandler with a level- + preserving tqdm-aware variant so log lines and bar redraws are serialised. + Unlike ``tqdm.contrib.logging.logging_redirect_tqdm``, the replacement + handler copies the original handler's *level* so DEBUG messages never leak + to the screen. +* The overall bar is advanced explicitly when each phase exits, so the filled + portion reflects work actually completed. +""" + +import contextlib +import logging +import sys +import threading +from collections.abc import Generator +from typing import Any, Optional + +import tqdm + +import settings + +# ────────────────────────────────────────────────────────────────────────────── +# Constants +# ────────────────────────────────────────────────────────────────────────────── + +# Sensible estimate (seconds) for the cluster initialisation phase when no +# explicit duration is available. Users can override via +# ``cluster.init_estimate_secs`` in their YAML config. +DEFAULT_CLUSTER_INIT_SECS: int = 180 + +# How often the ticker thread advances the phase bar (seconds). +# Coarse enough to avoid busy-looping; fine enough to feel responsive. +TICK_INTERVAL: float = 15.0 + +# tqdm bar_format strings — defined once here so they are not rebuilt on every +# context-manager entry and so all bars share a consistent layout. +_BAR_FORMAT_TIMED = "{l_bar}{bar}| {n_fmt}/{total_fmt}s [elapsed {elapsed} | eta {remaining}]" +_BAR_FORMAT_SPIN = "{desc}: {elapsed} elapsed" + +# ────────────────────────────────────────────────────────────────────────────── +# Module-level state (set once by setup()) +# ────────────────────────────────────────────────────────────────────────────── + +_disabled: bool = True # safe default — overridden by setup() +_overall: Optional["OverallBar"] = None # set by overall_bar() context manager + + +def setup(no_progress: bool = False) -> None: + """Resolve and store the global disabled flag. + + Call once after ``settings.initialize()`` and after CLI arg parsing. + + Args: + no_progress: True when ``--no-progress`` was supplied on the CLI. + """ + global _disabled # pylint: disable=global-statement + is_tty = sys.stdout.isatty() + is_teuthology = bool(settings.cluster.get("is_teuthology", False)) + _disabled = not is_tty or is_teuthology or no_progress + + +# ────────────────────────────────────────────────────────────────────────────── +# Public helpers +# ────────────────────────────────────────────────────────────────────────────── + + +def get_overall_bar() -> Optional["OverallBar"]: + """Return the active :class:`OverallBar`, or ``None`` when not running. + + Allows code deep in the call stack (e.g. :class:`~workloads.workloads.Workloads`) + to advance the overall bar without needing the handle passed through every + intermediate function signature. + """ + return _overall + + +def cluster_init_estimate() -> int: + """Return the estimated cluster-initialisation duration in seconds. + + Reads ``cluster.init_estimate_secs`` from the loaded settings, falling + back to :data:`DEFAULT_CLUSTER_INIT_SECS`. + """ + return int(settings.cluster.get("init_estimate_secs", DEFAULT_CLUSTER_INIT_SECS)) + + +# ────────────────────────────────────────────────────────────────────────────── +# Internal: level-preserving logging redirect +# ────────────────────────────────────────────────────────────────────────────── + + +class _TqdmHandler(logging.StreamHandler): # type: ignore[type-arg] + """StreamHandler that writes via ``tqdm.write()`` instead of directly. + + This ensures that log output and tqdm bar redraws are serialised, preventing + the blank-line / mid-line glitch seen when both write to the same terminal. + """ + + def emit(self, record: logging.LogRecord) -> None: + try: + msg = self.format(record) + tqdm.tqdm.write(msg, file=self.stream) + self.flush() + except Exception: # pylint: disable=broad-except + self.handleError(record) + + +@contextlib.contextmanager +def _logging_redirect(logger: logging.Logger) -> Generator[None, None, None]: + """Temporarily replace *logger*'s console handler with a tqdm-aware one. + + Unlike ``tqdm.contrib.logging.logging_redirect_tqdm``, this helper copies + the original handler's **level** and **formatter** so that the level + filtering configured in :func:`logging_configuration.setup_loggers` is + fully preserved (INFO+ to screen, DEBUG only to the file). + """ + # Find the existing console StreamHandler (not a FileHandler). + original: Optional[logging.StreamHandler] = next( # type: ignore[type-arg] + (h for h in logger.handlers if isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)), + None, + ) + + if original is None or _disabled: + # Nothing to redirect (bars are off, or no console handler found). + yield + return + + replacement = _TqdmHandler(stream=original.stream) + replacement.setLevel(original.level) # ← preserve INFO-only filtering + replacement.setFormatter(original.formatter) + + logger.removeHandler(original) + logger.addHandler(replacement) + try: + yield + finally: + logger.removeHandler(replacement) + logger.addHandler(original) + + +# ────────────────────────────────────────────────────────────────────────────── +# Progress bar context managers +# ────────────────────────────────────────────────────────────────────────────── + + +class OverallBar: # pylint: disable=too-few-public-methods + """Wrapper around the outer tqdm bar that exposes ``advance(seconds)``.""" + + def __init__(self, pbar: "tqdm.tqdm[Any]") -> None: + self._pbar = pbar + self._lock = threading.Lock() + + def advance(self, seconds: float) -> None: + """Advance the outer bar by *seconds* work-units.""" + with self._lock: + self._pbar.update(seconds) + + +@contextlib.contextmanager +def overall_bar(total_seconds: int) -> Generator[OverallBar, None, None]: + """Context manager for the outer overall-run progress bar. + + Activates the level-preserving logging redirect for its duration so that + all log messages are routed through ``tqdm.write()``. This ensures log + lines appear cleanly above both bars with no visual glitches, while the + INFO-only screen filter configured at startup is fully preserved. + + The bar tracks elapsed work in seconds. Callers advance it explicitly + via :meth:`OverallBar.advance` after each phase completes. + + Args: + total_seconds: Estimated total wall-clock seconds for the whole run. + + Yields: + An :class:`OverallBar` handle that callers use to advance the bar. + """ + pbar = tqdm.tqdm( + total=total_seconds, + desc="Overall run", + unit="s", + bar_format=_BAR_FORMAT_TIMED, + position=0, + leave=True, + file=sys.stderr, + dynamic_ncols=True, + disable=_disabled, + ) + handle = OverallBar(pbar) + global _overall # pylint: disable=global-statement + cbt_logger = logging.getLogger("cbt") + with _logging_redirect(cbt_logger): + _overall = handle + try: + yield handle + finally: + _overall = None + # Snap to 100% so the bar always closes complete, regardless of any + # estimation error or time spent outside phase bars (e.g. report generation). + if not _disabled and pbar.total is not None and pbar.n < pbar.total: + pbar.update(pbar.total - pbar.n) + pbar.close() + + +@contextlib.contextmanager +def phase_bar( + description: str, + duration_seconds: Optional[int] = None, + overall: Optional[OverallBar] = None, +) -> Generator[None, None, None]: + """Context manager for a single CBT phase (init, prefill, run, …). + + When *duration_seconds* is given a background ticker thread advances the + bar by :data:`TICK_INTERVAL` seconds every :data:`TICK_INTERVAL` wall-clock + seconds, so the filled portion moves in real time while CBT is blocked on + remote commands. The ticker only calls ``pbar.update()`` — it never + touches ``tqdm.write()`` or the logging system, so there is no deadlock + risk with the ``_TqdmHandler`` logging redirect. + + When ``duration_seconds`` is ``None`` the bar shows elapsed time only + (no fill, no ticker needed). + + On exit the outer *overall* bar is advanced by the estimated duration so + the overall progress reflects the completed work. + + Args: + description: Human-readable label shown on the bar. + duration_seconds: Expected duration of this phase in seconds. + ``None`` for indeterminate phases. + overall: The :class:`OverallBar` handle returned by + :func:`overall_bar`. When provided it is advanced + by *duration_seconds* on exit. + """ + pbar = tqdm.tqdm( + total=duration_seconds, + desc=description, + unit="s", + bar_format=_BAR_FORMAT_TIMED if duration_seconds is not None else _BAR_FORMAT_SPIN, + position=1, + leave=False, + file=sys.stderr, + dynamic_ncols=True, + disable=_disabled, + mininterval=1.0, + ) + + stop_event = threading.Event() + ticker: Optional[threading.Thread] = None + if duration_seconds is not None and not _disabled: + ticker = threading.Thread( + target=_tick, + args=(pbar, stop_event, overall), + daemon=True, + name=f"cbt-ticker-{description}", + ) + ticker.start() + + try: + yield + finally: + stop_event.set() + if ticker is not None: + ticker.join(timeout=TICK_INTERVAL + 1.0) + # Capture pbar.n before close() so we read a stable value. + already_ticked = pbar.n # how much the ticker already credited + pbar.close() + # Advance overall by any remaining seconds not yet ticked (e.g. phase + # finished early, or bars were disabled so no ticker ran). + if overall is not None and duration_seconds is not None: + remaining_credit = duration_seconds - already_ticked + if remaining_credit > 0: + overall.advance(remaining_credit) + + +def _tick( + pbar: "tqdm.tqdm[Any]", + stop_event: threading.Event, + overall: Optional[OverallBar] = None, +) -> None: + """Advance *pbar* and *overall* by :data:`TICK_INTERVAL` every :data:`TICK_INTERVAL` seconds. + + Clamps phase-bar updates so the bar never exceeds its total. Only calls + ``pbar.update()`` and ``overall.advance()`` — never ``tqdm.write()`` or + any logging function — so it is safe to run concurrently with + ``_TqdmHandler``. + """ + while not stop_event.wait(timeout=TICK_INTERVAL): + remaining = (pbar.total or 0) - pbar.n + if remaining > 0: + increment = min(TICK_INTERVAL, remaining) + pbar.update(increment) + if overall is not None: + overall.advance(increment) diff --git a/tests/test_progress.py b/tests/test_progress.py new file mode 100644 index 00000000..1e7c5ab3 --- /dev/null +++ b/tests/test_progress.py @@ -0,0 +1,298 @@ +""" +Tests for progress.py — CLI progress bars for CBT runs. +""" + +# pylint: disable=protected-access # deliberate in unit tests + +import threading +from collections.abc import Generator +from contextlib import contextmanager +from typing import Any, Optional +from unittest.mock import MagicMock, patch + +import pytest + +import progress +import settings as _settings +from benchmark.benchmark import Benchmark +from benchmark.radosbench import Radosbench + +# ────────────────────────────────────────────────────────────────────────────── +# Helpers +# ────────────────────────────────────────────────────────────────────────────── + + +@contextmanager +def _mock_settings(cluster_dict: dict[str, object]) -> Generator[None, None, None]: + """Temporarily replace settings.cluster with *cluster_dict*.""" + with patch("progress.settings") as mock_settings: + mock_settings.cluster = cluster_dict + yield + + +# ────────────────────────────────────────────────────────────────────────────── +# setup() +# ────────────────────────────────────────────────────────────────────────────── + + +class TestSetup: + """Tests for progress.setup().""" + + def test_disabled_when_not_a_tty(self) -> None: + """Progress is disabled when stdout is not a TTY.""" + with _mock_settings({}): + with patch("sys.stdout.isatty", return_value=False): + progress.setup(no_progress=False) + assert progress._disabled is True + + def test_disabled_when_is_teuthology(self) -> None: + """Progress is disabled when the teuthology flag is set.""" + with _mock_settings({"is_teuthology": True}): + with patch("sys.stdout.isatty", return_value=True): + progress.setup(no_progress=False) + assert progress._disabled is True + + def test_disabled_when_no_progress_flag(self) -> None: + """Progress is disabled when --no-progress was passed.""" + with _mock_settings({}): + with patch("sys.stdout.isatty", return_value=True): + progress.setup(no_progress=True) + assert progress._disabled is True + + def test_enabled_when_tty_and_no_flags(self) -> None: + """Progress is enabled when stdout is a TTY and no suppression flags are set.""" + with _mock_settings({}): + with patch("sys.stdout.isatty", return_value=True): + progress.setup(no_progress=False) + assert progress._disabled is False + + def test_disabled_when_all_suppression_conditions(self) -> None: + """Progress is disabled when multiple suppression conditions are present.""" + with _mock_settings({"is_teuthology": True}): + with patch("sys.stdout.isatty", return_value=False): + progress.setup(no_progress=True) + assert progress._disabled is True + + +# ────────────────────────────────────────────────────────────────────────────── +# cluster_init_estimate() +# ────────────────────────────────────────────────────────────────────────────── + + +class TestClusterInitEstimate: + """Tests for progress.cluster_init_estimate().""" + + def test_returns_default_when_not_configured(self) -> None: + """Returns the module default when the setting is absent.""" + with _mock_settings({}): + result = progress.cluster_init_estimate() + assert result == progress.DEFAULT_CLUSTER_INIT_SECS + + def test_returns_configured_value(self) -> None: + """Returns the value from settings when configured.""" + with _mock_settings({"init_estimate_secs": 300}): + result = progress.cluster_init_estimate() + assert result == 300 + + def test_returns_int(self) -> None: + """Always returns an int, even when the config value is a string.""" + with _mock_settings({"init_estimate_secs": "240"}): + result = progress.cluster_init_estimate() + assert isinstance(result, int) + assert result == 240 + + +# ────────────────────────────────────────────────────────────────────────────── +# overall_bar() +# ────────────────────────────────────────────────────────────────────────────── + + +class TestOverallBar: + """Tests for progress.overall_bar() context manager.""" + + def test_yields_overall_bar_handle(self) -> None: + """The context manager yields an OverallBar instance.""" + progress._disabled = True + with progress.overall_bar(100) as handle: + assert isinstance(handle, progress.OverallBar) + + def test_handle_advance_does_not_raise(self) -> None: + """advance() can be called without error.""" + progress._disabled = True + with progress.overall_bar(100) as handle: + handle.advance(30) + + def test_advance_is_clamped_implicitly(self) -> None: + """Advancing beyond the total (when disabled) doesn't raise.""" + progress._disabled = True + with progress.overall_bar(50) as handle: + handle.advance(200) + + +# ────────────────────────────────────────────────────────────────────────────── +# phase_bar() +# ────────────────────────────────────────────────────────────────────────────── + + +class TestPhaseBar: + """Tests for progress.phase_bar() context manager.""" + + def setup_method(self) -> None: + """Force disabled mode so bars don't write to stderr during tests.""" + progress._disabled = True + + def test_runs_body_with_known_duration(self) -> None: + """Body executes correctly with a known duration.""" + executed: list[bool] = [] + with progress.phase_bar("test phase", 60): + executed.append(True) + assert executed == [True] + + def test_runs_body_with_no_duration(self) -> None: + """Body executes correctly with an indeterminate duration.""" + executed: list[bool] = [] + with progress.phase_bar("indeterminate phase"): + executed.append(True) + assert executed == [True] + + def test_advances_overall_bar_on_exit(self) -> None: + """overall.advance() is called with the phase duration on exit.""" + mock_overall = MagicMock(spec=progress.OverallBar) + with progress.phase_bar("test phase", 45, overall=mock_overall): + pass + mock_overall.advance.assert_called_once_with(45) + + def test_does_not_advance_overall_when_no_duration(self) -> None: + """overall.advance() is NOT called when duration is None.""" + mock_overall = MagicMock(spec=progress.OverallBar) + with progress.phase_bar("no duration phase", None, overall=mock_overall): + pass + mock_overall.advance.assert_not_called() + + def test_does_not_advance_overall_when_not_provided(self) -> None: + """overall.advance() is not called when overall is not provided.""" + with progress.phase_bar("test phase", 30): + pass + + def test_exception_propagates(self) -> None: + """Exceptions raised in the body propagate correctly.""" + with pytest.raises(RuntimeError, match="boom"): + with progress.phase_bar("failing phase", 10): + raise RuntimeError("boom") + + def test_overall_still_advanced_on_exception(self) -> None: + """overall.advance() is still called even when the body raises.""" + mock_overall = MagicMock(spec=progress.OverallBar) + with pytest.raises(RuntimeError): + with progress.phase_bar("failing phase", 20, overall=mock_overall): + raise RuntimeError("oops") + mock_overall.advance.assert_called_once_with(20) + + def test_no_ticker_when_disabled(self) -> None: + """No ticker thread is started when progress bars are disabled.""" + progress._disabled = True + with patch("progress.threading.Thread") as mock_thread_cls: + with progress.phase_bar("test", 60): + pass + mock_thread_cls.assert_not_called() + + def test_no_ticker_when_no_duration(self) -> None: + """No ticker thread is started for indeterminate phases.""" + progress._disabled = False + with patch("progress.threading.Thread") as mock_thread_cls: + with patch("progress.tqdm.tqdm"): + with progress.phase_bar("test"): # no duration_seconds + pass + mock_thread_cls.assert_not_called() + + def test_ticker_started_when_enabled_with_duration(self) -> None: + """Ticker thread is started when bars are enabled and duration is known.""" + progress._disabled = False + with patch("progress.threading.Thread") as mock_thread_cls: + mock_thread = mock_thread_cls.return_value + mock_thread.join = lambda timeout=None: None + with patch("progress.tqdm.tqdm"): + with progress.phase_bar("test", 60): + pass + mock_thread_cls.assert_called_once() + mock_thread.start.assert_called_once() + + def test_overall_advanced_by_remaining_when_disabled(self) -> None: + """When bars are disabled the full duration is credited on exit (no ticker ran).""" + progress._disabled = True + mock_overall = MagicMock(spec=progress.OverallBar) + with progress.phase_bar("test phase", 90, overall=mock_overall): + pass + # pbar.n == 0 because disabled, so remaining_credit == 90 + mock_overall.advance.assert_called_once_with(90) + + def test_overall_advanced_incrementally_by_ticker(self) -> None: + """_tick advances overall on each tick; exit credits only the remainder.""" + mock_overall = MagicMock(spec=progress.OverallBar) + mock_pbar: MagicMock = MagicMock() + mock_pbar.total = 60 + mock_pbar.n = 0 + stop = threading.Event() + + # Simulate one tick: remaining=60, increment=15 + wait_responses = [False, True] + + def fake_wait(timeout: Optional[float] = None) -> bool: # pylint: disable=unused-argument + return wait_responses.pop(0) + + stop.wait = fake_wait # type: ignore[method-assign] + progress._tick(mock_pbar, stop, mock_overall) + + mock_pbar.update.assert_called_once_with(15) + mock_overall.advance.assert_called_once_with(15) + + +# ────────────────────────────────────────────────────────────────────────────── +# estimate_duration() — concrete benchmark overrides +# ────────────────────────────────────────────────────────────────────────────── + + +class TestEstimateDuration: + """Smoke tests to verify estimate_duration() returns sensible ints.""" + + def _make_radosbench(self, **kwargs: object) -> Any: + """Create a Radosbench instance with minimal settings for testing.""" + _settings.cluster = { + "tmp_dir": "/tmp/cbt_test", + "osd_ra": "0", + "clients": "localhost", + } + cluster_mock = MagicMock() + cluster_mock.config = {} + cluster_mock.tmp_conf = "/tmp/ceph.conf" + cluster_mock.mnt_dir = "/mnt" + config = { + "iteration": 0, + "time": "300", + "write_time": "300", + "read_time": "300", + **kwargs, + } + + return Radosbench("/tmp/archive", cluster_mock, config) # type: ignore[no-untyped-call] + + def test_radosbench_default_no_prefill(self) -> None: + """Write + read phases give 600s total.""" + b = self._make_radosbench() + assert b.estimate_duration() == 600 + + def test_radosbench_write_only(self) -> None: + """Write-only mode gives only the write duration.""" + b = self._make_radosbench(write_only=True) + assert b.estimate_duration() == 300 + + def test_radosbench_read_only_with_prefill(self) -> None: + """Prefill + read = 120 + 300 = 420.""" + b = self._make_radosbench(read_only=True, prefill_time="120") + assert b.estimate_duration() == 420 + + def test_base_benchmark_returns_zero(self) -> None: + """The base Benchmark.estimate_duration() always returns 0.""" + b = MagicMock(spec=Benchmark) + result = Benchmark.estimate_duration(b) + assert result == 0 diff --git a/workloads/workload.py b/workloads/workload.py index e28070e2..ad627c8e 100644 --- a/workloads/workload.py +++ b/workloads/workload.py @@ -76,6 +76,36 @@ def get_output_directories(self) -> Generator[str, None, None]: yield from unique_output_directories + def phase_duration_secs(self) -> Optional[int]: + """Return the estimated duration in seconds for a single parameter-set run. + + Uses the workload's own resolved options (workload-level values take + precedence over global benchmark values via :meth:`add_global_options`), + so this matches the ``--runtime`` and ``--ramp_time`` values that will + actually be passed to the I/O exerciser. + + Returns ``None`` when no ``time`` is configured. + """ + time_val = self._all_options.get("time") + if not time_val: + return None + run_secs = int(str(time_val)) + ramp_secs = int(str(self._all_options.get("ramp", 0))) or 0 + return run_secs + ramp_secs + + def param_set_count(self) -> int: + """Return the number of parameter-set iterations this workload will run. + + This is the product of the lengths of all list-valued options — the + same expansion that ``all_configs`` performs — without constructing + full config objects. Used by :meth:`Workloads.estimate_duration`. + """ + count = 1 + for value in self._all_options.values(): + if isinstance(value, list): + count *= len(value) + return max(count, 1) + def get_name(self) -> str: """ Return the name of this workload diff --git a/workloads/workloads.py b/workloads/workloads.py index 65f35947..042404a5 100644 --- a/workloads/workloads.py +++ b/workloads/workloads.py @@ -6,6 +6,7 @@ from time import sleep from typing import Optional, Union +import progress from common import CheckedPopen, CheckedPopenLocal, make_remote_dir, pdsh # pyright: ignore[reportUnknownVariableType] from monitoring.monitoring_factory import MonitoringFactory from settings import getnodes # pyright: ignore[reportUnknownVariableType] @@ -49,7 +50,28 @@ def exist(self) -> bool: """ return bool(self._workloads) - def run(self) -> None: + def estimate_duration(self) -> int: + """Estimate total wall-clock seconds for all workloads. + + For each workload, multiplies the per-param-set duration (``time`` + + ``ramp`` from the benchmark configuration) by the number of parameter + set combinations. Returns 0 when no workloads are configured or when + no time is set. + """ + per_set_secs = int(str(self._benchmark_configuration.get("time", 0))) or 0 + ramp_secs = int(str(self._benchmark_configuration.get("ramp", 0))) or 0 + if per_set_secs == 0: + return 0 + total = 0 + for workload in self._workloads: + # Each workload overrides time/ramp if set; fall back to global values. + w_options = workload._all_options # pylint: disable=protected-access + w_time = int(str(w_options.get("time", per_set_secs))) or per_set_secs + w_ramp = int(str(w_options.get("ramp", ramp_secs))) or ramp_secs + total += workload.param_set_count() * (w_time + w_ramp) + return total + + def run(self) -> None: # pylint: disable=too-many-locals """ Run all the I/O exerciser commands for each workload in turn, including any scripts that should be run between workloads @@ -69,7 +91,6 @@ def run(self) -> None: ramp_time: str = f"{self._benchmark_configuration.get('ramp', '')}" total_workloads = len(self._workloads) - processes: list[Union[CheckedPopen, CheckedPopenLocal]] = [] for workload_index, workload in enumerate(self._workloads, start=1): workload_name = workload.get_name() log.info("Starting workload '%s' (%d/%d)...", workload_name, workload_index, total_workloads) @@ -94,23 +115,31 @@ def run(self) -> None: total_param_sets, output_directory, ) - if script_command: - pdsh(getnodes("clients"), script_command).wait() # type: ignore[no-untyped-call] + phase_desc = f"Workload '{workload_name}' {param_index}/{total_param_sets}" + # Use the workload's own resolved time+ramp so the bar matches + # the --runtime/--ramp_time values actually passed to the exerciser. + phase_secs: Optional[int] = workload.phase_duration_secs() + + with progress.phase_bar(phase_desc, phase_secs, overall=progress.get_overall_bar()): + if script_command: + pdsh(getnodes("clients"), script_command).wait() # type: ignore[no-untyped-call] - for fio_command in fio_command_list: - processes.append(pdsh(getnodes("clients"), fio_command)) # type: ignore[no-untyped-call] + processes: list[Union[CheckedPopen, CheckedPopenLocal]] = [ + pdsh(getnodes("clients"), fio_command) # type: ignore[no-untyped-call] + for fio_command in fio_command_list + ] - # Sleep for the ramp time and then collect stats - if ramp_time: - log.info("Ramp time: waiting %ss before collecting stats...", ramp_time) - sleep(int(ramp_time)) + # Sleep for the ramp time and then collect stats + if ramp_time: + log.info("Ramp time: waiting %ss before collecting stats...", ramp_time) + sleep(int(ramp_time)) - MonitoringFactory.start(output_directory) + MonitoringFactory.start(output_directory) - for process in processes: - process.wait() # type: ignore[no-untyped-call] + for process in processes: + process.wait() # type: ignore[no-untyped-call] - MonitoringFactory.stop() + MonitoringFactory.stop() log.info("Workload '%s': parameter set %d/%d complete.", workload_name, param_index, total_param_sets) log.info("Workload '%s' complete (%d/%d).", workload_name, workload_index, total_workloads)