# Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import csv import io import logging import sys from functools import partial import tabulate from esrally import exceptions, metrics, types from esrally.utils import console, convert from esrally.utils import io as rio FINAL_SCORE = r""" ------------------------------------------------------ _______ __ _____ / ____(_)___ ____ _/ / / ___/_________ ________ / /_ / / __ \/ __ `/ / \__ \/ ___/ __ \/ ___/ _ \ / __/ / / / / / /_/ / / ___/ / /__/ /_/ / / / __/ /_/ /_/_/ /_/\__,_/_/ /____/\___/\____/_/ \___/ ------------------------------------------------------ """ def summarize(results, cfg: types.Config): SummaryReporter(results, cfg).report() def compare(cfg: types.Config, baseline_id, contender_id): if not baseline_id or not contender_id: raise exceptions.SystemSetupError("compare needs baseline and a contender") race_store = metrics.race_store(cfg) ComparisonReporter(cfg).report(race_store.find_by_race_id(baseline_id), race_store.find_by_race_id(contender_id)) def print_internal(message): console.println(message, logger=logging.getLogger(__name__).info) def print_header(message): print_internal(console.format.bold(message)) def write_single_report(report_file, report_format, cwd, numbers_align, headers, data_plain, data_rich): if report_format == "markdown": formatter = partial(format_as_markdown, numbers_align=numbers_align) elif report_format == "csv": formatter = format_as_csv else: raise exceptions.SystemSetupError("Unknown report format '%s'" % report_format) print_internal(formatter(headers, data_rich)) if len(report_file) > 0: normalized_report_file = rio.normalize_path(report_file, cwd) # ensure that the parent folder already exists when we try to write the file... rio.ensure_dir(rio.dirname(normalized_report_file)) with open(normalized_report_file, mode="a+", encoding="utf-8") as f: f.writelines(formatter(headers, data_plain)) def format_as_markdown(headers, data, numbers_align): rendered = tabulate.tabulate(data, headers=headers, tablefmt="pipe", numalign=numbers_align, stralign="right") return rendered + "\n" def format_as_csv(headers, data): with io.StringIO() as out: writer = csv.writer(out) writer.writerow(headers) for metric_record in data: writer.writerow(metric_record) return out.getvalue() def disk_usage_fields(stats): return { "inverted index": stats.disk_usage_inverted_index, "stored fields": stats.disk_usage_stored_fields, "doc values": stats.disk_usage_doc_values, "points": stats.disk_usage_points, "norms": stats.disk_usage_norms, "term vectors": stats.disk_usage_term_vectors, "total": stats.disk_usage_total, } def collate_disk_usage_stats(stats): collated = {} for stat, field_stats in disk_usage_fields(stats).items(): for field_stat in field_stats: collated.setdefault(field_stat["index"], {}).setdefault(field_stat["field"], {})[stat] = field_stat["value"] return collated def total_disk_usage_per_field(stats): totals = [] for field_stat in stats.disk_usage_total: totals.append([field_stat["index"], field_stat["value"], field_stat["field"]]) return totals class SummaryReporter: def __init__(self, results, config: types.Config): self.results = results self.report_file = config.opts("reporting", "output.path") self.report_format = config.opts("reporting", "format") self.numbers_align = config.opts("reporting", "numbers.align", mandatory=False, default_value="decimal") reporting_values = config.opts("reporting", "values") self.report_all_values = reporting_values == "all" self.report_all_percentile_values = reporting_values == "all-percentiles" self.show_processing_time = convert.to_bool(config.opts("reporting", "output.processingtime", mandatory=False, default_value=False)) self.cwd = config.opts("node", "rally.cwd") def report(self): print_header(FINAL_SCORE) stats = self.results warnings = [] metrics_table = [] metrics_table.extend(self._report_totals(stats)) metrics_table.extend(self._report_ml_processing_times(stats)) metrics_table.extend(self._report_gc_metrics(stats)) metrics_table.extend(self._report_disk_usage(stats)) metrics_table.extend(self._report_segment_memory(stats)) metrics_table.extend(self._report_segment_counts(stats)) metrics_table.extend(self._report_transform_stats(stats)) metrics_table.extend(self._report_ingest_pipeline_stats(stats)) metrics_table.extend(self._report_disk_usage_per_field(stats)) for record in stats.op_metrics: task = record["task"] metrics_table.extend(self._report_throughput(record, task)) metrics_table.extend(self._report_latency(record, task)) metrics_table.extend(self._report_service_time(record, task)) # this is mostly needed for debugging purposes but not so relevant to end users if self.show_processing_time: metrics_table.extend(self._report_processing_time(record, task)) metrics_table.extend(self._report_error_rate(record, task)) self.add_warnings(warnings, record, task) self.write_report(metrics_table) if warnings: for warning in warnings: console.warn(warning) def add_warnings(self, warnings, values, op): if values["error_rate"] > 0: warnings.append(f"Error rate is {round(values['error_rate'] * 100, 2)} for operation '{op}'. Please check the logs.") if values["throughput"]["median"] is None: error_rate = values["error_rate"] if error_rate: warnings.append( "No throughput metrics available for [%s]. Likely cause: Error rate is %.1f%%. Please check the logs." % (op, error_rate * 100) ) else: warnings.append("No throughput metrics available for [%s]. Likely cause: The benchmark ended already during warmup." % op) def write_report(self, metrics_table): write_single_report( self.report_file, self.report_format, self.cwd, self.numbers_align, headers=["Metric", "Task", "Value", "Unit"], data_plain=metrics_table, data_rich=metrics_table, ) def _report_throughput(self, values, task): throughput = values["throughput"] unit = throughput["unit"] return self._join( self._line("Min Throughput", task, throughput["min"], unit, lambda v: "%.2f" % v), self._line("Mean Throughput", task, throughput["mean"], unit, lambda v: "%.2f" % v), self._line("Median Throughput", task, throughput["median"], unit, lambda v: "%.2f" % v), self._line("Max Throughput", task, throughput["max"], unit, lambda v: "%.2f" % v), ) def _report_latency(self, values, task): return self._report_percentiles("latency", task, values["latency"]) def _report_service_time(self, values, task): return self._report_percentiles("service time", task, values["service_time"]) def _report_processing_time(self, values, task): return self._report_percentiles("processing time", task, values["processing_time"]) def _report_percentiles(self, name, task, value): lines = [] if value: for percentile in metrics.percentiles_for_sample_size(sys.maxsize): percentile_value = value.get(metrics.encode_float_key(percentile)) a_line = self._line( "%sth percentile %s" % (percentile, name), task, percentile_value, "ms", force=self.report_all_percentile_values ) self._append_non_empty(lines, a_line) return lines def _report_error_rate(self, values, task): return self._join(self._line("error rate", task, values["error_rate"], "%", lambda v: "%.2f" % (v * 100.0))) def _report_totals(self, stats): lines = [] lines.extend(self._report_total_time("indexing time", stats.total_time)) lines.extend(self._report_total_time_per_shard("indexing time", stats.total_time_per_shard)) lines.extend(self._report_total_time("indexing throttle time", stats.indexing_throttle_time)) lines.extend(self._report_total_time_per_shard("indexing throttle time", stats.indexing_throttle_time_per_shard)) lines.extend(self._report_total_time("merge time", stats.merge_time)) lines.extend(self._report_total_count("merge count", stats.merge_count)) lines.extend(self._report_total_time_per_shard("merge time", stats.merge_time_per_shard)) lines.extend(self._report_total_time("merge throttle time", stats.merge_throttle_time)) lines.extend(self._report_total_time_per_shard("merge throttle time", stats.merge_throttle_time_per_shard)) lines.extend(self._report_total_time("refresh time", stats.refresh_time)) lines.extend(self._report_total_count("refresh count", stats.refresh_count)) lines.extend(self._report_total_time_per_shard("refresh time", stats.refresh_time_per_shard)) lines.extend(self._report_total_time("flush time", stats.flush_time)) lines.extend(self._report_total_count("flush count", stats.flush_count)) lines.extend(self._report_total_time_per_shard("flush time", stats.flush_time_per_shard)) return lines def _report_total_time(self, name, total_time): unit = "min" return self._join( self._line(f"Cumulative {name} of primary shards", "", total_time, unit, convert.ms_to_minutes), ) def _report_total_time_per_shard(self, name, total_time_per_shard): unit = "min" return self._join( self._line(f"Min cumulative {name} across primary shards", "", total_time_per_shard.get("min"), unit, convert.ms_to_minutes), self._line( f"Median cumulative {name} across primary shards", "", total_time_per_shard.get("median"), unit, convert.ms_to_minutes, ), self._line(f"Max cumulative {name} across primary shards", "", total_time_per_shard.get("max"), unit, convert.ms_to_minutes), ) def _report_total_count(self, name, total_count): return self._join( self._line(f"Cumulative {name} of primary shards", "", total_count, ""), ) def _report_ml_processing_times(self, stats): lines = [] for processing_time in stats.ml_processing_time: job_name = processing_time["job"] unit = processing_time["unit"] lines.append(self._line("Min ML processing time", job_name, processing_time["min"], unit)) lines.append(self._line("Mean ML processing time", job_name, processing_time["mean"], unit)) lines.append(self._line("Median ML processing time", job_name, processing_time["median"], unit)) lines.append(self._line("Max ML processing time", job_name, processing_time["max"], unit)) return lines def _report_gc_metrics(self, stats): return self._join( self._line("Total Young Gen GC time", "", stats.young_gc_time, "s", convert.ms_to_seconds), self._line("Total Young Gen GC count", "", stats.young_gc_count, ""), self._line("Total Old Gen GC time", "", stats.old_gc_time, "s", convert.ms_to_seconds), self._line("Total Old Gen GC count", "", stats.old_gc_count, ""), self._line("Total ZGC Cycles GC time", "", stats.zgc_cycles_gc_time, "s", convert.ms_to_seconds), self._line("Total ZGC Cycles GC count", "", stats.zgc_cycles_gc_count, ""), self._line("Total ZGC Pauses GC time", "", stats.zgc_pauses_gc_time, "s", convert.ms_to_seconds), self._line("Total ZGC Pauses GC count", "", stats.zgc_pauses_gc_count, ""), ) def _report_disk_usage(self, stats): return self._join( self._line("Dataset size", "", stats.dataset_size, "GB", convert.bytes_to_gb), self._line("Store size", "", stats.store_size, "GB", convert.bytes_to_gb), self._line("Translog size", "", stats.translog_size, "GB", convert.bytes_to_gb), ) def _report_segment_memory(self, stats): unit = "MB" return self._join( self._line("Heap used for segments", "", stats.memory_segments, unit, convert.bytes_to_mb), self._line("Heap used for doc values", "", stats.memory_doc_values, unit, convert.bytes_to_mb), self._line("Heap used for terms", "", stats.memory_terms, unit, convert.bytes_to_mb), self._line("Heap used for norms", "", stats.memory_norms, unit, convert.bytes_to_mb), self._line("Heap used for points", "", stats.memory_points, unit, convert.bytes_to_mb), self._line("Heap used for stored fields", "", stats.memory_stored_fields, unit, convert.bytes_to_mb), ) def _report_segment_counts(self, stats): return self._join(self._line("Segment count", "", stats.segment_count, "")) def _report_transform_stats(self, stats): lines = [] for processing_time in stats.total_transform_processing_times: lines.append(self._line("Transform processing time", processing_time["id"], processing_time["mean"], processing_time["unit"])) for index_time in stats.total_transform_index_times: lines.append(self._line("Transform indexing time", index_time["id"], index_time["mean"], index_time["unit"])) for search_time in stats.total_transform_search_times: lines.append(self._line("Transform search time", search_time["id"], search_time["mean"], search_time["unit"])) for throughput in stats.total_transform_throughput: lines.append(self._line("Transform throughput", throughput["id"], throughput["mean"], throughput["unit"])) return lines def _report_ingest_pipeline_stats(self, stats): return self._join( self._line("Total Ingest Pipeline count", "", stats.ingest_pipeline_cluster_count, ""), self._line("Total Ingest Pipeline time", "", stats.ingest_pipeline_cluster_time, "s", convert.ms_to_seconds), self._line("Total Ingest Pipeline failed", "", stats.ingest_pipeline_cluster_failed, ""), ) def _report_disk_usage_per_field(self, stats): collated = collate_disk_usage_stats(stats) lines = [] for index, _total, field in sorted(total_disk_usage_per_field(stats)): for stat, value in collated[index][field].items(): lines.append( self._line(f"{index} {field} {stat}", "", value, convert.bytes_to_human_unit(value), convert.bytes_to_human_value) ) return lines def _join(self, *args): lines = [] for arg in args: self._append_non_empty(lines, arg) return lines def _append_non_empty(self, lines, line): if line and len(line) > 0: lines.append(line) def _line(self, k, task, v, unit, converter=lambda x: x, force=False): if v is not None or force or self.report_all_values: u = unit if v is not None else None return [k, task, converter(v), u] else: return [] class ComparisonReporter: def __init__(self, config: types.Config): self.report_file = config.opts("reporting", "output.path") self.report_format = config.opts("reporting", "format") self.numbers_align = config.opts("reporting", "numbers.align", mandatory=False, default_value="decimal") self.cwd = config.opts("node", "rally.cwd") self.show_processing_time = convert.to_bool(config.opts("reporting", "output.processingtime", mandatory=False, default_value=False)) self.plain = False def report(self, r1, r2): # we don't verify anything about the races as it is possible that the user benchmarks two different tracks intentionally baseline_stats = metrics.GlobalStats(r1.results) contender_stats = metrics.GlobalStats(r2.results) print_internal("") print_internal("Comparing baseline") print_internal(" Race ID: %s" % r1.race_id) print_internal(" Race timestamp: %s" % r1.race_timestamp) if r1.challenge_name: print_internal(" Challenge: %s" % r1.challenge_name) print_internal(" Car: %s" % r1.car_name) if r1.user_tags: r1_user_tags = ", ".join(["%s=%s" % (k, v) for k, v in sorted(r1.user_tags.items())]) print_internal(" User tags: %s" % r1_user_tags) print_internal("") print_internal("with contender") print_internal(" Race ID: %s" % r2.race_id) print_internal(" Race timestamp: %s" % r2.race_timestamp) if r2.challenge_name: print_internal(" Challenge: %s" % r2.challenge_name) print_internal(" Car: %s" % r2.car_name) if r2.user_tags: r2_user_tags = ", ".join(["%s=%s" % (k, v) for k, v in sorted(r2.user_tags.items())]) print_internal(" User tags: %s" % r2_user_tags) print_header(FINAL_SCORE) metric_table_plain = self._metrics_table(baseline_stats, contender_stats, plain=True) metric_table_rich = self._metrics_table(baseline_stats, contender_stats, plain=False) # Writes metric_table_rich to console, writes metric_table_plain to file self._write_report(metric_table_plain, metric_table_rich) def _metrics_table(self, baseline_stats, contender_stats, plain): self.plain = plain metrics_table = [] metrics_table.extend(self._report_total_times(baseline_stats, contender_stats)) metrics_table.extend(self._report_ml_processing_times(baseline_stats, contender_stats)) metrics_table.extend(self._report_gc_metrics(baseline_stats, contender_stats)) metrics_table.extend(self._report_disk_usage(baseline_stats, contender_stats)) metrics_table.extend(self._report_segment_memory(baseline_stats, contender_stats)) metrics_table.extend(self._report_segment_counts(baseline_stats, contender_stats)) metrics_table.extend(self._report_transform_processing_times(baseline_stats, contender_stats)) metrics_table.extend(self._report_ingest_pipeline_counts(baseline_stats, contender_stats)) metrics_table.extend(self._report_ingest_pipeline_times(baseline_stats, contender_stats)) metrics_table.extend(self._report_ingest_pipeline_failed(baseline_stats, contender_stats)) # Skip disk usage stats comparison if the disk_usage_total field does not exist if baseline_stats.disk_usage_total and contender_stats.disk_usage_total: metrics_table.extend(self._report_disk_usage_stats_per_field(baseline_stats, contender_stats)) for t in baseline_stats.tasks(): if t in contender_stats.tasks(): metrics_table.extend(self._report_throughput(baseline_stats, contender_stats, t)) metrics_table.extend(self._report_latency(baseline_stats, contender_stats, t)) metrics_table.extend(self._report_service_time(baseline_stats, contender_stats, t)) if self.show_processing_time: metrics_table.extend(self._report_processing_time(baseline_stats, contender_stats, t)) metrics_table.extend(self._report_error_rate(baseline_stats, contender_stats, t)) return metrics_table def _write_report(self, metrics_table, metrics_table_console): write_single_report( self.report_file, self.report_format, self.cwd, self.numbers_align, headers=["Metric", "Task", "Baseline", "Contender", "Diff", "Unit", "Diff %"], data_plain=metrics_table, data_rich=metrics_table_console, ) def _report_throughput(self, baseline_stats, contender_stats, task): b_min = baseline_stats.metrics(task)["throughput"]["min"] b_mean = baseline_stats.metrics(task)["throughput"]["mean"] b_median = baseline_stats.metrics(task)["throughput"]["median"] b_max = baseline_stats.metrics(task)["throughput"]["max"] b_unit = baseline_stats.metrics(task)["throughput"]["unit"] c_min = contender_stats.metrics(task)["throughput"]["min"] c_mean = contender_stats.metrics(task)["throughput"]["mean"] c_median = contender_stats.metrics(task)["throughput"]["median"] c_max = contender_stats.metrics(task)["throughput"]["max"] return self._join( self._line("Min Throughput", b_min, c_min, task, b_unit, treat_increase_as_improvement=True), self._line("Mean Throughput", b_mean, c_mean, task, b_unit, treat_increase_as_improvement=True), self._line("Median Throughput", b_median, c_median, task, b_unit, treat_increase_as_improvement=True), self._line("Max Throughput", b_max, c_max, task, b_unit, treat_increase_as_improvement=True), ) def _report_latency(self, baseline_stats, contender_stats, task): baseline_latency = baseline_stats.metrics(task)["latency"] contender_latency = contender_stats.metrics(task)["latency"] return self._report_percentiles("latency", task, baseline_latency, contender_latency) def _report_service_time(self, baseline_stats, contender_stats, task): baseline_service_time = baseline_stats.metrics(task)["service_time"] contender_service_time = contender_stats.metrics(task)["service_time"] return self._report_percentiles("service time", task, baseline_service_time, contender_service_time) def _report_processing_time(self, baseline_stats, contender_stats, task): baseline_processing_time = baseline_stats.metrics(task)["processing_time"] contender_processing_time = contender_stats.metrics(task)["processing_time"] return self._report_percentiles("processing time", task, baseline_processing_time, contender_processing_time) def _report_percentiles(self, name, task, baseline_values, contender_values): lines = [] for percentile in metrics.percentiles_for_sample_size(sys.maxsize): baseline_value = baseline_values.get(metrics.encode_float_key(percentile)) contender_value = contender_values.get(metrics.encode_float_key(percentile)) self._append_non_empty( lines, self._line( "%sth percentile %s" % (percentile, name), baseline_value, contender_value, task, "ms", treat_increase_as_improvement=False, ), ) return lines def _report_error_rate(self, baseline_stats, contender_stats, task): baseline_error_rate = baseline_stats.metrics(task)["error_rate"] contender_error_rate = contender_stats.metrics(task)["error_rate"] return self._join( self._line( "error rate", baseline_error_rate, contender_error_rate, task, "%", treat_increase_as_improvement=False, formatter=convert.factor(100.0), ) ) def _report_ml_processing_times(self, baseline_stats, contender_stats): lines = [] for baseline in baseline_stats.ml_processing_time: job_name = baseline["job"] unit = baseline["unit"] # O(n^2) but we assume here only a *very* limited number of jobs (usually just one) for contender in contender_stats.ml_processing_time: if contender["job"] == job_name: lines.append( self._line( "Min ML processing time", baseline["min"], contender["min"], job_name, unit, treat_increase_as_improvement=False ) ) lines.append( self._line( "Mean ML processing time", baseline["mean"], contender["mean"], job_name, unit, treat_increase_as_improvement=False, ) ) lines.append( self._line( "Median ML processing time", baseline["median"], contender["median"], job_name, unit, treat_increase_as_improvement=False, ) ) lines.append( self._line( "Max ML processing time", baseline["max"], contender["max"], job_name, unit, treat_increase_as_improvement=False ) ) return lines def _report_transform_processing_times(self, baseline_stats, contender_stats): lines = [] if baseline_stats.total_transform_processing_times is None: return lines for baseline in baseline_stats.total_transform_processing_times: transform_id = baseline["id"] for contender in contender_stats.total_transform_processing_times: if contender["id"] == transform_id: lines.append( self._line( "Transform processing time", baseline["mean"], contender["mean"], transform_id, baseline["unit"], treat_increase_as_improvement=False, ) ) for baseline in baseline_stats.total_transform_index_times: transform_id = baseline["id"] for contender in contender_stats.total_transform_index_times: if contender["id"] == transform_id: lines.append( self._line( "Transform indexing time", baseline["mean"], contender["mean"], transform_id, baseline["unit"], treat_increase_as_improvement=False, ) ) for baseline in baseline_stats.total_transform_search_times: transform_id = baseline["id"] for contender in contender_stats.total_transform_search_times: if contender["id"] == transform_id: lines.append( self._line( "Transform search time", baseline["mean"], contender["mean"], transform_id, baseline["unit"], treat_increase_as_improvement=False, ) ) for baseline in baseline_stats.total_transform_throughput: transform_id = baseline["id"] for contender in contender_stats.total_transform_throughput: if contender["id"] == transform_id: lines.append( self._line( "Transform throughput", baseline["mean"], contender["mean"], transform_id, baseline["unit"], treat_increase_as_improvement=True, ) ) return lines def _report_ingest_pipeline_counts(self, baseline_stats, contender_stats): if baseline_stats.ingest_pipeline_cluster_count is None: return [] return self._join( self._line( "Total Ingest Pipeline count", baseline_stats.ingest_pipeline_cluster_count, contender_stats.ingest_pipeline_cluster_count, "", "", treat_increase_as_improvement=False, ) ) def _report_ingest_pipeline_times(self, baseline_stats, contender_stats): if baseline_stats.ingest_pipeline_cluster_time is None: return [] return self._join( self._line( "Total Ingest Pipeline time", baseline_stats.ingest_pipeline_cluster_time, contender_stats.ingest_pipeline_cluster_time, "", "ms", treat_increase_as_improvement=False, ) ) def _report_ingest_pipeline_failed(self, baseline_stats, contender_stats): if baseline_stats.ingest_pipeline_cluster_failed is None: return [] return self._join( self._line( "Total Ingest Pipeline failed", baseline_stats.ingest_pipeline_cluster_failed, contender_stats.ingest_pipeline_cluster_failed, "", "", treat_increase_as_improvement=False, ) ) def _report_disk_usage_stats_per_field(self, baseline_stats, contender_stats): best = {} for index, total, field in total_disk_usage_per_field(baseline_stats): best.setdefault(index, {})[field] = total collated_baseline = collate_disk_usage_stats(baseline_stats) for index, total, field in total_disk_usage_per_field(contender_stats): for_idx = best.setdefault(index, {}) prev = for_idx.get(field, 0) if prev < total: for_idx[field] = total collated_contender = collate_disk_usage_stats(contender_stats) totals = [] for index, for_idx in best.items(): for field, total in for_idx.items(): totals.append([index, total, field]) totals.sort() lines = [] for index, _total, field in totals: for stat in disk_usage_fields(baseline_stats): if index in collated_baseline and index in collated_contender: baseline_value = collated_baseline[index].get(field, {}).get(stat, 0) contender_value = collated_contender[index].get(field, {}).get(stat, 0) if baseline_value == 0 and contender_value == 0: continue else: continue unit = convert.bytes_to_human_unit(min(baseline_value, contender_value)) lines.append( self._line( f"{index} {field} {stat}", baseline_value, contender_value, "", unit, treat_increase_as_improvement=False, formatter=partial(convert.bytes_to_unit, unit), ) ) return lines def _report_total_times(self, baseline_stats, contender_stats): lines = [] lines.extend( self._report_total_time( "indexing time", baseline_stats.total_time, contender_stats.total_time, ) ) lines.extend( self._report_total_time_per_shard( "indexing time", baseline_stats.total_time_per_shard, contender_stats.total_time_per_shard, ) ) lines.extend( self._report_total_time( "indexing throttle time", baseline_stats.indexing_throttle_time, contender_stats.indexing_throttle_time, ) ) lines.extend( self._report_total_time_per_shard( "indexing throttle time", baseline_stats.indexing_throttle_time_per_shard, contender_stats.indexing_throttle_time_per_shard, ) ) lines.extend( self._report_total_time( "merge time", baseline_stats.merge_time, contender_stats.merge_time, ) ) lines.extend( self._report_total_count( "merge count", baseline_stats.merge_count, contender_stats.merge_count, ) ) lines.extend( self._report_total_time_per_shard( "merge time", baseline_stats.merge_time_per_shard, contender_stats.merge_time_per_shard, ) ) lines.extend( self._report_total_time( "merge throttle time", baseline_stats.merge_throttle_time, contender_stats.merge_throttle_time, ) ) lines.extend( self._report_total_time_per_shard( "merge throttle time", baseline_stats.merge_throttle_time_per_shard, contender_stats.merge_throttle_time_per_shard, ) ) lines.extend( self._report_total_time( "refresh time", baseline_stats.refresh_time, contender_stats.refresh_time, ) ) lines.extend( self._report_total_count( "refresh count", baseline_stats.refresh_count, contender_stats.refresh_count, ) ) lines.extend( self._report_total_time_per_shard( "refresh time", baseline_stats.refresh_time_per_shard, contender_stats.refresh_time_per_shard, ) ) lines.extend( self._report_total_time( "flush time", baseline_stats.flush_time, contender_stats.flush_time, ) ) lines.extend( self._report_total_count( "flush count", baseline_stats.flush_count, contender_stats.flush_count, ) ) lines.extend( self._report_total_time_per_shard( "flush time", baseline_stats.flush_time_per_shard, contender_stats.flush_time_per_shard, ) ) return lines def _report_total_time(self, name, baseline_total, contender_total): unit = "min" return self._join( self._line( f"Cumulative {name} of primary shards", baseline_total, contender_total, "", unit, treat_increase_as_improvement=False, formatter=convert.ms_to_minutes, ), ) def _report_total_time_per_shard(self, name, baseline_per_shard, contender_per_shard): unit = "min" return self._join( self._line( f"Min cumulative {name} across primary shard", baseline_per_shard.get("min"), contender_per_shard.get("min"), "", unit, treat_increase_as_improvement=False, formatter=convert.ms_to_minutes, ), self._line( f"Median cumulative {name} across primary shard", baseline_per_shard.get("median"), contender_per_shard.get("median"), "", unit, treat_increase_as_improvement=False, formatter=convert.ms_to_minutes, ), self._line( f"Max cumulative {name} across primary shard", baseline_per_shard.get("max"), contender_per_shard.get("max"), "", unit, treat_increase_as_improvement=False, formatter=convert.ms_to_minutes, ), ) def _report_total_count(self, name, baseline_total, contender_total): return self._join( self._line(f"Cumulative {name} of primary shards", baseline_total, contender_total, "", "", treat_increase_as_improvement=False) ) def _report_gc_metrics(self, baseline_stats, contender_stats): line_method = self._line def _time_metric(metric_prefix, description): return line_method( f"Total {description} GC time", getattr(baseline_stats, f"{metric_prefix}_gc_time"), getattr(contender_stats, f"{metric_prefix}_gc_time"), "", "s", treat_increase_as_improvement=False, formatter=convert.ms_to_seconds, ) def _count_metric(metric_prefix, description): return line_method( f"Total {description} GC count", getattr(baseline_stats, f"{metric_prefix}_gc_count"), getattr(contender_stats, f"{metric_prefix}_gc_count"), "", "", treat_increase_as_improvement=False, ) return self._join( _time_metric("young", "Young Gen"), _count_metric("young", "Young Gen"), _time_metric("old", "Old Gen"), _count_metric("old", "Old Gen"), _time_metric("zgc_cycles", "ZGC Cycles"), _count_metric("zgc_cycles", "ZGC Cycles"), _time_metric("zgc_pauses", "ZGC Pauses"), _count_metric("zgc_pauses", "ZGC Pauses"), ) def _report_disk_usage(self, baseline_stats, contender_stats): return self._join( self._line( "Dataset size", baseline_stats.dataset_size, contender_stats.dataset_size, "", "GB", treat_increase_as_improvement=False, formatter=convert.bytes_to_gb, ), self._line( "Store size", baseline_stats.store_size, contender_stats.store_size, "", "GB", treat_increase_as_improvement=False, formatter=convert.bytes_to_gb, ), self._line( "Translog size", baseline_stats.translog_size, contender_stats.translog_size, "", "GB", treat_increase_as_improvement=False, formatter=convert.bytes_to_gb, ), ) def _report_segment_memory(self, baseline_stats, contender_stats): return self._join( self._line( "Heap used for segments", baseline_stats.memory_segments, contender_stats.memory_segments, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), self._line( "Heap used for doc values", baseline_stats.memory_doc_values, contender_stats.memory_doc_values, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), self._line( "Heap used for terms", baseline_stats.memory_terms, contender_stats.memory_terms, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), self._line( "Heap used for norms", baseline_stats.memory_norms, contender_stats.memory_norms, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), self._line( "Heap used for points", baseline_stats.memory_points, contender_stats.memory_points, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), self._line( "Heap used for stored fields", baseline_stats.memory_stored_fields, contender_stats.memory_stored_fields, "", "MB", treat_increase_as_improvement=False, formatter=convert.bytes_to_mb, ), ) def _report_segment_counts(self, baseline_stats, contender_stats): return self._join( self._line( "Segment count", baseline_stats.segment_count, contender_stats.segment_count, "", "", treat_increase_as_improvement=False, ) ) def _join(self, *args): lines = [] for arg in args: self._append_non_empty(lines, arg) return lines def _append_non_empty(self, lines, line): if line and len(line) > 0: lines.append(line) def _line(self, metric, baseline, contender, task, unit, treat_increase_as_improvement, formatter=lambda x: x): if baseline is not None and contender is not None: return [ metric, str(task), formatter(baseline), formatter(contender), self._diff(baseline, contender, treat_increase_as_improvement, formatter), unit, self._diff(baseline, contender, treat_increase_as_improvement, formatter, as_percentage=True), ] else: return [] def _diff(self, baseline, contender, treat_increase_as_improvement, formatter=lambda x: x, as_percentage=False): def identity(x): return x def _safe_divide(n, d): return n / d if d else 0 if self.plain: color_greater = identity color_smaller = identity color_neutral = identity elif treat_increase_as_improvement: color_greater = console.format.green color_smaller = console.format.red color_neutral = console.format.neutral else: color_greater = console.format.red color_smaller = console.format.green color_neutral = console.format.neutral if as_percentage: diff = _safe_divide(contender - baseline, baseline) * 100.0 precision = 2 suffix = "%" else: diff = formatter(contender - baseline) precision = 5 suffix = "" # ensures that numbers that appear as "zero" are also colored neutrally threshold = 10**-precision formatted = f"{diff:.{precision}f}{suffix}" if diff >= threshold: return color_greater(f"+{formatted}") elif diff <= -threshold: return color_smaller(formatted) else: return color_neutral(formatted)