petrbalvin/scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
scripts/vllm-deploy.py
T

2165 lines
76 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
# SPDX-License-Identifier: MIT
"""Idempotent vLLM deployment script for AMD ROCm GPUs.
Deploys vLLM behind nginx with self-signed TLS on Fedora, CentOS Stream, or openEuler.
vLLM binds to ::1 (IPv4 loopback fallback) on port 8000, internal only;
nginx proxies :443 → the loopback upstream with streaming (SSE) support.
The endpoint requires an API key, delivered to the service via a 0600
EnvironmentFile; nginx→vLLM proxying is allowed through SELinux on
enforcing systems. Fedora gets ROCm from its native repositories;
CentOS Stream and openEuler fall back to AMD's amdgpu-install (unsupported by AMD).
No external dependencies — stdlib and system tools only. Every operation
checks current state before acting; running the script repeatedly produces
the same result with no errors and no duplicate work.
Usage:
vllm-deploy.py # interactive model selection
vllm-deploy.py --model Qwen/Qwen3.6-35B-A3B # deploy specific model
vllm-deploy.py --model Qwen/Qwen3.6-35B-A3B --dry-run # preview
vllm-deploy.py --uninstall # tear down
vllm-deploy.py --version
"""
from __future__ import annotations
import argparse
import json
import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from typing import Any
BOLD = "\033[1m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
DIM = "\033[2m"
RESET = "\033[0m"
VERSION = "1.2.1"
SUPPORTED_OS: dict[str, str] = {
"fedora": "Fedora",
"centos": "CentOS Stream",
"openeuler": "openEuler",
}
# Tools this script itself needs (lspci, curl fetches, openssl for TLS).
DNF_PACKAGES: list[str] = ["pciutils", "curl", "openssl"]
# Fedora ships ROCm natively (maintained by the Fedora ROCm SIG) — preferred
# over AMD's amdgpu-install, which does not support Fedora.
FEDORA_ROCM_PACKAGES: list[str] = [
"rocm-hip",
"rocm-opencl",
"rocm-runtime",
"rocm-smi",
"rocminfo",
]
# Default models for interactive selection when --model is omitted.
# All IDs verified to exist (ungated) on Hugging Face as of 2026-07.
DEFAULT_MODELS: list[str] = [
"deepseek-ai/DeepSeek-V4-Pro",
"deepseek-ai/DeepSeek-V4-Flash",
"MiniMaxAI/MiniMax-M3",
"Qwen/Qwen3.6-35B-A3B",
"Qwen/Qwen3.6-27B",
]
# vLLM ROCm wheel resolution, per the official vLLM ROCm install docs:
# https://docs.vllm.ai/en/stable/getting_started/installation/gpu/
# The newest pin is scraped from the index at runtime; the constants below
# are the fallback (current stable example from the vLLM documentation).
WHEELS_INDEX = "https://wheels.vllm.ai/rocm"
FALLBACK_VLLM_VERSION = "0.25.0"
FALLBACK_ROCM_VARIANT = "rocm723"
AMDGPU_INSTALL_URL = "https://repo.radeon.com/amdgpu-install/latest/rhel"
# HuggingFace model IDs look like "org/name". The strict pattern also keeps
# systemd specifier characters (%) and whitespace out of unit files.
MODEL_ID_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$")
# ---------------------------------------------------------------------------
# Progress helpers — write to stderr so stdout stays clean
# ---------------------------------------------------------------------------
def _status(msg: str) -> None:
"""Print a progress message to stderr — stays on the same line until ``_status_done``."""
print(f" {msg}...", end="", flush=True, file=sys.stderr)
def _status_done(msg: str = "done") -> None:
"""Complete the current progress line on stderr."""
print(f" {msg}", file=sys.stderr)
def _info(msg: str) -> None:
"""Print an informational line to stderr."""
print(f" {DIM}{msg}{RESET}", file=sys.stderr)
def _warn(msg: str) -> None:
"""Print a warning to stderr."""
print(f" {YELLOW}⚠ {msg}{RESET}", file=sys.stderr)
def _ok(msg: str) -> None:
"""Print a success message to stderr."""
print(f" {GREEN}✓ {msg}{RESET}", file=sys.stderr)
def _fail(msg: str) -> None:
"""Print a failure message to stderr."""
print(f" {RED}✗ {msg}{RESET}", file=sys.stderr)
# ---------------------------------------------------------------------------
# Low-level command runner
# ---------------------------------------------------------------------------
def run(
cmd: list[str],
timeout: int = 300,
check: bool = False,
env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
"""Execute a command and return the CompletedProcess.
Sets LANG=C for reliable English-language output parsing.
Raises subprocess.CalledProcessError when *check* is True and the
command exits non-zero.
"""
merged_env = {**os.environ, "LANG": "C"}
if env:
merged_env.update(env)
return subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout, env=merged_env, check=check
)
def _find_uv() -> str | None:
"""Locate the uv binary: PATH first, then well-known install locations."""
uv = shutil.which("uv")
if uv:
return uv
candidates = (
"/usr/local/bin/uv",
"/root/.local/bin/uv",
os.path.expanduser("~/.local/bin/uv"),
)
for candidate in candidates:
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
return candidate
return None
def _fetch_text(url: str, timeout: int = 30) -> str:
"""Fetch a URL as text via curl -fsSL; return '' on failure (with a warning)."""
curl = shutil.which("curl")
if curl is None:
_warn("curl not found — cannot fetch remote version information")
return ""
try:
result = run([curl, "-fsSL", url], timeout=timeout)
except (OSError, subprocess.SubprocessError) as exc:
_warn(f"Failed to fetch {url}: {exc}")
return ""
if result.returncode != 0:
_warn(f"Failed to fetch {url} (curl exit {result.returncode})")
return ""
return result.stdout
# ---------------------------------------------------------------------------
# OS detection
# ---------------------------------------------------------------------------
def parse_os_release() -> dict[str, str]:
"""Parse /etc/os-release into a dict of KEY=VALUE pairs."""
result: dict[str, str] = {}
try:
with open("/etc/os-release", encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
key, _, value = line.partition("=")
value = value.strip().strip('"').strip("'")
result[key] = value
except OSError:
pass
return result
def detect_os() -> tuple[str, str, str]:
"""Detect the operating system.
Returns a tuple of (os_id, display_name, version_id).
Exits with an error message when the OS is not supported.
"""
release = parse_os_release()
os_id = release.get("ID", "").lower()
version_id = release.get("VERSION_ID", "unknown")
if os_id not in SUPPORTED_OS:
print(
f"{RED}{BOLD}Error:{RESET} Unsupported operating system: "
f"'{release.get('ID', 'unknown')}' (detected from /etc/os-release).",
file=sys.stderr,
)
print(
f" Supported systems: {', '.join(SUPPORTED_OS.values())}",
file=sys.stderr,
)
sys.exit(1)
return os_id, SUPPORTED_OS[os_id], version_id
# ---------------------------------------------------------------------------
# Root check
# ---------------------------------------------------------------------------
def check_root() -> None:
"""Exit with a clear message if not running as root (UID 0)."""
if os.geteuid() != 0:
print(
f"{RED}{BOLD}Error:{RESET} This script must be run as root (UID 0).",
file=sys.stderr,
)
print(
f" Current UID: {os.geteuid()}. Try: sudo python3 vllm-deploy.py --model ...",
file=sys.stderr,
)
sys.exit(1)
# ---------------------------------------------------------------------------
# Idempotency helpers
# ---------------------------------------------------------------------------
def _file_exists(path: str) -> bool:
"""Return True if the path exists as a file."""
return os.path.isfile(path)
def _dir_exists(path: str) -> bool:
"""Return True if the path exists as a directory."""
return os.path.isdir(path)
def _user_exists(username: str) -> bool:
"""Return True if the system user exists."""
result = run(["id", username])
return result.returncode == 0
def _group_exists(groupname: str) -> bool:
"""Return True if the system group exists."""
result = run(["getent", "group", groupname])
return result.returncode == 0
def _rpm_installed(pkg: str) -> bool:
"""Return True if the RPM package is already installed."""
result = run(["rpm", "-q", pkg])
return result.returncode == 0
def _systemctl_is_active(service: str) -> bool:
"""Return True if the systemd service is active."""
result = run(["systemctl", "is-active", "--quiet", service])
return result.returncode == 0
def _systemctl_is_enabled(service: str) -> bool:
"""Return True if the systemd service is enabled."""
result = run(["systemctl", "is-enabled", "--quiet", service])
return result.returncode == 0
def _map_to_el_major(os_id: str, version_id: str) -> str:
"""Map an OS version to the closest RHEL major ("8"/"9") for AMDGPU repos.
CentOS Stream VERSION_ID is already the RHEL major (e.g. "9", "10");
openEuler uses its own numbering (≥24 → "9", otherwise "8").
"""
if os_id.startswith("centos"):
try:
return str(int(version_id.split(".")[0]))
except (ValueError, IndexError):
return "9"
try:
major_ver = int(version_id.split(".")[0])
except (ValueError, IndexError):
major_ver = 0
return "9" if major_ver >= 24 else "8"
# ---------------------------------------------------------------------------
# GPU, ROCm and wheel-pin resolution helpers
# ---------------------------------------------------------------------------
def _detect_amd_gpus() -> list[str]:
"""Return AMD/ATI GPU descriptions from lspci -nn (empty list when none).
Only VGA/3D/Display controller class lines with the AMD/ATI vendor ID
(1002) match — AMD chipset, audio and USB lines found on AMD-CPU systems
are correctly ignored.
"""
lspci = shutil.which("lspci")
if lspci is None:
_fail("lspci not found — pciutils should have been installed in the dependencies step")
sys.exit(1)
result = run([lspci, "-nn"], timeout=30)
if result.returncode != 0:
_fail(f"lspci -nn failed: {result.stderr.strip()}")
sys.exit(1)
gpu_classes = ("VGA compatible controller", "3D controller", "Display controller")
gpus: list[str] = []
for line in result.stdout.split("\n"):
if "[1002:" in line and any(cls in line for cls in gpu_classes):
gpus.append(line.split(": ", 1)[-1].strip() if ": " in line else line.strip())
return gpus
def _detect_rocm_version() -> str | None:
"""Return the installed ROCm version (e.g. '7.1.0'), or None if not found."""
info_file = "/opt/rocm/.info/version"
if _file_exists(info_file):
try:
with open(info_file, encoding="utf-8") as fh:
version = fh.read().strip().split("-")[0]
if version:
return version
except OSError:
pass
# Fedora native packages install into /usr (no /opt/rocm).
result = run(["rpm", "-q", "--qf", "%{VERSION}", "rocm-runtime"], timeout=30)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip()
smi = shutil.which("rocm-smi")
if smi:
result = run([smi, "--version"], timeout=30)
match = re.search(r"(\d+\.\d+(?:\.\d+)?)", result.stdout + result.stderr)
if match:
return match.group(1)
return None
def _variant_to_version(variant: str) -> str:
"""Map a wheel ROCm variant ('rocm700', 'rocm723') to a dotted version."""
digits = variant.removeprefix("rocm")
return ".".join(digits) # '700' -> '7.0.0', '723' -> '7.2.3'
def _check_rocm_compat(rocm_version: str | None, variant: str) -> None:
"""Warn when the installed ROCm major version differs from the wheel's target."""
if rocm_version is None:
_warn("Could not determine the ROCm version — cannot verify wheel compatibility")
return
required = _variant_to_version(variant)
if rocm_version.split(".")[0] != required.split(".")[0]:
_warn(
f"ROCm {rocm_version} is installed but the vLLM wheel targets ROCm {required} "
f"({variant}) — runtime errors are likely; align the versions"
)
def _resolve_vllm_pin() -> tuple[str, str, str]:
"""Resolve (version, rocm_variant, index_url) for the newest stable ROCm wheel.
Scrapes the vLLM wheel index as recommended by the vLLM documentation;
falls back to FALLBACK_VLLM_VERSION/FALLBACK_ROCM_VARIANT when the index
is unreachable or unparsable.
"""
listing = _fetch_text(f"{WHEELS_INDEX}/vllm")
candidates = re.findall(r"vllm-(\d+\.\d+\.\d+)\+(rocm\d+)-", listing)
if not candidates:
if listing:
_warn("Could not parse the vLLM wheel index — using the fallback pin")
return (
FALLBACK_VLLM_VERSION,
FALLBACK_ROCM_VARIANT,
f"{WHEELS_INDEX}/{FALLBACK_VLLM_VERSION}/{FALLBACK_ROCM_VARIANT}",
)
version, variant = max(candidates, key=lambda c: ([int(p) for p in c[0].split(".")], c[1]))
return version, variant, f"{WHEELS_INDEX}/{version}/{variant}"
def _hf_model_status(model: str) -> str:
"""Probe Hugging Face for the model: 'ok', 'gated', 'missing', or 'unknown'.
Only a definitive 404 ('missing') is fatal; network problems yield
'unknown' so offline environments are not blocked.
"""
url = f"https://huggingface.co/api/models/{model}"
request = urllib.request.Request(url, headers={"User-Agent": f"vllm-deploy/{VERSION}"})
try:
with urllib.request.urlopen(request, timeout=15) as response:
payload = json.loads(response.read().decode("utf-8", "replace"))
return "gated" if payload.get("gated") not in (False, None) else "ok"
except urllib.error.HTTPError as exc:
if exc.code == 404:
return "missing"
if exc.code in (401, 403):
return "gated"
return "unknown"
except (urllib.error.URLError, OSError, json.JSONDecodeError):
return "unknown"
def _detect_loopback() -> tuple[str, str]:
"""Return (vllm_bind_host, nginx_upstream_host) — IPv6 ::1 first.
Falls back to 127.0.0.1 on kernels with IPv6 disabled
(net.ipv6.conf.all.disable_ipv6=1), where binding ::1 would fail.
"""
if os.path.exists("/proc/net/if_inet6"):
return "::1", "[::1]"
return "127.0.0.1", "127.0.0.1"
def _cert_san() -> str:
"""Build a subjectAltName value from the host FQDN and primary IPv4 (if resolvable)."""
fqdn = socket.getfqdn()
hostname = socket.gethostname()
entries = [f"DNS:{fqdn}"]
if hostname and hostname != fqdn:
entries.append(f"DNS:{hostname}")
try:
addr = socket.gethostbyname(fqdn)
if addr and not addr.startswith("127."):
entries.append(f"IP:{addr}")
except OSError:
pass
return ",".join(entries)
# ---------------------------------------------------------------------------
# 1. System dependencies
# ---------------------------------------------------------------------------
def install_system_deps(dry_run: bool = False) -> dict[str, Any]:
"""Install required system packages via dnf. Idempotent."""
results: dict[str, Any] = {"installed": [], "skipped": [], "failed": []}
missing: list[str] = []
for pkg in DNF_PACKAGES:
if _rpm_installed(pkg):
results["skipped"].append(pkg)
else:
missing.append(pkg)
if not missing:
_info(f"All {len(DNF_PACKAGES)} packages already installed")
else:
_status(f"Installing {len(missing)} package(s): {', '.join(missing)}")
if dry_run:
_status_done("dry run")
results["installed"] = missing
else:
result = run(["dnf", "install", "-y", *missing], timeout=600)
if result.returncode == 0:
results["installed"] = missing
_status_done("ok")
else:
_status_done("failed")
results["failed"] = missing
if result.stderr:
_info(f"dnf stderr: {result.stderr.strip()}")
# Install uv: dnf first (packaged in Fedora), standalone installer as
# fallback. Never pip-install into the system Python.
_status("Checking uv")
uv = _find_uv()
if uv is not None:
_status_done("already installed")
results["uv"] = uv
elif dry_run:
_status_done("would install via dnf")
results["uv"] = "dry run"
else:
uv_dnf = run(["dnf", "install", "-y", "uv"], timeout=300)
uv = _find_uv() if uv_dnf.returncode == 0 else None
if uv is not None:
_status_done(f"installed via dnf ({uv})")
results["uv"] = uv
else:
_info("uv not in distro repos — trying the official standalone installer")
installer = run(
["sh", "-c", "curl -fsSL https://astral.sh/uv/install.sh | sh"],
timeout=300,
)
uv = _find_uv() if installer.returncode == 0 else None
if uv is not None:
_status_done(f"installed via installer ({uv})")
results["uv"] = uv
else:
_status_done("failed")
results["uv"] = None
_fail("uv could not be installed — vLLM venv creation will fail")
# Packages expected from server-setup.py (warning only, not installed here).
required_server_packages: dict[str, str] = {
"nginx": "reverse proxy",
"git": "source control",
"python3": "Python runtime",
"python3-devel": "Python development headers",
}
for pkg, purpose in required_server_packages.items():
if not _rpm_installed(pkg):
_warn(f"{pkg} ({purpose}) is not installed — run server-setup.py first")
return results
# ---------------------------------------------------------------------------
# 2. Pre-flight checks
# ---------------------------------------------------------------------------
def _install_rocm_fedora(dry_run: bool) -> None:
"""Install ROCm from Fedora's native repositories (Fedora ROCm SIG)."""
_status("Installing ROCm from Fedora repositories (this may take several minutes)")
missing = [pkg for pkg in FEDORA_ROCM_PACKAGES if not _rpm_installed(pkg)]
if not missing:
_status_done("already installed")
return
if dry_run:
_status_done("dry run")
_info(f"Would run: dnf install -y {' '.join(missing)}")
return
result = run(["dnf", "install", "-y", *missing], timeout=1800)
if result.returncode != 0:
_status_done("failed")
print(
f"\n {RED}{BOLD}Error:{RESET} Failed to install ROCm packages.",
file=sys.stderr,
)
if result.stderr:
_info(f"dnf stderr: {result.stderr.strip()}")
sys.exit(1)
_status_done("ok")
def _install_rocm_openeuler(os_id: str, version_id: str, dry_run: bool) -> None:
"""Install ROCm on CentOS Stream / openEuler via AMD's amdgpu-install script.
AMD does not list CentOS Stream or openEuler as supported ROCm platforms —
the RHEL packages are installed strictly on a best-effort basis and may
fail. The RHEL point-release directory and RPM filename are resolved
dynamically from the repository (nothing is hardcoded).
"""
_warn(
"AMD does not support ROCm on CentOS Stream / openEuler — "
"amdgpu-install targets RHEL. Continuing best-effort; this may fail "
"or produce a broken install."
)
el_major = _map_to_el_major(os_id, version_id)
_status(f"Resolving latest amdgpu-install release for el{el_major}")
listing = _fetch_text(f"{AMDGPU_INSTALL_URL}/")
rhel_dirs = [
d for d in re.findall(r'href="(\d+\.\d+)/"', listing) if d.startswith(f"{el_major}.")
]
rhel_version = max(rhel_dirs, key=lambda v: [int(p) for p in v.split(".")], default=None)
rpm_filename = ""
if rhel_version:
rpm_listing = _fetch_text(f"{AMDGPU_INSTALL_URL}/{rhel_version}/")
match = re.search(rf"amdgpu-install-[\d.]+-\d+\.el{el_major}\.noarch\.rpm", rpm_listing)
if match:
rpm_filename = match.group(0)
if dry_run:
_status_done("dry run")
if rpm_filename:
_info(f"Would fetch: {AMDGPU_INSTALL_URL}/{rhel_version}/{rpm_filename}")
else:
_warn(f"Could not resolve the latest RPM — check {AMDGPU_INSTALL_URL}/")
_info("Would run: amdgpu-install -y --usecase=rocm")
return
if not rpm_filename or rhel_version is None:
_status_done("failed")
print(
f"\n {RED}{BOLD}Error:{RESET} Could not resolve the latest amdgpu-install "
f"RPM under {AMDGPU_INSTALL_URL}/.",
file=sys.stderr,
)
sys.exit(1)
rpm_url = f"{AMDGPU_INSTALL_URL}/{rhel_version}/{rpm_filename}"
ver_match = re.match(r"amdgpu-install-([\d.]+)", rpm_filename)
detected = ver_match.group(1) if ver_match else "latest"
_status_done(f"amdgpu-install {detected} (rhel {rhel_version})")
_status(f"Installing amdgpu-install {detected}")
install_result = run(["dnf", "install", "-y", rpm_url], timeout=300)
if install_result.returncode != 0:
_status_done("failed")
print(
f"\n {RED}{BOLD}Error:{RESET} Failed to install amdgpu-install.",
file=sys.stderr,
)
print(f" URL: {rpm_url}", file=sys.stderr)
if install_result.stderr:
_info(f"dnf stderr: {install_result.stderr.strip()}")
sys.exit(1)
_status_done("ok")
_status("Installing ROCm stack (this may take several minutes)")
rocm_result = run(["amdgpu-install", "-y", "--usecase=rocm"], timeout=1800)
if rocm_result.returncode != 0:
_status_done("failed")
print(
f"\n {RED}{BOLD}Error:{RESET} amdgpu-install failed.",
file=sys.stderr,
)
if rocm_result.stderr:
_info(f"stderr: {rocm_result.stderr.strip()}")
sys.exit(1)
_status_done("ok")
def preflight_checks(os_id: str, version_id: str, dry_run: bool = False) -> dict[str, Any]:
"""Run pre-flight checks: root, GPU hardware, wheel pin, ROCm.
Returns a dict with check results (including the resolved vLLM pin).
Exits with a clear error on any hard failure.
"""
results: dict[str, Any] = {}
# Root (idempotent no-op — main() already checked; keeps the status line).
_status("Checking root privileges")
check_root()
_status_done("root")
# OS (already detected by main for the header).
_status("Detecting operating system")
_status_done(f"{SUPPORTED_OS[os_id]} {version_id}")
results["os_id"] = os_id
results["os_display"] = SUPPORTED_OS[os_id]
results["version_id"] = version_id
# AMD GPU hardware — class + vendor filtered, so AMD chipsets never match.
_status("Checking GPU hardware")
gpus = _detect_amd_gpus()
if not gpus:
_status_done("not found")
print(
f"\n {RED}{BOLD}Error:{RESET} No AMD GPU detected in this system — "
"vLLM requires an AMD GPU with ROCm support.",
file=sys.stderr,
)
sys.exit(1)
if len(gpus) == 1:
_status_done(gpus[0])
else:
_status_done(f"{gpus[0]} (+{len(gpus) - 1} more)")
results["gpu_models"] = gpus
results["gpu_count"] = len(gpus)
# Resolve the vLLM ROCm wheel pin (also used for the ROCm version check).
_status("Resolving vLLM ROCm wheel version")
vllm_version, rocm_variant, wheel_index = _resolve_vllm_pin()
_status_done(f"vllm=={vllm_version}+{rocm_variant}")
results["vllm_version"] = vllm_version
results["rocm_variant"] = rocm_variant
results["wheel_index"] = wheel_index
# ROCm
_status("Checking ROCm installation")
rocm_version = _detect_rocm_version()
rocm_present = (
rocm_version is not None or _dir_exists("/opt/rocm") or shutil.which("rocm-smi") is not None
)
if rocm_present:
_status_done(f"present ({rocm_version or 'unknown version'})")
results["rocm"] = True
_check_rocm_compat(rocm_version, rocm_variant)
else:
_status_done("not found")
if os_id == "fedora":
_install_rocm_fedora(dry_run)
else:
_install_rocm_openeuler(os_id, version_id, dry_run)
results["rocm"] = True
# GPU verification via rocm-smi.
_status("Verifying GPU via rocm-smi")
if dry_run and not rocm_present:
_status_done("would verify")
_info("Would run: rocm-smi --showproductname")
return results
smi = shutil.which("rocm-smi")
if smi is None and _file_exists("/opt/rocm/bin/rocm-smi"):
smi = "/opt/rocm/bin/rocm-smi"
if smi is None:
_status_done("rocm-smi not found")
print(
f"\n {RED}{BOLD}Error:{RESET} rocm-smi not found — ROCm installation "
"appears incomplete.",
file=sys.stderr,
)
sys.exit(1)
gpu_result = run([smi, "--showproductname"], timeout=30)
if gpu_result.returncode != 0:
_status_done("no GPU detected")
print(
f"\n {RED}{BOLD}Error:{RESET} No AMD GPU detected via rocm-smi.",
file=sys.stderr,
)
print(
" ROCm installed but no compatible GPU found.",
file=sys.stderr,
)
if gpu_result.stderr:
print(f" rocm-smi stderr: {gpu_result.stderr.strip()}", file=sys.stderr)
sys.exit(1)
_status_done("available")
# Count unique GPU[n] indices — rocm-smi prints several lines per GPU.
indices = set(re.findall(r"GPU\[(\d+)\]", gpu_result.stdout))
if indices:
results["gpu_count"] = len(indices)
results["gpu_info"] = gpu_result.stdout.strip()
return results
# ---------------------------------------------------------------------------
# 3. Create vllm system user
# ---------------------------------------------------------------------------
def setup_vllm_user(dry_run: bool = False) -> dict[str, Any]:
"""Create the vllm system user and add to GPU groups. Idempotent."""
results: dict[str, Any] = {}
_status("Checking vllm system user")
if _user_exists("vllm"):
_status_done("already exists")
results["user_exists"] = True
else:
if dry_run:
_status_done("would create")
results["user_created"] = "dry run"
else:
result = run(
["useradd", "-r", "-s", "/sbin/nologin", "-d", "/opt/vllm", "vllm"],
timeout=30,
)
if result.returncode == 0:
_status_done("created")
results["user_created"] = True
else:
_status_done("failed")
_fail(f"useradd failed: {result.stderr.strip()}")
results["user_created"] = False
return results
# Add to render and video groups for GPU access.
for group in ("render", "video"):
_status(f"Adding vllm to {group} group")
if not _group_exists(group):
_status_done("group does not exist")
_warn(f"Group '{group}' not found — GPU access may be limited")
continue
if dry_run:
_status_done("dry run")
continue
result = run(["usermod", "-aG", group, "vllm"], timeout=30)
if result.returncode == 0:
_status_done("ok")
else:
_status_done("failed")
_warn(f"Failed to add vllm to {group}: {result.stderr.strip()}")
return results
# ---------------------------------------------------------------------------
# 4. Python venv with uv + vLLM for ROCm
# ---------------------------------------------------------------------------
def _chown_vllm(path: str, recursive: bool = False) -> None:
"""chown a path to vllm:vllm, warning (not crashing) on failure."""
cmd = ["chown"] + (["-R"] if recursive else []) + ["vllm:vllm", path]
result = run(cmd, timeout=60)
if result.returncode != 0:
_warn(f"chown failed for {path}: {result.stderr.strip()}")
def _installed_vllm_version(uv: str | None, venv_dir: str) -> str | None:
"""Return the vLLM version installed in the venv, or None if unknown."""
if uv is None:
return None
result = run([uv, "pip", "show", "--python", f"{venv_dir}/bin/python", "vllm"], timeout=60)
if result.returncode != 0:
return None
for line in result.stdout.split("\n"):
if line.startswith("Version:"):
return line.split(":", 1)[1].strip()
return None
def setup_venv(
vllm_version: str,
rocm_variant: str,
index_url: str,
venv_dir: str,
dry_run: bool = False,
) -> dict[str, Any]:
"""Create Python venv and install vLLM for ROCm via uv. Idempotent."""
results: dict[str, Any] = {}
# Ensure /opt/vllm exists and is owned by vllm.
_status("Ensuring /opt/vllm directory")
if not _dir_exists("/opt/vllm"):
if dry_run:
_status_done("would create")
else:
os.makedirs("/opt/vllm", exist_ok=True)
_chown_vllm("/opt/vllm")
_status_done("created")
else:
_status_done("exists")
# Create venv if missing. Stable ROCm wheels support Python 3.13 (uv downloads it automatically).
_status("Creating Python venv")
uv = _find_uv()
if _dir_exists(venv_dir):
_status_done("already exists")
results["venv_exists"] = True
else:
if dry_run:
_status_done("dry run")
results["venv_created"] = "dry run"
else:
if uv is None:
_status_done("uv not found")
_fail("uv is not installed — cannot create the venv")
results["venv_created"] = False
return results
venv_result = run([uv, "venv", "--python", "3.13", venv_dir], timeout=300)
if venv_result.returncode == 0:
_status_done("created")
results["venv_created"] = True
else:
_status_done("failed")
_fail(f"uv venv failed: {venv_result.stderr.strip()}")
results["venv_created"] = False
return results
if not dry_run and _dir_exists(venv_dir):
_chown_vllm(venv_dir, recursive=True)
# Idempotency: uv venvs have no pip — check for the vllm entry point and
# query the installed version via uv instead.
vllm_bin = f"{venv_dir}/bin/vllm"
_status("Checking vLLM installation")
if os.path.isfile(vllm_bin):
installed = _installed_vllm_version(uv, venv_dir)
_status_done(f"already installed (v{installed})" if installed else "already installed")
results["vllm_installed"] = True
results["vllm_version"] = installed or "unknown"
if installed and installed != vllm_version:
_warn(
f"Installed vLLM {installed} differs from the resolved pin {vllm_version} "
"— keeping the installed version (uninstall first to switch)"
)
return results
if dry_run:
_status_done("dry run")
_info(
f"Would run: uv pip install --python {venv_dir}/bin/python "
f"vllm=={vllm_version}+{rocm_variant} --extra-index-url {index_url}"
)
results["vllm_installed"] = "dry run"
return results
if uv is None:
_status_done("uv not found")
_fail("uv is not installed — cannot install vLLM")
results["vllm_installed"] = False
return results
_status(f"Installing vllm=={vllm_version}+{rocm_variant} (this may take several minutes)")
install_result = run(
[
uv,
"pip",
"install",
"--python",
f"{venv_dir}/bin/python",
f"vllm=={vllm_version}+{rocm_variant}",
"--extra-index-url",
index_url,
],
timeout=1800, # 30 min — vLLM is a large package
)
if install_result.returncode == 0:
_status_done("ok")
results["vllm_installed"] = True
results["vllm_version"] = vllm_version
else:
_status_done("failed")
_fail(f"vLLM install failed: {install_result.stderr.strip()[-500:]}")
results["vllm_installed"] = False
if not dry_run and _dir_exists(venv_dir):
_chown_vllm(venv_dir, recursive=True)
return results
# ---------------------------------------------------------------------------
# 5. TLS certificate (self-signed)
# ---------------------------------------------------------------------------
def setup_tls(cert_dir: str, dry_run: bool = False) -> dict[str, Any]:
"""Generate a self-signed TLS certificate (with SAN). Idempotent."""
results: dict[str, Any] = {}
key_path = os.path.join(cert_dir, "vllm.key")
crt_path = os.path.join(cert_dir, "vllm.crt")
_status("Checking TLS certificate")
if _file_exists(crt_path) and _file_exists(key_path):
_status_done("already exists")
results["cert_exists"] = True
return results
if dry_run:
_status_done("would generate")
results["cert_created"] = "dry run"
return results
openssl = shutil.which("openssl")
if openssl is None:
_status_done("openssl not found")
_fail("openssl is not installed — cannot generate a certificate")
results["cert_created"] = False
return results
if not _dir_exists(cert_dir):
os.makedirs(cert_dir, exist_ok=True)
_status("Generating self-signed certificate")
result = run(
[
openssl,
"req",
"-x509",
"-nodes",
"-days",
"365",
"-newkey",
"rsa:2048",
"-keyout",
key_path,
"-out",
crt_path,
"-subj",
f"/CN={socket.getfqdn()}",
"-addext",
f"subjectAltName={_cert_san()}",
],
timeout=30,
)
if result.returncode == 0:
os.chmod(key_path, 0o600)
os.chmod(crt_path, 0o644)
_status_done("generated")
results["cert_created"] = True
else:
_status_done("failed")
_fail(f"openssl failed: {result.stderr.strip()}")
results["cert_created"] = False
return results
# ---------------------------------------------------------------------------
# 6. API key (EnvironmentFile for the systemd unit)
# ---------------------------------------------------------------------------
def setup_api_key(
service_name: str,
api_key: str | None,
dry_run: bool = False,
) -> dict[str, Any]:
"""Store the vLLM API key in a 0600 EnvironmentFile. Idempotent.
The key comes from --api-key when given, otherwise it is generated once
and reused on subsequent runs. systemd (PID 1, root) reads the file, so
0600 root:vllm is sufficient — the service user never opens it.
"""
results: dict[str, Any] = {}
env_path = f"/etc/sysconfig/{service_name}"
current = ""
if _file_exists(env_path):
try:
with open(env_path, encoding="utf-8") as fh:
current = fh.read()
except OSError:
current = ""
_status("Checking API key environment file")
if api_key is None and "VLLM_API_KEY=" in current:
_status_done("already present — reusing existing key")
results["env_file"] = "exists"
return results
desired = f"VLLM_API_KEY={api_key}\n" if api_key else ""
if api_key is not None and current == desired:
_status_done("already configured")
results["env_file"] = "exists"
return results
if dry_run:
_status_done("would update" if current else "would create")
results["env_file"] = "dry run"
return results
key = api_key if api_key is not None else secrets.token_urlsafe(32)
try:
with open(env_path, "w", encoding="utf-8") as fh:
fh.write(f"VLLM_API_KEY={key}\n")
except OSError as exc:
_status_done("failed")
_fail(f"Could not write {env_path}: {exc}")
results["env_file"] = "failed"
return results
os.chmod(env_path, 0o600)
result = run(["chown", "root:vllm", env_path], timeout=30)
if result.returncode != 0:
_warn(f"chown failed for {env_path}: {result.stderr.strip()}")
_status_done("updated" if current else "created")
results["env_file"] = "updated" if current else "created"
if api_key is None:
results["generated_key"] = key
return results
# ---------------------------------------------------------------------------
# 7. SELinux
# ---------------------------------------------------------------------------
def setup_selinux(dry_run: bool = False) -> dict[str, Any]:
"""Allow nginx→vLLM proxying on SELinux-enforcing systems. Idempotent.
http_port_t covers 80/81/443/488/8008/8009/8443/9000 — not vLLM's port —
so on enforcing systems nginx needs httpd_can_network_connect.
"""
results: dict[str, Any] = {}
_status("Checking SELinux status")
getenforce = shutil.which("getenforce")
if getenforce is None:
_status_done("not installed — skipping")
results["selinux"] = "absent"
return results
mode_result = run([getenforce], timeout=10)
mode = mode_result.stdout.strip().lower()
if mode != "enforcing":
_status_done(f"{mode} — skipping")
results["selinux"] = mode
return results
_status_done("enforcing")
_status("Checking httpd_can_network_connect boolean")
getsebool = shutil.which("getsebool")
setsebool = shutil.which("setsebool")
if getsebool is None or setsebool is None:
_status_done("tools missing")
_warn("getsebool/setsebool not found — nginx proxying may be blocked (502)")
results["selinux"] = "failed"
return results
current = run([getsebool, "httpd_can_network_connect"], timeout=10)
if current.returncode == 0 and "--> on" in current.stdout:
_status_done("already on")
results["selinux"] = "on"
return results
if dry_run:
_status_done("would set on")
results["selinux"] = "dry run"
return results
# -P persists across reboots; the policy rebuild can take a while.
set_result = run([setsebool, "-P", "httpd_can_network_connect=1"], timeout=180)
if set_result.returncode == 0:
_status_done("set on")
results["selinux"] = "on"
else:
_status_done("failed")
_warn(f"setsebool failed: {set_result.stderr.strip()}")
results["selinux"] = "failed"
return results
# ---------------------------------------------------------------------------
# 8. nginx configuration
# ---------------------------------------------------------------------------
def _nginx_conf_content(port: int, upstream_host: str, cert_dir: str) -> str:
"""Return the nginx vllm server block content.
HTTP/1.1 + ``Connection ""`` + ``proxy_buffering off`` are required for
vLLM's SSE streaming — nginx's defaults (HTTP/1.0, buffering on) would
hold a whole streamed completion until generation finishes.
"""
return f"""server {{
listen [::]:443 ssl;
listen 443 ssl;
server_name _;
ssl_certificate {cert_dir}/vllm.crt;
ssl_certificate_key {cert_dir}/vllm.key;
ssl_protocols TLSv1.2 TLSv1.3;
client_max_body_size 50m;
location / {{
proxy_pass http://{upstream_host}:{port};
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_buffering off;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
}}
}}
"""
def setup_nginx(
port: int,
upstream_host: str,
cert_dir: str,
dry_run: bool = False,
) -> dict[str, Any]:
"""Configure nginx as reverse proxy for vLLM. Idempotent."""
results: dict[str, Any] = {}
conf_path = "/etc/nginx/conf.d/vllm.conf"
desired_content = _nginx_conf_content(port, upstream_host, cert_dir)
# Write nginx config if missing or different.
_status("Checking nginx configuration")
if _file_exists(conf_path):
try:
with open(conf_path, encoding="utf-8") as fh:
current = fh.read()
except OSError:
current = ""
if current.strip() == desired_content.strip():
_status_done("already configured")
results["nginx_configured"] = True
else:
if dry_run:
_status_done("would update")
results["nginx_configured"] = "dry run"
else:
with open(conf_path, "w", encoding="utf-8") as fh:
fh.write(desired_content)
_status_done("updated")
results["nginx_configured"] = True
else:
if dry_run:
_status_done("would create")
results["nginx_configured"] = "dry run"
else:
with open(conf_path, "w", encoding="utf-8") as fh:
fh.write(desired_content)
_status_done("created")
results["nginx_configured"] = True
# Ensure nginx is enabled and running (handles the enabled-but-stopped case).
_status("Ensuring nginx service is enabled and running")
nginx_running = _systemctl_is_active("nginx")
if _systemctl_is_enabled("nginx") and nginx_running:
_status_done("running")
results["nginx_running"] = True
elif dry_run:
_status_done("dry run")
results["nginx_running"] = "dry run"
else:
if not _systemctl_is_enabled("nginx"):
start_result = run(["systemctl", "enable", "--now", "nginx"], timeout=30)
else:
start_result = run(["systemctl", "start", "nginx"], timeout=30)
if start_result.returncode == 0:
_status_done("enabled and started")
results["nginx_running"] = True
else:
_status_done("failed")
_warn(f"Could not start nginx: {start_result.stderr.strip()}")
results["nginx_running"] = False
# Test and reload the configuration (only possible when nginx is running).
_status("Reloading nginx configuration")
if dry_run:
_status_done("dry run")
results["nginx_reloaded"] = "dry run"
elif not results.get("nginx_running"):
_status_done("skipped — nginx not running")
results["nginx_reloaded"] = False
else:
test_result = run(["nginx", "-t"], timeout=30)
if test_result.returncode != 0:
_status_done("config test failed")
_fail(f"nginx -t failed: {test_result.stderr.strip()}")
results["nginx_reloaded"] = False
else:
reload_result = run(["systemctl", "reload", "nginx"], timeout=30)
if reload_result.returncode == 0:
_status_done("reloaded")
results["nginx_reloaded"] = True
else:
_status_done("failed")
_fail(f"nginx reload failed: {reload_result.stderr.strip()}")
results["nginx_reloaded"] = False
return results
# ---------------------------------------------------------------------------
# 9. systemd service
# ---------------------------------------------------------------------------
def _systemd_content(
model: str,
port: int,
host: str,
tensor_parallel: int,
max_model_len: int,
gpu_memory_utilization: float,
venv_dir: str,
env_file: str,
) -> str:
"""Return the systemd unit file content.
The API key is substituted by systemd from the EnvironmentFile
(${VLLM_API_KEY}) — the key never appears in the unit file itself.
"""
return f"""[Unit]
Description=vLLM Inference Server ({model})
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=vllm
Group=vllm
WorkingDirectory=/opt/vllm
Environment=PYTHONUNBUFFERED=1
EnvironmentFile={env_file}
ExecStart={venv_dir}/bin/vllm serve {model} \\
--host {host} \\
--port {port} \\
--api-key ${{VLLM_API_KEY}} \\
--tensor-parallel-size {tensor_parallel} \\
--max-model-len {max_model_len} \\
--gpu-memory-utilization {gpu_memory_utilization}
Restart=on-failure
RestartSec=10
[Install]
WantedBy=multi-user.target
"""
def setup_systemd(
model: str,
port: int,
host: str,
tensor_parallel: int,
max_model_len: int,
gpu_memory_utilization: float,
venv_dir: str,
service_name: str,
dry_run: bool = False,
) -> dict[str, Any]:
"""Create and enable the vLLM systemd service. Idempotent.
When the unit file changed and the service is already active, it is
restarted (try-restart) so config changes actually take effect.
"""
results: dict[str, Any] = {}
unit_path = f"/etc/systemd/system/{service_name}.service"
env_file = f"/etc/sysconfig/{service_name}"
desired_content = _systemd_content(
model,
port,
host,
tensor_parallel,
max_model_len,
gpu_memory_utilization,
venv_dir,
env_file,
)
# Write unit file if missing or different.
unit_changed = False
_status(f"Checking {service_name}.service unit file")
if _file_exists(unit_path):
try:
with open(unit_path, encoding="utf-8") as fh:
current = fh.read()
except OSError:
current = ""
if current.strip() == desired_content.strip():
_status_done("already configured")
results["unit_configured"] = True
else:
if dry_run:
_status_done("would update")
results["unit_configured"] = "dry run"
else:
with open(unit_path, "w", encoding="utf-8") as fh:
fh.write(desired_content)
_status_done("updated")
results["unit_configured"] = True
unit_changed = True
else:
if dry_run:
_status_done("would create")
results["unit_configured"] = "dry run"
else:
with open(unit_path, "w", encoding="utf-8") as fh:
fh.write(desired_content)
_status_done("created")
results["unit_configured"] = True
unit_changed = True
results["unit_changed"] = unit_changed
# systemctl daemon-reload.
_status("Reloading systemd daemon")
if dry_run:
_status_done("dry run")
else:
run(["systemctl", "daemon-reload"], timeout=30)
_status_done("ok")
# Enable the service.
_status(f"Enabling {service_name} service")
if _systemctl_is_enabled(service_name):
_status_done("already enabled")
results["service_enabled"] = True
else:
if dry_run:
_status_done("dry run")
results["service_enabled"] = "dry run"
else:
enable_result = run(["systemctl", "enable", service_name], timeout=30)
if enable_result.returncode == 0:
_status_done("enabled")
results["service_enabled"] = True
else:
_status_done("failed")
_fail(f"systemctl enable failed: {enable_result.stderr.strip()}")
results["service_enabled"] = False
# Start the service — or restart it when the unit changed underneath it.
_status(f"Starting {service_name} service")
is_active = _systemctl_is_active(service_name)
if dry_run:
_status_done("dry run")
results["service_started"] = "dry run"
elif is_active and unit_changed:
restart_result = run(["systemctl", "try-restart", service_name], timeout=60)
if restart_result.returncode == 0:
_status_done("restarted (unit changed)")
results["service_started"] = "restarted"
else:
_status_done("failed")
_fail(
f"systemctl try-restart failed: {restart_result.stderr.strip()} "
f"(check logs with: journalctl -u {service_name} -f)"
)
results["service_started"] = False
elif is_active:
_status_done("already running")
results["service_started"] = True
else:
start_result = run(["systemctl", "start", service_name], timeout=30)
if start_result.returncode == 0:
_status_done("started")
results["service_started"] = True
else:
_status_done("failed")
_fail(
f"systemctl start failed: {start_result.stderr.strip()} "
f"(check logs with: journalctl -u {service_name} -f)"
)
results["service_started"] = False
return results
# ---------------------------------------------------------------------------
# 10. Firewall
# ---------------------------------------------------------------------------
def setup_firewall(dry_run: bool = False) -> dict[str, Any]:
"""Open HTTPS port (443) in firewalld. Idempotent."""
results: dict[str, Any] = {}
_status("Checking firewalld")
fw_result = run(["systemctl", "is-active", "--quiet", "firewalld"], timeout=10)
if fw_result.returncode != 0:
_status_done("not running — skipping")
results["firewall_active"] = False
return results
_status_done("active")
# Check if https service is already allowed.
_status("Checking HTTPS firewall rule")
list_result = run(["firewall-cmd", "--permanent", "--list-services"], timeout=30)
if list_result.returncode == 0 and "https" in list_result.stdout.split():
_status_done("already allowed")
results["firewall_configured"] = True
return results
if dry_run:
_status_done("dry run")
results["firewall_configured"] = "dry run"
return results
# Add https service.
add_result = run(["firewall-cmd", "--permanent", "--add-service=https"], timeout=30)
if add_result.returncode != 0:
_status_done("failed")
_warn(f"firewall-cmd failed: {add_result.stderr.strip()}")
results["firewall_configured"] = False
return results
# Reload to apply — a failed reload leaves the rule inactive.
reload_result = run(["firewall-cmd", "--reload"], timeout=30)
if reload_result.returncode == 0:
_status_done("added (permanent)")
results["firewall_configured"] = True
else:
_status_done("failed")
_warn(
f"firewall-cmd --reload failed: {reload_result.stderr.strip()} "
"— rule stored permanently but not active"
)
results["firewall_configured"] = False
return results
# ---------------------------------------------------------------------------
# 11. Uninstall
# ---------------------------------------------------------------------------
def uninstall(
venv_dir: str,
service_name: str,
cert_dir: str,
dry_run: bool = False,
) -> dict[str, Any]:
"""Tear down the vLLM deployment. Idempotent.
Deliberately kept (documented at the end): /opt/vllm with the
HuggingFace cache, the ROCm stack, the firewalld https rule and the
SELinux boolean.
"""
results: dict[str, Any] = {}
# Stop and disable the systemd service.
_status(f"Stopping {service_name} service")
if _systemctl_is_active(service_name) or _systemctl_is_enabled(service_name):
if dry_run:
_status_done("dry run")
else:
run(["systemctl", "stop", service_name], timeout=60)
run(["systemctl", "disable", service_name], timeout=30)
_status_done("stopped and disabled")
results["service_stopped"] = True
else:
_status_done("not running")
results["service_not_found"] = True
# Remove systemd unit file.
unit_path = f"/etc/systemd/system/{service_name}.service"
_status(f"Removing {service_name}.service unit file")
if _file_exists(unit_path):
if dry_run:
_status_done("dry run")
else:
os.remove(unit_path)
run(["systemctl", "daemon-reload"], timeout=30)
_status_done("removed")
results["unit_removed"] = True
else:
_status_done("not present")
results["unit_not_found"] = True
# Remove the API key environment file.
env_path = f"/etc/sysconfig/{service_name}"
_status("Removing API key environment file")
if _file_exists(env_path):
if dry_run:
_status_done("dry run")
else:
os.remove(env_path)
_status_done("removed")
results["env_removed"] = True
else:
_status_done("not present")
results["env_not_found"] = True
# Remove nginx config.
nginx_conf = "/etc/nginx/conf.d/vllm.conf"
_status("Removing nginx vllm configuration")
if _file_exists(nginx_conf):
if dry_run:
_status_done("dry run")
else:
os.remove(nginx_conf)
run(["systemctl", "reload", "nginx"], timeout=30)
_status_done("removed and nginx reloaded")
results["nginx_removed"] = True
else:
_status_done("not present")
results["nginx_not_found"] = True
# Remove TLS certificates.
_status("Removing TLS certificates")
if _dir_exists(cert_dir):
if dry_run:
_status_done("dry run")
else:
# Remove individual files first, then the directory.
for fname in ("vllm.key", "vllm.crt"):
fpath = os.path.join(cert_dir, fname)
if _file_exists(fpath):
os.remove(fpath)
try:
os.rmdir(cert_dir)
except OSError:
pass
_status_done("removed")
results["certs_removed"] = True
else:
_status_done("not present")
results["certs_not_found"] = True
# Remove venv.
_status(f"Removing venv at {venv_dir}")
if _dir_exists(venv_dir):
if dry_run:
_status_done("dry run")
else:
shutil.rmtree(venv_dir)
_status_done("removed")
results["venv_removed"] = True
else:
_status_done("not present")
results["venv_not_found"] = True
# Remove vllm user.
_status("Removing vllm system user")
if _user_exists("vllm"):
if dry_run:
_status_done("dry run")
else:
userdel_result = run(["userdel", "vllm"], timeout=30)
if userdel_result.returncode == 0:
_status_done("removed")
results["user_removed"] = True
else:
_status_done("failed")
_warn(f"userdel failed: {userdel_result.stderr.strip()}")
results["user_removed"] = False
else:
_status_done("not present")
results["user_not_found"] = True
# What deliberately persists after uninstall.
print(file=sys.stderr)
_info("Kept on the system (remove manually if unwanted):")
_info(" /opt/vllm — HuggingFace cache with downloaded model weights")
_info(" ROCm stack — via dnf or amdgpu-uninstall")
_info(" firewalld 'https' rule and the SELinux httpd_can_network_connect boolean")
return results
# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------
def print_summary(
model: str,
port: int,
service_name: str,
os_display: str,
version_id: str,
results: dict[str, Any],
warnings: list[str],
elapsed: float,
dry_run: bool = False,
uninstall_mode: bool = False,
bind_host: str = "",
generated_api_key: str | None = None,
) -> None:
"""Print a final summary of all sections to stderr."""
mode = "Uninstall" if uninstall_mode else ("Dry Run" if dry_run else "Setup")
print(f"\n{BOLD}── {mode} Summary ──{RESET}", file=sys.stderr)
print(f" OS: {os_display} {version_id}", file=sys.stderr)
if uninstall_mode:
# Uninstall summary.
ur = results.get("uninstall", {})
for key, label in [
("service_stopped", "vLLM service stopped"),
("unit_removed", "systemd unit removed"),
("env_removed", "API key environment file removed"),
("nginx_removed", "nginx config removed"),
("certs_removed", "TLS certificates removed"),
("venv_removed", "Python venv removed"),
("user_removed", "vLLM user removed"),
]:
if key in ur:
_ok(label)
for key, label in [
("service_not_found", "vLLM service: already absent"),
("unit_not_found", "systemd unit: already absent"),
("env_not_found", "API key environment file: already absent"),
("nginx_not_found", "nginx config: already absent"),
("certs_not_found", "TLS certs: already absent"),
("venv_not_found", "Python venv: already absent"),
("user_not_found", "vLLM user: already absent"),
]:
if key in ur:
_info(label)
else:
# Deploy summary.
_ok(f"Model: {model}")
sd = results.get("system_deps", {})
installed = len(sd.get("installed", []))
skipped = len(sd.get("skipped", []))
failed_pkgs = len(sd.get("failed", []))
if dry_run:
_ok(f"System packages: {skipped} present, {installed} would be installed")
else:
_ok(f"System packages: {skipped} already present, {installed} installed")
if failed_pkgs:
_fail(f" {failed_pkgs} package(s) failed to install")
vu = results.get("vllm_user", {})
if vu.get("user_created") is True or vu.get("user_exists"):
_ok("vLLM system user: configured")
elif vu.get("user_created") == "dry run":
_info("vLLM system user: would create")
venv = results.get("venv", {})
if venv.get("vllm_installed") is True:
ver = venv.get("vllm_version", "")
label = f"vLLM installed ({ver})" if ver else "vLLM installed"
_ok(label)
elif venv.get("vllm_installed") == "dry run":
pf = results.get("preflight", {})
_info(
f"vLLM: would install vllm=={pf.get('vllm_version', '?')}"
f"+{pf.get('rocm_variant', '?')}"
)
elif venv.get("venv_exists"):
_warn("vLLM: venv exists but vLLM not installed")
tls = results.get("tls", {})
if tls.get("cert_exists") or tls.get("cert_created") is True:
_ok("TLS certificate: configured (self-signed, 365-day validity)")
elif tls.get("cert_created") == "dry run":
_info("TLS: would generate self-signed certificate (SAN from host FQDN)")
ak = results.get("api_key", {})
env_state = ak.get("env_file")
if generated_api_key:
_ok(f"API key: stored in /etc/sysconfig/{service_name} (0600 root:vllm)")
print(
f" {YELLOW}{BOLD}API key (shown once — store it securely): "
f"{generated_api_key}{RESET}",
file=sys.stderr,
)
elif env_state == "exists":
_ok("API key: existing key reused")
elif env_state == "updated":
_ok("API key: updated from --api-key")
elif env_state == "created":
_ok(f"API key: stored in /etc/sysconfig/{service_name} (0600 root:vllm)")
elif env_state == "dry run":
_info(f"API key: would generate/store in /etc/sysconfig/{service_name} (0600)")
elif env_state == "failed":
_fail("API key: could not write the environment file")
se = results.get("selinux", {}).get("selinux")
if se == "on":
_ok("SELinux: httpd_can_network_connect on")
elif se == "dry run":
_info("SELinux: would set httpd_can_network_connect=1")
elif se == "failed":
_fail("SELinux: failed to set httpd_can_network_connect")
elif se in ("permissive", "disabled"):
_info(f"SELinux: {se} — skipped")
elif se == "absent":
_info("SELinux: not installed — skipped")
ng = results.get("nginx", {})
if ng.get("nginx_configured") is True and ng.get("nginx_reloaded") is True:
_ok("nginx: configured and reloaded")
elif ng.get("nginx_configured") is True:
_fail("nginx: config written but reload failed")
elif ng.get("nginx_configured") == "dry run":
_info("nginx: would write config and reload")
svc = results.get("systemd", {})
started = svc.get("service_started")
disp = f"[{bind_host}]:{port}" if ":" in bind_host else f"{bind_host}:{port}"
if started is True or started == "restarted":
verb = "restarted with updated unit" if started == "restarted" else "running"
_ok(f"systemd: {service_name} {verb} on {disp}")
_info("First load takes minutes — verify: curl -fk https://localhost/health (retry)")
elif started is False:
_fail(f"systemd: {service_name} failed to start")
elif started == "dry run":
_info(f"systemd: would enable and start {service_name}")
fw = results.get("firewall", {})
if fw.get("firewall_active") is False:
_info("Firewall: firewalld not running — skipped")
elif fw.get("firewall_configured") is True:
_ok("Firewall: HTTPS (443) allowed")
elif fw.get("firewall_configured") is False:
_fail("Firewall: failed to allow HTTPS (443)")
elif fw.get("firewall_configured") == "dry run":
_info("Firewall: would allow HTTPS (443)")
if warnings:
print(file=sys.stderr)
for w in warnings:
_warn(w)
print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
print(f" {BOLD}Total time: {elapsed:.1f}s{RESET}", file=sys.stderr)
print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
print(file=sys.stderr)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""Parse command-line arguments (no interaction — see _prompt_model)."""
parser = argparse.ArgumentParser(
description="Idempotent vLLM deployment for AMD ROCm GPUs",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--model",
type=str,
required=False,
default=None,
help="HuggingFace model ID (if omitted, shows interactive menu)",
)
parser.add_argument(
"--port",
type=int,
default=8000,
help="vLLM internal port (default: 8000)",
)
parser.add_argument(
"--tensor-parallel",
type=int,
default=1,
dest="tensor_parallel",
help="Number of GPUs for tensor parallelism (default: 1)",
)
parser.add_argument(
"--max-model-len",
type=int,
default=4096,
dest="max_model_len",
help="Maximum context length (default: 4096)",
)
parser.add_argument(
"--gpu-memory-utilization",
type=float,
default=0.90,
dest="gpu_memory_utilization",
help="GPU memory fraction (default: 0.90)",
)
parser.add_argument(
"--venv-dir",
type=str,
default="/opt/vllm/venv",
dest="venv_dir",
help="Path to Python venv (default: /opt/vllm/venv)",
)
parser.add_argument(
"--service-name",
type=str,
default="vllm",
dest="service_name",
help="systemd service name (default: vllm)",
)
parser.add_argument(
"--cert-dir",
type=str,
default="/etc/ssl/vllm",
dest="cert_dir",
help="TLS certificate directory (default: /etc/ssl/vllm)",
)
parser.add_argument(
"--api-key",
type=str,
default=None,
dest="api_key",
help="API key for the endpoint (default: generate and store in /etc/sysconfig)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Preview without making changes",
)
parser.add_argument(
"--uninstall",
action="store_true",
help="Tear down service, venv, nginx config, and certificates",
)
parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {VERSION}",
)
return parser.parse_args(argv)
def _prompt_model(args: argparse.Namespace) -> None:
"""Show the interactive model selection menu; set args.model."""
print(f"\n{BOLD}Select a model to deploy:{RESET}\n", file=sys.stderr)
for i, name in enumerate(DEFAULT_MODELS, 1):
print(f" {GREEN}{i}{RESET}. {name}", file=sys.stderr)
print(f" {DIM}{len(DEFAULT_MODELS) + 1}. Enter custom model ID{RESET}", file=sys.stderr)
print(file=sys.stderr)
try:
choice = input(f" Choice [{GREEN}1{RESET}]: ").strip()
except (EOFError, KeyboardInterrupt):
print("\nCancelled.", file=sys.stderr)
sys.exit(130)
if not choice:
choice = "1"
try:
idx = int(choice)
except ValueError:
print(f"{RED}Invalid input.{RESET}", file=sys.stderr)
sys.exit(1)
if 1 <= idx <= len(DEFAULT_MODELS):
args.model = DEFAULT_MODELS[idx - 1]
elif idx == len(DEFAULT_MODELS) + 1:
try:
args.model = input(" Model ID: ").strip()
except (EOFError, KeyboardInterrupt):
print("\nCancelled.", file=sys.stderr)
sys.exit(130)
if not args.model:
print(f"{RED}Model ID cannot be empty.{RESET}", file=sys.stderr)
sys.exit(1)
else:
print(f"{RED}Invalid choice.{RESET}", file=sys.stderr)
sys.exit(1)
def _validate_args(args: argparse.Namespace) -> None:
"""Validate CLI arguments; exit with a clear message on invalid input."""
if not re.match(r"^[A-Za-z0-9_.@-]+$", args.service_name):
_fail(f"Invalid --service-name: {args.service_name!r}")
sys.exit(1)
if args.uninstall:
return
if not MODEL_ID_RE.match(args.model or ""):
_fail(
f"Invalid model ID: {args.model!r} — expected 'org/name' "
"(letters, digits, dot, dash, underscore only)"
)
sys.exit(1)
if not 1 <= args.port <= 65535 or args.port == 443:
_fail(f"Invalid --port {args.port} — must be 1-65535 and not 443 (nginx)")
sys.exit(1)
if not 0.0 < args.gpu_memory_utilization <= 1.0:
_fail(f"Invalid --gpu-memory-utilization {args.gpu_memory_utilization} — must be in (0, 1]")
sys.exit(1)
if args.max_model_len <= 0:
_fail(f"Invalid --max-model-len {args.max_model_len} — must be positive")
sys.exit(1)
if args.tensor_parallel <= 0:
_fail(f"Invalid --tensor-parallel {args.tensor_parallel} — must be positive")
sys.exit(1)
if args.api_key is not None and (not args.api_key or any(ch.isspace() for ch in args.api_key)):
_fail("Invalid --api-key — must be non-empty and contain no whitespace")
sys.exit(1)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
"""Entry point — parse args, run pre-flight checks, deploy or uninstall."""
start_time = time.monotonic()
warnings: list[str] = []
results: dict[str, Any] = {}
failures: list[str] = []
# parse_args first: --help/--version must work without root privileges.
args = parse_args()
check_root()
if not args.uninstall and not args.model:
_prompt_model(args)
_validate_args(args)
os_id, os_display, version_id = detect_os()
# Header.
print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
print(
f"{BOLD} vLLM Deploy v{VERSION} — {os_display} {version_id} (ROCm){RESET}",
file=sys.stderr,
)
print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
if args.dry_run:
print(
f"\n {YELLOW}{BOLD}DRY RUN — no changes will be made{RESET}",
file=sys.stderr,
)
if args.uninstall and not args.dry_run:
print(
f"\n {YELLOW}{BOLD}UNINSTALL MODE — all vLLM components will be removed{RESET}",
file=sys.stderr,
)
print(file=sys.stderr)
# ── Uninstall path ──
if args.uninstall:
print(f"\n{BOLD}── Uninstall ──{RESET}", file=sys.stderr)
results["uninstall"] = uninstall(
venv_dir=args.venv_dir,
service_name=args.service_name,
cert_dir=args.cert_dir,
dry_run=args.dry_run,
)
elapsed = time.monotonic() - start_time
print_summary(
model="",
port=args.port,
service_name=args.service_name,
os_display=os_display,
version_id=version_id,
results=results,
warnings=warnings,
elapsed=elapsed,
dry_run=args.dry_run,
uninstall_mode=True,
)
if results["uninstall"].get("user_removed") is False:
sys.exit(1)
return
# ── Deploy path ──
bind_host, upstream_host = _detect_loopback()
try:
# 1. System dependencies (before preflight: provides lspci and curl).
print(f"\n{BOLD}── System Dependencies ──{RESET}", file=sys.stderr)
results["system_deps"] = install_system_deps(dry_run=args.dry_run)
failed = results["system_deps"].get("failed", [])
if failed:
failures.append(f"failed to install packages: {', '.join(failed)}")
# 2. Pre-flight checks.
print(f"\n{BOLD}── Pre-flight Checks ──{RESET}", file=sys.stderr)
preflight = preflight_checks(os_id=os_id, version_id=version_id, dry_run=args.dry_run)
results["preflight"] = preflight
gpu_count = preflight.get("gpu_count", 0)
_info(f"GPU count: {gpu_count}")
if gpu_count and gpu_count < args.tensor_parallel:
warnings.append(
f"--tensor-parallel={args.tensor_parallel} but only {gpu_count} GPU(s) detected"
)
# 3. vLLM system user.
print(f"\n{BOLD}── System User ──{RESET}", file=sys.stderr)
results["vllm_user"] = setup_vllm_user(dry_run=args.dry_run)
if results["vllm_user"].get("user_created") is False:
_fail("Could not create the vllm system user — cannot continue")
sys.exit(1)
# 4. Python venv + vLLM (fatal on failure — nothing works without it).
print(f"\n{BOLD}── Python Environment ──{RESET}", file=sys.stderr)
results["venv"] = setup_venv(
vllm_version=preflight["vllm_version"],
rocm_variant=preflight["rocm_variant"],
index_url=preflight["wheel_index"],
venv_dir=args.venv_dir,
dry_run=args.dry_run,
)
if (
results["venv"].get("venv_created") is False
or results["venv"].get("vllm_installed") is False
):
_fail("Python environment setup failed — cannot continue")
sys.exit(1)
# 5. TLS certificate (fatal — nginx cannot start without it).
print(f"\n{BOLD}── TLS Certificate ──{RESET}", file=sys.stderr)
results["tls"] = setup_tls(cert_dir=args.cert_dir, dry_run=args.dry_run)
if results["tls"].get("cert_created") is False:
_fail("TLS certificate setup failed — cannot continue")
sys.exit(1)
# 6. API key environment file.
print(f"\n{BOLD}── API Key ──{RESET}", file=sys.stderr)
results["api_key"] = setup_api_key(
service_name=args.service_name,
api_key=args.api_key,
dry_run=args.dry_run,
)
if results["api_key"].get("env_file") == "failed":
_fail("Could not store the API key — cannot continue")
sys.exit(1)
# 7. SELinux (non-fatal — only relevant on enforcing systems).
print(f"\n{BOLD}── SELinux ──{RESET}", file=sys.stderr)
results["selinux"] = setup_selinux(dry_run=args.dry_run)
if results["selinux"].get("selinux") == "failed":
failures.append("SELinux boolean httpd_can_network_connect not set")
# 8. nginx.
print(f"\n{BOLD}── nginx ──{RESET}", file=sys.stderr)
results["nginx"] = setup_nginx(
port=args.port,
upstream_host=upstream_host,
cert_dir=args.cert_dir,
dry_run=args.dry_run,
)
if results["nginx"].get("nginx_reloaded") is False:
failures.append("nginx configuration reload failed")
# 9. systemd service (model existence probed before writing the unit).
print(f"\n{BOLD}── systemd Service ──{RESET}", file=sys.stderr)
_status(f"Verifying {args.model} on Hugging Face")
hf_status = _hf_model_status(args.model)
if hf_status == "missing":
_status_done("not found")
_fail(f"Model '{args.model}' does not exist on Hugging Face")
sys.exit(1)
if hf_status == "gated":
_status_done("gated")
warnings.append(
f"Model '{args.model}' is gated on Hugging Face — "
"set HF_TOKEN for the vllm user or the first start will fail"
)
elif hf_status == "unknown":
_status_done("could not verify (offline?)")
else:
_status_done("found")
results["systemd"] = setup_systemd(
model=args.model,
port=args.port,
host=bind_host,
tensor_parallel=args.tensor_parallel,
max_model_len=args.max_model_len,
gpu_memory_utilization=args.gpu_memory_utilization,
venv_dir=args.venv_dir,
service_name=args.service_name,
dry_run=args.dry_run,
)
if results["systemd"].get("service_started") is False:
failures.append(f"{args.service_name} service failed to start")
# 10. Firewall.
print(f"\n{BOLD}── Firewall ──{RESET}", file=sys.stderr)
results["firewall"] = setup_firewall(dry_run=args.dry_run)
if results["firewall"].get("firewall_configured") is False:
failures.append("firewall rule for HTTPS (443) not applied")
except FileNotFoundError as exc:
print(file=sys.stderr)
_fail(f"Required tool not found: {exc.filename} — install it and re-run")
sys.exit(1)
elapsed = time.monotonic() - start_time
print_summary(
model=args.model,
port=args.port,
service_name=args.service_name,
os_display=os_display,
version_id=version_id,
results=results,
warnings=warnings,
elapsed=elapsed,
dry_run=args.dry_run,
uninstall_mode=False,
bind_host=bind_host,
generated_api_key=results.get("api_key", {}).get("generated_key"),
)
if failures:
_fail(f"Completed with {len(failures)} failed step(s):")
for failure in failures:
_info(f" - {failure}")
sys.exit(1)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink