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)