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: str, value: float, timestamp: datetime):
"""Record SLI measurement"""
if slo_name in self.slos:
self.measurements[slo_name].append({
'value': value,
'timestamp': timestamp
})
def calculate_slo_compliance(self, slo_name: str) -> Dict:
"""Calculate SLO compliance"""
slo = self.slos.get(slo_name)
if not slo:
return {}
measurements = self.measurements.get(slo_name, [])
window_start = datetime.now() - timedelta(days=slo.window_days)
recent_measurements = [
m for m in measurements
if m['timestamp'] > window_start
]
if not recent_measurements:
return {'status': 'no_data'}
values = [m['value'] for m in recent_measurements]
actual = np.mean(values)
return {
'slo_name': slo_name,
'target': slo.target,
'actual': actual,
'compliant': actual >= slo.target,
'window_days': slo.window_days,
'sample_count': len(recent_measurements)
}
def calculate_error_budget(self, slo_name: str) -> Dict:
"""Calculate remaining error budget"""
compliance = self.calculate_slo_compliance(slo_name)
if compliance.get('status') == 'no_data':
return {'status': 'no_data'}
target = compliance['target']
actual = compliance['actual']
error_budget_target = 100 - target
errors_actual = 100 - actual
remaining = error_budget_target - errors_actual
remaining_pct = (remaining / error_budget_target) * 100 if error_budget_target > 0 else 100
return {
'slo_name': slo_name,
'error_budget_target': error_budget_target,
'errors_actual': errors_actual,
'remaining': remaining,
'remaining_percentage': remaining_pct,
'exhausted': remaining < 0
}
# Example SLOs
def define_standard_slos() -> List[SLO]:
"""Define standard SLOs for a web service"""
return [
SLO(
name="api_availability",
sli=SLI(
name="availability",
description="Percentage of successful requests",
query="sum(rate(http_requests_total{code!~'5..'}[5m])) / sum(rate(http_requests_total[5m])) * 100",
unit="percentage"
),
target=99.9,
window_days=30
),
SLO(
name="api_latency",
sli=SLI(
name="latency_p95",
description="95th percentile latency",
query="histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))",
unit="seconds"
),
target=0.5, # 500ms
window_days=30
)
]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):
self.incidents: Dict[str, Incident] = {}
def create_incident(self, incident: Incident) -> str:
"""Create new incident"""
self.incidents[incident.incident_id] = incident
# Notify on-call
self.notify_oncall(incident)
# Start incident timeline
self.add_timeline_event(
incident.incident_id,
"Incident created",
datetime.now()
)
return incident.incident_id
def update_status(self, incident_id: str, new_status: IncidentStatus,
note: str):
"""Update incident status"""
if incident_id in self.incidents:
incident = self.incidents[incident_id]
incident.status = new_status
self.add_timeline_event(
incident_id,
f"Status changed to {new_status.value}: {note}",
datetime.now()
)
if new_status == IncidentStatus.RESOLVED:
incident.resolved_at = datetime.now()
def add_timeline_event(self, incident_id: str, event: str,
timestamp: datetime):
"""Add event to incident timeline"""
if incident_id in self.incidents:
self.incidents[incident_id].timeline.append({
'timestamp': timestamp,
'event': event
})
def calculate_mttr(self, incident_id: str) -> Optional[float]:
"""Calculate Mean Time To Resolution"""
incident = self.incidents.get(incident_id)
if incident and incident.resolved_at:
duration = incident.resolved_at - incident.detected_at
return duration.total_seconds() / 60 # minutes
return None
def generate_incident_report(self, incident_id: str) -> Dict:
"""Generate incident report"""
incident = self.incidents.get(incident_id)
if not incident:
return {}
return {
'incident_id': incident.incident_id,
'title': incident.title,
'severity': incident.severity.value,
'status': incident.status.value,
'duration_minutes': self.calculate_mttr(incident_id),
'affected_services': incident.affected_services,
'incident_commander': incident.incident_commander,
'responders': incident.responders,
'timeline': incident.timeline,
'root_cause': incident.root_cause
}
def notify_oncall(self, incident: Incident):
"""Notify on-call engineer (integrate with PagerDuty, etc.)"""
# Implementation would integrate with alerting system
passMonitoring 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,
'annotations': {
'summary': f'{name} alert triggered',
'runbook_url': f'https://runbooks.example.com/{name}'
}
}
self.alerts.append(alert)
return alert
def check_golden_signals(self, metrics: Dict) -> Dict:
"""Check the four golden signals"""
return {
'latency': self._check_latency(metrics.get('latency', [])),
'traffic': self._check_traffic(metrics.get('traffic', 0)),
'errors': self._check_errors(metrics.get('error_rate', 0)),
'saturation': self._check_saturation(metrics.get('cpu_usage', 0))
}
def _check_latency(self, latencies: List[float]) -> Dict:
if not latencies:
return {'status': 'unknown'}
p95 = np.percentile(latencies, 95)
return {
'status': 'critical' if p95 > 1000 else 'ok',
'p95_ms': p95
}
def _check_traffic(self, requests_per_second: float) -> Dict:
return {
'status': 'ok',
'rps': requests_per_second
}
def _check_errors(self, error_rate: float) -> Dict:
return {
'status': 'critical' if error_rate > 1.0 else 'ok',
'error_rate': error_rate
}
def _check_saturation(self, cpu_usage: float) -> Dict:
return {
'status': 'warning' if cpu_usage > 80 else 'ok',
'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) -> Dict:
"""Run chaos experiment"""
start_time = datetime.now()
try:
result = experiment_func()
status = "success"
error = None
except Exception as e:
result = None
status = "failed"
error = str(e)
end_time = datetime.now()
experiment_result = {
'name': self.name,
'hypothesis': self.hypothesis,
'status': status,
'result': result,
'error': error,
'duration': (end_time - start_time).total_seconds(),
'timestamp': start_time
}
self.results.append(experiment_result)
return experiment_resultBest 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/