Spaces:
Sleeping
Sleeping
File size: 4,025 Bytes
05a686e | 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 | from pydantic import BaseModel
from typing import Any, Dict, Optional
from server.actions import (
KubeAction,
ScaleAction,
DeletePodAction,
PatchAction,
RolloutRestartAction,
SetHPAAction,
DrainNodeAction,
DescribeAction,
WaitAction,
)
from server.models import ClusterObservation
class ExecutionResult(BaseModel):
observation: ClusterObservation
action_applied: str
tick_advanced: bool
describe_detail: Optional[Dict[str, Any]] = None
def execute(action: KubeAction, world) -> ExecutionResult:
if isinstance(action, ScaleAction):
return _execute_scale(action, world)
elif isinstance(action, DeletePodAction):
return _execute_delete_pod(action, world)
elif isinstance(action, PatchAction):
return _execute_patch(action, world)
elif isinstance(action, RolloutRestartAction):
return _execute_rollout_restart(action, world)
elif isinstance(action, SetHPAAction):
return _execute_set_hpa(action, world)
elif isinstance(action, DrainNodeAction):
return _execute_drain_node(action, world)
elif isinstance(action, DescribeAction):
return _execute_describe(action, world)
elif isinstance(action, WaitAction):
return _execute_wait(world)
else:
raise ValueError(f"Unknown action type: {type(action)}")
def _execute_scale(action: ScaleAction, world) -> ExecutionResult:
world.scale(action.deployment, action.replicas)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Scaled '{action.deployment}' to {action.replicas} replicas",
tick_advanced=True
)
def _execute_delete_pod(action: DeletePodAction, world) -> ExecutionResult:
world.delete_pod(action.pod_name)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Deleted pod '{action.pod_name}'",
tick_advanced=True
)
def _execute_patch(action: PatchAction, world) -> ExecutionResult:
world.apply_patch(action.resource_type, action.name, action.patch)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Patched {action.resource_type} '{action.name}'",
tick_advanced=True
)
def _execute_rollout_restart(action: RolloutRestartAction, world) -> ExecutionResult:
world.rollout_restart(action.deployment)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Rollout restarted '{action.deployment}'",
tick_advanced=True
)
def _execute_set_hpa(action: SetHPAAction, world) -> ExecutionResult:
world.set_hpa(
action.deployment,
action.min_replicas,
action.max_replicas,
action.cpu_target_percent
)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Set HPA for '{action.deployment}': {action.min_replicas}-{action.max_replicas} replicas, {action.cpu_target_percent}% CPU",
tick_advanced=True
)
def _execute_drain_node(action: DrainNodeAction, world) -> ExecutionResult:
world.drain_node(action.node_name)
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied=f"Drained node '{action.node_name}'",
tick_advanced=True
)
def _execute_describe(action: DescribeAction, world) -> ExecutionResult:
detail = world.describe(action.resource_type, action.name)
obs = world.get_observation()
return ExecutionResult(
observation=obs,
action_applied=f"Described {action.resource_type} '{action.name}'",
tick_advanced=False,
describe_detail=detail
)
def _execute_wait(world) -> ExecutionResult:
world.tick()
return ExecutionResult(
observation=world.get_observation(),
action_applied="Waited one simulation tick",
tick_advanced=True,
)
|