#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import re import shlex import shutil import socket import subprocess import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple LOCAL_HOST_ALIASES = {"127.0.0.1", "localhost", "host.docker.internal", "local", "test-service"} DEFAULT_ADMIN_USER = "forcad" DEFAULT_ADMIN_PASSWORD = "forcad_local" BUILDER_PROFILES = { "cpp-make": { "dockerfile": "builders/cpp-make.Dockerfile", "image": "forcad-local-builder-cpp-make:1", }, } K3D_PREWARM_IMAGES = [ "ghcr.io/k3d-io/k3d-tools:5.9.0", "rancher/k3s:v1.31.5-k3s1", ] @dataclass class ServiceSpec: name: str service_dir: Path checker_dir: Path profile: str checker_entry: Path checker_type: str places: int checker_timeout: int protocol: str host_port: Optional[int] requirements: List[str] = field(default_factory=list) checker_build_enabled: bool = False checker_build_profile: Optional[str] = None checker_build_command: Optional[str] = None checker_build_artifacts: List[str] = field(default_factory=list) start_cmd: Optional[str] = None stop_cmd: Optional[str] = None @dataclass class ValidationReport: errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) def ok(self) -> bool: return not self.errors def info(message: str) -> None: print(f"[INFO] {message}") def warn(message: str) -> None: print(f"[WARN] {message}") def err(message: str) -> None: print(f"[ERROR] {message}", file=sys.stderr) def run_command(command: Sequence[str], cwd: Optional[Path] = None, check: bool = True) -> subprocess.CompletedProcess: printable = " ".join(command) location = f" (cwd={cwd})" if cwd else "" info(f"Running: {printable}{location}") result = subprocess.run(command, cwd=cwd, text=True) if check and result.returncode != 0: raise RuntimeError(f"Command failed with exit code {result.returncode}: {printable}") return result def run_shell(command: str, cwd: Optional[Path] = None, check: bool = True) -> subprocess.CompletedProcess: location = f" (cwd={cwd})" if cwd else "" info(f"Running shell: {command}{location}") result = subprocess.run(["bash", "-lc", command], cwd=cwd, text=True) if check and result.returncode != 0: raise RuntimeError(f"Command failed with exit code {result.returncode}: {command}") return result def run_command_with_retries( command: Sequence[str], *, cwd: Optional[Path] = None, retries: int = 0, retry_delay: int = 5, purpose: str = "command", ) -> subprocess.CompletedProcess: attempts = 1 + max(0, retries) last_exception: Optional[Exception] = None for attempt in range(1, attempts + 1): try: return run_command(command, cwd=cwd, check=True) except Exception as exc: # noqa: BLE001 last_exception = exc if attempt >= attempts: raise warn( f"{purpose} failed ({attempt}/{attempts}): {exc}. " f"Retrying in {retry_delay}s..." ) time.sleep(retry_delay) if last_exception: raise last_exception raise RuntimeError(f"{purpose} failed for unknown reason") def require_command(command: str, report: ValidationReport) -> None: if shutil.which(command) is None: report.errors.append(f"Не найдена системная команда `{command}`.") def parse_manifest(services_root: Path) -> Dict[str, dict]: manifest_path = services_root / "forcad-local.json" if not manifest_path.exists(): return {} try: data = json.loads(manifest_path.read_text()) except json.JSONDecodeError as exc: raise RuntimeError(f"Некорректный JSON в {manifest_path}: {exc}") from exc services = data.get("services", {}) if not isinstance(services, dict): raise RuntimeError(f"`services` в {manifest_path} должен быть объектом.") return services def detect_profile(service_dir: Path, overrides: dict) -> Optional[str]: if "profile" in overrides: return str(overrides["profile"]) if (service_dir / "docker-compose.yml").exists(): return "compose" if (service_dir / "deploy.sh").exists(): return "deploy_script" if (service_dir / "redeploy.sh").exists() or (service_dir / "deploy" / "chart" / "values.yaml").exists(): return "helm_chart" return None def detect_checker_entry(checker_dir: Path) -> Optional[Path]: checker_py = checker_dir / "checker.py" if checker_py.exists(): return Path("checker.py") dotted = sorted(checker_dir.glob("*.checker.py")) if len(dotted) == 1: return Path(dotted[0].name) if len(dotted) > 1: return Path(dotted[0].name) checker_named = sorted(p for p in checker_dir.glob("*.py") if "checker" in p.name.lower()) if len(checker_named) == 1: return Path(checker_named[0].name) return None def checker_protocol(entry_content: str) -> str: if "https://" in entry_content: return "https" if "http://" in entry_content: return "http" if ".split(':')" in entry_content or "split(\":\")" in entry_content: return "tcp" return "unknown" def checker_type(entry_content: str) -> str: return "pfr" if "OK_WITH_FLAG_ID" in entry_content else "hackerdom" def checker_places(entry_content: str) -> int: count = len(re.findall(r"define_vuln", entry_content)) return max(1, count) def infer_make_artifacts(makefile_path: Path) -> List[str]: makefile_text = makefile_path.read_text() artifacts = re.findall(r"-o\s+([A-Za-z0-9_.\-\/]+)", makefile_text) return sorted(set(artifacts)) def infer_checker_artifacts(checker_dir: Path) -> List[str]: artifacts: List[str] = [] for py_file in checker_dir.glob("*.py"): try: text = py_file.read_text() except UnicodeDecodeError: continue if "./libgen" in text or "libgen" in text: artifacts.append("libgen") return sorted(set(artifacts)) def detect_checker_build(checker_dir: Path, service_override: dict) -> Tuple[bool, Optional[str], Optional[str], List[str]]: raw_cfg = service_override.get("checker_build") if isinstance(raw_cfg, dict): enabled = bool(raw_cfg.get("enabled", True)) if not enabled: return False, None, None, [] profile = str(raw_cfg.get("profile", "cpp-make")) command = str(raw_cfg.get("command", "make")) raw_artifacts = raw_cfg.get("artifacts", []) artifacts = [str(x) for x in raw_artifacts] if isinstance(raw_artifacts, list) else [] return True, profile, command, artifacts makefile = checker_dir / "Makefile" if makefile.exists(): artifacts = infer_make_artifacts(makefile) or infer_checker_artifacts(checker_dir) return True, "cpp-make", "make", artifacts return False, None, None, [] def load_requirements(checker_dir: Path) -> List[str]: requirements: List[str] = [] candidates = [checker_dir / "requirements.txt", checker_dir / "requirments.txt"] for candidate in candidates: if not candidate.exists(): continue for line in candidate.read_text().splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue requirements.append(stripped) return requirements def discover_host_port(service_dir: Path, profile: str, protocol: str, overrides: dict) -> Optional[int]: if "port" in overrides: try: return int(overrides["port"]) except (TypeError, ValueError): return None if profile == "compose": compose_path = service_dir / "docker-compose.yml" if not compose_path.exists(): return None text = compose_path.read_text() matches = re.findall(r"(\d{2,5})\s*:\s*(\d{2,5})(?:/\w+)?", text) if not matches: return None return int(matches[0][0]) values_files = list(service_dir.glob("**/values.yaml")) node_ports: List[int] = [] for path in values_files: text = path.read_text() for match in re.findall(r"nodePort:\s*(\d{2,5})", text): node_ports.append(int(match)) # Some charts (e.g. ingress-nginx values) keep NodePorts as: # nodePorts: # http: 31080 # https: 31443 if not node_ports: for match in re.findall(r":\s*(\d{2,5})\b", text): value = int(match) if value >= 30000: node_ports.append(value) if node_ports: if protocol == "http": return min(node_ports) return max(node_ports) deploy_script = service_dir / "deploy.sh" if deploy_script.exists(): script_text = deploy_script.read_text() direct = re.search(r"localhost:(\d{2,5})", script_text) if direct: return int(direct.group(1)) return None def discover_services( services_root: Path, checkers_root: Path, selected_services: Optional[set[str]] = None, ) -> Tuple[List[ServiceSpec], ValidationReport]: report = ValidationReport() overrides = parse_manifest(services_root) if not services_root.exists(): report.errors.append(f"Папка сервисов не найдена: {services_root}") return [], report if not checkers_root.exists(): report.errors.append(f"Папка чекеров не найдена: {checkers_root}") return [], report specs: List[ServiceSpec] = [] service_dirs = sorted(p for p in services_root.iterdir() if p.is_dir() and not p.name.startswith(".")) for service_dir in service_dirs: name = service_dir.name if selected_services and name not in selected_services: continue service_override = overrides.get(name, {}) profile = detect_profile(service_dir, service_override) if profile is None: report.warnings.append(f"Пропускаю `{name}`: не найден поддерживаемый профиль деплоя.") continue checker_name = str(service_override.get("checker_dir", name)) checker_dir = checkers_root / checker_name if not checker_dir.exists(): report.errors.append(f"`{name}`: не найдена папка чекера `{checker_dir}`.") continue entry_override = service_override.get("checker_entry") if entry_override: checker_entry = Path(str(entry_override)) if not (checker_dir / checker_entry).exists(): report.errors.append(f"`{name}`: не найден checker entry `{checker_entry}` в `{checker_dir}`.") continue else: detected_entry = detect_checker_entry(checker_dir) if detected_entry is None: report.errors.append(f"`{name}`: не найден исполняемый checker (`checker.py` или `*.checker.py`).") continue checker_entry = detected_entry entry_text = (checker_dir / checker_entry).read_text() protocol = checker_protocol(entry_text) detected_type = checker_type(entry_text) detected_places = checker_places(entry_text) task_type = str(service_override.get("checker_type", detected_type)) places = int(service_override.get("places", detected_places)) timeout = int(service_override.get("checker_timeout", 30)) host_port = discover_host_port(service_dir, profile, protocol, service_override) requirements = load_requirements(checker_dir) build_enabled, build_profile, build_command, build_artifacts = detect_checker_build(checker_dir, service_override) spec = ServiceSpec( name=name, service_dir=service_dir, checker_dir=checker_dir, profile=profile, checker_entry=checker_entry, checker_type=task_type, places=places, checker_timeout=timeout, protocol=protocol, host_port=host_port, requirements=requirements, checker_build_enabled=build_enabled, checker_build_profile=build_profile, checker_build_command=build_command, checker_build_artifacts=build_artifacts, start_cmd=service_override.get("start_cmd"), stop_cmd=service_override.get("stop_cmd"), ) specs.append(spec) if not specs: report.errors.append("Не найдено ни одного валидного сервиса для запуска.") return specs, report def validate_preflight(specs: List[ServiceSpec], board_port: int) -> ValidationReport: report = ValidationReport() require_command("docker", report) seen_ports: Dict[int, str] = {} for spec in specs: if spec.host_port is None: report.errors.append(f"`{spec.name}`: не удалось определить host-порт (укажи `port` в `forcad-local.json`).") else: if spec.host_port == board_port: report.errors.append( f"`{spec.name}`: host-порт {spec.host_port} конфликтует с портом борды {board_port}." ) if spec.host_port in seen_ports: report.errors.append( f"Конфликт портов: `{spec.name}` и `{seen_ports[spec.host_port]}` используют {spec.host_port}." ) seen_ports[spec.host_port] = spec.name if spec.profile == "compose" and not (spec.service_dir / "docker-compose.yml").exists(): report.errors.append(f"`{spec.name}`: профиль compose выбран, но `docker-compose.yml` отсутствует.") if spec.profile in {"deploy_script", "helm_chart"}: for dep in ("helm", "kubectl"): if shutil.which(dep) is None: report.warnings.append( f"`{spec.name}`: команда `{dep}` не найдена; сервис может не подняться." ) if (spec.service_dir / "k3d-config.yaml").exists() and shutil.which("k3d") is None: report.warnings.append(f"`{spec.name}`: найден `k3d-config.yaml`, но команда `k3d` отсутствует.") if spec.checker_build_enabled: if not spec.checker_build_profile: report.errors.append(f"`{spec.name}`: включена сборка чекера, но не указан checker_build_profile.") elif spec.checker_build_profile not in BUILDER_PROFILES: report.errors.append( f"`{spec.name}`: неизвестный checker build profile `{spec.checker_build_profile}`." ) if not spec.checker_build_command: report.errors.append(f"`{spec.name}`: включена сборка чекера, но не задан checker_build_command.") if not spec.checker_build_artifacts: report.warnings.append( f"`{spec.name}`: не заданы checker build artifacts, кэш сборки для него отключён." ) return report def merge_reports(*reports: ValidationReport) -> ValidationReport: result = ValidationReport() for report in reports: result.errors.extend(report.errors) result.warnings.extend(report.warnings) return result def print_report(report: ValidationReport) -> None: for warning in report.warnings: warn(warning) for error_message in report.errors: err(error_message) def resolve_checkers_root(services_root: Path, checkers_root_arg: Optional[Path]) -> Path: if checkers_root_arg: return checkers_root_arg.resolve() sibling = services_root.parent / "checkers" if sibling.exists(): return sibling.resolve() inside = services_root / "checkers" if inside.exists(): return inside.resolve() raise RuntimeError( "Не удалось определить папку чекеров автоматически. Укажи `--checkers-root` явно." ) def safe_service_var(service_name: str) -> str: normalized = re.sub(r"[^A-Za-z0-9]+", "_", service_name).upper() return f"AD_TARGET_{normalized}" def ensure_clean_dir(path: Path) -> None: if path.exists(): shutil.rmtree(path) path.mkdir(parents=True, exist_ok=True) def resolve_builder_image(orchestrator_root: Path, profile: str) -> str: profile_config = BUILDER_PROFILES.get(profile) if not profile_config: raise RuntimeError(f"Unknown builder profile: {profile}") image = profile_config["image"] inspect = run_command(["docker", "image", "inspect", image], check=False) if inspect.returncode == 0: return image dockerfile = orchestrator_root / profile_config["dockerfile"] if not dockerfile.exists(): raise RuntimeError(f"Builder dockerfile not found for profile `{profile}`: {dockerfile}") info(f"Building checker builder image `{image}` from `{dockerfile}`.") run_command( [ "docker", "build", "-f", str(dockerfile), "-t", image, str(orchestrator_root), ] ) return image def hash_checker_inputs(checker_dir: Path, build_profile: str, build_command: str) -> str: digest = hashlib.sha256() digest.update(build_profile.encode()) digest.update(b"\0") digest.update(build_command.encode()) digest.update(b"\0") for item in sorted(checker_dir.rglob("*")): if not item.is_file(): continue rel = item.relative_to(checker_dir).as_posix() digest.update(rel.encode()) digest.update(b"\0") digest.update(item.read_bytes()) digest.update(b"\0") return digest.hexdigest() def restore_build_cache( service: ServiceSpec, checker_runtime_dir: Path, cache_root: Path, cache_key: str, ) -> bool: if not service.checker_build_artifacts: return False entry = cache_root / service.name / cache_key if not entry.exists(): return False for artifact in service.checker_build_artifacts: src = entry / artifact dst = checker_runtime_dir / artifact if not src.exists(): return False if src.is_dir(): if dst.exists(): shutil.rmtree(dst) shutil.copytree(src, dst) else: dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) info(f"Checker build cache hit for `{service.name}`.") return True def store_build_cache( service: ServiceSpec, checker_runtime_dir: Path, cache_root: Path, cache_key: str, ) -> None: if not service.checker_build_artifacts: return entry = cache_root / service.name / cache_key if entry.exists(): shutil.rmtree(entry) entry.mkdir(parents=True, exist_ok=True) for artifact in service.checker_build_artifacts: src = checker_runtime_dir / artifact if not src.exists(): continue dst = entry / artifact if src.is_dir(): shutil.copytree(src, dst) else: dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) def create_wrapper(service: ServiceSpec, destination: Path) -> None: env_var = safe_service_var(service.name) default_target = f"host.docker.internal:{service.host_port}" wrapper = f"""#!/usr/bin/env python3 import os import subprocess import sys from pathlib import Path LOCAL_ALIASES = {sorted(LOCAL_HOST_ALIASES)!r} DEFAULT_TARGET = os.environ.get("{env_var}", "{default_target}") def patch_hostname(argv): if len(argv) < 3: return argv action = argv[1].lower() if action not in {{"check", "put", "get", "test"}}: return argv host = argv[2] if host in LOCAL_ALIASES: argv[2] = DEFAULT_TARGET elif ":" not in host and ":" in DEFAULT_TARGET: argv[2] = f"{{host}}:{{DEFAULT_TARGET.rsplit(':', 1)[1]}}" return argv def main(): checker_path = Path(__file__).with_name("{service.checker_entry.as_posix()}") argv = patch_hostname(sys.argv[:]) completed = subprocess.run([sys.executable, str(checker_path), *argv[1:]], cwd=Path(__file__).parent) raise SystemExit(completed.returncode) if __name__ == "__main__": main() """ destination.write_text(wrapper) destination.chmod(0o755) def generate_runtime( runtime_dir: Path, specs: List[ServiceSpec], team_count: int, board_port: int, round_time: int, flag_lifetime: int, start_delay_seconds: int, ) -> dict: ensure_clean_dir(runtime_dir) checkers_dir = runtime_dir / "checkers" env_dir = runtime_dir / "env" checkers_dir.mkdir(parents=True, exist_ok=True) env_dir.mkdir(parents=True, exist_ok=True) requirements = {"gornilo", "checklib==0.7.0"} task_chunks: List[str] = [] for spec in specs: target_dir = checkers_dir / spec.name shutil.copytree(spec.checker_dir, target_dir, dirs_exist_ok=True) create_wrapper(spec, target_dir / "forcad_wrapper.py") for requirement in spec.requirements: requirements.add(requirement) task_chunks.append( "\n".join( [ f" - name: {spec.name}", f" checker: {spec.name}/forcad_wrapper.py", f" checker_timeout: {spec.checker_timeout}", f" checker_type: {spec.checker_type}", " gets: 1", " puts: 1", f" places: {max(1, spec.places)}", ] ) ) (checkers_dir / "requirements.txt").write_text("\n".join(sorted(requirements)) + "\n") start_time = dt.datetime.now(dt.UTC) + dt.timedelta(seconds=start_delay_seconds) start_time_text = start_time.strftime("%Y-%m-%d %H:%M:%S") teams = "\n".join( [ "\n".join( [ " - ip: host.docker.internal", f" name: Team {idx + 1}", f" highlighted: {'true' if idx == 0 else 'false'}", ] ) for idx in range(team_count) ] ) config_text = "\n".join( [ "game:", " mode: classic", f" round_time: {round_time}", f" start_time: '{start_time_text}'", " timezone: UTC", " default_score: 2500", f" flag_lifetime: {flag_lifetime}", " game_hardness: 10.0", " inflation: true", " checkers_path: /checkers/", " env_path: /checkers/bin/", "admin:", f" username: {DEFAULT_ADMIN_USER}", f" password: {DEFAULT_ADMIN_PASSWORD}", "tasks:", *task_chunks, "teams:", teams, "storages:", " db:", f" user: {DEFAULT_ADMIN_USER}", f" password: {DEFAULT_ADMIN_PASSWORD}", " dbname: forcad", " host: postgres", " port: 5432", " redis:", f" password: {DEFAULT_ADMIN_PASSWORD}", " db: 0", " host: redis", " port: 6379", " rabbitmq:", f" user: {DEFAULT_ADMIN_USER}", f" password: {DEFAULT_ADMIN_PASSWORD}", " host: rabbitmq", " port: 5672", " vhost: forcad", "", ] ) config_path = runtime_dir / "config.yml" config_path.write_text(config_text) postgres_env = "\n".join( [ "POSTGRES_HOST=postgres", "POSTGRES_PORT=5432", f"POSTGRES_USER={DEFAULT_ADMIN_USER}", f"POSTGRES_PASSWORD={DEFAULT_ADMIN_PASSWORD}", "POSTGRES_DB=forcad", "", ] ) redis_env = "\n".join( [ "REDIS_HOST=redis", "REDIS_PORT=6379", f"REDIS_PASSWORD={DEFAULT_ADMIN_PASSWORD}", "", ] ) rabbit_env = "\n".join( [ "RABBITMQ_HOST=rabbitmq", "RABBITMQ_PORT=5672", f"RABBITMQ_DEFAULT_USER={DEFAULT_ADMIN_USER}", f"RABBITMQ_DEFAULT_PASS={DEFAULT_ADMIN_PASSWORD}", "RABBITMQ_DEFAULT_VHOST=forcad", f"BROKER_API_URL=http://{DEFAULT_ADMIN_USER}:{DEFAULT_ADMIN_PASSWORD}@rabbitmq:15672/api/", "", ] ) admin_env = "\n".join( [ f"ADMIN_USERNAME={DEFAULT_ADMIN_USER}", f"ADMIN_PASSWORD={DEFAULT_ADMIN_PASSWORD}", "", ] ) postgres_path = env_dir / "postgres_environment.env" redis_path = env_dir / "redis_environment.env" rabbit_path = env_dir / "rabbitmq_environment.env" admin_path = env_dir / "admin.env" postgres_path.write_text(postgres_env) redis_path.write_text(redis_env) rabbit_path.write_text(rabbit_env) admin_path.write_text(admin_env) target_env_lines = [f" {safe_service_var(spec.name)}: host.docker.internal:{spec.host_port}" for spec in specs] target_env = "\n".join(target_env_lines) forcad_override = "\n".join( [ "version: '3.8'", "services:", " celery:", " user: root", " volumes:", f" - \"{checkers_dir}:/checkers/\"", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", " environment:", " TEST: \"\"", " SERVICE: worker", target_env, " extra_hosts:", " - \"host.docker.internal:host-gateway\"", " command: bash -lc \"pip install --no-cache-dir -r /checkers/requirements.txt && /entrypoint.sh\"", "", " flower:", " user: root", " volumes:", f" - \"{checkers_dir}:/checkers/\"", " env_file:", f" - \"{admin_path}\"", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", " environment:", " TEST: \"\"", " SERVICE: flower", target_env, " extra_hosts:", " - \"host.docker.internal:host-gateway\"", " command: bash -lc \"pip install --no-cache-dir -r /checkers/requirements.txt && /entrypoint.sh\"", "", " initializer:", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", " volumes:", f" - \"{config_path}:/config.yml:ro\"", " environment:", " TEST: \"\"", " CONFIG_PATH: /config.yml", "", " ticker:", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", "", " client-api:", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", "", " admin-api:", " env_file:", f" - \"{admin_path}\"", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", "", " events:", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", "", " http-receiver:", " env_file:", f" - \"{postgres_path}\"", f" - \"{redis_path}\"", f" - \"{rabbit_path}\"", "", " redis:", " env_file:", f" - \"{redis_path}\"", "", " rabbitmq:", " env_file:", f" - \"{rabbit_path}\"", "", " postgres:", " env_file:", f" - \"{postgres_path}\"", "", " nginx:", " ports:", f" - \"{board_port}:80\"", "", ] ) override_path = runtime_dir / "forcad.override.yml" override_path.write_text(forcad_override) return { "runtime_dir": str(runtime_dir), "config_path": str(config_path), "checkers_dir": str(checkers_dir), "override_path": str(override_path), } def build_checker_artifacts( runtime_dir: Path, specs: List[ServiceSpec], cache_dir: Path, failure_policy: str = "strict", ) -> Tuple[List[ServiceSpec], Dict[str, str]]: orchestrator_root = Path(__file__).resolve().parent cache_root = cache_dir / "checker-builds" cache_root.mkdir(parents=True, exist_ok=True) builder_images: Dict[str, str] = {} ready_specs: List[ServiceSpec] = [] failed: Dict[str, str] = {} for spec in specs: try: if not spec.checker_build_enabled: ready_specs.append(spec) continue if not spec.checker_build_profile or not spec.checker_build_command: raise RuntimeError(f"Invalid checker build config for `{spec.name}`.") checker_runtime_dir = runtime_dir / "checkers" / spec.name cache_key = hash_checker_inputs( checker_dir=checker_runtime_dir, build_profile=spec.checker_build_profile, build_command=spec.checker_build_command, ) if restore_build_cache(spec, checker_runtime_dir, cache_root, cache_key): ready_specs.append(spec) continue if spec.checker_build_profile not in builder_images: builder_images[spec.checker_build_profile] = resolve_builder_image( orchestrator_root=orchestrator_root, profile=spec.checker_build_profile, ) image = builder_images[spec.checker_build_profile] info(f"Containerized checker build for `{spec.name}` ({spec.checker_build_profile}).") run_command( [ "docker", "run", "--rm", "--add-host", "host.docker.internal:host-gateway", "-v", f"{checker_runtime_dir}:/work", "-w", "/work", image, "bash", "-lc", spec.checker_build_command, ] ) store_build_cache(spec, checker_runtime_dir, cache_root, cache_key) ready_specs.append(spec) except Exception as exc: # noqa: BLE001 message = str(exc) failed[spec.name] = message if failure_policy == "strict": raise warn(f"Skipping `{spec.name}` due checker build failure: {message}") return ready_specs, failed def is_port_open(port: int) -> bool: with socket.socket() as sock: sock.settimeout(0.5) return sock.connect_ex(("127.0.0.1", int(port))) == 0 def ensure_service_port_available(spec: ServiceSpec) -> None: if spec.host_port is None: return if not is_port_open(spec.host_port): return warn(f"Port {spec.host_port} is busy before deploying `{spec.name}`, trying cleanup.") stop_service(spec, include_legacy_compose=True) time.sleep(1) if is_port_open(spec.host_port): raise RuntimeError( f"Port {spec.host_port} is still busy before deploying `{spec.name}`. " "Run cleanup and retry." ) def prewarm_k3d_images( specs: List[ServiceSpec], strict: bool, image_pull_retries: int, image_pull_retry_delay: int, ) -> None: if not any(is_k3d_service(spec) for spec in specs): return for image in K3D_PREWARM_IMAGES: try: run_command_with_retries( ["docker", "pull", image], retries=image_pull_retries, retry_delay=image_pull_retry_delay, purpose=f"prewarm pull `{image}`", ) except Exception as exc: # noqa: BLE001 message = f"Failed to prewarm image `{image}`: {exc}" if strict: raise RuntimeError(message) from exc warn(message) def auto_cleanup_services(specs: List[ServiceSpec]) -> None: info("Auto-cleanup previous service runs.") for spec in reversed(specs): stop_service(spec, include_legacy_compose=True) def rollback_services(specs: List[ServiceSpec]) -> None: if not specs: return warn("Rolling back already deployed services.") for spec in reversed(specs): stop_service(spec, include_legacy_compose=True) def deploy_service_with_retry( spec: ServiceSpec, k3d_retries: int, retry_delay: int, image_pull_retries: int, image_pull_retry_delay: int, ) -> None: attempts = 1 + (k3d_retries if is_k3d_service(spec) else 0) for attempt in range(1, attempts + 1): try: ensure_service_port_available(spec) deploy_service( spec, image_pull_retries=image_pull_retries, image_pull_retry_delay=image_pull_retry_delay, ) return except Exception as exc: # noqa: BLE001 if attempt >= attempts: raise warn(f"Deploy `{spec.name}` attempt {attempt}/{attempts} failed: {exc}") stop_service(spec, include_legacy_compose=True) time.sleep(retry_delay) def compose_project_names(spec: ServiceSpec) -> List[str]: names: List[str] = [f"forcadsvc_{spec.name}"] legacy = spec.service_dir.name if legacy not in names: names.append(legacy) return names def is_k3d_service(spec: ServiceSpec) -> bool: if (spec.service_dir / "k3d-config.yaml").exists(): return True deploy_script = spec.service_dir / "deploy.sh" if deploy_script.exists(): return "k3d cluster create" in deploy_script.read_text() return False def service_priority(spec: ServiceSpec) -> Tuple[int, str]: # Start k3d-heavy services first while Docker daemon is still idle. if is_k3d_service(spec): return (0, spec.name) if spec.profile == "compose": return (1, spec.name) return (2, spec.name) def order_services_for_deploy(specs: List[ServiceSpec]) -> List[ServiceSpec]: return sorted(specs, key=service_priority) def deploy_service( spec: ServiceSpec, *, image_pull_retries: int = 0, image_pull_retry_delay: int = 5, ) -> None: if spec.start_cmd: run_shell(spec.start_cmd, cwd=spec.service_dir) return if spec.profile == "compose": compose_path = spec.service_dir / "docker-compose.yml" project = compose_project_names(spec)[0] # Pull externally referenced images with retries before "up". run_command_with_retries( [ "docker", "compose", "-f", str(compose_path), "-p", project, "pull", ], retries=image_pull_retries, retry_delay=image_pull_retry_delay, purpose=f"image pull for `{spec.name}`", ) run_command( [ "docker", "compose", "-f", str(compose_path), "-p", project, "up", "-d", "--build", "--remove-orphans", ] ) return deploy_script = spec.service_dir / "deploy.sh" if deploy_script.exists(): run_command(["bash", str(deploy_script)], cwd=spec.service_dir) return redeploy_script = spec.service_dir / "redeploy.sh" if redeploy_script.exists(): run_command(["bash", str(redeploy_script)], cwd=spec.service_dir) return build_script = spec.service_dir / "build-docker.sh" chart_dir = spec.service_dir / "deploy" / "chart" if build_script.exists(): run_command(["bash", str(build_script)], cwd=spec.service_dir) if chart_dir.exists(): run_command( [ "helm", "upgrade", "--install", spec.name, str(chart_dir), "--create-namespace", "--namespace", spec.name, "--wait", "--timeout", "10m", ], cwd=spec.service_dir, ) return raise RuntimeError(f"Не знаю, как поднять сервис `{spec.name}` (profile={spec.profile}).") def stop_service(spec: ServiceSpec, include_legacy_compose: bool = True) -> None: if spec.stop_cmd: run_shell(spec.stop_cmd, cwd=spec.service_dir, check=False) return if spec.profile == "compose": compose_path = spec.service_dir / "docker-compose.yml" projects = compose_project_names(spec) if not include_legacy_compose: projects = projects[:1] for project in projects: run_command( ["docker", "compose", "-f", str(compose_path), "-p", project, "down", "-v", "--remove-orphans"], check=False, ) return if (spec.service_dir / "k3d-config.yaml").exists() and shutil.which("k3d"): config_text = (spec.service_dir / "k3d-config.yaml").read_text() match = re.search(r"\bname:\s*([A-Za-z0-9_.-]+)", config_text) cluster_name = match.group(1) if match else f"{spec.name}-cluster" run_command(["k3d", "cluster", "delete", cluster_name], check=False) if shutil.which("helm"): run_command(["helm", "uninstall", spec.name, "-n", spec.name], check=False) def wait_for_service(spec: ServiceSpec, timeout_seconds: int = 180) -> None: if spec.host_port is None: raise RuntimeError(f"У `{spec.name}` не определён host_port.") info(f"Ожидаю доступность `{spec.name}` на порту {spec.host_port} ...") deadline = time.time() + timeout_seconds while time.time() < deadline: with socket.socket() as sock: sock.settimeout(2.0) try: sock.connect(("127.0.0.1", int(spec.host_port))) info(f"`{spec.name}` доступен на 127.0.0.1:{spec.host_port}") return except OSError: time.sleep(2) raise RuntimeError(f"`{spec.name}` не стал доступен на порту {spec.host_port} за {timeout_seconds} сек.") def forcad_compose_files(forcad_root: Path, override_path: Path, fast: bool) -> List[str]: files = [str(forcad_root / "docker-compose.yml")] if fast: files.append(str(forcad_root / "docker-compose-fast.yml")) files.append(str(override_path)) return files def run_forcad_up(forcad_root: Path, override_path: Path, fast: bool, workers: int) -> None: compose_files = forcad_compose_files(forcad_root, override_path, fast) command = ["docker", "compose"] for path in compose_files: command += ["-f", path] command += ["up", "-d", "--build", "--scale", f"celery={workers}"] run_command(command, cwd=forcad_root) def run_forcad_down(forcad_root: Path, override_path: Path, fast: bool) -> None: compose_files = forcad_compose_files(forcad_root, override_path, fast) command = ["docker", "compose"] for path in compose_files: command += ["-f", path] command += ["down", "-v", "--remove-orphans"] run_command(command, cwd=forcad_root, check=False) def run_forcad_status(forcad_root: Path, override_path: Path, fast: bool) -> None: compose_files = forcad_compose_files(forcad_root, override_path, fast) command = ["docker", "compose"] for path in compose_files: command += ["-f", path] command += ["ps"] run_command(command, cwd=forcad_root, check=False) def run_smoke_checks(forcad_root: Path, override_path: Path, fast: bool, specs: List[ServiceSpec]) -> None: compose_files = forcad_compose_files(forcad_root, override_path, fast) for spec in specs: command = ["docker", "compose"] for path in compose_files: command += ["-f", path] command += [ "exec", "-T", "celery", f"/checkers/{spec.name}/forcad_wrapper.py", "check", "host.docker.internal", ] info(f"Smoke-check `{spec.name}` ...") result = run_command(command, cwd=forcad_root, check=False) if result.returncode == 0: info(f"Smoke-check `{spec.name}`: OK") else: warn(f"Smoke-check `{spec.name}` завершился с кодом {result.returncode}.") def run_light_smoke_checks(runtime_dir: Path, specs: List[ServiceSpec]) -> None: checkers_dir = runtime_dir / "checkers" quoted_services = " ".join(shlex.quote(spec.name) for spec in specs) command = ( "python3 -m pip install --no-cache-dir -r /checkers/requirements.txt >/tmp/pip.log 2>&1 && " "failed=0; " f"for svc in {quoted_services}; do " "echo \"[LIGHT] checker check $svc\"; " "python3 /checkers/$svc/forcad_wrapper.py check host.docker.internal || failed=1; " "done; " "exit $failed" ) result = run_command( [ "docker", "run", "--rm", "--add-host", "host.docker.internal:host-gateway", "-v", f"{checkers_dir}:/checkers", "python:3.11-slim", "bash", "-lc", command, ], check=False, ) if result.returncode == 0: info("Light smoke-checks completed successfully.") else: warn(f"Light smoke-checks completed with non-zero status: {result.returncode}") def save_state(runtime_dir: Path, payload: dict) -> None: state_path = runtime_dir / "state.json" state_path.write_text(json.dumps(payload, indent=2)) def load_state(runtime_dir: Path) -> Optional[dict]: state_path = runtime_dir / "state.json" if not state_path.exists(): return None return json.loads(state_path.read_text()) def load_context_from_args(args: argparse.Namespace, allow_state: bool = False) -> Tuple[Path, Path, Path]: forcad_root = args.forcad_root.resolve() runtime_dir = args.runtime_dir.resolve() services_root = args.services_root.resolve() if args.services_root else None checkers_root = args.checkers_root.resolve() if args.checkers_root else None if services_root is None and allow_state: state = load_state(runtime_dir) if state: services_root = Path(state["services_root"]) checkers_root = Path(state["checkers_root"]) forcad_root = Path(state["forcad_root"]) if services_root is None: raise RuntimeError("Нужно указать `--services-root` или иметь сохранённый state от `up`.") if checkers_root is None: checkers_root = resolve_checkers_root(services_root, None) return forcad_root, services_root, checkers_root def cmd_validate(args: argparse.Namespace) -> int: forcad_root, services_root, checkers_root = load_context_from_args(args, allow_state=False) if not forcad_root.exists(): err(f"Не найден ForcAD root: {forcad_root}") return 2 selected = set(args.service) if args.service else None specs, discovery_report = discover_services(services_root, checkers_root, selected) validation_report = validate_preflight(specs, args.board_port) report = merge_reports(discovery_report, validation_report) print_report(report) if not report.ok(): return 2 info("Preflight-проверка пройдена.") for spec in specs: info( f"service={spec.name} profile={spec.profile} checker={spec.checker_entry} " f"type={spec.checker_type} places={spec.places} port={spec.host_port}" ) return 0 def cmd_up(args: argparse.Namespace) -> int: forcad_root, services_root, checkers_root = load_context_from_args(args, allow_state=False) selected = set(args.service) if args.service else None specs, discovery_report = discover_services(services_root, checkers_root, selected) validation_report = validate_preflight(specs, args.board_port) report = merge_reports(discovery_report, validation_report) print_report(report) if not report.ok(): return 2 runtime_dir = args.runtime_dir.resolve() runtime_meta = generate_runtime( runtime_dir=runtime_dir, specs=specs, team_count=args.team_count, board_port=args.board_port, round_time=args.round_time, flag_lifetime=args.flag_lifetime, start_delay_seconds=args.start_delay_seconds, ) failure_policy = args.failure_policy.lower() strict = failure_policy == "strict" ordered_specs = order_services_for_deploy(specs) if args.auto_cleanup: auto_cleanup_services(ordered_specs) if args.prewarm_k3d: prewarm_k3d_images( ordered_specs, strict=strict, image_pull_retries=args.image_pull_retries, image_pull_retry_delay=args.image_pull_retry_delay, ) ready_specs, build_failed = build_checker_artifacts( runtime_dir, ordered_specs, cache_dir=args.cache_dir.resolve(), failure_policy=failure_policy, ) if build_failed and strict: raise RuntimeError(f"Checker build failed: {build_failed}") if build_failed and not strict: warn(f"Checker build failures (best-effort): {build_failed}") if not ready_specs: raise RuntimeError("No services left after checker build stage.") deployed_specs: List[ServiceSpec] = [] deploy_failed: Dict[str, str] = {} for spec in ready_specs: try: deploy_service_with_retry( spec, k3d_retries=args.k3d_retries, retry_delay=args.k3d_retry_delay, image_pull_retries=args.image_pull_retries, image_pull_retry_delay=args.image_pull_retry_delay, ) wait_for_service(spec, timeout_seconds=args.service_wait_timeout) deployed_specs.append(spec) except Exception as exc: # noqa: BLE001 deploy_failed[spec.name] = str(exc) if strict: if args.rollback_on_failure: rollback_services(deployed_specs + [spec]) raise warn(f"Skipping `{spec.name}` after deploy failure: {exc}") if not deployed_specs: raise RuntimeError("No services were deployed successfully.") if deploy_failed and not strict: warn(f"Deploy failures (best-effort): {deploy_failed}") mode = args.mode.lower() override_path = Path(runtime_meta["override_path"]) if mode == "full": run_forcad_up(forcad_root, override_path, fast=not args.no_fast, workers=args.workers) info("Ожидание стабилизации ForcAD (20 секунд)...") time.sleep(20) run_smoke_checks(forcad_root, override_path, fast=not args.no_fast, specs=deployed_specs) else: info("Running in light mode: full ForcAD stack is not started.") run_light_smoke_checks(runtime_dir, deployed_specs) state = { "forcad_root": str(forcad_root), "services_root": str(services_root), "checkers_root": str(checkers_root), "mode": mode, "fast": not args.no_fast, "board_port": args.board_port, "runtime": runtime_meta, "failure_policy": failure_policy, "deploy_failed": deploy_failed, "build_failed": build_failed, "services": [ { "name": spec.name, "profile": spec.profile, "service_dir": str(spec.service_dir), "checker_dir": str(spec.checker_dir), "checker_build_enabled": spec.checker_build_enabled, "checker_build_profile": spec.checker_build_profile, "checker_build_command": spec.checker_build_command, "checker_build_artifacts": spec.checker_build_artifacts, "host_port": spec.host_port, "start_cmd": spec.start_cmd, "stop_cmd": spec.stop_cmd, } for spec in deployed_specs ], } save_state(runtime_dir, state) if mode == "full": info(f"Борда должна быть доступна по адресу http://127.0.0.1:{args.board_port}/") else: info("Light mode complete: services are up and checker liveness checks were executed.") info("Готово.") return 0 def cmd_down(args: argparse.Namespace) -> int: runtime_dir = args.runtime_dir.resolve() state = load_state(runtime_dir) if state is None and not args.services_root: err("Не найден state.json. Укажи `--services-root` и `--checkers-root`.") return 2 if state is not None: forcad_root = Path(state["forcad_root"]) override_path = Path(state["runtime"]["override_path"]) fast = bool(state.get("fast", True)) mode = str(state.get("mode", "light")).lower() specs = [ ServiceSpec( name=s["name"], profile=s["profile"], service_dir=Path(s["service_dir"]), checker_dir=Path(s["checker_dir"]), checker_entry=Path("checker.py"), checker_type="hackerdom", places=1, checker_timeout=30, protocol="unknown", checker_build_enabled=bool(s.get("checker_build_enabled", False)), checker_build_profile=s.get("checker_build_profile"), checker_build_command=s.get("checker_build_command"), checker_build_artifacts=[str(x) for x in s.get("checker_build_artifacts", [])], host_port=s.get("host_port"), start_cmd=s.get("start_cmd"), stop_cmd=s.get("stop_cmd"), ) for s in state["services"] ] else: forcad_root, services_root, checkers_root = load_context_from_args(args, allow_state=False) discovered, report = discover_services(services_root, checkers_root, set(args.service) if args.service else None) print_report(report) if not discovered: return 2 specs = discovered override_path = runtime_dir / "forcad.override.yml" fast = not args.no_fast mode = args.mode.lower() if mode == "full": run_forcad_down(forcad_root, override_path, fast=fast) for spec in specs: stop_service(spec) info("Остановка завершена.") return 0 def cmd_status(args: argparse.Namespace) -> int: runtime_dir = args.runtime_dir.resolve() state = load_state(runtime_dir) if state is None: err("Не найден state.json. Сначала выполни `up`.") return 2 forcad_root = Path(state["forcad_root"]) override_path = Path(state["runtime"]["override_path"]) fast = bool(state.get("fast", True)) mode = str(state.get("mode", "light")).lower() if mode == "full": run_forcad_status(forcad_root, override_path, fast=fast) else: info("Light mode state: ForcAD stack was not started.") for service in state["services"]: port = service.get("host_port") if not port: warn(f"{service['name']}: порт неизвестен.") continue with socket.socket() as sock: sock.settimeout(1.0) try: sock.connect(("127.0.0.1", int(port))) info(f"{service['name']}: 127.0.0.1:{port} открыт") except OSError: warn(f"{service['name']}: 127.0.0.1:{port} недоступен") return 0 def build_parser() -> argparse.ArgumentParser: default_workspace = Path(__file__).resolve().parents[2] default_forcad = default_workspace / "ForcAD" parser = argparse.ArgumentParser( description="Локальный оркестратор ForcAD: auto-discovery сервисов, preflight, up/down/status." ) subparsers = parser.add_subparsers(dest="command", required=True) def add_common_flags(subparser: argparse.ArgumentParser, services_required: bool) -> None: subparser.add_argument("--forcad-root", type=Path, default=default_forcad) subparser.add_argument("--services-root", type=Path, required=services_required) subparser.add_argument("--checkers-root", type=Path) subparser.add_argument("--runtime-dir", type=Path, default=Path(__file__).resolve().parent / ".runtime") subparser.add_argument("--service", action="append", help="Имя сервиса (можно повторять). Если не задано, берутся все.") subparser.add_argument("--board-port", type=int, default=8080) subparser.add_argument("--no-fast", action="store_true", help="Не использовать docker-compose-fast.yml для ForcAD.") validate = subparsers.add_parser("validate", help="Проверка структуры сервисов/чекеров без запуска.") add_common_flags(validate, services_required=True) validate.set_defaults(handler=cmd_validate) up = subparsers.add_parser("up", help="Поднять сервисы, борду и запустить smoke-checks.") add_common_flags(up, services_required=True) up.add_argument("--mode", choices=["light", "full"], default="light", help="light: без ForcAD stack; full: с полным ForcAD stack.") up.add_argument("--cache-dir", type=Path, default=Path(__file__).resolve().parent / ".cache") up.add_argument("--failure-policy", choices=["strict", "best-effort"], default="strict") up.add_argument("--rollback-on-failure", action=argparse.BooleanOptionalAction, default=True) up.add_argument("--auto-cleanup", action=argparse.BooleanOptionalAction, default=True) up.add_argument("--prewarm-k3d", action=argparse.BooleanOptionalAction, default=True) up.add_argument("--k3d-retries", type=int, default=1) up.add_argument("--k3d-retry-delay", type=int, default=5) up.add_argument( "--image-pull-retries", type=int, default=2, help="Additional retries for docker image pulls (compose/prewarm).", ) up.add_argument( "--image-pull-retry-delay", type=int, default=10, help="Delay (seconds) between docker pull retries.", ) up.add_argument("--team-count", type=int, default=2) up.add_argument("--workers", type=int, default=2) up.add_argument("--round-time", type=int, default=120) up.add_argument("--flag-lifetime", type=int, default=5) up.add_argument("--start-delay-seconds", type=int, default=120) up.add_argument("--service-wait-timeout", type=int, default=180) up.set_defaults(handler=cmd_up) down = subparsers.add_parser("down", help="Остановить ForcAD и игровые сервисы.") add_common_flags(down, services_required=False) down.add_argument("--mode", choices=["light", "full"], default="light", help="Используется если state.json отсутствует.") down.set_defaults(handler=cmd_down) status = subparsers.add_parser("status", help="Показать статус запущенной инфраструктуры.") status.add_argument("--runtime-dir", type=Path, default=Path(__file__).resolve().parent / ".runtime") status.set_defaults(handler=cmd_status) return parser def main() -> int: parser = build_parser() args = parser.parse_args() try: return int(args.handler(args)) except KeyboardInterrupt: err("Прервано пользователем.") return 130 except Exception as exc: # noqa: BLE001 err(str(exc)) return 1 if __name__ == "__main__": sys.exit(main())