Compare commits

..
2 Commits
Author SHA1 Message Date
Ilker ManapandClaude Opus 5.5 746d4852af Describe v0.8.4
CI / test (push) Failing after 2m6s
Release / build (push) Successful in 2m56s
Release / notes (push) Successful in 2s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 12:27:47 +02:00
Ilker ManapandClaude Opus 5.5 33e678ab84 Show a node's logs on the admin page
Seeing what a node did meant ssh through a jump host, to a machine whose
address nobody remembers, to read a journal only root may open. When an
engine install failed on a new node today, the reason was one line — and it
was four steps away from the page that said the install had failed.

Each node's row on the cluster page now links to its logs: the agent's own,
and every engine running there, chosen from the list the node itself sends.
The page can follow, refreshing every five seconds, and is on both the admin
listener and /admin. `llmgw ctl logs NODE [--source SHARD] [-n N]` reads the
same.

They travel over the stream the agent already holds, so they need nothing
the control plane does not already have: a request goes down, the lines come
back, and a reply that does not come in ten seconds is an agent too old to
answer. The agent keeps its last 2000 lines in memory as well as writing them
wherever it did before; the engines' last few hundred lines were already kept
and sent only when one crashed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 12:27:47 +02:00
19 changed files with 822 additions and 43 deletions

No files matched your search

+10
View File
@@ -18,6 +18,16 @@ is only released once its **exit criterion** is met, not when its code lands.
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
APIs change between minors without a deprecation period.
## v0.8.4 — 2026-09-28
### Node logs on the admin page
Each node on the cluster page links to its logs — the agent's own, and the
output of every engine running there — with an option to follow them. `llmgw
ctl logs NODE` reads the same. They come over the agent's existing stream, so
no ssh, jump host or root is needed to see what a node did. The agent keeps its
last 2000 lines in memory; each engine its last few hundred.
## v0.8.3 — 2026-09-28
Found by building a node from nothing with v0.8.2.
+234 -42
View File
@@ -101,6 +101,7 @@ type AgentMessage struct {
// *AgentMessage_Register
// *AgentMessage_Heartbeat
// *AgentMessage_Shard
// *AgentMessage_Logs
Msg isAgentMessage_Msg `protobuf_oneof:"msg"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
@@ -170,6 +171,15 @@ func (x *AgentMessage) GetShard() *ShardStatus {
return nil
}
func (x *AgentMessage) GetLogs() *LogsReply {
if x != nil {
if x, ok := x.Msg.(*AgentMessage_Logs); ok {
return x.Logs
}
}
return nil
}
type isAgentMessage_Msg interface {
isAgentMessage_Msg()
}
@@ -186,12 +196,18 @@ type AgentMessage_Shard struct {
Shard *ShardStatus `protobuf:"bytes,3,opt,name=shard,proto3,oneof"`
}
type AgentMessage_Logs struct {
Logs *LogsReply `protobuf:"bytes,4,opt,name=logs,proto3,oneof"`
}
func (*AgentMessage_Register) isAgentMessage_Msg() {}
func (*AgentMessage_Heartbeat) isAgentMessage_Msg() {}
func (*AgentMessage_Shard) isAgentMessage_Msg() {}
func (*AgentMessage_Logs) isAgentMessage_Msg() {}
// ControlMessage is anything the control plane tells a node.
type ControlMessage struct {
state protoimpl.MessageState `protogen:"open.v1"`
@@ -203,6 +219,7 @@ type ControlMessage struct {
// *ControlMessage_Ack
// *ControlMessage_RemoveWeights
// *ControlMessage_Provision
// *ControlMessage_Logs
Msg isControlMessage_Msg `protobuf_oneof:"msg"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
@@ -299,6 +316,15 @@ func (x *ControlMessage) GetProvision() *Provision {
return nil
}
func (x *ControlMessage) GetLogs() *LogsRequest {
if x != nil {
if x, ok := x.Msg.(*ControlMessage_Logs); ok {
return x.Logs
}
}
return nil
}
type isControlMessage_Msg interface {
isControlMessage_Msg()
}
@@ -327,6 +353,10 @@ type ControlMessage_Provision struct {
Provision *Provision `protobuf:"bytes,6,opt,name=provision,proto3,oneof"`
}
type ControlMessage_Logs struct {
Logs *LogsRequest `protobuf:"bytes,7,opt,name=logs,proto3,oneof"`
}
func (*ControlMessage_Start) isControlMessage_Msg() {}
func (*ControlMessage_Stop) isControlMessage_Msg() {}
@@ -339,6 +369,8 @@ func (*ControlMessage_RemoveWeights) isControlMessage_Msg() {}
func (*ControlMessage_Provision) isControlMessage_Msg() {}
func (*ControlMessage_Logs) isControlMessage_Msg() {}
// Register is the first message on every stream. It carries the node's full
// hardware inventory, so the control plane needs no separate discovery call and
// a reconnecting agent automatically refreshes what the scheduler knows.
@@ -2168,23 +2200,166 @@ func (x *Ack) GetError() string {
return ""
}
// LogsRequest asks a node for the recent lines of one of its logs.
//
// Over the stream the agent already holds, because that is the one way into a
// node every deployment has: the admin page otherwise means ssh through a jump
// host to a machine whose address nobody remembers, to read a journal only
// root may open.
type LogsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
// request_id pairs the reply with its request; several may be in flight.
RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
// source is "agent" for the agent's own log, or a shard id for that
// engine's output.
Source string `protobuf:"bytes,2,opt,name=source,proto3" json:"source,omitempty"`
Lines int32 `protobuf:"varint,3,opt,name=lines,proto3" json:"lines,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *LogsRequest) Reset() {
*x = LogsRequest{}
mi := &file_llmgw_v1_agent_proto_msgTypes[21]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *LogsRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*LogsRequest) ProtoMessage() {}
func (x *LogsRequest) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[21]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use LogsRequest.ProtoReflect.Descriptor instead.
func (*LogsRequest) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{21}
}
func (x *LogsRequest) GetRequestId() string {
if x != nil {
return x.RequestId
}
return ""
}
func (x *LogsRequest) GetSource() string {
if x != nil {
return x.Source
}
return ""
}
func (x *LogsRequest) GetLines() int32 {
if x != nil {
return x.Lines
}
return 0
}
// LogsReply answers a LogsRequest.
type LogsReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
Lines []string `protobuf:"bytes,2,rep,name=lines,proto3" json:"lines,omitempty"`
// sources are every log this node can show, so a page can offer them
// without knowing what runs there.
Sources []string `protobuf:"bytes,3,rep,name=sources,proto3" json:"sources,omitempty"`
// error says why the lines are absent: an unknown source, say.
Error string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *LogsReply) Reset() {
*x = LogsReply{}
mi := &file_llmgw_v1_agent_proto_msgTypes[22]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *LogsReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*LogsReply) ProtoMessage() {}
func (x *LogsReply) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[22]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use LogsReply.ProtoReflect.Descriptor instead.
func (*LogsReply) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{22}
}
func (x *LogsReply) GetRequestId() string {
if x != nil {
return x.RequestId
}
return ""
}
func (x *LogsReply) GetLines() []string {
if x != nil {
return x.Lines
}
return nil
}
func (x *LogsReply) GetSources() []string {
if x != nil {
return x.Sources
}
return nil
}
func (x *LogsReply) GetError() string {
if x != nil {
return x.Error
}
return ""
}
var File_llmgw_v1_agent_proto protoreflect.FileDescriptor
const file_llmgw_v1_agent_proto_rawDesc = "" +
"\n" +
"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xab\x01\n" +
"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xd6\x01\n" +
"\fAgentMessage\x120\n" +
"\bregister\x18\x01 \x01(\v2\x12.llmgw.v1.RegisterH\x00R\bregister\x123\n" +
"\theartbeat\x18\x02 \x01(\v2\x13.llmgw.v1.HeartbeatH\x00R\theartbeat\x12-\n" +
"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shardB\x05\n" +
"\x03msg\"\xb8\x02\n" +
"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shard\x12)\n" +
"\x04logs\x18\x04 \x01(\v2\x13.llmgw.v1.LogsReplyH\x00R\x04logsB\x05\n" +
"\x03msg\"\xe5\x02\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" +
"\x05drain\x18\x03 \x01(\v2\x14.llmgw.v1.DrainShardH\x00R\x05drain\x12!\n" +
"\x03ack\x18\x04 \x01(\v2\r.llmgw.v1.AckH\x00R\x03ack\x12@\n" +
"\x0eremove_weights\x18\x05 \x01(\v2\x17.llmgw.v1.RemoveWeightsH\x00R\rremoveWeights\x123\n" +
"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovisionB\x05\n" +
"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovision\x12+\n" +
"\x04logs\x18\a \x01(\v2\x15.llmgw.v1.LogsRequestH\x00R\x04logsB\x05\n" +
"\x03msg\"\xf2\x05\n" +
"\bRegister\x12\x17\n" +
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" +
@@ -2379,7 +2554,18 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\n" +
"message_id\x18\x01 \x01(\tR\tmessageId\x12\x0e\n" +
"\x02ok\x18\x02 \x01(\bR\x02ok\x12\x14\n" +
"\x05error\x18\x03 \x01(\tR\x05error2O\n" +
"\x05error\x18\x03 \x01(\tR\x05error\"Z\n" +
"\vLogsRequest\x12\x1d\n" +
"\n" +
"request_id\x18\x01 \x01(\tR\trequestId\x12\x16\n" +
"\x06source\x18\x02 \x01(\tR\x06source\x12\x14\n" +
"\x05lines\x18\x03 \x01(\x05R\x05lines\"p\n" +
"\tLogsReply\x12\x1d\n" +
"\n" +
"request_id\x18\x01 \x01(\tR\trequestId\x12\x14\n" +
"\x05lines\x18\x02 \x03(\tR\x05lines\x12\x18\n" +
"\asources\x18\x03 \x03(\tR\asources\x12\x14\n" +
"\x05error\x18\x04 \x01(\tR\x05error2O\n" +
"\fAgentService\x12?\n" +
"\aConnect\x12\x16.llmgw.v1.AgentMessage\x1a\x18.llmgw.v1.ControlMessage(\x010\x01B\x93\x01\n" +
"\fcom.llmgw.v1B\n" +
@@ -2398,7 +2584,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, 25)
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 27)
var file_llmgw_v1_agent_proto_goTypes = []any{
(ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State
(*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage
@@ -2422,46 +2608,50 @@ var file_llmgw_v1_agent_proto_goTypes = []any{
(*StopShard)(nil), // 19: llmgw.v1.StopShard
(*DrainShard)(nil), // 20: llmgw.v1.DrainShard
(*Ack)(nil), // 21: llmgw.v1.Ack
nil, // 22: llmgw.v1.Register.LabelsEntry
nil, // 23: llmgw.v1.Register.EngineVersionsEntry
nil, // 24: llmgw.v1.Heartbeat.EngineVersionsEntry
nil, // 25: llmgw.v1.EngineProfile.EnvEntry
(*LogsRequest)(nil), // 22: llmgw.v1.LogsRequest
(*LogsReply)(nil), // 23: llmgw.v1.LogsReply
nil, // 24: llmgw.v1.Register.LabelsEntry
nil, // 25: llmgw.v1.Register.EngineVersionsEntry
nil, // 26: llmgw.v1.Heartbeat.EngineVersionsEntry
nil, // 27: llmgw.v1.EngineProfile.EnvEntry
}
var file_llmgw_v1_agent_proto_depIdxs = []int32{
3, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
10, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat
14, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus
16, // 3: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
19, // 4: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
20, // 5: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
21, // 6: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
18, // 7: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
11, // 8: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
22, // 9: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
7, // 10: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
8, // 11: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
23, // 12: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
6, // 13: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
9, // 14: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
5, // 15: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
4, // 16: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
0, // 17: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
5, // 18: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
9, // 19: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
24, // 20: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
13, // 21: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
12, // 22: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
0, // 23: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
15, // 24: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
17, // 25: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
25, // 26: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
1, // 27: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
2, // 28: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
28, // [28:29] is the sub-list for method output_type
27, // [27:28] is the sub-list for method input_type
27, // [27:27] is the sub-list for extension type_name
27, // [27:27] is the sub-list for extension extendee
0, // [0:27] is the sub-list for field type_name
23, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply
16, // 4: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
19, // 5: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
20, // 6: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
21, // 7: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
18, // 8: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
11, // 9: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
22, // 10: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest
24, // 11: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
7, // 12: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
8, // 13: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
25, // 14: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
6, // 15: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
9, // 16: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
5, // 17: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
4, // 18: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
0, // 19: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
5, // 20: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
9, // 21: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
26, // 22: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
13, // 23: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
12, // 24: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
0, // 25: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
15, // 26: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
17, // 27: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
27, // 28: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
1, // 29: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
2, // 30: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
30, // [30:31] is the sub-list for method output_type
29, // [29:30] is the sub-list for method input_type
29, // [29:29] is the sub-list for extension type_name
29, // [29:29] is the sub-list for extension extendee
0, // [0:29] is the sub-list for field type_name
}
func init() { file_llmgw_v1_agent_proto_init() }
@@ -2473,6 +2663,7 @@ func file_llmgw_v1_agent_proto_init() {
(*AgentMessage_Register)(nil),
(*AgentMessage_Heartbeat)(nil),
(*AgentMessage_Shard)(nil),
(*AgentMessage_Logs)(nil),
}
file_llmgw_v1_agent_proto_msgTypes[1].OneofWrappers = []any{
(*ControlMessage_Start)(nil),
@@ -2481,6 +2672,7 @@ func file_llmgw_v1_agent_proto_init() {
(*ControlMessage_Ack)(nil),
(*ControlMessage_RemoveWeights)(nil),
(*ControlMessage_Provision)(nil),
(*ControlMessage_Logs)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
@@ -2488,7 +2680,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: 25,
NumMessages: 27,
NumExtensions: 0,
NumServices: 1,
},
+28
View File
@@ -23,6 +23,7 @@ message AgentMessage {
Register register = 1;
Heartbeat heartbeat = 2;
ShardStatus shard = 3;
LogsReply logs = 4;
}
}
@@ -35,6 +36,7 @@ message ControlMessage {
Ack ack = 4;
RemoveWeights remove_weights = 5;
Provision provision = 6;
LogsRequest logs = 7;
}
}
@@ -418,3 +420,29 @@ message Ack {
bool ok = 2;
string error = 3;
}
// LogsRequest asks a node for the recent lines of one of its logs.
//
// Over the stream the agent already holds, because that is the one way into a
// node every deployment has: the admin page otherwise means ssh through a jump
// host to a machine whose address nobody remembers, to read a journal only
// root may open.
message LogsRequest {
// request_id pairs the reply with its request; several may be in flight.
string request_id = 1;
// source is "agent" for the agent's own log, or a shard id for that
// engine's output.
string source = 2;
int32 lines = 3;
}
// LogsReply answers a LogsRequest.
message LogsReply {
string request_id = 1;
repeated string lines = 2;
// sources are every log this node can show, so a page can offer them
// without knowing what runs there.
repeated string sources = 3;
// error says why the lines are absent: an unknown source, say.
string error = 4;
}
+45
View File
@@ -326,6 +326,8 @@ type Agent struct {
// it asked for, nil until it has; provStatus is where each stands and
// provSent the fingerprint of what the control plane last heard. All
// guarded by mu. provKick starts a pass without waiting for the timer.
logRing *logRing
prov *provision.Installer
provWants []provision.Want
provStatus map[string]provision.Status
@@ -372,6 +374,11 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
if log == nil {
log = slog.Default()
}
// Everything the agent logs also goes to a ring the admin page can read,
// so seeing what a node did does not mean ssh and a journal only root may
// open.
ring := newLogRing(agentLogLines)
log = slog.New(teeHandler{log.Handler(), slog.NewTextHandler(ring, nil)})
if httpClient == nil {
httpClient = http.DefaultClient
}
@@ -387,6 +394,7 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
http: httpClient,
shardReplica: map[string]string{},
provStatus: map[string]provision.Status{},
logRing: ring,
provKick: make(chan struct{}, 1),
}
a.prov = &provision.Installer{Root: cfg.EnginesDir, Log: log, OnStatus: a.provisionStatus}
@@ -960,6 +968,16 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
a.removeWeights(m.RemoveWeights.GetPaths())
case *llmgwv1.ControlMessage_Provision:
a.setProvision(m.Provision)
case *llmgwv1.ControlMessage_Logs:
reply := a.logsReply(m.Logs)
// Off the receive loop: a full send queue must not stop commands being
// read.
go func() {
select {
case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: reply}}:
case <-ctx.Done():
}
}()
case *llmgwv1.ControlMessage_Ack:
if !m.Ack.GetOk() {
a.log.Warn("control plane rejected a message",
@@ -1394,3 +1412,30 @@ func enginesFingerprint(vers map[string]string) string {
}
return b.String()
}
// logsReply answers a request for one of this node's logs.
func (a *Agent) logsReply(req *llmgwv1.LogsRequest) *llmgwv1.LogsReply {
n := int(req.GetLines())
if n <= 0 {
n = 200
}
n = min(n, agentLogLines)
shards := a.sup.IDs()
slices.Sort(shards)
out := &llmgwv1.LogsReply{
RequestId: req.GetRequestId(),
Sources: append([]string{"agent"}, shards...),
}
switch src := req.GetSource(); src {
case "", "agent":
out.Lines = a.logRing.tail(n)
default:
sh, err := a.sup.Get(src)
if err != nil {
out.Error = "no engine " + src + " runs on this node"
return out
}
out.Lines = sh.Logs(n)
}
return out
}
+84
View File
@@ -0,0 +1,84 @@
package agent
import (
"context"
"log/slog"
"strings"
"sync"
)
// agentLogLines is how much of its own log an agent keeps for the admin page.
// Enough to cover an engine install or a failed start; the journal keeps the
// rest for anyone who can reach it.
const agentLogLines = 2000
// logRing keeps the most recent lines written to it.
type logRing struct {
mu sync.Mutex
lines []string
next int
full bool
}
func newLogRing(n int) *logRing { return &logRing{lines: make([]string, n)} }
// Write takes whole lines, as a text handler writes them.
func (r *logRing) Write(p []byte) (int, error) {
r.mu.Lock()
defer r.mu.Unlock()
for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
r.lines[r.next] = line
r.next = (r.next + 1) % len(r.lines)
if r.next == 0 {
r.full = true
}
}
return len(p), nil
}
// tail returns up to n of the most recent lines, oldest first.
func (r *logRing) tail(n int) []string {
r.mu.Lock()
defer r.mu.Unlock()
size := r.next
if r.full {
size = len(r.lines)
}
n = min(n, size)
out := make([]string, 0, n)
for i := size - n; i < size; i++ {
idx := i
if r.full {
idx = (r.next + i) % len(r.lines)
}
out = append(out, r.lines[idx])
}
return out
}
// teeHandler sends every record to two handlers: wherever the agent already
// logs, and the ring the admin page reads.
type teeHandler struct{ a, b slog.Handler }
func (t teeHandler) Enabled(ctx context.Context, l slog.Level) bool {
return t.a.Enabled(ctx, l) || t.b.Enabled(ctx, l)
}
func (t teeHandler) Handle(ctx context.Context, r slog.Record) error {
var err error
if t.a.Enabled(ctx, r.Level) {
err = t.a.Handle(ctx, r.Clone())
}
if t.b.Enabled(ctx, r.Level) {
_ = t.b.Handle(ctx, r.Clone())
}
return err
}
func (t teeHandler) WithAttrs(as []slog.Attr) slog.Handler {
return teeHandler{t.a.WithAttrs(as), t.b.WithAttrs(as)}
}
func (t teeHandler) WithGroup(name string) slog.Handler {
return teeHandler{t.a.WithGroup(name), t.b.WithGroup(name)}
}
+49
View File
@@ -0,0 +1,49 @@
package agent
import (
"fmt"
"slices"
"strings"
"testing"
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
)
func TestTheLogRingKeepsTheNewestLinesInOrder(t *testing.T) {
r := newLogRing(3)
if got := r.tail(10); len(got) != 0 {
t.Fatalf("empty ring gave %v", got)
}
for i := 1; i <= 5; i++ {
fmt.Fprintf(r, "line %d\n", i)
}
if got := r.tail(10); !slices.Equal(got, []string{"line 3", "line 4", "line 5"}) {
t.Errorf("after wrapping: %v", got)
}
if got := r.tail(2); !slices.Equal(got, []string{"line 4", "line 5"}) {
t.Errorf("tail(2): %v", got)
}
}
func TestANodeAnswersForItsOwnLog(t *testing.T) {
cfg := Config{Server: "127.0.0.1:7443", NodeID: "n1", WeightsCache: t.TempDir(), EnginesDir: t.TempDir()}
cfg.Superv.WorkDir = t.TempDir()
a, err := New(cfg, nil, nil)
if err != nil {
t.Fatal(err)
}
a.log.Info("engine installation", "state", "installing")
r := a.logsReply(&llmgwv1.LogsRequest{RequestId: "x", Source: "agent", Lines: 50})
if r.GetRequestId() != "x" || !strings.Contains(strings.Join(r.GetLines(), "\n"), "engine installation") {
t.Errorf("the agent's own line is not in its log: %v", r.GetLines())
}
if !slices.Contains(r.GetSources(), "agent") {
t.Errorf("sources %v", r.GetSources())
}
r = a.logsReply(&llmgwv1.LogsRequest{RequestId: "y", Source: "no-such-shard"})
if r.GetError() == "" {
t.Error("an unknown engine was answered as if it existed")
}
}
+1
View File
@@ -116,6 +116,7 @@ func newCtlCmd() *cobra.Command {
cmd.AddCommand(
newNodesCmd(client, &asJSON),
newLogsCmd(client, &asJSON),
newModelsCmd(client, &asJSON),
newReplicasCmd(client, &asJSON),
newExplainCmd(client, &asJSON),
+58
View File
@@ -0,0 +1,58 @@
package cli
import (
"encoding/json"
"fmt"
"net/url"
"strconv"
"strings"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)
func newLogsCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
var (
source string
lines int
)
c := &cobra.Command{
Use: "logs NODE",
Short: "Show the recent lines of a node's agent log, or of an engine on it",
Long: "Read through the agent's own stream, so it needs no access to the node:\n" +
"no ssh, no jump host, no root for the journal. The agent keeps its last\n" +
"2000 lines and each engine's last few hundred.\n\n" +
" llmgw ctl logs llmgw-node06 the agent\n" +
" llmgw ctl logs llmgw-node06 --source SHARD an engine's output",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
q := url.Values{}
q.Set("source", source)
q.Set("lines", strconv.Itoa(lines))
var v adminapi.LogsView
err := client().get(cmd.Context(),
"/api/v1/nodes/"+url.PathEscape(args[0])+"/logs?"+q.Encode(), &v)
if *asJSON {
return json.NewEncoder(cmd.OutOrStdout()).Encode(v)
}
if err != nil {
return err
}
for _, l := range v.Lines {
fmt.Fprintln(cmd.OutOrStdout(), l)
}
if v.Error != "" {
return fmt.Errorf("%s", v.Error)
}
// Where else to look, since the source list is the one thing a
// person cannot guess.
fmt.Fprintf(cmd.ErrOrStderr(), "\n-- %s on %s; other sources: %s\n",
v.Source, v.Node, strings.Join(v.Sources, ", "))
return nil
},
}
c.Flags().StringVar(&source, "source", "agent", `"agent", or a shard id for that engine's output`)
c.Flags().IntVarP(&lines, "lines", "n", 200, "how many of the most recent lines")
return c
}
+16
View File
@@ -203,6 +203,7 @@ func newServerCmd() *cobra.Command {
api, usage, version.Version, 5*time.Second, true,
web.NavExtras{History: history, Window: cfg.Monitoring.Window}, log))
adminMux.Handle("GET /_static/", web.Static())
adminMux.HandleFunc("GET /nodes/{id}/logs", web.NodeLogsPage(api, true, version.Version))
// The same page on the client-facing port at /admin, behind a
// session and an admin flag. Handing the router the data rather
@@ -466,6 +467,21 @@ func (r *dispatcherRef) RemoveWeights(ctx context.Context, nodeID string, paths
return rm.RemoveWeights(ctx, nodeID, paths)
}
// Logs forwards a node log request, the same way RemoveWeights does: the
// controller finds it by type assertion, so a forwarder without it would make
// the capability silently absent at runtime.
func (r *dispatcherRef) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
d, err := r.get()
if err != nil {
return nil, nil, err
}
lr, ok := d.(control.LogReader)
if !ok {
return nil, nil, errors.New("this dispatcher cannot read node logs")
}
return lr.Logs(ctx, nodeID, source, lines)
}
func (r *dispatcherRef) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error {
d, err := r.get()
if err != nil {
+1
View File
@@ -56,6 +56,7 @@ func (a *API) Register(mux *http.ServeMux) {
mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights)
mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
}
// --- responses ---
+60
View File
@@ -0,0 +1,60 @@
package adminapi
import (
"context"
"net/http"
"strconv"
)
// LogsView is one of a node's logs, as the node sent it.
type LogsView struct {
Node string `json:"node"`
Source string `json:"source"`
Lines []string `json:"lines"`
// Sources are every log the node can show: "agent", and one per engine
// running there.
Sources []string `json:"sources"`
Error string `json:"error,omitempty"`
}
// maxLogLines bounds a request. The agent keeps two thousand lines of its own
// and a few hundred per engine, so more would only be refused further down.
const maxLogLines = 2000
// NodeLogs asks a node for the recent lines of one of its logs.
//
// Exported because the cluster page renders it server-side, the same as
// Nodes.
func (a *API) NodeLogs(ctx context.Context, node, source string, lines int) LogsView {
if source == "" {
source = "agent"
}
v := LogsView{Node: node, Source: source}
lr := a.ctl.LogReader()
if lr == nil {
v.Error = "this control plane cannot read node logs"
return v
}
if lines <= 0 {
lines = 300
}
out, sources, err := lr.Logs(ctx, node, source, min(lines, maxLogLines))
v.Lines, v.Sources = out, sources
if err != nil {
v.Error = err.Error()
}
if len(v.Sources) == 0 {
v.Sources = []string{"agent"}
}
return v
}
func (a *API) nodeLogs(w http.ResponseWriter, r *http.Request) {
lines, _ := strconv.Atoi(r.URL.Query().Get("lines"))
v := a.NodeLogs(r.Context(), r.PathValue("id"), r.URL.Query().Get("source"), lines)
code := http.StatusOK
if v.Error != "" && len(v.Lines) == 0 {
code = http.StatusBadGateway
}
writeJSON(w, code, v)
}
@@ -155,3 +155,36 @@ func TestAHeartbeatCarriesAnEngineInstalledAfterRegistration(t *testing.T) {
t.Errorf("the newly installed engine is not known: %v", n.EngineVersions)
}
}
func TestALogRequestIsAnsweredByTheNodeItWasSentTo(t *testing.T) {
s := newTestService()
c := &conn{nodeID: "n1", send: make(chan *llmgwv1.ControlMessage, 4), closed: make(chan struct{})}
s.conns["n1"] = c
type result struct {
lines, sources []string
err error
}
done := make(chan result, 1)
go func() {
l, src, err := s.Logs(t.Context(), "n1", "agent", 10)
done <- result{l, src, err}
}()
req := (<-c.send).GetLogs()
if req.GetSource() != "agent" || req.GetLines() != 10 {
t.Fatalf("sent %v", req)
}
s.handle("n1", &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: &llmgwv1.LogsReply{
RequestId: req.GetRequestId(), Lines: []string{"hello"}, Sources: []string{"agent", "shard-1"},
}}})
r := <-done
if r.err != nil || len(r.lines) != 1 || r.lines[0] != "hello" || len(r.sources) != 2 {
t.Errorf("got %+v", r)
}
if _, _, err := s.Logs(t.Context(), "nobody", "agent", 10); err == nil {
t.Error("a node with no stream answered")
}
}
+58
View File
@@ -10,6 +10,7 @@ import (
"net/http"
"slices"
"sync"
"time"
"connectrpc.com/connect"
@@ -52,6 +53,11 @@ type Service struct {
// engines is what every node is told to install, set once at start.
engines map[string]string
// logWaits are the log requests awaiting a reply, by request id.
logMu sync.Mutex
logWaits map[string]chan *llmgwv1.LogsReply
logSeq uint64
}
// SetEngines sets the engines every node installs, by name and version.
@@ -276,6 +282,15 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
"last_output", m.Shard.GetLastOutput())
}
case *llmgwv1.AgentMessage_Logs:
s.logMu.Lock()
ch, ok := s.logWaits[m.Logs.GetRequestId()]
delete(s.logWaits, m.Logs.GetRequestId())
s.logMu.Unlock()
if ok {
ch <- m.Logs // buffered; the asker may have given up
}
case *llmgwv1.AgentMessage_Register:
// A re-registration on a live stream means the agent's view of its own
// hardware changed — a GPU fell out, say. Take it.
@@ -511,3 +526,46 @@ func (s *Service) send(nodeID string, msg *llmgwv1.ControlMessage) error {
return fmt.Errorf("node %s is not keeping up with commands", nodeID)
}
}
// Logs asks a node for the recent lines of one of its logs: "agent", or a
// shard id for that engine's output.
//
// It returns the lines, and every source the node can show.
func (s *Service) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
s.logMu.Lock()
if s.logWaits == nil {
s.logWaits = map[string]chan *llmgwv1.LogsReply{}
}
s.logSeq++
id := fmt.Sprintf("logs-%d", s.logSeq)
ch := make(chan *llmgwv1.LogsReply, 1)
s.logWaits[id] = ch
s.logMu.Unlock()
defer func() {
s.logMu.Lock()
delete(s.logWaits, id)
s.logMu.Unlock()
}()
if source == "" {
source = "agent"
}
if err := s.send(nodeID, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_Logs{
Logs: &llmgwv1.LogsRequest{RequestId: id, Source: source, Lines: int32(lines)}, //nolint:gosec // bounded by the caller
}}); err != nil {
return nil, nil, err
}
// Bounded, because a page is waiting: an agent older than this request
// never answers it at all.
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
select {
case r := <-ch:
if r.GetError() != "" {
return nil, r.GetSources(), errors.New(r.GetError())
}
return r.GetLines(), r.GetSources(), nil
case <-ctx.Done():
return nil, nil, fmt.Errorf("node %s did not answer; an agent older than v0.8.4 cannot", nodeID)
}
}
+11
View File
@@ -204,6 +204,17 @@ func (c *Controller) WeightRemover() WeightRemover {
return r
}
// LogReader can ask a node for the recent lines of one of its logs.
type LogReader interface {
Logs(ctx context.Context, nodeID, source string, lines int) (out, sources []string, err error)
}
// LogReader returns the dispatcher's log capability, or nil.
func (c *Controller) LogReader() LogReader {
r, _ := c.disp.(LogReader)
return r
}
// Start runs the reconciler and begins publishing endpoints.
func (c *Controller) Start(ctx context.Context) {
if c.lead != nil {
+2
View File
@@ -157,6 +157,8 @@ func (r *Router) Handler() http.Handler {
web.NavExtras{
Mirror: r.mirror != nil, History: r.history, Window: r.histWindow,
}, r.log)))
mux.HandleFunc("GET /admin/nodes/{id}/logs", pages.AdminPage(
web.NodeLogsPage(r.clusterView, false, version.Version)))
pages.RegisterAdminUsers(mux, r.store)
// Quotas are written on the same tree the accounts page builds, so
// they are mounted together.
+68
View File
@@ -0,0 +1,68 @@
package web
import (
"context"
"net/http"
"strconv"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)
// LogSource reads a node's logs. Asked for by type assertion on the cluster
// source, so a source that cannot — a test's, an older one — leaves the page
// saying so rather than every Source having to grow the method.
type LogSource interface {
NodeLogs(ctx context.Context, node, source string, lines int) adminapi.LogsView
}
type nodeLogsData struct {
adminapi.LogsView
OnAdminListener bool
// Base is where this page lives, which differs between the admin
// listener and the client-facing port.
Base string
Back string
Count int
Counts []int
Follow bool
Version string
Now string
}
// NodeLogsPage shows the recent lines of one of a node's logs: the agent's
// own, or an engine's output.
//
// What a node did used to be learned by ssh through a jump host and a journal
// only root may read — which, when an engine install fails at two in the
// morning, is several steps between the page that says it failed and the line
// that says why.
func NodeLogsPage(src Source, onAdminListener bool, version string) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
node := r.PathValue("id")
d := nodeLogsData{
OnAdminListener: onAdminListener,
Base: "/admin/nodes/" + node + "/logs",
Back: "/admin",
Counts: []int{100, 300, 1000, 2000},
Follow: r.URL.Query().Get("follow") != "",
Version: version,
Now: time.Now().UTC().Format(time.RFC3339),
}
if onAdminListener {
d.Base, d.Back = "/nodes/"+node+"/logs", "/"
}
d.Count, _ = strconv.Atoi(r.URL.Query().Get("lines"))
if d.Count <= 0 {
d.Count = 300
}
ls, ok := src.(LogSource)
if !ok {
d.LogsView = adminapi.LogsView{Node: node, Error: "this control plane cannot read node logs"}
} else {
d.LogsView = ls.NodeLogs(r.Context(), node, r.URL.Query().Get("source"), d.Count)
}
noStore(w)
render(w, "node_logs", d)
}
}
+31
View File
@@ -0,0 +1,31 @@
package web
import (
"context"
"net/http/httptest"
"strings"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)
type logsFleet struct{ fakeSource }
func (logsFleet) NodeLogs(_ context.Context, node, source string, _ int) adminapi.LogsView {
return adminapi.LogsView{Node: node, Source: source, Lines: []string{"installed an engine <b>"},
Sources: []string{"agent", "shard-1"}}
}
func TestANodesLogIsShownWithItsOtherSources(t *testing.T) {
req := httptest.NewRequest("GET", "/admin/nodes/n6/logs?source=agent", nil)
req.SetPathValue("id", "n6")
rec := httptest.NewRecorder()
NodeLogsPage(logsFleet{}, false, "test")(rec, req)
body := rec.Body.String()
for _, want := range []string{"installed an engine &lt;b&gt;", "source=shard-1", "/admin/nodes/n6/logs"} {
if !strings.Contains(body, want) {
t.Errorf("page lacks %q", want)
}
}
}
+2 -1
View File
@@ -103,7 +103,8 @@ requested whenever a fallback served it.</p>
{{if .Nodes}}
<table><tr><th>Node</th><th>State</th><th class="num">GPUs</th><th class="num">RAM</th><th class="num">Disk</th><th>Engines</th><th>Classes</th><th>Fabrics</th><th class="num">Models</th><th class="num">Replicas</th><th>Last seen</th></tr>
{{range .Nodes}}<tr>
<td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}</td>
<td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}
<a class="note" href="{{if $.OnAdminListener}}/nodes/{{.ID}}/logs{{else}}/admin/nodes/{{.ID}}/logs{{end}}">logs</a></td>
<td class="state state-{{.State}}">{{.State}}</td>
<td class="num">{{.GPUFree}}/{{.GPUTotal}}</td>
<td class="num">{{if .MemoryGB}}{{.MemoryGB}}G{{else}}—{{end}}</td>
+31
View File
@@ -0,0 +1,31 @@
{{define "node_logs"}}<!doctype html>
<html lang="en"><head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
{{if .Follow}}<meta http-equiv="refresh" content="5">{{end}}
<title>llmgw — {{.Node}} logs</title><link rel="stylesheet" href="/_static/base.css">
</head><body><div class="wrap">
<header>
<h1>{{.Node}}</h1>
<p class="sub">{{if eq .Source "agent"}}the agent's log{{else}}output of engine <code>{{.Source}}</code>{{end}} · last {{.Count}} lines</p>
<nav><a href="{{.Back}}">cluster</a></nav>
</header>
<p>
{{- $d := . -}}
{{range .Sources}}<a href="{{$d.Base}}?source={{.}}&lines={{$d.Count}}{{if $d.Follow}}&follow=1{{end}}"{{if eq . $d.Source}} aria-current="page"><strong>{{.}}</strong>{{else}}>{{.}}{{end}}</a> &nbsp; {{end}}
</p>
<p class="note">
{{range .Counts}}<a href="{{$d.Base}}?source={{$d.Source}}&lines={{.}}{{if $d.Follow}}&follow=1{{end}}">{{.}}</a> {{end}}lines ·
{{if .Follow}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}">stop following</a>{{else}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}&follow=1">follow</a> (refresh every 5s){{end}}
</p>
{{if .Error}}<p class="bad">{{.Error}}</p>{{end}}
{{if .Lines}}<pre id="log">{{range .Lines}}{{.}}
{{end}}</pre>
{{/* Newest at the bottom, like a terminal, and scrolled there: the line
worth reading is almost always the last one. */}}
<script>window.scrollTo(0, document.body.scrollHeight);</script>
{{else if not .Error}}<p class="empty">Nothing logged yet.</p>{{end}}
<footer class="note">llmgw {{.Version}} · {{.Now}}</footer>
</div></body></html>{{end}}