From 3af436d48a1461c2054c42004c2ba5be31e34df0 Mon Sep 17 00:00:00 2001 From: Timofey Gulenok Date: Wed, 12 Aug 2026 21:11:39 +0300 Subject: [PATCH] little update --- .env.example | 20 ++ .gitignore | 8 + README.md | 98 ++++++ alembic.ini | 35 ++ alembic/env.py | 48 +++ .../20260812_01_create_network_pool.py | 32 ++ kernel.proto | 20 +- kernel/src/scripts/init_nats.py | 57 ++++ pyproject.toml | 23 ++ requirements.txt | 6 + scripts_live_subnet_test.py | 57 ++++ src/cloud/__init__.py | 56 ++++ src/cloud/iam.py | 53 +++ src/cloud/models.py | 254 ++++++++++++++ src/cloud/vm_manager.py | 315 ++++++++++++++++++ src/cloud/vpc_network_manager.py | 180 ++++++++++ src/database/__init__.py | 3 + src/database/models.py | 21 ++ src/database/session.py | 6 + src/kernel_consumer/__init__.py | 1 + src/kernel_consumer/main.py | 297 +++++++++++++++++ src/models/Vlan.py | 9 + src/models/__init__.py | 0 src/networking/__init__.py | 3 + src/networking/manager.py | 123 +++++++ src/scripts/__init__.py | 0 src/scripts/fire_game_cleanup.py | 44 +++ src/scripts/init_nats.py | 60 ++++ src/scripts/start_game_debug.py | 60 ++++ tests/test_environment.py | 23 ++ tests/test_iam.py | 26 ++ tests/test_kernel_consumer.py | 46 +++ tests/test_networking_manager.py | 151 +++++++++ tests/test_vm_api_errors.py | 10 + tests/test_vm_manager.py | 102 ++++++ tests/test_vpc_network_manager.py | 49 +++ token.json | 12 + 37 files changed, 2299 insertions(+), 9 deletions(-) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 alembic.ini create mode 100644 alembic/env.py create mode 100644 alembic/versions/20260812_01_create_network_pool.py create mode 100644 kernel/src/scripts/init_nats.py create mode 100644 pyproject.toml create mode 100644 requirements.txt create mode 100644 scripts_live_subnet_test.py create mode 100644 src/cloud/__init__.py create mode 100644 src/cloud/iam.py create mode 100644 src/cloud/models.py create mode 100644 src/cloud/vm_manager.py create mode 100644 src/cloud/vpc_network_manager.py create mode 100644 src/database/__init__.py create mode 100644 src/database/models.py create mode 100644 src/database/session.py create mode 100644 src/kernel_consumer/__init__.py create mode 100644 src/kernel_consumer/main.py create mode 100644 src/models/Vlan.py create mode 100644 src/models/__init__.py create mode 100644 src/networking/__init__.py create mode 100644 src/networking/manager.py create mode 100644 src/scripts/__init__.py create mode 100644 src/scripts/fire_game_cleanup.py create mode 100644 src/scripts/init_nats.py create mode 100644 src/scripts/start_game_debug.py create mode 100644 tests/test_environment.py create mode 100644 tests/test_iam.py create mode 100644 tests/test_kernel_consumer.py create mode 100644 tests/test_networking_manager.py create mode 100644 tests/test_vm_api_errors.py create mode 100644 tests/test_vm_manager.py create mode 100644 tests/test_vpc_network_manager.py create mode 100644 token.json diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..c59847d --- /dev/null +++ b/.env.example @@ -0,0 +1,20 @@ +NATS_URL=nats://127.0.0.1:4222 +NATS_STREAM=GAME +NATS_SUBJECT=game.start +NATS_CLEANUP_SUBJECT=game.cleanup +NATS_CONSUMER=kernel-consumer +DATABASE_URL=sqlite:///kernel.db +CLOUD_TOKEN= +CLOUD_KEY_ID= +CLOUD_SECRET= +CLOUD_PROJECT_ID= +CLOUD_VPC_ID= +CLOUD_AVAILABILITY_ZONE_ID= +CLOUD_AVAILABILITY_ZONE_NAME= +CLOUD_VM_FLAVOR_NAME=low-1-1 +CLOUD_VM_IMAGE_NAME=ubuntu-22.04 +CLOUD_VM_DISK_TYPE_NAME=SSD +CLOUD_VM_USERNAME=user1 +CLOUD_VM_PASSWORD= +CLOUD_VM_PUBLIC_KEY= +LOG_LEVEL=INFO diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cf5a7db --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.env +.env.* +!.env.example +*.db +__pycache__/ +.pytest_cache/ +.idea/ +kernel.db \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..ec1676e --- /dev/null +++ b/README.md @@ -0,0 +1,98 @@ +# Kernel NATS consumer + +Consumes JSON `StartGameRequest` messages. `kernel.proto` is used only as field schema reference; no protobuf runtime or encoding is used. + +Expected message shape: + +```json +{ + "team_count": 2, + "services": [], + "teams": [{"team_name": "red", "team_id": "red-1"}, {"team_name": "blue", "team_id": "blue-1"}], + "game_starts_at": 1730000000, + "rounds": 10, + "network_opens_at": 1729999900 +} +``` + +Run from `kernel/`: + +```sh +python -m pip install -e . +init-nats +NATS_URL=nats://127.0.0.1:4222 kernel-consumer +``` + +`init-nats` idempotently creates stream `GAME` and its `game.start` channel. It also adds configured `NATS_SUBJECT` to existing configured `NATS_STREAM`. + +Environment variables: `NATS_URL`, `NATS_STREAM` (`GAME`), `NATS_SUBJECT` (`game.start`), `NATS_CONSUMER` (`kernel-consumer`). + +## Cloud.ru Evolution VPC + +`VPCNetworkManager` is a synchronous client for documented Evolution VPC API operations. Pass a Cloud.ru Bearer JWT directly; do not commit it. + +```python +from cloud import CreateVPCRequest, VPCNetworkManager + +manager = VPCNetworkManager(token="") +operation = manager.create_vpc( + CreateVPCRequest(projectId="", name="game-vpc") +) +vpcs = manager.list_vpcs(project_id="") +``` + +It supports VPC CRUD, operation lookup, and static-route read/create/delete. Evolution VPC public API does not expose endpoints for listing or creating subnets. + +## Cloud.ru Virtual Machines + +`VMManager` is a synchronous client for documented Evolution Virtual Machines API methods. Pass a Cloud.ru Bearer JWT directly; do not commit it. + +```python +from cloud import VMCreateRequest, VMManager + +manager = VMManager(token="") +vms = manager.list_vms(project_id="") +vms = manager.create_vm(VMCreateRequest( + project_id="", + name="game-vm", + disks=[{"disk_id": ""}], +)) +manager.start_vm(vms[0].id) +``` + +It supports VM list/create/read/update/delete, batch creation, start/stop, power actions, rebuild, VNC and remote-console creation, password management, and guest-agent checks. Managers accept Pydantic request models and return validated Pydantic response models. `VMCreateRequest` requires `project_id`, `name`, and non-empty `disks`. + +## Network Pool + +Copy environment template and set Cloud.ru credentials: + +```sh +cp .env.example .env +``` + +Run database migrations before starting consumer: + +```sh +DATABASE_URL="sqlite:///$PWD/kernel.db" PYTHONPATH=src alembic upgrade head +``` + +`.env` loads automatically for consumer and Alembic. Shell environment variables override `.env`. Set `DATABASE_URL`, `CLOUD_PROJECT_ID`, `CLOUD_KEY_ID`, and `CLOUD_SECRET`. Consumer requests fresh Cloud.ru JWT from IAM at startup; it uses `CLOUD_TOKEN` only when IAM credentials are absent. Set `CLOUD_VPC_ID` and `CLOUD_AVAILABILITY_ZONE_ID` (UUID) or `CLOUD_AVAILABILITY_ZONE_NAME` (resolved to an ID via `GET /api/v1/availability-zones`) when project defaults do not select VPC and availability zone. The subnet API expects the zone as a flat `availability_zone_id` UUID. Without `DATABASE_URL`, consumer uses absolute `kernel/kernel.db`, independent of launch folder. Alembic uses `DATABASE_URL` when set. For every valid game message, consumer creates UUID and allocates network through `NetworkingManager`. Free pool networks are verified using Cloud.ru subnet API before allocation. Missing cloud subnet records are removed and recreated as `10.N.0.0/20`, where `N` ranges from `0` to `240` in steps of 16. `NetworkPoolExhaustedError` stops allocation when all CIDRs are used. + +For each valid game event, the consumer creates `team_count + 1` Ubuntu VMs. Every VM is attached to the allocated game subnet; the extra VM receives a new public IPv4 address. Configure `CLOUD_VM_PASSWORD` or `CLOUD_VM_PUBLIC_KEY` for Ubuntu authentication, plus optional `CLOUD_VM_FLAVOR_NAME`, `CLOUD_VM_IMAGE_NAME`, and `CLOUD_VM_DISK_TYPE_NAME` overrides. + +## Game Cleanup + +Initialize both JetStream subjects with `init-nats`, then queue cleanup for a game: + +```sh +init-nats +fire-game-cleanup 4bc5200f-09ba-4e18-8da4-64f3950d342e +``` + +The cleanup message is published to `NATS_CLEANUP_SUBJECT` (`game.cleanup` by default): + +```json +{"game_id": "4bc5200f-09ba-4e18-8da4-64f3950d342e"} +``` + +The consumer finds VMs named for that game, deletes them, waits until Cloud.ru removes them, and only then releases the game subnet back to the pool. Cleanup is idempotent if the VMs or network were already removed. diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..4da2d45 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,35 @@ +[alembic] +script_location = alembic +sqlalchemy.url = sqlite:///kernel.db + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..d3a7916 --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,48 @@ +from logging.config import fileConfig +import os +from pathlib import Path +import sys + +from alembic import context +from dotenv import load_dotenv +from sqlalchemy import engine_from_config, pool + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT / "src")) +load_dotenv(PROJECT_ROOT / ".env") + +from database.models import Base + +config = context.config +if database_url := os.getenv("DATABASE_URL"): + config.set_main_option("sqlalchemy.url", database_url) +if config.config_file_name: + fileConfig(config.config_file_name) +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + context.configure( + url=config.get_main_option("sqlalchemy.url"), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool + ) + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/versions/20260812_01_create_network_pool.py b/alembic/versions/20260812_01_create_network_pool.py new file mode 100644 index 0000000..6c136c5 --- /dev/null +++ b/alembic/versions/20260812_01_create_network_pool.py @@ -0,0 +1,32 @@ +"""create network pool + +Revision ID: 20260812_01 +Revises: +Create Date: 2026-08-12 +""" + +from alembic import op +import sqlalchemy as sa + +revision = "20260812_01" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "network_pool", + sa.Column("cidr", sa.String(length=18), nullable=False), + sa.Column("cloud_subnet_id", sa.String(length=64), nullable=False), + sa.Column("game_id", sa.Uuid(), nullable=True), + sa.Column("allocated_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False), + sa.PrimaryKeyConstraint("cidr"), + sa.UniqueConstraint("cloud_subnet_id"), + sa.UniqueConstraint("game_id"), + ) + + +def downgrade() -> None: + op.drop_table("network_pool") diff --git a/kernel.proto b/kernel.proto index 466c489..73ea0b3 100644 --- a/kernel.proto +++ b/kernel.proto @@ -1,3 +1,5 @@ +syntax = "proto3"; + message Team { string team_name = 1; string team_id = 2; @@ -10,22 +12,22 @@ message GameService { //board config string checker_type =3; - int gets = 4; + int32 gets = 4; string name = 5; - int places = 6; - int puts =7; - int default_score = 8; - int checker_timeout = 9; + int32 places = 6; + int32 puts =7; + int32 default_score = 8; + int32 checker_timeout = 9; } // all time is in unixtime message StartGameRequest { - int teams = 1; + int32 team_count = 1; repeated GameService services = 2; repeated Team teams = 3; - int game_starts_at = 4; - int rounds = 5; - int network_opens_at = 6; + int32 game_starts_at = 4; + int32 rounds = 5; + int32 network_opens_at = 6; } diff --git a/kernel/src/scripts/init_nats.py b/kernel/src/scripts/init_nats.py new file mode 100644 index 0000000..98fd074 --- /dev/null +++ b/kernel/src/scripts/init_nats.py @@ -0,0 +1,57 @@ +import argparse +import asyncio +import os + +import nats +from nats.js.api import StorageType, StreamConfig +from nats.js.errors import NotFoundError + + +async def ensure_stream(url: str, stream: str, subject: str) -> None: + nc = await nats.connect(url) + try: + js = nc.jetstream() + try: + stream_info = await js.stream_info(stream) + except NotFoundError: + await js.add_stream( + StreamConfig( + name=stream, + subjects=[subject], + storage=StorageType.FILE, + ) + ) + print(f"created stream {stream} for subject {subject}") + return + + subjects = set(stream_info.config.subjects or []) + if subject in subjects: + print(f"stream {stream} already contains subject {subject}") + return + + subjects.add(subject) + config = stream_info.config + config.subjects = sorted(subjects) + await js.update_stream(config) + print(f"added subject {subject} to stream {stream}") + finally: + await nc.drain() + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Create NATS JetStream channels required by kernel." + ) + parser.add_argument( + "--url", + default=os.getenv("NATS_URL", "nats://127.0.0.1:4222"), + ) + parser.add_argument("--stream", default=os.getenv("NATS_STREAM", "GAME")) + parser.add_argument("--subject", default=os.getenv("NATS_SUBJECT", "game.start")) + args = parser.parse_args() + + asyncio.run(ensure_stream(args.url, args.stream, args.subject)) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1056830 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "kernel-consumer" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = [ + "nats-py>=2.9,<3", + "pydantic>=2.9,<3", + "SQLAlchemy>=2.0,<3", + "alembic>=1.14,<2", + "python-dotenv>=1.0,<2", +] + +[project.scripts] +kernel-consumer = "kernel_consumer.main:main" +init-nats = "scripts.init_nats:main" +fire-game-cleanup = "scripts.fire_game_cleanup:main" + +[tool.pytest.ini_options] +pythonpath = ["src"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..23b08a0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +nats-py>=2.9,<3 +pydantic>=2.9,<3 +SQLAlchemy>=2.0,<3 +alembic>=1.14,<2 +python-dotenv>=1.0,<2 +pytest>=8,<9 diff --git a/scripts_live_subnet_test.py b/scripts_live_subnet_test.py new file mode 100644 index 0000000..06ad3a8 --- /dev/null +++ b/scripts_live_subnet_test.py @@ -0,0 +1,57 @@ +"""Live smoke test: list, create, verify, delete subnet on real Cloud.ru.""" +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) + +from dotenv import load_dotenv +load_dotenv(Path(__file__).resolve().parent / ".env") + +import os +from cloud.vm_manager import VMManager, VMAPIError +from cloud.models import SubnetCreateRequest + +token = os.environ["CLOUD_TOKEN"] +project_id = os.environ["CLOUD_PROJECT_ID"] +vpc_id = os.getenv("CLOUD_VPC_ID") +zone_id = "479a4ab3-3ff3-4972-95c5-7610bac5c0bb" # ru.AZ-2, from list_subnets + +m = VMManager(token) +print("== list_subnets (before) ==") +try: + existing = m.list_subnets(project_id) + print(json.dumps(existing.model_dump(), indent=2)[:2000]) +except VMAPIError as e: + print(f"LIST FAILED: status={e.status} msg={e} details={e.details}") + +print("\n== create_subnet ==") +created = None +try: + created = m.create_subnet(SubnetCreateRequest( + name="kernel-live-test-subnet", + project_id=project_id, + availability_zone_id=zone_id, + subnet_address="10.250.0.0/20", + vpc_id=vpc_id, + description="kernel live test, safe to delete", + )) + print(f"CREATED: id={created.id} name={created.name} addr={created.subnet_address} state={created.state}") +except VMAPIError as e: + print(f"CREATE FAILED: status={e.status} msg={e}") + print(f"details={json.dumps(e.details)[:2000]}") + sys.exit(1) + +print("\n== get_subnet ==") +try: + got = m.get_subnet(created.id) + print(f"GET OK: id={got.id} addr={got.subnet_address} state={got.state}") +except VMAPIError as e: + print(f"GET FAILED: status={e.status} msg={e} details={json.dumps(e.details)[:1000]}") + +print("\n== delete_subnet ==") +try: + m._request("DELETE", f"/api/v1/subnets/{created.id}") + print("DELETE OK") +except VMAPIError as e: + print(f"DELETE FAILED: status={e.status} msg={e} details={json.dumps(e.details)[:1000]}") diff --git a/src/cloud/__init__.py b/src/cloud/__init__.py new file mode 100644 index 0000000..1ab8334 --- /dev/null +++ b/src/cloud/__init__.py @@ -0,0 +1,56 @@ +from .vpc_network_manager import VPCAPIError, VPCNetworkManager +from .vm_manager import VMAPIError, VMManager +from .iam import IAMAPIError, IAMClient +from .models import ( + AvailabilityZone, + CreateStaticRouteRequest, + CreateVPCRequest, + ListStaticRoutesResponse, + ListVMsResponse, + ListVPCsResponse, + Operation, + RemoteConsoleRequest, + RouteNextHop, + StaticRoute, + Subnet, + SubnetCreateRequest, + VM, + VMCreateRequest, + VMDeletePasswordRequest, + VMDeleteRequest, + VMPasswordRequest, + VMPowerRequest, + VMRebuildRequest, + VMStateRequest, + VMUpdateRequest, +) + +__all__ = [ + "AvailabilityZone", + "CreateStaticRouteRequest", + "CreateVPCRequest", + "ListStaticRoutesResponse", + "ListVMsResponse", + "ListVPCsResponse", + "Operation", + "RemoteConsoleRequest", + "RouteNextHop", + "StaticRoute", + "Subnet", + "SubnetCreateRequest", + "VMAPIError", + "VM", + "VMCreateRequest", + "VMDeletePasswordRequest", + "VMDeleteRequest", + "VMManager", + "VMPasswordRequest", + "VMPowerRequest", + "VMRebuildRequest", + "VMStateRequest", + "VMUpdateRequest", + "VPCAPIError", + "VPCNetworkManager", + "IAMAPIError", + "IAMClient", +] diff --git a/src/cloud/iam.py b/src/cloud/iam.py new file mode 100644 index 0000000..ad3069d --- /dev/null +++ b/src/cloud/iam.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import json +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +class IAMAPIError(RuntimeError): + """Cloud.ru IAM API request failed.""" + + +class IAMClient: + TOKEN_URL = "https://iam.api.cloud.ru/api/v1/auth/token" + + def __init__(self, key_id: str, secret: str, *, timeout: float = 30) -> None: + if not key_id: + raise ValueError("key_id must not be empty") + if not secret: + raise ValueError("secret must not be empty") + self._key_id = key_id + self._secret = secret + self._timeout = timeout + + def get_access_token(self) -> str: + request = Request( + self.TOKEN_URL, + data=json.dumps({"keyId": self._key_id, "secret": self._secret}).encode(), + method="POST", + headers={"Accept": "application/json", "Content-Type": "application/json"}, + ) + try: + with urlopen(request, timeout=self._timeout) as response: + payload = self._decode(response.read()) + except HTTPError as error: + raise IAMAPIError(f"IAM token request failed: HTTP {error.code}") from error + except URLError as error: + raise IAMAPIError(f"IAM token request failed: {error.reason}") from error + + token = payload.get("access_token") + if not isinstance(token, str) or not token: + raise IAMAPIError("IAM token response has no access_token") + return token + + @staticmethod + def _decode(data: bytes) -> dict[str, Any]: + try: + payload = json.loads(data) + except json.JSONDecodeError as error: + raise IAMAPIError("IAM token response is not JSON") from error + if not isinstance(payload, dict): + raise IAMAPIError("IAM token response is not an object") + return payload diff --git a/src/cloud/models.py b/src/cloud/models.py new file mode 100644 index 0000000..ffd424c --- /dev/null +++ b/src/cloud/models.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +class CloudRequest(BaseModel): + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + +class CloudResponse(BaseModel): + model_config = ConfigDict(extra="allow", populate_by_name=True) + + +class APIStatus(CloudResponse): + code: int | str | None = None + message: str | None = None + details: list[Any] | None = None + + +class Operation(CloudResponse): + id: str + resource_name: str | None = Field(default=None, alias="resourceName") + resource_id: str | None = Field(default=None, alias="resourceId") + done: bool | None = None + description: str | None = None + error: APIStatus | None = None + + +class CreateVPCRequest(CloudRequest): + project_id: str = Field(alias="projectId") + name: str = Field(min_length=1, max_length=64) + description: str | None = Field(default=None, max_length=255) + + +class UpdateVPCRequest(CloudRequest): + vpc_id: str = Field(alias="vpcId") + name: str | None = Field(default=None, min_length=1, max_length=64) + description: str | None = Field(default=None, max_length=255) + + @model_validator(mode="after") + def require_change(self) -> "UpdateVPCRequest": + if self.name is None and self.description is None: + raise ValueError("name or description must be set") + return self + + +class VPC(CloudResponse): + id: str + name: str + project_id: str = Field(alias="projectId") + description: str | None = None + is_default: bool | None = Field(default=None, alias="isDefault") + + +class ListVPCsResponse(CloudResponse): + vpcs: list[VPC] = Field(default_factory=list) + next_page_token: str | None = Field(default=None, alias="nextPageToken") + + +class MagicRouterNextHop(CloudRequest): + magic_router: dict[str, str] = Field(alias="magicRouter") + az_name: str = Field(alias="azName") + + +class VirtualMachineNextHop(CloudRequest): + virtual_machine: dict[str, str] = Field(alias="virtualMachine") + interface: dict[str, str] + + +class VIPNextHop(CloudRequest): + vips: list[dict[str, str]] = Field(min_length=1) + + +class RouteNextHop(CloudRequest): + magic_router: MagicRouterNextHop | None = Field(default=None, alias="magicRouter") + virtual_machine: VirtualMachineNextHop | None = Field(default=None, alias="virtualMachine") + vip: VIPNextHop | None = None + + @model_validator(mode="after") + def require_single_target(self) -> "RouteNextHop": + if sum(value is not None for value in (self.magic_router, self.virtual_machine, self.vip)) != 1: + raise ValueError("exactly one next-hop target must be set") + return self + + +class CreateStaticRouteRequest(CloudRequest): + vpc_id: str = Field(alias="vpcId") + project_id: str = Field(alias="projectId") + subnet: str + next_hop: RouteNextHop = Field(alias="nextHop") + description: str | None = Field(default=None, max_length=255) + + +class UpdateStaticRouteRequest(CloudRequest): + vpc_id: str = Field(alias="vpcId") + static_route_id: str = Field(alias="staticRouteId") + description: str | None = Field(default=None, max_length=255) + + +class StaticRoute(CloudResponse): + id: str | None = None + vpc_id: str = Field(alias="vpcId") + project_id: str = Field(alias="projectId") + subnet: str + next_hop: RouteNextHop = Field(alias="nextHop") + description: str | None = None + status: str | None = None + + +class ListStaticRoutesResponse(CloudResponse): + routes: list[StaticRoute] = Field(default_factory=list) + next_page_token: str | None = Field(default=None, alias="nextPageToken") + total_count: int | None = Field(default=None, alias="totalCount") + + +VMState = Literal[ + "running", "stopped", "error", "error_creating", "error_deleting", "creating", + "starting", "stopping", "rebooting", "deleting", "updating", "rebuilding", +] +VMPowerState = Literal["power_on", "power_off", "reboot"] +RemoteConsoleProtocol = Literal["vnc", "serial"] + + +class VMCreateRequest(CloudRequest): + project_id: str + name: str = Field(min_length=1, max_length=64, pattern=r"^[a-zA-Z][a-zA-Z0-9.\-_]*$") + disks: list[dict[str, Any]] = Field(min_length=1) + availability_zone_id: str | None = None + availability_zone_name: str | None = None + description: str | None = Field(default=None, max_length=255) + flavor_id: str | None = None + flavor_name: str | None = None + image_id: str | None = None + image_name: str | None = None + interfaces: list[dict[str, Any]] | None = None + subnets: list[dict[str, Any]] | None = None + metadata_fields: dict[str, Any] | None = None + image_metadata: dict[str, Any] | None = None + tag_ids: list[str] | None = None + tag_names: list[str] | None = None + cloud_init: str | None = None + + +class VMUpdateRequest(CloudRequest): + name: str | None = Field(default=None, min_length=1, max_length=64) + flavor_id: str | None = None + flavor_name: str | None = None + description: str | None = Field(default=None, max_length=255) + tag_ids: list[str] | None = None + tag_names: list[str] | None = None + + +class VMDeleteRequest(CloudRequest): + delete_attachments: dict[str, Any] | None = None + + +class VMStateRequest(CloudRequest): + id: str + state: Literal["running", "stopped"] + + +class VMPowerRequest(CloudRequest): + state: VMPowerState + + +class VMRebuildRequest(CloudRequest): + image_id: str | None = None + image_name: str | None = None + image_metadata: dict[str, Any] | None = None + metadata_fields: dict[str, Any] | None = None + + +class RemoteConsoleRequest(CloudRequest): + protocol: RemoteConsoleProtocol + + +class VMPasswordRequest(CloudRequest): + password: str = Field(min_length=8) + + +class VMDeletePasswordRequest(CloudRequest): + login: str | None = None + + +class VM(CloudResponse): + id: str + name: str + state: VMState | None = None + description: str | None = None + locked: bool | None = None + project: dict[str, Any] | None = None + availability_zone: dict[str, Any] | None = None + + +class ListVMsResponse(CloudResponse): + items: list[VM] = Field(default_factory=list) + total: int | None = None + offset: int | None = None + limit: int | None = None + + +class RemoteConsole(CloudResponse): + remote_console_url: str | None = None + remote_console_ws: str | None = None + vnc_url: str | None = None + vnc_ws: str | None = None + + +class VNCConsole(CloudResponse): + url: str | None = None + ws_url: str | None = None + +class SubnetCreateRequest(CloudRequest): + name: str = Field(min_length=1, max_length=64) + project_id: str + availability_zone_id: str + subnet_address: str | None = None + prefix_length: int | None = Field(default=None, ge=8, le=29) + vpc_id: str | None = None + description: str | None = Field(default=None, max_length=255) + routed_network: bool = True + + @model_validator(mode="after") + def require_address_or_prefix(self) -> "SubnetCreateRequest": + if (self.subnet_address is None) == (self.prefix_length is None): + raise ValueError("exactly one of subnet_address or prefix_length must be set") + return self + + +class Subnet(CloudResponse): + id: str + name: str + subnet_address: str + project: dict[str, Any] | None = None + vpc_id: str | None = None + state: str | None = None + + +class ListSubnetsResponse(CloudResponse): + items: list[Subnet] = Field(default_factory=list) + total: int | None = None + offset: int | None = None + limit: int | None = None + + +class AvailabilityZone(CloudResponse): + id: str + name: str + display_name: str | None = None + short_name: str | None = None + enabled: bool | None = None + default: bool | None = None diff --git a/src/cloud/vm_manager.py b/src/cloud/vm_manager.py new file mode 100644 index 0000000..9debf8b --- /dev/null +++ b/src/cloud/vm_manager.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import json +import time +from collections.abc import Mapping, Sequence +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +from .models import ( + AvailabilityZone, + ListVMsResponse, + RemoteConsole, + RemoteConsoleRequest, + VMCreateRequest, + VMDeletePasswordRequest, + VMDeleteRequest, + VMPasswordRequest, + VMPowerRequest, + VMRebuildRequest, + VMStateRequest, + VMUpdateRequest, + VM, + ListSubnetsResponse, + Subnet, + SubnetCreateRequest, +) + + +class VMAPIError(RuntimeError): + """Cloud.ru Virtual Machines API request failed.""" + + def __init__( + self, + status: int | None, + message: str, + details: Any = None, + *, + content_type: str | None = None, + ) -> None: + super().__init__(message) + self.status = status + self.details = details + self.content_type = content_type + + +class VMManager: + """Client for documented Cloud.ru Evolution Virtual Machines API methods.""" + + DEFAULT_BASE_URL = "https://compute.api.cloud.ru" + + def __init__(self, token: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 30) -> None: + if not token: + raise ValueError("token must not be empty") + if timeout <= 0: + raise ValueError("timeout must be positive") + + self._token = token + self._base_url = base_url.rstrip("/") + self._timeout = timeout + + def list_vms( + self, + project_id: str, + *, + vm_ids: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + offset: int | None = None, + limit: int | None = None, + ) -> ListVMsResponse: + return ListVMsResponse.model_validate(self._request( + "GET", + "/api/v1/vms", + query=self._without_none( + project_id=project_id, + vm_ids=vm_ids, + statuses=statuses, + offset=offset, + limit=limit, + ), + )) + + def get_vm(self, vm_id: str) -> VM: + return VM.model_validate(self._request("GET", f"/api/v1/vms/{vm_id}")) + + def create_vm(self, vm: VMCreateRequest) -> list[VM]: + """Create VM from documented `/api/v1/vms` request payload.""" + response = self._request( + "POST", "/api/v1/vms", body=[vm.model_dump(exclude_none=True)] + ) + return [VM.model_validate(item) for item in self._require_list(response)] + + def create_vms(self, vms: Sequence[VMCreateRequest]) -> list[VM]: + """Create one or more VMs using the documented batch endpoint.""" + response = self._request( + "POST", + "/api/v1/vms", + body=[vm.model_dump(exclude_none=True) for vm in vms], + ) + return [VM.model_validate(item) for item in self._require_list(response)] + + def wait_for_vms_ready( + self, + vms: Sequence[VM], + *, + timeout: float = 600, + poll_interval: float = 5, + ) -> list[VM]: + if timeout <= 0 or poll_interval <= 0: + raise ValueError("timeout and poll_interval must be positive") + pending = {vm.id for vm in vms} + latest = {vm.id: vm for vm in vms} + deadline = time.monotonic() + timeout + while pending: + for vm_id in tuple(pending): + vm = self.get_vm(vm_id) + latest[vm_id] = vm + if vm.state == "running": + pending.remove(vm_id) + elif vm.state in {"error", "error_creating", "deleting", "deleted"}: + raise VMAPIError(None, f"VM {vm_id} entered terminal state {vm.state!r}") + if pending: + if time.monotonic() >= deadline: + raise TimeoutError(f"timed out waiting for VMs: {', '.join(sorted(pending))}") + time.sleep(poll_interval) + return [latest[vm.id] for vm in vms] + + def delete_game_vms( + self, + game_id: str, + project_id: str, + *, + timeout: float = 600, + poll_interval: float = 5, + ) -> None: + """Delete VMs created by this consumer for one game and wait for removal.""" + prefix = f"game-{game_id}-" + vms = [vm for vm in self.list_vms(project_id).items if vm.name.startswith(prefix)] + for vm in vms: + self.delete_vm(vm.id) + + pending = {vm.id for vm in vms} + deadline = time.monotonic() + timeout + while pending: + for vm_id in tuple(pending): + try: + self.get_vm(vm_id) + except VMAPIError as error: + if error.status == 404: + pending.remove(vm_id) + else: + raise + if pending: + if time.monotonic() >= deadline: + raise TimeoutError(f"timed out waiting for VMs to delete: {', '.join(sorted(pending))}") + time.sleep(poll_interval) + + def update_vm(self, vm_id: str, changes: VMUpdateRequest) -> VM: + return VM.model_validate(self._request( + "PUT", f"/api/v1/vms/{vm_id}", body=changes.model_dump(exclude_none=True) + )) + + def delete_vm( + self, vm_id: str, request: VMDeleteRequest | None = None + ) -> None: + self._request( + "DELETE", + f"/api/v1/vms/{vm_id}", + body=request.model_dump(exclude_none=True) if request else None, + ) + + def set_state(self, request: VMStateRequest) -> None: + self._request("PUT", "/api/v1/vms", body=[request.model_dump()]) + + def start_vm(self, vm_id: str) -> None: + self.set_state(VMStateRequest(id=vm_id, state="running")) + + def stop_vm(self, vm_id: str) -> None: + self.set_state(VMStateRequest(id=vm_id, state="stopped")) + + def set_power(self, vm_id: str, request: VMPowerRequest) -> None: + self._request("POST", f"/api/v1/vms/{vm_id}/set-power", body=request.model_dump()) + + def rebuild_vm(self, vm_id: str, request: VMRebuildRequest) -> VM: + return VM.model_validate(self._request( + "POST", f"/api/v1/vms/{vm_id}/rebuild", body=request.model_dump(exclude_none=True) + )) + + def get_vnc(self, vm_id: str) -> RemoteConsole: + return RemoteConsole.model_validate(self._request("POST", f"/api/v1/vms/{vm_id}/get-vnc")) + + def create_remote_console(self, vm_id: str, request: RemoteConsoleRequest) -> RemoteConsole: + return RemoteConsole.model_validate(self._request( + "POST", f"/api/v1/vms/{vm_id}/remote-console", body=request.model_dump() + )) + + def set_password(self, vm_id: str, request: VMPasswordRequest) -> None: + self._request("POST", f"/api/v1/vms/{vm_id}/set-password", body=request.model_dump()) + + def delete_password(self, vm_id: str, request: VMDeletePasswordRequest | None = None) -> None: + self._request( + "POST", + f"/api/v1/vms/{vm_id}/delete-password", + body=request.model_dump(exclude_none=True) if request else None, + ) + + def check_guest_agent(self, vm_id: str) -> None: + self._request("POST", f"/api/v1/vms/{vm_id}/check-guest-agent") + + def list_availability_zones(self) -> list[AvailabilityZone]: + """List availability zones from documented `/api/v1/availability-zones`.""" + payload = self._request("GET", "/api/v1/availability-zones") + return [AvailabilityZone.model_validate(item) for item in self._require_list(payload)] + + def list_subnets(self, project_id: str, *, subnet_address: str | None = None) -> ListSubnetsResponse: + return ListSubnetsResponse.model_validate(self._request( + "GET", "/api/v1/subnets", query=self._without_none( + project_id=project_id, subnet_address=subnet_address + ) + )) + + def get_subnet(self, subnet_id: str) -> Subnet: + return Subnet.model_validate(self._request("GET", f"/api/v1/subnets/{subnet_id}")) + + def create_subnet(self, request: SubnetCreateRequest) -> Subnet: + return Subnet.model_validate(self._request( + "POST", "/api/v1/subnets", body=request.model_dump(exclude_none=True) + )) + + def _request( + self, + method: str, + path: str, + *, + query: Mapping[str, Any] | None = None, + body: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, + ) -> Any: + url = f"{self._base_url}{path}" + if query: + url = f"{url}?{urlencode(query, doseq=True)}" + + data = json.dumps(body).encode() if body is not None else None + request = Request( + url, + data=data, + method=method, + headers={ + "Authorization": f"Bearer {self._token}", + "Accept": "application/json", + "Content-Type": "application/json", + }, + ) + try: + with urlopen(request, timeout=self._timeout) as response: + return self._decode_response(response.read()) + except HTTPError as error: + raw_payload = error.read() + payload = self._decode_error_response(raw_payload) + message = self._error_message(payload, error.reason) + raise VMAPIError( + error.code, + message, + payload, + content_type=error.headers.get_content_type(), + ) from error + except URLError as error: + raise VMAPIError(None, str(error.reason)) from error + + @staticmethod + def _decode_response(data: bytes) -> Any: + if not data: + return {} + try: + payload = json.loads(data) + except json.JSONDecodeError as error: + raise VMAPIError(None, "API returned invalid JSON") from error + return payload + + @staticmethod + def _decode_error_response(data: bytes) -> Any: + if not data: + return None + try: + return json.loads(data) + except json.JSONDecodeError: + text = data.decode("utf-8", errors="replace").strip() + return {"raw_response": text[:512]} + + @staticmethod + def _error_message(payload: Any, fallback: str) -> str: + if isinstance(payload, list): + messages = [item.get("message") for item in payload if isinstance(item, dict)] + if messages: + return "; ".join(message for message in messages if isinstance(message, str)) + if not isinstance(payload, dict): + return fallback + if isinstance(payload.get("raw_response"), str): + return f"{fallback}: {payload['raw_response']}" + message = payload.get("message") + if isinstance(message, str): + return message + if isinstance(payload.get("detail"), str): + return payload["detail"] + return fallback + + @staticmethod + def _require_list(payload: Any) -> list[dict[str, Any]]: + if not isinstance(payload, list) or not all(isinstance(item, dict) for item in payload): + raise VMAPIError(None, "API returned non-array VM response") + return payload + + @staticmethod + def _without_none(**values: Any) -> dict[str, Any]: + return {key: value for key, value in values.items() if value is not None} diff --git a/src/cloud/vpc_network_manager.py b/src/cloud/vpc_network_manager.py new file mode 100644 index 0000000..f3ca9e2 --- /dev/null +++ b/src/cloud/vpc_network_manager.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +from .models import ( + CreateStaticRouteRequest, + CreateVPCRequest, + ListStaticRoutesResponse, + ListVPCsResponse, + Operation, + RouteNextHop, + StaticRoute, + UpdateVPCRequest, + VPC, +) + + +class VPCAPIError(RuntimeError): + """Cloud.ru Evolution VPC API request failed.""" + + def __init__(self, status: int | None, message: str, details: Any = None) -> None: + super().__init__(message) + self.status = status + self.details = details + + +class VPCNetworkManager: + """Client for documented Cloud.ru Evolution VPC API operations.""" + + DEFAULT_BASE_URL = "https://vpc.api.cloud.ru" + + def __init__(self, token: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 30) -> None: + if not token: + raise ValueError("token must not be empty") + if timeout <= 0: + raise ValueError("timeout must be positive") + + self._token = token + self._base_url = base_url.rstrip("/") + self._timeout = timeout + + def list_vpcs( + self, + project_id: str, + *, + page_size: int | None = None, + page_token: str | None = None, + filter: str | None = None, + ) -> ListVPCsResponse: + return ListVPCsResponse.model_validate(self._request( + "GET", + "/v1/vpcs", + query=self._without_none( + projectId=project_id, + pageSize=page_size, + pageToken=page_token, + filter=filter, + ), + )) + + def get_vpc(self, vpc_id: str) -> VPC: + return VPC.model_validate(self._request("GET", f"/v1/vpcs/{vpc_id}")) + + def create_vpc( + self, request: CreateVPCRequest + ) -> Operation: + return Operation.model_validate(self._request( + "POST", + "/v1/vpcs", + body=request.model_dump(by_alias=True, exclude_none=True), + )) + + def update_vpc(self, request: UpdateVPCRequest) -> Operation: + return Operation.model_validate(self._request( + "PUT", + f"/v1/vpcs/{request.vpc_id}", + body=request.model_dump(by_alias=True, exclude_none=True), + )) + + def delete_vpc(self, vpc_id: str) -> Operation: + return Operation.model_validate(self._request("DELETE", f"/v1/vpcs/{vpc_id}")) + + def get_operation(self, operation_id: str) -> Operation: + return Operation.model_validate(self._request("GET", f"/v1/vpcs/operations/{operation_id}")) + + def list_static_routes( + self, + vpc_id: str, + *, + page_size: int | None = None, + page_token: str | None = None, + filter: str | None = None, + ) -> ListStaticRoutesResponse: + return ListStaticRoutesResponse.model_validate(self._request( + "GET", + f"/v1/vpcs/{vpc_id}/routes/static", + query=self._without_none( + pageSize=page_size, pageToken=page_token, filter=filter + ), + )) + + def get_static_route(self, vpc_id: str, static_route_id: str) -> StaticRoute: + return StaticRoute.model_validate(self._request( + "GET", f"/v1/vpcs/{vpc_id}/routes/static/{static_route_id}" + )) + + def create_static_route( + self, + vpc_id: str, + request: CreateStaticRouteRequest, + ) -> Operation: + if request.vpc_id != vpc_id: + raise ValueError("vpc_id must match request.vpc_id") + return Operation.model_validate(self._request( + "POST", + f"/v1/vpcs/{vpc_id}/routes/static", + body=request.model_dump(by_alias=True, exclude_none=True), + )) + + def delete_static_route(self, vpc_id: str, static_route_id: str) -> Operation: + return Operation.model_validate(self._request( + "DELETE", f"/v1/vpcs/{vpc_id}/routes/static/{static_route_id}" + )) + + def _request( + self, + method: str, + path: str, + *, + query: Mapping[str, Any] | None = None, + body: Mapping[str, Any] | None = None, + ) -> dict[str, Any]: + url = f"{self._base_url}{path}" + if query: + url = f"{url}?{urlencode(query)}" + + data = json.dumps(body).encode() if body is not None else None + request = Request( + url, + data=data, + method=method, + headers={ + "Authorization": f"Bearer {self._token}", + "Accept": "application/json", + "Content-Type": "application/json", + }, + ) + try: + with urlopen(request, timeout=self._timeout) as response: + return self._decode_response(response.read()) + except HTTPError as error: + payload = self._decode_response(error.read()) + raise VPCAPIError( + error.code, + str(payload.get("message", error.reason)), + payload.get("details"), + ) from error + except URLError as error: + raise VPCAPIError(None, str(error.reason)) from error + + @staticmethod + def _decode_response(data: bytes) -> dict[str, Any]: + if not data: + return {} + try: + payload = json.loads(data) + except json.JSONDecodeError as error: + raise VPCAPIError(None, "API returned invalid JSON") from error + if not isinstance(payload, dict): + raise VPCAPIError(None, "API returned non-object JSON") + return payload + + @staticmethod + def _without_none(**values: Any) -> dict[str, Any]: + return {key: value for key, value in values.items() if value is not None} diff --git a/src/database/__init__.py b/src/database/__init__.py new file mode 100644 index 0000000..517e857 --- /dev/null +++ b/src/database/__init__.py @@ -0,0 +1,3 @@ +from .models import Base, NetworkPoolEntry + +__all__ = ["Base", "NetworkPoolEntry"] diff --git a/src/database/models.py b/src/database/models.py new file mode 100644 index 0000000..0d05dc9 --- /dev/null +++ b/src/database/models.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from datetime import datetime +from uuid import UUID + +from sqlalchemy import DateTime, String, Uuid, func +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class Base(DeclarativeBase): + pass + + +class NetworkPoolEntry(Base): + __tablename__ = "network_pool" + + cidr: Mapped[str] = mapped_column(String(18), primary_key=True) + cloud_subnet_id: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + game_id: Mapped[UUID | None] = mapped_column(Uuid, unique=True, nullable=True) + allocated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/src/database/session.py b/src/database/session.py new file mode 100644 index 0000000..002ed9c --- /dev/null +++ b/src/database/session.py @@ -0,0 +1,6 @@ +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + + +def create_session_factory(database_url: str) -> sessionmaker[Session]: + return sessionmaker(create_engine(database_url), expire_on_commit=False) diff --git a/src/kernel_consumer/__init__.py b/src/kernel_consumer/__init__.py new file mode 100644 index 0000000..fca7f9f --- /dev/null +++ b/src/kernel_consumer/__init__.py @@ -0,0 +1 @@ +"""NATS consumer for game-start requests.""" diff --git a/src/kernel_consumer/main.py b/src/kernel_consumer/main.py new file mode 100644 index 0000000..14021e1 --- /dev/null +++ b/src/kernel_consumer/main.py @@ -0,0 +1,297 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import os +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from uuid import UUID, uuid4 + +if __package__ is None or __package__ == "": + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import nats +from nats.errors import TimeoutError +from nats.js import JetStreamContext +from nats.js.api import AckPolicy, ConsumerConfig, DeliverPolicy +from dotenv import load_dotenv + +from cloud.models import VMCreateRequest +from cloud.vm_manager import VMManager +from cloud.iam import IAMClient +from database.session import create_session_factory +from networking import NetworkingManager + +LOGGER = logging.getLogger("kernel-consumer") +DEFAULT_DATABASE_URL = f"sqlite:///{Path(__file__).resolve().parents[2] / 'kernel.db'}" +load_dotenv(Path(__file__).resolve().parents[2] / ".env") + + +@dataclass(frozen=True) +class Settings: + nats_url: str = os.getenv("NATS_URL", "nats://127.0.0.1:4222") + stream: str = os.getenv("NATS_STREAM", "GAME") + subject: str = os.getenv("NATS_SUBJECT", "game.start") + cleanup_subject: str = os.getenv("NATS_CLEANUP_SUBJECT", "game.cleanup") + consumer: str = os.getenv("NATS_CONSUMER", "kernel-consumer") + database_url: str = os.getenv("DATABASE_URL", DEFAULT_DATABASE_URL) + cloud_token: str | None = os.getenv("CLOUD_TOKEN") + cloud_key_id: str | None = os.getenv("CLOUD_KEY_ID") + cloud_secret: str | None = os.getenv("CLOUD_SECRET") + cloud_project_id: str | None = os.getenv("CLOUD_PROJECT_ID") + cloud_vpc_id: str | None = os.getenv("CLOUD_VPC_ID") + cloud_availability_zone_id: str | None = os.getenv("CLOUD_AVAILABILITY_ZONE_ID") + cloud_availability_zone_name: str | None = os.getenv("CLOUD_AVAILABILITY_ZONE_NAME") + cloud_flavor_name: str = os.getenv("CLOUD_VM_FLAVOR_NAME", "low-1-1") + cloud_image_name: str = os.getenv("CLOUD_VM_IMAGE_NAME", "ubuntu-22.04") + cloud_disk_type_name: str = os.getenv("CLOUD_VM_DISK_TYPE_NAME", "SSD") + cloud_vm_username: str = os.getenv("CLOUD_VM_USERNAME", "user1") + cloud_vm_password: str | None = os.getenv("CLOUD_VM_PASSWORD") + cloud_vm_public_key: str | None = os.getenv("CLOUD_VM_PUBLIC_KEY") + + +def validate_start_game_request(request: dict[str, Any]) -> None: + """Validate fields described by kernel.proto without using protobuf.""" + required = {"team_count", "services", "teams", "game_starts_at", "rounds", "network_opens_at"} + missing = required - request.keys() + if missing: + raise ValueError(f"missing fields: {', '.join(sorted(missing))}") + + if not isinstance(request["teams"], list) or not isinstance(request["services"], list): + raise ValueError("teams and services must be arrays") + if not isinstance(request["team_count"], int) or request["team_count"] < 0: + raise ValueError("team_count must be a non-negative integer") + if request["team_count"] != len(request["teams"]): + raise ValueError("team_count does not match teams length") + + +def create_game_vms( + request: dict[str, Any], + game_id: UUID, + network: Any, + cloud: VMManager, + settings: Settings, +) -> list[Any]: + print("Creating game VMs...") + vm_requests = [] + for index in range(request["team_count"] + 1): + is_extra = index == request["team_count"] + vm_requests.append(VMCreateRequest( + project_id=settings.cloud_project_id, + name=f"game-{game_id}-{'extra' if is_extra else f'team-{index + 1}'}", + flavor_name=settings.cloud_flavor_name, + availability_zone_id=settings.cloud_availability_zone_id or None, + availability_zone_name=settings.cloud_availability_zone_name or None, + image_name=settings.cloud_image_name, + disks=[{ + "name": f"game-{game_id}-disk-{index + 1}", + "size": 10, + "disk_type_name": settings.cloud_disk_type_name, + }], + subnets=[{ + "subnet_id": network.cloud_subnet_id, + "new_floating_ip": is_extra, + }], + image_metadata=_image_metadata(settings, f"game-{game_id}-{index + 1}"), + description=f"game {game_id}", + )) + return cloud.wait_for_vms_ready(cloud.create_vms(vm_requests)) + + +def _image_metadata(settings: Settings, hostname: str) -> dict[str, str]: + metadata = {"name": settings.cloud_vm_username, "hostname": hostname} + if settings.cloud_vm_password: + if not any(char.isascii() and char.isupper() for char in settings.cloud_vm_password): + raise ValueError("CLOUD_VM_PASSWORD must contain at least one uppercase Latin letter") + metadata["linux_password"] = settings.cloud_vm_password + if settings.cloud_vm_public_key: + metadata["public_key"] = settings.cloud_vm_public_key + return metadata + + +async def handle_start_game( + request: dict[str, Any], + game_id: UUID, + *, + network: Any = None, + cloud: VMManager | None = None, + settings: Settings | None = None, +) -> None: + """Create one VM per team plus one public management VM.""" + LOGGER.info( + "starting game: id=%s teams=%d services=%d rounds=%d starts_at=%d network_opens_at=%d", + game_id, + request["team_count"], + len(request["services"]), + request["rounds"], + request["game_starts_at"], + request["network_opens_at"], + ) + if network is not None and cloud is not None and settings is not None: + vms = await asyncio.to_thread(create_game_vms, request, game_id, network, cloud, settings) + LOGGER.info("created %d VMs for game %s", len(vms), game_id) + + +async def handle_cleanup_game( + game_id: UUID, + *, + networking: NetworkingManager, + cloud: VMManager, + project_id: str, +) -> None: + """Delete a game's VMs before releasing its network back to the pool.""" + await asyncio.to_thread(cloud.delete_game_vms, str(game_id), project_id) + try: + await asyncio.to_thread(networking.releaseNetworkToPool, game_id) + except LookupError: + LOGGER.info("network for game %s was already released", game_id) + LOGGER.info("cleaned up game %s", game_id) + + +def validate_cleanup_request(request: dict[str, Any]) -> UUID: + if set(request) != {"game_id"}: + raise ValueError("cleanup message must contain only game_id") + try: + return UUID(str(request["game_id"])) + except (TypeError, ValueError) as error: + raise ValueError("game_id must be a UUID") from error + + +async def consume(settings: Settings = Settings()) -> None: + + + networking_manager: NetworkingManager | None = None + cloud_manager: VMManager | None = None + cloud_token = settings.cloud_token + if settings.cloud_key_id or settings.cloud_secret: + if not (settings.cloud_key_id and settings.cloud_secret): + raise ValueError("CLOUD_KEY_ID and CLOUD_SECRET must be set together") + cloud_token = IAMClient(settings.cloud_key_id, settings.cloud_secret).get_access_token() + + if cloud_token and settings.cloud_project_id: + if not settings.cloud_vm_password and not settings.cloud_vm_public_key: + raise ValueError("CLOUD_VM_PASSWORD or CLOUD_VM_PUBLIC_KEY must be set for Ubuntu VMs") + cloud_manager = VMManager(cloud_token) + networking_manager = NetworkingManager( + create_session_factory(settings.database_url), + cloud_manager, + settings.cloud_project_id, + vpc_id=settings.cloud_vpc_id, + availability_zone_id=settings.cloud_availability_zone_id, + availability_zone_name=settings.cloud_availability_zone_name, + ) + elif cloud_token or settings.cloud_project_id: + raise ValueError("Cloud credentials and CLOUD_PROJECT_ID must be set together") + + nc = await nats.connect(settings.nats_url) + try: + js: JetStreamContext = nc.jetstream() + async def on_message(message: Any) -> None: + game_id: UUID | None = None + network: Any = None + LOGGER.info("onMessage") + try: + request = json.loads(message.data) + if not isinstance(request, dict): + raise ValueError("message must contain a JSON object") + validate_start_game_request(request) + game_id = uuid4() + LOGGER.info("setting up game %s ", game_id) + if networking_manager: + network = await asyncio.to_thread( + networking_manager.getNetworkFromPool, game_id + ) + LOGGER.info("allocated network %s for game %s", network.cidr, game_id) + await handle_start_game( + request, + game_id, + network=network if networking_manager else None, + cloud=cloud_manager, + settings=settings, + ) + except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError): + LOGGER.exception("invalid StartGameRequest; dropping message") + await message.term() + except Exception: + if networking_manager and game_id and network: + try: + await asyncio.to_thread(networking_manager.releaseNetworkToPool, game_id) + except Exception: + LOGGER.exception("failed to release network for game %s", game_id) + LOGGER.exception("StartGameRequest handling failed; message will retry") + return + else: + await message.ack() + + subscription = await js.pull_subscribe( + settings.subject, + durable=settings.consumer, + stream=settings.stream, + config=ConsumerConfig( + durable_name=settings.consumer, + filter_subject=settings.subject, + ack_policy=AckPolicy.EXPLICIT, + deliver_policy=DeliverPolicy.ALL, + ), + ) + cleanup_subscription = await js.pull_subscribe( + settings.cleanup_subject, + durable=f"{settings.consumer}-cleanup", + stream=settings.stream, + config=ConsumerConfig( + durable_name=f"{settings.consumer}-cleanup", + filter_subject=settings.cleanup_subject, + ack_policy=AckPolicy.EXPLICIT, + deliver_policy=DeliverPolicy.ALL, + ), + ) + LOGGER.info("listening on %s (stream=%s consumer=%s)", settings.subject, settings.stream, settings.consumer) + while True: + try: + messages = await subscription.fetch(batch=1, timeout=0.2) + except TimeoutError: + messages = [] + for message in messages: + await on_message(message) + try: + cleanup_messages = await cleanup_subscription.fetch(batch=1, timeout=0.2) + except TimeoutError: + cleanup_messages = [] + for message in cleanup_messages: + try: + payload = json.loads(message.data) + if not isinstance(payload, dict): + raise ValueError("message must contain a JSON object") + cleanup_game_id = validate_cleanup_request(payload) + if networking_manager is None or cloud_manager is None or not settings.cloud_project_id: + raise ValueError("cloud configuration is required for cleanup") + await handle_cleanup_game( + cleanup_game_id, + networking=networking_manager, + cloud=cloud_manager, + project_id=settings.cloud_project_id, + ) + except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError): + LOGGER.exception("invalid GameCleanupRequest; dropping message") + await message.term() + except Exception: + LOGGER.exception("GameCleanupRequest handling failed; message will retry") + else: + await message.ack() + finally: + await nc.drain() + + +def main() -> None: + logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s") + try: + asyncio.run(consume()) + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() diff --git a/src/models/Vlan.py b/src/models/Vlan.py new file mode 100644 index 0000000..98587fa --- /dev/null +++ b/src/models/Vlan.py @@ -0,0 +1,9 @@ +from uuid import UUID + +from pydantic import BaseModel + + +class VLAN(BaseModel): + id: UUID + cidr: str + diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/networking/__init__.py b/src/networking/__init__.py new file mode 100644 index 0000000..5e7ecc4 --- /dev/null +++ b/src/networking/__init__.py @@ -0,0 +1,3 @@ +from .manager import NetworkingManager, NetworkPoolExhaustedError + +__all__ = ["NetworkingManager", "NetworkPoolExhaustedError"] diff --git a/src/networking/manager.py b/src/networking/manager.py new file mode 100644 index 0000000..b49e0cc --- /dev/null +++ b/src/networking/manager.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import time +from datetime import datetime, timezone +from typing import Protocol +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from cloud.models import AvailabilityZone, Subnet, SubnetCreateRequest +from cloud.vm_manager import VMAPIError +from database.models import NetworkPoolEntry + + +class NetworkPoolExhaustedError(RuntimeError): + pass + + +class CloudSubnetClient(Protocol): + def get_subnet(self, subnet_id: str) -> Subnet: ... + def create_subnet(self, request: SubnetCreateRequest) -> Subnet: ... + def list_availability_zones(self) -> list[AvailabilityZone]: ... + + +class NetworkingManager: + def __init__( + self, + sessions: sessionmaker[Session], + cloud: CloudSubnetClient, + project_id: str, + *, + vpc_id: str | None = None, + availability_zone_id: str | None = None, + availability_zone_name: str | None = None, + readiness_timeout: float = 300, + poll_interval: float = 2, + ) -> None: + self._sessions = sessions + self._cloud = cloud + self._project_id = project_id + self._vpc_id = vpc_id + if availability_zone_id is None and availability_zone_name is None: + raise ValueError("availability_zone_id or availability_zone_name is required to create subnets") + self._availability_zone_id = availability_zone_id + self._availability_zone_name = availability_zone_name + if readiness_timeout <= 0 or poll_interval <= 0: + raise ValueError("readiness_timeout and poll_interval must be positive") + self._readiness_timeout = readiness_timeout + self._poll_interval = poll_interval + + def getNetworkFromPool(self, game_id: UUID) -> NetworkPoolEntry: + with self._sessions.begin() as session: + existing = session.scalar(select(NetworkPoolEntry).where(NetworkPoolEntry.game_id == game_id)) + if existing: + return existing + entry = session.scalar( + select(NetworkPoolEntry).where(NetworkPoolEntry.game_id.is_(None)).order_by(NetworkPoolEntry.cidr).with_for_update() + ) + if entry: + try: + self._wait_for_subnet(entry.cloud_subnet_id) + except VMAPIError as error: + if error.status != 404: + raise + session.delete(entry) + session.flush() + entry = None + if entry is None: + entry = self._create_network(session) + entry.game_id = game_id + entry.allocated_at = datetime.now(timezone.utc) + session.flush() + return entry + + def releaseNetworkToPool(self, game_id: UUID) -> None: + with self._sessions.begin() as session: + entry = session.scalar(select(NetworkPoolEntry).where(NetworkPoolEntry.game_id == game_id).with_for_update()) + if entry is None: + raise LookupError(f"network for game {game_id} is not allocated") + entry.game_id = None + entry.allocated_at = None + + def _create_network(self, session: Session) -> NetworkPoolEntry: + used = {int(row.cidr.split('.')[1]) for row in session.scalars(select(NetworkPoolEntry))} + octet = next((value for value in range(0, 256, 16) if value not in used), None) + if octet is None: + raise NetworkPoolExhaustedError("network pool exhausted: no free 10.N.0.0/20 CIDRs") + cidr = f"10.{octet}.0.0/20" + subnet = self._cloud.create_subnet(SubnetCreateRequest( + name=f"game-network-{octet}", project_id=self._project_id, + availability_zone_id=self._zone_id(), + subnet_address=cidr, + vpc_id=self._vpc_id, + )) + subnet = self._wait_for_subnet(subnet.id) + entry = NetworkPoolEntry(cidr=cidr, cloud_subnet_id=subnet.id) + session.add(entry) + return entry + + def _wait_for_subnet(self, subnet_id: str) -> Subnet: + deadline = time.monotonic() + self._readiness_timeout + while True: + subnet = self._cloud.get_subnet(subnet_id) + if subnet.state in {"created", "available", "active", None}: + return subnet + if subnet.state in {"error", "error_creating", "deleting", "deleted"}: + raise RuntimeError(f"Cloud.ru subnet {subnet_id} entered terminal state {subnet.state!r}") + if time.monotonic() >= deadline: + raise TimeoutError(f"timed out waiting for Cloud.ru subnet {subnet_id} to become ready") + time.sleep(self._poll_interval) + + def _zone_id(self) -> str: + if self._availability_zone_id is None: + zones = self._cloud.list_availability_zones() + zone = next((z for z in zones if z.name == self._availability_zone_name), None) + if zone is None: + raise LookupError( + f"availability zone {self._availability_zone_name!r} not found; " + f"available: {', '.join(z.name for z in zones)}" + ) + self._availability_zone_id = zone.id + return self._availability_zone_id diff --git a/src/scripts/__init__.py b/src/scripts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/scripts/fire_game_cleanup.py b/src/scripts/fire_game_cleanup.py new file mode 100644 index 0000000..6384421 --- /dev/null +++ b/src/scripts/fire_game_cleanup.py @@ -0,0 +1,44 @@ +"""Publish a game cleanup request to NATS JetStream.""" + +import argparse +import asyncio +import json +import os +import sys +from pathlib import Path +from uuid import UUID + +import nats + +if __package__ is None or __package__ == "": + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from scripts.init_nats import ensure_stream + + +async def publish(url: str, stream: str, subject: str, start_subject: str, game_id: UUID) -> None: + await ensure_stream(url, stream, [start_subject, subject]) + nc = await nats.connect(url) + try: + await nc.jetstream().publish( + subject, + json.dumps({"game_id": str(game_id)}).encode(), + ) + print(f"published cleanup for game {game_id} to {subject} ({stream})") + finally: + await nc.drain() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Queue cleanup of one game") + parser.add_argument("game_id", type=UUID) + parser.add_argument("--url", default=os.getenv("NATS_URL", "nats://127.0.0.1:4222")) + parser.add_argument("--stream", default=os.getenv("NATS_STREAM", "GAME")) + parser.add_argument("--subject", default=os.getenv("NATS_CLEANUP_SUBJECT", "game.cleanup")) + parser.add_argument("--start-subject", default=os.getenv("NATS_SUBJECT", "game.start")) + args = parser.parse_args() + asyncio.run(publish(args.url, args.stream, args.subject, args.start_subject, args.game_id)) + + +if __name__ == "__main__": + main() diff --git a/src/scripts/init_nats.py b/src/scripts/init_nats.py new file mode 100644 index 0000000..13a5d72 --- /dev/null +++ b/src/scripts/init_nats.py @@ -0,0 +1,60 @@ +import argparse +import asyncio +import os + +import nats +from nats.js.api import StorageType, StreamConfig +from nats.js.errors import NotFoundError + + +async def ensure_stream(url: str, stream: str, subjects: list[str]) -> None: + configured_subjects = set(subjects) + nc = await nats.connect(url) + try: + js = nc.jetstream() + try: + stream_info = await js.stream_info(stream) + except NotFoundError: + await js.add_stream( + StreamConfig( + name=stream, + subjects=sorted(configured_subjects), + storage=StorageType.FILE, + ) + ) + print(f"created stream {stream} for subjects {', '.join(subjects)}") + return + + stream_subjects = set(stream_info.config.subjects or []) + missing = configured_subjects - stream_subjects + if not missing: + print(f"stream {stream} already contains requested subjects") + return + + stream_subjects.update(missing) + config = stream_info.config + config.subjects = sorted(stream_subjects) + await js.update_stream(config) + print(f"added subjects {', '.join(sorted(missing))} to stream {stream}") + finally: + await nc.drain() + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Create NATS JetStream channels required by kernel." + ) + parser.add_argument( + "--url", + default=os.getenv("NATS_URL", "nats://127.0.0.1:4222"), + ) + parser.add_argument("--stream", default=os.getenv("NATS_STREAM", "GAME")) + parser.add_argument("--subject", default=os.getenv("NATS_SUBJECT", "game.start")) + parser.add_argument("--cleanup-subject", default=os.getenv("NATS_CLEANUP_SUBJECT", "game.cleanup")) + args = parser.parse_args() + + asyncio.run(ensure_stream(args.url, args.stream, [args.subject, args.cleanup_subject])) + + +if __name__ == "__main__": + main() diff --git a/src/scripts/start_game_debug.py b/src/scripts/start_game_debug.py new file mode 100644 index 0000000..6b652ec --- /dev/null +++ b/src/scripts/start_game_debug.py @@ -0,0 +1,60 @@ +import argparse +import asyncio +import json +import os +from pathlib import Path + +import nats + +DEFAULT_REQUEST = { + "team_count": 2, + "services": [], + "teams": [ + {"team_name": "red", "team_id": "red-1"}, + {"team_name": "blue", "team_id": "blue-1"}, + ], + "game_starts_at": 0, + "rounds": 10, + "network_opens_at": 0, +} + + +async def publish(url: str, subject: str, payload: dict) -> None: + nc = await nats.connect(url) + try: + await nc.publish(subject, json.dumps(payload).encode()) + await nc.flush() + finally: + await nc.close() + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument( + "--url", + default=os.getenv("NATS_URL", "nats://127.0.0.1:4222"), + ) + parser.add_argument( + "--subject", + default=os.getenv("NATS_SUBJECT", "game.start"), + ) + parser.add_argument("--file", type=Path) + parser.add_argument("--pretty", action="store_true") + args = parser.parse_args() + + payload = ( + json.loads(args.file.read_text()) + if args.file + else DEFAULT_REQUEST + ) + + asyncio.run(publish(args.url, args.subject, payload)) + + if args.pretty: + print(json.dumps(payload, indent=2)) + else: + print(f"published StartGameRequest to {args.subject}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/test_environment.py b/tests/test_environment.py new file mode 100644 index 0000000..1a913d7 --- /dev/null +++ b/tests/test_environment.py @@ -0,0 +1,23 @@ +import os +from pathlib import Path + +from dotenv import load_dotenv + + +def test_dotenv_loads_missing_environment_value(tmp_path: Path, monkeypatch) -> None: + env_file = tmp_path / ".env" + env_file.write_text("CLOUD_PROJECT_ID=project-from-file\n") + monkeypatch.delenv("CLOUD_PROJECT_ID", raising=False) + + assert load_dotenv(env_file) is True + assert os.environ["CLOUD_PROJECT_ID"] == "project-from-file" + + +def test_dotenv_does_not_override_shell_environment(tmp_path: Path, monkeypatch) -> None: + env_file = tmp_path / ".env" + env_file.write_text("CLOUD_PROJECT_ID=project-from-file\n") + monkeypatch.setenv("CLOUD_PROJECT_ID", "project-from-shell") + + load_dotenv(env_file) + + assert os.environ["CLOUD_PROJECT_ID"] == "project-from-shell" diff --git a/tests/test_iam.py b/tests/test_iam.py new file mode 100644 index 0000000..9b1378e --- /dev/null +++ b/tests/test_iam.py @@ -0,0 +1,26 @@ +import json +from unittest.mock import patch + +from cloud.iam import IAMClient + + +class Response: + def read(self) -> bytes: + return json.dumps({"access_token": "fresh-token"}).encode() + + def __enter__(self) -> "Response": + return self + + def __exit__(self, *args: object) -> None: + return None + + +@patch("cloud.iam.urlopen") +def test_fetches_iam_token(urlopen) -> None: + urlopen.return_value = Response() + + assert IAMClient("key-id", "secret").get_access_token() == "fresh-token" + + request = urlopen.call_args.args[0] + assert request.full_url == "https://iam.api.cloud.ru/api/v1/auth/token" + assert json.loads(request.data) == {"keyId": "key-id", "secret": "secret"} diff --git a/tests/test_kernel_consumer.py b/tests/test_kernel_consumer.py new file mode 100644 index 0000000..c1a28c6 --- /dev/null +++ b/tests/test_kernel_consumer.py @@ -0,0 +1,46 @@ +from uuid import uuid4 + +from kernel_consumer.main import Settings, _image_metadata, create_game_vms + + +class Cloud: + def __init__(self) -> None: + self.requests = [] + + def create_vms(self, requests): + self.requests.extend(requests) + return [] + + def wait_for_vms_ready(self, vms): + return vms + + +def test_creates_one_vm_per_team_plus_public_extra() -> None: + cloud = Cloud() + settings = Settings( + cloud_project_id="project-id", + cloud_availability_zone_id="zone-id", + cloud_availability_zone_name=None, + cloud_vm_password="Password123", + ) + network = type("Network", (), {"cloud_subnet_id": "subnet-id"})() + request = {"team_count": 2} + + create_game_vms(request, uuid4(), network, cloud, settings) + + assert len(cloud.requests) == 3 + assert [item.subnets[0]["new_floating_ip"] for item in cloud.requests] == [False, False, True] + assert all(item.subnets[0]["subnet_id"] == "subnet-id" for item in cloud.requests) + assert all(item.availability_zone_id == "zone-id" for item in cloud.requests) + assert all(item.availability_zone_name is None for item in cloud.requests) + + +def test_rejects_password_without_uppercase_latin_letter() -> None: + settings = Settings(cloud_vm_password="lowercase123") + + try: + _image_metadata(settings, "game-host") + except ValueError as error: + assert str(error) == "CLOUD_VM_PASSWORD must contain at least one uppercase Latin letter" + else: + raise AssertionError("expected invalid password to be rejected") diff --git a/tests/test_networking_manager.py b/tests/test_networking_manager.py new file mode 100644 index 0000000..787febb --- /dev/null +++ b/tests/test_networking_manager.py @@ -0,0 +1,151 @@ +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from cloud.models import AvailabilityZone, Subnet +from cloud.vm_manager import VMAPIError +from database.models import Base, NetworkPoolEntry +from networking import NetworkingManager, NetworkPoolExhaustedError + + +class Cloud: + def __init__(self) -> None: + self.subnets: dict[str, Subnet] = {} + self.created: list[str] = [] + self.requests = [] + self.zones: list[AvailabilityZone] = [] + self.subnet_states: list[str] = [] + + def get_subnet(self, subnet_id: str) -> Subnet: + if subnet_id not in self.subnets: + raise VMAPIError(404, "not found") + subnet = self.subnets[subnet_id] + if self.subnet_states: + subnet.state = self.subnet_states.pop(0) + return subnet + + def list_availability_zones(self) -> list[AvailabilityZone]: + return self.zones + + def create_subnet(self, request): + self.requests.append(request) + subnet = Subnet( + id=f"subnet-{len(self.subnets) + 1}", + name=request.name, + subnet_address=request.subnet_address, + ) + self.subnets[subnet.id] = subnet + self.created.append(subnet.subnet_address) + return subnet + + +@pytest.fixture +def session_factory(): + engine = create_engine("sqlite://") + Base.metadata.create_all(engine) + return sessionmaker(engine, expire_on_commit=False) + + +def test_allocates_new_network_and_reuses_after_release(session_factory) -> None: + cloud = Cloud() + manager = NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id") + first_game = uuid4() + + first = manager.getNetworkFromPool(first_game) + assert first.cidr == "10.0.0.0/20" + assert first.game_id == first_game + assert cloud.created == ["10.0.0.0/20"] + + manager.releaseNetworkToPool(first_game) + reused = manager.getNetworkFromPool(uuid4()) + assert reused.cidr == first.cidr + assert len(cloud.created) == 1 + + +def test_waits_for_new_subnet_to_be_ready(session_factory) -> None: + cloud = Cloud() + cloud.subnet_states = ["creating", "created"] + + entry = NetworkingManager( + session_factory, + cloud, + "project-id", + availability_zone_id="zone-id", + poll_interval=0.001, + ).getNetworkFromPool(uuid4()) + + assert entry.cidr == "10.0.0.0/20" + + +def test_discards_missing_cloud_network_and_creates_replacement(session_factory) -> None: + cloud = Cloud() + with session_factory.begin() as session: + session.add(NetworkPoolEntry(cidr="10.0.0.0/20", cloud_subnet_id="gone")) + + entry = NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id").getNetworkFromPool(uuid4()) + + assert entry.cidr == "10.0.0.0/20" + assert entry.cloud_subnet_id == "subnet-1" + + +def test_passes_vpc_and_availability_zone_to_subnet_create(session_factory) -> None: + cloud = Cloud() + + NetworkingManager( + session_factory, + cloud, + "project-id", + vpc_id="vpc-id", + availability_zone_id="zone-id", + ).getNetworkFromPool(uuid4()) + + request = cloud.requests[0] + assert request.subnet_address == "10.0.0.0/20" + assert request.vpc_id == "vpc-id" + assert request.availability_zone_id == "zone-id" + assert request.model_dump(exclude_none=True)["availability_zone_id"] == "zone-id" + + +def test_resolves_availability_zone_name_to_id(session_factory) -> None: + cloud = Cloud() + cloud.zones.append(AvailabilityZone(id="zone-uuid", name="ru.AZ-2")) + + NetworkingManager( + session_factory, + cloud, + "project-id", + availability_zone_name="ru.AZ-2", + ).getNetworkFromPool(uuid4()) + + assert cloud.requests[0].availability_zone_id == "zone-uuid" + + +def test_raises_for_unknown_availability_zone_name(session_factory) -> None: + cloud = Cloud() + cloud.zones.append(AvailabilityZone(id="zone-uuid", name="ru.AZ-2")) + + with pytest.raises(LookupError, match="ru.AZ-9"): + NetworkingManager( + session_factory, + cloud, + "project-id", + availability_zone_name="ru.AZ-9", + ).getNetworkFromPool(uuid4()) + + +def test_raises_when_all_pool_cidrs_are_used(session_factory) -> None: + cloud = Cloud() + with session_factory.begin() as session: + for octet in range(0, 256, 16): + session.add( + NetworkPoolEntry( + cidr=f"10.{octet}.0.0/20", + cloud_subnet_id=str(octet), + game_id=uuid4(), + ) + ) + + with pytest.raises(NetworkPoolExhaustedError, match="network pool exhausted"): + NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id").getNetworkFromPool(uuid4()) diff --git a/tests/test_vm_api_errors.py b/tests/test_vm_api_errors.py new file mode 100644 index 0000000..f353ec6 --- /dev/null +++ b/tests/test_vm_api_errors.py @@ -0,0 +1,10 @@ +from cloud.vm_manager import VMManager + + +def test_error_response_keeps_non_json_body() -> None: + payload = VMManager._decode_error_response(b"upstream error") + + assert payload == {"raw_response": "upstream error"} + assert VMManager._error_message(payload, "Bad Gateway") == ( + "Bad Gateway: upstream error" + ) diff --git a/tests/test_vm_manager.py b/tests/test_vm_manager.py new file mode 100644 index 0000000..fd9bae6 --- /dev/null +++ b/tests/test_vm_manager.py @@ -0,0 +1,102 @@ +import json +from unittest.mock import patch + +import pytest + +from cloud import VM, VMCreateRequest, VMManager, VMStateRequest + + +class Response: + def __init__(self, payload: dict | None = None) -> None: + self.payload = payload + + def read(self) -> bytes: + return b"" if self.payload is None else json.dumps(self.payload).encode() + + def __enter__(self) -> "Response": + return self + + def __exit__(self, *args: object) -> None: + return None + + +@patch("cloud.vm_manager.urlopen") +def test_list_vms_encodes_repeated_filters(urlopen) -> None: + urlopen.return_value = Response({"items": []}) + manager = VMManager("token", base_url="https://api.example") + + response = manager.list_vms("project-id", vm_ids=["vm-1", "vm-2"], statuses=["running"]) + assert response.items == [] + + request = urlopen.call_args.args[0] + assert request.full_url == ( + "https://api.example/api/v1/vms?project_id=project-id&vm_ids=vm-1&vm_ids=vm-2&statuses=running" + ) + assert request.get_header("Authorization") == "Bearer token" + + +@patch("cloud.vm_manager.urlopen") +def test_create_vm_sends_documented_payload(urlopen) -> None: + urlopen.return_value = Response([{"id": "vm-id", "name": "game-vm"}]) + manager = VMManager("token", base_url="https://api.example") + payload = VMCreateRequest( + project_id="project-id", + name="game-vm", + disks=[{"disk_id": "disk-id"}], + ) + + assert manager.create_vm(payload)[0].id == "vm-id" + + request = urlopen.call_args.args[0] + assert request.method == "POST" + assert request.full_url == "https://api.example/api/v1/vms" + assert json.loads(request.data) == [payload.model_dump(exclude_none=True)] + + +@patch("cloud.vm_manager.urlopen") +def test_create_vms_uses_v1_batch_endpoint(urlopen) -> None: + urlopen.return_value = Response([{"id": "vm-id", "name": "game-vm"}]) + manager = VMManager("token", base_url="https://api.example") + payload = VMCreateRequest( + project_id="project-id", + name="game-vm", + disks=[{"disk_id": "disk-id"}], + subnets=[{"subnet_id": "subnet-id", "new_floating_ip": True}], + ) + + assert manager.create_vms([payload])[0].id == "vm-id" + + request = urlopen.call_args.args[0] + assert request.full_url == "https://api.example/api/v1/vms" + assert json.loads(request.data) == [payload.model_dump(exclude_none=True)] + + +@patch("cloud.vm_manager.urlopen") +def test_start_vm_uses_bulk_state_endpoint(urlopen) -> None: + urlopen.return_value = Response() + manager = VMManager("token", base_url="https://api.example") + + manager.start_vm("vm-id") + + request = urlopen.call_args.args[0] + assert request.method == "PUT" + assert request.full_url == "https://api.example/api/v1/vms" + assert json.loads(request.data) == [{"id": "vm-id", "state": "running"}] + + +@patch("cloud.vm_manager.time.sleep") +def test_wait_for_vms_ready_polls_until_running(sleep) -> None: + manager = VMManager("token") + initial = VM(id="vm-id", name="game-vm", state="creating") + states = iter(["creating", "running"]) + + with patch.object(manager, "get_vm", side_effect=lambda _: initial.model_copy(update={"state": next(states)})): + ready = manager.wait_for_vms_ready([initial], timeout=1, poll_interval=0.01) + + assert ready[0].state == "running" + sleep.assert_called_once_with(0.01) + + +def test_set_state_rejects_unknown_state() -> None: + with pytest.raises(ValueError, match="Input should be 'running' or 'stopped'"): + VMStateRequest(id="vm-id", state="rebooting") diff --git a/tests/test_vpc_network_manager.py b/tests/test_vpc_network_manager.py new file mode 100644 index 0000000..1a89886 --- /dev/null +++ b/tests/test_vpc_network_manager.py @@ -0,0 +1,49 @@ +import json +from unittest.mock import patch + +import pytest + +from cloud import CreateVPCRequest, VPCAPIError, VPCNetworkManager + + +class Response: + def __init__(self, payload: dict) -> None: + self.payload = payload + + def read(self) -> bytes: + return json.dumps(self.payload).encode() + + def __enter__(self) -> "Response": + return self + + def __exit__(self, *args: object) -> None: + return None + + +@patch("cloud.vpc_network_manager.urlopen") +def test_create_vpc_sends_bearer_request(urlopen) -> None: + urlopen.return_value = Response({"id": "operation-id"}) + manager = VPCNetworkManager("token", base_url="https://api.example") + + operation = manager.create_vpc( + CreateVPCRequest(projectId="project-id", name="game-vpc") + ) + assert operation.id == "operation-id" + + request = urlopen.call_args.args[0] + assert request.full_url == "https://api.example/v1/vpcs" + assert request.method == "POST" + assert request.get_header("Authorization") == "Bearer token" + assert json.loads(request.data) == {"projectId": "project-id", "name": "game-vpc"} + + +def test_manager_rejects_empty_token() -> None: + with pytest.raises(ValueError, match="token must not be empty"): + VPCNetworkManager("") + + +def test_api_error_keeps_status_and_details() -> None: + error = VPCAPIError(403, "forbidden", [{"reason": "denied"}]) + + assert error.status == 403 + assert error.details == [{"reason": "denied"}] diff --git a/token.json b/token.json new file mode 100644 index 0000000..41ddf6e --- /dev/null +++ b/token.json @@ -0,0 +1,12 @@ +{ + "access_token": "eyJhbGciOiJSUzI1NiIsImtpZCI6ImEyMjM3ZDhhLWQ0ZDQtNDA5Yi04ZTMxLWM3NGJhYTZhM2NjYiIsInR5cCI6IkpXVCJ9.eyJhdWQiOlsiaWFtIl0sImF1dGhfdGltZSI6MTc4NjU0OTQzMCwiYXpwIjoiMTJmN2UxZTcwYTgyZTg2NTgxZjkzNGU4MDNiYjdhMTEiLCJlbWFpbCI6InRpbW9mZXlAZ3VsZW5vay5ydSIsImVtYWlsX3ZlcmlmaWVkIjp0cnVlLCJleHAiOjE3ODY1NTMwMzAsImZhbWlseV9uYW1lIjoi0JPRg9C70LXQvdC-0LoiLCJnaXZlbl9uYW1lIjoi0KLQuNC80L7RhNC10LkiLCJpYXQiOjE3ODY1NDk0MzAsImlzcyI6Imh0dHBzOi8vaWQuY2xvdWQucnUvYXV0aC9zeXN0ZW0iLCJqdGkiOiIwOTVjNzVlMi00OTkyLTRhZTQtYTc4Mi04MzBlZjkxN2EzMjIiLCJuYmYiOjE3ODY1NDk0MzAsIm5vbmNlIjoiIiwicGhvbmVfbnVtYmVyIjoiNzkxNjY5ODAzMDYiLCJwcmVmZXJyZWRfdXNlcm5hbWUiOiJ0aW1vZmV5QGd1bGVub2sucnUiLCJyZXNvdXJjZV9hY2Nlc3MiOm51bGwsInNjb3BlIjoicm9sZXMgZW1haWwgb3BlbmlkIHByb2ZpbGUiLCJzdWIiOiI5ZmZmYmQyNy02YWUwLTRiNjMtYWFkNC0wYjZjOWI1YjJkOTAiLCJzdWJfaWQiOiI5ZmZmYmQyNy02YWUwLTRiNjMtYWFkNC0wYjZjOWI1YjJkOTAiLCJzdWJfdHlwZSI6InVzZXIiLCJ0eXAiOiJCZWFyZXIifQ.uYBzeL1rCzxCwLabrUnUGa1VGkkHnHIQstAYllt3vje_NpVzDGVH0AkBDqnAzV-2N031rk7Q-DWXsnFAEmJnVGT_B5dgjvrAG-hcRmluQgsWyOLF0QksUgkMb6pXI5vmUsQwnyDwqii7oki1LHP32-aohtPqb7WltPP3FiyRxkENz6DLdMf7xk4At72TQgbnTMYhuJE9lU1DBhnIx7RMh1Z4oDjE45XaXeWnJk_spMBQCCMGy-QVayLVKiIumaFu3lNgH52lo13at7ozu1MHVDmU0lqXXAgNtjctwevOqwrpUE_Hbjja01L86hnbw1QyV6J6fwDWi-COkRVsqztMAw", + "id_token": "eyJhbGciOiJSUzI1NiIsImtpZCI6ImEyMjM3ZDhhLWQ0ZDQtNDA5Yi04ZTMxLWM3NGJhYTZhM2NjYiIsInR5cCI6IkpXVCJ9.eyJhdWQiOlsiaWFtIl0sImF1dGhfdGltZSI6MTc4NjU0OTQzMCwiYXpwIjoiMTJmN2UxZTcwYTgyZTg2NTgxZjkzNGU4MDNiYjdhMTEiLCJlbWFpbCI6InRpbW9mZXlAZ3VsZW5vay5ydSIsImVtYWlsX3ZlcmlmaWVkIjp0cnVlLCJleHAiOjE3ODY1NTMwMzAsImZhbWlseV9uYW1lIjoi0JPRg9C70LXQvdC-0LoiLCJnaXZlbl9uYW1lIjoi0KLQuNC80L7RhNC10LkiLCJpYXQiOjE3ODY1NDk0MzAsImlzcyI6Imh0dHBzOi8vaWQuY2xvdWQucnUvYXV0aC9zeXN0ZW0iLCJqdGkiOiJjYjI4YzE4MC05ZmQ5LTRhMmUtYjY1My0wNTJmYjZkOGZiOWUiLCJuYmYiOjE3ODY1NDk0MzAsIm5vbmNlIjoiIiwicGhvbmVfbnVtYmVyIjoiNzkxNjY5ODAzMDYiLCJwcmVmZXJyZWRfdXNlcm5hbWUiOiJ0aW1vZmV5QGd1bGVub2sucnUiLCJyZXNvdXJjZV9hY2Nlc3MiOm51bGwsInNjb3BlIjoicm9sZXMgZW1haWwgb3BlbmlkIHByb2ZpbGUiLCJzdWIiOiI5ZmZmYmQyNy02YWUwLTRiNjMtYWFkNC0wYjZjOWI1YjJkOTAiLCJzdWJfaWQiOiI5ZmZmYmQyNy02YWUwLTRiNjMtYWFkNC0wYjZjOWI1YjJkOTAiLCJzdWJfdHlwZSI6InVzZXIiLCJ0eXAiOiJJRCJ9.azbZqrLyBY5mRGneLCS_7ss_JhlamXUHgcjJ_QM2mMJOvPBZDBDjG-ORgUkKXtoRiS9mPbBBQcOw5RCko7ECwwBNQsqkDrYS6bTuvSQR7YRKNlbRWGHP9ejn_B4ZlIHn-KwS9xmyKeN8lxoSpMLQWwOzVe7jWK2hoAxGHhgXjZYUmlp-XMGqqeRtIl_WVJP0a2MokX-qwQsaCVTRs6FTiIdff7qAr49NqqBWbR7vSAi_oIFrIhUz78UVwZMM_cZFv715q9CgJoOuRzfPn8EPur1ffvjMwWgaqXigRrLv53z9pgC7wKKpbnU0Rm9SmZCx5kk-LBmomm3O-2AFOiWnlg", + "refresh_token": "", + "expires_in": 3600, + "refresh_expires_in": 0, + "scopes": [ + "roles email openid profile" + ], + "token_type": "Bearer", + "not_before": 0 +}