Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
40c3d83110 | ||
|
|
069d334652 | ||
|
|
4865a96ec2 | ||
|
|
2a381ad669 | ||
|
|
746d4852af | ||
|
|
33e678ab84 |
No files matched your search
@@ -55,6 +55,13 @@ jobs:
|
||||
(cd dist && sha256sum "${name}" >> checksums.txt)
|
||||
done
|
||||
|
||||
# The reverse proxy configurations, so a release page carries
|
||||
# everything install.md asks for and the repository is not needed.
|
||||
for f in deploy/proxy/Caddyfile deploy/proxy/nginx.conf; do
|
||||
cp "$f" dist/
|
||||
(cd dist && sha256sum "$(basename "$f")" >> checksums.txt)
|
||||
done
|
||||
|
||||
# An NVML build for nodes with NVLink, which the portable binary
|
||||
# cannot serve.
|
||||
#
|
||||
|
||||
@@ -18,6 +18,55 @@ is only released once its **exit criterion** is met, not when its code lands.
|
||||
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
|
||||
APIs change between minors without a deprecation period.
|
||||
|
||||
## v0.8.6 — 2026-09-28
|
||||
|
||||
### Reverse proxy configurations
|
||||
|
||||
`deploy/proxy/` has a Caddyfile and an nginx configuration for the client
|
||||
port, and both are attached to every release beside the binaries. They do not
|
||||
buffer streamed replies, their read timeout outlasts a model's wait before its
|
||||
first token, their body limit matches the router's, and they set
|
||||
`X-Forwarded-Proto` and `X-Forwarded-Host` themselves — llmgw marks session
|
||||
cookies `Secure` from the first and shows the second as the base URL on its
|
||||
landing page.
|
||||
|
||||
## v0.8.5 — 2026-09-28
|
||||
|
||||
### A control plane in one command
|
||||
|
||||
```
|
||||
sudo llmgw server init --admin you@example.com
|
||||
```
|
||||
|
||||
Creates the llmgw user and its directories, the certificate authority, the
|
||||
control plane's certificate for the machine's hostname and every address it
|
||||
has, `server.yaml`, the default project, and an administrator whose password is
|
||||
generated and printed once — then installs the systemd unit, starts it and
|
||||
waits until it answers. What used to be two steps of install.md is now this.
|
||||
|
||||
Each step is skipped when its result exists, so running it again finishes an
|
||||
install that stopped and changes nothing on one that did not. `--host` adds an
|
||||
address the machine cannot see of itself (a DNS alias, a NAT or floating
|
||||
address) and reissues the certificate keeping the old ones. It never overwrites
|
||||
`server.yaml` or an authority, and certificates issued by an authority kept on
|
||||
another machine are left as they are.
|
||||
|
||||
### Fixes
|
||||
|
||||
- **The configuration install.md showed could not start.** An empty catalog
|
||||
without `control.enabled: true` is refused as having nothing to serve. The
|
||||
documentation and the file `init` writes both set it.
|
||||
|
||||
## v0.8.4 — 2026-09-28
|
||||
|
||||
### Node logs on the admin page
|
||||
|
||||
Each node on the cluster page links to its logs — the agent's own, and the
|
||||
output of every engine running there — with an option to follow them. `llmgw
|
||||
ctl logs NODE` reads the same. They come over the agent's existing stream, so
|
||||
no ssh, jump host or root is needed to see what a node did. The agent keeps its
|
||||
last 2000 lines in memory; each engine its last few hundred.
|
||||
|
||||
## v0.8.3 — 2026-09-28
|
||||
|
||||
Found by building a node from nothing with v0.8.2.
|
||||
|
||||
@@ -66,6 +66,8 @@ on a machine that has them.
|
||||
does not exist or miss one that does
|
||||
- [deploy/mirror/](deploy/mirror/) — a model mirror, which is the cheapest
|
||||
and most useful thing to build around llmgw
|
||||
- [deploy/proxy/](deploy/proxy/) — Caddy and nginx in front of the client
|
||||
port, with streaming and the forwarded headers right
|
||||
- [deploy/ha/](deploy/ha/) — running two control planes, and the one health
|
||||
check that has to be right
|
||||
- [docs/node-onboarding.md](docs/node-onboarding.md),
|
||||
|
||||
+234
-42
@@ -101,6 +101,7 @@ type AgentMessage struct {
|
||||
// *AgentMessage_Register
|
||||
// *AgentMessage_Heartbeat
|
||||
// *AgentMessage_Shard
|
||||
// *AgentMessage_Logs
|
||||
Msg isAgentMessage_Msg `protobuf_oneof:"msg"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
@@ -170,6 +171,15 @@ func (x *AgentMessage) GetShard() *ShardStatus {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *AgentMessage) GetLogs() *LogsReply {
|
||||
if x != nil {
|
||||
if x, ok := x.Msg.(*AgentMessage_Logs); ok {
|
||||
return x.Logs
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type isAgentMessage_Msg interface {
|
||||
isAgentMessage_Msg()
|
||||
}
|
||||
@@ -186,12 +196,18 @@ type AgentMessage_Shard struct {
|
||||
Shard *ShardStatus `protobuf:"bytes,3,opt,name=shard,proto3,oneof"`
|
||||
}
|
||||
|
||||
type AgentMessage_Logs struct {
|
||||
Logs *LogsReply `protobuf:"bytes,4,opt,name=logs,proto3,oneof"`
|
||||
}
|
||||
|
||||
func (*AgentMessage_Register) isAgentMessage_Msg() {}
|
||||
|
||||
func (*AgentMessage_Heartbeat) isAgentMessage_Msg() {}
|
||||
|
||||
func (*AgentMessage_Shard) isAgentMessage_Msg() {}
|
||||
|
||||
func (*AgentMessage_Logs) isAgentMessage_Msg() {}
|
||||
|
||||
// ControlMessage is anything the control plane tells a node.
|
||||
type ControlMessage struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
@@ -203,6 +219,7 @@ type ControlMessage struct {
|
||||
// *ControlMessage_Ack
|
||||
// *ControlMessage_RemoveWeights
|
||||
// *ControlMessage_Provision
|
||||
// *ControlMessage_Logs
|
||||
Msg isControlMessage_Msg `protobuf_oneof:"msg"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
@@ -299,6 +316,15 @@ func (x *ControlMessage) GetProvision() *Provision {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *ControlMessage) GetLogs() *LogsRequest {
|
||||
if x != nil {
|
||||
if x, ok := x.Msg.(*ControlMessage_Logs); ok {
|
||||
return x.Logs
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type isControlMessage_Msg interface {
|
||||
isControlMessage_Msg()
|
||||
}
|
||||
@@ -327,6 +353,10 @@ type ControlMessage_Provision struct {
|
||||
Provision *Provision `protobuf:"bytes,6,opt,name=provision,proto3,oneof"`
|
||||
}
|
||||
|
||||
type ControlMessage_Logs struct {
|
||||
Logs *LogsRequest `protobuf:"bytes,7,opt,name=logs,proto3,oneof"`
|
||||
}
|
||||
|
||||
func (*ControlMessage_Start) isControlMessage_Msg() {}
|
||||
|
||||
func (*ControlMessage_Stop) isControlMessage_Msg() {}
|
||||
@@ -339,6 +369,8 @@ func (*ControlMessage_RemoveWeights) isControlMessage_Msg() {}
|
||||
|
||||
func (*ControlMessage_Provision) isControlMessage_Msg() {}
|
||||
|
||||
func (*ControlMessage_Logs) isControlMessage_Msg() {}
|
||||
|
||||
// Register is the first message on every stream. It carries the node's full
|
||||
// hardware inventory, so the control plane needs no separate discovery call and
|
||||
// a reconnecting agent automatically refreshes what the scheduler knows.
|
||||
@@ -2168,23 +2200,166 @@ func (x *Ack) GetError() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// LogsRequest asks a node for the recent lines of one of its logs.
|
||||
//
|
||||
// Over the stream the agent already holds, because that is the one way into a
|
||||
// node every deployment has: the admin page otherwise means ssh through a jump
|
||||
// host to a machine whose address nobody remembers, to read a journal only
|
||||
// root may open.
|
||||
type LogsRequest struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
// request_id pairs the reply with its request; several may be in flight.
|
||||
RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
|
||||
// source is "agent" for the agent's own log, or a shard id for that
|
||||
// engine's output.
|
||||
Source string `protobuf:"bytes,2,opt,name=source,proto3" json:"source,omitempty"`
|
||||
Lines int32 `protobuf:"varint,3,opt,name=lines,proto3" json:"lines,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *LogsRequest) Reset() {
|
||||
*x = LogsRequest{}
|
||||
mi := &file_llmgw_v1_agent_proto_msgTypes[21]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *LogsRequest) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*LogsRequest) ProtoMessage() {}
|
||||
|
||||
func (x *LogsRequest) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_llmgw_v1_agent_proto_msgTypes[21]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use LogsRequest.ProtoReflect.Descriptor instead.
|
||||
func (*LogsRequest) Descriptor() ([]byte, []int) {
|
||||
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{21}
|
||||
}
|
||||
|
||||
func (x *LogsRequest) GetRequestId() string {
|
||||
if x != nil {
|
||||
return x.RequestId
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *LogsRequest) GetSource() string {
|
||||
if x != nil {
|
||||
return x.Source
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *LogsRequest) GetLines() int32 {
|
||||
if x != nil {
|
||||
return x.Lines
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// LogsReply answers a LogsRequest.
|
||||
type LogsReply struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
|
||||
Lines []string `protobuf:"bytes,2,rep,name=lines,proto3" json:"lines,omitempty"`
|
||||
// sources are every log this node can show, so a page can offer them
|
||||
// without knowing what runs there.
|
||||
Sources []string `protobuf:"bytes,3,rep,name=sources,proto3" json:"sources,omitempty"`
|
||||
// error says why the lines are absent: an unknown source, say.
|
||||
Error string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *LogsReply) Reset() {
|
||||
*x = LogsReply{}
|
||||
mi := &file_llmgw_v1_agent_proto_msgTypes[22]
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
|
||||
func (x *LogsReply) String() string {
|
||||
return protoimpl.X.MessageStringOf(x)
|
||||
}
|
||||
|
||||
func (*LogsReply) ProtoMessage() {}
|
||||
|
||||
func (x *LogsReply) ProtoReflect() protoreflect.Message {
|
||||
mi := &file_llmgw_v1_agent_proto_msgTypes[22]
|
||||
if x != nil {
|
||||
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
|
||||
if ms.LoadMessageInfo() == nil {
|
||||
ms.StoreMessageInfo(mi)
|
||||
}
|
||||
return ms
|
||||
}
|
||||
return mi.MessageOf(x)
|
||||
}
|
||||
|
||||
// Deprecated: Use LogsReply.ProtoReflect.Descriptor instead.
|
||||
func (*LogsReply) Descriptor() ([]byte, []int) {
|
||||
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{22}
|
||||
}
|
||||
|
||||
func (x *LogsReply) GetRequestId() string {
|
||||
if x != nil {
|
||||
return x.RequestId
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *LogsReply) GetLines() []string {
|
||||
if x != nil {
|
||||
return x.Lines
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *LogsReply) GetSources() []string {
|
||||
if x != nil {
|
||||
return x.Sources
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *LogsReply) GetError() string {
|
||||
if x != nil {
|
||||
return x.Error
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
var File_llmgw_v1_agent_proto protoreflect.FileDescriptor
|
||||
|
||||
const file_llmgw_v1_agent_proto_rawDesc = "" +
|
||||
"\n" +
|
||||
"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xab\x01\n" +
|
||||
"\x14llmgw/v1/agent.proto\x12\bllmgw.v1\"\xd6\x01\n" +
|
||||
"\fAgentMessage\x120\n" +
|
||||
"\bregister\x18\x01 \x01(\v2\x12.llmgw.v1.RegisterH\x00R\bregister\x123\n" +
|
||||
"\theartbeat\x18\x02 \x01(\v2\x13.llmgw.v1.HeartbeatH\x00R\theartbeat\x12-\n" +
|
||||
"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shardB\x05\n" +
|
||||
"\x03msg\"\xb8\x02\n" +
|
||||
"\x05shard\x18\x03 \x01(\v2\x15.llmgw.v1.ShardStatusH\x00R\x05shard\x12)\n" +
|
||||
"\x04logs\x18\x04 \x01(\v2\x13.llmgw.v1.LogsReplyH\x00R\x04logsB\x05\n" +
|
||||
"\x03msg\"\xe5\x02\n" +
|
||||
"\x0eControlMessage\x12,\n" +
|
||||
"\x05start\x18\x01 \x01(\v2\x14.llmgw.v1.StartShardH\x00R\x05start\x12)\n" +
|
||||
"\x04stop\x18\x02 \x01(\v2\x13.llmgw.v1.StopShardH\x00R\x04stop\x12,\n" +
|
||||
"\x05drain\x18\x03 \x01(\v2\x14.llmgw.v1.DrainShardH\x00R\x05drain\x12!\n" +
|
||||
"\x03ack\x18\x04 \x01(\v2\r.llmgw.v1.AckH\x00R\x03ack\x12@\n" +
|
||||
"\x0eremove_weights\x18\x05 \x01(\v2\x17.llmgw.v1.RemoveWeightsH\x00R\rremoveWeights\x123\n" +
|
||||
"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovisionB\x05\n" +
|
||||
"\tprovision\x18\x06 \x01(\v2\x13.llmgw.v1.ProvisionH\x00R\tprovision\x12+\n" +
|
||||
"\x04logs\x18\a \x01(\v2\x15.llmgw.v1.LogsRequestH\x00R\x04logsB\x05\n" +
|
||||
"\x03msg\"\xf2\x05\n" +
|
||||
"\bRegister\x12\x17\n" +
|
||||
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" +
|
||||
@@ -2379,7 +2554,18 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
|
||||
"\n" +
|
||||
"message_id\x18\x01 \x01(\tR\tmessageId\x12\x0e\n" +
|
||||
"\x02ok\x18\x02 \x01(\bR\x02ok\x12\x14\n" +
|
||||
"\x05error\x18\x03 \x01(\tR\x05error2O\n" +
|
||||
"\x05error\x18\x03 \x01(\tR\x05error\"Z\n" +
|
||||
"\vLogsRequest\x12\x1d\n" +
|
||||
"\n" +
|
||||
"request_id\x18\x01 \x01(\tR\trequestId\x12\x16\n" +
|
||||
"\x06source\x18\x02 \x01(\tR\x06source\x12\x14\n" +
|
||||
"\x05lines\x18\x03 \x01(\x05R\x05lines\"p\n" +
|
||||
"\tLogsReply\x12\x1d\n" +
|
||||
"\n" +
|
||||
"request_id\x18\x01 \x01(\tR\trequestId\x12\x14\n" +
|
||||
"\x05lines\x18\x02 \x03(\tR\x05lines\x12\x18\n" +
|
||||
"\asources\x18\x03 \x03(\tR\asources\x12\x14\n" +
|
||||
"\x05error\x18\x04 \x01(\tR\x05error2O\n" +
|
||||
"\fAgentService\x12?\n" +
|
||||
"\aConnect\x12\x16.llmgw.v1.AgentMessage\x1a\x18.llmgw.v1.ControlMessage(\x010\x01B\x93\x01\n" +
|
||||
"\fcom.llmgw.v1B\n" +
|
||||
@@ -2398,7 +2584,7 @@ func file_llmgw_v1_agent_proto_rawDescGZIP() []byte {
|
||||
}
|
||||
|
||||
var file_llmgw_v1_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
|
||||
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
|
||||
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 27)
|
||||
var file_llmgw_v1_agent_proto_goTypes = []any{
|
||||
(ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State
|
||||
(*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage
|
||||
@@ -2422,46 +2608,50 @@ var file_llmgw_v1_agent_proto_goTypes = []any{
|
||||
(*StopShard)(nil), // 19: llmgw.v1.StopShard
|
||||
(*DrainShard)(nil), // 20: llmgw.v1.DrainShard
|
||||
(*Ack)(nil), // 21: llmgw.v1.Ack
|
||||
nil, // 22: llmgw.v1.Register.LabelsEntry
|
||||
nil, // 23: llmgw.v1.Register.EngineVersionsEntry
|
||||
nil, // 24: llmgw.v1.Heartbeat.EngineVersionsEntry
|
||||
nil, // 25: llmgw.v1.EngineProfile.EnvEntry
|
||||
(*LogsRequest)(nil), // 22: llmgw.v1.LogsRequest
|
||||
(*LogsReply)(nil), // 23: llmgw.v1.LogsReply
|
||||
nil, // 24: llmgw.v1.Register.LabelsEntry
|
||||
nil, // 25: llmgw.v1.Register.EngineVersionsEntry
|
||||
nil, // 26: llmgw.v1.Heartbeat.EngineVersionsEntry
|
||||
nil, // 27: llmgw.v1.EngineProfile.EnvEntry
|
||||
}
|
||||
var file_llmgw_v1_agent_proto_depIdxs = []int32{
|
||||
3, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
|
||||
10, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat
|
||||
14, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus
|
||||
16, // 3: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
|
||||
19, // 4: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
|
||||
20, // 5: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
|
||||
21, // 6: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
|
||||
18, // 7: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
|
||||
11, // 8: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
|
||||
22, // 9: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
|
||||
7, // 10: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
|
||||
8, // 11: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
|
||||
23, // 12: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
|
||||
6, // 13: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
|
||||
9, // 14: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
|
||||
5, // 15: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
|
||||
4, // 16: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
|
||||
0, // 17: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
|
||||
5, // 18: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
|
||||
9, // 19: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
|
||||
24, // 20: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
|
||||
13, // 21: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
|
||||
12, // 22: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
|
||||
0, // 23: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
|
||||
15, // 24: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
|
||||
17, // 25: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
|
||||
25, // 26: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
|
||||
1, // 27: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
|
||||
2, // 28: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
|
||||
28, // [28:29] is the sub-list for method output_type
|
||||
27, // [27:28] is the sub-list for method input_type
|
||||
27, // [27:27] is the sub-list for extension type_name
|
||||
27, // [27:27] is the sub-list for extension extendee
|
||||
0, // [0:27] is the sub-list for field type_name
|
||||
23, // 3: llmgw.v1.AgentMessage.logs:type_name -> llmgw.v1.LogsReply
|
||||
16, // 4: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
|
||||
19, // 5: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
|
||||
20, // 6: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
|
||||
21, // 7: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
|
||||
18, // 8: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
|
||||
11, // 9: llmgw.v1.ControlMessage.provision:type_name -> llmgw.v1.Provision
|
||||
22, // 10: llmgw.v1.ControlMessage.logs:type_name -> llmgw.v1.LogsRequest
|
||||
24, // 11: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
|
||||
7, // 12: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
|
||||
8, // 13: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
|
||||
25, // 14: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
|
||||
6, // 15: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
|
||||
9, // 16: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
|
||||
5, // 17: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
|
||||
4, // 18: llmgw.v1.Register.host:type_name -> llmgw.v1.Host
|
||||
0, // 19: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
|
||||
5, // 20: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
|
||||
9, // 21: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
|
||||
26, // 22: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
|
||||
13, // 23: llmgw.v1.Heartbeat.provision:type_name -> llmgw.v1.ProvisionStatus
|
||||
12, // 24: llmgw.v1.Provision.engines:type_name -> llmgw.v1.EngineWant
|
||||
0, // 25: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
|
||||
15, // 26: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
|
||||
17, // 27: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
|
||||
27, // 28: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
|
||||
1, // 29: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
|
||||
2, // 30: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
|
||||
30, // [30:31] is the sub-list for method output_type
|
||||
29, // [29:30] is the sub-list for method input_type
|
||||
29, // [29:29] is the sub-list for extension type_name
|
||||
29, // [29:29] is the sub-list for extension extendee
|
||||
0, // [0:29] is the sub-list for field type_name
|
||||
}
|
||||
|
||||
func init() { file_llmgw_v1_agent_proto_init() }
|
||||
@@ -2473,6 +2663,7 @@ func file_llmgw_v1_agent_proto_init() {
|
||||
(*AgentMessage_Register)(nil),
|
||||
(*AgentMessage_Heartbeat)(nil),
|
||||
(*AgentMessage_Shard)(nil),
|
||||
(*AgentMessage_Logs)(nil),
|
||||
}
|
||||
file_llmgw_v1_agent_proto_msgTypes[1].OneofWrappers = []any{
|
||||
(*ControlMessage_Start)(nil),
|
||||
@@ -2481,6 +2672,7 @@ func file_llmgw_v1_agent_proto_init() {
|
||||
(*ControlMessage_Ack)(nil),
|
||||
(*ControlMessage_RemoveWeights)(nil),
|
||||
(*ControlMessage_Provision)(nil),
|
||||
(*ControlMessage_Logs)(nil),
|
||||
}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
@@ -2488,7 +2680,7 @@ func file_llmgw_v1_agent_proto_init() {
|
||||
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
|
||||
RawDescriptor: unsafe.Slice(unsafe.StringData(file_llmgw_v1_agent_proto_rawDesc), len(file_llmgw_v1_agent_proto_rawDesc)),
|
||||
NumEnums: 1,
|
||||
NumMessages: 25,
|
||||
NumMessages: 27,
|
||||
NumExtensions: 0,
|
||||
NumServices: 1,
|
||||
},
|
||||
|
||||
@@ -23,6 +23,7 @@ message AgentMessage {
|
||||
Register register = 1;
|
||||
Heartbeat heartbeat = 2;
|
||||
ShardStatus shard = 3;
|
||||
LogsReply logs = 4;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,6 +36,7 @@ message ControlMessage {
|
||||
Ack ack = 4;
|
||||
RemoveWeights remove_weights = 5;
|
||||
Provision provision = 6;
|
||||
LogsRequest logs = 7;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -418,3 +420,29 @@ message Ack {
|
||||
bool ok = 2;
|
||||
string error = 3;
|
||||
}
|
||||
|
||||
// LogsRequest asks a node for the recent lines of one of its logs.
|
||||
//
|
||||
// Over the stream the agent already holds, because that is the one way into a
|
||||
// node every deployment has: the admin page otherwise means ssh through a jump
|
||||
// host to a machine whose address nobody remembers, to read a journal only
|
||||
// root may open.
|
||||
message LogsRequest {
|
||||
// request_id pairs the reply with its request; several may be in flight.
|
||||
string request_id = 1;
|
||||
// source is "agent" for the agent's own log, or a shard id for that
|
||||
// engine's output.
|
||||
string source = 2;
|
||||
int32 lines = 3;
|
||||
}
|
||||
|
||||
// LogsReply answers a LogsRequest.
|
||||
message LogsReply {
|
||||
string request_id = 1;
|
||||
repeated string lines = 2;
|
||||
// sources are every log this node can show, so a page can offer them
|
||||
// without knowing what runs there.
|
||||
repeated string sources = 3;
|
||||
// error says why the lines are absent: an unknown source, say.
|
||||
string error = 4;
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
# llmgw behind Caddy: the client-facing port (8080) on a public name, with a
|
||||
# certificate Caddy obtains and renews by itself.
|
||||
#
|
||||
# Replace llm.example.com, point its DNS at this machine, open 80 and 443, and
|
||||
# run: caddy run --config /etc/caddy/Caddyfile
|
||||
#
|
||||
# Only 8080 goes through here. 7443 (agents) is mutual TLS with the cluster's
|
||||
# own authority and must reach llmgw directly; 9090 and 9091 stay private.
|
||||
|
||||
llm.example.com {
|
||||
# The router reads at most 32 MiB of body (router.max_body_bytes). Refusing
|
||||
# more here gives the client a 413 before the upload, not after it.
|
||||
request_body {
|
||||
max_size 32MB
|
||||
}
|
||||
|
||||
reverse_proxy 127.0.0.1:8080 {
|
||||
# Tokens are written the moment they are generated. Without this a
|
||||
# streamed reply can arrive in bursts, and a client measuring time to
|
||||
# first token measures the proxy instead.
|
||||
flush_interval -1
|
||||
|
||||
# Caddy sets X-Forwarded-Proto and X-Forwarded-Host itself and, with
|
||||
# no trusted_proxies, discards whatever the client sent. llmgw relies
|
||||
# on that: the proto decides whether session cookies are Secure, and
|
||||
# the host is the base URL the landing page tells people to use.
|
||||
# Behind a load balancer of your own, name it in trusted_proxies.
|
||||
|
||||
transport http {
|
||||
# A long generation can be silent for a while before the first
|
||||
# token. llmgw has its own limits (timeouts.*); this must not be
|
||||
# the shorter one.
|
||||
response_header_timeout 15m
|
||||
}
|
||||
}
|
||||
|
||||
# No `encode` on purpose: compressing a stream buffers it.
|
||||
|
||||
# Access log to stderr, which under systemd is the journal.
|
||||
log
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
# A reverse proxy in front of llmgw
|
||||
|
||||
The client-facing port (8080) is plain HTTP. On anything but a trusted network,
|
||||
put it behind a proxy that holds a publicly trusted certificate.
|
||||
|
||||
| File | For |
|
||||
| --- | --- |
|
||||
| [Caddyfile](Caddyfile) | Caddy — obtains and renews its own certificate; the least work |
|
||||
| [nginx.conf](nginx.conf) | nginx, with a certificate from certbot |
|
||||
|
||||
Both are also attached to every release.
|
||||
|
||||
Four things both files get right, and any other proxy has to as well:
|
||||
|
||||
- **Streaming is not buffered.** Replies are server-sent events; a buffering
|
||||
proxy delivers them in one burst at the end.
|
||||
- **The read timeout is longer than a model's silence.** nginx's default of 60
|
||||
seconds turns a slow first token into a 504.
|
||||
- **`X-Forwarded-Proto` and `X-Forwarded-Host` are set by the proxy**, replacing
|
||||
whatever the client sent. llmgw uses the first to mark session cookies
|
||||
`Secure` and the second for the base URL on its landing page.
|
||||
- **The body limit matches llmgw's** (32 MiB, `router.max_body_bytes`), so an
|
||||
oversized request is refused before it is uploaded.
|
||||
|
||||
Only port 8080 is proxied. Agents reach 7443 directly — it is mutual TLS with
|
||||
the cluster's own authority, and a proxy terminating it would take away the
|
||||
node identity it carries. 9090 (admin, pprof) stays on loopback; 9091
|
||||
(metrics) goes to your monitoring host only.
|
||||
|
||||
With a proxy on the same machine, bind llmgw to loopback so nothing reaches it
|
||||
around the proxy:
|
||||
|
||||
```yaml
|
||||
# /etc/llmgw/server.yaml
|
||||
listen: "127.0.0.1:8080"
|
||||
```
|
||||
@@ -0,0 +1,63 @@
|
||||
# llmgw behind nginx: the client-facing port (8080) on a public name.
|
||||
#
|
||||
# Put this in /etc/nginx/sites-available/llmgw (or conf.d/llmgw.conf), replace
|
||||
# llm.example.com, and get a certificate — certbot --nginx -d llm.example.com
|
||||
# fills in the ssl_ lines below. Then: nginx -t && systemctl reload nginx
|
||||
#
|
||||
# Only 8080 goes through here. 7443 (agents) is mutual TLS with the cluster's
|
||||
# own authority and must reach llmgw directly; 9090 and 9091 stay private.
|
||||
|
||||
upstream llmgw {
|
||||
server 127.0.0.1:8080;
|
||||
keepalive 32;
|
||||
}
|
||||
|
||||
server {
|
||||
listen 80;
|
||||
listen [::]:80;
|
||||
server_name llm.example.com;
|
||||
return 301 https://$host$request_uri;
|
||||
}
|
||||
|
||||
server {
|
||||
# `listen ... http2` rather than `http2 on;`: the latter needs nginx
|
||||
# 1.25.1, and Ubuntu 22.04 and Debian 12 ship older ones. Newer nginx
|
||||
# accepts this form with a deprecation warning.
|
||||
listen 443 ssl http2;
|
||||
listen [::]:443 ssl http2;
|
||||
server_name llm.example.com;
|
||||
|
||||
ssl_certificate /etc/letsencrypt/live/llm.example.com/fullchain.pem;
|
||||
ssl_certificate_key /etc/letsencrypt/live/llm.example.com/privkey.pem;
|
||||
|
||||
# The router reads at most 32 MiB of body (router.max_body_bytes).
|
||||
client_max_body_size 32m;
|
||||
|
||||
location / {
|
||||
proxy_pass http://llmgw;
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Connection "";
|
||||
|
||||
# Set, never appended: llmgw decides from X-Forwarded-Proto whether
|
||||
# session cookies are Secure, and builds the base URL the landing page
|
||||
# shows from X-Forwarded-Host. A value passed through from the client
|
||||
# would be the client's choice.
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Forwarded-Host $host;
|
||||
proxy_set_header X-Forwarded-Proto $scheme;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
|
||||
# Streamed replies: tokens go out as they are generated. Buffering
|
||||
# turns a stream into one late burst, and gzip buffers too.
|
||||
proxy_buffering off;
|
||||
proxy_cache off;
|
||||
gzip off;
|
||||
|
||||
# nginx's default is 60 seconds between two reads from upstream. A
|
||||
# large model can think longer than that before its first token, and
|
||||
# the client then sees a 504 for a request that was about to succeed.
|
||||
# llmgw has its own limits (timeouts.*); this must not be the shorter.
|
||||
proxy_read_timeout 15m;
|
||||
proxy_send_timeout 15m;
|
||||
}
|
||||
}
|
||||
+94
-128
@@ -29,7 +29,7 @@ The reason is boring and unavoidable: drivers and engines are platform decisions
|
||||
with kernel modules and version matrices attached.
|
||||
|
||||
Weights are the one item on that list llmgw does handle. It includes a mirror —
|
||||
step 4 — which holds the files on your own network and hands them to the nodes,
|
||||
step 3 — which holds the files on your own network and hands them to the nodes,
|
||||
so getting weights onto every machine is not a problem you solve before
|
||||
starting. What llmgw will not do is choose the models for you.
|
||||
|
||||
@@ -58,7 +58,8 @@ them, so nodes need no inbound access except your own SSH.
|
||||
| 8082 | the control plane administering the mirror | the control plane only — this port can delete models and start downloads |
|
||||
|
||||
Decide now which address nodes will use to reach the control plane. It goes into
|
||||
a certificate and changing it later means reissuing one.
|
||||
a certificate: the machine's own addresses are found for you, and any other name
|
||||
— a DNS alias, a NAT or floating address — has to be given with `--host`.
|
||||
|
||||
## 1. The binary
|
||||
|
||||
@@ -83,141 +84,106 @@ git clone https://gitea.mynodes.xyz/ilker/llmgw && cd llmgw
|
||||
CGO_ENABLED=0 go build -o llmgw ./cmd/llmgw
|
||||
```
|
||||
|
||||
## 2. The certificate authority
|
||||
## 2. The control plane
|
||||
|
||||
Agents authenticate to the control plane with mutual TLS, and a node's identity
|
||||
is taken from its certificate's common name. A public authority cannot supply
|
||||
these: Let's Encrypt and its peers stopped issuing certificates carrying the
|
||||
`clientAuth` usage, so nothing they hand out can authenticate an agent.
|
||||
|
||||
Run this wherever you intend to keep the authority. It talks to nothing and
|
||||
works on files, so it does not have to be a cluster machine — and there is an
|
||||
argument that it should not be.
|
||||
On the machine that will be the control plane, as root:
|
||||
|
||||
```
|
||||
sudo llmgw ctl cert init-ca --dir /etc/llmgw/pki
|
||||
sudo llmgw server init --admin you@example.com
|
||||
```
|
||||
|
||||
That writes `ca.pem` and `ca-key.pem`. **Whoever holds `ca-key.pem` can mint any
|
||||
node's identity**, so it stays here; `ca.pem` is copied to every machine and is
|
||||
not a secret.
|
||||
That is the whole step. In order, it:
|
||||
|
||||
Now the control plane's own certificate. Every address an agent might dial must
|
||||
appear in `--host`, bare IP addresses included — hostname verification reads
|
||||
subject alternative names and ignores the common name entirely, so a certificate
|
||||
naming the host only in its subject is rejected by every client:
|
||||
1. creates the `llmgw` system user, `/etc/llmgw` and `/var/lib/llmgw`
|
||||
2. creates the cluster's certificate authority in `/etc/llmgw/pki`
|
||||
3. issues the control plane's certificate for this machine's hostname and
|
||||
every address it has, and copies `ca.pem` beside it
|
||||
4. writes `/etc/llmgw/server.yaml`: mutual TLS, authentication, default limits
|
||||
and a price, and a control plane with an empty catalog
|
||||
5. creates the store, the `default` project, and your account, already approved
|
||||
and promoted to administrator — the password is generated and **printed
|
||||
once**
|
||||
6. installs the systemd unit, starts it, and waits until `/health` answers
|
||||
|
||||
Every step looks before it acts. Running it again finishes an install that
|
||||
stopped halfway and changes nothing on one that did not; it never overwrites
|
||||
`server.yaml`, the authority, or an existing account's password.
|
||||
|
||||
**The addresses in the certificate matter.** Agents check that the address
|
||||
they dial is named in it, and the certificate names only what this machine
|
||||
could see of itself. If nodes will reach it by another name — a DNS alias, a
|
||||
NAT address, a floating IP — give it:
|
||||
|
||||
```
|
||||
sudo llmgw ctl cert issue-server llmgw-ctl \
|
||||
--host 10.0.0.140 --host llmgw-ctl.internal \
|
||||
--dir /etc/llmgw/pki --out /etc/llmgw
|
||||
sudo llmgw server init --host llmgw-ctl.internal --host 203.0.113.10
|
||||
```
|
||||
|
||||
Run that at any time: the certificate is reissued with the new names added to
|
||||
the old ones, and the service restarted.
|
||||
|
||||
**`ca-key.pem` signs every node's identity.** `init` leaves it in
|
||||
`/etc/llmgw/pki`, because onboarding a node needs it and the control plane is
|
||||
where most people run onboarding from. Back it up; there is no recovering it.
|
||||
If this machine is exposed, move the directory elsewhere once the nodes are in
|
||||
and pass `--pki` to onboarding.
|
||||
|
||||
This covers the fleet-facing port only. The client-facing API is public and
|
||||
belongs behind an ordinary publicly trusted certificate, usually terminated by a
|
||||
reverse proxy in front of port 8080.
|
||||
|
||||
## 3. The control plane
|
||||
|
||||
### The user and directories
|
||||
|
||||
```
|
||||
sudo useradd --system --home /var/lib/llmgw --shell /usr/sbin/nologin llmgw
|
||||
sudo install -d -o llmgw -g llmgw -m 0750 /var/lib/llmgw
|
||||
sudo install -d -m 0755 /etc/llmgw
|
||||
sudo chown root:llmgw /etc/llmgw/ca.pem /etc/llmgw/server.pem /etc/llmgw/server-key.pem
|
||||
sudo chmod 0640 /etc/llmgw/server-key.pem
|
||||
```
|
||||
|
||||
### The configuration
|
||||
|
||||
`/etc/llmgw/server.yaml`. This is the smallest file that is not a toy: mutual
|
||||
TLS on, authentication on, one model, and limits that apply to anyone nobody has
|
||||
given limits to.
|
||||
|
||||
```yaml
|
||||
listen: ":8080"
|
||||
admin_listen: "127.0.0.1:9090"
|
||||
agent_listen: ":7443"
|
||||
|
||||
agent_tls:
|
||||
ca: /etc/llmgw/ca.pem
|
||||
cert: /etc/llmgw/server.pem
|
||||
key: /etc/llmgw/server-key.pem
|
||||
|
||||
auth:
|
||||
enabled: true
|
||||
store: /var/lib/llmgw/llmgw.db
|
||||
signup_open: true
|
||||
# Applied to any key no scope gives a limit to. Without these, a project
|
||||
# created without limits is metered by nothing at all, and the first sign of
|
||||
# that is the bill or the outage.
|
||||
default_rpm: 60
|
||||
default_tpm: 200000
|
||||
# What a request costs, which is what makes a budget mean anything. Without
|
||||
# a price every request costs zero, so a budget is displayed, counted against
|
||||
# nothing, and never binds — and nobody finds out until they expected it to
|
||||
# stop something. Self-hosted models have no per-token price, so the unit is
|
||||
# yours to choose: GPU-seconds, a notional rate, or 1000 everywhere, which
|
||||
# turns a budget into a plain token quota.
|
||||
default_price:
|
||||
prompt_per_1k: 1000
|
||||
completion_per_1k: 1000
|
||||
# pricing:
|
||||
# qwen2.5-coder-7b-instruct: { prompt_per_1k: 500, completion_per_1k: 1500 }
|
||||
|
||||
landing:
|
||||
contact: "platform@example.com"
|
||||
|
||||
# Start with none. Models are added while the cluster runs — see step 8 —
|
||||
# and anything listed here is declarative: it is in the catalog on every start
|
||||
# and cannot be removed without editing this file.
|
||||
models: []
|
||||
```
|
||||
|
||||
`admin_listen` is on loopback on purpose. It serves pprof, which can dump the
|
||||
process heap, and API keys sit in that heap in plaintext. To see the dashboard
|
||||
from elsewhere, promote an account and use `/admin` on the client port — do not
|
||||
move this listener.
|
||||
|
||||
The store file is created on first start; there is no separate migration step.
|
||||
|
||||
### The service
|
||||
|
||||
```
|
||||
sudo cp deploy/systemd/llmgw-server.service /etc/systemd/system/
|
||||
sudo systemctl daemon-reload
|
||||
sudo systemctl enable --now llmgw-server
|
||||
systemctl status llmgw-server
|
||||
```
|
||||
|
||||
### Your account
|
||||
|
||||
Accounts are created and approved from a shell on this machine, deliberately:
|
||||
having a shell here is what makes someone an administrator, and there are no
|
||||
roles to get subtly wrong.
|
||||
|
||||
```
|
||||
sudo -u llmgw llmgw users create you@example.com --password 'change me now'
|
||||
# prints a user id, usr-xxxxxxxx
|
||||
sudo -u llmgw llmgw users promote usr-xxxxxxxx
|
||||
```
|
||||
|
||||
`promote` is what grants `/admin`. Sign in at `http://<control-plane>:8080/login`,
|
||||
change the password, and issue yourself a key from the profile page.
|
||||
belongs behind an ordinary publicly trusted certificate, terminated by a
|
||||
reverse proxy in front of port 8080. [deploy/proxy/](../deploy/proxy/) has a
|
||||
Caddyfile and an nginx configuration that get streaming, timeouts and the
|
||||
forwarded headers right; both are attached to every release.
|
||||
|
||||
### Check it
|
||||
|
||||
```
|
||||
curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8080/health # 200, with an empty body
|
||||
sudo -u llmgw llmgw ctl status
|
||||
curl -s -o /dev/null -w '%{http_code}\n' http://127.0.0.1:8080/health # 200
|
||||
llmgw ctl status
|
||||
```
|
||||
|
||||
`ctl status` will report no nodes and no models, which is correct — nothing has
|
||||
been added yet. It is worth running now so its output is familiar before you
|
||||
need it.
|
||||
been added yet. Sign in at `http://<control-plane>:8080/login` with the printed
|
||||
password, change it, and issue yourself a key from the profile page.
|
||||
|
||||
## 4. The mirror
|
||||
### The configuration
|
||||
|
||||
`/etc/llmgw/server.yaml` is yours from here on. Three things in it are worth
|
||||
knowing before you edit it:
|
||||
|
||||
- `admin_listen` is on loopback on purpose. It serves pprof, which can dump the
|
||||
process heap, and API keys sit in that heap in plaintext. To see the dashboard
|
||||
from elsewhere use `/admin` on the client port — do not move this listener.
|
||||
- `default_rpm`, `default_tpm` and `default_price` are what make limits and
|
||||
budgets mean anything. Without a price every request costs zero and a budget
|
||||
never binds; without the defaults, a project created without limits is
|
||||
metered by nothing at all.
|
||||
- `control.enabled: true` is what makes this a control plane while its catalog
|
||||
is empty. Without it, a file with no models is refused as having nothing to
|
||||
serve.
|
||||
|
||||
Every key is in [reference.md](reference.md).
|
||||
|
||||
### Keeping the authority off the control plane
|
||||
|
||||
Create it on another machine, issue the control plane's certificate there, and
|
||||
bring only the results:
|
||||
|
||||
```
|
||||
# on the authority's machine
|
||||
llmgw ctl cert init-ca --dir /etc/llmgw/pki
|
||||
llmgw ctl cert issue-server llmgw-ctl --host 10.0.0.140 --host llmgw-ctl.internal \
|
||||
--dir /etc/llmgw/pki --out ./ctl
|
||||
|
||||
# copy ca.pem, ctl/server.pem and ctl/server-key.pem into /etc/llmgw on the
|
||||
# control plane, then there:
|
||||
sudo llmgw server init --admin you@example.com
|
||||
```
|
||||
|
||||
`init` finds certificates in `/etc/llmgw` and no authority beside them, and
|
||||
keeps them as they are. Adding an address later is then done where the
|
||||
authority is; `init --host` says so rather than minting a second authority no
|
||||
node trusts.
|
||||
|
||||
## 3. The mirror
|
||||
|
||||
Do this before the nodes, not after. Every node fetches its own copy of a
|
||||
model's weights, and where they fetch from decides whether bringing up a cluster
|
||||
@@ -313,7 +279,7 @@ llmgw mirror fetch --model qwen2.5-14b-instruct --version 2026-03 \
|
||||
or from the admin page, or by copying files into the directory by hand — the
|
||||
mirror is a directory, and anything that puts files in it works.
|
||||
|
||||
## 5. Preparing a node
|
||||
## 4. Preparing a node
|
||||
|
||||
**Do this before onboarding.** A node missing its driver or engine registers
|
||||
healthy and fails on first placement, and the reason is three logs away.
|
||||
@@ -418,7 +384,7 @@ llmgw ctl engine-check vllm --bin /opt/venvs/vllm/bin/vllm
|
||||
|
||||
### Weights
|
||||
|
||||
Nothing to do here if you built the mirror in step 4 — the agent fetches what a
|
||||
Nothing to do here if you built the mirror in step 3 — the agent fetches what a
|
||||
model's `source` names, and that source is your mirror. This section is for the
|
||||
arrangements where the files are already on the machine.
|
||||
|
||||
@@ -441,7 +407,7 @@ model's `source` at a file to download and the agent fetches it once, before the
|
||||
engine is launched:
|
||||
|
||||
```yaml
|
||||
# your mirror, which is what step 4 was for
|
||||
# your mirror, which is what step 3 was for
|
||||
source: http://models.internal:8081/qwen2.5-coder-3b-instruct/2026-03/q4_k_m.gguf
|
||||
# a whole version, for a model of several files — the trailing slash is the signal
|
||||
source: http://models.internal:8081/llama-3.1-70b-instruct/2026-03/
|
||||
@@ -498,7 +464,7 @@ How much disk a node needs follows from the models rather than a rule of thumb.
|
||||
`llmgw ctl model plan <repo>` prints the sizes; as a starting point, a 70B at
|
||||
bf16 is 141 GB and the same model at q4 is 43 GB.
|
||||
|
||||
## 6. Adding the node
|
||||
## 5. Adding the node
|
||||
|
||||
Two halves, because they need different access: building the machine needs the
|
||||
hypervisor, and issuing its identity needs the certificate authority. Where one
|
||||
@@ -555,7 +521,7 @@ Repeat for each node. Nodes in the same fabric must be given the same `--fabric`
|
||||
name; nodes with no InfiniBand take no `--fabric` at all and are each their own
|
||||
island, which is fine for single-node models.
|
||||
|
||||
## 7. Setting the node's CPU model, if it is a VM
|
||||
## 6. Setting the node's CPU model, if it is a VM
|
||||
|
||||
Proxmox defaults a VM to `kvm64`, which hides AVX, AVX2, FMA and even SSE4.2
|
||||
from the guest. Inference then falls back to scalar code and is not slightly
|
||||
@@ -572,7 +538,7 @@ only change being `--cpu host`:
|
||||
probably did not, and nothing about it looks broken — every number is simply
|
||||
small. Check with `grep -qw avx2 /proc/cpuinfo` on the node.
|
||||
|
||||
## 8. Adding a model
|
||||
## 7. Adding a model
|
||||
|
||||
Write the definition. This is the same shape a model takes in `server.yaml`, so
|
||||
an entry moves between the two by being copied:
|
||||
@@ -714,7 +680,7 @@ An engine that publishes no metrics endpoint is measured from the outside only.
|
||||
The run says so, and queue depth and KV occupancy are then not part of the
|
||||
answer.
|
||||
|
||||
## 9. The first request
|
||||
## 8. The first request
|
||||
|
||||
```
|
||||
curl http://<control-plane>:8080/v1/chat/completions \
|
||||
@@ -957,7 +923,7 @@ the choice is yours rather than silent.
|
||||
|
||||
Three things, in descending order of how badly you want them.
|
||||
|
||||
**The certificate authority.** `ca-key.pem`, wherever you put it in step 2.
|
||||
**The certificate authority.** `ca-key.pem`, `/etc/llmgw/pki` unless you kept it elsewhere (step 2).
|
||||
Losing it does not stop the cluster, and it means every node identity has to be
|
||||
reissued the next time you touch one. Nothing in llmgw can reconstruct it.
|
||||
|
||||
@@ -1028,7 +994,7 @@ output, which is the actual diagnosis.
|
||||
node and the reason: wrong GPU class, not enough free cards, too little memory,
|
||||
no fabric, a label it does not carry.
|
||||
|
||||
**Everything is slow.** On a VM, check AVX2 (step 7). On CPU, check the
|
||||
**Everything is slow.** On a VM, check AVX2 (step 6). On CPU, check the
|
||||
quantisation — the difference between `q8_0` and `q4_k_m` is roughly a factor of
|
||||
two. `llmgw ctl status` shows how full the cluster is.
|
||||
|
||||
@@ -1065,7 +1031,7 @@ requests. On a slow model, set `timeouts.generation` higher.
|
||||
those eight local, which is most of the problem; on a fabric where the nodes
|
||||
can see each other, seeding between them is the obvious next thing to build.
|
||||
- **A budget without a price never binds.** Set `default_price` or `pricing`
|
||||
(step 3) or every request costs zero, and a budget is something the dashboard
|
||||
(step 2) or every request costs zero, and a budget is something the dashboard
|
||||
displays rather than something that stops anything. The quotas page says so
|
||||
where it can, which is not the same as being told before you rely on it.
|
||||
- **Engines are yours to install.** Onboarding verifies one runs; it installs
|
||||
|
||||
@@ -326,6 +326,8 @@ type Agent struct {
|
||||
// it asked for, nil until it has; provStatus is where each stands and
|
||||
// provSent the fingerprint of what the control plane last heard. All
|
||||
// guarded by mu. provKick starts a pass without waiting for the timer.
|
||||
logRing *logRing
|
||||
|
||||
prov *provision.Installer
|
||||
provWants []provision.Want
|
||||
provStatus map[string]provision.Status
|
||||
@@ -372,6 +374,11 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
// Everything the agent logs also goes to a ring the admin page can read,
|
||||
// so seeing what a node did does not mean ssh and a journal only root may
|
||||
// open.
|
||||
ring := newLogRing(agentLogLines)
|
||||
log = slog.New(teeHandler{log.Handler(), slog.NewTextHandler(ring, nil)})
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
}
|
||||
@@ -387,6 +394,7 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
|
||||
http: httpClient,
|
||||
shardReplica: map[string]string{},
|
||||
provStatus: map[string]provision.Status{},
|
||||
logRing: ring,
|
||||
provKick: make(chan struct{}, 1),
|
||||
}
|
||||
a.prov = &provision.Installer{Root: cfg.EnginesDir, Log: log, OnStatus: a.provisionStatus}
|
||||
@@ -960,6 +968,16 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
|
||||
a.removeWeights(m.RemoveWeights.GetPaths())
|
||||
case *llmgwv1.ControlMessage_Provision:
|
||||
a.setProvision(m.Provision)
|
||||
case *llmgwv1.ControlMessage_Logs:
|
||||
reply := a.logsReply(m.Logs)
|
||||
// Off the receive loop: a full send queue must not stop commands being
|
||||
// read.
|
||||
go func() {
|
||||
select {
|
||||
case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: reply}}:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}()
|
||||
case *llmgwv1.ControlMessage_Ack:
|
||||
if !m.Ack.GetOk() {
|
||||
a.log.Warn("control plane rejected a message",
|
||||
@@ -1394,3 +1412,30 @@ func enginesFingerprint(vers map[string]string) string {
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// logsReply answers a request for one of this node's logs.
|
||||
func (a *Agent) logsReply(req *llmgwv1.LogsRequest) *llmgwv1.LogsReply {
|
||||
n := int(req.GetLines())
|
||||
if n <= 0 {
|
||||
n = 200
|
||||
}
|
||||
n = min(n, agentLogLines)
|
||||
shards := a.sup.IDs()
|
||||
slices.Sort(shards)
|
||||
out := &llmgwv1.LogsReply{
|
||||
RequestId: req.GetRequestId(),
|
||||
Sources: append([]string{"agent"}, shards...),
|
||||
}
|
||||
switch src := req.GetSource(); src {
|
||||
case "", "agent":
|
||||
out.Lines = a.logRing.tail(n)
|
||||
default:
|
||||
sh, err := a.sup.Get(src)
|
||||
if err != nil {
|
||||
out.Error = "no engine " + src + " runs on this node"
|
||||
return out
|
||||
}
|
||||
out.Lines = sh.Logs(n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// agentLogLines is how much of its own log an agent keeps for the admin page.
|
||||
// Enough to cover an engine install or a failed start; the journal keeps the
|
||||
// rest for anyone who can reach it.
|
||||
const agentLogLines = 2000
|
||||
|
||||
// logRing keeps the most recent lines written to it.
|
||||
type logRing struct {
|
||||
mu sync.Mutex
|
||||
lines []string
|
||||
next int
|
||||
full bool
|
||||
}
|
||||
|
||||
func newLogRing(n int) *logRing { return &logRing{lines: make([]string, n)} }
|
||||
|
||||
// Write takes whole lines, as a text handler writes them.
|
||||
func (r *logRing) Write(p []byte) (int, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
|
||||
r.lines[r.next] = line
|
||||
r.next = (r.next + 1) % len(r.lines)
|
||||
if r.next == 0 {
|
||||
r.full = true
|
||||
}
|
||||
}
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
// tail returns up to n of the most recent lines, oldest first.
|
||||
func (r *logRing) tail(n int) []string {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
size := r.next
|
||||
if r.full {
|
||||
size = len(r.lines)
|
||||
}
|
||||
n = min(n, size)
|
||||
out := make([]string, 0, n)
|
||||
for i := size - n; i < size; i++ {
|
||||
idx := i
|
||||
if r.full {
|
||||
idx = (r.next + i) % len(r.lines)
|
||||
}
|
||||
out = append(out, r.lines[idx])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// teeHandler sends every record to two handlers: wherever the agent already
|
||||
// logs, and the ring the admin page reads.
|
||||
type teeHandler struct{ a, b slog.Handler }
|
||||
|
||||
func (t teeHandler) Enabled(ctx context.Context, l slog.Level) bool {
|
||||
return t.a.Enabled(ctx, l) || t.b.Enabled(ctx, l)
|
||||
}
|
||||
|
||||
func (t teeHandler) Handle(ctx context.Context, r slog.Record) error {
|
||||
var err error
|
||||
if t.a.Enabled(ctx, r.Level) {
|
||||
err = t.a.Handle(ctx, r.Clone())
|
||||
}
|
||||
if t.b.Enabled(ctx, r.Level) {
|
||||
_ = t.b.Handle(ctx, r.Clone())
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (t teeHandler) WithAttrs(as []slog.Attr) slog.Handler {
|
||||
return teeHandler{t.a.WithAttrs(as), t.b.WithAttrs(as)}
|
||||
}
|
||||
|
||||
func (t teeHandler) WithGroup(name string) slog.Handler {
|
||||
return teeHandler{t.a.WithGroup(name), t.b.WithGroup(name)}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
|
||||
)
|
||||
|
||||
func TestTheLogRingKeepsTheNewestLinesInOrder(t *testing.T) {
|
||||
r := newLogRing(3)
|
||||
if got := r.tail(10); len(got) != 0 {
|
||||
t.Fatalf("empty ring gave %v", got)
|
||||
}
|
||||
for i := 1; i <= 5; i++ {
|
||||
fmt.Fprintf(r, "line %d\n", i)
|
||||
}
|
||||
if got := r.tail(10); !slices.Equal(got, []string{"line 3", "line 4", "line 5"}) {
|
||||
t.Errorf("after wrapping: %v", got)
|
||||
}
|
||||
if got := r.tail(2); !slices.Equal(got, []string{"line 4", "line 5"}) {
|
||||
t.Errorf("tail(2): %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestANodeAnswersForItsOwnLog(t *testing.T) {
|
||||
cfg := Config{Server: "127.0.0.1:7443", NodeID: "n1", WeightsCache: t.TempDir(), EnginesDir: t.TempDir()}
|
||||
cfg.Superv.WorkDir = t.TempDir()
|
||||
a, err := New(cfg, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a.log.Info("engine installation", "state", "installing")
|
||||
|
||||
r := a.logsReply(&llmgwv1.LogsRequest{RequestId: "x", Source: "agent", Lines: 50})
|
||||
if r.GetRequestId() != "x" || !strings.Contains(strings.Join(r.GetLines(), "\n"), "engine installation") {
|
||||
t.Errorf("the agent's own line is not in its log: %v", r.GetLines())
|
||||
}
|
||||
if !slices.Contains(r.GetSources(), "agent") {
|
||||
t.Errorf("sources %v", r.GetSources())
|
||||
}
|
||||
|
||||
r = a.logsReply(&llmgwv1.LogsRequest{RequestId: "y", Source: "no-such-shard"})
|
||||
if r.GetError() == "" {
|
||||
t.Error("an unknown engine was answered as if it existed")
|
||||
}
|
||||
}
|
||||
@@ -116,6 +116,7 @@ func newCtlCmd() *cobra.Command {
|
||||
|
||||
cmd.AddCommand(
|
||||
newNodesCmd(client, &asJSON),
|
||||
newLogsCmd(client, &asJSON),
|
||||
newModelsCmd(client, &asJSON),
|
||||
newReplicasCmd(client, &asJSON),
|
||||
newExplainCmd(client, &asJSON),
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
||||
)
|
||||
|
||||
func newLogsCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
var (
|
||||
source string
|
||||
lines int
|
||||
)
|
||||
c := &cobra.Command{
|
||||
Use: "logs NODE",
|
||||
Short: "Show the recent lines of a node's agent log, or of an engine on it",
|
||||
Long: "Read through the agent's own stream, so it needs no access to the node:\n" +
|
||||
"no ssh, no jump host, no root for the journal. The agent keeps its last\n" +
|
||||
"2000 lines and each engine's last few hundred.\n\n" +
|
||||
" llmgw ctl logs llmgw-node06 the agent\n" +
|
||||
" llmgw ctl logs llmgw-node06 --source SHARD an engine's output",
|
||||
Args: cobra.ExactArgs(1),
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
q := url.Values{}
|
||||
q.Set("source", source)
|
||||
q.Set("lines", strconv.Itoa(lines))
|
||||
var v adminapi.LogsView
|
||||
err := client().get(cmd.Context(),
|
||||
"/api/v1/nodes/"+url.PathEscape(args[0])+"/logs?"+q.Encode(), &v)
|
||||
if *asJSON {
|
||||
return json.NewEncoder(cmd.OutOrStdout()).Encode(v)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, l := range v.Lines {
|
||||
fmt.Fprintln(cmd.OutOrStdout(), l)
|
||||
}
|
||||
if v.Error != "" {
|
||||
return fmt.Errorf("%s", v.Error)
|
||||
}
|
||||
// Where else to look, since the source list is the one thing a
|
||||
// person cannot guess.
|
||||
fmt.Fprintf(cmd.ErrOrStderr(), "\n-- %s on %s; other sources: %s\n",
|
||||
v.Source, v.Node, strings.Join(v.Sources, ", "))
|
||||
return nil
|
||||
},
|
||||
}
|
||||
c.Flags().StringVar(&source, "source", "agent", `"agent", or a shard id for that engine's output`)
|
||||
c.Flags().IntVarP(&lines, "lines", "n", 200, "how many of the most recent lines")
|
||||
return c
|
||||
}
|
||||
@@ -203,6 +203,7 @@ func newServerCmd() *cobra.Command {
|
||||
api, usage, version.Version, 5*time.Second, true,
|
||||
web.NavExtras{History: history, Window: cfg.Monitoring.Window}, log))
|
||||
adminMux.Handle("GET /_static/", web.Static())
|
||||
adminMux.HandleFunc("GET /nodes/{id}/logs", web.NodeLogsPage(api, true, version.Version))
|
||||
|
||||
// The same page on the client-facing port at /admin, behind a
|
||||
// session and an admin flag. Handing the router the data rather
|
||||
@@ -349,6 +350,7 @@ func newServerCmd() *cobra.Command {
|
||||
f.StringVar(&metricsListen, "metrics-listen", "", "serve metrics only on this address, safe to expose")
|
||||
f.StringArrayVar(&endpoints, "endpoint", nil, "static endpoint: id=URL=model[,model] (repeatable)")
|
||||
|
||||
cmd.AddCommand(newServerInitCmd())
|
||||
return cmd
|
||||
}
|
||||
|
||||
@@ -466,6 +468,21 @@ func (r *dispatcherRef) RemoveWeights(ctx context.Context, nodeID string, paths
|
||||
return rm.RemoveWeights(ctx, nodeID, paths)
|
||||
}
|
||||
|
||||
// Logs forwards a node log request, the same way RemoveWeights does: the
|
||||
// controller finds it by type assertion, so a forwarder without it would make
|
||||
// the capability silently absent at runtime.
|
||||
func (r *dispatcherRef) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
|
||||
d, err := r.get()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
lr, ok := d.(control.LogReader)
|
||||
if !ok {
|
||||
return nil, nil, errors.New("this dispatcher cannot read node logs")
|
||||
}
|
||||
return lr.Logs(ctx, nodeID, source, lines)
|
||||
}
|
||||
|
||||
func (r *dispatcherRef) StartShard(ctx context.Context, cmd reconciler.ShardCommand) error {
|
||||
d, err := r.get()
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,609 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/base32"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/user"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/account"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
|
||||
)
|
||||
|
||||
// serverUnit is deploy/systemd/llmgw-server.service. A test holds the two
|
||||
// together, because a machine set up by init and one set up from the docs
|
||||
// should not differ in a way nobody chose.
|
||||
const serverUnit = `[Unit]
|
||||
Description=llmgw control plane
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
ExecStart=/usr/local/bin/llmgw server --config /etc/llmgw/server.yaml
|
||||
User=llmgw
|
||||
Group=llmgw
|
||||
Restart=on-failure
|
||||
RestartSec=2s
|
||||
NoNewPrivileges=true
|
||||
ProtectSystem=strict
|
||||
ProtectHome=true
|
||||
PrivateTmp=true
|
||||
StateDirectory=llmgw
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
`
|
||||
|
||||
// serverInit is everything a control plane needs before it can start, done in
|
||||
// the order it has to be done in. Every step looks before it acts, so running
|
||||
// it again finishes an install that stopped halfway and changes nothing on
|
||||
// one that did not.
|
||||
type serverInit struct {
|
||||
etcDir string // /etc/llmgw
|
||||
stateDir string // /var/lib/llmgw
|
||||
pkiDir string // where the authority lives
|
||||
unitPath string // "" leaves systemd alone
|
||||
|
||||
// user owns the state and reads the key; "" means no user is created and
|
||||
// nothing is chowned, which is what a test wants.
|
||||
user string
|
||||
hosts []string
|
||||
admin string
|
||||
|
||||
out io.Writer
|
||||
|
||||
uid, gid int
|
||||
changed bool
|
||||
}
|
||||
|
||||
func newServerInitCmd() *cobra.Command {
|
||||
in := serverInit{
|
||||
etcDir: "/etc/llmgw",
|
||||
stateDir: "/var/lib/llmgw",
|
||||
pkiDir: defaultPKIDir,
|
||||
unitPath: "/etc/systemd/system/llmgw-server.service",
|
||||
user: "llmgw",
|
||||
}
|
||||
cmd := &cobra.Command{
|
||||
Use: "init",
|
||||
Short: "Set this machine up as a control plane and start it",
|
||||
Long: "Creates the llmgw user and its directories, a certificate authority, the\n" +
|
||||
"control plane's certificate for every address this machine has, a\n" +
|
||||
"configuration, an administrator account, and the systemd unit — then\n" +
|
||||
"starts it and waits until it answers.\n\n" +
|
||||
"Each step is skipped when its result already exists, so it is safe to run\n" +
|
||||
"again: to finish an install that stopped, or with another --host to\n" +
|
||||
"reissue the certificate. It never overwrites server.yaml or the authority.\n\n" +
|
||||
"The authority's key stays in " + defaultPKIDir + " on this machine. Move it\n" +
|
||||
"somewhere safer once the nodes are onboarded if this machine is exposed.",
|
||||
Args: cobra.NoArgs,
|
||||
RunE: func(cmd *cobra.Command, _ []string) error {
|
||||
if runtime.GOOS != "linux" {
|
||||
return fmt.Errorf("init sets up a Linux machine; this is %s", runtime.GOOS)
|
||||
}
|
||||
if os.Geteuid() != 0 {
|
||||
return errors.New("init creates a user, writes /etc and starts a service: run it as root")
|
||||
}
|
||||
in.out = cmd.OutOrStdout()
|
||||
return in.run(cmd.Context())
|
||||
},
|
||||
}
|
||||
f := cmd.Flags()
|
||||
f.StringArrayVar(&in.hosts, "host", nil,
|
||||
"an address agents dial besides the ones found on this machine; repeatable")
|
||||
f.StringVar(&in.admin, "admin", "", "email of the first administrator; a password is generated and printed once")
|
||||
f.StringVar(&in.pkiDir, "pki", in.pkiDir, "directory holding the certificate authority")
|
||||
return cmd
|
||||
}
|
||||
|
||||
func (in *serverInit) run(ctx context.Context) error {
|
||||
steps := []struct {
|
||||
name string
|
||||
fn func(context.Context) error
|
||||
}{
|
||||
{"user", in.ensureUser},
|
||||
{"directories", in.ensureDirs},
|
||||
{"certificates", in.ensureCerts},
|
||||
{"configuration", in.ensureConfig},
|
||||
{"store", in.ensureStore},
|
||||
}
|
||||
for _, s := range steps {
|
||||
if err := s.fn(ctx); err != nil {
|
||||
return fmt.Errorf("%s: %w", s.name, err)
|
||||
}
|
||||
}
|
||||
if in.unitPath == "" {
|
||||
return nil
|
||||
}
|
||||
if err := in.ensureService(ctx); err != nil {
|
||||
return fmt.Errorf("service: %w", err)
|
||||
}
|
||||
in.nextSteps()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (in *serverInit) done(format string, a ...any) {
|
||||
fmt.Fprintf(in.out, " ok "+format+"\n", a...)
|
||||
in.changed = true
|
||||
}
|
||||
|
||||
func (in *serverInit) kept(format string, a ...any) {
|
||||
fmt.Fprintf(in.out, " kept "+format+"\n", a...)
|
||||
}
|
||||
|
||||
func (in *serverInit) ensureUser(context.Context) error {
|
||||
if in.user == "" {
|
||||
return nil
|
||||
}
|
||||
u, err := user.Lookup(in.user)
|
||||
if err != nil {
|
||||
var unknown user.UnknownUserError
|
||||
if !errors.As(err, &unknown) {
|
||||
return err
|
||||
}
|
||||
out, err := exec.Command("useradd", "--system", "--home", in.stateDir,
|
||||
"--shell", "/usr/sbin/nologin", in.user).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("useradd: %v: %s", err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
if u, err = user.Lookup(in.user); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("system user %s", in.user)
|
||||
} else {
|
||||
in.kept("system user %s", in.user)
|
||||
}
|
||||
in.uid, _ = strconv.Atoi(u.Uid)
|
||||
in.gid, _ = strconv.Atoi(u.Gid)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (in *serverInit) ensureDirs(context.Context) error {
|
||||
for _, d := range []struct {
|
||||
path string
|
||||
mode os.FileMode
|
||||
owned bool
|
||||
}{
|
||||
{in.etcDir, 0o755, false},
|
||||
{in.pkiDir, 0o700, false},
|
||||
{in.stateDir, 0o750, true},
|
||||
} {
|
||||
if err := os.MkdirAll(d.path, d.mode); err != nil {
|
||||
return err
|
||||
}
|
||||
if d.owned {
|
||||
if err := in.chown(d.path, true); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// chown gives a path to the service user, or only its group when the user
|
||||
// should read and not write.
|
||||
func (in *serverInit) chown(path string, own bool) error {
|
||||
if in.user == "" {
|
||||
return nil
|
||||
}
|
||||
uid := 0
|
||||
if own {
|
||||
uid = in.uid
|
||||
}
|
||||
return os.Chown(path, uid, in.gid)
|
||||
}
|
||||
|
||||
func (in *serverInit) ensureCerts(context.Context) error {
|
||||
certPath := filepath.Join(in.etcDir, "server.pem")
|
||||
_, keyErr := os.Stat(filepath.Join(in.pkiDir, "ca-key.pem"))
|
||||
_, caErr := os.Stat(filepath.Join(in.etcDir, "ca.pem"))
|
||||
_, certErr := os.Stat(certPath)
|
||||
if keyErr != nil && caErr == nil && certErr == nil {
|
||||
// Issued by an authority kept elsewhere, which is the arrangement for
|
||||
// anyone who will not have the CA key on the control plane. Making a
|
||||
// new authority here would replace their certificate with one no node
|
||||
// trusts.
|
||||
if len(in.hosts) > 0 {
|
||||
return fmt.Errorf("%s was issued by an authority that is not in %s; "+
|
||||
"reissue it there with the new --host", certPath, in.pkiDir)
|
||||
}
|
||||
in.kept("certificates in %s, issued elsewhere", in.etcDir)
|
||||
return nil
|
||||
}
|
||||
|
||||
var (
|
||||
ca *tlsconf.CA
|
||||
err error
|
||||
)
|
||||
if keyErr == nil {
|
||||
if ca, err = loadCA(in.pkiDir); err != nil {
|
||||
return err
|
||||
}
|
||||
in.kept("certificate authority in %s", in.pkiDir)
|
||||
} else {
|
||||
if ca, err = tlsconf.NewCA("llmgw cluster CA"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ca.Write(in.pkiDir); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("certificate authority in %s (ca-key.pem signs every node: back it up)", in.pkiDir)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(in.etcDir, "ca.pem"), ca.PEM, 0o644); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
want := in.certHosts()
|
||||
if cert, err := readCert(certPath); err == nil {
|
||||
have := strings.Split(sanNames(cert), ", ")
|
||||
var missing []string
|
||||
for _, h := range want {
|
||||
if !slices.Contains(have, h) {
|
||||
missing = append(missing, h)
|
||||
}
|
||||
}
|
||||
// A certificate from this authority naming every address is left
|
||||
// alone. One naming fewer is reissued with the union, so an address
|
||||
// that worked yesterday is never dropped by adding one today.
|
||||
if len(missing) == 0 && cert.CheckSignatureFrom(ca.Cert) == nil {
|
||||
in.kept("server certificate for %s", strings.Join(have, ", "))
|
||||
return nil
|
||||
}
|
||||
for _, h := range have {
|
||||
if h != "" && !slices.Contains(want, h) {
|
||||
want = append(want, h)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
name, _ := os.Hostname()
|
||||
if name == "" {
|
||||
name = "llmgw-control"
|
||||
}
|
||||
leaf, err := ca.IssueServer(name, want)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := leaf.Write(in.etcDir, "server"); err != nil {
|
||||
return err
|
||||
}
|
||||
// The service runs as llmgw and must read its key; nothing else should.
|
||||
keyPath := filepath.Join(in.etcDir, "server-key.pem")
|
||||
if err := in.chown(keyPath, false); err != nil {
|
||||
return err
|
||||
}
|
||||
if in.user != "" {
|
||||
if err := os.Chmod(keyPath, 0o640); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
in.done("server certificate for %s", strings.Join(want, ", "))
|
||||
return nil
|
||||
}
|
||||
|
||||
// certHosts is every address an agent could plausibly dial: the hostname and
|
||||
// each address on this machine, plus whatever --host added. Asking the
|
||||
// operator to list them is how a certificate ends up naming the one address
|
||||
// nobody uses, and that is found out by the first node refusing it.
|
||||
func (in *serverInit) certHosts() []string {
|
||||
var out []string
|
||||
add := func(h string) {
|
||||
if h != "" && !slices.Contains(out, h) {
|
||||
out = append(out, h)
|
||||
}
|
||||
}
|
||||
for _, h := range in.hosts {
|
||||
add(h)
|
||||
}
|
||||
if name, err := os.Hostname(); err == nil {
|
||||
add(name)
|
||||
}
|
||||
for _, ip := range localAddrs() {
|
||||
add(ip.String())
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// localAddrs are this machine's addresses another machine could reach:
|
||||
// loopback and link-local are left out.
|
||||
func localAddrs() []net.IP {
|
||||
addrs, err := net.InterfaceAddrs()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
var out []net.IP
|
||||
for _, a := range addrs {
|
||||
n, ok := a.(*net.IPNet)
|
||||
if !ok || !n.IP.IsGlobalUnicast() {
|
||||
continue
|
||||
}
|
||||
out = append(out, n.IP)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (in *serverInit) configPath() string { return filepath.Join(in.etcDir, "server.yaml") }
|
||||
|
||||
func (in *serverInit) ensureConfig(context.Context) error {
|
||||
path := in.configPath()
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
in.kept("%s (never overwritten)", path)
|
||||
} else {
|
||||
contact := in.admin
|
||||
if contact == "" {
|
||||
contact = "platform@example.com"
|
||||
}
|
||||
body := fmt.Sprintf(serverConfigTemplate, in.etcDir, in.etcDir, in.etcDir,
|
||||
filepath.Join(in.stateDir, "llmgw.db"), contact)
|
||||
if err := os.WriteFile(path, []byte(body), 0o640); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := in.chown(path, false); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("%s", path)
|
||||
}
|
||||
|
||||
// Checked whether written now or years ago: a configuration that will not
|
||||
// load is better found here than as a unit restarting every two seconds.
|
||||
cfg, err := config.Load(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := cfg.Validate(); err != nil {
|
||||
return fmt.Errorf("%s: %w", path, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// serverConfigTemplate is the smallest file that is not a toy: mutual TLS,
|
||||
// authentication, default limits and a price, and a control plane with an
|
||||
// empty catalog.
|
||||
const serverConfigTemplate = `# Written by 'llmgw server init'. Every key: docs/reference.md
|
||||
listen: ":8080"
|
||||
admin_listen: "127.0.0.1:9090" # serves pprof, which can dump API keys: keep it local
|
||||
agent_listen: ":7443"
|
||||
|
||||
agent_tls:
|
||||
ca: %s/ca.pem
|
||||
cert: %s/server.pem
|
||||
key: %s/server-key.pem
|
||||
|
||||
# A control plane with nothing in its catalog yet. Models are added while it
|
||||
# runs, with 'llmgw ctl model add' or from /admin/mirror.
|
||||
control:
|
||||
enabled: true
|
||||
|
||||
auth:
|
||||
enabled: true
|
||||
store: %s
|
||||
signup_open: true
|
||||
# Applied to any key no scope gives a limit to. Without them, a project
|
||||
# created without limits is metered by nothing at all.
|
||||
default_rpm: 60
|
||||
default_tpm: 200000
|
||||
# Without a price every request costs zero and no budget ever binds. 1000
|
||||
# everywhere turns a budget into a plain token quota.
|
||||
default_price:
|
||||
prompt_per_1k: 1000
|
||||
completion_per_1k: 1000
|
||||
|
||||
landing:
|
||||
contact: "%s"
|
||||
|
||||
# The engines every node installs for itself. Omitted, each gets the llama.cpp
|
||||
# build this release was checked against.
|
||||
# engines:
|
||||
# llamacpp: b10991
|
||||
|
||||
# A model mirror, once there is one (docs/install.md, step 3):
|
||||
# mirror:
|
||||
# api: https://models.internal:8082
|
||||
# tls: { ca: /etc/llmgw/ca.pem, cert: /etc/llmgw/server.pem, key: /etc/llmgw/server-key.pem }
|
||||
`
|
||||
|
||||
func (in *serverInit) ensureStore(ctx context.Context) error {
|
||||
cfg, err := config.Load(in.configPath())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
address := cfg.Auth.Store
|
||||
if address == "" {
|
||||
address = filepath.Join(in.stateDir, "llmgw.db")
|
||||
}
|
||||
s, err := store.Open(ctx, address)
|
||||
if err != nil {
|
||||
return fmt.Errorf("opening %s: %w", address, err)
|
||||
}
|
||||
// Opened as root, so a SQLite file created just now belongs to root and
|
||||
// the service could not write it. Given back before anything else runs.
|
||||
defer in.chownSQLite(address)
|
||||
defer s.Close()
|
||||
|
||||
// The tree an approved account hangs off. Approving into a project that
|
||||
// does not exist succeeds, and the first key that person makes does not.
|
||||
if err := s.PutOrg(ctx, store.Org{ID: "default", Name: "default"}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.PutTeam(ctx, store.Team{ID: "default", OrgID: "default", Name: "default"}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.PutProject(ctx, store.Project{ID: "default", TeamID: "default", Name: "default"}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if in.admin == "" {
|
||||
in.kept("store at %s (no --admin: create an account with 'llmgw users create')", address)
|
||||
return nil
|
||||
}
|
||||
if u, err := s.UserByEmail(ctx, in.admin); err == nil {
|
||||
if !u.IsAdmin {
|
||||
if err := s.SetAdmin(ctx, u.ID, true); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
in.kept("administrator %s (%s); its password is unchanged", in.admin, u.ID)
|
||||
return nil
|
||||
} else if !errors.Is(err, store.ErrNotFound) {
|
||||
return err
|
||||
}
|
||||
|
||||
password := randomPassword()
|
||||
hash, err := account.HashPassword(password)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
id := randomID("usr")
|
||||
now := time.Now()
|
||||
if err := s.CreateUser(ctx, store.User{
|
||||
ID: id, Email: in.admin, Name: in.admin,
|
||||
PasswordHash: hash, State: store.UserPending, CreatedAt: now,
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.ApproveUser(ctx, id, "default", "init", now); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.SetAdmin(ctx, id, true); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("administrator %s (%s)", in.admin, id)
|
||||
fmt.Fprintf(in.out, "\n password: %s\n shown once — sign in and change it from /profile\n\n", password)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (in *serverInit) chownSQLite(address string) {
|
||||
if strings.Contains(address, "://") || strings.HasPrefix(address, "@") {
|
||||
return
|
||||
}
|
||||
for _, suffix := range []string{"", "-wal", "-shm"} {
|
||||
if _, err := os.Stat(address + suffix); err == nil {
|
||||
_ = in.chown(address+suffix, true)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func randomPassword() string {
|
||||
var buf [15]byte
|
||||
_, _ = rand.Read(buf[:])
|
||||
return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(buf[:]))
|
||||
}
|
||||
|
||||
func (in *serverInit) ensureService(ctx context.Context) error {
|
||||
if _, err := exec.LookPath("systemctl"); err != nil {
|
||||
fmt.Fprintf(in.out, " skip no systemd here; start it with:\n"+
|
||||
" sudo -u %s llmgw server --config %s\n", in.user, in.configPath())
|
||||
return nil
|
||||
}
|
||||
if b, err := os.ReadFile(in.unitPath); err == nil && string(b) == serverUnit {
|
||||
in.kept("%s", in.unitPath)
|
||||
} else {
|
||||
if err == nil {
|
||||
// Somebody's edits. Kept beside it, because the unit is also where
|
||||
// people put environment overrides and those are not ours to lose.
|
||||
_ = os.WriteFile(in.unitPath+".bak", b, 0o644)
|
||||
}
|
||||
if err := os.WriteFile(in.unitPath, []byte(serverUnit), 0o644); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("%s", in.unitPath)
|
||||
}
|
||||
|
||||
systemctl := func(args ...string) error {
|
||||
out, err := exec.CommandContext(ctx, "systemctl", args...).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("systemctl %s: %v: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err := systemctl("daemon-reload"); err != nil {
|
||||
return err
|
||||
}
|
||||
active := exec.CommandContext(ctx, "systemctl", "is-active", "--quiet", "llmgw-server").Run() == nil
|
||||
switch {
|
||||
case !active:
|
||||
if err := systemctl("enable", "--now", "llmgw-server"); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("llmgw-server started")
|
||||
case in.changed:
|
||||
// A certificate or configuration it has not read is one it is not
|
||||
// using, and the difference shows up as a node that cannot connect.
|
||||
if err := systemctl("restart", "llmgw-server"); err != nil {
|
||||
return err
|
||||
}
|
||||
in.done("llmgw-server restarted to pick up the changes")
|
||||
default:
|
||||
in.kept("llmgw-server running")
|
||||
}
|
||||
return in.waitHealthy(ctx)
|
||||
}
|
||||
|
||||
func (in *serverInit) waitHealthy(ctx context.Context) error {
|
||||
cfg, err := config.Load(in.configPath())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, port, err := net.SplitHostPort(cfg.Listen)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listen %q: %w", cfg.Listen, err)
|
||||
}
|
||||
url := "http://127.0.0.1:" + port + "/health"
|
||||
deadline := time.Now().Add(30 * time.Second)
|
||||
for {
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
in.kept("answering on %s", url)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("not answering on %s after 30s; 'journalctl -u llmgw-server -n 50' says why", url)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(time.Second):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (in *serverInit) nextSteps() {
|
||||
server := "https://<this machine>:7443"
|
||||
if ips := localAddrs(); len(ips) > 0 {
|
||||
server = "https://" + net.JoinHostPort(ips[0].String(), "7443")
|
||||
}
|
||||
fmt.Fprintf(in.out, `
|
||||
The control plane is up. Next:
|
||||
|
||||
llmgw ctl status # no nodes, no models: correct for now
|
||||
|
||||
# each node, from this machine (it holds the authority):
|
||||
deploy/bootstrap/onboard-node.sh --node <name> --addr <node ip> --server %s
|
||||
|
||||
# then a model:
|
||||
llmgw ctl model add -f model.yaml
|
||||
|
||||
The web UI is on port 8080; /admin once you are signed in as the administrator.
|
||||
`, server)
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
|
||||
)
|
||||
|
||||
func newTestInit(t *testing.T) (*serverInit, *bytes.Buffer) {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
var out bytes.Buffer
|
||||
return &serverInit{
|
||||
etcDir: filepath.Join(root, "etc"),
|
||||
stateDir: filepath.Join(root, "state"),
|
||||
pkiDir: filepath.Join(root, "etc", "pki"),
|
||||
admin: "admin@example.com",
|
||||
out: &out,
|
||||
}, &out
|
||||
}
|
||||
|
||||
func TestInitLeavesAControlPlaneThatStarts(t *testing.T) {
|
||||
in, out := newTestInit(t)
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
|
||||
cfg, err := config.Load(in.configPath())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The configuration install.md used to show had an empty catalog and no
|
||||
// control.enabled, which the server refuses as having nothing to serve.
|
||||
if !cfg.ControlPlaneMode() {
|
||||
t.Error("the written configuration does not run a control plane")
|
||||
}
|
||||
if !cfg.AgentTLS.Enabled() || !cfg.Auth.Enabled {
|
||||
t.Error("mutual TLS and authentication should both be on")
|
||||
}
|
||||
for _, f := range []string{"ca.pem", "server.pem", "server-key.pem"} {
|
||||
if _, err := os.Stat(filepath.Join(in.etcDir, f)); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
s, err := store.Open(context.Background(), cfg.Auth.Store)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer s.Close()
|
||||
u, err := s.UserByEmail(context.Background(), "admin@example.com")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !u.IsAdmin || u.State != store.UserActive || u.ProjectID != "default" {
|
||||
t.Errorf("administrator not ready to use: %+v", u)
|
||||
}
|
||||
if !strings.Contains(out.String(), "password: ") {
|
||||
t.Error("the generated password was not shown")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitAgainChangesNothing(t *testing.T) {
|
||||
in, out := newTestInit(t)
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
cert, _ := os.ReadFile(filepath.Join(in.etcDir, "server.pem"))
|
||||
|
||||
again := *in
|
||||
again.changed = false
|
||||
out.Reset()
|
||||
if err := again.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
if again.changed {
|
||||
t.Errorf("a second run changed something:\n%s", out)
|
||||
}
|
||||
if strings.Contains(out.String(), "password: ") {
|
||||
t.Error("a second run must not reset the administrator's password")
|
||||
}
|
||||
if now, _ := os.ReadFile(filepath.Join(in.etcDir, "server.pem")); !bytes.Equal(now, cert) {
|
||||
t.Error("the certificate was reissued without reason")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitWithANewHostKeepsTheOldOnes(t *testing.T) {
|
||||
in, out := newTestInit(t)
|
||||
in.hosts = []string{"old.internal"}
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
in.hosts = []string{"new.internal"}
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
cert, err := readCert(filepath.Join(in.etcDir, "server.pem"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, h := range []string{"old.internal", "new.internal"} {
|
||||
if !slices.Contains(cert.DNSNames, h) {
|
||||
t.Errorf("certificate lost %s: %v", h, cert.DNSNames)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitNeverOverwritesTheConfiguration(t *testing.T) {
|
||||
in, out := newTestInit(t)
|
||||
if err := os.MkdirAll(in.etcDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mine := "# mine\ncontrol:\n enabled: true\nauth:\n enabled: true\n store: " +
|
||||
filepath.Join(in.stateDir, "mine.db") + "\n"
|
||||
if err := os.WriteFile(in.configPath(), []byte(mine), 0o640); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
if got, _ := os.ReadFile(in.configPath()); string(got) != mine {
|
||||
t.Error("an existing server.yaml was rewritten")
|
||||
}
|
||||
// And the store it names is the one used.
|
||||
if _, err := os.Stat(filepath.Join(in.stateDir, "mine.db")); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheInitUnitIsTheShippedUnit(t *testing.T) {
|
||||
b, err := os.ReadFile("../../deploy/systemd/llmgw-server.service")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(b) != serverUnit {
|
||||
t.Error("deploy/systemd/llmgw-server.service and the unit 'server init' writes have drifted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitKeepsCertificatesFromAnAuthorityElsewhere(t *testing.T) {
|
||||
in, out := newTestInit(t)
|
||||
// Issued on another machine by an authority that never comes here.
|
||||
elsewhere := &serverInit{etcDir: in.etcDir, pkiDir: filepath.Join(t.TempDir(), "pki"), out: &bytes.Buffer{}}
|
||||
if err := os.MkdirAll(in.etcDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := elsewhere.ensureCerts(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before, _ := os.ReadFile(filepath.Join(in.etcDir, "server.pem"))
|
||||
|
||||
if err := in.run(context.Background()); err != nil {
|
||||
t.Fatalf("%v\n%s", err, out)
|
||||
}
|
||||
if after, _ := os.ReadFile(filepath.Join(in.etcDir, "server.pem")); !bytes.Equal(before, after) {
|
||||
t.Error("a certificate from another authority was replaced")
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(in.pkiDir, "ca-key.pem")); err == nil {
|
||||
t.Error("an authority was created beside one that already exists elsewhere")
|
||||
}
|
||||
|
||||
in.hosts = []string{"new.internal"}
|
||||
if err := in.run(context.Background()); err == nil {
|
||||
t.Error("--host cannot be honoured without the authority, and should say so")
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,7 @@ func (a *API) Register(mux *http.ServeMux) {
|
||||
mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights)
|
||||
mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
|
||||
mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
|
||||
mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
|
||||
}
|
||||
|
||||
// --- responses ---
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
package adminapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// LogsView is one of a node's logs, as the node sent it.
|
||||
type LogsView struct {
|
||||
Node string `json:"node"`
|
||||
Source string `json:"source"`
|
||||
Lines []string `json:"lines"`
|
||||
// Sources are every log the node can show: "agent", and one per engine
|
||||
// running there.
|
||||
Sources []string `json:"sources"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// maxLogLines bounds a request. The agent keeps two thousand lines of its own
|
||||
// and a few hundred per engine, so more would only be refused further down.
|
||||
const maxLogLines = 2000
|
||||
|
||||
// NodeLogs asks a node for the recent lines of one of its logs.
|
||||
//
|
||||
// Exported because the cluster page renders it server-side, the same as
|
||||
// Nodes.
|
||||
func (a *API) NodeLogs(ctx context.Context, node, source string, lines int) LogsView {
|
||||
if source == "" {
|
||||
source = "agent"
|
||||
}
|
||||
v := LogsView{Node: node, Source: source}
|
||||
lr := a.ctl.LogReader()
|
||||
if lr == nil {
|
||||
v.Error = "this control plane cannot read node logs"
|
||||
return v
|
||||
}
|
||||
if lines <= 0 {
|
||||
lines = 300
|
||||
}
|
||||
out, sources, err := lr.Logs(ctx, node, source, min(lines, maxLogLines))
|
||||
v.Lines, v.Sources = out, sources
|
||||
if err != nil {
|
||||
v.Error = err.Error()
|
||||
}
|
||||
if len(v.Sources) == 0 {
|
||||
v.Sources = []string{"agent"}
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func (a *API) nodeLogs(w http.ResponseWriter, r *http.Request) {
|
||||
lines, _ := strconv.Atoi(r.URL.Query().Get("lines"))
|
||||
v := a.NodeLogs(r.Context(), r.PathValue("id"), r.URL.Query().Get("source"), lines)
|
||||
code := http.StatusOK
|
||||
if v.Error != "" && len(v.Lines) == 0 {
|
||||
code = http.StatusBadGateway
|
||||
}
|
||||
writeJSON(w, code, v)
|
||||
}
|
||||
@@ -155,3 +155,36 @@ func TestAHeartbeatCarriesAnEngineInstalledAfterRegistration(t *testing.T) {
|
||||
t.Errorf("the newly installed engine is not known: %v", n.EngineVersions)
|
||||
}
|
||||
}
|
||||
|
||||
func TestALogRequestIsAnsweredByTheNodeItWasSentTo(t *testing.T) {
|
||||
s := newTestService()
|
||||
c := &conn{nodeID: "n1", send: make(chan *llmgwv1.ControlMessage, 4), closed: make(chan struct{})}
|
||||
s.conns["n1"] = c
|
||||
|
||||
type result struct {
|
||||
lines, sources []string
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
l, src, err := s.Logs(t.Context(), "n1", "agent", 10)
|
||||
done <- result{l, src, err}
|
||||
}()
|
||||
|
||||
req := (<-c.send).GetLogs()
|
||||
if req.GetSource() != "agent" || req.GetLines() != 10 {
|
||||
t.Fatalf("sent %v", req)
|
||||
}
|
||||
s.handle("n1", &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Logs{Logs: &llmgwv1.LogsReply{
|
||||
RequestId: req.GetRequestId(), Lines: []string{"hello"}, Sources: []string{"agent", "shard-1"},
|
||||
}}})
|
||||
|
||||
r := <-done
|
||||
if r.err != nil || len(r.lines) != 1 || r.lines[0] != "hello" || len(r.sources) != 2 {
|
||||
t.Errorf("got %+v", r)
|
||||
}
|
||||
|
||||
if _, _, err := s.Logs(t.Context(), "nobody", "agent", 10); err == nil {
|
||||
t.Error("a node with no stream answered")
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"net/http"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"connectrpc.com/connect"
|
||||
|
||||
@@ -52,6 +53,11 @@ type Service struct {
|
||||
|
||||
// engines is what every node is told to install, set once at start.
|
||||
engines map[string]string
|
||||
|
||||
// logWaits are the log requests awaiting a reply, by request id.
|
||||
logMu sync.Mutex
|
||||
logWaits map[string]chan *llmgwv1.LogsReply
|
||||
logSeq uint64
|
||||
}
|
||||
|
||||
// SetEngines sets the engines every node installs, by name and version.
|
||||
@@ -276,6 +282,15 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) {
|
||||
"last_output", m.Shard.GetLastOutput())
|
||||
}
|
||||
|
||||
case *llmgwv1.AgentMessage_Logs:
|
||||
s.logMu.Lock()
|
||||
ch, ok := s.logWaits[m.Logs.GetRequestId()]
|
||||
delete(s.logWaits, m.Logs.GetRequestId())
|
||||
s.logMu.Unlock()
|
||||
if ok {
|
||||
ch <- m.Logs // buffered; the asker may have given up
|
||||
}
|
||||
|
||||
case *llmgwv1.AgentMessage_Register:
|
||||
// A re-registration on a live stream means the agent's view of its own
|
||||
// hardware changed — a GPU fell out, say. Take it.
|
||||
@@ -511,3 +526,46 @@ func (s *Service) send(nodeID string, msg *llmgwv1.ControlMessage) error {
|
||||
return fmt.Errorf("node %s is not keeping up with commands", nodeID)
|
||||
}
|
||||
}
|
||||
|
||||
// Logs asks a node for the recent lines of one of its logs: "agent", or a
|
||||
// shard id for that engine's output.
|
||||
//
|
||||
// It returns the lines, and every source the node can show.
|
||||
func (s *Service) Logs(ctx context.Context, nodeID, source string, lines int) ([]string, []string, error) {
|
||||
s.logMu.Lock()
|
||||
if s.logWaits == nil {
|
||||
s.logWaits = map[string]chan *llmgwv1.LogsReply{}
|
||||
}
|
||||
s.logSeq++
|
||||
id := fmt.Sprintf("logs-%d", s.logSeq)
|
||||
ch := make(chan *llmgwv1.LogsReply, 1)
|
||||
s.logWaits[id] = ch
|
||||
s.logMu.Unlock()
|
||||
defer func() {
|
||||
s.logMu.Lock()
|
||||
delete(s.logWaits, id)
|
||||
s.logMu.Unlock()
|
||||
}()
|
||||
|
||||
if source == "" {
|
||||
source = "agent"
|
||||
}
|
||||
if err := s.send(nodeID, &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_Logs{
|
||||
Logs: &llmgwv1.LogsRequest{RequestId: id, Source: source, Lines: int32(lines)}, //nolint:gosec // bounded by the caller
|
||||
}}); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
// Bounded, because a page is waiting: an agent older than this request
|
||||
// never answers it at all.
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
select {
|
||||
case r := <-ch:
|
||||
if r.GetError() != "" {
|
||||
return nil, r.GetSources(), errors.New(r.GetError())
|
||||
}
|
||||
return r.GetLines(), r.GetSources(), nil
|
||||
case <-ctx.Done():
|
||||
return nil, nil, fmt.Errorf("node %s did not answer; an agent older than v0.8.4 cannot", nodeID)
|
||||
}
|
||||
}
|
||||
@@ -204,6 +204,17 @@ func (c *Controller) WeightRemover() WeightRemover {
|
||||
return r
|
||||
}
|
||||
|
||||
// LogReader can ask a node for the recent lines of one of its logs.
|
||||
type LogReader interface {
|
||||
Logs(ctx context.Context, nodeID, source string, lines int) (out, sources []string, err error)
|
||||
}
|
||||
|
||||
// LogReader returns the dispatcher's log capability, or nil.
|
||||
func (c *Controller) LogReader() LogReader {
|
||||
r, _ := c.disp.(LogReader)
|
||||
return r
|
||||
}
|
||||
|
||||
// Start runs the reconciler and begins publishing endpoints.
|
||||
func (c *Controller) Start(ctx context.Context) {
|
||||
if c.lead != nil {
|
||||
|
||||
@@ -157,6 +157,8 @@ func (r *Router) Handler() http.Handler {
|
||||
web.NavExtras{
|
||||
Mirror: r.mirror != nil, History: r.history, Window: r.histWindow,
|
||||
}, r.log)))
|
||||
mux.HandleFunc("GET /admin/nodes/{id}/logs", pages.AdminPage(
|
||||
web.NodeLogsPage(r.clusterView, false, version.Version)))
|
||||
pages.RegisterAdminUsers(mux, r.store)
|
||||
// Quotas are written on the same tree the accounts page builds, so
|
||||
// they are mounted together.
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
||||
)
|
||||
|
||||
// LogSource reads a node's logs. Asked for by type assertion on the cluster
|
||||
// source, so a source that cannot — a test's, an older one — leaves the page
|
||||
// saying so rather than every Source having to grow the method.
|
||||
type LogSource interface {
|
||||
NodeLogs(ctx context.Context, node, source string, lines int) adminapi.LogsView
|
||||
}
|
||||
|
||||
type nodeLogsData struct {
|
||||
adminapi.LogsView
|
||||
OnAdminListener bool
|
||||
// Base is where this page lives, which differs between the admin
|
||||
// listener and the client-facing port.
|
||||
Base string
|
||||
Back string
|
||||
Count int
|
||||
Counts []int
|
||||
Follow bool
|
||||
Version string
|
||||
Now string
|
||||
}
|
||||
|
||||
// NodeLogsPage shows the recent lines of one of a node's logs: the agent's
|
||||
// own, or an engine's output.
|
||||
//
|
||||
// What a node did used to be learned by ssh through a jump host and a journal
|
||||
// only root may read — which, when an engine install fails at two in the
|
||||
// morning, is several steps between the page that says it failed and the line
|
||||
// that says why.
|
||||
func NodeLogsPage(src Source, onAdminListener bool, version string) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
node := r.PathValue("id")
|
||||
d := nodeLogsData{
|
||||
OnAdminListener: onAdminListener,
|
||||
Base: "/admin/nodes/" + node + "/logs",
|
||||
Back: "/admin",
|
||||
Counts: []int{100, 300, 1000, 2000},
|
||||
Follow: r.URL.Query().Get("follow") != "",
|
||||
Version: version,
|
||||
Now: time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
if onAdminListener {
|
||||
d.Base, d.Back = "/nodes/"+node+"/logs", "/"
|
||||
}
|
||||
d.Count, _ = strconv.Atoi(r.URL.Query().Get("lines"))
|
||||
if d.Count <= 0 {
|
||||
d.Count = 300
|
||||
}
|
||||
ls, ok := src.(LogSource)
|
||||
if !ok {
|
||||
d.LogsView = adminapi.LogsView{Node: node, Error: "this control plane cannot read node logs"}
|
||||
} else {
|
||||
d.LogsView = ls.NodeLogs(r.Context(), node, r.URL.Query().Get("source"), d.Count)
|
||||
}
|
||||
noStore(w)
|
||||
render(w, "node_logs", d)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
|
||||
)
|
||||
|
||||
type logsFleet struct{ fakeSource }
|
||||
|
||||
func (logsFleet) NodeLogs(_ context.Context, node, source string, _ int) adminapi.LogsView {
|
||||
return adminapi.LogsView{Node: node, Source: source, Lines: []string{"installed an engine <b>"},
|
||||
Sources: []string{"agent", "shard-1"}}
|
||||
}
|
||||
|
||||
func TestANodesLogIsShownWithItsOtherSources(t *testing.T) {
|
||||
req := httptest.NewRequest("GET", "/admin/nodes/n6/logs?source=agent", nil)
|
||||
req.SetPathValue("id", "n6")
|
||||
rec := httptest.NewRecorder()
|
||||
NodeLogsPage(logsFleet{}, false, "test")(rec, req)
|
||||
|
||||
body := rec.Body.String()
|
||||
for _, want := range []string{"installed an engine <b>", "source=shard-1", "/admin/nodes/n6/logs"} {
|
||||
if !strings.Contains(body, want) {
|
||||
t.Errorf("page lacks %q", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -103,7 +103,8 @@ requested whenever a fallback served it.</p>
|
||||
{{if .Nodes}}
|
||||
<table><tr><th>Node</th><th>State</th><th class="num">GPUs</th><th class="num">RAM</th><th class="num">Disk</th><th>Engines</th><th>Classes</th><th>Fabrics</th><th class="num">Models</th><th class="num">Replicas</th><th>Last seen</th></tr>
|
||||
{{range .Nodes}}<tr>
|
||||
<td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}</td>
|
||||
<td><code>{{.ID}}</code>{{if .MgmtAddr}} <span class="note">{{.MgmtAddr}}</span>{{end}}
|
||||
<a class="note" href="{{if $.OnAdminListener}}/nodes/{{.ID}}/logs{{else}}/admin/nodes/{{.ID}}/logs{{end}}">logs</a></td>
|
||||
<td class="state state-{{.State}}">{{.State}}</td>
|
||||
<td class="num">{{.GPUFree}}/{{.GPUTotal}}</td>
|
||||
<td class="num">{{if .MemoryGB}}{{.MemoryGB}}G{{else}}—{{end}}</td>
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
{{define "node_logs"}}<!doctype html>
|
||||
<html lang="en"><head>
|
||||
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
||||
{{if .Follow}}<meta http-equiv="refresh" content="5">{{end}}
|
||||
<title>llmgw — {{.Node}} logs</title><link rel="stylesheet" href="/_static/base.css">
|
||||
</head><body><div class="wrap">
|
||||
<header>
|
||||
<h1>{{.Node}}</h1>
|
||||
<p class="sub">{{if eq .Source "agent"}}the agent's log{{else}}output of engine <code>{{.Source}}</code>{{end}} · last {{.Count}} lines</p>
|
||||
<nav><a href="{{.Back}}">cluster</a></nav>
|
||||
</header>
|
||||
|
||||
<p>
|
||||
{{- $d := . -}}
|
||||
{{range .Sources}}<a href="{{$d.Base}}?source={{.}}&lines={{$d.Count}}{{if $d.Follow}}&follow=1{{end}}"{{if eq . $d.Source}} aria-current="page"><strong>{{.}}</strong>{{else}}>{{.}}{{end}}</a> {{end}}
|
||||
</p>
|
||||
<p class="note">
|
||||
{{range .Counts}}<a href="{{$d.Base}}?source={{$d.Source}}&lines={{.}}{{if $d.Follow}}&follow=1{{end}}">{{.}}</a> {{end}}lines ·
|
||||
{{if .Follow}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}">stop following</a>{{else}}<a href="{{.Base}}?source={{.Source}}&lines={{.Count}}&follow=1">follow</a> (refresh every 5s){{end}}
|
||||
</p>
|
||||
|
||||
{{if .Error}}<p class="bad">{{.Error}}</p>{{end}}
|
||||
{{if .Lines}}<pre id="log">{{range .Lines}}{{.}}
|
||||
{{end}}</pre>
|
||||
{{/* Newest at the bottom, like a terminal, and scrolled there: the line
|
||||
worth reading is almost always the last one. */}}
|
||||
<script>window.scrollTo(0, document.body.scrollHeight);</script>
|
||||
{{else if not .Error}}<p class="empty">Nothing logged yet.</p>{{end}}
|
||||
|
||||
<footer class="note">llmgw {{.Version}} · {{.Now}}</footer>
|
||||
</div></body></html>{{end}}
|
||||
Reference in new issue
Block a user