Service 给一组 Pod 一个稳定的虚拟地址

"""用小型服务网段模拟 ClusterIP 分配。"""

import ipaddress


class ClusterIPPool:
    """分配虚拟地址；无头 Service 绕过地址池。"""

    def __init__(self, cidr, reserved=1):
        network = ipaddress.ip_network(cidr)
        hosts = list(network.hosts())
        self._all = set(hosts)
        self._reserved = set(hosts[:reserved])
        self.free = sorted(set(hosts) - self._reserved)
        self.allocated = {}

    def allocate(self, service, requested=None, headless=False):
        if headless:
            return None
        if service in self.allocated:
            raise ValueError(f"{service} 已持有 {self.allocated[service]}")
        if requested is not None:
            ip = ipaddress.ip_address(requested)
            if ip in self._reserved or ip not in self._all:
                raise ValueError(f"{requested} 超出可分配范围")
            if ip not in self.free:
                raise ValueError(f"{requested} 已被占用")
            self.free.remove(ip)
            self.allocated[service] = ip
            return str(ip)
        if not self.free:
            raise RuntimeError("服务网段耗尽：没有可用 ClusterIP")
        ip = self.free.pop(0)
        self.allocated[service] = ip
        return str(ip)

    def release(self, service):
        ip = self.allocated.pop(service, None)
        if ip is None:
            raise ValueError(f"{service} 没有可释放的地址")
        self.free.append(ip)
        self.free.sort()
        return str(ip)


def check(condition, message):
    if not condition:
        raise AssertionError(message)


def main():
    pool = ClusterIPPool("10.96.0.0/29", reserved=1)
    web = pool.allocate("web")
    api = pool.allocate("api", requested="10.96.0.5")
    check(web != api, "顺序创建的服务必须拿到不同地址")
    print("allocated:", web, api)

    try:
        pool.allocate("worker", requested="10.96.0.5")
        raise SystemExit("冲突被悄悄共享了")
    except ValueError as exc:
        print("conflict rejected:", exc)

    try:
        pool.allocate("evil", requested="10.99.0.1")
        raise SystemExit("越界申请被接受了")
    except ValueError as exc:
        print("out-of-range rejected:", exc)

    headless = pool.allocate("db-headless", headless=True)
    check(headless is None, "无头服务必须绕过地址池")
    check(len(pool.allocated) == 2, "无头服务不得占用地址池")
    print("headless bypass: None")

    freed = pool.release("web")
    worker = pool.allocate("worker")
    check(worker == freed, "释放的地址必须可复用")
    print("reused:", freed, "-> worker")

    pool.allocate("extra-1")
    pool.allocate("extra-2")
    pool.allocate("extra-3")
    try:
        pool.allocate("overflow")
        raise SystemExit("耗尽没有报错")
    except RuntimeError as exc:
        print("exhaustion reported:", exc)

    print("PASS: allocation, conflict rejection, release reuse, and exhaustion hold")


if __name__ == "__main__":
    main()