scripts/deploy_queue_qa.py
Rama main · réplica de lectura del repositorio original.
"""Queued cancellation and rollback priority on the disposable Deploy QA host."""
import concurrent.futures
import http.client
import subprocess
import threading
import time
import uuid
def run_queue(api, wait_for, traffic, network, spec_a, digest_b, port, evidence):
services, held_ips, threads = [], [], []
stop = threading.Event()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
started = time.monotonic()
metrics = {"requests": {}, "httpFailures": [], "monitorErrors": [], "stages": [],
"maxLiveVMs": 0, "maxLiveVCPUs": 0, "httpTimeoutSeconds": 5}
evidence["deployQueue"] = metrics
def path(service, suffix=""):
return "/api/micro-services/" + service["id"] + suffix
def current(service):
return api.call("GET", path(service, "/deployment"))
def operation(service, identifier):
return api.call("GET", path(service, "/deployments/" + identifier))
def routes(service):
return [route for route in api.call("GET", "/api/pipe")["config"]["routes"]
if (route.get("owner") or {}).get("serviceId") == service["id"]]
def ready(service):
record = current(service)
assert record["phase"] not in ["failed", "stopped", "cancelled"], record
return record if record["phase"] == "running" else None
def create(deploy=True):
name = "queue-" + uuid.uuid4().hex[:8]
spec = {**spec_a, "name": name, "domain": name + ".example.test", "replicas": 1,
"drainTimeout": 90}
service = {"id": api.call("POST", "/api/micro-services", spec)["id"], "spec": spec}
services.append(service)
if deploy:
api.call("POST", path(service, "/deployment"), {"microNetworkId": network})
service["initial"] = wait_for(lambda: ready(service), timeout=180)
return service
def update(service, previous):
return api.call("POST", path(service, "/updates"), {
"operationId": str(uuid.uuid4()), "expectedReleaseId": previous["id"],
"spec": {**service["spec"], "image": "ghcr.io/team/deploy-qa@" + digest_b},
"strategy": "rolling"})
def cancel(service, identifier):
record = api.call("POST", path(service, "/deployments/" + identifier + "/cancel"))
assert record["id"] == identifier and record["phase"] == "cancelled", record
return record
def hold(service):
address = current(service)["instances"][0]["ipv4"]
held_ips.append(address)
def request():
# The fixture sends progress bytes. Bound total time rather than
# relying on the HTTP client's inactivity timeout during cleanup.
result = subprocess.run(["curl", "--silent", "--show-error", "--max-time", "120",
"--header", "Host: " + service["spec"]["domain"],
"--write-out", "\n%{http_code}",
f"http://127.0.0.1:{port}/cgi-bin/hold"],
capture_output=True, text=True, check=True, timeout=125)
body, status = result.stdout.rsplit("\n", 1)
return int(status), body.strip()
future = executor.submit(request)
wait_for(lambda: traffic(address, 8080, "fixture", "/cgi-bin/holding") == (200, "holding"), timeout=10)
return address, future
def release(address, future, label):
traffic(address, 8080, "fixture", "/cgi-bin/release")
assert future.result(timeout=15) == (200, "release-" + label)
held_ips.remove(address)
def load(service):
domain = service["spec"]["domain"]
metrics["requests"][domain] = 0
while not stop.is_set():
try:
result = traffic("127.0.0.1", port, domain)
assert result in [(200, "release-A"), (200, "release-B")], result
except Exception as error:
if len(metrics["httpFailures"]) < 10:
metrics["httpFailures"].append({"domain": domain, "error": str(error)})
metrics["requests"][domain] += 1
stop.wait(.03)
def observe():
while not stop.is_set():
try:
live = [vm for vm in api.call("GET", "/api/vms")
if vm["state"] in ["running", "starting", "paused"]]
cpus = sum(vm["cpu"] for vm in live)
assert len(live) <= 4 and cpus <= 4, "Deploy queue exceeded the shared surge limit"
metrics["maxLiveVMs"] = max(metrics["maxLiveVMs"], len(live))
metrics["maxLiveVCPUs"] = max(metrics["maxLiveVCPUs"], cpus)
except (OSError, http.client.HTTPException):
pass # The API restart does not stop independent ingress traffic.
except Exception as error:
if len(metrics["monitorErrors"]) < 10:
metrics["monitorErrors"].append(str(error))
stop.wait(.1)
def cleanup(service):
if current(service) is None:
api.call("DELETE", path(service))
return
record = wait_for(lambda: (r if (r := current(service)) and r["phase"] in
["running", "failed", "degraded", "stopped", "cancelled", "queued"] else None),
timeout=180)
if record["phase"] == "queued":
cancel(service, record["id"])
record = current(service)
if record["phase"] != "cancelled":
api.call("POST", path(service, "/stop"))
wait_for(lambda: current(service)["phase"] == "stopped", timeout=180)
api.call("DELETE", path(service))
try:
active, normal, rollback = create(), create(), create()
previous = rollback["initial"]
update(rollback, previous)
rollback_b = wait_for(lambda: ready(rollback), timeout=180)
for service in [active, normal, rollback]:
thread = threading.Thread(target=load, args=(service,), daemon=True)
threads.append(thread)
thread.start()
monitor = threading.Thread(target=observe, daemon=True)
threads.append(monitor)
monitor.start()
active_ip, active_hold = hold(active)
running = update(active, active["initial"])
wait_for(lambda: current(active)["phase"] == "draining", timeout=180)
# Nginx drains a whole worker generation. Start the second hold on the
# new workers so it cannot block completion of the active transition.
rollback_ip, rollback_hold = hold(rollback)
assert not active_hold.done() and not rollback_hold.done()
initial = create(deploy=False)
queued_initial = api.call("POST", path(initial, "/deployment"), {"microNetworkId": network})
assert queued_initial["phase"] == "queued"
cancel(initial, queued_initial["id"])
cancel(initial, queued_initial["id"])
assert not current(initial)["instances"] and not routes(initial)
before = current(normal)
original_routes = routes(normal)
queued = update(normal, before)
assert queued["phase"] == "queued"
cancel(normal, queued["id"])
cancel(normal, queued["id"])
retained = current(normal)
assert all(retained[key] == before[key] for key in ["id", "spec", "digest", "request"])
assert {vm["vmId"] for vm in retained["instances"]} == {vm["vmId"] for vm in before["instances"]}
assert routes(normal) == original_routes
metrics["cancelledInitial"] = queued_initial["id"]
metrics["cancelledUpdate"] = queued["id"]
metrics["stages"].append("queued initial and update cancellation was idempotent; active release and VMs retained")
next_normal = update(normal, before)
next_rollback = api.call("POST", path(rollback, "/rollback"), {
"operationId": str(uuid.uuid4()), "expectedReleaseId": rollback_b["id"],
"releaseId": previous["id"], "strategy": "rolling"})
assert next_normal["phase"] == next_rollback["phase"] == "queued"
metrics["normalOperation"] = next_normal["id"]
metrics["rollbackOperation"] = next_rollback["id"]
ingress_pid = api.call("GET", "/api/pipe")["units"][0]["mainPid"]
subprocess.run(["systemctl", "restart", "firecrab-api.service"], check=True)
api.ready()
assert api.call("GET", "/api/pipe")["units"][0]["mainPid"] == ingress_pid
assert not active_hold.done() and not rollback_hold.done()
assert operation(normal, next_normal["id"])["phase"] == "queued"
assert operation(rollback, next_rollback["id"])["phase"] == "queued"
assert operation(initial, queued_initial["id"])["phase"] == "cancelled"
assert operation(normal, queued["id"])["phase"] == "cancelled"
release(active_ip, active_hold, "A")
def rollback_draining():
assert operation(normal, next_normal["id"])["phase"] == "queued", "ordinary Deploy overtook rollback"
record = operation(rollback, next_rollback["id"])
assert record["phase"] not in ["failed", "cancelled", "stopped"], record
return record if record["phase"] == "draining" else None
wait_for(rollback_draining, timeout=180)
assert not rollback_hold.done()
assert current(active)["id"] == running["id"] and current(active)["phase"] == "running"
metrics["stages"].append("API restart resumed active drain first; rollback overtook the earlier ordinary Deploy")
release(rollback_ip, rollback_hold, "B")
wait_for(lambda: ready(rollback), timeout=180)
wait_for(lambda: ready(normal), timeout=180)
assert current(rollback)["digest"] == previous["digest"]
assert current(normal)["digest"] == digest_b
assert operation(normal, queued["id"])["phase"] == "cancelled"
assert not metrics["httpFailures"] and not metrics["monitorErrors"], metrics
assert min(metrics["requests"].values()) > 10, metrics
metrics["stages"].append("waiting Deploy completed after rollback; cancelled work did not resurrect")
finally:
stop.set()
for thread in threads:
thread.join(timeout=10)
errors = ["traffic monitor did not exit" for thread in threads if thread.is_alive()]
for address in held_ips:
try:
traffic(address, 8080, "fixture", "/cgi-bin/release")
except Exception as error:
errors.append("release held request: " + str(error))
executor.shutdown(wait=True, cancel_futures=True)
for service in reversed(services):
try:
cleanup(service)
except Exception as error:
errors.append(str(error))
metrics["cleanupErrors"] = errors
metrics["elapsedSeconds"] = round(time.monotonic() - started, 2)
assert not errors, errors
assert not metrics["httpFailures"] and not metrics["monitorErrors"], metrics
print("PASS queued cancellation, rollback priority, API restart and retained HTTP traffic", flush=True)