Skip to content

test largescale mq #34

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ActivePeter wants to merge 5 commits into
base: main
Choose a base branch
from
Open

test largescale mq #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3c38f26
test largescale mq
ActivePeter Jul 2, 2026
3258161
test
ActivePeter Jul 2, 2026
5f9b186
test
ActivePeter Jul 3, 2026
65a3b22
test
ActivePeter Jul 3, 2026
45ef655
test
ActivePeter Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 139 additions & 27 deletions .github/workflows/all_test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ jobs:
run: |
python3 - <<'PY'
import copy
import shlex
from pathlib import Path
import yaml

Expand All @@ -64,7 +65,7 @@ jobs:

top_attention_scene_prefix = "ci_top_attention_"

def top_attention_command(scene_id, *, case_config, timeout_seconds=21600):
def top_attention_command(scene_id, *, case_config, extra_args=None, timeout_seconds=21600):
if not scene_id.startswith(top_attention_scene_prefix):
raise ValueError(f"not a top-attention CI scene id: {scene_id}")
suffix = scene_id[len(top_attention_scene_prefix):]
Expand All @@ -74,6 +75,8 @@ jobs:
)
if case_config:
command += " --case-config __RUN_DIR__/configs/ci_scene_config.yaml"
for arg in extra_args or []:
command += " " + shlex.quote(str(arg))
return {
"id": f"top_attention_{suffix}",
"command": command,
Expand Down Expand Up @@ -151,16 +154,70 @@ jobs:
"case_config": True,
"scene_config": {},
},
"ci_top_attention_largescale_mq": {
"subject": "mq",
"runtime_contract": "rust_self_managed",
"scale": "n1_kvowner_dram_3gib",
"case_config": False,
"command_variants": [
{
"id_suffix": f"p{producer_count}_c{consumer_count}",
"extra_args": [
"--single-host-logical-targets",
"--testbed-bundle-source",
"__TEST_BED_BUNDLE_ROOT__",
"--workdir",
f"__WORKDIR_ROOT__/largescale_mq_ci_single_host/p{producer_count}_c{consumer_count}",
"--owner-count",
"4",
"--owner-dram-gib",
"1",
"--producer-count",
str(producer_count),
"--consumer-count",
str(consumer_count),
"--duration-seconds",
"30",
"--value-size",
"256",
"--op-timeout-seconds",
"5",
"--cluster-ready-timeout-seconds",
"1800",
"--consumer-sim-min-ms",
"1",
"--consumer-sim-max-ms",
"1",
],
}
for producer_count, consumer_count in ((8, 8), (32, 32), (160, 8))
],
"scene_config": {},
},
}

for scene_id, scene_def in top_attention_ci_scenes.items():
commands = [
top_attention_command(
scene_id,
case_config=scene_def["case_config"],
timeout_seconds=scene_def.get("timeout_seconds", 21600),
)
]
command_variants = scene_def.get("command_variants")
if command_variants is None:
commands = [
top_attention_command(
scene_id,
case_config=scene_def["case_config"],
extra_args=scene_def.get("extra_args"),
timeout_seconds=scene_def.get("timeout_seconds", 21600),
)
]
else:
commands = []
for variant in command_variants:
command = top_attention_command(
scene_id,
case_config=scene_def["case_config"],
extra_args=variant["extra_args"],
timeout_seconds=scene_def.get("timeout_seconds", 21600),
)
command["id"] = f"{command['id']}_{variant['id_suffix']}"
commands.append(command)
existing_scene = suite["scenes"].get(scene_id)
if existing_scene is None:
suite["scenes"][scene_id] = {
Expand Down Expand Up @@ -189,6 +246,7 @@ jobs:
# - ci_top_attention_mq_mpsc keeps MPSC API channel coverage inside the same CI testbed contract.
# - ci_top_attention_mq_mpmc keeps MPMC API channel coverage inside the same CI testbed contract.
# - ci_top_attention_mq_mpmc_bench keeps heavier MPMC benchmark-style coverage in this workflow.
# - ci_top_attention_largescale_mq runs a bounded same-host MQ TEST_STACK workload in GitHub Actions.
suite["scenes"] = {
key: value
for key, value in suite["scenes"].items()
Expand Down Expand Up @@ -221,6 +279,7 @@ jobs:
# - ci_top_attention_mq_mpsc stays on n1_kvowner_dram_20gib.
# - ci_top_attention_mq_mpmc stays on n1_kvowner_dram_20gib.
# - ci_top_attention_mq_mpmc_bench stays on n1_kvowner_dram_20gib.
# - ci_top_attention_largescale_mq stays on n1_kvowner_dram_3gib because the nested workload owns its bounded scale.

out_path.write_text(
yaml.safe_dump(suite, sort_keys=False, allow_unicode=False),
Expand All @@ -231,14 +290,17 @@ jobs:

- name: Run ci_2_virt_node full flow
run: |
python3 fluxon_test_stack/ci_2_virt_node.py \
--suite-path "$RUNNER_TEMP/ci_test_list.ci.yaml" \
--print-generated \
--testbed-hostworkdir "$RUNNER_TEMP/fluxon_deploy"

- name: Print ci_2_virt_node failure summary
if: ${{ failure() }}
run: |
set +e
timeout --preserve-status --signal=INT 17000s \
python3 fluxon_test_stack/ci_2_virt_node.py \
--suite-path "$RUNNER_TEMP/ci_test_list.ci.yaml" \
--print-generated \
--testbed-hostworkdir "$RUNNER_TEMP/fluxon_deploy"
rc=$?
if [ "$rc" -eq 0 ]; then
exit 0
fi
echo "ci_2_virt_node failed or timed out before GitHub job cancellation: rc=$rc"
python3 - <<'PY'
from pathlib import Path
import yaml
Expand All @@ -249,6 +311,21 @@ jobs:
print(f"missing {case_runs_path}")
raise SystemExit(0)

def print_file(path):
print(f"=== {path} ===")
if path.exists():
print(path.read_text(encoding="utf-8", errors="replace"))
else:
print(f"missing {path}")

def print_tail(path, *, line_count=240):
print(f"=== {path} tail ===")
if path.exists():
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
print("\n".join(lines[-line_count:]))
else:
print(f"missing {path}")

case_runs = yaml.safe_load(case_runs_path.read_text(encoding="utf-8"))
print("=== case_runs.yaml ===")
print(yaml.safe_dump(case_runs, sort_keys=False, allow_unicode=False))
Expand All @@ -264,22 +341,57 @@ jobs:
"summary.yaml",
"exception.txt",
"logs/ci_runner/exit_code.txt",
"logs/ci_runner/restart_count.txt",
"logs/ci_runner/inflight_attempt.txt",
):
path = run_dir / rel
print(f"=== {path} ===")
if path.exists():
print(path.read_text(encoding="utf-8", errors="replace"))
else:
print(f"missing {path}")
print_file(path)

stdout_path = run_dir / "logs" / "ci_runner" / "stdout.log"
print(f"=== {stdout_path} tail ===")
if stdout_path.exists():
lines = stdout_path.read_text(encoding="utf-8", errors="replace").splitlines()
print("\n".join(lines[-240:]))
else:
print(f"missing {stdout_path}")
print_tail(stdout_path)

nested_root = workdir / "largescale_mq_ci_single_host"
print(f"=== nested largescale MQ diagnostics: {nested_root} ===")
if not nested_root.exists():
print(f"missing {nested_root}")
else:
nested_failed_run_dirs = []
for path in sorted(nested_root.glob("*/case_runs.yaml")):
print_file(path)
try:
nested_case_runs = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except Exception as exc:
print(f"failed to parse {path}: {type(exc).__name__}: {exc}")
continue
for case in nested_case_runs.get("cases", []):
last_run = case.get("last_run", {})
if last_run.get("outcome") == "SUCCESS":
continue
case_id = case.get("case_id")
run_index = last_run.get("run_index")
nested_failed_run_dirs.append(path.parent / "results" / str(case_id) / f"run_{run_index}")

print("=== nested largescale MQ failed run diagnostics ===")
if not nested_failed_run_dirs:
print("no failed nested run dirs found from nested case_runs.yaml")
for run_dir in nested_failed_run_dirs:
print(f"--- nested failed run: {run_dir} ---")
for rel in (
"summary.yaml",
"exception.txt",
"benchmark_result.json",
"deploy_result.yaml",
"logs/ci_runner/exit_code.txt",
"logs/ci_runner/restart_count.txt",
"logs/ci_runner/inflight_attempt.txt",
):
print_file(run_dir / rel)
print_tail(run_dir / "logs" / "ci_runner" / "stdout.log", line_count=320)

for path in sorted(nested_root.glob("*/test_runner.log")):
print_tail(path, line_count=320)
PY
exit "$rc"

- name: Normalize ci_2_virt_node debug artifact permissions
if: ${{ always() }}
Expand Down
Loading
Loading