Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import argparse | |
| from pathlib import Path | |
| import sys | |
| ROOT = Path(__file__).resolve().parents[1] | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from incident_commander_env.models import IncidentAction | |
| from incident_commander_env.server.incident_commander_environment import IncidentCommanderEnvironment | |
| SOLUTIONS = { | |
| "cache_saturation": { | |
| "root": "cache saturation from memory pressure and evictions", | |
| "mitigation": "scale cache nodes and enable cache bypass for hot keys", | |
| "topic": "cache", | |
| }, | |
| "bad_deploy_auth": { | |
| "root": "bad deploy enabled strict audience JWT validation", | |
| "mitigation": "rollback auth release and disable strict audience flag", | |
| "topic": "auth", | |
| }, | |
| "payment_provider_degradation": { | |
| "root": "payment provider degradation on card authorization route", | |
| "mitigation": "fail over to secondary provider and reroute card traffic", | |
| "topic": "payments", | |
| }, | |
| "database_connection_leak": { | |
| "root": "database connection leak from worker idle transactions", | |
| "mitigation": "pause worker and lower database pool size", | |
| "topic": "database", | |
| }, | |
| "queue_worker_crash": { | |
| "root": "worker crash left queue consumers at zero", | |
| "mitigation": "rollback worker image and scale workers with smaller batch size", | |
| "topic": "queue", | |
| }, | |
| } | |
| def call(env: IncidentCommanderEnvironment, tool_name: str, **arguments): | |
| observation = env.step(IncidentAction(tool_name=tool_name, arguments=arguments)) | |
| print(f"\n> {tool_name}({arguments})") | |
| print(observation.tool_result) | |
| return observation | |
| def first_relevant_metric(env: IncidentCommanderEnvironment) -> tuple[str, str]: | |
| for service, metric in env.scenario.relevant_metric_keys: | |
| if service in env.scenario.metrics and metric in env.scenario.metrics[service]: | |
| return service, metric | |
| service = next(iter(env.scenario.metrics)) | |
| metric = next(iter(env.scenario.metrics[service])) | |
| return service, metric | |
| def main() -> None: | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--scenario", default="cache_saturation") | |
| args = parser.parse_args() | |
| env = IncidentCommanderEnvironment() | |
| obs = env.reset(scenario_id=args.scenario) | |
| print(obs.message) | |
| solution = SOLUTIONS[env.scenario.family] | |
| first_service = env.scenario.services[0] | |
| call(env, "query_logs", service=first_service, query="error timeout failed", minutes=20) | |
| metric_service, metric = first_relevant_metric(env) | |
| call(env, "inspect_metric", service=metric_service, metric=metric, minutes=30) | |
| call(env, "read_runbook", topic=solution["topic"]) | |
| call(env, "test_hypothesis", root_cause=solution["root"]) | |
| call(env, "apply_mitigation", mitigation=solution["mitigation"]) | |
| final = call( | |
| env, | |
| "final_report", | |
| root_cause=solution["root"], | |
| mitigation=solution["mitigation"], | |
| customer_update=( | |
| "Customers are impacted by elevated failures. We have applied mitigation and " | |
| "expect recovery within 15 minutes; next update in 10 minutes." | |
| ), | |
| ) | |
| print("\nScore breakdown:", final.score_breakdown) | |
| if __name__ == "__main__": | |
| main() | |