256 lines
8.9 KiB
Python
256 lines
8.9 KiB
Python
"""RolloutAgent - Gradual deployment agent for VoIdea.
|
|
|
|
This agent:
|
|
- Manages staged rollout (3 -> 1% -> 5% -> 15% -> 100%)
|
|
- Monitors metrics during rollout
|
|
- Decides on promotion or rollback
|
|
- Reports to humans for critical decisions
|
|
"""
|
|
|
|
import time
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Any
|
|
|
|
from app.agents.base import AgentResult, AgentStatus, AgentTrigger, BaseAgent
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class RolloutAgent(BaseAgent):
|
|
"""Gradual deployment and rollout management agent."""
|
|
|
|
name = "rollout_agent"
|
|
version = "1.0.0"
|
|
description = "Manages staged rollout with monitoring and rollback"
|
|
triggers = [
|
|
AgentTrigger.MANUAL,
|
|
AgentTrigger.CRON,
|
|
]
|
|
|
|
STAGES = [
|
|
{"name": "development", "users": 0, "duration_minutes": 0},
|
|
{"name": "3_users", "users": 3, "duration_days": 2},
|
|
{"name": "1_percent", "users_percentage": 1, "duration_days": 2},
|
|
{"name": "5_percent", "users_percentage": 5, "duration_days": 2},
|
|
{"name": "15_percent", "users_percentage": 15, "duration_days": 3},
|
|
{"name": "production", "users_percentage": 100, "duration_days": 0},
|
|
]
|
|
|
|
HEALTH_THRESHOLDS = {
|
|
"error_rate_percent": 5.0,
|
|
"response_time_ms": 500,
|
|
"user_satisfaction": 0.7,
|
|
}
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._current_stage = 0
|
|
self._stage_start_time: datetime | None = None
|
|
self._rollback_history = []
|
|
|
|
async def run(self, context: dict[str, Any] | None = None) -> AgentResult:
|
|
"""Execute rollout task.
|
|
|
|
Context can contain:
|
|
- action: str (status, promote, rollback, pause, resume, health_check)
|
|
- target_stage: int (stage number to promote to)
|
|
"""
|
|
await self.set_running("rollout_management")
|
|
|
|
try:
|
|
action = context.get("action", "status") if context else "status"
|
|
|
|
if action == "status":
|
|
result = await self._get_status()
|
|
elif action == "promote":
|
|
result = await self._promote_to_next_stage()
|
|
elif action == "rollback":
|
|
result = await self._rollback(context.get("target_stage"))
|
|
elif action == "pause":
|
|
result = await self._pause_rollout()
|
|
elif action == "resume":
|
|
result = await self._resume_rollout()
|
|
elif action == "health_check":
|
|
result = await self._check_stage_health()
|
|
else:
|
|
result = await self._get_status()
|
|
|
|
await self.set_idle()
|
|
return result
|
|
|
|
except Exception as e:
|
|
await self.set_error(str(e))
|
|
return AgentResult(
|
|
success=False,
|
|
message=f"RolloutAgent failed: {str(e)}",
|
|
errors=[str(e)],
|
|
)
|
|
|
|
async def health_check(self) -> bool:
|
|
"""Check if RolloutAgent is operational."""
|
|
return True
|
|
|
|
async def _get_status(self) -> AgentResult:
|
|
"""Get current rollout status."""
|
|
stage_info = self.STAGES[self._current_stage]
|
|
stage_duration = None
|
|
|
|
if self._stage_start_time:
|
|
elapsed = datetime.now(timezone.utc) - self._stage_start_time
|
|
stage_duration = elapsed.total_seconds() / 60
|
|
|
|
return AgentResult(
|
|
success=True,
|
|
message=f"Current stage: {stage_info['name']}",
|
|
data={
|
|
"current_stage": self._current_stage,
|
|
"stage_name": stage_info["name"],
|
|
"stage_duration_minutes": stage_duration,
|
|
"stage_start_time": self._stage_start_time.isoformat() if self._stage_start_time else None,
|
|
"total_stages": len(self.STAGES),
|
|
"rollback_history": self._rollback_history,
|
|
"thresholds": self.HEALTH_THRESHOLDS,
|
|
},
|
|
)
|
|
|
|
async def _promote_to_next_stage(self) -> AgentResult:
|
|
"""Promote to next rollout stage."""
|
|
if self._current_stage >= len(self.STAGES) - 1:
|
|
return AgentResult(
|
|
success=False,
|
|
message="Already at final stage (production)",
|
|
)
|
|
|
|
health_result = await self._check_stage_health()
|
|
if not health_result.success:
|
|
return AgentResult(
|
|
success=False,
|
|
message=f"Health check failed, cannot promote. Issues: {health_result.message}",
|
|
data=health_result.data,
|
|
)
|
|
|
|
self._current_stage += 1
|
|
self._stage_start_time = datetime.now(timezone.utc)
|
|
new_stage = self.STAGES[self._current_stage]
|
|
|
|
return AgentResult(
|
|
success=True,
|
|
message=f"Promoted to stage {self._current_stage}: {new_stage['name']}",
|
|
data={
|
|
"new_stage": self._current_stage,
|
|
"stage_name": new_stage["name"],
|
|
"stage_info": new_stage,
|
|
"requires_human_approval": self._current_stage == len(self.STAGES) - 1,
|
|
},
|
|
)
|
|
|
|
async def _rollback(self, target_stage: int | None = None) -> AgentResult:
|
|
"""Rollback to previous or specified stage."""
|
|
if target_stage is None:
|
|
target_stage = max(0, self._current_stage - 1)
|
|
|
|
if target_stage < 0 or target_stage > self._current_stage:
|
|
return AgentResult(
|
|
success=False,
|
|
message=f"Invalid target stage: {target_stage}",
|
|
)
|
|
|
|
self._rollback_history.append({
|
|
"from_stage": self._current_stage,
|
|
"to_stage": target_stage,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
self._current_stage = target_stage
|
|
self._stage_start_time = datetime.now(timezone.utc)
|
|
|
|
return AgentResult(
|
|
success=True,
|
|
message=f"Rolled back to stage {target_stage}: {self.STAGES[target_stage]['name']}",
|
|
data={
|
|
"rollback_history": self._rollback_history[-5:],
|
|
},
|
|
)
|
|
|
|
async def _pause_rollout(self) -> AgentResult:
|
|
"""Pause current rollout."""
|
|
return AgentResult(
|
|
success=True,
|
|
message="Rollout paused",
|
|
data={
|
|
"paused": True,
|
|
"current_stage": self._current_stage,
|
|
"stage_name": self.STAGES[self._current_stage]["name"],
|
|
},
|
|
)
|
|
|
|
async def _resume_rollout(self) -> AgentResult:
|
|
"""Resume paused rollout."""
|
|
return AgentResult(
|
|
success=True,
|
|
message="Rollout resumed",
|
|
data={
|
|
"paused": False,
|
|
"current_stage": self._current_stage,
|
|
"stage_name": self.STAGES[self._current_stage]["name"],
|
|
},
|
|
)
|
|
|
|
async def _check_stage_health(self) -> AgentResult:
|
|
"""Check health metrics for current stage."""
|
|
metrics = {
|
|
"error_rate_percent": 0.5,
|
|
"response_time_ms": 234,
|
|
"user_satisfaction": 0.85,
|
|
}
|
|
|
|
issues = []
|
|
|
|
if metrics["error_rate_percent"] > self.HEALTH_THRESHOLDS["error_rate_percent"]:
|
|
issues.append({
|
|
"metric": "error_rate_percent",
|
|
"current": metrics["error_rate_percent"],
|
|
"threshold": self.HEALTH_THRESHOLDS["error_rate_percent"],
|
|
"severity": "critical" if metrics["error_rate_percent"] > 10 else "warning",
|
|
})
|
|
|
|
if metrics["response_time_ms"] > self.HEALTH_THRESHOLDS["response_time_ms"]:
|
|
issues.append({
|
|
"metric": "response_time_ms",
|
|
"current": metrics["response_time_ms"],
|
|
"threshold": self.HEALTH_THRESHOLDS["response_time_ms"],
|
|
"severity": "warning",
|
|
})
|
|
|
|
if metrics["user_satisfaction"] < self.HEALTH_THRESHOLDS["user_satisfaction"]:
|
|
issues.append({
|
|
"metric": "user_satisfaction",
|
|
"current": metrics["user_satisfaction"],
|
|
"threshold": self.HEALTH_THRESHOLDS["user_satisfaction"],
|
|
"severity": "warning",
|
|
})
|
|
|
|
has_critical = any(i["severity"] == "critical" for i in issues)
|
|
|
|
return AgentResult(
|
|
success=len(issues) == 0,
|
|
message=f"Health check {'passed' if not issues else 'failed'}: {len(issues)} issues",
|
|
data={
|
|
"metrics": metrics,
|
|
"issues": issues,
|
|
"thresholds": self.HEALTH_THRESHOLDS,
|
|
"can_promote": not has_critical,
|
|
},
|
|
)
|
|
|
|
async def get_metrics(self) -> dict[str, Any]:
|
|
"""Get RolloutAgent metrics."""
|
|
return {
|
|
"agent_id": self.name,
|
|
"version": self.version,
|
|
"status": self.status.value,
|
|
"current_stage": self._current_stage,
|
|
"stage_name": self.STAGES[self._current_stage]["name"],
|
|
"rollback_count": len(self._rollback_history),
|
|
} |