Spaces:
Sleeping
Sleeping
File size: 4,855 Bytes
44c4c2d 0ece8e9 a78a875 0ece8e9 44c4c2d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """Base class for all SRE incident tasks."""
from abc import ABC, abstractmethod
from typing import Dict, Any, Tuple, List
from app.models import Observation, Alert, ServiceStatus, LogEntry, MetricPoint
from datetime import datetime, timezone
BASE_INCIDENT_TIME = "2024-11-15T09:47:00Z"
AVAILABLE_ACTIONS = [
"query_logs",
"check_metrics",
"restart_service",
"rollback_deployment",
"scale_service",
"kill_query",
"acknowledge_alert",
"examine_trace",
"check_config",
"resolve_incident",
]
ACTION_SCHEMA = {
"query_logs": {
"description": "Fetch recent log entries for a service",
"parameters": {
"service": {"type": "string", "required": True, "description": "Service name"},
"lines": {"type": "integer", "required": False, "default": 50},
},
},
"check_metrics": {
"description": "Retrieve current metrics for a service",
"parameters": {
"service": {"type": "string", "required": True},
},
},
"restart_service": {
"description": "Restart a service (rolling restart, brief downtime)",
"parameters": {
"service": {"type": "string", "required": True},
},
},
"rollback_deployment": {
"description": "Roll back a service to its previous deployment version",
"parameters": {
"service": {"type": "string", "required": True},
},
},
"scale_service": {
"description": "Change the number of replicas for a service",
"parameters": {
"service": {"type": "string", "required": True},
"replicas": {"type": "integer", "required": True, "min": 1, "max": 20},
},
},
"kill_query": {
"description": "Kill long-running database queries from a specific source/application",
"parameters": {
"source": {"type": "string", "required": True, "description": "Application or service holding queries"},
},
},
"acknowledge_alert": {
"description": "Acknowledge an alert to stop paging",
"parameters": {
"alert_id": {"type": "string", "required": True},
},
},
"examine_trace": {
"description": "Examine a distributed trace to identify slow spans",
"parameters": {
"trace_id": {"type": "string", "required": True},
},
},
"check_config": {
"description": "Inspect the live runtime configuration of a service",
"parameters": {
"service": {"type": "string", "required": True},
},
},
"resolve_incident": {
"description": "Mark the incident as resolved. Terminal action — ends the episode.",
"parameters": {},
},
}
class BaseTask(ABC):
task_id: str
name: str
description: str
difficulty: str
max_steps: int
passing_score: float = 0.6
@abstractmethod
def initial_state(self, seed: int = 42) -> Dict[str, Any]:
"""Return initial world state dict."""
...
@abstractmethod
def process_action(
self, action_type: str, params: Dict[str, Any], state: Dict[str, Any]
) -> Tuple[Dict[str, Any], float, bool, str]:
"""
Apply action to state.
Returns: (new_state, step_reward, done, message)
"""
...
@abstractmethod
def get_observation(self, state: Dict[str, Any], session_id: str, step: int) -> Observation:
"""Build Observation from state."""
...
@abstractmethod
def grade(self, state: Dict[str, Any], history: List[Dict]) -> Tuple[float, Dict[str, float]]:
"""
Grade the episode.
Returns: (score 0.0–1.0, breakdown dict)
"""
...
@staticmethod
def clamp_score_strict(score: float, eps: float = 0.01) -> float:
"""
Hackathon validator requirement: scores must be strictly within (0, 1).
Use this at the end of task graders to avoid returning exactly 0.0 or 1.0.
"""
try:
s = float(score)
except Exception:
s = 0.0
if s <= 0.0:
return eps
if s >= 1.0:
return 1.0 - eps
return s
def _make_service(self, name: str, status: str, cpu: float, mem: float,
err: float, **kwargs) -> ServiceStatus:
return ServiceStatus(
name=name, status=status,
cpu_percent=cpu, memory_percent=mem, error_rate=err,
**kwargs,
)
def _make_alert(self, alert_id: str, severity: str, service: str,
message: str, ack: bool = False) -> Alert:
return Alert(
alert_id=alert_id, severity=severity, service=service,
message=message, triggered_at=BASE_INCIDENT_TIME, acknowledged=ack,
)
|