To implement a safe rollback orchestrator, use a deterministic sequence: (1) validate current state and health endpoints, (2) shift a small canary percentage (10%) to the previous model, (3) monitor health checks for a fixed validation window, and (4) either complete rollback (move 100% traffic) or revert the canary. Below is pseudo-Python orchestration logic (no real networking calls — placeholders simulate actions).python
import time
from typing import Callable
# Placeholders for traffic control and health check operations
def set_traffic_split(model_a_pct: int, model_b_pct: int):
# API call to traffic router / service mesh would go here
print(f"Traffic: model_new={model_a_pct}%, model_prev={model_b_pct}%")
def check_health(endpoint: str) -> bool:
# Replace with HTTP GET + status parsing in real system
return True
def wait_and_validate(endpoint: str, duration_s: int, interval_s: int, validator: Callable[[str], bool]):
end = time.time() + duration_s
while time.time() < end:
if not validator(endpoint):
return False
time.sleep(interval_s)
return True
def orchestrate_safe_rollback(prev_model_endpoint: str, new_model_endpoint: str,
validation_duration=300, check_interval=10):
# 1. pre-checks
if not check_health(prev_model_endpoint):
raise RuntimeError("Previous model unhealthy; abort rollback")
if not check_health(new_model_endpoint):
print("New model unhealthy — initiating rollback canary")
# 2. shift 10% traffic to previous model as canary
set_traffic_split(model_a_pct=90, model_b_pct=10) # new=90%, prev=10%
# 3. validate health for fixed period
ok_prev = wait_and_validate(prev_model_endpoint, validation_duration, check_interval, check_health)
ok_new = wait_and_validate(new_model_endpoint, validation_duration, check_interval, check_health)
# 4. decide
if ok_prev and not ok_new:
print("Completing rollback: routing 100% to previous model")
set_traffic_split(model_a_pct=0, model_b_pct=100)
else:
print("Reverting canary: restoring 100% to new model")
set_traffic_split(model_a_pct=100, model_b_pct=0)
