|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import Awaitable, Callable, Optional
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def percentile(sorted_values: list[float], q: float) -> float:
|
|
if not sorted_values:
|
|
return 0.0
|
|
if len(sorted_values) == 1:
|
|
return float(sorted_values[0])
|
|
rank = (len(sorted_values) - 1) * q
|
|
low = int(rank)
|
|
high = min(low + 1, len(sorted_values) - 1)
|
|
frac = rank - low
|
|
return float(sorted_values[low] + (sorted_values[high] - sorted_values[low]) * frac)
|
|
|
|
|
|
class CircuitBreaker:
|
|
def __init__(self, threshold: int, cooldown_seconds: int):
|
|
self.threshold = threshold
|
|
self.cooldown_seconds = cooldown_seconds
|
|
self.failures = 0
|
|
self.opened_at: Optional[float] = None
|
|
|
|
def configure(self, threshold: int, cooldown_seconds: int) -> None:
|
|
self.threshold = threshold
|
|
self.cooldown_seconds = cooldown_seconds
|
|
|
|
@property
|
|
def is_open(self) -> bool:
|
|
return self.opened_at is not None
|
|
|
|
def allow(self) -> bool:
|
|
if self.opened_at is None:
|
|
return True
|
|
if (time.monotonic() - self.opened_at) >= self.cooldown_seconds:
|
|
self.opened_at = None
|
|
self.failures = 0
|
|
return True
|
|
return False
|
|
|
|
def record_success(self) -> None:
|
|
self.failures = 0
|
|
self.opened_at = None
|
|
|
|
def record_failure(self) -> None:
|
|
self.failures += 1
|
|
if self.threshold > 0 and self.failures >= self.threshold:
|
|
self.opened_at = time.monotonic()
|
|
|
|
|
|
async def _backoff(backoff_ms: int, attempt: int) -> None:
|
|
delay = max(0, backoff_ms) * attempt / 1000.0
|
|
if delay > 0:
|
|
await asyncio.sleep(delay)
|
|
|
|
|
|
async def retry_send(
|
|
do_call: Callable[[], Awaitable[httpx.Response]],
|
|
max_retries: int,
|
|
backoff_ms: int,
|
|
log: Optional[Callable[[str], None]] = None,
|
|
) -> tuple[Optional[httpx.Response], Optional[Exception], int]:
|
|
log = log or (lambda message: None)
|
|
attempts = 0
|
|
last_exc: Optional[Exception] = None
|
|
while attempts <= max_retries:
|
|
attempts += 1
|
|
try:
|
|
resp = await do_call()
|
|
except httpx.RequestError as exc:
|
|
last_exc = exc
|
|
if attempts > max_retries:
|
|
return None, exc, attempts
|
|
log(
|
|
f"upstream connection failed, retrying ({attempts}/{max_retries}): {exc}"
|
|
)
|
|
await _backoff(backoff_ms, attempts)
|
|
continue
|
|
if resp.status_code >= 500 and attempts <= max_retries:
|
|
log(f"upstream {resp.status_code}, retrying ({attempts}/{max_retries})")
|
|
await _backoff(backoff_ms, attempts)
|
|
continue
|
|
return resp, None, attempts
|
|
return None, last_exc, attempts
|