from datetime import datetime from typing import Optional import yaml from frostfs_testlib.load.interfaces.summarized import SummarizedStats from frostfs_testlib.load.load_config import K6ProcessAllocationStrategy, LoadParams, LoadScenario from frostfs_testlib.utils.converting_utils import calc_unit class LoadReport: def __init__(self, load_test) -> None: self.load_test = load_test # List of load summaries dict self.load_summaries_list: Optional[list[dict]] = [] self.load_params: Optional[LoadParams] = None self.start_time: Optional[datetime] = None self.end_time: Optional[datetime] = None def set_start_time(self, time: datetime = None): if time is None: time = datetime.utcnow() self.start_time = time def set_end_time(self, time: datetime = None): if time is None: time = datetime.utcnow() self.end_time = time def add_summaries(self, load_summaries: dict): self.load_summaries_list.append(load_summaries) def set_load_params(self, load_params: LoadParams): self.load_params = load_params def get_report_html(self): report_sections = [ [self.load_params, self._get_load_id_section_html], [self.load_test, self._get_load_params_section_html], [self.load_summaries_list, self._get_totals_section_html], [self.end_time, self._get_test_time_html], ] html = "" for section in report_sections: if section[0] is not None: html += section[1]() return html def _get_load_params_section_html(self) -> str: params: str = yaml.safe_dump([self.load_test], sort_keys=False, indent=2, explicit_start=True) params = params.replace("\n", "
").replace(" ", " ") section_html = f"""

Scenario params

{params}
""" return section_html def _get_load_id_section_html(self) -> str: section_html = f"""

Load ID: {self.load_params.load_id}

""" return section_html def _get_test_time_html(self) -> str: if not self.start_time or not self.end_time: return "" html = f"""

Scenario duration

{self.start_time} - {self.end_time}
""" return html def _seconds_to_formatted_duration(self, seconds: int) -> str: """Converts N number of seconds to formatted output ignoring zeroes. Examples: 186399 -> "2d3h46m39s" 86399 -> "23h59m59s" 86399 -> "23h59m59s" 3605 -> "1h5s" 123 -> "2m3s" """ units = {"d": 86400, "h": 3600, "m": 60, "s": 1} parts = [] remaining = seconds for divisor in units.values(): part = remaining // divisor remaining -= divisor * part parts.append(part) return "".join([f"{val}{unit}" for unit, val in zip(units, parts) if val > 0]) def _row(self, caption: str, value: str) -> str: return f"{caption}{value}" def _get_model_string(self): if self.load_params.min_iteration_duration is not None: return f"min_iteration_duration={self.load_params.min_iteration_duration}" model_map = { LoadScenario.gRPC: "closed model", LoadScenario.S3: "closed model", LoadScenario.S3_MULTIPART: "closed model", LoadScenario.HTTP: "closed model", LoadScenario.gRPC_CAR: "open model", LoadScenario.S3_CAR: "open model", LoadScenario.LOCAL: "local fill", LoadScenario.S3_LOCAL: "local fill", } return model_map[self.load_params.scenario] def _get_operations_sub_section_html(self, operation_type: str, stats: SummarizedStats): throughput_html = "" if stats.throughput > 0: throughput, unit = calc_unit(stats.throughput) throughput_html = self._row("Throughput", f"{throughput:.2f} {unit}/sec") per_node_errors_html = "" for node_key, errors in stats.errors.by_node.items(): if self.load_params.k6_process_allocation_strategy == K6ProcessAllocationStrategy.PER_ENDPOINT: per_node_errors_html += self._row(f"At {node_key}", errors) latency_html = "" for node_key, latencies in stats.latencies.by_node.items(): latency_values = "N/A" if latencies: latency_values = "" for param_name, param_val in latencies.items(): latency_values += f"{param_name}={param_val:.2f}ms " latency_html += self._row(f"{operation_type} latency {node_key.split(':')[0]}", latency_values) object_size, object_size_unit = calc_unit(self.load_params.object_size, 1) duration = self._seconds_to_formatted_duration(self.load_params.load_time) model = self._get_model_string() requested_rate_str = f"{stats.requested_rate}op/sec" if stats.requested_rate else "" # write 8KB 15h49m 50op/sec 50th open model/closed model/min_iteration duration=1s - 1.636MB/s 199.57451/s short_summary = f"{operation_type} {object_size}{object_size_unit} {duration} {requested_rate_str} {stats.threads}th {model} - {throughput:.2f}{unit}/s {stats.rate:.2f}/s" html = f""" {self._row("Total operations", stats.operations)} {self._row("OP/sec", f"{stats.rate:.2f}")} {throughput_html} {latency_html} {per_node_errors_html} {self._row("Total", f"{stats.errors.total} ({stats.errors.percent:.2f}%)")} {self._row("Threshold", f"{stats.errors.threshold:.2f}%")}
{short_summary}
Metrics
Errors

""" return html def _get_totals_section_html(self): html = "" for i in range(len(self.load_summaries_list)): html += f"

Load Results for load #{i+1}

" summarized = SummarizedStats.collect(self.load_params, self.load_summaries_list[i]) for operation_type, stats in summarized.items(): if stats.operations: html += self._get_operations_sub_section_html(operation_type, stats) return html