This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr-orbit.git
The following commit(s) were added to refs/heads/main by this push:
new 05f15a43 Fix: aggregate crashes when an operation reports null metrics
(#58)
05f15a43 is described below
commit 05f15a4321c6399fe4240fa231747886f8e34249
Author: Serhiy Bzhezytskyy <[email protected]>
AuthorDate: Thu Jul 30 02:19:41 2026 +0300
Fix: aggregate crashes when an operation reports null metrics (#58)
---
solrorbit/aggregator.py | 42 ++++++++++++++++++++++++++++++++----------
tests/aggregator_test.py | 41 +++++++++++++++++++++++++++++++++++++++++
2 files changed, 73 insertions(+), 10 deletions(-)
diff --git a/solrorbit/aggregator.py b/solrorbit/aggregator.py
index 6d0a7790..fb31f3e6 100644
--- a/solrorbit/aggregator.py
+++ b/solrorbit/aggregator.py
@@ -1,6 +1,6 @@
import os
import statistics
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, Union
import uuid
from solrorbit.metrics import FileTestRunStore, TestRun
@@ -213,7 +213,6 @@ class Aggregator:
# Get iterations for each test run
iterations_per_run = [self.accumulated_iterations[test_id][task_name]
for test_id in self.test_runs.keys()]
- total_iterations = sum(iterations_per_run)
for metric, values in task_metrics.items():
if isinstance(values[0], dict):
@@ -222,23 +221,46 @@ class Aggregator:
if metric_field == 'unit':
weighted_metrics[metric][metric_field] =
values[0][metric_field]
elif metric_field == 'min':
- weighted_metrics[metric]['overall_min'] =
min(value.get(metric_field, 0) for value in values)
+ item_values = [value.get(metric_field) for value in
values]
+ weighted_metrics[metric]['overall_min'] = min(
+ (value for value in item_values if value is not
None), default=None)
elif metric_field == 'max':
- weighted_metrics[metric]['overall_max'] =
max(value.get(metric_field, 0) for value in values)
+ item_values = [value.get(metric_field) for value in
values]
+ weighted_metrics[metric]['overall_max'] = max(
+ (value for value in item_values if value is not
None), default=None)
else:
# for items like median or containing percentile values
- item_values = [value.get(metric_field, 0) for value in
values]
- weighted_sum = sum(value * iterations for value,
iterations in zip(item_values, iterations_per_run))
- weighted_metrics[metric][metric_field] = weighted_sum
/ total_iterations
+ item_values = [value.get(metric_field) for value in
values]
+ weighted_metrics[metric][metric_field] =
self.weighted_mean(item_values, iterations_per_run)
else:
- weighted_sum = sum(value * iterations for value, iterations in
zip(values, iterations_per_run))
- weighted_metrics[metric] = weighted_sum / total_iterations
+ weighted_metrics[metric] = self.weighted_mean(values,
iterations_per_run)
return weighted_metrics
- def calculate_rsd(self, values: List[Union[int, float]], metric_name: str)
-> Union[float, str]:
+ @staticmethod
+ def weighted_mean(values: List[Optional[Union[int, float]]],
+ iterations_per_run: List[int]) -> Optional[float]:
+ """
+ Weights each test run's value by that run's iteration count.
+ Operations that produced no valid samples report their metrics as
None; those are left out
+ of both the sum and the divisor, and the result is None when no run
contributed a value,
+ so that "not measured" stays distinct from a measured zero
+ """
+ contributions = [(value, iterations) for value, iterations in
zip(values, iterations_per_run)
+ if value is not None]
+ total_iterations = sum(iterations for _, iterations in contributions)
+ if not total_iterations:
+ return None
+ return sum(value * iterations for value, iterations in contributions)
/ total_iterations
+
+ def calculate_rsd(self, values: List[Optional[Union[int, float]]],
+ metric_name: str) -> Union[float, str]:
if not values:
raise ValueError(f"Cannot calculate RSD for metric
'{metric_name}': empty list of values")
+ # operations that produced no valid samples report None, which cannot
contribute to a deviation
+ values = [value for value in values if value is not None]
+ if not values:
+ return "NA" # no test run measured this metric
if len(values) == 1:
return "NA" # RSD is not applicable for a single value
mean = statistics.mean(values)
diff --git a/tests/aggregator_test.py b/tests/aggregator_test.py
index 50a3d9d9..44efecbc 100644
--- a/tests/aggregator_test.py
+++ b/tests/aggregator_test.py
@@ -121,6 +121,47 @@ def test_calculate_weighted_average(aggregator):
assert result["latency"]["avg"] == 16 # (10*2 + 20*3) / (2+3)
assert result["latency"]["unit"] == "ms"
+def test_calculate_weighted_average_with_null_metric_fields(aggregator):
+ # An operation that produced no valid samples reports null metric fields,
e.g. `optimize`
+ # in the geonames workload, which reports error_rate 1.0 and a fully null
throughput.
+ task_metrics = {
+ "throughput": [
+ {"min": None, "mean": None, "median": None, "max": None, "unit":
"ops/s"},
+ {"min": None, "mean": None, "median": None, "max": None, "unit":
"ops/s"}
+ ]
+ }
+ aggregator.accumulated_iterations = {"test1": {"op1": 1}, "test2": {"op1":
1}}
+ aggregator.test_runs = {"test1": Mock(), "test2": Mock()}
+
+ result = aggregator.calculate_weighted_average(task_metrics, "op1")
+
+ # null is carried through rather than replaced by 0, which would read as a
real
+ # measurement of zero throughput
+ assert result["throughput"]["overall_min"] is None
+ assert result["throughput"]["overall_max"] is None
+ assert result["throughput"]["mean"] is None
+ assert result["throughput"]["median"] is None
+ assert result["throughput"]["unit"] == "ops/s"
+
+def
test_calculate_weighted_average_with_partially_null_metric_fields(aggregator):
+ # A metric that has samples in one test run but not another: the valid
values are
+ # still aggregated, weighted only by the runs that contributed them.
+ task_metrics = {
+ "throughput": [
+ {"min": 10, "mean": 20, "median": 20, "max": 30, "unit": "ops/s"},
+ {"min": None, "mean": None, "median": None, "max": None, "unit":
"ops/s"}
+ ]
+ }
+ aggregator.accumulated_iterations = {"test1": {"op1": 2}, "test2": {"op1":
3}}
+ aggregator.test_runs = {"test1": Mock(), "test2": Mock()}
+
+ result = aggregator.calculate_weighted_average(task_metrics, "op1")
+
+ assert result["throughput"]["overall_min"] == 10
+ assert result["throughput"]["overall_max"] == 30
+ assert result["throughput"]["mean"] == 20
+ assert result["throughput"]["unit"] == "ops/s"
+
def
test_update_config_object_reads_attributes_that_test_runs_actually_have(aggregator):
# a real TestRun, not a Mock: a Mock creates whatever attribute is asked
of it, so it cannot
# tell us whether update_config_object reads attributes that a test run
really carries
