diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f8c1bc..e6a2a6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -150,6 +150,37 @@ The account page (was "Your keys") shows the last thirty days of a person's own use and, for each place they spend, a bar for every budget that applies — their own, or a project's shared with others, said as such. +### `llmgw node add`: a node in one command + +Adding a node was a bash script from the repository, not installed anywhere, +that needed the node's name, the control plane's address and a dozen flags. +It is now part of the binary: + +``` +sudo llmgw node add ubuntu@10.0.0.21 +``` + +run on the control plane. The node id is the machine's hostname and the +address its agent dials is this machine's on the route to it, checked against +the server certificate. It installs llmgw on the node — the release for its +architecture, fetched and checked there, which works on RHEL too — runs +`llmgw doctor` there and stops on a problem, issues and installs the +certificate, writes `agent.yaml` and the unit, starts the agent, and waits for +the node to join and install llama.cpp. + +In a terminal it is a checklist that fills in, and asks what a script cannot: +which fabric each group of active adapters is on — grouped by the subnet +manager or RoCE network they report — and whether to go on past a problem. +Elsewhere it prints one line per step. Running it again on a node is safe. + +`llmgw node remove` forgets a node that is gone for good, which otherwise +stays listed as lost until the control plane restarts; only a lost node, or an +unreachable one with nothing placed on it, is removed. `llmgw ctl nodes` no +longer lists vLLM as skipped on every CPU node. + +Tried on an empty VM, from the terminal screen and plain: joined with +llama.cpp ready in under a minute, and 17 seconds to add it again. + ### vLLM installs itself on GPU nodes The agent installs vLLM as it does llama.cpp: on every node with an NVIDIA GPU diff --git a/docs/install.md b/docs/install.md index 6f661be..c6d7bc9 100644 --- a/docs/install.md +++ b/docs/install.md @@ -488,40 +488,44 @@ bf16 is 141 GB and the same model at q4 is 43 GB. ## 5. Adding the node -Two halves, because they need different access: building the machine needs the -hypervisor, and issuing its identity needs the certificate authority. Where one -person holds both — the common case — one command does it: +On the control plane, which holds the certificate authority: ``` -llmgw-add-node \ - --name llmgw-node01 --ip 10.0.0.141/24 \ - --server https://10.0.0.140:7443 \ - --cores 16 --memory 131072 --disk 200 \ - --fabric ib-a:mlx5_0,mlx5_1 \ - --llamacpp-bin /opt/llama.cpp/llama-server +sudo llmgw node add ubuntu@10.0.0.141 --fabric ib-a:mlx5_0,mlx5_1 ``` -**The VM name, the hostname, the node id and the certificate's common name are -all the same string.** They do not have to be, and every time they are not, -somebody spends an afternoon working out which of the four a message refers to. +It logs in to the machine over SSH as that user, who needs passwordless sudo +there, and: -For a machine that already exists — bare metal, or a VM you built yourself — skip -provisioning and issue the identity only: +1. installs llmgw: the release for the machine's architecture, downloaded by + the machine and checked against the release's checksums, or this machine's + copy when it cannot download; +2. runs `llmgw doctor` there and stops on any problem, printing the fix. Run it + again once fixed, or add `--ignore-checks`; +3. issues the node's certificate and installs it, with the cluster's root, + trusted system-wide; +4. writes `agent.yaml` — only what the node cannot find for itself — and the + systemd unit, and starts the agent; +5. waits for the node to join and to install llama.cpp (`--wait` also waits + for vLLM, `--no-wait` for neither). -``` -llmgw-onboard-node \ - --node llmgw-node01 --addr 10.0.0.141 \ - --server https://10.0.0.140:7443 \ - --fabric ib-a:mlx5_0,mlx5_1 \ - --llamacpp-bin /opt/llama.cpp/llama-server -``` +In a terminal it draws this as a checklist and asks what a script cannot: which +fabric each group of the node's active adapters is on, when `--fabric` was not +given, and whether to go on past a problem doctor found. -Onboarding needs SSH from the CA host to the node. It waits for SSH rather than -failing on it, and learns the host key of a machine it has not seen — a new -machine spends its first minute regenerating host keys and restarting sshd. +**The node id is the machine's hostname**, and the address its agent dials is +this machine's on the route to it; `--node` and `--server` override them. The +VM name, the hostname, the node id and the certificate's common name being the +same string saves an afternoon each time they would differ. -It also checks that a configured engine actually runs, and says so loudly when -it does not. That check exists because the failure is otherwise invisible. +**SSH runs as root under sudo**, with root's keys. To use yours, log in to the +control plane with `ssh -A` and run `sudo --preserve-env=SSH_AUTH_SOCK llmgw node +add …`. It waits for SSH rather than failing on it, and learns the host key of +a machine it has not seen — a new machine spends its first minute regenerating +host keys and restarting sshd. + +Running it again on a node is safe: a fresh certificate and configuration, and +the engines it runs are adopted by the restarted agent. Within a few seconds: diff --git a/docs/node-onboarding.md b/docs/node-onboarding.md index f8f3444..385aa62 100644 --- a/docs/node-onboarding.md +++ b/docs/node-onboarding.md @@ -4,24 +4,16 @@ Written for whoever runs the cluster. ## The short version -Where one person holds both the hypervisor and the certificate authority — -which is the common case — one command does the whole thing in about -45 seconds: +On the control plane: ``` -llmgw-add-node \ - --name llmgw-node05 --ip 10.0.0.145/24 \ - --server https://10.0.0.140:7443 \ - --cores 8 --memory 32768 --disk 60 \ - --llamacpp-bin /opt/llama.cpp/llama-server +sudo llmgw node add ubuntu@10.0.0.145 ``` -The VMID defaults to the last octet of the address, so `qm list` and -`llmgw ctl nodes` stay in the same order. `--provision-only` and -`--onboard-only` split it back into the two halves below when the two kinds -of access belong to different people. - -The rest of this page is what that command does, and what it still does not. +That is all a machine that already exists needs — bare metal, or a VM you +built. [install.md](install.md#5-adding-the-node) says what it does. On +Proxmox, the machine itself can be built first by the script below, on the +hypervisor. ## The two halves @@ -54,41 +46,31 @@ their working tree. ## 2. The identity -On the host that holds the certificate authority — this is the part that cannot -be delegated, because it issues the credential by which the control plane -believes the node is who it says it is: +On the control plane, which holds the certificate authority — this is the +part that cannot be delegated, because it issues the credential by which the +control plane believes the node is who it says it is: ``` -llmgw-onboard-node \ - --node llmgw-node04 \ - --addr 10.0.0.144 \ - --server https://10.0.0.140:7443 \ - --fabric ib-b +sudo llmgw node add ilker@10.0.0.144 --fabric ib-b:mlx5_0,mlx5_1 ``` -It issues a client certificate whose common name is the node id, installs it -along with the CA root, trusts that root system-wide so other tooling on the -machine stops treating llmgw's certificates as strangers, installs the binary -and the unit, and starts the agent. The node appears in `llmgw ctl nodes` within -a few seconds. +It checks the machine with `llmgw doctor`, issues a client certificate whose +common name is the node id (the hostname), installs it along with the CA root, +trusts that root system-wide, installs the binary and the unit, starts the +agent, and waits for the node to join. -If it does not, the control plane's log says why. A certificate whose common -name does not match the claimed node id is refused by design — that check is -what makes a node's identity something it proves rather than something it picks. +If it does not join, the agent's log on the node says why. A certificate whose +common name does not match the claimed node id is refused by design — that +check is what makes a node's identity something it proves rather than +something it picks. -**It needs ssh from the CA host to the node.** Provisioning copies that host's -`authorized_keys` into the new machine, so a node built by the script above -already satisfies it; an older machine may need the key added. +**It needs SSH from the control plane to the node.** Provisioning copies the +hypervisor's `authorized_keys` into the new machine; with `ssh -A` to the +control plane and `sudo --preserve-env=SSH_AUTH_SOCK`, the same key that +reaches the hypervisor reaches the node. -It waits for ssh rather than failing on it, and learns the host key of a machine -it has not seen (`accept-new`, which still refuses a *changed* key — the risk -being guarded against is a substituted machine, not a new one). Both are needed -to run the halves back to back: for the first minute of its life a new machine -is still regenerating host keys and restarting sshd, so it refuses connections -it accepted a moment earlier. - -It also checks that a configured engine actually runs, which is cheap here and -expensive later — see below. +`deploy/bootstrap/onboard-node.sh` is the script this replaced, for a machine +holding an authority but not a current llmgw. ## The CPU model matters, and its default is wrong diff --git a/docs/quick-install.md b/docs/quick-install.md index ca1183b..267cf50 100644 --- a/docs/quick-install.md +++ b/docs/quick-install.md @@ -11,7 +11,7 @@ explains every choice. | **One control plane machine** | Linux, x86-64 or ARM64. Debian 12, Ubuntu 22.04+ or RHEL 10. No GPU needed; 2 GB RAM, 20 GB disk. | | **GPU nodes** (optional to start) | The same Linux, with the NVIDIA driver installed. vLLM needs driver 580 or newer and a Turing GPU or newer (not V100). | | **root** | on the control plane, and `sudo` without a password on every node. | -| **SSH** | from root on the control plane to each node. | +| **SSH** | from the control plane to each node, as a user with passwordless `sudo` there. | | **Network** | nodes reach the control plane on 7443 and 8081; users reach it on 8080. The machines need internet access to download engines and models. | You do not install llama.cpp, vLLM, Python or CUDA. The nodes do that @@ -60,59 +60,49 @@ the account page. ## 3. Add the GPU nodes -**On each node**, check it first. Set `VERSION` and `U` as in step 1, then: +On the control plane, for each node: ``` -curl -fsSLO $U/llmgw-$VERSION-linux-amd64 -sudo install -m0755 llmgw-$VERSION-linux-amd64 /usr/local/bin/llmgw -sudo llmgw doctor +sudo llmgw node add ubuntu@10.0.0.21 ``` -Fix every `FAIL` line before going on. The fix is printed under each line. +Replace `ubuntu` with the user you log in to the node as, and `10.0.0.21` with +its address. That user needs `sudo` without a password on the node. -**On the control plane**, get the onboarding script once: +It logs in over SSH and, while you watch: + +1. installs llmgw on the node, the release for its processor, checksum checked; +2. runs `llmgw doctor` there, and stops if the node has a problem. Each problem + comes with the command that fixes it. Fix it and run the same command again; +3. gives the node its certificate and configuration, and starts its agent; +4. waits for the node to join and install llama.cpp. + +The node's name in the cluster is its hostname. vLLM installs by itself after +that, in a few more minutes on a GPU node (about 8 GB). The cluster page +(`/admin`) shows its progress, or why the node cannot run it. + +**InfiniBand.** When the node's adapters are up, `node add` shows them grouped +by the fabric they are on and asks for each fabric's name. Give every node on +the same fabric the same name. In a script, give it instead: ``` -curl -fsSLO https://gitea.mynodes.xyz/ilker/llmgw/raw/tag/$VERSION/deploy/bootstrap/onboard-node.sh -chmod +x onboard-node.sh +sudo llmgw node add ubuntu@10.0.0.21 --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3 ``` -Check that root can reach the node over SSH. This must print `ok` without asking -for a password: +**If it cannot log in.** `sudo` runs SSH as root, with root's keys. To use your +own keys instead, log in to the control plane with `ssh -A` and keep your +agent through sudo: ``` -sudo ssh ubuntu@10.0.0.21 'sudo true && echo ok' +sudo --preserve-env=SSH_AUTH_SOCK llmgw node add ubuntu@10.0.0.21 ``` -Then add the node. `--node` is its name in the cluster; use its hostname. +**Check:** `llmgw ctl nodes` lists the node. Running `node add` again on the +same node is safe: it renews its certificate and configuration, and keeps what +runs there. -``` -sudo ./onboard-node.sh --node gpu01 --addr 10.0.0.21 \ - --server https://10.0.0.10:7443 --ssh-user ubuntu -``` - -Replace `10.0.0.21` with the node's address, `10.0.0.10` with the control -plane's, and `ubuntu` with the user you SSH in as. - -If the node has InfiniBand, name the fabric and its adapters, and give every -node on the same fabric the same name: - -``` - --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3 -``` - -**Check:** - -``` -llmgw ctl nodes -``` - -The node appears within seconds. It then installs its engines by itself: -llama.cpp in about a minute, vLLM in a few more (about 8 GB). The cluster page -(`/admin`) shows the progress under each node. If a node cannot run vLLM, it -says why on the same page. - -Repeat for each node. +A machine retired for good is removed with `llmgw node remove ` once its +agent is stopped. ## 4. Add a model @@ -174,7 +164,8 @@ through the proxy. There is an nginx configuration too: | What you see | What to do | |---|---| -| A node does not appear in `llmgw ctl nodes` | `sudo journalctl -u llmgw-agent -n 50` on the node. Usually it cannot reach port 7443, or the control plane's address is not in its certificate: run `server init` again with `--host`. | +| `node add` stops at `join` | `sudo journalctl -u llmgw-agent -n 50` on the node. Usually it cannot reach port 7443 on the control plane. | +| `node add` says the address is not in the certificate | Run `sudo llmgw server init --host `, then `node add` again. | | A node says it cannot run vLLM | The reason is on `/admin`: the driver is older than 580, the GPU is older than Turing, or there is less than 25 GB free. | | A model shows `not placed` | `llmgw ctl explain-placement ` names each node and why it was not used. | | A reply stops after 60 seconds behind nginx | Use the nginx configuration from the release. Its timeouts are set for slow models. | diff --git a/go.mod b/go.mod index efb422b..66eaace 100644 --- a/go.mod +++ b/go.mod @@ -6,11 +6,15 @@ require ( connectrpc.com/connect v1.21.0 github.com/NVIDIA/go-nvml v0.13.4-0 github.com/cespare/xxhash/v2 v2.3.0 + github.com/charmbracelet/bubbles v1.0.0 + github.com/charmbracelet/bubbletea v1.3.10 + github.com/charmbracelet/lipgloss v1.1.0 github.com/jackc/pgx/v5 v5.11.0 github.com/knadh/koanf/parsers/yaml v1.1.1 github.com/knadh/koanf/providers/env/v2 v2.0.1 github.com/knadh/koanf/providers/file v1.2.1 github.com/knadh/koanf/v2 v2.3.6 + github.com/mattn/go-isatty v0.0.24 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_model v0.6.3 github.com/prometheus/common v0.71.0 @@ -24,8 +28,18 @@ require ( ) require ( + github.com/atotto/clipboard v0.1.4 // indirect + github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect github.com/beorn7/perks v1.0.1 // indirect + github.com/charmbracelet/colorprofile v0.4.1 // indirect + github.com/charmbracelet/x/ansi v0.11.6 // indirect + github.com/charmbracelet/x/cellbuf v0.0.15 // indirect + github.com/charmbracelet/x/term v0.2.2 // indirect + github.com/clipperhouse/displaywidth v0.9.0 // indirect + github.com/clipperhouse/stringish v0.1.1 // indirect + github.com/clipperhouse/uax29/v2 v2.5.0 // indirect github.com/dustin/go-humanize v1.0.1 // indirect + github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/google/uuid v1.6.0 // indirect @@ -34,14 +48,21 @@ require ( github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect - github.com/mattn/go-isatty v0.0.24 // indirect + github.com/lucasb-eyer/go-colorful v1.3.0 // indirect + github.com/mattn/go-localereader v0.0.1 // indirect + github.com/mattn/go-runewidth v0.0.19 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect + github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect + github.com/muesli/cancelreader v0.2.2 // indirect + github.com/muesli/termenv v0.16.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/rivo/uniseg v0.4.7 // indirect github.com/spf13/pflag v1.0.9 // indirect + github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/sys v0.48.0 // indirect golang.org/x/text v0.42.0 // indirect diff --git a/go.sum b/go.sum index 01cd524..7f0cc48 100644 --- a/go.sum +++ b/go.sum @@ -2,14 +2,40 @@ connectrpc.com/connect v1.21.0 h1:LhqSJt7jHf5NJBo9Jq/t/9FjcYAideif0mg+qe2jCUs= connectrpc.com/connect v1.21.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4= github.com/NVIDIA/go-nvml v0.13.4-0 h1:o3jp9u2x1R9ShFE3v+Aesp55XOSIQFMJz/VGNUcJaNE= github.com/NVIDIA/go-nvml v0.13.4-0/go.mod h1:id63qwpoDWpFXwnwM6psDCSqW4BmNu6mWpr4YeQtPGo= +github.com/atotto/clipboard v0.1.4 h1:EH0zSVneZPSuFR11BlR9YppQTVDbh5+16AmcJi4g1z4= +github.com/atotto/clipboard v0.1.4/go.mod h1:ZY9tmq7sm5xIbd9bOK4onWV4S6X0u6GY7Vn0Yu86PYI= +github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= +github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/charmbracelet/bubbles v1.0.0 h1:12J8/ak/uCZEMQ6KU7pcfwceyjLlWsDLAxB5fXonfvc= +github.com/charmbracelet/bubbles v1.0.0/go.mod h1:9d/Zd5GdnauMI5ivUIVisuEm3ave1XwXtD1ckyV6r3E= +github.com/charmbracelet/bubbletea v1.3.10 h1:otUDHWMMzQSB0Pkc87rm691KZ3SWa4KUlvF9nRvCICw= +github.com/charmbracelet/bubbletea v1.3.10/go.mod h1:ORQfo0fk8U+po9VaNvnV95UPWA1BitP1E0N6xJPlHr4= +github.com/charmbracelet/colorprofile v0.4.1 h1:a1lO03qTrSIRaK8c3JRxJDZOvhvIeSco3ej+ngLk1kk= +github.com/charmbracelet/colorprofile v0.4.1/go.mod h1:U1d9Dljmdf9DLegaJ0nGZNJvoXAhayhmidOdcBwAvKk= +github.com/charmbracelet/lipgloss v1.1.0 h1:vYXsiLHVkK7fp74RkV7b2kq9+zDLoEU4MZoFqR/noCY= +github.com/charmbracelet/lipgloss v1.1.0/go.mod h1:/6Q8FR2o+kj8rz4Dq0zQc3vYf7X+B0binUUBwA0aL30= +github.com/charmbracelet/x/ansi v0.11.6 h1:GhV21SiDz/45W9AnV2R61xZMRri5NlLnl6CVF7ihZW8= +github.com/charmbracelet/x/ansi v0.11.6/go.mod h1:2JNYLgQUsyqaiLovhU2Rv/pb8r6ydXKS3NIttu3VGZQ= +github.com/charmbracelet/x/cellbuf v0.0.15 h1:ur3pZy0o6z/R7EylET877CBxaiE1Sp1GMxoFPAIztPI= +github.com/charmbracelet/x/cellbuf v0.0.15/go.mod h1:J1YVbR7MUuEGIFPCaaZ96KDl5NoS0DAWkskup+mOY+Q= +github.com/charmbracelet/x/term v0.2.2 h1:xVRT/S2ZcKdhhOuSP4t5cLi5o+JxklsoEObBSgfgZRk= +github.com/charmbracelet/x/term v0.2.2/go.mod h1:kF8CY5RddLWrsgVwpw4kAa6TESp6EB5y3uxGLeCqzAI= +github.com/clipperhouse/displaywidth v0.9.0 h1:Qb4KOhYwRiN3viMv1v/3cTBlz3AcAZX3+y9OLhMtAtA= +github.com/clipperhouse/displaywidth v0.9.0/go.mod h1:aCAAqTlh4GIVkhQnJpbL0T/WfcrJXHcj8C0yjYcjOZA= +github.com/clipperhouse/stringish v0.1.1 h1:+NSqMOr3GR6k1FdRhhnXrLfztGzuG+VuFDfatpWHKCs= +github.com/clipperhouse/stringish v0.1.1/go.mod h1:v/WhFtE1q0ovMta2+m+UbpZ+2/HEXNWYXQgCt4hdOzA= +github.com/clipperhouse/uax29/v2 v2.5.0 h1:x7T0T4eTHDONxFJsL94uKNKPHrclyFI0lm7+w94cO8U= +github.com/clipperhouse/uax29/v2 v2.5.0/go.mod h1:Wn1g7MK6OoeDT0vL+Q0SQLDz/KpfsVRgg6W7ihQeh4g= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f h1:Y/CXytFA4m6baUTXGLOoWe4PQhGxaX0KpnayAqC48p4= +github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f/go.mod h1:vw97MGsxSvLiUE2X8qFplwetxpGLQrlU1Q9AUEIzCaM= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= @@ -46,12 +72,24 @@ github.com/knadh/koanf/v2 v2.3.6 h1:JoQPSJmvS4aP0xNc8xMDr5tcrkSEInL23/Il7pITAKo= github.com/knadh/koanf/v2 v2.3.6/go.mod h1:gRb40VRAbd4iJMYYD5IxZ6hfuopFcXBpc9bbQpZwo28= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag= +github.com/lucasb-eyer/go-colorful v1.3.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0= github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/mattn/go-localereader v0.0.1 h1:ygSAOl7ZXTx4RdPYinUpg6W99U8jWvWi9Ye2JC/oIi4= +github.com/mattn/go-localereader v0.0.1/go.mod h1:8fBrzywKY7BI3czFoHkuzRoWE9C+EiG4R1k4Cjx5p88= +github.com/mattn/go-runewidth v0.0.19 h1:v++JhqYnZuu5jSKrk9RbgF5v4CGUjqRfBm05byFGLdw= +github.com/mattn/go-runewidth v0.0.19/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw= github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s= github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ= github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 h1:ZK8zHtRHOkbHy6Mmr5D264iyp3TiX5OmNcI5cIARiQI= +github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6/go.mod h1:CJlz5H+gyd6CUWT45Oy4q24RdLyn7Md9Vj2/ldJBSIo= +github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA= +github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo= +github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc= +github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= @@ -67,6 +105,8 @@ github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+ github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= +github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= @@ -77,6 +117,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no= +github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM= github.com/yuin/goldmark v1.8.6 h1:d0VcaP1sx9GkFVkoW+KtggpGi2KZ965i14b0+bDQST4= github.com/yuin/goldmark v1.8.6/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -88,12 +130,15 @@ go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 1c56eda..4f14fd3 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -199,7 +199,8 @@ func newNodesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { said = true } for _, p := range n.Provision { - if p.State == "ready" { + // Skipped is vLLM on a node with no GPU: as it should be. + if p.State == "ready" || p.State == "skipped" { continue } fmt.Printf("\n%s: %s %s %s: %s", n.ID, p.Engine, p.Version, p.State, p.Detail) diff --git a/internal/cli/node.go b/internal/cli/node.go new file mode 100644 index 0000000..36be281 --- /dev/null +++ b/internal/cli/node.go @@ -0,0 +1,349 @@ +package cli + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "strings" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/config" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" + "gitea.mynodes.xyz/ilker/llmgw/internal/version" +) + +func newNodeCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "node", + Short: "Add machines to the cluster", + } + cmd.AddCommand(newNodeAddCmd(), newNodeRemoveCmd()) + return cmd +} + +func newNodeRemoveCmd() *cobra.Command { + var admin, serverCfg string + cmd := &cobra.Command{ + Use: "remove ", + Short: "Forget a node that is gone for good", + Long: `For a machine retired, or rebuilt under another name, which would otherwise +stay listed as lost until the control plane restarts. + +Only a node that is not there is removed: stop its agent first, and wait for +it to be lost, a minute later. Its certificate stays valid, since the +authority keeps no revocation list: a machine that still has it can join +again, which is how a node rebuilt under the same name comes back.`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + na := nodeAdd{admin: admin, serverCfg: serverCfg} + var out map[string]string + if err := na.adminAPI().delete(cmd.Context(), "/api/v1/nodes/"+args[0], &out); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "%s removed\n", args[0]) + return nil + }, + } + cmd.Flags().StringVar(&admin, "admin", "", "the control plane's admin API (default: from server.yaml, else 127.0.0.1:9090)") + cmd.Flags().StringVar(&serverCfg, "config", "/etc/llmgw/server.yaml", "the control plane's configuration") + return cmd +} + +// nodeAdd is what `llmgw node add` was asked to do. +type nodeAdd struct { + opts onboard.Options + fabrics []string + servers []string + admin string + waitAll bool + noWait bool + plain bool + serverCfg string +} + +func newNodeAddCmd() *cobra.Command { + var na nodeAdd + cmd := &cobra.Command{ + Use: "add [user@]address", + Short: "Make a machine a node: check it, give it an identity, start its agent", + Long: `Run on the machine holding the certificate authority — the control plane, +after 'llmgw server init' — as root. It logs in to the machine with ssh, +as you would by hand, and: + + 1. installs llmgw there: the release for its architecture, checksum verified + 2. runs 'llmgw doctor' there, and stops if it finds a problem + 3. issues the node's certificate and installs it + 4. writes agent.yaml and the systemd unit, and starts the agent + 5. waits for the node to join and install llama.cpp + +The node's id is its hostname, and the address its agent dials is this +machine's on the route to it, unless --node or --server say otherwise. +Nothing else needs giving, except --fabric for InfiniBand or RoCE: which +fabric an adapter is on has a name only you can choose, and it must be the +same on every node attached to it. Doctor suggests one when it sees adapters. + +The user needs sudo on the machine without a password. Run it again at any +time: the node gets a fresh certificate and configuration, and its running +engines are kept.`, + Example: ` sudo llmgw node add 10.0.0.21 + sudo llmgw node add ubuntu@gpu01 --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + return na.run(cmd.Context(), args[0], cmd.OutOrStdout()) + }, + } + f := cmd.Flags() + f.StringVar(&na.opts.Node, "node", "", "node id (default: the machine's hostname)") + f.StringVar(&na.opts.User, "ssh-user", "", "user to log in as (default: the one before @ in the address, else whoever ran sudo, else you)") + f.StringArrayVar(&na.servers, "server", nil, "control plane address the agent dials, https://host:7443; repeatable for failover (default: this machine)") + f.StringArrayVar(&na.fabrics, "fabric", nil, "InfiniBand or RoCE fabric and its adapters, name:mlx5_0,mlx5_1; repeatable") + f.StringVar(&na.opts.MgmtAddr, "mgmt-addr", "", "address engines are reached at (default: found by the agent)") + f.StringVar(&na.opts.MgmtIface, "mgmt-if", "", "interface NCCL bootstraps over (default: the one holding that address)") + f.StringVar(&na.opts.LlamaCppBin, "llamacpp-bin", "", "a llama-server you installed, instead of the one the agent installs") + f.StringVar(&na.opts.VLLMBin, "vllm-bin", "", "a vllm you installed; also set engines.vllm: \"\" in server.yaml") + f.StringVar(&na.opts.SGLangPython, "sglang-python", "", "python of an SGLang virtualenv you installed") + f.StringVar(&na.opts.NvidiaSMI, "nvidia-smi", "", "where nvidia-smi is, when it is not on PATH") + f.StringVar(&na.opts.IBRoot, "ib-root", "", "the sysfs InfiniBand directory, when not /sys/class/infiniband") + f.StringVar(&na.opts.PKI, "pki", defaultPKIDir, "directory holding the certificate authority") + f.StringVar(&na.opts.Releases, "releases", "https://gitea.mynodes.xyz/ilker/llmgw/releases/download", "where release files are downloaded from") + f.BoolVar(&na.opts.IgnoreChecks, "ignore-checks", false, "add the node even when doctor finds a problem") + f.StringVar(&na.admin, "admin", "", "the control plane's admin API, to watch the node join (default: from server.yaml, else 127.0.0.1:9090)") + f.StringVar(&na.serverCfg, "config", "/etc/llmgw/server.yaml", "the control plane's configuration, for the admin API's address") + f.BoolVar(&na.waitAll, "wait", false, "wait for every engine, vLLM included, not only llama.cpp") + f.BoolVar(&na.noWait, "no-wait", false, "do not wait for the node to join") + f.BoolVar(&na.plain, "plain", false, "plain output, even in a terminal") + return cmd +} + +func (na *nodeAdd) prepare(addr string) error { + o := &na.opts + o.Addr = addr + if user, host, ok := strings.Cut(addr, "@"); ok { + o.User, o.Addr = user, host + } + if o.User == "" { + // Under sudo, the person who ran it — their account is usually the + // one that exists on the nodes too. root's own keys still do the + // logging in, since that is who is running ssh. + o.User = os.Getenv("SUDO_USER") + } + for _, s := range na.fabrics { + fab, err := onboard.ParseFabric(s) + if err != nil { + return err + } + if len(fab.HCAs) == 0 { + return fmt.Errorf("--fabric %s names no adapters: give them as %s:mlx5_0,mlx5_1", s, fab.Name) + } + o.Fabrics = append(o.Fabrics, fab) + } + for _, s := range na.servers { + for _, one := range strings.Split(s, ",") { + if one = strings.TrimSpace(one); one != "" { + o.Servers = append(o.Servers, one) + } + } + } + o.Version = version.Version + if o.Binary == "" { + exe, err := os.Executable() + if err != nil { + return err + } + o.Binary = exe + } + onboard.AgentUnit = agentUnit + return nil +} + +func (na *nodeAdd) run(ctx context.Context, addr string, out io.Writer) error { + if err := na.prepare(addr); err != nil { + return err + } + if !na.plain && isTerminal(out) { + return na.runTUI(ctx) + } + p := &plainProgress{out: out} + fmt.Fprintf(out, "adding %s\n", na.opts.Addr) + r := onboard.New(na.opts, p.event) + res, err := r.Run(ctx) + if err != nil { + return err + } + if na.noWait { + fmt.Fprintf(out, "\n%s is added. 'llmgw ctl nodes' shows it once it joins.\n", res.Node) + return nil + } + if err := na.watch(ctx, res.Node, p.event); err != nil { + return err + } + fmt.Fprintf(out, "\n%s is in the cluster.\n", res.Node) + return nil +} + +// adminAPI is the control plane's admin API, from its configuration when +// this machine has it. +func (na *nodeAdd) adminAPI() *ctlClient { + addr := na.admin + if addr == "" { + addr = "127.0.0.1:9090" + if cfg, err := config.Load(na.serverCfg); err == nil && cfg.AdminListen != "" { + addr = cfg.AdminListen + if strings.HasPrefix(addr, ":") { + addr = "127.0.0.1" + addr + } + } + } + return newCtlClient(addr) +} + +// errNoAdmin is the watch giving up because the admin API is not here, which +// is not a failure of onboarding. +var errNoAdmin = errors.New("no admin API") + +// watch follows the node from the control plane: joining, then installing +// its engines. llama.cpp is waited for, since without it the node serves +// nothing; vLLM takes minutes more and is waited for only with --wait. +func (na *nodeAdd) watch(ctx context.Context, node string, emit func(onboard.Event)) error { + client := na.adminAPI() + var nodes []adminapi.NodeView + if err := client.get(ctx, "/api/v1/nodes", &nodes); err != nil { + emit(onboard.Event{Step: onboard.StepJoin, Level: onboard.Warn, + Detail: "cannot reach the control plane's admin API here, so the node's joining is not followed", + Fix: "llmgw ctl nodes on the control plane shows it"}) + return nil + } + emit(onboard.Event{Step: onboard.StepJoin, Level: onboard.Running, Detail: "waiting for " + node + " to join"}) + var view adminapi.NodeView + started := time.Now() + err := waitFor(ctx, 2*time.Minute, node+" to join", func() (bool, error) { + if err := client.get(ctx, "/api/v1/nodes", &nodes); err != nil { + return false, nil + } + for _, n := range nodes { + if n.ID == node && heardSince(n, started) { + view = n + return true, nil + } + } + return false, nil + }) + if err != nil { + return fmt.Errorf("%s did not join: its agent's log says why (journalctl -u llmgw-agent on the node); a certificate whose name differs from the node id is refused", node) + } + emit(onboard.Event{Step: onboard.StepJoin, Level: onboard.Done, Detail: node + " joined: " + describeNode(view)}) + + emit(onboard.Event{Step: onboard.StepEngines, Level: onboard.Running, Detail: "installing engines"}) + seen := map[string]string{} + limit := 20 * time.Minute + if na.waitAll { + limit = time.Hour + } + err = waitFor(ctx, limit, "the engines on "+node, func() (bool, error) { + if err := client.get(ctx, "/api/v1/nodes", &nodes); err != nil { + return false, nil + } + for _, n := range nodes { + if n.ID != node { + continue + } + finished := true + for _, p := range n.Provision { + line := p.State + " " + p.Detail + if seen[p.Engine] != line { + seen[p.Engine] = line + emit(engineEvent(p.Engine, p.Version, p.State, p.Detail)) + } + switch p.State { + case "installing": + if p.Engine == "llamacpp" || na.waitAll { + finished = false + } + case "failed": + if p.Engine == "llamacpp" { + return false, fmt.Errorf("llama.cpp did not install: %s", p.Detail) + } + } + } + return finished && len(n.Provision) > 0, nil + } + return false, nil + }) + if err != nil { + return err + } + msg := "engines ready" + if v, ok := seen["vllm"]; ok && strings.HasPrefix(v, "installing") && !na.waitAll { + msg = "llama.cpp ready; vLLM is still installing, which takes a few minutes more (watch /admin, or add --wait)" + } + emit(onboard.Event{Step: onboard.StepEngines, Level: onboard.Done, Detail: msg}) + return nil +} + +// heardSince reports whether the node has been heard from since t, by the +// agent just started. A node added again is already listed, from its agent +// before the restart, and a cordoned one is not "ready": neither says whether +// the new agent connected. +func heardSince(n adminapi.NodeView, t time.Time) bool { + switch n.State { + case "unreachable", "lost", "unknown": + return false + } + ago, err := time.ParseDuration(n.LastSeenAgo) + if err != nil { + return false + } + // A second's margin: the age is rounded to one. + return ago+time.Second < time.Since(t) +} + +func engineEvent(engine, version, state, detail string) onboard.Event { + level := onboard.Info + switch state { + case "failed", "unsupported": + level = onboard.Warn + } + return onboard.Event{Step: onboard.StepEngines, Level: level, + Detail: fmt.Sprintf("%s %s: %s%s", engine, version, state, prefixed(": ", detail))} +} + +func prefixed(p, s string) string { + if s == "" { + return "" + } + return p + s +} + +func describeNode(n adminapi.NodeView) string { + parts := []string{} + if n.GPUTotal > 0 { + parts = append(parts, fmt.Sprintf("%d GPU(s) %s", n.GPUTotal, strings.Join(n.Classes, ", "))) + } else { + parts = append(parts, "no GPU") + } + if len(n.Fabrics) > 0 { + parts = append(parts, "fabrics "+strings.Join(n.Fabrics, ", ")) + } + return strings.Join(parts, ", ") +} + +// plainProgress prints events as lines, for a log or a pipe. +type plainProgress struct { + out io.Writer +} + +func (p *plainProgress) event(e onboard.Event) { + mark := map[string]string{ + onboard.Running: " .. ", onboard.Done: " ok ", onboard.Info: " ", + onboard.Warn: " !! ", onboard.Fail: " FAIL", + }[e.Level] + fmt.Fprintf(p.out, "%s %-13s %s\n", mark, e.Step, e.Detail) + if e.Fix != "" { + fmt.Fprintf(p.out, " %-13s → %s\n", "", e.Fix) + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 39f895d..0ed9c07 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -46,6 +46,7 @@ func Root() *cobra.Command { newMirrorCmd(), newSimEngineCmd(), newDoctorCmd(), + newNodeCmd(), ) return root } diff --git a/internal/cli/serverinit.go b/internal/cli/serverinit.go index bfc0649..36c8706 100644 --- a/internal/cli/serverinit.go +++ b/internal/cli/serverinit.go @@ -639,21 +639,17 @@ func (in *serverInit) waitHealthy(ctx context.Context) error { } func (in *serverInit) nextSteps() { - server := "https://:7443" - if ips := localAddrs(); len(ips) > 0 { - server = "https://" + net.JoinHostPort(ips[0].String(), "7443") - } - fmt.Fprintf(in.out, ` + fmt.Fprint(in.out, ` The control plane is up. Next: - llmgw ctl status # no nodes, no models: correct for now + llmgw ctl status # the fleet in one answer # each node, from this machine (it holds the authority): - deploy/bootstrap/onboard-node.sh --node --addr --server %s + sudo llmgw node add @ - # then a model: + # then a model: /admin/mirror, "Explore models", or llmgw ctl model add -f model.yaml The web UI is on port 8080; /admin once you are signed in as the administrator. -`, server) +`) } diff --git a/internal/cli/tui_node.go b/internal/cli/tui_node.go new file mode 100644 index 0000000..29bf69e --- /dev/null +++ b/internal/cli/tui_node.go @@ -0,0 +1,354 @@ +package cli + +import ( + "context" + "fmt" + "io" + "os" + "strings" + + "github.com/charmbracelet/bubbles/spinner" + "github.com/charmbracelet/bubbles/textinput" + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" + "github.com/mattn/go-isatty" + + "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" +) + +// The full-screen way through 'llmgw node add': the same steps as the plain +// one, drawn as a checklist that fills in, with the questions a script +// cannot ask — which fabric the adapters are on, whether to go on past a +// problem — asked in place. + +func isTerminal(w io.Writer) bool { + f, ok := w.(*os.File) + return ok && isatty.IsTerminal(f.Fd()) && os.Getenv("TERM") != "dumb" +} + +var ( + tuiAccent = lipgloss.AdaptiveColor{Light: "#0B7F7D", Dark: "#3FC2BC"} + tuiOK = lipgloss.AdaptiveColor{Light: "#1F8A4C", Dark: "#6FD3A6"} + tuiWarn = lipgloss.AdaptiveColor{Light: "#A15C00", Dark: "#E9B35A"} + tuiBad = lipgloss.AdaptiveColor{Light: "#B42318", Dark: "#F07C6C"} + tuiDim = lipgloss.AdaptiveColor{Light: "#56696C", Dark: "#8FA5A4"} + + sTitle = lipgloss.NewStyle().Bold(true).Foreground(tuiAccent) + sStep = lipgloss.NewStyle().Width(14) + sDim = lipgloss.NewStyle().Foreground(tuiDim) + sOK = lipgloss.NewStyle().Foreground(tuiOK) + sWarn = lipgloss.NewStyle().Foreground(tuiWarn) + sBad = lipgloss.NewStyle().Foreground(tuiBad) + sBox = lipgloss.NewStyle().Border(lipgloss.RoundedBorder()).BorderForeground(tuiAccent).Padding(0, 1) +) + +// Messages from the onboarding goroutine to the screen. +type ( + evMsg onboard.Event + finishMsg struct { + node, summary string + err error + } + askFabricsMsg struct { + suggested []onboard.Fabric + reply chan []onboard.Fabric + } + askContinueMsg struct { + fails int + reply chan bool + } +) + +type stepState struct { + level, detail string +} + +type nodeModel struct { + title string + steps map[string]stepState + notes []onboard.Event // warnings and failures, with their fixes + spin spinner.Model + cancel context.CancelFunc + width int + done bool + err error + node string + summary string + + // The question being asked, if any. + fabrics *askFabricsMsg + inputs []textinput.Model + focus int + confirm *askContinueMsg + quitAsked bool +} + +func newNodeModel(title string, cancel context.CancelFunc) nodeModel { + sp := spinner.New() + sp.Spinner = spinner.MiniDot + sp.Style = lipgloss.NewStyle().Foreground(tuiAccent) + return nodeModel{title: title, steps: map[string]stepState{}, spin: sp, cancel: cancel, width: 100} +} + +func (m nodeModel) Init() tea.Cmd { return m.spin.Tick } + +func (m nodeModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { + switch msg := msg.(type) { + case tea.WindowSizeMsg: + m.width = msg.Width + case spinner.TickMsg: + var cmd tea.Cmd + m.spin, cmd = m.spin.Update(msg) + return m, cmd + case evMsg: + e := onboard.Event(msg) + switch e.Level { + case onboard.Running, onboard.Done: + m.steps[e.Step] = stepState{e.Level, e.Detail} + case onboard.Fail: + m.steps[e.Step] = stepState{e.Level, e.Detail} + if e.Step == onboard.StepCheck && strings.Contains(e.Detail, ":") { + m.notes = append(m.notes, e) + } + case onboard.Warn: + m.notes = append(m.notes, e) + case onboard.Info: + if e.Step == onboard.StepEngines { + m.steps[e.Step] = stepState{onboard.Running, e.Detail} + } + } + case askFabricsMsg: + m.fabrics = &msg + m.inputs = nil + for i, f := range msg.suggested { + in := textinput.New() + in.Prompt = "" + in.SetValue(f.Name) + in.CharLimit = 40 + in.Width = 24 + if i == 0 { + in.Focus() + } + m.inputs = append(m.inputs, in) + } + m.focus = 0 + case askContinueMsg: + m.confirm = &msg + case finishMsg: + m.done, m.err, m.node, m.summary = true, msg.err, msg.node, msg.summary + return m, tea.Quit + case tea.KeyMsg: + return m.key(msg) + } + return m, nil +} + +func (m nodeModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) { + if k.Type == tea.KeyCtrlC { + m.cancel() + m.quitAsked = true + return m, tea.Quit + } + switch { + case m.confirm != nil: + switch strings.ToLower(k.String()) { + case "y": + m.confirm.reply <- true + m.confirm = nil + case "n", "enter", "esc": + m.confirm.reply <- false + m.confirm = nil + } + return m, nil + case m.fabrics != nil: + switch k.Type { + case tea.KeyTab, tea.KeyDown: + m.move(1) + return m, nil + case tea.KeyShiftTab, tea.KeyUp: + m.move(-1) + return m, nil + case tea.KeyEsc: + m.fabrics.reply <- nil + m.fabrics = nil + return m, nil + case tea.KeyEnter: + var out []onboard.Fabric + for i, in := range m.inputs { + if name := strings.TrimSpace(in.Value()); name != "" { + out = append(out, onboard.Fabric{Name: name, HCAs: m.fabrics.suggested[i].HCAs}) + } + } + m.fabrics.reply <- out + m.fabrics = nil + return m, nil + } + var cmd tea.Cmd + m.inputs[m.focus], cmd = m.inputs[m.focus].Update(k) + return m, cmd + } + if k.String() == "q" { + m.cancel() + m.quitAsked = true + return m, tea.Quit + } + return m, nil +} + +func (m *nodeModel) move(d int) { + m.inputs[m.focus].Blur() + m.focus = (m.focus + d + len(m.inputs)) % len(m.inputs) + m.inputs[m.focus].Focus() +} + +func (m nodeModel) View() string { + var b strings.Builder + b.WriteString(sTitle.Render(m.title) + "\n\n") + detailWidth := max(20, m.width-22) + for _, step := range onboard.Steps { + st, ok := m.steps[step] + var mark string + switch { + case !ok: + mark = sDim.Render("·") + case st.level == onboard.Running && !m.done: + mark = m.spin.View() + case st.level == onboard.Fail: + mark = sBad.Render("✗") + case st.level == onboard.Running: + mark = sDim.Render("·") + default: + mark = sOK.Render("✓") + } + name := sStep.Render(step) + if !ok { + name = sDim.Render(sStep.Render(step)) + } + fmt.Fprintf(&b, " %s %s %s\n", mark, name, truncate(st.detail, detailWidth)) + } + + if len(m.notes) > 0 { + b.WriteString("\n") + for _, n := range m.notes { + mark := sWarn.Render("!") + if n.Level == onboard.Fail { + mark = sBad.Render("✗") + } + fmt.Fprintf(&b, " %s %s\n", mark, wrap(n.Detail, m.width-6, " ")) + if n.Fix != "" { + fmt.Fprintf(&b, " %s\n", sDim.Render("→ "+wrap(n.Fix, m.width-8, " "))) + } + } + } + + switch { + case m.fabrics != nil: + var q strings.Builder + q.WriteString("These adapters are up. Which fabric is each group on?\n") + q.WriteString(sDim.Render("Give each the name it has on every other node attached to it.\nLeave a name empty to leave that group out.") + "\n\n") + for i, f := range m.fabrics.suggested { + fmt.Fprintf(&q, " %s %s\n", m.inputs[i].View(), sDim.Render(strings.Join(f.HCAs, ", "))) + } + q.WriteString("\n" + sDim.Render("tab next · enter use these · esc none")) + b.WriteString("\n" + sBox.Render(q.String()) + "\n") + case m.confirm != nil: + q := fmt.Sprintf("doctor found %d problem(s) above.\nAdd the node anyway? %s", m.confirm.fails, sDim.Render("y / N")) + b.WriteString("\n" + sBox.BorderForeground(tuiWarn).Render(q) + "\n") + case m.done && m.err != nil: + b.WriteString("\n" + sBad.Render("✗ ") + wrap(m.err.Error(), m.width-4, " ") + "\n") + case m.done: + b.WriteString("\n" + sOK.Render("✓ ") + m.summary + "\n") + case m.quitAsked: + b.WriteString("\n" + sDim.Render("stopped") + "\n") + default: + b.WriteString("\n" + sDim.Render("ctrl+c to stop") + "\n") + } + return b.String() +} + +func truncate(s string, n int) string { + if n <= 1 || len([]rune(s)) <= n { + return s + } + r := []rune(s) + return string(r[:n-1]) + "…" +} + +// wrap breaks s at spaces to fit width, indenting every line after the first. +func wrap(s string, width int, indent string) string { + if width < 20 { + return s + } + var lines []string + line := "" + for _, w := range strings.Fields(s) { + if line != "" && len(line)+1+len(w) > width { + lines = append(lines, line) + line = w + continue + } + if line == "" { + line = w + } else { + line += " " + w + } + } + lines = append(lines, line) + return strings.Join(lines, "\n"+indent) +} + +// runTUI adds the node with the screen above. +func (na *nodeAdd) runTUI(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + m := newNodeModel("llmgw · adding "+na.opts.Addr, cancel) + p := tea.NewProgram(m) + + emit := func(e onboard.Event) { p.Send(evMsg(e)) } + na.opts.AskFabrics = func(s []onboard.Fabric) []onboard.Fabric { + reply := make(chan []onboard.Fabric) + p.Send(askFabricsMsg{s, reply}) + select { + case f := <-reply: + return f + case <-ctx.Done(): + return nil + } + } + na.opts.AskContinue = func(fails int) bool { + reply := make(chan bool) + p.Send(askContinueMsg{fails, reply}) + select { + case ok := <-reply: + return ok + case <-ctx.Done(): + return false + } + } + + go func() { + res, err := onboard.New(na.opts, emit).Run(ctx) + node := "" + if res != nil { + node = res.Node + } + if err == nil && !na.noWait { + err = na.watch(ctx, node, emit) + } + summary := "" + if err == nil { + summary = node + " is in the cluster" + } + p.Send(finishMsg{node, summary, err}) + }() + + final, err := p.Run() + if err != nil { + return err + } + fm := final.(nodeModel) + if fm.quitAsked && !fm.done { + return fmt.Errorf("stopped; the node may be half set up, and running this again finishes it") + } + return fm.err +} diff --git a/internal/cli/tui_node_test.go b/internal/cli/tui_node_test.go new file mode 100644 index 0000000..79de13e --- /dev/null +++ b/internal/cli/tui_node_test.go @@ -0,0 +1,96 @@ +package cli + +import ( + "strings" + "testing" + + "time" + + tea "github.com/charmbracelet/bubbletea" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" +) + +func send(m tea.Model, msgs ...tea.Msg) tea.Model { + for _, msg := range msgs { + m, _ = m.Update(msg) + } + return m +} + +func TestTheNodeScreenFillsInAsStepsFinish(t *testing.T) { + var m tea.Model = newNodeModel("adding gpu01", func() {}) + m = send(m, + evMsg{Step: onboard.StepConnect, Level: onboard.Done, Detail: "gpu01 (amd64)"}, + evMsg{Step: onboard.StepCheck, Level: onboard.Warn, Detail: "fabric: nvidia_peermem is not loaded", Fix: "modprobe nvidia_peermem"}, + evMsg{Step: onboard.StepCheck, Level: onboard.Running, Detail: "running llmgw doctor"}, + ) + v := m.View() + for _, want := range []string{"✓", "gpu01 (amd64)", "nvidia_peermem is not loaded", "→ modprobe nvidia_peermem", "running llmgw doctor", "engines"} { + if !strings.Contains(v, want) { + t.Errorf("screen lacks %q:\n%s", want, v) + } + } +} + +func TestTheFabricQuestionAnswersWithWhatWasTyped(t *testing.T) { + var m tea.Model = newNodeModel("adding gpu01", func() {}) + reply := make(chan []onboard.Fabric, 1) + m = send(m, askFabricsMsg{suggested: []onboard.Fabric{ + {Name: "fabric1", HCAs: []string{"mlx5_0", "mlx5_1"}}, + {Name: "fabric2", HCAs: []string{"mlx5_4"}}, + }, reply: reply}) + if v := m.View(); !strings.Contains(v, "Which fabric") || !strings.Contains(v, "mlx5_0, mlx5_1") { + t.Fatalf("no question on screen:\n%s", m.View()) + } + // Rename the first to ib-a, empty the second to leave it out. + for range len("fabric1") { + m = send(m, tea.KeyMsg{Type: tea.KeyBackspace}) + } + m = send(m, tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("ib-a")}, tea.KeyMsg{Type: tea.KeyTab}) + for range len("fabric2") { + m = send(m, tea.KeyMsg{Type: tea.KeyBackspace}) + } + m = send(m, tea.KeyMsg{Type: tea.KeyEnter}) + got := <-reply + if len(got) != 1 || got[0].Name != "ib-a" || strings.Join(got[0].HCAs, ",") != "mlx5_0,mlx5_1" { + t.Errorf("got %+v", got) + } + if strings.Contains(m.View(), "Which fabric") { + t.Error("the question stayed on screen after it was answered") + } +} + +func TestContinuingPastAProblemIsNoUnlessSaidYes(t *testing.T) { + var m tea.Model = newNodeModel("adding gpu01", func() {}) + reply := make(chan bool, 1) + m = send(m, askContinueMsg{fails: 2, reply: reply}, tea.KeyMsg{Type: tea.KeyEnter}) + if <-reply { + t.Error("enter went on past the problems") + } + m = send(m, askContinueMsg{fails: 2, reply: reply}, tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("y")}) + if !<-reply { + t.Error("y did not go on") + } + _ = m +} + +func TestANodeCountsAsJoinedOnlyWhenHeardSinceItsAgentStarted(t *testing.T) { + start := time.Now().Add(-10 * time.Second) + cases := []struct { + state, ago string + want bool + }{ + {"ready", "3s", true}, + {"cordoned", "3s", true}, // added again while cordoned + {"ready", "30s", false}, // the agent before the restart + {"unreachable", "3s", false}, + {"ready", "", false}, + } + for _, c := range cases { + if got := heardSince(adminapi.NodeView{State: c.state, LastSeenAgo: c.ago}, start); got != c.want { + t.Errorf("%s seen %q ago: got %v", c.state, c.ago, got) + } + } +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 1d17e38..8906a80 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -74,6 +74,7 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights) mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement) mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState) + mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode) mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs) } @@ -812,6 +813,35 @@ func (a *API) Explain(model string) ExplainView { // One endpoint rather than three verbs, because they are the same operation // with different targets and the difference matters: cordon stops new // placement, drain also moves existing work off. +// removeNode forgets a node that has gone for good: a machine retired or +// rebuilt under another name, which otherwise stays listed as lost until the +// control plane restarts. +// +// Only one that is not there: lost, or unreachable with nothing placed on +// it. A connected node removed here would register again on its next +// heartbeat, and one with replicas would take them out of the reconciler's +// sight while they may still be running. +func (a *API) removeNode(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + inv := a.ctl.Inventory() + n, ok := inv.Get(id) + if !ok { + writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id)) + return + } + switch { + case n.State == inventory.Lost: + case n.State == inventory.Unreachable && len(inv.Allocations(id)) == 0: + default: + writeError(w, http.StatusConflict, fmt.Sprintf( + "%s is %s: stop its agent (systemctl stop llmgw-agent) and remove it once it is lost, a minute later", id, n.State)) + return + } + inv.Remove(id) + a.ctl.Reconciler().Notify() + writeJSON(w, http.StatusOK, map[string]string{"id": id, "removed": "true"}) +} + func (a *API) setNodeState(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") diff --git a/internal/control/adminapi/api_test.go b/internal/control/adminapi/api_test.go index 47af26c..21cc382 100644 --- a/internal/control/adminapi/api_test.go +++ b/internal/control/adminapi/api_test.go @@ -667,3 +667,24 @@ func TestAProfileThatIsGoneIsNotReportedAsStale(t *testing.T) { t.Error("a missing model is not a changed profile") } } + +func TestOnlyANodeThatIsGoneCanBeRemoved(t *testing.T) { + a, ctl := setup(t, testModels(), gpuNode("n1", 8)) + + // Connected, even cordoned: it would register again on its next + // heartbeat, so removing it would do nothing but confuse. + ctl.Inventory().SetState("n1", inventory.Cordoned) + if code := call(t, a, "DELETE", "/api/v1/nodes/n1", "", nil); code != 409 { + t.Errorf("a connected node gave %d, expected 409", code) + } + ctl.Inventory().SetState("n1", inventory.Lost) + if code := call(t, a, "DELETE", "/api/v1/nodes/n1", "", nil); code != 200 { + t.Fatalf("a lost node gave %d, expected 200", code) + } + if _, ok := ctl.Inventory().Get("n1"); ok { + t.Error("the node is still listed") + } + if code := call(t, a, "DELETE", "/api/v1/nodes/n1", "", nil); code != 404 { + t.Errorf("removing it again gave %d, expected 404", code) + } +} diff --git a/internal/onboard/onboard.go b/internal/onboard/onboard.go new file mode 100644 index 0000000..c69c40a --- /dev/null +++ b/internal/onboard/onboard.go @@ -0,0 +1,670 @@ +// Package onboard makes a machine a node of the cluster, over SSH, from the +// machine holding the certificate authority. +// +// It is what deploy/bootstrap/onboard-node.sh did, moved into the binary so +// there is nothing else to fetch, and extended by what a script could not +// easily do: it asks the machine its own name and architecture instead of +// being told, runs llmgw doctor there before changing anything, and suggests +// the fabric its adapters are on. +// +// It reports each step as an Event, so a plain terminal and a full-screen one +// show the same thing from the same code. +package onboard + +import ( + "context" + "crypto/x509" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "regexp" + "slices" + "strings" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/doctor" + "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" +) + +// Fabric is one InfiniBand or RoCE fabric the node is on, and its adapters. +type Fabric struct { + Name string + HCAs []string +} + +// ParseFabric reads "name:hca1,hca2", or "name" alone. +func ParseFabric(s string) (Fabric, error) { + name, hcas, _ := strings.Cut(strings.TrimSpace(s), ":") + if name == "" { + return Fabric{}, fmt.Errorf("fabric %q has no name: give it as name:mlx5_0,mlx5_1", s) + } + f := Fabric{Name: name} + for _, h := range strings.Split(hcas, ",") { + if h = strings.TrimSpace(h); h != "" { + f.HCAs = append(f.HCAs, h) + } + } + return f, nil +} + +// Options say which machine, and what to tell its agent. +type Options struct { + // Addr is how to reach the machine over SSH, and User who to log in as. + Addr, User string + // Node is its id in the cluster. Empty means its hostname, which is what + // it should be: the node id, the hostname and the certificate's name + // being one string saves an afternoon each time they would differ. + Node string + // Servers are the control planes the agent dials. Empty means this + // machine, at the address it has on the route to the node. + Servers []string + Fabrics []Fabric + + MgmtAddr, MgmtIface string + NvidiaSMI, IBRoot string + LlamaCppBin string + VLLMBin string + SGLangPython string + + // PKI is the authority's directory; EtcDir is where this machine's own + // server certificate is, to check the chosen address is named in it. + PKI, EtcDir string + // Binary is the llmgw to install when the node cannot download the + // release, and Version what it is. + Binary, Version string + // Releases is where release files are downloaded from. + Releases string + // IgnoreChecks goes on when doctor finds a problem. + IgnoreChecks bool + // SSHWait is how long an unreachable machine is waited for. + SSHWait time.Duration + + // AskFabrics, when set, is asked which fabrics to declare for a node + // given none whose adapters are up, with the adapters grouped by the + // fabric they report. It returns the fabrics to use, none to go on + // without. Nil leaves it at a warning, as a script needs. + AskFabrics func(suggested []Fabric) []Fabric + // AskContinue, when set, is asked whether to go on after doctor found + // problems. Nil stops unless IgnoreChecks. + AskContinue func(fails int) bool +} + +// Levels of an Event. +const ( + Running = "running" + Done = "done" + Info = "info" + Warn = "warn" + Fail = "fail" +) + +// Event is one thing that happened: a step starting or ending, or something +// worth saying along the way. +type Event struct { + Step string + Level string + Detail string + // Fix is what to do about a warning or a failure, when there is a known + // answer. + Fix string +} + +// The steps, in order. +const ( + StepConnect = "connect" + StepCheck = "check" + StepBinary = "binary" + StepCert = "certificate" + StepConfig = "configuration" + StepStart = "start" + // StepJoin and StepEngines are watched from the control plane once the + // agent runs; the caller does that, since it has the admin API. + StepJoin = "join" + StepEngines = "engines" +) + +// Steps lists them for a display that shows what is still to come. +var Steps = []string{StepConnect, StepBinary, StepCheck, StepCert, StepConfig, StepStart, StepJoin, StepEngines} + +// Result is what onboarding learned and did. +type Result struct { + Node string + Arch string + Hostname string + Servers []string + Facts doctor.Facts + Findings []doctor.Finding + // Suggested is the fabrics the node's active adapters are on, for a node + // given none: grouped by what the adapters report, named for the operator + // to choose. + Suggested []Fabric + // EngineProblems are engines named by path that will not run there. + EngineProblems []string +} + +// Runner onboards one machine. +type Runner struct { + Opts Options + Remote Remote + // Emit hears every event. Nil discards them. + Emit func(Event) + + ca *tlsconf.CA +} + +// New makes a Runner that reaches the machine with the system's ssh. +func New(o Options, emit func(Event)) *Runner { + return &Runner{Opts: o, Remote: &SSH{Addr: o.Addr, User: o.User}, Emit: emit} +} + +func (r *Runner) emit(step, level, detail, fix string) { + if r.Emit != nil { + r.Emit(Event{Step: step, Level: level, Detail: detail, Fix: fix}) + } +} + +// ErrChecks is returned when doctor found problems and IgnoreChecks is off. +var ErrChecks = errors.New("the node has problems to fix first") + +// Run does every step, stopping at the first that fails. +func (r *Runner) Run(ctx context.Context) (*Result, error) { + res := &Result{} + // The authority first: without it nothing can be finished, and finding + // out after the node has been changed leaves it half done. + pki := r.Opts.PKI + ca, err := tlsconf.LoadCA(filepath.Join(pki, "ca.pem"), filepath.Join(pki, "ca-key.pem")) + if err != nil { + err = fmt.Errorf("cannot read the certificate authority in %s: %v. Run this on the machine that holds it, as root (sudo), or give --pki", pki, err) + r.emit(StepConnect, Fail, err.Error(), "") + return res, err + } + r.ca = ca + for _, step := range []struct { + name string + fn func(context.Context, *Result) error + }{ + {StepConnect, r.connect}, + {StepBinary, r.binary}, + {StepCheck, r.check}, + {StepCert, r.certificate}, + {StepConfig, r.configure}, + {StepStart, r.start}, + } { + if err := step.fn(ctx, res); err != nil { + r.emit(step.name, Fail, err.Error(), "") + return res, err + } + } + return res, nil +} + +// --- connect --- + +var nodeName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,62}$`) + +func (r *Runner) connect(ctx context.Context, res *Result) error { + r.emit(StepConnect, Running, "connecting to "+r.target(), "") + wait := r.Opts.SSHWait + if wait == 0 { + wait = 3 * time.Minute + } + // Waited for rather than failed on: a machine a minute old restarts sshd + // after cloud-init regenerates its host keys, and refuses connections for + // a few seconds after answering. + deadline := time.Now().Add(wait) + var last error + for { + out, err := r.Remote.Run(ctx, "hostname && uname -m", nil) + if err == nil { + lines := strings.Fields(string(out)) + if len(lines) >= 2 { + res.Hostname, res.Arch = lines[0], lines[1] + break + } + err = fmt.Errorf("unexpected answer: %q", out) + } + last = err + if time.Now().After(deadline) { + return fmt.Errorf("cannot log in to %s: %v. Check that `ssh %s true` works from this machine, as the user running this, without a password", + r.target(), last, r.target()) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(5 * time.Second): + } + } + if _, err := r.Remote.Run(ctx, "sudo -n true", nil); err != nil { + return fmt.Errorf("%s can log in but cannot use sudo without a password, which installing the agent needs", r.target()) + } + switch res.Arch { + case "x86_64", "amd64": + res.Arch = "amd64" + case "aarch64", "arm64": + res.Arch = "arm64" + default: + return fmt.Errorf("%s is %s; llmgw is built for x86-64 and ARM64", r.target(), res.Arch) + } + res.Node = r.Opts.Node + if res.Node == "" { + res.Node = res.Hostname + } + if !nodeName.MatchString(res.Node) { + return fmt.Errorf("%q cannot be a node id: letters, digits, dots, dashes and underscores; give one with --node", res.Node) + } + servers, err := r.servers() + if err != nil { + return err + } + res.Servers = servers + r.emit(StepConnect, Done, fmt.Sprintf("%s (%s), joining as %s, agent will dial %s", + res.Hostname, res.Arch, res.Node, strings.Join(servers, ", ")), "") + return nil +} + +func (r *Runner) target() string { + if r.Opts.User == "" { + return r.Opts.Addr + } + return r.Opts.User + "@" + r.Opts.Addr +} + +// servers are the given control planes, or this machine at the address it +// uses to reach the node — which is the one the node can reach it at, and +// is named in the server certificate when init issued it here. +func (r *Runner) servers() ([]string, error) { + if len(r.Opts.Servers) > 0 { + return r.Opts.Servers, nil + } + host := r.Opts.Addr + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + c, err := net.Dial("udp", net.JoinHostPort(host, "7443")) + if err != nil { + return nil, fmt.Errorf("cannot tell which address the node would reach this machine at: %v; give it with --server", err) + } + local := c.LocalAddr().(*net.UDPAddr).IP.String() + _ = c.Close() + if names, err := certNames(filepath.Join(r.etcDir(), "server.pem")); err == nil && !slices.Contains(names, local) { + return nil, fmt.Errorf("the node would reach this machine at %s, which the server certificate does not name (%s): "+ + "run `sudo llmgw server init --host %s` to add it, or give the address with --server", + local, strings.Join(names, ", "), local) + } + return []string{"https://" + net.JoinHostPort(local, "7443")}, nil +} + +func (r *Runner) etcDir() string { + if r.Opts.EtcDir != "" { + return r.Opts.EtcDir + } + return "/etc/llmgw" +} + +func certNames(path string) ([]string, error) { + b, err := os.ReadFile(path) + if err != nil { + return nil, err + } + blk, _ := pem.Decode(b) + if blk == nil { + return nil, errors.New("no certificate in " + path) + } + c, err := x509.ParseCertificate(blk.Bytes) + if err != nil { + return nil, err + } + names := slices.Clone(c.DNSNames) + for _, ip := range c.IPAddresses { + names = append(names, ip.String()) + } + return names, nil +} + +// --- binary --- + +var releaseTag = regexp.MustCompile(`^v\d+\.\d+\.\d+$`) + +func (r *Runner) binary(ctx context.Context, res *Result) error { + v := r.Opts.Version + if releaseTag.MatchString(v) { + r.emit(StepBinary, Running, "downloading llmgw "+v+" on the node", "") + name := "llmgw-" + v + "-linux-" + res.Arch + base := strings.TrimSuffix(r.Opts.Releases, "/") + "/" + v + // Fetched by the node, for its own architecture, and checked against + // the release's checksums, rather than handed whatever this machine + // happens to run. + script := fmt.Sprintf(`set -e +d=$(mktemp -d); cd "$d" +curl -fsSL -o %[1]s '%[2]s/%[1]s' +curl -fsSL -o checksums.txt '%[2]s/checksums.txt' +grep " %[1]s$" checksums.txt | sha256sum -c - >/dev/null +sudo install -m 0755 %[1]s /usr/local/bin/llmgw +cd /; rm -rf "$d"`, name, base) + out, err := r.Remote.Run(ctx, script, nil) + if err == nil { + r.emit(StepBinary, Done, "llmgw "+v+" from the release, checksum verified", "") + return nil + } + r.emit(StepBinary, Warn, "the node could not download the release: "+lastLine(out), "") + } + // A development build has no release to fetch, and a node that cannot + // reach the releases gets this machine's copy — when it can run it. + if res.Arch != hostArch() { + return fmt.Errorf("the node is %s and this machine %s, so its binary cannot be copied: "+ + "let the node reach %s, or install llmgw there yourself", res.Arch, hostArch(), r.Opts.Releases) + } + r.emit(StepBinary, Running, "copying this machine's llmgw ("+orUnknown(v)+")", "") + b, err := os.ReadFile(r.Opts.Binary) + if err != nil { + return fmt.Errorf("reading %s: %w", r.Opts.Binary, err) + } + if out, err := r.Remote.Run(ctx, `set -e; t=$(mktemp); cat > "$t"; sudo install -m 0755 "$t" /usr/local/bin/llmgw; rm -f "$t"`, b); err != nil { + return fmt.Errorf("installing the binary: %v: %s", err, lastLine(out)) + } + r.emit(StepBinary, Done, "llmgw "+orUnknown(v)+", copied from this machine", "") + return nil +} + +// --- check --- + +func (r *Runner) check(ctx context.Context, res *Result) error { + r.emit(StepCheck, Running, "running llmgw doctor on the node", "") + args := "sudo llmgw doctor --json" + if r.Opts.NvidiaSMI != "" { + args += " --nvidia-smi " + shellQuote(r.Opts.NvidiaSMI) + } + if r.Opts.IBRoot != "" { + args += " --ib-root " + shellQuote(r.Opts.IBRoot) + } + out, err := r.Remote.Run(ctx, args, nil) + var rep struct { + Findings []doctor.Finding `json:"findings"` + Facts doctor.Facts `json:"facts"` + } + if jerr := json.Unmarshal(out, &rep); jerr != nil { + if err == nil { + err = jerr + } + return fmt.Errorf("llmgw doctor did not run on the node: %v: %s", err, lastLine(out)) + } + res.Facts, res.Findings = rep.Facts, rep.Findings + + fails := 0 + for _, f := range rep.Findings { + switch f.Level { + case doctor.Fail: + fails++ + r.emit(StepCheck, Fail, f.Area+": "+f.Message, f.Fix) + case doctor.Warn: + r.emit(StepCheck, Warn, f.Area+": "+f.Message, f.Fix) + case doctor.OK: + r.emit(StepCheck, Info, f.Area+": "+f.Message, "") + } + } + res.Suggested = suggestFabrics(rep.Facts.HCAs) + if len(r.Opts.Fabrics) == 0 && len(res.Suggested) > 0 && r.Opts.AskFabrics != nil { + r.Opts.Fabrics = r.Opts.AskFabrics(res.Suggested) + for _, f := range r.Opts.Fabrics { + r.emit(StepCheck, Info, "fabric "+f.Name+": "+strings.Join(f.HCAs, ", "), "") + } + } + if len(r.Opts.Fabrics) == 0 && len(res.Suggested) > 0 { + var flags []string + for _, f := range res.Suggested { + flags = append(flags, "--fabric "+f.Name+":"+strings.Join(f.HCAs, ",")) + } + r.emit(StepCheck, Warn, "InfiniBand or RoCE adapters are up, but no fabric was given: this node would never be given a replica spanning nodes", + "add "+strings.Join(flags, " ")+", naming each fabric the same on every node attached to it") + } + goOn := r.Opts.IgnoreChecks + if fails > 0 && !goOn && r.Opts.AskContinue != nil && r.Opts.AskContinue(fails) { + r.emit(StepCheck, Warn, fmt.Sprintf("going on despite %d problem(s)", fails), "") + goOn = true + } + if fails > 0 && !goOn { + return fmt.Errorf("%w: doctor found %d; fix them and run this again, or add --ignore-checks", ErrChecks, fails) + } + r.emit(StepCheck, Done, fmt.Sprintf("%d problem(s), %d warning(s)", fails, countLevel(rep.Findings, doctor.Warn)), "") + return nil +} + +// suggestFabrics groups the active adapters by the fabric they report being +// on: the subnet manager they see, or the network of their RoCE address. +// Names are placeholders for the operator: a name is what ties one node's +// adapters to another's, and nothing on one node can know what the others +// were called. +func suggestFabrics(hcas []discovery.HCA) []Fabric { + byKey := map[string][]string{} + var keys []string + for _, h := range hcas { + for _, p := range h.ActivePorts() { + k := p.FabricKey() + if k == "" { + k = "unknown" + } + if _, ok := byKey[k]; !ok { + keys = append(keys, k) + } + if !slices.Contains(byKey[k], h.Name) { + byKey[k] = append(byKey[k], h.Name) + } + } + } + var out []Fabric + for i, k := range keys { + name := "fabric" + if len(keys) > 1 { + name = fmt.Sprintf("fabric%d", i+1) + } + out = append(out, Fabric{Name: name, HCAs: byKey[k]}) + } + return out +} + +// --- certificate --- + +func (r *Runner) certificate(ctx context.Context, res *Result) error { + r.emit(StepCert, Running, "issuing a certificate for "+res.Node, "") + ca := r.ca + // Issued fresh every time: onboarding again should not depend on finding + // an old key, and the old one keeps working until revoked. + leaf, err := ca.IssueClient(res.Node) + if err != nil { + return err + } + for _, f := range []struct { + path, mode string + body []byte + }{ + {"/etc/llmgw/ca.pem", "0644", ca.PEM}, + {"/etc/llmgw/agent.pem", "0644", leaf.CertPEM}, + {"/etc/llmgw/agent-key.pem", "0600", leaf.KeyPEM}, + } { + if err := r.put(ctx, f.path, f.mode, f.body); err != nil { + return err + } + } + // Trusting the root too, so other tools on the machine do not treat the + // cluster's certificates as strangers. Best effort: either family's way. + _, _ = r.Remote.Run(ctx, `if [ -d /usr/local/share/ca-certificates ]; then + sudo install -m 0644 /etc/llmgw/ca.pem /usr/local/share/ca-certificates/llmgw-ca.crt && sudo update-ca-certificates >/dev/null 2>&1 +elif [ -d /etc/pki/ca-trust/source/anchors ]; then + sudo install -m 0644 /etc/llmgw/ca.pem /etc/pki/ca-trust/source/anchors/llmgw-ca.pem && sudo update-ca-trust >/dev/null 2>&1 +fi; true`, nil) + r.emit(StepCert, Done, "certificate for "+res.Node+" installed, the cluster's root trusted", "") + return nil +} + +// put writes a file on the node as root, never readable by others on the +// way when its mode says it should not be. +func (r *Runner) put(ctx context.Context, path, mode string, body []byte) error { + script := fmt.Sprintf(`set -e +sudo install -d -m 0755 %[1]s +sudo sh -c 'umask 077; cat > %[2]s.new' +sudo chmod %[3]s %[2]s.new +sudo chown root:root %[2]s.new +sudo mv -f %[2]s.new %[2]s`, filepath.Dir(path), path, mode) + if out, err := r.Remote.Run(ctx, script, body); err != nil { + return fmt.Errorf("writing %s: %v: %s", path, err, lastLine(out)) + } + return nil +} + +// --- configuration --- + +// AgentUnit is the agent's systemd unit, the same one server init writes. +var AgentUnit string + +func (r *Runner) configure(ctx context.Context, res *Result) error { + r.emit(StepConfig, Running, "writing /etc/llmgw/agent.yaml", "") + if err := r.put(ctx, "/etc/llmgw/agent.yaml", "0644", []byte(AgentYAML(r.Opts, res))); err != nil { + return err + } + if AgentUnit == "" { + return errors.New("onboard: AgentUnit is not set") + } + if err := r.put(ctx, "/etc/systemd/system/llmgw-agent.service", "0644", []byte(AgentUnit)); err != nil { + return err + } + res.EngineProblems = r.checkEngines(ctx) + for _, p := range res.EngineProblems { + r.emit(StepConfig, Warn, p, "") + } + r.emit(StepConfig, Done, "agent.yaml and the systemd unit", "") + return nil +} + +// AgentYAML is the node's configuration: only what the node cannot find out +// for itself. Its address, GPUs, adapters and engines it discovers, and every +// line written here is one somebody has to keep true by hand. +func AgentYAML(o Options, res *Result) string { + var b strings.Builder + fmt.Fprintf(&b, "# %s, written by 'llmgw node add'.\n", res.Node) + if len(res.Servers) == 1 { + fmt.Fprintf(&b, "server: %q\n", res.Servers[0]) + } else { + // Every control plane it may fail over to; a standby refuses it and + // it moves on to the next. + b.WriteString("servers:\n") + for _, s := range res.Servers { + fmt.Fprintf(&b, " - %q\n", s) + } + } + b.WriteString("\ntls:\n ca: /etc/llmgw/ca.pem\n cert: /etc/llmgw/agent.pem\n key: /etc/llmgw/agent-key.pem\n") + // The control plane refuses a node whose id differs from its + // certificate's name; said only when the hostname would say otherwise. + if res.Node != res.Hostname { + fmt.Fprintf(&b, "node_id: %q\n", res.Node) + } + if o.MgmtAddr != "" { + fmt.Fprintf(&b, "mgmt_addr: %q\n", o.MgmtAddr) + } + if o.MgmtIface != "" { + fmt.Fprintf(&b, "mgmt_interface: %q\n", o.MgmtIface) + } + if len(o.Fabrics) > 0 { + // The one thing a node cannot find out: which name its fabric goes + // by on every other node. + b.WriteString("fabrics:\n") + for _, f := range o.Fabrics { + fmt.Fprintf(&b, " - name: %q\n hcas: [%s]\n", f.Name, strings.Join(f.HCAs, ", ")) + } + } + if o.NvidiaSMI != "" || o.IBRoot != "" { + b.WriteString("discovery:\n") + if o.NvidiaSMI != "" { + fmt.Fprintf(&b, " nvidia_smi: %q\n", o.NvidiaSMI) + } + if o.IBRoot != "" { + fmt.Fprintf(&b, " ib_root: %q\n", o.IBRoot) + } + } + if o.LlamaCppBin != "" || o.VLLMBin != "" || o.SGLangPython != "" { + b.WriteString("engines:\n") + if o.LlamaCppBin != "" { + fmt.Fprintf(&b, " llamacpp:\n bin: %q\n", o.LlamaCppBin) + } + if o.VLLMBin != "" { + fmt.Fprintf(&b, " vllm:\n bin: %q\n", o.VLLMBin) + } + if o.SGLangPython != "" { + fmt.Fprintf(&b, " sglang:\n python: %q\n", o.SGLangPython) + } + } + return b.String() +} + +// checkEngines runs each engine named by path. A wrong one registers as +// healthy and fails every shard placed on it, with the reason three logs +// away: llama-server on node05 was present, executable, and missing libgomp. +func (r *Runner) checkEngines(ctx context.Context) []string { + var problems []string + for _, e := range []struct{ name, path string }{ + {"llama.cpp", r.Opts.LlamaCppBin}, {"vLLM", r.Opts.VLLMBin}, {"SGLang", r.Opts.SGLangPython}, + } { + if e.path == "" { + continue + } + q := shellQuote(e.path) + if _, err := r.Remote.Run(ctx, "test -x "+q, nil); err != nil { + problems = append(problems, fmt.Sprintf("%s is set to %s, which is not an executable on the node", e.name, e.path)) + continue + } + if _, err := r.Remote.Run(ctx, q+" --version >/dev/null 2>&1 || "+q+" --help >/dev/null 2>&1", nil); err != nil { + problems = append(problems, fmt.Sprintf("%s at %s exists but will not run; usually a missing library: ldd %s | grep 'not found'", + e.name, e.path, e.path)) + } + } + return problems +} + +// --- start --- + +func (r *Runner) start(ctx context.Context, _ *Result) error { + r.emit(StepStart, Running, "starting the agent", "") + if out, err := r.Remote.Run(ctx, `set -e +sudo systemctl daemon-reload +sudo systemctl enable llmgw-agent >/dev/null 2>&1 +sudo systemctl restart llmgw-agent +sleep 2 +systemctl is-active llmgw-agent`, nil); err != nil { + return fmt.Errorf("the agent did not start: %s; see journalctl -u llmgw-agent on the node", lastLine(out)) + } + r.emit(StepStart, Done, "llmgw-agent is running", "") + return nil +} + +// --- helpers --- + +func lastLine(b []byte) string { + lines := strings.Split(strings.TrimSpace(string(b)), "\n") + return strings.TrimSpace(lines[len(lines)-1]) +} + +func orUnknown(s string) string { + if s == "" { + return "unversioned" + } + return s +} + +func countLevel(fs []doctor.Finding, l doctor.Level) int { + n := 0 + for _, f := range fs { + if f.Level == l { + n++ + } + } + return n +} + +// shellQuote quotes a string for a POSIX shell. +func shellQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" +} diff --git a/internal/onboard/onboard_test.go b/internal/onboard/onboard_test.go new file mode 100644 index 0000000..8a40cd7 --- /dev/null +++ b/internal/onboard/onboard_test.go @@ -0,0 +1,282 @@ +package onboard + +import ( + "context" + "encoding/json" + "errors" + "regexp" + "runtime" + "strings" + "sync" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/doctor" + "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" +) + +// fakeMachine answers what onboarding asks a machine, and keeps what it is +// given. +type fakeMachine struct { + mu sync.Mutex + hostname string + arch string + noSudo bool + report doctorReport + files map[string][]byte + scripts []string + download error +} + +type doctorReport struct { + Findings []doctor.Finding `json:"findings"` + Facts doctor.Facts `json:"facts"` +} + +var putPath = regexp.MustCompile(`cat > (\S+)\.new`) + +func (f *fakeMachine) Run(_ context.Context, script string, stdin []byte) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.scripts = append(f.scripts, script) + switch { + case script == "hostname && uname -m": + return []byte(f.hostname + "\n" + f.arch + "\n"), nil + case script == "sudo -n true": + if f.noSudo { + return []byte("sudo: a password is required"), errors.New("exit 1") + } + return nil, nil + case strings.Contains(script, "checksums.txt"): + if f.download != nil { + return []byte("curl: (6) Could not resolve host"), f.download + } + return nil, nil + case strings.Contains(script, "install -m 0755") && stdin != nil: + f.files["/usr/local/bin/llmgw"] = stdin + return nil, nil + case strings.HasPrefix(script, "sudo llmgw doctor --json"): + b, _ := json.Marshal(f.report) + return b, nil + case putPath.MatchString(script): + f.files[putPath.FindStringSubmatch(script)[1]] = stdin + return nil, nil + case strings.Contains(script, "systemctl restart llmgw-agent"): + return []byte("active\n"), nil + } + return nil, nil +} + +func (f *fakeMachine) ran(s string) bool { + for _, x := range f.scripts { + if strings.Contains(x, s) { + return true + } + } + return false +} + +func newFake() *fakeMachine { + return &fakeMachine{hostname: "gpu01", arch: map[string]string{"amd64": "x86_64", "arm64": "aarch64"}[runtime.GOARCH], + files: map[string][]byte{}} +} + +func runner(t *testing.T, m *fakeMachine, o Options) (*Runner, *[]Event) { + t.Helper() + pki := t.TempDir() + ca, err := tlsconf.NewCA("test") + if err != nil { + t.Fatal(err) + } + if err := ca.Write(pki); err != nil { + t.Fatal(err) + } + o.Addr = "10.0.0.21" + o.PKI, o.EtcDir = pki, t.TempDir() + if o.Servers == nil { + o.Servers = []string{"https://10.0.0.10:7443"} + } + o.Binary = "onboard_test.go" // anything readable + AgentUnit = "[Unit]\n" + var events []Event + var mu sync.Mutex + return &Runner{Opts: o, Remote: m, Emit: func(e Event) { mu.Lock(); events = append(events, e); mu.Unlock() }}, &events +} + +func TestANodeIsAddedFromItsAddressAlone(t *testing.T) { + m := newFake() + r, _ := runner(t, m, Options{Version: "v1.2.3-4-gabc"}) + res, err := r.Run(t.Context()) + if err != nil { + t.Fatal(err) + } + if res.Node != "gpu01" { + t.Errorf("node id %q, want the machine's hostname", res.Node) + } + for _, f := range []string{"/etc/llmgw/ca.pem", "/etc/llmgw/agent.pem", "/etc/llmgw/agent-key.pem", + "/etc/llmgw/agent.yaml", "/etc/systemd/system/llmgw-agent.service", "/usr/local/bin/llmgw"} { + if len(m.files[f]) == 0 { + t.Errorf("%s was not written", f) + } + } + yaml := string(m.files["/etc/llmgw/agent.yaml"]) + if !strings.Contains(yaml, `server: "https://10.0.0.10:7443"`) || strings.Contains(yaml, "node_id") { + t.Errorf("agent.yaml:\n%s", yaml) + } + // The key is never readable by others, not even for a moment on the way. + if !m.ran("umask 077; cat > /etc/llmgw/agent-key.pem.new") || !m.ran("chmod 0600 /etc/llmgw/agent-key.pem.new") { + t.Error("the key was not written private") + } + if !m.ran("systemctl restart llmgw-agent") { + t.Error("the agent was not started") + } +} + +func TestAReleaseIsFetchedByTheNodeAndCheckedThere(t *testing.T) { + m := newFake() + r, _ := runner(t, m, Options{Version: "v1.2.3", Releases: "https://example.test/releases/download"}) + if _, err := r.Run(t.Context()); err != nil { + t.Fatal(err) + } + if !m.ran("sha256sum -c") || !m.ran("https://example.test/releases/download/v1.2.3") { + t.Error("the release was not fetched and verified on the node") + } + if _, copied := m.files["/usr/local/bin/llmgw"]; copied { + t.Error("this machine's binary was copied although the node fetched the release") + } +} + +func TestANodeThatCannotDownloadGetsThisMachinesCopy(t *testing.T) { + m := newFake() + m.download = errors.New("exit 6") + r, events := runner(t, m, Options{Version: "v1.2.3", Releases: "https://example.test"}) + if _, err := r.Run(t.Context()); err != nil { + t.Fatal(err) + } + if len(m.files["/usr/local/bin/llmgw"]) == 0 { + t.Error("the binary was not copied") + } + if !said(*events, Warn, "could not download") { + t.Error("the failed download was not mentioned") + } +} + +func TestAnotherArchitectureIsNotGivenThisMachinesBinary(t *testing.T) { + m := newFake() + m.arch = map[string]string{"amd64": "aarch64", "arm64": "x86_64"}[runtime.GOARCH] + r, _ := runner(t, m, Options{Version: "dev"}) + _, err := r.Run(t.Context()) + if err == nil || !strings.Contains(err.Error(), "cannot be copied") { + t.Errorf("got %v", err) + } +} + +func TestNoSudoIsSaidPlainly(t *testing.T) { + m := newFake() + m.noSudo = true + r, _ := runner(t, m, Options{}) + _, err := r.Run(t.Context()) + if err == nil || !strings.Contains(err.Error(), "sudo without a password") { + t.Errorf("got %v", err) + } + if len(m.files) > 0 { + t.Error("the machine was changed before finding it could not be finished") + } +} + +func TestADoctorProblemStopsItUnlessAnswered(t *testing.T) { + fail := doctor.Finding{Level: doctor.Fail, Area: "nvlink", Message: "fabric manager inactive", Fix: "systemctl enable --now nvidia-fabricmanager"} + + m := newFake() + m.report.Findings = []doctor.Finding{fail} + r, events := runner(t, m, Options{Version: "dev"}) + if _, err := r.Run(t.Context()); !errors.Is(err, ErrChecks) { + t.Fatalf("got %v, want ErrChecks", err) + } + if !said(*events, Fail, "fabric manager inactive") { + t.Error("the problem itself was not reported") + } + if _, ok := m.files["/etc/llmgw/agent.pem"]; ok { + t.Error("a certificate was installed on a machine that failed its checks") + } + + m = newFake() + m.report.Findings = []doctor.Finding{fail} + asked := 0 + r, _ = runner(t, m, Options{Version: "dev", AskContinue: func(n int) bool { asked = n; return true }}) + if _, err := r.Run(t.Context()); err != nil { + t.Fatal(err) + } + if asked != 1 { + t.Errorf("asked about %d problems, want 1", asked) + } +} + +func ibPort(sm string) discovery.Port { + return discovery.Port{Num: 1, State: "ACTIVE", LinkLayer: "InfiniBand", SM: &discovery.SMInfo{GUID: sm}} +} + +func TestAdaptersAreGroupedByTheFabricTheyReport(t *testing.T) { + hcas := []discovery.HCA{ + {Name: "mlx5_0", Ports: []discovery.Port{ibPort("0xa")}}, + {Name: "mlx5_1", Ports: []discovery.Port{ibPort("0xa")}}, + {Name: "mlx5_4", Ports: []discovery.Port{ibPort("0xb")}}, + {Name: "mlx5_5", Ports: []discovery.Port{{Num: 1, State: "DOWN"}}}, + } + got := suggestFabrics(hcas) + if len(got) != 2 || strings.Join(got[0].HCAs, ",") != "mlx5_0,mlx5_1" || strings.Join(got[1].HCAs, ",") != "mlx5_4" { + t.Errorf("got %+v", got) + } +} + +func TestTheFabricsChosenAreWrittenToTheAgent(t *testing.T) { + m := newFake() + m.report.Facts.HCAs = []discovery.HCA{{Name: "mlx5_0", Ports: []discovery.Port{ibPort("0xa")}}} + r, _ := runner(t, m, Options{Version: "dev", AskFabrics: func(s []Fabric) []Fabric { + return []Fabric{{Name: "ib-a", HCAs: s[0].HCAs}} + }}) + if _, err := r.Run(t.Context()); err != nil { + t.Fatal(err) + } + if yaml := string(m.files["/etc/llmgw/agent.yaml"]); !strings.Contains(yaml, `- name: "ib-a"`) || !strings.Contains(yaml, "hcas: [mlx5_0]") { + t.Errorf("agent.yaml:\n%s", yaml) + } + + // Asked nothing, a script is told what to add instead. + m = newFake() + m.report.Facts.HCAs = []discovery.HCA{{Name: "mlx5_0", Ports: []discovery.Port{ibPort("0xa")}}} + r, events := runner(t, m, Options{Version: "dev"}) + if _, err := r.Run(t.Context()); err != nil { + t.Fatal(err) + } + if !said(*events, Warn, "no fabric was given") { + t.Error("adapters up with no fabric given went unmentioned") + } +} + +func TestANodeNamedOtherThanItsHostnameSaysSo(t *testing.T) { + res := &Result{Node: "gpu-a", Hostname: "ip-10-0-0-21", Servers: []string{"https://a:7443", "https://b:7443"}} + y := AgentYAML(Options{}, res) + if !strings.Contains(y, `node_id: "gpu-a"`) || !strings.Contains(y, "servers:\n - \"https://a:7443\"") { + t.Errorf("agent.yaml:\n%s", y) + } +} + +func TestParseFabric(t *testing.T) { + f, err := ParseFabric("ib-a:mlx5_0, mlx5_1") + if err != nil || f.Name != "ib-a" || strings.Join(f.HCAs, ",") != "mlx5_0,mlx5_1" { + t.Errorf("got %+v %v", f, err) + } + if _, err := ParseFabric(":mlx5_0"); err == nil { + t.Error("a fabric with no name was accepted") + } +} + +func said(events []Event, level, s string) bool { + for _, e := range events { + if e.Level == level && strings.Contains(e.Detail, s) { + return true + } + } + return false +} diff --git a/internal/onboard/ssh.go b/internal/onboard/ssh.go new file mode 100644 index 0000000..85fdcc1 --- /dev/null +++ b/internal/onboard/ssh.go @@ -0,0 +1,62 @@ +package onboard + +import ( + "bytes" + "context" + "os/exec" + "runtime" +) + +// Remote runs a shell command on the machine being onboarded, with input. +type Remote interface { + Run(ctx context.Context, script string, stdin []byte) ([]byte, error) +} + +// SSH is Remote through the system's ssh, so the operator's keys, agent and +// ~/.ssh/config apply exactly as they do when they log in by hand. +type SSH struct { + Addr, User string +} + +// sshOptions never prompt: a question nobody is there to answer would hang +// onboarding instead of failing it. +// +// accept-new rather than refusing unknown host keys: a machine provisioned a +// minute ago has a key nobody has seen, and the risk guarded against is a +// substituted machine, whose changed key is still refused. +var sshOptions = []string{ + "-o", "BatchMode=yes", + "-o", "ConnectTimeout=10", + "-o", "StrictHostKeyChecking=accept-new", + "-o", "ServerAliveInterval=15", + // "Permanently added … to the list of known hosts" and the like are + // for a person at a prompt; here they would be read as the answer. + "-o", "LogLevel=ERROR", +} + +// Run runs script with sh on the machine. Files travel on standard input, +// not scp, which a hardened sshd may not offer. +func (s *SSH) Run(ctx context.Context, script string, stdin []byte) ([]byte, error) { + target := s.Addr + if s.User != "" { + target = s.User + "@" + s.Addr + } + args := append(append([]string{}, sshOptions...), target, "sh", "-c", shellQuote(script)) + cmd := exec.CommandContext(ctx, "ssh", args...) + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + // The answer is standard output alone: ssh and the machine's login + // scripts write to standard error, and mixed in they were taken for the + // answer — the first connection's "Warning: Permanently added" read as + // the machine's hostname. Standard error comes back only with a failure, + // where it is usually the reason. + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + return append(stdout.Bytes(), stderr.Bytes()...), err + } + return stdout.Bytes(), nil +} + +func hostArch() string { return runtime.GOARCH }