sre-expert

name	sre-expert
version	1.0.0
description	Expert-level site reliability engineering, SLOs, incident management, and operational excellence
category	devops
tags	["sre","reliability","monitoring","incident-management","slo","observability"]
allowed-tools	["Read","Write","Edit","Bash(*)"]

Site Reliability Engineering Expert

Expert guidance for SRE practices, reliability engineering, SLOs/SLIs, incident management, and operational excellence.

Core Concepts

SRE Fundamentals

• Service Level Objectives (SLOs)
• Service Level Indicators (SLIs)
• Error budgets
• Toil reduction
• Monitoring and alerting
• Capacity planning

Reliability Practices

• Incident management
• Post-incident reviews (PIRs)
• On-call rotations
• Chaos engineering
• Disaster recovery
• Change management

Automation

• Infrastructure as Code
• Configuration management
• Deployment automation
• Self-healing systems
• Runbook automation
• Automated remediation

SLO/SLI Management

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict
import numpy as np

@dataclass
class SLI:
    """Service Level Indicator"""
    name: str
    description: str
    query: str
    unit: str  # 'percentage', 'milliseconds', etc.

@dataclass
class SLO:
    """Service Level Objective"""
    name: str
    sli: SLI
    target: float
    window_days: int

class SLOTracker:
    """Track and manage SLOs"""

    def __init__(self):
        self.slos: Dict[str, SLO] = {}
        self.measurements: Dict[str, List[Dict]] = {}

    def define_slo(self, slo: SLO):
        """Define a new SLO"""
        self.slos[slo.name] = slo
        self.measurements[slo.name] = []

    def record_measurement(self, slo_name: , value: , timestamp: datetime):
        
         slo_name  .slos:
            .measurements[slo_name].append({
                : value,
                : timestamp
            })

     () -> :
        
        slo = .slos.get(slo_name)
          slo:
             {}

        measurements = .measurements.get(slo_name, [])
        window_start = datetime.now() - timedelta(days=slo.window_days)

        recent_measurements = [
            m  m  measurements
             m[] > window_start
        ]

          recent_measurements:
             {: }

        values = [m[]  m  recent_measurements]
        actual = np.mean(values)

         {
            : slo_name,
            : slo.target,
            : actual,
            : actual >= slo.target,
            : slo.window_days,
            : (recent_measurements)
        }

     () -> :
        
        compliance = .calculate_slo_compliance(slo_name)

         compliance.get() == :
             {: }

        target = compliance[]
        actual = compliance[]

        error_budget_target =  - target
        errors_actual =  - actual

        remaining = error_budget_target - errors_actual
        remaining_pct = (remaining / error_budget_target) *   error_budget_target >   

         {
            : slo_name,
            : error_budget_target,
            : errors_actual,
            : remaining,
            : remaining_pct,
            : remaining < 
        }


 () -> [SLO]:
    
     [
        SLO(
            name=,
            sli=SLI(
                name=,
                description=,
                query=,
                unit=
            ),
            target=,
            window_days=
        ),
        SLO(
            name=,
            sli=SLI(
                name=,
                description=,
                query=,
                unit=
            ),
            target=,  
            window_days=
        )
    ]

Incident Management

from enum import Enum
from datetime import datetime
from typing import List, Optional

class Severity(Enum):
    SEV1 = "sev1"  # Critical
    SEV2 = "sev2"  # High
    SEV3 = "sev3"  # Medium
    SEV4 = "sev4"  # Low

class IncidentStatus(Enum):
    INVESTIGATING = "investigating"
    IDENTIFIED = "identified"
    MONITORING = "monitoring"
    RESOLVED = "resolved"

@dataclass
class Incident:
    incident_id: str
    title: str
    severity: Severity
    status: IncidentStatus
    started_at: datetime
    detected_at: datetime
    resolved_at: Optional[datetime]
    incident_commander: str
    responders: List[str]
    affected_services: List[str]
    timeline: List[Dict]
    root_cause: Optional[str] = None

class IncidentManager:
    """Manage incidents following SRE best practices"""

    def __init__(self):
        .incidents: [, Incident] = {}

     () -> :
        
        .incidents[incident.incident_id] = incident

        
        .notify_oncall(incident)

        
        .add_timeline_event(
            incident.incident_id,
            ,
            datetime.now()
        )

         incident.incident_id

     ():
        
         incident_id  .incidents:
            incident = .incidents[incident_id]
            incident.status = new_status

            .add_timeline_event(
                incident_id,
                ,
                datetime.now()
            )

             new_status == IncidentStatus.RESOLVED:
                incident.resolved_at = datetime.now()

     ():
        
         incident_id  .incidents:
            .incidents[incident_id].timeline.append({
                : timestamp,
                : event
            })

     () -> []:
        
        incident = .incidents.get(incident_id)

         incident  incident.resolved_at:
            duration = incident.resolved_at - incident.detected_at
             duration.total_seconds() /   

         

     () -> :
        
        incident = .incidents.get(incident_id)

          incident:
             {}

         {
            : incident.incident_id,
            : incident.title,
            : incident.severity.value,
            : incident.status.value,
            : .calculate_mttr(incident_id),
            : incident.affected_services,
            : incident.incident_commander,
            : incident.responders,
            : incident.timeline,
            : incident.root_cause
        }

     ():
        
        
        

Monitoring and Alerting

from prometheus_client import Counter, Histogram, Gauge
import time

# Metrics
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
request_duration = Histogram('http_request_duration_seconds', 'HTTP request duration')
active_connections = Gauge('active_connections', 'Number of active connections')

class MonitoringSystem:
    """Implement monitoring best practices"""

    def __init__(self):
        self.alerts = []

    def record_request(self, method: str, endpoint: str, status: int, duration: float):
        """Record HTTP request metrics"""
        request_count.labels(method=method, endpoint=endpoint, status=status).inc()
        request_duration.observe(duration)

    def define_alert(self, name: str, expression: str, threshold: float,
                    duration: str, severity: str) -> Dict:
        """Define alerting rule"""
        alert = {
            'name': name,
            'expression': expression,
            'threshold': threshold,
            'duration': duration,
            'severity': severity,
            : {
                : ,
                : 
            }
        }

        .alerts.append(alert)
         alert

     () -> :
        
         {
            : ._check_latency(metrics.get(, [])),
            : ._check_traffic(metrics.get(, )),
            : ._check_errors(metrics.get(, )),
            : ._check_saturation(metrics.get(, ))
        }

     () -> :
          latencies:
             {: }

        p95 = np.percentile(latencies, )
         {
            :   p95 >   ,
            : p95
        }

     () -> :
         {
            : ,
            : requests_per_second
        }

     () -> :
         {
            :   error_rate >   ,
            : error_rate
        }

     () -> :
         {
            :   cpu_usage >   ,
            : cpu_usage
        }

Chaos Engineering

import random
from typing import Callable

class ChaosExperiment:
    """Run chaos engineering experiments"""

    def __init__(self, name: str, hypothesis: str):
        self.name = name
        self.hypothesis = hypothesis
        self.results = []

    def inject_latency(self, service_call: Callable, delay_ms: int):
        """Inject latency into service call"""
        time.sleep(delay_ms / 1000)
        return service_call()

    def inject_failure(self, service_call: Callable, failure_rate: float):
        """Randomly fail service calls"""
        if random.random() < failure_rate:
            raise Exception("Chaos: Simulated failure")
        return service_call()

    def kill_random_instance(self, instances: List[str]) -> str:
        """Kill random instance"""
        victim = random.choice(instances)
        # Implementation would actually kill the instance
        return victim

    def run_experiment(self, experiment_func: Callable) -> :
        
        start_time = datetime.now()

        :
            result = experiment_func()
            status = 
            error = 
         Exception  e:
            result = 
            status = 
            error = (e)

        end_time = datetime.now()

        experiment_result = {
            : .name,
            : .hypothesis,
            : status,
            : result,
            : error,
            : (end_time - start_time).total_seconds(),
            : start_time
        }

        .results.append(experiment_result)
         experiment_result

Best Practices

SRE Principles

• Embrace risk management
• Set SLOs based on user experience
• Use error budgets for decision making
• Automate toil away
• Monitor the four golden signals
• Practice blameless post-mortems
• Gradual rollouts and canary deployments

Incident Management

• Clear incident severity definitions
• Defined incident commander role
• Communicate proactively
• Document timeline during incident
• Conduct post-incident reviews
• Track action items to completion
• Share learnings across teams

On-Call

• Reasonable on-call rotations
• Comprehensive runbooks
• Alert on symptoms, not causes
• Actionable alerts only
• Escalation policies
• Support on-call engineers
• Measure and reduce alert fatigue

Anti-Patterns

❌ No SLOs defined ❌ Alerts without runbooks ❌ Blame culture for incidents ❌ No post-incident reviews ❌ 100% uptime expectations ❌ Toil not tracked or reduced ❌ Manual processes for common tasks

Resources

• Google SRE Book: https://sre.google/sre-book/table-of-contents/
• Site Reliability Engineering: https://sre.google/
• SLO Workshop: https://github.com/google/slo-workshop
• Chaos Engineering: https://principlesofchaos.org/
• Prometheus: https://prometheus.io/