diff --git a/scripts/benchmark/capacity_local_docker.py b/scripts/benchmark/capacity_local_docker.py index 184a224b5..89543e7b0 100644 --- a/scripts/benchmark/capacity_local_docker.py +++ b/scripts/benchmark/capacity_local_docker.py @@ -633,7 +633,11 @@ def _execute_matrix( topology, topology["component_services"]["load-generator"] ) environment = _execution_environment(context) - source_revision = _run(["git", "rev-parse", "HEAD"], timeout=30).strip() + try: + git = capacity_suite.git_executable() + except FileNotFoundError as error: + raise TopologyError(str(error)) from error + source_revision = _run([git, "rev-parse", "HEAD"], timeout=30).strip() if not re.fullmatch(r"[0-9a-f]{40}", source_revision): raise TopologyError("source revision must be a full Git commit") matrix_arguments = [command, *arguments] diff --git a/scripts/benchmark/capacity_matrix.py b/scripts/benchmark/capacity_matrix.py index 028430481..22cd3198f 100644 --- a/scripts/benchmark/capacity_matrix.py +++ b/scripts/benchmark/capacity_matrix.py @@ -38,6 +38,44 @@ DEFAULT_COLLECTOR = SUITE_ROOT / "collectors/local-docker/collector.json" DEFAULT_SMOKE = SUITE_ROOT / "topologies/local-docker/smoke.json" +# Container worker lifecycle files (Server #1051). Each script takes only the +# file name and keeps it in one private directory under the container's temp +# directory: created 0700, and used only when it is a real directory owned by +# the adapter user, so another user cannot plant or redirect a pid file. Each +# file is created exclusively (noclobber), so a stale or planted file refuses +# the start instead of being overwritten. Cleanup checks the same ownership +# and still kills only the process whose recorded start time matches. +_PIDFILE_CUSTODY = ( + 'case "$1" in ""|.*|*/*) exit 42;; esac; ' + 'dir="${TMPDIR:-/tmp}/durable-workflow-capacity-workers"; pidfile="$dir/$1"; ' +) +_PIDFILE_DIRECTORY_OWNED = ( + '{ [ -d "$dir" ] && [ ! -L "$dir" ] && [ -O "$dir" ]; } || exit 42; ' +) +WORKER_PIDFILE_SCRIPT = ( + _PIDFILE_CUSTODY + + 'shift; umask 077; mkdir -m 700 "$dir" 2>/dev/null; ' + + _PIDFILE_DIRECTORY_OWNED + + 'start="$(awk \'{print $22}\' "/proc/$$/stat")"; ' + 'set -C; printf "%s %s\\n" "$$" "$start" >"$pidfile" || exit 43; set +C; ' + 'exec "$@"' +) +WORKER_CLEANUP_SCRIPT = ( + _PIDFILE_CUSTODY + + 'if [ ! -e "$pidfile" ] && [ ! -L "$pidfile" ]; then exit 0; fi; ' + + _PIDFILE_DIRECTORY_OWNED + + '{ [ -f "$pidfile" ] && [ ! -L "$pidfile" ] && [ -O "$pidfile" ]; } || exit 42; ' + 'read -r pid expected_start <"$pidfile"; ' + 'if [ -r "/proc/$pid/stat" ]; then ' + 'actual_start="$(awk \'{print $22}\' "/proc/$pid/stat")"; ' + '[ "$actual_start" = "$expected_start" ] || exit 41; ' + 'kill -TERM "$pid"; attempts=0; ' + 'while kill -0 "$pid" 2>/dev/null && [ "$attempts" -lt 150 ]; do ' + "sleep 0.1; attempts=$((attempts + 1)); done; " + 'if kill -0 "$pid" 2>/dev/null; then kill -KILL "$pid"; fi; ' + 'fi; rm -f "$pidfile"' +) + class MatrixError(RuntimeError): """The executable benchmark matrix failed closed.""" @@ -833,11 +871,7 @@ def _entrypoint( [ "sh", "-c", - ( - 'pidfile="$1"; shift; umask 077; ' - 'start="$(awk \'{print $22}\' "/proc/$$/stat")"; ' - 'printf "%s %s\\n" "$$" "$start" >"$pidfile"; exec "$@"' - ), + WORKER_PIDFILE_SCRIPT, "capacity-worker", self._container_worker_pid_file(descriptor, environment), *command, @@ -860,7 +894,7 @@ def _container_worker_pid_file( ) ) digest = hashlib.sha256(identity.encode()).hexdigest()[:24] - return f"/tmp/durable-workflow-capacity-worker-{digest}.pid" + return f"{digest}.pid" def _container_worker_cleanup_command( self, @@ -875,26 +909,13 @@ def _container_worker_cleanup_command( if not container: return None pidfile = self._container_worker_pid_file(descriptor, environment) - script = ( - 'pidfile="$1"; ' - 'if [ ! -f "$pidfile" ]; then exit 0; fi; ' - 'read -r pid expected_start <"$pidfile"; ' - 'if [ -r "/proc/$pid/stat" ]; then ' - 'actual_start="$(awk \'{print $22}\' "/proc/$pid/stat")"; ' - '[ "$actual_start" = "$expected_start" ] || exit 41; ' - 'kill -TERM "$pid"; attempts=0; ' - 'while kill -0 "$pid" 2>/dev/null && [ "$attempts" -lt 150 ]; do ' - "sleep 0.1; attempts=$((attempts + 1)); done; " - 'if kill -0 "$pid" 2>/dev/null; then kill -KILL "$pid"; fi; ' - 'fi; rm -f "$pidfile"' - ) return [ "docker", "exec", container, "sh", "-c", - script, + WORKER_CLEANUP_SCRIPT, "capacity-worker-cleanup", pidfile, ] diff --git a/scripts/benchmark/capacity_suite.py b/scripts/benchmark/capacity_suite.py index 17b9f6c48..bca58e07a 100755 --- a/scripts/benchmark/capacity_suite.py +++ b/scripts/benchmark/capacity_suite.py @@ -7,6 +7,7 @@ import hashlib import json import math +import os import platform import re import subprocess @@ -2589,10 +2590,31 @@ def _reference_observations( return observations +# System Git installs, checked in order. The capacity tools never resolve Git +# through PATH (Server #1051); CAPACITY_GIT names another absolute binary. +GIT_CANDIDATES = ("/usr/bin/git", "/usr/local/bin/git", "/opt/homebrew/bin/git") + + +def git_executable() -> str: + configured = os.environ.get("CAPACITY_GIT", "").strip() + for candidate in (configured,) if configured else GIT_CANDIDATES: + path = Path(candidate) + if path.is_absolute() and path.is_file() and os.access(path, os.X_OK): + return str(path) + raise FileNotFoundError( + "CAPACITY_GIT must name an absolute, executable Git binary" + if configured + else "no Git executable at " + ", ".join(GIT_CANDIDATES) + ) + + def _git_revision() -> str: try: return subprocess.check_output( - ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, stderr=subprocess.DEVNULL + [git_executable(), "rev-parse", "HEAD"], + cwd=ROOT, + text=True, + stderr=subprocess.DEVNULL, ).strip() except (OSError, subprocess.CalledProcessError): return "unknown-local-revision" diff --git a/tests/Unit/Support/capacity_benchmark_test.py b/tests/Unit/Support/capacity_benchmark_test.py index bca340802..4637bd008 100644 --- a/tests/Unit/Support/capacity_benchmark_test.py +++ b/tests/Unit/Support/capacity_benchmark_test.py @@ -4,6 +4,7 @@ import importlib.util import io import json +import os from pathlib import Path import shutil import subprocess @@ -643,12 +644,10 @@ def test_matrix_routes_adapter_roles_to_resolved_language_containers( self.assertEqual(["python3", "capacity_adapter.py", "worker"], command[-3:]) self.assertNotIn("--interactive", command) self.assertIn("capacity-worker", command) - self.assertTrue( - any( - value.startswith("/tmp/durable-workflow-capacity-worker-") - for value in command - ) + self.assertRegex( + runner._container_worker_pid_file(adapter, environment), r"^[0-9a-f]{24}\.pid$" ) + self.assertIn(runner._container_worker_pid_file(adapter, environment), command) self.assertTrue( any( value == "DURABLE_WORKFLOW_TASK_QUEUE=capacity-smoke" @@ -679,6 +678,84 @@ def test_matrix_routes_adapter_roles_to_resolved_language_containers( cleanup[-1], ) + def test_worker_pid_files_are_private_and_exclusive(self) -> None: + # Server #1051: run the exact container scripts with the local sh, + # with TMPDIR standing in for the container's temp directory. + def run(script: str, temp: Path, *arguments: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["sh", "-c", script, "capacity-worker", *arguments], + text=True, capture_output=True, check=False, + env={"PATH": os.environ.get("PATH", ""), "TMPDIR": str(temp)}, + ) + + def worker(temp: Path, name: str) -> subprocess.CompletedProcess[str]: + return run(capacity_matrix.WORKER_PIDFILE_SCRIPT, temp, name, "true") + + def cleanup(temp: Path, name: str) -> subprocess.CompletedProcess[str]: + return run(capacity_matrix.WORKER_CLEANUP_SCRIPT, temp, name) + + with tempfile.TemporaryDirectory() as scratch: + temp = Path(scratch) / "temp" + temp.mkdir() + directory = temp / "durable-workflow-capacity-workers" + pidfile = directory / "worker.pid" + started = worker(temp, "worker.pid") + self.assertEqual(0, started.returncode, started.stderr) + self.assertEqual(0o700, directory.stat().st_mode & 0o777) + self.assertEqual(0o600, pidfile.stat().st_mode & 0o777) + recorded = pidfile.read_text() + self.assertRegex(recorded, r"^\d+ \d+\n$") + + # A second start with the same identity refuses; nothing is overwritten. + self.assertEqual(43, worker(temp, "worker.pid").returncode) + self.assertEqual(recorded, pidfile.read_text()) + + # The recorded process has exited, so cleanup only removes the file. + self.assertEqual(0, cleanup(temp, "worker.pid").returncode) + self.assertFalse(pidfile.exists()) + self.assertEqual(0, cleanup(temp, "worker.pid").returncode, "a missing pid file is already clean") + + # Only a plain file name is accepted. + for name in ("", "../escape.pid", ".hidden", "a/b.pid"): + self.assertEqual(42, worker(temp, name).returncode, name) + + # A symlinked directory or pid file is refused, never followed. + target = Path(scratch) / "elsewhere" + target.mkdir(mode=0o700) + linked = Path(scratch) / "linked-temp" + linked.mkdir() + (linked / "durable-workflow-capacity-workers").symlink_to(target) + self.assertEqual(42, worker(linked, "worker.pid").returncode) + self.assertEqual([], list(target.iterdir())) + (directory / "planted.pid").symlink_to(target / "victim") + self.assertEqual(42, cleanup(temp, "planted.pid").returncode) + self.assertFalse((target / "victim").exists()) + + # A live process with a different recorded start is left alone. + with open(pidfile, "x") as handle: + handle.write(f"{os.getpid()} 1\n") + self.assertEqual(41, cleanup(temp, "worker.pid").returncode) + self.assertTrue(pidfile.exists()) + + def test_git_is_resolved_without_path(self) -> None: + with tempfile.TemporaryDirectory() as scratch: + fake = Path(scratch) / "git" + fake.write_text("#!/bin/sh\necho 0123456789abcdef0123456789abcdef01234567\n") + fake.chmod(0o700) + with mock.patch.dict(capacity_suite.os.environ, {"CAPACITY_GIT": str(fake), "PATH": scratch}): + self.assertEqual(str(fake), capacity_suite.git_executable()) + self.assertEqual("0123456789abcdef0123456789abcdef01234567", capacity_suite._git_revision()) + for configured in ("git", str(Path(scratch) / "missing")): + with mock.patch.dict(capacity_suite.os.environ, {"CAPACITY_GIT": configured}): + with self.assertRaises(FileNotFoundError): + capacity_suite.git_executable() + self.assertEqual("unknown-local-revision", capacity_suite._git_revision()) + # Unset, a PATH entry named git is never chosen. + with mock.patch.dict(capacity_suite.os.environ, {"PATH": scratch}, clear=True): + with mock.patch.object(capacity_suite, "GIT_CANDIDATES", ("/nonexistent/git",)): + with self.assertRaises(FileNotFoundError): + capacity_suite.git_executable() + def test_worker_process_runs_container_cleanup_after_client_exit(self) -> None: process = mock.Mock() process.poll.return_value = 0