| import random |
| import datetime |
| import json |
| import uuid |
| from typing import Dict, List, Any, Optional |
|
|
| class SyntheticIncidentSimulator: |
| """ |
| Generates high-fidelity synthetic microservices data (logs, metrics, alerts, deploys) |
| for SREBench-Env scenarios. |
| """ |
| def __init__(self, seed: int = 42): |
| self.seed = seed |
| self.random = random.Random(seed) |
| self.services = ["api-gateway", "auth-service", "payment-service", "db-proxy"] |
|
|
| def generate_scenario(self, task_name: str) -> Dict[str, Any]: |
| """Creates a deterministic, timestamp-perfect scenario.""" |
| |
| self.start_time = datetime.datetime(2026, 4, 3, 10, 0, 0) |
| |
| short_id = str(uuid.UUID(int=self.random.getrandbits(128)))[:8] |
| self.incident_id = f"INC-{short_id}" |
| |
| if task_name == "alert-classifier": |
| return self._create_easy_scenario() |
| elif task_name == "root-cause-correlator": |
| return self._create_medium_scenario() |
| elif task_name == "incident-commander": |
| return self._create_hard_scenario() |
| else: |
| raise ValueError(f"Unknown task: {task_name}") |
|
|
| def _create_easy_scenario(self) -> Dict[str, Any]: |
| """Task 1: Alert Classifier — Signal in logs/metrics for triage.""" |
| alerts = [ |
| {"id": "A1", "severity": "P1", "service": "auth-service", "title": "High Error Rate (>40%)", "triggered_at": self.start_time.isoformat(), "description": "Auth service is failing to validate tokens."}, |
| {"id": "A2", "severity": "P2", "service": "api-gateway", "title": "Monitor Flapping", "triggered_at": (self.start_time + datetime.timedelta(minutes=2)).isoformat(), "description": "Connectivity test is intermittent."}, |
| {"id": "A3", "severity": "P3", "service": "payment-service", "title": "Slight Latency Bump", "triggered_at": self.start_time.isoformat(), "description": "Slight p99 increase observed."} |
| ] |
| |
| |
| auth_logs = [ |
| self._make_log("auth-service", "ERROR", "Invalid JWT: Signature verification failed", offset_sec=i*10) |
| for i in range(5) |
| ] |
| |
| |
| error_metrics = { |
| (self.start_time - datetime.timedelta(minutes=i)).isoformat(): 0.45 |
| for i in range(5) |
| } |
| |
| return { |
| "incident_id": self.incident_id, |
| "scenario": "alert-classifier", |
| "ground_truth": {"root_cause_service": "auth-service", "root_cause_severity": "P1"}, |
| "services": self.services, |
| "alerts": alerts, |
| "logs": {"auth-service": auth_logs, "api-gateway": [], "payment-service": [], "db-proxy": []}, |
| "metrics": {"auth-service": {"error_rate": error_metrics}}, |
| "deployments": [], |
| "runbooks": self.get_runbook_store() |
| } |
|
|
| def _create_medium_scenario(self) -> Dict[str, Any]: |
| """Task 2: Root Cause Correlator — Chronological latency spike.""" |
| alerts = [ |
| {"id": "A1", "severity": "P2", "service": "payment-service", "title": "High Latency (p99 > 2000ms)", "triggered_at": self.start_time.isoformat(), "description": "Payment processing is timing out. High latency detected downstream of db-proxy."} |
| ] |
| |
| |
| latency_history = {} |
| for i in range(15): |
| ts = (self.start_time - datetime.timedelta(minutes=14-i)).isoformat() |
| |
| latency_history[ts] = 120 if i < 3 else (2100 + i*10) |
| |
| payment_logs = [ |
| self._make_log("payment-service", "ERROR", "Timeout waiting for database response", latency_ms=2150, offset_sec=i*12) |
| for i in range(5) |
| ] |
| |
| |
| db_logs = [ |
| self._make_log("db-proxy", "WARN", "Slow query detected: SELECT * FROM tokens WHERE id = ?", latency_ms=1800, offset_sec=i*5) |
| for i in range(3) |
| ] |
| |
| deploys = [ |
| {"timestamp": (self.start_time - datetime.timedelta(minutes=12)).isoformat(), "service": "db-proxy", "version": "v1.2.4", "commit": "badc0de", "author": "dev-a"}, |
| {"timestamp": (self.start_time - datetime.timedelta(hours=2)).isoformat(), "service": "payment-service", "version": "v2.1.0", "commit": "a1b2c3d", "author": "dev-b"} |
| ] |
| |
| return { |
| "incident_id": self.incident_id, |
| "scenario": "root-cause-correlator", |
| "ground_truth": {"root_cause_service": "db-proxy", "correct_commit": "badc0de"}, |
| "services": self.services, |
| "alerts": alerts, |
| "logs": {"payment-service": payment_logs, "db-proxy": db_logs, "api-gateway": [], "auth-service": []}, |
| "metrics": {"payment-service": {"latency_p99": latency_history}}, |
| "deployments": deploys, |
| "runbooks": self.get_runbook_store() |
| } |
|
|
| def _create_hard_scenario(self) -> Dict[str, Any]: |
| """Task 3: Incident Commander — Red Herrings & Cascading errors.""" |
| alerts = [ |
| {"id": "A1", "severity": "P1", "service": "api-gateway", "title": "High Error Rate (>45%)", "triggered_at": self.start_time.isoformat(), "description": "Major breakage at edge."}, |
| {"id": "A2", "severity": "P2", "service": "auth-service", "title": "High Memory Usage (>90%)", "triggered_at": self.start_time.isoformat(), "description": "Memory saturation."}, |
| {"id": "A3", "severity": "P3", "service": "payment-service", "title": "Inbound Errors", "triggered_at": self.start_time.isoformat(), "description": "Failure cascading upstream."} |
| ] |
| |
| |
| deploys = [ |
| {"timestamp": (self.start_time + datetime.timedelta(minutes=1)).isoformat(), "service": "payment-service", "version": "v2.2.0", "commit": "badf00d", "author": "dev-x"}, |
| {"timestamp": (self.start_time - datetime.timedelta(days=1)).isoformat(), "service": "auth-service", "version": "v1.9.0", "commit": "fedcba9", "author": "dev-y"} |
| ] |
| |
| auth_logs = [ |
| self._make_log("auth-service", "WARN", "Memory usage at 91%. GC overhead limit exceeded.", offset_sec=i*8) |
| for i in range(10) |
| ] |
| |
| |
| api_logs = [self._make_log("api-gateway", "ERROR", "Upstream auth-service returned 503", offset_sec=i*12) for i in range(5)] |
| payment_logs = [self._make_log("payment-service", "ERROR", "Auth verification failed: service unavailable", offset_sec=i*15) for i in range(5)] |
| |
| |
| memory_progression = [0.70, 0.78, 0.85, 0.90, 0.92] |
| memory_metrics = { |
| (self.start_time - datetime.timedelta(minutes=4-i)).isoformat(): memory_progression[i] |
| for i in range(5) |
| } |
| error_metrics = { |
| (self.start_time - datetime.timedelta(minutes=i)).isoformat(): 0.45 |
| for i in range(5) |
| } |
| |
| return { |
| "incident_id": self.incident_id, |
| "scenario": "incident-commander", |
| "ground_truth": {"root_cause_service": "auth-service", "root_cause_type": "memory_leak", "correct_action": "restart_service"}, |
| "services": self.services, |
| "alerts": alerts, |
| "logs": {"auth-service": auth_logs, "api-gateway": api_logs, "payment-service": payment_logs, "db-proxy": []}, |
| "metrics": { |
| "auth-service": {"memory_usage": memory_metrics}, |
| "api-gateway": {"error_rate": error_metrics} |
| }, |
| "deployments": deploys, |
| "runbooks": self.get_runbook_store() |
| } |
|
|
| def _make_log(self, service: str, level: str, msg: str, latency_ms: int = 100, offset_sec: int = 0) -> str: |
| """Helper to create a structured JSON log line with variable timestamp.""" |
| ts = (self.start_time + datetime.timedelta(seconds=offset_sec)).isoformat() |
| log_data = { |
| "timestamp": ts, |
| "level": level, |
| "service": service, |
| "msg": msg, |
| "latency_ms": latency_ms, |
| "request_id": f"req-{self.random.randint(100000, 999999)}" |
| } |
| return json.dumps(log_data) |
|
|
| def get_runbook_store(self) -> List[Dict[str, str]]: |
| """Provides a searchable list of runbooks.""" |
| return [ |
| { |
| "title": "Memory Leak Mitigation", |
| "content": "If a service hits >90% heap usage, restart it immediately to clear the leak. Check GC logs for confirmation.", |
| "tags": "memory, leak, saturation, auth-service, restart" |
| }, |
| { |
| "title": "Rollback Procedure", |
| "content": "In case of latent degradation following a deployment, revert the service to the last known stable commit hash.", |
| "tags": "deploy, rollback, rollback_deploy, revert" |
| }, |
| { |
| "title": "Database Connectivity Issues", |
| "content": "High db-proxy latency often points to slow queries or connection pool exhaustion. Check db logs for slow queries.", |
| "tags": "db, sql, query, latency, db-proxy" |
| }, |
| { |
| "title": "Auth Failures Triage", |
| "content": "P1 auth failures often involve invalid signatures or token expiration. Verify JWT metadata in auth-service logs.", |
| "tags": "auth, jwt, token, p1" |
| } |
| ] |
|
|
