Leonid Pershin
2005b00175
- Added `ollama-models.yaml` to .gitignore and implemented logic to copy it in gpu-rent.ps1 and gpu-rent.sh. - Enhanced env.example to include new variables for LLM runtime options and local watchdog configuration. - Updated CLI commands to support LLM options during setup and execution, including new flags for Ollama and llama.cpp. - Improved documentation in cli.md and README.md to reflect changes in LLM integration and local watchdog functionality. - Adjusted architecture and decisions documentation to clarify the role of LLMs and local watchdog in the system.
568 lines
19 KiB
Python
568 lines
19 KiB
Python
"""Find-or-create Selectel OpenStack resources for one gpu-rent session."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any, Callable
|
||
|
||
import httpx
|
||
|
||
from gpu_rent.errors import CloudError
|
||
from gpu_rent.os_client import (
|
||
BOOT_VOLUME_NAME,
|
||
BOOT_VOLUME_SIZE_GB,
|
||
DATA_VOLUME_NAME,
|
||
KEYPAIR_NAME,
|
||
NET_NAME,
|
||
PREEMPTIBLE_TAG,
|
||
RESOURCE_TAG,
|
||
ROUTER_NAME,
|
||
SERVER_NAME,
|
||
SG_NAME,
|
||
SUBNET_CIDR,
|
||
SUBNET_NAME,
|
||
find_snapshot_by_name,
|
||
find_tagged_servers,
|
||
find_volumes_by_name,
|
||
)
|
||
|
||
|
||
def _wrap(exc: BaseException, what: str) -> CloudError:
|
||
text = str(exc)
|
||
low = text.lower()
|
||
if "402" in text or "payment" in low:
|
||
return CloudError(f"{what}: нет средств / 402. {text}")
|
||
if "403" in text or "forbidden" in low or "quota" in low:
|
||
return CloudError(
|
||
f"{what}: квота или запрет (403). Часто GPU=0 — тикет в поддержку, docs/setup.md. {text}"
|
||
)
|
||
return CloudError(f"{what}: {text}")
|
||
|
||
|
||
def _oid(obj: Any) -> str:
|
||
return str(getattr(obj, "id", "") or "")
|
||
|
||
|
||
def guess_operator_cidr() -> str:
|
||
"""Public egress /32 for SSH SG. Override: GPU_RENT_SSH_CIDR=0.0.0.0/0."""
|
||
import os
|
||
|
||
override = (os.environ.get("GPU_RENT_SSH_CIDR") or "").strip()
|
||
if override:
|
||
return override
|
||
for url in (
|
||
"https://ifconfig.me/ip",
|
||
"https://api.ipify.org",
|
||
"https://icanhazip.com",
|
||
):
|
||
try:
|
||
response = httpx.get(url, timeout=5.0, follow_redirects=True)
|
||
ip = response.text.strip().split()[0]
|
||
if ip.count(".") == 3 and all(p.isdigit() for p in ip.split(".")):
|
||
return f"{ip}/32"
|
||
except httpx.HTTPError:
|
||
continue
|
||
return "0.0.0.0/0"
|
||
|
||
|
||
def wait_volume(conn, volume, status: str = "available", timeout: int = 900) -> Any:
|
||
deadline = time.time() + timeout
|
||
vid = _oid(volume)
|
||
last = None
|
||
while time.time() < deadline:
|
||
current = conn.block_storage.get_volume(vid)
|
||
last = getattr(current, "status", None)
|
||
if last == status:
|
||
return current
|
||
if last in {"error", "error_restoring"}:
|
||
raise CloudError(f"том {vid} статус {last}")
|
||
time.sleep(5)
|
||
raise CloudError(f"том {vid} не стал {status} (последний {last})")
|
||
|
||
|
||
def wait_server(conn, server, status: str = "ACTIVE", timeout: int = 900) -> Any:
|
||
deadline = time.time() + timeout
|
||
sid = _oid(server)
|
||
last = None
|
||
while time.time() < deadline:
|
||
current = conn.compute.get_server(sid)
|
||
last = (getattr(current, "status", None) or "").upper()
|
||
if last == status.upper():
|
||
return current
|
||
if last in {"ERROR"}:
|
||
fault = getattr(current, "fault", None)
|
||
raise CloudError(f"сервер {sid} ERROR{f' {fault}' if fault else ''}")
|
||
time.sleep(5)
|
||
raise CloudError(f"сервер {sid} не стал {status} (последний {last})")
|
||
|
||
|
||
def wait_gone(fetch: Callable[[], Any], timeout: int = 300) -> None:
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
try:
|
||
obj = fetch()
|
||
except Exception:
|
||
return
|
||
if obj is None:
|
||
return
|
||
time.sleep(4)
|
||
raise CloudError("ресурс не исчез вовремя")
|
||
|
||
|
||
def find_external_network(conn) -> Any:
|
||
for net in conn.network.networks():
|
||
if getattr(net, "is_router_external", False):
|
||
return net
|
||
for net in conn.network.networks():
|
||
name = (getattr(net, "name", "") or "").lower()
|
||
if name in {"external-network", "wan", "public"}:
|
||
return net
|
||
raise CloudError("нет внешней сети для floating IP / router gateway")
|
||
|
||
|
||
def ensure_network(conn, log: Callable[[str], None]) -> tuple[Any, Any]:
|
||
net = conn.network.find_network(NET_NAME)
|
||
if net:
|
||
subnet = next(iter(conn.network.subnets(network_id=net.id)), None)
|
||
if not subnet:
|
||
raise CloudError(f"сеть {NET_NAME} есть, subnet нет")
|
||
log(f"сеть {NET_NAME} уже есть")
|
||
return net, subnet
|
||
try:
|
||
net = conn.network.create_network(name=NET_NAME)
|
||
subnet = conn.network.create_subnet(
|
||
name=SUBNET_NAME,
|
||
network_id=net.id,
|
||
ip_version=4,
|
||
cidr=SUBNET_CIDR,
|
||
dns_nameservers=["1.1.1.1", "8.8.8.8"],
|
||
)
|
||
ext = find_external_network(conn)
|
||
router = conn.network.find_router(ROUTER_NAME)
|
||
if not router:
|
||
router = conn.network.create_router(
|
||
name=ROUTER_NAME,
|
||
external_gateway_info={"network_id": ext.id},
|
||
)
|
||
conn.network.add_interface_to_router(router, subnet_id=subnet.id)
|
||
log(f"создана сеть {NET_NAME} + router")
|
||
return net, subnet
|
||
except Exception as exc:
|
||
raise _wrap(exc, "сеть") from exc
|
||
|
||
|
||
def ensure_security_group(conn, cidr: str, log: Callable[[str], None]) -> Any:
|
||
sg = conn.network.find_security_group(SG_NAME)
|
||
created = False
|
||
if not sg:
|
||
try:
|
||
sg = conn.network.create_security_group(
|
||
name=SG_NAME,
|
||
description="gpu-rent SSH only",
|
||
)
|
||
created = True
|
||
except Exception as exc:
|
||
raise _wrap(exc, "security group") from exc
|
||
|
||
# Ensure TCP/22 from operator CIDR; drop stale SSH /32s from old VPN/WARP IPs.
|
||
have_cidr = False
|
||
stale_ssh: list[Any] = []
|
||
try:
|
||
for rule in conn.network.security_group_rules(security_group_id=sg.id):
|
||
if (
|
||
getattr(rule, "direction", None) != "ingress"
|
||
or getattr(rule, "protocol", None) != "tcp"
|
||
or int(getattr(rule, "port_range_min", 0) or 0) != 22
|
||
or int(getattr(rule, "port_range_max", 0) or 0) != 22
|
||
):
|
||
continue
|
||
prefix = getattr(rule, "remote_ip_prefix", None) or ""
|
||
if prefix == cidr:
|
||
have_cidr = True
|
||
elif prefix:
|
||
stale_ssh.append(rule)
|
||
except Exception:
|
||
have_cidr = False
|
||
stale_ssh = []
|
||
|
||
for rule in stale_ssh:
|
||
old = getattr(rule, "remote_ip_prefix", None) or "?"
|
||
try:
|
||
conn.network.delete_security_group_rule(rule.id)
|
||
log(f"SG {SG_NAME}: − устаревший TCP/22 с {old}")
|
||
except Exception as exc:
|
||
log(f"SG {SG_NAME}: не удалить stale {old}: {exc}")
|
||
|
||
if not have_cidr:
|
||
try:
|
||
conn.network.create_security_group_rule(
|
||
security_group_id=sg.id,
|
||
direction="ingress",
|
||
ethertype="IPv4",
|
||
protocol="tcp",
|
||
port_range_min=22,
|
||
port_range_max=22,
|
||
remote_ip_prefix=cidr,
|
||
)
|
||
log(f"SG {SG_NAME}: + TCP/22 с {cidr}")
|
||
except Exception as exc:
|
||
text = str(exc).lower()
|
||
if "already" not in text and "conflict" not in text and "409" not in text:
|
||
raise _wrap(exc, "security group rule") from exc
|
||
log(f"SG {SG_NAME}: TCP/22 с {cidr} уже есть")
|
||
elif created:
|
||
log(f"SG {SG_NAME}: TCP/22 с {cidr}")
|
||
else:
|
||
log(f"security group {SG_NAME} уже есть (SSH {cidr})")
|
||
return sg
|
||
|
||
|
||
def ensure_keypair(conn, public_key: str, log: Callable[[str], None]) -> Any:
|
||
existing = conn.compute.find_keypair(KEYPAIR_NAME)
|
||
pub = public_key.strip()
|
||
if existing:
|
||
have = (getattr(existing, "public_key", "") or "").strip()
|
||
if have == pub:
|
||
log(f"keypair {KEYPAIR_NAME} совпадает")
|
||
return existing
|
||
log(f"keypair {KEYPAIR_NAME} другой — пересоздаём")
|
||
try:
|
||
conn.compute.delete_keypair(existing)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "удалить keypair") from exc
|
||
try:
|
||
key = conn.compute.create_keypair(name=KEYPAIR_NAME, public_key=pub)
|
||
log(f"keypair {KEYPAIR_NAME} создан")
|
||
return key
|
||
except Exception as exc:
|
||
raise _wrap(exc, "keypair") from exc
|
||
|
||
|
||
def _get_volume_or_none(conn, volume_id: str) -> Any | None:
|
||
if not volume_id:
|
||
return None
|
||
try:
|
||
return conn.block_storage.get_volume(volume_id)
|
||
except Exception as exc:
|
||
text = str(exc).lower()
|
||
if "404" in text or "not found" in text or "could not be found" in text:
|
||
return None
|
||
raise
|
||
|
||
|
||
def ensure_boot_volume(
|
||
conn,
|
||
*,
|
||
az: str,
|
||
volume_type: str | None,
|
||
image_id: str | None,
|
||
snapshot_name: str,
|
||
existing_id: str,
|
||
log: Callable[[str], None],
|
||
) -> Any:
|
||
if existing_id:
|
||
vol = _get_volume_or_none(conn, existing_id)
|
||
if vol is not None:
|
||
log(f"boot volume из state {existing_id}")
|
||
return vol
|
||
log(f"boot volume {existing_id} нет в облаке — создаём заново")
|
||
found = find_volumes_by_name(conn, BOOT_VOLUME_NAME)
|
||
if found:
|
||
log(f"boot volume {BOOT_VOLUME_NAME} уже в проекте")
|
||
return found[0]
|
||
snap = find_snapshot_by_name(conn, snapshot_name)
|
||
kwargs: dict[str, Any] = {
|
||
"name": BOOT_VOLUME_NAME,
|
||
"size": BOOT_VOLUME_SIZE_GB,
|
||
"availability_zone": az,
|
||
}
|
||
if volume_type:
|
||
kwargs["volume_type"] = volume_type
|
||
if snap:
|
||
kwargs["snapshot_id"] = snap.id
|
||
log(f"boot volume из snapshot {snapshot_name}")
|
||
elif image_id:
|
||
kwargs["image_id"] = image_id
|
||
log("boot volume из GPU-образа")
|
||
else:
|
||
raise CloudError("нет ни snapshot, ни image_id для boot volume")
|
||
try:
|
||
vol = conn.block_storage.create_volume(**kwargs)
|
||
return wait_volume(conn, vol)
|
||
except TypeError:
|
||
if "image_id" in kwargs:
|
||
kwargs["imageRef"] = kwargs.pop("image_id")
|
||
try:
|
||
vol = conn.block_storage.create_volume(**kwargs)
|
||
return wait_volume(conn, vol)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "boot volume") from exc
|
||
except Exception as exc:
|
||
raise _wrap(exc, "boot volume") from exc
|
||
|
||
|
||
def ensure_data_volume(
|
||
conn,
|
||
*,
|
||
az: str,
|
||
volume_type: str | None,
|
||
size_gb: int,
|
||
existing_id: str,
|
||
log: Callable[[str], None],
|
||
) -> Any:
|
||
if existing_id:
|
||
vol = _get_volume_or_none(conn, existing_id)
|
||
if vol is not None:
|
||
log(f"data volume из state {existing_id}")
|
||
return vol
|
||
log(f"data volume {existing_id} нет в облаке — создаём заново")
|
||
found = find_volumes_by_name(conn, DATA_VOLUME_NAME)
|
||
if found:
|
||
log(f"data volume {DATA_VOLUME_NAME} уже в проекте")
|
||
return found[0]
|
||
kwargs: dict[str, Any] = {
|
||
"name": DATA_VOLUME_NAME,
|
||
"size": size_gb,
|
||
"availability_zone": az,
|
||
}
|
||
if volume_type:
|
||
kwargs["volume_type"] = volume_type
|
||
try:
|
||
vol = conn.block_storage.create_volume(**kwargs)
|
||
log(f"data volume {size_gb} GB")
|
||
return wait_volume(conn, vol)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "data volume") from exc
|
||
|
||
|
||
def _tag_server(conn, server, spot: bool) -> None:
|
||
tags = [RESOURCE_TAG]
|
||
if spot:
|
||
tags.append(PREEMPTIBLE_TAG)
|
||
try:
|
||
conn.compute.set_server_tags(server, tags)
|
||
except Exception:
|
||
try:
|
||
for tag in tags:
|
||
conn.compute.add_tag_to_server(server, tag)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _ssh_user_data(public_key: str) -> str:
|
||
"""Minimal cloud-config (Selectel docs: ssh_authorized_keys)."""
|
||
key = public_key.strip()
|
||
return (
|
||
"#cloud-config\n"
|
||
"ssh_authorized_keys:\n"
|
||
f" - {key}\n"
|
||
"users:\n"
|
||
" - default\n"
|
||
" - name: ubuntu\n"
|
||
" lock_passwd: true\n"
|
||
" shell: /bin/bash\n"
|
||
" sudo: ['ALL=(ALL) NOPASSWD:ALL']\n"
|
||
" ssh_authorized_keys:\n"
|
||
f" - {key}\n"
|
||
)
|
||
|
||
|
||
def _user_data_b64(public_key: str) -> str:
|
||
import base64
|
||
|
||
return base64.b64encode(_ssh_user_data(public_key).encode("utf-8")).decode("ascii")
|
||
|
||
|
||
def create_gpu_server(
|
||
conn,
|
||
*,
|
||
flavor_id: str,
|
||
net_id: str,
|
||
sg_name: str,
|
||
boot_volume_id: str,
|
||
data_volume_id: str,
|
||
az: str,
|
||
spot: bool,
|
||
public_key: str,
|
||
log: Callable[[str], None],
|
||
) -> Any:
|
||
bdm = [
|
||
{
|
||
"boot_index": 0,
|
||
"uuid": boot_volume_id,
|
||
"source_type": "volume",
|
||
"destination_type": "volume",
|
||
"delete_on_termination": False,
|
||
},
|
||
{
|
||
"boot_index": 1,
|
||
"uuid": data_volume_id,
|
||
"source_type": "volume",
|
||
"destination_type": "volume",
|
||
"delete_on_termination": False,
|
||
},
|
||
]
|
||
tags = [RESOURCE_TAG]
|
||
if spot:
|
||
tags.append(PREEMPTIBLE_TAG)
|
||
base: dict[str, Any] = {
|
||
"name": SERVER_NAME,
|
||
"flavor_id": flavor_id,
|
||
"networks": [{"uuid": net_id}],
|
||
"key_name": KEYPAIR_NAME,
|
||
"availability_zone": az,
|
||
"block_device_mapping_v2": bdm,
|
||
"security_groups": [{"name": sg_name}],
|
||
}
|
||
ud_b64 = _user_data_b64(public_key)
|
||
ud_plain = _ssh_user_data(public_key)
|
||
|
||
# Selectel: OpenStack API expects Base64 user_data; config_drive often unsupported.
|
||
attempts: list[tuple[str, dict[str, Any]]] = [
|
||
("tags+user_data_b64", {**base, "tags": tags, "user_data": ud_b64}),
|
||
("user_data_b64", {**base, "user_data": ud_b64}),
|
||
("user_data_plain", {**base, "user_data": ud_plain}),
|
||
("key_name_only", {**base, "tags": tags}),
|
||
("minimal", dict(base)),
|
||
]
|
||
server = None
|
||
last_exc: BaseException | None = None
|
||
used = ""
|
||
for label, kwargs in attempts:
|
||
try:
|
||
server = conn.compute.create_server(**kwargs)
|
||
used = label
|
||
break
|
||
except Exception as exc:
|
||
last_exc = exc
|
||
log(f"create {label}: {exc}")
|
||
if server is None:
|
||
raise _wrap(last_exc or RuntimeError("create failed"), "create server")
|
||
|
||
if "user_data" not in used:
|
||
log(
|
||
f"create OK ({used}) без user_data — ключ может не попасть на boot-from-volume; "
|
||
"смотри docs Selectel user-data (Base64)"
|
||
)
|
||
else:
|
||
log(f"create OK ({used})")
|
||
|
||
log("ждём Nova ACTIVE (GPU create может занять несколько минут)")
|
||
server = wait_server(conn, server, "ACTIVE")
|
||
_tag_server(conn, server, spot)
|
||
return server
|
||
|
||
|
||
def unshelve(conn, server, log: Callable[[str], None]) -> Any:
|
||
log("unshelve (EXPIRED / shelved)")
|
||
try:
|
||
conn.compute.unshelve_server(server)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "unshelve") from exc
|
||
return wait_server(conn, server, "ACTIVE", timeout=900)
|
||
|
||
|
||
def associate_floating_ip(conn, server, log: Callable[[str], None]) -> tuple[str, str]:
|
||
ports = list(conn.network.ports(device_id=server.id))
|
||
if not ports:
|
||
raise CloudError("у сервера нет neutron-порта — не к чему привязать FIP")
|
||
ext = find_external_network(conn)
|
||
try:
|
||
fip = conn.network.create_ip(floating_network_id=ext.id)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "floating IP allocate") from exc
|
||
try:
|
||
fip = conn.network.update_ip(fip, port_id=ports[0].id)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "floating IP associate") from exc
|
||
addr = getattr(fip, "floating_ip_address", None) or getattr(fip, "name", None)
|
||
log(f"floating IP {addr}")
|
||
return str(addr), _oid(fip)
|
||
|
||
|
||
def ensure_floating_ip(
|
||
conn,
|
||
server,
|
||
existing_id: str | None,
|
||
existing_addr: str | None,
|
||
log: Callable[[str], None],
|
||
) -> tuple[str, str]:
|
||
have = server_floating_ip(server)
|
||
if have:
|
||
return have, existing_id or ""
|
||
ports = list(conn.network.ports(device_id=server.id))
|
||
if existing_id and ports:
|
||
try:
|
||
fip = conn.network.update_ip(existing_id, port_id=ports[0].id)
|
||
addr = getattr(fip, "floating_ip_address", None) or existing_addr
|
||
log(f"вернули FIP {addr}")
|
||
return str(addr), existing_id
|
||
except Exception:
|
||
log("старый FIP не привязался — выделяем новый")
|
||
return associate_floating_ip(conn, server, log)
|
||
|
||
|
||
def delete_floating_ip(conn, fip_id: str | None, address: str | None, log: Callable[[str], None]) -> None:
|
||
if fip_id:
|
||
try:
|
||
conn.network.delete_ip(fip_id, ignore_missing=True)
|
||
log(f"удалён FIP {fip_id}")
|
||
return
|
||
except Exception:
|
||
pass
|
||
if address:
|
||
for ip in conn.network.ips():
|
||
if getattr(ip, "floating_ip_address", None) == address:
|
||
conn.network.delete_ip(ip, ignore_missing=True)
|
||
log(f"удалён FIP {address}")
|
||
return
|
||
|
||
|
||
def delete_server(conn, server, log: Callable[[str], None]) -> None:
|
||
sid = _oid(server)
|
||
log(f"удаляем compute {sid} (диски оставляем)")
|
||
try:
|
||
conn.compute.delete_server(server, ignore_missing=True)
|
||
except Exception as exc:
|
||
raise _wrap(exc, "delete server") from exc
|
||
|
||
def _get():
|
||
try:
|
||
current = conn.compute.get_server(sid)
|
||
except Exception:
|
||
return None
|
||
if current is None:
|
||
return None
|
||
if server_status(current) in {"DELETED", "SOFT_DELETED"}:
|
||
return None
|
||
return current
|
||
|
||
wait_gone(_get, timeout=420)
|
||
|
||
|
||
def server_status(server: Any) -> str:
|
||
return (getattr(server, "status", None) or "").upper()
|
||
|
||
|
||
def server_floating_ip(server: Any) -> str | None:
|
||
addrs = getattr(server, "addresses", None) or {}
|
||
if not isinstance(addrs, dict):
|
||
return None
|
||
for nets in addrs.values():
|
||
if not isinstance(nets, list):
|
||
continue
|
||
for item in nets:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
if item.get("OS-EXT-IPS:type") == "floating":
|
||
return str(item.get("addr") or "") or None
|
||
return None
|
||
|
||
|
||
def pick_existing_server(conn) -> Any | None:
|
||
found = find_tagged_servers(conn)
|
||
if len(found) > 1:
|
||
ids = ", ".join(_oid(s) for s in found)
|
||
raise CloudError(f"несколько серверов gpu-rent: {ids}. Разбери вручную или --adopt один.")
|
||
return found[0] if found else None
|