SLO Readiness Check¶

add_slo_readiness_check makes Kubernetes /ready return 503 when a named SLO's error budget falls below a threshold — preventing traffic from being routed to a service that is burning through its budget.

Quick Start¶

Python

from obskit import add_slo_readiness_check, configure_observability

obs = configure_observability(service_name="orders-api")

# Register the SLO first
obs.metrics.register_slo("availability", target=0.999)

# Then gate readiness on budget health
add_slo_readiness_check("availability", critical_threshold=0.10)
# /health/ready → 503 when error budget < 10 %

Parameters¶

Parameter	Default	Description
`slo_name`	—	Name of a registered SLO
`critical_threshold`	`0.10`	Budget fraction below which readiness fails
`warning_threshold`	`0.25`	Budget fraction that triggers a warning (still healthy)
`health_checker`	global	`HealthChecker` instance to register with

Budget thresholds¶

Text Only

budget ≥ warning_threshold              → healthy
warning_threshold > budget ≥ critical   → warning (still healthy, logged)
budget < critical_threshold             → CRITICAL — /ready returns 503

Multiple SLOs¶

Python

add_slo_readiness_check("availability", critical_threshold=0.05)
add_slo_readiness_check("latency_p99",  critical_threshold=0.20)
# Both must pass for /ready to return 200

Direct check¶

Python

from obskit.health.slo_check import SLOReadinessCheck

check = SLOReadinessCheck("availability", critical_threshold=0.10)
result = check.check()
print(result.healthy, result.error_budget_remaining)

Retrieve a registered check¶

New in v1.7.0. Use get_slo_readiness_check to retrieve an already-registered check by name — useful for tests or monitoring dashboards that want to inspect the current budget without re-registering:

Python

from obskit import add_slo_readiness_check, get_slo_readiness_check

add_slo_readiness_check("availability", critical_threshold=0.10)

# Later — retrieve the same check object
check = get_slo_readiness_check("availability")
result = check.check()
print(result.healthy, result.error_budget_remaining)

add_slo_readiness_check is idempotent — calling it a second time with the same name returns the existing check without creating a duplicate.

API Reference¶

obskit.health.slo_check.add_slo_readiness_check ¶

Python

add_slo_readiness_check(
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
    health_checker: HealthChecker | None = None,
) -> SLOReadinessCheck

Register an SLO-based readiness check with the global health checker.

Idempotent — calling this function multiple times with the same slo_name (e.g. on every /_ready request) registers the handler only once and returns the existing :class:SLOReadinessCheck on subsequent calls. It is therefore safe to call per-request, but the recommended pattern is to call it once at startup (in the app lifespan) and use :func:get_slo_readiness_check for per-request budget checks.

Parameters¶

slo_name : str Name of the SLO to check. critical_threshold : float Error budget threshold for failing readiness (default: 0.1). warning_threshold : float Error budget threshold for warning (default: 0.25). health_checker : HealthChecker, optional Health checker to add check to (defaults to global).

Returns¶

SLOReadinessCheck The check instance (existing one if already registered).

Example — startup registration (recommended)¶

In your app lifespan:¶

add_slo_readiness_check("availability", critical_threshold=0.1) add_slo_readiness_check("latency_p95", critical_threshold=0.1)

Example — per-request (also safe due to idempotency)¶

check = add_slo_readiness_check("availability", critical_threshold=0.1) result = check.check()

Source code in src/obskit/health/slo_check.py

Python
def add_slo_readiness_check(
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
    health_checker: HealthChecker | None = None,
) -> SLOReadinessCheck:
    """
    Register an SLO-based readiness check with the global health checker.

    **Idempotent** — calling this function multiple times with the same
    *slo_name* (e.g. on every ``/_ready`` request) registers the handler only
    once and returns the existing :class:`SLOReadinessCheck` on subsequent
    calls.  It is therefore safe to call per-request, but the recommended
    pattern is to call it once at startup (in the app lifespan) and use
    :func:`get_slo_readiness_check` for per-request budget checks.

    Parameters
    ----------
    slo_name : str
        Name of the SLO to check.
    critical_threshold : float
        Error budget threshold for failing readiness (default: 0.1).
    warning_threshold : float
        Error budget threshold for warning (default: 0.25).
    health_checker : HealthChecker, optional
        Health checker to add check to (defaults to global).

    Returns
    -------
    SLOReadinessCheck
        The check instance (existing one if already registered).

    Example — startup registration (recommended)
    --------------------------------------------
    >>> # In your app lifespan:
    >>> add_slo_readiness_check("availability", critical_threshold=0.1)
    >>> add_slo_readiness_check("latency_p95", critical_threshold=0.1)

    Example — per-request (also safe due to idempotency)
    -----------------------------------------------------
    >>> check = add_slo_readiness_check("availability", critical_threshold=0.1)
    >>> result = check.check()
    """
    if slo_name in _REGISTERED_SLO_CHECKS:
        return _REGISTERED_SLO_CHECKS[slo_name]

    checker = health_checker or get_health_checker()
    slo_check = SLOReadinessCheck(
        slo_name=slo_name,
        critical_threshold=critical_threshold,
        warning_threshold=warning_threshold,
    )

    # Create async check function
    async def check_slo() -> dict[str, Any] | bool:  # NOSONAR
        result = slo_check.check()
        if not result.healthy:
            return {
                "healthy": False,
                "message": result.message,
                "details": {
                    "error_budget_remaining": result.error_budget_remaining,
                    "current_value": result.current_value,
                    "target_value": result.target_value,
                },
            }
        return True

    # Register with health checker
    checker.add_readiness_check(f"slo_{slo_name}")(check_slo)

    _REGISTERED_SLO_CHECKS[slo_name] = slo_check

    logger.info(
        "slo_readiness_check_added",
        slo_name=slo_name,
        critical_threshold=critical_threshold,
        warning_threshold=warning_threshold,
    )

    return slo_check

obskit.health.slo_check.get_slo_readiness_check ¶

Python

get_slo_readiness_check(
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
) -> SLOReadinessCheck

Return a lightweight SLO readiness check — safe to call per-request.

Unlike :func:add_slo_readiness_check, this function does not register anything with the global health checker. It creates a plain :class:SLOReadinessCheck that you can call .check() on directly.

Use this inside per-request handlers (e.g. /_ready) once the SLO has already been registered at startup with :func:add_slo_readiness_check.

Parameters¶

slo_name : str Name of the SLO to check. critical_threshold : float Error budget threshold below which the check is unhealthy (default: 0.1). warning_threshold : float Error budget threshold below which the check is in warning (default: 0.25).

Returns¶

SLOReadinessCheck A check object whose .check() method can be called immediately.

Example¶

In a /_ready handler (called on every request):¶

check = get_slo_readiness_check("api_availability", critical_threshold=0.1) if not check.check().healthy: ... return JSONResponse({"ready": False}, status_code=503)

Source code in src/obskit/health/slo_check.py

Python
def get_slo_readiness_check(
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
) -> SLOReadinessCheck:
    """
    Return a lightweight SLO readiness check — safe to call per-request.

    Unlike :func:`add_slo_readiness_check`, this function does **not** register
    anything with the global health checker.  It creates a plain
    :class:`SLOReadinessCheck` that you can call ``.check()`` on directly.

    Use this inside per-request handlers (e.g. ``/_ready``) once the SLO has
    already been registered at startup with :func:`add_slo_readiness_check`.

    Parameters
    ----------
    slo_name : str
        Name of the SLO to check.
    critical_threshold : float
        Error budget threshold below which the check is unhealthy (default: 0.1).
    warning_threshold : float
        Error budget threshold below which the check is in warning (default: 0.25).

    Returns
    -------
    SLOReadinessCheck
        A check object whose ``.check()`` method can be called immediately.

    Example
    -------
    >>> # In a /_ready handler (called on every request):
    >>> check = get_slo_readiness_check("api_availability", critical_threshold=0.1)
    >>> if not check.check().healthy:
    ...     return JSONResponse({"ready": False}, status_code=503)
    """
    return SLOReadinessCheck(
        slo_name=slo_name,
        critical_threshold=critical_threshold,
        warning_threshold=warning_threshold,
    )

obskit.health.slo_check.SLOReadinessCheck ¶

Health check based on SLO compliance.

Fails readiness if error budget falls below critical threshold.

Example¶

check = SLOReadinessCheck( ... slo_name="availability", ... critical_threshold=0.1, ... warning_threshold=0.25, ... ) result = check.check() if not result.healthy: ... print(f"SLO failing: {result.message}")

Source code in src/obskit/health/slo_check.py

Python
class SLOReadinessCheck:
    """
    Health check based on SLO compliance.

    Fails readiness if error budget falls below critical threshold.

    Example
    -------
    >>> check = SLOReadinessCheck(
    ...     slo_name="availability",
    ...     critical_threshold=0.1,
    ...     warning_threshold=0.25,
    ... )
    >>> result = check.check()
    >>> if not result.healthy:
    ...     print(f"SLO failing: {result.message}")
    """

    def __init__(
        self,
        slo_name: str,
        critical_threshold: float = 0.1,
        warning_threshold: float = 0.25,
        slo_tracker: SLOTracker | None = None,
    ):
        """
        Initialize SLO readiness check.

        Parameters
        ----------
        slo_name : str
            Name of the SLO to check.
        critical_threshold : float
            Error budget threshold for critical/unhealthy (default: 0.1 = 10%).
        warning_threshold : float
            Error budget threshold for warning (default: 0.25 = 25%).
        slo_tracker : SLOTracker, optional
            SLO tracker instance (defaults to global).
        """
        self.slo_name = slo_name
        self.critical_threshold = critical_threshold
        self.warning_threshold = warning_threshold
        self._slo_tracker = slo_tracker

    @property
    def slo_tracker(self) -> SLOTracker | None:
        """Get SLO tracker lazily."""
        if self._slo_tracker is None:
            self._slo_tracker = get_slo_tracker()
        return self._slo_tracker

    def check(self) -> SLOCheckResult:
        """
        Perform the SLO health check.

        Returns
        -------
        SLOCheckResult
            Check result with status and details.
        """
        if not self.slo_tracker:
            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.UNKNOWN,
                healthy=True,  # Don't fail if tracker not available
                error_budget_remaining=1.0,
                current_value=0.0,
                target_value=0.0,
                message="SLO tracker not available",
            )

        try:
            status = self.slo_tracker.get_status(self.slo_name)

            if status is None:
                return SLOCheckResult(
                    slo_name=self.slo_name,
                    status=SLOHealthStatus.UNKNOWN,
                    healthy=True,  # Don't fail for unregistered SLO
                    error_budget_remaining=1.0,
                    current_value=0.0,
                    target_value=0.0,
                    message=f"SLO '{self.slo_name}' not registered",
                )

            budget = status.error_budget_remaining
            current = status.current_value
            target = status.target.target_value

            # Determine health status
            if budget < self.critical_threshold:
                return SLOCheckResult(
                    slo_name=self.slo_name,
                    status=SLOHealthStatus.CRITICAL,
                    healthy=False,
                    error_budget_remaining=budget,
                    current_value=current,
                    target_value=target,
                    message=f"Error budget critically low: {budget:.1%} (threshold: {self.critical_threshold:.0%})",
                    details={
                        "burn_rate": status.error_budget_burn_rate,
                        "critical_threshold": self.critical_threshold,
                        "warning_threshold": self.warning_threshold,
                    },
                )

            if budget < self.warning_threshold:
                return SLOCheckResult(
                    slo_name=self.slo_name,
                    status=SLOHealthStatus.WARNING,
                    healthy=True,  # Still healthy, just warning
                    error_budget_remaining=budget,
                    current_value=current,
                    target_value=target,
                    message=f"Error budget low: {budget:.1%} (warning: {self.warning_threshold:.0%})",
                    details={
                        "burn_rate": status.error_budget_burn_rate,
                    },
                )

            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.HEALTHY,
                healthy=True,
                error_budget_remaining=budget,
                current_value=current,
                target_value=target,
                message=f"SLO healthy: {budget:.1%} error budget remaining",
            )

        except Exception as e:
            logger.error("slo_check_failed", slo_name=self.slo_name, error=str(e))
            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.UNKNOWN,
                healthy=True,  # Don't fail on check errors
                error_budget_remaining=1.0,
                current_value=0.0,
                target_value=0.0,
                message=f"SLO check error: {str(e)}",
            )

slo_tracker `property` ¶

Python

slo_tracker: SLOTracker | None

Get SLO tracker lazily.

init ¶

Python

__init__(
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
    slo_tracker: SLOTracker | None = None,
)

Initialize SLO readiness check.

Parameters¶

slo_name : str Name of the SLO to check. critical_threshold : float Error budget threshold for critical/unhealthy (default: 0.1 = 10%). warning_threshold : float Error budget threshold for warning (default: 0.25 = 25%). slo_tracker : SLOTracker, optional SLO tracker instance (defaults to global).

Source code in src/obskit/health/slo_check.py

Python
def __init__(
    self,
    slo_name: str,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
    slo_tracker: SLOTracker | None = None,
):
    """
    Initialize SLO readiness check.

    Parameters
    ----------
    slo_name : str
        Name of the SLO to check.
    critical_threshold : float
        Error budget threshold for critical/unhealthy (default: 0.1 = 10%).
    warning_threshold : float
        Error budget threshold for warning (default: 0.25 = 25%).
    slo_tracker : SLOTracker, optional
        SLO tracker instance (defaults to global).
    """
    self.slo_name = slo_name
    self.critical_threshold = critical_threshold
    self.warning_threshold = warning_threshold
    self._slo_tracker = slo_tracker

check ¶

Python

check() -> SLOCheckResult

Perform the SLO health check.

Returns¶

SLOCheckResult Check result with status and details.

Source code in src/obskit/health/slo_check.py

Python
def check(self) -> SLOCheckResult:
    """
    Perform the SLO health check.

    Returns
    -------
    SLOCheckResult
        Check result with status and details.
    """
    if not self.slo_tracker:
        return SLOCheckResult(
            slo_name=self.slo_name,
            status=SLOHealthStatus.UNKNOWN,
            healthy=True,  # Don't fail if tracker not available
            error_budget_remaining=1.0,
            current_value=0.0,
            target_value=0.0,
            message="SLO tracker not available",
        )

    try:
        status = self.slo_tracker.get_status(self.slo_name)

        if status is None:
            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.UNKNOWN,
                healthy=True,  # Don't fail for unregistered SLO
                error_budget_remaining=1.0,
                current_value=0.0,
                target_value=0.0,
                message=f"SLO '{self.slo_name}' not registered",
            )

        budget = status.error_budget_remaining
        current = status.current_value
        target = status.target.target_value

        # Determine health status
        if budget < self.critical_threshold:
            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.CRITICAL,
                healthy=False,
                error_budget_remaining=budget,
                current_value=current,
                target_value=target,
                message=f"Error budget critically low: {budget:.1%} (threshold: {self.critical_threshold:.0%})",
                details={
                    "burn_rate": status.error_budget_burn_rate,
                    "critical_threshold": self.critical_threshold,
                    "warning_threshold": self.warning_threshold,
                },
            )

        if budget < self.warning_threshold:
            return SLOCheckResult(
                slo_name=self.slo_name,
                status=SLOHealthStatus.WARNING,
                healthy=True,  # Still healthy, just warning
                error_budget_remaining=budget,
                current_value=current,
                target_value=target,
                message=f"Error budget low: {budget:.1%} (warning: {self.warning_threshold:.0%})",
                details={
                    "burn_rate": status.error_budget_burn_rate,
                },
            )

        return SLOCheckResult(
            slo_name=self.slo_name,
            status=SLOHealthStatus.HEALTHY,
            healthy=True,
            error_budget_remaining=budget,
            current_value=current,
            target_value=target,
            message=f"SLO healthy: {budget:.1%} error budget remaining",
        )

    except Exception as e:
        logger.error("slo_check_failed", slo_name=self.slo_name, error=str(e))
        return SLOCheckResult(
            slo_name=self.slo_name,
            status=SLOHealthStatus.UNKNOWN,
            healthy=True,  # Don't fail on check errors
            error_budget_remaining=1.0,
            current_value=0.0,
            target_value=0.0,
            message=f"SLO check error: {str(e)}",
        )

obskit.health.slo_check.get_slo_health_status ¶

Python

get_slo_health_status(
    slo_names: list[str] | None = None,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
) -> dict[str, Any]

Get health status for multiple SLOs.

Parameters¶

slo_names : list of str, optional SLO names to check (defaults to all registered). critical_threshold : float Threshold for critical status. warning_threshold : float Threshold for warning status.

Returns¶

dict Health status summary.

Example¶

status = get_slo_health_status(["availability", "latency_p99"]) print(f"Overall healthy: {status['healthy']}")

Source code in src/obskit/health/slo_check.py

Python
def get_slo_health_status(
    slo_names: list[str] | None = None,
    critical_threshold: float = 0.1,
    warning_threshold: float = 0.25,
) -> dict[str, Any]:
    """
    Get health status for multiple SLOs.

    Parameters
    ----------
    slo_names : list of str, optional
        SLO names to check (defaults to all registered).
    critical_threshold : float
        Threshold for critical status.
    warning_threshold : float
        Threshold for warning status.

    Returns
    -------
    dict
        Health status summary.

    Example
    -------
    >>> status = get_slo_health_status(["availability", "latency_p99"])
    >>> print(f"Overall healthy: {status['healthy']}")
    """
    tracker = get_slo_tracker()

    if not tracker:
        return {
            "healthy": True,
            "status": "unknown",
            "message": "SLO tracker not available",
            "slos": {},
        }

    # Get all SLO names if not specified
    if slo_names is None:
        try:
            slo_names = list(tracker._slos.keys()) if hasattr(tracker, "_slos") else []
        except Exception:
            slo_names = []

    results = {}
    overall_healthy = True
    overall_status = SLOHealthStatus.HEALTHY

    for name in slo_names:
        check = SLOReadinessCheck(
            slo_name=name,
            critical_threshold=critical_threshold,
            warning_threshold=warning_threshold,
            slo_tracker=tracker,
        )
        result = check.check()

        results[name] = {
            "status": result.status.value,
            "healthy": result.healthy,
            "error_budget_remaining": round(result.error_budget_remaining, 4),
            "current_value": round(result.current_value, 4),
            "target_value": round(result.target_value, 4),
            "message": result.message,
        }

        if not result.healthy:
            overall_healthy = False

        if result.status == SLOHealthStatus.CRITICAL:
            overall_status = SLOHealthStatus.CRITICAL
        elif result.status == SLOHealthStatus.WARNING and overall_status == SLOHealthStatus.HEALTHY:
            overall_status = SLOHealthStatus.WARNING

    return {
        "healthy": overall_healthy,
        "status": overall_status.value,
        "slos_checked": len(results),
        "slos": results,
    }

SLO Readiness Check¶

Quick Start¶

Parameters¶

Budget thresholds¶

Multiple SLOs¶

Direct check¶

Retrieve a registered check¶

API Reference¶

obskit.health.slo_check.add_slo_readiness_check ¶

Parameters¶

Returns¶

Example — startup registration (recommended)¶

In your app lifespan:¶

Example — per-request (also safe due to idempotency)¶

obskit.health.slo_check.get_slo_readiness_check ¶

Parameters¶

Returns¶

Example¶

In a /_ready handler (called on every request):¶

obskit.health.slo_check.SLOReadinessCheck ¶

Example¶

slo_tracker property ¶

__init__ ¶

Parameters¶

check ¶

Returns¶

obskit.health.slo_check.get_slo_health_status ¶

Parameters¶

Returns¶

Example¶

slo_tracker `property` ¶

init ¶