---
name: aiops
description: Generic AIOps (AI for IT Operations) patterns and best practices for 2025. Provides comprehensive implementation strategies for intelligent monitoring, automation, incident response, and observability across any infrastructure. Framework-agnostic approach supporting multiple monitoring platforms, cloud providers, and automation tools.
license: MIT
---

# AIOps - AI for IT Operations

This skill provides comprehensive patterns for implementing AIOps strategies in 2025, including intelligent monitoring, automated incident response, predictive analytics, and observability best practices. The patterns are designed to be framework-agnostic and applicable across different infrastructure platforms.

## When to Use This Skill

Use this skill when you need to:
- Implement AIOps strategies for modern infrastructure
- Build intelligent monitoring and alerting systems
- Create automated incident response workflows
- Deploy predictive maintenance solutions
- Implement self-healing capabilities
- Build observability platforms with AI/ML
- Optimize multi-cloud operations
- Create chaos engineering practices
- Implement generative AI for operations
- Build digital twins for infrastructure

## 1. AIOps Architecture Patterns

### Core AIOps Platform Architecture

```python
# aiops/core/architecture.py
from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import asyncio
import json
import logging

logger = logging.getLogger(__name__)

class AlertSeverity(str, Enum):
    """Alert severity levels"""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

class IncidentStatus(str, Enum):
    """Incident status"""
    OPEN = "open"
    INVESTIGATING = "investigating"
    IDENTIFIED = "identified"
    MONITORING = "monitoring"
    RESOLVED = "resolved"
    CLOSED = "closed"

@dataclass
class Metric:
    """Metric data point"""
    name: str
    value: float
    timestamp: datetime
    labels: Dict[str, str] = field(default_factory=dict)
    source: str = ""
    unit: str = ""

@dataclass
class LogEntry:
    """Structured log entry"""
    timestamp: datetime
    level: str
    message: str
    service: str
    trace_id: Optional[str] = None
    span_id: Optional[str] = None
    labels: Dict[str, str] = field(default_factory=dict)
    stack_trace: Optional[str] = None

@dataclass
class Trace:
    """Distributed trace data"""
    trace_id: str
    spans: List[Dict[str, Any]]
    duration: timedelta
    service_map: Dict[str, List[str]]
    error_count: int = 0

@dataclass
class Alert:
    """Alert representation"""
    id: str
    name: str
    severity: AlertSeverity
    status: IncidentStatus
    description: str
    source: str
    timestamp: datetime
    labels: Dict[str, str] = field(default_factory=dict)
    annotations: Dict[str, str] = field(default_factory=dict)
    fingerprint: str = ""
    related_entities: List[str] = field(default_factory=list)

class DataSource(ABC):
    """Abstract data source interface"""

    @abstractmethod
    async def collect_metrics(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime
    ) -> List[Metric]:
        """Collect metrics from data source"""
        pass

    @abstractmethod
    async def query_logs(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime,
        limit: int = 100
    ) -> List[LogEntry]:
        """Query logs from data source"""
        pass

    @abstractmethod
    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """Get trace data"""
        pass

class AIOpsEngine:
    """Core AIOps processing engine"""

    def __init__(self):
        self.data_sources: List[DataSource] = []
        self.alert_processors: List[AlertProcessor] = []
        self.ml_models: Dict[str, Any] = {}
        self.knowledge_base: KnowledgeBase = KnowledgeBase()
        self.automation_engine = AutomationEngine()

    def register_data_source(self, data_source: DataSource) -> None:
        """Register a data source"""
        self.data_sources.append(data_source)
        logger.info(f"Registered data source: {type(data_source).__name__}")

    def register_alert_processor(self, processor: 'AlertProcessor') -> None:
        """Register an alert processor"""
        self.alert_processors.append(processor)
        logger.info(f"Registered alert processor: {type(processor).__name__}")

    async def process_alert(self, alert: Alert) -> Dict[str, Any]:
        """Process incoming alert through AIOps pipeline"""
        logger.info(f"Processing alert: {alert.id} - {alert.name}")

        # 1. Enrich alert with context
        enriched_alert = await self._enrich_alert(alert)

        # 2. Run through ML models for classification and prediction
        analysis = await self._analyze_alert(enriched_alert)

        # 3. Determine appropriate response
        response_plan = await self._generate_response_plan(enriched_alert, analysis)

        # 4. Execute automated actions if applicable
        if response_plan.get("auto_execute", False):
            await self._execute_response(enriched_alert, response_plan)

        # 5. Update knowledge base
        await self.knowledge_base.update_from_alert(enriched_alert, analysis, response_plan)

        return {
            "alert_id": alert.id,
            "status": "processed",
            "analysis": analysis,
            "response_plan": response_plan
        }

    async def _enrich_alert(self, alert: Alert) -> Alert:
        """Enrich alert with additional context"""
        # Get related metrics
        for source in self.data_sources:
            try:
                metrics = await source.collect_metrics(
                    query=f"{alert.name}[5m]",
                    start_time=alert.timestamp - timedelta(minutes=5),
                    end_time=alert.timestamp
                )

                # Add metrics to alert annotations
                alert.annotations[f"metrics_{type(source).__name__}"] = json.dumps([
                    {"name": m.name, "value": m.value} for m in metrics
                ])
            except Exception as e:
                logger.error(f"Failed to enrich alert with metrics: {e}")

        # Get related logs
        for source in self.data_sources:
            try:
                logs = await source.query_logs(
                    query=f"service:{alert.source} level:error",
                    start_time=alert.timestamp - timedelta(minutes=5),
                    end_time=alert.timestamp,
                    limit=10
                )

                # Add logs to alert context
                alert.annotations["recent_errors"] = json.dumps([
                    {"timestamp": l.timestamp.isoformat(), "message": l.message}
                    for l in logs
                ])
            except Exception as e:
                logger.error(f"Failed to enrich alert with logs: {e}")

        return alert

    async def _analyze_alert(self, alert: Alert) -> Dict[str, Any]:
        """Analyze alert using ML models"""
        analysis = {
            "classification": None,
            "severity_score": 0.0,
            "predicted_impact": "low",
            "recommendations": [],
            "related_incidents": []
        }

        # Classification model
        if "classification" in self.ml_models:
            features = self._extract_features(alert)
            analysis["classification"] = self.ml_models["classification"].predict([features])[0]

        # Severity prediction
        if "severity" in self.ml_models:
            features = self._extract_features(alert)
            analysis["severity_score"] = self.ml_models["severity"].predict_proba([features])[0][1]

        # Impact prediction
        if "impact" in self.ml_models:
            features = self._extract_features(alert)
            analysis["predicted_impact"] = self.ml_models["impact"].predict([features])[0]

        # Find related incidents from knowledge base
        analysis["related_incidents"] = await self.knowledge_base.find_similar_incidents(alert)

        return analysis

    async def _generate_response_plan(
        self,
        alert: Alert,
        analysis: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Generate automated response plan"""
        plan = {
            "priority": alert.severity,
            "actions": [],
            "escalation_rules": [],
            "auto_execute": False,
            "estimated_resolution_time": None
        }

        # Determine if auto-execution is safe
        if (analysis["classification"] in ["known_issue", "false_positive"] and
            alert.severity in [AlertSeverity.LOW, AlertSeverity.MEDIUM]):
            plan["auto_execute"] = True

        # Generate actions based on classification
        classification = analysis.get("classification")

        if classification == "resource_exhaustion":
            plan["actions"].append({
                "type": "scale",
                "target": alert.source,
                "params": {"replicas": "+20%"}
            })
            plan["actions"].append({
                "type": "notify",
                "channel": "devops",
                "message": f"Resource exhaustion detected in {alert.source}"
            })

        elif classification == "known_issue":
            # Apply known fix
            known_fix = await self.knowledge_base.get_known_fix(alert.name)
            if known_fix:
                plan["actions"].append(known_fix)

        elif classification == "false_positive":
            plan["actions"].append({
                "type": "suppress",
                "duration": "1h",
                "reason": "False positive detected"
            })

        # Add escalation rules
        if alert.severity == AlertSeverity.CRITICAL:
            plan["escalation_rules"].append({
                "condition": "no_response_5m",
                "action": "escalate_to_oncall",
                "timeout": 300
            })

        return plan

    def _extract_features(self, alert: Alert) -> List[float]:
        """Extract features for ML models"""
        # Simple feature extraction - in production, use more sophisticated methods
        features = [
            float(alert.severity == AlertSeverity.CRITICAL),
            float(alert.severity == AlertSeverity.HIGH),
            float(alert.severity == AlertSeverity.MEDIUM),
            len(alert.description),
            len(alert.labels),
            alert.timestamp.hour / 24.0,
            alert.timestamp.weekday() / 7.0
        ]
        return features

    async def _execute_response(
        self,
        alert: Alert,
        response_plan: Dict[str, Any]
    ) -> None:
        """Execute automated response actions"""
        for action in response_plan.get("actions", []):
            try:
                await self.automation_engine.execute(action)
                logger.info(f"Executed action for alert {alert.id}: {action['type']}")
            except Exception as e:
                logger.error(f"Failed to execute action: {e}")
```

### Monitoring Platform Integrations

```python
# aiops/integrations/monitoring.py
import aiohttp
import asyncio
from typing import List, Dict, Any, Optional
from datetime import datetime
from ..core.architecture import DataSource, Metric, LogEntry, Trace, Alert

class PrometheusDataSource(DataSource):
    """Prometheus metrics integration"""

    def __init__(self, url: str):
        self.url = url.rstrip("/")
        self.session = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if not self.session:
            self.session = aiohttp.ClientSession()
        return self.session

    async def collect_metrics(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime
    ) -> List[Metric]:
        """Query Prometheus for metrics"""
        session = await self._get_session()

        params = {
            "query": query,
            "start": start_time.timestamp(),
            "end": end_time.timestamp(),
            "step": "15s"
        }

        url = f"{self.url}/api/v1/query_range"

        async with session.get(url, params=params) as response:
            data = await response.json()
            metrics = []

            if data["status"] == "success":
                result = data["data"]
                if result["resultType"] == "matrix":
                    for series in result["result"]:
                        labels = series["metric"]
                        values = series["values"]

                        for timestamp, value in values:
                            metrics.append(Metric(
                                name=query,
                                value=float(value),
                                timestamp=datetime.fromtimestamp(float(timestamp)),
                                labels=labels,
                                source="prometheus"
                            ))

            return metrics

    async def query_logs(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime,
        limit: int = 100
    ) -> List[LogEntry]:
        """Prometheus doesn't store logs - use Loki integration"""
        return []

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """Prometheus doesn't store traces - use Jaeger/Tempo integration"""
        return None

class LokiDataSource(DataSource):
    """Loki log aggregation integration"""

    def __init__(self, url: str):
        self.url = url.rstrip("/")
        self.session = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if not self.session:
            self.session = aiohttp.ClientSession()
        return self.session

    async def collect_metrics(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime
    ) -> List[Metric]:
        """Loki doesn't store metrics"""
        return []

    async def query_logs(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime,
        limit: int = 100
    ) -> List[LogEntry]:
        """Query Loki for logs"""
        session = await self._get_session()

        params = {
            "query": query,
            "start": start_time.isoformat(),
            "end": end_time.isoformat(),
            "limit": str(limit)
        }

        url = f"{self.url}/loki/api/v1/query_range"

        async with session.get(url, params=params) as response:
            data = await response.json()
            logs = []

            if data["status"] == "success":
                result = data["data"]
                for series in result["result"]:
                    stream = series["stream"]
                    for timestamp, line in series["values"]:
                        logs.append(LogEntry(
                            timestamp=datetime.fromtimestamp(float(timestamp) / 1e9),
                            level=stream.get("level", "info"),
                            message=line,
                            service=stream.get("service", "unknown"),
                            labels=stream
                        ))

            return logs

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """Loki doesn't store traces"""
        return None

class JaegerDataSource(DataSource):
    """Jaeger distributed tracing integration"""

    def __init__(self, url: str):
        self.url = url.rstrip("/")
        self.session = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if not self.session:
            self.session = aiohttp.ClientSession()
        return self.session

    async def collect_metrics(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime
    ) -> List[Metric]:
        """Jaeger doesn't store metrics"""
        return []

    async def query_logs(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime,
        limit: int = 100
    ) -> List[LogEntry]:
        """Jaeger doesn't store logs"""
        return []

    async def get_trace(self, trace_id: str) -> Optional[Trace]:
        """Get trace from Jaeger"""
        session = await self._get_session()

        url = f"{self.url}/api/traces/{trace_id}"

        async with session.get(url) as response:
            data = await response.json()

            if data["data"] and data["data"][0]["spans"]:
                spans = data["data"][0]["spans"]
                service_map = {}
                error_count = 0

                for span in spans:
                    service = span["process"]["serviceName"]
                    if service not in service_map:
                        service_map[service] = []
                    service_map[service].append(span["spanID"])

                    if span.get("tags"):
                        for tag in span["tags"]:
                            if tag.get("key") == "error" and tag.get("value"):
                                error_count += 1

                return Trace(
                    trace_id=trace_id,
                    spans=spans,
                    duration=timedelta(
                        microseconds=data["data"][0]["duration"]
                    ),
                    service_map=service_map,
                    error_count=error_count
                )

            return None

class DatadogDataSource(DataSource):
    """Datadog APM integration"""

    def __init__(self, api_key: str, app_key: str, site: str = "datadoghq.com"):
        self.api_key = api_key
        self.app_key = app_key
        self.site = site
        self.base_url = f"https://api.{site}"
        self.session = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if not self.session:
            headers = {
                "DD-API-KEY": self.api_key,
                "DD-APPLICATION-KEY": self.app_key
            }
            self.session = aiohttp.ClientSession(headers=headers)
        return self.session

    async def collect_metrics(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime
    ) -> List[Metric]:
        """Query Datadog metrics"""
        session = await self._get_session()

        params = {
            "query": query,
            "from": int(start_time.timestamp()),
            "to": int(end_time.timestamp())
        }

        url = f"{self.base_url}/api/v1/query"

        async with session.get(url, params=params) as response:
            data = await response.json()
            metrics = []

            if "series" in data:
                for series in data["series"]:
                    for point in series["pointlist"]:
                        metrics.append(Metric(
                            name=series["metric"],
                            value=point[1],
                            timestamp=datetime.fromtimestamp(point[0]),
                            labels={k: str(v) for k, v in series["tag_set"].items()},
                            source="datadog",
                            unit=series.get("unit", "")
                        ))

            return metrics

    async def query_logs(
        self,
        query: str,
        start_time: datetime,
        end_time: datetime,
        limit: int = 100
    ) -> List[LogEntry]:
        """Query Datadog logs"""
        session = await self._get_session()

        payload = {
            "query": query,
            "time": {
                "from": start_time.isoformat(),
                "to": end_time.isoformat()
            },
            "limit": limit
        }

        url = f"{self.base_url}/api/v2/logs/events/search"

        async with session.post(url, json=payload) as response: