Onboard a node with a script, and find two examples that never worked
New nodes had been joined by hand, four times, by me. There was no procedure, so there was nothing to be wrong — and nothing to be right either. Two scripts now: provision-node.sh makes the VM on Proxmox, onboard-node.sh gives it an identity. They are separate because they are run by different people with different access, and the second must run where the certificate authority is, which is the one part that cannot be delegated. Running them turned up why phase 4 was never verified. FabricAttachment carried no koanf tags, so the `name:` key the shipped example documents bound to nothing and `rate_gbps` was never read at all. An agent configured exactly as instructed refused to start with "fabrics[0]: name is required" — naming the key the operator had just written. Every fabric in this project had been configured in a Go test, where the tags do not matter. The server example did not load either: its models had no source, so it failed validation the moment anything tried to parse it. Both examples are now parsed by a test. A published example that has never been loaded is a guess, and these two had been guesses since they were written. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5 committed
1 parent
a075ff4120
commit
c1ac2f022f
11 files changed
+687
-9
No files matched your search
Executable
+174
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# Give a machine an identity in an llmgw cluster.
|
||||
#
|
||||
# Run on the host that holds the certificate authority, because that is the one
|
||||
# thing that cannot be delegated: this script issues the credential by which the
|
||||
# control plane will believe the node is who it says it is.
|
||||
#
|
||||
# What it does, in the order it matters:
|
||||
# 1. issue a client certificate whose common name is the node id
|
||||
# 2. copy the CA root, the certificate and its key to the machine
|
||||
# 3. trust the root system-wide, so other tooling there stops complaining
|
||||
# 4. install the binary, the configuration and the unit
|
||||
# 5. start the agent
|
||||
#
|
||||
# The node id, the VM name, the hostname and the certificate common name are all
|
||||
# the same string. They do not have to be, and every time they are not, someone
|
||||
# spends an afternoon working out which of the four a message is talking about.
|
||||
set -euo pipefail
|
||||
|
||||
usage() {
|
||||
cat >&2 <<USAGE
|
||||
usage: $0 --node ID --addr HOST --server URL [options]
|
||||
|
||||
--node ID node id, hostname and certificate common name
|
||||
--addr HOST how to reach the machine over ssh
|
||||
--server URL the control plane's agent endpoint, e.g. https://10.0.0.140:7443
|
||||
--fabric NAME InfiniBand fabric this node is attached to; repeatable as
|
||||
--fabric NAME:hca1,hca2 . Omit for a node with no fabric,
|
||||
which can host single-node replicas and will never be given
|
||||
a multi-node one.
|
||||
--mgmt-addr ADDR address engines are reachable on (default: --addr)
|
||||
--mgmt-if NAME interface NCCL bootstraps over (default eth0)
|
||||
--pki DIR where the authority lives (default /etc/llmgw/pki)
|
||||
--binary PATH llmgw binary to install (default: the one running this)
|
||||
--ssh-user NAME (default ilker)
|
||||
USAGE
|
||||
exit 2
|
||||
}
|
||||
|
||||
NODE= ADDR= SERVER= MGMT= MGMT_IF=eth0
|
||||
PKI=/etc/llmgw/pki BINARY= SSH_USER=ilker
|
||||
FABRIC_NAME= FABRIC_HCAS="mlx5_0,mlx5_1"
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
case "$1" in
|
||||
--node) NODE="$2"; shift 2;;
|
||||
--addr) ADDR="$2"; shift 2;;
|
||||
--server) SERVER="$2"; shift 2;;
|
||||
--fabric) FABRIC_NAME="${2%%:*}"; [ "${2#*:}" != "$2" ] && FABRIC_HCAS="${2#*:}"; shift 2;;
|
||||
--mgmt-addr) MGMT="$2"; shift 2;;
|
||||
--mgmt-if) MGMT_IF="$2"; shift 2;;
|
||||
--pki) PKI="$2"; shift 2;;
|
||||
--binary) BINARY="$2"; shift 2;;
|
||||
--ssh-user) SSH_USER="$2"; shift 2;;
|
||||
*) usage;;
|
||||
esac
|
||||
done
|
||||
[ -n "$NODE" ] && [ -n "$ADDR" ] && [ -n "$SERVER" ] || usage
|
||||
MGMT="${MGMT:-$ADDR}"
|
||||
BINARY="${BINARY:-$(command -v llmgw)}"
|
||||
[ -x "$BINARY" ] || { echo "no llmgw binary; pass --binary" >&2; exit 1; }
|
||||
[ -f "$PKI/ca-key.pem" ] || { echo "no authority in $PKI; run 'llmgw ctl cert init-ca' first" >&2; exit 1; }
|
||||
|
||||
SSH="ssh -o BatchMode=yes -o ConnectTimeout=10 $SSH_USER@$ADDR"
|
||||
say() { printf ' %s\n' "$*"; }
|
||||
|
||||
echo "onboarding $NODE at $ADDR"
|
||||
$SSH true || { echo "cannot reach $SSH_USER@$ADDR over ssh" >&2; exit 1; }
|
||||
|
||||
# 1. The certificate. Issued fresh every time: re-onboarding a machine should
|
||||
# not depend on finding an old key, and the old one keeps working until it
|
||||
# is revoked, which is a separate decision.
|
||||
say "issuing a certificate"
|
||||
TMP="$(mktemp -d)"
|
||||
trap 'rm -rf "$TMP"' EXIT
|
||||
"$BINARY" ctl cert issue-client "$NODE" --dir "$PKI" --out "$TMP" >/dev/null
|
||||
|
||||
# 2 and 3. Identity, and trusting the root so future tooling on that machine
|
||||
# does not treat our certificates as strangers.
|
||||
say "installing the certificate and trusting the root"
|
||||
scp -q "$PKI/ca.pem" "$TMP/$NODE.pem" "$TMP/$NODE-key.pem" "$SSH_USER@$ADDR:/tmp/"
|
||||
$SSH "set -e
|
||||
sudo install -d -m 0755 /etc/llmgw
|
||||
sudo install -m 0644 /tmp/ca.pem /etc/llmgw/ca.pem
|
||||
sudo install -m 0644 /tmp/$NODE.pem /etc/llmgw/agent.pem
|
||||
sudo install -m 0600 /tmp/$NODE-key.pem /etc/llmgw/agent-key.pem
|
||||
sudo chown root:root /etc/llmgw/*
|
||||
sudo install -m 0644 /tmp/ca.pem /usr/local/share/ca-certificates/llmgw-ca.crt
|
||||
sudo update-ca-certificates >/dev/null 2>&1 || true
|
||||
rm -f /tmp/ca.pem /tmp/$NODE.pem /tmp/$NODE-key.pem"
|
||||
|
||||
# 4. The binary, the configuration and the unit.
|
||||
say "installing the binary"
|
||||
scp -q "$BINARY" "$SSH_USER@$ADDR:/tmp/llmgw"
|
||||
$SSH 'sudo install -m 0755 /tmp/llmgw /usr/local/bin/llmgw && rm -f /tmp/llmgw'
|
||||
|
||||
say "writing the configuration"
|
||||
FABRICS="fabrics: []"
|
||||
if [ -n "$FABRIC_NAME" ]; then
|
||||
FABRICS="fabrics:
|
||||
- name: $FABRIC_NAME
|
||||
hcas: [$(echo "$FABRIC_HCAS" | sed 's/,/, /g')]"
|
||||
fi
|
||||
cat > "$TMP/agent.yaml" <<YAML
|
||||
# $NODE
|
||||
#
|
||||
# Written by onboard-node.sh. Fabric membership is configuration and there is no
|
||||
# way around it: OpenSM's default subnet prefix is the same on every fabric, so
|
||||
# a matching prefix proves nothing and a wrong guess produces a replica that
|
||||
# starts, passes health checks and then hangs on its first all-reduce.
|
||||
node_id: $NODE
|
||||
server: "$SERVER"
|
||||
|
||||
tls:
|
||||
ca: /etc/llmgw/ca.pem
|
||||
cert: /etc/llmgw/agent.pem
|
||||
key: /etc/llmgw/agent-key.pem
|
||||
|
||||
mgmt_addr: "$MGMT"
|
||||
mgmt_interface: $MGMT_IF
|
||||
|
||||
heartbeat_interval: 5s
|
||||
reconnect_base: 1s
|
||||
reconnect_max: 30s
|
||||
|
||||
$FABRICS
|
||||
|
||||
discovery:
|
||||
prefer_nvml: false
|
||||
skip_gpus: false
|
||||
|
||||
supervisor:
|
||||
work_dir: /var/lib/llmgw-agent
|
||||
log_lines: 512
|
||||
stop_timeout: 60s
|
||||
health_interval: 2s
|
||||
YAML
|
||||
scp -q "$TMP/agent.yaml" "$SSH_USER@$ADDR:/tmp/agent.yaml"
|
||||
|
||||
cat > "$TMP/llmgw-agent.service" <<'UNIT'
|
||||
[Unit]
|
||||
Description=llmgw node agent
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
ExecStart=/usr/local/bin/llmgw agent --config /etc/llmgw/agent.yaml
|
||||
# Root, because the agent launches engines and needs device access. Engines
|
||||
# outlive a restart of this unit and are adopted again on the next start, so
|
||||
# KillMode must not take the whole process group with it.
|
||||
KillMode=process
|
||||
Restart=on-failure
|
||||
RestartSec=2s
|
||||
StateDirectory=llmgw-agent
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
UNIT
|
||||
scp -q "$TMP/llmgw-agent.service" "$SSH_USER@$ADDR:/tmp/"
|
||||
|
||||
say "starting the agent"
|
||||
$SSH 'set -e
|
||||
sudo install -m 0644 -o root -g root /tmp/agent.yaml /etc/llmgw/agent.yaml
|
||||
sudo install -m 0644 /tmp/llmgw-agent.service /etc/systemd/system/llmgw-agent.service
|
||||
rm -f /tmp/agent.yaml /tmp/llmgw-agent.service
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now llmgw-agent >/dev/null
|
||||
sudo systemctl restart llmgw-agent'
|
||||
|
||||
echo
|
||||
echo "$NODE onboarded. It should appear in 'llmgw ctl nodes' within a few seconds."
|
||||
echo "If it does not, the control plane's log says why — a certificate whose"
|
||||
echo "common name does not match the claimed node id is refused by design."
|
||||
Executable
+83
@@ -0,0 +1,83 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# Create a VM on Proxmox for a new llmgw node.
|
||||
#
|
||||
# Run on the Proxmox host. It does the half of onboarding that is about
|
||||
# hardware; onboard-node.sh does the half that is about identity, and the two
|
||||
# are separate because they are done by different people with different access.
|
||||
#
|
||||
# Idempotent only in the sense that it refuses to touch an existing VMID. A node
|
||||
# that already exists is not something a provisioning script should have an
|
||||
# opinion about.
|
||||
set -euo pipefail
|
||||
|
||||
usage() {
|
||||
cat >&2 <<USAGE
|
||||
usage: $0 --vmid N --name NAME --ip ADDR [options]
|
||||
|
||||
--vmid N Proxmox VM id; must be free
|
||||
--name NAME VM name; use the node id, so the machine, the hostname and
|
||||
the certificate common name are the same string
|
||||
--ip ADDR address with prefix, e.g. 10.0.0.144/24
|
||||
--gw ADDR gateway (default 10.0.0.1)
|
||||
--cores N (default 4)
|
||||
--memory MB (default 8192)
|
||||
--disk GB (default 20)
|
||||
--bridge NAME (default vmbr0)
|
||||
--storage NAME (default local-lvm)
|
||||
--image PATH cloud image (default Ubuntu 24.04 in the ISO store)
|
||||
--user NAME cloud-init user (default ilker)
|
||||
--sshkeys PATH authorized_keys to install (default /root/.ssh/authorized_keys)
|
||||
USAGE
|
||||
exit 2
|
||||
}
|
||||
|
||||
VMID= NAME= IP= GW=10.0.0.1 CORES=4 MEMORY=8192 DISK=20
|
||||
BRIDGE=vmbr0 STORAGE=local-lvm USER=ilker
|
||||
IMAGE=/var/lib/vz/template/iso/noble-server-cloudimg-amd64.img
|
||||
SSHKEYS=/root/.ssh/authorized_keys
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
case "$1" in
|
||||
--vmid) VMID="$2"; shift 2;;
|
||||
--name) NAME="$2"; shift 2;;
|
||||
--ip) IP="$2"; shift 2;;
|
||||
--gw) GW="$2"; shift 2;;
|
||||
--cores) CORES="$2"; shift 2;;
|
||||
--memory) MEMORY="$2"; shift 2;;
|
||||
--disk) DISK="$2"; shift 2;;
|
||||
--bridge) BRIDGE="$2"; shift 2;;
|
||||
--storage) STORAGE="$2"; shift 2;;
|
||||
--image) IMAGE="$2"; shift 2;;
|
||||
--user) USER="$2"; shift 2;;
|
||||
--sshkeys) SSHKEYS="$2"; shift 2;;
|
||||
*) usage;;
|
||||
esac
|
||||
done
|
||||
[ -n "$VMID" ] && [ -n "$NAME" ] && [ -n "$IP" ] || usage
|
||||
|
||||
if qm status "$VMID" >/dev/null 2>&1; then
|
||||
echo "VM $VMID already exists. Pick a free id; this script will not adopt one." >&2
|
||||
exit 1
|
||||
fi
|
||||
[ -f "$IMAGE" ] || { echo "no cloud image at $IMAGE" >&2; exit 1; }
|
||||
|
||||
echo "creating $NAME as VM $VMID at $IP"
|
||||
qm create "$VMID" \
|
||||
--name "$NAME" --cores "$CORES" --memory "$MEMORY" \
|
||||
--net0 "virtio,bridge=$BRIDGE" --ostype l26 --agent enabled=1
|
||||
|
||||
# Imported rather than installed: a cloud image plus cloud-init is the whole
|
||||
# reason a node takes a minute to build instead of an afternoon.
|
||||
qm importdisk "$VMID" "$IMAGE" "$STORAGE" >/dev/null
|
||||
qm set "$VMID" --scsihw virtio-scsi-pci --scsi0 "$STORAGE:vm-$VMID-disk-0,discard=on,ssd=1" >/dev/null
|
||||
qm disk resize "$VMID" scsi0 "${DISK}G" >/dev/null
|
||||
qm set "$VMID" --ide2 "$STORAGE:cloudinit" --boot order=scsi0 >/dev/null
|
||||
qm set "$VMID" \
|
||||
--ipconfig0 "ip=$IP,gw=$GW" \
|
||||
--nameserver "$GW 8.8.8.8" \
|
||||
--ciuser "$USER" \
|
||||
--sshkeys "$SSHKEYS" >/dev/null
|
||||
|
||||
qm start "$VMID"
|
||||
echo "started. cloud-init takes a minute; then run onboard-node.sh from the CA host."
|
||||
@@ -206,15 +206,51 @@ control:
|
||||
port_range: [20000, 20999]
|
||||
mgmt_interface: eth0
|
||||
|
||||
# Each model needs a source — a weights path or a HuggingFace id — and at least
|
||||
# one profile if the control plane is to place it. Without profiles the model is
|
||||
# catalogued but never started, which is the router-only mode below.
|
||||
models:
|
||||
- name: llama-3.3-70b
|
||||
source: meta-llama/Llama-3.3-70B-Instruct
|
||||
aliases: [llama]
|
||||
fallbacks: [qwen2.5-32b]
|
||||
routing_profile: chat
|
||||
scaling: { min: 1, max: 4 }
|
||||
profiles:
|
||||
# Two nodes of four H100s each. A multi-node profile requires a fabric,
|
||||
# and a replica never spans two of them.
|
||||
- name: tp8
|
||||
engine: vllm
|
||||
gpu_class: h100-80g
|
||||
tp: 8
|
||||
nodes: 2
|
||||
gpus_per_node: 4
|
||||
priority: 100
|
||||
weight: 8
|
||||
startup_timeout: 30m
|
||||
# Where a whole node is free, one node is cheaper to schedule and faster
|
||||
# to start than two.
|
||||
- name: tp4-a100
|
||||
engine: vllm
|
||||
gpu_class: a100-40g
|
||||
tp: 4
|
||||
priority: 50
|
||||
weight: 3
|
||||
startup_timeout: 20m
|
||||
|
||||
- name: qwen2.5-32b
|
||||
source: Qwen/Qwen2.5-32B-Instruct
|
||||
aliases: [qwen]
|
||||
routing_profile: batch
|
||||
scaling: { min: 1, max: 2 }
|
||||
profiles:
|
||||
- name: tp2
|
||||
engine: sglang
|
||||
gpu_class: h100-80g
|
||||
tp: 2
|
||||
priority: 100
|
||||
weight: 2
|
||||
startup_timeout: 15m
|
||||
|
||||
# A static replica list, used only when no model declares a profile. That is the
|
||||
# router-only mode for testing against sim-engines; with the control plane
|
||||
|
||||
@@ -142,6 +142,7 @@ func NewRegistry(cfg Config) *Registry {
|
||||
r.Add(&VLLM{cfg: cfg.VLLM})
|
||||
r.Add(&SGLang{cfg: cfg.SGLang})
|
||||
r.Add(&LlamaCpp{cfg: cfg.LlamaCpp})
|
||||
r.Add(&Sim{cfg: cfg.Sim})
|
||||
return r
|
||||
}
|
||||
|
||||
@@ -172,6 +173,7 @@ type Config struct {
|
||||
VLLM VLLMConfig `koanf:"vllm"`
|
||||
SGLang SGLangConfig `koanf:"sglang"`
|
||||
LlamaCpp LlamaCppConfig `koanf:"llamacpp"`
|
||||
Sim SimConfig `koanf:"sim"`
|
||||
}
|
||||
|
||||
// --- shared helpers ---
|
||||
|
||||
@@ -397,14 +397,14 @@ func TestValidateRequiresModelPath(t *testing.T) {
|
||||
|
||||
func TestRegistry(t *testing.T) {
|
||||
r := NewRegistry(Config{})
|
||||
if got := r.Names(); !slices.Equal(got, []string{"llamacpp", "sglang", "vllm"}) {
|
||||
if got := r.Names(); !slices.Equal(got, []string{"llamacpp", "sglang", "sim", "vllm"}) {
|
||||
t.Errorf("registry contents: %v", got)
|
||||
}
|
||||
if _, err := r.Get("vllm"); err != nil {
|
||||
t.Errorf("vllm missing: %v", err)
|
||||
}
|
||||
err := func() error { _, err := r.Get("nope"); return err }()
|
||||
if err == nil || !strings.Contains(err.Error(), "llamacpp, sglang, vllm") {
|
||||
if err == nil || !strings.Contains(err.Error(), "llamacpp, sglang, sim, vllm") {
|
||||
t.Errorf("an unknown engine should list what is available: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
package engines
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// SimConfig locates the simulated engine.
|
||||
type SimConfig struct {
|
||||
// Bin is the llmgw binary itself; the engine is one of its subcommands.
|
||||
// Empty means whatever started this agent, which is right almost always.
|
||||
Bin string `koanf:"bin"`
|
||||
}
|
||||
|
||||
// Sim launches llmgw's own simulated engine.
|
||||
//
|
||||
// It is in the production registry on purpose. A cluster can be brought up and
|
||||
// exercised end to end — placement, fabrics, multi-node sequencing, failure and
|
||||
// rebuild — on machines that have no GPUs, which is the only way most of that
|
||||
// path gets tested at all. The multi-node launch sequence went four phases
|
||||
// without ever running, because the only things exercising it were stub
|
||||
// processes that exercised the command line and nothing else.
|
||||
//
|
||||
// Nothing about it pretends to be real: the model name it serves is whatever it
|
||||
// is told, the tokens are invented, and it says so in its own logs.
|
||||
type Sim struct{ cfg SimConfig }
|
||||
|
||||
// Name identifies the engine in the catalog.
|
||||
func (*Sim) Name() string { return "sim" }
|
||||
|
||||
// HealthPath is the simulated engine's readiness endpoint.
|
||||
func (*Sim) HealthPath() string { return "/health" }
|
||||
|
||||
// MetricsPath is its Prometheus endpoint, in vLLM's vocabulary.
|
||||
func (*Sim) MetricsPath() string { return "/metrics" }
|
||||
|
||||
// ServesHTTP reports whether this rank answers the API.
|
||||
//
|
||||
// Only the leader, exactly as vLLM and SGLang behave — which is the behaviour
|
||||
// this engine exists to let us test.
|
||||
func (*Sim) ServesHTTP(s ShardSpec) bool { return s.Leader() }
|
||||
|
||||
// Validate accepts anything the arithmetic allows.
|
||||
//
|
||||
// No compute-capability checks and no quantization rules: there is no hardware
|
||||
// to be wrong about, and inventing refusals would make this a worse stand-in
|
||||
// for the engines it stands in for.
|
||||
func (*Sim) Validate(p Profile, gpus []GPUInfo) error {
|
||||
return validateCommon(p, gpus)
|
||||
}
|
||||
|
||||
// Commands returns the single process that makes up one shard.
|
||||
func (s *Sim) Commands(spec ShardSpec) ([]Command, error) {
|
||||
bin := s.cfg.Bin
|
||||
if bin == "" {
|
||||
exe, err := os.Executable()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("locating the llmgw binary for the sim engine: %w", err)
|
||||
}
|
||||
bin = exe
|
||||
}
|
||||
p := spec.Profile
|
||||
|
||||
args := []string{"sim-engine",
|
||||
"--listen", "0.0.0.0:" + strconv.Itoa(spec.HTTPPort),
|
||||
"--model", p.ServedModelName,
|
||||
}
|
||||
if spec.NNodes > 1 {
|
||||
args = append(args,
|
||||
"--nnodes", strconv.Itoa(spec.NNodes),
|
||||
"--node-rank", strconv.Itoa(spec.Rank),
|
||||
"--dist-init-addr", spec.LeaderAddr+":"+strconv.Itoa(spec.LeaderPort),
|
||||
)
|
||||
}
|
||||
args = append(args, p.Args...)
|
||||
|
||||
return []Command{{
|
||||
Name: "sim-engine", Path: bin, Args: args,
|
||||
Env: materializeEnv(baseEnv(spec)),
|
||||
// Workers dial the leader, so the leader has to be listening first.
|
||||
// Both ranks express the same wait because both know the same address.
|
||||
WaitForLeaderPort: spec.NNodes > 1 && !spec.Leader(),
|
||||
}}, nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package engines
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func simSpec(rank, nnodes int) ShardSpec {
|
||||
return ShardSpec{
|
||||
ReplicaID: "rep-1", ShardID: "sh-1", Rank: rank, NNodes: nnodes,
|
||||
LeaderAddr: "10.0.0.5", LeaderPort: 20001,
|
||||
GPUIndices: []int{0, 1}, HTTPPort: 20000, MgmtInterface: "eth0",
|
||||
HCAs: []string{"mlx5_0", "mlx5_1"},
|
||||
Profile: Profile{
|
||||
Name: "tp4", Engine: "sim", TP: 4, GPUsPerNode: 2, Nodes: nnodes,
|
||||
ModelPath: "/models/x", ServedModelName: "big-model",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestOnlyTheLeaderServes(t *testing.T) {
|
||||
// The behaviour that made replicas fail while they were coming up perfectly
|
||||
// well: health checking a worker's port, which never opens.
|
||||
if !(&Sim{}).ServesHTTP(simSpec(0, 2)) {
|
||||
t.Error("the leader does not serve")
|
||||
}
|
||||
if (&Sim{}).ServesHTTP(simSpec(1, 2)) {
|
||||
t.Error("a worker claims to serve")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMultiNodeShardCarriesItsRendezvous(t *testing.T) {
|
||||
cmds, err := (&Sim{cfg: SimConfig{Bin: "/usr/local/bin/llmgw"}}).Commands(simSpec(1, 2))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(cmds) != 1 {
|
||||
t.Fatalf("%d commands", len(cmds))
|
||||
}
|
||||
c := cmds[0]
|
||||
|
||||
if c.Args[0] != "sim-engine" {
|
||||
t.Errorf("the engine is a subcommand of llmgw: %v", c.Args)
|
||||
}
|
||||
for flag, want := range map[string]string{
|
||||
"--nnodes": "2", "--node-rank": "1", "--dist-init-addr": "10.0.0.5:20001",
|
||||
} {
|
||||
if got, _ := flagValue(c.Args, flag); got != want {
|
||||
t.Errorf("%s = %q, want %q", flag, got, want)
|
||||
}
|
||||
}
|
||||
// A worker must wait for the leader's port, or the launch is a race.
|
||||
if !c.WaitForLeaderPort {
|
||||
t.Error("a worker does not wait for the leader")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheLeaderDoesNotWaitForItself(t *testing.T) {
|
||||
cmds, _ := (&Sim{cfg: SimConfig{Bin: "x"}}).Commands(simSpec(0, 2))
|
||||
if cmds[0].WaitForLeaderPort {
|
||||
t.Error("the leader waits for a port it is about to open")
|
||||
}
|
||||
}
|
||||
|
||||
func TestASingleNodeShardCarriesNoRendezvous(t *testing.T) {
|
||||
cmds, _ := (&Sim{cfg: SimConfig{Bin: "x"}}).Commands(simSpec(0, 1))
|
||||
for _, flag := range []string{"--nnodes", "--node-rank", "--dist-init-addr"} {
|
||||
if slices.Contains(cmds[0].Args, flag) {
|
||||
t.Errorf("a single-node shard was given %s", flag)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheShardIsStillPinnedToItsFabric(t *testing.T) {
|
||||
// The simulated engine does not use NCCL, but the environment is generated
|
||||
// by the same code that generates it for the ones that do — so a cluster
|
||||
// exercised with this engine is exercising that too.
|
||||
cmds, _ := (&Sim{cfg: SimConfig{Bin: "x"}}).Commands(simSpec(1, 2))
|
||||
env := cmds[0].Env
|
||||
for _, want := range []string{
|
||||
"NCCL_IB_HCA=mlx5_0,mlx5_1", "NCCL_SOCKET_IFNAME=eth0",
|
||||
"CUDA_VISIBLE_DEVICES=0,1", "NCCL_IB_DISABLE=0",
|
||||
} {
|
||||
if !slices.Contains(env, want) {
|
||||
t.Errorf("the environment is missing %s", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -17,15 +17,39 @@ import (
|
||||
func newSimEngineCmd() *cobra.Command {
|
||||
cfg := sim.Default()
|
||||
var listen string
|
||||
var rv sim.Rendezvous
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "sim-engine",
|
||||
Short: "OpenAI-compatible fake inference engine (for testing)",
|
||||
Long: "Simulates the engine behaviours that matter to the router:\n" +
|
||||
"continuous batching, KV block accounting, prefix caching, and vLLM-shaped\n" +
|
||||
"/metrics. Lets the router be developed and tested without GPUs.",
|
||||
"/metrics. Lets the router be developed and tested without GPUs.\n\n" +
|
||||
"With --nnodes above 1 it also simulates a multi-node replica: rank 0\n" +
|
||||
"listens at --dist-init-addr and does not serve until every other rank has\n" +
|
||||
"joined, and only rank 0 answers the API. That is the part of the\n" +
|
||||
"multi-node path stub processes cannot exercise — the sequencing, the\n" +
|
||||
"waiting, and one dead shard taking the replica with it.",
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
log := logger()
|
||||
|
||||
ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
// Before anything binds the API port: a worker must never look
|
||||
// ready, and a leader must not serve until the replica exists.
|
||||
if err := rv.Join(ctx, log); err != nil {
|
||||
return err
|
||||
}
|
||||
if rv.Multi() && !rv.Leader() {
|
||||
// A worker contributes GPUs and answers nothing. Health
|
||||
// checking its port is what used to fail replicas that were
|
||||
// coming up perfectly well.
|
||||
log.Info("worker rank running; this rank serves no API", "rank", rv.Rank)
|
||||
<-ctx.Done()
|
||||
return nil
|
||||
}
|
||||
|
||||
eng := sim.New(cfg)
|
||||
defer eng.Close()
|
||||
|
||||
@@ -47,9 +71,6 @@ func newSimEngineCmd() *cobra.Command {
|
||||
"prefix_cache_blocks", cfg.PrefixCacheBlocks,
|
||||
)
|
||||
|
||||
ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- srv.Serve(ln) }()
|
||||
|
||||
@@ -70,6 +91,11 @@ func newSimEngineCmd() *cobra.Command {
|
||||
|
||||
f := cmd.Flags()
|
||||
f.StringVar(&listen, "listen", ":8000", "address to listen on")
|
||||
f.IntVar(&rv.NNodes, "nnodes", 1, "ranks in this replica; above 1 simulates a multi-node launch")
|
||||
f.IntVar(&rv.Rank, "node-rank", 0, "this rank; 0 is the leader and the only one that serves")
|
||||
f.StringVar(&rv.Addr, "dist-init-addr", "", "the leader's rendezvous address")
|
||||
f.DurationVar(&rv.Timeout, "rendezvous-timeout", 5*time.Minute,
|
||||
"how long to wait for the replica to assemble before failing")
|
||||
f.StringVar(&cfg.Model, "model", cfg.Model, "model name to serve")
|
||||
f.IntVar(&cfg.MaxNumSeqs, "max-num-seqs", cfg.MaxNumSeqs, "concurrent sequences (continuous batch capacity)")
|
||||
f.IntVar(&cfg.BlockSize, "block-size", cfg.BlockSize, "tokens per KV cache block")
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestTheDocumentedAgentExampleLoads parses the file shipped in deploy/examples.
|
||||
//
|
||||
// It exists because the example was wrong for the entire life of the project
|
||||
// and nothing noticed. The fabric struct carried no koanf tags, so the `name:`
|
||||
// key the example documents bound to nothing, and an agent configured exactly
|
||||
// as instructed refused to start with "fabrics[0]: name is required" — naming
|
||||
// the key the operator had just written.
|
||||
//
|
||||
// Every fabric until then had been configured in a Go test, where the tags do
|
||||
// not matter. A published example that has never been parsed is a guess.
|
||||
func TestTheDocumentedAgentExampleLoads(t *testing.T) {
|
||||
path := filepath.Join("..", "..", "deploy", "examples", "agent.yaml")
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
t.Skipf("no example to check: %v", err)
|
||||
}
|
||||
|
||||
cfg, err := LoadAgent(path)
|
||||
if err != nil {
|
||||
t.Fatalf("the documented example does not load: %v", err)
|
||||
}
|
||||
if len(cfg.Fabrics) == 0 {
|
||||
t.Fatal("the example documents a fabric and none was parsed")
|
||||
}
|
||||
f := cfg.Fabrics[0]
|
||||
if f.Fabric == "" {
|
||||
t.Error("the fabric name did not bind")
|
||||
}
|
||||
if len(f.HCAs) == 0 {
|
||||
t.Error("the fabric's adapters did not bind")
|
||||
}
|
||||
if f.RateGbps == 0 {
|
||||
t.Error("the fabric's rate did not bind")
|
||||
}
|
||||
if cfg.NodeID == "" || cfg.Server == "" {
|
||||
t.Errorf("node_id=%q server=%q", cfg.NodeID, cfg.Server)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheDocumentedServerExampleLoads(t *testing.T) {
|
||||
path := filepath.Join("..", "..", "deploy", "examples", "server.yaml")
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
t.Skipf("no example to check: %v", err)
|
||||
}
|
||||
cfg, err := Load(path)
|
||||
if err != nil {
|
||||
t.Fatalf("the documented example does not load: %v", err)
|
||||
}
|
||||
if err := cfg.Validate(); err != nil {
|
||||
t.Fatalf("the documented example does not validate: %v", err)
|
||||
}
|
||||
if cfg.Listen == "" || len(cfg.Models) == 0 {
|
||||
t.Errorf("listen=%q models=%d", cfg.Listen, len(cfg.Models))
|
||||
}
|
||||
}
|
||||
@@ -50,16 +50,22 @@ func (s NodeState) String() string {
|
||||
func (s NodeState) Schedulable() bool { return s == Ready }
|
||||
|
||||
// FabricAttachment is a node's connection to one fabric.
|
||||
// The koanf tags are not decoration. Without them the fields bind from their Go
|
||||
// names, so the documented `name:` key silently produced an empty fabric and
|
||||
// `rate_gbps` was never read at all — and the agent then refused to start with
|
||||
// "fabrics[0]: name is required", naming the key the operator had just written.
|
||||
// Every fabric in this project was configured in a test until the first one was
|
||||
// configured from a file.
|
||||
type FabricAttachment struct {
|
||||
// Fabric is the operator-assigned name. It is not derived from the GID
|
||||
// prefix: OpenSM's default is fe80:: everywhere, so a matching prefix
|
||||
// proves nothing (plan §5.2).
|
||||
Fabric string `json:"fabric"`
|
||||
Fabric string `json:"fabric" koanf:"name"`
|
||||
// HCAs are the adapters attached to this fabric.
|
||||
HCAs []string `json:"hcas"`
|
||||
HCAs []string `json:"hcas" koanf:"hcas"`
|
||||
// RateGbps is the slowest active port's rate, since a collective runs at
|
||||
// the speed of its slowest link.
|
||||
RateGbps int `json:"rate_gbps"`
|
||||
RateGbps int `json:"rate_gbps" koanf:"rate_gbps"`
|
||||
}
|
||||
|
||||
// Node is one machine in the cluster.
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
package sim
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Rendezvous makes a sim-engine behave like a multi-node one.
|
||||
//
|
||||
// It exists because the multi-node launch path — leader first, workers joining,
|
||||
// only rank 0 serving, the whole replica failing when one shard dies — can only
|
||||
// be tested against something that actually rendezvouses. Stub processes
|
||||
// exercise the command construction and nothing else, which is how that path
|
||||
// stayed unverified through four phases.
|
||||
//
|
||||
// The protocol is the smallest thing that has the same shape as vLLM's Ray head
|
||||
// or SGLang's dist-init: rank 0 listens, every other rank dials, and rank 0
|
||||
// does not serve until all of them have.
|
||||
type Rendezvous struct {
|
||||
// Addr is the leader's dist-init endpoint. Empty means single node.
|
||||
Addr string
|
||||
// Rank 0 is the leader. NNodes counts every rank including it.
|
||||
Rank int
|
||||
NNodes int
|
||||
// Timeout bounds the wait. A rendezvous that never completes must fail the
|
||||
// shard rather than hang it, which is the difference between a replica that
|
||||
// gets rebuilt and one that occupies its GPUs forever.
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// Multi reports whether this shard is part of a multi-node replica.
|
||||
func (r Rendezvous) Multi() bool { return r.NNodes > 1 && r.Addr != "" }
|
||||
|
||||
// Leader reports whether this rank serves the API.
|
||||
func (r Rendezvous) Leader() bool { return r.Rank == 0 }
|
||||
|
||||
// Join blocks until the replica is assembled.
|
||||
//
|
||||
// The leader waits for every worker to dial in; a worker dials the leader and
|
||||
// holds the connection open, so the leader learns immediately when one dies.
|
||||
func (r Rendezvous) Join(ctx context.Context, log *slog.Logger) error {
|
||||
if !r.Multi() {
|
||||
return nil
|
||||
}
|
||||
if r.Timeout <= 0 {
|
||||
r.Timeout = 5 * time.Minute
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, r.Timeout)
|
||||
defer cancel()
|
||||
|
||||
if r.Leader() {
|
||||
return r.host(ctx, log)
|
||||
}
|
||||
return r.dial(ctx, log)
|
||||
}
|
||||
|
||||
func (r Rendezvous) host(ctx context.Context, log *slog.Logger) error {
|
||||
ln, err := net.Listen("tcp", r.Addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("binding the rendezvous address %s: %w", r.Addr, err)
|
||||
}
|
||||
// Deliberately not closed: holding it open is how a worker that restarts
|
||||
// can reconnect, and how the leader notices one going away.
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
_ = ln.Close()
|
||||
}()
|
||||
|
||||
want := r.NNodes - 1
|
||||
log.Info("waiting for workers", "rank", r.Rank, "want", want, "addr", r.Addr)
|
||||
|
||||
joined := 0
|
||||
for joined < want {
|
||||
conn, err := ln.Accept()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return fmt.Errorf("only %d of %d workers joined before the deadline", joined, want)
|
||||
}
|
||||
return err
|
||||
}
|
||||
joined++
|
||||
log.Info("a worker joined", "joined", joined, "want", want, "from", conn.RemoteAddr())
|
||||
// Held open rather than closed: the connection is the worker's
|
||||
// liveness, exactly as a real distributed runtime would treat it.
|
||||
go func() { <-ctx.Done(); _ = conn.Close() }()
|
||||
}
|
||||
log.Info("every worker joined; serving", "nnodes", r.NNodes)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r Rendezvous) dial(ctx context.Context, log *slog.Logger) error {
|
||||
log.Info("joining the leader", "rank", r.Rank, "addr", r.Addr)
|
||||
|
||||
// Retried, because the leader is started first but "started" and "listening"
|
||||
// are not the same instant, and a worker that gives up on the first refusal
|
||||
// would make the whole replica a race.
|
||||
var lastErr error
|
||||
for ctx.Err() == nil {
|
||||
conn, err := (&net.Dialer{Timeout: 5 * time.Second}).DialContext(ctx, "tcp", r.Addr)
|
||||
if err == nil {
|
||||
log.Info("joined", "rank", r.Rank)
|
||||
go func() { <-ctx.Done(); _ = conn.Close() }()
|
||||
return nil
|
||||
}
|
||||
lastErr = err
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(250 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("could not reach the leader at %s: %w", r.Addr, errors.Join(lastErr, ctx.Err()))
|
||||
}
|
||||
Reference in new issue
Block a user