# dependency_discovery.py
import psutil
import numpy as np
from collections import defaultdict
from datetime import datetime, timedelta

class DependencyDiscovery:
    """基于流量模式的依赖关系发现"""
    
    def __init__(self, window_minutes=60):
        self.window = timedelta(minutes=window_minutes)
        self.flow_records = defaultdict(list)
        self.correlation_cache = {}
    
    def record_flow(self, src_port, dst_ip, dst_port, bytes_sent, timestamp):
        """记录一次网络连接"""
        key = f"{src_port}->{dst_ip}:{dst_port}"
        self.flow_records[key].append({
            'timestamp': timestamp,
            'bytes': bytes_sent
        })
    
    def compute_correlation(self, service_a, service_b, hours=24):
        """计算两个服务之间的流量相关性"""
        pattern_a = self._get_traffic_pattern(service_a, hours)
        pattern_b = self._get_traffic_pattern(service_b, hours)
        
        if len(pattern_a) < 10 or len(pattern_b) < 10:
            return 0.0
        
        # 使用皮尔逊相关系数
        corr = np.corrcoef(pattern_a, pattern_b)[0, 1]
        return corr if not np.isnan(corr) else 0.0
    
    def _get_traffic_pattern(self, service, hours):
        """提取服务在指定时间窗口内的流量模式"""
        records = self.flow_records.get(service, [])
        cutoff = datetime.now() - timedelta(hours=hours)
        values = [r['bytes'] for r in records if r['timestamp'] >= cutoff]
        
        # 按小时聚合
        hourly = defaultdict(int)
        for r in records:
            if r['timestamp'] >= cutoff:
                hour_key = r['timestamp'].replace(minute=0, second=0, microsecond=0)
                hourly[hour_key] += r['bytes']
        
        return [hourly.get(k, 0) for k in sorted(hourly.keys())]
    
    def build_topology(self, all_services, threshold=0.7):
        """构建依赖拓扑图"""
        edges = []
        for i, svc_a in enumerate(all_services):
            for svc_b in all_services[i+1:]:
                corr = self.compute_correlation(svc_a, svc_b)
                if abs(corr) >= threshold:
                    direction = "downstream" if corr > 0 else "independent"
                    edges.append({
                        'from': svc_a,
                        'from_port': svc_a.split(':')[1] if ':' in svc_a else '0',
                        'to': svc_b,
                        'to_port': svc_b.split(':')[1] if ':' in svc_b else '0',
                        'correlation': round(corr, 3),
                        'direction': direction
                    })
        return edges

# prometheus.service-discovery.yml
scrape_configs:
  - job_name: 'kubernetes-services'
    kubernetes_sd_configs:
      - role: service
    relabel_configs:
      - source_labels: [__meta_kubernetes_service_label_app]
        target_label: app
      - source_labels: [__meta_kubernetes_service_annotation_dependency_of]
        target_label: depends_on

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  annotations:
    dependency.of: "api-gateway"
    criticality: "high"
    fallback: "cache-fallback"
spec:
  # ...

# topology_visualizer.py
import networkx as nx
import json

def render_topology(edges, format="mermaid"):
    """生成依赖拓扑图"""
    if format == "mermaid":
        lines = ["graph TD"]
        for e in edges:
            strength = "strong" if e['correlation'] > 0.85 else "medium"
            lines.append(f"    {e['from']} -->|{strength}| {e['to']}")
        return "\n".join(lines)
    elif format == "json":
        return {
            "nodes": list(set(
                [e['from'] for e in edges] + [e['to'] for e in edges]
            )),
            "edges": [
                {"source": e['from'], "target": e['to'], 
                 "weight": e['correlation']}
                for e in edges
            ]
        }

# failure_propagation_predictor.py
import torch
import torch.nn as nn
import numpy as np

class FailurePropagationGNN(nn.Module):
    """基于图神经网络的故障传播预测模型"""
    
    def __init__(self, node_features=16, hidden=32, num_layers=3):
        super().__init__()
        self.gnn_layers = nn.ModuleList([
            nn.Linear(node_features if i == 0 else hidden, hidden)
            for i in range(num_layers)
        ])
        self.prediction_head = nn.Linear(hidden, 1)
        self.dropout = nn.Dropout(0.2)
    
    def forward(self, adj_matrix, node_features):
        """
        adj_matrix: (N, N) 邻接矩阵（依赖关系）
        node_features: (N, F) 节点特征（CPU、内存、连接数等）
        """
        h = node_features
        for layer in self.gnn_layers:
            # 图卷积：聚合邻居信息
            h = torch.matmul(adj_matrix, h)
            h = torch.relu(layer(h))
            h = self.dropout(h)
        
        # 预测每个节点的健康风险分数
        risk_scores = torch.sigmoid(self.prediction_head(h)).squeeze(-1)
        return risk_scores
    
    def predict_cascade(self, adj_matrix, incident_node, node_features, time_steps=5):
        """
        模拟故障从 incident_node 开始的传播
        返回每个时间步的影响范围
        """
        cascade_history = []
        affected = {incident_node}
        current_features = node_features.clone()
        
        for t in range(time_steps):
            # 受影响的节点特征恶化
            for node in affected:
                current_features[node] *= 0.8  # 资源恶化
            
            # 预测下一轮影响
            risk_scores = self.forward(adj_matrix, current_features)
            
            # 找出风险超过阈值的节点
            threshold = 0.6
            next_affected = {
                i for i, score in enumerate(risk_scores)
                if score > threshold and i not in affected
            }
            
            if not next_affected:
                break
            
            cascade_history.append({
                'step': t,
                'newly_affected': list(next_affected),
                'total_affected': len(affected | next_affected),
                'max_risk': risk_scores.max().item()
            })
            
            affected |= next_affected
        
        return cascade_history

# llm_root_cause_analyzer.py
import json
from datetime import datetime

ALERT_PROMPT = """
你是一个VPS运维专家。以下是当前触发的告警序列：

{alerts_json}

服务依赖拓扑：
{topology_json}

请分析：
1. 最可能的根因节点是哪个？
2. 故障传播路径是什么？
3. 应该优先处理哪个服务？
4. 如果根因节点暂时无法恢复，有哪些隔离措施？

请以JSON格式回答：
{{
  "root_cause": "服务名",
  "propagation_path": ["服务A", "服务B", "服务C"],
  "priority_action": "操作描述",
  "isolation_measures": ["措施1", "措施2"],
  "confidence": 0.95
}}
"""

def analyze_root_cause(alerts, topology, llm_client):
    """使用LLM分析告警序列，推断根因"""
    alerts_json = json.dumps(alerts, indent=2, ensure_ascii=False)
    topology_json = json.dumps(topology, indent=2, ensure_ascii=False)
    
    prompt = ALERT_PROMPT.format(
        alerts_json=alerts_json,
        topology_json=topology_json
    )
    
    response = llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    
    result = json.loads(response.choices[0].message.content)
    return result

# 模拟告警数据
alerts = [
    {"service": "nginx", "metric": "latency_p99", "value": 15000, "unit": "ms", "threshold": 5000},
    {"service": "api-gateway", "metric": "error_rate", "value": 0.45, "threshold": 0.05},
    {"service": "order-service", "metric": "cpu", "value": 98.5, "unit": "%", "threshold": 85},
    {"service": "mysql", "metric": "connections", "value": 500, "max": 500, "threshold": 450},
]

topology = {
    "nginx": ["api-gateway"],
    "api-gateway": ["order-service", "user-service"],
    "order-service": ["mysql"],
    "user-service": ["mysql", "redis"],
}

result = analyze_root_cause(alerts, topology, llm_client)
# 预期输出：root_cause = "mysql", propagation_path = ["mysql", "order-service", "api-gateway", "nginx"]

# adaptive_circuit_breaker.py
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 熔断
    HALF_OPEN = "half_open" # 半开，试探恢复

class AdaptiveCircuitBreaker:
    """AI驱动的自适应熔断器"""
    
    def __init__(self, service_name, dependency_graph, llm_analyzer):
        self.service = service_name
        self.graph = dependency_graph
        self.llm = llm_analyzer
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0
        self.base_timeout = 30  # 默认熔断30秒
        self.dynamic_threshold = 5  # 默认5次失败触发熔断
    
    def get_dynamic_threshold(self, current_risk_score):
        """根据AI风险评分动态调整熔断阈值"""
        # 风险越高，越容易熔断（阈值降低）
        # 风险越低，越宽容（阈值提高）
        if current_risk_score > 0.8:
            return 2  # 高风险：2次失败就熔断
        elif current_risk_score > 0.5:
            return 4
        else:
            return 8  # 低风险：宽容
    
    def record_failure(self, risk_score=0.5):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.get_dynamic_threshold(risk_score):
            self._trip_circuit(risk_score)
    
    def _trip_circuit(self, risk_score):
        """触发熔断，并通知依赖方"""
        self.state = CircuitState.OPEN
        timeout = self.base_timeout * (1 + risk_score)  # 高风险等更久
        
        # 通知上游依赖服务
        dependents = self.graph.get_reverse_deps(self.service)
        for dep in dependents:
            dep.breaker.notify_downstream_failure(self.service)
        
        # 启动超时恢复
        print(f"[CIRCUIT OPEN] {self.service} 熔断 {timeout}s，风险评分: {risk_score}")
        time.sleep(timeout)
        self.state = CircuitState.HALF_OPEN
    
    def record_success(self):
        self.success_count += 1
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            self.failure_count = 0
            print(f"[CIRCUIT CLOSED] {self.service} 恢复")
    
    def can_execute(self):
        if self.state == CircuitState.CLOSED:
            return True
        elif self.state == CircuitState.OPEN:
            return False
        else:  # HALF_OPEN
            return True  # 允许少量试探请求

# intelligent_degradation.py
from enum import Enum
import asyncio

class DegradationLevel(Enum):
    NONE = 0        # 全功能
    BASIC = 1       # 保留核心功能
    MINIMAL = 2     # 仅保留只读
    MAINTENANCE = 3 # 维护模式

class IntelligentDegradation:
    """AI驱动的主动降级"""
    
    def __init__(self, cascade_predictor, topology):
        self.predictor = cascade_predictor
        self.topology = topology
        self.current_level = DegradationLevel.NONE
        self.degradation_rules = {
            "order-service": {
                DegradationLevel.BASIC: ["query orders"],
                DegradationLevel.MINIMAL: ["read-only query"],
                DegradationLevel.MAINTENANCE: ["return 503"],
            },
            "user-service": {
                DegradationLevel.BASIC: ["read user profiles"],
                DegradationLevel.MINIMAL: ["cache-only responses"],
            },
            "recommendation-service": {
                DegradationLevel.BASIC: ["return cached recommendations"],
                DegradationLevel.MINIMAL: ["return empty list"],
                DegradationLevel.MAINTENANCE: ["disabled"],
            },
        }
    
    def assess_degradation_needed(self, cascade_risk):
        """评估是否需要降级"""
        if cascade_risk < 0.3:
            return DegradationLevel.NONE
        elif cascade_risk < 0.6:
            return DegradationLevel.BASIC
        elif cascade_risk < 0.85:
            return DegradationLevel.MINIMAL
        else:
            return DegradationLevel.MAINTENANCE
    
    def apply_degradation(self, service, level):
        """应用降级策略"""
        rules = self.degradation_rules.get(service, {})
        actions = rules.get(level, [f"return error for {service}"])
        
        print(f"[DEGRADE] {service} → {level.name}: {actions}")
        self.current_level = level
        return actions

# docker-compose.cascade-guard.yml
version: '3.8'
services:
  cascade-guard:
    build: ./cascade-guard
    container_name: cascade-guard
    volumes:
      - ./config:/app/config
      - ./data:/app/data
    environment:
      - OPENAI_API_KEY=${LLM_API_KEY}
      - PROMETHEUS_URL=http://prometheus:9090
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
      - prometheus
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    container_name: cascade-redis
    volumes:
      - cascade-redis-data:/data
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    container_name: cascade-prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    container_name: cascade-grafana
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
    depends_on:
      - prometheus
    restart: unless-stopped

volumes:
  cascade-redis-data:
  prometheus-data:
  grafana-data:

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'cascade-guard'
    static_configs:
      - targets: ['cascade-guard:8080']

  - job_name: 'node-exporter'
    static_configs:
      - targets: ['node-exporter:9100']

  - job_name: 'application'
    static_configs:
      - targets: ['order-service:8080', 'user-service:8080', 'api-gateway:8080']

rule_files:
  - 'cascade-rules.yml'

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

# cascade-rules.yml
groups:
  - name: cascade_detection
    rules:
      - alert: CascadingFailureRisk
        expr: |
          vector(cascade_risk_score{service!="cascade-guard"}) > 0.7
        for: 2m
        labels:
          severity: critical
          team: platform
        annotations:
          summary: "级联故障高风险: {{ $labels.service }}"
          description: "服务 {{ $labels.service }} 的级联风险评分为 {{ $value }}，建议立即隔离"

      - alert: DependencyBreak
        expr: |
          increase(circuit_breaker_trips_total[5m]) > 3
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "依赖断裂告警: 5分钟内熔断触发 {{ $value }} 次"

{
  "incident_id": "INC-20260817-001",
  "root_cause": "Redis内存满导致拒绝写入",
  "propagation_path": ["redis", "user-service", "api-gateway", "nginx"],
  "impact": {
    "total_affected_services": 4,
    "max_downtime_seconds": 180,
    "user_impact": "API超时率峰值45%，持续约3分钟"
  },
  "prevention_suggestions": [
    "为Redis设置maxmemory-policy=allkeys-lru，避免内存满后直接拒绝",
    "为user-service配置Redis熔断，超时200ms自动fallback到MySQL",
    "在Redis内存使用达到80%时提前触发扩容告警",
    "添加Redis Sentinel或Cluster，提升可用性"
  ],
  "simulation_result": {
    "cascades_prevented": 2,
    "avg_response_time_seconds": 4.2
  }
}