diff --git a/controller/docker_socket_proxy.go b/controller/docker_socket_proxy.go index c46fd895..a4a0d9ad 100644 --- a/controller/docker_socket_proxy.go +++ b/controller/docker_socket_proxy.go @@ -22,6 +22,7 @@ import ( "time" "github.com/google/uuid" + "golang.org/x/sys/unix" ) const dockerProxySocket = "/run/ci-fleet/locks/docker.sock" @@ -30,6 +31,7 @@ const dockerProxyHealth = "ci-fleet-docker-socket-proxy-v1\n" const dockerBuildContextLimit = 1 << 30 const dockerBuildUploadLimit = 2 const dockerMetadataBodyLimit = 1 << 20 +const dockerSpoolCacheWindow = 4 << 20 var dockerAPIVersion = regexp.MustCompile(`^/v[0-9]+\.[0-9]+/`) @@ -308,15 +310,79 @@ func (body *dockerProxyBody) Close() error { } type dockerBuildSpool struct { - *os.File - closed sync.Once - slots chan struct{} - err error + file *os.File + offset int64 + dropped int64 + closed sync.Once + slots chan struct{} + err error +} + +func (spool *dockerBuildSpool) requireDisk() error { + var filesystem unix.Statfs_t + if err := unix.Fstatfs(int(spool.file.Fd()), &filesystem); err != nil { + return err + } + if filesystem.Type == unix.TMPFS_MAGIC || filesystem.Type == unix.RAMFS_MAGIC { + return errors.New("Docker upload spool requires disk-backed temporary storage") + } + return nil +} + +func (spool *dockerBuildSpool) Write(buffer []byte) (int, error) { + count, err := spool.file.Write(buffer) + spool.offset += int64(count) + if err == nil && spool.offset-spool.dropped >= dockerSpoolCacheWindow { + err = spool.releaseWritten() + } + return count, err +} + +func (spool *dockerBuildSpool) releaseWritten() error { + // Dirty and reread file pages count against the proxy's memory cgroup too. + if err := unix.Fdatasync(int(spool.file.Fd())); err != nil { + return err + } + return spool.releaseCache() +} + +func (spool *dockerBuildSpool) releaseCache() error { + length := spool.offset - spool.dropped + if length == 0 { + return nil + } + if err := unix.Fadvise(int(spool.file.Fd()), spool.dropped, length, unix.FADV_DONTNEED); err != nil { + return err + } + spool.dropped = spool.offset + return nil +} + +func (spool *dockerBuildSpool) rewind() error { + if err := spool.releaseWritten(); err != nil { + return err + } + if _, err := spool.file.Seek(0, io.SeekStart); err != nil { + return err + } + spool.offset, spool.dropped = 0, 0 + return nil +} + +func (spool *dockerBuildSpool) Read(buffer []byte) (int, error) { + count, err := spool.file.Read(buffer) + spool.offset += int64(count) + if spool.offset-spool.dropped >= dockerSpoolCacheWindow || errors.Is(err, io.EOF) { + if cacheErr := spool.releaseCache(); cacheErr != nil { + err = errors.Join(err, cacheErr) + } + } + return count, err } func (spool *dockerBuildSpool) Close() error { spool.closed.Do(func() { - spool.err = spool.File.Close() + spool.err = spool.file.Close() <-spool.slots }) return spool.err @@ -375,14 +441,18 @@ func newDockerSocketProxy(upstream, lockPath, bootID string) (http.Handler, *htt http.Error(w, err.Error(), http.StatusServiceUnavailable) return } - spool := &dockerBuildSpool{File: file, slots: buildSlots} + spool := &dockerBuildSpool{file: file, slots: buildSlots} defer spool.Close() // Unlink before writing so a killed proxy cannot leave spooled contexts behind. if err := os.Remove(file.Name()); err != nil { http.Error(w, err.Error(), http.StatusServiceUnavailable) return } - length, err := io.Copy(file, http.MaxBytesReader(w, request.Body, dockerBuildContextLimit)) + if err := spool.requireDisk(); err != nil { + http.Error(w, err.Error(), http.StatusServiceUnavailable) + return + } + length, err := io.Copy(spool, http.MaxBytesReader(w, request.Body, dockerBuildContextLimit)) if err != nil { var tooLarge *http.MaxBytesError if errors.As(err, &tooLarge) { @@ -392,7 +462,7 @@ func newDockerSocketProxy(upstream, lockPath, bootID string) (http.Handler, *htt http.Error(w, err.Error(), http.StatusBadRequest) return } - if _, err := file.Seek(0, io.SeekStart); err != nil { + if err := spool.rewind(); err != nil { http.Error(w, err.Error(), http.StatusServiceUnavailable) return } diff --git a/controller/docker_socket_proxy_spool_test.go b/controller/docker_socket_proxy_spool_test.go new file mode 100644 index 00000000..fadc0abd --- /dev/null +++ b/controller/docker_socket_proxy_spool_test.go @@ -0,0 +1,110 @@ +package main + +import ( + "bytes" + "crypto/sha256" + "io" + "net/http" + "os" + "strings" + "testing" +) + +func TestDockerBuildSpoolRetainsBytesAcrossWriteAndReadCacheWindows(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "spool-") + if err != nil { + t.Fatal(err) + } + slots := make(chan struct{}, 1) + slots <- struct{}{} + spool := &dockerBuildSpool{file: file, slots: slots} + defer spool.Close() + if err := spool.requireDisk(); err != nil { + t.Fatal(err) + } + input := bytes.Repeat([]byte("exact-upload-content\n"), dockerSpoolCacheWindow/10+17) + want := sha256.Sum256(input) + if count, err := io.Copy(spool, bytes.NewReader(input)); err != nil || count != int64(len(input)) { + t.Fatalf("spooling lost bytes: count=%d, error=%v", count, err) + } + if err := spool.rewind(); err != nil { + t.Fatal(err) + } + actual := sha256.New() + if count, err := io.Copy(actual, spool); err != nil || count != int64(len(input)) { + t.Fatalf("forwarding lost bytes: count=%d, error=%v", count, err) + } + if !bytes.Equal(actual.Sum(nil), want[:]) { + t.Fatal("cache release changed the forwarded payload") + } + if err := spool.Close(); err != nil { + t.Fatal(err) + } + if err := spool.Close(); err != nil || len(slots) != 0 { + t.Fatalf("closing must release exactly one upload slot: %v", err) + } +} + +func TestDockerBuildSpoolRefusesFailedCachePreparation(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "closed-spool-") + if err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + spool := &dockerBuildSpool{file: file} + if err := spool.requireDisk(); err == nil { + t.Fatal("unreadable filesystem identity was accepted") + } + if err := spool.rewind(); err == nil { + t.Fatal("failed file synchronization must not reach forwarding") + } + if _, err := spool.Write([]byte("upload")); err == nil { + t.Fatal("failed spool write was accepted") + } + if _, err := spool.Read(make([]byte, 8)); err == nil { + t.Fatal("failed spool read was accepted") + } +} + +func TestDockerBuildSpoolRefusesMemoryBackedTemporaryFiles(t *testing.T) { + file, err := os.CreateTemp("/dev/shm", "proxy-memory-spool-") + if err != nil { + t.Fatal(err) + } + defer os.Remove(file.Name()) + defer file.Close() + spool := &dockerBuildSpool{file: file} + if err := spool.requireDisk(); err == nil { + t.Fatal("memory-backed spooling cannot fit a full upload within the proxy memory limit") + } +} + +func TestDockerSocketProxyRefusesMemoryBackedSpoolBeforeDispatch(t *testing.T) { + directory, err := os.MkdirTemp("/dev/shm", "proxy-spool-refusal-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(directory) + dispatched := make(chan struct{}, 1) + fixture := newSocketProxyTest(t, http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + dispatched <- struct{}{} + w.WriteHeader(http.StatusOK) + })) + t.Setenv("TMPDIR", directory) + response, err := fixture.client.Post("http://docker/images/load", "application/x-tar", strings.NewReader("fixture")) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || response.StatusCode != http.StatusServiceUnavailable || !strings.Contains(string(body), "disk-backed") { + t.Fatalf("memory-backed spool was not refused: %d %s, %v", response.StatusCode, body, err) + } + select { + case <-dispatched: + t.Fatal("memory-backed spool reached Docker") + default: + } +} diff --git a/controller/go.mod b/controller/go.mod index cc062a9d..912d94e7 100644 --- a/controller/go.mod +++ b/controller/go.mod @@ -7,6 +7,7 @@ require ( github.com/docker/docker v28.5.2+incompatible github.com/google/uuid v1.6.0 github.com/moby/docker-image-spec v1.3.1 + golang.org/x/sys v0.49.0 ) require ( @@ -35,5 +36,4 @@ require ( go.opentelemetry.io/otel/metric v1.47.0 // indirect go.opentelemetry.io/otel/trace v1.47.0 // indirect go.opentelemetry.io/proto/otlp v1.11.0 // indirect - golang.org/x/sys v0.49.0 // indirect ) diff --git a/docs/HOST-MAINTENANCE.md b/docs/HOST-MAINTENANCE.md index 9c18b34e..b77e046d 100644 --- a/docs/HOST-MAINTENANCE.md +++ b/docs/HOST-MAINTENANCE.md @@ -129,6 +129,17 @@ image removal, and builder pruning participate in proxy protocol `v2` alongside builds and container reference mutations. Storage removal defers with older proxies. Image-load uploads use the same 1 GiB spool limit as build contexts. Image imports use that bounded spool too and share its two upload slots. +Upload spools synchronize written file data and discard its cached pages in +4 MiB windows. Forwarding also discards consumed pages. File-cache charges +belong to the proxy's memory cgroup, so an unlinked disk file alone does not +keep a large upload within its unchanged 128 MiB memory limit. Synchronization +or cache-advice failures during spooling refuse the request before forwarding. +A cache-advice error during forwarding retains the existing uncertainty guard. +The body limit and two upload slots are unchanged. +The temporary directory must have disk backing. Memory-backed `tmpfs` or `ramfs` +spools are refused before forwarding because their data cannot be evicted from +the memory limit as a disk page can. Do not move spools to a memory filesystem +to avoid synchronization costs. Registry pulls and imports from remote URLs bypass upload slots while retaining the maintenance lock. Other guarded request bodies have a 1 MiB metadata limit; oversized bodies receive HTTP 413 before reaching Docker. diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh index b2cf7d50..35e4da2c 100755 --- a/scripts/install-worker-controller.sh +++ b/scripts/install-worker-controller.sh @@ -824,7 +824,7 @@ PY } runtime_release_complete() { - local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required stored_digest actual_digest policy_script + local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required stored_digest actual_digest policy_script enabled_capabilities local -a capability_args=() [[ -d "$path" && -f "$path/.ci-fleet-engine-ref" && -f "$path/.ci-fleet-tree-sha256" && -f "$path/deploy/compose.yaml" ]] || return 1 [[ -x "$path/scripts/preflight.sh" && -x "$path/scripts/healthcheck.sh" && -x "$path/scripts/cleanup.sh" ]] || return 1 @@ -834,14 +834,10 @@ runtime_release_complete() { [[ "$require_status" != 1 ]] || capability_args+=(--require-status-reporting) python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities \ --manifest "$path/engine-capabilities.json" "${capability_args[@]}" >/dev/null || return 1 + enabled_capabilities=$(jq -c '.capabilities | to_entries[] | select(.value == true) | .key' "$path/engine-capabilities.json") || return 1 for required in docker_network_policy_config:apply-docker-network-policy.sh docker_network_policy_adapter:docker-network-policy-adapter.sh reproducible_managed_images:build-managed-images.sh portable_managed_image_ids:managed_image_identity.py apt_cache:apt_cache.py; do policy_script=${required#*:} - if python3 - "$path/engine-capabilities.json" "${required%%:*}" <<'PY' -import json, sys -value = json.load(open(sys.argv[1], encoding="utf-8")) -raise SystemExit(value.get("capabilities", {}).get(sys.argv[2]) is not True) -PY - then + if grep -Fxq -- "\"${required%%:*}\"" <<<"$enabled_capabilities"; then [[ -f "$path/scripts/$policy_script" && ! -L "$path/scripts/$policy_script" ]] || return 1 [[ "$policy_script" == *.py || -x "$path/scripts/$policy_script" ]] || return 1 fi diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh index 0c7eaa4b..276adce3 100755 --- a/scripts/test-install-worker-controller.sh +++ b/scripts/test-install-worker-controller.sh @@ -1930,7 +1930,7 @@ retry_result=$tmp/interrupted-retry-result.json : >"$retry_result" FAKE_PAUSE_SYSTEMCTL_ONCE=$retry_pause CI_FLEET_TRANSACTION_RESULT_FD=7 "$installer" --upgrade "${base_args[@]}" --ref "$interrupted_ref" 7>"$retry_result" >"$tmp/interrupted-retry.out" 2>&1 & retry_pid=$! -wait_for_file "$retry_pause.entered" 'interrupted policy retry did not reach nested activation' +(wait_for_file "$retry_pause.entered" 'interrupted policy retry did not reach nested activation') || fail "interrupted policy retry did not reach nested activation: $(<"$tmp/interrupted-retry.out")" for repaired_checkpoint_target in "$stale_checkpoint_release" "$stale_checkpoint_manager"; do if find "$repaired_checkpoint_target" \( -type d -o -type f \) -perm /022 -print -quit | grep -q .; then fail "pending checkpoint target remained permission-unsafe: $repaired_checkpoint_target" diff --git a/scripts/test_local_validation.py b/scripts/test_local_validation.py index d5638303..e99e1c32 100644 --- a/scripts/test_local_validation.py +++ b/scripts/test_local_validation.py @@ -452,12 +452,44 @@ def test_successful_leader_allows_descendant_cleanup(self): run(["python3", "-c", leader], root, os.environ.copy(), root / "test.log", 5) self.assertEqual((root / "cleaned").read_text(), "done") + def test_proxy_memory_timeout_unwinds_owned_fixture_through_shell(self): + for logs_status in (0, 1): + with self.subTest(logs_status=logs_status), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + docker = root / "docker" + docker.write_text('''#!/usr/bin/python3 +import json, os, sys, time +from pathlib import Path +args = sys.argv[1:] +with Path(os.environ["PROXY_COMMANDS"]).open("a") as log: + log.write(json.dumps(args) + "\\n") +if args[:2] == ["run", "-d"]: + print("fixture-proxy") +if args[:2] == ["inspect", "--format"]: + time.sleep(60) +if args[:1] == ["logs"]: + sys.exit(int(os.environ["PROXY_LOGS_STATUS"])) +''') + docker.chmod(0o755) + commands = root / "commands.jsonl" + env = os.environ | {"PATH": str(root) + ":" + os.environ["PATH"], + "PROXY_COMMANDS": str(commands), "PROXY_LOGS_STATUS": str(logs_status)} + command = ["bash", "-c", 'python3 "$1" --image fixture-image; exit "$?"', + "bash", str(ROOT / "scripts/test_proxy_memory.py")] + with self.assertRaises(subprocess.TimeoutExpired): + run(command, ROOT, env, root / "fixture.log", 2) + recorded = [json.loads(line) for line in commands.read_text().splitlines()] + self.assertTrue(any(args[:2] == ["run", "-d"] for args in recorded)) + self.assertIn(["logs", "fixture-proxy"], recorded) + self.assertIn(["rm", "-f", "fixture-proxy"], recorded) + def test_container_stage_propagates_cleanup_failure(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) scripts = root / "scripts"; scripts.mkdir() (scripts / "test_apt_cache_e2e.py").write_text( "import os, sys\nsys.exit(int(os.environ['TEST_STAGE_STATUS']))\n") + (scripts / "test_proxy_memory.py").write_text("") (scripts / "scan_committed_secrets.py").write_text("") for name in ("test-cleanup-e2e.sh", "cleanup.sh"): path = scripts / name diff --git a/scripts/test_portable_image_lifecycle.py b/scripts/test_portable_image_lifecycle.py index f79f0a56..55907cb7 100644 --- a/scripts/test_portable_image_lifecycle.py +++ b/scripts/test_portable_image_lifecycle.py @@ -60,6 +60,57 @@ def shell(self, names, body, *, portable=True): result = subprocess.run(["bash", "-c", script], capture_output=True, text=True, env=os.environ | {"PYTHONDONTWRITEBYTECODE": "1"}) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + def test_runtime_capability_helpers_fail_closed(self): + self.shell(("release_tree_digest", "runtime_release_complete"), r""" +mkdir -p "$release_dir/"{deploy,scripts,controller,runner} +cp "$repo_root/engine-capabilities.json" "$release_dir/engine-capabilities.json" +for required in deploy/compose.yaml controller/Dockerfile controller/go.mod controller/main.go controller/config.go controller/scaler.go controller/state.go runner/Dockerfile; do + : >"$release_dir/$required" +done +helpers=(apply-docker-network-policy.sh docker-network-policy-adapter.sh build-managed-images.sh managed_image_identity.py apt_cache.py) +for helper in preflight.sh healthcheck.sh cleanup.sh "${helpers[@]}"; do + printf '#!/bin/sh\n' >"$release_dir/scripts/$helper" + chmod 755 "$release_dir/scripts/$helper" +done +refresh() { release_tree_digest "$release_dir" >"$release_dir/.ci-fleet-tree-sha256"; } +refresh +runtime_release_complete "$release_dir" "$engine_ref" +for helper in "${helpers[@]}"; do + target=$release_dir/scripts/$helper + mv "$target" "$temporary/helper" + refresh + if runtime_release_complete "$release_dir" "$engine_ref"; then exit 1; fi + ln -s "$temporary/helper" "$target" + refresh + if runtime_release_complete "$release_dir" "$engine_ref"; then exit 1; fi + rm "$target" + mv "$temporary/helper" "$target" + chmod 644 "$target" + refresh + if [[ $helper == *.py ]]; then + runtime_release_complete "$release_dir" "$engine_ref" + elif runtime_release_complete "$release_dir" "$engine_ref"; then exit 1; fi + chmod 755 "$target" +done +refresh +runtime_release_complete "$release_dir" "$engine_ref" +python3 - "$release_dir/engine-capabilities.json" <<'DISABLED' +import json, sys +from pathlib import Path +p = Path(sys.argv[1]) +value = json.loads(p.read_text()) +for key in ("docker_network_policy_config", "docker_network_policy_adapter", "reproducible_managed_images", "portable_managed_image_ids", "apt_cache"): + value["capabilities"][key] = False + value["capabilities"]["unknown\n" + key] = True +p.write_text(json.dumps(value)) +DISABLED +for helper in "${helpers[@]}"; do rm "$release_dir/scripts/$helper"; done +refresh +runtime_release_complete "$release_dir" "$engine_ref" +jq() { return 127; } +if runtime_release_complete "$release_dir" "$engine_ref"; then exit 1; fi +""") + def test_runtime_source_selects_semantics_not_scalar_syntax(self): for portable in (False, True): self.shell(("portable_managed_images",), diff --git a/scripts/test_proxy_memory.py b/scripts/test_proxy_memory.py new file mode 100644 index 00000000..9ddbfec0 --- /dev/null +++ b/scripts/test_proxy_memory.py @@ -0,0 +1,185 @@ +"""Verify large concurrent uploads under the unchanged proxy cgroup envelope. + +Run as root only on a disposable Docker/systemd/cgroup-v2 validation worker. +The backend drains fixture bytes, never loads images or runs application code. +""" +import argparse +import datetime +import hashlib +import http.server +import importlib.util +import json +import os +import pathlib +import signal +import socket +import socketserver +import subprocess +import threading +import tempfile +import time + +parser = argparse.ArgumentParser() +parser.add_argument('--output') +parser.add_argument('--image', required=True) +args = parser.parse_args() +assert os.geteuid() == 0 and pathlib.Path('/sys/fs/cgroup/cgroup.controllers').is_file(), 'disposable root cgroup-v2 worker required' +temporary = tempfile.TemporaryDirectory(prefix='proxy-memory-', dir='/var/tmp') if not args.output else None +root = pathlib.Path(args.output or temporary.name) +if args.output: + root.mkdir(mode=0o700) +locks = root / 'locks' +locks.mkdir(mode=0o755) +upstream = root / 'upstream.sock' +name = 'ci-fleet-proxy-memory-' + root.name +image = args.image +upload_size = 1 << 30 +events = [] +stop = threading.Event() +samples = [] +clients = [] +phase = {'name': 'start'} + +def event(kind, **values): + events.append(dict(at=datetime.datetime.now(datetime.timezone.utc).isoformat(), monotonic=time.monotonic(), event=kind, **values)) + +class Backend(http.server.BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + def log_message(self, *unused): + pass + def do_GET(self): + self.send_response(200) + self.send_header('Content-Length', '2') + self.end_headers() + self.wfile.write(b'OK') + def do_POST(self): + total = int(self.headers['Content-Length']) + event('forward_started', length=total) + phase['name'] = 'forwarding' + digest = hashlib.sha256() + count = 0 + while count < total: + chunk = self.rfile.read(min(256 * 1024, total - count)) + if not chunk: + break + count += len(chunk) + digest.update(chunk) + time.sleep(0.001) + event('forward_complete', bytes=count, sha256=digest.hexdigest()) + body = b'{"stream":"loaded fixture"}\n' + self.send_response(200 if count == total else 400) + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + +class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): + daemon_threads = True + +server = Server(str(upstream), Backend) +threading.Thread(target=server.serve_forever, daemon=True).start() + +def docker(*words): + return subprocess.check_output(['docker', *words], text=True, timeout=30).strip() + +def upload(index): + try: + data = bytes(256 * 1024) + digest = hashlib.sha256() + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(120) + sock.connect(str(locks / 'docker.sock')) + sock.sendall(('POST /v1.55/images/load?quiet=1 HTTP/1.1\r\nHost: docker\r\nContent-Type: application/x-tar\r\nContent-Length: ' + str(upload_size) + '\r\nConnection: close\r\n\r\n').encode()) + count = 0 + while count < upload_size: + part = data[:min(len(data), upload_size - count)] + sock.sendall(part) + count += len(part) + digest.update(part) + event('upload_sent', client=index, bytes=count, sha256=digest.hexdigest()) + response = bytearray() + while chunk := sock.recv(8192): + response.extend(chunk) + event('client_complete', client=index, response=response.decode(errors='replace')[:1024]) + except Exception as error: + event('client_failed', client=index, error=type(error).__name__ + ': ' + str(error)) + +def terminate(signum, _frame): + raise SystemExit(128 + signum) + + +signal.signal(signal.SIGTERM, terminate) +container = None +try: + cmd = ['run', '-d', '--name', name, '--memory=128m', '--cpus=1', '--user=0:0', '--network=none', '--security-opt=no-new-privileges', + '--health-cmd', '/usr/local/bin/ci-fleet-controller --check-docker-socket-proxy', + '--health-interval=5s', '--health-timeout=5s', '--health-retries=6', + '--mount', 'type=bind,src=' + str(upstream) + ',dst=/var/run/docker.sock', + '--mount', 'type=bind,src=' + str(locks) + ',dst=/run/ci-fleet/locks'] + cmd += [image, '--docker-socket-proxy'] + container = docker(*cmd) + initial = json.loads(docker('inspect', '--format', '{"pid":{{.State.Pid}},"image":{{json .Image}},"memory":{{.HostConfig.Memory}}}', container)) + cgroup_line = next(line for line in pathlib.Path('/proc', str(initial['pid']), 'cgroup').read_text().splitlines() if line.startswith('0::')) + group = pathlib.Path('/sys/fs/cgroup') / cgroup_line[3:].lstrip('/') + event('started', container=container, image=initial['image'], memory_limit=initial['memory'], cgroup=str(group), size_per_upload=upload_size) + def monitor(): + while not stop.wait(0.01): + try: + values = {key: int(value) for key, value in (line.split() for line in (group / 'memory.stat').read_text().splitlines())} + sample = {'at':time.monotonic(), 'phase':phase['name'], 'current':int((group / 'memory.current').read_text()), 'peak':int((group / 'memory.peak').read_text()), + 'events':dict(line.split() for line in (group / 'memory.events').read_text().splitlines())} + sample.update({key:values[key] for key in ['anon','file','kernel','sock','shmem','file_dirty','file_writeback','active_file','inactive_file'] if key in values}) + samples.append(sample) + except FileNotFoundError: + return + monitor_thread = threading.Thread(target=monitor) + monitor_thread.start() + limit = time.monotonic() + 20 + while not (locks / 'docker.sock').exists() and time.monotonic() < limit: + time.sleep(0.02) + assert (locks / 'docker.sock').exists(), 'fixture socket did not appear' + phase['name'] = 'spooling' + threads = [threading.Thread(target=upload, args=(index,)) for index in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(150) + assert not thread.is_alive(), 'fixture upload deadline exceeded' + final = json.loads(docker('inspect', '--format', '{"state":{{json .State}},"restart_count":{{.RestartCount}}}', container)) + event('container_result', state=final) +finally: + stop.set() + if 'monitor_thread' in locals(): + monitor_thread.join(3) + server.shutdown() + server.server_close() + if container: + try: + (root / 'proxy.log').write_text(docker('logs', container)) + finally: + docker('rm', '-f', container) + upstream.unlink(missing_ok=True) + (root / 'events.json').write_text(json.dumps(events, indent=2) + '\n') + (root / 'memory.json').write_text(json.dumps(samples) + '\n') + summary = {'at':datetime.datetime.now(datetime.timezone.utc).isoformat(), 'image':image, + 'uploads':len([e for e in events if e['event'] == 'client_complete']), + 'failures':len([e for e in events if e['event'] == 'client_failed']), + 'memory_peak':max((s['peak'] for s in samples), default=None), + 'file_peak':max((s.get('file', 0) for s in samples), default=None), + 'file_dirty_peak':max((s.get('file_dirty', 0) for s in samples), default=None), + 'anon_peak':max((s.get('anon', 0) for s in samples), default=None), + 'oom_kill':max((int(s['events'].get('oom_kill', '0')) for s in samples), default=None), + 'container_oom_killed':final['state']['OOMKilled'] if 'final' in locals() else None, + 'fixture_container_removed':True} + (root / 'summary.json').write_text(json.dumps(summary, indent=2) + '\n') + print(json.dumps(summary, indent=2)) + +assert summary['uploads'] == 2 and summary['failures'] == 0, 'concurrent full-size upload failed' +assert summary['oom_kill'] == 0 and summary['container_oom_killed'] is False and summary['memory_peak'] < 96 * 1024**2, 'upload exceeded proxy memory headroom' +completed = [e for e in events if e['event'] == 'forward_complete'] +sent = [e for e in events if e['event'] == 'upload_sent'] +assert len(completed) == len(sent) == 2 +assert all(e['bytes'] == upload_size and e['sha256'] == sent[0]['sha256'] for e in completed), 'forwarding changed upload bytes' +assert all(' 200 ' in e['response'] for e in events if e['event'] == 'client_complete') +assert any(s['phase'] == 'spooling' for s in samples) and any(s['phase'] == 'forwarding' for s in samples), 'both memory phases must be observed' +if temporary: + temporary.cleanup() diff --git a/scripts/validate.sh b/scripts/validate.sh index 1c1cedc4..366e86d9 100755 --- a/scripts/validate.sh +++ b/scripts/validate.sh @@ -28,6 +28,7 @@ python3 -m py_compile \ scripts/cleanup_networks.py \ scripts/cleanup_storage.py \ scripts/test_cleanup_storage.py \ + scripts/test_proxy_memory.py \ scripts/health.py \ scripts/endpoint_snapshot.py \ scripts/endpoint_snapshot_sources.py \ @@ -148,6 +149,7 @@ docker compose -f deploy/compose.yaml build runner-image docker buildx build --load --build-arg CI_FLEET_COMMIT --build-arg CI_FLEET_VERSION \ --tag "$CI_FLEET_CONTROLLER_IMAGE" --file controller/Dockerfile . docker run --name "${CI_FLEET_VALIDATION_ID:-ci-fleet-validation}-runner-inspect" --rm --entrypoint /bin/bash "$CI_FLEET_RUNNER_IMAGE" -c './bin/Runner.Listener --version && docker --version && docker compose version && git --version' +python3 scripts/test_proxy_memory.py --image "$CI_FLEET_CONTROLLER_IMAGE" python3 scripts/test_apt_cache_e2e.py bash scripts/test-cleanup-e2e.sh CI_FLEET_INSTANCE=${CI_FLEET_VALIDATION_ID:-validation} scripts/cleanup.sh