From 0bd0d881c5d8474089963c8eb4860ff4e5104e09 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 7 Oct 2026 13:02:17 +0200 Subject: [PATCH] A replica says what its engine computes on, and llama.cpp's word on the CPU refuses a GPU placement llama.cpp's start-up output is read as a shard becomes ready: the devices it used, its backend, and the layers it offloaded. That goes up with READY in a new ShardStatus.compute, is kept on the shard, and is shown by the admin API, ctl replicas (ON) and the console (COMPUTE). A shard given GPUs whose engine says it found no usable GPU or offloaded no layer is stopped and failed, even where nvidia-smi cannot be asked. Co-Authored-By: Claude Opus 5.5 (1M context) --- api/gen/llmgw/v1/agent.pb.go | 319 ++++++++++++++-------- api/proto/llmgw/v1/agent.proto | 16 ++ docs/admin-api.openapi.json | 27 ++ internal/agent/agent.go | 14 +- internal/agent/agent_test.go | 17 +- internal/agent/engines/compute.go | 79 ++++++ internal/agent/engines/compute_test.go | 32 +++ internal/agent/gpucheck.go | 26 ++ internal/agent/gpucheck_test.go | 25 ++ internal/cli/ctl.go | 14 +- internal/cli/tui_console.go | 8 +- internal/control/adminapi/api.go | 28 ++ internal/control/agentsvc/service.go | 8 + internal/control/reconciler/reconciler.go | 16 ++ internal/control/reconciler/replica.go | 14 + 15 files changed, 517 insertions(+), 126 deletions(-) create mode 100644 internal/agent/engines/compute.go create mode 100644 internal/agent/engines/compute_test.go diff --git a/api/gen/llmgw/v1/agent.pb.go b/api/gen/llmgw/v1/agent.pb.go index e7c6dbcf..ed765c8d 100644 --- a/api/gen/llmgw/v1/agent.pb.go +++ b/api/gen/llmgw/v1/agent.pb.go @@ -90,7 +90,7 @@ func (x ShardStatus_State) Number() protoreflect.EnumNumber { // Deprecated: Use ShardStatus_State.Descriptor instead. func (ShardStatus_State) EnumDescriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{24, 0} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{25, 0} } // AgentMessage is anything a node tells the control plane. @@ -3125,6 +3125,79 @@ func (x *ProvisionStatus) GetDetail() string { return "" } +// Compute is where an engine runs its model, as it said when it loaded it. +type Compute struct { + state protoimpl.MessageState `protogen:"open.v1"` + // backend is "CUDA", "Vulkan", "Metal" or "CPU". + Backend string `protobuf:"bytes,1,opt,name=backend,proto3" json:"backend,omitempty"` + // devices are the devices it named: "CUDA0: Quadro RTX 6000 (7.5)". + Devices []string `protobuf:"bytes,2,rep,name=devices,proto3" json:"devices,omitempty"` + // layers_offloaded of layers_total went to the devices; both 0 when the + // engine did not say. + LayersOffloaded int32 `protobuf:"varint,3,opt,name=layers_offloaded,json=layersOffloaded,proto3" json:"layers_offloaded,omitempty"` + LayersTotal int32 `protobuf:"varint,4,opt,name=layers_total,json=layersTotal,proto3" json:"layers_total,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Compute) Reset() { + *x = Compute{} + mi := &file_llmgw_v1_agent_proto_msgTypes[24] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Compute) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Compute) ProtoMessage() {} + +func (x *Compute) ProtoReflect() protoreflect.Message { + mi := &file_llmgw_v1_agent_proto_msgTypes[24] + 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 Compute.ProtoReflect.Descriptor instead. +func (*Compute) Descriptor() ([]byte, []int) { + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{24} +} + +func (x *Compute) GetBackend() string { + if x != nil { + return x.Backend + } + return "" +} + +func (x *Compute) GetDevices() []string { + if x != nil { + return x.Devices + } + return nil +} + +func (x *Compute) GetLayersOffloaded() int32 { + if x != nil { + return x.LayersOffloaded + } + return 0 +} + +func (x *Compute) GetLayersTotal() int32 { + if x != nil { + return x.LayersTotal + } + return 0 +} + // ShardStatus reports a shard's lifecycle transition. type ShardStatus struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3144,14 +3217,17 @@ type ShardStatus struct { // wedged are the same thing — a shard that has been starting for ten // minutes — and telling them apart is the difference between waiting and // going to look. - Fetch *FetchProgress `protobuf:"bytes,8,opt,name=fetch,proto3" json:"fetch,omitempty"` + Fetch *FetchProgress `protobuf:"bytes,8,opt,name=fetch,proto3" json:"fetch,omitempty"` + // compute is what the engine says it computes on, read from its own output + // as it became ready: sent with STATE_READY when the engine says. + Compute *Compute `protobuf:"bytes,9,opt,name=compute,proto3" json:"compute,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *ShardStatus) Reset() { *x = ShardStatus{} - mi := &file_llmgw_v1_agent_proto_msgTypes[24] + mi := &file_llmgw_v1_agent_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3163,7 +3239,7 @@ func (x *ShardStatus) String() string { func (*ShardStatus) ProtoMessage() {} func (x *ShardStatus) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[24] + mi := &file_llmgw_v1_agent_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3176,7 +3252,7 @@ func (x *ShardStatus) ProtoReflect() protoreflect.Message { // Deprecated: Use ShardStatus.ProtoReflect.Descriptor instead. func (*ShardStatus) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{24} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{25} } func (x *ShardStatus) GetNodeId() string { @@ -3235,6 +3311,13 @@ func (x *ShardStatus) GetFetch() *FetchProgress { return nil } +func (x *ShardStatus) GetCompute() *Compute { + if x != nil { + return x.Compute + } + return nil +} + // FetchProgress is a download in flight. type FetchProgress struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3250,7 +3333,7 @@ type FetchProgress struct { func (x *FetchProgress) Reset() { *x = FetchProgress{} - mi := &file_llmgw_v1_agent_proto_msgTypes[25] + mi := &file_llmgw_v1_agent_proto_msgTypes[26] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3262,7 +3345,7 @@ func (x *FetchProgress) String() string { func (*FetchProgress) ProtoMessage() {} func (x *FetchProgress) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[25] + mi := &file_llmgw_v1_agent_proto_msgTypes[26] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3275,7 +3358,7 @@ func (x *FetchProgress) ProtoReflect() protoreflect.Message { // Deprecated: Use FetchProgress.ProtoReflect.Descriptor instead. func (*FetchProgress) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{25} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{26} } func (x *FetchProgress) GetSource() string { @@ -3338,7 +3421,7 @@ type StartShard struct { func (x *StartShard) Reset() { *x = StartShard{} - mi := &file_llmgw_v1_agent_proto_msgTypes[26] + mi := &file_llmgw_v1_agent_proto_msgTypes[27] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3350,7 +3433,7 @@ func (x *StartShard) String() string { func (*StartShard) ProtoMessage() {} func (x *StartShard) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[26] + mi := &file_llmgw_v1_agent_proto_msgTypes[27] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3363,7 +3446,7 @@ func (x *StartShard) ProtoReflect() protoreflect.Message { // Deprecated: Use StartShard.ProtoReflect.Descriptor instead. func (*StartShard) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{26} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{27} } func (x *StartShard) GetReplicaId() string { @@ -3510,7 +3593,7 @@ type EngineProfile struct { func (x *EngineProfile) Reset() { *x = EngineProfile{} - mi := &file_llmgw_v1_agent_proto_msgTypes[27] + mi := &file_llmgw_v1_agent_proto_msgTypes[28] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3522,7 +3605,7 @@ func (x *EngineProfile) String() string { func (*EngineProfile) ProtoMessage() {} func (x *EngineProfile) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[27] + mi := &file_llmgw_v1_agent_proto_msgTypes[28] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3535,7 +3618,7 @@ func (x *EngineProfile) ProtoReflect() protoreflect.Message { // Deprecated: Use EngineProfile.ProtoReflect.Descriptor instead. func (*EngineProfile) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{27} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{28} } func (x *EngineProfile) GetName() string { @@ -3645,7 +3728,7 @@ type RemoveWeights struct { func (x *RemoveWeights) Reset() { *x = RemoveWeights{} - mi := &file_llmgw_v1_agent_proto_msgTypes[28] + mi := &file_llmgw_v1_agent_proto_msgTypes[29] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3657,7 +3740,7 @@ func (x *RemoveWeights) String() string { func (*RemoveWeights) ProtoMessage() {} func (x *RemoveWeights) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[28] + mi := &file_llmgw_v1_agent_proto_msgTypes[29] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3670,7 +3753,7 @@ func (x *RemoveWeights) ProtoReflect() protoreflect.Message { // Deprecated: Use RemoveWeights.ProtoReflect.Descriptor instead. func (*RemoveWeights) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{28} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{29} } func (x *RemoveWeights) GetPaths() []string { @@ -3690,7 +3773,7 @@ type StopShard struct { func (x *StopShard) Reset() { *x = StopShard{} - mi := &file_llmgw_v1_agent_proto_msgTypes[29] + mi := &file_llmgw_v1_agent_proto_msgTypes[30] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3702,7 +3785,7 @@ func (x *StopShard) String() string { func (*StopShard) ProtoMessage() {} func (x *StopShard) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[29] + mi := &file_llmgw_v1_agent_proto_msgTypes[30] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3715,7 +3798,7 @@ func (x *StopShard) ProtoReflect() protoreflect.Message { // Deprecated: Use StopShard.ProtoReflect.Descriptor instead. func (*StopShard) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{29} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{30} } func (x *StopShard) GetShardId() string { @@ -3735,7 +3818,7 @@ type DrainShard struct { func (x *DrainShard) Reset() { *x = DrainShard{} - mi := &file_llmgw_v1_agent_proto_msgTypes[30] + mi := &file_llmgw_v1_agent_proto_msgTypes[31] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3747,7 +3830,7 @@ func (x *DrainShard) String() string { func (*DrainShard) ProtoMessage() {} func (x *DrainShard) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[30] + mi := &file_llmgw_v1_agent_proto_msgTypes[31] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3760,7 +3843,7 @@ func (x *DrainShard) ProtoReflect() protoreflect.Message { // Deprecated: Use DrainShard.ProtoReflect.Descriptor instead. func (*DrainShard) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{30} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{31} } func (x *DrainShard) GetShardId() string { @@ -3782,7 +3865,7 @@ type Ack struct { func (x *Ack) Reset() { *x = Ack{} - mi := &file_llmgw_v1_agent_proto_msgTypes[31] + mi := &file_llmgw_v1_agent_proto_msgTypes[32] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3794,7 +3877,7 @@ func (x *Ack) String() string { func (*Ack) ProtoMessage() {} func (x *Ack) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[31] + mi := &file_llmgw_v1_agent_proto_msgTypes[32] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3807,7 +3890,7 @@ func (x *Ack) ProtoReflect() protoreflect.Message { // Deprecated: Use Ack.ProtoReflect.Descriptor instead. func (*Ack) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{31} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{32} } func (x *Ack) GetMessageId() string { @@ -3851,7 +3934,7 @@ type LogsRequest struct { func (x *LogsRequest) Reset() { *x = LogsRequest{} - mi := &file_llmgw_v1_agent_proto_msgTypes[32] + mi := &file_llmgw_v1_agent_proto_msgTypes[33] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3863,7 +3946,7 @@ func (x *LogsRequest) String() string { func (*LogsRequest) ProtoMessage() {} func (x *LogsRequest) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[32] + mi := &file_llmgw_v1_agent_proto_msgTypes[33] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3876,7 +3959,7 @@ func (x *LogsRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use LogsRequest.ProtoReflect.Descriptor instead. func (*LogsRequest) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{32} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{33} } func (x *LogsRequest) GetRequestId() string { @@ -3916,7 +3999,7 @@ type LogsReply struct { func (x *LogsReply) Reset() { *x = LogsReply{} - mi := &file_llmgw_v1_agent_proto_msgTypes[33] + mi := &file_llmgw_v1_agent_proto_msgTypes[34] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3928,7 +4011,7 @@ func (x *LogsReply) String() string { func (*LogsReply) ProtoMessage() {} func (x *LogsReply) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[33] + mi := &file_llmgw_v1_agent_proto_msgTypes[34] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3941,7 +4024,7 @@ func (x *LogsReply) ProtoReflect() protoreflect.Message { // Deprecated: Use LogsReply.ProtoReflect.Descriptor instead. func (*LogsReply) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{33} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{34} } func (x *LogsReply) GetRequestId() string { @@ -3987,7 +4070,7 @@ type FabricScanRequest struct { func (x *FabricScanRequest) Reset() { *x = FabricScanRequest{} - mi := &file_llmgw_v1_agent_proto_msgTypes[34] + mi := &file_llmgw_v1_agent_proto_msgTypes[35] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3999,7 +4082,7 @@ func (x *FabricScanRequest) String() string { func (*FabricScanRequest) ProtoMessage() {} func (x *FabricScanRequest) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[34] + mi := &file_llmgw_v1_agent_proto_msgTypes[35] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4012,7 +4095,7 @@ func (x *FabricScanRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use FabricScanRequest.ProtoReflect.Descriptor instead. func (*FabricScanRequest) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{34} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{35} } func (x *FabricScanRequest) GetRequestId() string { @@ -4051,7 +4134,7 @@ type FabricScanReply struct { func (x *FabricScanReply) Reset() { *x = FabricScanReply{} - mi := &file_llmgw_v1_agent_proto_msgTypes[35] + mi := &file_llmgw_v1_agent_proto_msgTypes[36] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4063,7 +4146,7 @@ func (x *FabricScanReply) String() string { func (*FabricScanReply) ProtoMessage() {} func (x *FabricScanReply) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[35] + mi := &file_llmgw_v1_agent_proto_msgTypes[36] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4076,7 +4159,7 @@ func (x *FabricScanReply) ProtoReflect() protoreflect.Message { // Deprecated: Use FabricScanReply.ProtoReflect.Descriptor instead. func (*FabricScanReply) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{35} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{36} } func (x *FabricScanReply) GetRequestId() string { @@ -4132,7 +4215,7 @@ type FabricTestRequest struct { func (x *FabricTestRequest) Reset() { *x = FabricTestRequest{} - mi := &file_llmgw_v1_agent_proto_msgTypes[36] + mi := &file_llmgw_v1_agent_proto_msgTypes[37] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4144,7 +4227,7 @@ func (x *FabricTestRequest) String() string { func (*FabricTestRequest) ProtoMessage() {} func (x *FabricTestRequest) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[36] + mi := &file_llmgw_v1_agent_proto_msgTypes[37] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4157,7 +4240,7 @@ func (x *FabricTestRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use FabricTestRequest.ProtoReflect.Descriptor instead. func (*FabricTestRequest) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{36} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{37} } func (x *FabricTestRequest) GetRequestId() string { @@ -4226,7 +4309,7 @@ type FabricTestReply struct { func (x *FabricTestReply) Reset() { *x = FabricTestReply{} - mi := &file_llmgw_v1_agent_proto_msgTypes[37] + mi := &file_llmgw_v1_agent_proto_msgTypes[38] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4238,7 +4321,7 @@ func (x *FabricTestReply) String() string { func (*FabricTestReply) ProtoMessage() {} func (x *FabricTestReply) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[37] + mi := &file_llmgw_v1_agent_proto_msgTypes[38] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4251,7 +4334,7 @@ func (x *FabricTestReply) ProtoReflect() protoreflect.Message { // Deprecated: Use FabricTestReply.ProtoReflect.Descriptor instead. func (*FabricTestReply) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{37} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{38} } func (x *FabricTestReply) GetRequestId() string { @@ -4311,7 +4394,7 @@ type UpgradeAgent struct { func (x *UpgradeAgent) Reset() { *x = UpgradeAgent{} - mi := &file_llmgw_v1_agent_proto_msgTypes[38] + mi := &file_llmgw_v1_agent_proto_msgTypes[39] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4323,7 +4406,7 @@ func (x *UpgradeAgent) String() string { func (*UpgradeAgent) ProtoMessage() {} func (x *UpgradeAgent) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[38] + mi := &file_llmgw_v1_agent_proto_msgTypes[39] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4336,7 +4419,7 @@ func (x *UpgradeAgent) ProtoReflect() protoreflect.Message { // Deprecated: Use UpgradeAgent.ProtoReflect.Descriptor instead. func (*UpgradeAgent) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{38} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{39} } func (x *UpgradeAgent) GetRequestId() string { @@ -4373,7 +4456,7 @@ type UpgradeReply struct { func (x *UpgradeReply) Reset() { *x = UpgradeReply{} - mi := &file_llmgw_v1_agent_proto_msgTypes[39] + mi := &file_llmgw_v1_agent_proto_msgTypes[40] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4385,7 +4468,7 @@ func (x *UpgradeReply) String() string { func (*UpgradeReply) ProtoMessage() {} func (x *UpgradeReply) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[39] + mi := &file_llmgw_v1_agent_proto_msgTypes[40] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4398,7 +4481,7 @@ func (x *UpgradeReply) ProtoReflect() protoreflect.Message { // Deprecated: Use UpgradeReply.ProtoReflect.Descriptor instead. func (*UpgradeReply) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{39} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{40} } func (x *UpgradeReply) GetRequestId() string { @@ -4434,7 +4517,7 @@ type LoraAdapter struct { func (x *LoraAdapter) Reset() { *x = LoraAdapter{} - mi := &file_llmgw_v1_agent_proto_msgTypes[40] + mi := &file_llmgw_v1_agent_proto_msgTypes[41] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -4446,7 +4529,7 @@ func (x *LoraAdapter) String() string { func (*LoraAdapter) ProtoMessage() {} func (x *LoraAdapter) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[40] + mi := &file_llmgw_v1_agent_proto_msgTypes[41] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -4459,7 +4542,7 @@ func (x *LoraAdapter) ProtoReflect() protoreflect.Message { // Deprecated: Use LoraAdapter.ProtoReflect.Descriptor instead. func (*LoraAdapter) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{40} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{41} } func (x *LoraAdapter) GetName() string { @@ -4780,7 +4863,12 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\aversion\x18\x02 \x01(\tR\aversion\x12\x18\n" + "\avariant\x18\x03 \x01(\tR\avariant\x12\x14\n" + "\x05state\x18\x04 \x01(\tR\x05state\x12\x16\n" + - "\x06detail\x18\x05 \x01(\tR\x06detail\"\xbb\x03\n" + + "\x06detail\x18\x05 \x01(\tR\x06detail\"\x8b\x01\n" + + "\aCompute\x12\x18\n" + + "\abackend\x18\x01 \x01(\tR\abackend\x12\x18\n" + + "\adevices\x18\x02 \x03(\tR\adevices\x12)\n" + + "\x10layers_offloaded\x18\x03 \x01(\x05R\x0flayersOffloaded\x12!\n" + + "\flayers_total\x18\x04 \x01(\x05R\vlayersTotal\"\xe8\x03\n" + "\vShardStatus\x12\x17\n" + "\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1d\n" + "\n" + @@ -4791,7 +4879,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\texit_code\x18\x06 \x01(\x05R\bexitCode\x12\x1f\n" + "\vlast_output\x18\a \x03(\tR\n" + "lastOutput\x12-\n" + - "\x05fetch\x18\b \x01(\v2\x17.llmgw.v1.FetchProgressR\x05fetch\"\xa2\x01\n" + + "\x05fetch\x18\b \x01(\v2\x17.llmgw.v1.FetchProgressR\x05fetch\x12+\n" + + "\acompute\x18\t \x01(\v2\x11.llmgw.v1.ComputeR\acompute\"\xa2\x01\n" + "\x05State\x12\x15\n" + "\x11STATE_UNSPECIFIED\x10\x00\x12\x11\n" + "\rSTATE_PENDING\x10\x01\x12\x12\n" + @@ -4933,7 +5022,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, 49) +var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 50) var file_llmgw_v1_agent_proto_goTypes = []any{ (ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State (*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage @@ -4960,73 +5049,74 @@ var file_llmgw_v1_agent_proto_goTypes = []any{ (*Provision)(nil), // 22: llmgw.v1.Provision (*EngineWant)(nil), // 23: llmgw.v1.EngineWant (*ProvisionStatus)(nil), // 24: llmgw.v1.ProvisionStatus - (*ShardStatus)(nil), // 25: llmgw.v1.ShardStatus - (*FetchProgress)(nil), // 26: llmgw.v1.FetchProgress - (*StartShard)(nil), // 27: llmgw.v1.StartShard - (*EngineProfile)(nil), // 28: llmgw.v1.EngineProfile - (*RemoveWeights)(nil), // 29: llmgw.v1.RemoveWeights - (*StopShard)(nil), // 30: llmgw.v1.StopShard - (*DrainShard)(nil), // 31: llmgw.v1.DrainShard - (*Ack)(nil), // 32: llmgw.v1.Ack - (*LogsRequest)(nil), // 33: llmgw.v1.LogsRequest - (*LogsReply)(nil), // 34: llmgw.v1.LogsReply - (*FabricScanRequest)(nil), // 35: llmgw.v1.FabricScanRequest - (*FabricScanReply)(nil), // 36: llmgw.v1.FabricScanReply - (*FabricTestRequest)(nil), // 37: llmgw.v1.FabricTestRequest - (*FabricTestReply)(nil), // 38: llmgw.v1.FabricTestReply - (*UpgradeAgent)(nil), // 39: llmgw.v1.UpgradeAgent - (*UpgradeReply)(nil), // 40: llmgw.v1.UpgradeReply - (*LoraAdapter)(nil), // 41: llmgw.v1.LoraAdapter - nil, // 42: llmgw.v1.ControlMessage.TraceEntry - nil, // 43: llmgw.v1.Register.LabelsEntry - nil, // 44: llmgw.v1.Register.EngineVersionsEntry - nil, // 45: llmgw.v1.GPU.NicPathsEntry - nil, // 46: llmgw.v1.IBPort.CountersEntry - nil, // 47: llmgw.v1.Heartbeat.EngineVersionsEntry - nil, // 48: llmgw.v1.EngineWant.OfflineSha256Entry - nil, // 49: llmgw.v1.EngineProfile.EnvEntry + (*Compute)(nil), // 25: llmgw.v1.Compute + (*ShardStatus)(nil), // 26: llmgw.v1.ShardStatus + (*FetchProgress)(nil), // 27: llmgw.v1.FetchProgress + (*StartShard)(nil), // 28: llmgw.v1.StartShard + (*EngineProfile)(nil), // 29: llmgw.v1.EngineProfile + (*RemoveWeights)(nil), // 30: llmgw.v1.RemoveWeights + (*StopShard)(nil), // 31: llmgw.v1.StopShard + (*DrainShard)(nil), // 32: llmgw.v1.DrainShard + (*Ack)(nil), // 33: llmgw.v1.Ack + (*LogsRequest)(nil), // 34: llmgw.v1.LogsRequest + (*LogsReply)(nil), // 35: llmgw.v1.LogsReply + (*FabricScanRequest)(nil), // 36: llmgw.v1.FabricScanRequest + (*FabricScanReply)(nil), // 37: llmgw.v1.FabricScanReply + (*FabricTestRequest)(nil), // 38: llmgw.v1.FabricTestRequest + (*FabricTestReply)(nil), // 39: llmgw.v1.FabricTestReply + (*UpgradeAgent)(nil), // 40: llmgw.v1.UpgradeAgent + (*UpgradeReply)(nil), // 41: llmgw.v1.UpgradeReply + (*LoraAdapter)(nil), // 42: llmgw.v1.LoraAdapter + nil, // 43: llmgw.v1.ControlMessage.TraceEntry + nil, // 44: llmgw.v1.Register.LabelsEntry + nil, // 45: llmgw.v1.Register.EngineVersionsEntry + nil, // 46: llmgw.v1.GPU.NicPathsEntry + nil, // 47: llmgw.v1.IBPort.CountersEntry + nil, // 48: llmgw.v1.Heartbeat.EngineVersionsEntry + nil, // 49: llmgw.v1.EngineWant.OfflineSha256Entry + nil, // 50: llmgw.v1.EngineProfile.EnvEntry } var file_llmgw_v1_agent_proto_depIdxs = []int32{ 6, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register 21, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat - 25, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus - 34, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply - 36, // 4: llmgw.v1.AgentMessage.fabric_scan:type_name -> llmgw.v1.FabricScanReply - 38, // 5: llmgw.v1.AgentMessage.fabric_test:type_name -> llmgw.v1.FabricTestReply - 40, // 6: llmgw.v1.AgentMessage.upgrade:type_name -> llmgw.v1.UpgradeReply + 26, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus + 35, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply + 37, // 4: llmgw.v1.AgentMessage.fabric_scan:type_name -> llmgw.v1.FabricScanReply + 39, // 5: llmgw.v1.AgentMessage.fabric_test:type_name -> llmgw.v1.FabricTestReply + 41, // 6: llmgw.v1.AgentMessage.upgrade:type_name -> llmgw.v1.UpgradeReply 2, // 7: llmgw.v1.AgentMessage.kv_events:type_name -> llmgw.v1.KVEvents - 27, // 8: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard - 30, // 9: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard - 31, // 10: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard - 32, // 11: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack - 29, // 12: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights + 28, // 8: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard + 31, // 9: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard + 32, // 10: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard + 33, // 11: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack + 30, // 12: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights 22, // 13: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision - 33, // 14: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest - 35, // 15: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest - 37, // 16: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest - 39, // 17: llmgw.v1.ControlMessage.upgrade:type_name -> llmgw.v1.UpgradeAgent + 34, // 14: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest + 36, // 15: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest + 38, // 16: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest + 40, // 17: llmgw.v1.ControlMessage.upgrade:type_name -> llmgw.v1.UpgradeAgent 5, // 18: llmgw.v1.ControlMessage.control_planes:type_name -> llmgw.v1.ControlPlanes 4, // 19: llmgw.v1.ControlMessage.firewall:type_name -> llmgw.v1.Firewall - 42, // 20: llmgw.v1.ControlMessage.trace:type_name -> llmgw.v1.ControlMessage.TraceEntry - 43, // 21: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry + 43, // 20: llmgw.v1.ControlMessage.trace:type_name -> llmgw.v1.ControlMessage.TraceEntry + 44, // 21: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry 10, // 22: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU 12, // 23: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment - 44, // 24: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry + 45, // 24: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry 9, // 25: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard 13, // 26: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights 8, // 27: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage 7, // 28: llmgw.v1.Register.host:type_name -> llmgw.v1.Host 19, // 29: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX 0, // 30: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State - 45, // 31: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry + 46, // 31: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry 11, // 32: llmgw.v1.GPU.mig:type_name -> llmgw.v1.MIGDevice 16, // 33: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad 15, // 34: llmgw.v1.NodeLoad.gpus:type_name -> llmgw.v1.GPUSensor - 46, // 35: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry + 47, // 35: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry 18, // 36: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager 8, // 37: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage 13, // 38: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights - 47, // 39: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry + 48, // 39: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry 24, // 40: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus 14, // 41: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad 17, // 42: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort @@ -5034,19 +5124,20 @@ var file_llmgw_v1_agent_proto_depIdxs = []int32{ 20, // 44: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState 19, // 45: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX 23, // 46: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant - 48, // 47: llmgw.v1.EngineWant.offline_sha256:type_name -> llmgw.v1.EngineWant.OfflineSha256Entry + 49, // 47: llmgw.v1.EngineWant.offline_sha256:type_name -> llmgw.v1.EngineWant.OfflineSha256Entry 0, // 48: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State - 26, // 49: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress - 28, // 50: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile - 49, // 51: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry - 41, // 52: llmgw.v1.EngineProfile.adapters:type_name -> llmgw.v1.LoraAdapter - 1, // 53: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage - 3, // 54: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage - 54, // [54:55] is the sub-list for method output_type - 53, // [53:54] is the sub-list for method input_type - 53, // [53:53] is the sub-list for extension type_name - 53, // [53:53] is the sub-list for extension extendee - 0, // [0:53] is the sub-list for field type_name + 27, // 49: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress + 25, // 50: llmgw.v1.ShardStatus.compute:type_name -> llmgw.v1.Compute + 29, // 51: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile + 50, // 52: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry + 42, // 53: llmgw.v1.EngineProfile.adapters:type_name -> llmgw.v1.LoraAdapter + 1, // 54: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage + 3, // 55: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage + 55, // [55:56] is the sub-list for method output_type + 54, // [54:55] is the sub-list for method input_type + 54, // [54:54] is the sub-list for extension type_name + 54, // [54:54] is the sub-list for extension extendee + 0, // [0:54] is the sub-list for field type_name } func init() { file_llmgw_v1_agent_proto_init() } @@ -5084,7 +5175,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: 49, + NumMessages: 50, NumExtensions: 0, NumServices: 1, }, diff --git a/api/proto/llmgw/v1/agent.proto b/api/proto/llmgw/v1/agent.proto index 387cc205..c95af4bb 100644 --- a/api/proto/llmgw/v1/agent.proto +++ b/api/proto/llmgw/v1/agent.proto @@ -605,6 +605,18 @@ message ProvisionStatus { string detail = 5; } +// Compute is where an engine runs its model, as it said when it loaded it. +message Compute { + // backend is "CUDA", "Vulkan", "Metal" or "CPU". + string backend = 1; + // devices are the devices it named: "CUDA0: Quadro RTX 6000 (7.5)". + repeated string devices = 2; + // layers_offloaded of layers_total went to the devices; both 0 when the + // engine did not say. + int32 layers_offloaded = 3; + int32 layers_total = 4; +} + // ShardStatus reports a shard's lifecycle transition. message ShardStatus { string node_id = 1; @@ -626,6 +638,10 @@ message ShardStatus { // going to look. FetchProgress fetch = 8; + // compute is what the engine says it computes on, read from its own output + // as it became ready: sent with STATE_READY when the engine says. + Compute compute = 9; + enum State { STATE_UNSPECIFIED = 0; STATE_PENDING = 1; diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 45152ac0..88f11a96 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -3714,6 +3714,29 @@ } } }, + "ComputeView": { + "type": "object", + "description": "ComputeView is where an engine runs its model, by its own account.", + "properties": { + "backend": { + "type": "string", + "description": "Backend is CUDA, Vulkan, Metal, ROCm, GPU or CPU." + }, + "devices": { + "type": "array", + "items": { + "type": "string" + } + }, + "layers_offloaded": { + "type": "integer", + "description": "LayersOffloaded of LayersTotal went to the devices; absent when unsaid." + }, + "layers_total": { + "type": "integer" + } + } + }, "ControlPlane": { "type": "object", "description": "ControlPlane is one control plane.", @@ -5779,6 +5802,10 @@ "memory_bytes": { "type": "integer", "format": "int64" + }, + "compute": { + "$ref": "#/components/schemas/ComputeView", + "description": "Compute is where the engine said it put the model as it became ready: absent when it did not say." } } }, diff --git a/internal/agent/agent.go b/internal/agent/agent.go index e067ee7e..1c6d3549 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -1130,6 +1130,14 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan< } else { err = a.sup.WaitAlive(ctx, sh) } + // Where it put the model, by its own account: said with the readiness, + // and a GPU placement it computes on the CPU refused even where + // nvidia-smi cannot be asked. + compute, cerr := engineCompute(eng, spec.GPUIndices, sh.Logs(512)) + if err == nil && cerr != nil { + err = cerr + _ = a.sup.Stop(ctx, shardID) + } if err == nil { err = a.onItsGPUs(ctx, eng, spec, sh.Status().PID) if err != nil { @@ -1142,7 +1150,11 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan< report(supervisor.Failed, err.Error(), sh.Logs(20)) return } - report(supervisor.Ready, "", nil) + if compute != nil { + send(&llmgwv1.ShardStatus{State: agentsvc.StateToWire(supervisor.Ready), Compute: compute}) + } else { + report(supervisor.Ready, "", nil) + } // Its dying later is reported by the session's watchExits. } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 8bf49290..441bbf5c 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -37,11 +37,12 @@ import ( type statusRecorder struct { ch chan status - mu sync.Mutex - calls []adoptCall - adopted []reconciler.AdoptedShard - unclaim []string // shard ids to report back as unclaimed - fetches []fetchReport + mu sync.Mutex + calls []adoptCall + adopted []reconciler.AdoptedShard + unclaim []string // shard ids to report back as unclaimed + fetches []fetchReport + computes []reconciler.Compute } type status struct { @@ -61,6 +62,12 @@ func (r *statusRecorder) OnShardStatus(replicaID, shardID string, st supervisor. // fetches records download progress, so a test can assert that a fetch is // visible from the control plane rather than only in the node's log. +func (r *statusRecorder) OnShardCompute(replicaID, shardID string, c reconciler.Compute) { + r.mu.Lock() + defer r.mu.Unlock() + r.computes = append(r.computes, c) +} + func (r *statusRecorder) OnFetchProgress(replicaID, shardID, source string, done, total int64) { r.mu.Lock() defer r.mu.Unlock() diff --git a/internal/agent/engines/compute.go b/internal/agent/engines/compute.go new file mode 100644 index 00000000..34ae7022 --- /dev/null +++ b/internal/agent/engines/compute.go @@ -0,0 +1,79 @@ +package engines + +import ( + "regexp" + "strconv" + "strings" +) + +// Compute is where an engine runs its model, as it said while loading it. +type Compute struct { + // Backend is CUDA, Vulkan, Metal, ROCm or CPU. + Backend string + // Devices are those it named: "CUDA0: Quadro RTX 6000 (7.5)". + Devices []string + // Offloaded of Total layers went to the devices; 0 and 0 when it did + // not say. + Offloaded, Total int + // NoGPU is the engine saying it found no device it could use. + NoGPU bool +} + +var ( + offloadedRE = regexp.MustCompile(`offloaded (\d+)/(\d+) layers to GPU`) + usingRE = regexp.MustCompile(`using device ([A-Za-z]+\d+) \(([^)]*)\)`) + ccRE = regexp.MustCompile(`Device (\d+): ([^,]+), compute capability ([0-9.]+)`) + noDevicesRE = regexp.MustCompile(`found 0 (CUDA|Vulkan|ROCm) devices|no usable GPU found`) +) + +// ReadLlamaCompute reads llama.cpp's start-up output for where it put the +// model, reporting false when the output says nothing of it. +func ReadLlamaCompute(lines []string) (Compute, bool) { + var c Compute + known := false + cc := map[string]string{} // device index → compute capability + for _, l := range lines { + if m := ccRE.FindStringSubmatch(l); m != nil { + cc[m[1]] = m[3] + } + } + seen := map[string]bool{} + for _, l := range lines { + if m := offloadedRE.FindStringSubmatch(l); m != nil { + c.Offloaded, _ = strconv.Atoi(m[1]) + c.Total, _ = strconv.Atoi(m[2]) + known = true + } + if m := usingRE.FindStringSubmatch(l); m != nil && !seen[m[1]] { + seen[m[1]] = true + d := m[1] + ": " + strings.TrimSpace(m[2]) + idx := strings.TrimLeft(m[1], "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") + if v, ok := cc[idx]; ok && strings.HasPrefix(m[1], "CUDA") { + d += " (" + v + ")" + } + c.Devices = append(c.Devices, d) + if c.Backend == "" { + c.Backend = strings.TrimRight(m[1], "0123456789") + } + known = true + } + if noDevicesRE.MatchString(l) { + c.NoGPU, known = true, true + } + } + switch { + case c.OnTheCPU(): + c.Backend = "CPU" + case c.Backend == "" && c.Offloaded > 0: + c.Backend = "GPU" + case c.Backend == "": + c.Backend = "CPU" + } + return c, known +} + +// OnTheCPU is whether a model the engine was given GPUs for is computed on +// the CPU, by its own account: no device found, or no layer offloaded. +func (c Compute) OnTheCPU() bool { + return c.NoGPU || (c.Total > 0 && c.Offloaded == 0) +} diff --git a/internal/agent/engines/compute_test.go b/internal/agent/engines/compute_test.go new file mode 100644 index 00000000..d113afb9 --- /dev/null +++ b/internal/agent/engines/compute_test.go @@ -0,0 +1,32 @@ +package engines + +import ( + "strings" + "testing" +) + +func TestLlamaCppSaysWhereItPutTheModel(t *testing.T) { + gpu := strings.Split(`ggml_cuda_init: found 1 CUDA devices: + Device 0: Quadro RTX 6000, compute capability 7.5, VMM: yes +load_backend: loaded CUDA backend from /opt/llmgw/engines/llamacpp/b10991/libggml-cuda.so +llama_model_load_from_file_impl: using device CUDA0 (Quadro RTX 6000) - 23800 MiB free +load_tensors: offloading 48 repeating layers to GPU +load_tensors: offloaded 49/49 layers to GPU`, "\n") + c, ok := ReadLlamaCompute(gpu) + if !ok || c.Backend != "CUDA" || c.Offloaded != 49 || c.Total != 49 || c.OnTheCPU() || + len(c.Devices) != 1 || c.Devices[0] != "CUDA0: Quadro RTX 6000 (7.5)" { + t.Errorf("on the GPU: %+v %v", c, ok) + } + + cpu := strings.Split(`warning: no usable GPU found, --gpu-layers option will be ignored +warning: one possible reason is that llama.cpp was compiled without GPU support +load_tensors: offloaded 0/49 layers to GPU`, "\n") + c, ok = ReadLlamaCompute(cpu) + if !ok || c.Backend != "CPU" || !c.OnTheCPU() { + t.Errorf("fell back to the CPU: %+v %v", c, ok) + } + + if _, ok := ReadLlamaCompute([]string{"main: server is listening on http://127.0.0.1:20001"}); ok { + t.Error("output that says nothing was read as saying something") + } +} diff --git a/internal/agent/gpucheck.go b/internal/agent/gpucheck.go index 0e4c086a..3bed7fc6 100644 --- a/internal/agent/gpucheck.go +++ b/internal/agent/gpucheck.go @@ -8,6 +8,7 @@ import ( "strings" "time" + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines" ) @@ -153,3 +154,28 @@ func joinIndices(is []int) string { } return strings.Join(s, ", ") } + +// engineCompute is where an engine said it put its model, from its output +// so far — llama.cpp's, which says — nil for one that said nothing. A shard +// given GPUs that its engine says it computes on the CPU is an error, found +// this way even where nvidia-smi cannot be asked. +func engineCompute(eng engines.Engine, gpus []int, lines []string) (*llmgwv1.Compute, error) { + if _, ok := eng.(engines.CPUFallback); !ok { + return nil, nil + } + read, ok := engines.ReadLlamaCompute(lines) + if !ok { + return nil, nil + } + c := &llmgwv1.Compute{Backend: read.Backend, Devices: read.Devices, + LayersOffloaded: int32(read.Offloaded), LayersTotal: int32(read.Total)} //nolint:gosec // layer counts + if len(gpus) > 0 && read.OnTheCPU() { + why := "its GPU backend saw no device it could use" + if read.Total > 0 && !read.NoGPU { + why = fmt.Sprintf("it offloaded %d of %d layers", read.Offloaded, read.Total) + } + return c, fmt.Errorf("placed on GPU %v but computing on the CPU: %s, by llama.cpp's own account "+ + "(llmgw ctl engine-check on this node says why)", gpus, why) + } + return c, nil +} diff --git a/internal/agent/gpucheck_test.go b/internal/agent/gpucheck_test.go index a6a2bb33..f73dc376 100644 --- a/internal/agent/gpucheck_test.go +++ b/internal/agent/gpucheck_test.go @@ -122,3 +122,28 @@ func TestForeignGPUMemoryLeavesOutTheAgentsOwnShards(t *testing.T) { t.Errorf("no reading came back as %v", foreign) } } + +func TestLlamaCppsOwnWordOnTheCPUIsEnoughToRefuse(t *testing.T) { + eng := fallbackEngine{} + // Without nvidia-smi to ask, the engine's own output says it. + c, err := engineCompute(eng, []int{0}, []string{ + "warning: no usable GPU found, --gpu-layers option will be ignored", + "load_tensors: offloaded 0/49 layers to GPU"}) + if err == nil || !strings.Contains(err.Error(), "computing on the CPU") || c.GetBackend() != "CPU" { + t.Fatalf("a GPU shard llama.cpp put on the CPU: %v %+v", err, c) + } + c, err = engineCompute(eng, []int{0}, []string{ + "llama_model_load_from_file_impl: using device CUDA0 (Quadro RTX 6000) - 23800 MiB free", + "load_tensors: offloaded 49/49 layers to GPU"}) + if err != nil || c.GetBackend() != "CUDA" || c.GetLayersOffloaded() != 49 { + t.Errorf("on its GPU: %v %+v", err, c) + } + // A CPU shard is not refused for being on the CPU, and says so. + if c, err := engineCompute(eng, nil, []string{"load_tensors: offloaded 0/49 layers to GPU"}); err != nil || c.GetBackend() != "CPU" { + t.Errorf("a CPU shard: %v %+v", err, c) + } + // An engine that said nothing says nothing. + if c, err := engineCompute(eng, []int{0}, []string{"listening"}); err != nil || c != nil { + t.Errorf("nothing said: %v %+v", err, c) + } +} diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 10c25643..3e432024 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -450,7 +450,7 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { } w := tab() - fmt.Fprintln(w, "REPLICA\tMODEL\tSTATE\tPOOL\tNODES\tAGE\tIDLE\tENDPOINT") + fmt.Fprintln(w, "REPLICA\tMODEL\tSTATE\tPOOL\tON\tNODES\tAGE\tIDLE\tENDPOINT") for _, r := range reps { var nodes []string for _, s := range r.Shards { @@ -468,8 +468,14 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { if r.Restart != "" { state += " (restart " + r.Restart + ")" } - fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", - r.ID, r.Model, state, r.Pool, join(nodes), + // What the engine computes on, by its own account: the first + // shard's, which serves. + on := "-" + if len(r.Shards) > 0 { + on = r.Shards[0].Compute.Short() + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", + r.ID, r.Model, state, r.Pool, on, join(nodes), (time.Duration(r.AgeSec) * time.Second).String(), idleColumn(r.IdleSec), r.Endpoint) // A line rather than a column, because most replicas are not @@ -479,7 +485,7 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { if s.Fetch == nil { continue } - fmt.Fprintf(w, " %s: fetching %s\t\t\t\t\t\t\t\n", + fmt.Fprintf(w, " %s: fetching %s\t\t\t\t\t\t\t\t\n", s.Node, fetchColumn(s.Fetch)) } } diff --git a/internal/cli/tui_console.go b/internal/cli/tui_console.go index 0dda53b1..8164c253 100644 --- a/internal/cli/tui_console.go +++ b/internal/cli/tui_console.go @@ -1230,13 +1230,17 @@ func (m consoleModel) panel() (title, body, keys string) { for _, s := range r.Shards { nodes = append(nodes, fmt.Sprintf("%s%v", s.Node, discovery.DeviceLabels(s.GPUs))) } - rows = append(rows, []string{r.ID, r.Model, r.State, r.Pool, strings.Join(nodes, " "), + compute := "-" + if len(r.Shards) > 0 { + compute = r.Shards[0].Compute.Short() + } + rows = append(rows, []string{r.ID, r.Model, r.State, r.Pool, compute, strings.Join(nodes, " "), (time.Duration(r.AgeSec) * time.Second).Round(time.Second).String()}) } if m.inPanel { keys = rowKeys } - return "Replicas", "\n" + table([]string{"REPLICA", "MODEL", "STATE", "POOL", "ON", "AGE"}, rows, m.row, pw, m.inPanel), keys + return "Replicas", "\n" + table([]string{"REPLICA", "MODEL", "STATE", "POOL", "COMPUTE", "ON", "AGE"}, rows, m.row, pw, m.inPanel), keys case cUsage: head := "\n Last 24 hours, by " + lipgloss.NewStyle().Bold(true).Render(usageGroupings[m.usageBy]) + sDim.Render(" (→ to group by "+usageGroupings[(m.usageBy+1)%len(usageGroupings)]+")") + "\n\n" diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 554c5336..63b79b8f 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -518,6 +518,30 @@ type ShardView struct { // node's last heartbeat; zero when not measured. CPUCores float64 `json:"cpu_cores,omitempty"` MemoryBytes int64 `json:"memory_bytes,omitempty"` + // Compute is where the engine said it put the model as it became + // ready: absent when it did not say. + Compute *ComputeView `json:"compute,omitempty"` +} + +// ComputeView is where an engine runs its model, by its own account. +type ComputeView struct { + // Backend is CUDA, Vulkan, Metal, ROCm, GPU or CPU. + Backend string `json:"backend"` + Devices []string `json:"devices,omitempty"` + // LayersOffloaded of LayersTotal went to the devices; absent when unsaid. + LayersOffloaded int `json:"layers_offloaded,omitempty"` + LayersTotal int `json:"layers_total,omitempty"` +} + +// Short is the compute in a few characters: "CUDA 49/49", "CPU". +func (c *ComputeView) Short() string { + if c == nil { + return "-" + } + if c.LayersTotal > 0 && c.Backend != "CPU" { + return fmt.Sprintf("%s %d/%d", c.Backend, c.LayersOffloaded, c.LayersTotal) + } + return c.Backend } // FetchView is a download in flight. @@ -829,6 +853,10 @@ func (a *API) Replicas(model string) []ReplicaView { sv.Fetch.Percent = &p } } + if c := s.Compute; c != nil { + sv.Compute = &ComputeView{Backend: c.Backend, Devices: c.Devices, + LayersOffloaded: c.Offloaded, LayersTotal: c.Total} + } if l := loads[s.NodeID]; l != nil { if use, ok := l.Shards[s.ID]; ok { sv.CPUCores, sv.MemoryBytes = use.CPUCores, use.RSSBytes diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index 9fdf1f30..d477cb35 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -36,6 +36,8 @@ type StatusSink interface { // still starting, and treating each report as a state change would reset // the timers that decide when a slow start has become a failed one. OnFetchProgress(replicaID, shardID, source string, done, total int64) + // OnShardCompute records where a shard's engine said it runs the model. + OnShardCompute(replicaID, shardID string, c reconciler.Compute) Adopt(nodeID string, shards []reconciler.AdoptedShard) (unclaimed []reconciler.AdoptedShard) } @@ -449,6 +451,12 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) { } st := StateFromWire(m.Shard.GetState()) s.status.OnShardStatus(m.Shard.GetReplicaId(), m.Shard.GetShardId(), st, m.Shard.GetError()) + if c := m.Shard.GetCompute(); c != nil { + s.status.OnShardCompute(m.Shard.GetReplicaId(), m.Shard.GetShardId(), reconciler.Compute{ + Backend: c.GetBackend(), Devices: c.GetDevices(), + Offloaded: int(c.GetLayersOffloaded()), Total: int(c.GetLayersTotal()), + }) + } if m.Shard.GetState() == llmgwv1.ShardStatus_STATE_FAILED { s.log.Error("shard failed", "node", nodeID, "shard", m.Shard.GetShardId(), diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 2308a0ab..104f83c2 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -1246,6 +1246,22 @@ func (r *Reconciler) OnFetchProgress(replicaID, shardID, source string, done, to } } +// OnShardCompute records where a shard's engine said it put the model. +func (r *Reconciler) OnShardCompute(replicaID, shardID string, c Compute) { + r.mu.Lock() + defer r.mu.Unlock() + rep, ok := r.replicas[replicaID] + if !ok { + return + } + for i := range rep.Shards { + if rep.Shards[i].ID == shardID { + cc := c + rep.Shards[i].Compute = &cc + } + } +} + // OnShardStatus records a status report from an agent. func (r *Reconciler) OnShardStatus(replicaID, shardID string, st supervisor.State, errMsg string) { r.markShard(replicaID, shardID, st, errMsg) diff --git a/internal/control/reconciler/replica.go b/internal/control/reconciler/replica.go index ee4de60b..1f9a213e 100644 --- a/internal/control/reconciler/replica.go +++ b/internal/control/reconciler/replica.go @@ -72,6 +72,20 @@ type Shard struct { // shard reaches any other state, so a finished download does not sit in // the listing looking like one still running. Fetch *FetchProgress + + // Compute is where the engine said it put the model as it became ready: + // nil when it did not say. + Compute *Compute +} + +// Compute is where an engine runs its model, by its own account. +type Compute struct { + // Backend is CUDA, Vulkan, Metal, ROCm, GPU or CPU. + Backend string + // Devices it named: "CUDA0: Quadro RTX 6000 (7.5)". + Devices []string + // Offloaded of Total layers went to the devices; 0 and 0 when unsaid. + Offloaded, Total int } // FetchProgress is how far a shard's weights have got.