Slurm's ports for the cluster only, and srun in a range of its own

slurm.conf sets SrunPortRange=60001-63000, so a site's firewall can let
srun's step connections in. A node given a part in Slurm adds 6817-6818
and that range to llmgw's nftables table, open to the cluster's networks
and closed to everyone else, as the engines' ports are.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -41,6 +41,15 @@ APIs change between minors without a deprecation period.
   organization's name and the rest of the cluster stay with the cluster's
   administrators; an admin cannot take away their own role. Every change is in
   the audit log.
+- **Slurm beside llmgw, for training and batch work.** `llmgw ctl slurm setup
+  --controller NODE --nodes NODE,...` (or `--all-nodes`, or `server init
+  --slurm`) has the agents install munge and Slurm from the machines' package
+  sources: slurmctld on one node, slurmd configless on the others, slurm.conf
+  and gres.conf written from what each node reports. Every node stays llmgw's,
+  drained in Slurm, until `ctl slurm lend NODE`: its replicas move off, then
+  Slurm runs jobs there. `ctl slurm reclaim NODE` takes it back once its jobs
+  end. `ctl slurm status` shows each node. Slurm's ports (6817–6818, srun's
+  60001–63000) answer the cluster's networks only, as llmgw's own do.
 
 ### Changed
 

docs/install.md

@@ -184,7 +184,9 @@ takes it back once its jobs end. Jobs are submitted on this machine with
 `sbatch` and `srun`; `llmgw ctl slurm status` shows where each node is. It
 needs this machine to be a node, and its hostname to be a name Slurm takes.
 On a cluster installed without it, `llmgw ctl slurm setup --controller NODE
---all-nodes` does the same later.
+--all-nodes` does the same later. Slurm's ports (6817–6818, and srun's 60001–63000) answer the cluster's
+networks only, as llmgw's own do; a firewall of the site's own between the
+nodes must let them through.
 
 **Secure by default.** What init writes is the hardened configuration, not a
 demonstration one:
@@ -3158,7 +3160,7 @@ loaded at boot by `llmgw-firewall.service`):
 | Machine | Ports, for the cluster only |
 | --- | --- |
 | control plane | 7443 (agents), metrics when not on loopback, 8081 (its mirror), engines and 9100 when it is a node, 2379/2380 when it holds the store's third etcd vote |
-| node | engine ports (`control.port_range`, 20000–20999), 9100 |
+| node | engine ports (`control.port_range`, 20000–20999), 9100; with a part in Slurm also 6817–6818 (slurmctld, slurmd) and 60001–63000 (srun's `SrunPortRange`) |
 | store member | 5432, 8008, 2379, 2380, 9100 |
 | mirror | 8081, 8082, 9100 |
 

internal/agent/agent.go

@@ -98,7 +98,10 @@ type Agent struct {
 	// (nil: this machine's shell).
 	fwMu     sync.Mutex
 	fwRunner firewall.Runner
-	inv      discovery.Inventory
+	// fwLast is the last firewall message, applied again when the node
+	// is given a part in Slurm. Under fwMu.
+	fwLast *llmgwv1.Firewall
+	inv    discovery.Inventory
 	// shardReplica remembers which replica a shard belongs to, so a status
 	// report can name both. The control plane keys on the replica.
 	shardReplica map[string]string
@@ -961,7 +964,7 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
 	case *llmgwv1.ControlMessage_TrainStart:
 		go a.trainJob(ctx, m.TrainStart, out)
 	case *llmgwv1.ControlMessage_Slurm:
-		a.slurm.Apply(m.Slurm)
+		a.slurmMessage(m.Slurm)
 	case *llmgwv1.ControlMessage_TrainCancel:
 		a.trainer.Cancel(m.TrainCancel.GetJobId())
 	case *llmgwv1.ControlMessage_RestartAgent:

internal/agent/firewall.go

@@ -10,8 +10,16 @@ import (
 
 	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/firewall"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/slurm"
 )
 
+// slurmPorts are Slurm's on a node that runs any of it: slurmctld's and
+// slurmd's, and srun's range for the steps it starts.
+var slurmPorts = []firewall.Port{
+	{From: slurm.CtldPort, To: slurm.SlurmdPort, What: "slurm: slurmctld and slurmd"},
+	{From: slurm.SrunPortsFrom, To: slurm.SrunPortsTo, What: "slurm: srun, for the steps it starts"},
+}
+
 // applyFirewall keeps llmgw's firewall on this node as the cluster's
 // settings say: the engines' ports and the node exporter's for the networks
 // the node reaches its control planes on, and those named, or no table at
@@ -19,6 +27,7 @@ import (
 func (a *Agent) applyFirewall(m *llmgwv1.Firewall) {
 	a.fwMu.Lock()
 	defer a.fwMu.Unlock()
+	a.fwLast = m
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 	defer cancel()
 	r := a.fwRunner
@@ -55,6 +64,11 @@ func (a *Agent) applyFirewall(m *llmgwv1.Firewall) {
 	if from > 0 && to >= from {
 		ports = append(ports, firewall.Engines(from, to))
 	}
+	// Once there, they stay: the ruleset keeps every port it held, so a
+	// node that leaves Slurm keeps them closed to everyone but the cluster.
+	if a.slurm != nil && a.slurm.HasPart() {
+		ports = append(ports, slurmPorts...)
+	}
 	changed, err := firewall.Install(ctx, r, firewall.Spec{Ports: ports, Allow: nets})
 	switch {
 	case err != nil:
@@ -64,6 +78,22 @@ func (a *Agent) applyFirewall(m *llmgwv1.Firewall) {
 	}
 }
 
+// slurmMessage is the leader's word on this node's part in Slurm. A node
+// that gets one has the firewall applied again, with Slurm's ports.
+func (a *Agent) slurmMessage(m *llmgwv1.Slurm) {
+	had := a.slurm.HasPart()
+	a.slurm.Apply(m)
+	if had || !a.slurm.HasPart() {
+		return
+	}
+	a.fwMu.Lock()
+	last := a.fwLast
+	a.fwMu.Unlock()
+	if last != nil && last.GetEnabled() {
+		go a.applyFirewall(last)
+	}
+}
+
 // controlPlaneIPs are the addresses of the control planes an agent dials.
 func controlPlaneIPs(ctx context.Context, servers []string) []string {
 	var out []string

internal/agent/firewall_test.go

@@ -6,8 +6,10 @@ import (
 	"log/slog"
 	"strings"
 	"testing"
+	"time"
 
 	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/slurmnode"
 )
 
 // fakeShell answers the firewall package's scripts and records what was
@@ -55,3 +57,28 @@ func TestANodeKeepsOrDropsItsFirewallAsTheClusterSays(t *testing.T) {
 		t.Error("turned off, the table stayed")
 	}
 }
+
+func TestANodeGivenAPartInSlurmLetsSlurmInFromTheClusterOnly(t *testing.T) {
+	sh := &fakeShell{}
+	a := &Agent{log: slog.New(slog.DiscardHandler), fwRunner: sh, serverList: []string{"https://10.0.0.140:7443"},
+		slurm: slurmnode.New(nil, nil, "", slog.New(slog.DiscardHandler))}
+	a.applyFirewall(&llmgwv1.Firewall{Enabled: true, EnginePortsFrom: 20000, EnginePortsTo: 20999})
+	if strings.Contains(sh.installed, "6817") {
+		t.Fatalf("Slurm's ports before the node has a part in it:\n%s", sh.installed)
+	}
+
+	a.slurmMessage(&llmgwv1.Slurm{Name: "node05", Node: true, ConfServer: "10.0.0.148:6817"})
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		a.fwMu.Lock()
+		got := sh.installed
+		a.fwMu.Unlock()
+		if strings.Contains(got, "6817-6818") && strings.Contains(got, "60001-63000") {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("given a part in Slurm, the rules still lack its ports:\n%s", got)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}

internal/agent/slurmnode/slurmnode.go

@@ -101,6 +101,13 @@ func (k *Keeper) Apply(m *llmgwv1.Slurm) {
 	}
 }
 
+// HasPart is whether the last message gives this machine a part in Slurm.
+func (k *Keeper) HasPart() bool {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	return k.want.GetController() || k.want.GetNode()
+}
+
 // Resend has the status sent again on the next pass, which it starts now.
 func (k *Keeper) Resend() {
 	k.mu.Lock()

internal/slurm/slurm.go

@@ -27,6 +27,13 @@ const (
 	// CtldPort is slurmctld's, which configless slurmd fetch their files
 	// from.
 	CtldPort = 6817
+	// SlurmdPort is slurmd's, which the controller and srun reach.
+	SlurmdPort = 6818
+	// SrunPortsFrom and SrunPortsTo are where srun listens for the steps it
+	// starts: a range of its own, so a firewall can let it in (Slurm's own
+	// example range). Without one srun takes any ephemeral port.
+	SrunPortsFrom = 60001
+	SrunPortsTo   = 63000
 )
 
 // ReasonNew is why a node new to slurm.conf starts drained. Every reason
@@ -189,6 +196,7 @@ func (c Cluster) SlurmConf() string {
 	b.WriteString(`SlurmctldParameters=enable_configless
 SlurmctldPort=6817
 SlurmdPort=6818
+SrunPortRange=60001-63000
 AuthType=auth/munge
 CredType=cred/munge
 SlurmUser=slurm