SLIs, SLOs, and SLAs.

Product Manager: "Is the API reliable?"
Engineer: "Yeah, it's pretty reliable."
PM: "What does 'pretty reliable' mean?"
Engineer: "It's usually up."
PM: "Usually? We just lost a $500K customer who said the API was down."
Engineer: "It was only down for 10 minutes!"
PM: "They said it was down for an hour."
Engineer: "Well, it depends on what you mean by 'down'..."

# error_budget/policy.py

"""
Error budget policy implementation.

When error budget is depleted, we shift focus from features to reliability.
"""

from dataclasses import dataclass
from enum import Enum
from datetime import datetime, timedelta, timezone
from typing import List, Optional
import logging

logger = logging.getLogger(__name__)


class BudgetStatus(Enum):
    """Error budget health status."""
    HEALTHY = "healthy"          # > 50% budget remaining
    WARNING = "warning"          # 20-50% remaining
    CRITICAL = "critical"        # 0-20% remaining
    EXHAUSTED = "exhausted"      # Budget depleted


@dataclass
class ErrorBudget:
    """Error budget state for an SLO."""
    slo_name: str
    window_days: int
    target_percentage: float  # e.g., 99.9
    
    # Current state
    current_percentage: float
    budget_remaining_minutes: float
    budget_consumed_minutes: float
    
    # Burn rate
    burn_rate_1h: float   # How fast we're consuming (1 = normal)
    burn_rate_24h: float  # Smoothed over 24 hours
    
    @property
    def total_budget_minutes(self) -> float:
        """Total error budget in minutes."""
        error_percentage = 100 - self.target_percentage
        return (error_percentage / 100) * self.window_days * 24 * 60
    
    @property
    def budget_remaining_percentage(self) -> float:
        """Percentage of budget remaining."""
        return (self.budget_remaining_minutes / self.total_budget_minutes) * 100
    
    @property
    def status(self) -> BudgetStatus:
        """Current budget status."""
        remaining_pct = self.budget_remaining_percentage
        
        if remaining_pct <= 0:
            return BudgetStatus.EXHAUSTED
        elif remaining_pct < 20:
            return BudgetStatus.CRITICAL
        elif remaining_pct <= 50:
            return BudgetStatus.WARNING
        else:
            return BudgetStatus.HEALTHY
    
    @property
    def projected_exhaustion(self) -> Optional[datetime]:
        """When budget will be exhausted at current burn rate."""
        if self.burn_rate_24h <= 0:
            return None  # Not consuming budget
        
        # At burn rate 1x the WHOLE budget lasts exactly one window, so the
        # budget burns at burn_rate * (total budget / window length) per minute
        window_minutes = self.window_days * 24 * 60
        burn_per_minute = self.burn_rate_24h * self.total_budget_minutes / window_minutes
        minutes_until_exhausted = self.budget_remaining_minutes / burn_per_minute
        return datetime.now(timezone.utc) + timedelta(minutes=minutes_until_exhausted)


class ErrorBudgetPolicy:
    """
    Defines actions based on error budget status.
    
    Philosophy:
    - Budget healthy: Ship features freely
    - Budget warning: Increase caution
    - Budget critical: Focus on reliability
    - Budget exhausted: Feature freeze
    """
    
    def __init__(self, config):
        self.config = config
    
    def get_allowed_actions(self, budget: ErrorBudget) -> dict:
        """
        Get allowed actions based on budget status.
        """
        status = budget.status
        
        if status == BudgetStatus.HEALTHY:
            return {
                "feature_development": True,
                "risky_deployments": True,
                "experiments": True,
                "on_call_escalation": "normal",
                "deployment_approval": "team_lead",
                "rollback_threshold": "5xx_rate > 5%",
                "message": "Budget healthy. Ship features!"
            }
        
        elif status == BudgetStatus.WARNING:
            return {
                "feature_development": True,
                "risky_deployments": False,  # No risky changes
                "experiments": True,
                "on_call_escalation": "normal",
                "deployment_approval": "team_lead",
                "rollback_threshold": "5xx_rate > 2%",  # Lower threshold
                "message": "Budget warning. Exercise caution with deployments."
            }
        
        elif status == BudgetStatus.CRITICAL:
            return {
                "feature_development": False,  # Pause features
                "risky_deployments": False,
                "experiments": False,
                "on_call_escalation": "heightened",  # More eyes on alerts
                "deployment_approval": "engineering_manager",
                "rollback_threshold": "5xx_rate > 1%",  # Very sensitive
                "message": "Budget critical. Focus on reliability only."
            }
        
        else:  # EXHAUSTED
            return {
                "feature_development": False,
                "risky_deployments": False,
                "experiments": False,
                "on_call_escalation": "incident_mode",
                "deployment_approval": "director",
                "rollback_threshold": "any_degradation",
                "message": "Budget exhausted. Full reliability focus until recovered."
            }
    
    def should_alert(self, budget: ErrorBudget) -> List[dict]:
        """
        Determine what alerts should fire based on budget state.
        """
        alerts = []
        
        # Budget status alerts
        if budget.status == BudgetStatus.EXHAUSTED:
            alerts.append({
                "severity": "critical",
                "title": f"Error budget exhausted: {budget.slo_name}",
                "message": f"SLO {budget.slo_name} has exhausted its error budget. "
                          f"Feature development should halt.",
                "route": "engineering_leadership"
            })
        
        elif budget.status == BudgetStatus.CRITICAL:
            alerts.append({
                "severity": "warning",
                "title": f"Error budget critical: {budget.slo_name}",
                "message": f"Only {budget.budget_remaining_percentage:.1f}% budget remaining.",
                "route": "team_lead"
            })
        
        # Burn rate alerts (predicting future problems).
        # Time to exhaust a full budget = window / burn rate
        hours_to_exhaust = budget.window_days * 24 / max(budget.burn_rate_1h, 1e-9)
        
        if budget.burn_rate_1h > 14.4:  # 2% of a 30-day budget per hour (~50h to empty)
            alerts.append({
                "severity": "critical",
                "title": f"High burn rate: {budget.slo_name}",
                "message": f"At this rate a full budget lasts ~{hours_to_exhaust:.0f} hours. "
                          f"Burn rate: {budget.burn_rate_1h:.1f}x normal.",
                "route": "on_call"
            })
        
        elif budget.burn_rate_1h > 6:  # 5% of a 30-day budget per 6h (~5 days to empty)
            alerts.append({
                "severity": "warning",
                "title": f"Elevated burn rate: {budget.slo_name}",
                "message": f"Burn rate: {budget.burn_rate_1h:.1f}x normal.",
                "route": "on_call"
            })
        
        return alerts


class ErrorBudgetCalculator:
    """
    Calculates error budget from SLI measurements.
    """
    
    def __init__(self, metrics_client):
        self.metrics = metrics_client
    
    async def calculate_budget(
        self,
        slo_name: str,
        sli_query: str,
        target: float,
        window_days: int = 30
    ) -> ErrorBudget:
        """
        Calculate current error budget state.
        """
        # Get SLI value over window
        sli_value = await self.metrics.query(
            sli_query,
            window=f"{window_days}d"
        )
        
        # Calculate budget
        target_error_rate = 100 - target
        actual_error_rate = 100 - sli_value
        
        total_budget_minutes = (target_error_rate / 100) * window_days * 24 * 60
        consumed_minutes = (actual_error_rate / 100) * window_days * 24 * 60
        remaining_minutes = total_budget_minutes - consumed_minutes
        
        # Calculate burn rates
        sli_1h = await self.metrics.query(sli_query, window="1h")
        sli_24h = await self.metrics.query(sli_query, window="24h")
        
        expected_error_rate_1h = target_error_rate
        actual_error_rate_1h = 100 - sli_1h
        burn_rate_1h = actual_error_rate_1h / expected_error_rate_1h if expected_error_rate_1h > 0 else 0
        
        actual_error_rate_24h = 100 - sli_24h
        burn_rate_24h = actual_error_rate_24h / expected_error_rate_1h if expected_error_rate_1h > 0 else 0
        
        return ErrorBudget(
            slo_name=slo_name,
            window_days=window_days,
            target_percentage=target,
            current_percentage=sli_value,
            budget_remaining_minutes=max(0, remaining_minutes),
            budget_consumed_minutes=consumed_minutes,
            burn_rate_1h=burn_rate_1h,
            burn_rate_24h=burn_rate_24h
        )

# error_budget/alerting.py

"""
Multi-window, multi-burn-rate alerting.
"""

from dataclasses import dataclass
from typing import List


@dataclass
class BurnRateAlert:
    """A burn rate alert configuration."""
    name: str
    burn_rate: float
    short_window: str
    long_window: str
    severity: str
    action: str


class BurnRateAlerting:
    """
    Implements multi-window burn rate alerting.
    """
    
    # Alert configurations
    ALERTS = [
        BurnRateAlert(
            name="budget_burn_critical",
            burn_rate=14.4,
            short_window="5m",
            long_window="1h",
            severity="critical",
            action="page"
        ),
        BurnRateAlert(
            name="budget_burn_high",
            burn_rate=6.0,
            short_window="30m",
            long_window="6h",
            severity="warning",
            action="page_business_hours"
        ),
        BurnRateAlert(
            name="budget_burn_medium",
            burn_rate=3.0,
            short_window="2h",
            long_window="1d",
            severity="info",
            action="ticket"
        ),
        BurnRateAlert(
            name="budget_burn_low",
            burn_rate=1.0,
            short_window="6h",
            long_window="3d",
            severity="info",
            action="review"
        ),
    ]
    
    def generate_prometheus_rules(
        self,
        slo_name: str,
        error_ratio_metric: str,
        target: float
    ) -> str:
        """
        Generate Prometheus alerting rules for an SLO.
        """
        error_budget_ratio = 1 - (target / 100)  # e.g., 0.001 for 99.9%
        
        rules = []
        
        for alert in self.ALERTS:
            # Multi-window condition: Both windows must exceed burn rate
            rule = f"""
- alert: {slo_name}_{alert.name}
  expr: |
    (
      avg_over_time({error_ratio_metric}[{alert.short_window}]) > {error_budget_ratio * alert.burn_rate}
      and
      avg_over_time({error_ratio_metric}[{alert.long_window}]) > {error_budget_ratio * alert.burn_rate}
    )
  for: 2m
  labels:
    severity: {alert.severity}
    slo: {slo_name}
  annotations:
    summary: "SLO {slo_name} burning error budget at {alert.burn_rate}x rate"
    description: "Error budget burn rate is {alert.burn_rate}x normal over {alert.long_window}"
    runbook: "https://wiki/runbooks/{slo_name}"
    action: "{alert.action}"
"""
            rules.append(rule)
        
        return "\n".join(rules)

# slo/dashboard.py

"""
SLO Dashboard data provider.

Provides all the data needed for an SLO status dashboard.
"""

from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from enum import Enum


@dataclass
class SLOStatus:
    """Current status of an SLO."""
    name: str
    description: str
    target: float
    current_value: float
    window_days: int
    
    # Budget state
    budget_total_minutes: float
    budget_consumed_minutes: float
    budget_remaining_minutes: float
    budget_remaining_percentage: float
    
    # Burn rate
    burn_rate_1h: float
    burn_rate_6h: float
    burn_rate_24h: float
    
    # Trend
    trend_direction: str  # "improving", "stable", "degrading"
    projected_end_of_window: float  # Projected SLI at window end
    
    # Status
    status: str  # "healthy", "warning", "critical", "violated"
    status_changed_at: Optional[datetime]


class SLODashboardService:
    """
    Provides data for SLO dashboards.
    """
    
    def __init__(self, metrics_client, slo_config):
        self.metrics = metrics_client
        self.config = slo_config
    
    async def get_all_slo_status(self) -> List[SLOStatus]:
        """Get status of all configured SLOs."""
        statuses = []
        
        for slo in self.config.slos:
            status = await self.get_slo_status(slo)
            statuses.append(status)
        
        return statuses
    
    async def get_slo_status(self, slo_config) -> SLOStatus:
        """Get detailed status for a single SLO."""
        
        # Query current SLI value
        current_value = await self.metrics.query(
            slo_config.sli_query,
            window=f"{slo_config.window_days}d"
        )
        
        # Calculate error budget
        target = slo_config.target
        error_rate_allowed = 100 - target
        error_rate_actual = 100 - current_value
        
        budget_total = (error_rate_allowed / 100) * slo_config.window_days * 24 * 60
        budget_consumed = (error_rate_actual / 100) * slo_config.window_days * 24 * 60
        budget_remaining = max(0, budget_total - budget_consumed)
        budget_remaining_pct = (budget_remaining / budget_total) * 100 if budget_total > 0 else 0
        
        # Calculate burn rates at different windows
        burn_rate_1h = await self._calculate_burn_rate(slo_config, "1h")
        burn_rate_6h = await self._calculate_burn_rate(slo_config, "6h")
        burn_rate_24h = await self._calculate_burn_rate(slo_config, "24h")
        
        # Determine trend
        trend = self._calculate_trend(burn_rate_1h, burn_rate_6h, burn_rate_24h)
        
        # Project end-of-window value
        projected = self._project_eow_value(current_value, burn_rate_24h, slo_config)
        
        # Determine status
        status = self._determine_status(budget_remaining_pct, burn_rate_1h)
        
        return SLOStatus(
            name=slo_config.name,
            description=slo_config.description,
            target=target,
            current_value=current_value,
            window_days=slo_config.window_days,
            budget_total_minutes=budget_total,
            budget_consumed_minutes=budget_consumed,
            budget_remaining_minutes=budget_remaining,
            budget_remaining_percentage=budget_remaining_pct,
            burn_rate_1h=burn_rate_1h,
            burn_rate_6h=burn_rate_6h,
            burn_rate_24h=burn_rate_24h,
            trend_direction=trend,
            projected_end_of_window=projected,
            status=status,
            status_changed_at=None  # Would track in database
        )
    
    async def _calculate_burn_rate(self, slo_config, window: str) -> float:
        """Calculate burn rate for a time window."""
        value = await self.metrics.query(slo_config.sli_query, window=window)
        
        error_rate_allowed = 100 - slo_config.target
        error_rate_actual = 100 - value
        
        if error_rate_allowed == 0:
            return 0
        
        return error_rate_actual / error_rate_allowed
    
    def _calculate_trend(
        self, 
        burn_1h: float, 
        burn_6h: float, 
        burn_24h: float
    ) -> str:
        """Determine trend from burn rates."""
        # If recent burn rate is lower than longer-term, improving
        if burn_1h < burn_6h < burn_24h:
            return "improving"
        elif burn_1h > burn_6h > burn_24h:
            return "degrading"
        else:
            return "stable"
    
    def _project_eow_value(
        self,
        current: float,
        burn_rate_24h: float,
        slo_config
    ) -> float:
        """Project SLI value at end of window (simplified)."""
        # Assume we're mid-window: the elapsed half ran at the current SLI,
        # the remaining half runs at the 24h burn rate. Burn rate b means
        # an error rate of b x the allowed error rate.
        allowed_error = 100 - slo_config.target
        future_error = burn_rate_24h * allowed_error
        projected_error = 0.5 * (100 - current) + 0.5 * future_error
        
        return max(0, 100 - projected_error)
    
    def _determine_status(
        self,
        budget_remaining_pct: float,
        burn_rate_1h: float
    ) -> str:
        """Determine SLO status."""
        # Violated if budget exhausted
        if budget_remaining_pct <= 0:
            return "violated"
        
        # Critical if budget very low OR burning very fast
        if budget_remaining_pct < 20 or burn_rate_1h > 10:
            return "critical"
        
        # Warning if budget getting low OR elevated burn rate
        if budget_remaining_pct < 50 or burn_rate_1h > 5:
            return "warning"
        
        return "healthy"
    
    async def get_slo_history(
        self,
        slo_name: str,
        days: int = 30
    ) -> List[Dict]:
        """Get historical SLO performance."""
        # Query historical data points
        data_points = await self.metrics.query_range(
            self.config.get_slo(slo_name).sli_query,
            start=f"-{days}d",
            end="now",
            step="1h"
        )
        
        return [
            {
                "timestamp": dp.timestamp,
                "value": dp.value,
                "target": self.config.get_slo(slo_name).target
            }
            for dp in data_points
        ]

{
  "dashboard": {
    "title": "Service SLO Dashboard",
    "panels": [
      {
        "title": "SLO Status Overview",
        "type": "stat",
        "gridPos": {"x": 0, "y": 0, "w": 24, "h": 4},
        "targets": [
          {
            "expr": "slo:current_value{service=\"api\"}",
            "legendFormat": "{{slo_name}}"
          }
        ],
        "options": {
          "colorMode": "background",
          "graphMode": "none",
          "justifyMode": "center",
          "textMode": "value_and_name"
        },
        "fieldConfig": {
          "defaults": {
            "thresholds": {
              "steps": [
                {"color": "red", "value": null},
                {"color": "red", "value": 99},
                {"color": "yellow", "value": 99.5},
                {"color": "green", "value": 99.9}
              ]
            },
            "unit": "percent"
          }
        }
      },
      {
        "title": "Error Budget Remaining",
        "type": "gauge",
        "gridPos": {"x": 0, "y": 4, "w": 8, "h": 6},
        "targets": [
          {
            "expr": "slo:error_budget_remaining_ratio{service=\"api\"} * 100"
          }
        ],
        "options": {
          "showThresholdLabels": true,
          "showThresholdMarkers": true
        },
        "fieldConfig": {
          "defaults": {
            "min": 0,
            "max": 100,
            "thresholds": {
              "steps": [
                {"color": "red", "value": 0},
                {"color": "orange", "value": 20},
                {"color": "yellow", "value": 50},
                {"color": "green", "value": 75}
              ]
            },
            "unit": "percent"
          }
        }
      },
      {
        "title": "Burn Rate (1h, 6h, 24h)",
        "type": "timeseries",
        "gridPos": {"x": 8, "y": 4, "w": 16, "h": 6},
        "targets": [
          {
            "expr": "slo:burn_rate_1h{service=\"api\"}",
            "legendFormat": "1h burn rate"
          },
          {
            "expr": "slo:burn_rate_6h{service=\"api\"}",
            "legendFormat": "6h burn rate"
          },
          {
            "expr": "slo:burn_rate_24h{service=\"api\"}",
            "legendFormat": "24h burn rate"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "custom": {
              "thresholdsStyle": {
                "mode": "line"
              }
            },
            "thresholds": {
              "steps": [
                {"color": "green", "value": null},
                {"color": "yellow", "value": 1},
                {"color": "orange", "value": 6},
                {"color": "red", "value": 14.4}
              ]
            }
          }
        }
      },
      {
        "title": "SLI Over Time",
        "type": "timeseries",
        "gridPos": {"x": 0, "y": 10, "w": 24, "h": 8},
        "targets": [
          {
            "expr": "slo:sli_value{service=\"api\"}",
            "legendFormat": "{{slo_name}}"
          },
          {
            "expr": "slo:target{service=\"api\"}",
            "legendFormat": "Target"
          }
        ]
      }
    ]
  }
}

# slo/tiered.py

"""
Tiered SLO management for different customer segments.
"""

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict
from enum import Enum


class CustomerTier(Enum):
    """Customer tier levels."""
    FREE = "free"
    STANDARD = "standard"
    PROFESSIONAL = "professional"
    ENTERPRISE = "enterprise"


@dataclass
class TieredSLO:
    """SLO with different targets per tier."""
    sli_name: str
    description: str
    
    # Targets by tier
    targets: Dict[CustomerTier, float]
    
    # Whether to enforce (vs aspirational)
    enforced: bool = True


# Example tiered SLOs for our notification platform
NOTIFICATION_SLOS = [
    TieredSLO(
        sli_name="api_availability",
        description="API request success rate",
        targets={
            CustomerTier.FREE: 99.0,        # 7.2 hours/month downtime OK
            CustomerTier.STANDARD: 99.5,    # 3.6 hours/month
            CustomerTier.PROFESSIONAL: 99.9, # 43 minutes/month
            CustomerTier.ENTERPRISE: 99.99,  # 4.3 minutes/month
        }
    ),
    TieredSLO(
        sli_name="api_latency_p99",
        description="99th percentile API latency",
        targets={
            CustomerTier.FREE: 1000,         # 1 second
            CustomerTier.STANDARD: 500,      # 500ms
            CustomerTier.PROFESSIONAL: 200,  # 200ms
            CustomerTier.ENTERPRISE: 100,    # 100ms
        }
    ),
    TieredSLO(
        sli_name="delivery_rate",
        description="Notification delivery success rate",
        targets={
            CustomerTier.FREE: 95.0,
            CustomerTier.STANDARD: 98.0,
            CustomerTier.PROFESSIONAL: 99.0,
            CustomerTier.ENTERPRISE: 99.9,
        }
    ),
]


class TieredSLOService:
    """
    Manages tiered SLOs across customer segments.
    """
    
    def __init__(self, metrics_client, customer_service):
        self.metrics = metrics_client
        self.customers = customer_service
    
    async def check_customer_slo(
        self,
        customer_id: str,
        slo_name: str
    ) -> Dict:
        """Check SLO status for a specific customer."""
        
        # Get customer tier
        customer = await self.customers.get(customer_id)
        tier = CustomerTier(customer.tier)
        
        # Find SLO config
        slo_config = next(s for s in NOTIFICATION_SLOS if s.sli_name == slo_name)
        target = slo_config.targets[tier]
        
        # Query SLI for this customer
        current_value = await self.metrics.query(
            f'sli:{slo_name}{{customer_id="{customer_id}"}}',
            window="30d"
        )
        
        return {
            "customer_id": customer_id,
            "tier": tier.value,
            "slo_name": slo_name,
            "target": target,
            "current_value": current_value,
            "meeting_slo": current_value >= target,
            "gap": current_value - target
        }
    
    async def get_slo_compliance_report(self) -> Dict:
        """Generate compliance report across all tiers."""
        report = {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "by_tier": {},
            "by_slo": {}
        }
        
        for tier in CustomerTier:
            tier_customers = await self.customers.get_by_tier(tier.value)
            
            compliance = {
                "total_customers": len(tier_customers),
                "meeting_all_slos": 0,
                "violating_slos": 0
            }
            
            for customer in tier_customers:
                meeting_all = True
                
                for slo in NOTIFICATION_SLOS:
                    result = await self.check_customer_slo(
                        customer.id, 
                        slo.sli_name
                    )
                    if not result["meeting_slo"]:
                        meeting_all = False
                        break
                
                if meeting_all:
                    compliance["meeting_all_slos"] += 1
                else:
                    compliance["violating_slos"] += 1
            
            report["by_tier"][tier.value] = compliance
        
        return report