sre-expert

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
        )
    ]

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
        pass

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
        }

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_result