From 7bb360fc789875527ef1a0b7ab2bf468a22cdd9c Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Tue, 29 Sep 2026 21:29:23 +0200 Subject: [PATCH] Measure two nodes' fabric rail by rail: llmgw ctl fabric-test doctor checks a node alone, and every check can pass while two nodes still cannot reach each other. ctl fabric-test A B asks B's agent for an ib_write_bw server and A's for the client, adapter by adapter as NCCL pairs them, one rail at a time, and grades each against its link's rate. It goes through the agents' streams, so there is no SSH between nodes; the agent installs perftest the first time, and names the RoCE v2 GID itself. The result is read from perftest's table by its heading, not a column number. Agents now list the requests they answer beyond the usual (Register.features), and the control plane refuses a test on one too old to answer instead of waiting minutes for it. The forwarding dispatcher carries the new capability, and its test says so: a missing forward would have made the command quietly unavailable in production. Not yet run over real InfiniBand or RoCE; this cluster has neither. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 18 + api/gen/llmgw/v1/agent.pb.go | 373 +++++++++++++++--- api/proto/llmgw/v1/agent.proto | 40 ++ docs/fabrics.md | 35 +- docs/quick-install.md | 7 + internal/agent/agent.go | 75 ++++ internal/agent/agent_test.go | 32 ++ internal/agent/discovery/perftest.go | 133 +++++++ internal/agent/discovery/perftest_test.go | 49 +++ internal/cli/ctl.go | 1 + internal/cli/ctl_fabrictest.go | 75 ++++ internal/cli/dispatcherref_test.go | 8 + internal/cli/server.go | 14 + internal/control/adminapi/api.go | 1 + internal/control/adminapi/fabrictest.go | 71 ++++ internal/control/agentsvc/convert.go | 1 + internal/control/agentsvc/service.go | 47 +++ internal/control/control.go | 7 + internal/control/fabrictest/fabrictest.go | 235 +++++++++++ .../control/fabrictest/fabrictest_test.go | 100 +++++ internal/control/inventory/inventory.go | 5 +- 21 files changed, 1265 insertions(+), 62 deletions(-) create mode 100644 internal/agent/discovery/perftest.go create mode 100644 internal/agent/discovery/perftest_test.go create mode 100644 internal/cli/ctl_fabrictest.go create mode 100644 internal/control/adminapi/fabrictest.go create mode 100644 internal/control/fabrictest/fabrictest.go create mode 100644 internal/control/fabrictest/fabrictest_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e020a0..0acf0e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,24 @@ 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. +## Unreleased + +### `llmgw ctl fabric-test`: two nodes, rail by rail + +`llmgw doctor` checks a node on its own, and every check can pass while two +nodes still cannot reach each other: a cable into the wrong switch, a +partition key, a RoCE network with no route. `llmgw ctl fabric-test A B` runs +`ib_write_bw` from each of A's adapters on a fabric to the same adapter on B, +one rail at a time, through the agents — no SSH between nodes, and perftest is +installed the first time — and grades each against its link's rate: ok at 80%, +slow at 50%, fail below or when nothing got through. + +Agents now say which requests they answer beyond the usual (`features`), so +one too old for the test is refused at once instead of waited for. + +Written against perftest's real output and tested with it, but not yet run +over real InfiniBand or RoCE: this cluster has neither. + ## v0.10.0 — 2026-09-29 Independent InfiniBand fabrics and NVLink domains, told apart, drawn and diff --git a/api/gen/llmgw/v1/agent.pb.go b/api/gen/llmgw/v1/agent.pb.go index d8a2932..5171941 100644 --- a/api/gen/llmgw/v1/agent.pb.go +++ b/api/gen/llmgw/v1/agent.pb.go @@ -103,6 +103,7 @@ type AgentMessage struct { // *AgentMessage_Shard // *AgentMessage_Logs // *AgentMessage_FabricScan + // *AgentMessage_FabricTest Msg isAgentMessage_Msg `protobuf_oneof:"msg"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -190,6 +191,15 @@ func (x *AgentMessage) GetFabricScan() *FabricScanReply { return nil } +func (x *AgentMessage) GetFabricTest() *FabricTestReply { + if x != nil { + if x, ok := x.Msg.(*AgentMessage_FabricTest); ok { + return x.FabricTest + } + } + return nil +} + type isAgentMessage_Msg interface { isAgentMessage_Msg() } @@ -214,6 +224,10 @@ type AgentMessage_FabricScan struct { FabricScan *FabricScanReply `protobuf:"bytes,5,opt,name=fabric_scan,json=fabricScan,proto3,oneof"` } +type AgentMessage_FabricTest struct { + FabricTest *FabricTestReply `protobuf:"bytes,6,opt,name=fabric_test,json=fabricTest,proto3,oneof"` +} + func (*AgentMessage_Register) isAgentMessage_Msg() {} func (*AgentMessage_Heartbeat) isAgentMessage_Msg() {} @@ -224,6 +238,8 @@ func (*AgentMessage_Logs) isAgentMessage_Msg() {} func (*AgentMessage_FabricScan) isAgentMessage_Msg() {} +func (*AgentMessage_FabricTest) isAgentMessage_Msg() {} + // ControlMessage is anything the control plane tells a node. type ControlMessage struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -237,6 +253,7 @@ type ControlMessage struct { // *ControlMessage_Provision // *ControlMessage_Logs // *ControlMessage_FabricScan + // *ControlMessage_FabricTest Msg isControlMessage_Msg `protobuf_oneof:"msg"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -351,6 +368,15 @@ func (x *ControlMessage) GetFabricScan() *FabricScanRequest { return nil } +func (x *ControlMessage) GetFabricTest() *FabricTestRequest { + if x != nil { + if x, ok := x.Msg.(*ControlMessage_FabricTest); ok { + return x.FabricTest + } + } + return nil +} + type isControlMessage_Msg interface { isControlMessage_Msg() } @@ -387,6 +413,10 @@ type ControlMessage_FabricScan struct { FabricScan *FabricScanRequest `protobuf:"bytes,8,opt,name=fabric_scan,json=fabricScan,proto3,oneof"` } +type ControlMessage_FabricTest struct { + FabricTest *FabricTestRequest `protobuf:"bytes,9,opt,name=fabric_test,json=fabricTest,proto3,oneof"` +} + func (*ControlMessage_Start) isControlMessage_Msg() {} func (*ControlMessage_Stop) isControlMessage_Msg() {} @@ -403,6 +433,8 @@ func (*ControlMessage_Logs) isControlMessage_Msg() {} func (*ControlMessage_FabricScan) isControlMessage_Msg() {} +func (*ControlMessage_FabricTest) 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. @@ -478,8 +510,13 @@ type Register struct { // vLLM, it answered "unsupported", and every node of a cluster half way // through an upgrade showed as failing. InstallableEngines []string `protobuf:"bytes,16,rep,name=installable_engines,json=installableEngines,proto3" json:"installable_engines,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // features are the requests this agent answers beyond the ones every agent + // does: "fabric-test". An agent that does not know a request ignores it, so + // without this the control plane would wait minutes for an answer that is + // never coming. + Features []string `protobuf:"bytes,17,rep,name=features,proto3" json:"features,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Register) Reset() { @@ -624,6 +661,13 @@ func (x *Register) GetInstallableEngines() []string { return nil } +func (x *Register) GetFeatures() []string { + if x != nil { + return x.Features + } + return nil +} + // Host describes the operating system and processor of a node. type Host struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3223,19 +3267,203 @@ func (x *FabricScanReply) GetTookMs() int64 { return 0 } +// FabricTestRequest asks a node for one side of an RDMA bandwidth test +// between two nodes: ib_write_bw, from perftest, over one adapter. The +// server side is asked first and answers when the test is over; the client +// dials it. What it proves is what a replica spanning nodes needs and nothing +// on one node can show: that these two adapters reach each other through the +// fabric, and how fast. +type FabricTestRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + // role is "server" or "client". + Role string `protobuf:"bytes,2,opt,name=role,proto3" json:"role,omitempty"` + Hca string `protobuf:"bytes,3,opt,name=hca,proto3" json:"hca,omitempty"` + Port int32 `protobuf:"varint,4,opt,name=port,proto3" json:"port,omitempty"` + // peer is the server's management address, for the client. perftest + // exchanges its connection details over TCP before any RDMA. + Peer string `protobuf:"bytes,5,opt,name=peer,proto3" json:"peer,omitempty"` + // tcp_port is that exchange's port, the same on both sides. + TcpPort int32 `protobuf:"varint,6,opt,name=tcp_port,json=tcpPort,proto3" json:"tcp_port,omitempty"` + Seconds int32 `protobuf:"varint,7,opt,name=seconds,proto3" json:"seconds,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FabricTestRequest) Reset() { + *x = FabricTestRequest{} + mi := &file_llmgw_v1_agent_proto_msgTypes[31] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FabricTestRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FabricTestRequest) ProtoMessage() {} + +func (x *FabricTestRequest) ProtoReflect() protoreflect.Message { + mi := &file_llmgw_v1_agent_proto_msgTypes[31] + 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 FabricTestRequest.ProtoReflect.Descriptor instead. +func (*FabricTestRequest) Descriptor() ([]byte, []int) { + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{31} +} + +func (x *FabricTestRequest) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + +func (x *FabricTestRequest) GetRole() string { + if x != nil { + return x.Role + } + return "" +} + +func (x *FabricTestRequest) GetHca() string { + if x != nil { + return x.Hca + } + return "" +} + +func (x *FabricTestRequest) GetPort() int32 { + if x != nil { + return x.Port + } + return 0 +} + +func (x *FabricTestRequest) GetPeer() string { + if x != nil { + return x.Peer + } + return "" +} + +func (x *FabricTestRequest) GetTcpPort() int32 { + if x != nil { + return x.TcpPort + } + return 0 +} + +func (x *FabricTestRequest) GetSeconds() int32 { + if x != nil { + return x.Seconds + } + return 0 +} + +// FabricTestReply is what one side measured. +type FabricTestReply struct { + state protoimpl.MessageState `protogen:"open.v1"` + RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` + // gbps is the average bandwidth, in gigabits a second. + Gbps float64 `protobuf:"fixed64,2,opt,name=gbps,proto3" json:"gbps,omitempty"` + // output is the end of the tool's output, for when the number is not + // enough. + Output string `protobuf:"bytes,3,opt,name=output,proto3" json:"output,omitempty"` + Error string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"` + TookMs int64 `protobuf:"varint,5,opt,name=took_ms,json=tookMs,proto3" json:"took_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FabricTestReply) Reset() { + *x = FabricTestReply{} + mi := &file_llmgw_v1_agent_proto_msgTypes[32] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FabricTestReply) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FabricTestReply) ProtoMessage() {} + +func (x *FabricTestReply) ProtoReflect() protoreflect.Message { + mi := &file_llmgw_v1_agent_proto_msgTypes[32] + 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 FabricTestReply.ProtoReflect.Descriptor instead. +func (*FabricTestReply) Descriptor() ([]byte, []int) { + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{32} +} + +func (x *FabricTestReply) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + +func (x *FabricTestReply) GetGbps() float64 { + if x != nil { + return x.Gbps + } + return 0 +} + +func (x *FabricTestReply) GetOutput() string { + if x != nil { + return x.Output + } + return "" +} + +func (x *FabricTestReply) GetError() string { + if x != nil { + return x.Error + } + return "" +} + +func (x *FabricTestReply) GetTookMs() int64 { + if x != nil { + return x.TookMs + } + return 0 +} + var File_llmgw_v1_agent_proto protoreflect.FileDescriptor const file_llmgw_v1_agent_proto_rawDesc = "" + "\n" + - "\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\x94\x02\n" + + "\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xd2\x02\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\x05shard\x12)\n" + "\x04logs\x18\x04 \x01(\v2\x13.llmgw.v1.LogsReplyH\x00R\x04logs\x12<\n" + "\vfabric_scan\x18\x05 \x01(\v2\x19.llmgw.v1.FabricScanReplyH\x00R\n" + - "fabricScanB\x05\n" + - "\x03msg\"\xa5\x03\n" + + "fabricScan\x12<\n" + + "\vfabric_test\x18\x06 \x01(\v2\x19.llmgw.v1.FabricTestReplyH\x00R\n" + + "fabricTestB\x05\n" + + "\x03msg\"\xe5\x03\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" + @@ -3245,8 +3473,10 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovision\x12+\n" + "\x04logs\x18\a \x01(\v2\x15.llmgw.v1.LogsRequestH\x00R\x04logs\x12>\n" + "\vfabric_scan\x18\b \x01(\v2\x1b.llmgw.v1.FabricScanRequestH\x00R\n" + - "fabricScanB\x05\n" + - "\x03msg\"\xc7\x06\n" + + "fabricScan\x12>\n" + + "\vfabric_test\x18\t \x01(\v2\x1b.llmgw.v1.FabricTestRequestH\x00R\n" + + "fabricTestB\x05\n" + + "\x03msg\"\xe3\x06\n" + "\bRegister\x12\x17\n" + "\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" + "\bhostname\x18\x02 \x01(\tR\bhostname\x12\x1b\n" + @@ -3264,7 +3494,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "\x04host\x18\r \x01(\v2\x0e.llmgw.v1.HostR\x04host\x12%\n" + "\x0eengines_probed\x18\x0e \x01(\bR\renginesProbed\x12\"\n" + "\x04imex\x18\x0f \x01(\v2\x0e.llmgw.v1.IMEXR\x04imex\x12/\n" + - "\x13installable_engines\x18\x10 \x03(\tR\x12installableEngines\x1a9\n" + + "\x13installable_engines\x18\x10 \x03(\tR\x12installableEngines\x12\x1a\n" + + "\bfeatures\x18\x11 \x03(\tR\bfeatures\x1a9\n" + "\vLabelsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1aA\n" + @@ -3536,7 +3767,23 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "request_id\x18\x01 \x01(\tR\trequestId\x12\x16\n" + "\x06output\x18\x02 \x01(\tR\x06output\x12\x14\n" + "\x05error\x18\x03 \x01(\tR\x05error\x12\x17\n" + - "\atook_ms\x18\x04 \x01(\x03R\x06tookMs2O\n" + + "\atook_ms\x18\x04 \x01(\x03R\x06tookMs\"\xb5\x01\n" + + "\x11FabricTestRequest\x12\x1d\n" + + "\n" + + "request_id\x18\x01 \x01(\tR\trequestId\x12\x12\n" + + "\x04role\x18\x02 \x01(\tR\x04role\x12\x10\n" + + "\x03hca\x18\x03 \x01(\tR\x03hca\x12\x12\n" + + "\x04port\x18\x04 \x01(\x05R\x04port\x12\x12\n" + + "\x04peer\x18\x05 \x01(\tR\x04peer\x12\x19\n" + + "\btcp_port\x18\x06 \x01(\x05R\atcpPort\x12\x18\n" + + "\aseconds\x18\a \x01(\x05R\aseconds\"\x8b\x01\n" + + "\x0fFabricTestReply\x12\x1d\n" + + "\n" + + "request_id\x18\x01 \x01(\tR\trequestId\x12\x12\n" + + "\x04gbps\x18\x02 \x01(\x01R\x04gbps\x12\x16\n" + + "\x06output\x18\x03 \x01(\tR\x06output\x12\x14\n" + + "\x05error\x18\x04 \x01(\tR\x05error\x12\x17\n" + + "\atook_ms\x18\x05 \x01(\x03R\x06tookMs2O\n" + "\fAgentService\x12?\n" + "\aConnect\x12\x16.llmgw.v1.AgentMessage\x1a\x18.llmgw.v1.ControlMessage(\x010\x01B\x93\x01\n" + "\fcom.llmgw.v1B\n" + @@ -3555,7 +3802,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, 37) +var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 39) var file_llmgw_v1_agent_proto_goTypes = []any{ (ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State (*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage @@ -3589,12 +3836,14 @@ var file_llmgw_v1_agent_proto_goTypes = []any{ (*LogsReply)(nil), // 29: llmgw.v1.LogsReply (*FabricScanRequest)(nil), // 30: llmgw.v1.FabricScanRequest (*FabricScanReply)(nil), // 31: llmgw.v1.FabricScanReply - nil, // 32: llmgw.v1.Register.LabelsEntry - nil, // 33: llmgw.v1.Register.EngineVersionsEntry - nil, // 34: llmgw.v1.GPU.NicPathsEntry - nil, // 35: llmgw.v1.IBPort.CountersEntry - nil, // 36: llmgw.v1.Heartbeat.EngineVersionsEntry - nil, // 37: llmgw.v1.EngineProfile.EnvEntry + (*FabricTestRequest)(nil), // 32: llmgw.v1.FabricTestRequest + (*FabricTestReply)(nil), // 33: llmgw.v1.FabricTestReply + nil, // 34: llmgw.v1.Register.LabelsEntry + nil, // 35: llmgw.v1.Register.EngineVersionsEntry + nil, // 36: llmgw.v1.GPU.NicPathsEntry + nil, // 37: llmgw.v1.IBPort.CountersEntry + nil, // 38: llmgw.v1.Heartbeat.EngineVersionsEntry + nil, // 39: llmgw.v1.EngineProfile.EnvEntry } var file_llmgw_v1_agent_proto_depIdxs = []int32{ 3, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register @@ -3602,49 +3851,51 @@ var file_llmgw_v1_agent_proto_depIdxs = []int32{ 20, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus 29, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply 31, // 4: llmgw.v1.AgentMessage.fabric_scan:type_name -> llmgw.v1.FabricScanReply - 22, // 5: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard - 25, // 6: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard - 26, // 7: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard - 27, // 8: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack - 24, // 9: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights - 17, // 10: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision - 28, // 11: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest - 30, // 12: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest - 32, // 13: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry - 7, // 14: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU - 8, // 15: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment - 33, // 16: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry - 6, // 17: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard - 9, // 18: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights - 5, // 19: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage - 4, // 20: llmgw.v1.Register.host:type_name -> llmgw.v1.Host - 14, // 21: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX - 0, // 22: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State - 34, // 23: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry - 11, // 24: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad - 35, // 25: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry - 13, // 26: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager - 5, // 27: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage - 9, // 28: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights - 36, // 29: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry - 19, // 30: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus - 10, // 31: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad - 12, // 32: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort - 8, // 33: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment - 15, // 34: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState - 14, // 35: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX - 18, // 36: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant - 0, // 37: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State - 21, // 38: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress - 23, // 39: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile - 37, // 40: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry - 1, // 41: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage - 2, // 42: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage - 42, // [42:43] is the sub-list for method output_type - 41, // [41:42] is the sub-list for method input_type - 41, // [41:41] is the sub-list for extension type_name - 41, // [41:41] is the sub-list for extension extendee - 0, // [0:41] is the sub-list for field type_name + 33, // 5: llmgw.v1.AgentMessage.fabric_test:type_name -> llmgw.v1.FabricTestReply + 22, // 6: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard + 25, // 7: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard + 26, // 8: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard + 27, // 9: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack + 24, // 10: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights + 17, // 11: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision + 28, // 12: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest + 30, // 13: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest + 32, // 14: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest + 34, // 15: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry + 7, // 16: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU + 8, // 17: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment + 35, // 18: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry + 6, // 19: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard + 9, // 20: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights + 5, // 21: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage + 4, // 22: llmgw.v1.Register.host:type_name -> llmgw.v1.Host + 14, // 23: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX + 0, // 24: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State + 36, // 25: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry + 11, // 26: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad + 37, // 27: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry + 13, // 28: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager + 5, // 29: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage + 9, // 30: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights + 38, // 31: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry + 19, // 32: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus + 10, // 33: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad + 12, // 34: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort + 8, // 35: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment + 15, // 36: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState + 14, // 37: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX + 18, // 38: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant + 0, // 39: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State + 21, // 40: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress + 23, // 41: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile + 39, // 42: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry + 1, // 43: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage + 2, // 44: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage + 44, // [44:45] is the sub-list for method output_type + 43, // [43:44] is the sub-list for method input_type + 43, // [43:43] is the sub-list for extension type_name + 43, // [43:43] is the sub-list for extension extendee + 0, // [0:43] is the sub-list for field type_name } func init() { file_llmgw_v1_agent_proto_init() } @@ -3658,6 +3909,7 @@ func file_llmgw_v1_agent_proto_init() { (*AgentMessage_Shard)(nil), (*AgentMessage_Logs)(nil), (*AgentMessage_FabricScan)(nil), + (*AgentMessage_FabricTest)(nil), } file_llmgw_v1_agent_proto_msgTypes[1].OneofWrappers = []any{ (*ControlMessage_Start)(nil), @@ -3668,6 +3920,7 @@ func file_llmgw_v1_agent_proto_init() { (*ControlMessage_Provision)(nil), (*ControlMessage_Logs)(nil), (*ControlMessage_FabricScan)(nil), + (*ControlMessage_FabricTest)(nil), } type x struct{} out := protoimpl.TypeBuilder{ @@ -3675,7 +3928,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: 37, + NumMessages: 39, NumExtensions: 0, NumServices: 1, }, diff --git a/api/proto/llmgw/v1/agent.proto b/api/proto/llmgw/v1/agent.proto index 29bea3b..9894a2a 100644 --- a/api/proto/llmgw/v1/agent.proto +++ b/api/proto/llmgw/v1/agent.proto @@ -25,6 +25,7 @@ message AgentMessage { ShardStatus shard = 3; LogsReply logs = 4; FabricScanReply fabric_scan = 5; + FabricTestReply fabric_test = 6; } } @@ -39,6 +40,7 @@ message ControlMessage { Provision provision = 6; LogsRequest logs = 7; FabricScanRequest fabric_scan = 8; + FabricTestRequest fabric_test = 9; } } @@ -125,6 +127,12 @@ message Register { // vLLM, it answered "unsupported", and every node of a cluster half way // through an upgrade showed as failing. repeated string installable_engines = 16; + + // features are the requests this agent answers beyond the ones every agent + // does: "fabric-test". An agent that does not know a request ignores it, so + // without this the control plane would wait minutes for an answer that is + // never coming. + repeated string features = 17; } // Host describes the operating system and processor of a node. @@ -605,3 +613,35 @@ message FabricScanReply { string error = 3; int64 took_ms = 4; } + +// FabricTestRequest asks a node for one side of an RDMA bandwidth test +// between two nodes: ib_write_bw, from perftest, over one adapter. The +// server side is asked first and answers when the test is over; the client +// dials it. What it proves is what a replica spanning nodes needs and nothing +// on one node can show: that these two adapters reach each other through the +// fabric, and how fast. +message FabricTestRequest { + string request_id = 1; + // role is "server" or "client". + string role = 2; + string hca = 3; + int32 port = 4; + // peer is the server's management address, for the client. perftest + // exchanges its connection details over TCP before any RDMA. + string peer = 5; + // tcp_port is that exchange's port, the same on both sides. + int32 tcp_port = 6; + int32 seconds = 7; +} + +// FabricTestReply is what one side measured. +message FabricTestReply { + string request_id = 1; + // gbps is the average bandwidth, in gigabits a second. + double gbps = 2; + // output is the end of the tool's output, for when the number is not + // enough. + string output = 3; + string error = 4; + int64 took_ms = 5; +} diff --git a/docs/fabrics.md b/docs/fabrics.md index a1ea392..20d178a 100644 --- a/docs/fabrics.md +++ b/docs/fabrics.md @@ -121,6 +121,39 @@ rate and errors, and the NCCL environment a shard on each fabric would be given. `--json` writes every fact it read as well as what it concluded — send that file when the report is not enough. +## Checking two nodes: `llmgw ctl fabric-test` + +Every check on one node can pass while two nodes still cannot reach each other +— a cable into the wrong switch, a partition key, a RoCE network with no route +— and a replica spanning them then hangs on its first all-reduce. Before +placing one, on the control plane: + +``` +llmgw ctl fabric-test gpu01 gpu02 +``` + +``` +gpu01 → gpu02 on ib-compute, 41s + +RAIL LINK MEASURED VERDICT DETAIL +mlx5_0 → mlx5_0 400 Gb/s 388.2 Gb/s ok 97% of 400 Gb/s +mlx5_1 → mlx5_1 400 Gb/s 241.0 Gb/s slow 60% of 400 Gb/s: check the cable, and the card's PCIe width and NUMA node +mlx5_2 → mlx5_2 400 Gb/s 0.0 Gb/s fail gpu01: ib_write_bw: Couldn't connect to 10.1.0.22:18517 +``` + +It runs `ib_write_bw` (perftest, installed on a node the first time) from each +of the first node's adapters on the fabric to the same adapter on the second — +mlx5_0 to mlx5_0, the way NCCL pairs them — one rail at a time, through the +agents, so no SSH between nodes is needed. Each rail is graded against its +link's rate: **ok** at 80% or more, **slow** at 50%, **fail** below that or when +nothing got through. `--fabric` picks one when the nodes share several, +`--seconds` the length of each rail's run. The traffic is real: run it when the +fabric is quiet. + +It measures RDMA between adapters, not GPUDirect or NCCL itself: a pass says +the fabric carries traffic between these two nodes at its rate, which is what +a first all-reduce needs and what nothing on one node can show. + ## Watching them - **Cluster page, Fabrics.** Each declared fabric and what it physically is; @@ -172,6 +205,6 @@ workers headless. `launch_mode: ray` is there for a vLLM too old for that. | It | Look at | |---|---| | is not placed | `llmgw ctl explain-placement `: a fabric split by identity, a node whose IMEX is down, a node refused because its adapters are cabled to two fabrics | -| starts and hangs on its first collective | the cluster page's fabric problems; `doctor` on each node — a port down, the wrong interface, RoCE without a v2 GID | +| starts and hangs on its first collective | `llmgw ctl fabric-test` between two of its nodes; the cluster page's fabric problems; `doctor` on each node — a port down, the wrong interface, RoCE without a v2 GID | | runs, slowly | `nvidia_peermem`; the map for a degraded cable; the node page's NVLink matrix for a group that crossed PCIe | | worked yesterday | the subnet manager table for a failover; `link_downed` on the node's ports | diff --git a/docs/quick-install.md b/docs/quick-install.md index 267cf50..e16f2fb 100644 --- a/docs/quick-install.md +++ b/docs/quick-install.md @@ -89,6 +89,13 @@ the same fabric the same name. In a script, give it instead: sudo llmgw node add ubuntu@10.0.0.21 --fabric ib-a:mlx5_0,mlx5_1,mlx5_2,mlx5_3 ``` +Once two nodes on a fabric are in, check they reach each other at the link's +rate before placing anything across them: + +``` +llmgw ctl fabric-test gpu01 gpu02 +``` + **If it cannot log in.** `sudo` runs SSH as root, with root's keys. To use your own keys instead, log in to the control plane with `ssh -A` and keep your agent through sudo: diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 5dc6342..5282735 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -844,6 +844,7 @@ func (a *Agent) session(ctx context.Context, server string) error { reg.EngineVersions = a.engines(parent) reg.EnginesProbed = true reg.InstallableEngines = provision.Installable() + reg.Features = []string{FeatureFabricTest} reg.RunningShards = a.runningShards() a.mu.Lock() h := a.prov.Host @@ -1145,6 +1146,16 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha case <-ctx.Done(): } }(m.FabricScan) + case *llmgwv1.ControlMessage_FabricTest: + // Seconds of traffic, and a package install the first time: off the + // receive loop too. + go func(req *llmgwv1.FabricTestRequest) { + reply := a.fabricTest(ctx, req) + select { + case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_FabricTest{FabricTest: reply}}: + case <-ctx.Done(): + } + }(m.FabricTest) case *llmgwv1.ControlMessage_Ack: if !m.Ack.GetOk() { a.log.Warn("control plane rejected a message", @@ -1758,6 +1769,70 @@ func (a *Agent) ensureDiags(ctx context.Context) { } } +// FeatureFabricTest is the agent answering FabricTestRequest. +const FeatureFabricTest = "fabric-test" + +// runWriteBW is discovery.RunWriteBW, and haveWriteBW whether it is +// installed; both replaced in tests. +var ( + runWriteBW = discovery.RunWriteBW + haveWriteBW = func() bool { _, err := exec.LookPath("ib_write_bw"); return err == nil } +) + +// fabricTest runs this node's side of a bandwidth test with another. +func (a *Agent) fabricTest(ctx context.Context, req *llmgwv1.FabricTestRequest) *llmgwv1.FabricTestReply { + start := time.Now() + reply := &llmgwv1.FabricTestReply{RequestId: req.GetRequestId()} + defer func() { reply.TookMs = time.Since(start).Milliseconds() }() + + w := discovery.WriteBW{HCA: req.GetHca(), Port: int(req.GetPort()), GIDIndex: -1, + TCPPort: int(req.GetTcpPort()), Seconds: int(req.GetSeconds())} + if req.GetRole() == "client" { + w.Peer = req.GetPeer() + } + a.mu.Lock() + found := false + for _, h := range a.inv.HCAs { + for _, p := range h.Ports { + if h.Name == w.HCA && p.Num == w.Port { + found = true + // RoCE needs its v2 GID named, as NCCL does; InfiniBand uses the LID. + if p.LinkLayer == discovery.LinkEthernet && p.RoCEv2GID >= 0 { + w.GIDIndex = p.RoCEv2GID + } + } + } + } + a.mu.Unlock() + if !found { + reply.Error = fmt.Sprintf("this node has no port %s/%d", w.HCA, w.Port) + return reply + } + if !haveWriteBW() && a.prov != nil { + a.log.Info("installing perftest, for a bandwidth test with another node") + pkgs := map[string][]string{"deb": {"perftest"}, "rpm": {"perftest"}} + if err := a.prov.InstallPackages(ctx, pkgs); err != nil { + reply.Error = "installing perftest: " + err.Error() + return reply + } + } + gbps, out, err := runWriteBW(ctx, w) + reply.Gbps = gbps + reply.Output = tailLines(out, 12) + if err != nil { + reply.Error = err.Error() + } + return reply +} + +func tailLines(s string, n int) string { + lines := strings.Split(strings.TrimRight(s, "\n"), "\n") + if len(lines) > n { + lines = lines[len(lines)-n:] + } + return strings.Join(lines, "\n") +} + // discoverGPUFabric and discoverIMEX are replaced in tests. var ( discoverGPUFabric = discovery.DiscoverGPUFabric diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 07309bc..a3dc052 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -23,6 +23,7 @@ import ( "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" + 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" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" @@ -881,3 +882,34 @@ func TestAFabricScanIsAskedForAndAnswered(t *testing.T) { t.Errorf("asked %q, got %+v", asked, r) } } + +func TestABandwidthTestIsRunOnTheAdapterAskedFor(t *testing.T) { + oldRun, oldHave := runWriteBW, haveWriteBW + t.Cleanup(func() { runWriteBW, haveWriteBW = oldRun, oldHave }) + haveWriteBW = func() bool { return true } + var got discovery.WriteBW + runWriteBW = func(_ context.Context, w discovery.WriteBW) (float64, string, error) { + got = w + return 388.2, " #bytes #iterations BW peak[Gb/sec] BW average[Gb/sec]\n", nil + } + h := start(t, func(c *Config) { c.Discovery.IBRoot = ibTreeWithActivePorts(t, "mlx5_0") }) + + r, err := h.svc.FabricTest(t.Context(), "n1", &llmgwv1.FabricTestRequest{ + Role: "client", Hca: "mlx5_0", Port: 1, Peer: "10.0.0.2", TcpPort: 18515, Seconds: 3}) + if err != nil { + t.Fatal(err) + } + if r.GetGbps() != 388.2 || r.GetError() != "" { + t.Errorf("reply %+v", r) + } + if got.HCA != "mlx5_0" || got.Peer != "10.0.0.2" || got.TCPPort != 18515 || got.GIDIndex != -1 { + t.Errorf("ran %+v", got) + } + + // An adapter the node does not have is said so, without running anything. + got = discovery.WriteBW{} + r, _ = h.svc.FabricTest(t.Context(), "n1", &llmgwv1.FabricTestRequest{Role: "server", Hca: "mlx5_9", Port: 1, Seconds: 1}) + if !strings.Contains(r.GetError(), "no port mlx5_9/1") || got.HCA != "" { + t.Errorf("reply %+v, ran %+v", r, got) + } +} diff --git a/internal/agent/discovery/perftest.go b/internal/agent/discovery/perftest.go new file mode 100644 index 0000000..d2e2983 --- /dev/null +++ b/internal/agent/discovery/perftest.go @@ -0,0 +1,133 @@ +package discovery + +import ( + "context" + "errors" + "fmt" + "os/exec" + "regexp" + "strconv" + "strings" + "time" +) + +// WriteBW is one side of an ib_write_bw run, from perftest: an RDMA write +// bandwidth test between two adapters, which is what a collective spanning +// nodes rides on. +type WriteBW struct { + HCA string + Port int + // GIDIndex is the RoCE v2 GID to use; negative on InfiniBand, where the + // port's LID is used. + GIDIndex int + // Peer is the server's address, for the client; empty makes the server. + Peer string + TCPPort int + Seconds int +} + +// Args are the command line: a duration rather than a count, so the test +// takes as long on a slow link as a fast one, and results in Gb/s. +func (w WriteBW) Args() []string { + args := []string{"-d", w.HCA, "-i", strconv.Itoa(max(w.Port, 1)), + "-p", strconv.Itoa(w.TCPPort), "-D", strconv.Itoa(max(w.Seconds, 1)), + "-F", "--report_gbits"} + if w.GIDIndex >= 0 { + args = append(args, "-x", strconv.Itoa(w.GIDIndex)) + } + if w.Peer != "" { + args = append(args, w.Peer) + } + return args +} + +// RunWriteBW runs one side and returns the average bandwidth the output +// reports, in Gb/s, and the output. The client retries its connection for a +// while: the server is started first, but a few seconds may pass before it +// listens. +func RunWriteBW(ctx context.Context, w WriteBW) (float64, string, error) { + bin, err := exec.LookPath("ib_write_bw") + if err != nil { + return 0, "", errors.New("ib_write_bw is not installed (the perftest package)") + } + limit := time.Duration(w.Seconds+60) * time.Second + var out []byte + for attempt := 0; ; attempt++ { + cctx, cancel := context.WithTimeout(ctx, limit) + out, err = exec.CommandContext(cctx, bin, w.Args()...).CombinedOutput() //nolint:gosec // fixed tool + cancel() + if err == nil || w.Peer == "" || !refused(string(out)) || attempt >= 10 { + break + } + select { + case <-ctx.Done(): + return 0, string(out), ctx.Err() + case <-time.After(time.Second): + } + } + text := string(out) + if err != nil { + return 0, text, fmt.Errorf("ib_write_bw: %v: %s", err, lastNonEmpty(text)) + } + gbps, perr := ParseWriteBW(text) + if perr != nil && w.Peer != "" { + // Only the client is sure to print the result; a server that does not + // is not a failure. + return 0, text, perr + } + return gbps, text, nil +} + +func refused(out string) bool { + for _, s := range []string{"Couldn't connect", "Unable to init the socket connection", "Connection refused"} { + if strings.Contains(out, s) { + return true + } + } + return false +} + +var columnSplit = regexp.MustCompile(`\s{2,}`) + +// ParseWriteBW finds the average bandwidth in perftest's result table: +// +// #bytes #iterations BW peak[Gb/sec] BW average[Gb/sec] MsgRate[Mpps] +// 65536 887036 0.00 96.93 0.184877 +// +// by the heading's position rather than a fixed column, since versions add +// columns and the peak reads 0.00 when the test runs for a duration. +func ParseWriteBW(out string) (float64, error) { + lines := strings.Split(out, "\n") + for i, l := range lines { + if !strings.Contains(l, "BW average") { + continue + } + heads := columnSplit.Split(strings.TrimSpace(l), -1) + col := -1 + for j, h := range heads { + if strings.HasPrefix(h, "BW average") { + col = j + } + } + for _, d := range lines[i+1:] { + f := strings.Fields(d) + if len(f) == 0 || strings.HasPrefix(f[0], "-") { + continue + } + if len(f) != len(heads) || col < 0 { + return 0, fmt.Errorf("perftest's result table has a shape this does not know: %q / %q", l, d) + } + v, err := strconv.ParseFloat(f[col], 64) + if err != nil { + return 0, fmt.Errorf("reading the average bandwidth %q: %w", f[col], err) + } + return v, nil + } + } + return 0, errors.New("perftest printed no result") +} + +func lastNonEmpty(s string) string { + lines := strings.Split(strings.TrimSpace(s), "\n") + return strings.TrimSpace(lines[len(lines)-1]) +} diff --git a/internal/agent/discovery/perftest_test.go b/internal/agent/discovery/perftest_test.go new file mode 100644 index 0000000..4298e99 --- /dev/null +++ b/internal/agent/discovery/perftest_test.go @@ -0,0 +1,49 @@ +package discovery + +import ( + "slices" + "testing" +) + +// The client's output of perftest 24.x, with a duration. +const writeBWClient = `--------------------------------------------------------------------------------------- + RDMA_Write BW Test + Dual-port : OFF Device : mlx5_0 + Number of qps : 1 Transport type : IB + Connection type : RC Using SRQ : OFF + TX depth : 128 + CQ Moderation : 100 + Mtu : 4096[B] + Link type : IB + Max inline data : 0[B] + rdma_cm QPs : OFF + Data ex. method : Ethernet +--------------------------------------------------------------------------------------- + local address: LID 0x05 QPN 0x0112 PSN 0x7b1f02 RKey 0x1fffbd VAddr 0x007f1d2a5f0000 + remote address: LID 0x09 QPN 0x0113 PSN 0x2c1e44 RKey 0x1ffeaf VAddr 0x007fa4e1a60000 +--------------------------------------------------------------------------------------- + #bytes #iterations BW peak[Gb/sec] BW average[Gb/sec] MsgRate[Mpps] + 65536 1851214 0.00 388.21 0.740422 +--------------------------------------------------------------------------------------- +` + +func TestTheAverageBandwidthIsReadFromPerftest(t *testing.T) { + got, err := ParseWriteBW(writeBWClient) + if err != nil || got != 388.21 { + t.Errorf("got %v, %v", got, err) + } + if _, err := ParseWriteBW(" Couldn't connect to 10.0.0.2:18515\n"); err == nil { + t.Error("output with no result table gave a number") + } +} + +func TestWriteBWArguments(t *testing.T) { + ib := WriteBW{HCA: "mlx5_0", Port: 1, GIDIndex: -1, TCPPort: 18515, Seconds: 3}.Args() + if slices.Contains(ib, "-x") { + t.Errorf("InfiniBand was given a GID index: %v", ib) + } + roce := WriteBW{HCA: "mlx5_0", Port: 1, GIDIndex: 3, Peer: "10.0.0.2", TCPPort: 18515, Seconds: 3}.Args() + if !slices.Contains(roce, "-x") || roce[len(roce)-1] != "10.0.0.2" { + t.Errorf("RoCE client: %v", roce) + } +} diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 4f14fd3..4ebecea 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -120,6 +120,7 @@ func newCtlCmd() *cobra.Command { newModelsCmd(client, &asJSON), newReplicasCmd(client, &asJSON), newExplainCmd(client, &asJSON), + newFabricTestCmd(client, &asJSON), newNodeStateCmd(client, "cordon", "cordoned", "Stop placing new work on a node; what runs there keeps running"), newNodeStateCmd(client, "uncordon", "ready", diff --git a/internal/cli/ctl_fabrictest.go b/internal/cli/ctl_fabrictest.go new file mode 100644 index 0000000..6336960 --- /dev/null +++ b/internal/cli/ctl_fabrictest.go @@ -0,0 +1,75 @@ +package cli + +import ( + "encoding/json" + "fmt" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" +) + +func newFabricTestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + var ( + fabric string + seconds int + ) + c := &cobra.Command{ + Use: "fabric-test NODE-A NODE-B", + Short: "Measure RDMA bandwidth between two nodes, rail by rail", + Long: `Runs ib_write_bw from each of NODE-A's adapters on a fabric to the same +adapter on NODE-B — mlx5_0 to mlx5_0, and so on, the way NCCL pairs them — +one rail at a time, and grades each against its link's rate: ok at 80% or +more, slow at 50%, fail below or when nothing got through. + +Every check on one node can pass while two nodes still cannot reach each +other: a cable into the wrong switch, a partition key, a RoCE network with no +route. A replica spanning them then hangs on its first all-reduce. Run this +before placing one. + +It goes through the agents, so it needs no ssh between nodes. perftest is +installed on a node the first time it is asked. The traffic is real and +takes a few seconds a rail; run it when the fabric is quiet.`, + Example: ` llmgw ctl fabric-test gpu01 gpu02 + llmgw ctl fabric-test gpu01 gpu02 --fabric ib-compute --seconds 10`, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + body, _ := json.Marshal(map[string]any{"a": args[0], "b": args[1], "fabric": fabric, "seconds": seconds}) + cl := client() + // Rails one after another, each for its seconds and a settle; the + // client's usual timeout is for questions, not measurements. + cl.http.Timeout = 30 * time.Minute + var v adminapi.FabricTestView + if err := cl.post(cmd.Context(), "/api/v1/fabric-test", body, &v); err != nil { + return err + } + if *asJSON { + return json.NewEncoder(cmd.OutOrStdout()).Encode(v) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "%s → %s on %s, %s\n\n", v.A, v.B, v.Fabric, v.Took) + w := tabwriter.NewWriter(out, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "RAIL\tLINK\tMEASURED\tVERDICT\tDETAIL") + for _, r := range v.Rails { + link := "-" + if r.RateGbps > 0 { + link = fmt.Sprintf("%d Gb/s", r.RateGbps) + } + fmt.Fprintf(w, "%s → %s\t%s\t%.1f Gb/s\t%s\t%s\n", r.A, r.B, link, r.Gbps, r.Verdict, r.Detail) + } + _ = w.Flush() + for _, n := range v.Notes { + fmt.Fprintf(out, "\nnote: %s\n", n) + } + if !v.OK { + return fmt.Errorf("not every rail passed") + } + return nil + }, + } + c.Flags().StringVar(&fabric, "fabric", "", "the fabric to test (default: the one the two nodes share)") + c.Flags().IntVar(&seconds, "seconds", 3, "how long each rail runs, up to 30") + return c +} diff --git a/internal/cli/dispatcherref_test.go b/internal/cli/dispatcherref_test.go index e9413d1..b9fbd72 100644 --- a/internal/cli/dispatcherref_test.go +++ b/internal/cli/dispatcherref_test.go @@ -5,6 +5,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" ) func TestTheForwardingDispatcherCarriesEveryCapability(t *testing.T) { @@ -23,10 +24,17 @@ func TestTheForwardingDispatcherCarriesEveryCapability(t *testing.T) { t.Error("dispatcherRef drops WeightRemover, so the control plane cannot reclaim disk") } + if _, ok := ref.(fabrictest.Asker); !ok { + t.Error("dispatcherRef drops the bandwidth test, so ctl fabric-test cannot reach the agents") + } + // And the thing it forwards to must have it, or the forwarding has nothing // to reach. var svc any = &agentsvc.Service{} if _, ok := svc.(control.WeightRemover); !ok { t.Error("the agent service does not implement WeightRemover") } + if _, ok := svc.(fabrictest.Asker); !ok { + t.Error("the agent service does not implement the bandwidth test") + } } diff --git a/internal/cli/server.go b/internal/cli/server.go index cd62ad6..82a0f3a 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -21,11 +21,13 @@ import ( "golang.org/x/net/http2/h2c" "golang.org/x/sync/errgroup" + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" "gitea.mynodes.xyz/ilker/llmgw/internal/config" "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" @@ -493,6 +495,18 @@ func (r *dispatcherRef) Logs(ctx context.Context, nodeID, source string, lines i return lr.Logs(ctx, nodeID, source, lines) } +func (r *dispatcherRef) FabricTest(ctx context.Context, nodeID string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) { + d, err := r.get() + if err != nil { + return nil, err + } + ft, ok := d.(fabrictest.Asker) + if !ok { + return nil, errors.New("this dispatcher cannot run a bandwidth test") + } + return ft.FabricTest(ctx, nodeID, req) +} + func (r *dispatcherRef) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error { d, err := r.get() if err != nil { diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 8906a80..34f5867 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -75,6 +75,7 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement) mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState) mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode) + mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest) mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs) } diff --git a/internal/control/adminapi/fabrictest.go b/internal/control/adminapi/fabrictest.go new file mode 100644 index 0000000..fbe4f4d --- /dev/null +++ b/internal/control/adminapi/fabrictest.go @@ -0,0 +1,71 @@ +package adminapi + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" +) + +// FabricTestView is a bandwidth test between two nodes. +type FabricTestView struct { + A string `json:"a"` + B string `json:"b"` + Fabric string `json:"fabric"` + OK bool `json:"ok"` + Rails []RailView `json:"rails"` + Notes []string `json:"notes,omitempty"` + Took string `json:"took"` +} + +// RailView is one rail's result. +type RailView struct { + A string `json:"a"` + B string `json:"b"` + RateGbps int `json:"rate_gbps"` + Gbps float64 `json:"gbps"` + Verdict string `json:"verdict"` + Detail string `json:"detail,omitempty"` +} + +// fabricTest measures RDMA bandwidth between two nodes, rail by rail. It +// sends traffic over the fabric for seconds a rail, so it is a POST and only +// ever run when asked. +func (a *API) fabricTest(w http.ResponseWriter, r *http.Request) { + var body struct { + A, B, Fabric string + Seconds int + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + ask := a.ctl.FabricTester() + if ask == nil { + writeError(w, http.StatusNotImplemented, "this control plane cannot reach its nodes' agents") + return + } + inv := a.ctl.Inventory() + na, okA := inv.Get(body.A) + nb, okB := inv.Get(body.B) + if !okA || !okB { + writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", map[bool]string{true: body.B, false: body.A}[okA])) + return + } + fabric, rails, notes, err := fabrictest.Plan(na, nb, body.Fabric) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + start := time.Now() + rep := fabrictest.Run(r.Context(), ask, na, nb, fabric, rails, + fabrictest.Options{Seconds: min(max(body.Seconds, 1), 30), Settle: 2 * time.Second}) + v := FabricTestView{A: na.ID, B: nb.ID, Fabric: fabric, OK: rep.OK(), Notes: append(notes, rep.Notes...), + Took: time.Since(start).Round(time.Second).String()} + for _, x := range rep.Results { + v.Rails = append(v.Rails, RailView{x.A, x.B, x.RateGbps, x.Gbps, x.Verdict, x.Detail}) + } + writeJSON(w, http.StatusOK, v) +} diff --git a/internal/control/agentsvc/convert.go b/internal/control/agentsvc/convert.go index a98d22b..69ba97c 100644 --- a/internal/control/agentsvc/convert.go +++ b/internal/control/agentsvc/convert.go @@ -23,6 +23,7 @@ func NodeFromRegister(r *llmgwv1.Register) inventory.Node { Labels: r.GetLabels(), State: inventory.Ready, AgentVersion: r.GetAgentVersion(), + Features: r.GetFeatures(), EngineVersions: r.GetEngineVersions(), EnginesProbed: r.GetEnginesProbed(), MemoryTotalMB: int(r.GetMemoryTotalMb()), diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index 9a4207d..91207c1 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -19,6 +19,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1/llmgwv1connect" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" ) @@ -61,6 +62,7 @@ type Service struct { logSeq uint64 // scanWaits are the fabric scans awaiting a reply, by request id. scanWaits map[string]chan *llmgwv1.FabricScanReply + testWaits map[string]chan *llmgwv1.FabricTestReply } // SetEngines sets the engines every node installs, by name and version. @@ -289,6 +291,15 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) { "last_output", m.Shard.GetLastOutput()) } + case *llmgwv1.AgentMessage_FabricTest: + s.logMu.Lock() + ch, ok := s.testWaits[m.FabricTest.GetRequestId()] + delete(s.testWaits, m.FabricTest.GetRequestId()) + s.logMu.Unlock() + if ok { + ch <- m.FabricTest + } + case *llmgwv1.AgentMessage_FabricScan: s.logMu.Lock() ch, ok := s.scanWaits[m.FabricScan.GetRequestId()] @@ -717,6 +728,38 @@ func (s *Service) ScanFabric(ctx context.Context, nodeID, hca string, port int) } } +// FabricTest asks a node for one side of a bandwidth test, and waits for +// what it measured. The request's id is filled in here. +func (s *Service) FabricTest(ctx context.Context, nodeID string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) { + s.logMu.Lock() + if s.testWaits == nil { + s.testWaits = map[string]chan *llmgwv1.FabricTestReply{} + } + s.logSeq++ + id := fmt.Sprintf("test-%d", s.logSeq) + ch := make(chan *llmgwv1.FabricTestReply, 1) + s.testWaits[id] = ch + s.logMu.Unlock() + defer func() { + s.logMu.Lock() + delete(s.testWaits, id) + s.logMu.Unlock() + }() + req.RequestId = id + if err := s.send(nodeID, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_FabricTest{FabricTest: req}}); err != nil { + return nil, err + } + // The test's own length, and a package install the first time. + ctx, cancel := context.WithTimeout(ctx, time.Duration(req.GetSeconds())*time.Second+5*time.Minute) + defer cancel() + select { + case r := <-ch: + return r, nil + case <-ctx.Done(): + return nil, fmt.Errorf("node %s did not answer the bandwidth test", nodeID) + } +} + // ScanFabrics maps every physical InfiniBand fabric every so often, asking // one node on each: the first, by name, whose port on it is up. A fabric // looks the same from all its ports, so one is enough, and the rest of its @@ -795,3 +838,7 @@ func installable(reg *llmgwv1.Register) []string { } return []string{"llamacpp"} } + +// The admin API finds the bandwidth test by asserting this; a signature that +// drifted would make it quietly nil. +var _ fabrictest.Asker = (*Service)(nil) diff --git a/internal/control/control.go b/internal/control/control.go index 5b69327..57b97cb 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -11,6 +11,7 @@ import ( "time" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" @@ -215,6 +216,12 @@ func (c *Controller) LogReader() LogReader { return r } +// FabricTester returns the dispatcher's bandwidth test capability, or nil. +func (c *Controller) FabricTester() fabrictest.Asker { + r, _ := c.disp.(fabrictest.Asker) + return r +} + // Start runs the reconciler and begins publishing endpoints. func (c *Controller) Start(ctx context.Context) { if c.lead != nil { diff --git a/internal/control/fabrictest/fabrictest.go b/internal/control/fabrictest/fabrictest.go new file mode 100644 index 0000000..1753682 --- /dev/null +++ b/internal/control/fabrictest/fabrictest.go @@ -0,0 +1,235 @@ +// Package fabrictest measures the fabric between two nodes, rail by rail. +// +// Every check on one node — ports up, a subnet manager, the right modules — +// can pass while the two nodes still cannot reach each other: a cable into the +// wrong switch, a partition key, a RoCE network with no route. A replica +// spanning them then starts, and hangs on its first all-reduce with nothing to +// say why. This runs what that all-reduce would ride on — an RDMA write from +// each adapter to its counterpart — and says how it went against the link's +// rate, before a replica does. +package fabrictest + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + "time" + + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" +) + +// Asker sends one side of a test to a node and waits for its answer. +type Asker interface { + FabricTest(ctx context.Context, node string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) +} + +// Rail is one adapter on each node, paired by their position in the fabric's +// declaration: mlx5_0 with mlx5_0, the way NCCL pairs them. +type Rail struct { + A, B string // adapters + PortA, PortB int + // RateGbps is the slower of the two links. + RateGbps int +} + +// Result is one rail's test. +type Result struct { + Rail + Gbps float64 `json:"gbps"` + // Verdict is "ok" at 80% of the link's rate or more, "slow" at 50%, + // "fail" below that or when it did not run. + Verdict string `json:"verdict"` + Detail string `json:"detail,omitempty"` +} + +// Report is a whole test. +type Report struct { + Fabric string `json:"fabric"` + A, B string `json:"-"` + Results []Result `json:"results"` + // Notes say what the test could not cover. + Notes []string `json:"notes,omitempty"` +} + +// OK reports whether every rail passed. +func (r Report) OK() bool { + for _, x := range r.Results { + if x.Verdict != "ok" { + return false + } + } + return len(r.Results) > 0 +} + +// Plan pairs the adapters two nodes have on a fabric. With no fabric named, +// the one fabric they share is used. +func Plan(a, b inventory.Node, fabric string) (string, []Rail, []string, error) { + if a.ID == b.ID { + return "", nil, nil, errors.New("give two different nodes") + } + if fabric == "" { + var common []string + for _, f := range a.FabricNames() { + if b.HasFabric(f) { + common = append(common, f) + } + } + switch len(common) { + case 0: + return "", nil, nil, fmt.Errorf("%s (%s) and %s (%s) declare no fabric in common", + a.ID, orNone(a.FabricNames()), b.ID, orNone(b.FabricNames())) + case 1: + fabric = common[0] + default: + return "", nil, nil, fmt.Errorf("%s and %s share %s: name one with --fabric", a.ID, b.ID, strings.Join(common, ", ")) + } + } + for _, n := range []inventory.Node{a, b} { + if !slices.Contains(n.Features, "fabric-test") { + return "", nil, nil, fmt.Errorf("%s's agent (%s) is too old to run a bandwidth test: upgrade it", n.ID, orNone([]string{n.AgentVersion})) + } + } + ha, hb := a.HCAsFor(fabric), b.HCAsFor(fabric) + if len(ha) == 0 || len(hb) == 0 { + return "", nil, nil, fmt.Errorf("%s is not declared on both %s and %s", fabric, a.ID, b.ID) + } + var notes []string + if len(ha) != len(hb) { + notes = append(notes, fmt.Sprintf("%s has %d adapters on %s and %s has %d: only the first %d are paired", + a.ID, len(ha), fabric, b.ID, len(hb), min(len(ha), len(hb)))) + } + var rails []Rail + for i := range min(len(ha), len(hb)) { + pa, ra := port(a, ha[i]) + pb, rb := port(b, hb[i]) + rails = append(rails, Rail{A: ha[i], B: hb[i], PortA: pa, PortB: pb, RateGbps: minPositive(ra, rb)}) + } + return fabric, rails, notes, nil +} + +// port is an adapter's first active port and its rate, or port 1 when the +// node reported none. +func port(n inventory.Node, hca string) (int, int) { + for _, p := range n.IBPorts { + if p.HCA == hca && strings.EqualFold(p.State, "ACTIVE") { + return p.Port, p.RateGbps + } + } + return 1, 0 +} + +// Options for a run. +type Options struct { + Seconds int + // BasePort is perftest's TCP exchange port for the first rail; each rail + // takes the next, so a rail whose server is still closing never meets the + // next one's. + BasePort int + // Settle is how long the server is given to listen before the client + // dials; the client retries besides. + Settle time.Duration +} + +// Run tests every rail, one after another: rails at once would share the +// PCIe and memory bandwidth they are measuring. +func Run(ctx context.Context, ask Asker, a, b inventory.Node, fabric string, rails []Rail, o Options) Report { + if o.Seconds <= 0 { + o.Seconds = 3 + } + if o.BasePort == 0 { + o.BasePort = 18515 + } + rep := Report{Fabric: fabric, A: a.ID, B: b.ID} + for i, r := range rails { + rep.Results = append(rep.Results, runRail(ctx, ask, a, b, r, o, o.BasePort+i)) + } + return rep +} + +func runRail(ctx context.Context, ask Asker, a, b inventory.Node, r Rail, o Options, tcp int) Result { + res := Result{Rail: r, Verdict: "fail"} + if b.MgmtAddr == "" { + res.Detail = b.ID + " reported no management address for the client to dial" + return res + } + type answer struct { + r *llmgwv1.FabricTestReply + err error + } + server := make(chan answer, 1) + go func() { + rep, err := ask.FabricTest(ctx, b.ID, &llmgwv1.FabricTestRequest{Role: "server", Hca: r.B, + Port: int32(r.PortB), TcpPort: int32(tcp), Seconds: int32(o.Seconds)}) //nolint:gosec // small numbers + server <- answer{rep, err} + }() + select { + case <-ctx.Done(): + res.Detail = ctx.Err().Error() + return res + case <-time.After(o.Settle): + } + client, err := ask.FabricTest(ctx, a.ID, &llmgwv1.FabricTestRequest{Role: "client", Hca: r.A, + Port: int32(r.PortA), Peer: b.MgmtAddr, TcpPort: int32(tcp), Seconds: int32(o.Seconds)}) //nolint:gosec // small numbers + var srv answer + select { + case srv = <-server: + case <-time.After(time.Duration(o.Seconds)*time.Second + time.Minute): + srv.err = errors.New("the server side did not finish") + } + switch { + case err != nil: + res.Detail = a.ID + ": " + err.Error() + case client.GetError() != "": + res.Detail = a.ID + ": " + client.GetError() + // The server's reason is often the better one: it saw no one arrive. + if srv.r != nil && srv.r.GetError() != "" { + res.Detail += "; " + b.ID + ": " + srv.r.GetError() + } + default: + res.Gbps = client.GetGbps() + res.Verdict, res.Detail = grade(res.Gbps, r.RateGbps) + } + return res +} + +// grade compares a measurement with the link's rate. RDMA writes of 64 KiB +// reach 90–97% of the signalling rate on a healthy link; well under that is +// a degraded cable, a card in a narrow PCIe slot, or traffic crossing a +// socket. +func grade(gbps float64, rate int) (string, string) { + if gbps <= 0 { + return "fail", "no traffic got through" + } + if rate <= 0 { + return "ok", "the link's rate is not known to compare with" + } + pct := gbps / float64(rate) * 100 + switch { + case pct >= 80: + return "ok", fmt.Sprintf("%.0f%% of %d Gb/s", pct, rate) + case pct >= 50: + return "slow", fmt.Sprintf("%.0f%% of %d Gb/s: check the cable, and the card's PCIe width and NUMA node", pct, rate) + default: + return "fail", fmt.Sprintf("%.0f%% of %d Gb/s", pct, rate) + } +} + +func minPositive(a, b int) int { + switch { + case a <= 0: + return b + case b <= 0: + return a + } + return min(a, b) +} + +func orNone(s []string) string { + if len(s) == 0 { + return "none" + } + return strings.Join(slices.Clone(s), ", ") +} diff --git a/internal/control/fabrictest/fabrictest_test.go b/internal/control/fabrictest/fabrictest_test.go new file mode 100644 index 0000000..4c2bb8b --- /dev/null +++ b/internal/control/fabrictest/fabrictest_test.go @@ -0,0 +1,100 @@ +package fabrictest + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" +) + +func node(id string, hcas ...string) inventory.Node { + n := inventory.Node{ID: id, MgmtAddr: "10.0.0." + id[len(id)-1:], Features: []string{"fabric-test"}, + Fabrics: []inventory.FabricAttachment{{Fabric: "ib-a", HCAs: hcas}}} + for _, h := range hcas { + n.IBPorts = append(n.IBPorts, inventory.IBPort{HCA: h, Port: 1, State: "ACTIVE", RateGbps: 400}) + } + return n +} + +// fakeFabric answers as two agents would: the server waits for its client. +type fakeFabric struct { + mu sync.Mutex + gbps map[string]float64 // by client adapter + fails map[string]string + asked []string +} + +func (f *fakeFabric) FabricTest(_ context.Context, n string, req *llmgwv1.FabricTestRequest) (*llmgwv1.FabricTestReply, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.asked = append(f.asked, n+" "+req.GetRole()+" "+req.GetHca()) + if req.GetRole() == "server" { + return &llmgwv1.FabricTestReply{}, nil + } + if e := f.fails[req.GetHca()]; e != "" { + return &llmgwv1.FabricTestReply{Error: e}, nil + } + return &llmgwv1.FabricTestReply{Gbps: f.gbps[req.GetHca()]}, nil +} + +func TestRailsArePairedByPositionOnTheSharedFabric(t *testing.T) { + a, b := node("n1", "mlx5_0", "mlx5_1"), node("n2", "mlx5_0", "mlx5_1", "mlx5_2") + fabric, rails, notes, err := Plan(a, b, "") + if err != nil { + t.Fatal(err) + } + if fabric != "ib-a" || len(rails) != 2 || rails[1].A != "mlx5_1" || rails[1].B != "mlx5_1" || rails[0].RateGbps != 400 { + t.Errorf("got %s %+v", fabric, rails) + } + if len(notes) != 1 || !strings.Contains(notes[0], "only the first 2") { + t.Errorf("an unequal count went unmentioned: %v", notes) + } +} + +func TestNodesWithNoFabricInCommonAreSaidSo(t *testing.T) { + b := node("n2", "mlx5_0") + b.Fabrics[0].Fabric = "ib-b" + if _, _, _, err := Plan(node("n1", "mlx5_0"), b, ""); err == nil || !strings.Contains(err.Error(), "no fabric in common") { + t.Errorf("got %v", err) + } +} + +func TestEachRailIsGradedAgainstItsLink(t *testing.T) { + a, b := node("n1", "mlx5_0", "mlx5_1", "mlx5_2"), node("n2", "mlx5_0", "mlx5_1", "mlx5_2") + _, rails, _, _ := Plan(a, b, "ib-a") + f := &fakeFabric{ + gbps: map[string]float64{"mlx5_0": 388, "mlx5_1": 240}, + fails: map[string]string{"mlx5_2": "ib_write_bw: Couldn't connect to 10.0.0.2:18517"}, + } + rep := Run(context.Background(), f, a, b, "ib-a", rails, Options{Seconds: 1, Settle: time.Millisecond}) + want := []string{"ok", "slow", "fail"} + for i, r := range rep.Results { + if r.Verdict != want[i] { + t.Errorf("rail %d: %s (%s), want %s", i, r.Verdict, r.Detail, want[i]) + } + } + if rep.OK() { + t.Error("a report with a failed rail is OK") + } + if !strings.Contains(rep.Results[2].Detail, "Couldn't connect") { + t.Errorf("the reason was lost: %q", rep.Results[2].Detail) + } + // Servers on the second node, clients on the first, one rail at a time. + if f.asked[0] != "n2 server mlx5_0" || f.asked[1] != "n1 client mlx5_0" { + t.Errorf("asked %v", f.asked) + } +} + +func TestAnAgentTooOldToAnswerIsRefusedAtOnce(t *testing.T) { + // It would ignore the request, and the control plane would wait minutes. + old := node("n2", "mlx5_0") + old.Features, old.AgentVersion = nil, "v0.10.0" + _, _, _, err := Plan(node("n1", "mlx5_0"), old, "") + if err == nil || !strings.Contains(err.Error(), "n2's agent (v0.10.0) is too old") { + t.Errorf("got %v", err) + } +} diff --git a/internal/control/inventory/inventory.go b/internal/control/inventory/inventory.go index 52e3e9c..7cef569 100644 --- a/internal/control/inventory/inventory.go +++ b/internal/control/inventory/inventory.go @@ -100,7 +100,10 @@ type Node struct { LastSeen time.Time `json:"last_seen"` // AgentVersion and EngineVersions are reported at registration. The engine // version decides which metric names and CLI flags apply (plan §14). - AgentVersion string `json:"agent_version,omitempty"` + AgentVersion string `json:"agent_version,omitempty"` + // Features are the requests the agent answers beyond the ones every + // agent does. + Features []string `json:"features,omitempty"` EngineVersions map[string]string `json:"engine_versions,omitempty"` // EnginesProbed says EngineVersions is an answer, so an empty map means // "has none" rather than "an agent too old to say".