CI / test (push) Successful in 5m10s
The setup screen asks for up to two judge machines after the store's two, and server init takes --judge-node for each. Each is added as a node labelled llmgw.role: judge, and the judge's profile selects that label, with cores by init's rule over those machines. A judge beside a busy control plane refused 38 of 60 requests under load; alone, 2. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
991 lines
34 KiB
Go
991 lines
34 KiB
Go
package cli
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/base32"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"os/user"
|
|
"path/filepath"
|
|
"runtime"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/account"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/pgha"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
|
|
)
|
|
|
|
// serverUnit is deploy/systemd/llmgw-server.service. A test holds the two
|
|
// together, because a machine set up by init and one set up from the docs
|
|
// should not differ in a way nobody chose.
|
|
const serverUnit = `[Unit]
|
|
Description=llmgw control plane
|
|
After=network-online.target
|
|
Wants=network-online.target
|
|
|
|
[Service]
|
|
ExecStart=/usr/local/bin/llmgw server --config /etc/llmgw/server.yaml
|
|
User=llmgw
|
|
Group=llmgw
|
|
Restart=on-failure
|
|
RestartSec=2s
|
|
NoNewPrivileges=true
|
|
ProtectSystem=strict
|
|
ProtectHome=true
|
|
PrivateTmp=true
|
|
StateDirectory=llmgw
|
|
|
|
[Install]
|
|
WantedBy=multi-user.target
|
|
`
|
|
|
|
// serverInit is everything a control plane needs before it can start, done in
|
|
// the order it has to be done in. Every step looks before it acts, so running
|
|
// it again finishes an install that stopped halfway and changes nothing on
|
|
// one that did not.
|
|
type serverInit struct {
|
|
etcDir string // /etc/llmgw
|
|
stateDir string // /var/lib/llmgw
|
|
pkiDir string // where the authority lives
|
|
unitPath string // "" leaves systemd alone
|
|
|
|
// user owns the state and reads the key; "" means no user is created and
|
|
// nothing is chowned, which is what a test wants.
|
|
user string
|
|
hosts []string
|
|
admin string
|
|
// postgres keeps the store in a PostgreSQL on this machine, installed if
|
|
// need be, moving whatever SQLite already holds.
|
|
postgres bool
|
|
// patroni are user@host of the machines to keep the store on under
|
|
// Patroni, replicated; this machine is the third etcd vote for two.
|
|
patroni []string
|
|
// webhooks are where the cluster's alerts go: put into its settings
|
|
// once it runs, and each sent a test.
|
|
webhooks []string
|
|
// patroniNodes are the same machines already connected to, which the
|
|
// setup screen does with the password it was given; when set, patroni
|
|
// only names them.
|
|
patroniNodes []pgha.Node
|
|
// judgeMachines are machines for the guardrails' judges alone, added as
|
|
// nodes labelled for them; judgeCPUs their cores, once added.
|
|
judgeMachines []judgeMachine
|
|
judgeCPUs []int
|
|
// noMirror, noNode and noModel turn off the parts of a first install that
|
|
// make it serve something: a mirror, this machine as a node, and a small
|
|
// model on it.
|
|
noMirror, noNode, noModel bool
|
|
// noGuardModel leaves out the guardrails' judge model; the pattern
|
|
// checks in the baseline policy stay.
|
|
noGuardModel bool
|
|
// noFirewall leaves this machine's firewall, and the store members',
|
|
// as they are.
|
|
noFirewall bool
|
|
// mirrorURL is a mirror on another machine to use instead of running one.
|
|
mirrorURL string
|
|
// bundle is an offline bundle to install from, and bundleSHA its
|
|
// manifest's sha256: the cluster is then set up offline.
|
|
bundle, bundleSHA string
|
|
// mirrorPath is the mirror's directory here; "" is /srv/models.
|
|
mirrorPath string
|
|
|
|
// What setup lets a person choose besides the parts: zero is the
|
|
// default (serverinit_choices.go).
|
|
starter starterSpec
|
|
starterCtx int
|
|
judge judgeSpec
|
|
// language is the portal's for a browser that asks for neither of its
|
|
// own; "" leaves it English.
|
|
language string
|
|
signupOpen bool
|
|
defaultRPM, defaultTPM int
|
|
maxKeyTTL time.Duration
|
|
// media has the nodes install ComfyUI and the leader add a picture model
|
|
// and a 3D model once a card fits one.
|
|
media bool
|
|
// evictAfterIdle and minLifetime are on_demand's; zero leaves its own.
|
|
evictAfterIdle, minLifetime time.Duration
|
|
// standbys makes each machine of the replicated store a control plane
|
|
// too, standing by while this one leads: three machines, each able to
|
|
// take over. vip is a floating address, as 10.0.0.160/24, kept on
|
|
// whichever leads.
|
|
standbys bool
|
|
vip string
|
|
// cpus is this machine's CPU count for dividing it between init's
|
|
// models; zero asks the machine. A seam for tests.
|
|
cpus int
|
|
// pgSetup prepares PostgreSQL and returns the store address; a seam for
|
|
// tests. Nil does it for real.
|
|
pgSetup func(context.Context) (string, error)
|
|
pg pgServer
|
|
|
|
out io.Writer
|
|
|
|
uid, gid int
|
|
changed bool
|
|
}
|
|
|
|
// newDefaultServerInit is init as it sets up a real machine.
|
|
func newDefaultServerInit() serverInit {
|
|
return serverInit{
|
|
etcDir: "/etc/llmgw",
|
|
stateDir: "/var/lib/llmgw",
|
|
pkiDir: defaultPKIDir,
|
|
unitPath: "/etc/systemd/system/llmgw-server.service",
|
|
user: "llmgw",
|
|
}
|
|
}
|
|
|
|
func newServerInitCmd() *cobra.Command {
|
|
in := newDefaultServerInit()
|
|
var (
|
|
plain bool
|
|
starter, judgeCats string
|
|
)
|
|
cmd := &cobra.Command{
|
|
Use: "init",
|
|
Short: "Set this machine up as a control plane and start it",
|
|
Long: "Creates the llmgw user and its directories, a certificate authority, the\n" +
|
|
"control plane's certificate for every address this machine has, a\n" +
|
|
"configuration, an administrator account, and the systemd unit — then\n" +
|
|
"starts it and waits until it answers.\n\n" +
|
|
"Then it makes the machine serve something: a model mirror beside the\n" +
|
|
"control plane, this machine as a node of it (on its GPUs when nvidia-smi\n" +
|
|
"finds any, on CPU otherwise), and " + starterName + " fetched into the\n" +
|
|
"mirror and served on this node. --no-mirror, --no-node and --no-model\n" +
|
|
"leave those out.\n\n" +
|
|
"With --postgres the store is a PostgreSQL database on this machine:\n" +
|
|
"installed if there is none, with its own role and a generated password\n" +
|
|
"kept in /etc/llmgw/secrets. An existing SQLite store is copied into it,\n" +
|
|
"so accounts and keys survive, and server.yaml is pointed at it.\n\n" +
|
|
"With --patroni, given two or more machines, the store is PostgreSQL\n" +
|
|
"replicated across them under Patroni, which promotes a standby when the\n" +
|
|
"primary is lost. etcd holds its leader lock on those machines — and on\n" +
|
|
"this one, when they are two, as the third vote. Installed over ssh with\n" +
|
|
"passwordless sudo, as 'llmgw node add' does; the store already in use,\n" +
|
|
"SQLite or PostgreSQL, is moved in. Run again, it changes nothing.\n\n" +
|
|
"Each step is skipped when its result already exists, so it is safe to run\n" +
|
|
"again: to finish an install that stopped, or with another --host to\n" +
|
|
"reissue the certificate. It never overwrites server.yaml or the authority.\n\n" +
|
|
"The authority's key stays in " + defaultPKIDir + " on this machine. Move it\n" +
|
|
"somewhere safer once the nodes are onboarded if this machine is exposed.",
|
|
Args: cobra.NoArgs,
|
|
RunE: func(cmd *cobra.Command, _ []string) error {
|
|
if runtime.GOOS != "linux" {
|
|
return fmt.Errorf("init sets up a Linux machine; this is %s", runtime.GOOS)
|
|
}
|
|
if os.Geteuid() != 0 {
|
|
return errors.New("init creates a user, writes /etc and starts a service: run it as root")
|
|
}
|
|
if err := in.readChoices(starter, judgeCats); err != nil {
|
|
return err
|
|
}
|
|
in.out = cmd.OutOrStdout()
|
|
if !plain && isTerminal(in.out) {
|
|
return in.runInitTUI(cmd.Context(), in.out)
|
|
}
|
|
return in.run(cmd.Context())
|
|
},
|
|
}
|
|
f := cmd.Flags()
|
|
f.BoolVar(&plain, "plain", false, "plain output, even in a terminal")
|
|
f.StringArrayVar(&in.hosts, "host", nil,
|
|
"an address agents dial besides the ones found on this machine; repeatable")
|
|
f.StringVar(&in.admin, "admin", "", "email of the first administrator; a password is generated and printed once")
|
|
f.StringVar(&in.pkiDir, "pki", in.pkiDir, "directory holding the certificate authority")
|
|
f.BoolVar(&in.noMirror, "no-mirror", false, "do not run a model mirror on this machine")
|
|
f.StringVar(&in.mirrorURL, "mirror", "",
|
|
"use the mirror on another machine, set up there by 'llmgw mirror init': its management API, http://host:8082")
|
|
f.StringVar(&in.bundle, "bundle", "",
|
|
"install offline from this bundle ('llmgw bundle create'): into the mirror, with the cluster set to reach nothing outside")
|
|
f.StringVar(&in.bundleSHA, "bundle-sha256", "", "the bundle's sha256, as 'llmgw bundle create' printed it")
|
|
f.BoolVar(&in.noNode, "no-node", false, "do not make this machine a node (and serve no model on it)")
|
|
f.BoolVar(&in.noModel, "no-model", false, "do not fetch and serve the starter model ("+starterName+")")
|
|
f.BoolVar(&in.noGuardModel, "no-guard-model", false,
|
|
"do not fetch and serve the guardrails' judge model ("+guardModelName+"); the pattern checks stay")
|
|
f.BoolVar(&in.noFirewall, "no-firewall", false,
|
|
"leave the firewall alone, instead of keeping llmgw's internal ports to the cluster's networks")
|
|
f.BoolVar(&in.postgres, "postgres", false,
|
|
"keep the store in PostgreSQL on this machine, installing it and moving an existing SQLite store into it")
|
|
f.StringArrayVar(&in.webhooks, "webhook", nil,
|
|
"where alerts go — a model parked, a node lost, a budget spent: an http(s) address, Slack's, Mattermost's "+
|
|
"and Rocket.Chat's sent Slack's text, any other the event as JSON (slack: or json: in front says which); repeatable")
|
|
f.StringArrayVar(&in.patroni, "patroni", nil,
|
|
"user@host of a machine to keep the store on, replicated under Patroni; give two or more. "+
|
|
"Installs PostgreSQL, Patroni and etcd there over ssh, and moves the existing store in")
|
|
|
|
// What server.yaml is written with; each is in it only when init writes
|
|
// it, never into one already there.
|
|
f.StringVar(&in.mirrorPath, "mirror-dir", mirrorDir, "the directory the mirror on this machine keeps model files in")
|
|
f.StringVar(&starter, "starter", starterName,
|
|
"the starter model: "+strings.Join(starterNames(), ", ")+", or hf:<owner>/<repo>/<file>.gguf")
|
|
f.IntVar(&in.starterCtx, "starter-context", defaultContext, "the context, in tokens, a client may use of the starter model")
|
|
f.StringVar(&in.judge.Fail, "judge-fail", "closed",
|
|
"a request no judge read in time: closed refuses it, open serves it flagged")
|
|
f.StringVar(&judgeCats, "judge-categories", strings.Join(defaultJudgeCategories, ","),
|
|
"what the judge refuses, comma-separated, of: "+strings.Join(judgeCategories, ", "))
|
|
f.IntVar(&in.judge.Replicas, "judge-replicas", 2, "how many copies of the judge run, each on CPUs")
|
|
f.Var(judgeMachineFlag{&in.judgeMachines}, "judge-node",
|
|
"user@host of a machine for the guardrails' judges alone: added as a node, and the judges run only there; repeatable")
|
|
f.DurationVar(&in.judge.Budget, "judge-budget", 5*time.Second, "how long a request waits for the judge")
|
|
f.StringVar(&in.language, "language", "", "the portal's language for a browser that asks for neither: en or tr")
|
|
f.BoolVar(&in.signupOpen, "signup-open", false, "let anyone ask for an account on the portal, to be approved by an administrator")
|
|
f.IntVar(&in.defaultRPM, "default-rpm", defaultRPM, "requests a minute a key may make when nothing gives it a limit")
|
|
f.IntVar(&in.defaultTPM, "default-tpm", defaultTPM, "tokens a minute a key may use when nothing gives it a limit")
|
|
f.DurationVar(&in.maxKeyTTL, "max-key-ttl", defaultMaxKeyTTL, "the longest an API key lives")
|
|
f.BoolVar(&in.media, "media", false,
|
|
"pictures and 3D: the nodes install ComfyUI, and a picture model and a 3D model are added once a card fits one")
|
|
f.DurationVar(&in.evictAfterIdle, "evict-after-idle", 0,
|
|
"how long a model loaded on demand goes unasked-for before another may have its card (0: 2m)")
|
|
f.DurationVar(&in.minLifetime, "min-lifetime", 0,
|
|
"how long a model just loaded on demand is safe from that (0: 5m)")
|
|
f.BoolVar(&in.standbys, "standbys", false,
|
|
"with --patroni, make each of those machines a control plane too, standing by while this one leads")
|
|
f.StringVar(&in.vip, "vip", "",
|
|
"with --standbys, a floating address for whichever control plane leads, as 10.0.0.160/24 (keepalived on each)")
|
|
return cmd
|
|
}
|
|
|
|
func (in *serverInit) run(ctx context.Context) error {
|
|
if err := in.checkBundle(); err != nil {
|
|
return fmt.Errorf("offline bundle: %w", err)
|
|
}
|
|
steps := []struct {
|
|
name string
|
|
fn func(context.Context) error
|
|
}{
|
|
{"user", in.ensureUser},
|
|
{"directories", in.ensureDirs},
|
|
{"certificates", in.ensureCerts},
|
|
{"configuration", in.ensureConfig},
|
|
{"database", in.ensureDatabase},
|
|
{"store", in.ensureStore},
|
|
}
|
|
for _, s := range steps {
|
|
if err := s.fn(ctx); err != nil {
|
|
return fmt.Errorf("%s: %w", s.name, err)
|
|
}
|
|
}
|
|
if in.unitPath == "" {
|
|
return nil
|
|
}
|
|
hasSystemd := true
|
|
if _, err := exec.LookPath("systemctl"); err != nil {
|
|
hasSystemd = false
|
|
}
|
|
// Before the service starts, because the mirror goes into server.yaml.
|
|
switch {
|
|
case in.mirrorURL != "":
|
|
if err := in.useMirror(ctx); err != nil {
|
|
return fmt.Errorf("mirror: %w", err)
|
|
}
|
|
case hasSystemd && !in.noMirror:
|
|
if err := in.ensureMirror(ctx); err != nil {
|
|
return fmt.Errorf("mirror: %w", err)
|
|
}
|
|
}
|
|
// Into the mirror, and server.yaml told, before the service starts:
|
|
// the control plane reads at start that the cluster is offline.
|
|
if in.bundle != "" {
|
|
if err := in.ensureBundle(ctx); err != nil {
|
|
return fmt.Errorf("offline bundle: %w", err)
|
|
}
|
|
}
|
|
if err := in.ensureService(ctx); err != nil {
|
|
return fmt.Errorf("service: %w", err)
|
|
}
|
|
if hasSystemd && !in.noNode {
|
|
if _, err := in.ensureNode(ctx); err != nil {
|
|
return fmt.Errorf("node: %w", err)
|
|
}
|
|
}
|
|
// The protections first: a starter model that fits nowhere must not
|
|
// leave the machine without its firewall or its judge.
|
|
if hasSystemd && !in.noFirewall {
|
|
if err := in.ensureFirewall(ctx); err != nil {
|
|
// The cluster works without it; said, not fatal.
|
|
fmt.Fprintf(in.out, " !! firewall: %v; llmgw's ports answer every network this machine is on\n", err)
|
|
}
|
|
}
|
|
if err := in.ensureWebhooks(ctx); err != nil {
|
|
fmt.Fprintf(in.out, " !! webhooks: %v\n", err)
|
|
}
|
|
if hasSystemd && len(in.judgeMachines) > 0 {
|
|
if err := in.ensureJudgeNodes(ctx); err != nil {
|
|
return fmt.Errorf("judge nodes: %w", err)
|
|
}
|
|
}
|
|
if in.judgeWanted() && in.configJudgesWith(guardModelName) {
|
|
// Not fatal: without its judge the baseline policy still checks for
|
|
// secrets and injection, and a judged request is served flagged.
|
|
if err := in.ensureGuardModel(ctx); err != nil {
|
|
fmt.Fprintf(in.out, " !! guard model: %v; requests are not judged until it runs\n", err)
|
|
}
|
|
}
|
|
switch {
|
|
case !hasSystemd || in.noModel:
|
|
case in.noMirror && in.mirrorURL == "":
|
|
fmt.Fprintln(in.out, " skip no starter model without the mirror")
|
|
default:
|
|
if err := in.ensureStarter(ctx); err != nil {
|
|
return fmt.Errorf("starter model: %w", err)
|
|
}
|
|
}
|
|
if in.standbys {
|
|
if err := in.ensureStandbys(ctx); err != nil {
|
|
return fmt.Errorf("standby control planes: %w", err)
|
|
}
|
|
}
|
|
in.nextSteps()
|
|
return nil
|
|
}
|
|
|
|
// judgeWanted is whether this install fetches the judge model, and so
|
|
// whether the baseline policy asks it.
|
|
func (in *serverInit) judgeWanted() bool {
|
|
if in.noGuardModel || (in.noMirror && in.mirrorURL == "") || in.unitPath == "" {
|
|
return false
|
|
}
|
|
// Offline, only a judge the bundle carries: a policy asking a judge
|
|
// that never comes refuses every request.
|
|
if in.bundle != "" && !in.bundleCarries(guardModelName) {
|
|
return false
|
|
}
|
|
_, err := exec.LookPath("systemctl")
|
|
return err == nil
|
|
}
|
|
|
|
// configJudgesWith is whether server.yaml has a policy asking that judge: a
|
|
// configuration written before init installed one is never changed, so the
|
|
// model is not fetched for nothing.
|
|
func (in *serverInit) configJudgesWith(model string) bool {
|
|
cfg, err := loadEffectiveConfig(in.configPath())
|
|
if err != nil {
|
|
return false
|
|
}
|
|
for _, p := range cfg.Guard.Policies {
|
|
if slices.Contains(p.Judge.Judges(), model) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// guardSection is server.yaml's guard: the baseline every request passes,
|
|
// with the judge model, as init sets it up unless told otherwise, when there
|
|
// is one.
|
|
func guardSection(judge bool) string {
|
|
if !judge {
|
|
return guardSectionFor(nil)
|
|
}
|
|
j := (&serverInit{}).judgeFor()
|
|
return guardSectionFor(&j)
|
|
}
|
|
|
|
// guardSectionFor is server.yaml's guard with this judge, or none.
|
|
func guardSectionFor(j *judgeSpec) string {
|
|
fail := "closed"
|
|
jb := ` # A judge model reads what patterns cannot: violence, weapons, abuse,
|
|
# self-harm. Installed by init; --no-guard-model leaves it out.
|
|
# judge: {model: ` + guardModelName + `, input: block, output: block, ...}
|
|
`
|
|
if j != nil {
|
|
fail = j.Fail
|
|
copies := map[int]string{1: "One replica runs", 2: "Two replicas run", 3: "Three replicas run"}[j.Replicas]
|
|
if copies == "" {
|
|
copies = fmt.Sprintf("%d replicas run", j.Replicas)
|
|
}
|
|
jb = ` # A judge model reads what patterns cannot: violence, crime, sexual
|
|
# content, self-harm, jailbreaks. ` + copies + ` on CPUs, about two
|
|
# seconds for a short text; an answer is read every 600 characters as
|
|
# it streams. With fail: closed, a request no judge can read within
|
|
# latency_budget is refused (503); open serves it, flagged, unjudged.
|
|
# PII is the patterns' (above), and copyright, unethical acts and
|
|
# politically sensitive topics are left out unless chosen: they refuse
|
|
# too much ordinary work. Controversial counts as safe; controversial:
|
|
# unsafe makes it count. With a GPU: llmgw ctl guard judge enable --gpu,
|
|
# then llmgw ctl guard judge use ` + defaultGPUJudge + ` ` + guardModelName + `.
|
|
judge:
|
|
models: [` + guardModelName + `]
|
|
input: block
|
|
output: block
|
|
categories: [` + strings.Join(j.Categories, ", ") + `]
|
|
latency_budget: ` + durationYAML(j.Budget) + `
|
|
`
|
|
}
|
|
return `# Guardrails (docs/guardrails.md). The baseline applies to every request;
|
|
# attach stricter policies to parts of the tree with 'llmgw ctl guard attach'.
|
|
# mode: shadow records what would be done without doing it.
|
|
guard:
|
|
default: [baseline]
|
|
policies:
|
|
baseline:
|
|
fail: ` + fail + `
|
|
input:
|
|
pii: {secret: block, card: redact}
|
|
injection: flag
|
|
output:
|
|
pii: {secret: block, card: redact}
|
|
` + jb
|
|
}
|
|
|
|
func (in *serverInit) done(format string, a ...any) {
|
|
fmt.Fprintf(in.out, " ok "+format+"\n", a...)
|
|
in.changed = true
|
|
}
|
|
|
|
func (in *serverInit) kept(format string, a ...any) {
|
|
fmt.Fprintf(in.out, " kept "+format+"\n", a...)
|
|
}
|
|
|
|
func (in *serverInit) ensureUser(context.Context) error {
|
|
if in.user == "" {
|
|
return nil
|
|
}
|
|
u, err := user.Lookup(in.user)
|
|
if err != nil {
|
|
var unknown user.UnknownUserError
|
|
if !errors.As(err, &unknown) {
|
|
return err
|
|
}
|
|
out, err := exec.Command("useradd", "--system", "--home", in.stateDir,
|
|
"--shell", "/usr/sbin/nologin", in.user).CombinedOutput()
|
|
if err != nil {
|
|
return fmt.Errorf("useradd: %v: %s", err, strings.TrimSpace(string(out)))
|
|
}
|
|
if u, err = user.Lookup(in.user); err != nil {
|
|
return err
|
|
}
|
|
in.done("system user %s", in.user)
|
|
} else {
|
|
in.kept("system user %s", in.user)
|
|
}
|
|
in.uid, _ = strconv.Atoi(u.Uid)
|
|
in.gid, _ = strconv.Atoi(u.Gid)
|
|
return nil
|
|
}
|
|
|
|
func (in *serverInit) ensureDirs(context.Context) error {
|
|
for _, d := range []struct {
|
|
path string
|
|
mode os.FileMode
|
|
owned bool
|
|
}{
|
|
{in.etcDir, 0o755, false},
|
|
{in.pkiDir, 0o700, false},
|
|
{in.stateDir, 0o750, true},
|
|
} {
|
|
if err := os.MkdirAll(d.path, d.mode); err != nil {
|
|
return err
|
|
}
|
|
if d.owned {
|
|
if err := in.chown(d.path, true); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// chown gives a path to the service user, or only its group when the user
|
|
// should read and not write.
|
|
func (in *serverInit) chown(path string, own bool) error {
|
|
if in.user == "" {
|
|
return nil
|
|
}
|
|
uid := 0
|
|
if own {
|
|
uid = in.uid
|
|
}
|
|
return os.Chown(path, uid, in.gid)
|
|
}
|
|
|
|
func (in *serverInit) ensureCerts(context.Context) error {
|
|
certPath := filepath.Join(in.etcDir, "server.pem")
|
|
_, keyErr := os.Stat(filepath.Join(in.pkiDir, "ca-key.pem"))
|
|
_, caErr := os.Stat(filepath.Join(in.etcDir, "ca.pem"))
|
|
_, certErr := os.Stat(certPath)
|
|
if keyErr != nil && caErr == nil && certErr == nil {
|
|
// Issued by an authority kept elsewhere, which is the arrangement for
|
|
// anyone who will not have the CA key on the control plane. Making a
|
|
// new authority here would replace their certificate with one no node
|
|
// trusts.
|
|
if len(in.hosts) > 0 {
|
|
return fmt.Errorf("%s was issued by an authority that is not in %s; "+
|
|
"reissue it there with the new --host", certPath, in.pkiDir)
|
|
}
|
|
in.kept("certificates in %s, issued elsewhere", in.etcDir)
|
|
return nil
|
|
}
|
|
|
|
var (
|
|
ca *tlsconf.CA
|
|
err error
|
|
)
|
|
if keyErr == nil {
|
|
if ca, err = loadCA(in.pkiDir); err != nil {
|
|
return err
|
|
}
|
|
in.kept("certificate authority in %s", in.pkiDir)
|
|
} else {
|
|
if ca, err = tlsconf.NewCA("llmgw cluster CA"); err != nil {
|
|
return err
|
|
}
|
|
if err := ca.Write(in.pkiDir); err != nil {
|
|
return err
|
|
}
|
|
in.done("certificate authority in %s (ca-key.pem signs every node: back it up)", in.pkiDir)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(in.etcDir, "ca.pem"), ca.PEM, 0o644); err != nil {
|
|
return err
|
|
}
|
|
|
|
want := in.certHosts()
|
|
if cert, err := readCert(certPath); err == nil {
|
|
have := strings.Split(sanNames(cert), ", ")
|
|
var missing []string
|
|
for _, h := range want {
|
|
if !slices.Contains(have, h) {
|
|
missing = append(missing, h)
|
|
}
|
|
}
|
|
// A certificate from this authority naming every address is left
|
|
// alone. One naming fewer is reissued with the union, so an address
|
|
// that worked yesterday is never dropped by adding one today.
|
|
// And one that can also present itself as a client, which a
|
|
// standby's admin API needs to reach the leader's; certificates
|
|
// from before that are reissued.
|
|
if len(missing) == 0 && cert.CheckSignatureFrom(ca.Cert) == nil && clientCapable(cert) {
|
|
in.kept("server certificate for %s", strings.Join(have, ", "))
|
|
return nil
|
|
}
|
|
for _, h := range have {
|
|
if h != "" && !slices.Contains(want, h) {
|
|
want = append(want, h)
|
|
}
|
|
}
|
|
}
|
|
|
|
name, _ := os.Hostname()
|
|
if name == "" {
|
|
name = "llmgw-control"
|
|
}
|
|
leaf, err := ca.IssuePeer(name, want)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := leaf.Write(in.etcDir, "server"); err != nil {
|
|
return err
|
|
}
|
|
// The service runs as llmgw and must read its key; nothing else should.
|
|
keyPath := filepath.Join(in.etcDir, "server-key.pem")
|
|
if err := in.chown(keyPath, false); err != nil {
|
|
return err
|
|
}
|
|
if in.user != "" {
|
|
if err := os.Chmod(keyPath, 0o640); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
in.done("server certificate for %s", strings.Join(want, ", "))
|
|
return nil
|
|
}
|
|
|
|
// certHosts is every address an agent could plausibly dial: the hostname and
|
|
// each address on this machine, plus whatever --host added. Asking the
|
|
// operator to list them is how a certificate ends up naming the one address
|
|
// nobody uses, and that is found out by the first node refusing it.
|
|
func (in *serverInit) certHosts() []string {
|
|
var out []string
|
|
add := func(h string) {
|
|
if h != "" && !slices.Contains(out, h) {
|
|
out = append(out, h)
|
|
}
|
|
}
|
|
for _, h := range in.hosts {
|
|
add(h)
|
|
}
|
|
// The floating address is this machine's whenever it leads.
|
|
if ip, _, err := net.ParseCIDR(in.vip); err == nil {
|
|
add(ip.String())
|
|
}
|
|
if name, err := os.Hostname(); err == nil {
|
|
add(name)
|
|
}
|
|
for _, ip := range localAddrs() {
|
|
add(ip.String())
|
|
}
|
|
return out
|
|
}
|
|
|
|
// localAddrs are this machine's addresses another machine could reach:
|
|
// loopback and link-local are left out.
|
|
func localAddrs() []net.IP {
|
|
addrs, err := net.InterfaceAddrs()
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
var out []net.IP
|
|
for _, a := range addrs {
|
|
n, ok := a.(*net.IPNet)
|
|
if !ok || !n.IP.IsGlobalUnicast() {
|
|
continue
|
|
}
|
|
out = append(out, n.IP)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (in *serverInit) configPath() string { return filepath.Join(in.etcDir, "server.yaml") }
|
|
|
|
func (in *serverInit) ensureConfig(context.Context) error {
|
|
path := in.configPath()
|
|
if _, err := os.Stat(path); err == nil {
|
|
in.kept("%s (never overwritten)", path)
|
|
} else {
|
|
contact := in.admin
|
|
if contact == "" {
|
|
contact = "platform@example.com"
|
|
}
|
|
var judge *judgeSpec
|
|
if in.judgeWanted() {
|
|
j := in.judgeFor()
|
|
judge = &j
|
|
}
|
|
body := in.tuneConfig(fmt.Sprintf(serverConfigTemplate, in.etcDir, in.etcDir, in.etcDir,
|
|
in.storeAddress(), contact, guardSectionFor(judge)))
|
|
if err := os.WriteFile(path, []byte(body), 0o640); err != nil {
|
|
return err
|
|
}
|
|
if err := in.chown(path, false); err != nil {
|
|
return err
|
|
}
|
|
in.done("%s", path)
|
|
}
|
|
|
|
// Checked whether written now or years ago: a configuration that will not
|
|
// load is better found here than as a unit restarting every two seconds.
|
|
cfg, err := config.Load(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := cfg.Validate(); err != nil {
|
|
return fmt.Errorf("%s: %w", path, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// serverConfigTemplate is the smallest file that is not a toy: mutual TLS,
|
|
// authentication, default limits and a price, and a control plane with an
|
|
// empty catalog.
|
|
const serverConfigTemplate = `# Written by 'llmgw server init'. Every key: docs/reference.md
|
|
listen: ":8080"
|
|
admin_listen: "127.0.0.1:9090" # serves pprof, which can dump API keys: keep it local
|
|
agent_listen: ":7443"
|
|
|
|
agent_tls:
|
|
ca: %s/ca.pem
|
|
cert: %s/server.pem
|
|
key: %s/server-key.pem
|
|
|
|
# A control plane with nothing in its catalog yet. Models are added while it
|
|
# runs, with 'llmgw ctl model add' or from /admin/mirror.
|
|
control:
|
|
enabled: true
|
|
|
|
auth:
|
|
enabled: true
|
|
store: "%s" # a SQLite path, or @file holding a PostgreSQL address
|
|
# The signup form: accounts it makes wait for an administrator's approval,
|
|
# but anyone reaching port 8080 can make them. Off until you want it.
|
|
signup_open: false
|
|
# Applied to any key no scope gives a limit to. Without them, a project
|
|
# created without limits is metered by nothing at all.
|
|
default_rpm: 60
|
|
default_tpm: 200000
|
|
# Without a price every request costs zero and no budget ever binds. 1000
|
|
# everywhere turns a budget into a plain token quota.
|
|
default_price:
|
|
prompt_per_1k: 1000
|
|
completion_per_1k: 1000
|
|
# The longest any API key lives; a key made without an expiry gets this.
|
|
# Shorter per organization, team or project with 'llmgw keys ttl'.
|
|
max_key_ttl: 2160h # 90 days
|
|
|
|
landing:
|
|
contact: "%s"
|
|
# What the web portal speaks to someone whose browser asks for neither
|
|
# language it has: en or tr. Each person can switch at the foot of a page.
|
|
# language: tr
|
|
|
|
%s
|
|
# The engines every node installs for itself. Omitted, each gets the llama.cpp
|
|
# build this release was checked against.
|
|
# engines:
|
|
# llamacpp: b10991
|
|
|
|
# A model mirror, once there is one (docs/install.md, step 3):
|
|
# mirror:
|
|
# api: https://models.internal:8082
|
|
# tls: { ca: /etc/llmgw/ca.pem, cert: /etc/llmgw/server.pem, key: /etc/llmgw/server-key.pem }
|
|
`
|
|
|
|
func (in *serverInit) ensureStore(ctx context.Context) error {
|
|
cfg, err := config.Load(in.configPath())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
address := cfg.Auth.Store
|
|
if address == "" {
|
|
address = filepath.Join(in.stateDir, "llmgw.db")
|
|
}
|
|
s, err := store.Open(ctx, address)
|
|
if err != nil {
|
|
return fmt.Errorf("opening %s: %w", address, err)
|
|
}
|
|
// Opened as root, so a SQLite file created just now belongs to root and
|
|
// the service could not write it. Given back before anything else runs.
|
|
defer in.chownSQLite(address)
|
|
defer s.Close()
|
|
|
|
// The tree an approved account hangs off. Approving into a project that
|
|
// does not exist succeeds, and the first key that person makes does not.
|
|
if err := s.PutOrg(ctx, store.Org{ID: "default", Name: "default"}); err != nil {
|
|
return err
|
|
}
|
|
if err := s.PutTeam(ctx, store.Team{ID: "default", OrgID: "default", Name: "default"}); err != nil {
|
|
return err
|
|
}
|
|
if err := s.PutProject(ctx, store.Project{ID: "default", TeamID: "default", Name: "default"}); err != nil {
|
|
return err
|
|
}
|
|
|
|
if in.admin == "" {
|
|
in.kept("store at %s (no --admin: create an account with 'llmgw users create')", address)
|
|
return nil
|
|
}
|
|
if u, err := s.UserByEmail(ctx, in.admin); err == nil {
|
|
if !u.IsAdmin {
|
|
if err := s.SetAdmin(ctx, u.ID, true); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
in.kept("administrator %s (%s); its password is unchanged", in.admin, u.ID)
|
|
return nil
|
|
} else if !errors.Is(err, store.ErrNotFound) {
|
|
return err
|
|
}
|
|
|
|
password := randomPassword()
|
|
hash, err := account.HashPassword(password)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id := randomID("usr")
|
|
now := time.Now()
|
|
if err := s.CreateUser(ctx, store.User{
|
|
ID: id, Email: in.admin, Name: in.admin,
|
|
PasswordHash: hash, State: store.UserPending, CreatedAt: now,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if err := s.ApproveUser(ctx, id, "default", "init", now); err != nil {
|
|
return err
|
|
}
|
|
if err := s.SetAdmin(ctx, id, true); err != nil {
|
|
return err
|
|
}
|
|
in.done("administrator %s (%s)", in.admin, id)
|
|
fmt.Fprintf(in.out, "\n password: %s\n shown once — sign in and change it from /profile\n\n", password)
|
|
return nil
|
|
}
|
|
|
|
func (in *serverInit) chownSQLite(address string) {
|
|
if strings.Contains(address, "://") || strings.HasPrefix(address, "@") {
|
|
return
|
|
}
|
|
for _, suffix := range []string{"", "-wal", "-shm"} {
|
|
if _, err := os.Stat(address + suffix); err == nil {
|
|
_ = in.chown(address+suffix, true)
|
|
}
|
|
}
|
|
}
|
|
|
|
func randomPassword() string {
|
|
var buf [15]byte
|
|
_, _ = rand.Read(buf[:])
|
|
return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(buf[:]))
|
|
}
|
|
|
|
func (in *serverInit) ensureService(ctx context.Context) error {
|
|
if _, err := exec.LookPath("systemctl"); err != nil {
|
|
fmt.Fprintf(in.out, " skip no systemd here; start it with:\n"+
|
|
" sudo -u %s llmgw server --config %s\n", in.user, in.configPath())
|
|
return nil
|
|
}
|
|
if b, err := os.ReadFile(in.unitPath); err == nil && string(b) == serverUnit {
|
|
in.kept("%s", in.unitPath)
|
|
} else {
|
|
if err == nil {
|
|
// Somebody's edits. Kept beside it, because the unit is also where
|
|
// people put environment overrides and those are not ours to lose.
|
|
_ = os.WriteFile(in.unitPath+".bak", b, 0o644)
|
|
}
|
|
if err := os.WriteFile(in.unitPath, []byte(serverUnit), 0o644); err != nil {
|
|
return err
|
|
}
|
|
in.done("%s", in.unitPath)
|
|
}
|
|
|
|
systemctl := func(args ...string) error {
|
|
out, err := exec.CommandContext(ctx, "systemctl", args...).CombinedOutput()
|
|
if err != nil {
|
|
return fmt.Errorf("systemctl %s: %v: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
|
|
}
|
|
return nil
|
|
}
|
|
if err := systemctl("daemon-reload"); err != nil {
|
|
return err
|
|
}
|
|
active := exec.CommandContext(ctx, "systemctl", "is-active", "--quiet", "llmgw-server").Run() == nil
|
|
switch {
|
|
case !active:
|
|
if err := systemctl("enable", "--now", "llmgw-server"); err != nil {
|
|
return err
|
|
}
|
|
in.done("llmgw-server started")
|
|
case in.changed:
|
|
// A certificate or configuration it has not read is one it is not
|
|
// using, and the difference shows up as a node that cannot connect.
|
|
if err := systemctl("restart", "llmgw-server"); err != nil {
|
|
return err
|
|
}
|
|
in.done("llmgw-server restarted to pick up the changes")
|
|
default:
|
|
in.kept("llmgw-server running")
|
|
}
|
|
return in.waitHealthy(ctx)
|
|
}
|
|
|
|
func (in *serverInit) waitHealthy(ctx context.Context) error {
|
|
cfg, err := config.Load(in.configPath())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, port, err := net.SplitHostPort(cfg.Listen)
|
|
if err != nil {
|
|
return fmt.Errorf("listen %q: %w", cfg.Listen, err)
|
|
}
|
|
url := "http://127.0.0.1:" + port + "/health"
|
|
deadline := time.Now().Add(30 * time.Second)
|
|
for {
|
|
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err == nil {
|
|
resp.Body.Close()
|
|
if resp.StatusCode == http.StatusOK {
|
|
in.kept("answering on %s", url)
|
|
return nil
|
|
}
|
|
}
|
|
if time.Now().After(deadline) {
|
|
return fmt.Errorf("not answering on %s after 30s; 'journalctl -u llmgw-server -n 50' says why", url)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-time.After(time.Second):
|
|
}
|
|
}
|
|
}
|
|
|
|
func (in *serverInit) nextSteps() {
|
|
fmt.Fprint(in.out, `
|
|
The control plane is up. Next:
|
|
|
|
llmgw ctl status # the fleet in one answer
|
|
|
|
# each node, from this machine (it holds the authority):
|
|
sudo llmgw node add <user>@<node address>
|
|
|
|
# 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.
|
|
|
|
Before people outside this network use it:
|
|
# port 8080 is plain HTTP: put a proxy with a certificate in front
|
|
# (deploy/proxy/ in the release; Caddy gets the certificate itself)
|
|
# backups carry every secret: name who may open them, backup_recipients
|
|
# in server.yaml, and copy them off this machine
|
|
`)
|
|
}
|
|
|
|
// primaryAddr is the address other machines most likely reach this one on:
|
|
// the source of its default route, else the first address on an interface
|
|
// that is not a container's or a bridge's. The first address of all is not
|
|
// it on a machine with Docker, whose 172.17.0.1 a node takes for its own.
|
|
func primaryAddr() string {
|
|
if out, err := exec.Command("ip", "-4", "route", "get", "1.1.1.1").Output(); err == nil {
|
|
if src := routeSource(string(out)); src != "" {
|
|
return src
|
|
}
|
|
}
|
|
ifaces, err := net.Interfaces()
|
|
if err == nil {
|
|
for _, ifc := range ifaces {
|
|
if ifc.Flags&net.FlagUp == 0 || virtualInterface(ifc.Name) {
|
|
continue
|
|
}
|
|
addrs, _ := ifc.Addrs()
|
|
for _, a := range addrs {
|
|
if n, ok := a.(*net.IPNet); ok && n.IP.To4() != nil && n.IP.IsGlobalUnicast() {
|
|
return n.IP.String()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if ips := localAddrs(); len(ips) > 0 {
|
|
return ips[0].String()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// routeSource reads the src of `ip route get`.
|
|
func routeSource(out string) string {
|
|
f := strings.Fields(out)
|
|
for i := 0; i+1 < len(f); i++ {
|
|
if f[i] == "src" {
|
|
if ip := net.ParseIP(f[i+1]); ip != nil && !ip.IsLoopback() {
|
|
return ip.String()
|
|
}
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// virtualInterface is an interface containers, bridges or overlays made,
|
|
// whose address no other machine reaches this one on.
|
|
func virtualInterface(name string) bool {
|
|
for _, p := range []string{"docker", "br-", "virbr", "veth", "cni", "flannel", "cali", "vxlan",
|
|
"kube", "lxc", "lxd", "podman", "cilium", "weave", "tun", "tap"} {
|
|
if strings.HasPrefix(name, p) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|