Slurm beside llmgw, P3 and P4: the leader hands nodes over; ctl slurm
CI / test (push) Successful in 9m40s
CI / test (push) Successful in 9m40s
internal/control/slurmsvc tells each node what it is, writes slurm.conf and gres.conf from the nodes' reports and the inventory, and moves each node of slurm.nodes between llmgw and Slurm: lent, drained of replicas first, then resumed in Slurm; taken back, drained in Slurm and cordoned in llmgw until its last job step ends. Kept every tick from the settings and the agents' reports, so a new leader carries on. A broken Slurm never keeps a node from llmgw. GET /api/v1/slurm; ctl slurm status, setup, lend, reclaim, off. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
35de47444c
commit
410fe545ce
18 files changed
+1178
-11
No files matched your search
@@ -2028,6 +2028,15 @@ lent, partition`); the leader tells each named node's agent (root) what to be, t
|
||||
slurmd/slurmctld, status); P3 the leader side (conf from reports, lend/reclaim); P4 `ctl slurm`
|
||||
(status, lend, reclaim) and a TUI screen; P5 `ctl train import` of a Slurm job's output into the mirror.
|
||||
Later: a backup slurmctld on a second node, accounting (slurmdbd on the Patroni PG), users from LDAP/AD.
|
||||
- Built 2026-11-11, not tried on a machine yet: P1 (`internal/slurm`, `slurm.*` settings), P2
|
||||
(`internal/agent/slurmnode`, proto Slurm/SlurmStatus), P3 (`internal/control/slurmsvc`, wired in the
|
||||
leader; `GET /api/v1/slurm`), P4's commands (`ctl slurm status|setup|lend|reclaim|off`). Decisions on the
|
||||
way: a node not heard from or with Slurm jobs is cordoned, not drained (nothing of llmgw's to move, and a
|
||||
new leader must not move models off every Slurm node); jobs are counted from the node's own slurmstepd,
|
||||
so a controller that is down neither hides them nor, with none, keeps the node from llmgw; a drain whose
|
||||
reason does not start with "llmgw" is a Slurm administrator's and never resumed. Left: the TUI screen, P5,
|
||||
a try on the PoC (needs agents built from this; slurm-wlm 23.11 on Ubuntu 24.04), GPU minor numbers are
|
||||
taken to be nvidia-smi's index.
|
||||
|
||||
## D. Facts worth knowing before starting
|
||||
|
||||
|
||||
@@ -5401,7 +5401,9 @@ type SlurmStatus struct {
|
||||
// state and reason are the controller's for this node, from scontrol.
|
||||
State string `protobuf:"bytes,5,opt,name=state,proto3" json:"state,omitempty"`
|
||||
Reason string `protobuf:"bytes,6,opt,name=reason,proto3" json:"reason,omitempty"`
|
||||
// jobs running on this node now.
|
||||
// jobs is how many job steps run on this node now, counted from their
|
||||
// slurmstepd so a controller that is down does not hide them; -1 when the
|
||||
// agent cannot tell.
|
||||
Jobs int32 `protobuf:"varint,7,opt,name=jobs,proto3" json:"jobs,omitempty"`
|
||||
// error is the last thing that went wrong keeping Slurm, empty when
|
||||
// nothing did.
|
||||
|
||||
@@ -1024,7 +1024,9 @@ message SlurmStatus {
|
||||
// state and reason are the controller's for this node, from scontrol.
|
||||
string state = 5;
|
||||
string reason = 6;
|
||||
// jobs running on this node now.
|
||||
// jobs is how many job steps run on this node now, counted from their
|
||||
// slurmstepd so a controller that is down does not hide them; -1 when the
|
||||
// agent cannot tell.
|
||||
int32 jobs = 7;
|
||||
// error is the last thing that went wrong keeping Slurm, empty when
|
||||
// nothing did.
|
||||
|
||||
@@ -116,6 +116,9 @@
|
||||
{
|
||||
"name": "settings"
|
||||
},
|
||||
{
|
||||
"name": "slurm"
|
||||
},
|
||||
{
|
||||
"name": "summary"
|
||||
},
|
||||
@@ -5078,6 +5081,34 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/v1/slurm": {
|
||||
"get": {
|
||||
"operationId": "getSlurm",
|
||||
"summary": "The Slurm cluster beside llmgw, and where each of its nodes is",
|
||||
"description": "From the cluster's settings (slurm.*) and what each node's agent last said. A node's phase is llmgw (drained in Slurm, no job left), lending (its replicas moving off), lent, reclaiming (drained in Slurm, jobs still running) or unknown (its agent has not said).",
|
||||
"tags": [
|
||||
"slurm"
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "OK",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/slurmsvc.Status"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"401": {
|
||||
"$ref": "#/components/responses/Unauthorized"
|
||||
},
|
||||
"default": {
|
||||
"$ref": "#/components/responses/Error"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/v1/summary": {
|
||||
"get": {
|
||||
"operationId": "getSummary",
|
||||
@@ -12178,6 +12209,98 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"slurmsvc.NodeStatus": {
|
||||
"type": "object",
|
||||
"description": "NodeStatus is one node as the service sees it.",
|
||||
"properties": {
|
||||
"node": {
|
||||
"type": "string"
|
||||
},
|
||||
"phase": {
|
||||
"type": "string"
|
||||
},
|
||||
"replicas": {
|
||||
"type": "integer"
|
||||
},
|
||||
"slurm": {
|
||||
"$ref": "#/components/schemas/slurmsvc.Reported"
|
||||
},
|
||||
"connected": {
|
||||
"type": "boolean"
|
||||
}
|
||||
}
|
||||
},
|
||||
"slurmsvc.Reported": {
|
||||
"type": "object",
|
||||
"description": "Reported is the agent's SlurmStatus, for the admin API.",
|
||||
"properties": {
|
||||
"version": {
|
||||
"type": "string"
|
||||
},
|
||||
"hardware": {
|
||||
"type": "string"
|
||||
},
|
||||
"slurmd_running": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"slurmctld_running": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"state": {
|
||||
"type": "string"
|
||||
},
|
||||
"reason": {
|
||||
"type": "string"
|
||||
},
|
||||
"jobs": {
|
||||
"type": "integer"
|
||||
},
|
||||
"error": {
|
||||
"type": "string"
|
||||
},
|
||||
"files_sha256": {
|
||||
"type": "string"
|
||||
},
|
||||
"at": {
|
||||
"type": "string",
|
||||
"format": "date-time"
|
||||
}
|
||||
}
|
||||
},
|
||||
"slurmsvc.Status": {
|
||||
"type": "object",
|
||||
"description": "Status is the whole Slurm cluster, for `ctl slurm status`.",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"cluster": {
|
||||
"type": "string"
|
||||
},
|
||||
"controller": {
|
||||
"type": "string"
|
||||
},
|
||||
"partition": {
|
||||
"type": "string"
|
||||
},
|
||||
"controller_status": {
|
||||
"$ref": "#/components/schemas/slurmsvc.Reported"
|
||||
},
|
||||
"nodes": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"$ref": "#/components/schemas/slurmsvc.NodeStatus"
|
||||
}
|
||||
},
|
||||
"waiting": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"description": "Waiting are nodes left out of slurm.conf until they report their hardware."
|
||||
}
|
||||
}
|
||||
},
|
||||
"store.AuditEvent": {
|
||||
"type": "object",
|
||||
"description": "AuditEvent is one change: who made it, through what, to what, and how it ended. Detail never holds a secret; the writer takes them out.",
|
||||
|
||||
@@ -390,6 +390,7 @@ func (k *Keeper) node(ctx context.Context, want *llmgwv1.Slurm, st *llmgwv1.Slur
|
||||
}
|
||||
st.SlurmdRunning = k.active(ctx, "slurmd")
|
||||
|
||||
st.Jobs = k.localJobs(ctx)
|
||||
env := k.clientEnv()
|
||||
out, err := k.host.Run(ctx, env, "scontrol", "show", "node", "-o", name)
|
||||
if err != nil {
|
||||
@@ -421,14 +422,23 @@ func (k *Keeper) node(ctx context.Context, want *llmgwv1.Slurm, st *llmgwv1.Slur
|
||||
}
|
||||
}
|
||||
st.State, st.Reason = ns.State, ns.Reason
|
||||
if out, err := k.host.Run(ctx, env, "squeue", "-h", "-w", name, "-o", "%i"); err == nil {
|
||||
st.Jobs = int32(len(strings.Fields(string(out)))) //nolint:gosec // a count of jobs
|
||||
} else {
|
||||
st.Jobs = -1
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// localJobs counts the job steps running here, from their slurmstepd, not
|
||||
// from the controller: a job outlives a controller that is down, and is
|
||||
// what decides whether llmgw may have the cards back. -1 when it cannot
|
||||
// tell.
|
||||
func (k *Keeper) localJobs(ctx context.Context) int32 {
|
||||
// pgrep -c prints 0 and exits 1 when there is none.
|
||||
out, _ := k.host.Run(ctx, nil, "pgrep", "-c", "-x", "slurmstepd")
|
||||
n, err := strconv.Atoi(strings.TrimSpace(string(out)))
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
return int32(n) //nolint:gosec // a count of processes
|
||||
}
|
||||
|
||||
// clientEnv points scontrol at slurmd's copy of the files on a node that is
|
||||
// not the controller, which has none of its own.
|
||||
func (k *Keeper) clientEnv() []string {
|
||||
|
||||
@@ -68,8 +68,11 @@ func (f *fakeHost) Run(_ context.Context, env []string, name string, args ...str
|
||||
f.state, f.reason = "IDLE+DRAIN", ReasonServing
|
||||
}
|
||||
return nil, nil
|
||||
case strings.HasPrefix(c, "squeue"):
|
||||
return []byte(f.outputs["squeue"]), nil
|
||||
case strings.HasPrefix(c, "pgrep"):
|
||||
if f.outputs["jobs"] == "" {
|
||||
return []byte("0\n"), errors.New("exit status 1")
|
||||
}
|
||||
return []byte(f.outputs["jobs"]), nil
|
||||
}
|
||||
return []byte(f.outputs[c]), nil
|
||||
}
|
||||
@@ -175,7 +178,7 @@ func TestLendAndReclaim(t *testing.T) {
|
||||
if h.count("scontrol update NodeName=cuda01 State=RESUME") != 1 || st.GetState() != "IDLE" {
|
||||
t.Fatalf("a lent node was not resumed: %v %v", h.ran, st)
|
||||
}
|
||||
h.outputs["squeue"] = "101\n102\n"
|
||||
h.outputs["jobs"] = "2\n"
|
||||
st = k.Reconcile(context.Background(), nodeMsg(false))
|
||||
if h.count("scontrol update NodeName=cuda01 State=DRAIN Reason="+ReasonServing) != 1 {
|
||||
t.Fatalf("a node taken back was not drained: %v", h.ran)
|
||||
|
||||
@@ -247,6 +247,8 @@ var liveSettings = []string{
|
||||
"presets",
|
||||
// Handed to the reconciler as they change.
|
||||
"on_demand",
|
||||
// Read by the Slurm service each pass, kicked when they change.
|
||||
"slurm",
|
||||
// Read by server init, node add and backups, not by the server.
|
||||
"backup_recipients", "backup_copy_to",
|
||||
}
|
||||
|
||||
@@ -189,6 +189,7 @@ func newCtlCmd() *cobra.Command {
|
||||
newMCPCmd(client, &asJSON),
|
||||
newKBCmd(client, &asJSON),
|
||||
newTrainCmd(client, &asJSON),
|
||||
newSlurmCmd(client, &asJSON),
|
||||
newConfigCmd(client, &asJSON),
|
||||
newAuditCmd(client, &asJSON),
|
||||
newNotifyCmd(client, &asJSON),
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"text/tabwriter"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/slurmsvc"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
|
||||
)
|
||||
|
||||
// newSlurmCmd is the Slurm cluster run beside llmgw (BACKLOG P): set up on
|
||||
// nodes llmgw already has, and nodes handed between the two.
|
||||
func newSlurmCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
cmd := &cobra.Command{
|
||||
Use: "slurm",
|
||||
Short: "A Slurm cluster beside llmgw, for training and other batch work",
|
||||
Long: "Slurm runs on llmgw's own nodes, installed and kept by their agents: slurmctld on one node,\n" +
|
||||
"slurmd on those named. A node is llmgw's, drained in Slurm, until it is lent; lent, its\n" +
|
||||
"replicas move off first, then Slurm may run jobs there. Taken back, Slurm drains it and\n" +
|
||||
"llmgw has it again when its last job ends. Jobs are submitted on the controller node\n" +
|
||||
"(sbatch, srun) by its own users, as on any Slurm cluster.",
|
||||
}
|
||||
cmd.AddCommand(newSlurmStatusCmd(client, asJSON), newSlurmSetupCmd(client), newSlurmOffCmd(client),
|
||||
newSlurmListCmd(client, "lend", "Let Slurm run jobs on nodes; llmgw's replicas move off first", true),
|
||||
newSlurmListCmd(client, "reclaim", "Take nodes back from Slurm; llmgw has them when their jobs end", false))
|
||||
return cmd
|
||||
}
|
||||
|
||||
func newSlurmStatusCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
return &cobra.Command{
|
||||
Use: "status",
|
||||
Short: "The Slurm cluster, and where each of its nodes is",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
var st slurmsvc.Status
|
||||
if err := client().get(cmd.Context(), "/api/v1/slurm", &st); err != nil {
|
||||
return err
|
||||
}
|
||||
w := cmd.OutOrStdout()
|
||||
if *asJSON {
|
||||
return json.NewEncoder(w).Encode(st)
|
||||
}
|
||||
if !st.Enabled {
|
||||
fmt.Fprintln(w, "Slurm is off. `llmgw ctl slurm setup --controller NODE --nodes NODE,...` turns it on.")
|
||||
return nil
|
||||
}
|
||||
fmt.Fprintf(w, "cluster %s, partition %s, controller %s", st.Cluster, st.Partition, st.Controller)
|
||||
switch c := st.Ctld; {
|
||||
case c == nil:
|
||||
fmt.Fprintln(w, ": not heard from yet")
|
||||
case c.Error != "":
|
||||
fmt.Fprintf(w, ": %s\n", c.Error)
|
||||
case c.SlurmctldRunning:
|
||||
fmt.Fprintf(w, ": slurmctld %s running\n", c.Version)
|
||||
default:
|
||||
fmt.Fprintln(w, ": slurmctld not running")
|
||||
}
|
||||
tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
|
||||
fmt.Fprintln(tw, "NODE\tPHASE\tSLURM\tJOBS\tSLURMD\tNOTE")
|
||||
for _, n := range st.Nodes {
|
||||
state, jobs, slurmd, note := "-", "-", "-", n.Describe()
|
||||
if r := n.Status; r != nil {
|
||||
state = or(r.State, "-")
|
||||
jobs = fmt.Sprint(r.Jobs)
|
||||
slurmd = map[bool]string{true: "running", false: "stopped"}[r.SlurmdRunning]
|
||||
if r.Error != "" {
|
||||
note = r.Error
|
||||
}
|
||||
}
|
||||
if !n.Connected {
|
||||
note = "agent not connected; " + note
|
||||
}
|
||||
fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\n", n.Node, n.Phase, state, jobs, slurmd, note)
|
||||
}
|
||||
if err := tw.Flush(); err != nil {
|
||||
return err
|
||||
}
|
||||
if len(st.Waiting) > 0 {
|
||||
fmt.Fprintf(w, "not in slurm.conf until they report their hardware: %s\n", strings.Join(st.Waiting, ", "))
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func newSlurmSetupCmd(client func() *ctlClient) *cobra.Command {
|
||||
var controller, cluster, partition, note string
|
||||
var nodes []string
|
||||
c := &cobra.Command{
|
||||
Use: "setup --controller NODE --nodes NODE,...",
|
||||
Short: "Run Slurm on llmgw's nodes: slurmctld on one, slurmd on those named",
|
||||
Long: "Turns slurm.enabled on and names the nodes. Their agents install munge and Slurm (22.05 or\n" +
|
||||
"newer) from the machines' own package sources. Every node starts drained in Slurm and stays\n" +
|
||||
"llmgw's until `llmgw ctl slurm lend` gives it to Slurm.",
|
||||
Example: " llmgw ctl slurm setup --controller node06 --nodes cuda01,cuda02",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
if controller == "" || len(nodes) == 0 {
|
||||
return fmt.Errorf("give --controller and --nodes")
|
||||
}
|
||||
return changeSettings(cmd, client, note, func(doc *settings.Doc) error {
|
||||
if err := doc.SetYAML([]string{"slurm", "enabled"}, "true"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := doc.SetYAML([]string{"slurm", "controller"}, yamlString(controller)); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := doc.SetYAML([]string{"slurm", "nodes"}, yamlList(nodes)); err != nil {
|
||||
return err
|
||||
}
|
||||
// Lent nodes no longer named are not lent.
|
||||
lent := slices.DeleteFunc(docList(doc, []string{"slurm", "lent"}),
|
||||
func(n string) bool { return !slices.Contains(nodes, n) })
|
||||
if err := setList(doc, []string{"slurm", "lent"}, lent); err != nil {
|
||||
return err
|
||||
}
|
||||
for k, v := range map[string]string{"cluster_name": cluster, "partition": partition} {
|
||||
if v != "" {
|
||||
if err := doc.SetYAML([]string{"slurm", k}, yamlString(v)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
},
|
||||
}
|
||||
c.Flags().StringVar(&controller, "controller", "", "the node that runs slurmctld")
|
||||
c.Flags().StringSliceVar(&nodes, "nodes", nil, "the nodes that run slurmd")
|
||||
c.Flags().StringVar(&cluster, "cluster-name", "", "Slurm's ClusterName (llmgw)")
|
||||
c.Flags().StringVar(&partition, "partition", "", "the partition every node is in (gpu)")
|
||||
c.Flags().StringVar(¬e, "note", "", "why, kept with the settings' version")
|
||||
return c
|
||||
}
|
||||
|
||||
func newSlurmOffCmd(client func() *ctlClient) *cobra.Command {
|
||||
var note string
|
||||
var force bool
|
||||
c := &cobra.Command{
|
||||
Use: "off",
|
||||
Short: "Stop running Slurm; every node is llmgw's again",
|
||||
Long: "slurmd and slurmctld are stopped where llmgw started them; the packages stay. A node still\n" +
|
||||
"lent is refused unless --force: take it back first, so its jobs end before the daemons do.",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
return changeSettings(cmd, client, note, func(doc *settings.Doc) error {
|
||||
if lent := docList(doc, []string{"slurm", "lent"}); len(lent) > 0 && !force {
|
||||
return fmt.Errorf("%s still lent: `llmgw ctl slurm reclaim` them first, or --force",
|
||||
strings.Join(lent, ", "))
|
||||
}
|
||||
setList(doc, []string{"slurm", "lent"}, nil) //nolint:errcheck // an empty list
|
||||
return doc.SetYAML([]string{"slurm", "enabled"}, "false")
|
||||
})
|
||||
},
|
||||
}
|
||||
c.Flags().BoolVar(&force, "force", false, "even with nodes still lent")
|
||||
c.Flags().StringVar(¬e, "note", "", "why, kept with the settings' version")
|
||||
return c
|
||||
}
|
||||
|
||||
// newSlurmListCmd adds nodes to slurm.lent, or takes them out.
|
||||
func newSlurmListCmd(client func() *ctlClient, use, short string, add bool) *cobra.Command {
|
||||
var note string
|
||||
c := &cobra.Command{
|
||||
Use: use + " NODE...",
|
||||
Short: short,
|
||||
Args: cobra.MinimumNArgs(1),
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
return changeSettings(cmd, client, note, func(doc *settings.Doc) error {
|
||||
nodes := docList(doc, []string{"slurm", "nodes"})
|
||||
lent := docList(doc, []string{"slurm", "lent"})
|
||||
for _, n := range args {
|
||||
switch {
|
||||
case add && !slices.Contains(nodes, n):
|
||||
return fmt.Errorf("%s does not run slurmd: add it with `llmgw ctl slurm setup --nodes`", n)
|
||||
case add && !slices.Contains(lent, n):
|
||||
lent = append(lent, n)
|
||||
case !add:
|
||||
lent = slices.DeleteFunc(lent, func(l string) bool { return l == n })
|
||||
}
|
||||
}
|
||||
return setList(doc, []string{"slurm", "lent"}, lent)
|
||||
})
|
||||
},
|
||||
}
|
||||
c.Flags().StringVar(¬e, "note", "", "why, kept with the settings' version")
|
||||
return c
|
||||
}
|
||||
|
||||
// docList reads a list of strings from the settings, empty when unset.
|
||||
func docList(doc *settings.Doc, path []string) []string {
|
||||
n := doc.Node(path)
|
||||
if n == nil {
|
||||
return nil
|
||||
}
|
||||
var out []string
|
||||
if err := n.Decode(&out); err != nil {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func setList(doc *settings.Doc, path []string, list []string) error {
|
||||
if len(list) == 0 {
|
||||
if doc.Node(path) != nil {
|
||||
doc.Unset(path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return doc.SetYAML(path, yamlList(list))
|
||||
}
|
||||
|
||||
func yamlString(s string) string {
|
||||
b, _ := json.Marshal(s)
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func yamlList(list []string) string {
|
||||
b, _ := json.Marshal(list)
|
||||
return string(b)
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
|
||||
)
|
||||
|
||||
func TestSlurmDocLists(t *testing.T) {
|
||||
doc, err := settings.FromBytes("settings", []byte("slurm:\n enabled: true\n nodes: [cuda01, site2]\n"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := docList(doc, []string{"slurm", "nodes"}); !slices.Equal(got, []string{"cuda01", "site2"}) {
|
||||
t.Fatalf("nodes %v", got)
|
||||
}
|
||||
if got := docList(doc, []string{"slurm", "lent"}); got != nil {
|
||||
t.Fatalf("lent %v", got)
|
||||
}
|
||||
if err := setList(doc, []string{"slurm", "lent"}, []string{"cuda01"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := docList(doc, []string{"slurm", "lent"}); !slices.Equal(got, []string{"cuda01"}) {
|
||||
t.Fatalf("lent %v", got)
|
||||
}
|
||||
if err := setList(doc, []string{"slurm", "lent"}, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, _ := doc.Bytes()
|
||||
if strings.Contains(string(b), "lent") {
|
||||
t.Fatalf("an empty list left:\n%s", b)
|
||||
}
|
||||
}
|
||||
@@ -49,6 +49,7 @@ import (
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/slurmsvc"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/train"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
|
||||
@@ -59,6 +60,7 @@ import (
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/promq"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/router"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/secrets"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/slurm"
|
||||
storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/version"
|
||||
@@ -261,6 +263,8 @@ func newServerCmd() *cobra.Command {
|
||||
// What the settings say of the firewall, for the mirror and the
|
||||
// store's members; nil without a control plane.
|
||||
var fwPub *firewallPublisher
|
||||
// The Slurm cluster beside llmgw, when there is a store (BACKLOG P).
|
||||
var slurmSvc *slurmsvc.Service
|
||||
updateMetrics := &modelserve.UpdateCollector{}
|
||||
// The store's cluster, when it is one; read by the metrics and
|
||||
// by `ctl status`.
|
||||
@@ -436,6 +440,19 @@ func newServerCmd() *cobra.Command {
|
||||
agentSvc.SetTrain(trainer)
|
||||
go trainer.Run(ctx)
|
||||
api.SetTrain(&adminapi.TrainAdmin{Store: st, Service: trainer})
|
||||
// Slurm on the nodes the settings name, its munge key
|
||||
// made from a cluster secret.
|
||||
slurmSvc = slurmsvc.New(func() config.SlurmSettings { return current.Load().Slurm },
|
||||
admin.Inventory(), agentSvc, func(ctx context.Context) ([]byte, error) {
|
||||
secret, err := st.ClusterSecret(ctx, "slurm-munge")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return slurm.MungeKey(secret)
|
||||
}, log)
|
||||
agentSvc.SetSlurm(slurmSvc)
|
||||
go slurmSvc.Run(ctx)
|
||||
api.SetSlurm(slurmSvc)
|
||||
}
|
||||
// The cluster's settings, each change checked against this
|
||||
// machine's own keys before it is kept.
|
||||
@@ -841,6 +858,9 @@ func newServerCmd() *cobra.Command {
|
||||
fwPub.Kick()
|
||||
}
|
||||
}
|
||||
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "slurm") }) && slurmSvc != nil {
|
||||
slurmSvc.Kick()
|
||||
}
|
||||
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "on_demand") }) && admin != nil {
|
||||
admin.Reconciler().SetEvictionTimes(next.OnDemand.EvictAfterIdle, next.OnDemand.MinLifetime)
|
||||
}
|
||||
|
||||
@@ -78,7 +78,9 @@ type API struct {
|
||||
// tenancy is organizations, keys and people; nil without a store.
|
||||
tenancy TenancyStore
|
||||
// train is adapter training; nil without a store and agents.
|
||||
train *TrainAdmin
|
||||
train *TrainAdmin
|
||||
// slurm is the Slurm cluster beside llmgw; nil without a store and agents.
|
||||
slurm SlurmService
|
||||
settings *Settings
|
||||
auth *AuthAdmin
|
||||
mcp MCPStore
|
||||
@@ -136,6 +138,7 @@ func (a *API) Register(mux Mux) {
|
||||
a.registerKB(mux)
|
||||
a.registerTenancy(mux)
|
||||
a.registerTrain(mux)
|
||||
a.registerSlurm(mux)
|
||||
a.registerGuard(mux)
|
||||
a.registerSettings(mux)
|
||||
a.registerAuth(mux)
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/federation"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/slurmsvc"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
||||
@@ -392,6 +393,12 @@ var Routes = []Route{
|
||||
Description: "Their keys in projects they no longer reach are revoked.",
|
||||
Request: MembershipRequest{}, Response: UserView{}},
|
||||
|
||||
{Method: "GET", Path: "/api/v1/slurm", Summary: "The Slurm cluster beside llmgw, and where each of its nodes is",
|
||||
Description: "From the cluster's settings (slurm.*) and what each node's agent last said. A node's phase is llmgw " +
|
||||
"(drained in Slurm, no job left), lending (its replicas moving off), lent, reclaiming (drained in Slurm, jobs " +
|
||||
"still running) or unknown (its agent has not said).",
|
||||
Response: slurmsvc.Status{}},
|
||||
|
||||
{Method: "POST", Path: "/api/v1/train/jobs", Summary: "Queue a LoRA adapter's training on a node",
|
||||
Description: "The node trains when its agent is there, nothing else trains on it, and a card has gpu_memory_mib free; " +
|
||||
"the pairs are NDJSON of {\"q\", \"a\"}. An adapter learns how answers sound far better than what they say.",
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
package adminapi
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/slurmsvc"
|
||||
)
|
||||
|
||||
// SlurmService is the Slurm cluster beside llmgw (BACKLOG P), as the leader
|
||||
// keeps it.
|
||||
type SlurmService interface {
|
||||
Status() slurmsvc.Status
|
||||
}
|
||||
|
||||
// SetSlurm gives the API the Slurm service.
|
||||
func (a *API) SetSlurm(s SlurmService) { a.slurm = s }
|
||||
|
||||
func (a *API) registerSlurm(mux Mux) {
|
||||
mux.HandleFunc("GET /api/v1/slurm", a.slurmStatus)
|
||||
}
|
||||
|
||||
func (a *API) slurmStatus(w http.ResponseWriter, _ *http.Request) {
|
||||
if a.slurm == nil {
|
||||
writeError(w, http.StatusNotImplemented, "Slurm needs the store and the nodes' agents, which this control plane runs without")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, a.slurm.Status())
|
||||
}
|
||||
@@ -84,6 +84,8 @@ type Service struct {
|
||||
agentSettings func(node string) string
|
||||
// train is told how training jobs go; see SetTrain.
|
||||
train TrainHandler
|
||||
// slurm is told of the nodes' part in Slurm; see SetSlurm.
|
||||
slurm SlurmHandler
|
||||
|
||||
// logWaits are the log requests awaiting a reply, by request id.
|
||||
logMu sync.Mutex
|
||||
@@ -308,6 +310,10 @@ func (s *Service) Connect(
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Its part in Slurm, sent by the Slurm service's next pass.
|
||||
if h := s.slurmHandler(); h != nil {
|
||||
h.OnRegister(nodeID)
|
||||
}
|
||||
|
||||
// Recover replica state from what this node is already running, before any
|
||||
// reconciliation pass can conclude the cluster is empty and place
|
||||
@@ -515,6 +521,11 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
|
||||
h.OnTrainStatus(nodeID, m.TrainStatus)
|
||||
}
|
||||
|
||||
case *llmgwv1.AgentMessage_SlurmStatus:
|
||||
if h := s.slurmHandler(); h != nil {
|
||||
h.OnSlurmStatus(nodeID, m.SlurmStatus)
|
||||
}
|
||||
|
||||
case *llmgwv1.AgentMessage_TrainArtifact:
|
||||
if h := s.trainHandler(); h != nil {
|
||||
h.OnTrainArtifact(nodeID, m.TrainArtifact)
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
package agentsvc
|
||||
|
||||
import (
|
||||
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
|
||||
)
|
||||
|
||||
// SlurmHandler is told of the nodes' part in Slurm (BACKLOG P).
|
||||
type SlurmHandler interface {
|
||||
OnSlurmStatus(node string, st *llmgwv1.SlurmStatus)
|
||||
// OnRegister is a node's new stream, which is told again what it is.
|
||||
OnRegister(node string)
|
||||
}
|
||||
|
||||
// SetSlurm gives the service what Slurm's messages go to.
|
||||
func (s *Service) SetSlurm(h SlurmHandler) {
|
||||
s.mu.Lock()
|
||||
s.slurm = h
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
func (s *Service) slurmHandler() SlurmHandler {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
return s.slurm
|
||||
}
|
||||
@@ -0,0 +1,417 @@
|
||||
// Package slurmsvc is the leader's side of the Slurm cluster run beside
|
||||
// llmgw (BACKLOG P): it tells each node's agent what it is, writes the
|
||||
// controller's files from what the nodes report, and hands nodes between
|
||||
// llmgw and Slurm as the cluster's settings say.
|
||||
//
|
||||
// A node in slurm.lent is drained of llmgw's replicas, as `ctl drain` does,
|
||||
// and lent to Slurm once none is left. A node taken out of slurm.lent is
|
||||
// drained in Slurm, and is llmgw's again when its last job has ended. Both
|
||||
// are kept in force every tick, from the settings and what the agents say,
|
||||
// so a leader that takes over carries on where the last one was.
|
||||
package slurmsvc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"maps"
|
||||
"net"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/slurm"
|
||||
)
|
||||
|
||||
// Agents is how the service reaches the nodes.
|
||||
type Agents interface {
|
||||
Send(node string, msg *llmgwv1.ControlMessage) error
|
||||
ConnectedNodes() []string
|
||||
}
|
||||
|
||||
// Inventory is what the service reads and changes of llmgw's nodes.
|
||||
type Inventory interface {
|
||||
Get(id string) (inventory.Node, bool)
|
||||
SetState(id string, st inventory.NodeState) bool
|
||||
Allocations(nodeID string) []inventory.Allocation
|
||||
}
|
||||
|
||||
// Phase is where a node of slurm.nodes is between llmgw and Slurm.
|
||||
type Phase string
|
||||
|
||||
const (
|
||||
// Llmgw: drained in Slurm with no job left, llmgw's to place on.
|
||||
Llmgw Phase = "llmgw"
|
||||
// Lending: in slurm.lent, its replicas being moved off.
|
||||
Lending Phase = "lending"
|
||||
// Lent: Slurm's; llmgw places nothing there.
|
||||
Lent Phase = "lent"
|
||||
// Reclaiming: out of slurm.lent, drained in Slurm, jobs still running.
|
||||
Reclaiming Phase = "reclaiming"
|
||||
// Unknown: its agent has not said how it is in Slurm yet.
|
||||
Unknown Phase = "unknown"
|
||||
)
|
||||
|
||||
// NodeStatus is one node as the service sees it.
|
||||
type NodeStatus struct {
|
||||
Node string `json:"node"`
|
||||
Phase Phase `json:"phase"`
|
||||
Replicas int `json:"replicas"`
|
||||
Status *Reported `json:"slurm,omitempty"`
|
||||
Connected bool `json:"connected"`
|
||||
}
|
||||
|
||||
// Reported is the agent's SlurmStatus, for the admin API.
|
||||
type Reported struct {
|
||||
Version string `json:"version,omitempty"`
|
||||
Hardware string `json:"hardware,omitempty"`
|
||||
SlurmdRunning bool `json:"slurmd_running"`
|
||||
SlurmctldRunning bool `json:"slurmctld_running"`
|
||||
State string `json:"state,omitempty"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
Jobs int `json:"jobs"`
|
||||
Error string `json:"error,omitempty"`
|
||||
FilesSHA256 string `json:"files_sha256,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Status is the whole Slurm cluster, for `ctl slurm status`.
|
||||
type Status struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Cluster string `json:"cluster"`
|
||||
Controller string `json:"controller"`
|
||||
Partition string `json:"partition"`
|
||||
Ctld *Reported `json:"controller_status,omitempty"`
|
||||
Nodes []NodeStatus `json:"nodes"`
|
||||
// Waiting are nodes left out of slurm.conf until they report their
|
||||
// hardware.
|
||||
Waiting []string `json:"waiting,omitempty"`
|
||||
}
|
||||
|
||||
// Service keeps the Slurm cluster.
|
||||
type Service struct {
|
||||
settings func() config.SlurmSettings
|
||||
inv Inventory
|
||||
agents Agents
|
||||
key func(ctx context.Context) ([]byte, error)
|
||||
log *slog.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
// status is each node's last SlurmStatus.
|
||||
status map[string]*Reported
|
||||
// drained are nodes this service put in Draining, which it alone makes
|
||||
// ready again.
|
||||
drained map[string]bool
|
||||
// sent is what each node was last told, so it is told again only when
|
||||
// that changes or its stream is new.
|
||||
sent map[string]*llmgwv1.Slurm
|
||||
mungeKy []byte
|
||||
kick chan struct{}
|
||||
}
|
||||
|
||||
// New is a Service; key gives the munge key, made once from a cluster
|
||||
// secret.
|
||||
func New(settings func() config.SlurmSettings, inv Inventory, agents Agents,
|
||||
key func(ctx context.Context) ([]byte, error), log *slog.Logger) *Service {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &Service{settings: settings, inv: inv, agents: agents, key: key, log: log,
|
||||
status: map[string]*Reported{}, drained: map[string]bool{}, sent: map[string]*llmgwv1.Slurm{},
|
||||
kick: make(chan struct{}, 1)}
|
||||
}
|
||||
|
||||
// Every is how often the service looks again without being asked.
|
||||
var Every = 10 * time.Second
|
||||
|
||||
// Run keeps the cluster until ctx ends.
|
||||
func (s *Service) Run(ctx context.Context) {
|
||||
t := time.NewTicker(Every)
|
||||
defer t.Stop()
|
||||
for {
|
||||
s.Tick(ctx)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
case <-s.kick:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Kick has the service look again now: the settings changed.
|
||||
func (s *Service) Kick() {
|
||||
select {
|
||||
case s.kick <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// OnSlurmStatus records what a node's agent says.
|
||||
func (s *Service) OnSlurmStatus(node string, st *llmgwv1.SlurmStatus) {
|
||||
r := &Reported{Version: st.GetVersion(), Hardware: st.GetHardware(), SlurmdRunning: st.GetSlurmdRunning(),
|
||||
SlurmctldRunning: st.GetSlurmctldRunning(), State: st.GetState(), Reason: st.GetReason(),
|
||||
Jobs: int(st.GetJobs()), Error: st.GetError(), FilesSHA256: st.GetFilesSha256(), At: time.Now()}
|
||||
s.mu.Lock()
|
||||
prev := s.status[node]
|
||||
s.status[node] = r
|
||||
s.mu.Unlock()
|
||||
if r.Error != "" && (prev == nil || prev.Error != r.Error) {
|
||||
s.log.Warn("a node could not keep its part in Slurm", "node", node, "err", r.Error)
|
||||
}
|
||||
if prev == nil || prev.Hardware != r.Hardware || prev.State != r.State || prev.Jobs != r.Jobs {
|
||||
s.Kick()
|
||||
}
|
||||
}
|
||||
|
||||
// OnRegister is a node's new stream: it is told again what it is.
|
||||
func (s *Service) OnRegister(node string) {
|
||||
s.mu.Lock()
|
||||
delete(s.sent, node)
|
||||
s.mu.Unlock()
|
||||
s.Kick()
|
||||
}
|
||||
|
||||
// Tick does one pass: the nodes handed over as far as they can be, and
|
||||
// each told what it is.
|
||||
func (s *Service) Tick(ctx context.Context) {
|
||||
set := s.settings()
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
phases := s.handOver(set)
|
||||
var key []byte
|
||||
if set.Enabled {
|
||||
if s.mungeKy == nil {
|
||||
k, err := s.key(ctx)
|
||||
if err != nil {
|
||||
s.log.Warn("no munge key for Slurm; nothing is sent to the nodes", "err", err)
|
||||
return
|
||||
}
|
||||
s.mungeKy = k
|
||||
}
|
||||
key = s.mungeKy
|
||||
}
|
||||
files, _ := s.files(set)
|
||||
for _, node := range s.agents.ConnectedNodes() {
|
||||
m := s.message(set, node, phases[node] == Lent, key, files)
|
||||
if old, ok := s.sent[node]; ok && proto.Equal(old, m) {
|
||||
continue
|
||||
}
|
||||
if err := s.agents.Send(node, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_Slurm{Slurm: m}}); err != nil {
|
||||
s.log.Warn("could not tell a node its part in Slurm", "node", node, "err", err)
|
||||
continue
|
||||
}
|
||||
s.sent[node] = m
|
||||
}
|
||||
}
|
||||
|
||||
// handOver moves each node of slurm.nodes a step between llmgw and Slurm,
|
||||
// answering where each is. Under s.mu.
|
||||
func (s *Service) handOver(set config.SlurmSettings) map[string]Phase {
|
||||
out := map[string]Phase{}
|
||||
if !set.Enabled {
|
||||
// Off: every node is llmgw's again. What Slurm still runs is left
|
||||
// to finish where it is; turning Slurm off with work lent is the
|
||||
// operator's choice.
|
||||
for node := range s.drained {
|
||||
s.ready(node)
|
||||
}
|
||||
return out
|
||||
}
|
||||
for _, node := range set.Nodes {
|
||||
out[node] = s.phase(set, node)
|
||||
}
|
||||
// A node no longer in slurm.nodes is llmgw's.
|
||||
for node := range s.drained {
|
||||
if !slices.Contains(set.Nodes, node) {
|
||||
s.ready(node)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *Service) phase(set config.SlurmSettings, node string) Phase {
|
||||
n, ok := s.inv.Get(node)
|
||||
st := s.status[node]
|
||||
if set.IsLent(node) {
|
||||
if ok && (n.State == inventory.Ready || n.State == inventory.Cordoned) {
|
||||
s.inv.SetState(node, inventory.Draining)
|
||||
s.drained[node] = true
|
||||
s.log.Info("lending a node to Slurm: its replicas are moved off first", "node", node)
|
||||
}
|
||||
if !ok || len(s.inv.Allocations(node)) > 0 {
|
||||
return Lending
|
||||
}
|
||||
return Lent
|
||||
}
|
||||
// Not lent. Until the agent says Slurm has drained it with no job left,
|
||||
// llmgw places nothing new on it: a new leader starts with every node
|
||||
// ready. Cordoned, not drained: llmgw has nothing there to move.
|
||||
if st == nil || slurmBusy(st) {
|
||||
if ok && n.State == inventory.Ready {
|
||||
s.inv.SetState(node, inventory.Cordoned)
|
||||
s.drained[node] = true
|
||||
if st != nil {
|
||||
s.log.Info("taking a node back from Slurm: llmgw places nothing there until its jobs end",
|
||||
"node", node, "jobs", st.Jobs)
|
||||
}
|
||||
}
|
||||
if st == nil {
|
||||
return Unknown
|
||||
}
|
||||
return Reclaiming
|
||||
}
|
||||
if s.drained[node] {
|
||||
s.ready(node)
|
||||
}
|
||||
return Llmgw
|
||||
}
|
||||
|
||||
// slurmBusy is whether Slurm may still run work on a node: a job step left
|
||||
// on it (or no telling), or a controller that would still give it one. A
|
||||
// node the controller does not know of, or cannot be asked about, gets no
|
||||
// job from it, so a Slurm that is broken never keeps a node from llmgw.
|
||||
func slurmBusy(st *Reported) bool {
|
||||
if st.Jobs != 0 {
|
||||
return true
|
||||
}
|
||||
if st.State == "" {
|
||||
return false
|
||||
}
|
||||
return !strings.Contains(st.State, "DRAIN") && !strings.Contains(st.State, "DOWN")
|
||||
}
|
||||
|
||||
func (s *Service) ready(node string) {
|
||||
if n, ok := s.inv.Get(node); ok && (n.State == inventory.Draining || n.State == inventory.Cordoned) {
|
||||
s.inv.SetState(node, inventory.Ready)
|
||||
s.log.Info("a node is llmgw's again: Slurm has no job there", "node", node)
|
||||
}
|
||||
delete(s.drained, node)
|
||||
}
|
||||
|
||||
// files are the controller's files, and the nodes left out of them for not
|
||||
// having reported their hardware. Under s.mu.
|
||||
func (s *Service) files(set config.SlurmSettings) (map[string]string, []string) {
|
||||
ctl, ok := s.inv.Get(set.Controller)
|
||||
if !set.Enabled || !ok {
|
||||
return nil, nil
|
||||
}
|
||||
c := slurm.Cluster{Name: set.Cluster(), Controller: set.Controller, ControllerAddr: ctl.MgmtAddr,
|
||||
Partition: set.PartitionName()}
|
||||
var waiting []string
|
||||
for _, node := range set.Nodes {
|
||||
n, ok := s.inv.Get(node)
|
||||
st := s.status[node]
|
||||
if !ok || st == nil || st.Hardware == "" {
|
||||
waiting = append(waiting, node)
|
||||
continue
|
||||
}
|
||||
hw, err := slurm.ParseHardware(st.Hardware)
|
||||
if err != nil {
|
||||
waiting = append(waiting, node)
|
||||
continue
|
||||
}
|
||||
sn := slurm.Node{Name: node, Addr: n.MgmtAddr, Hardware: hw}
|
||||
for _, g := range n.GPUs {
|
||||
sn.GPUs = append(sn.GPUs, slurm.GPU{Minor: g.Index, Name: g.Name})
|
||||
}
|
||||
c.Nodes = append(c.Nodes, sn)
|
||||
}
|
||||
return c.Files(), waiting
|
||||
}
|
||||
|
||||
// message is what a node is told. Under s.mu.
|
||||
func (s *Service) message(set config.SlurmSettings, node string, lent bool, key []byte,
|
||||
files map[string]string) *llmgwv1.Slurm {
|
||||
m := &llmgwv1.Slurm{}
|
||||
if !set.Enabled {
|
||||
return m
|
||||
}
|
||||
isNode := slices.Contains(set.Nodes, node)
|
||||
isCtl := set.Controller == node && files != nil
|
||||
if !isNode && !isCtl {
|
||||
return m
|
||||
}
|
||||
m.Name, m.MungeKey, m.Lent = node, key, lent
|
||||
m.Node = isNode
|
||||
if ctl, ok := s.inv.Get(set.Controller); ok {
|
||||
m.ConfServer = net.JoinHostPort(ctl.MgmtAddr, strconv.Itoa(slurm.CtldPort))
|
||||
}
|
||||
if isNode && m.ConfServer == "" {
|
||||
// Without its controller a node cannot start; it is told when the
|
||||
// controller registers.
|
||||
m.Node = false
|
||||
}
|
||||
if isCtl {
|
||||
m.Controller = true
|
||||
m.Files = maps.Clone(files)
|
||||
}
|
||||
if !m.Node && !m.Controller {
|
||||
return &llmgwv1.Slurm{}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// Status is the cluster as the service sees it now.
|
||||
func (s *Service) Status() Status {
|
||||
set := s.settings()
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := Status{Enabled: set.Enabled, Cluster: set.Cluster(), Controller: set.Controller,
|
||||
Partition: set.PartitionName(), Ctld: s.status[set.Controller]}
|
||||
if !set.Enabled {
|
||||
return out
|
||||
}
|
||||
_, out.Waiting = s.files(set)
|
||||
conn := s.agents.ConnectedNodes()
|
||||
for _, node := range set.Nodes {
|
||||
ns := NodeStatus{Node: node, Status: s.status[node], Replicas: len(s.inv.Allocations(node)),
|
||||
Connected: slices.Contains(conn, node)}
|
||||
ns.Phase = s.peek(set, node)
|
||||
out.Nodes = append(out.Nodes, ns)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// peek is phase without changing anything. Under s.mu.
|
||||
func (s *Service) peek(set config.SlurmSettings, node string) Phase {
|
||||
st := s.status[node]
|
||||
if set.IsLent(node) {
|
||||
if len(s.inv.Allocations(node)) > 0 {
|
||||
return Lending
|
||||
}
|
||||
return Lent
|
||||
}
|
||||
switch {
|
||||
case st == nil:
|
||||
return Unknown
|
||||
case slurmBusy(st):
|
||||
return Reclaiming
|
||||
}
|
||||
return Llmgw
|
||||
}
|
||||
|
||||
// Describe is one line for a log or a screen.
|
||||
func (n NodeStatus) Describe() string {
|
||||
switch n.Phase {
|
||||
case Lending:
|
||||
return fmt.Sprintf("being lent: %d replica(s) still to move off", n.Replicas)
|
||||
case Lent:
|
||||
return "lent to Slurm"
|
||||
case Reclaiming:
|
||||
if n.Status != nil {
|
||||
return fmt.Sprintf("being taken back: %d job(s) still running", n.Status.Jobs)
|
||||
}
|
||||
return "being taken back"
|
||||
case Llmgw:
|
||||
return "llmgw's"
|
||||
}
|
||||
return "not heard from yet"
|
||||
}
|
||||
@@ -0,0 +1,243 @@
|
||||
package slurmsvc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
|
||||
)
|
||||
|
||||
type fakeInv struct {
|
||||
nodes map[string]*inventory.Node
|
||||
allocs map[string]int
|
||||
}
|
||||
|
||||
func (f *fakeInv) Get(id string) (inventory.Node, bool) {
|
||||
n, ok := f.nodes[id]
|
||||
if !ok {
|
||||
return inventory.Node{}, false
|
||||
}
|
||||
return *n, true
|
||||
}
|
||||
|
||||
func (f *fakeInv) SetState(id string, st inventory.NodeState) bool {
|
||||
n, ok := f.nodes[id]
|
||||
if ok {
|
||||
n.State = st
|
||||
}
|
||||
return ok
|
||||
}
|
||||
|
||||
func (f *fakeInv) Allocations(id string) []inventory.Allocation {
|
||||
return make([]inventory.Allocation, f.allocs[id])
|
||||
}
|
||||
|
||||
type fakeAgents struct {
|
||||
conn []string
|
||||
got map[string][]*llmgwv1.Slurm
|
||||
}
|
||||
|
||||
func (f *fakeAgents) Send(node string, m *llmgwv1.ControlMessage) error {
|
||||
f.got[node] = append(f.got[node], m.GetSlurm())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeAgents) ConnectedNodes() []string { return f.conn }
|
||||
|
||||
func (f *fakeAgents) last(node string) *llmgwv1.Slurm {
|
||||
ms := f.got[node]
|
||||
if len(ms) == 0 {
|
||||
return nil
|
||||
}
|
||||
return ms[len(ms)-1]
|
||||
}
|
||||
|
||||
type rig struct {
|
||||
set config.SlurmSettings
|
||||
inv *fakeInv
|
||||
agents *fakeAgents
|
||||
svc *Service
|
||||
}
|
||||
|
||||
func newRig() *rig {
|
||||
r := &rig{
|
||||
set: config.SlurmSettings{Enabled: true, Controller: "node06", Nodes: []string{"cuda01"}},
|
||||
inv: &fakeInv{nodes: map[string]*inventory.Node{
|
||||
"node06": {ID: "node06", MgmtAddr: "10.0.0.146"},
|
||||
"cuda01": {ID: "cuda01", MgmtAddr: "10.0.0.147", GPUs: []discovery.GPU{{Index: 0, Name: "NVIDIA GeForce RTX 3090"}}},
|
||||
}, allocs: map[string]int{"cuda01": 2}},
|
||||
agents: &fakeAgents{conn: []string{"node06", "cuda01"}, got: map[string][]*llmgwv1.Slurm{}},
|
||||
}
|
||||
r.svc = New(func() config.SlurmSettings { return r.set }, r.inv, r.agents,
|
||||
func(context.Context) ([]byte, error) { return []byte(strings.Repeat("k", 1024)), nil }, nil)
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *rig) report(node string, st *llmgwv1.SlurmStatus) { r.svc.OnSlurmStatus(node, st) }
|
||||
|
||||
func (r *rig) tick() { r.svc.Tick(context.Background()) }
|
||||
|
||||
const hw = "NodeName=cuda01 CPUs=16 RealMemory=64000"
|
||||
|
||||
func TestMessagesAndFiles(t *testing.T) {
|
||||
r := newRig()
|
||||
r.tick()
|
||||
ctl, node := r.agents.last("node06"), r.agents.last("cuda01")
|
||||
if !ctl.GetController() || ctl.GetNode() || len(ctl.GetMungeKey()) != 1024 {
|
||||
t.Fatalf("controller told %v", ctl)
|
||||
}
|
||||
if !node.GetNode() || node.GetController() || node.GetConfServer() != "10.0.0.146:6817" || node.GetLent() ||
|
||||
node.GetName() != "cuda01" || len(node.GetFiles()) != 0 {
|
||||
t.Fatalf("node told %v", node)
|
||||
}
|
||||
if strings.Contains(ctl.GetFiles()["slurm.conf"], "NodeName=cuda01") {
|
||||
t.Fatal("a node with no hardware reported is in slurm.conf")
|
||||
}
|
||||
if w := r.svc.Status().Waiting; len(w) != 1 || w[0] != "cuda01" {
|
||||
t.Fatalf("waiting %v", w)
|
||||
}
|
||||
// Same again: nobody is told twice.
|
||||
r.tick()
|
||||
if len(r.agents.got["node06"]) != 1 || len(r.agents.got["cuda01"]) != 1 {
|
||||
t.Fatal("told again with nothing changed")
|
||||
}
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "IDLE+DRAIN", Reason: "llmgw: not lent yet"})
|
||||
r.tick()
|
||||
conf := r.agents.last("node06").GetFiles()["slurm.conf"]
|
||||
if !strings.Contains(conf, "NodeName=cuda01 NodeAddr=10.0.0.147 CPUs=16 RealMemory=62080 Gres=gpu:rtx3090:1") {
|
||||
t.Fatalf("slurm.conf:\n%s", conf)
|
||||
}
|
||||
if !strings.Contains(r.agents.last("node06").GetFiles()["gres.conf"], "File=/dev/nvidia0") {
|
||||
t.Fatal("gres.conf has no card")
|
||||
}
|
||||
if len(r.agents.got["cuda01"]) != 1 {
|
||||
t.Fatal("a node was told again when only the controller's files changed")
|
||||
}
|
||||
// A new stream is told again.
|
||||
r.svc.OnRegister("cuda01")
|
||||
r.tick()
|
||||
if len(r.agents.got["cuda01"]) != 2 {
|
||||
t.Fatal("a new stream was not told")
|
||||
}
|
||||
if r.inv.nodes["cuda01"].State != inventory.Ready {
|
||||
t.Fatal("a node drained in Slurm with no job was taken from llmgw")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLendAndReclaim(t *testing.T) {
|
||||
r := newRig()
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "IDLE+DRAIN", Reason: "llmgw: not lent yet"})
|
||||
r.set.Lent = []string{"cuda01"}
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State != inventory.Draining {
|
||||
t.Fatal("a lent node was not drained of its replicas")
|
||||
}
|
||||
if r.agents.last("cuda01").GetLent() {
|
||||
t.Fatal("lent to Slurm with replicas still on it")
|
||||
}
|
||||
if p := r.svc.Status().Nodes[0].Phase; p != Lending {
|
||||
t.Fatalf("phase %s", p)
|
||||
}
|
||||
r.inv.allocs["cuda01"] = 0
|
||||
r.tick()
|
||||
if !r.agents.last("cuda01").GetLent() || r.svc.Status().Nodes[0].Phase != Lent {
|
||||
t.Fatal("not lent once its replicas were gone")
|
||||
}
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "ALLOCATED", Jobs: 3})
|
||||
|
||||
// Taken back: Slurm drains it; llmgw keeps off until its jobs end.
|
||||
r.set.Lent = nil
|
||||
r.tick()
|
||||
if r.agents.last("cuda01").GetLent() {
|
||||
t.Fatal("still lent")
|
||||
}
|
||||
if s := r.inv.nodes["cuda01"].State; s != inventory.Draining && s != inventory.Cordoned {
|
||||
t.Fatalf("llmgw may place on a node with Slurm jobs: %v", s)
|
||||
}
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "ALLOCATED+DRAIN", Jobs: 1})
|
||||
r.tick()
|
||||
if r.svc.Status().Nodes[0].Phase != Reclaiming || r.inv.nodes["cuda01"].State == inventory.Ready {
|
||||
t.Fatal("given back with a job left")
|
||||
}
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "IDLE+DRAIN", Reason: "llmgw: serving models"})
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State != inventory.Ready || r.svc.Status().Nodes[0].Phase != Llmgw {
|
||||
t.Fatal("not llmgw's when its last job ended")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewLeaderKeepsOffUntilHeard(t *testing.T) {
|
||||
r := newRig()
|
||||
r.inv.allocs["cuda01"] = 0
|
||||
r.tick()
|
||||
// Nothing heard yet: cordoned, nothing moved.
|
||||
if r.inv.nodes["cuda01"].State != inventory.Cordoned {
|
||||
t.Fatalf("state %v before the agent said anything", r.inv.nodes["cuda01"].State)
|
||||
}
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "MIXED", Jobs: 2})
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State == inventory.Ready {
|
||||
t.Fatal("ready with Slurm jobs on it")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBrokenSlurmLeavesTheNodeToLlmgw(t *testing.T) {
|
||||
r := newRig()
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Error: "installing Slurm: no package", Jobs: 0})
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State != inventory.Ready {
|
||||
t.Fatal("a Slurm that does not work took a node from llmgw")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOperatorsCordonIsLeft(t *testing.T) {
|
||||
r := newRig()
|
||||
r.inv.nodes["cuda01"].State = inventory.Cordoned
|
||||
r.report("cuda01", &llmgwv1.SlurmStatus{Hardware: hw, State: "IDLE+DRAIN"})
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State != inventory.Cordoned {
|
||||
t.Fatal("llmgw uncordoned a node an operator cordoned")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOffAndOut(t *testing.T) {
|
||||
r := newRig()
|
||||
r.inv.allocs["cuda01"] = 0
|
||||
r.set.Lent = []string{"cuda01"}
|
||||
r.tick()
|
||||
r.set.Enabled = false
|
||||
r.tick()
|
||||
if m := r.agents.last("cuda01"); m.GetNode() || m.GetController() {
|
||||
t.Fatalf("told %v with Slurm off", m)
|
||||
}
|
||||
if r.inv.nodes["cuda01"].State != inventory.Ready {
|
||||
t.Fatal("not llmgw's with Slurm off")
|
||||
}
|
||||
|
||||
r = newRig()
|
||||
r.inv.allocs["cuda01"] = 0
|
||||
r.set.Lent = []string{"cuda01"}
|
||||
r.tick()
|
||||
r.set.Nodes, r.set.Lent = nil, nil
|
||||
r.tick()
|
||||
if r.inv.nodes["cuda01"].State != inventory.Ready || r.agents.last("cuda01").GetNode() {
|
||||
t.Fatal("a node taken out of slurm.nodes is not llmgw's")
|
||||
}
|
||||
if !r.agents.last("node06").GetController() {
|
||||
t.Fatal("the controller stopped with no nodes")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoControllerYet(t *testing.T) {
|
||||
r := newRig()
|
||||
delete(r.inv.nodes, "node06")
|
||||
r.agents.conn = []string{"cuda01"}
|
||||
r.tick()
|
||||
if m := r.agents.last("cuda01"); m.GetNode() {
|
||||
t.Fatalf("a node was started with no controller to fetch its files from: %v", m)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user