Ilker ManapandClaude Opus 5.5
e725adc2f0
A reconcile pass that does something is a trace, with placement (where it went, or every candidate turned down), scale-up/down, drain, stop, eviction, preemption, reaping and restarts beneath it. A replica's start is a span open from placement until it serves; the start command carries its context in a new ControlMessage.trace field, and the agent continues the trace through the weights fetch, engine launch, rendezvous and readiness wait. Stops, drains, adoption and engine installs are traced on the node, model jobs are a trace with a span per step linked to the replica's start, and agent.yaml takes the same tracing block as server.yaml. Off, every span is the no-op one and commands carry nothing. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
66 lines
2.2 KiB
Go
66 lines
2.2 KiB
Go
package modelserve
|
|
|
|
import (
|
|
"strings"
|
|
"testing"
|
|
|
|
"go.opentelemetry.io/otel/codes"
|
|
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
|
|
"gitea.mynodes.xyz/ilker/llmgw/internal/obs/obstest"
|
|
)
|
|
|
|
func TestAJobIsATraceWithASpanPerStep(t *testing.T) {
|
|
rec := obstest.Record(t)
|
|
files := []hf.File{{Path: "Qwen2.5-1.5B-Instruct-Q4_K_M.gguf", Bytes: 1 << 30, SHA256: "bb"}}
|
|
s, _, _, _ := setup(t, files, map[string]string{"Qwen2.5-1.5B-Instruct-Q4_K_M.gguf": "BB"})
|
|
job, err := s.Start(Request{Repo: "bartowski/Qwen2.5-1.5B-Instruct-GGUF", Replicas: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if j := wait(t, s, job.ID); j.State != Done {
|
|
t.Fatalf("job %s: %s", j.State, j.Error)
|
|
}
|
|
|
|
root := rec.One(t, [16]byte{}, "serve model")
|
|
tid := root.SpanContext().TraceID()
|
|
if a := obstest.Attrs(root); a["llmgw.job"] != job.ID || a["llmgw.model"] != "qwen2.5-1.5b-instruct" {
|
|
t.Errorf("the job's span does not say which job: %v", a)
|
|
}
|
|
var steps []string
|
|
for _, sp := range rec.Ended() {
|
|
if sp.SpanContext().TraceID() == tid && obstest.ChildOf(sp, root) {
|
|
steps = append(steps, sp.Name())
|
|
}
|
|
}
|
|
if strings.Join(steps, ",") != "registry,mirror,check,catalog,placement,smoke test" {
|
|
t.Errorf("step spans %v", steps)
|
|
}
|
|
smoke := rec.One(t, tid, "smoke test")
|
|
if a := obstest.Attrs(smoke); a["llmgw.state"] != Done || !strings.Contains(a["llmgw.detail"], "Hello!") {
|
|
t.Errorf("the smoke test's span says %v", a)
|
|
}
|
|
}
|
|
|
|
func TestAFailedJobsTraceEndsAtTheStepThatFailed(t *testing.T) {
|
|
rec := obstest.Record(t)
|
|
files := []hf.File{{Path: "m-Q4_K_M.gguf", Bytes: 1 << 30, SHA256: "aaaaaaaaaaaaaaaa"}}
|
|
s, _, _, _ := setup(t, files, map[string]string{"m-Q4_K_M.gguf": "bbbbbbbbbbbbbbbb"})
|
|
job, _ := s.Start(Request{Repo: "a/b", Replicas: 1})
|
|
if j := wait(t, s, job.ID); j.State != Failed {
|
|
t.Fatalf("job %s", j.State)
|
|
}
|
|
|
|
root := rec.One(t, [16]byte{}, "serve model")
|
|
if root.Status().Code != codes.Error || !strings.Contains(root.Status().Description, "registry published") {
|
|
t.Errorf("the failed job's span says %+v", root.Status())
|
|
}
|
|
mirror := rec.One(t, root.SpanContext().TraceID(), "mirror")
|
|
if mirror.Status().Code != codes.Error {
|
|
t.Errorf("the step that failed is %+v", mirror.Status())
|
|
}
|
|
if len(rec.Named("catalog")) != 0 {
|
|
t.Error("a step after the failure was traced")
|
|
}
|
|
}
|