diff --git a/BACKLOG.md b/BACKLOG.md index b93f3cf..073ff4d 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -793,7 +793,11 @@ leader 30s later with /health answering throughout; rollback did the same. facility state from the store (it kept what it read at start). - Retires `appendGuardIfMissing` (server init): the import does it. -### E2. Generated files from settings (L) +### E2. Generated files from settings (L) — firewall on/off done 2026-10-06 +`firewall.enabled` in the cluster's settings reaches nodes over the agent +stream and control planes through llmgw-firewall-sync.path. Left: the mirror +machine and store members, the allowed set from membership, the other files. + Every machine runs the agent (root) with roles — node, control plane, mirror, store member — and the agent renders and applies what was written once by init: the llmgw nft table, the mirror's unit flags, keepalived, etcd and diff --git a/CHANGELOG.md b/CHANGELOG.md index 7957de7..20630c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,19 @@ APIs change between minors without a deprecation period. guardrails; after that, those keys in a server.yaml are ignored, with a warning. +### Added + +- **llmgw's firewall can be turned off and on for the whole cluster**: + `llmgw ctl config set firewall.enabled false`. Every node's agent takes its + table out within seconds, and each control plane's new + `llmgw-firewall-sync.path` does the same there; `true` puts the rules back + as they were. `firewall_networks` changes reach the nodes the same way. +- **`llmgw wipe`** removes llmgw from a machine — services, firewall table, + configuration, the authority, secrets, state, users — after listing it and + asking for the machine's name; `--models`, `--store` and `--all` take + weights, the store's database here, backups and the binary too. Also + **Do → Wipe this machine** in the console. + ### Fixed - **A standby that becomes the leader reads the store again**: models added diff --git a/api/gen/llmgw/v1/agent.pb.go b/api/gen/llmgw/v1/agent.pb.go index 32aed9e..e7c6dbc 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{23, 0} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{24, 0} } // AgentMessage is anything a node tells the control plane. @@ -389,6 +389,7 @@ type ControlMessage struct { // *ControlMessage_FabricTest // *ControlMessage_Upgrade // *ControlMessage_ControlPlanes + // *ControlMessage_Firewall Msg isControlMessage_Msg `protobuf_oneof:"msg"` // trace is the W3C trace context (traceparent, and tracestate when there // is one) of the control plane's span that sent this command, so what the @@ -536,6 +537,15 @@ func (x *ControlMessage) GetControlPlanes() *ControlPlanes { return nil } +func (x *ControlMessage) GetFirewall() *Firewall { + if x != nil { + if x, ok := x.Msg.(*ControlMessage_Firewall); ok { + return x.Firewall + } + } + return nil +} + func (x *ControlMessage) GetTrace() map[string]string { if x != nil { return x.Trace @@ -591,6 +601,10 @@ type ControlMessage_ControlPlanes struct { ControlPlanes *ControlPlanes `protobuf:"bytes,11,opt,name=control_planes,json=controlPlanes,proto3,oneof"` } +type ControlMessage_Firewall struct { + Firewall *Firewall `protobuf:"bytes,13,opt,name=firewall,proto3,oneof"` +} + func (*ControlMessage_Start) isControlMessage_Msg() {} func (*ControlMessage_Stop) isControlMessage_Msg() {} @@ -613,6 +627,84 @@ func (*ControlMessage_Upgrade) isControlMessage_Msg() {} func (*ControlMessage_ControlPlanes) isControlMessage_Msg() {} +func (*ControlMessage_Firewall) isControlMessage_Msg() {} + +// Firewall is whether llmgw's firewall is kept on the node — its nftables +// table, which lets the cluster's networks alone reach the engines' ports and +// the node exporter — sent when an agent registers and whenever the +// cluster's settings change it (firewall.enabled). +type Firewall struct { + state protoimpl.MessageState `protogen:"open.v1"` + Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"` + // networks are let in besides those the node reaches the control planes + // on: firewall_networks, networks or single addresses. + Networks []string `protobuf:"bytes,2,rep,name=networks,proto3" json:"networks,omitempty"` + // engine_ports_from and engine_ports_to are the engines' ports + // (control.port_range). + EnginePortsFrom int32 `protobuf:"varint,3,opt,name=engine_ports_from,json=enginePortsFrom,proto3" json:"engine_ports_from,omitempty"` + EnginePortsTo int32 `protobuf:"varint,4,opt,name=engine_ports_to,json=enginePortsTo,proto3" json:"engine_ports_to,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Firewall) Reset() { + *x = Firewall{} + mi := &file_llmgw_v1_agent_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Firewall) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Firewall) ProtoMessage() {} + +func (x *Firewall) ProtoReflect() protoreflect.Message { + mi := &file_llmgw_v1_agent_proto_msgTypes[3] + 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 Firewall.ProtoReflect.Descriptor instead. +func (*Firewall) Descriptor() ([]byte, []int) { + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{3} +} + +func (x *Firewall) GetEnabled() bool { + if x != nil { + return x.Enabled + } + return false +} + +func (x *Firewall) GetNetworks() []string { + if x != nil { + return x.Networks + } + return nil +} + +func (x *Firewall) GetEnginePortsFrom() int32 { + if x != nil { + return x.EnginePortsFrom + } + return 0 +} + +func (x *Firewall) GetEnginePortsTo() int32 { + if x != nil { + return x.EnginePortsTo + } + return 0 +} + // ControlPlanes is every control plane an agent may dial, sent by the leader // when an agent registers and whenever the list changes. The agent keeps it // beside its configuration, so a second control plane added after the node @@ -626,7 +718,7 @@ type ControlPlanes struct { func (x *ControlPlanes) Reset() { *x = ControlPlanes{} - mi := &file_llmgw_v1_agent_proto_msgTypes[3] + mi := &file_llmgw_v1_agent_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -638,7 +730,7 @@ func (x *ControlPlanes) String() string { func (*ControlPlanes) ProtoMessage() {} func (x *ControlPlanes) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[3] + mi := &file_llmgw_v1_agent_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -651,7 +743,7 @@ func (x *ControlPlanes) ProtoReflect() protoreflect.Message { // Deprecated: Use ControlPlanes.ProtoReflect.Descriptor instead. func (*ControlPlanes) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{3} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{4} } func (x *ControlPlanes) GetServers() []string { @@ -751,7 +843,7 @@ type Register struct { func (x *Register) Reset() { *x = Register{} - mi := &file_llmgw_v1_agent_proto_msgTypes[4] + mi := &file_llmgw_v1_agent_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -763,7 +855,7 @@ func (x *Register) String() string { func (*Register) ProtoMessage() {} func (x *Register) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[4] + mi := &file_llmgw_v1_agent_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -776,7 +868,7 @@ func (x *Register) ProtoReflect() protoreflect.Message { // Deprecated: Use Register.ProtoReflect.Descriptor instead. func (*Register) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{4} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{5} } func (x *Register) GetNodeId() string { @@ -928,7 +1020,7 @@ type Host struct { func (x *Host) Reset() { *x = Host{} - mi := &file_llmgw_v1_agent_proto_msgTypes[5] + mi := &file_llmgw_v1_agent_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -940,7 +1032,7 @@ func (x *Host) String() string { func (*Host) ProtoMessage() {} func (x *Host) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[5] + mi := &file_llmgw_v1_agent_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -953,7 +1045,7 @@ func (x *Host) ProtoReflect() protoreflect.Message { // Deprecated: Use Host.ProtoReflect.Descriptor instead. func (*Host) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{5} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{6} } func (x *Host) GetOsId() string { @@ -1020,7 +1112,7 @@ type Storage struct { func (x *Storage) Reset() { *x = Storage{} - mi := &file_llmgw_v1_agent_proto_msgTypes[6] + mi := &file_llmgw_v1_agent_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1032,7 +1124,7 @@ func (x *Storage) String() string { func (*Storage) ProtoMessage() {} func (x *Storage) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[6] + mi := &file_llmgw_v1_agent_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1045,7 +1137,7 @@ func (x *Storage) ProtoReflect() protoreflect.Message { // Deprecated: Use Storage.ProtoReflect.Descriptor instead. func (*Storage) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{6} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{7} } func (x *Storage) GetPath() string { @@ -1112,7 +1204,7 @@ type RunningShard struct { func (x *RunningShard) Reset() { *x = RunningShard{} - mi := &file_llmgw_v1_agent_proto_msgTypes[7] + mi := &file_llmgw_v1_agent_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1124,7 +1216,7 @@ func (x *RunningShard) String() string { func (*RunningShard) ProtoMessage() {} func (x *RunningShard) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[7] + mi := &file_llmgw_v1_agent_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1137,7 +1229,7 @@ func (x *RunningShard) ProtoReflect() protoreflect.Message { // Deprecated: Use RunningShard.ProtoReflect.Descriptor instead. func (*RunningShard) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{7} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{8} } func (x *RunningShard) GetReplicaId() string { @@ -1290,7 +1382,7 @@ type GPU struct { func (x *GPU) Reset() { *x = GPU{} - mi := &file_llmgw_v1_agent_proto_msgTypes[8] + mi := &file_llmgw_v1_agent_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1302,7 +1394,7 @@ func (x *GPU) String() string { func (*GPU) ProtoMessage() {} func (x *GPU) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[8] + mi := &file_llmgw_v1_agent_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1315,7 +1407,7 @@ func (x *GPU) ProtoReflect() protoreflect.Message { // Deprecated: Use GPU.ProtoReflect.Descriptor instead. func (*GPU) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{8} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{9} } func (x *GPU) GetIndex() int32 { @@ -1491,7 +1583,7 @@ type MIGDevice struct { func (x *MIGDevice) Reset() { *x = MIGDevice{} - mi := &file_llmgw_v1_agent_proto_msgTypes[9] + mi := &file_llmgw_v1_agent_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1503,7 +1595,7 @@ func (x *MIGDevice) String() string { func (*MIGDevice) ProtoMessage() {} func (x *MIGDevice) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[9] + mi := &file_llmgw_v1_agent_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1516,7 +1608,7 @@ func (x *MIGDevice) ProtoReflect() protoreflect.Message { // Deprecated: Use MIGDevice.ProtoReflect.Descriptor instead. func (*MIGDevice) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{9} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{10} } func (x *MIGDevice) GetParentIndex() int32 { @@ -1582,7 +1674,7 @@ type FabricAttachment struct { func (x *FabricAttachment) Reset() { *x = FabricAttachment{} - mi := &file_llmgw_v1_agent_proto_msgTypes[10] + mi := &file_llmgw_v1_agent_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1594,7 +1686,7 @@ func (x *FabricAttachment) String() string { func (*FabricAttachment) ProtoMessage() {} func (x *FabricAttachment) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[10] + mi := &file_llmgw_v1_agent_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1607,7 +1699,7 @@ func (x *FabricAttachment) ProtoReflect() protoreflect.Message { // Deprecated: Use FabricAttachment.ProtoReflect.Descriptor instead. func (*FabricAttachment) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{10} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{11} } func (x *FabricAttachment) GetFabric() string { @@ -1649,7 +1741,7 @@ type LocalWeights struct { func (x *LocalWeights) Reset() { *x = LocalWeights{} - mi := &file_llmgw_v1_agent_proto_msgTypes[11] + mi := &file_llmgw_v1_agent_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1661,7 +1753,7 @@ func (x *LocalWeights) String() string { func (*LocalWeights) ProtoMessage() {} func (x *LocalWeights) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[11] + mi := &file_llmgw_v1_agent_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1674,7 +1766,7 @@ func (x *LocalWeights) ProtoReflect() protoreflect.Message { // Deprecated: Use LocalWeights.ProtoReflect.Descriptor instead. func (*LocalWeights) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{11} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{12} } func (x *LocalWeights) GetPath() string { @@ -1730,7 +1822,7 @@ type NodeLoad struct { func (x *NodeLoad) Reset() { *x = NodeLoad{} - mi := &file_llmgw_v1_agent_proto_msgTypes[12] + mi := &file_llmgw_v1_agent_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1742,7 +1834,7 @@ func (x *NodeLoad) String() string { func (*NodeLoad) ProtoMessage() {} func (x *NodeLoad) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[12] + mi := &file_llmgw_v1_agent_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1755,7 +1847,7 @@ func (x *NodeLoad) ProtoReflect() protoreflect.Message { // Deprecated: Use NodeLoad.ProtoReflect.Descriptor instead. func (*NodeLoad) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{12} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{13} } func (x *NodeLoad) GetCpuPercent() float64 { @@ -1888,7 +1980,7 @@ type GPUSensor struct { func (x *GPUSensor) Reset() { *x = GPUSensor{} - mi := &file_llmgw_v1_agent_proto_msgTypes[13] + mi := &file_llmgw_v1_agent_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1900,7 +1992,7 @@ func (x *GPUSensor) String() string { func (*GPUSensor) ProtoMessage() {} func (x *GPUSensor) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[13] + mi := &file_llmgw_v1_agent_proto_msgTypes[14] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1913,7 +2005,7 @@ func (x *GPUSensor) ProtoReflect() protoreflect.Message { // Deprecated: Use GPUSensor.ProtoReflect.Descriptor instead. func (*GPUSensor) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{13} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{14} } func (x *GPUSensor) GetIndex() int32 { @@ -2127,7 +2219,7 @@ type ShardLoad struct { func (x *ShardLoad) Reset() { *x = ShardLoad{} - mi := &file_llmgw_v1_agent_proto_msgTypes[14] + mi := &file_llmgw_v1_agent_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2139,7 +2231,7 @@ func (x *ShardLoad) String() string { func (*ShardLoad) ProtoMessage() {} func (x *ShardLoad) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[14] + mi := &file_llmgw_v1_agent_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2152,7 +2244,7 @@ func (x *ShardLoad) ProtoReflect() protoreflect.Message { // Deprecated: Use ShardLoad.ProtoReflect.Descriptor instead. func (*ShardLoad) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{14} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{15} } func (x *ShardLoad) GetShardId() string { @@ -2220,7 +2312,7 @@ type IBPort struct { func (x *IBPort) Reset() { *x = IBPort{} - mi := &file_llmgw_v1_agent_proto_msgTypes[15] + mi := &file_llmgw_v1_agent_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2232,7 +2324,7 @@ func (x *IBPort) String() string { func (*IBPort) ProtoMessage() {} func (x *IBPort) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[15] + mi := &file_llmgw_v1_agent_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2245,7 +2337,7 @@ func (x *IBPort) ProtoReflect() protoreflect.Message { // Deprecated: Use IBPort.ProtoReflect.Descriptor instead. func (*IBPort) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{15} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{16} } func (x *IBPort) GetHca() string { @@ -2388,7 +2480,7 @@ type SubnetManager struct { func (x *SubnetManager) Reset() { *x = SubnetManager{} - mi := &file_llmgw_v1_agent_proto_msgTypes[16] + mi := &file_llmgw_v1_agent_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2400,7 +2492,7 @@ func (x *SubnetManager) String() string { func (*SubnetManager) ProtoMessage() {} func (x *SubnetManager) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[16] + mi := &file_llmgw_v1_agent_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2413,7 +2505,7 @@ func (x *SubnetManager) ProtoReflect() protoreflect.Message { // Deprecated: Use SubnetManager.ProtoReflect.Descriptor instead. func (*SubnetManager) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{16} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{17} } func (x *SubnetManager) GetGuid() string { @@ -2456,7 +2548,7 @@ type IMEX struct { func (x *IMEX) Reset() { *x = IMEX{} - mi := &file_llmgw_v1_agent_proto_msgTypes[17] + mi := &file_llmgw_v1_agent_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2468,7 +2560,7 @@ func (x *IMEX) String() string { func (*IMEX) ProtoMessage() {} func (x *IMEX) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[17] + mi := &file_llmgw_v1_agent_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2481,7 +2573,7 @@ func (x *IMEX) ProtoReflect() protoreflect.Message { // Deprecated: Use IMEX.ProtoReflect.Descriptor instead. func (*IMEX) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{17} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{18} } func (x *IMEX) GetState() string { @@ -2513,7 +2605,7 @@ type GPUFabricState struct { func (x *GPUFabricState) Reset() { *x = GPUFabricState{} - mi := &file_llmgw_v1_agent_proto_msgTypes[18] + mi := &file_llmgw_v1_agent_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2525,7 +2617,7 @@ func (x *GPUFabricState) String() string { func (*GPUFabricState) ProtoMessage() {} func (x *GPUFabricState) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[18] + mi := &file_llmgw_v1_agent_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2538,7 +2630,7 @@ func (x *GPUFabricState) ProtoReflect() protoreflect.Message { // Deprecated: Use GPUFabricState.ProtoReflect.Descriptor instead. func (*GPUFabricState) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{18} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{19} } func (x *GPUFabricState) GetIndex() int32 { @@ -2653,7 +2745,7 @@ type Heartbeat struct { func (x *Heartbeat) Reset() { *x = Heartbeat{} - mi := &file_llmgw_v1_agent_proto_msgTypes[19] + mi := &file_llmgw_v1_agent_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2665,7 +2757,7 @@ func (x *Heartbeat) String() string { func (*Heartbeat) ProtoMessage() {} func (x *Heartbeat) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[19] + mi := &file_llmgw_v1_agent_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2678,7 +2770,7 @@ func (x *Heartbeat) ProtoReflect() protoreflect.Message { // Deprecated: Use Heartbeat.ProtoReflect.Descriptor instead. func (*Heartbeat) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{19} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{20} } func (x *Heartbeat) GetNodeId() string { @@ -2833,7 +2925,7 @@ type Provision struct { func (x *Provision) Reset() { *x = Provision{} - mi := &file_llmgw_v1_agent_proto_msgTypes[20] + mi := &file_llmgw_v1_agent_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2845,7 +2937,7 @@ func (x *Provision) String() string { func (*Provision) ProtoMessage() {} func (x *Provision) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[20] + mi := &file_llmgw_v1_agent_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2858,7 +2950,7 @@ func (x *Provision) ProtoReflect() protoreflect.Message { // Deprecated: Use Provision.ProtoReflect.Descriptor instead. func (*Provision) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{20} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{21} } func (x *Provision) GetEngines() []*EngineWant { @@ -2891,7 +2983,7 @@ type EngineWant struct { func (x *EngineWant) Reset() { *x = EngineWant{} - mi := &file_llmgw_v1_agent_proto_msgTypes[21] + mi := &file_llmgw_v1_agent_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2903,7 +2995,7 @@ func (x *EngineWant) String() string { func (*EngineWant) ProtoMessage() {} func (x *EngineWant) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[21] + mi := &file_llmgw_v1_agent_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2916,7 +3008,7 @@ func (x *EngineWant) ProtoReflect() protoreflect.Message { // Deprecated: Use EngineWant.ProtoReflect.Descriptor instead. func (*EngineWant) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{21} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{22} } func (x *EngineWant) GetEngine() string { @@ -2970,7 +3062,7 @@ type ProvisionStatus struct { func (x *ProvisionStatus) Reset() { *x = ProvisionStatus{} - mi := &file_llmgw_v1_agent_proto_msgTypes[22] + mi := &file_llmgw_v1_agent_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -2982,7 +3074,7 @@ func (x *ProvisionStatus) String() string { func (*ProvisionStatus) ProtoMessage() {} func (x *ProvisionStatus) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[22] + mi := &file_llmgw_v1_agent_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -2995,7 +3087,7 @@ func (x *ProvisionStatus) ProtoReflect() protoreflect.Message { // Deprecated: Use ProvisionStatus.ProtoReflect.Descriptor instead. func (*ProvisionStatus) Descriptor() ([]byte, []int) { - return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{22} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{23} } func (x *ProvisionStatus) GetEngine() string { @@ -3059,7 +3151,7 @@ type ShardStatus struct { func (x *ShardStatus) Reset() { *x = ShardStatus{} - mi := &file_llmgw_v1_agent_proto_msgTypes[23] + mi := &file_llmgw_v1_agent_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -3071,7 +3163,7 @@ func (x *ShardStatus) String() string { func (*ShardStatus) ProtoMessage() {} func (x *ShardStatus) ProtoReflect() protoreflect.Message { - mi := &file_llmgw_v1_agent_proto_msgTypes[23] + mi := &file_llmgw_v1_agent_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -3084,7 +3176,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{23} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{24} } func (x *ShardStatus) GetNodeId() string { @@ -3158,7 +3250,7 @@ type FetchProgress struct { func (x *FetchProgress) Reset() { *x = FetchProgress{} - 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) } @@ -3170,7 +3262,7 @@ func (x *FetchProgress) String() string { func (*FetchProgress) ProtoMessage() {} func (x *FetchProgress) 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 { @@ -3183,7 +3275,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{24} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{25} } func (x *FetchProgress) GetSource() string { @@ -3246,7 +3338,7 @@ type StartShard struct { func (x *StartShard) Reset() { *x = StartShard{} - 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) } @@ -3258,7 +3350,7 @@ func (x *StartShard) String() string { func (*StartShard) ProtoMessage() {} func (x *StartShard) 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 { @@ -3271,7 +3363,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{25} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{26} } func (x *StartShard) GetReplicaId() string { @@ -3418,7 +3510,7 @@ type EngineProfile struct { func (x *EngineProfile) Reset() { *x = EngineProfile{} - 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) } @@ -3430,7 +3522,7 @@ func (x *EngineProfile) String() string { func (*EngineProfile) ProtoMessage() {} func (x *EngineProfile) 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 { @@ -3443,7 +3535,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{26} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{27} } func (x *EngineProfile) GetName() string { @@ -3553,7 +3645,7 @@ type RemoveWeights struct { func (x *RemoveWeights) Reset() { *x = RemoveWeights{} - 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) } @@ -3565,7 +3657,7 @@ func (x *RemoveWeights) String() string { func (*RemoveWeights) ProtoMessage() {} func (x *RemoveWeights) 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 { @@ -3578,7 +3670,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{27} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{28} } func (x *RemoveWeights) GetPaths() []string { @@ -3598,7 +3690,7 @@ type StopShard struct { func (x *StopShard) Reset() { *x = StopShard{} - 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) } @@ -3610,7 +3702,7 @@ func (x *StopShard) String() string { func (*StopShard) ProtoMessage() {} func (x *StopShard) 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 { @@ -3623,7 +3715,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{28} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{29} } func (x *StopShard) GetShardId() string { @@ -3643,7 +3735,7 @@ type DrainShard struct { func (x *DrainShard) Reset() { *x = DrainShard{} - 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) } @@ -3655,7 +3747,7 @@ func (x *DrainShard) String() string { func (*DrainShard) ProtoMessage() {} func (x *DrainShard) 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 { @@ -3668,7 +3760,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{29} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{30} } func (x *DrainShard) GetShardId() string { @@ -3690,7 +3782,7 @@ type Ack struct { func (x *Ack) Reset() { *x = Ack{} - 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) } @@ -3702,7 +3794,7 @@ func (x *Ack) String() string { func (*Ack) ProtoMessage() {} func (x *Ack) 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 { @@ -3715,7 +3807,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{30} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{31} } func (x *Ack) GetMessageId() string { @@ -3759,7 +3851,7 @@ type LogsRequest struct { func (x *LogsRequest) Reset() { *x = LogsRequest{} - 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) } @@ -3771,7 +3863,7 @@ func (x *LogsRequest) String() string { func (*LogsRequest) ProtoMessage() {} func (x *LogsRequest) 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 { @@ -3784,7 +3876,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{31} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{32} } func (x *LogsRequest) GetRequestId() string { @@ -3824,7 +3916,7 @@ type LogsReply struct { func (x *LogsReply) Reset() { *x = LogsReply{} - 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) } @@ -3836,7 +3928,7 @@ func (x *LogsReply) String() string { func (*LogsReply) ProtoMessage() {} func (x *LogsReply) 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 { @@ -3849,7 +3941,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{32} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{33} } func (x *LogsReply) GetRequestId() string { @@ -3895,7 +3987,7 @@ type FabricScanRequest struct { func (x *FabricScanRequest) Reset() { *x = FabricScanRequest{} - 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) } @@ -3907,7 +3999,7 @@ func (x *FabricScanRequest) String() string { func (*FabricScanRequest) ProtoMessage() {} func (x *FabricScanRequest) 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 { @@ -3920,7 +4012,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{33} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{34} } func (x *FabricScanRequest) GetRequestId() string { @@ -3959,7 +4051,7 @@ type FabricScanReply struct { func (x *FabricScanReply) Reset() { *x = FabricScanReply{} - 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) } @@ -3971,7 +4063,7 @@ func (x *FabricScanReply) String() string { func (*FabricScanReply) ProtoMessage() {} func (x *FabricScanReply) 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 { @@ -3984,7 +4076,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{34} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{35} } func (x *FabricScanReply) GetRequestId() string { @@ -4040,7 +4132,7 @@ type FabricTestRequest struct { func (x *FabricTestRequest) Reset() { *x = FabricTestRequest{} - 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) } @@ -4052,7 +4144,7 @@ func (x *FabricTestRequest) String() string { func (*FabricTestRequest) ProtoMessage() {} func (x *FabricTestRequest) 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 { @@ -4065,7 +4157,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{35} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{36} } func (x *FabricTestRequest) GetRequestId() string { @@ -4134,7 +4226,7 @@ type FabricTestReply struct { func (x *FabricTestReply) Reset() { *x = FabricTestReply{} - 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) } @@ -4146,7 +4238,7 @@ func (x *FabricTestReply) String() string { func (*FabricTestReply) ProtoMessage() {} func (x *FabricTestReply) 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 { @@ -4159,7 +4251,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{36} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{37} } func (x *FabricTestReply) GetRequestId() string { @@ -4219,7 +4311,7 @@ type UpgradeAgent struct { func (x *UpgradeAgent) Reset() { *x = UpgradeAgent{} - 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) } @@ -4231,7 +4323,7 @@ func (x *UpgradeAgent) String() string { func (*UpgradeAgent) ProtoMessage() {} func (x *UpgradeAgent) 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 { @@ -4244,7 +4336,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{37} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{38} } func (x *UpgradeAgent) GetRequestId() string { @@ -4281,7 +4373,7 @@ type UpgradeReply struct { func (x *UpgradeReply) Reset() { *x = UpgradeReply{} - 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) } @@ -4293,7 +4385,7 @@ func (x *UpgradeReply) String() string { func (*UpgradeReply) ProtoMessage() {} func (x *UpgradeReply) 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 { @@ -4306,7 +4398,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{38} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{39} } func (x *UpgradeReply) GetRequestId() string { @@ -4342,7 +4434,7 @@ type LoraAdapter struct { func (x *LoraAdapter) Reset() { *x = LoraAdapter{} - 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) } @@ -4354,7 +4446,7 @@ func (x *LoraAdapter) String() string { func (*LoraAdapter) ProtoMessage() {} func (x *LoraAdapter) 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 { @@ -4367,7 +4459,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{39} + return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{40} } func (x *LoraAdapter) GetName() string { @@ -4410,7 +4502,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "clearFirst\x12\x16\n" + "\x06stored\x18\x04 \x03(\x06R\x06stored\x12\x18\n" + "\aremoved\x18\x05 \x03(\x06R\aremoved\x12\x18\n" + - "\aproblem\x18\x06 \x01(\tR\aproblem\"\xd0\x05\n" + + "\aproblem\x18\x06 \x01(\tR\aproblem\"\x82\x06\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" + @@ -4425,13 +4517,19 @@ const file_llmgw_v1_agent_proto_rawDesc = "" + "fabricTest\x122\n" + "\aupgrade\x18\n" + " \x01(\v2\x16.llmgw.v1.UpgradeAgentH\x00R\aupgrade\x12@\n" + - "\x0econtrol_planes\x18\v \x01(\v2\x17.llmgw.v1.ControlPlanesH\x00R\rcontrolPlanes\x129\n" + + "\x0econtrol_planes\x18\v \x01(\v2\x17.llmgw.v1.ControlPlanesH\x00R\rcontrolPlanes\x120\n" + + "\bfirewall\x18\r \x01(\v2\x12.llmgw.v1.FirewallH\x00R\bfirewall\x129\n" + "\x05trace\x18\f \x03(\v2#.llmgw.v1.ControlMessage.TraceEntryR\x05trace\x1a8\n" + "\n" + "TraceEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\x05\n" + - "\x03msg\")\n" + + "\x03msg\"\x94\x01\n" + + "\bFirewall\x12\x18\n" + + "\aenabled\x18\x01 \x01(\bR\aenabled\x12\x1a\n" + + "\bnetworks\x18\x02 \x03(\tR\bnetworks\x12*\n" + + "\x11engine_ports_from\x18\x03 \x01(\x05R\x0fenginePortsFrom\x12&\n" + + "\x0fengine_ports_to\x18\x04 \x01(\x05R\renginePortsTo\")\n" + "\rControlPlanes\x12\x18\n" + "\aservers\x18\x01 \x03(\tR\aservers\"\x80\a\n" + "\bRegister\x12\x17\n" + @@ -4835,118 +4933,120 @@ 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, 48) +var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 49) var file_llmgw_v1_agent_proto_goTypes = []any{ (ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State (*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage (*KVEvents)(nil), // 2: llmgw.v1.KVEvents (*ControlMessage)(nil), // 3: llmgw.v1.ControlMessage - (*ControlPlanes)(nil), // 4: llmgw.v1.ControlPlanes - (*Register)(nil), // 5: llmgw.v1.Register - (*Host)(nil), // 6: llmgw.v1.Host - (*Storage)(nil), // 7: llmgw.v1.Storage - (*RunningShard)(nil), // 8: llmgw.v1.RunningShard - (*GPU)(nil), // 9: llmgw.v1.GPU - (*MIGDevice)(nil), // 10: llmgw.v1.MIGDevice - (*FabricAttachment)(nil), // 11: llmgw.v1.FabricAttachment - (*LocalWeights)(nil), // 12: llmgw.v1.LocalWeights - (*NodeLoad)(nil), // 13: llmgw.v1.NodeLoad - (*GPUSensor)(nil), // 14: llmgw.v1.GPUSensor - (*ShardLoad)(nil), // 15: llmgw.v1.ShardLoad - (*IBPort)(nil), // 16: llmgw.v1.IBPort - (*SubnetManager)(nil), // 17: llmgw.v1.SubnetManager - (*IMEX)(nil), // 18: llmgw.v1.IMEX - (*GPUFabricState)(nil), // 19: llmgw.v1.GPUFabricState - (*Heartbeat)(nil), // 20: llmgw.v1.Heartbeat - (*Provision)(nil), // 21: llmgw.v1.Provision - (*EngineWant)(nil), // 22: llmgw.v1.EngineWant - (*ProvisionStatus)(nil), // 23: llmgw.v1.ProvisionStatus - (*ShardStatus)(nil), // 24: llmgw.v1.ShardStatus - (*FetchProgress)(nil), // 25: llmgw.v1.FetchProgress - (*StartShard)(nil), // 26: llmgw.v1.StartShard - (*EngineProfile)(nil), // 27: llmgw.v1.EngineProfile - (*RemoveWeights)(nil), // 28: llmgw.v1.RemoveWeights - (*StopShard)(nil), // 29: llmgw.v1.StopShard - (*DrainShard)(nil), // 30: llmgw.v1.DrainShard - (*Ack)(nil), // 31: llmgw.v1.Ack - (*LogsRequest)(nil), // 32: llmgw.v1.LogsRequest - (*LogsReply)(nil), // 33: llmgw.v1.LogsReply - (*FabricScanRequest)(nil), // 34: llmgw.v1.FabricScanRequest - (*FabricScanReply)(nil), // 35: llmgw.v1.FabricScanReply - (*FabricTestRequest)(nil), // 36: llmgw.v1.FabricTestRequest - (*FabricTestReply)(nil), // 37: llmgw.v1.FabricTestReply - (*UpgradeAgent)(nil), // 38: llmgw.v1.UpgradeAgent - (*UpgradeReply)(nil), // 39: llmgw.v1.UpgradeReply - (*LoraAdapter)(nil), // 40: llmgw.v1.LoraAdapter - nil, // 41: llmgw.v1.ControlMessage.TraceEntry - nil, // 42: llmgw.v1.Register.LabelsEntry - nil, // 43: llmgw.v1.Register.EngineVersionsEntry - nil, // 44: llmgw.v1.GPU.NicPathsEntry - nil, // 45: llmgw.v1.IBPort.CountersEntry - nil, // 46: llmgw.v1.Heartbeat.EngineVersionsEntry - nil, // 47: llmgw.v1.EngineWant.OfflineSha256Entry - nil, // 48: llmgw.v1.EngineProfile.EnvEntry + (*Firewall)(nil), // 4: llmgw.v1.Firewall + (*ControlPlanes)(nil), // 5: llmgw.v1.ControlPlanes + (*Register)(nil), // 6: llmgw.v1.Register + (*Host)(nil), // 7: llmgw.v1.Host + (*Storage)(nil), // 8: llmgw.v1.Storage + (*RunningShard)(nil), // 9: llmgw.v1.RunningShard + (*GPU)(nil), // 10: llmgw.v1.GPU + (*MIGDevice)(nil), // 11: llmgw.v1.MIGDevice + (*FabricAttachment)(nil), // 12: llmgw.v1.FabricAttachment + (*LocalWeights)(nil), // 13: llmgw.v1.LocalWeights + (*NodeLoad)(nil), // 14: llmgw.v1.NodeLoad + (*GPUSensor)(nil), // 15: llmgw.v1.GPUSensor + (*ShardLoad)(nil), // 16: llmgw.v1.ShardLoad + (*IBPort)(nil), // 17: llmgw.v1.IBPort + (*SubnetManager)(nil), // 18: llmgw.v1.SubnetManager + (*IMEX)(nil), // 19: llmgw.v1.IMEX + (*GPUFabricState)(nil), // 20: llmgw.v1.GPUFabricState + (*Heartbeat)(nil), // 21: llmgw.v1.Heartbeat + (*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 } var file_llmgw_v1_agent_proto_depIdxs = []int32{ - 5, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register - 20, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat - 24, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus - 33, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply - 35, // 4: llmgw.v1.AgentMessage.fabric_scan:type_name -> llmgw.v1.FabricScanReply - 37, // 5: llmgw.v1.AgentMessage.fabric_test:type_name -> llmgw.v1.FabricTestReply - 39, // 6: llmgw.v1.AgentMessage.upgrade:type_name -> llmgw.v1.UpgradeReply + 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 2, // 7: llmgw.v1.AgentMessage.kv_events:type_name -> llmgw.v1.KVEvents - 26, // 8: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard - 29, // 9: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard - 30, // 10: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard - 31, // 11: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack - 28, // 12: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights - 21, // 13: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision - 32, // 14: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest - 34, // 15: llmgw.v1.ControlMessage.fabric_scan:type_name -> llmgw.v1.FabricScanRequest - 36, // 16: llmgw.v1.ControlMessage.fabric_test:type_name -> llmgw.v1.FabricTestRequest - 38, // 17: llmgw.v1.ControlMessage.upgrade:type_name -> llmgw.v1.UpgradeAgent - 4, // 18: llmgw.v1.ControlMessage.control_planes:type_name -> llmgw.v1.ControlPlanes - 41, // 19: llmgw.v1.ControlMessage.trace:type_name -> llmgw.v1.ControlMessage.TraceEntry - 42, // 20: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry - 9, // 21: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU - 11, // 22: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment - 43, // 23: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry - 8, // 24: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard - 12, // 25: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights - 7, // 26: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage - 6, // 27: llmgw.v1.Register.host:type_name -> llmgw.v1.Host - 18, // 28: llmgw.v1.Register.imex:type_name -> llmgw.v1.IMEX - 0, // 29: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State - 44, // 30: llmgw.v1.GPU.nic_paths:type_name -> llmgw.v1.GPU.NicPathsEntry - 10, // 31: llmgw.v1.GPU.mig:type_name -> llmgw.v1.MIGDevice - 15, // 32: llmgw.v1.NodeLoad.shards:type_name -> llmgw.v1.ShardLoad - 14, // 33: llmgw.v1.NodeLoad.gpus:type_name -> llmgw.v1.GPUSensor - 45, // 34: llmgw.v1.IBPort.counters:type_name -> llmgw.v1.IBPort.CountersEntry - 17, // 35: llmgw.v1.IBPort.subnet_managers:type_name -> llmgw.v1.SubnetManager - 7, // 36: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage - 12, // 37: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights - 46, // 38: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry - 23, // 39: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus - 13, // 40: llmgw.v1.Heartbeat.load:type_name -> llmgw.v1.NodeLoad - 16, // 41: llmgw.v1.Heartbeat.ib_ports:type_name -> llmgw.v1.IBPort - 11, // 42: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment - 19, // 43: llmgw.v1.Heartbeat.gpu_fabric:type_name -> llmgw.v1.GPUFabricState - 18, // 44: llmgw.v1.Heartbeat.imex:type_name -> llmgw.v1.IMEX - 22, // 45: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant - 47, // 46: llmgw.v1.EngineWant.offline_sha256:type_name -> llmgw.v1.EngineWant.OfflineSha256Entry - 0, // 47: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State - 25, // 48: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress - 27, // 49: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile - 48, // 50: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry - 40, // 51: llmgw.v1.EngineProfile.adapters:type_name -> llmgw.v1.LoraAdapter - 1, // 52: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage - 3, // 53: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage - 53, // [53:54] is the sub-list for method output_type - 52, // [52:53] is the sub-list for method input_type - 52, // [52:52] is the sub-list for extension type_name - 52, // [52:52] is the sub-list for extension extendee - 0, // [0:52] is the sub-list for field type_name + 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 + 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 + 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 + 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 + 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 + 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 + 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 + 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 + 12, // 43: llmgw.v1.Heartbeat.fabrics:type_name -> llmgw.v1.FabricAttachment + 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 + 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 } func init() { file_llmgw_v1_agent_proto_init() } @@ -4976,6 +5076,7 @@ func file_llmgw_v1_agent_proto_init() { (*ControlMessage_FabricTest)(nil), (*ControlMessage_Upgrade)(nil), (*ControlMessage_ControlPlanes)(nil), + (*ControlMessage_Firewall)(nil), } type x struct{} out := protoimpl.TypeBuilder{ @@ -4983,7 +5084,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: 48, + NumMessages: 49, NumExtensions: 0, NumServices: 1, }, diff --git a/api/proto/llmgw/v1/agent.proto b/api/proto/llmgw/v1/agent.proto index 606399a..387cc20 100644 --- a/api/proto/llmgw/v1/agent.proto +++ b/api/proto/llmgw/v1/agent.proto @@ -71,6 +71,7 @@ message ControlMessage { FabricTestRequest fabric_test = 9; UpgradeAgent upgrade = 10; ControlPlanes control_planes = 11; + Firewall firewall = 13; } // trace is the W3C trace context (traceparent, and tracestate when there // is one) of the control plane's span that sent this command, so what the @@ -80,6 +81,21 @@ message ControlMessage { map trace = 12; } +// Firewall is whether llmgw's firewall is kept on the node — its nftables +// table, which lets the cluster's networks alone reach the engines' ports and +// the node exporter — sent when an agent registers and whenever the +// cluster's settings change it (firewall.enabled). +message Firewall { + bool enabled = 1; + // networks are let in besides those the node reaches the control planes + // on: firewall_networks, networks or single addresses. + repeated string networks = 2; + // engine_ports_from and engine_ports_to are the engines' ports + // (control.port_range). + int32 engine_ports_from = 3; + int32 engine_ports_to = 4; +} + // ControlPlanes is every control plane an agent may dial, sent by the leader // when an agent registers and whenever the list changes. The agent keeps it // beside its configuration, so a second control plane added after the node diff --git a/docs/console.md b/docs/console.md index 7833faf..e9cb6ca 100644 --- a/docs/console.md +++ b/docs/console.md @@ -319,6 +319,9 @@ to restart it. - **Restart all replicas**, one at a time. - **Back up now** — the store and `/etc/llmgw` into one archive (`sudo llmgw server backup`). +- **Wipe this machine** — removes llmgw's services, configuration, state and + users from this machine, after listing them and asking for its name + (`sudo llmgw wipe`; on a node too). - **Leave**. ## The node's console diff --git a/docs/install.md b/docs/install.md index 222bb18..9d5d780 100644 --- a/docs/install.md +++ b/docs/install.md @@ -2142,6 +2142,22 @@ for nodes behind a router, goes in `server.yaml`: firewall_networks: [10.0.5.0/24, 192.168.1.20] ``` +**Turning it off and on.** Whether llmgw keeps this firewall is one of the +cluster's settings: + +``` +llmgw ctl config set firewall.enabled false # every machine's table taken out +llmgw ctl config set firewall.enabled true # and put back as it was +``` + +It is in force within seconds: each node's agent takes its table out or puts +it back, and each control plane's `llmgw-firewall-sync.path` (installed by +`server init` and `server add`; `sudo llmgw firewall sync --install` on one +set up before) does the same there. The rules files stay, so turning it back +on restores them. Nodes added while it is off get none. A separate mirror +machine and the store's members keep theirs for now. `firewall_networks` +changes reach the nodes the same way. + Only those ports are touched. The table's policy is to accept, so whatever else the machine serves — and whatever firewall it already has — is left as it was; the client API (`:8080`) is not closed, being meant to be reached. @@ -2278,6 +2294,31 @@ A cluster set up online becomes offline the same way: `bundle import` on the mirror and control plane, then the `offline:` block in `server.yaml` and a restart. +## Removing llmgw from a machine + +``` +sudo llmgw wipe # services, configuration, certificates, state, users +sudo llmgw wipe --models # and model weights, the mirror's files, engines +sudo llmgw wipe --store # and the store's database here (Patroni member, etcd, or local PostgreSQL) +sudo llmgw wipe --all # all of that, backups and the binary +``` + +It lists everything it will remove and asks for the machine's name (`--yes` +does not ask), then stops llmgw's services and the engines, takes the +firewall table out, and deletes `/etc/llmgw` — the authority and secrets +with it — `/var/lib/llmgw`, the agent's state, the sudo grant, the +authority from the system's trust, keepalived's configuration when llmgw +wrote it, and the `llmgw` and `llmgw-engine` users. Packages it installed +(PostgreSQL, Patroni, etcd, nftables, the node exporter) stay. Run it again +to finish what failed. A node wiped is then forgotten on the control plane +with `llmgw node remove `. The console has it too: **Do → Wipe this +machine**. + +This is how a machine with an older install is set up again from nothing: +wipe it, then install as above. Wiping a control plane without `--store` +leaves the store with the cluster's accounts, keys and settings in it, and a +new control plane against that store takes them up. + ## Kubernetes The control plane runs in Kubernetes from the Helm chart in diff --git a/docs/reference.md b/docs/reference.md index d1e9e58..1965dac 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -98,6 +98,7 @@ rather than repeated. | `other_watts` | `float64` | [FacilityBudget](#facilitybudget) | | `name` | `string` | [Federation](#federation) | | `interval` | `time.Duration` | [Federation](#federation) | +| `enabled` | `bool` | [FirewallSettings](#firewallsettings) | | `format` | `string` | [GuardModel](#guardmodel) | | `categories` | `[]string` | [GuardModel](#guardmodel) | | `model` | `string` | [JudgePolicy](#judgepolicy) | @@ -287,6 +288,7 @@ rather than repeated. | `guard` | `guard.Config` | [Server](#server) | | `secrets` | `Secrets` | [Server](#server) | | `notify` | `notify.Config` | [Server](#server) | +| `firewall` | `FirewallSettings` | [Server](#server) | | `firewall_networks` | `[]string` | [Server](#server) | | `backup_recipients` | `[]string` | [Server](#server) | | `backup_copy_to` | `[]string` | [Server](#server) | @@ -963,6 +965,19 @@ Interval is how often each peer is asked whether it is ready, timed, and what it offers. A peer that does not answer is taken out of routing until it does. Zero is 10s. +## FirewallSettings + +### `enabled` + +`bool` + +Enabled keeps llmgw's internal ports — engines, node exporter, the +agents' stream, the mirror, the store — to the cluster's networks on +every machine. False removes llmgw's nftables table everywhere, for a +site whose own firewall does that job; the rules are kept, and true +puts them back. Changed in the cluster's settings, it is in force on +every node and control plane within seconds. + ## GuardModel ### `format` @@ -2441,6 +2456,14 @@ Prometheus and Alertmanager in between. Empty sends nothing. Takes a [notify.Config](#notifyconfig) block. +### `firewall` + +`FirewallSettings` + +Firewall turns llmgw's firewall on every machine on or off. + +Takes a [FirewallSettings](#firewallsettings) block. + ### `firewall_networks` `[]string` diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 8e3ebd4..e067ee7 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -36,6 +36,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights" "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" + "gitea.mynodes.xyz/ilker/llmgw/internal/firewall" "gitea.mynodes.xyz/ilker/llmgw/internal/kvevents" "gitea.mynodes.xyz/ilker/llmgw/internal/obs" "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" @@ -87,7 +88,11 @@ type Agent struct { // serverList is every control plane to dial: the configured ones, and // any the leader has named since. Read through servers(). serverList []string - inv discovery.Inventory + // fwMu keeps one firewall change at a time; fwRunner runs its scripts + // (nil: this machine's shell). + fwMu sync.Mutex + fwRunner firewall.Runner + inv discovery.Inventory // shardReplica remembers which replica a shard belongs to, so a status // report can name both. The control plane keys on the replica. shardReplica map[string]string @@ -854,6 +859,8 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha go a.upgrade(ctx, m.Upgrade, out) case *llmgwv1.ControlMessage_ControlPlanes: a.setControlPlanes(m.ControlPlanes) + case *llmgwv1.ControlMessage_Firewall: + go a.applyFirewall(m.Firewall) case *llmgwv1.ControlMessage_Logs: reply := a.logsReply(m.Logs) // Off the receive loop: a full send queue must not stop commands being diff --git a/internal/agent/firewall.go b/internal/agent/firewall.go new file mode 100644 index 0000000..e1e324a --- /dev/null +++ b/internal/agent/firewall.go @@ -0,0 +1,89 @@ +package agent + +import ( + "context" + "net" + "net/netip" + "net/url" + "strings" + "time" + + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" + "gitea.mynodes.xyz/ilker/llmgw/internal/firewall" +) + +// applyFirewall keeps llmgw's firewall on this node as the cluster's +// settings say: the engines' ports and the node exporter's for the networks +// the node reaches its control planes on, and those named, or no table at +// all. Off the receive loop, one at a time. +func (a *Agent) applyFirewall(m *llmgwv1.Firewall) { + a.fwMu.Lock() + defer a.fwMu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + r := a.fwRunner + if r == nil { + r = firewall.Local{} + } + if !m.GetEnabled() { + had, err := firewall.Disable(ctx, r) + switch { + case err != nil: + a.log.Warn("could not take llmgw's firewall off this node", "err", err) + case had: + a.log.Info("llmgw's firewall is off on this node, as the cluster's settings say") + } + return + } + a.mu.Lock() + servers := append([]string(nil), a.serverList...) + a.mu.Unlock() + nets, err := firewall.Networks(ctx, r, controlPlaneIPs(ctx, servers)) + if err != nil { + a.log.Warn("could not tell which networks reach the control planes; the firewall is left as it is", "err", err) + return + } + for _, n := range m.GetNetworks() { + if p, err := netip.ParsePrefix(n); err == nil { + nets = append(nets, p) + } else if ad, err := netip.ParseAddr(n); err == nil { + nets = append(nets, netip.PrefixFrom(ad, ad.BitLen())) + } + } + from, to := int(m.GetEnginePortsFrom()), int(m.GetEnginePortsTo()) + ports := []firewall.Port{firewall.NodeExporter} + if from > 0 && to >= from { + ports = append(ports, firewall.Engines(from, to)) + } + changed, err := firewall.Install(ctx, r, firewall.Spec{Ports: ports, Allow: nets}) + switch { + case err != nil: + a.log.Warn("could not put llmgw's firewall on this node", "err", err) + case changed: + a.log.Info("llmgw's firewall is on this node", "networks", len(nets)) + } +} + +// controlPlaneIPs are the addresses of the control planes an agent dials. +func controlPlaneIPs(ctx context.Context, servers []string) []string { + var out []string + for _, s := range servers { + host := s + if u, err := url.Parse(s); err == nil && u.Host != "" { + host = u.Hostname() + } + host = strings.Trim(host, "[]") + if a, err := netip.ParseAddr(host); err == nil { + out = append(out, a.String()) + continue + } + ips, err := net.DefaultResolver.LookupNetIP(ctx, "ip", host) + if err != nil { + continue + } + for _, ip := range ips { + out = append(out, ip.Unmap().String()) + } + } + return out +} diff --git a/internal/agent/firewall_test.go b/internal/agent/firewall_test.go new file mode 100644 index 0000000..dd58f6e --- /dev/null +++ b/internal/agent/firewall_test.go @@ -0,0 +1,57 @@ +package agent + +import ( + "context" + "errors" + "log/slog" + "strings" + "testing" + + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" +) + +// fakeShell answers the firewall package's scripts and records what was +// installed or removed. +type fakeShell struct { + present bool + installed string + deleted bool +} + +func (f *fakeShell) Run(_ context.Context, script string, stdin []byte) ([]byte, error) { + switch { + case strings.HasPrefix(script, "ip -o route get 10.0.0.140"): + return []byte("10.0.0.140 dev eth0 src 10.0.0.141 uid 0\n2: eth0 inet 10.0.0.141/24 brd 10.0.0.255 scope global eth0\n"), nil + case strings.HasPrefix(script, "sudo -n /usr/sbin/nft list table"): + if f.present { + return nil, nil + } + return nil, errors.New("no table") + case strings.Contains(script, "nft delete table") && !strings.Contains(script, "cat > "): + f.deleted, f.present = true, false + return nil, nil + case strings.HasPrefix(script, "sudo cat "): + return nil, nil + default: + f.installed, f.present = string(stdin), true + return []byte("changed\n"), nil + } +} + +func TestANodeKeepsOrDropsItsFirewallAsTheClusterSays(t *testing.T) { + sh := &fakeShell{} + a := &Agent{log: slog.New(slog.DiscardHandler), fwRunner: sh, serverList: []string{"https://10.0.0.140:7443"}} + + a.applyFirewall(&llmgwv1.Firewall{Enabled: true, Networks: []string{"192.168.5.0/24", "10.9.9.9"}, + EnginePortsFrom: 20000, EnginePortsTo: 20999}) + for _, want := range []string{"10.0.0.0/24", "192.168.5.0/24", "10.9.9.9/32", "20000-20999", "9100"} { + if !strings.Contains(sh.installed, want) { + t.Errorf("the rules lack %s:\n%s", want, sh.installed) + } + } + + a.applyFirewall(&llmgwv1.Firewall{Enabled: false}) + if !sh.deleted { + t.Error("turned off, the table stayed") + } +} diff --git a/internal/cli/clustersettings.go b/internal/cli/clustersettings.go index 446e377..c474670 100644 --- a/internal/cli/clustersettings.go +++ b/internal/cli/clustersettings.go @@ -237,8 +237,10 @@ func (w *settingsWatch) check(ctx context.Context) { // again: what it reads at every use, or not at all. var liveSettings = []string{ "guard", + // Sent to the nodes and this machine's firewall unit as they change. + "firewall", "firewall_networks", // Read by server init, node add and backups, not by the server. - "firewall_networks", "backup_recipients", "backup_copy_to", + "backup_recipients", "backup_copy_to", } // onlyLive is whether every changed key is one of liveSettings. diff --git a/internal/cli/firewallcmd.go b/internal/cli/firewallcmd.go new file mode 100644 index 0000000..405fae7 --- /dev/null +++ b/internal/cli/firewallcmd.go @@ -0,0 +1,176 @@ +package cli + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/netip" + "os" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/config" + "gitea.mynodes.xyz/ilker/llmgw/internal/firewall" +) + +// firewallStateFile is where a control plane writes what the cluster's +// settings say of the firewall, for llmgw-firewall-sync to apply as root: +// the server itself runs as llmgw and cannot change the ruleset. +const firewallStateFile = "/var/lib/llmgw/firewall.json" + +// firewallState is what is written there. +type firewallState struct { + Enabled bool `json:"enabled"` + Networks []string `json:"networks,omitempty"` +} + +// writeFirewallState records the settings' firewall for this machine's sync +// unit. Without the state directory — a server not run by systemd — there +// is no unit to read it, and nothing is written. +func writeFirewallState(cfg config.Server) error { + if _, err := os.Stat("/var/lib/llmgw"); err != nil { + return nil //nolint:nilerr // not a control plane set up by init + } + b, _ := json.Marshal(firewallState{Enabled: cfg.Firewall.Enabled, Networks: cfg.FirewallNetworks}) + if old, err := os.ReadFile(firewallStateFile); err == nil && string(old) == string(b) { + return nil + } + tmp := firewallStateFile + ".new" + if err := os.WriteFile(tmp, b, 0o644); err != nil { + return err + } + return os.Rename(tmp, firewallStateFile) +} + +// The units that apply it: a path unit watching the file, and the oneshot it +// starts. +const ( + firewallSyncPathUnit = "llmgw-firewall-sync.path" + firewallSyncUnitFile = `[Unit] +Description=llmgw firewall: apply the cluster's settings on this control plane + +[Service] +Type=oneshot +ExecStart=/usr/local/bin/llmgw firewall sync +` + firewallSyncPathFile = `[Unit] +Description=llmgw firewall: follow the cluster's settings on this control plane + +[Path] +PathChanged=` + firewallStateFile + ` + +[Install] +WantedBy=multi-user.target +` +) + +func newFirewallCmd() *cobra.Command { + c := &cobra.Command{ + Use: "firewall", + Short: "llmgw's firewall on this machine", + Long: "Whether llmgw keeps a firewall is the cluster's setting firewall.enabled\n" + + "(llmgw ctl config set firewall.enabled false). Nodes are told by the\n" + + "control plane; a control plane writes it to " + firewallStateFile + ",\n" + + "which " + firewallSyncPathUnit + " applies with 'llmgw firewall sync'.", + } + var install bool + sync := &cobra.Command{ + Use: "sync", + Short: "Apply what the cluster's settings say of the firewall, here (root)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if os.Geteuid() != 0 { + return errors.New("the firewall is changed as root: run it with sudo") + } + if install { + if err := installFirewallSync(cmd.Context(), firewall.Local{}); err != nil { + return err + } + fmt.Fprintln(cmd.OutOrStdout(), firewallSyncPathUnit+" follows the cluster's settings here") + if _, err := os.Stat(firewallStateFile); err != nil { + return nil // the server writes it when it starts + } + } + b, err := os.ReadFile(firewallStateFile) + if err != nil { + return err + } + var st firewallState + if err := json.Unmarshal(b, &st); err != nil { + return fmt.Errorf("%s: %w", firewallStateFile, err) + } + ctx, cancel := context.WithTimeout(cmd.Context(), 2*time.Minute) + defer cancel() + msg, err := syncFirewall(ctx, firewall.Local{}, st) + if err != nil { + return err + } + fmt.Fprintln(cmd.OutOrStdout(), msg) + return nil + }, + } + sync.Flags().BoolVar(&install, "install", false, "first install the units that run this whenever the settings change") + c.AddCommand(sync) + return c +} + +// installFirewallSync writes the sync units on a machine and has the path +// unit follow the state file. +func installFirewallSync(ctx context.Context, r firewall.Runner) error { + body := firewallSyncUnitFile + "\x00" + firewallSyncPathFile + out, err := r.Run(ctx, `set -e +t=$(mktemp -d); trap 'rm -rf "$t"' EXIT +cat > "$t/all" +python3 - "$t" <<'PY' +import sys +d = sys.argv[1] +a, b = open(d + "/all", "rb").read().split(b"\x00") +open(d + "/service", "wb").write(a) +open(d + "/path", "wb").write(b) +PY +sudo install -m 644 "$t/service" /etc/systemd/system/llmgw-firewall-sync.service +sudo install -m 644 "$t/path" /etc/systemd/system/`+firewallSyncPathUnit+` +sudo systemctl daemon-reload +sudo systemctl enable --now `+firewallSyncPathUnit+` >/dev/null 2>&1`, []byte(body)) + if err != nil { + return fmt.Errorf("installing %s: %v: %s", firewallSyncPathUnit, err, string(out)) + } + return nil +} + +// syncFirewall puts a firewall state in force on this machine. +func syncFirewall(ctx context.Context, r firewall.Runner, st firewallState) (string, error) { + if !st.Enabled { + had, err := firewall.Disable(ctx, r) + if err != nil { + return "", err + } + if had { + return "llmgw's firewall is off here", nil + } + return "llmgw's firewall was already off here", nil + } + loaded, err := firewall.Enable(ctx, r) + if err != nil { + return "", err + } + if !loaded { + return "no llmgw rules on this machine to put back; 'llmgw server init' writes them", nil + } + var extra []netip.Prefix + for _, n := range st.Networks { + if p, err := netip.ParsePrefix(n); err == nil { + extra = append(extra, p) + } else if a, err := netip.ParseAddr(n); err == nil { + extra = append(extra, netip.PrefixFrom(a, a.BitLen())) + } + } + if len(extra) > 0 { + if _, err := firewall.Install(ctx, r, firewall.Spec{Allow: extra}); err != nil { + return "", err + } + } + return "llmgw's firewall is on here", nil +} diff --git a/internal/cli/firewallcmd_test.go b/internal/cli/firewallcmd_test.go new file mode 100644 index 0000000..4a05ffc --- /dev/null +++ b/internal/cli/firewallcmd_test.go @@ -0,0 +1,51 @@ +package cli + +import ( + "context" + "strings" + "testing" +) + +type scriptLog struct { + scripts []string + reply func(script string) string +} + +func (s *scriptLog) Run(_ context.Context, script string, _ []byte) ([]byte, error) { + s.scripts = append(s.scripts, script) + if s.reply != nil { + return []byte(s.reply(script)), nil + } + return nil, nil +} + +func TestAControlPlaneFollowsTheFirewallSetting(t *testing.T) { + off := &scriptLog{} + if _, err := syncFirewall(context.Background(), off, firewallState{Enabled: false}); err != nil { + t.Fatal(err) + } + if !strings.Contains(strings.Join(off.scripts, "\n"), "nft delete table inet llmgw") { + t.Errorf("off did not take the table out: %q", off.scripts) + } + + none := &scriptLog{reply: func(s string) string { + if strings.Contains(s, "echo loaded") { + return "none\n" + } + return "" + }} + msg, err := syncFirewall(context.Background(), none, firewallState{Enabled: true}) + if err != nil || !strings.Contains(msg, "server init") { + t.Errorf("on, with no rules here: %q %v", msg, err) + } + + on := &scriptLog{reply: func(s string) string { + if strings.Contains(s, "echo loaded") { + return "loaded\n" + } + return "" + }} + if msg, err := syncFirewall(context.Background(), on, firewallState{Enabled: true}); err != nil || msg != "llmgw's firewall is on here" { + t.Errorf("on: %q %v", msg, err) + } +} diff --git a/internal/cli/node.go b/internal/cli/node.go index 940ed3a..8f35330 100644 --- a/internal/cli/node.go +++ b/internal/cli/node.go @@ -174,8 +174,11 @@ func (na *nodeAdd) prepare() error { } } } - // The node's ports — its engines', its metrics — for the cluster only. - if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil && !o.NoFirewall { + // The node's ports — its engines', its metrics — for the cluster only, + // unless the cluster's settings turned llmgw's firewall off. + if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil && !cfg.Firewall.Enabled { + o.NoFirewall = true + } else if err == nil && !o.NoFirewall { from, to := cfg.EnginePorts() o.FirewallPorts = []firewall.Port{firewall.Engines(from, to), firewall.NodeExporter} o.FirewallAllow, _ = cfg.FirewallAllow() diff --git a/internal/cli/root.go b/internal/cli/root.go index 1f45211..0414b9e 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -59,6 +59,8 @@ func Root() *cobra.Command { newSelfUpgradeCmd(), newReleaseCmd(), newJoinCmd(), + newFirewallCmd(), + newWipeCmd(), newDocsCmd(), ) return root diff --git a/internal/cli/server.go b/internal/cli/server.go index e384518..0dc9798 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -223,6 +223,10 @@ func newServerCmd() *cobra.Command { // The control plane replaces the static endpoint list: agents // register, the reconciler places replicas, and Ready ones are // published to the router (plan §6.4). + // The settings in force, for what reads them at each use. + var current atomic.Pointer[config.Server] + current.Store(&cfg) + var agentSvc *agentsvc.Service var agentHandler http.Handler var agentPath string var admin *control.Controller @@ -247,6 +251,14 @@ func newServerCmd() *cobra.Command { return err } defer ctl.Close() + agentSvc = svc + // Whether nodes keep llmgw's firewall, as the settings say now. + svc.SetFirewall(func() *llmgwv1.Firewall { + c := current.Load() + from, to := c.EnginePorts() + return &llmgwv1.Firewall{Enabled: c.Firewall.Enabled, Networks: c.FirewallNetworks, + EnginePortsFrom: int32(from), EnginePortsTo: int32(to)} //nolint:gosec // ports + }) ctl.Reconciler().SetNotifier(notifier) admin = ctl agentPath, agentHandler = svc.Handler() @@ -653,11 +665,24 @@ func newServerCmd() *cobra.Command { // Every change to the settings in force: what can change live // does, the rest starts this process again, which systemd brings // back on them. + // And this machine's own firewall, through its sync unit. + if err := writeFirewallState(cfg); err != nil { + log.Warn("could not record the firewall's settings for this machine", "err", err) + } watch := &settingsWatch{path: configPath, st: rt.Store(), cur: boot, log: log, live: func(next config.Server, changed []string) bool { if !onlyLive(changed) { return false } + current.Store(&next) + if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "firewall") }) { + if err := writeFirewallState(next); err != nil { + log.Warn("could not record the firewall's settings for this machine", "err", err) + } + if agentSvc != nil { + agentSvc.BroadcastFirewall() + } + } if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "guard") }) { if gate == nil { // Built at start, and only when there are policies. diff --git a/internal/cli/serveradd.go b/internal/cli/serveradd.go index 7dff1fa..efb932c 100644 --- a/internal/cli/serveradd.go +++ b/internal/cli/serveradd.go @@ -159,6 +159,14 @@ func (sa *serverAdd) firewall(ctx context.Context) error { if err != nil { return err } + // What applies the cluster's firewall setting there when it changes. + if err := installFirewallSync(ctx, sa.remote); err != nil { + fmt.Fprintf(sa.out, " !! firewall: %v\n", err) + } + if !cfg.Firewall.Enabled { + sa.kept("firewall: off, as the cluster's settings say (firewall.enabled)") + return nil + } self, err := sourceIPToward(sa.ip) if err != nil { return err diff --git a/internal/cli/serverinit_firewall.go b/internal/cli/serverinit_firewall.go index a272327..baa2231 100644 --- a/internal/cli/serverinit_firewall.go +++ b/internal/cli/serverinit_firewall.go @@ -23,6 +23,15 @@ func (in *serverInit) ensureFirewall(ctx context.Context) error { if err != nil { return err } + // What applies the cluster's setting here when it changes: the server + // writes it, as llmgw; this applies it, as root. + if err := in.ensureFirewallSync(ctx); err != nil { + return err + } + if !cfg.Firewall.Enabled { + in.kept("firewall: off, as the cluster's settings say (firewall.enabled)") + return nil + } ports := controlPlanePorts(cfg) if !in.noMirror && in.mirrorURL == "" { ports = append(ports, firewall.MirrorFiles) @@ -122,3 +131,9 @@ func loopbackListen(listen string) bool { ip := net.ParseIP(host) return host == "localhost" || (ip != nil && ip.IsLoopback()) } + +// ensureFirewallSync installs the units that apply the cluster's firewall +// setting on this control plane. +func (in *serverInit) ensureFirewallSync(ctx context.Context) error { + return installFirewallSync(ctx, firewall.Local{}) +} diff --git a/internal/cli/tui_console.go b/internal/cli/tui_console.go index 1f88bbc..3d41670 100644 --- a/internal/cli/tui_console.go +++ b/internal/cli/tui_console.go @@ -62,13 +62,14 @@ const ( cUpgrade cRestartAll cBackup + cWipe cLeave cCount ) var consoleLabels = [cCount]string{"Overview", "Monitor", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Guardrails", "Federation", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Audit", "Manual", "server.yaml", "Add a node", "Add a control plane", "Replicate the store", "Serve a model", "Upgrade the cluster", - "Restart all replicas", "Back up now", "Leave"} + "Restart all replicas", "Back up now", "Wipe this machine", "Leave"} // consoleAction is what is done once the screen closes: each runs its own // screen, and the console comes back after. @@ -150,7 +151,7 @@ type consoleModel struct { saveNote string // A question open as a dialog. - dialog string // "", "target", "edit", "addname", "saved", "confirm-leave", "error" + dialog string // "", "target", "edit", "addname", "wipe", "saved", "note", "confirm-leave", "error" dialogTitle string dialogText string prompt textField @@ -407,7 +408,7 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { func (m consoleModel) entryOK(i int) bool { if m.node { - return i == cSettings || i == cManual || i == cLeave + return i == cSettings || i == cManual || i == cWipe || i == cLeave } return i != cReplicate || !m.storeReplicated() } @@ -493,6 +494,8 @@ func (m consoleModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) { return m.ask("Replicate the store", "user@address of database machine", 2) case cLeave: return m.leave() + case cWipe: + return m.askWipe() case cUpgrade, cRestartAll, cBackup: return m, m.doEntry() case cOverview: @@ -744,6 +747,13 @@ func (m consoleModel) dialogKey(k tea.KeyMsg) (tea.Model, tea.Cmd) { m.dialogText = floatingQuestion m.prompt.Placeholder = "10.0.0.161/24" } + case "wipe": + if host, _ := os.Hostname(); v != host { + m.dialogText = m.wipeQuestion() + "\n\n" + sBad.Render("that is not this machine's name; nothing is removed") + return m, nil + } + m.action = consoleAction{kind: "wipe"} + return m, tea.Quit case "edit": if err := m.doc.Set(m.editPath, m.editKind, m.prompt.Value()); err != nil { m.dialogText = m.editQuestion() + "\n\n" + sBad.Render(err.Error()) @@ -1055,7 +1065,7 @@ func (m consoleModel) View() string { nav = append(nav, navItem{Label: "", Heading: true}, navItem{Label: "Help", Heading: true}) case cSettings: nav = append(nav, navItem{Label: "", Heading: true}, navItem{Label: "Settings", Heading: true}) - case cAddNode, cLeave: + case cAddNode, cWipe: if (i == cAddNode) != m.node { nav = append(nav, navItem{Label: "", Heading: true}, navItem{Label: "Do", Heading: true}) } @@ -1096,7 +1106,7 @@ func (m consoleModel) View() string { f.DialogTitle = m.dialogTitle f.Dialog = m.dialogText switch m.dialog { - case "target", "edit", "addname": + case "target", "edit", "addname", "wipe": q := m.dialogText if m.dialog == "target" && m.editKind > 1 { q = fmt.Sprintf("%s %d of %d", q, len(m.asked)+1, int(m.editKind)) @@ -1298,6 +1308,9 @@ func (m consoleModel) panel() (title, body, keys string) { case cBackup: item = "\n Writes the store and /etc/llmgw to one archive in " + DefaultBackupDir + ",\n and keeps the last few. Copy it off this machine.\n\n " + sDim.Render("On the command line:") + "\n sudo llmgw server backup" + case cWipe: + item = "\n Removes llmgw from this machine: its services, firewall table,\n configuration, certificates, the authority, secrets, state and users,\n so it can be set up again from nothing. Model weights, engines, the\n store's database and backups stay. Asks for the machine's name first.\n\n " + + sDim.Render("On the command line, with --models, --store or --all for more:") + "\n sudo llmgw wipe" case cLeave: item = "\n Closes the console. The cluster keeps running." } @@ -1621,6 +1634,16 @@ func runConsole(ctx context.Context, out io.Writer, file string) error { // Already set up: only the store changes. in.noMirror, in.noNode, in.noModel = true, true, true err = in.runInitTUI(ctx, out) + case "wipe": + c := Root() + c.SetArgs([]string{"wipe", "--yes"}) + c.SetOut(out) + c.SetErr(out) + if err := c.ExecuteContext(ctx); err != nil { + fmt.Fprintln(out, sBad.Render("✗ ")+err.Error()) + } + // Nothing is left here for the console to show. + return nil case "run": c := Root() c.SetArgs(act.targets) @@ -1666,3 +1689,31 @@ func (m consoleModel) standbyNote() string { } return "\n " + sDim.Render("Standing by. Shown, and changed, on the leader: "+leader) + "\n" } + +// askWipe asks for this machine's name before removing llmgw from it, saying +// what would go. +func (m consoleModel) askWipe() (tea.Model, tea.Cmd) { + if os.Geteuid() != 0 { + m.dialog, m.dialogTitle = "error", "Wipe this machine" + m.dialogText = "Removing services, users and files under /etc needs root:\nstart the console with sudo llmgw." + return m, nil + } + m.dialog, m.dialogTitle, m.dialogText = "wipe", "Wipe this machine", m.wipeQuestion() + m.prompt = newTextField() + m.prompt.Width = 40 + m.prompt.Focus() + return m, nil +} + +func (m consoleModel) wipeQuestion() string { + host, _ := os.Hostname() + w := &wiper{} + var b strings.Builder + b.WriteString("This removes, on " + host + ":\n\n") + for _, s := range w.plan() { + b.WriteString(" - " + s.what + "\n") + } + b.WriteString("\nModel weights, engines, the store's database and backups stay\n" + + "(sudo llmgw wipe --all takes them too).\n\nType the machine's name to remove it all:") + return b.String() +} diff --git a/internal/cli/tui_setup_test.go b/internal/cli/tui_setup_test.go index 16ae936..714cfdc 100644 --- a/internal/cli/tui_setup_test.go +++ b/internal/cli/tui_setup_test.go @@ -300,3 +300,27 @@ func TestTheConsoleEditsTheClustersSettingsInTheStore(t *testing.T) { t.Errorf("the cluster's key went into the machine's file:\n%s", b) } } + +func TestTheConsoleWipesOnlyOnTheMachinesName(t *testing.T) { + m := newConsoleModel("127.0.0.1:1", filepath.Join(t.TempDir(), "server.yaml")) + m.width, m.height, m.loaded = 120, 40, true + m.menu = cWipe + m.dialog, m.dialogTitle, m.dialogText = "wipe", "Wipe this machine", m.wipeQuestion() + m.prompt = newTextField() + m.prompt.Focus() + if v := m.View(); !strings.Contains(v, "Type the machine's name") { + t.Errorf("the dialog does not ask for the name:\n%s", v) + } + m.prompt.SetValue("not-this-one") + mm := press(m, "enter") + if a := mm.(consoleModel).action; a.kind != "" { + t.Fatalf("a wrong name wiped: %+v", a) + } + cm := mm.(consoleModel) + host, _ := os.Hostname() + cm.prompt.SetValue(host) + mm = press(cm, "enter") + if a := mm.(consoleModel).action; a.kind != "wipe" { + t.Errorf("the right name did not: %+v", a) + } +} diff --git a/internal/cli/wipe.go b/internal/cli/wipe.go new file mode 100644 index 0000000..50ed04b --- /dev/null +++ b/internal/cli/wipe.go @@ -0,0 +1,277 @@ +package cli + +import ( + "bufio" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + + "github.com/spf13/cobra" +) + +// wipeUnits are the systemd units llmgw installs, stopped first so nothing +// writes while the rest goes. +var wipeUnits = []string{ + "llmgw-server.service", "llmgw-agent.service", "llmgw-mirror.service", + firewallSyncPathUnit, "llmgw-firewall-sync.service", "llmgw-firewall.service", + "llmgw-backup.timer", "llmgw-backup.service", +} + +// wiper removes llmgw from a machine. root prefixes every path, for tests; +// run runs a command. +type wiper struct { + root string + run func(name string, args ...string) error + store bool + models bool + all bool +} + +// wipeStep is one thing to remove, said before it is done. +type wipeStep struct { + what string + do func() error +} + +func (w *wiper) path(p string) string { return filepath.Join(w.root, p) } + +func (w *wiper) exists(p string) bool { + _, err := os.Lstat(w.path(p)) + return err == nil +} + +func (w *wiper) glob(pattern string) []string { + m, _ := filepath.Glob(w.path(pattern)) + out := make([]string, len(m)) + for i, p := range m { + out[i] = strings.TrimPrefix(p, w.root) + } + return out +} + +// mentions is whether a file is there and says llmgw: a file of a name +// another tool also uses (keepalived's, etcd's) is removed only when llmgw +// wrote it. +func (w *wiper) mentions(p string) bool { + b, err := os.ReadFile(w.path(p)) + return err == nil && strings.Contains(string(b), "llmgw") +} + +func (w *wiper) remove(paths ...string) func() error { + return func() error { + var errs []error + for _, p := range paths { + if err := os.RemoveAll(w.path(p)); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) + } +} + +// plan is every step, in order, for what is on this machine. +func (w *wiper) plan() []wipeStep { + var steps []wipeStep + add := func(what string, do func() error) { steps = append(steps, wipeStep{what, do}) } + files := func(what string, paths ...string) { + var there []string + for _, p := range paths { + if w.exists(p) { + there = append(there, p) + } + } + if len(there) > 0 { + add(what+": "+strings.Join(there, ", "), w.remove(there...)) + } + } + + // The services, stopped and gone. + var units []string + for _, u := range wipeUnits { + if w.exists("/etc/systemd/system/" + u) { + units = append(units, u) + } + } + if len(units) > 0 { + add("stop and remove "+strings.Join(units, ", "), func() error { + for _, u := range units { + _ = w.run("systemctl", "disable", "--now", u) + if err := os.Remove(w.path("/etc/systemd/system/" + u)); err != nil && !os.IsNotExist(err) { + return err + } + } + return w.run("systemctl", "daemon-reload") + }) + } + // Engines run as their own user, kept alive across the agent's restarts: + // stopped by user, or they would keep their GPUs. + if w.userExists("llmgw-engine") { + add("stop the engines (every process of llmgw-engine)", func() error { + _ = w.run("pkill", "-KILL", "-u", "llmgw-engine") + return nil + }) + } + add("take llmgw's table out of the firewall (nft inet llmgw), if loaded", func() error { + _ = w.run("nft", "delete", "table", "inet", "llmgw") + return nil + }) + if w.mentions("/etc/keepalived/keepalived.conf") { + add("stop keepalived and remove llmgw's floating address (/etc/keepalived/keepalived.conf)", func() error { + _ = w.run("systemctl", "disable", "--now", "keepalived") + return w.remove("/etc/keepalived/keepalived.conf")() + }) + } + + files("configuration, certificates, the authority and secrets", "/etc/llmgw") + files("state: the admin token, a SQLite store, what the agent kept", + "/var/lib/llmgw", "/var/lib/llmgw-agent") + files("the sudo grant setup left", "/etc/sudoers.d/llmgw-setup") + if w.exists("/usr/local/share/ca-certificates/llmgw-ca.crt") { + add("the cluster's authority from the system's trust (llmgw-ca.crt)", func() error { + if err := w.remove("/usr/local/share/ca-certificates/llmgw-ca.crt")(); err != nil { + return err + } + _ = w.run("update-ca-certificates", "--fresh") + return nil + }) + } + if w.exists("/etc/pki/ca-trust/source/anchors/llmgw-ca.pem") { + add("the cluster's authority from the system's trust (llmgw-ca.pem)", func() error { + if err := w.remove("/etc/pki/ca-trust/source/anchors/llmgw-ca.pem")(); err != nil { + return err + } + _ = w.run("update-ca-trust") + return nil + }) + } + + if w.store || w.all { + w.planStore(add, files) + } + if w.models || w.all { + files("model weights and the mirror", "/srv/models", "/var/lib/llmgw-weights", "/var/lib/llmgw-models") + files("installed engines", "/opt/llmgw") + } + if w.all { + files("backups", DefaultBackupDir) + } + + for _, u := range []string{"llmgw-engine", "llmgw"} { + if w.userExists(u) { + add("the user "+u, func() error { return w.run("userdel", u) }) + } + } + if w.all { + files("the llmgw binary", "/usr/local/bin/llmgw", "/usr/local/bin/llmgw.prev") + } + return steps +} + +// planStore is the store, when it is on this machine: a member of the +// replicated one (Patroni, etcd), or llmgw's database in a local PostgreSQL. +func (w *wiper) planStore(add func(string, func() error), files func(string, ...string)) { + patroni := w.mentions("/etc/patroni/config.yml") + if patroni || w.mentions("/etc/default/etcd") { + add("stop Patroni and etcd", func() error { + _ = w.run("systemctl", "disable", "--now", "patroni") + _ = w.run("systemctl", "disable", "--now", "etcd") + return nil + }) + } + if patroni { + data := append(w.glob("/var/lib/postgresql/*/llmgw"), w.glob("/var/lib/pgsql/*/llmgw")...) + files("the store's database on this member, and Patroni's configuration", + append([]string{"/etc/patroni", "/var/lib/postgresql/.pgpass_patroni", "/root/llmgw-pg"}, data...)...) + } else if w.exists("/usr/bin/psql") { + add("drop the llmgw database and role from the local PostgreSQL, if there", func() error { + _ = w.run("runuser", "-u", "postgres", "--", "psql", "-q", + "-c", "DROP DATABASE IF EXISTS llmgw", "-c", "DROP ROLE IF EXISTS llmgw") + return nil + }) + } + if w.mentions("/etc/default/etcd") { + files("etcd's data and configuration", "/var/lib/etcd/default", "/etc/etcd/tls", "/etc/default/etcd") + } +} + +func (w *wiper) userExists(name string) bool { + b, err := os.ReadFile(w.path("/etc/passwd")) + return err == nil && strings.Contains("\n"+string(b), "\n"+name+":") +} + +func newWipeCmd() *cobra.Command { + var w wiper + var yes bool + c := &cobra.Command{ + Use: "wipe", + Short: "Remove llmgw from this machine: its services, configuration and state", + Long: "Stops and removes llmgw's services, takes its firewall table out, and\n" + + "deletes its configuration, certificates, the authority, secrets, state\n" + + "and users — what server init, node add, server add and mirror init left —\n" + + "so the machine can be set up again from nothing. It says everything it\n" + + "will remove first, and asks for the machine's name.\n\n" + + "Kept unless asked: model weights, the mirror and engines (--models), the\n" + + "store's database when it is on this machine (--store), backups and the\n" + + "binary (--all, which is all of these). Packages llmgw installed —\n" + + "PostgreSQL, Patroni, etcd, nftables, the node exporter — stay.\n\n" + + "A node wiped is then forgotten on the control plane with\n" + + "'llmgw node remove '.", + Example: " sudo llmgw wipe\n" + + " sudo llmgw wipe --models\n" + + " sudo llmgw wipe --all --yes", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if runtime.GOOS != "linux" { + return fmt.Errorf("wipe removes llmgw from a Linux machine; this is %s", runtime.GOOS) + } + if os.Geteuid() != 0 { + return errors.New("wipe removes services, users and files under /etc: run it with sudo") + } + out := cmd.OutOrStdout() + w.run = func(name string, args ...string) error { + return exec.CommandContext(cmd.Context(), name, args...).Run() //nolint:gosec // fixed commands + } + steps := w.plan() + host, _ := os.Hostname() + fmt.Fprintf(out, "On %s, this removes:\n", host) + for _, s := range steps { + fmt.Fprintf(out, " - %s\n", s.what) + } + if !yes { + if !isTerminal(out) { + return errors.New("nothing removed: give --yes to wipe without being asked") + } + fmt.Fprintf(out, "\nType this machine's name (%s) to remove all of it: ", host) + answer, _ := bufio.NewReader(cmd.InOrStdin()).ReadString('\n') + if strings.TrimSpace(answer) != host { + return errors.New("nothing removed") + } + } + var failed []string + for _, s := range steps { + if err := s.do(); err != nil { + failed = append(failed, s.what+": "+err.Error()) + fmt.Fprintf(out, " !! %s: %v\n", s.what, err) + continue + } + fmt.Fprintf(out, " ok %s\n", s.what) + } + if len(failed) > 0 { + return fmt.Errorf("%d step(s) failed; running wipe again finishes them", len(failed)) + } + fmt.Fprintln(out, "\nllmgw is gone from this machine. If it was a node, forget it on the control plane: "+ + "llmgw node remove "+host) + return nil + }, + } + f := c.Flags() + f.BoolVar(&yes, "yes", false, "do not ask") + f.BoolVar(&w.models, "models", false, "also remove model weights, the mirror's files and installed engines") + f.BoolVar(&w.store, "store", false, "also remove the store's database here: a Patroni member's data, etcd's, or llmgw's database in a local PostgreSQL") + f.BoolVar(&w.all, "all", false, "everything: --models, --store, backups and the binary") + return c +} diff --git a/internal/cli/wipe_test.go b/internal/cli/wipe_test.go new file mode 100644 index 0000000..bb15f67 --- /dev/null +++ b/internal/cli/wipe_test.go @@ -0,0 +1,106 @@ +package cli + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func wipeFixture(t *testing.T) string { + t.Helper() + root := t.TempDir() + for p, body := range map[string]string{ + "/etc/llmgw/server.yaml": "listen: :8080\n", + "/etc/llmgw/pki/ca-key.pem": "key", + "/var/lib/llmgw/admin-token": "t", + "/var/lib/llmgw-agent/control-planes.json": "[]", + "/etc/systemd/system/llmgw-server.service": "[Unit]", + "/etc/systemd/system/llmgw-agent.service": "[Unit]", + "/etc/sudoers.d/llmgw-setup": "x", + "/etc/keepalived/keepalived.conf": "# llmgw: the gateway's floating address\n", + "/etc/passwd": "root:x:0:0::/root:/bin/sh\nllmgw:x:999:999::/var/lib/llmgw:/usr/sbin/nologin\nllmgw-engine:x:998:998::/:/usr/sbin/nologin\n", + "/srv/models/qwen/v1/q.gguf": "w", + "/opt/llmgw/engines/llamacpp/b1/llama-server": "e", + "/etc/patroni/config.yml": "scope: llmgw\n", + "/var/lib/postgresql/16/llmgw/PG_VERSION": "16", + "/etc/default/etcd": "ETCD_INITIAL_CLUSTER_TOKEN=llmgw\n", + "/var/lib/etcd/default/member/snap": "s", + "/usr/local/bin/llmgw": "bin", + } { + full := filepath.Join(root, p) + if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(full, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + } + return root +} + +func TestWipeRemovesTheConfigurationAndKeepsWhatWasNotAsked(t *testing.T) { + root := wipeFixture(t) + var ran []string + w := &wiper{root: root, run: func(name string, args ...string) error { + ran = append(ran, name+" "+strings.Join(args, " ")) + return nil + }} + for _, s := range w.plan() { + if err := s.do(); err != nil { + t.Fatalf("%s: %v", s.what, err) + } + } + for _, gone := range []string{"/etc/llmgw", "/var/lib/llmgw", "/var/lib/llmgw-agent", + "/etc/systemd/system/llmgw-server.service", "/etc/sudoers.d/llmgw-setup", "/etc/keepalived/keepalived.conf"} { + if _, err := os.Stat(filepath.Join(root, gone)); err == nil { + t.Errorf("%s is still there", gone) + } + } + for _, kept := range []string{"/srv/models/qwen/v1/q.gguf", "/opt/llmgw/engines", "/etc/patroni/config.yml", + "/var/lib/postgresql/16/llmgw", "/var/lib/etcd/default", "/usr/local/bin/llmgw"} { + if _, err := os.Stat(filepath.Join(root, kept)); err != nil { + t.Errorf("%s was removed without being asked", kept) + } + } + all := strings.Join(ran, "\n") + for _, want := range []string{"systemctl disable --now llmgw-server.service", "pkill -KILL -u llmgw-engine", + "nft delete table inet llmgw", "systemctl disable --now keepalived", "userdel llmgw-engine", "userdel llmgw"} { + if !strings.Contains(all, want) { + t.Errorf("did not run %q:\n%s", want, all) + } + } + if strings.Contains(all, "patroni") { + t.Error("the store was touched without --store") + } +} + +func TestWipeAllTakesTheStoreAndTheModelsToo(t *testing.T) { + root := wipeFixture(t) + w := &wiper{root: root, all: true, run: func(string, ...string) error { return nil }} + for _, s := range w.plan() { + if err := s.do(); err != nil { + t.Fatalf("%s: %v", s.what, err) + } + } + for _, gone := range []string{"/srv/models", "/opt/llmgw", "/etc/patroni", "/var/lib/postgresql/16/llmgw", + "/var/lib/etcd/default", "/etc/default/etcd", "/usr/local/bin/llmgw"} { + if _, err := os.Stat(filepath.Join(root, gone)); err == nil { + t.Errorf("%s is still there", gone) + } + } +} + +func TestWipeLeavesAKeepalivedItDidNotWrite(t *testing.T) { + root := t.TempDir() + p := filepath.Join(root, "/etc/keepalived/keepalived.conf") + _ = os.MkdirAll(filepath.Dir(p), 0o755) + _ = os.WriteFile(p, []byte("vrrp_instance site_vip {}\n"), 0o600) + w := &wiper{root: root, run: func(string, ...string) error { return nil }} + for _, s := range w.plan() { + _ = s.do() + } + if _, err := os.Stat(p); err != nil { + t.Error("someone else's keepalived configuration was removed") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 6356a7b..d8cc39d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -146,6 +146,9 @@ type Server struct { // Prometheus and Alertmanager in between. Empty sends nothing. Notify notify.Config `koanf:"notify"` + // Firewall turns llmgw's firewall on every machine on or off. + Firewall FirewallSettings `koanf:"firewall"` + // FirewallNetworks are networks, besides those each machine reaches the // cluster on, that llmgw's internal ports answer: a monitoring host on // another subnet, nodes behind a router. Written into every machine's @@ -763,6 +766,7 @@ func Default() Server { AgentListen: ":7443", MaxConnections: 10_000, Mirror: Mirror{UpdateCheck: 6 * time.Hour}, + Firewall: FirewallSettings{Enabled: true}, Control: Control{ Interval: 5 * time.Second, BaseBackoff: 10 * time.Second, diff --git a/internal/config/firewall.go b/internal/config/firewall.go index 86d6147..505dfac 100644 --- a/internal/config/firewall.go +++ b/internal/config/firewall.go @@ -32,3 +32,15 @@ func (s Server) EnginePorts() (int, int) { d := Default().Control.PortRange return d[0], d[1] } + +// FirewallSettings is whether llmgw keeps a firewall on the cluster's +// machines. +type FirewallSettings struct { + // Enabled keeps llmgw's internal ports — engines, node exporter, the + // agents' stream, the mirror, the store — to the cluster's networks on + // every machine. False removes llmgw's nftables table everywhere, for a + // site whose own firewall does that job; the rules are kept, and true + // puts them back. Changed in the cluster's settings, it is in force on + // every node and control plane within seconds. + Enabled bool `koanf:"enabled"` +} diff --git a/internal/control/agentsvc/firewall.go b/internal/control/agentsvc/firewall.go new file mode 100644 index 0000000..7d10377 --- /dev/null +++ b/internal/control/agentsvc/firewall.go @@ -0,0 +1,42 @@ +package agentsvc + +import ( + llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" +) + +// SetFirewall gives the service whether nodes keep llmgw's firewall, read +// at each use so a change to the cluster's settings is what is sent next. +func (s *Service) SetFirewall(f func() *llmgwv1.Firewall) { + s.mu.Lock() + s.firewall = f + s.mu.Unlock() +} + +func (s *Service) firewallMsg() *llmgwv1.ControlMessage { + s.mu.RLock() + f := s.firewall + s.mu.RUnlock() + if f == nil { + return nil + } + fw := f() + if fw == nil { + return nil + } + return &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_Firewall{Firewall: fw}} +} + +// BroadcastFirewall tells every connected agent what the cluster's settings +// now say of the firewall. A node not connected is told when it registers. +func (s *Service) BroadcastFirewall() { + msg := s.firewallMsg() + if msg == nil { + return + } + for _, node := range s.ConnectedNodes() { + if err := s.send(node, msg); err != nil { + s.log.Warn("could not tell a node about the firewall; it is told when it next registers", + "node", node, "err", err) + } + } +} diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index 533dc39..9fdf1f3 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -75,6 +75,8 @@ type Service struct { // controlPlanes lists what agents may dial; see SetControlPlanes. controlPlanes func(context.Context) ([]string, error) + // firewall is what nodes are told of llmgw's firewall; see SetFirewall. + firewall func() *llmgwv1.Firewall // logWaits are the log requests awaiting a reply, by request id. logMu sync.Mutex @@ -286,6 +288,12 @@ func (s *Service) Connect( return err } } + // Whether to keep llmgw's firewall, as the cluster's settings say now. + if msg := s.firewallMsg(); msg != nil { + if err := stream.Send(msg); err != nil { + return err + } + } // Recover replica state from what this node is already running, before any // reconciliation pass can conclude the cluster is empty and place diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 6b4c73e..e9d04ab 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -117,6 +117,9 @@ var Reference = []Block{ {Key: "name", Type: "string", Doc: "Name is this cluster as a peer reads it, in the federated_from tag\nsent with a request to a peer set to receive tags (`ctl federation add\n--send-tags`). Empty is this machine's host name.", Nested: ""}, {Key: "interval", Type: "time.Duration", Doc: "Interval is how often each peer is asked whether it is ready, timed,\nand what it offers. A peer that does not answer is taken out of\nrouting until it does. Zero is 10s.", Nested: ""}, }}, + {Package: "config", Name: "FirewallSettings", Title: "FirewallSettings", Doc: "", Fields: []Field{ + {Key: "enabled", Type: "bool", Doc: "Enabled keeps llmgw's internal ports — engines, node exporter, the\nagents' stream, the mirror, the store — to the cluster's networks on\nevery machine. False removes llmgw's nftables table everywhere, for a\nsite whose own firewall does that job; the rules are kept, and true\nputs them back. Changed in the cluster's settings, it is in force on\nevery node and control plane within seconds.", Nested: ""}, + }}, {Package: "catalog", Name: "GuardModel", Title: "GuardModel", Doc: "", Fields: []Field{ {Key: "format", Type: "string", Doc: "Format is llamaguard3 (Meta's Llama Guard 3: \"safe\", or \"unsafe\" and\nthe categories), shieldgemma (Google's ShieldGemma: Yes or No), or\njson ({\"safe\": false, \"categories\": [\"...\"]}).", Nested: ""}, {Key: "categories", Type: "[]string", Doc: "Categories the model is asked about, for shieldgemma — each a policy\nsentence — and the names reported for llamaguard3's S1, S2, ….", Nested: ""}, @@ -363,6 +366,7 @@ var Reference = []Block{ {Key: "guard", Type: "guard.Config", Doc: "Guard is the library of guardrail policies: what requests and answers\nare checked for, and what is done. Applied to every request through\nguard.default, and to parts of the tenancy tree by attaching them\n(llmgw ctl guard attach). Empty checks nothing.", Nested: "guard.Config"}, {Key: "secrets", Type: "Secrets", Doc: "Secrets is where secrets kept outside files are read from, and the key\nthe store's own secrets are sealed under.", Nested: "config.Secrets"}, {Key: "notify", Type: "notify.Config", Doc: "Notify sends what needs a person — a model parked, a node lost, a\nbudget spent, new weights published — to webhooks, signed, without a\nPrometheus and Alertmanager in between. Empty sends nothing.", Nested: "notify.Config"}, + {Key: "firewall", Type: "FirewallSettings", Doc: "Firewall turns llmgw's firewall on every machine on or off.", Nested: "config.FirewallSettings"}, {Key: "firewall_networks", Type: "[]string", Doc: "FirewallNetworks are networks, besides those each machine reaches the\ncluster on, that llmgw's internal ports answer: a monitoring host on\nanother subnet, nodes behind a router. Written into every machine's\nfirewall by `server init`, `node add` and `server add`.", Nested: ""}, {Key: "backup_recipients", Type: "[]string", Doc: "BackupRecipients encrypt every archive `llmgw server backup` writes, to\nwhoever holds one of their keys: age recipients (age1…) or SSH public\nkeys (ssh-ed25519 …, ssh-rsa …), as `age` takes them. An archive holds\nthe certificate authority's key and every secret in the store, and is\ncopied off the machine; encrypted, the copy is safe wherever it lands.\nEmpty writes archives unencrypted, readable by root only.", Nested: ""}, {Key: "backup_copy_to", Type: "[]string", Doc: "BackupCopyTo is where each archive is copied once written, so it\nsurvives this machine: a directory (a mounted share, another disk) or\nssh://user@host[:port]/path, reached with BackupCopyKey and checked\nagainst /etc/llmgw/backup_known_hosts, which learns a host the first\ntime. The same number of archives is kept there as here. Refused\nwithout backup_recipients: an archive leaves the machine encrypted.", Nested: ""}, diff --git a/internal/docgen/render.go b/internal/docgen/render.go index a686027..bb92a9f 100644 --- a/internal/docgen/render.go +++ b/internal/docgen/render.go @@ -30,6 +30,7 @@ var Sources = []string{ "internal/guard/policy.go", "internal/guard/judge.go", "internal/config/oidc.go", + "internal/config/firewall.go", "internal/notify/notify.go", "internal/secrets/secrets.go", "internal/obs/trace.go", diff --git a/internal/firewall/firewall.go b/internal/firewall/firewall.go index e70468d..ac99506 100644 --- a/internal/firewall/firewall.go +++ b/internal/firewall/firewall.go @@ -11,10 +11,13 @@ package firewall import ( + "bytes" "context" "errors" "fmt" "net/netip" + "os" + "os/exec" "slices" "strconv" "strings" @@ -353,3 +356,51 @@ func Admit(ctx context.Context, r Runner, addrs []string) ([]netip.Prefix, error } return nets, nil } + +// Disable takes llmgw's table out of a machine's ruleset and stops it being +// loaded at boot, keeping the rules file: Install, or enabling the unit, +// puts it back as it was. Reports whether there was anything to take out. +func Disable(ctx context.Context, r Runner) (bool, error) { + had := Present(ctx, r) + out, err := r.Run(ctx, `sudo systemctl disable --now `+Unit+` >/dev/null 2>&1 || true +if sudo /usr/sbin/nft list table inet `+Table+` >/dev/null 2>&1; then + sudo /usr/sbin/nft delete table inet `+Table+` +fi`, nil) + if err != nil { + return false, fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out))) + } + return had, nil +} + +// Enable loads the rules file a machine already has and has it loaded at +// boot again, after Disable. Reports false when there is no rules file. +func Enable(ctx context.Context, r Runner) (bool, error) { + out, err := r.Run(ctx, `[ -f `+RulesFile+` ] || { echo none; exit 0; } +sudo systemctl enable `+Unit+` >/dev/null 2>&1 +sudo /usr/sbin/nft -f `+RulesFile+` +sudo systemctl start `+Unit+` >/dev/null 2>&1 || true +echo loaded`, nil) + if err != nil { + return false, fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out))) + } + return strings.TrimSpace(string(out)) == "loaded", nil +} + +// Local runs scripts on this machine. As root, sudo in them runs its command +// directly, for a machine that has no sudo. +type Local struct{} + +// Run runs script with sh. +func (Local) Run(ctx context.Context, script string, stdin []byte) ([]byte, error) { + if os.Geteuid() == 0 { + script = "sudo() { \"$@\"; }\n" + script + } + cmd := exec.CommandContext(ctx, "sh", "-c", script) + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + var out bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &out + err := cmd.Run() + return out.Bytes(), err +}