Spaces:
Sleeping
Sleeping
| """ | |
| Symptom Engine for Repair Strategy System V2. | |
| HARD RULES ENFORCED: | |
| 1. Log ambiguity: No single log message uniquely identifies one root cause. | |
| Every log maps to >= 2 root causes. | |
| 2. Metrics are derived from the cascade-affected symptomatic state. | |
| """ | |
| import random | |
| from typing import Dict, List, Set | |
| from models_v2 import LogEntry, LogSeverity, ComponentMetrics | |
| # Hard Rule #1: Ambiguous overlapping log pool | |
| # Every message MUST appear in the list for >= 2 distinct components. | |
| # The keys represent the 'symptomatic' state of a component (which may be caused by a different root cause) | |
| LOG_POOL: Dict[str, Dict[str, List[str]]] = { | |
| "database": { | |
| "down": [ | |
| "Connection timeout", # Shared with Cache down | |
| "Service unavailable", # Shared with API down | |
| "Upstream dependency failure", # Shared with Cache down | |
| "Request queue backing up", # Shared with Queue overloaded | |
| "High response time", # Shared with Queue overloaded | |
| "Elevated read pressure", # Shared with Cache down | |
| "Consumer lag detected", # Shared with Queue overloaded | |
| "504 Gateway Timeout", # Shared with API degraded | |
| "Slow slow-query log rising", # Shared with API degraded | |
| "Cache miss rate critical", # Shared with Cache down | |
| "Fatal IO error" # Shared with Queue down | |
| ] | |
| }, | |
| "api": { | |
| "down": [ | |
| "Service unavailable", # Shared with DB down | |
| "Endpoint unhandled exception", # Shared with Cache down | |
| "Process exited unexpectedly" # Shared with Queue overloaded | |
| ], | |
| "degraded": [ | |
| "504 Gateway Timeout", # Shared with DB down | |
| "Slow slow-query log rising", # Shared with DB down | |
| "Thread pool exhaustion" # Shared with Queue overloaded | |
| ] | |
| }, | |
| "cache": { | |
| "down": [ | |
| "Connection timeout", # Shared with DB down | |
| "Upstream dependency failure", # Shared with DB down | |
| "Cache miss rate critical", # Shared with DB down (read pressure) | |
| "Elevated read pressure", # Shared with DB down | |
| "Memory pool exhausted", # Shared with Queue overloaded | |
| "Queue backlog exceeding threshold", # Shared with Queue overloaded | |
| "Endpoint unhandled exception", # Shared with API down | |
| "Broker connection reset" # Shared with Queue down | |
| ] | |
| }, | |
| "queue": { | |
| "down": [ | |
| "Process exited unexpectedly", # Shared with API down | |
| "Broker connection reset", # Shared with Cache down | |
| "Fatal IO error" # Shared with DB down | |
| ], | |
| "degraded": [ # Corresponds to "overloaded" | |
| "High response time", # Shared with DB down | |
| "Request queue backing up", # Shared with DB down | |
| "Memory pool exhausted", # Shared with Cache down | |
| "Queue backlog exceeding threshold", # Shared with Cache down | |
| "Consumer lag detected", # Shared with DB down | |
| "Thread pool exhaustion" # Shared with API degraded | |
| ] | |
| } | |
| } | |
| # Healthy baseline metrics | |
| BASE_METRICS = { | |
| "database": {"latency": 30, "error_rate": 0.0, "queue_load": 0.0, "cpu_usage": 15.0}, | |
| "api": {"latency": 50, "error_rate": 0.0, "queue_load": 0.1, "cpu_usage": 20.0}, | |
| "cache": {"latency": 10, "error_rate": 0.0, "queue_load": 0.0, "cpu_usage": 30.0}, | |
| "queue": {"latency": 20, "error_rate": 0.0, "queue_load": 0.2, "cpu_usage": 10.0}, | |
| } | |
| class SymptomEngine: | |
| def __init__(self, seed: int): | |
| self._rng = random.Random(seed) | |
| def _get_log_variations(self, base_log: str) -> str: | |
| variations = { | |
| "Connection timeout": [ | |
| "Connection timeout on socket {}...", | |
| "Err: socket hangup {}ms", | |
| "Timeout Error: Target unreachable" | |
| ], | |
| "Service unavailable": [ | |
| "Service unavailable (503) at path /{}", | |
| "CRITICAL: Application unresponsive", | |
| "HTTP 503 Service Temporarily Unavailable" | |
| ], | |
| "Upstream dependency failure": [ | |
| "[DependencyError] Upstream failed to respond", | |
| "Upstream dependency failure: {} retry", | |
| "Failed to ping upstream node {}" | |
| ], | |
| "Request queue backing up": [ | |
| "Request queue backing up, count: {}", | |
| "WARNING: Queue depth exceeding safe limits", | |
| "High request buffering ({})" | |
| ], | |
| "High response time": [ | |
| "High response time > {}ms", | |
| "Latency spike detected in APM: {}", | |
| "Slow response: {}ms avg" | |
| ], | |
| "Elevated read pressure": [ | |
| "Elevated read pressure on node {}", | |
| "High read IOPS detected ({})", | |
| "Warning: Read volume approaching limit" | |
| ], | |
| "Consumer lag detected": [ | |
| "Consumer lag detected: {} units", | |
| "Processing delay on consumer group {}", | |
| "Lag spike: {} offset difference" | |
| ], | |
| "504 Gateway Timeout": [ | |
| "504 Gateway Timeout on {}", | |
| "Gateway timeout contacting upstream", | |
| "Nginx 504: Target took too long" | |
| ], | |
| "Slow slow-query log rising": [ | |
| "Slow slow-query log rising", | |
| "Slow query detected: {} sec", | |
| "DB Query execution time > {}ms" | |
| ], | |
| "Cache miss rate critical": [ | |
| "Cache miss rate critical ({}%)", | |
| "Warning: Cache hit/miss ratio inverted", | |
| "Excessive cache misses on key prefix {}" | |
| ], | |
| "Fatal IO error": [ | |
| "Fatal IO error at block {}", | |
| "Disk IO exception volume mounted at {}", | |
| "IOError: operation failed" | |
| ], | |
| "Memory pool exhausted": [ | |
| "Memory pool exhausted (alloc: {}MB)", | |
| "OOM Warning: pool size limit reached", | |
| "Cannot allocate memory for {}" | |
| ], | |
| "Queue backlog exceeding threshold": [ | |
| "Queue backlog exceeding threshold: {} reqs", | |
| "Backlog critical: {} items", | |
| "Spooling exceeded maximum partition limit" | |
| ], | |
| "Endpoint unhandled exception": [ | |
| "Endpoint unhandled exception in route {}", | |
| "Unhandled runtime Error at GET /{}", | |
| "500 Internal Server Error: {}" | |
| ], | |
| "Broker connection reset": [ | |
| "Broker connection reset by peer", | |
| "TCP RST received from broker {}", | |
| "Connection dropped: Broker unavailable" | |
| ] | |
| } | |
| pool = variations.get(base_log, [base_log]) | |
| choice = self._rng.choice(pool) | |
| # Random template filling for realism | |
| if "{}" in choice: | |
| filler = self._rng.randint(10, 9999) | |
| choice = choice.format(filler) | |
| return choice | |
| self._verify_ambiguity_constraint() | |
| def _verify_ambiguity_constraint(self): | |
| """ | |
| Hard Rule verification at startup: | |
| Ensures NO log exists that maps to exactly 1 root cause. | |
| (We map to symptomatic components here, but the principle is that | |
| different root causes trigger these symptoms, making the log ambiguous). | |
| """ | |
| message_to_sources: Dict[str, Set[str]] = {} | |
| for comp, statuses in LOG_POOL.items(): | |
| for status, msgs in statuses.items(): | |
| for msg in msgs: | |
| if msg not in message_to_sources: | |
| message_to_sources[msg] = set() | |
| message_to_sources[msg].add(comp) | |
| violations = [] | |
| for msg, sources in message_to_sources.items(): | |
| if len(sources) < 2: | |
| # Some logs map to different *statuses* of the same component, | |
| # but to be truly ambiguous across root causes, we need strict overlap. | |
| # The manual pool defines them carefully, but this safety check ensures | |
| # future additions don't break the rule. | |
| violations.append(f"'{msg}' only appears in {sources}") | |
| # If the pool is perfectly designed, violations will be empty. | |
| if violations: | |
| raise ValueError(f"Ambiguity Constraint Violation! Logs uniquely identifying a component: {violations}") | |
| def generate_metrics(self, state: Dict[str, Dict[str, any]]) -> Dict[str, ComponentMetrics]: | |
| """ | |
| Generate metrics based on the current (cascade-applied) state. | |
| """ | |
| metrics = {} | |
| # 1. Base translation from status to metrics | |
| for comp, data in state.items(): | |
| status = data["status"] | |
| base = BASE_METRICS[comp] | |
| m = {"latency": base["latency"], "error_rate": base["error_rate"], | |
| "queue_load": base["queue_load"], "cpu_usage": base["cpu_usage"]} | |
| if status == "down": | |
| m["latency"] = 9999 | |
| m["error_rate"] = 1.0 | |
| m["cpu_usage"] = 0.0 # Process is dead | |
| elif status == "degraded": | |
| m["latency"] *= int(3 + self._rng.random() * 2) | |
| m["error_rate"] = 0.3 + self._rng.random() * 0.2 | |
| m["cpu_usage"] = min(100.0, m["cpu_usage"] * 4) | |
| metrics[comp] = m | |
| # 2. Cross-component metric leaking (Symptoms) | |
| # If DB is down, API naturally struggles | |
| if state["database"]["status"] == "down" and state["api"]["status"] != "down": | |
| metrics["api"]["latency"] += int(300 + self._rng.random() * 100) | |
| metrics["api"]["error_rate"] = min(1.0, metrics["api"]["error_rate"] + 0.5) | |
| metrics["api"]["cpu_usage"] = min(100.0, metrics["api"]["cpu_usage"] * 2) | |
| # If Cache is down, Queue load naturally spikes due to DB pressure | |
| if state["cache"]["status"] == "down": | |
| metrics["queue"]["queue_load"] += 0.3 + self._rng.random() * 0.2 | |
| # Format output | |
| formatted_metrics = {} | |
| for comp, m in metrics.items(): | |
| formatted_metrics[comp] = ComponentMetrics( | |
| latency=int(m["latency"]), | |
| error_rate=round(m["error_rate"], 2), | |
| queue_load=round(m["queue_load"], 2), | |
| cpu_usage=round(m["cpu_usage"], 1) | |
| ) | |
| return formatted_metrics | |
| def generate_logs(self, state: Dict[str, Dict[str, any]], step: int) -> List[LogEntry]: | |
| """ | |
| Generate ambiguous logs based on active symptoms. | |
| Returns 0-3 logs per broken component. | |
| """ | |
| logs = [] | |
| for comp, data in state.items(): | |
| status = data["status"] | |
| if status == "healthy": | |
| continue | |
| pool = LOG_POOL.get(comp, {}).get(status, []) | |
| if not pool: | |
| # Fallback if specific status pool isn't defined | |
| pool = LOG_POOL.get(comp, {}).get("down", ["System error"]) | |
| # Randomly select 1-2 logs for this symptom | |
| num_logs = self._rng.randint(1, 2) | |
| selected_msgs = self._rng.sample(pool, min(num_logs, len(pool))) | |
| for msg in selected_msgs: | |
| severity = LogSeverity.ERROR if status == "down" else LogSeverity.WARN | |
| logs.append(LogEntry( | |
| timestamp=step * 1000 + self._rng.randint(10, 990), | |
| message=msg, | |
| severity=severity | |
| )) | |
| # Shuffle to prevent component ordering from being a tell | |
| self._rng.shuffle(logs) | |
| return logs | |