scripts/test-deploy-nginx.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Check Deploy's generated data plane with real Nginx and local HTTP apps.
This verifies reload/drain behavior. It does not replace MicroVM/KVM acceptance.
"""
import argparse
import base64
import concurrent.futures
import contextlib
import hashlib
import http.client
import http.server
import json
from pathlib import Path
import re
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
def receive(stream, count):
result = b""
while len(result) < count:
part = stream.recv(count - len(result))
if not part:
raise EOFError("connection closed during drain")
result += part
return result
class App:
def __init__(self, name):
self.name = name
self.started = threading.Event()
self.release = threading.Event()
app = self
class Handler(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_GET(self):
if self.path == "/ws":
key = self.headers["Sec-WebSocket-Key"]
accept = base64.b64encode(hashlib.sha1(
(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()
).digest()).decode()
self.send_response(101)
self.send_header("Upgrade", "websocket")
self.send_header("Connection", "Upgrade")
self.send_header("Sec-WebSocket-Accept", accept)
self.end_headers()
try:
while True:
header = receive(self.connection, 2)
if header[0] & 15 == 8:
break
size = header[1] & 127
assert size < 126 and header[1] & 128
mask = receive(self.connection, 4)
payload = receive(self.connection, size)
decoded = bytes(value ^ mask[i % 4] for i, value in enumerate(payload))
reply = app.name.encode() + b":" + decoded
self.connection.sendall(bytes([0x81, len(reply)]) + reply)
except (OSError, EOFError):
pass
self.close_connection = True
return
if self.path == "/held":
app.started.set()
assert app.release.wait(20), "held request was not released"
body = (app.name + ":" + self.path).encode()
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
@property
def address(self):
return f"127.0.0.1:{self.server.server_port}"
def close(self):
self.release.set()
self.server.shutdown()
self.server.server_close()
self.thread.join()
def request(port, host="deploy.example.com", path="/"):
with contextlib.closing(http.client.HTTPConnection("127.0.0.1", port, timeout=10)) as conn:
conn.request("GET", path, headers={"Host": host})
response = conn.getresponse()
return response.status, response.read().decode()
def websocket(port):
stream = socket.create_connection(("127.0.0.1", port), timeout=10)
stream.sendall(b"GET /ws HTTP/1.1\r\nHost: deploy.example.com\r\n"
b"Connection: Upgrade\r\nUpgrade: websocket\r\n"
b"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
b"Sec-WebSocket-Version: 13\r\n\r\n")
header = b""
while not header.endswith(b"\r\n\r\n"):
header += receive(stream, 1)
assert header.startswith(b"HTTP/1.1 101"), header
return stream
def echo(stream, text):
payload = text.encode()
mask = b"test"
stream.sendall(bytes([0x81, 128 | len(payload)]) + mask
+ bytes(value ^ mask[i % 4] for i, value in enumerate(payload)))
header = receive(stream, 2)
assert header[0] == 0x81 and header[1] < 126, header
return receive(stream, header[1]).decode()
def run(firecrab, nginx):
with contextlib.ExitStack() as stack:
directory = Path(stack.enter_context(tempfile.TemporaryDirectory(prefix="fc-deploy-", dir="/tmp")))
old, new, other = [App(name) for name in ["old", "new", "other"]]
for app in [old, new, other]:
stack.callback(app.close)
with socket.socket() as reserved:
reserved.bind(("127.0.0.1", 0))
port = reserved.getsockname()[1]
manifest = {"listen": f"127.0.0.1:{port}", "routes": [
{"host": "deploy.example.com", "upstreams": [old.address]},
{"host": "other.example.com", "upstreams": [other.address]}]}
config = directory / "nginx.conf"
manifest_file = directory / "pipe.json"
status_socket = directory / "status.sock"
def render(upstreams):
manifest["routes"][0]["upstreams"] = upstreams
manifest_file.write_text(json.dumps(manifest))
text = subprocess.check_output([firecrab, "pipe", "render", "--config", str(manifest_file)], text=True)
assert "worker_shutdown_timeout 0;" in text
if sys.platform == "darwin":
# Native ARM macOS uses 16 KiB pages. Linux production uses 4 KiB.
text = re.sub(r"(zone fc_route_\d+) 64k;", r"\1 256k;", text)
config.write_text(text.replace("/run/firecrab-ingress/status.sock", str(status_socket)))
return re.search(r'return 200 "([a-f0-9]+)"', text)[1]
def check():
return subprocess.run([nginx, "-t", "-p", str(directory) + "/", "-c", str(config)], capture_output=True)
def wait_generation(expected):
for _ in range(100):
try:
with socket.socket(socket.AF_UNIX) as conn:
conn.settimeout(1)
conn.connect(str(status_socket))
conn.sendall(b"GET / HTTP/1.0\r\nHost: localhost\r\n\r\n")
response = b""
while part := conn.recv(4096):
response += part
if response.partition(b"\r\n\r\n")[2].decode() == expected:
return
except OSError:
pass
time.sleep(.05)
raise AssertionError("Nginx did not acknowledge the route generation")
generation = render([old.address])
result = check()
assert result.returncode == 0, result.stderr
log = stack.enter_context((directory / "nginx.log").open("w+"))
process = subprocess.Popen([nginx, "-p", str(directory) + "/", "-c", str(config), "-g", "daemon off;"], stdout=log, stderr=log)
def stop():
if process.poll() is None:
process.send_signal(signal.SIGQUIT)
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
stack.callback(stop)
wait_generation(generation)
assert request(port) == (200, "old:/")
connection = stack.enter_context(websocket(port))
assert echo(connection, "before") == "old:before"
errors, samples = [], []
done = threading.Event()
def load():
while not done.is_set():
try:
response = request(port)
samples.append(response)
if response not in [(200, "old:/"), (200, "new:/")]:
errors.append(response)
except Exception as error:
errors.append(str(error))
time.sleep(.005)
worker = threading.Thread(target=load, daemon=True)
worker.start()
try:
with concurrent.futures.ThreadPoolExecutor() as executor:
held = executor.submit(request, port, "deploy.example.com", "/held")
assert old.started.wait(5)
for upstreams, expected in [([old.address, new.address], {"old:/", "new:/"}),
([new.address], {"new:/"}),
([old.address], {"old:/"})]:
generation = render(upstreams)
result = check()
assert result.returncode == 0, result.stderr
process.send_signal(signal.SIGHUP)
wait_generation(generation)
# Generation acknowledgement can precede old workers closing
# their listening sockets. Deploy also waits for worker drain.
deadline = time.monotonic() + 5
observed = set()
while time.monotonic() < deadline:
observed = {request(port)[1] for _ in range(40)}
if observed == expected:
break
time.sleep(.05)
assert observed == expected, (upstreams, observed)
assert request(port, "other.example.com") == (200, "other:/")
assert echo(connection, "after") == "old:after"
assert process.poll() is None
# An invalid candidate never replaces the saved, serving config.
valid = config.read_text()
config.write_text(valid + "invalid_deploy_directive;\n")
assert check().returncode != 0
config.write_text(valid)
assert request(port) == (200, "old:/")
old.release.set()
assert held.result(timeout=10) == (200, "old:/held")
finally:
old.release.set()
done.set()
worker.join(timeout=10)
assert len(samples) >= 10, len(samples)
assert not errors, errors
print(f"PASS Deploy rolling switch and rollback: {len(samples)} continuous requests, zero errors")
print("PASS held HTTP and WebSocket survive reloads; unrelated route and invalid-config recovery remain intact")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--firecrab", required=True)
parser.add_argument("--nginx", required=True)
args = parser.parse_args()
run(str(Path(args.firecrab).resolve()), str(Path(args.nginx).resolve()))