Ilker ManapandClaude Opus 5
40f5c930c4
CI / test (push) Successful in 1m43s
Two defects, both of the same kind: something the CLI knew and the rest of the system did not. A profile can be edited while a replica is running against the old one. The reconciler deliberately leaves it alone — churning a replica that is serving costs more than the imperfection — but there was no way to see it, so the cluster could disagree with its own configuration indefinitely and read as healthy. Profile now carries a fingerprint, a replica records the one it was placed with, and both the API and `ctl replicas` say "stale profile" when they differ. An empty fingerprint is not stale: it means a replica adopted from an older agent, which is unknown rather than wrong. Adding a model from the admin page skipped the source check `ctl model add` has done all along, so a typo or a mirror that is down became a replica failing on a node minutes later, with an error naming the node. The check moves out of the CLI into a package both use. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
404 lines
15 KiB
Go
404 lines
15 KiB
Go
package web
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"strings"
|
|
"testing"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
|
|
)
|
|
|
|
type fakeCatalog struct {
|
|
added []catalog.Model
|
|
declared string
|
|
err error
|
|
}
|
|
|
|
func (f *fakeCatalog) Add(_ context.Context, m catalog.Model, _ string) error {
|
|
if f.err != nil {
|
|
return f.err
|
|
}
|
|
f.added = append(f.added, m)
|
|
return nil
|
|
}
|
|
func (f *fakeCatalog) Declared(name string) bool { return name == f.declared }
|
|
|
|
// fleet reports a cluster with a given number of cards and engines.
|
|
type fakeFleet struct {
|
|
gpus int
|
|
engines []string
|
|
memoryGB int
|
|
}
|
|
|
|
func (f fakeFleet) Nodes() []adminapi.NodeView {
|
|
return []adminapi.NodeView{
|
|
{ID: "n1", GPUTotal: f.gpus, Engines: f.engines, MemoryGB: f.memoryGB},
|
|
}
|
|
}
|
|
func (f fakeFleet) Models() []adminapi.ModelView { return nil }
|
|
func (f fakeFleet) Replicas(string) []adminapi.ReplicaView { return nil }
|
|
func (f fakeFleet) Explain(string) adminapi.ExplainView { return adminapi.ExplainView{} }
|
|
func (f fakeFleet) Summary() adminapi.Summary { return adminapi.Summary{} }
|
|
|
|
func held(model, version, file string) mirror.Status {
|
|
return mirror.Status{Model: model, Version: version, File: file,
|
|
State: mirror.Present, BytesDone: 10, BytesTotal: 10}
|
|
}
|
|
|
|
func servePage(t *testing.T, models []mirror.Status, cat CatalogAdmin, gpus int,
|
|
engines ...string,
|
|
) *http.ServeMux {
|
|
t.Helper()
|
|
return mirrorPageDeps(t, MirrorDeps{
|
|
Mirror: &fakeMirror{models: models}, Catalog: cat,
|
|
Fleet: fakeFleet{gpus: gpus, engines: engines},
|
|
}, true)
|
|
}
|
|
|
|
// reachableSource stands in for a mirror holding the file. The serve handler
|
|
// checks the source before the catalog sees it, the same as the CLI does, so a
|
|
// test that means to exercise the form has to offer something that answers.
|
|
func reachableSource(t *testing.T) string {
|
|
t.Helper()
|
|
srv := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
|
|
t.Cleanup(srv.Close)
|
|
return srv.URL + "/coder/v1/m.gguf"
|
|
}
|
|
|
|
func TestAGGUFCanBeServedStraightFromTheList(t *testing.T) {
|
|
cat := &fakeCatalog{}
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen2.5-coder-7b", "c03e6d35", "qwen-q4_k_m.gguf")}, cat, 0),
|
|
"/admin/mirror?serve=qwen2.5-coder-7b:c03e6d35").Body.String()
|
|
|
|
// The mirror already knows the name, the URL and what reads that kind of
|
|
// file; leaving an operator to write YAML for it was the gap.
|
|
if !strings.Contains(body, `name="engine" value="llamacpp"`) {
|
|
t.Errorf("the engine should be inferred from the extension; got %s", body)
|
|
}
|
|
if !strings.Contains(body, "/qwen2.5-coder-7b/c03e6d35/qwen-q4_k_m.gguf") {
|
|
t.Error("the source should be the mirror's own URL")
|
|
}
|
|
}
|
|
|
|
func TestAShardedVersionIsServedAsTheWholeVersion(t *testing.T) {
|
|
models := []mirror.Status{
|
|
held("llama", "v1", "model-00001-of-00002.safetensors"),
|
|
held("llama", "v1", "model-00002-of-00002.safetensors"),
|
|
}
|
|
body := mirrorGet(t, servePage(t, models, &fakeCatalog{}, 8, "vllm"),
|
|
"/admin/mirror?serve=llama:v1").Body.String()
|
|
|
|
// A catalogue entry names one source, and for a version of several files
|
|
// that source is the version. Refusing it meant a sharded model could be
|
|
// mirrored and never served — which is most of what is not a GGUF.
|
|
if !strings.Contains(body, `value="http://mirror:8081/llama/v1/"`) {
|
|
t.Errorf("the source should be the version directory; got %s", body)
|
|
}
|
|
if !strings.Contains(body, "The whole version, 2 files") {
|
|
t.Error("the form should say it is fetching a set")
|
|
}
|
|
if !strings.Contains(body, `action="/admin/mirror/serve"`) {
|
|
t.Error("it should be servable")
|
|
}
|
|
}
|
|
|
|
func TestTheEngineForASetComesFromItsWeights(t *testing.T) {
|
|
// A version is a mixture only when somebody has put a file there by hand.
|
|
// config.json decides nothing; the safetensors do.
|
|
models := []mirror.Status{
|
|
held("llama", "v1", "config.json"),
|
|
held("llama", "v1", "model-00001-of-00002.safetensors"),
|
|
}
|
|
body := mirrorGet(t, servePage(t, models, &fakeCatalog{}, 8, "vllm"),
|
|
"/admin/mirror?serve=llama:v1").Body.String()
|
|
|
|
if !strings.Contains(body, `name="engine" value="vllm"`) {
|
|
t.Errorf("the weights should pick the engine; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestSafetensorsOnACPUFleetWarnsBeforeAnybodyTries(t *testing.T) {
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen", "v1", "model.safetensors")}, &fakeCatalog{}, 0, "llamacpp"),
|
|
"/admin/mirror?serve=qwen:v1").Body.String()
|
|
|
|
// Fifteen gigabytes arrived before anybody noticed the fleet had no card
|
|
// and no vLLM to run them with. The page can say so.
|
|
if !strings.Contains(body, "no node reports having vllm installed") {
|
|
t.Errorf("expected the warning; got %s", body)
|
|
}
|
|
// And the way out, which on a fleet with no cards is a different build of
|
|
// the same model rather than installing vLLM.
|
|
if !strings.Contains(body, "GGUF build of the same model") {
|
|
t.Error("a fleet with no GPU should be pointed at a GGUF")
|
|
}
|
|
// A warning, not a refusal: a card could be added tomorrow.
|
|
if !strings.Contains(body, `action="/admin/mirror/serve"`) {
|
|
t.Error("it should still be possible to add it")
|
|
}
|
|
}
|
|
|
|
func TestSafetensorsOnAGPUFleetIsNotWarnedAbout(t *testing.T) {
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen", "v1", "model.safetensors")}, &fakeCatalog{}, 8, "vllm"),
|
|
"/admin/mirror?serve=qwen:v1").Body.String()
|
|
|
|
if strings.Contains(body, "no node reports having") {
|
|
t.Error("a fleet that runs vllm should not be warned about")
|
|
}
|
|
if !strings.Contains(body, `value="vllm"`) {
|
|
t.Error("safetensors should propose vllm")
|
|
}
|
|
}
|
|
|
|
func TestAModelDeclaredInConfigurationIsNotOfferedTwice(t *testing.T) {
|
|
cat := &fakeCatalog{declared: "qwen"}
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen", "v1", "m.gguf")}, cat, 0),
|
|
"/admin/mirror?serve=qwen:v1").Body.String()
|
|
|
|
if !strings.Contains(body, "declared in configuration") {
|
|
t.Errorf("expected the refusal; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestServingBuildsTheEntryFromTheForm(t *testing.T) {
|
|
cat := &fakeCatalog{}
|
|
source := reachableSource(t)
|
|
mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "aliases": {"c7b, coder-7b"},
|
|
"source": {source}, "engine": {"llamacpp"},
|
|
"replicas": {"3"}, "context_length": {"4096"}, "max_output": {"1024"},
|
|
"ctx_size": {"16384"}, "profile": {"cpu-large"},
|
|
"min_memory_gb": {"8"}, "selector_key": {"role"}, "selector_value": {"cpu-large"},
|
|
})
|
|
|
|
if len(cat.added) != 1 {
|
|
t.Fatalf("expected one model added, got %d", len(cat.added))
|
|
}
|
|
m := cat.added[0]
|
|
if m.Name != "coder" || len(m.Aliases) != 2 {
|
|
t.Errorf("name and aliases did not arrive: %+v", m)
|
|
}
|
|
if m.Scaling.Min != 3 || m.Scaling.Max != 3 {
|
|
t.Errorf("expected three replicas, got %+v", m.Scaling)
|
|
}
|
|
if len(m.Profiles) != 1 {
|
|
t.Fatalf("expected one profile, got %+v", m.Profiles)
|
|
}
|
|
p := m.Profiles[0]
|
|
if p.Engine != "llamacpp" || p.Name != "cpu-large" || p.MinMemoryGB != 8 {
|
|
t.Errorf("the profile is wrong: %+v", p)
|
|
}
|
|
// The label is what keeps a large model off the small machines.
|
|
if p.NodeSelector["role"] != "cpu-large" {
|
|
t.Errorf("the node selector was lost: %+v", p.NodeSelector)
|
|
}
|
|
if strings.Join(p.Args, " ") != "--ctx-size 16384" {
|
|
t.Errorf("the engine's context pool was lost: %v", p.Args)
|
|
}
|
|
}
|
|
|
|
func TestRubbishInTheFormIsRefusedBeforeTheCatalog(t *testing.T) {
|
|
cat := &fakeCatalog{}
|
|
mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {"http://x/m.gguf"},
|
|
"engine": {"llamacpp"}, "replicas": {"lots"}, "context_length": {"4096"},
|
|
})
|
|
if len(cat.added) != 0 {
|
|
t.Error("nothing should have reached the catalog")
|
|
}
|
|
}
|
|
|
|
func TestTheCatalogsOwnRefusalReachesTheOperator(t *testing.T) {
|
|
cat := &fakeCatalog{err: errors.New("that name collides with an alias")}
|
|
source := reachableSource(t)
|
|
body := mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {source},
|
|
"engine": {"llamacpp"}, "replicas": {"1"}, "context_length": {"4096"},
|
|
}).Body.String()
|
|
|
|
if !strings.Contains(body, "collides with an alias") {
|
|
t.Errorf("the catalog's reason should be shown; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestWithNoCatalogThereIsNoServeAction(t *testing.T) {
|
|
body := mirrorGet(t, mirrorPageDeps(t, MirrorDeps{
|
|
Mirror: &fakeMirror{models: []mirror.Status{held("qwen", "v1", "m.gguf")}},
|
|
}, true), "/admin/mirror").Body.String()
|
|
|
|
// Without a store a model added at runtime would not survive a restart, so
|
|
// the action is absent rather than offered and then refused.
|
|
if strings.Contains(body, "?serve=") {
|
|
t.Error("serving should not be offered without somewhere to persist it")
|
|
}
|
|
}
|
|
|
|
func TestAGPUNodeWithoutVLLMIsStillWarnedAbout(t *testing.T) {
|
|
// The first version of this checked cards, which would have waved through
|
|
// a fleet of eight-GPU nodes that had never had vLLM installed. Agents
|
|
// report their engines now, so it is a fact.
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen", "v1", "model.safetensors")}, &fakeCatalog{}, 8, "llamacpp"),
|
|
"/admin/mirror?serve=qwen:v1").Body.String()
|
|
|
|
if !strings.Contains(body, "no node reports having vllm installed") {
|
|
t.Errorf("cards are not the question; the engine is. Got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestAFleetOfOlderAgentsIsNotToldEverythingIsUnservable(t *testing.T) {
|
|
// An agent older than the engine probe reports nothing, which is not the
|
|
// same as having no engines. Placement makes the same allowance, and the
|
|
// two disagreeing would be worse than either.
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
held("qwen", "v1", "m.gguf")}, &fakeCatalog{}, 0),
|
|
"/admin/mirror?serve=qwen:v1").Body.String()
|
|
|
|
if strings.Contains(body, "no node reports having") {
|
|
t.Error("silence from older agents must not read as a refusal")
|
|
}
|
|
}
|
|
|
|
func sized(model, version, file string, gb float64) mirror.Status {
|
|
n := int64(gb * (1 << 30))
|
|
return mirror.Status{Model: model, Version: version, File: file,
|
|
State: mirror.Present, BytesDone: n, BytesTotal: n}
|
|
}
|
|
|
|
func TestTheMemoryFieldIsFilledInFromTheWeights(t *testing.T) {
|
|
// The real case: 4.68 GB of weights went onto a 7.9 GB node and left it
|
|
// 3.5 GB for a KV pool, because the field was optional and blank. Blank
|
|
// writes no constraint, and a CPU profile has no card to be short of.
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
sized("coder", "v1", "m.gguf", 4.68)}, &fakeCatalog{}, 0, "llamacpp"),
|
|
"/admin/mirror?serve=coder:v1").Body.String()
|
|
|
|
if !strings.Contains(body, `name="min_memory_gb" value="9"`) {
|
|
t.Errorf("expected a figure derived from 4.68 GB of weights; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestASmallModelDoesNotDemandALargeNode(t *testing.T) {
|
|
// The estimate has to leave the small nodes usable, or every model ends up
|
|
// pinned to the one big machine.
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
sized("small", "v1", "m.gguf", 1.12)}, &fakeCatalog{}, 0, "llamacpp"),
|
|
"/admin/mirror?serve=small:v1").Body.String()
|
|
|
|
if !strings.Contains(body, `name="min_memory_gb" value="3"`) {
|
|
t.Errorf("a 1.12 GB model should fit a small node; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestAModelTooBigForEveryNodeSaysSo(t *testing.T) {
|
|
page := mirrorPageDeps(t, MirrorDeps{
|
|
Mirror: &fakeMirror{models: []mirror.Status{sized("huge", "v1", "m.gguf", 40)}},
|
|
Catalog: &fakeCatalog{},
|
|
Fleet: fakeFleet{gpus: 0, engines: []string{"llamacpp"}, memoryGB: 31},
|
|
}, true)
|
|
body := mirrorGet(t, page, "/admin/mirror?serve=huge:v1").Body.String()
|
|
|
|
// Better said here than discovered by a node being OOM-killed twenty
|
|
// minutes into a load.
|
|
if !strings.Contains(body, "largest node has 31 GB") {
|
|
t.Errorf("expected the size warning; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestAnEstimateIsNotMadeWhenTheSizeIsUnknown(t *testing.T) {
|
|
// A version whose size nobody reported would otherwise be given a
|
|
// requirement of two, which is a constraint that excludes nothing and
|
|
// reads as though somebody measured it.
|
|
body := mirrorGet(t, servePage(t, []mirror.Status{
|
|
{Model: "m", Version: "v", File: "f.gguf", State: mirror.Present}},
|
|
&fakeCatalog{}, 0, "llamacpp"), "/admin/mirror?serve=m:v").Body.String()
|
|
|
|
if !strings.Contains(body, `name="min_memory_gb" value="0"`) {
|
|
t.Errorf("an unknown size should leave no figure; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestADescriptionReachesTheCatalog(t *testing.T) {
|
|
cat := &fakeCatalog{}
|
|
source := reachableSource(t)
|
|
mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {source},
|
|
"engine": {"llamacpp"}, "replicas": {"1"}, "context_length": {"4096"},
|
|
"description": {"code completion and review; 7B, fits the large node"},
|
|
})
|
|
|
|
if len(cat.added) != 1 {
|
|
t.Fatalf("expected one model, got %d", len(cat.added))
|
|
}
|
|
// The line a client sees in /v1/models, which for most of them is the only
|
|
// view of what this gateway serves.
|
|
if cat.added[0].Description != "code completion and review; 7B, fits the large node" {
|
|
t.Errorf("the description was lost: %q", cat.added[0].Description)
|
|
}
|
|
}
|
|
|
|
func TestAModelWithNoDescriptionIsStillAccepted(t *testing.T) {
|
|
cat := &fakeCatalog{}
|
|
source := reachableSource(t)
|
|
mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {source},
|
|
"engine": {"llamacpp"}, "replicas": {"1"}, "context_length": {"4096"},
|
|
})
|
|
if len(cat.added) != 1 || cat.added[0].Description != "" {
|
|
t.Errorf("a description is optional: %+v", cat.added)
|
|
}
|
|
}
|
|
|
|
func TestAnUnreachableSourceIsRefusedBeforeTheCatalog(t *testing.T) {
|
|
// A port nothing is listening on: refused at once rather than after a
|
|
// timeout, and the same shape of failure as a mirror that is down.
|
|
srv := httptest.NewServer(http.HandlerFunc(
|
|
func(http.ResponseWriter, *http.Request) {}))
|
|
dead := srv.URL
|
|
srv.Close()
|
|
|
|
cat := &fakeCatalog{}
|
|
body := mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {dead + "/m.gguf"},
|
|
"engine": {"llamacpp"}, "replicas": {"1"}, "context_length": {"4096"},
|
|
}).Body.String()
|
|
|
|
if len(cat.added) != 0 {
|
|
t.Error("a model whose weights are not there should not reach the catalog")
|
|
}
|
|
// The address, so the operator can see which one it could not reach; the
|
|
// page used to accept this and leave the failure to a node, minutes later.
|
|
if !strings.Contains(body, "cannot reach") {
|
|
t.Errorf("the reason should name the source; got %s", body)
|
|
}
|
|
}
|
|
|
|
func TestASourceTheMirrorNoLongerHoldsIsRefused(t *testing.T) {
|
|
gone := httptest.NewServer(http.HandlerFunc(
|
|
func(w http.ResponseWriter, _ *http.Request) { http.Error(w, "gone", http.StatusNotFound) }))
|
|
t.Cleanup(gone.Close)
|
|
|
|
cat := &fakeCatalog{}
|
|
body := mirrorPost(t, servePage(t, nil, cat, 0), "/admin/mirror/serve", url.Values{
|
|
"csrf": {"tok"}, "name": {"coder"}, "source": {gone.URL + "/m.gguf"},
|
|
"engine": {"llamacpp"}, "replicas": {"1"}, "context_length": {"4096"},
|
|
}).Body.String()
|
|
|
|
if len(cat.added) != 0 {
|
|
t.Error("a file removed between the listing and the click should be caught")
|
|
}
|
|
if !strings.Contains(body, "404") {
|
|
t.Errorf("the server's answer should be shown; got %s", body)
|
|
}
|
|
}
|