impl/python.py

from .monitor_error_budget_types import ErrorBudget


def _whole(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _check_counts(target_basis_points: int, total_events: int, bad_events: int) -> None:
    if not _whole(target_basis_points) or target_basis_points < 1 or target_basis_points > 9999:
        raise ValueError("targetBasisPoints must be a whole number from 1 to 9999 (10000 leaves no error budget), received %s" % (target_basis_points,))
    if not _whole(total_events) or total_events < 0:
        raise ValueError("totalEvents must be a whole number of at least 0, received %s" % (total_events,))
    if not _whole(bad_events) or bad_events < 0:
        raise ValueError("badEvents must be a whole number of at least 0, received %s" % (bad_events,))
    if bad_events > total_events:
        raise ValueError("badEvents must not exceed totalEvents: %d > %d" % (bad_events, total_events))


def _half_up(n: int, d: int) -> int:
    # Non-negative operands only, so half-up needs no sign handling.
    return (2 * n + d) // (2 * d)


def error_budget(target_basis_points: int, total_events: int, bad_events: int) -> ErrorBudget:
    """The error budget of an SLO over one window: how many bad events it
    allows, how many are left, and how much of it is spent. Exact integers
    throughout."""
    _check_counts(target_basis_points, total_events, bad_events)
    # The exact budget is budget_times_10000 / 10000 events; keep it as a fraction.
    budget_times_10000 = total_events * (10000 - target_basis_points)
    allowed_bad = budget_times_10000 // 10000
    consumed = 0 if total_events == 0 else _half_up(bad_events * 10000 * 10000, budget_times_10000)
    return ErrorBudget(
        allowed_bad=allowed_bad,
        remaining_bad=allowed_bad - bad_events,
        consumed_basis_points=consumed,
        remaining_basis_points=10000 - consumed,
        exhausted=bad_events * 10000 > budget_times_10000,
    )