Site Reliability Engineering: Practical Guide to SLOs, SLIs, and Error Budgets

# slo_calculator.py
from prometheus_api_client import PrometheusConnect
from datetime import datetime, timedelta

class SLOCalculator:
    def __init__(self, prometheus_url):
        self.prom = PrometheusConnect(url=prometheus_url, disable_ssl=True)
    
    def calculate_availability_slo(self, service, window_days=30):
        """Calculate availability SLO for a service"""
        end_time = datetime.now()
        start_time = end_time - timedelta(days=window_days)
        
        # Query for successful requests
        success_query = f'''
            sum(rate(http_requests_total{{
                job="{service}",
                code!~"5.."
            }}[5m]))
        '''
        
        # Query for total requests
        total_query = f'''
            sum(rate(http_requests_total{{
                job="{service}"
            }}[5m]))
        '''
        
        success_data = self.prom.custom_query_range(
            success_query,
            start_time=start_time,
            end_time=end_time,
            step='5m'
        )
        
        total_data = self.prom.custom_query_range(
            total_query,
            start_time=start_time,
            end_time=end_time,
            step='5m'
        )
        
        # Calculate SLI
        total_successful = sum([float(d[1]) for d in success_data[0]['values']])
        total_requests = sum([float(d[1]) for d in total_data[0]['values']])
        
        sli = (total_successful / total_requests) * 100 if total_requests > 0 else 0
        
        return {
            'service': service,
            'window_days': window_days,
            'sli': round(sli, 4),
            'total_requests': int(total_requests),
            'successful_requests': int(total_successful),
            'failed_requests': int(total_requests - total_successful)
        }
    
    def calculate_error_budget(self, service, slo_target=99.9, window_days=30):
        """Calculate remaining error budget"""
        result = self.calculate_availability_slo(service, window_days)
        current_sli = result['sli']
        
        # Calculate error budget
        allowed_failure_rate = 100 - slo_target
        actual_failure_rate = 100 - current_sli
        
        error_budget_consumed = (actual_failure_rate / allowed_failure_rate) * 100
        error_budget_remaining = 100 - error_budget_consumed
        
        return {
            **result,
            'slo_target': slo_target,
            'allowed_failure_rate': allowed_failure_rate,
            'actual_failure_rate': round(actual_failure_rate, 4),
            'error_budget_consumed_percent': round(error_budget_consumed, 2),
            'error_budget_remaining_percent': round(error_budget_remaining, 2),
            'status': 'HEALTHY' if error_budget_remaining > 10 else 'WARNING' if error_budget_remaining > 0 else 'CRITICAL'
        }

# Usage
calculator = SLOCalculator('http://prometheus:9090')
budget = calculator.calculate_error_budget('api-service', slo_target=99.9)

print(f"Service: {budget['service']}")
print(f"Current SLI: {budget['sli']}%")
print(f"SLO Target: {budget['slo_target']}%")
print(f"Error Budget Remaining: {budget['error_budget_remaining_percent']}%")
print(f"Status: {budget['status']}")

# error_budget_enforcer.py
class ErrorBudgetEnforcer:
    def __init__(self, calculator, policy):
        self.calculator = calculator
        self.policy = policy
    
    def check_deployment_allowed(self, service):
        """Check if deployment is allowed based on error budget"""
        budget = self.calculator.calculate_error_budget(
            service,
            slo_target=self.policy['slo_target']
        )
        
        remaining = budget['error_budget_remaining_percent']
        
        if remaining > 50:
            return {
                'allowed': True,
                'risk_level': 'LOW',
                'max_deployments_per_day': 2,
                'message': 'Normal operations - proceed with deployment'
            }
        elif remaining > 10:
            return {
                'allowed': True,
                'risk_level': 'MEDIUM',
                'max_deployments_per_day': 1,
                'message': 'Increased caution - require additional validation'
            }
        else:
            return {
                'allowed': False,
                'risk_level': 'HIGH',
                'max_deployments_per_day': 0,
                'message': 'Feature freeze - focus on reliability improvements'
            }
    
    def generate_report(self, service):
        """Generate error budget report"""
        budget = self.calculator.calculate_error_budget(service)
        deployment_status = self.check_deployment_allowed(service)
        
        report = f"""
        ═══════════════════════════════════════════
        ERROR BUDGET REPORT - {service.upper()}
        ═══════════════════════════════════════════
        
        Current SLI:          {budget['sli']}%
        SLO Target:           {budget['slo_target']}%
        
        Error Budget:
        - Consumed:           {budget['error_budget_consumed_percent']}%
        - Remaining:          {budget['error_budget_remaining_percent']}%
        - Status:             {budget['status']}
        
        Deployment Status:
        - Allowed:            {'✓ YES' if deployment_status['allowed'] else '✗ NO'}
        - Risk Level:         {deployment_status['risk_level']}
        - Max Deployments:    {deployment_status['max_deployments_per_day']}/day
        
        Recommendation:
        {deployment_status['message']}
        
        ═══════════════════════════════════════════
        """
        
        return report

# Usage
enforcer = ErrorBudgetEnforcer(calculator, error_budget_policy)
print(enforcer.generate_report('api-service'))

# toil_tracker.py
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List

@dataclass
class ToilTask:
    name: str
    time_minutes: int
    frequency: str  # daily, weekly, monthly
    category: str
    automatable: bool
    
class ToilCalculator:
    def __init__(self):
        self.tasks: List[ToilTask] = []
    
    def add_task(self, task: ToilTask):
        self.tasks.append(task)
    
    def calculate_monthly_toil(self):
        """Calculate total monthly toil hours"""
        total_minutes = 0
        
        for task in self.tasks:
            if task.frequency == 'daily':
                total_minutes += task.time_minutes * 30
            elif task.frequency == 'weekly':
                total_minutes += task.time_minutes * 4
            elif task.frequency == 'monthly':
                total_minutes += task.time_minutes
        
        return total_minutes / 60  # Convert to hours
    
    def toil_by_category(self):
        """Group toil by category"""
        categories = {}
        
        for task in self.tasks:
            freq_multiplier = {'daily': 30, 'weekly': 4, 'monthly': 1}
            minutes = task.time_minutes * freq_multiplier[task.frequency]
            
            if task.category not in categories:
                categories[task.category] = 0
            categories[task.category] += minutes / 60
        
        return categories
    
    def automation_opportunities(self):
        """Identify high-value automation opportunities"""
        opportunities = []
        
        for task in self.tasks:
            if task.automatable:
                freq_multiplier = {'daily': 30, 'weekly': 4, 'monthly': 1}
                monthly_hours = (task.time_minutes * freq_multiplier[task.frequency]) / 60
                
                opportunities.append({
                    'task': task.name,
                    'monthly_hours_saved': monthly_hours,
                    'category': task.category
                })
        
        return sorted(opportunities, key=lambda x: x['monthly_hours_saved'], reverse=True)

# Example usage
tracker = ToilCalculator()

# Add toil tasks
tracker.add_task(ToilTask("Manual deployment verification", 30, "daily", "Deployments", True))
tracker.add_task(ToilTask("Certificate renewal", 45, "monthly", "Security", True))
tracker.add_task(ToilTask("Log analysis for errors", 60, "daily", "Monitoring", True))
tracker.add_task(ToilTask("Database backup verification", 15, "daily", "Backups", True))
tracker.add_task(ToilTask("Capacity planning review", 120, "weekly", "Planning", False))

print(f"Total monthly toil: {tracker.calculate_monthly_toil():.1f} hours")
print(f"\nToil by category: {tracker.toil_by_category()}")
print(f"\nTop automation opportunities:")
for opp in tracker.automation_opportunities()[:3]:
    print(f"  - {opp['task']}: {opp['monthly_hours_saved']:.1f} hours/month")

# chaos_monkey.py
import random
import time
from kubernetes import client, config

class ChaosMonkey:
    def __init__(self):
        config.load_kube_config()
        self.api = client.CoreV1Api()
    
    def kill_random_pod(self, namespace, label_selector):
        """Kill a random pod matching the selector"""
        pods = self.api.list_namespaced_pod(
            namespace=namespace,
            label_selector=label_selector
        )
        
        if not pods.items:
            print("No pods found")
            return
        
        target_pod = random.choice(pods.items)
        print(f"Terminating pod: {target_pod.metadata.name}")
        
        self.api.delete_namespaced_pod(
            name=target_pod.metadata.name,
            namespace=namespace
        )
    
    def introduce_latency(self, namespace, deployment, delay_ms=1000):
        """Add network latency to a deployment"""
        # Using toxiproxy or similar tool
        pass
    
    def run_experiment(self, namespace, label_selector, duration_minutes=5):
        """Run chaos experiment"""
        print(f"Starting chaos experiment for {duration_minutes} minutes")
        end_time = time.time() + (duration_minutes * 60)
        
        while time.time() < end_time:
            self.kill_random_pod(namespace, label_selector)
            time.sleep(random.randint(30, 120))  # Wait 30-120 seconds
        
        print("Chaos experiment complete")

# Usage
chaos = ChaosMonkey()
chaos.run_experiment(
    namespace="production",
    label_selector="app=api,tier=backend",
    duration_minutes=5
)