Tracing beyond the router: reconcile passes, placement and a replica's start traced down to the node

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>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-05 00:03:17 +02:00
1 parent e59ff52932
commit e725adc2f0
33 files changed
+1607 -143

No files matched your search

+68 -48
View File
@@ -272,7 +272,13 @@ type ControlMessage struct {
// *ControlMessage_FabricTest
// *ControlMessage_Upgrade
// *ControlMessage_ControlPlanes
Msg isControlMessage_Msg `protobuf_oneof:"msg"`
Msg isControlMessage_Msg `protobuf_oneof:"msg"`
// trace is the W3C trace context (traceparent, and tracestate when there
// is one) of the control plane's span that sent this command, so what the
// agent does for it — fetching weights, launching the engine, waiting for
// it to answer — lands in the same trace. Empty when the control plane is
// not tracing, and ignored by an agent that is not.
Trace map[string]string `protobuf:"bytes,12,rep,name=trace,proto3" json:"trace,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -413,6 +419,13 @@ func (x *ControlMessage) GetControlPlanes() *ControlPlanes {
return nil
}
func (x *ControlMessage) GetTrace() map[string]string {
if x != nil {
return x.Trace
}
return nil
}
type isControlMessage_Msg interface {
isControlMessage_Msg()
}
@@ -4128,7 +4141,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\vfabric_test\x18\x06 \x01(\v2\x19.llmgw.v1.FabricTestReplyH\x00R\n" +
"fabricTest\x122\n" +
"\aupgrade\x18\a \x01(\v2\x16.llmgw.v1.UpgradeReplyH\x00R\aupgradeB\x05\n" +
"\x03msg\"\xdb\x04\n" +
"\x03msg\"\xd0\x05\n" +
"\x0eControlMessage\x12,\n" +
"\x05start\x18\x01 \x01(\v2\x14.llmgw.v1.StartShardH\x00R\x05start\x12)\n" +
"\x04stop\x18\x02 \x01(\v2\x13.llmgw.v1.StopShardH\x00R\x04stop\x12,\n" +
@@ -4143,7 +4156,12 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"fabricTest\x122\n" +
"\aupgrade\x18\n" +
" \x01(\v2\x16.llmgw.v1.UpgradeAgentH\x00R\aupgrade\x12@\n" +
"\x0econtrol_planes\x18\v \x01(\v2\x17.llmgw.v1.ControlPlanesH\x00R\rcontrolPlanesB\x05\n" +
"\x0econtrol_planes\x18\v \x01(\v2\x17.llmgw.v1.ControlPlanesH\x00R\rcontrolPlanes\x129\n" +
"\x05trace\x18\f \x03(\v2#.llmgw.v1.ControlMessage.TraceEntryR\x05trace\x1a8\n" +
"\n" +
"TraceEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x05\n" +
"\x03msg\")\n" +
"\rControlPlanes\x12\x18\n" +
"\aservers\x18\x01 \x03(\tR\aservers\"\x80\a\n" +
@@ -4532,7 +4550,7 @@ func file_llmgw_v1_agent_proto_rawDescGZIP() []byte {
}
var file_llmgw_v1_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 44)
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 45)
var file_llmgw_v1_agent_proto_goTypes = []any{
(ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State
(*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage
@@ -4573,12 +4591,13 @@ var file_llmgw_v1_agent_proto_goTypes = []any{
(*UpgradeAgent)(nil), // 36: llmgw.v1.UpgradeAgent
(*UpgradeReply)(nil), // 37: llmgw.v1.UpgradeReply
(*LoraAdapter)(nil), // 38: llmgw.v1.LoraAdapter
nil, // 39: llmgw.v1.Register.LabelsEntry
nil, // 40: llmgw.v1.Register.EngineVersionsEntry
nil, // 41: llmgw.v1.GPU.NicPathsEntry
nil, // 42: llmgw.v1.IBPort.CountersEntry
nil, // 43: llmgw.v1.Heartbeat.EngineVersionsEntry
nil, // 44: llmgw.v1.EngineProfile.EnvEntry
nil, // 39: llmgw.v1.ControlMessage.TraceEntry
nil, // 40: llmgw.v1.Register.LabelsEntry
nil, // 41: llmgw.v1.Register.EngineVersionsEntry
nil, // 42: llmgw.v1.GPU.NicPathsEntry
nil, // 43: llmgw.v1.IBPort.CountersEntry
nil, // 44: llmgw.v1.Heartbeat.EngineVersionsEntry
nil, // 45: llmgw.v1.EngineProfile.EnvEntry
}
var file_llmgw_v1_agent_proto_depIdxs = []int32{
4, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
@@ -4599,43 +4618,44 @@ var file_llmgw_v1_agent_proto_depIdxs = []int32{
34, // 15: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest
36, // 16: llmgw.v1.ControlMessage.upgrade:type_name -> llmgw.v1.UpgradeAgent
3, // 17: llmgw.v1.ControlMessage.control_planes:type_name -> llmgw.v1.ControlPlanes
39, // 18: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
8, // 19: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
9, // 20: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
40, // 21: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
7, // 22: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
10, // 23: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
6, // 24: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
5, // 25: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
16, // 26: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX
0, // 27: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
41, // 28: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry
13, // 29: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad
12, // 30: llmgw.v1.NodeLoad.gpus:type_name -> llmgw.v1.GPUSensor
42, // 31: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry
15, // 32: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager
6, // 33: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
10, // 34: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
43, // 35: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
21, // 36: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
11, // 37: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad
14, // 38: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort
9, // 39: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment
17, // 40: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState
16, // 41: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX
20, // 42: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
0, // 43: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
23, // 44: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
25, // 45: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
44, // 46: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
38, // 47: llmgw.v1.EngineProfile.adapters:type_name -> llmgw.v1.LoraAdapter
1, // 48: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
2, // 49: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
49, // [49:50] is the sub-list for method output_type
48, // [48:49] is the sub-list for method input_type
48, // [48:48] is the sub-list for extension type_name
48, // [48:48] is the sub-list for extension extendee
0, // [0:48] is the sub-list for field type_name
39, // 18: llmgw.v1.ControlMessage.trace:type_name -> llmgw.v1.ControlMessage.TraceEntry
40, // 19: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
8, // 20: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
9, // 21: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
41, // 22: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
7, // 23: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
10, // 24: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
6, // 25: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
5, // 26: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
16, // 27: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX
0, // 28: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
42, // 29: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry
13, // 30: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad
12, // 31: llmgw.v1.NodeLoad.gpus:type_name -> llmgw.v1.GPUSensor
43, // 32: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry
15, // 33: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager
6, // 34: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
10, // 35: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
44, // 36: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
21, // 37: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
11, // 38: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad
14, // 39: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort
9, // 40: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment
17, // 41: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState
16, // 42: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX
20, // 43: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
0, // 44: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
23, // 45: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
25, // 46: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
45, // 47: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
38, // 48: llmgw.v1.EngineProfile.adapters:type_name -> llmgw.v1.LoraAdapter
1, // 49: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
2, // 50: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
50, // [50:51] is the sub-list for method output_type
49, // [49:50] is the sub-list for method input_type
49, // [49:49] is the sub-list for extension type_name
49, // [49:49] is the sub-list for extension extendee
0, // [0:49] is the sub-list for field type_name
}
func init() { file_llmgw_v1_agent_proto_init() }
@@ -4671,7 +4691,7 @@ func file_llmgw_v1_agent_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_llmgw_v1_agent_proto_rawDesc), len(file_llmgw_v1_agent_proto_rawDesc)),
NumEnums: 1,
NumMessages: 44,
NumMessages: 45,
NumExtensions: 0,
NumServices: 1,
},
+6
View File
@@ -45,6 +45,12 @@ message ControlMessage {
UpgradeAgent upgrade = 10;
ControlPlanes control_planes = 11;
}
// trace is the W3C trace context (traceparent, and tracestate when there
// is one) of the control plane's span that sent this command, so what the
// agent does for it — fetching weights, launching the engine, waiting for
// it to answer — lands in the same trace. Empty when the control plane is
// not tracing, and ignored by an agent that is not.
map<string, string> trace = 12;
}
// ControlPlanes is every control plane an agent may dial, sent by the leader
+40
View File
@@ -1002,6 +1002,46 @@ same event about the same thing is sent at most once an hour, so a budget that
refuses a thousand requests is one message. `llmgw ctl notify test` sends a
test event to each webhook and says how each answered.
### Tracing
Metrics say that a model took four minutes to start; a trace says where they
went. Give the control plane and every agent the same OpenTelemetry collector,
over OTLP/HTTP — Jaeger, Tempo, Honeycomb, anything that takes OTLP:
```yaml
# /etc/llmgw/server.yaml, and the same block in each node's agent.yaml
tracing:
endpoint: http://otel.internal:4318
sample_ratio: 1 # of requests that arrive without a decision of their own
```
`OTEL_EXPORTER_OTLP_ENDPOINT` does the same when the file does not say. With
neither, nothing is traced and nothing is sent: every span is a no-op, and a
command to a node carries no trace context.
What is traced:
| Trace | Spans |
|---|---|
| An inference request (`llmgw-server`) | `inference /v1/…` with `admission`, `schedule` and an `upstream` per attempt beneath it; a caller's `traceparent` is continued and the engine is sent one. |
| A reconcile pass that did something | `reconcile`, and beneath it what it did: `scale up` with `place` (the profile, GPU class, pool, fabric and nodes chosen, or every candidate turned down and why) and `replica start`; `scale down`; `drain replica` and `stop replica`, each with its reason; `evict`, `preempt` with its victims; `reap replica` with the error and when it is tried again; `restart replica`, `restart refused`; `node unreachable`, `node lost`. A pass that found nothing to do is not traced. |
| A replica's start | `replica start` stays open from placement until the replica serves, fails or is stopped, with each shard's progress as an event on it and how long it took. A `send start` per shard beneath it is the command to the node, and the node's spans (`llmgw-agent`) continue the same trace: `start shard`, with `fetch weights` (local, cached, downloaded or fetched by another shard, and a `download` with its bytes), `launch engine` (and `wait for rendezvous` on a worker of a multi-node replica) and `wait for engine` (how often the engine was asked, and its last answer if it never came up). |
| A stop on a node | `stop shard` and `stop engine` in the trace of the control plane's `stop replica`, saying whether the engine had to be killed. |
| An adoption | `adopt`, when a node that registers reports engines already running, with how many nobody claimed. |
| A model job | `serve model` or `update model`, with a span per step — `registry`, `mirror`, `check`, `catalog`, `placement`, `smoke test`, `replicas` — and from the placement step a link to the replica's own start. |
| An engine install | `install engine` on the node, a trace of its own. |
So "why did qwen take four minutes to start" is one trace: find `replica start`
with `llmgw.model` = qwen, and the node's spans beneath it say whether the
time went to the download, the launch or the engine loading.
A node follows the control plane's decision: a start the control plane kept is
traced on the node whatever the agent's own `sample_ratio`, and one it sampled
out is not; the agent's ratio decides only for what arrives with no trace, as
from a control plane that is not tracing. The agent needs its own `tracing`
block either way — without one its spans go nowhere, and the control plane's
trace simply ends at `send start`.
## PostgreSQL, when SQLite is no longer the right shape
SQLite is the default and stays the right answer for most installations: one
+22 -6
View File
@@ -224,7 +224,7 @@ rather than repeated.
| `debug_pprof` | `bool` | [Server](#server) |
| `agent_insecure` | `bool` | [Server](#server) |
| `monitoring` | `Monitoring` | [Server](#server) |
| `tracing` | `Tracing` | [Server](#server) |
| `tracing` | `obs.Tracing` | [Server](#server) |
| `mirror` | `Mirror` | [Server](#server) |
| `engines` | `map[string]string` | [Server](#server) |
| `engine_overrides` | `[]EngineOverride` | [Server](#server) |
@@ -291,6 +291,7 @@ rather than repeated.
| `discovery` | `discovery.Options` | [agent.Config](#agentconfig) |
| `engines` | `engines.Config` | [agent.Config](#agentconfig) |
| `supervisor` | `supervisor.Config` | [agent.Config](#agentconfig) |
| `tracing` | `obs.Tracing` | [agent.Config](#agentconfig) |
| `name` | `string` | [config.Band](#configband) |
| `priority` | `int` | [config.Band](#configband) |
| `saturation_ceiling` | `float64` | [config.Band](#configband) |
@@ -1817,10 +1818,11 @@ Takes a [Monitoring](#monitoring) block.
### `tracing`
`Tracing`
`obs.Tracing`
Tracing sends a trace of each inference request to an OpenTelemetry
collector. Off unless an endpoint is set here or in
Tracing sends traces to an OpenTelemetry collector: each inference
request, and on a control plane each placement, replica start and
model job. Off unless an endpoint is set here or in
OTEL_EXPORTER_OTLP_ENDPOINT.
Takes a [Tracing](#tracing) block.
@@ -2044,8 +2046,10 @@ Empty falls back to OTEL_EXPORTER_OTLP_ENDPOINT, and to no tracing.
`float64`
SampleRatio is the share of requests traced, 0 to 1, when the caller
did not decide it with a sampled traceparent of its own. Zero takes 1.
SampleRatio is the share of traces kept, 0 to 1, of those nobody
upstream already decided. Zero takes 1. A request with a sampled
traceparent of its own is kept, and an agent keeps every start the
control plane kept, whatever their own ratio says.
## VLLMConfig
@@ -2316,6 +2320,18 @@ Takes a [engines.Config](#enginesconfig) block.
Takes a [supervisor.Config](#supervisorconfig) block.
### `tracing`
`obs.Tracing`
Tracing sends this node's part of each replica start — the weights
fetched, the engine launched, the wait for it to answer — to an
OpenTelemetry collector, inside the control plane's trace of that
start. Give it the collector server.yaml's tracing names. Off unless
an endpoint is set here or in OTEL_EXPORTER_OTLP_ENDPOINT.
Takes a [Tracing](#tracing) block.
## config.Band
### `name`
+57 -4
View File
@@ -23,6 +23,8 @@ import (
"sync"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"golang.org/x/net/http2"
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
@@ -33,6 +35,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
@@ -419,6 +422,17 @@ func int32Slice(v []int) []int32 {
return out
}
func int32Ints(v []int32) []int {
out := make([]int, len(v))
for i, x := range v {
out[i] = int(x)
}
return out
}
// tracerName names the spans the agent opens.
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/agent"
// scan refreshes the node's hardware inventory.
func (a *Agent) scan(ctx context.Context) error {
// Carried from the agent's own configuration rather than duplicated into
@@ -786,15 +800,23 @@ func (a *Agent) removeWeights(paths []string) {
}
func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out chan<- *llmgwv1.AgentMessage) {
// The control plane's span that sent this, when it is tracing: what is
// done here for a start or a stop goes into the same trace.
tctx := obs.Extract(ctx, msg.GetTrace())
switch m := msg.Msg.(type) {
case *llmgwv1.ControlMessage_Start:
go a.startShard(ctx, m.Start, out)
go a.startShard(tctx, m.Start, out)
case *llmgwv1.ControlMessage_Stop:
go a.stopShard(ctx, m.Stop.GetShardId(), out)
go a.stopShard(tctx, m.Stop.GetShardId(), out)
case *llmgwv1.ControlMessage_Drain:
if err := a.sup.Drain(m.Drain.GetShardId()); err != nil {
_, span := obs.Tracer(tracerName).Start(tctx, "drain shard",
trace.WithSpanKind(trace.SpanKindConsumer),
trace.WithAttributes(attribute.String("llmgw.shard", m.Drain.GetShardId())))
err := a.sup.Drain(m.Drain.GetShardId())
if err != nil {
a.log.Warn("draining failed", "shard", m.Drain.GetShardId(), "err", err)
}
obs.End(span, err)
case *llmgwv1.ControlMessage_RemoveWeights:
a.removeWeights(m.RemoveWeights.GetPaths())
case *llmgwv1.ControlMessage_Provision:
@@ -927,6 +949,25 @@ func localAddr(ip netip.Addr) bool {
func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan<- *llmgwv1.AgentMessage) {
shardID, replicaID := m.GetShardId(), m.GetReplicaId()
// This node's whole part of the start, from the command arriving to the
// engine answering, beneath the control plane's span that sent it. The
// steps — weights, launch, readiness — are its children.
ctx, span := obs.Tracer(tracerName).Start(ctx, "start shard",
trace.WithSpanKind(trace.SpanKindConsumer),
trace.WithAttributes(
attribute.String("llmgw.node", a.cfg.NodeID),
attribute.String("llmgw.replica", replicaID),
attribute.String("llmgw.shard", shardID),
attribute.Int("llmgw.rank", int(m.GetRank())),
attribute.String("llmgw.model", m.GetProfile().GetServedModelName()),
attribute.String("llmgw.profile", m.GetProfile().GetName()),
attribute.String("llmgw.engine", m.GetProfile().GetEngine()),
attribute.String("llmgw.gpu_class", m.GetProfile().GetGpuClass()),
attribute.IntSlice("llmgw.gpus", int32Ints(m.GetGpuIndices())),
))
var failed error
defer func() { obs.End(span, failed) }()
a.mu.Lock()
a.shardReplica[shardID] = replicaID
a.mu.Unlock()
@@ -939,6 +980,9 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan<
}
}
report := func(st supervisor.State, errMsg string, tail []string) {
if st == supervisor.Failed {
failed = errors.New(errMsg)
}
send(&llmgwv1.ShardStatus{
State: agentsvc.StateToWire(st), Error: errMsg, LastOutput: tail,
})
@@ -1109,9 +1153,18 @@ func (a *Agent) stopShard(ctx context.Context, shardID string, out chan<- *llmgw
delete(a.shardReplica, shardID)
a.mu.Unlock()
if err := a.sup.Stop(ctx, shardID); err != nil {
ctx, span := obs.Tracer(tracerName).Start(ctx, "stop shard",
trace.WithSpanKind(trace.SpanKindConsumer),
trace.WithAttributes(
attribute.String("llmgw.node", a.cfg.NodeID),
attribute.String("llmgw.replica", replicaID),
attribute.String("llmgw.shard", shardID),
))
err := a.sup.Stop(ctx, shardID)
if err != nil {
a.log.Warn("stopping a shard failed", "shard", shardID, "err", err)
}
obs.End(span, err)
a.sup.Remove(shardID)
msg := &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Shard{Shard: &llmgwv1.ShardStatus{
+8
View File
@@ -14,6 +14,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
@@ -94,6 +95,13 @@ type Config struct {
Engines engines.Config `koanf:"engines"`
Superv supervisor.Config `koanf:"supervisor"`
// Tracing sends this node's part of each replica start — the weights
// fetched, the engine launched, the wait for it to answer — to an
// OpenTelemetry collector, inside the control plane's trace of that
// start. Give it the collector server.yaml's tracing names. Off unless
// an endpoint is set here or in OTEL_EXPORTER_OTLP_ENDPOINT.
Tracing obs.Tracing `koanf:"tracing"`
Version string
}
+15 -1
View File
@@ -20,7 +20,11 @@ import (
"strings"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// Want is one engine at one version.
@@ -251,7 +255,17 @@ func (in *Installer) ready(st Status, b Build, bin, version string) Status {
//
// Into a scratch directory renamed into place at the end, so a failure halfway
// leaves nothing that looks installed.
func (in *Installer) install(ctx context.Context, r Recipe, w Want, b Build, dir string, st *Status) error {
//
// Traced as "install engine", a trace of its own: an install is not part of
// any one start, and runs once per version rather than once per replica.
func (in *Installer) install(ctx context.Context, r Recipe, w Want, b Build, dir string, st *Status) (err error) {
ctx, span := obs.Tracer("gitea.mynodes.xyz/ilker/llmgw/internal/agent/provision").Start(ctx, "install engine",
trace.WithAttributes(
attribute.String("llmgw.engine", w.Engine),
attribute.String("llmgw.engine_version", w.Version),
attribute.String("llmgw.engine_build", w.Variant),
))
defer func() { obs.End(span, err) }()
if b.Python != "" {
return in.installPython(ctx, r, w, b, dir, st)
}
+75 -6
View File
@@ -18,9 +18,19 @@ import (
"syscall"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// tracerName names the spans the supervisor opens. There is no restart span
// here because the supervisor never restarts an engine on its own: a restart
// is the control plane's decision, traced there, and reaches a node as a stop
// and a start.
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
// Config configures the supervisor.
type Config struct {
// WorkDir holds pid files and per-shard scratch space.
@@ -195,7 +205,25 @@ var ErrAlreadyRunning = errors.New("shard is already running")
// It is idempotent by ID: a repeated StartShard for the same shard is a no-op
// rather than a second process, because the control plane retries commands
// (plan §5.6).
func (s *Supervisor) Start(ctx context.Context, id string, spec engines.ShardSpec, cmds []engines.Command) (*Shard, error) {
//
// Traced as "launch engine": from the first process to the engine's being
// started, which for a multi-node worker includes waiting for its leader's
// rendezvous point. Loading the model is the readiness wait's, not this.
func (s *Supervisor) Start(ctx context.Context, id string, spec engines.ShardSpec, cmds []engines.Command) (_ *Shard, err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "launch engine", trace.WithAttributes(
attribute.String("llmgw.shard", id),
attribute.String("llmgw.engine", spec.Profile.Engine),
attribute.Int("llmgw.processes", len(cmds)),
))
defer func() {
if errors.Is(err, ErrAlreadyRunning) {
// A repeated command, which is not this start failing.
span.SetAttributes(attribute.Bool("llmgw.already_running", true))
span.End()
return
}
obs.End(span, err)
}()
if len(cmds) == 0 {
return nil, fmt.Errorf("shard %s: no commands to run", id)
}
@@ -301,6 +329,7 @@ func (s *Supervisor) launch(ctx context.Context, sh *Shard, cmds []engines.Comma
sh.pid = c.Process.Pid
sh.state = Starting
sh.mu.Unlock()
trace.SpanFromContext(ctx).SetAttributes(attribute.Int("llmgw.pid", c.Process.Pid))
if err := s.writeState(sh); err != nil {
s.log.Warn("could not write pid file; adoption after a restart will not work",
@@ -337,7 +366,13 @@ func (s *Supervisor) spawn(ctx context.Context, sh *Shard, c engines.Command) (*
}
// waitForPort blocks until addr accepts a connection.
func waitForPort(ctx context.Context, addr string, timeout time.Duration) error {
//
// Traced: a worker's start that took ten minutes because its leader was
// still loading reads as exactly that.
func waitForPort(ctx context.Context, addr string, timeout time.Duration) (err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "wait for rendezvous",
trace.WithAttributes(attribute.String("llmgw.rendezvous", addr)))
defer func() { obs.End(span, err) }()
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -416,7 +451,26 @@ func (s *Supervisor) wait(sh *Shard) {
// The timeout comes from the profile because it is model-dependent: loading a
// 70B model across two nodes legitimately takes tens of minutes, while a small
// model failing to load for ten minutes means something is wrong.
func (s *Supervisor) WaitReady(ctx context.Context, sh *Shard, baseURL, model string, timeout time.Duration) error {
//
// Traced as "wait for engine": the model loading, which is usually most of a
// start. The span says how many times the engine was asked and what it last
// answered, so one that never came up says why.
func (s *Supervisor) WaitReady(ctx context.Context, sh *Shard, baseURL, model string, timeout time.Duration) (err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "wait for engine", trace.WithAttributes(
attribute.String("llmgw.shard", sh.ID),
attribute.String("llmgw.readiness", "health check"),
attribute.String("llmgw.timeout", timeout.String()),
))
polls := 0
var answer error
defer func() {
span.SetAttributes(attribute.Int("llmgw.polls", polls))
if answer != nil && err != nil {
span.SetAttributes(attribute.String("llmgw.last_answer", answer.Error()))
}
obs.End(span, err)
}()
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
@@ -432,7 +486,8 @@ func (s *Supervisor) WaitReady(ctx context.Context, sh *Shard, baseURL, model st
last := strings.Join(sh.Logs(10), "\n")
return fmt.Errorf("shard %s not ready within %v; last output:\n%s", sh.ID, timeout, last)
case <-t.C:
if err := s.health.Ready(ctx, baseURL, model, sh.Spec.APIKey); err != nil {
polls++
if answer = s.health.Ready(ctx, baseURL, model, sh.Spec.APIKey); answer != nil {
continue
}
sh.setState(Ready, nil)
@@ -449,7 +504,13 @@ func (s *Supervisor) WaitReady(ctx context.Context, sh *Shard, baseURL, model st
// leader answering, which only happens once every worker has joined. Polling a
// worker's port instead would time out and fail a replica that is coming up
// perfectly well.
func (s *Supervisor) WaitAlive(ctx context.Context, sh *Shard) error {
func (s *Supervisor) WaitAlive(ctx context.Context, sh *Shard) (err error) {
_, span := obs.Tracer(tracerName).Start(ctx, "wait for engine", trace.WithAttributes(
attribute.String("llmgw.shard", sh.ID),
attribute.String("llmgw.readiness", "stays alive"),
attribute.String("llmgw.timeout", s.cfg.AliveGrace.String()),
))
defer func() { obs.End(span, err) }()
sh.setState(Loading, nil)
select {
@@ -483,7 +544,14 @@ func (s *Supervisor) Drain(id string) error {
// Stop terminates a shard: SIGTERM to the whole group, then SIGKILL after
// StopTimeout.
func (s *Supervisor) Stop(ctx context.Context, id string) error {
//
// Traced as "stop engine", saying whether it had to be killed: an engine
// that ignores SIGTERM for a minute is a minute its replacement waits for
// the GPU.
func (s *Supervisor) Stop(ctx context.Context, id string) (err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "stop engine",
trace.WithAttributes(attribute.String("llmgw.shard", id)))
defer func() { obs.End(span, err) }()
sh, err := s.Get(id)
if err != nil {
return err
@@ -527,6 +595,7 @@ func (s *Supervisor) Stop(ctx context.Context, id string) error {
// force kills the group outright. exec.Cmd's cancel does this for shards we
// started; an adopted shard has none, so signal its group directly.
force := func() {
span.SetAttributes(attribute.Bool("llmgw.killed", true))
if sh.cancel != nil {
sh.cancel()
return
+135
View File
@@ -0,0 +1,135 @@
package agent
import (
"testing"
"time"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs/obstest"
)
func TestAStartOnTheNodeIsInTheControlPlanesTrace(t *testing.T) {
rec := obstest.Record(t)
h := start(t, func(c *Config) {
c.Engines.VLLM.Bin = servingEngine(t)
})
// The control plane's side, as the reconciler sends it: the command
// dispatched under a span of its own, over the real stream.
ctx, sent := obs.Tracer("test").Start(t.Context(), "send start",
trace.WithSpanKind(trace.SpanKindProducer))
if err := h.svc.StartShard(ctx, shardCommand(t, "shard-a")); err != nil {
t.Fatal(err)
}
sent.End()
h.rec.await(t, 30*time.Second, func(s status) bool {
return s.shardID == "shard-a" && s.state == supervisor.Ready
})
// The node's part ends just after it reports the shard ready.
tid := sent.SpanContext().TraceID()
node := rec.Await(t, tid, "start shard", 5*time.Second)
if node.Parent().SpanID() != sent.SpanContext().SpanID() {
t.Fatal("the node's start is not beneath the control plane's command")
}
if node.SpanKind() != trace.SpanKindConsumer {
t.Errorf("the node's start is a %v span", node.SpanKind())
}
if a := obstest.Attrs(node); a["llmgw.node"] != "n1" || a["llmgw.replica"] != "rep-1" ||
a["llmgw.model"] != "test-model" || a["llmgw.engine"] != "vllm" {
t.Errorf("the node's start does not say what it started: %v", a)
}
if node.Status().Code == codes.Error {
t.Errorf("a start that served is %v", node.Status())
}
for _, name := range []string{"fetch weights", "launch engine", "wait for engine"} {
if !obstest.ChildOf(rec.One(t, tid, name), node) {
t.Errorf("%s is not beneath the node's start", name)
}
}
if a := obstest.Attrs(rec.One(t, tid, "fetch weights")); a["llmgw.weights.outcome"] != "local" {
t.Errorf("the weights do not say how they were found: %v", a)
}
if a := obstest.Attrs(rec.One(t, tid, "wait for engine")); a["llmgw.polls"] == "" || a["llmgw.polls"] == "0" {
t.Errorf("the readiness wait does not say how often it asked: %v", a)
}
// A stop joins the trace it was sent in, too.
sctx, stop := obs.Tracer("test").Start(t.Context(), "stop replica")
if err := h.svc.StopShard(sctx, "n1", "shard-a"); err != nil {
t.Fatal(err)
}
stop.End()
h.rec.await(t, 15*time.Second, func(s status) bool {
return s.shardID == "shard-a" && s.state == supervisor.Stopped
})
stid := stop.SpanContext().TraceID()
nodeStop := rec.Await(t, stid, "stop shard", 5*time.Second)
if !obstest.ChildOf(nodeStop, rec.One(t, stid, "stop replica")) {
t.Error("the node's stop is not beneath the control plane's")
}
if !obstest.ChildOf(rec.One(t, stid, "stop engine"), nodeStop) {
t.Error("stopping the engine is not beneath the node's stop")
}
}
func TestAnAdoptionIsTracedAndItsStopsAreInIt(t *testing.T) {
rec := obstest.Record(t)
h := start(t, func(c *Config) {
c.Engines.VLLM.Bin = fakeEngineScript(t, `sleep 60`)
})
if err := h.svc.StartShard(t.Context(), shardCommand(t, "shard-orphaned")); err != nil {
t.Fatal(err)
}
h.rec.await(t, 15*time.Second, func(s status) bool {
return s.shardID == "shard-orphaned" && s.state == supervisor.Starting
})
// The control plane, restarted, claims nothing the node is running.
h.rec.markUnclaimed("shard-orphaned")
h.ln.drop()
h.rec.await(t, 20*time.Second, func(s status) bool {
return s.shardID == "shard-orphaned" && s.state == supervisor.Stopped
})
adopt := rec.Named("adopt")
if len(adopt) != 1 {
t.Fatalf("%d adoptions traced", len(adopt))
}
if a := obstest.Attrs(adopt[0]); a["llmgw.node"] != "n1" || a["llmgw.shards"] != "1" || a["llmgw.unclaimed"] != "1" {
t.Errorf("the adoption does not say what it found: %v", a)
}
stop := rec.Await(t, adopt[0].SpanContext().TraceID(), "stop shard", 5*time.Second)
if !obstest.ChildOf(stop, adopt[0]) {
t.Error("stopping the unclaimed shard is not beneath the adoption")
}
}
func TestAStartThatFailsOnTheNodeSaysWhyInTheTrace(t *testing.T) {
rec := obstest.Record(t)
h := start(t, func(c *Config) {
c.Engines.VLLM.Bin = fakeEngineScript(t, `echo "CUDA out of memory" >&2; exit 1`)
})
ctx, sent := obs.Tracer("test").Start(t.Context(), "send start")
if err := h.svc.StartShard(ctx, shardCommand(t, "shard-b")); err != nil {
t.Fatal(err)
}
sent.End()
h.rec.await(t, 30*time.Second, func(s status) bool {
return s.shardID == "shard-b" && s.state == supervisor.Failed
})
tid := sent.SpanContext().TraceID()
node := rec.Await(t, tid, "start shard", 5*time.Second)
if node.Status().Code != codes.Error {
t.Errorf("a start that failed is %v", node.Status())
}
if wait := rec.One(t, tid, "wait for engine"); wait.Status().Code != codes.Error {
t.Errorf("the readiness wait of an engine that exited is %v", wait.Status())
}
}
+59 -2
View File
@@ -26,6 +26,11 @@ import (
"strings"
"sync"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// Progress reports how a fetch is going, for anything that wants to say so.
@@ -80,13 +85,48 @@ var ErrNotLocal = errors.New("the weights are not on this node and the source sa
// local disk, a Lustre mount or an image exported over the fabric, and a
// downloader that second-guessed that would copy a hundred gigabytes that were
// already present.
func (f *Fetcher) Ensure(ctx context.Context, source string) (string, error) {
//
// Traced as "fetch weights", with what it came to — the file was local,
// already cached, downloaded, or downloaded by another shard this one waited
// for — since that is most of the difference between a replica that starts
// in a minute and one that takes an hour.
func (f *Fetcher) Ensure(ctx context.Context, source string) (path string, err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "fetch weights",
trace.WithAttributes(attribute.String("llmgw.weights.source", redact(source))))
defer func() {
span.SetAttributes(attribute.String("llmgw.weights.path", path))
obs.End(span, err)
}()
return f.ensure(ctx, source)
}
// tracerName names the spans this package opens.
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights"
// outcome says on the current span how a source was made present.
func outcome(ctx context.Context, how string) {
trace.SpanFromContext(ctx).SetAttributes(attribute.String("llmgw.weights.outcome", how))
}
// redact is a source fit for a trace: a URL without its credentials or
// query, which is where a token would be.
func redact(source string) string {
u, err := url.Parse(source)
if err != nil || u.Scheme == "" || u.Host == "" {
return source
}
u.User, u.RawQuery, u.Fragment = nil, "", ""
return u.String()
}
func (f *Fetcher) ensure(ctx context.Context, source string) (string, error) {
if source == "" {
return "", fmt.Errorf("no source given")
}
// A version rather than a file. Everything below assumes one URL is one
// file, and a set has to be handled before any of it.
if IsSet(source) {
outcome(ctx, "set")
return f.ensureSet(ctx, source)
}
if !isRemote(source) {
@@ -95,9 +135,11 @@ func (f *Fetcher) Ensure(ctx context.Context, source string) (string, error) {
// was never meant to be one. vLLM and SGLang resolve those
// themselves, and doing it for them would be both a duplicate
// downloader and a change in documented behaviour.
outcome(ctx, "left to the engine")
return source, nil
}
if _, err := os.Stat(source); err == nil {
outcome(ctx, "local")
return source, nil
}
return "", fmt.Errorf("%w: %s", ErrNotLocal, source)
@@ -108,6 +150,7 @@ func (f *Fetcher) Ensure(ctx context.Context, source string) (string, error) {
return "", err
}
if _, err := os.Stat(target); err == nil {
outcome(ctx, "cached")
return target, nil
}
@@ -117,6 +160,7 @@ func (f *Fetcher) Ensure(ctx context.Context, source string) (string, error) {
}
if call, running := f.inflight[source]; running {
f.mu.Unlock()
outcome(ctx, "fetched by another shard")
select {
case <-call.done:
return call.path, call.err
@@ -128,6 +172,7 @@ func (f *Fetcher) Ensure(ctx context.Context, source string) (string, error) {
f.inflight[source] = call
f.mu.Unlock()
outcome(ctx, "downloaded")
call.path, call.err = f.fetch(ctx, source, target)
close(call.done)
@@ -147,7 +192,15 @@ func (f *Fetcher) fetch(ctx context.Context, source, target string) (string, err
// The two differ for a set: the shards waiting on it asked for the version, not
// for the fourth file in it, and progress reported under the file's own URL
// reaches nobody. What they see is one number climbing across the whole set.
func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) (string, error) {
//
// Each file is a "download" span, with how many bytes it took and where a
// resumed one started from.
func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) (_ string, err error) {
ctx, span := obs.Tracer(tracerName).Start(ctx, "download",
trace.WithSpanKind(trace.SpanKindClient),
trace.WithAttributes(attribute.String("llmgw.weights.source", redact(source))))
defer func() { obs.End(span, err) }()
url, err := resolveURL(source)
if err != nil {
return "", err
@@ -208,6 +261,10 @@ func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string)
total := existing + resp.ContentLength
f.report(Progress{Source: reportAs, BytesDone: existing, BytesTotal: total})
written, err := f.copyWithProgress(ctx, out, resp.Body, reportAs, existing, total)
span.SetAttributes(
attribute.Int64("llmgw.weights.resumed_from", existing),
attribute.Int64("llmgw.weights.bytes", written),
)
closeErr := out.Close()
if err != nil {
// The part file is kept on purpose, so the next attempt resumes.
+43
View File
@@ -0,0 +1,43 @@
package weights
import (
"strings"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs/obstest"
)
func TestAFetchSaysInItsTraceWhetherItDownloadedAndHowMuch(t *testing.T) {
rec := obstest.Record(t)
srv := serveBytes(t, []byte("forty gigabytes, give or take"), nil)
f := &Fetcher{CacheDir: t.TempDir()}
src := strings.Replace(srv.URL, "http://", "http://user:secret@", 1) + "/m/model.gguf?token=hunter2"
if _, err := f.Ensure(t.Context(), src); err != nil {
t.Fatal(err)
}
fetch := rec.One(t, [16]byte{}, "fetch weights")
a := obstest.Attrs(fetch)
if a["llmgw.weights.outcome"] != "downloaded" {
t.Errorf("the first fetch says %v", a)
}
if strings.Contains(a["llmgw.weights.source"], "secret") || strings.Contains(a["llmgw.weights.source"], "hunter2") {
t.Errorf("the trace holds the source's credentials: %s", a["llmgw.weights.source"])
}
dl := rec.One(t, fetch.SpanContext().TraceID(), "download")
if !obstest.ChildOf(dl, fetch) || obstest.Attrs(dl)["llmgw.weights.bytes"] != "29" {
t.Errorf("the download does not say how much it moved: %v", obstest.Attrs(dl))
}
// The second start on this node finds it here.
if _, err := f.Ensure(t.Context(), src); err != nil {
t.Fatal(err)
}
again := rec.Named("fetch weights")
if len(again) != 2 || obstest.Attrs(again[1])["llmgw.weights.outcome"] != "cached" {
t.Errorf("the second fetch says %v", obstest.Attrs(again[len(again)-1]))
}
if len(rec.Named("download")) != 1 {
t.Error("a cached file was downloaded again")
}
}
+16
View File
@@ -16,6 +16,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/agent"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/version"
)
@@ -64,6 +65,21 @@ func newAgentCmd() *cobra.Command {
}
cfg.Version = version.Version
// Before the agent, so its tracer is the configured one.
shutdownTracing, err := obs.SetupTracing(cmd.Context(), obs.TracingOptions{
Endpoint: cfg.Tracing.Endpoint, SampleRatio: cfg.Tracing.SampleRatio,
Service: "llmgw-agent", Version: version.Version, Instance: cfg.NodeID,
})
if err != nil {
return err
}
defer func() {
// Bounded: a collector that is gone must not hold up a shutdown.
sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_ = shutdownTracing(sctx)
}()
httpClient, err := agent.Client(cfg.TLS, 30*time.Second)
if err != nil {
return err
+5 -2
View File
@@ -9,6 +9,7 @@ import (
"maps"
"net"
"net/http"
"os"
"os/signal"
"slices"
"strings"
@@ -101,10 +102,12 @@ func newServerCmd() *cobra.Command {
"or give --endpoint for a router-only run")
}
// Before the router, so its tracer is the configured one.
// Before the router, so its tracer is the configured one. The
// host tells two control planes' traces apart.
host, _ := os.Hostname()
shutdownTracing, err := obs.SetupTracing(cmd.Context(), obs.TracingOptions{
Endpoint: cfg.Tracing.Endpoint, SampleRatio: cfg.Tracing.SampleRatio,
Service: "llmgw-server", Version: version.Version,
Service: "llmgw-server", Version: version.Version, Instance: host,
})
if err != nil {
return err
+5 -18
View File
@@ -20,6 +20,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
@@ -93,10 +94,11 @@ type Server struct {
// Monitoring is where the dashboard reads its history from.
Monitoring Monitoring `koanf:"monitoring"`
// Tracing sends a trace of each inference request to an OpenTelemetry
// collector. Off unless an endpoint is set here or in
// Tracing sends traces to an OpenTelemetry collector: each inference
// request, and on a control plane each placement, replica start and
// model job. Off unless an endpoint is set here or in
// OTEL_EXPORTER_OTLP_ENDPOINT.
Tracing Tracing `koanf:"tracing"`
Tracing obs.Tracing `koanf:"tracing"`
// Mirror points the control plane at a model mirror it can administer.
// Empty means there is none, and /admin/mirror stays unmounted.
@@ -182,21 +184,6 @@ type Monitoring struct {
Timeout time.Duration `koanf:"timeout"`
}
// Tracing is where request traces go, over OTLP/HTTP.
//
// One span per inference request, with admission, scheduling and each attempt
// at an engine beneath it, and the trace carried on to the engine in a
// traceparent header. The request's X-Request-Id is on the span, so a trace
// and `llmgw ctl request` name the same request.
type Tracing struct {
// Endpoint is the collector, for example http://otel.internal:4318.
// Empty falls back to OTEL_EXPORTER_OTLP_ENDPOINT, and to no tracing.
Endpoint string `koanf:"endpoint"`
// SampleRatio is the share of requests traced, 0 to 1, when the caller
// did not decide it with a sampled traceparent of its own. Zero takes 1.
SampleRatio float64 `koanf:"sample_ratio"`
}
// Mirror is a model mirror this control plane administers over its API.
//
// The mirror is a separate machine with a large disk, not part of the gateway,
+34 -7
View File
@@ -14,6 +14,8 @@ import (
"time"
"connectrpc.com/connect"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/trace"
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
"gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1/llmgwv1connect"
@@ -22,6 +24,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// StatusSink receives shard transitions and recovers replica state from what
@@ -311,7 +314,20 @@ func (s *Service) Connect(
// unreachable in production while its unit tests, which call the reconciler
// directly, kept passing.
func (s *Service) adopt(ctx context.Context, nodeID string, shards []*llmgwv1.RunningShard, c *conn) {
// A trace of its own, and only for a node that reported something: a
// node running nothing registering is not an event anyone looks for.
tctx, span := ctx, trace.SpanFromContext(context.Background()) // the no-op span
if len(shards) > 0 {
tctx, span = obs.Tracer(tracerName).Start(ctx, "adopt",
trace.WithNewRoot(),
trace.WithAttributes(
attribute.String("llmgw.node", nodeID),
attribute.Int("llmgw.shards", len(shards)),
))
}
defer span.End()
unclaimed := s.status.Adopt(nodeID, AdoptedFromRegister(nodeID, shards))
span.SetAttributes(attribute.Int("llmgw.unclaimed", len(unclaimed)))
for _, u := range unclaimed {
// The model or profile left the catalog while this was running. Nobody
// will ever route to it, and it is holding GPUs.
@@ -319,7 +335,8 @@ func (s *Service) adopt(ctx context.Context, nodeID string, shards []*llmgwv1.Ru
"node", nodeID, "shard", u.ShardID, "model", u.Model, "profile", u.Profile)
select {
case c.send <- &llmgwv1.ControlMessage{
Msg: &llmgwv1.ControlMessage_Stop{Stop: &llmgwv1.StopShard{ShardId: u.ShardID}},
Msg: &llmgwv1.ControlMessage_Stop{Stop: &llmgwv1.StopShard{ShardId: u.ShardID}},
Trace: obs.Inject(tctx),
}:
case <-ctx.Done():
return
@@ -715,27 +732,37 @@ type Leader interface{ IsLeader() bool }
// and what every deployment before leadership had.
func (s *Service) SetLeadership(l Leader) { s.leader = l }
// tracerName names the spans this package opens.
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
// ErrNotConnected means no agent stream is open for that node.
var ErrNotConnected = errors.New("no agent connected")
// StartShard implements reconciler.Dispatcher.
func (s *Service) StartShard(_ context.Context, cmd reconciler.ShardCommand) error {
//
// The command carries the trace of ctx, so the agent's spans for this start
// — the weights, the engine's launch, its readiness — are in the trace of
// the replica they are starting rather than in one of their own.
func (s *Service) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error {
return s.send(cmd.NodeID, &llmgwv1.ControlMessage{
Msg: &llmgwv1.ControlMessage_Start{Start: StartShardFromCommand(cmd)},
Msg: &llmgwv1.ControlMessage_Start{Start: StartShardFromCommand(cmd)},
Trace: obs.Inject(ctx),
})
}
// StopShard implements reconciler.Dispatcher.
func (s *Service) StopShard(_ context.Context, nodeID, shardID string) error {
func (s *Service) StopShard(ctx context.Context, nodeID, shardID string) error {
return s.send(nodeID, &llmgwv1.ControlMessage{
Msg: &llmgwv1.ControlMessage_Stop{Stop: &llmgwv1.StopShard{ShardId: shardID}},
Msg: &llmgwv1.ControlMessage_Stop{Stop: &llmgwv1.StopShard{ShardId: shardID}},
Trace: obs.Inject(ctx),
})
}
// DrainShard implements reconciler.Dispatcher.
func (s *Service) DrainShard(_ context.Context, nodeID, shardID string) error {
func (s *Service) DrainShard(ctx context.Context, nodeID, shardID string) error {
return s.send(nodeID, &llmgwv1.ControlMessage{
Msg: &llmgwv1.ControlMessage_Drain{Drain: &llmgwv1.DrainShard{ShardId: shardID}},
Msg: &llmgwv1.ControlMessage_Drain{Drain: &llmgwv1.DrainShard{ShardId: shardID}},
Trace: obs.Inject(ctx),
})
}
@@ -172,6 +172,9 @@ type Job struct {
// how fast it said it.
Reply string `json:"reply,omitempty"`
TokensPerSecond float64 `json:"tokens_per_second,omitempty"`
// trace is the job's spans, when tracing is on.
trace jobTrace
}
// Server runs jobs and remembers them.
@@ -318,6 +321,7 @@ func (s *Server) snapshot(j *Job) Job {
func (s *Server) step(j *Job, name, state, detail string) {
s.mu.Lock()
defer s.mu.Unlock()
j.traceStep(name, state, detail)
if n := len(j.Steps); n > 0 && j.Steps[n-1].Name == name {
j.Steps[n-1].State, j.Steps[n-1].Detail, j.Steps[n-1].At = state, detail, time.Now()
return
@@ -329,6 +333,7 @@ func (s *Server) fail(j *Job, name string, err error) {
s.step(j, name, Failed, err.Error())
s.mu.Lock()
j.State, j.Error = Failed, err.Error()
j.traceEnd(err)
s.mu.Unlock()
s.deps.Log.Warn("serving a model failed", "job", j.ID, "repo", j.Request.Repo, "step", name, "err", err)
}
@@ -617,6 +622,7 @@ func (s *Server) addAndStart(ctx context.Context, j *Job, repo hf.Repo, set hf.F
func (s *Server) finish(j *Job) {
s.mu.Lock()
j.State = Done
j.traceEnd(nil)
s.mu.Unlock()
s.deps.Log.Info("model served", "job", j.ID, "repo", j.Request.Repo, "model", j.Model)
}
@@ -785,6 +791,7 @@ func (s *Server) waitReady(ctx context.Context, j *Job, name string, hardware bo
continue
}
if r.State == reconciler.Ready {
s.linkReplica(j, r)
s.step(j, "placement", Done, fmt.Sprintf("%s ready on %s at %s", r.ID,
strings.Join(r.NodeIDs(), ","), r.Endpoint))
return r, nil
+99
View File
@@ -0,0 +1,99 @@
package modelserve
import (
"context"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// A job is a trace of its own, and each of its steps a span beneath it —
// registry, mirror, check, catalog, placement, smoke test — opened when the
// step begins and closed when it is done, failed or skipped. The steps are
// what a job already records for `ctl model jobs`; the trace adds where the
// time went, and from the placement step a link to the replica's own start,
// whose trace goes on down to the node.
//
// Begun by the first step rather than where a job is made, so every kind of
// job is traced by recording steps and nothing else.
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
// jobTrace is a job's spans. Read and written with Server.mu held.
type jobTrace struct {
span trace.Span
step trace.Span
stepName string
}
// traceStep records a step on the job's trace. Called with s.mu held.
func (j *Job) traceStep(name, state, detail string) {
if j.trace.span == nil {
kind := j.Kind
if kind == "" {
kind = "serve"
}
_, j.trace.span = obs.Tracer(tracerName).Start(context.Background(), kind+" model",
trace.WithAttributes(
attribute.String("llmgw.job", j.ID),
attribute.String("llmgw.repo", j.Request.Repo),
attribute.String("llmgw.model", j.Model),
))
}
if j.trace.step != nil && j.trace.stepName != name {
// A step left running when the next began: over, whatever it said.
j.trace.step.End()
j.trace.step = nil
}
if j.trace.step == nil {
ctx := trace.ContextWithSpan(context.Background(), j.trace.span)
_, j.trace.step = obs.Tracer(tracerName).Start(ctx, name)
j.trace.stepName = name
}
span := j.trace.step
span.SetAttributes(attribute.String("llmgw.state", state), attribute.String("llmgw.detail", detail))
switch state {
case Running:
return
case Failed:
span.SetStatus(codes.Error, detail)
}
span.End()
j.trace.step = nil
}
// traceEnd closes the job's trace, failed when err is set. Called with s.mu
// held.
func (j *Job) traceEnd(err error) {
if j.trace.step != nil {
j.trace.step.End()
j.trace.step = nil
}
if j.trace.span == nil {
return
}
if j.Model != "" {
j.trace.span.SetAttributes(attribute.String("llmgw.model", j.Model))
}
obs.End(j.trace.span, err)
j.trace.span = nil
}
// linkReplica links the step in progress to the start of a replica it waited
// for, which is in the reconciler's trace rather than this one.
func (s *Server) linkReplica(j *Job, r reconciler.Replica) {
sc := r.StartTrace()
if !sc.IsValid() {
return
}
s.mu.Lock()
defer s.mu.Unlock()
if j.trace.step != nil {
j.trace.step.AddLink(trace.Link{SpanContext: sc,
Attributes: []attribute.KeyValue{attribute.String("llmgw.replica", r.ID)}})
}
}
+65
View File
@@ -0,0 +1,65 @@
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")
}
}
+3
View File
@@ -609,6 +609,9 @@ func (s *Server) replace(ctx context.Context, j *Job, name string) error {
}
}
if left == 0 && len(ready) > 0 {
for _, r := range ready {
s.linkReplica(j, r)
}
s.step(j, "replicas", Done, fmt.Sprintf("%d replaced", len(old)))
s.step(j, "smoke test", Running, "asking "+ready[0].ID+" to say hello")
reply, tps, err := s.smoke(ctx, ready[0], name)
+11 -1
View File
@@ -6,6 +6,8 @@ import (
"sort"
"time"
"go.opentelemetry.io/otel/attribute"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
)
@@ -53,7 +55,15 @@ func (r *Reconciler) makeRoomFor(ctx context.Context, m catalog.Model) bool {
r.log.Info("evicting an idle replica to make room",
"stopping", best.ID, "idle_model", best.Model,
"idle", best.Idle.Round(time.Second), "gpus", best.GPUs, "for", m.Name)
r.stopReplica(ctx, best.ID)
ctx, span := startSpan(ctx, "evict",
attribute.String("llmgw.for", m.Name),
attribute.String("llmgw.replica", best.ID),
attribute.String("llmgw.model", best.Model),
attribute.Float64("llmgw.idle_seconds", best.Idle.Seconds()),
attribute.Int("llmgw.gpus", best.GPUs),
attribute.Int("llmgw.candidates", len(candidates)))
defer span.End()
r.stopReplica(ctx, best.ID, "evicted, idle, to make room for "+m.Name)
return true
}
+16 -1
View File
@@ -2,10 +2,13 @@ package reconciler
import (
"context"
"fmt"
"slices"
"sort"
"time"
"go.opentelemetry.io/otel/attribute"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/placement"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/quota"
@@ -113,12 +116,24 @@ func (r *Reconciler) preempt(ctx context.Context, m catalog.Model, chosen []vict
r.preempting[m.Name] = ids
r.mu.Unlock()
victims := make([]string, 0, len(chosen))
for _, v := range chosen {
r.log.Warn("preempting a replica to make room",
"replica", v.ID, "model", v.Model, "class", v.Class, "above_min", v.AboveMin,
"for", m.Name, "for_class", m.EffectiveClass())
desc := fmt.Sprintf("%s (%s, %s)", v.ID, v.Model, v.Class)
if v.AboveMin {
desc = fmt.Sprintf("%s (%s, %s, above its minimum)", v.ID, v.Model, v.Class)
}
victims = append(victims, desc)
}
r.scaleDown(ctx, ids)
ctx, span := startSpan(ctx, "preempt",
attribute.String("llmgw.for", m.Name),
attribute.String("llmgw.for_class", m.EffectiveClass()),
attribute.StringSlice("llmgw.victims", victims),
attribute.String("llmgw.hold", r.cfg.PreemptHold.String()))
defer span.End()
r.scaleDown(ctx, ids, "preempted for "+m.Name)
}
// awaitingPreemption reports whether replicas stopped for a model are still
+184 -25
View File
@@ -11,6 +11,10 @@ import (
"sync/atomic"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
@@ -18,6 +22,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/placement"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/quota"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// ShardCommand is everything an agent needs to launch one shard.
@@ -402,7 +407,9 @@ func (r *Reconciler) Reconcile(ctx context.Context) {
if r.leader != nil && !r.leader.IsLeader() {
return
}
r.expireNodes()
ctx, p := beginPass(ctx)
defer p.end()
r.expireNodes(ctx)
r.expireStuckReplicas()
r.drainCordonedNodes(ctx)
r.reapFailed(ctx)
@@ -449,7 +456,7 @@ func (r *Reconciler) stopReplicasOfRemovedModels(ctx context.Context) {
for _, rep := range gone {
r.log.Info("stopping a replica whose model left the catalog",
"replica", rep.ID, "model", rep.Model)
r.stopReplica(ctx, rep.ID)
r.stopReplica(ctx, rep.ID, "its model left the catalog")
}
}
@@ -520,6 +527,7 @@ func (r *Reconciler) expireStuckReplicas() {
}
rep.State = Failed
rep.Err = fmt.Sprintf("did not come up within %s", r.startupTimeout(rep))
rep.startDone(now)
stuck = append(stuck, rep)
}
r.mu.Unlock()
@@ -552,7 +560,7 @@ func (r *Reconciler) startupTimeout(rep *Replica) time.Duration {
// The two-stage delay is deliberate: a brief network blip must not tear down
// replicas that are still serving traffic perfectly well, and rebuilding a
// multi-node replica costs tens of minutes of model loading.
func (r *Reconciler) expireNodes() {
func (r *Reconciler) expireNodes(ctx context.Context) {
now := r.now()
for _, n := range r.inv.Nodes() {
silent := now.Sub(n.LastSeen)
@@ -560,14 +568,21 @@ func (r *Reconciler) expireNodes() {
case n.State == inventory.Ready && silent > r.cfg.UnreachableAfter:
r.inv.SetState(n.ID, inventory.Unreachable)
r.log.Warn("node is unreachable", "node", n.ID, "silent_for", silent.Round(time.Second))
_, span := startSpan(ctx, "node unreachable", attribute.String("llmgw.node", n.ID),
attribute.Float64("llmgw.silent_seconds", silent.Seconds()))
span.End()
case n.State == inventory.Unreachable && silent > r.cfg.LostAfter:
_, span := startSpan(ctx, "node lost", attribute.String("llmgw.node", n.ID),
attribute.Float64("llmgw.silent_seconds", silent.Seconds()))
r.inv.SetState(n.ID, inventory.Lost)
r.log.Error("node lost; its replicas will be rebuilt elsewhere",
"node", n.ID, "silent_for", silent.Round(time.Second))
r.notify.Load().Notify(notify.NodeLost, n.ID,
fmt.Sprintf("node %s was lost: silent for %s; its replicas are rebuilt elsewhere", n.ID, silent.Round(time.Second)),
map[string]any{"node": n.ID, "silent_seconds": int(silent.Seconds())})
r.failReplicasOn(n.ID, "node "+n.ID+" was lost")
span.SetAttributes(attribute.StringSlice("llmgw.replicas_failed",
r.failReplicasOn(n.ID, "node "+n.ID+" was lost")))
span.End()
}
}
}
@@ -606,24 +621,26 @@ func (r *Reconciler) drainCordonedNodes(ctx context.Context) {
for _, rep := range toDrain {
r.log.Info("draining replica off a node being taken out of service",
"replica", rep.ID, "nodes", rep.NodeIDs())
for _, s := range shards[rep.ID] {
if err := r.disp.DrainShard(ctx, s.NodeID, s.ID); err != nil {
r.log.Warn("draining a shard failed", "shard", s.ID, "err", err)
}
}
r.drainShards(ctx, rep.ID, shards[rep.ID], "its node is being drained")
}
}
func (r *Reconciler) failReplicasOn(nodeID, reason string) {
// failReplicasOn fails every replica on a node, and says which.
func (r *Reconciler) failReplicasOn(nodeID, reason string) []string {
r.mu.Lock()
defer r.mu.Unlock()
var ids []string
for _, rep := range r.replicas {
if rep.State.Terminal() || !rep.OnNode(nodeID) {
continue
}
rep.State = Failed
rep.Err = reason
rep.startDone(r.now())
ids = append(ids, rep.ID)
}
slices.Sort(ids)
return ids
}
// reapFailed releases the resources of terminal replicas and tells the agents
@@ -636,8 +653,10 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
doomed = append(doomed, rep)
}
}
parkedNow := map[string]bool{}
for _, rep := range doomed {
delete(r.replicas, rep.ID)
rep.startDone(r.now())
if rep.State == Failed {
// Carry the failure count on the model, not the replica: the
// replica is about to disappear, and the backoff has to survive it.
@@ -649,6 +668,7 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
r.parked[rep.Model] = parkRecord{
Since: r.now(), Reason: rep.Err, Failures: n, fp: modelFingerprint(m),
}
parkedNow[rep.ID] = true
r.log.Error("model parked: it failed every time it started, and is not retried "+
"until its definition changes or `llmgw ctl retry` is run",
"model", rep.Model, "failures", n, "last_error", rep.Err)
@@ -664,23 +684,52 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
for _, rep := range doomed {
// rep is out of the map now, so no agent report can reach it any more.
sctx, span := startSpan(ctx, "reap replica", replicaAttrs(rep)...)
span.SetAttributes(attribute.String("llmgw.state", rep.State.String()))
if rep.State == Failed {
r.mu.Lock()
retry := r.backoff(r.attempts[rep.Model])
attempts := r.attempts[rep.Model]
retry := r.backoff(attempts)
r.mu.Unlock()
r.log.Error("replica failed",
"replica", rep.ID, "model", rep.Model, "err", rep.Err,
"retry_in", retry.Round(time.Second))
span.SetAttributes(
attribute.String("llmgw.error", rep.Err),
attribute.Int("llmgw.failures_in_a_row", attempts),
attribute.Float64("llmgw.retry_in_seconds", retry.Seconds()),
attribute.Bool("llmgw.parked", parkedNow[rep.ID]),
)
}
// Stop every shard, including the ones that looked healthy: a partially
// alive replica holds GPUs while serving nothing.
for _, s := range rep.Shards {
if err := r.disp.StopShard(ctx, s.NodeID, s.ID); err != nil {
if err := r.disp.StopShard(sctx, s.NodeID, s.ID); err != nil {
r.log.Warn("stopping a shard of a dead replica failed",
"replica", rep.ID, "shard", s.ID, "err", err)
}
}
r.inv.ReleaseReplica(rep.ID)
span.End()
}
}
// drainShards tells a replica's nodes to stop routing to its shards, as one
// "drain replica" span saying why.
func (r *Reconciler) drainShards(ctx context.Context, id string, shards []Shard, why string) {
nodes := make([]string, 0, len(shards))
for _, s := range shards {
nodes = append(nodes, s.NodeID)
}
ctx, span := startSpan(ctx, "drain replica",
attribute.String("llmgw.replica", id),
attribute.StringSlice("llmgw.nodes", nodes),
attribute.String("llmgw.reason", why))
defer span.End()
for _, s := range shards {
if err := r.disp.DrainShard(ctx, s.NodeID, s.ID); err != nil {
r.log.Warn("draining a shard failed", "shard", s.ID, "err", err)
}
}
}
@@ -724,12 +773,14 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
}
}
// The next restart an operator queued, once the last one is done.
restarting := ""
if len(retiringIDs) == 0 {
if id, ok := r.nextRestartLocked(m.Name); ok {
r.replicas[id].RetiringAt = r.now()
retiringIDs = append(retiringIDs, id)
liveIDs = slices.DeleteFunc(liveIDs, func(x string) bool { return x == id })
readyLive--
restarting = id
r.log.Info("restarting a replica: placing its replacement first",
"replica", id, "model", m.Name)
}
@@ -746,6 +797,11 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
r.log.Info("model unparked: its definition changed", "model", m.Name)
}
r.mu.Unlock()
if restarting != "" {
_, span := startSpan(ctx, "restart replica",
attribute.String("llmgw.replica", restarting), attribute.String("llmgw.model", m.Name))
span.End()
}
desired := r.desiredFor(m)
outcome := notTried
@@ -762,9 +818,20 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
// the real error in noise.
return
}
outcome = r.scaleUp(ctx, m, avoid)
sctx, span := startSpan(ctx, "scale up",
attribute.String("llmgw.model", m.Name),
attribute.Int("llmgw.replicas", len(liveIDs)),
attribute.Int("llmgw.desired", desired))
outcome = r.scaleUp(sctx, m, avoid)
span.SetAttributes(attribute.String("llmgw.outcome", outcome.String()))
span.End()
case len(liveIDs) > desired:
r.scaleDown(ctx, liveIDs[desired:])
sctx, span := startSpan(ctx, "scale down",
attribute.String("llmgw.model", m.Name),
attribute.Int("llmgw.replicas", len(liveIDs)),
attribute.Int("llmgw.desired", desired))
r.scaleDown(sctx, liveIDs[desired:], "more replicas than desired")
span.End()
}
// A drained replica is stopped once its replacement is serving, or once the
@@ -787,17 +854,17 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
switch {
case !ok:
case replaced:
r.stopReplica(ctx, id)
r.stopReplica(ctx, id, "drained, and its replacement serves")
case expired && preemptedFor != "":
r.log.Info("stopping a preempted replica", "replica", id,
"for", preemptedFor, "after", grace)
r.stopReplica(ctx, id)
r.stopReplica(ctx, id, "preempted for "+preemptedFor+", and its grace is over")
case expired:
// The replacement never arrived — often because nowhere else can
// host it. Draining forever would leave the node undrainable.
r.log.Warn("stopping a drained replica without a replacement",
"replica", id, "after", grace)
r.stopReplica(ctx, id)
r.stopReplica(ctx, id, "drained, and no replacement arrived within "+grace.String())
}
}
@@ -810,6 +877,9 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
// allocated, and starting several at once from the same inventory snapshot
// would hand them the same GPUs.
func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []string) scaleOutcome {
_, pspan := startSpan(ctx, "place",
attribute.String("llmgw.model", m.Name),
attribute.StringSlice("llmgw.avoid", avoid))
profiles := r.orderedProfiles(m)
res := r.sched.Place(profiles, avoid)
if res.Placement == nil && len(avoid) > 0 {
@@ -817,7 +887,10 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
// is the only thing standing in the way, take the placement anyway:
// a second replica on one node still survives an engine crash.
res = r.sched.Place(profiles, nil)
pspan.SetAttributes(attribute.Bool("llmgw.spread_given_up", res.Placement != nil))
}
placementAttrs(pspan, res)
pspan.End()
r.mu.Lock()
r.lastExplain[m.Name] = res
@@ -849,10 +922,12 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
{Profile: res.Placement.Profile.Name, Reason: why}}}
r.mu.Unlock()
r.notePlacementFailure(m.Name, why)
trace.SpanFromContext(ctx).SetAttributes(attribute.String("llmgw.over_limit", why))
return overLimit
}
if err := r.startReplica(ctx, m, res.Placement); err != nil {
r.log.Error("starting replica failed", "model", m.Name, "err", err)
trace.SpanFromContext(ctx).SetStatus(codes.Error, err.Error())
return noRoom
}
return placed
@@ -869,6 +944,51 @@ const (
overLimit // there is room, but the owner's quota does not allow it
)
// String names an outcome for a trace.
func (o scaleOutcome) String() string {
switch o {
case placed:
return "placed"
case makingRoom:
return "making room"
case noRoom:
return "no room"
case overLimit:
return "over its quota"
default:
return "not tried"
}
}
// placementAttrs puts a placement's answer on its span: where it went, or
// every candidate it turned down and why, which is what `ctl
// explain-placement` would have printed at that moment.
func placementAttrs(span trace.Span, res placement.Result) {
span.SetAttributes(attribute.Int("llmgw.rejections", len(res.Rejections)))
pl := res.Placement
if pl == nil {
explain := res.Explain()
if len(explain) > 4096 {
explain = explain[:4096] + "…"
}
span.SetAttributes(attribute.String("llmgw.explain", explain))
span.SetStatus(codes.Error, "fits nowhere")
return
}
nodes := make([]string, 0, len(pl.Nodes))
for _, a := range pl.Nodes {
nodes = append(nodes, a.NodeID)
}
span.SetAttributes(
attribute.String("llmgw.profile", pl.Profile.Name),
attribute.String("llmgw.engine", pl.Profile.Engine),
attribute.String("llmgw.gpu_class", pl.Profile.GPUClass),
attribute.String("llmgw.pool", pl.Pool),
attribute.String("llmgw.fabric", pl.Fabric),
attribute.StringSlice("llmgw.nodes", nodes),
)
}
func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *placement.Placement) error {
prof, ok := m.ProfileByName(pl.Profile.Name)
if !ok {
@@ -933,6 +1053,19 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac
slices.SortFunc(order, func(a, b Shard) int { return a.Rank - b.Rank })
nodeIDs := rep.NodeIDs()
// Open until the replica serves, fails or is stopped, and the parent of
// every node's part of the start: the agents' spans arrive under it.
rctx, rspan := startSpan(ctx, "replica start", append(replicaAttrs(rep),
attribute.String("llmgw.engine", prof.Engine),
attribute.String("llmgw.gpu_class", prof.GPUClass),
attribute.String("llmgw.pool", pl.Pool),
attribute.String("llmgw.fabric", pl.Fabric),
attribute.String("llmgw.endpoint", rep.Endpoint),
attribute.Int("llmgw.attempt", attempt),
attribute.String("llmgw.startup_timeout", r.startupTimeout(rep).String()),
)...)
rep.startSpan, rep.startTrace = rspan, rspan.SpanContext()
r.mu.Lock()
r.replicas[id] = rep
r.mu.Unlock()
@@ -962,7 +1095,19 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac
Fabric: pl.Fabric,
ProfileFP: prof.Fingerprint(),
}
if err := r.disp.StartShard(ctx, cmd); err != nil {
// The command, sent: its context travels with it to the agent.
sctx, sspan := tracer().Start(rctx, "send start",
trace.WithSpanKind(trace.SpanKindProducer),
trace.WithAttributes(
attribute.String("llmgw.shard", s.ID),
attribute.String("llmgw.node", s.NodeID),
attribute.Int("llmgw.rank", s.Rank),
attribute.IntSlice("llmgw.gpus", s.GPUIndices),
attribute.Int("llmgw.port", s.HTTPPort),
))
err := r.disp.StartShard(sctx, cmd)
obs.End(sspan, err)
if err != nil {
r.markShard(id, s.ID, supervisor.Failed, err.Error())
return fmt.Errorf("starting shard %s: %w", s.ID, err)
}
@@ -1028,7 +1173,8 @@ func (r *Reconciler) allocatePorts(pl *placement.Placement) int {
return r.cfg.PortRange[0]
}
func (r *Reconciler) scaleDown(ctx context.Context, excess []string) {
// scaleDown drains replicas; why is for the trace.
func (r *Reconciler) scaleDown(ctx context.Context, excess []string, why string) {
for _, id := range excess {
r.mu.Lock()
rep, ok := r.replicas[id]
@@ -1042,15 +1188,12 @@ func (r *Reconciler) scaleDown(ctx context.Context, excess []string) {
r.mu.Unlock()
r.log.Info("draining replica", "replica", id)
for _, s := range shards {
if err := r.disp.DrainShard(ctx, s.NodeID, s.ID); err != nil {
r.log.Warn("draining a shard failed", "shard", s.ID, "err", err)
}
}
r.drainShards(ctx, id, shards, why)
}
}
func (r *Reconciler) stopReplica(ctx context.Context, id string) {
// stopReplica stops a replica's shards and forgets it; why is for the trace.
func (r *Reconciler) stopReplica(ctx context.Context, id, why string) {
r.mu.Lock()
rep, ok := r.replicas[id]
if !ok {
@@ -1059,6 +1202,10 @@ func (r *Reconciler) stopReplica(ctx context.Context, id string) {
}
shards := slices.Clone(rep.Shards)
rep.State = Stopped
rep.startDone(r.now())
ctx, span := startSpan(ctx, "stop replica", append(replicaAttrs(rep),
attribute.String("llmgw.reason", why))...)
defer span.End()
r.mu.Unlock()
for _, s := range shards {
@@ -1127,6 +1274,18 @@ func (r *Reconciler) markShard(replicaID, shardID string, st supervisor.State, e
prev := rep.State
rep.deriveState(r.now())
if rep.startSpan != nil {
ev := []attribute.KeyValue{
attribute.String("llmgw.shard", shardID), attribute.String("llmgw.state", st.String()),
}
if errMsg != "" {
ev = append(ev, attribute.String("llmgw.error", errMsg))
}
rep.startSpan.AddEvent("shard "+st.String(), trace.WithAttributes(ev...))
}
if rep.State == Ready || rep.State.Terminal() {
rep.startDone(r.now())
}
if rep.State == Ready && prev != Ready {
// A model that comes up clears its backoff; the next failure starts
// counting from scratch rather than from an old streak.
@@ -11,6 +11,8 @@ import (
"testing"
"time"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
@@ -24,9 +26,12 @@ type fakeDispatcher struct {
stopped []string
drained []string
failNext error
// traces are the span each start was sent under, which is what the
// real dispatcher puts on the command for the agent.
traces []trace.SpanContext
}
func (d *fakeDispatcher) StartShard(_ context.Context, cmd ShardCommand) error {
func (d *fakeDispatcher) StartShard(ctx context.Context, cmd ShardCommand) error {
d.mu.Lock()
defer d.mu.Unlock()
if d.failNext != nil {
@@ -35,6 +40,7 @@ func (d *fakeDispatcher) StartShard(_ context.Context, cmd ShardCommand) error {
return err
}
d.started = append(d.started, cmd)
d.traces = append(d.traces, trace.SpanContextFromContext(ctx))
return nil
}
+13
View File
@@ -7,6 +7,8 @@ import (
"slices"
"time"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
)
@@ -166,8 +168,19 @@ type Replica struct {
// carried across rebuilds so the backoff does not reset on every attempt.
Attempt int
Err string
// startSpan is the "replica start" span, open from placement until the
// replica serves, fails or is stopped; nil once closed, and for a replica
// adopted rather than started. startTrace names it for good.
startSpan trace.Span
startTrace trace.SpanContext
}
// StartTrace names the span of this replica's start, for something that
// waited on it — a job serving a model — to link to. Invalid for a replica
// adopted rather than started, and whenever tracing is off.
func (r Replica) StartTrace() trace.SpanContext { return r.startTrace }
// NodeIDs lists the nodes this replica occupies.
func (r Replica) NodeIDs() []string {
out := make([]string, 0, len(r.Shards))
+18 -15
View File
@@ -7,6 +7,8 @@ import (
"slices"
"time"
"go.opentelemetry.io/otel/attribute"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
)
@@ -48,39 +50,39 @@ func (r *Reconciler) retire(ctx context.Context, m catalog.Model, ids []string,
}
switch {
case readyLive >= desired:
r.drainForRestart(ctx, id)
r.drainForRestart(ctx, id, "restarted: its replacement serves")
case inPlace:
r.log.Warn("restarting a replica in place, as asked: stopped before its replacement starts", "replica", id)
r.stopForRestart(ctx, id)
r.stopForRestart(ctx, id, "restarted in place, as asked")
case live >= desired:
// A replacement is placed and loading; it is waited for.
case parked:
r.refuseRestart(id, "its replacements fail to start; the model is parked (llmgw ctl retry "+m.Name+")")
r.refuseRestart(ctx, id, "its replacements fail to start; the model is parked (llmgw ctl retry "+m.Name+")")
case outcome == noRoom && r.fitsWithout(m, nil, []victim{{ID: id, Model: m.Name}}):
r.log.Warn("restarting a replica in place: it is the only thing in its replacement's way",
"replica", id, "model", m.Name)
r.stopForRestart(ctx, id)
r.stopForRestart(ctx, id, "restarted in place: it was the only thing in its replacement's way")
case outcome == noRoom || outcome == overLimit:
r.refuseRestart(id, "its replacement fits on no node even with this replica stopped "+
r.refuseRestart(ctx, id, "its replacement fits on no node even with this replica stopped "+
"(llmgw ctl explain-placement "+m.Name+")")
case outcome == notTried && r.now().Sub(since) > r.cfg.DrainGrace:
r.log.Warn("restarting a replica in place: no replacement beside it",
"replica", id, "after", r.cfg.DrainGrace)
r.stopForRestart(ctx, id)
r.stopForRestart(ctx, id, "restarted in place: no replacement beside it within "+r.cfg.DrainGrace.String())
}
}
}
// stopForRestart drains a replica and stops it at once, for its replacement
// to take its place.
func (r *Reconciler) stopForRestart(ctx context.Context, id string) {
r.drainForRestart(ctx, id)
r.stopReplica(ctx, id)
func (r *Reconciler) stopForRestart(ctx context.Context, id, why string) {
r.drainForRestart(ctx, id, why)
r.stopReplica(ctx, id, why)
}
// refuseRestart gives up a replica's restart, leaving it serving, and keeps
// why for whoever looks at the replica.
func (r *Reconciler) refuseRestart(id, why string) {
func (r *Reconciler) refuseRestart(ctx context.Context, id, why string) {
r.mu.Lock()
if rep, ok := r.replicas[id]; ok {
rep.RetiringAt = time.Time{}
@@ -89,9 +91,12 @@ func (r *Reconciler) refuseRestart(id, why string) {
r.restartRefused[id] = why
r.mu.Unlock()
r.log.Warn("not restarting a replica: it keeps serving", "replica", id, "because", why)
_, span := startSpan(ctx, "restart refused",
attribute.String("llmgw.replica", id), attribute.String("llmgw.reason", why))
span.End()
}
func (r *Reconciler) drainForRestart(ctx context.Context, id string) {
func (r *Reconciler) drainForRestart(ctx context.Context, id, why string) {
r.mu.Lock()
rep, ok := r.replicas[id]
var shards []Shard
@@ -101,10 +106,8 @@ func (r *Reconciler) drainForRestart(ctx context.Context, id string) {
}
delete(r.restartInPlace, id)
r.mu.Unlock()
for _, s := range shards {
if err := r.disp.DrainShard(ctx, s.NodeID, s.ID); err != nil {
r.log.Warn("draining a shard failed", "shard", s.ID, "err", err)
}
if ok {
r.drainShards(ctx, id, shards, why)
}
}
+122
View File
@@ -0,0 +1,122 @@
package reconciler
import (
"context"
"errors"
"sync"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
)
// What is traced here, and why it is shaped this way.
//
// A pass runs every few seconds and nearly always finds nothing to do. A
// trace per pass would bury the few that placed or stopped something under
// seventeen thousand empty ones a day, so a pass's span is begun only when
// the pass first does something, and backdated to when the pass began.
//
// A replica's start outlives the pass that placed it by minutes. Its span,
// "replica start", is opened beneath the placement and closed when the
// replica serves, fails, or is stopped, and the command to each node carries
// it, so the agent's spans for that shard — weights, launch, readiness — are
// its descendants. That one trace is the answer to "why did this model take
// four minutes to start".
const tracerName = "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
func tracer() trace.Tracer { return obs.Tracer(tracerName) }
// pass is one reconcile pass's span, begun on first use.
type pass struct {
start time.Time
base context.Context
once sync.Once
ctx context.Context
span trace.Span
}
type passKey struct{}
// beginPass marks ctx as a pass's, for startSpan to hang its spans on.
func beginPass(ctx context.Context) (context.Context, *pass) {
p := &pass{start: time.Now(), base: ctx}
return context.WithValue(ctx, passKey{}, p), p
}
// context is the pass's span's context, beginning the span if it has not
// been.
func (p *pass) context() context.Context {
p.once.Do(func() {
p.ctx, p.span = tracer().Start(p.base, "reconcile", trace.WithTimestamp(p.start))
})
return p.ctx
}
// end closes the pass's span, if anything began it.
func (p *pass) end() {
p.once.Do(func() {}) // nothing may begin it after this
if p.span != nil {
p.span.End()
}
}
// startSpan opens a span for something the reconciler did: beneath the span
// already in ctx, or beneath the pass's — begun now — when ctx holds the
// pass and nothing else.
func startSpan(ctx context.Context, name string, attrs ...attribute.KeyValue) (context.Context, trace.Span) {
if !trace.SpanContextFromContext(ctx).IsValid() {
if p, ok := ctx.Value(passKey{}).(*pass); ok {
if pctx := p.context(); pctx != nil {
ctx = pctx
}
}
}
return tracer().Start(ctx, name, trace.WithAttributes(attrs...))
}
// endSpan closes a span, marking it failed when why is not empty.
func endSpan(span trace.Span, why string) {
if why != "" {
obs.End(span, errors.New(why))
return
}
span.End()
}
// replicaAttrs are what every span about one replica says.
func replicaAttrs(rep *Replica) []attribute.KeyValue {
return []attribute.KeyValue{
attribute.String("llmgw.replica", rep.ID),
attribute.String("llmgw.model", rep.Model),
attribute.String("llmgw.profile", rep.Profile),
attribute.StringSlice("llmgw.nodes", rep.NodeIDs()),
}
}
// startDone closes a replica's "replica start" span, once: when it serves,
// with how long that took, or when it fails or is stopped first, with why.
// Called with r.mu held, or before the replica is in the map.
func (rep *Replica) startDone(now time.Time) {
if rep.startSpan == nil {
return
}
span := rep.startSpan
rep.startSpan = nil
span.SetAttributes(attribute.String("llmgw.state", rep.State.String()))
switch rep.State {
case Ready:
span.SetAttributes(attribute.Float64("llmgw.ready_after_seconds", now.Sub(rep.CreatedAt).Seconds()))
span.SetStatus(codes.Ok, "")
case Failed:
span.SetStatus(codes.Error, rep.Err)
default:
span.SetStatus(codes.Error, "stopped before it was ready")
}
span.End()
}
+174
View File
@@ -0,0 +1,174 @@
package reconciler
import (
"errors"
"strings"
"testing"
"go.opentelemetry.io/otel/codes"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs/obstest"
)
func TestAPlacementIsTracedFromThePassToTheCommandAndTheReplicaServing(t *testing.T) {
rec := obstest.Record(t)
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.r.Reconcile(t.Context())
pass := rec.One(t, [16]byte{}, "reconcile")
tid := pass.SpanContext().TraceID()
up := rec.One(t, tid, "scale up")
place := rec.One(t, tid, "place")
if !obstest.ChildOf(up, pass) || !obstest.ChildOf(place, up) {
t.Fatal("the placement is not beneath the scale-up, beneath the pass")
}
if a := obstest.Attrs(place); a["llmgw.model"] != "llama" || a["llmgw.profile"] != "tp4" ||
a["llmgw.gpu_class"] != "h100-80g" || a["llmgw.nodes"] != `["1"]` {
t.Errorf("the placement does not say where it went: %v", a)
}
if a := obstest.Attrs(up); a["llmgw.outcome"] != "placed" || a["llmgw.desired"] != "1" {
t.Errorf("the scale-up does not say what it did: %v", a)
}
// The command went under a span of its own, beneath the replica's start,
// which is still open: the replica is loading.
send := rec.One(t, tid, "send start")
if got := h.disp.traces; len(got) != 1 || got[0].SpanID() != send.SpanContext().SpanID() {
t.Fatalf("the command was not sent under its span: %v", got)
}
if len(rec.Named("replica start")) != 0 {
t.Fatal("the replica's start ended before the replica served")
}
h.markReady()
start := rec.One(t, tid, "replica start")
if !obstest.ChildOf(start, up) || !obstest.ChildOf(send, start) {
t.Error("the replica's start is not beneath the scale-up, or the command beneath it")
}
a := obstest.Attrs(start)
if a["llmgw.state"] != "ready" || a["llmgw.profile"] != "tp4" || a["llmgw.ready_after_seconds"] == "" {
t.Errorf("the replica's start does not say how it ended: %v", a)
}
if start.Status().Code != codes.Ok {
t.Errorf("a replica that serves is %v", start.Status())
}
events := start.Events()
if len(events) == 0 || events[len(events)-1].Name != "shard ready" {
t.Errorf("the shard's progress is not on the replica's start: %v", events)
}
if sc := h.r.Replicas()[0].StartTrace(); sc.SpanID() != start.SpanContext().SpanID() {
t.Error("the replica does not name its start's span for a job to link to")
}
}
func TestAPassThatDoesNothingLeavesNoTrace(t *testing.T) {
rec := obstest.Record(t)
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.r.Reconcile(t.Context())
h.markReady()
before := len(rec.Ended())
// Every pass from here finds the model as it should be.
for range 3 {
h.r.Reconcile(t.Context())
}
if got := rec.Ended()[before:]; len(got) != 0 {
var names []string
for _, s := range got {
names = append(names, s.Name())
}
t.Errorf("passes with nothing to do were traced: %v", names)
}
}
func TestAPlacementThatFitsNowhereSaysWhy(t *testing.T) {
rec := obstest.Record(t)
m := testModel(1, catalog.Profile{Name: "tp16", GPUClass: "h100-80g", TP: 16})
h := newHarness(t, m, gpuNode("1", 8))
h.r.Reconcile(t.Context())
place := rec.One(t, [16]byte{}, "place")
if place.Status().Code != codes.Error {
t.Errorf("a placement that found nowhere is %v", place.Status())
}
if a := obstest.Attrs(place); a["llmgw.rejections"] == "0" || !strings.Contains(a["llmgw.explain"], "tp16") {
t.Errorf("the placement does not say what it turned down: %v", a)
}
if a := obstest.Attrs(rec.One(t, place.SpanContext().TraceID(), "scale up")); a["llmgw.outcome"] != "no room" {
t.Errorf("the scale-up says %v", a)
}
}
func TestAFailedStartEndsItsTraceWithTheEnginesWords(t *testing.T) {
rec := obstest.Record(t)
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.r.Reconcile(t.Context())
cmd := h.disp.startedShards()[0]
h.r.OnShardStatus(cmd.ReplicaID, cmd.ShardID, supervisor.Failed, "CUDA out of memory")
start := rec.One(t, [16]byte{}, "replica start")
if start.Status().Code != codes.Error || !strings.Contains(start.Status().Description, "CUDA out of memory") {
t.Errorf("the failed start says %+v", start.Status())
}
// The next pass reaps it, in a trace of its own, saying when it is tried
// again.
h.r.Reconcile(t.Context())
reap := rec.One(t, [16]byte{}, "reap replica")
a := obstest.Attrs(reap)
if a["llmgw.replica"] != cmd.ReplicaID || a["llmgw.failures_in_a_row"] != "1" || a["llmgw.retry_in_seconds"] == "" {
t.Errorf("the reaping does not say what happens next: %v", a)
}
}
func TestADispatchThatFailsIsTracedAsFailed(t *testing.T) {
rec := obstest.Record(t)
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.disp.failNext = errors.New("no agent connected")
h.r.Reconcile(t.Context())
send := rec.One(t, [16]byte{}, "send start")
if send.Status().Code != codes.Error {
t.Errorf("a command that was not sent is %v", send.Status())
}
if start := rec.One(t, [16]byte{}, "replica start"); start.Status().Code != codes.Error {
t.Errorf("a replica none of whose shards was sent is %v", start.Status())
}
}
func TestAPreemptionIsTracedWithItsVictims(t *testing.T) {
h := fillThenWant(t, gpuModel("vital", catalog.ClassCritical, 1), gpuModel("chat", "", 1), gpuNode("1", 8))
rec := obstest.Record(t)
h.r.Reconcile(t.Context())
pre := rec.One(t, [16]byte{}, "preempt")
a := obstest.Attrs(pre)
if a["llmgw.for"] != "vital" || !strings.Contains(a["llmgw.victims"], "chat") {
t.Errorf("the preemption does not say who gave way to whom: %v", a)
}
drain := rec.One(t, pre.SpanContext().TraceID(), "drain replica")
if !obstest.ChildOf(drain, pre) || obstest.Attrs(drain)["llmgw.reason"] != "preempted for vital" {
t.Errorf("the victim's drain is not beneath the preemption: %v", obstest.Attrs(drain))
}
}
func TestAStopSaysWhy(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.r.Reconcile(t.Context())
h.markReady()
rec := obstest.Record(t)
h.setModels(t) // the model leaves the catalog
h.r.Reconcile(t.Context())
stop := rec.One(t, [16]byte{}, "stop replica")
if a := obstest.Attrs(stop); a["llmgw.model"] != "llama" || a["llmgw.reason"] != "its model left the catalog" {
t.Errorf("the stop does not say why: %v", a)
}
}
+4 -3
View File
@@ -278,7 +278,7 @@ var Reference = []Block{
{Key: "debug_pprof", Type: "bool", Doc: "DebugPprof serves Go's profiler on the admin listener, behind the token.\nOff by default: a heap profile holds the API keys requests presented.", Nested: ""},
{Key: "agent_insecure", Type: "bool", Doc: "AgentInsecure lets a control plane run its agent port without\nagent_tls: plaintext, and any machine that reaches it may register as\nany node — be given replicas, and so be sent users' prompts — or, in\nthe other direction, a machine in the path may tell every agent what\nto run. Refused unless said here, and then the port listens on\nloopback unless agent_listen names an address. For one machine, a\ntest, or a network nothing untrusted reaches.", Nested: ""},
{Key: "monitoring", Type: "Monitoring", Doc: "Monitoring is where the dashboard reads its history from.", Nested: "config.Monitoring"},
{Key: "tracing", Type: "Tracing", Doc: "Tracing sends a trace of each inference request to an OpenTelemetry\ncollector. Off unless an endpoint is set here or in\nOTEL_EXPORTER_OTLP_ENDPOINT.", Nested: "config.Tracing"},
{Key: "tracing", Type: "obs.Tracing", Doc: "Tracing sends traces to an OpenTelemetry collector: each inference\nrequest, and on a control plane each placement, replica start and\nmodel job. Off unless an endpoint is set here or in\nOTEL_EXPORTER_OTLP_ENDPOINT.", Nested: "obs.Tracing"},
{Key: "mirror", Type: "Mirror", Doc: "Mirror points the control plane at a model mirror it can administer.\nEmpty means there is none, and /admin/mirror stays unmounted.", Nested: "config.Mirror"},
{Key: "engines", Type: "map[string]string", Doc: "Engines are the engines every node installs, and at which version. The\nagents install them themselves — the right build for their hardware,\nchecked against its publisher's checksum and seen to run — so a node is\nnever trusted to already have what somebody once put there.\n\nUnset keeps DefaultEngines. A version replaces the default; an empty\none leaves that engine to be installed by hand.", Nested: ""},
{Key: "engine_overrides", Type: "[]EngineOverride", Doc: "EngineOverrides give some nodes different engines from the rest: a\nvLLM built for CUDA 12 on the Volta cards CUDA 13 dropped, or none at\nall on nodes meant only for llama.cpp. Each picks nodes and lays its\nengines over the cluster's, in order, later ones last; an empty version\nremoves that engine from the nodes it picks.", Nested: ""},
@@ -315,9 +315,9 @@ var Reference = []Block{
{Key: "max_calls_per_turn", Type: "int", Doc: "MaxCallsPerTurn stops an answer calling more tools than this.", Nested: ""},
{Key: "validate_arguments", Type: "bool", Doc: "ValidateArguments checks a call's arguments against the parameters\nschema the request declared for the tool.", Nested: ""},
}},
{Package: "config", Name: "Tracing", Title: "Tracing", Doc: "", Fields: []Field{
{Package: "obs", Name: "Tracing", Title: "Tracing", Doc: "", Fields: []Field{
{Key: "endpoint", Type: "string", Doc: "Endpoint is the collector, for example http://otel.internal:4318.\nEmpty falls back to OTEL_EXPORTER_OTLP_ENDPOINT, and to no tracing.", Nested: ""},
{Key: "sample_ratio", Type: "float64", Doc: "SampleRatio is the share of requests traced, 0 to 1, when the caller\ndid not decide it with a sampled traceparent of its own. Zero takes 1.", Nested: ""},
{Key: "sample_ratio", Type: "float64", Doc: "SampleRatio is the share of traces kept, 0 to 1, of those nobody\nupstream already decided. Zero takes 1. A request with a sampled\ntraceparent of its own is kept, and an agent keeps every start the\ncontrol plane kept, whatever their own ratio says.", Nested: ""},
}},
{Package: "engines", Name: "VLLMConfig", Title: "VLLMConfig", Doc: "", Fields: []Field{
{Key: "bin", Type: "string", Doc: "Bin is the vllm executable.", Nested: ""},
@@ -367,6 +367,7 @@ var Reference = []Block{
{Key: "discovery", Type: "discovery.Options", Doc: "", Nested: "discovery.Options"},
{Key: "engines", Type: "engines.Config", Doc: "", Nested: "engines.Config"},
{Key: "supervisor", Type: "supervisor.Config", Doc: "", Nested: "supervisor.Config"},
{Key: "tracing", Type: "obs.Tracing", Doc: "Tracing sends this node's part of each replica start — the weights\nfetched, the engine launched, the wait for it to answer — to an\nOpenTelemetry collector, inside the control plane's trace of that\nstart. Give it the collector server.yaml's tracing names. Off unless\nan endpoint is set here or in OTEL_EXPORTER_OTLP_ENDPOINT.", Nested: "obs.Tracing"},
}},
{Package: "config", Name: "Band", Title: "config.Band", Doc: "", Fields: []Field{
{Key: "name", Type: "string", Doc: "", Nested: ""},
+1
View File
@@ -30,6 +30,7 @@ var Sources = []string{
"internal/guard/judge.go",
"internal/config/oidc.go",
"internal/notify/notify.go",
"internal/obs/trace.go",
}
// notReference are files with koanf tags that an operator never writes.
+98
View File
@@ -0,0 +1,98 @@
// Package obstest records spans for tests that assert what was traced.
package obstest
import (
"testing"
"time"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/propagation"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
"go.opentelemetry.io/otel/trace"
)
// Recorder is every span ended while a test runs.
type Recorder struct{ *tracetest.SpanRecorder }
// Record installs a provider that keeps every span, and the W3C propagator,
// for the length of the test, putting back what was there after.
//
// Global, as the code under test finds it, so no test that records may run
// in parallel with another that traces.
func Record(t *testing.T) *Recorder {
t.Helper()
rec := tracetest.NewSpanRecorder()
prevTP, prevProp := otel.GetTracerProvider(), otel.GetTextMapPropagator()
otel.SetTracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)))
otel.SetTextMapPropagator(propagation.TraceContext{})
t.Cleanup(func() { otel.SetTracerProvider(prevTP); otel.SetTextMapPropagator(prevProp) })
return &Recorder{rec}
}
// Named is the ended spans called name, oldest first.
func (r *Recorder) Named(name string) []sdktrace.ReadOnlySpan {
var out []sdktrace.ReadOnlySpan
for _, s := range r.Ended() {
if s.Name() == name {
out = append(out, s)
}
}
return out
}
// One is the single ended span called name in trace tid, failing the test
// when there is not exactly one. A zero tid is any trace.
func (r *Recorder) One(t *testing.T, tid trace.TraceID, name string) sdktrace.ReadOnlySpan {
t.Helper()
var found []sdktrace.ReadOnlySpan
for _, s := range r.Named(name) {
if !tid.IsValid() || s.SpanContext().TraceID() == tid {
found = append(found, s)
}
}
if len(found) != 1 {
t.Fatalf("%d spans called %q in the trace; ended: %v", len(found), name, r.names())
}
return found[0]
}
// Await is One, waiting up to limit for the span to end: the code that
// ends it may run after whatever the test waited on.
func (r *Recorder) Await(t *testing.T, tid trace.TraceID, name string, limit time.Duration) sdktrace.ReadOnlySpan {
t.Helper()
deadline := time.Now().Add(limit)
for time.Now().Before(deadline) {
for _, s := range r.Named(name) {
if s.SpanContext().TraceID() == tid {
return r.One(t, tid, name)
}
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("no span called %q ended within %v; ended: %v", name, limit, r.names())
return nil
}
func (r *Recorder) names() []string {
var out []string
for _, s := range r.Ended() {
out = append(out, s.Name())
}
return out
}
// Attrs is a span's attributes, as strings.
func Attrs(s sdktrace.ReadOnlySpan) map[string]string {
out := map[string]string{}
for _, a := range s.Attributes() {
out[string(a.Key)] = a.Value.Emit()
}
return out
}
// ChildOf reports whether child's parent is parent.
func ChildOf(child, parent sdktrace.ReadOnlySpan) bool {
return child.Parent().SpanID() == parent.SpanContext().SpanID() &&
child.SpanContext().TraceID() == parent.SpanContext().TraceID()
}
+74 -3
View File
@@ -9,18 +9,47 @@ import (
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/trace"
)
// Tracing is where traces go, over OTLP/HTTP.
//
// In server.yaml: one span per inference request, with admission, scheduling
// and each attempt at an engine beneath it, and the trace carried on to the
// engine in a traceparent header — the request's X-Request-Id is on the span,
// so a trace and `llmgw ctl request` name the same request. A control plane
// also traces every reconcile pass that does something, each replica from
// its placement until it serves, and each job `ctl model serve` and `ctl
// model update` start.
//
// In agent.yaml: the node's part of starting a replica — fetching the
// weights, launching the engine, waiting for it to answer — inside the
// control plane's trace of that start. Point both at the same collector.
type Tracing struct {
// Endpoint is the collector, for example http://otel.internal:4318.
// Empty falls back to OTEL_EXPORTER_OTLP_ENDPOINT, and to no tracing.
Endpoint string `koanf:"endpoint"`
// SampleRatio is the share of traces kept, 0 to 1, of those nobody
// upstream already decided. Zero takes 1. A request with a sampled
// traceparent of its own is kept, and an agent keeps every start the
// control plane kept, whatever their own ratio says.
SampleRatio float64 `koanf:"sample_ratio"`
}
// TracingOptions says where traces go.
type TracingOptions struct {
Endpoint string // http(s)://host:4318; empty reads OTEL_EXPORTER_OTLP_ENDPOINT
SampleRatio float64 // of traces the caller did not already decide; zero takes 1
Service string
Version string
// Instance tells apart the processes of one service: the node, for an
// agent. Empty leaves it out.
Instance string
}
// SetupTracing installs the global tracer provider and the W3C propagator,
@@ -55,13 +84,16 @@ func SetupTracing(ctx context.Context, o TracingOptions) (func(context.Context)
if ratio <= 0 || ratio > 1 {
ratio = 1
}
res := resource.NewSchemaless(
attrs := []attribute.KeyValue{
attribute.String("service.name", o.Service),
attribute.String("service.version", o.Version),
)
}
if o.Instance != "" {
attrs = append(attrs, attribute.String("service.instance.id", o.Instance))
}
tp := sdktrace.NewTracerProvider(
sdktrace.WithBatcher(exp, sdktrace.WithBatchTimeout(5*time.Second)),
sdktrace.WithResource(res),
sdktrace.WithResource(resource.NewSchemaless(attrs...)),
// A caller that sent a sampled traceparent has already decided; the
// ratio is for requests that arrive without one.
sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(ratio))),
@@ -70,3 +102,42 @@ func SetupTracing(ctx context.Context, o TracingOptions) (func(context.Context)
otel.SetTextMapPropagator(propagation.TraceContext{})
return tp.Shutdown, nil
}
// Tracer is the named tracer of whatever provider is installed when it is
// called. Looked up per use rather than kept, so a provider installed after a
// component was built — as tests do — is the one its spans go to. With none
// installed it is the no-op tracer, and a span costs a function call.
func Tracer(name string) trace.Tracer { return otel.Tracer(name) }
// End closes a span, marking it failed when err is set.
func End(span trace.Span, err error) {
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
}
span.End()
}
// Inject is the trace context of ctx as a message carries it: traceparent
// and, when there is one, tracestate. Nil when ctx holds no span, which is
// every context while tracing is off, so a message then carries nothing.
func Inject(ctx context.Context) map[string]string {
if !trace.SpanContextFromContext(ctx).IsValid() {
return nil
}
m := propagation.MapCarrier{}
otel.GetTextMapPropagator().Inject(ctx, m)
if len(m) == 0 {
return nil
}
return m
}
// Extract is ctx continuing the trace a message carried, as Inject wrote it.
// A message that carried none leaves ctx as it was.
func Extract(ctx context.Context, carried map[string]string) context.Context {
if len(carried) == 0 {
return ctx
}
return otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(carried))
}
+63
View File
@@ -2,7 +2,16 @@ package obs
import (
"context"
"errors"
"testing"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/propagation"
sdktrace "go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
"go.opentelemetry.io/otel/trace"
"go.opentelemetry.io/otel/trace/noop"
)
func TestTracingIsOffUntilGivenSomewhereToSend(t *testing.T) {
@@ -17,3 +26,57 @@ func TestTracingIsOffUntilGivenSomewhereToSend(t *testing.T) {
}
}
}
func TestATraceCrossesAMessageAndContinuesOnTheOtherSide(t *testing.T) {
rec := tracetest.NewSpanRecorder()
prevTP, prevProp := otel.GetTracerProvider(), otel.GetTextMapPropagator()
otel.SetTracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)))
otel.SetTextMapPropagator(propagation.TraceContext{})
t.Cleanup(func() { otel.SetTracerProvider(prevTP); otel.SetTextMapPropagator(prevProp) })
if m := Inject(context.Background()); m != nil {
t.Errorf("a context with no span carried %v", m)
}
if ctx := Extract(context.Background(), nil); trace.SpanContextFromContext(ctx).IsValid() {
t.Error("nothing was carried, yet a trace continued")
}
ctx, sent := Tracer("t").Start(context.Background(), "send")
carried := Inject(ctx)
sent.End()
if carried["traceparent"] == "" {
t.Fatalf("no traceparent carried: %v", carried)
}
_, got := Tracer("t").Start(Extract(context.Background(), carried), "receive")
End(got, errors.New("it broke"))
spans := rec.Ended()
if len(spans) != 2 {
t.Fatalf("%d spans", len(spans))
}
recv := spans[1]
if recv.Parent().SpanID() != sent.SpanContext().SpanID() || recv.SpanContext().TraceID() != sent.SpanContext().TraceID() {
t.Error("the receiving span is not the sending one's child")
}
if recv.Status().Code != codes.Error || recv.Status().Description != "it broke" {
t.Errorf("a failure was not recorded: %+v", recv.Status())
}
}
func TestWithTracingOffAMessageCarriesNothing(t *testing.T) {
// What a process that never called SetupTracing has: spans that are not
// recorded and contexts with nothing to propagate.
prevTP, prevProp := otel.GetTracerProvider(), otel.GetTextMapPropagator()
otel.SetTracerProvider(noop.NewTracerProvider())
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator())
t.Cleanup(func() { otel.SetTracerProvider(prevTP); otel.SetTextMapPropagator(prevProp) })
ctx, span := Tracer("t").Start(context.Background(), "send")
defer span.End()
if span.IsRecording() {
t.Error("a span is recorded with tracing off")
}
if m := Inject(ctx); m != nil {
t.Errorf("a message carries %v with tracing off", m)
}
}
+60
View File
@@ -0,0 +1,60 @@
## CHANGELOG
### Tracing beyond the router
- **The control plane traces what it does to the fleet.** A reconcile pass
that does something is a `reconcile` trace — passes with nothing to do are
not traced at all — with what it did beneath it: `scale up` with `place`
(the profile, engine, GPU class, pool, fabric and nodes chosen, or every
candidate turned down and why, as `ctl explain-placement` would print it),
`scale down`, `drain replica` and `stop replica` each with its reason,
`evict`, `preempt` with its victims, `reap replica` with the error and
when the model is tried again, `restart replica`, `restart refused`,
`node unreachable` and `node lost`.
- **A replica's start is one trace, down to the node.** `replica start`
stays open from placement until the replica serves, fails or is stopped,
with each shard's progress as an event and how long it took. The command
to each node is a `send start` span whose context travels in the command
(a new `trace` field on the agent protocol's `ControlMessage`), and the
agent continues the same trace: `start shard`, with `fetch weights`
(local, cached, downloaded, or downloaded by another shard; a `download`
with its bytes and where a resume began, the source's credentials and
query left out), `launch engine` (and `wait for rendezvous` on a worker
of a multi-node replica), and `wait for engine` with how often the engine
was asked and its last answer. "Why did qwen take four minutes to start"
is that one trace.
- **Stops, drains and adoptions on a node** are in the control plane's
trace too: `stop shard` and `stop engine`, saying whether the engine had
to be killed; `drain shard`; and `adopt` when a registering node reports
engines already running, with how many nobody claimed. An engine install
is an `install engine` trace on the node.
- **Model jobs are traces.** `ctl model serve` and `ctl model update` are
`serve model` and `update model`, with a span per step — registry,
mirror, check, catalog, placement, smoke test, replicas — and from the
placement step a link to the replica's own start.
- **The agent has a `tracing` block** in `agent.yaml`, the same as
`server.yaml`'s (`endpoint`, `sample_ratio`; `OTEL_EXPORTER_OTLP_ENDPOINT`
when unset). Point both at one collector. A node follows the control
plane's sampling for every start it is sent. Traces from a node carry its
id as `service.instance.id`, and a control plane's its host name.
- With no endpoint nothing changes: every span is the no-op one, and a
command carries no trace context.
## BACKLOG
**Done (branch `features-c2`):** spans for the reconcile pass (begun only
when a pass does something), placement with its answer or rejections,
scale-up/down, drain, stop, eviction, preemption, reaping, restarts and
lost nodes; a `replica start` span from placement to serving whose context
travels to the agent in the start command (`ControlMessage.trace`), where
weights fetch, engine launch, rendezvous and readiness wait continue the
trace; stops, drains, adoption and engine installs; modelserve jobs as a
trace with a span per step linked to the replica's start; an agent
`tracing` block; docs in install.md's "Tracing". Left: the router's span
for a request does not link to the replica it was sent to's start;
heartbeats, fabric scans, logs and agent upgrades are not traced; the
supervisor has no restart span because it never restarts an engine itself
(a restart is the reconciler's `restart replica` and reaches a node as a
stop and a start); canary and promote jobs, which landed on `features-c2`
after this was branched, are traced only through the shared step recording
(as `<kind> model`), unchecked by a test.