# incident-response-automation.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: incident-response-config
  namespace: incident-management
data:
  response-automation.yaml: |
    incident_classification:
      severity_levels:
        sev1:
          name: "Critical - Service Down"
          description: "Complete service outage affecting all users"
          response_time: "5 minutes"
          escalation_time: "15 minutes"
          stakeholders: ["sre", "engineering_manager", "cto"]
          automatic_actions:
            - create_war_room
            - notify_executives
            - trigger_emergency_response
            - start_status_page_incident
        
        sev2:
          name: "High - Major Functionality Impaired"
          description: "Core functionality severely degraded"
          response_time: "15 minutes"
          escalation_time: "30 minutes"
          stakeholders: ["sre", "engineering_manager"]
          automatic_actions:
            - create_war_room
            - notify_stakeholders
            - start_status_page_incident
        
        sev3:
          name: "Medium - Minor Functionality Impaired"
          description: "Non-critical functionality affected"
          response_time: "1 hour"
          escalation_time: "2 hours"
          stakeholders: ["sre", "product_owner"]
          automatic_actions:
            - create_slack_channel
            - notify_team
        
        sev4:
          name: "Low - Minimal Impact"
          description: "Minor issues with workarounds available"
          response_time: "4 hours"
          escalation_time: "8 hours"
          stakeholders: ["sre"]
          automatic_actions:
            - create_ticket
            - schedule_fix
    
    automated_response_workflows:
      service_down_detection:
        triggers:
          - alert: "service_availability < 50%"
            duration: "2 minutes"
          - alert: "error_rate > 10%"
            duration: "3 minutes"
          - alert: "response_time > 5000ms"
            duration: "5 minutes"
        
        actions:
          - type: "create_incident"
            severity: "sev1"
            title: "Service {{ $labels.service }} experiencing outage"
          
          - type: "execute_runbook"
            runbook: "service_recovery_playbook"
            parameters:
              service: "{{ $labels.service }}"
              environment: "{{ $labels.environment }}"
          
          - type: "scale_resources"
            target: "deployment/{{ $labels.service }}"
            replicas: "{{ .current_replicas * 2 }}"
          
          - type: "notify_stakeholders"
            channels: ["#incidents", "#sre-alerts"]
            message: "Critical incident detected for {{ $labels.service }}"
      
      database_performance_degradation:
        triggers:
          - alert: "database_connection_pool_utilization > 90%"
            duration: "5 minutes"
          - alert: "database_query_duration_p95 > 1000ms"
            duration: "10 minutes"
        
        actions:
          - type: "create_incident"
            severity: "sev2"
            title: "Database performance degradation detected"
          
          - type: "execute_runbook"
            runbook: "database_performance_optimization"
          
          - type: "enable_read_replicas"
            database: "{{ $labels.database }}"
          
          - type: "throttle_non_critical_queries"
            priority_threshold: "low"
    
    escalation_policies:
      primary_oncall:
        - type: "pagerduty"
          escalation_delay: "5 minutes"
          retry_count: 3
        
        - type: "phone"
          escalation_delay: "2 minutes"
          retry_count: 2
        
        - type: "sms"
          escalation_delay: "1 minute"
          retry_count: 3
      
      management_escalation:
        - type: "slack"
          channel: "#leadership"
          conditions:
            - "severity == 'sev1'"
            - "duration > '30 minutes'"
        
        - type: "email"
          recipients: ["cto@company.com", "vp-engineering@company.com"]
          conditions:
            - "severity == 'sev1'"
            - "duration > '60 minutes'"
---
# Automated runbook execution
apiVersion: batch/v1
kind: Job
metadata:
  name: incident-response-automation
  namespace: incident-management
spec:
  template:
    spec:
      serviceAccountName: incident-responder
      containers:
      - name: incident-automation
        image: company/incident-automation:v2.1.0
        command:
        - /bin/sh
        - -c
        - |
          echo "Starting incident response automation..."
          
          # Process incoming alerts
          python3 /app/scripts/alert_processor.py
          
          # Execute automated responses
          python3 /app/scripts/response_executor.py
          
          # Update incident status
          python3 /app/scripts/status_updater.py
          
          echo "Incident response automation completed"
        
        env:
        - name: PAGERDUTY_API_KEY
          valueFrom:
            secretKeyRef:
              name: incident-secrets
              key: pagerduty_api_key
        
        - name: SLACK_BOT_TOKEN
          valueFrom:
            secretKeyRef:
              name: incident-secrets
              key: slack_bot_token
        
        - name: KUBERNETES_SERVICE_ACCOUNT
          valueFrom:
            fieldRef:
              fieldPath: spec.serviceAccountName
        
        volumeMounts:
        - name: runbooks
          mountPath: /app/runbooks
          readOnly: true
        - name: incident-data
          mountPath: /app/data
      
      volumes:
      - name: runbooks
        configMap:
          name: incident-runbooks
      - name: incident-data
        persistentVolumeClaim:
          claimName: incident-data
      
      restartPolicy: OnFailure

#!/usr/bin/env python3
# post-mortem-automation.py

import asyncio
import json
import yaml
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
import pandas as pd
import numpy as np
import re
import logging
from jinja2 import Template
import requests
import openai

@dataclass
class IncidentTimeline:
    """Represents an incident timeline event."""
    timestamp: datetime
    event_type: str
    description: str
    actor: str
    data: Dict[str, Any]

@dataclass
class PostMortemAnalysis:
    """Represents post-mortem analysis results."""
    incident_id: str
    title: str
    severity: str
    start_time: datetime
    end_time: datetime
    duration_minutes: int
    impact_assessment: Dict[str, Any]
    timeline: List[IncidentTimeline]
    root_cause: str
    contributing_factors: List[str]
    action_items: List[Dict[str, Any]]
    lessons_learned: List[str]
    preventive_measures: List[str]
    confidence_score: float

class PostMortemAutomationEngine:
    """Advanced post-mortem automation and analysis engine."""
    
    def __init__(self, config: Dict):
        self.config = config
        self.openai_client = openai.OpenAI(api_key=config.get('openai_api_key'))
        
        # Configure logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
        # Initialize integrations
        self.pagerduty_api = config.get('pagerduty_api_url')
        self.pagerduty_token = config.get('pagerduty_token')
        self.prometheus_url = config.get('prometheus_url')
        self.github_token = config.get('github_token')
    
    async def generate_post_mortem(self, incident_id: str) -> PostMortemAnalysis:
        """Generate comprehensive post-mortem analysis."""
        try:
            self.logger.info(f"Generating post-mortem for incident {incident_id}")
            
            # Gather incident data
            incident_data = await self._gather_incident_data(incident_id)
            
            # Analyze timeline
            timeline = await self._analyze_incident_timeline(incident_data)
            
            # Extract metrics and impact
            impact_assessment = await self._assess_incident_impact(incident_data, timeline)
            
            # Perform root cause analysis
            root_cause_analysis = await self._perform_root_cause_analysis(
                incident_data, timeline, impact_assessment
            )
            
            # Generate action items
            action_items = await self._generate_action_items(
                incident_data, root_cause_analysis
            )
            
            # Extract lessons learned
            lessons_learned = await self._extract_lessons_learned(
                incident_data, root_cause_analysis
            )
            
            # Create post-mortem analysis
            post_mortem = PostMortemAnalysis(
                incident_id=incident_id,
                title=incident_data.get('title', f'Incident {incident_id}'),
                severity=incident_data.get('severity', 'unknown'),
                start_time=datetime.fromisoformat(incident_data['start_time']),
                end_time=datetime.fromisoformat(incident_data['end_time']),
                duration_minutes=incident_data.get('duration_minutes', 0),
                impact_assessment=impact_assessment,
                timeline=timeline,
                root_cause=root_cause_analysis['primary_cause'],
                contributing_factors=root_cause_analysis['contributing_factors'],
                action_items=action_items,
                lessons_learned=lessons_learned,
                preventive_measures=root_cause_analysis['preventive_measures'],
                confidence_score=root_cause_analysis['confidence_score']
            )
            
            # Save post-mortem
            await self._save_post_mortem(post_mortem)
            
            # Create GitHub issue for tracking
            await self._create_github_issue(post_mortem)
            
            # Generate and publish report
            await self._publish_post_mortem_report(post_mortem)
            
            return post_mortem
            
        except Exception as e:
            self.logger.error(f"Error generating post-mortem: {e}")
            raise
    
    async def _gather_incident_data(self, incident_id: str) -> Dict:
        """Gather comprehensive incident data from multiple sources."""
        try:
            # Get incident details from PagerDuty
            pagerduty_data = await self._get_pagerduty_incident(incident_id)
            
            # Get alerts and metrics from Prometheus
            metrics_data = await self._get_incident_metrics(
                pagerduty_data['start_time'],
                pagerduty_data['end_time']
            )
            
            # Get log data
            log_data = await self._get_incident_logs(
                pagerduty_data['start_time'],
                pagerduty_data['end_time']
            )
            
            # Get deployment and change data
            change_data = await self._get_recent_changes(
                pagerduty_data['start_time']
            )
            
            return {
                'incident_details': pagerduty_data,
                'metrics': metrics_data,
                'logs': log_data,
                'changes': change_data,
                'title': pagerduty_data.get('title', ''),
                'severity': pagerduty_data.get('severity', ''),
                'start_time': pagerduty_data.get('start_time', ''),
                'end_time': pagerduty_data.get('end_time', ''),
                'duration_minutes': pagerduty_data.get('duration_minutes', 0)
            }
            
        except Exception as e:
            self.logger.error(f"Error gathering incident data: {e}")
            return {}
    
    async def _analyze_incident_timeline(self, incident_data: Dict) -> List[IncidentTimeline]:
        """Analyze and construct incident timeline."""
        timeline_events = []
        
        try:
            # Process PagerDuty timeline
            for event in incident_data.get('incident_details', {}).get('timeline', []):
                timeline_events.append(IncidentTimeline(
                    timestamp=datetime.fromisoformat(event['timestamp']),
                    event_type='pagerduty',
                    description=event['description'],
                    actor=event.get('actor', 'system'),
                    data=event
                ))
            
            # Process metrics anomalies
            for anomaly in incident_data.get('metrics', {}).get('anomalies', []):
                timeline_events.append(IncidentTimeline(
                    timestamp=datetime.fromisoformat(anomaly['timestamp']),
                    event_type='metric_anomaly',
                    description=f"Metric {anomaly['metric']} exceeded threshold",
                    actor='monitoring_system',
                    data=anomaly
                ))
            
            # Process deployment events
            for deployment in incident_data.get('changes', {}).get('deployments', []):
                timeline_events.append(IncidentTimeline(
                    timestamp=datetime.fromisoformat(deployment['timestamp']),
                    event_type='deployment',
                    description=f"Deployment {deployment['version']} to {deployment['environment']}",
                    actor=deployment.get('deployer', 'unknown'),
                    data=deployment
                ))
            
            # Sort timeline by timestamp
            timeline_events.sort(key=lambda x: x.timestamp)
            
            return timeline_events
            
        except Exception as e:
            self.logger.error(f"Error analyzing timeline: {e}")
            return []
    
    async def _assess_incident_impact(self, incident_data: Dict, timeline: List[IncidentTimeline]) -> Dict:
        """Assess the impact of the incident."""
        try:
            metrics = incident_data.get('metrics', {})
            
            # Calculate user impact
            user_impact = {
                'affected_users': metrics.get('affected_users', 0),
                'error_rate_peak': metrics.get('error_rate_peak', 0),
                'response_time_impact': metrics.get('response_time_impact', 0),
                'availability_impact': metrics.get('availability_impact', 0)
            }
            
            # Calculate business impact
            business_impact = {
                'revenue_loss_estimate': metrics.get('revenue_loss', 0),
                'transaction_loss': metrics.get('transaction_loss', 0),
                'customer_complaints': metrics.get('customer_complaints', 0),
                'sla_breach': metrics.get('sla_breach', False)
            }
            
            # Calculate technical impact
            technical_impact = {
                'services_affected': metrics.get('services_affected', []),
                'infrastructure_impact': metrics.get('infrastructure_impact', {}),
                'data_integrity_issues': metrics.get('data_integrity_issues', False),
                'security_implications': metrics.get('security_implications', False)
            }
            
            return {
                'user_impact': user_impact,
                'business_impact': business_impact,
                'technical_impact': technical_impact,
                'overall_severity': self._calculate_overall_severity(
                    user_impact, business_impact, technical_impact
                )
            }
            
        except Exception as e:
            self.logger.error(f"Error assessing impact: {e}")
            return {}
    
    async def _perform_root_cause_analysis(
        self, 
        incident_data: Dict, 
        timeline: List[IncidentTimeline], 
        impact_assessment: Dict
    ) -> Dict:
        """Perform AI-assisted root cause analysis."""
        try:
            # Prepare context for AI analysis
            context = {
                'incident_summary': incident_data.get('title', ''),
                'severity': incident_data.get('severity', ''),
                'timeline_events': [
                    {
                        'timestamp': event.timestamp.isoformat(),
                        'type': event.event_type,
                        'description': event.description,
                        'actor': event.actor
                    }
                    for event in timeline
                ],
                'metrics_data': incident_data.get('metrics', {}),
                'recent_changes': incident_data.get('changes', {}),
                'impact_assessment': impact_assessment
            }
            
            # Use AI to analyze root cause
            analysis_prompt = f"""
            Analyze the following incident data and provide a root cause analysis:
            
            {json.dumps(context, indent=2)}
            
            Please provide:
            1. Primary root cause
            2. Contributing factors
            3. Preventive measures
            4. Confidence score (0-1)
            
            Focus on technical accuracy and actionable insights.
            """
            
            response = await self._query_ai_assistant(analysis_prompt)
            
            # Parse AI response
            root_cause_analysis = self._parse_ai_response(response)
            
            # Validate and enrich analysis
            enriched_analysis = await self._enrich_root_cause_analysis(
                root_cause_analysis, context
            )
            
            return enriched_analysis
            
        except Exception as e:
            self.logger.error(f"Error performing root cause analysis: {e}")
            return {
                'primary_cause': 'Analysis failed - manual investigation required',
                'contributing_factors': [],
                'preventive_measures': [],
                'confidence_score': 0.0
            }
    
    async def _generate_action_items(self, incident_data: Dict, root_cause_analysis: Dict) -> List[Dict]:
        """Generate actionable items from incident analysis."""
        action_items = []
        
        try:
            # System-generated action items based on root cause
            primary_cause = root_cause_analysis.get('primary_cause', '')
            
            if 'deployment' in primary_cause.lower():
                action_items.extend([
                    {
                        'title': 'Improve deployment validation',
                        'description': 'Enhance pre-deployment testing and validation processes',
                        'priority': 'high',
                        'owner': 'engineering_team',
                        'due_date': (datetime.now() + timedelta(days=14)).isoformat(),
                        'category': 'process_improvement'
                    },
                    {
                        'title': 'Implement deployment canary analysis',
                        'description': 'Add automated canary deployment with rollback triggers',
                        'priority': 'high',
                        'owner': 'platform_team',
                        'due_date': (datetime.now() + timedelta(days=30)).isoformat(),
                        'category': 'technical_improvement'
                    }
                ])
            
            if 'monitoring' in primary_cause.lower() or 'detection' in primary_cause.lower():
                action_items.extend([
                    {
                        'title': 'Enhance monitoring coverage',
                        'description': 'Add missing alerts and improve detection time',
                        'priority': 'high',
                        'owner': 'sre_team',
                        'due_date': (datetime.now() + timedelta(days=7)).isoformat(),
                        'category': 'monitoring_improvement'
                    }
                ])
            
            # Add preventive measures as action items
            for measure in root_cause_analysis.get('preventive_measures', []):
                action_items.append({
                    'title': f'Implement: {measure}',
                    'description': measure,
                    'priority': 'medium',
                    'owner': 'tbd',
                    'due_date': (datetime.now() + timedelta(days=21)).isoformat(),
                    'category': 'prevention'
                })
            
            return action_items
            
        except Exception as e:
            self.logger.error(f"Error generating action items: {e}")
            return []
    
    async def _query_ai_assistant(self, prompt: str) -> str:
        """Query AI assistant for analysis."""
        try:
            response = await self.openai_client.chat.completions.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": "You are an expert site reliability engineer analyzing production incidents."},
                    {"role": "user", "content": prompt}
                ],
                max_tokens=2000,
                temperature=0.3
            )
            
            return response.choices[0].message.content
            
        except Exception as e:
            self.logger.error(f"Error querying AI assistant: {e}")
            return "AI analysis unavailable"
    
    async def _publish_post_mortem_report(self, post_mortem: PostMortemAnalysis) -> None:
        """Generate and publish post-mortem report."""
        try:
            # Generate report using template
            template = Template("""
# Post-Mortem: {{ post_mortem.title }}

## Incident Summary
- **Incident ID**: {{ post_mortem.incident_id }}
- **Severity**: {{ post_mortem.severity }}
- **Start Time**: {{ post_mortem.start_time }}
- **End Time**: {{ post_mortem.end_time }}
- **Duration**: {{ post_mortem.duration_minutes }} minutes

## Impact Assessment
{{ post_mortem.impact_assessment | tojson(indent=2) }}

## Timeline
{% for event in post_mortem.timeline %}
- **{{ event.timestamp }}** ({{ event.event_type }}): {{ event.description }} - {{ event.actor }}
{% endfor %}

## Root Cause Analysis
**Primary Cause**: {{ post_mortem.root_cause }}

**Contributing Factors**:
{% for factor in post_mortem.contributing_factors %}
- {{ factor }}
{% endfor %}

## Action Items
{% for item in post_mortem.action_items %}
### {{ item.title }}
- **Priority**: {{ item.priority }}
- **Owner**: {{ item.owner }}
- **Due Date**: {{ item.due_date }}
- **Description**: {{ item.description }}
{% endfor %}

## Lessons Learned
{% for lesson in post_mortem.lessons_learned %}
- {{ lesson }}
{% endfor %}

## Preventive Measures
{% for measure in post_mortem.preventive_measures %}
- {{ measure }}
{% endfor %}

---
*This post-mortem was generated automatically and reviewed by the SRE team.*
            """)
            
            report_content = template.render(post_mortem=post_mortem)
            
            # Save to repository
            await self._save_report_to_repository(post_mortem.incident_id, report_content)
            
            # Notify stakeholders
            await self._notify_stakeholders(post_mortem)
            
            self.logger.info(f"Post-mortem report published for {post_mortem.incident_id}")
            
        except Exception as e:
            self.logger.error(f"Error publishing report: {e}")

async def main():
    """Main function for post-mortem automation."""
    config = {
        'openai_api_key': 'sk-...',
        'pagerduty_api_url': 'https://api.pagerduty.com',
        'pagerduty_token': 'u+...',
        'prometheus_url': 'https://prometheus.company.com',
        'github_token': 'ghp_...'
    }
    
    engine = PostMortemAutomationEngine(config)
    
    # Example: Generate post-mortem for incident
    incident_id = "INC-2026-001"
    post_mortem = await engine.generate_post_mortem(incident_id)
    
    print(f"Post-mortem generated for incident {incident_id}")
    print(f"Root cause: {post_mortem.root_cause}")
    print(f"Action items: {len(post_mortem.action_items)}")

if __name__ == '__main__':
    asyncio.run(main())