diff --git a/internal/agent/agent.go b/internal/agent/agent.go index d1f1c2ae..e094687c 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -34,8 +34,18 @@ import ( // Config configures the agent. type Config struct { - NodeID string `koanf:"node_id"` + NodeID string `koanf:"node_id"` + // Server is the control plane, and Servers is the same thing when there is + // more than one. Both are accepted: every configuration written before + // there could be a second control plane names one address, and none of + // them should have to change. + // + // The order is tried in order, and a refusal moves to the next. Only the + // leader accepts an agent, so a standby is skipped in the time it takes to + // be told so — which is what makes failover a reconnect rather than an + // edit to every node's configuration. Server string `koanf:"server"` + Servers []string `koanf:"servers"` Hostname string `koanf:"hostname"` MgmtAddr string `koanf:"mgmt_addr"` Labels map[string]string `koanf:"labels"` @@ -108,9 +118,6 @@ func (c *Config) normalize() error { if c.NodeID == "" { return errors.New("node_id is required") } - if c.Server == "" { - return errors.New("server is required") - } if err := c.TLS.Validate(); err != nil { return err } @@ -140,26 +147,59 @@ func (c *Config) normalize() error { // that says plaintext while TLS is configured. Silently connecting anyway would // mean a cluster believing it has mutual authentication when it has none. func (c *Config) normalizeServerURL() error { - secure := c.TLS.Enabled() - switch { - case strings.HasPrefix(c.Server, "https://"): - if !secure { - return errors.New("server is an https:// address but no tls certificates are configured") - } - case strings.HasPrefix(c.Server, "http://"): - if secure { - return errors.New("tls is configured but server is an http:// address; use https://") - } - case strings.Contains(c.Server, "://"): - return fmt.Errorf("server %q must be an http:// or https:// address", c.Server) - case secure: - c.Server = "https://" + c.Server - default: - c.Server = "http://" + c.Server + // One list, however it was written. Server first when both are given, + // because an operator adding a standby to an existing file expects the + // one already there to stay preferred. + all := make([]string, 0, 1+len(c.Servers)) + if c.Server != "" { + all = append(all, c.Server) } + all = append(all, c.Servers...) + + seen := map[string]struct{}{} + out := make([]string, 0, len(all)) + for _, s := range all { + fixed, err := normalizeOne(s, c.TLS.Enabled()) + if err != nil { + return err + } + // A duplicate would be tried twice in a row on failover, which reads + // as the rotation not working. + if _, dup := seen[fixed]; dup { + continue + } + seen[fixed] = struct{}{} + out = append(out, fixed) + } + if len(out) == 0 { + return errors.New("server is required") + } + c.Servers = out + c.Server = out[0] return nil } +// normalizeOne applies the scheme rules to a single address. +func normalizeOne(addr string, secure bool) (string, error) { + switch { + case strings.HasPrefix(addr, "https://"): + if !secure { + return "", fmt.Errorf("%s is an https:// address but no tls certificates are configured", addr) + } + case strings.HasPrefix(addr, "http://"): + if secure { + return "", fmt.Errorf("tls is configured but %s is an http:// address; use https://", addr) + } + case strings.Contains(addr, "://"): + return "", fmt.Errorf("server %q must be an http:// or https:// address", addr) + case secure: + return "https://" + addr, nil + default: + return "http://" + addr, nil + } + return addr, nil +} + // Agent runs on a GPU node. type Agent struct { cfg Config @@ -230,15 +270,30 @@ func (a *Agent) Run(ctx context.Context) error { a.recoverOrphans() attempt := 0 + next := 0 for ctx.Err() == nil { - err := a.session(ctx) + server := a.cfg.Servers[next%len(a.cfg.Servers)] + err := a.session(ctx, server) if ctx.Err() != nil { break } + // On to the next address whatever went wrong. A standby refuses, a + // stopped control plane refuses the connection, and a crashed one + // never answers — from here they are the same thing, and trying + // somewhere else is the answer to all three. + next++ attempt++ delay := a.reconnectDelay(attempt) + // No wait when there is somewhere else to try: the delay exists to + // stop hammering one control plane, and the next address has not been + // hammered. With a single address this is the old behaviour exactly. + if len(a.cfg.Servers) > 1 && next%len(a.cfg.Servers) != 0 { + delay = 0 + } a.log.Warn("disconnected from the control plane, retrying", - "err", err, "in", delay.Round(time.Millisecond), "attempt", attempt) + "err", err, "server", server, + "next", a.cfg.Servers[next%len(a.cfg.Servers)], + "in", delay.Round(time.Millisecond), "attempt", attempt) select { case <-ctx.Done(): @@ -469,8 +524,8 @@ func (a *Agent) scan(ctx context.Context) error { } // session runs one connection until it breaks. -func (a *Agent) session(ctx context.Context) error { - client := llmgwv1connect.NewAgentServiceClient(a.http, a.cfg.Server) +func (a *Agent) session(ctx context.Context, server string) error { + client := llmgwv1connect.NewAgentServiceClient(a.http, server) // parent is kept because the derived context is cancelled from inside this // function; asking the derived one whether we are shutting down would diff --git a/internal/agent/failover_test.go b/internal/agent/failover_test.go new file mode 100644 index 00000000..96a524aa --- /dev/null +++ b/internal/agent/failover_test.go @@ -0,0 +1,107 @@ +package agent + +import ( + "context" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "golang.org/x/net/http2" + "golang.org/x/net/http2/h2c" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" +) + +type standby bool + +func (s standby) IsLeader() bool { return !bool(s) } + +// startPlain runs an agent service over h2c, which is enough to test who +// accepts a connection and who refuses it. +// +// A nil leader means none was configured, which is every deployment that +// existed before leadership did and has to keep accepting agents. +func startPlain(t *testing.T, leader agentsvc.Leader) (*inventory.Store, string) { + t.Helper() + inv := inventory.NewStore() + svc := agentsvc.New(inv, newRecorder(), false, slog.New(slog.DiscardHandler)) + if leader != nil { + svc.SetLeadership(leader) + } + + path, handler := svc.Handler() + mux := http.NewServeMux() + mux.Handle(path, handler) + // h2c, because the agent speaks HTTP/2 with prior knowledge and a plain + // server negotiates HTTP/1.1 — which turns a clear refusal into a frame + // parsing error that says nothing about why. + srv := httptest.NewServer(h2c.NewHandler(mux, &http2.Server{})) + t.Cleanup(srv.Close) + return inv, srv.URL +} + +func TestAStandbyRefusesAnAgent(t *testing.T) { + // Commands travel down the stream the agent opened, so whichever control + // plane holds it is the only one that can start anything on that node. A + // standby holding it means the leader's placements fail with "no agent + // connected" for a node that is plainly connected. + _, addr := startPlain(t, standby(true)) + + ag := runAgent(t, Config{NodeID: "n1", Server: addr, MgmtAddr: "127.0.0.1"}) + err := ag.session(context.Background(), ag.cfg.Server) + if err == nil { + t.Fatal("a standby accepted an agent") + } + if !strings.Contains(err.Error(), "standby") { + t.Errorf("the refusal does not say why: %v", err) + } +} + +func TestAnAgentReachesTheLeaderPastAStandby(t *testing.T) { + // The point of the list. The standby is first, so the agent only registers + // if it moved on rather than sitting on the refusal. + _, standbyAddr := startPlain(t, standby(true)) + leaderInv, leaderAddr := startPlain(t, standby(false)) + + ag := runAgent(t, Config{ + NodeID: "n1", MgmtAddr: "127.0.0.1", + Servers: []string{standbyAddr, leaderAddr}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + go ag.Run(ctx) + + deadline := time.Now().Add(8 * time.Second) + for time.Now().Before(deadline) { + if _, ok := leaderInv.Get("n1"); ok { + return + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("the agent never reached the leader past the standby") +} + +func TestALeaderStillAcceptsWithNoLeadershipConfigured(t *testing.T) { + // Every deployment before this had no leadership at all, and must keep + // accepting agents with nothing configured. + inv, addr := startPlain(t, nil) + + ag := runAgent(t, Config{NodeID: "n1", Server: addr, MgmtAddr: "127.0.0.1"}) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + go ag.Run(ctx) + + deadline := time.Now().Add(4 * time.Second) + for time.Now().Before(deadline) { + if _, ok := inv.Get("n1"); ok { + return + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("a service with no leadership configured refused an agent") +} diff --git a/internal/agent/mtls_test.go b/internal/agent/mtls_test.go index 5b173207..fda74f9c 100644 --- a/internal/agent/mtls_test.go +++ b/internal/agent/mtls_test.go @@ -255,7 +255,7 @@ func TestRefusedAgentLearnsWhy(t *testing.T) { t.Fatal(err) } - err = ag.session(context.Background()) + err = ag.session(context.Background(), ag.cfg.Server) if err == nil { t.Fatal("a refused session reported success") } diff --git a/internal/agent/servers_test.go b/internal/agent/servers_test.go new file mode 100644 index 00000000..ae9c940c --- /dev/null +++ b/internal/agent/servers_test.go @@ -0,0 +1,90 @@ +package agent + +import ( + "slices" + "strings" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" +) + +func normalized(t *testing.T, c Config) Config { + t.Helper() + if err := c.normalizeServerURL(); err != nil { + t.Fatalf("normalising %+v: %v", c, err) + } + return c +} + +func TestOneServerStillWorksExactlyAsBefore(t *testing.T) { + // Every agent configuration written before there could be a second control + // plane names one address, and none of them should have to change. + c := normalized(t, Config{Server: "10.0.0.140:7443"}) + if !slices.Equal(c.Servers, []string{"http://10.0.0.140:7443"}) { + t.Errorf("servers = %v", c.Servers) + } + if c.Server != "http://10.0.0.140:7443" { + t.Errorf("server = %q", c.Server) + } +} + +func TestTheAddressAlreadyThereStaysPreferred(t *testing.T) { + // An operator adding a standby to an existing file expects the control + // plane that was already working to keep being tried first. + c := normalized(t, Config{ + Server: "first:7443", + Servers: []string{"second:7443", "third:7443"}, + }) + want := []string{"http://first:7443", "http://second:7443", "http://third:7443"} + if !slices.Equal(c.Servers, want) { + t.Errorf("servers = %v, want %v", c.Servers, want) + } +} + +func TestADuplicateAddressIsNotTriedTwiceInARow(t *testing.T) { + // Which would look exactly like the rotation not working. + c := normalized(t, Config{ + Server: "a:7443", + Servers: []string{"http://a:7443", "b:7443"}, + }) + want := []string{"http://a:7443", "http://b:7443"} + if !slices.Equal(c.Servers, want) { + t.Errorf("servers = %v, want %v", c.Servers, want) + } +} + +func TestEveryAddressIsCheckedAgainstTheTLSSetting(t *testing.T) { + // The check existed for one address and has to hold for all of them: a + // standby reached over plaintext while the leader is authenticated would + // be a way around mutual TLS, not a fallback. + c := Config{ + Servers: []string{"https://a:7443", "http://b:7443"}, + TLS: tlsconf.Config{CA: "ca.pem", Cert: "c.pem", Key: "k.pem"}, + } + err := c.normalizeServerURL() + if err == nil { + t.Fatal("a plaintext standby was accepted alongside TLS") + } + if !strings.Contains(err.Error(), "b:7443") { + t.Errorf("the error does not say which address: %v", err) + } +} + +func TestNoServerAtAllIsStillAnError(t *testing.T) { + c := Config{} + if err := c.normalizeServerURL(); err == nil { + t.Error("an agent with no control plane configured was accepted") + } +} + +func TestSchemeIsAddedPerAddress(t *testing.T) { + c := normalized(t, Config{ + Servers: []string{"a:7443", "b:7443"}, + TLS: tlsconf.Config{CA: "ca.pem", Cert: "c.pem", Key: "k.pem"}, + }) + for _, s := range c.Servers { + if !strings.HasPrefix(s, "https://") { + t.Errorf("%q did not get the secure scheme", s) + } + } +} diff --git a/internal/cli/server.go b/internal/cli/server.go index e58efb32..a0114bed 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -342,15 +342,23 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live // The same shape as the usage assertion above, and for the same reason: // the sink is the router when the router is serving, and nothing when a // control plane runs on its own. + var lead *control.Leadership if h, ok := sink.(interface{ Store() *storepkg.SQLite }); ok { if st := h.Store(); st != nil { - lead := control.NewLeadership(st, cfg.Control.LeaseTTL, log) + lead = control.NewLeadership(st, cfg.Control.LeaseTTL, log) ctl.SetLeadership(lead) log.Info("competing to be the control plane that places work", "owner", lead.Owner()) } } svc := agentsvc.New(ctl.Inventory(), ctl.Reconciler(), requireIdentity, log) + // A standby must refuse agents as well as refusing to place. Commands + // travel down the stream the agent opened, so a standby holding one is a + // node the leader cannot command — and the leader's placements would fail + // with "no agent connected" for a node that is plainly connected. + if lead != nil { + svc.SetLeadership(lead) + } fwd.set(svc) ctl.Start(ctx) diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index f68d8931..0b937b6d 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -31,6 +31,9 @@ type StatusSink interface { // time the reconciler's Dispatcher: it holds the open streams, so it is the // only thing that can deliver a command to a node. type Service struct { + // leader gates whether this process accepts agent streams at all. + leader Leader + inv *inventory.Store status StatusSink log *slog.Logger @@ -86,6 +89,19 @@ func (s *Service) Connect( ctx context.Context, stream *connect.BidiStream[llmgwv1.AgentMessage, llmgwv1.ControlMessage], ) error { + // A standby must not hold an agent's stream. Commands travel down the + // stream the agent opened, so whichever control plane holds it is the only + // one that can start or stop anything on that node — and if that is not + // the leader, the leader's placements fail with "no agent connected" for a + // node that is plainly connected. + // + // Refused before the opening message is read, so the agent spends nothing + // on a connection that cannot serve it. + if s.leader != nil && !s.leader.IsLeader() { + return connect.NewError(connect.CodeUnavailable, + errors.New("this control plane is a standby; connect to the leader")) + } + first, err := stream.Receive() if err != nil { return connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("no opening message: %w", err)) @@ -345,6 +361,16 @@ func (s *Service) ConnectedNodes() []string { // --- Dispatcher --- +// Leader reports whether this process is the one that should be holding agent +// streams. +type Leader interface{ IsLeader() bool } + +// SetLeadership makes this service refuse agents while it is a standby. +// +// Nil accepts them unconditionally, which is what a single control plane wants +// and what every deployment before leadership had. +func (s *Service) SetLeadership(l Leader) { s.leader = l } + // ErrNotConnected means no agent stream is open for that node. var ErrNotConnected = errors.New("no agent connected")