tegwick
beab2a04d1
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0260c-4067-7052-9647-ad000d576e38
73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from .model import Outcome
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CapacitySample:
|
|
consumer: str
|
|
latency_ms: float
|
|
error_rate: float
|
|
throughput_per_second: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CapacityResult:
|
|
outcome: Outcome
|
|
governor_bound: bool
|
|
aggressor_peak: float
|
|
aggressor_ceiling: float
|
|
neighbour_degradation: dict[str, dict[str, float]]
|
|
reasons: list[str]
|
|
|
|
|
|
def characterize(*, baseline: list[CapacitySample], loaded: list[CapacitySample],
|
|
governor_bound: bool, aggressor_peak: float,
|
|
aggressor_ceiling: float) -> CapacityResult:
|
|
before = {sample.consumer: sample for sample in baseline}
|
|
after = {sample.consumer: sample for sample in loaded}
|
|
reasons: list[str] = []
|
|
if not governor_bound:
|
|
reasons.append("declared governor did not bind")
|
|
if aggressor_peak > aggressor_ceiling:
|
|
reasons.append("probe exceeded its authorized allowance")
|
|
missing = sorted(before.keys() - after.keys())
|
|
if missing:
|
|
reasons.append("loaded samples missing consumers: " + ", ".join(missing))
|
|
|
|
degradation: dict[str, dict[str, float]] = {}
|
|
for consumer in sorted(before.keys() & after.keys()):
|
|
old, new = before[consumer], after[consumer]
|
|
degradation[consumer] = {
|
|
"latency_increase_percent": _change(old.latency_ms, new.latency_ms),
|
|
"error_rate_increase_points": new.error_rate - old.error_rate,
|
|
"throughput_decrease_percent": _decrease(old.throughput_per_second,
|
|
new.throughput_per_second),
|
|
}
|
|
outcome: Outcome = "pass"
|
|
if aggressor_peak > aggressor_ceiling:
|
|
outcome = "aborted"
|
|
elif reasons:
|
|
outcome = "finding"
|
|
return CapacityResult(
|
|
outcome=outcome,
|
|
governor_bound=governor_bound,
|
|
aggressor_peak=aggressor_peak,
|
|
aggressor_ceiling=aggressor_ceiling,
|
|
neighbour_degradation=degradation,
|
|
reasons=reasons,
|
|
)
|
|
|
|
|
|
def _change(before: float, after: float) -> float:
|
|
if before == 0:
|
|
return 0.0 if after == 0 else float("inf")
|
|
return round((after - before) / before * 100, 3)
|
|
|
|
|
|
def _decrease(before: float, after: float) -> float:
|
|
if before == 0:
|
|
return 0.0
|
|
return round((before - after) / before * 100, 3)
|