Ilker ManapandClaude Opus 5.5
63f7c8cfd7
Every model of a stopped system asks for no replica and so has what it asks for. Until the control plane took the setting up, seconds later, start --wait saw 18 of 18 satisfied and returned at once, with nothing started. It now counts only once the control plane says it is in the state asked for, both ways. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
233 lines
9.0 KiB
Go
233 lines
9.0 KiB
Go
package cli
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"os/exec"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
|
|
)
|
|
|
|
// systemClient is the admin API of the control plane on this machine, as its
|
|
// server.yaml says, with its token.
|
|
func systemClient(configPath string) func() *ctlClient {
|
|
return func() *ctlClient {
|
|
addr := "127.0.0.1:9090"
|
|
var tokens []string
|
|
if cfg, ok := loadServerConfig(configPath); ok {
|
|
if cfg.AdminListen != "" {
|
|
addr = cfg.AdminListen
|
|
if strings.HasPrefix(addr, ":") {
|
|
addr = "127.0.0.1" + addr
|
|
}
|
|
}
|
|
tokens = append(tokens, adminTokenPath(configPath, cfg))
|
|
}
|
|
return newCtlClient(addr, tokens...)
|
|
}
|
|
}
|
|
|
|
// newSystemCmds are llmgw status, stop, start and restart: the whole system,
|
|
// from a control plane.
|
|
func newSystemCmds() []*cobra.Command {
|
|
var configPath string
|
|
client := systemClient("/etc/llmgw/server.yaml")
|
|
bind := func(c *cobra.Command) *cobra.Command {
|
|
c.Flags().StringVar(&configPath, "config", "/etc/llmgw/server.yaml", "this control plane's configuration")
|
|
c.PreRun = func(*cobra.Command, []string) { client = systemClient(configPath) }
|
|
return c
|
|
}
|
|
call := func() *ctlClient { return client() }
|
|
|
|
status := bind(&cobra.Command{
|
|
Use: "status",
|
|
Short: "The whole system: whether it serves, its control planes, nodes, models and store",
|
|
Long: "Run on a control plane. Says whether the service is stopped (llmgw stop),\n" +
|
|
"and everything 'llmgw ctl status' does. One machine's services are\n" +
|
|
"'llmgw service status'.",
|
|
Args: cobra.NoArgs,
|
|
RunE: func(cmd *cobra.Command, _ []string) error {
|
|
var s adminapi.Summary
|
|
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err != nil {
|
|
return notAControlPlane(err)
|
|
}
|
|
printSummary(cmd, s)
|
|
return nil
|
|
},
|
|
})
|
|
|
|
switchService := func(stop bool) *cobra.Command {
|
|
use, short, long := "start", "Start the whole system serving again, after llmgw stop",
|
|
"Every model comes back as its catalog entry asks: placed, loaded and\n"+
|
|
"serving again, on every control plane within seconds. Run on a control\n"+
|
|
"plane. The services themselves are never stopped by llmgw stop; a\n"+
|
|
"machine's own are started with 'llmgw service start'."
|
|
if stop {
|
|
use, short, long = "stop", "Stop the whole system serving: every model taken down, requests answered 503",
|
|
"Every model's replicas are taken down, freeing every GPU, and every\n"+
|
|
"request for a model is answered 503 (system_stopped) until 'llmgw start'.\n"+
|
|
"The control planes, agents and mirror keep running, so the console, the\n"+
|
|
"admin pages and llmgw start still work, and no model's definition is\n"+
|
|
"changed. In force on every control plane within seconds; kept in the\n"+
|
|
"cluster's settings (serving.stopped), so a restart does not undo it."
|
|
}
|
|
var note string
|
|
var wait bool
|
|
c := bind(&cobra.Command{
|
|
Use: use,
|
|
Short: short,
|
|
Long: long,
|
|
Args: cobra.NoArgs,
|
|
RunE: func(cmd *cobra.Command, _ []string) error {
|
|
if note == "" {
|
|
note = "llmgw " + use
|
|
}
|
|
var s adminapi.Summary
|
|
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err != nil {
|
|
return notAControlPlane(err)
|
|
}
|
|
if s.Stopped == stop {
|
|
fmt.Fprintf(cmd.OutOrStdout(), "the system is already %s\n", map[bool]string{true: "stopped", false: "serving"}[stop])
|
|
return nil
|
|
}
|
|
if err := changeSettings(cmd, call, note, func(doc *settings.Doc) error {
|
|
return doc.SetYAML([]string{"serving", "stopped"}, fmt.Sprint(stop))
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
out := cmd.OutOrStdout()
|
|
if !wait {
|
|
if stop {
|
|
fmt.Fprintln(out, "stopping: every model's replicas are being taken down; llmgw status follows it")
|
|
} else {
|
|
fmt.Fprintln(out, "starting: every model is placed again as it asks; llmgw status follows it")
|
|
}
|
|
return nil
|
|
}
|
|
return waitForService(cmd, call, stop)
|
|
},
|
|
})
|
|
c.Flags().StringVar(¬e, "note", "", "why, kept with the settings' version and in the audit")
|
|
c.Flags().BoolVar(&wait, "wait", false, "wait until every replica is down (stop) or the models have theirs (start)")
|
|
return c
|
|
}
|
|
|
|
var sshUser string
|
|
restart := bind(&cobra.Command{
|
|
Use: "restart",
|
|
Short: "Restart every control plane and node agent; the engines keep serving",
|
|
Long: "Every node's agent starts again, then the other control planes over ssh\n" +
|
|
"as --ssh-user, then this one — the leader last, so it hands over to a\n" +
|
|
"standby already restarted. The engines keep serving throughout and are\n" +
|
|
"taken over again; nothing is placed anew. Run on a control plane, with\n" +
|
|
"sudo. One machine's services are 'llmgw service restart'.",
|
|
Args: cobra.NoArgs,
|
|
RunE: func(cmd *cobra.Command, _ []string) error {
|
|
if os.Geteuid() != 0 {
|
|
return errors.New("restarting this control plane needs root: run it with sudo")
|
|
}
|
|
ctx, out := cmd.Context(), cmd.OutOrStdout()
|
|
var s adminapi.Summary
|
|
if err := call().get(ctx, "/api/v1/summary", &s); err != nil {
|
|
return notAControlPlane(err)
|
|
}
|
|
var agents adminapi.AgentRestartView
|
|
if err := call().post(ctx, "/api/v1/agents/restart", []byte("{}"), &agents); err != nil {
|
|
return fmt.Errorf("restarting the agents: %w", err)
|
|
}
|
|
fmt.Fprintf(out, "agents %d told to restart: %s\n", len(agents.Nodes), strings.Join(agents.Nodes, ", "))
|
|
|
|
// The other control planes, standbys before a leader.
|
|
var others []adminapi.ControlPlane
|
|
for _, cp := range s.ControlPlanes.Members {
|
|
if !cp.This {
|
|
others = append(others, cp)
|
|
}
|
|
}
|
|
// One that cannot be reached is said and left: the rest of the
|
|
// system still restarts, and the leader still goes last.
|
|
var failed []string
|
|
for pass := 0; pass < 2; pass++ {
|
|
for _, cp := range others {
|
|
leading := cp.State == "leading"
|
|
if leading != (pass == 1) {
|
|
continue
|
|
}
|
|
r := &onboard.SSH{Addr: cp.Address, User: sshUser}
|
|
if res, err := r.Run(ctx, "sudo systemctl restart llmgw-server", nil); err != nil {
|
|
fmt.Fprintf(out, "%-10s NOT restarted: ssh as %q: %v %s\n", cp.Name, sshUser, err, strings.TrimSpace(string(res)))
|
|
failed = append(failed, cp.Name+" ("+cp.Address+")")
|
|
continue
|
|
}
|
|
fmt.Fprintf(out, "%-10s restarted (%s)\n", cp.Name, cp.State)
|
|
time.Sleep(5 * time.Second)
|
|
}
|
|
}
|
|
if _, err := os.Stat(unitDir + "/llmgw-server.service"); err == nil {
|
|
fmt.Fprintln(out, "this restarting")
|
|
if b, err := exec.CommandContext(ctx, "systemctl", "restart", "llmgw-server").CombinedOutput(); err != nil { //nolint:gosec // fixed command
|
|
return fmt.Errorf("restarting this control plane: %v: %s", err, strings.TrimSpace(string(b)))
|
|
}
|
|
fmt.Fprintln(out, "this restarted")
|
|
}
|
|
fmt.Fprintln(out, "\nllmgw status shows the nodes coming back; their replicas are taken over as they are.")
|
|
if len(failed) > 0 {
|
|
return fmt.Errorf("not restarted: %s; run `sudo llmgw service restart server` there, "+
|
|
"or give --ssh-user someone this machine's root can log in as", strings.Join(failed, ", "))
|
|
}
|
|
return nil
|
|
},
|
|
})
|
|
restart.Flags().StringVar(&sshUser, "ssh-user", os.Getenv("SUDO_USER"), "who to log in to the other control planes as")
|
|
|
|
return []*cobra.Command{status, switchService(true), switchService(false), restart}
|
|
}
|
|
|
|
// servicePoll is how often --wait asks; a variable so tests need not wait.
|
|
var servicePoll = 5 * time.Second
|
|
|
|
// waitForService waits until no replica runs (stopped) or every model has
|
|
// the replicas it asks for (started).
|
|
//
|
|
// Only once the control plane says it is in the state asked for: it takes the
|
|
// setting up at its next look at the store, seconds later, and until then a
|
|
// stopped system's models, which ask for nothing, all have what they ask for —
|
|
// so llmgw start --wait returned at once, with nothing started.
|
|
func waitForService(cmd *cobra.Command, call func() *ctlClient, stop bool) error {
|
|
deadline := time.Now().Add(15 * time.Minute)
|
|
for time.Now().Before(deadline) {
|
|
var s adminapi.Summary
|
|
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err == nil && s.Stopped == stop {
|
|
if stop && s.Replicas.Ready+s.Replicas.Starting+s.Replicas.Draining == 0 {
|
|
fmt.Fprintln(cmd.OutOrStdout(), "stopped: no replica runs")
|
|
return nil
|
|
}
|
|
if !stop && s.Models.Satisfied >= s.Models.Total {
|
|
fmt.Fprintf(cmd.OutOrStdout(), "serving: %d of %d models have the replicas they ask for\n",
|
|
s.Models.Satisfied, s.Models.Total)
|
|
return nil
|
|
}
|
|
}
|
|
select {
|
|
case <-cmd.Context().Done():
|
|
return cmd.Context().Err()
|
|
case <-time.After(servicePoll):
|
|
}
|
|
}
|
|
return errors.New("not there within fifteen minutes; llmgw status says where it stands")
|
|
}
|
|
|
|
// notAControlPlane is a failure to reach the admin API, said as what it
|
|
// most often means.
|
|
func notAControlPlane(err error) error {
|
|
return fmt.Errorf("%w\n(the whole system is stopped and started from a control plane; this machine's own "+
|
|
"services are 'llmgw service status|start|stop|restart')", err)
|
|
}
|