from datetime import datetime
from typing import Optional
import yaml
from frostfs_testlib.load.load_config import K6ProcessAllocationStrategy, LoadParams, LoadScenario
from frostfs_testlib.load.load_metrics import get_metrics_object
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,
total_operations: int,
requested_rate_str: str,
vus_str: str,
total_rate: float,
throughput: float,
errors: dict[str, int],
latency: dict[str, dict],
):
throughput_html = ""
if throughput > 0:
throughput, unit = calc_unit(throughput)
throughput_html = self._row("Throughput", f"{throughput:.2f} {unit}/sec")
per_node_errors_html = ""
total_errors = 0
if errors:
total_errors: int = 0
for node_key, errors in errors.items():
total_errors += errors
if self.load_params.k6_process_allocation_strategy == K6ProcessAllocationStrategy.PER_ENDPOINT:
per_node_errors_html += self._row(f"At {node_key}", errors)
latency_html = ""
if latency:
for node_key, latency_dict in latency.items():
latency_values = "N/A"
if latency_dict:
latency_values = ""
for param_name, param_val in latency_dict.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()
# 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} {vus_str} {model} - {throughput:.2f}{unit}/s {total_rate:.2f}/s"
errors_percent = 0
if total_operations:
errors_percent = total_errors / total_operations * 100.0
html = f"""
| {short_summary} |
|---|
| Metrics |
|---|
{self._row("Total operations", total_operations)}
{self._row("OP/sec", f"{total_rate:.2f}")}
{throughput_html}
{latency_html}
| Errors |
|---|
{per_node_errors_html}
{self._row("Total", f"{total_errors} ({errors_percent:.2f}%)")}
{self._row("Threshold", f"{self.load_params.error_threshold:.2f}%")}
"""
return html
def _get_totals_section_html(self):
html = ""
for i, load_summaries in enumerate(self.load_summaries_list, 1):
html += f"Load Results for load #{i}
"
write_operations = 0
write_op_sec = 0
write_throughput = 0
write_latency = {}
write_errors = {}
requested_write_rate = self.load_params.write_rate
requested_write_rate_str = f"{requested_write_rate}op/sec" if requested_write_rate else ""
read_operations = 0
read_op_sec = 0
read_throughput = 0
read_latency = {}
read_errors = {}
requested_read_rate = self.load_params.read_rate
requested_read_rate_str = f"{requested_read_rate}op/sec" if requested_read_rate else ""
delete_operations = 0
delete_op_sec = 0
delete_latency = {}
delete_errors = {}
requested_delete_rate = self.load_params.delete_rate
requested_delete_rate_str = f"{requested_delete_rate}op/sec" if requested_delete_rate else ""
if self.load_params.scenario in [LoadScenario.gRPC_CAR, LoadScenario.S3_CAR]:
delete_vus = max(self.load_params.preallocated_deleters or 0, self.load_params.max_deleters or 0)
write_vus = max(self.load_params.preallocated_writers or 0, self.load_params.max_writers or 0)
read_vus = max(self.load_params.preallocated_readers or 0, self.load_params.max_readers or 0)
else:
write_vus = self.load_params.writers
read_vus = self.load_params.readers
delete_vus = self.load_params.deleters
write_vus_str = f"{write_vus}th"
read_vus_str = f"{read_vus}th"
delete_vus_str = f"{delete_vus}th"
write_section_required = False
read_section_required = False
delete_section_required = False
for node_key, load_summary in load_summaries.items():
metrics = get_metrics_object(self.load_params.scenario, load_summary)
write_operations += metrics.write_total_iterations
if write_operations:
write_section_required = True
write_op_sec += metrics.write_rate
write_latency[node_key] = metrics.write_latency
write_throughput += metrics.write_throughput
if metrics.write_failed_iterations:
write_errors[node_key] = metrics.write_failed_iterations
read_operations += metrics.read_total_iterations
if read_operations:
read_section_required = True
read_op_sec += metrics.read_rate
read_throughput += metrics.read_throughput
read_latency[node_key] = metrics.read_latency
if metrics.read_failed_iterations:
read_errors[node_key] = metrics.read_failed_iterations
delete_operations += metrics.delete_total_iterations
if delete_operations:
delete_section_required = True
delete_op_sec += metrics.delete_rate
delete_latency[node_key] = metrics.delete_latency
if metrics.delete_failed_iterations:
delete_errors[node_key] = metrics.delete_failed_iterations
if write_section_required:
html += self._get_operations_sub_section_html(
"Write",
write_operations,
requested_write_rate_str,
write_vus_str,
write_op_sec,
write_throughput,
write_errors,
write_latency,
)
if read_section_required:
html += self._get_operations_sub_section_html(
"Read",
read_operations,
requested_read_rate_str,
read_vus_str,
read_op_sec,
read_throughput,
read_errors,
read_latency,
)
if delete_section_required:
html += self._get_operations_sub_section_html(
"Delete",
delete_operations,
requested_delete_rate_str,
delete_vus_str,
delete_op_sec,
0,
delete_errors,
delete_latency,
)
return html