This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-performance-tests.git
commit 9175ca3aece6b8b5b02178bb97cf7eef847b529f Author: Claus Ibsen <[email protected]> AuthorDate: Fri Nov 6 10:20:10 2026 +0100 ai-benchmark: the harness stops only its own Camel apps and the infra services it started On Nov 6 a reference pass of test-citrus-orders stopped chaosdemo, a Camel app that another session ran on this machine: run_test stopped every Camel app that started while camel test run ran. run-stepwise-ladder.sh also ran camel stop with no name after each example and camel infra stop for every service of the steps folder after a run, and infra_start stopped a service before it started its own. The apps of the harness are now the ones that run from its folders: the runtime.directory of the app's status file, as camel restart takes it, or its working directory. That is the dev app, the peer, the integrations a Citrus test starts, and the apps the model starts with camel_run under any name. run_test stops those the test left running, the harness stops what still runs of them when an example ends, and the runners do the same as a safety net for an interrupted run. Another session's apps run from its own folders and stay. The harness records the PID of each infra service it starts, in stepwise/infra-started.json, and stops only those: before the next pass of an example and when the run is over. A service that another session runs already is used as it is and left running, with a line in run.log. run-kamelet-check.sh gets the same two changes. agent_local.py, the one-shot harness, is unchanged. Verified with reference and negative passes of test-citrus-orders, connect-http-client and connect-service-jms-request-reply, with a Camel app from another folder running beside them: it ran through all passes, and after each pass no app or infra service of the harness was left. An artemis started outside the harness was used as it is and not stopped. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_016Y4dDi5iQNgtHrDdz2NpwF --- ai-benchmark/agent_mcp_stepwise.py | 134 ++++++++++++++++++++++++++++++------ ai-benchmark/run-kamelet-check.sh | 19 ++--- ai-benchmark/run-stepwise-ladder.sh | 21 +++--- 3 files changed, 128 insertions(+), 46 deletions(-) diff --git a/ai-benchmark/agent_mcp_stepwise.py b/ai-benchmark/agent_mcp_stepwise.py index 8366d01..4886fa9 100755 --- a/ai-benchmark/agent_mcp_stepwise.py +++ b/ai-benchmark/agent_mcp_stepwise.py @@ -254,7 +254,51 @@ SERVER_INSTRUCTIONS = os.environ.get("BENCH_SERVER_INSTRUCTIONS") == "1" INFRA_HOME = {} -def infra_start(services, seeds=None): +def infra_running(): + """{pid: alias} of the running camel infra services (camel infra ps).""" + out = subprocess.run(["camel", "infra", "ps", "--json"], capture_output=True, text=True).stdout + try: + return {str(r.get("pid")): r.get("alias") for r in json.loads(out[out.index("["):])} + except (ValueError, AttributeError): + return {} + + +# the infra services the harness started, {alias: pid}: the harness stops only those, at the next pass and when the run +# is over (run-stepwise-ladder.sh). A service that another session runs on this machine is not one of them +INFRA_STARTED = os.path.join(HERE, "stepwise", "infra-started.json") + + +def infra_started(): + try: + with open(INFRA_STARTED) as f: + return json.load(f) + except Exception: + return {} + + +def infra_save(started): + os.makedirs(os.path.dirname(INFRA_STARTED), exist_ok=True) + with open(INFRA_STARTED, "w") as f: + json.dump(started, f) + + +def infra_stop_started(aliases=None): + """Stops the infra services the harness started (all, or those of the aliases) that still run, by their PID, and + forgets them. Answers the aliases it stopped.""" + started, running, stopped = infra_started(), infra_running(), [] + for alias, pid in list(started.items()): + if aliases is not None and alias not in aliases: + continue + if running.get(str(pid)) == alias: + subprocess.run(["camel", "infra", "stop", str(pid)], check=False, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + stopped.append(alias) + del started[alias] + infra_save(started) + return stopped + + +def infra_start(services, seeds=None, log=None): """Start the services the example needs with `camel infra run <svc> --background` and wait until each answers. Stopped first, so every pass starts against an empty database. A service kept running between passes would carry @@ -262,6 +306,9 @@ def infra_start(services, seeds=None): of the model: step 3 sets C-207 to NL, and the next pass would find it already NL. Postgres takes about 80 s to come up, which is the price of that. The runner stops them when the whole run is over. + Only a service the harness started is stopped: one that another session runs already is used as it is and left + running, as a second one would not get its ports (the pass then starts against what that service holds). + seeds: {service: folder under seed/} for a service that serves files (sftp): the folder is copied to infra-<svc>/ in the pass's output folder, and the service is started in it, so it serves a fresh copy in every pass. @@ -269,10 +316,15 @@ def infra_start(services, seeds=None): `camel infra run aws s3`; camel infra get and stop know the service by its alias (aws) only. """ seeds = seeds or {} + infra_stop_started([svc.split()[0] for svc in services]) + running = infra_running() + others = [svc for svc in services if svc.split()[0] in running.values()] + if others and log: + print(f"camel infra: {', '.join(others)} runs already, not started by the harness: used as it is", + file=log, flush=True) for svc in services: - subprocess.run(["camel", "infra", "stop", svc.split()[0]], check=False, - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - for svc in services: + if svc in others: + continue cwd = HERE if seeds.get(svc): cwd = os.path.join(OUT, "infra-" + svc.replace(" ", "-")) @@ -298,6 +350,13 @@ def infra_start(services, seeds=None): time.sleep(5) else: data[svc] = {"error": f"{svc} did not come up within {INFRA_TIMEOUT}s"} + # the PIDs of the ones it started, to stop those and only those + started, now = infra_started(), infra_running() + for svc in services: + pids = [pid for pid, alias in now.items() if alias == svc.split()[0] and pid not in running] + if svc not in others and pids: + started[svc.split()[0]] = pids[0] + infra_save(started) return data @@ -610,13 +669,50 @@ def start_dev_app(project, name, mode="a"): return proc -def camel_pids(): - """The PIDs of the running Camel integrations (camel ps).""" +def harness_pids(): + """The PIDs of the running Camel apps of the harness (camel ps): the ones that run from a folder of it, as the dev + app, the peer, the apps the model starts with camel_run (in the project) and the integrations a Citrus test starts + do. Another session's apps run from its own folders and are left alone: on Nov 6 run_test stopped chaosdemo of + another session, as it then stopped every app that started while a test ran.""" out = subprocess.run(["camel", "ps", "--json"], capture_output=True, text=True).stdout try: - return {str(p.get("pid")) for p in json.loads(out[out.index("["):])} + pids = [str(p.get("pid")) for p in json.loads(out[out.index("["):])] except (ValueError, AttributeError): return set() + here = os.path.realpath(HERE) + return {pid for pid in pids if (os.path.realpath(app_folder(pid) or "/") + os.sep).startswith(here + os.sep)} + + +def app_folder(pid): + """The folder a Camel app runs from: runtime.directory of its status file, as camel restart takes it, else its + working directory (an app that has not written its status file yet).""" + try: + with open(os.path.expanduser(f"~/.camel/{pid}-status.json")) as f: + folder = (json.load(f).get("runtime") or {}).get("directory") + if folder: + return folder + except Exception: + pass + out = subprocess.run(["lsof", "-a", "-p", pid, "-d", "cwd", "-Fn"], capture_output=True, text=True).stdout + m = re.search(r"^n(.+)$", out, re.M) + return m.group(1) if m else None + + +def stop_harness_apps(keep=()): + """Stops the running Camel apps of the harness but the PIDs in keep, waits until they are gone (their ports with + them) and kills what is left after 30 s. Answers the PIDs it stopped.""" + pids = sorted(harness_pids() - set(keep)) + for pid in pids: + subprocess.run(["camel", "stop", pid], capture_output=True) + deadline = time.time() + 30 + while pids and time.time() < deadline and harness_pids() & set(pids): + time.sleep(1) + for pid in harness_pids() & set(pids): + try: + os.kill(int(pid), signal.SIGKILL) + except (OSError, ValueError): + pass + return pids def run_test(project, test_file, name, sid, log): @@ -640,7 +736,7 @@ def run_test(project, test_file, name, sid, log): stop.wait(timeout=30) except subprocess.TimeoutExpired: stop.kill() - before = camel_pids() + before = harness_pids() out_path = os.path.join(OUT, f"step{sid}.test-run.out") rc = None with open(out_path, "w") as out: @@ -654,18 +750,8 @@ def run_test(project, test_file, name, sid, log): test.wait(timeout=15) except subprocess.TimeoutExpired: os.killpg(test.pid, signal.SIGKILL) - left = sorted(camel_pids() - before) - for pid in left: - subprocess.run(["camel", "stop", pid], capture_output=True) - # until they are gone (the app's port with them), else killed - deadline = time.time() + 30 - while time.time() < deadline and camel_pids() - before: - time.sleep(1) - for pid in camel_pids() - before: - try: - os.kill(int(pid), signal.SIGKILL) - except (OSError, ValueError): - pass + # what the test left running of its own, until it is gone (the app's port with it), else killed + left = stop_harness_apps(keep=before) text = re.sub(r"\x1b\[[0-9;]*m", "", read(out_path)) # without the colours of the console summary = re.findall(r"Total tests run: \d+, Passes: \d+, Failures: \d+, Skips: \d+", text) # passed: the exit code and the summary agree; a test file that does not load fails with exit code 1 @@ -902,7 +988,7 @@ def main(): # when the pass ends, so the next pass of the same example does not pay the startup again (about 80 s); the # runner stops them when the whole run is over t0 = time.time() - data = infra_start(cfg["infra"], cfg.get("infra_seed")) + data = infra_start(cfg["infra"], cfg.get("infra_seed"), log) print(f"camel infra run {' '.join(cfg['infra'])} -> {json.dumps(data)[:300]} ({time.time() - t0:.0f}s)", file=log, flush=True) for svc, home in INFRA_HOME.items(): @@ -1188,6 +1274,12 @@ def main(): proc.wait(timeout=20) except Exception: proc.kill() + # what else of the harness still runs (an app the model started with camel_run under another name, one a test + # left behind), and only that: run-stepwise-ladder.sh stopped every Camel app after an example, also another + # session's + left = stop_harness_apps() + if left: + print(f"stopped the harness's apps still running: {left}", file=log, flush=True) print(f"DONE passed={sum(1 for r in results if r['ok'])}/{len(results)}", file=log, flush=True) log.close() diff --git a/ai-benchmark/run-kamelet-check.sh b/ai-benchmark/run-kamelet-check.sh index bf15339..668f339 100755 --- a/ai-benchmark/run-kamelet-check.sh +++ b/ai-benchmark/run-kamelet-check.sh @@ -20,23 +20,18 @@ for i in $(seq 1 "$K"); do echo "[$T] $name start $(date +%T)" | tee -a "stepwise-$T.log" BENCH_STEPS="$f" BENCH_TAG="$T/$name" python3 agent_mcp_stepwise.py > "stepwise-$T-$name.out" 2>&1 tail -1 "stepwise/$T/$name/run.log" | tee -a "stepwise-$T.log" - # safety net: the harness stops its integration, but an interrupted run leaves one behind, and an app the model - # started itself with camel_run, or renamed, is not stopped by name (r5: an orphan held port 8080 for six hours - # and failed every later HTTP rung). The machine runs only the benchmark during a series, so stop them all. - camel stop > /dev/null 2>&1 || true + # safety net: the harness stops its apps, but an interrupted run leaves one behind (r5: an orphan held port 8080 for + # six hours and failed every later HTTP rung). Only the harness's apps, the ones that run from its folders, as in + # run-stepwise-ladder.sh: another session's apps on this machine stay + python3 -c "import agent_mcp_stepwise as h; h.stop_harness_apps()" > /dev/null 2>&1 || true sleep 3 if lsof -nP -iTCP:8080 -sTCP:LISTEN > /dev/null 2>&1; then echo "[$T] WARNING port 8080 still in use after $name: $(lsof -nP -iTCP:8080 -sTCP:LISTEN | tail -1)" | tee -a "stepwise-$T.log" fi done - # the services an example needed stay up across its passes; stop them once the run is over - for svc in $(python3 -c " -import glob, json -out = set() -for f in glob.glob('steps-kamelet/*.json'): - try: out.update(json.load(open(f)).get('infra') or []) - except Exception: pass -print(' '.join(sorted(out)))"); do camel infra stop "$svc" > /dev/null 2>&1 || true; done + # the services an example needed stay up across its passes; stop them once the run is over, only those the harness + # started (stepwise/infra-started.json): another session may run the same service + python3 -c "import agent_mcp_stepwise as h; h.infra_stop_started()" > /dev/null 2>&1 || true echo "[$T] DONE $(date +%T)" | tee -a "stepwise-$T.log" done BENCH_STEPS_DIR=steps-kamelet python3 summarize_stepwise.py "$TAG" "$K" diff --git a/ai-benchmark/run-stepwise-ladder.sh b/ai-benchmark/run-stepwise-ladder.sh index 63774a2..bfae4df 100755 --- a/ai-benchmark/run-stepwise-ladder.sh +++ b/ai-benchmark/run-stepwise-ladder.sh @@ -21,24 +21,19 @@ for i in $(seq 1 "$K"); do echo "[$T] $name start $(date +%T)" | tee -a "stepwise-$T.log" BENCH_STEPS="$f" BENCH_TAG="$T/$name" python3 agent_mcp_stepwise.py > "stepwise-$T-$name.out" 2>&1 tail -1 "stepwise/$T/$name/run.log" | tee -a "stepwise-$T.log" - # safety net: the harness stops its integration, but an interrupted run leaves one behind, and an app the model - # started itself with camel_run, or renamed, is not stopped by name (r5: an orphan held port 8080 for six hours - # and failed every later HTTP rung). The machine runs only the benchmark during a series, so stop them all. - camel stop > /dev/null 2>&1 || true + # safety net: the harness stops its apps, but an interrupted run leaves one behind (r5: an orphan held port 8080 for + # six hours and failed every later HTTP rung). Only the harness's apps, the ones that run from its folders, also + # those the model started with camel_run under another name: another session's apps on this machine stay (Nov 6: + # `camel stop` with no name stopped every app) + python3 -c "import agent_mcp_stepwise as h; h.stop_harness_apps()" > /dev/null 2>&1 || true sleep 3 if lsof -nP -iTCP:8080 -sTCP:LISTEN > /dev/null 2>&1; then echo "[$T] WARNING port 8080 still in use after $name: $(lsof -nP -iTCP:8080 -sTCP:LISTEN | tail -1)" | tee -a "stepwise-$T.log" fi done - # the services an example needed stay up across its passes; stop them once the run is over - for svc in $(python3 -c " -import glob, json, os -out = set() -for f in glob.glob(os.environ.get('BENCH_STEPS_DIR', 'steps-ladder') + '/*.json'): - # an entry may name the implementation (aws s3): camel infra stop takes the alias - try: out.update(s.split()[0] for s in json.load(open(f)).get('infra') or []) - except Exception: pass -print(' '.join(sorted(out)))"); do camel infra stop "$svc" > /dev/null 2>&1 || true; done + # the services an example needed stay up across its passes; stop them once the run is over, only those the harness + # started (stepwise/infra-started.json): another session may run the same service + python3 -c "import agent_mcp_stepwise as h; h.infra_stop_started()" > /dev/null 2>&1 || true echo "[$T] DONE $(date +%T)" | tee -a "stepwise-$T.log" done python3 summarize_stepwise.py "$TAG" "$K"
