Testing Against SLOs and SLAs: Error Budgets, Availability, and Reliability Verification

Error budget = (1 - 0.999) × 30 days × 24 hours × 60 minutes
             = 0.001 × 43,200 minutes
             = 43.2 minutes

from dataclasses import dataclass
from typing import Callable
import statistics
import time

@dataclass
class SLO:
    name: str
    description: str
    sli_query: Callable  # Function that returns current SLI value
    target: float  # e.g., 0.999 for 99.9%
    window_days: int
    alert_burn_rate: float = 14.4  # Alert if burning budget 14.4x faster (1-hour window)

class SLOTest:
    def __init__(self, slo: SLO, prometheus_client):
        self.slo = slo
        self.prom = prometheus_client
    
    def verify_current_compliance(self) -> bool:
        """Check if SLO is currently being met."""
        current_sli = self.slo.sli_query(self.prom)
        return current_sli >= self.slo.target
    
    def calculate_burn_rate(self, window_hours: int = 1) -> float:
        """Calculate current error budget burn rate."""
        error_budget = 1 - self.slo.target
        
        # Current error rate over the window
        current_error_rate = 1 - self.slo.sli_query(self.prom, window_hours=window_hours)
        
        # Burn rate = actual error rate / allowable error rate
        return current_error_rate / error_budget
    
    def calculate_budget_remaining(self) -> float:
        """Return fraction of error budget remaining in current window."""
        window_start = time.time() - (self.slo.window_days * 86400)
        errors_consumed = self.prom.query_range(
            f"1 - ({self.slo.sli_query.__name__})",
            start=window_start,
            end=time.time()
        )
        total_budget = (1 - self.slo.target) * self.slo.window_days * 24 * 60  # in minutes
        consumed_minutes = sum(errors_consumed) / 60
        return max(0, (total_budget - consumed_minutes) / total_budget)

# Define SLOs as code — single source of truth
AVAILABILITY_SLO = SLO(
    name="api_availability",
    description="API returns 2xx or 4xx for at least 99.9% of requests",
    sli_query=lambda prom, window_hours=24: prom.query(
        f'sum(rate(http_requests_total{{status!~"5.."}}[{window_hours}h])) / '
        f'sum(rate(http_requests_total[{window_hours}h]))'
    ),
    target=0.999,
    window_days=30
)

LATENCY_SLO = SLO(
    name="api_latency_p99",
    description="99% of requests complete within 500ms",
    sli_query=lambda prom, window_hours=24: prom.query(
        f'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[{window_hours}h])) by (le)) < 0.5'
    ),
    target=0.99,
    window_days=30
)

# Test: Current SLO compliance
def test_availability_slo_currently_met():
    slo_test = SLOTest(AVAILABILITY_SLO, prometheus)
    assert slo_test.verify_current_compliance(), \
        f"Availability SLO violated. Current SLI: {AVAILABILITY_SLO.sli_query(prometheus):.4f}, " \
        f"Target: {AVAILABILITY_SLO.target}"

def test_latency_slo_currently_met():
    slo_test = SLOTest(LATENCY_SLO, prometheus)
    assert slo_test.verify_current_compliance(), \
        f"Latency SLO violated. p99 latency exceeds 500ms."

# Test: Error budget not burning too fast
def test_error_budget_burn_rate_acceptable():
    slo_test = SLOTest(AVAILABILITY_SLO, prometheus)
    burn_rate = slo_test.calculate_burn_rate(window_hours=1)
    
    # Alert threshold: 14.4x burn rate means budget exhausted in 2.1 days
    assert burn_rate < 14.4, \
        f"Error budget burning too fast: {burn_rate:.1f}x. " \
        f"At this rate, budget exhausted in " \
        f"{(30 / burn_rate):.1f} days instead of 30."

def test_error_budget_has_sufficient_remaining():
    slo_test = SLOTest(AVAILABILITY_SLO, prometheus)
    remaining = slo_test.calculate_budget_remaining()
    
    assert remaining > 0.1, \
        f"Less than 10% error budget remaining: {remaining:.1%}. " \
        f"Freeze risky deployments."

import asyncio
import aiohttp
import time
from collections import defaultdict

async def slo_load_test(
    base_url: str,
    target_rps: int,
    duration_seconds: int,
    availability_target: float = 0.999,
    latency_p99_target_ms: float = 500
):
    results = defaultdict(list)
    
    async def make_request(session, request_id):
        start = time.time()
        try:
            async with session.get(f"{base_url}/api/health") as resp:
                latency_ms = (time.time() - start) * 1000
                results['latencies'].append(latency_ms)
                results['status_codes'].append(resp.status)
                if resp.status >= 500:
                    results['errors'].append(request_id)
        except Exception as e:
            latency_ms = (time.time() - start) * 1000
            results['latencies'].append(latency_ms)
            results['errors'].append(request_id)
    
    async with aiohttp.ClientSession() as session:
        end_time = time.time() + duration_seconds
        request_id = 0
        
        while time.time() < end_time:
            batch = []
            for _ in range(target_rps):
                batch.append(make_request(session, request_id))
                request_id += 1
            
            await asyncio.gather(*batch)
            await asyncio.sleep(1)
    
    # Analyze results against SLO targets
    total = len(results['latencies'])
    errors = len(results['errors'])
    
    availability = (total - errors) / total
    sorted_latencies = sorted(results['latencies'])
    p99_latency = sorted_latencies[int(total * 0.99)]
    p999_latency = sorted_latencies[int(total * 0.999)]
    
    print(f"\nLoad Test Results ({total} requests @ {target_rps} RPS)")
    print(f"Availability: {availability:.4%} (target: {availability_target:.3%}) "
          f"{'PASS' if availability >= availability_target else 'FAIL'}")
    print(f"p99 Latency: {p99_latency:.0f}ms (target: {latency_p99_target_ms}ms) "
          f"{'PASS' if p99_latency <= latency_p99_target_ms else 'FAIL'}")
    print(f"p99.9 Latency: {p999_latency:.0f}ms")
    
    assert availability >= availability_target, \
        f"Availability SLO not met under load: {availability:.4%} < {availability_target:.3%}"
    assert p99_latency <= latency_p99_target_ms, \
        f"Latency SLO not met under load: p99={p99_latency:.0f}ms > {latency_p99_target_ms}ms"

# Run
asyncio.run(slo_load_test(
    base_url="https://api.example.com",
    target_rps=1000,
    duration_seconds=300,
    availability_target=0.999,
    latency_p99_target_ms=500
))

@pytest.fixture
def normal_conditions():
    """All dependencies healthy."""
    yield

@pytest.fixture
def degraded_conditions(wiremock):
    """Cache unavailable, database at 50% capacity."""
    wiremock.stub(url="/cache/*", response=ServiceUnavailable())
    with database.connection_limit(0.5):
        yield

def test_availability_slo_normal_conditions(normal_conditions, load_generator):
    results = load_generator.run(rps=1000, duration=60)
    assert results.availability >= 0.999, "Normal conditions SLO violated"

def test_availability_slo_degraded_conditions(degraded_conditions, load_generator):
    results = load_generator.run(rps=1000, duration=60)
    # Degraded SLO: 99.5% instead of 99.9%
    assert results.availability >= 0.995, "Degraded conditions SLO violated"
    assert results.p99_latency_ms <= 2000, "Degraded conditions latency SLO violated"

def calculate_availability(requests_total: int, errors_5xx: int) -> float:
    """Calculate availability as fraction of non-5xx responses."""
    if requests_total == 0:
        return 1.0  # No traffic = perfect (debatable, but common)
    return (requests_total - errors_5xx) / requests_total

def test_availability_calculation_basic():
    # 1 error in 1000 requests = 99.9%
    assert calculate_availability(1000, 1) == pytest.approx(0.999)

def test_availability_calculation_no_errors():
    assert calculate_availability(1000, 0) == 1.0

def test_availability_calculation_all_errors():
    assert calculate_availability(1000, 1000) == 0.0

def test_availability_calculation_no_traffic():
    assert calculate_availability(0, 0) == 1.0

def test_availability_99_9_requires_less_than_864_errors_per_day():
    """Verify SLO arithmetic: 99.9% over 1 day = max 86.4 error-seconds."""
    requests_per_second = 1000
    seconds_per_day = 86400
    total_requests = requests_per_second * seconds_per_day
    
    # 0.1% error budget
    max_errors = total_requests * 0.001
    assert max_errors == 86400.0  # 86,400 allowed errors per day at 1000 RPS
    
    availability = calculate_availability(total_requests, int(max_errors))
    assert availability >= 0.999

# prometheus-rules.yaml
groups:
  - name: slo_alerts
    rules:
      - alert: ErrorBudgetBurnRateCritical
        expr: |
          (
            sum(rate(http_requests_total{status=~"5.."}[1h])) /
            sum(rate(http_requests_total[1h]))
          ) / 0.001 > 14.4
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Error budget burning at {{ $value }}x rate"
          
      - alert: ErrorBudgetBurnRateWarning
        expr: |
          (
            sum(rate(http_requests_total{status=~"5.."}[6h])) /
            sum(rate(http_requests_total[6h]))
          ) / 0.001 > 6
        for: 15m
        labels:
          severity: warning

# promtool test rules rules-test.yaml

# rules-test.yaml
rule_files:
  - prometheus-rules.yaml

tests:
  - interval: 1m
    input_series:
      - series: 'http_requests_total{status="200"}'
        values: '1000+1000x60'  # 1000/s for 60 minutes
      - series: 'http_requests_total{status="500"}'
        values: '0+15x60'  # 15/s — that's 1.5% error rate, 15x burn rate
    
    alert_rule_test:
      - eval_time: 3m
        alertname: ErrorBudgetBurnRateCritical
        exp_alerts:
          - exp_labels:
              severity: critical
            exp_annotations:
              summary: "Error budget burning at 15.0x rate"

  - interval: 1m
    input_series:
      - series: 'http_requests_total{status="200"}'
        values: '1000+1000x60'
      - series: 'http_requests_total{status="500"}'
        values: '0+1x60'   # 0.1% error rate — exactly at budget, burn rate 1.0
    
    alert_rule_test:
      - eval_time: 30m
        alertname: ErrorBudgetBurnRateCritical
        exp_alerts: []  # Should NOT fire at 1.0x burn rate

def estimate_deployment_slo_impact(
    deployment_duration_minutes: int,
    error_rate_during_deployment: float,
    requests_per_minute: int,
    slo_target: float,
    window_days: int
) -> dict:
    """
    Estimate error budget consumed by a deployment.
    
    Returns dict with budget consumed and whether deployment is safe to proceed.
    """
    # Total error budget in minutes
    total_budget_minutes = (1 - slo_target) * window_days * 24 * 60
    
    # Errors during deployment
    total_requests = requests_per_minute * deployment_duration_minutes
    error_requests = total_requests * error_rate_during_deployment
    
    # Convert to "downtime equivalent minutes"
    downtime_equivalent = (error_requests / requests_per_minute)
    
    # Budget consumed as fraction
    budget_consumed_fraction = downtime_equivalent / total_budget_minutes
    
    return {
        'deployment_duration_minutes': deployment_duration_minutes,
        'error_rate': error_rate_during_deployment,
        'downtime_equivalent_minutes': downtime_equivalent,
        'budget_consumed_fraction': budget_consumed_fraction,
        'safe_to_deploy': budget_consumed_fraction < 0.1  # Don't use >10% budget in one deploy
    }

def test_deployment_impact_within_budget():
    impact = estimate_deployment_slo_impact(
        deployment_duration_minutes=5,
        error_rate_during_deployment=0.01,  # 1% errors during rolling deploy
        requests_per_minute=60000,
        slo_target=0.999,
        window_days=30
    )
    
    assert impact['safe_to_deploy'], \
        f"Deployment consumes {impact['budget_consumed_fraction']:.1%} of error budget"

class SLACalculator:
    """Calculates SLA compliance per contract terms."""
    
    def __init__(self, contract: SLAContract):
        self.contract = contract
    
    def calculate_monthly_availability(self, year: int, month: int) -> float:
        """
        Calculate availability per contract: 
        - Excludes scheduled maintenance windows
        - Uses calendar month window
        - Counts partial minutes as full minutes for outage calculation
        """
        total_minutes = self._get_calendar_minutes(year, month)
        maintenance_minutes = self._get_scheduled_maintenance_minutes(year, month)
        billable_minutes = total_minutes - maintenance_minutes
        
        outage_minutes = self._get_outage_minutes(year, month)
        # Exclude maintenance from outage calculation
        outage_minutes_excl_maintenance = max(0, outage_minutes - maintenance_minutes)
        
        uptime_minutes = billable_minutes - outage_minutes_excl_maintenance
        return uptime_minutes / billable_minutes

def test_sla_calculation_excludes_maintenance():
    calculator = SLACalculator(contract=monthly_sla_contract)
    
    # Inject 2 hours of scheduled maintenance and 30 minutes of incident
    with mock_maintenance_window(hours=2) and mock_outage(minutes=30):
        availability = calculator.calculate_monthly_availability(2024, 1)
    
    # 30 minutes outage in 44,640 minute month (minus 120 maintenance) = 44,490 usable
    # 44,490 - 30 = 44,460 uptime
    expected = 44460 / 44490
    assert availability == pytest.approx(expected, rel=1e-4)

def test_sla_99_9_threshold_met():
    calculator = SLACalculator(contract=monthly_sla_contract)
    availability = calculator.calculate_monthly_availability(2024, 1)
    assert availability >= 0.999, f"SLA violated: {availability:.4%}"

def test_sla_credit_calculation():
    """Verify service credits are calculated correctly per contract."""
    calculator = SLACalculator(contract=monthly_sla_contract)
    
    # 99.5% availability — below 99.9% SLA
    credit = calculator.calculate_service_credit(availability=0.995, monthly_fee=10000)
    
    # Contract: 10% credit for 99.0-99.9%, 25% for 95.0-99.0%
    assert credit == 1000  # 10% of $10,000

# .github/workflows/deploy.yml
- name: Check error budget before deploy
  run: |
    BURN_RATE=$(curl -s "$PROMETHEUS_URL/api/v1/query" \
      --data-urlencode 'query=(sum(rate(http_requests_total{status=~"5.."}[1h])) / sum(rate(http_requests_total[1h]))) / 0.001' \
      | jq -r '.data.result[0].value[1]')
    
    echo "Current burn rate: ${BURN_RATE}x"
    
    if (( $(echo "$BURN_RATE > 14.4" | bc -l) )); then
      echo "ERROR: Error budget burning too fast (${BURN_RATE}x). Deployment blocked."
      exit 1
    fi
    
    BUDGET_REMAINING=$(./scripts/calculate-budget-remaining.sh)
    echo "Budget remaining: ${BUDGET_REMAINING}%"
    
    if (( $(echo "$BUDGET_REMAINING < 10" | bc -l) )); then
      echo "ERROR: Less than 10% error budget remaining. Deployment blocked."
      exit 1
    fi
    
    echo "Error budget check passed. Safe to deploy."