diff --git a/docs/reference.md b/docs/reference.md index 04a5e8a6..c12ffbcd 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -180,6 +180,7 @@ rather than repeated. | `parallel` | `int` | [LlamaCppConfig](#llamacppconfig) | | `threads` | `int` | [LlamaCppConfig](#llamacppconfig) | | `api` | `string` | [Mirror](#mirror) | +| `peers` | `[]string` | [Mirror](#mirror) | | `tls` | `tlsconf.Config` | [Mirror](#mirror) | | `registry` | `string` | [Mirror](#mirror) | | `timeout` | `time.Duration` | [Mirror](#mirror) | @@ -1722,6 +1723,14 @@ The management port, not the one nodes fetch bytes from. Those are two listeners on the mirror precisely so this one can be locked down without re-certifying every agent's fetcher. +### `peers` + +`[]string` + +Peers are the management addresses of mirrors kept in step with this +one (llmgw mirror serve --peer): asked when it cannot be reached, and +a node given the same file on one of them while it is down. + ### `tls` `tlsconf.Config` diff --git a/internal/cli/node.go b/internal/cli/node.go index 049cceb2..6da7de57 100644 --- a/internal/cli/node.go +++ b/internal/cli/node.go @@ -186,7 +186,7 @@ func (na *nodeAdd) prepare() error { } // Where the mirror serves model files, for the node to be asked to reach. if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil && cfg.Mirror.API != "" { - if mc, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 5*time.Second); err == nil { + if mc, err := mirror.NewFailoverClient(append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...), cfg.Mirror.TLS, 5*time.Second); err == nil { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) if list, err := mc.List(ctx); err == nil { o.MirrorBase = list.Base diff --git a/internal/cli/server.go b/internal/cli/server.go index 6276bb5c..113545a6 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -567,10 +567,21 @@ func newServerCmd() *cobra.Command { // rather than turning into a broken admin page hours later when // somebody finally opens it. if cfg.Mirror.API != "" { - mc, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, cfg.Mirror.Timeout) + mirrors := append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...) + mc, err := mirror.NewFailoverClient(mirrors, cfg.Mirror.TLS, cfg.Mirror.Timeout) if err != nil { return fmt.Errorf("the model mirror: %w", err) } + // Mirrors kept in step: a node is given a model's file on one + // that answers while the one the catalog names is down. + if len(cfg.Mirror.Peers) > 0 { + src, err := mirror.NewSources(mirrors, cfg.Mirror.TLS, log) + if err != nil { + return fmt.Errorf("the model mirrors: %w", err) + } + go src.Run(ctx) + agentsvc.SetSourceRewrite(src.Rewrite) + } rt.SetMirror(mc) if fwPub != nil { fwPub.setMirror(mc) @@ -578,7 +589,7 @@ func newServerCmd() *cobra.Command { // Adapters trained on the nodes are kept there too; sending one // takes longer than the mirror's usual answers. if adminAPI != nil { - if up, err := mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 10*time.Minute); err == nil { + if up, err := mirror.NewFailoverClient(mirrors, cfg.Mirror.TLS, 10*time.Minute); err == nil { adminAPI.SetTrainMirror(up) } } diff --git a/internal/cli/tui_mirror.go b/internal/cli/tui_mirror.go index 63e159bd..6de5c7df 100644 --- a/internal/cli/tui_mirror.go +++ b/internal/cli/tui_mirror.go @@ -32,7 +32,7 @@ func (m consoleModel) mirrorClient() (*mirror.Client, error) { if cfg.Mirror.API == "" { return nil, errors.New("no mirror: mirror.api in server.yaml is empty") } - return mirror.NewClient(cfg.Mirror.API, cfg.Mirror.TLS, 10*time.Second) + return mirror.NewFailoverClient(append([]string{cfg.Mirror.API}, cfg.Mirror.Peers...), cfg.Mirror.TLS, 10*time.Second) } func (m consoleModel) fetchMirror() tea.Cmd { diff --git a/internal/config/config.go b/internal/config/config.go index ea394294..6d129d79 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -345,6 +345,10 @@ type Mirror struct { // listeners on the mirror precisely so this one can be locked down without // re-certifying every agent's fetcher. API string `koanf:"api"` + // Peers are the management addresses of mirrors kept in step with this + // one (llmgw mirror serve --peer): asked when it cannot be reached, and + // a node given the same file on one of them while it is down. + Peers []string `koanf:"peers"` // TLS authenticates to the mirror's management API, with the same CA the // agents use. Empty leaves the connection in plain HTTP, which is only // acceptable on a network where nothing untrusted can reach that port — diff --git a/internal/control/agentsvc/convert.go b/internal/control/agentsvc/convert.go index e07617a9..d4f84662 100644 --- a/internal/control/agentsvc/convert.go +++ b/internal/control/agentsvc/convert.go @@ -5,6 +5,8 @@ import ( "fmt" "maps" "runtime" + "slices" + "sync/atomic" "time" llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" @@ -189,8 +191,31 @@ func AdoptedFromRegister(nodeID string, shards []*llmgwv1.RunningShard) []reconc } // StartShardFromCommand converts a reconciler command into a wire message. +// sourceRewrite is where a model is fetched while its mirror is down: set +// when mirrors are kept in step, else nil and the catalog's address as it is. +var sourceRewrite atomic.Pointer[func(string) string] + +// SetSourceRewrite has every shard started from now on fetch its model, and +// adapters, from the address f gives: one on a mirror that answers. +func SetSourceRewrite(f func(string) string) { sourceRewrite.Store(&f) } + +func rewriteSource(src string) string { + if f := sourceRewrite.Load(); f != nil && src != "" { + return (*f)(src) + } + return src +} + func StartShardFromCommand(cmd reconciler.ShardCommand) *llmgwv1.StartShard { p := cmd.Spec.Profile + p.ModelPath = rewriteSource(p.ModelPath) + if len(p.Adapters) > 0 { + ads := slices.Clone(p.Adapters) + for i := range ads { + ads[i].Path = rewriteSource(ads[i].Path) + } + p.Adapters = ads + } return &llmgwv1.StartShard{ ReplicaId: cmd.ReplicaID, ShardId: cmd.ShardID, diff --git a/internal/control/agentsvc/convert_test.go b/internal/control/agentsvc/convert_test.go index 68880066..8310f850 100644 --- a/internal/control/agentsvc/convert_test.go +++ b/internal/control/agentsvc/convert_test.go @@ -1,6 +1,7 @@ package agentsvc import ( + "strings" "testing" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines" @@ -18,3 +19,20 @@ func TestAdaptersTravelToTheAgentInOrder(t *testing.T) { t.Errorf("adapters %+v", got) } } + +// While its mirror is down, a shard is sent its model and adapters on the +// mirror the rewrite names, and the command it was made from is left as it is. +func TestAShardIsSentItsModelOnAMirrorThatAnswers(t *testing.T) { + SetSourceRewrite(func(s string) string { return strings.Replace(s, "http://mirror-a", "http://mirror-b", 1) }) + t.Cleanup(func() { SetSourceRewrite(func(s string) string { return s }) }) + ads := []engines.Adapter{{Name: "sql", Path: "http://mirror-a/sql.gguf"}} + cmd := reconciler.ShardCommand{ReplicaID: "r", ShardID: "s", Spec: engines.ShardSpec{ + Profile: engines.Profile{Name: "cpu", Engine: "llamacpp", ModelPath: "http://mirror-a/qwen/v1/", Adapters: ads}}} + spec, _ := ShardSpecFromStart(StartShardFromCommand(cmd)) + if spec.Profile.ModelPath != "http://mirror-b/qwen/v1/" || spec.Profile.Adapters[0].Path != "http://mirror-b/sql.gguf" { + t.Errorf("sent %+v", spec.Profile) + } + if ads[0].Path != "http://mirror-a/sql.gguf" { + t.Error("the catalog's adapter was changed in place") + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index b0e52afd..d8c8618f 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -230,6 +230,7 @@ var Reference = []Block{ }}, {Package: "config", Name: "Mirror", Title: "Mirror", Doc: "", Fields: []Field{ {Key: "api", Type: "string", Doc: "API is the mirror's management address, scheme included, for example\nhttps://models.internal:8082. Empty means no mirror is configured.\n\nThe management port, not the one nodes fetch bytes from. Those are two\nlisteners on the mirror precisely so this one can be locked down without\nre-certifying every agent's fetcher.", Nested: ""}, + {Key: "peers", Type: "[]string", Doc: "Peers are the management addresses of mirrors kept in step with this\none (llmgw mirror serve --peer): asked when it cannot be reached, and\na node given the same file on one of them while it is down.", Nested: ""}, {Key: "tls", Type: "tlsconf.Config", Doc: "TLS authenticates to the mirror's management API, with the same CA the\nagents use. Empty leaves the connection in plain HTTP, which is only\nacceptable on a network where nothing untrusted can reach that port —\nthe mirror says as much in its own log when it starts that way.", Nested: "tlsconf.Config"}, {Key: "registry", Type: "string", Doc: "Registry is the model registry the mirror page browses, empty for the\npublic Hugging Face. Set it to an internal mirror of the registry where\nthe control plane cannot reach the internet — or leave the whole mirror\nblock out, which turns the explorer off rather than giving it a search\nbox that can only time out.", Nested: ""}, {Key: "timeout", Type: "time.Duration", Doc: "Timeout bounds a management call. It does not bound a download: the\nmirror answers a fetch as soon as it has started one, and the bytes keep\narriving long after this deadline has passed.", Nested: ""}, diff --git a/internal/mirror/client.go b/internal/mirror/client.go index ef25b3b7..c1b538ea 100644 --- a/internal/mirror/client.go +++ b/internal/mirror/client.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -24,6 +25,29 @@ import ( type Client struct { base string hc *http.Client + // others are the mirrors kept in step with this one, asked in turn when + // it cannot be reached: the same models, on another machine. + others []string +} + +// NewFailoverClient is NewClient over several mirrors kept in step: the +// first answers, and the next is asked when it cannot be reached. +func NewFailoverClient(addrs []string, tc tlsconf.Config, timeout time.Duration) (*Client, error) { + if len(addrs) == 0 { + return nil, fmt.Errorf("no mirror address") + } + c, err := NewClient(addrs[0], tc, timeout) + if err != nil { + return nil, err + } + for _, a := range addrs[1:] { + u, err := url.Parse(a) + if err != nil || u.Host == "" { + return nil, fmt.Errorf("the mirror address %q is not a URL with a host", a) + } + c.others = append(c.others, strings.TrimSuffix(a, "/")) + } + return c, nil } // NewClient dials a mirror's management API, with mutual TLS when configured. @@ -149,28 +173,42 @@ func (c *Client) Remove(ctx context.Context, model, version string) ([]string, e } func (c *Client) do(ctx context.Context, method, path string, body, out any) error { - var rdr io.Reader + var b []byte if body != nil { - b, err := json.Marshal(body) + var err error + if b, err = json.Marshal(body); err != nil { + return err + } + } + // The mirror first, then those kept in step with it, while each cannot + // be reached at all: one that answers, even with a refusal, is the one. + var resp *http.Response + var errs []error + for _, base := range append([]string{c.base}, c.others...) { + var rdr io.Reader + if body != nil { + rdr = bytes.NewReader(b) + } + req, err := http.NewRequestWithContext(ctx, method, base+path, rdr) if err != nil { return err } - rdr = bytes.NewReader(b) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + if resp, err = c.hc.Do(req); err == nil { + break + } + // Wrapped with the address because the failure an operator sees on + // a page is otherwise "connection refused" with no clue which of + // the machines refused it. + errs = append(errs, fmt.Errorf("reaching the mirror at %s: %w", base, err)) + if ctx.Err() != nil { + break + } } - req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr) - if err != nil { - return err - } - if body != nil { - req.Header.Set("Content-Type", "application/json") - } - - resp, err := c.hc.Do(req) - if err != nil { - // Wrapped with the address because the failure an operator sees on a - // page is otherwise "connection refused" with no clue which of the two - // machines refused it. - return fmt.Errorf("reaching the mirror at %s: %w", c.base, err) + if resp == nil { + return errors.Join(errs...) } defer resp.Body.Close() diff --git a/internal/mirror/peer_test.go b/internal/mirror/peer_test.go index 9fbdbfd4..6ff5e29c 100644 --- a/internal/mirror/peer_test.go +++ b/internal/mirror/peer_test.go @@ -125,3 +125,50 @@ func TestACopyWhoseChecksumIsNotThePeersIsNotKept(t *testing.T) { t.Errorf("a copy with the wrong checksum was kept: %v", err) } } + +// A mirror that cannot be reached is passed over for one kept in step with it. +func TestTheNextMirrorIsAskedWhenTheFirstCannotBeReached(t *testing.T) { + a, b := newTestMirror(t), newTestMirror(t) + c, err := NewFailoverClient([]string{a.api.URL, b.api.URL}, tlsconf.Config{}, 2*time.Second) + if err != nil { + t.Fatal(err) + } + if list, err := c.List(t.Context()); err != nil || list.Base != a.files.URL { + t.Fatalf("both up: %+v %v", list, err) + } + a.api.Close() + if list, err := c.List(t.Context()); err != nil || list.Base != b.files.URL { + t.Errorf("the first down: %+v %v", list, err) + } + b.api.Close() + if _, err := c.List(t.Context()); err == nil { + t.Error("both down, and an answer") + } +} + +// A node is given a model's file on a mirror that answers. +func TestAModelsAddressIsMovedToAMirrorThatAnswers(t *testing.T) { + a, b := newTestMirror(t), newTestMirror(t) + s, err := NewSources([]string{a.api.URL, b.api.URL}, tlsconf.Config{}, discardLog()) + if err != nil { + t.Fatal(err) + } + s.Check(t.Context()) + src := a.files.URL + "/qwen/v1/m.gguf" + if got := s.Rewrite(src); got != src { + t.Errorf("its own mirror up: %s", got) + } + a.api.Close() + s.Check(t.Context()) + if got := s.Rewrite(src); got != b.files.URL+"/qwen/v1/m.gguf" { + t.Errorf("its own mirror down: %s", got) + } + if got := s.Rewrite("https://huggingface.co/x/y"); got != "https://huggingface.co/x/y" { + t.Errorf("an address on no mirror: %s", got) + } + b.api.Close() + s.Check(t.Context()) + if got := s.Rewrite(src); got != src { + t.Errorf("none up: %s", got) + } +} diff --git a/internal/mirror/sources.go b/internal/mirror/sources.go new file mode 100644 index 00000000..4c19a14c --- /dev/null +++ b/internal/mirror/sources.go @@ -0,0 +1,102 @@ +package mirror + +import ( + "context" + "log/slog" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" +) + +// Sources is where a node fetches a model while one mirror is down. A model +// in the catalog names one mirror's address; when that mirror does not +// answer, the address a node is given is the same file on one that does. + +// SourcesInterval is how often each mirror is asked whether it answers. +const SourcesInterval = 15 * time.Second + +// Sources watches mirrors kept in step, and rewrites a model's address to +// one that answers. +type Sources struct { + clients []*Client + log *slog.Logger + + mu sync.Mutex + bases []string + healthy []bool +} + +// NewSources watches the mirrors at these management addresses. +func NewSources(addrs []string, tc tlsconf.Config, log *slog.Logger) (*Sources, error) { + s := &Sources{log: log, bases: make([]string, len(addrs)), healthy: make([]bool, len(addrs))} + for _, a := range addrs { + c, err := NewClient(a, tc, 5*time.Second) + if err != nil { + return nil, err + } + s.clients = append(s.clients, c) + } + return s, nil +} + +// Run asks each mirror, every while, until ctx ends. +func (s *Sources) Run(ctx context.Context) { + t := time.NewTicker(SourcesInterval) + defer t.Stop() + for { + s.Check(ctx) + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// Check asks each mirror once: whether it answers, and the address its +// files are served at. +func (s *Sources) Check(ctx context.Context) { + for i, c := range s.clients { + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + list, err := c.List(ctx) + cancel() + s.mu.Lock() + was := s.healthy[i] + s.healthy[i] = err == nil + if err == nil && list.Base != "" { + s.bases[i] = strings.TrimSuffix(list.Base, "/") + } + s.mu.Unlock() + if was && err != nil { + s.log.Warn("a mirror does not answer; models it holds are fetched from another", "mirror", c.Addr(), "err", err) + } else if !was && err == nil { + s.log.Info("a mirror answers", "mirror", c.Addr(), "files", list.Base) + } + } +} + +// Rewrite is a model's address on a mirror that answers: as it is when its +// own mirror answers or no other does, else the same path on the first that +// does. +func (s *Sources) Rewrite(src string) string { + s.mu.Lock() + defer s.mu.Unlock() + from := -1 + for i, b := range s.bases { + if b != "" && (src == b || strings.HasPrefix(src, b+"/")) { + from = i + break + } + } + if from < 0 || s.healthy[from] { + return src + } + for i, b := range s.bases { + if i != from && b != "" && s.healthy[i] { + return b + strings.TrimPrefix(src, s.bases[from]) + } + } + return src +}