local-ci-cd-system/src/ci_orchestrator/ip_pool.py

"""IP slot pool — pre-assigns static IPs to build VMs via VMX guestinfo.

When ``[ip_pool]`` is configured, the orchestrator allocates one IP from the
pool before starting each VM, injects it into the cloned VMX as::

    guestinfo.ip-assignment = "192.168.79.201"
    guestinfo.netmask       = "255.255.255.0"
    guestinfo.gateway       = "192.168.79.2"

and releases it on VM destroy.  The guest applies the static IP at boot:

* **Linux**: ``/usr/local/bin/ci-report-ip.sh`` (updated version in
  ``template/guest-setup/linux/``) reads ``guestinfo.ip-assignment`` and
  configures a static NIC address before ``network-online.target`` is reached.

* **Windows**: the ``CI-StaticIp`` Task Scheduler task
  (``template/guest-setup/windows/ci-static-ip.ps1``) runs at system startup
  as SYSTEM and applies the static IP before WinRM starts.

This eliminates the DHCP / VMware-Tools IP-detect polling loop (20–180 s
variance observed in B7 benchmarks).  After the VM reaches *running* state the
host knows the IP immediately and goes straight to the transport probe.

Lease file — JSON dict mapping each pool IP to its current owner string, or
``null`` if free::

    {
      "192.168.79.201": "Clone_job123_20260524_abc123",
      "192.168.79.202": null,
      "192.168.79.203": null,
      "192.168.79.204": null
    }

File locking: ``fcntl.flock`` (POSIX/Linux); no-op on Windows (low risk —
pool size matches concurrency, failures are detectable via transport probe).
"""

from __future__ import annotations

import json
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Generator


class IpSlotPool:
    """Process-safe IP slot allocator backed by a JSON lease file."""

    def __init__(
        self,
        addresses: list[str],
        netmask: str,
        gateway: str,
        lease_file: Path,
    ) -> None:
        if not addresses:
            raise ValueError("IpSlotPool requires at least one address.")
        self._addresses = list(addresses)
        self.netmask = netmask
        self.gateway = gateway
        self._lease_file = lease_file
        self._lock_file = lease_file.with_suffix(".lock")

    # ── public API ─────────────────────────────────────────────────────────

    def acquire(self, owner: str, timeout: float = 60.0) -> str:
        """Allocate a free IP slot.  Blocks up to *timeout* seconds.

        Raises :exc:`RuntimeError` if the pool stays exhausted past *timeout*.
        """
        deadline = time.monotonic() + timeout
        while True:
            with _file_lock(self._lock_file):
                leases = self._read_leases()
                for ip in self._addresses:
                    if leases.get(ip) is None:
                        leases[ip] = owner
                        self._write_leases(leases)
                        return ip
            if time.monotonic() >= deadline:
                raise RuntimeError(
                    f"IP pool exhausted after {timeout:.0f}s; "
                    f"addresses={self._addresses}"
                )
            time.sleep(2.0)

    def release(self, ip: str) -> None:
        """Return *ip* to the pool (no-op if not in pool)."""
        with _file_lock(self._lock_file):
            leases = self._read_leases()
            if ip in leases:
                leases[ip] = None
                self._write_leases(leases)

    # ── internals ──────────────────────────────────────────────────────────

    def _read_leases(self) -> dict[str, str | None]:
        result: dict[str, str | None] = {ip: None for ip in self._addresses}
        if not self._lease_file.exists():
            return result
        try:
            raw: object = json.loads(self._lease_file.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            return result
        if not isinstance(raw, dict):
            return result
        for ip in self._addresses:
            val = raw.get(ip)
            result[ip] = val if isinstance(val, str) else None
        return result

    def _write_leases(self, leases: dict[str, str | None]) -> None:
        self._lease_file.parent.mkdir(parents=True, exist_ok=True)
        tmp = self._lease_file.with_suffix(".tmp")
        tmp.write_text(json.dumps(leases, indent=2) + "\n", encoding="utf-8")
        tmp.replace(self._lease_file)


@contextmanager
def _file_lock(path: Path) -> Generator[None, None, None]:
    """Exclusive advisory lock on *path* via ``fcntl.flock`` (POSIX only)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w") as fd:
        try:
            import fcntl
            fcntl.flock(fd, fcntl.LOCK_EX)
        except ImportError:
            pass
        try:
            yield
        finally:
            try:
                import fcntl
                fcntl.flock(fd, fcntl.LOCK_UN)
            except ImportError:
                pass


__all__ = ["IpSlotPool"]