Compare commits

...
4 Commits
Author SHA1 Message Date
Ilker ManapandClaude Opus 5.5 8097a5616d Describe v0.8.2
CI / test (push) Successful in 2m11s
Release / build (push) Successful in 2m56s
Release / notes (push) Successful in 3s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 11:50:12 +02:00
Ilker ManapandClaude Opus 5.5 f5d6210b86 Install engines on any glibc system, with its own package manager
The llama.cpp build is published from Ubuntu but is not Ubuntu's in any way
that matters: it needs glibc 2.34 and GCC 12's libstdc++, which Debian 12
and RHEL 10 have and RHEL 9 does not. Refusing every system not called
"ubuntu" turned away the first two for their name. The build is now tried
everywhere, and ldd decides.

What ldd says is made readable. "/lib64/libstdc++.so.6: version
`GLIBCXX_3.4.30' not found" is a system too old for the build, and is
reported as "libstdc++.so.6 is too old (needs GLIBCXX_3.4.30)", once —
rather than as a missing library that is plainly installed.

Packages are named per family — libgomp1 for Debian's, libgomp for Red
Hat's — and installed with whichever of apt-get, dnf, yum or zypper the
machine has, decided from ID_LIKE rather than a list of distribution names.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 11:10:06 +02:00
Ilker ManapandClaude Opus 5.5 7d97c6b74f Install a node's engines on the node, instead of trusting what is there
A node ran whatever llama-server somebody had once unpacked on it. Onboarding
checked an engine only when told its path, installed none, and a node with
nothing on it registered as ready. The one time that went wrong it went wrong
invisibly: the binary was there, executable, and could not start for want of
libgomp1, while the node reported healthy.

The control plane now says which engines every node has, at which version —
`engines:` in server.yaml, defaulting to the llama.cpp build this release was
checked against — and each agent converges on it. What only the node can
know, the node works out: the build for its processor and driver (CUDA with
its runtime when an NVIDIA driver is loaded, CPU otherwise), the OS packages
that build needs from its own package sources, and whether the result runs.
Every archive is checked against the sha256 GitHub publishes for it and
unpacked refusing any path or link that leaves the install directory. ldd
names a missing library outright; running `--version` proves the rest. Only
then is a directory marked installed, and it is checked again every ten
minutes, because an unrelated upgrade can remove a library.

An engine is reported only once installed and seen to run, or when the
agent's configuration names it on purpose — never because something happened
to be on PATH. The agent now says when it has probed, so an empty list means
"has none" rather than "too old to say": a node rebuilt from nothing is no
longer a candidate for every model at the moment it can run none of them.
Placement's refusal carries the node's own words, "installing: downloading
llama-b10991-bin-ubuntu-x64.tar.gz (16 MB)", and so do `ctl nodes` and the
cluster page, along with a warning for a CPU without AVX2 — the VM CPU type
that makes everything five times slower and nothing look wrong.

Agents also report their OS, architecture, vector extensions and NVIDIA
driver. Onboarding installs llmgw itself from the CI release for the node's
architecture, checked against the release's checksums, rather than copying
whatever binary the operator's machine had.

vLLM and SGLang are not installed yet: they are Python environments of
several gigabytes, not an archive, and still have to be named in the agent's
configuration.

Also fixes a race on the weights fingerprint, which two heartbeat loops touch
during a failover.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 10:59:59 +02:00
Ilker ManapandClaude Opus 5.5 fa7f90880d Put the changelog on the release page
CI / test (push) Successful in 1m42s
Every release on Gitea was a list of binaries and nothing else. Gitea makes
an empty release for each pushed tag, and the workflow only ever attached
files to it, so what had changed was written in CHANGELOG.md and nowhere a
release was looked at.

A job after the build now copies the tag's section of the changelog onto its
release, and fails if there is none — a version nobody described should say
so where it is seen. Run by hand, it syncs every version the changelog has,
which is how the releases made before this get their notes. The changelog
stays the one place notes are written; this copies it.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 10:32:58 +02:00
30 changed files with 2471 additions and 192 deletions

No files matched your search

+28
View File
@@ -150,3 +150,31 @@ jobs:
|| { echo "upload of $(basename "$f") was refused" >&2; exit 1; }
done
echo "attached $(ls dist | wc -l) files to $tag"
# The release page says what changed, copied from CHANGELOG.md. Gitea makes
# an empty release for every tag and the job above only attaches files, so
# every release until this existed was a list of binaries and nothing else.
#
# After the build, because that is what guarantees the release exists. Run by
# hand from a branch it syncs every version in the changelog, which is how
# the releases made before this got their notes.
notes:
needs: build
runs-on: ubuntu-latest
permissions:
contents: write
steps:
- uses: actions/checkout@v4
- name: Copy the changelog onto the release
env:
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
GITEA_INSTANCE: ${{ github.server_url }}
GITEA_REPO: ${{ github.repository }}
run: |
set -euo pipefail
if [ "${GITHUB_REF_TYPE:-}" = "tag" ]; then
python3 scripts/gitea/release_notes.py --tag "${GITHUB_REF_NAME}"
else
python3 scripts/gitea/release_notes.py
fi
+39
View File
@@ -18,6 +18,45 @@ 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.2 — 2026-09-28
A node no longer runs whatever engine somebody once put on it. It installs its
own.
### Engines are installed by the node
`engines:` in server.yaml names the engines every node has and their versions,
defaulting to the llama.cpp build this release was checked against (b10991).
Each agent converges on it: it picks the build for its hardware — CUDA with its
runtime when an NVIDIA driver is loaded, CPU otherwise — installs the OS
packages that build needs with the machine's own package manager, downloads the
archive and checks it against the sha256 GitHub publishes, unpacks it refusing
anything that would land outside its directory, and runs it. `ldd` names a
missing library or a system too old for the build ("libstdc++.so.6 is too old
(needs GLIBCXX_3.4.30)"); `--version` proves the rest. Installs are checked
again every ten minutes.
An engine is reported only once installed and seen to run, or when the agent's
configuration names it on purpose. A node that has installed nothing yet is not
a candidate for any model, where before it was a candidate for all of them.
`ctl nodes`, the cluster page and placement's refusals say where each
installation stands, and warn about a CPU without AVX2.
Not every system is refused for its name: the build runs on Ubuntu 22.04+,
Debian 12+ and RHEL-family 10, and `ldd` decides. vLLM and SGLang are not yet
installed this way and must still be named in the agent's configuration.
### Onboarding
Onboarding installs llmgw from the CI release for the node's architecture,
checked against the release's checksums, and writes an agent configuration of
only the control plane's addresses and the TLS files.
### Fixes
- A race on the weights fingerprint, which two heartbeat loops touched during a
failover.
## v0.8.1 — 2026-09-28
Nothing here changes a configuration format: every file that worked with
+475 -103
View File
@@ -90,7 +90,7 @@ func (x ShardStatus_State) Number() protoreflect.EnumNumber {
// Deprecated: Use ShardStatus_State.Descriptor instead.
func (ShardStatus_State) EnumDescriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{9, 0}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{13, 0}
}
// AgentMessage is anything a node tells the control plane.
@@ -202,6 +202,7 @@ type ControlMessage struct {
// *ControlMessage_Drain
// *ControlMessage_Ack
// *ControlMessage_RemoveWeights
// *ControlMessage_Provision
Msg isControlMessage_Msg `protobuf_oneof:"msg"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
@@ -289,6 +290,15 @@ func (x *ControlMessage) GetRemoveWeights() *RemoveWeights {
return nil
}
func (x *ControlMessage) GetProvision() *Provision {
if x != nil {
if x, ok := x.Msg.(*ControlMessage_Provision); ok {
return x.Provision
}
}
return nil
}
type isControlMessage_Msg interface {
isControlMessage_Msg()
}
@@ -313,6 +323,10 @@ type ControlMessage_RemoveWeights struct {
RemoveWeights *RemoveWeights `protobuf:"bytes,5,opt,name=remove_weights,json=removeWeights,proto3,oneof"`
}
type ControlMessage_Provision struct {
Provision *Provision `protobuf:"bytes,6,opt,name=provision,proto3,oneof"`
}
func (*ControlMessage_Start) isControlMessage_Msg() {}
func (*ControlMessage_Stop) isControlMessage_Msg() {}
@@ -323,6 +337,8 @@ func (*ControlMessage_Ack) isControlMessage_Msg() {}
func (*ControlMessage_RemoveWeights) isControlMessage_Msg() {}
func (*ControlMessage_Provision) 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.
@@ -381,7 +397,15 @@ type Register struct {
//
// Empty means the node was not asked to look, exactly as for weights, and
// constrains nothing.
Storage []*Storage `protobuf:"bytes,12,rep,name=storage,proto3" json:"storage,omitempty"`
Storage []*Storage `protobuf:"bytes,12,rep,name=storage,proto3" json:"storage,omitempty"`
// host is what the machine is, so the control plane can have the mirror
// fetch software built for it rather than assuming every node matches the
// first.
Host *Host `protobuf:"bytes,13,opt,name=host,proto3" json:"host,omitempty"`
// engines_probed says engine_versions is an answer. Without it an empty map
// meant "an agent too old to say" and constrained nothing — so a freshly
// built node with no engine installed yet was a candidate for every model.
EnginesProbed bool `protobuf:"varint,14,opt,name=engines_probed,json=enginesProbed,proto3" json:"engines_probed,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -500,6 +524,113 @@ func (x *Register) GetStorage() []*Storage {
return nil
}
func (x *Register) GetHost() *Host {
if x != nil {
return x.Host
}
return nil
}
func (x *Register) GetEnginesProbed() bool {
if x != nil {
return x.EnginesProbed
}
return false
}
// Host describes the operating system and processor of a node.
type Host struct {
state protoimpl.MessageState `protogen:"open.v1"`
// From /etc/os-release: "ubuntu", "24.04", "noble".
OsId string `protobuf:"bytes,1,opt,name=os_id,json=osId,proto3" json:"os_id,omitempty"`
OsVersion string `protobuf:"bytes,2,opt,name=os_version,json=osVersion,proto3" json:"os_version,omitempty"`
OsCodename string `protobuf:"bytes,3,opt,name=os_codename,json=osCodename,proto3" json:"os_codename,omitempty"`
// Debian's spelling, which is what package indexes are keyed by: "amd64",
// "arm64".
Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
// The vector extensions inference cares about, of those the CPU has. A VM
// given QEMU's default CPU model has none of them, and runs, and is five
// times slower, and nothing says why.
CpuFlags []string `protobuf:"bytes,5,rep,name=cpu_flags,json=cpuFlags,proto3" json:"cpu_flags,omitempty"`
// nvidia_driver is the driver version, empty when there is none. A card
// without a driver is not a GPU anything can use.
NvidiaDriver string `protobuf:"bytes,6,opt,name=nvidia_driver,json=nvidiaDriver,proto3" json:"nvidia_driver,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Host) Reset() {
*x = Host{}
mi := &file_llmgw_v1_agent_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Host) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Host) ProtoMessage() {}
func (x *Host) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Host.ProtoReflect.Descriptor instead.
func (*Host) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{3}
}
func (x *Host) GetOsId() string {
if x != nil {
return x.OsId
}
return ""
}
func (x *Host) GetOsVersion() string {
if x != nil {
return x.OsVersion
}
return ""
}
func (x *Host) GetOsCodename() string {
if x != nil {
return x.OsCodename
}
return ""
}
func (x *Host) GetArch() string {
if x != nil {
return x.Arch
}
return ""
}
func (x *Host) GetCpuFlags() []string {
if x != nil {
return x.CpuFlags
}
return nil
}
func (x *Host) GetNvidiaDriver() string {
if x != nil {
return x.NvidiaDriver
}
return ""
}
// Storage is one filesystem a node keeps weights on.
//
// Per filesystem rather than per configured directory: free space belongs to
@@ -522,7 +653,7 @@ type Storage struct {
func (x *Storage) Reset() {
*x = Storage{}
mi := &file_llmgw_v1_agent_proto_msgTypes[3]
mi := &file_llmgw_v1_agent_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -534,7 +665,7 @@ func (x *Storage) String() string {
func (*Storage) ProtoMessage() {}
func (x *Storage) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[3]
mi := &file_llmgw_v1_agent_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -547,7 +678,7 @@ func (x *Storage) ProtoReflect() protoreflect.Message {
// Deprecated: Use Storage.ProtoReflect.Descriptor instead.
func (*Storage) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{3}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{4}
}
func (x *Storage) GetPath() string {
@@ -614,7 +745,7 @@ type RunningShard struct {
func (x *RunningShard) Reset() {
*x = RunningShard{}
mi := &file_llmgw_v1_agent_proto_msgTypes[4]
mi := &file_llmgw_v1_agent_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -626,7 +757,7 @@ func (x *RunningShard) String() string {
func (*RunningShard) ProtoMessage() {}
func (x *RunningShard) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[4]
mi := &file_llmgw_v1_agent_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -639,7 +770,7 @@ func (x *RunningShard) ProtoReflect() protoreflect.Message {
// Deprecated: Use RunningShard.ProtoReflect.Descriptor instead.
func (*RunningShard) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{4}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{5}
}
func (x *RunningShard) GetReplicaId() string {
@@ -776,7 +907,7 @@ type GPU struct {
func (x *GPU) Reset() {
*x = GPU{}
mi := &file_llmgw_v1_agent_proto_msgTypes[5]
mi := &file_llmgw_v1_agent_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -788,7 +919,7 @@ func (x *GPU) String() string {
func (*GPU) ProtoMessage() {}
func (x *GPU) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[5]
mi := &file_llmgw_v1_agent_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -801,7 +932,7 @@ func (x *GPU) ProtoReflect() protoreflect.Message {
// Deprecated: Use GPU.ProtoReflect.Descriptor instead.
func (*GPU) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{5}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{6}
}
func (x *GPU) GetIndex() int32 {
@@ -923,7 +1054,7 @@ type FabricAttachment struct {
func (x *FabricAttachment) Reset() {
*x = FabricAttachment{}
mi := &file_llmgw_v1_agent_proto_msgTypes[6]
mi := &file_llmgw_v1_agent_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -935,7 +1066,7 @@ func (x *FabricAttachment) String() string {
func (*FabricAttachment) ProtoMessage() {}
func (x *FabricAttachment) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[6]
mi := &file_llmgw_v1_agent_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -948,7 +1079,7 @@ func (x *FabricAttachment) ProtoReflect() protoreflect.Message {
// Deprecated: Use FabricAttachment.ProtoReflect.Descriptor instead.
func (*FabricAttachment) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{6}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{7}
}
func (x *FabricAttachment) GetFabric() string {
@@ -990,7 +1121,7 @@ type LocalWeights struct {
func (x *LocalWeights) Reset() {
*x = LocalWeights{}
mi := &file_llmgw_v1_agent_proto_msgTypes[7]
mi := &file_llmgw_v1_agent_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1002,7 +1133,7 @@ func (x *LocalWeights) String() string {
func (*LocalWeights) ProtoMessage() {}
func (x *LocalWeights) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[7]
mi := &file_llmgw_v1_agent_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1015,7 +1146,7 @@ func (x *LocalWeights) ProtoReflect() protoreflect.Message {
// Deprecated: Use LocalWeights.ProtoReflect.Descriptor instead.
func (*LocalWeights) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{7}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{8}
}
func (x *LocalWeights) GetPath() string {
@@ -1070,13 +1201,18 @@ type Heartbeat struct {
// somebody restarts the agent — which is a step nobody knows to take.
EnginesChanged bool `protobuf:"varint,8,opt,name=engines_changed,json=enginesChanged,proto3" json:"engines_changed,omitempty"`
EngineVersions map[string]string `protobuf:"bytes,9,rep,name=engine_versions,json=engineVersions,proto3" json:"engine_versions,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// provision is where installing the node's software stands, sent when it
// changed. The flag makes an empty list mean "nothing to install" rather
// than "nothing to say", as for weights.
ProvisionChanged bool `protobuf:"varint,10,opt,name=provision_changed,json=provisionChanged,proto3" json:"provision_changed,omitempty"`
Provision []*ProvisionStatus `protobuf:"bytes,11,rep,name=provision,proto3" json:"provision,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Heartbeat) Reset() {
*x = Heartbeat{}
mi := &file_llmgw_v1_agent_proto_msgTypes[8]
mi := &file_llmgw_v1_agent_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1088,7 +1224,7 @@ func (x *Heartbeat) String() string {
func (*Heartbeat) ProtoMessage() {}
func (x *Heartbeat) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[8]
mi := &file_llmgw_v1_agent_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1101,7 +1237,7 @@ func (x *Heartbeat) ProtoReflect() protoreflect.Message {
// Deprecated: Use Heartbeat.ProtoReflect.Descriptor instead.
func (*Heartbeat) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{8}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{9}
}
func (x *Heartbeat) GetNodeId() string {
@@ -1167,6 +1303,206 @@ func (x *Heartbeat) GetEngineVersions() map[string]string {
return nil
}
func (x *Heartbeat) GetProvisionChanged() bool {
if x != nil {
return x.ProvisionChanged
}
return false
}
func (x *Heartbeat) GetProvision() []*ProvisionStatus {
if x != nil {
return x.Provision
}
return nil
}
// Provision is the software a node should have: which engines, at which
// versions. Nothing more.
//
// Everything else is the node's to work out, because only the node knows it:
// which build fits its hardware, where that build is published, what its
// operating system is missing, and whether the result actually runs. The agent
// converges on this — installs what is absent, checks what is there — and
// reports each engine through ProvisionStatus. Nothing is taken on trust from
// what happens to be on the machine: an engine is offered to placement only
// once the agent has installed it and seen it run.
type Provision struct {
state protoimpl.MessageState `protogen:"open.v1"`
Engines []*EngineWant `protobuf:"bytes,1,rep,name=engines,proto3" json:"engines,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Provision) Reset() {
*x = Provision{}
mi := &file_llmgw_v1_agent_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Provision) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Provision) ProtoMessage() {}
func (x *Provision) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[10]
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 Provision.ProtoReflect.Descriptor instead.
func (*Provision) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{10}
}
func (x *Provision) GetEngines() []*EngineWant {
if x != nil {
return x.Engines
}
return nil
}
// EngineWant names one engine at one version.
type EngineWant struct {
state protoimpl.MessageState `protogen:"open.v1"`
Engine string `protobuf:"bytes,1,opt,name=engine,proto3" json:"engine,omitempty"` // "llamacpp"
Version string `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"` // "b10991"
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *EngineWant) Reset() {
*x = EngineWant{}
mi := &file_llmgw_v1_agent_proto_msgTypes[11]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *EngineWant) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*EngineWant) ProtoMessage() {}
func (x *EngineWant) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[11]
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 EngineWant.ProtoReflect.Descriptor instead.
func (*EngineWant) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{11}
}
func (x *EngineWant) GetEngine() string {
if x != nil {
return x.Engine
}
return ""
}
func (x *EngineWant) GetVersion() string {
if x != nil {
return x.Version
}
return ""
}
// ProvisionStatus is where one engine's installation stands on a node.
type ProvisionStatus struct {
state protoimpl.MessageState `protogen:"open.v1"`
Engine string `protobuf:"bytes,1,opt,name=engine,proto3" json:"engine,omitempty"`
Version string `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
Variant string `protobuf:"bytes,3,opt,name=variant,proto3" json:"variant,omitempty"`
// "installing", "ready", "failed", or "unsupported" when no build fits.
State string `protobuf:"bytes,4,opt,name=state,proto3" json:"state,omitempty"`
// detail says what is happening or what went wrong, in words for a person.
Detail string `protobuf:"bytes,5,opt,name=detail,proto3" json:"detail,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ProvisionStatus) Reset() {
*x = ProvisionStatus{}
mi := &file_llmgw_v1_agent_proto_msgTypes[12]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ProvisionStatus) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ProvisionStatus) ProtoMessage() {}
func (x *ProvisionStatus) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[12]
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 ProvisionStatus.ProtoReflect.Descriptor instead.
func (*ProvisionStatus) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{12}
}
func (x *ProvisionStatus) GetEngine() string {
if x != nil {
return x.Engine
}
return ""
}
func (x *ProvisionStatus) GetVersion() string {
if x != nil {
return x.Version
}
return ""
}
func (x *ProvisionStatus) GetVariant() string {
if x != nil {
return x.Variant
}
return ""
}
func (x *ProvisionStatus) GetState() string {
if x != nil {
return x.State
}
return ""
}
func (x *ProvisionStatus) GetDetail() string {
if x != nil {
return x.Detail
}
return ""
}
// ShardStatus reports a shard's lifecycle transition.
type ShardStatus struct {
state protoimpl.MessageState `protogen:"open.v1"`
@@ -1193,7 +1529,7 @@ type ShardStatus struct {
func (x *ShardStatus) Reset() {
*x = ShardStatus{}
mi := &file_llmgw_v1_agent_proto_msgTypes[9]
mi := &file_llmgw_v1_agent_proto_msgTypes[13]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1205,7 +1541,7 @@ func (x *ShardStatus) String() string {
func (*ShardStatus) ProtoMessage() {}
func (x *ShardStatus) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[9]
mi := &file_llmgw_v1_agent_proto_msgTypes[13]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1218,7 +1554,7 @@ func (x *ShardStatus) ProtoReflect() protoreflect.Message {
// Deprecated: Use ShardStatus.ProtoReflect.Descriptor instead.
func (*ShardStatus) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{9}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{13}
}
func (x *ShardStatus) GetNodeId() string {
@@ -1292,7 +1628,7 @@ type FetchProgress struct {
func (x *FetchProgress) Reset() {
*x = FetchProgress{}
mi := &file_llmgw_v1_agent_proto_msgTypes[10]
mi := &file_llmgw_v1_agent_proto_msgTypes[14]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1304,7 +1640,7 @@ func (x *FetchProgress) String() string {
func (*FetchProgress) ProtoMessage() {}
func (x *FetchProgress) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[10]
mi := &file_llmgw_v1_agent_proto_msgTypes[14]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1317,7 +1653,7 @@ func (x *FetchProgress) ProtoReflect() protoreflect.Message {
// Deprecated: Use FetchProgress.ProtoReflect.Descriptor instead.
func (*FetchProgress) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{10}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{14}
}
func (x *FetchProgress) GetSource() string {
@@ -1373,7 +1709,7 @@ type StartShard struct {
func (x *StartShard) Reset() {
*x = StartShard{}
mi := &file_llmgw_v1_agent_proto_msgTypes[11]
mi := &file_llmgw_v1_agent_proto_msgTypes[15]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1385,7 +1721,7 @@ func (x *StartShard) String() string {
func (*StartShard) ProtoMessage() {}
func (x *StartShard) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[11]
mi := &file_llmgw_v1_agent_proto_msgTypes[15]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1398,7 +1734,7 @@ func (x *StartShard) ProtoReflect() protoreflect.Message {
// Deprecated: Use StartShard.ProtoReflect.Descriptor instead.
func (*StartShard) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{11}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{15}
}
func (x *StartShard) GetReplicaId() string {
@@ -1525,7 +1861,7 @@ type EngineProfile struct {
func (x *EngineProfile) Reset() {
*x = EngineProfile{}
mi := &file_llmgw_v1_agent_proto_msgTypes[12]
mi := &file_llmgw_v1_agent_proto_msgTypes[16]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1537,7 +1873,7 @@ func (x *EngineProfile) String() string {
func (*EngineProfile) ProtoMessage() {}
func (x *EngineProfile) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[12]
mi := &file_llmgw_v1_agent_proto_msgTypes[16]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1550,7 +1886,7 @@ func (x *EngineProfile) ProtoReflect() protoreflect.Message {
// Deprecated: Use EngineProfile.ProtoReflect.Descriptor instead.
func (*EngineProfile) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{12}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{16}
}
func (x *EngineProfile) GetName() string {
@@ -1646,7 +1982,7 @@ type RemoveWeights struct {
func (x *RemoveWeights) Reset() {
*x = RemoveWeights{}
mi := &file_llmgw_v1_agent_proto_msgTypes[13]
mi := &file_llmgw_v1_agent_proto_msgTypes[17]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1658,7 +1994,7 @@ func (x *RemoveWeights) String() string {
func (*RemoveWeights) ProtoMessage() {}
func (x *RemoveWeights) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[13]
mi := &file_llmgw_v1_agent_proto_msgTypes[17]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1671,7 +2007,7 @@ func (x *RemoveWeights) ProtoReflect() protoreflect.Message {
// Deprecated: Use RemoveWeights.ProtoReflect.Descriptor instead.
func (*RemoveWeights) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{13}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{17}
}
func (x *RemoveWeights) GetPaths() []string {
@@ -1691,7 +2027,7 @@ type StopShard struct {
func (x *StopShard) Reset() {
*x = StopShard{}
mi := &file_llmgw_v1_agent_proto_msgTypes[14]
mi := &file_llmgw_v1_agent_proto_msgTypes[18]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1703,7 +2039,7 @@ func (x *StopShard) String() string {
func (*StopShard) ProtoMessage() {}
func (x *StopShard) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[14]
mi := &file_llmgw_v1_agent_proto_msgTypes[18]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1716,7 +2052,7 @@ func (x *StopShard) ProtoReflect() protoreflect.Message {
// Deprecated: Use StopShard.ProtoReflect.Descriptor instead.
func (*StopShard) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{14}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{18}
}
func (x *StopShard) GetShardId() string {
@@ -1736,7 +2072,7 @@ type DrainShard struct {
func (x *DrainShard) Reset() {
*x = DrainShard{}
mi := &file_llmgw_v1_agent_proto_msgTypes[15]
mi := &file_llmgw_v1_agent_proto_msgTypes[19]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1748,7 +2084,7 @@ func (x *DrainShard) String() string {
func (*DrainShard) ProtoMessage() {}
func (x *DrainShard) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[15]
mi := &file_llmgw_v1_agent_proto_msgTypes[19]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1761,7 +2097,7 @@ func (x *DrainShard) ProtoReflect() protoreflect.Message {
// Deprecated: Use DrainShard.ProtoReflect.Descriptor instead.
func (*DrainShard) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{15}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{19}
}
func (x *DrainShard) GetShardId() string {
@@ -1783,7 +2119,7 @@ type Ack struct {
func (x *Ack) Reset() {
*x = Ack{}
mi := &file_llmgw_v1_agent_proto_msgTypes[16]
mi := &file_llmgw_v1_agent_proto_msgTypes[20]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -1795,7 +2131,7 @@ func (x *Ack) String() string {
func (*Ack) ProtoMessage() {}
func (x *Ack) ProtoReflect() protoreflect.Message {
mi := &file_llmgw_v1_agent_proto_msgTypes[16]
mi := &file_llmgw_v1_agent_proto_msgTypes[20]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -1808,7 +2144,7 @@ func (x *Ack) ProtoReflect() protoreflect.Message {
// Deprecated: Use Ack.ProtoReflect.Descriptor instead.
func (*Ack) Descriptor() ([]byte, []int) {
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{16}
return file_llmgw_v1_agent_proto_rawDescGZIP(), []int{20}
}
func (x *Ack) GetMessageId() string {
@@ -1841,14 +2177,15 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\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\"\x83\x02\n" +
"\x03msg\"\xb8\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\rremoveWeightsB\x05\n" +
"\x03msg\"\xa7\x05\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" +
"\x03msg\"\xf2\x05\n" +
"\bRegister\x12\x17\n" +
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1a\n" +
"\bhostname\x18\x02 \x01(\tR\bhostname\x12\x1b\n" +
@@ -1862,13 +2199,24 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x0fmemory_total_mb\x18\n" +
" \x01(\x03R\rmemoryTotalMb\x120\n" +
"\aweights\x18\v \x03(\v2\x16.llmgw.v1.LocalWeightsR\aweights\x12+\n" +
"\astorage\x18\f \x03(\v2\x11.llmgw.v1.StorageR\astorage\x1a9\n" +
"\astorage\x18\f \x03(\v2\x11.llmgw.v1.StorageR\astorage\x12\"\n" +
"\x04host\x18\r \x01(\v2\x0e.llmgw.v1.HostR\x04host\x12%\n" +
"\x0eengines_probed\x18\x0e \x01(\bR\renginesProbed\x1a9\n" +
"\vLabelsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1aA\n" +
"\x13EngineVersionsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"u\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xb1\x01\n" +
"\x04Host\x12\x13\n" +
"\x05os_id\x18\x01 \x01(\tR\x04osId\x12\x1d\n" +
"\n" +
"os_version\x18\x02 \x01(\tR\tosVersion\x12\x1f\n" +
"\vos_codename\x18\x03 \x01(\tR\n" +
"osCodename\x12\x12\n" +
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1b\n" +
"\tcpu_flags\x18\x05 \x03(\tR\bcpuFlags\x12#\n" +
"\rnvidia_driver\x18\x06 \x01(\tR\fnvidiaDriver\"u\n" +
"\aStorage\x12\x12\n" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x1d\n" +
"\n" +
@@ -1925,7 +2273,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\x04path\x18\x01 \x01(\tR\x04path\x12\x1d\n" +
"\n" +
"size_bytes\x18\x02 \x01(\x03R\tsizeBytes\x12\x16\n" +
"\x06shared\x18\x03 \x01(\bR\x06shared\"\xe2\x03\n" +
"\x06shared\x18\x03 \x01(\bR\x06shared\"\xc8\x04\n" +
"\tHeartbeat\x12\x17\n" +
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12+\n" +
"\x12gpu_memory_used_mb\x18\x02 \x03(\x05R\x0fgpuMemoryUsedMb\x12!\n" +
@@ -1935,10 +2283,25 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
"\aweights\x18\x06 \x03(\v2\x16.llmgw.v1.LocalWeightsR\aweights\x12&\n" +
"\x0fmemory_total_mb\x18\a \x01(\x03R\rmemoryTotalMb\x12'\n" +
"\x0fengines_changed\x18\b \x01(\bR\x0eenginesChanged\x12P\n" +
"\x0fengine_versions\x18\t \x03(\v2'.llmgw.v1.Heartbeat.EngineVersionsEntryR\x0eengineVersions\x1aA\n" +
"\x0fengine_versions\x18\t \x03(\v2'.llmgw.v1.Heartbeat.EngineVersionsEntryR\x0eengineVersions\x12+\n" +
"\x11provision_changed\x18\n" +
" \x01(\bR\x10provisionChanged\x127\n" +
"\tprovision\x18\v \x03(\v2\x19.llmgw.v1.ProvisionStatusR\tprovision\x1aA\n" +
"\x13EngineVersionsEntry\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xbb\x03\n" +
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\";\n" +
"\tProvision\x12.\n" +
"\aengines\x18\x01 \x03(\v2\x14.llmgw.v1.EngineWantR\aengines\">\n" +
"\n" +
"EngineWant\x12\x16\n" +
"\x06engine\x18\x01 \x01(\tR\x06engine\x12\x18\n" +
"\aversion\x18\x02 \x01(\tR\aversion\"\x8b\x01\n" +
"\x0fProvisionStatus\x12\x16\n" +
"\x06engine\x18\x01 \x01(\tR\x06engine\x12\x18\n" +
"\aversion\x18\x02 \x01(\tR\aversion\x12\x18\n" +
"\avariant\x18\x03 \x01(\tR\avariant\x12\x14\n" +
"\x05state\x18\x04 \x01(\tR\x05state\x12\x16\n" +
"\x06detail\x18\x05 \x01(\tR\x06detail\"\xbb\x03\n" +
"\vShardStatus\x12\x17\n" +
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x1d\n" +
"\n" +
@@ -2035,62 +2398,70 @@ 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, 21)
var file_llmgw_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
var file_llmgw_v1_agent_proto_goTypes = []any{
(ShardStatus_State)(0), // 0: llmgw.v1.ShardStatus.State
(*AgentMessage)(nil), // 1: llmgw.v1.AgentMessage
(*ControlMessage)(nil), // 2: llmgw.v1.ControlMessage
(*Register)(nil), // 3: llmgw.v1.Register
(*Storage)(nil), // 4: llmgw.v1.Storage
(*RunningShard)(nil), // 5: llmgw.v1.RunningShard
(*GPU)(nil), // 6: llmgw.v1.GPU
(*FabricAttachment)(nil), // 7: llmgw.v1.FabricAttachment
(*LocalWeights)(nil), // 8: llmgw.v1.LocalWeights
(*Heartbeat)(nil), // 9: llmgw.v1.Heartbeat
(*ShardStatus)(nil), // 10: llmgw.v1.ShardStatus
(*FetchProgress)(nil), // 11: llmgw.v1.FetchProgress
(*StartShard)(nil), // 12: llmgw.v1.StartShard
(*EngineProfile)(nil), // 13: llmgw.v1.EngineProfile
(*RemoveWeights)(nil), // 14: llmgw.v1.RemoveWeights
(*StopShard)(nil), // 15: llmgw.v1.StopShard
(*DrainShard)(nil), // 16: llmgw.v1.DrainShard
(*Ack)(nil), // 17: llmgw.v1.Ack
nil, // 18: llmgw.v1.Register.LabelsEntry
nil, // 19: llmgw.v1.Register.EngineVersionsEntry
nil, // 20: llmgw.v1.Heartbeat.EngineVersionsEntry
nil, // 21: llmgw.v1.EngineProfile.EnvEntry
(*Host)(nil), // 4: llmgw.v1.Host
(*Storage)(nil), // 5: llmgw.v1.Storage
(*RunningShard)(nil), // 6: llmgw.v1.RunningShard
(*GPU)(nil), // 7: llmgw.v1.GPU
(*FabricAttachment)(nil), // 8: llmgw.v1.FabricAttachment
(*LocalWeights)(nil), // 9: llmgw.v1.LocalWeights
(*Heartbeat)(nil), // 10: llmgw.v1.Heartbeat
(*Provision)(nil), // 11: llmgw.v1.Provision
(*EngineWant)(nil), // 12: llmgw.v1.EngineWant
(*ProvisionStatus)(nil), // 13: llmgw.v1.ProvisionStatus
(*ShardStatus)(nil), // 14: llmgw.v1.ShardStatus
(*FetchProgress)(nil), // 15: llmgw.v1.FetchProgress
(*StartShard)(nil), // 16: llmgw.v1.StartShard
(*EngineProfile)(nil), // 17: llmgw.v1.EngineProfile
(*RemoveWeights)(nil), // 18: llmgw.v1.RemoveWeights
(*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
}
var file_llmgw_v1_agent_proto_depIdxs = []int32{
3, // 0: llmgw.v1.AgentMessage.register:type_name -> llmgw.v1.Register
9, // 1: llmgw.v1.AgentMessage.heartbeat:type_name -> llmgw.v1.Heartbeat
10, // 2: llmgw.v1.AgentMessage.shard:type_name -> llmgw.v1.ShardStatus
12, // 3: llmgw.v1.ControlMessage.start:type_name -> llmgw.v1.StartShard
15, // 4: llmgw.v1.ControlMessage.stop:type_name -> llmgw.v1.StopShard
16, // 5: llmgw.v1.ControlMessage.drain:type_name -> llmgw.v1.DrainShard
17, // 6: llmgw.v1.ControlMessage.ack:type_name -> llmgw.v1.Ack
14, // 7: llmgw.v1.ControlMessage.remove_weights:type_name -> llmgw.v1.RemoveWeights
18, // 8: llmgw.v1.Register.labels:type_name -> llmgw.v1.Register.LabelsEntry
6, // 9: llmgw.v1.Register.gpus:type_name -> llmgw.v1.GPU
7, // 10: llmgw.v1.Register.fabrics:type_name -> llmgw.v1.FabricAttachment
19, // 11: llmgw.v1.Register.engine_versions:type_name -> llmgw.v1.Register.EngineVersionsEntry
5, // 12: llmgw.v1.Register.running_shards:type_name -> llmgw.v1.RunningShard
8, // 13: llmgw.v1.Register.weights:type_name -> llmgw.v1.LocalWeights
4, // 14: llmgw.v1.Register.storage:type_name -> llmgw.v1.Storage
0, // 15: llmgw.v1.RunningShard.state:type_name -> llmgw.v1.ShardStatus.State
4, // 16: llmgw.v1.Heartbeat.storage:type_name -> llmgw.v1.Storage
8, // 17: llmgw.v1.Heartbeat.weights:type_name -> llmgw.v1.LocalWeights
20, // 18: llmgw.v1.Heartbeat.engine_versions:type_name -> llmgw.v1.Heartbeat.EngineVersionsEntry
0, // 19: llmgw.v1.ShardStatus.state:type_name -> llmgw.v1.ShardStatus.State
11, // 20: llmgw.v1.ShardStatus.fetch:type_name -> llmgw.v1.FetchProgress
13, // 21: llmgw.v1.StartShard.profile:type_name -> llmgw.v1.EngineProfile
21, // 22: llmgw.v1.EngineProfile.env:type_name -> llmgw.v1.EngineProfile.EnvEntry
1, // 23: llmgw.v1.AgentService.Connect:input_type -> llmgw.v1.AgentMessage
2, // 24: llmgw.v1.AgentService.Connect:output_type -> llmgw.v1.ControlMessage
24, // [24:25] is the sub-list for method output_type
23, // [23:24] is the sub-list for method input_type
23, // [23:23] is the sub-list for extension type_name
23, // [23:23] is the sub-list for extension extendee
0, // [0:23] is the sub-list for field type_name
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
}
func init() { file_llmgw_v1_agent_proto_init() }
@@ -2109,6 +2480,7 @@ func file_llmgw_v1_agent_proto_init() {
(*ControlMessage_Drain)(nil),
(*ControlMessage_Ack)(nil),
(*ControlMessage_RemoveWeights)(nil),
(*ControlMessage_Provision)(nil),
}
type x struct{}
out := protoimpl.TypeBuilder{
@@ -2116,7 +2488,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: 21,
NumMessages: 25,
NumExtensions: 0,
NumServices: 1,
},
+66
View File
@@ -34,6 +34,7 @@ message ControlMessage {
DrainShard drain = 3;
Ack ack = 4;
RemoveWeights remove_weights = 5;
Provision provision = 6;
}
}
@@ -101,6 +102,34 @@ message Register {
// Empty means the node was not asked to look, exactly as for weights, and
// constrains nothing.
repeated Storage storage = 12;
// host is what the machine is, so the control plane can have the mirror
// fetch software built for it rather than assuming every node matches the
// first.
Host host = 13;
// engines_probed says engine_versions is an answer. Without it an empty map
// meant "an agent too old to say" and constrained nothing — so a freshly
// built node with no engine installed yet was a candidate for every model.
bool engines_probed = 14;
}
// Host describes the operating system and processor of a node.
message Host {
// From /etc/os-release: "ubuntu", "24.04", "noble".
string os_id = 1;
string os_version = 2;
string os_codename = 3;
// Debian's spelling, which is what package indexes are keyed by: "amd64",
// "arm64".
string arch = 4;
// The vector extensions inference cares about, of those the CPU has. A VM
// given QEMU's default CPU model has none of them, and runs, and is five
// times slower, and nothing says why.
repeated string cpu_flags = 5;
// nvidia_driver is the driver version, empty when there is none. A card
// without a driver is not a GPU anything can use.
string nvidia_driver = 6;
}
// Storage is one filesystem a node keeps weights on.
@@ -234,6 +263,43 @@ message Heartbeat {
// somebody restarts the agent — which is a step nobody knows to take.
bool engines_changed = 8;
map<string, string> engine_versions = 9;
// provision is where installing the node's software stands, sent when it
// changed. The flag makes an empty list mean "nothing to install" rather
// than "nothing to say", as for weights.
bool provision_changed = 10;
repeated ProvisionStatus provision = 11;
}
// Provision is the software a node should have: which engines, at which
// versions. Nothing more.
//
// Everything else is the node's to work out, because only the node knows it:
// which build fits its hardware, where that build is published, what its
// operating system is missing, and whether the result actually runs. The agent
// converges on this — installs what is absent, checks what is there — and
// reports each engine through ProvisionStatus. Nothing is taken on trust from
// what happens to be on the machine: an engine is offered to placement only
// once the agent has installed it and seen it run.
message Provision {
repeated EngineWant engines = 1;
}
// EngineWant names one engine at one version.
message EngineWant {
string engine = 1; // "llamacpp"
string version = 2; // "b10991"
}
// ProvisionStatus is where one engine's installation stands on a node.
message ProvisionStatus {
string engine = 1;
string version = 2;
string variant = 3;
// "installing", "ready", "failed", or "unsupported" when no build fits.
string state = 4;
// detail says what is happening or what went wrong, in words for a person.
string detail = 5;
}
// ShardStatus reports a shard's lifecycle transition.
+2 -3
View File
@@ -108,6 +108,5 @@ echo "== 2/2 onboarding $NAME"
echo
echo "Check it with: llmgw ctl nodes"
echo
echo "A node is not finished until the engines its profiles name are installed"
echo "and the weights are on disk. Onboarding checks that a configured engine"
echo "runs, but it installs neither — see docs/node-onboarding.md."
echo "The agent installs its engines itself; the node takes no model until they"
echo "are installed and seen to run. 'llmgw ctl nodes' shows where that stands."
+31 -7
View File
@@ -33,8 +33,8 @@ usage: $0 --node ID --addr HOST --server URL [options]
address on the route to --server, found by the agent)
--mgmt-if NAME interface NCCL bootstraps over (default: the one holding
that address)
--llamacpp-bin P path to llama-server (default: PATH, then /opt/llama.cpp)
--vllm-bin P path to the vllm entry point
--llamacpp-bin P use this llama-server instead of the one the agent installs
--vllm-bin P path to the vllm entry point (not yet installed by the agent)
--sglang-python P python interpreter of the SGLang virtualenv
--nvidia-smi PATH where nvidia-smi lives, when it is not on PATH. Containerized
driver installs routinely put it somewhere else.
@@ -44,6 +44,8 @@ usage: $0 --node ID --addr HOST --server URL [options]
Releases carry static linux/amd64 and linux/arm64 builds;
downloading one is better than copying whatever happens to
be on the machine you are running this from.
--releases URL where release binaries are downloaded from (default the
project's Gitea releases)
--ssh-user NAME (default ilker)
USAGE
exit 2
@@ -54,6 +56,7 @@ NVIDIA_SMI= IB_ROOT=/sys/class/infiniband
LLAMACPP_BIN= VLLM_BIN= SGLANG_PYTHON=
SSH_WAIT=180
PKI=/etc/llmgw/pki BINARY= SSH_USER=ilker
RELEASES=https://gitea.mynodes.xyz/ilker/llmgw/releases/download
FABRIC_NAME= FABRIC_HCAS="mlx5_0,mlx5_1"
while [ $# -gt 0 ]; do
@@ -71,6 +74,7 @@ while [ $# -gt 0 ]; do
--ib-root) IB_ROOT="$2"; shift 2;;
--pki) PKI="$2"; shift 2;;
--binary) BINARY="$2"; shift 2;;
--releases) RELEASES="$2"; shift 2;;
--ssh-user) SSH_USER="$2"; shift 2;;
*) usage;;
esac
@@ -131,9 +135,28 @@ $SSH "set -e
rm -f /tmp/ca.pem /tmp/$NODE.pem /tmp/$NODE-key.pem"
# 4. The binary, the configuration and the unit.
say "installing the binary"
scp -q $SSHOPTS "$BINARY" "$SSH_USER@$ADDR:/tmp/llmgw"
$SSH 'sudo install -m 0755 /tmp/llmgw /usr/local/bin/llmgw && rm -f /tmp/llmgw'
# The node fetches the release CI built for this version, for its own
# architecture, and checks it against the release's checksums — rather than
# being handed whatever binary is on the machine running this script. A build
# that is not a release (a developer's, with commits past the tag) has no such
# file, and is copied as before.
TAG="$("$BINARY" --version 2>/dev/null | awk '{print $NF}')"
if printf '%s' "$TAG" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+$'; then
say "installing llmgw $TAG from the release"
$SSH "set -e
arch=\$(dpkg --print-architecture)
name=llmgw-$TAG-linux-\$arch
cd /tmp && rm -f \$name checksums.txt
curl -fsSL -o \$name '$RELEASES/$TAG/'\$name
curl -fsSL -o checksums.txt '$RELEASES/$TAG/checksums.txt'
grep \" \$name\\\$\" checksums.txt | sha256sum -c - >/dev/null
sudo install -m 0755 \$name /usr/local/bin/llmgw
rm -f \$name checksums.txt"
else
say "installing the binary (${TAG:-unversioned}, not a release)"
scp -q $SSHOPTS "$BINARY" "$SSH_USER@$ADDR:/tmp/llmgw"
$SSH 'sudo install -m 0755 /tmp/llmgw /usr/local/bin/llmgw && rm -f /tmp/llmgw'
fi
say "writing the configuration"
NODE_IS_HOSTNAME=no
@@ -242,8 +265,9 @@ $SSH 'set -e
echo
if [ -z "$LLAMACPP_BIN$VLLM_BIN$SGLANG_PYTHON" ]; then
say "no engine paths given; this node can only run the 'sim' engine until"
say "one is installed and the configuration points at it"
say "the agent now installs the engines the control plane asks for: the"
say "build for this machine, checked against its publisher's checksum and run"
say "before the node is offered any model. 'llmgw ctl nodes' shows progress."
fi
if [ -n "$ENGINE_PROBLEMS" ]; then
echo "$NODE is onboarded but an engine will not run:$ENGINE_PROBLEMS"
+9
View File
@@ -84,6 +84,15 @@ discovery:
# - /var/lib/llmgw-models
# - /mnt/lustre/models
# The agent installs the engines the control plane names (see `engines:` in
# server.yaml) under /opt/llmgw/engines: the build for this machine — CUDA when
# an NVIDIA driver is loaded — checked against its publisher's checksum, with
# the OS packages it needs, and run before the node takes any model. llama.cpp
# is installed this way today. Naming a binary below opts that engine out and
# uses yours instead; vLLM and SGLang still have to be named until the agent
# installs them too.
# engines_dir: /opt/llmgw/engines
engines:
vllm:
bin: /opt/venvs/vllm/bin/vllm
+7
View File
@@ -197,6 +197,13 @@ router:
session_load_factor: 1.25
# How aggressively the control plane reconciles.
# The engines every node installs, and at which version. Unset keeps the
# versions this release was checked against (llama.cpp b10991). Each agent
# picks the build for its own hardware and installs it itself; changing a
# version here reaches every node on its next connect.
# engines:
# llamacpp: b10991
control:
interval: 5s
base_backoff: 10s
+23 -2
View File
@@ -54,6 +54,7 @@ rather than repeated.
| `adopt_ready_timeout` | `time.Duration` | [Config](#config) |
| `tls` | `tlsconf.Config` | [Config](#config) |
| `model_dirs` | `[]string` | [Config](#config) |
| `engines_dir` | `string` | [Config](#config) |
| `weights_cache` | `string` | [Config](#config) |
| `discovery` | `discovery.Options` | [Config](#config) |
| `engines` | `engines.Config` | [Config](#config) |
@@ -160,6 +161,7 @@ rather than repeated.
| `agent_tls` | `tlsconf.Config` | [Server](#server) |
| `monitoring` | `Monitoring` | [Server](#server) |
| `mirror` | `Mirror` | [Server](#server) |
| `engines` | `map[string]string` | [Server](#server) |
| `router` | `Router` | [Server](#server) |
| `control` | `Control` | [Server](#server) |
| `models` | `[]catalog.Model` | [Server](#server) |
@@ -432,6 +434,13 @@ exported over the fabric, local NVMe and a copy made by hand all look
identical from here, and placement only needs to know whether the file
is present.
### `engines_dir`
`string`
EnginesDir is where the agent installs engines, one directory per
build. Defaults to /opt/llmgw/engines.
### `weights_cache`
`string`
@@ -589,8 +598,8 @@ know that this page cannot guess.
Bin is the llama-server binary. It is the server, not the CLI: the
gateway talks OpenAI over HTTP and needs the one that serves it.
Empty looks on PATH, then where release archives are unpacked
(/opt/llama.cpp).
Empty means the agent installs llama.cpp itself, at the version the
control plane asks for. Setting it opts out of that.
### `parallel`
@@ -1193,6 +1202,18 @@ Empty means there is none, and /admin/mirror stays unmounted.
Takes a [Mirror](#mirror) block.
### `engines`
`map[string]string`
Engines are the engines every node installs, and at which version. The
agents install them themselves — the right build for their hardware,
checked against its publisher's checksum and seen to run — so a node is
never trusted to already have what somebody once put there.
Unset keeps DefaultEngines. A version replaces the default; an empty
one leaves that engine to be installed by hand.
### `router`
`Router`
+189 -31
View File
@@ -28,6 +28,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1/llmgwv1connect"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/provision"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/weights"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
@@ -94,6 +95,10 @@ type Config struct {
// is present.
ModelDirs []string `koanf:"model_dirs"`
// EnginesDir is where the agent installs engines, one directory per
// build. Defaults to /opt/llmgw/engines.
EnginesDir string `koanf:"engines_dir"`
// WeightsCache is where a remote source is downloaded to. Defaults to the
// first entry of ModelDirs, because a node that says where its models live
// has already answered this question, and otherwise to
@@ -304,16 +309,28 @@ type Agent struct {
reg *engines.Registry
http connectHTTPClient
// engineOnce probes the installed engines once before registering.
// engineOnce probes the configured engines once before registering.
//
// Once, because probing SGLang imports it in a Python interpreter — seconds
// on a cold cache — and a reconnect is not a reason to pay that again.
// Later changes are found by the slower re-probe the heartbeat starts, not
// by registering again: an engine installed while the agent runs used to
// stay invisible until somebody thought to restart the agent.
// Later changes are found by the slower re-probe the heartbeat starts.
engineOnce sync.Once
// engineVers is guarded by mu once the first probe has run.
engineVers map[string]string
// configuredVers are the engines the configuration names and that answer.
// Guarded by mu once the first probe has run.
configuredVers map[string]string
// installedVers are the engines this agent installed and saw run.
// Guarded by mu.
installedVers map[string]string
// prov installs the engines the control plane asks for. provWants is what
// 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.
prov *provision.Installer
provWants []provision.Want
provStatus map[string]provision.Status
provSent string
provKick chan struct{}
// enginesSent is the fingerprint of the engines the control plane last
// heard about, from registration or a heartbeat. Guarded by mu.
enginesSent string
@@ -329,8 +346,9 @@ type Agent struct {
fetcher *weights.Fetcher
// lastWeights is the fingerprint of the file list most recently sent, so a
// heartbeat carries the list only when it changed. Touched only by the
// heartbeat goroutine, which is the only thing that reads or writes it.
// heartbeat carries the list only when it changed. Guarded by mu: during a
// failover the old stream's heartbeat loop can still be running when the
// new one's starts.
lastWeights string
// fetchWatchers are the shards waiting on each source, so progress on one
@@ -363,12 +381,16 @@ func New(cfg Config, httpClient connectHTTPClient, log *slog.Logger) (*Agent, er
if err != nil {
return nil, err
}
return &Agent{
a := &Agent{
cfg: cfg, log: log, sup: sup,
reg: engines.NewRegistry(cfg.Engines),
http: httpClient,
shardReplica: map[string]string{},
}, nil
provStatus: map[string]provision.Status{},
provKick: make(chan struct{}, 1),
}
a.prov = &provision.Installer{Root: cfg.EnginesDir, Log: log, OnStatus: a.provisionStatus}
return a, nil
}
// Supervisor exposes the process supervisor, for tests and `ctl logs`.
@@ -387,6 +409,7 @@ func (a *Agent) Run(ctx context.Context) error {
a.log.Error("hardware discovery failed", "err", err)
}
a.recoverOrphans()
go a.provisionLoop(ctx)
attempt := 0
next := 0
@@ -645,9 +668,13 @@ func (a *Agent) scan(ctx context.Context) error {
opts := a.cfg.Discovery
opts.ModelDirs = a.cfg.ModelDirs
inv, err := discovery.Discover(ctx, opts, a.log)
host := discovery.DiscoverHost()
a.mu.Lock()
a.inv = inv
a.prov.Host = host
a.mu.Unlock()
a.log.Info("this node", "os", host.OSID+" "+host.OSVersion, "arch", host.Arch,
"cpu_flags", host.CPUFlags, "nvidia_driver", host.NVIDIADriver)
return err
}
@@ -685,7 +712,18 @@ func (a *Agent) session(ctx context.Context, server string) error {
// engine whether or not its binary exists, so without this a node with no
// vLLM accepted a vLLM shard and failed it minutes later.
reg.EngineVersions = a.engines(parent)
reg.EnginesProbed = true
reg.RunningShards = a.runningShards()
a.mu.Lock()
h := a.prov.Host
// A new stream is a control plane that knows nothing of where installs
// stand, so the next heartbeat tells it again.
a.provSent = ""
a.mu.Unlock()
reg.Host = &llmgwv1.Host{
OsId: h.OSID, OsVersion: h.OSVersion, OsCodename: h.OSCodename,
Arch: h.Arch, CpuFlags: h.CPUFlags, NvidiaDriver: h.NVIDIADriver,
}
if err := stream.Send(&llmgwv1.AgentMessage{
Msg: &llmgwv1.AgentMessage_Register{Register: reg},
@@ -780,10 +818,22 @@ func (a *Agent) heartbeat() *llmgwv1.AgentMessage {
a.reprobeEngines()
a.mu.Lock()
if fp := enginesFingerprint(a.engineVers); a.engineVers != nil && fp != a.enginesSent {
a.enginesSent = fp
if vers := a.engineReportLocked(); a.configuredVers != nil &&
enginesFingerprint(vers) != a.enginesSent {
a.enginesSent = enginesFingerprint(vers)
hb.EnginesChanged = true
hb.EngineVersions = maps.Clone(a.engineVers)
hb.EngineVersions = vers
}
if fp := provisionFingerprint(a.provStatus); fp != a.provSent {
a.provSent = fp
hb.ProvisionChanged = true
for _, e := range slices.Sorted(maps.Keys(a.provStatus)) {
st := a.provStatus[e]
hb.Provision = append(hb.Provision, &llmgwv1.ProvisionStatus{
Engine: st.Engine, Version: st.Version, Variant: st.Variant,
State: st.State, Detail: st.Detail,
})
}
}
a.mu.Unlock()
@@ -807,8 +857,12 @@ func (a *Agent) heartbeat() *llmgwv1.AgentMessage {
// rather than "nothing to report" — without it, the first heartbeat after
// a model was deleted would be indistinguishable from silence.
weights := discovery.ScanWeights(a.cfg.ModelDirs)
if fp := weightsFingerprint(weights); fp != a.lastWeights {
a.lastWeights = fp
fp := weightsFingerprint(weights)
a.mu.Lock()
changed := fp != a.lastWeights
a.lastWeights = fp
a.mu.Unlock()
if changed {
hb.WeightsChanged = true
for _, w := range weights {
hb.Weights = append(hb.Weights, &llmgwv1.LocalWeights{
@@ -899,6 +953,8 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
}
case *llmgwv1.ControlMessage_RemoveWeights:
a.removeWeights(m.RemoveWeights.GetPaths())
case *llmgwv1.ControlMessage_Provision:
a.setProvision(m.Provision)
case *llmgwv1.ControlMessage_Ack:
if !m.Ack.GetOk() {
a.log.Warn("control plane rejected a message",
@@ -1152,27 +1208,129 @@ func Client(cfg tlsconf.Config, timeout time.Duration) (*http.Client, error) {
return &http.Client{Transport: &http2.Transport{TLSClientConfig: tc}}, nil
}
// engines is what this node can run, probed once.
//
// Failures are silence rather than an error: an engine that is not installed is
// the ordinary case on most nodes, and an agent that refused to register over
// one would take a working node out of the fleet for a missing engine it was
// never going to be asked to run.
// engines is what this node can run: what the agent installed and saw run,
// and what the configuration names and answers. Nothing merely found on the
// machine — a llama-server on PATH is somebody's, not necessarily one that
// works, and placing on it is how a replica fails on a node nobody watches.
func (a *Agent) engines(ctx context.Context) map[string]string {
a.engineOnce.Do(func() {
vers := a.reg.Probe(ctx)
vers := a.reg.ProbeConfigured(ctx)
a.mu.Lock()
a.engineVers, a.lastEngineProbe = vers, time.Now()
a.configuredVers, a.lastEngineProbe = vers, time.Now()
a.mu.Unlock()
a.log.Info("engines available on this node",
"engines", engines.ProbedNames(vers))
if len(vers) > 0 {
a.log.Info("engines named in the configuration",
"engines", engines.ProbedNames(vers))
}
})
a.mu.Lock()
defer a.mu.Unlock()
vers := a.engineReportLocked()
// Whatever registration sends is what the control plane now knows, so a
// heartbeat repeats it only once it differs.
a.enginesSent = enginesFingerprint(a.engineVers)
return maps.Clone(a.engineVers)
a.enginesSent = enginesFingerprint(vers)
return vers
}
// engineReportLocked merges the two sources. Installed wins: it is the one
// that was seen to run. Called with mu held.
func (a *Agent) engineReportLocked() map[string]string {
out := maps.Clone(a.configuredVers)
if out == nil {
out = map[string]string{}
}
maps.Copy(out, a.installedVers)
return out
}
// provisionEvery is how often installed engines are checked again. A library
// removed by an unrelated upgrade is found here rather than by the next
// replica to start.
const provisionEvery = 10 * time.Minute
// setProvision takes the control plane's list of wanted engines.
func (a *Agent) setProvision(p *llmgwv1.Provision) {
wants := []provision.Want{}
for _, e := range p.GetEngines() {
wants = append(wants, provision.Want{Engine: e.GetEngine(), Version: e.GetVersion()})
}
a.mu.Lock()
a.provWants = wants
a.mu.Unlock()
select {
case a.provKick <- struct{}{}:
default:
}
}
// provisionLoop installs and re-checks the wanted engines, one pass at a time.
func (a *Agent) provisionLoop(ctx context.Context) {
t := time.NewTicker(provisionEvery)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-a.provKick:
case <-t.C:
}
a.mu.Lock()
wants := slices.Clone(a.provWants)
managed := a.provWants != nil
a.mu.Unlock()
if !managed {
continue
}
statuses := a.prov.Converge(ctx, wants)
installed := map[string]string{}
keep := map[string]bool{}
for _, st := range statuses {
keep[st.Engine] = true
if st.State == provision.Ready {
a.reg.UseInstalled(st.Engine, st.Binary)
installed[st.Engine] = st.Version
}
}
a.mu.Lock()
if enginesFingerprint(installed) != enginesFingerprint(a.installedVers) {
a.log.Info("engines installed and running", "engines", engines.ProbedNames(installed))
}
a.installedVers = installed
// An engine no longer asked for is no longer reported on.
for e := range a.provStatus {
if !keep[e] {
delete(a.provStatus, e)
}
}
a.mu.Unlock()
}
}
// provisionStatus records a change in one engine's installation.
func (a *Agent) provisionStatus(st provision.Status) {
a.mu.Lock()
prev, had := a.provStatus[st.Engine]
a.provStatus[st.Engine] = st
a.mu.Unlock()
if !had || prev.State != st.State || prev.Detail != st.Detail {
lvl := slog.LevelInfo
if st.State == provision.Failed || st.State == provision.Unsupported {
lvl = slog.LevelWarn
}
a.log.Log(context.Background(), lvl, "engine installation",
"engine", st.Engine, "version", st.Version, "variant", st.Variant,
"state", st.State, "detail", st.Detail)
}
}
func provisionFingerprint(m map[string]provision.Status) string {
var b strings.Builder
for _, k := range slices.Sorted(maps.Keys(m)) {
st := m[k]
fmt.Fprintf(&b, "%s|%s|%s|%s|%s\n", st.Engine, st.Version, st.Variant, st.State, st.Detail)
}
return b.String()
}
// engineReprobe is how often the installed engines are looked for again.
@@ -1198,14 +1356,14 @@ func (a *Agent) reprobeEngines() {
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
vers := a.reg.Probe(ctx)
vers := a.reg.ProbeConfigured(ctx)
a.mu.Lock()
changed := enginesFingerprint(vers) != enginesFingerprint(a.engineVers)
a.engineVers, a.lastEngineProbe, a.probingEngines = vers, time.Now(), false
changed := enginesFingerprint(vers) != enginesFingerprint(a.configuredVers)
a.configuredVers, a.lastEngineProbe, a.probingEngines = vers, time.Now(), false
a.mu.Unlock()
if changed {
a.log.Info("engines available on this node changed",
a.log.Info("engines named in the configuration changed",
"engines", engines.ProbedNames(vers))
}
}()
+154
View File
@@ -0,0 +1,154 @@
package discovery
import (
"bufio"
"os"
"runtime"
"strings"
)
// Host is what a node is, as far as installing software on it is concerned.
type Host struct {
// OSID, OSVersion and OSCodename come from /etc/os-release: "ubuntu",
// "24.04", "noble".
OSID string `json:"os_id,omitempty"`
OSVersion string `json:"os_version,omitempty"`
OSCodename string `json:"os_codename,omitempty"`
// OSLike is ID_LIKE: the families this system says it follows, "debian"
// for Ubuntu, "rhel centos fedora" for Rocky. It is what decides the
// package manager, which the name alone does not for every derivative.
OSLike []string `json:"os_like,omitempty"`
// Arch is Debian's spelling, "amd64" or "arm64", because that is what
// package names and release archives are keyed by.
Arch string `json:"arch,omitempty"`
// CPUFlags are the vector extensions inference cares about, of those this
// processor has.
CPUFlags []string `json:"cpu_flags,omitempty"`
// NVIDIADriver is the loaded driver's version, empty when there is none.
NVIDIADriver string `json:"nvidia_driver,omitempty"`
}
// Family is the packaging family: "deb", "rpm", or "" when neither is known.
func (h Host) Family() string {
for _, id := range append([]string{h.OSID}, h.OSLike...) {
switch id {
case "debian", "ubuntu":
return "deb"
case "rhel", "fedora", "centos", "rocky", "almalinux", "suse", "opensuse":
return "rpm"
}
}
return ""
}
// HasCPUFlag reports whether the processor has an extension.
func (h Host) HasCPUFlag(f string) bool {
for _, have := range h.CPUFlags {
if have == f {
return true
}
}
return false
}
// interestingFlags are the extensions that decide which build of an engine
// runs well. A VM on QEMU's default CPU model has none of them — and still
// runs, five times slower, with nothing saying why.
var interestingFlags = []string{"avx", "avx2", "fma", "f16c", "avx512f", "avx512_bf16", "amx_tile", "asimd", "sve"}
// DiscoverHost reads what the machine is.
func DiscoverHost() Host {
h := hostFromOSRelease("/etc/os-release")
h.Arch = debianArch(runtime.GOARCH)
h.CPUFlags = cpuFlagsFrom("/proc/cpuinfo")
h.NVIDIADriver = nvidiaDriverFrom("/proc/driver/nvidia/version")
return h
}
func hostFromOSRelease(path string) Host {
var h Host
f, err := os.Open(path) //nolint:gosec // a fixed system path
if err != nil {
return h
}
defer f.Close()
sc := bufio.NewScanner(f)
for sc.Scan() {
k, v, ok := strings.Cut(sc.Text(), "=")
if !ok {
continue
}
v = strings.Trim(v, `"'`)
switch k {
case "ID":
h.OSID = v
case "VERSION_ID":
h.OSVersion = v
case "VERSION_CODENAME":
h.OSCodename = v
case "ID_LIKE":
h.OSLike = strings.Fields(v)
}
}
return h
}
func debianArch(goarch string) string {
switch goarch {
case "amd64", "arm64":
return goarch
case "386":
return "i386"
case "arm":
return "armhf"
}
return goarch
}
func cpuFlagsFrom(path string) []string {
b, err := os.ReadFile(path) //nolint:gosec // a fixed system path
if err != nil {
return nil
}
have := map[string]bool{}
for _, line := range strings.Split(string(b), "\n") {
k, v, ok := strings.Cut(line, ":")
if !ok {
continue
}
// "flags" on x86, "Features" on ARM. The first processor is enough:
// a machine does not mix instruction sets across its cores.
if k = strings.TrimSpace(k); k != "flags" && k != "Features" {
continue
}
for _, f := range strings.Fields(v) {
have[f] = true
}
break
}
var out []string
for _, f := range interestingFlags {
if have[f] {
out = append(out, f)
}
}
return out
}
// nvidiaDriverFrom reads the loaded kernel module's version, which exists
// exactly when a driver is loaded — unlike nvidia-smi, which can be installed
// on a machine whose module failed to load.
func nvidiaDriverFrom(path string) string {
b, err := os.ReadFile(path) //nolint:gosec // a fixed system path
if err != nil {
return ""
}
// "NVRM version: NVIDIA UNIX x86_64 Kernel Module 570.86.15 Thu ..."
fields := strings.Fields(string(b))
for i, f := range fields {
if f == "Module" && i+1 < len(fields) {
return fields[i+1]
}
}
return ""
}
+10
View File
@@ -151,6 +151,16 @@ func NewRegistry(cfg Config) *Registry {
return r
}
// UseInstalled points an engine at a build the agent installed, and reports
// whether the engine takes one.
func (r *Registry) UseInstalled(name, path string) bool {
u, ok := r.engines[name].(interface{ UseInstalled(string) })
if ok {
u.UseInstalled(path)
}
return ok
}
// Add registers an engine, replacing any engine of the same name.
func (r *Registry) Add(e Engine) { r.engines[e.Name()] = e }
+22 -27
View File
@@ -3,19 +3,19 @@ package engines
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"sort"
"strconv"
"strings"
"sync/atomic"
)
// LlamaCppConfig locates the llama.cpp installation.
type LlamaCppConfig struct {
// Bin is the llama-server binary. It is the server, not the CLI: the
// gateway talks OpenAI over HTTP and needs the one that serves it.
// Empty looks on PATH, then where release archives are unpacked
// (/opt/llama.cpp).
// Empty means the agent installs llama.cpp itself, at the version the
// control plane asks for. Setting it opts out of that.
Bin string `koanf:"bin"`
// Parallel is how many requests a server handles at once, when a profile
// does not say. llama.cpp defaults to one, which would make every second
@@ -29,7 +29,10 @@ type LlamaCppConfig struct {
// small model on CPU at a speed that is fine for testing the whole path, and it
// serves the same OpenAI API as the others, so nothing downstream can tell the
// difference.
type LlamaCpp struct{ cfg LlamaCppConfig }
type LlamaCpp struct {
cfg LlamaCppConfig
installed atomic.Pointer[string]
}
// Name identifies the engine in the catalog.
func (*LlamaCpp) Name() string { return "llamacpp" }
@@ -170,33 +173,25 @@ func ggufIn(path string) (string, error) {
return filepath.Join(path, names[0]), nil
}
// llamaServerPlaces are where llama.cpp's release archives are unpacked when
// they are not on PATH. The archive carries its shared libraries beside the
// binary, so it is usually left in one directory rather than installed.
var llamaServerPlaces = []string{
"/opt/llama.cpp/llama-server",
"/opt/llama.cpp/bin/llama-server",
"/opt/llama.cpp/build/bin/llama-server",
"/usr/local/llama.cpp/llama-server",
}
// UseInstalled points the engine at a build the agent installed and checked.
// It takes precedence over everything else, because it is the only one that
// was seen to run.
func (l *LlamaCpp) UseInstalled(path string) { l.installed.Store(&path) }
// bin is the llama-server to run: the configured one, else the one on PATH,
// else the first of the usual unpack locations that exists.
//
// Found rather than required in the configuration, because every node here
// had the same path written into its file by hand, and a node set up without
// it reported no engine at all.
// Configured reports whether the agent's configuration names a binary. A
// named binary is the operator's deliberate choice; one merely found on PATH
// is not, and is not reported as installed.
func (l *LlamaCpp) Configured() bool { return l.cfg.Bin != "" }
// bin is the llama-server to run: the installed build, else the configured
// one, else whatever PATH has — which is only ever reached by a node not
// provisioned by the agent and not told where its engine is.
func (l *LlamaCpp) bin() string {
if p := l.installed.Load(); p != nil {
return *p
}
if l.cfg.Bin != "" {
return l.cfg.Bin
}
if p, err := exec.LookPath("llama-server"); err == nil {
return p
}
for _, p := range llamaServerPlaces {
if info, err := os.Stat(p); err == nil && !info.IsDir() && info.Mode()&0o111 != 0 {
return p
}
}
return "llama-server"
}
+38
View File
@@ -157,11 +157,49 @@ func firstVersion(s string) string {
return Unknown
}
// Configured is an engine whose location the agent's configuration names.
type Configured interface {
Configured() bool
}
// ProbeConfigured probes only the engines the configuration names.
//
// Probe looks everywhere — PATH, the default interpreter — and reports what
// it finds, which is exactly the trust in whatever happens to be on a machine
// that installing engines replaces. A node the agent provisions reports what
// it installed, plus what the operator named on purpose; nothing else.
func (r *Registry) ProbeConfigured(ctx context.Context) map[string]string {
out := map[string]string{}
for name, e := range r.engines {
c, ok := e.(Configured)
if !ok || !c.Configured() {
continue
}
p, ok := e.(Prober)
if !ok {
continue
}
if v, err := p.Probe(ctx); err == nil {
if v == "" {
v = Unknown
}
out[name] = v
}
}
return out
}
// Probe reports whether the llama.cpp server is installed.
func (l *LlamaCpp) Probe(context.Context) (string, error) {
return installedBinary(l.bin())
}
// Configured reports whether the configuration names vLLM.
func (v *VLLM) Configured() bool { return v.cfg.Bin != "" }
// Configured reports whether the configuration names SGLang's interpreter.
func (g *SGLang) Configured() bool { return g.cfg.Python != "" }
// Probe reports whether vLLM is installed.
func (v *VLLM) Probe(context.Context) (string, error) {
bin := v.cfg.Bin
+614
View File
@@ -0,0 +1,614 @@
package provision
import (
"archive/tar"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"log/slog"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
)
// Want is one engine at one version.
type Want struct {
Engine string
Version string
}
// The states an engine's installation can be in.
const (
Installing = "installing"
Ready = "ready"
Failed = "failed"
Unsupported = "unsupported"
)
// Status is where one engine's installation stands.
type Status struct {
Engine string
Version string
Variant string
State string
// Detail is for a person: what is happening, or what went wrong.
Detail string
// Binary is the executable, once ready.
Binary string
}
// Installer installs engines under one directory.
type Installer struct {
// Root is where engines are installed, one directory per build:
// <root>/<engine>/<version>-<variant>.
Root string
Host discovery.Host
Log *slog.Logger
HTTP *http.Client
// OnStatus hears every change, so a download that takes minutes is seen
// to be under way rather than silent.
OnStatus func(Status)
// Run executes a command; a seam for tests. Nil runs it for real.
Run func(ctx context.Context, env []string, name string, args ...string) ([]byte, error)
// APIBase and DownloadBase default to GitHub; tests point them at a fake.
APIBase string
DownloadBase string
}
// DefaultRoot is where engines are installed when nothing says otherwise.
const DefaultRoot = "/opt/llmgw/engines"
// marker is written into a build's directory once it has been installed and
// seen to run. Its absence means the directory is not to be trusted.
const marker = ".llmgw-installed"
type markerFile struct {
Engine string `json:"engine"`
Version string `json:"version"`
Variant string `json:"variant"`
Binary string `json:"binary"`
SHA256 map[string]string `json:"sha256"`
Installed time.Time `json:"installed"`
}
// Converge brings every wanted engine to ready, or says why it cannot.
//
// An engine already installed is checked again rather than believed: its
// directory can be damaged, or a library it needs removed by somebody's
// upgrade, and that is found here rather than by the next replica to fail.
func (in *Installer) Converge(ctx context.Context, wants []Want) []Status {
out := make([]Status, 0, len(wants))
for _, w := range wants {
st := in.one(ctx, w)
in.report(st)
out = append(out, st)
}
return out
}
func (in *Installer) one(ctx context.Context, w Want) Status {
st := Status{Engine: w.Engine, Version: w.Version}
recipe, ok := Recipes[w.Engine]
if !ok {
st.State = Unsupported
st.Detail = "this agent does not know how to install " + w.Engine +
"; install it by hand and name it in the agent's configuration"
return st
}
build, err := recipe.Choose(w.Version, in.Host)
if err != nil {
st.State, st.Detail = Unsupported, err.Error()
return st
}
st.Variant = build.Variant
dir := filepath.Join(in.root(), w.Engine, w.Version+"-"+build.Variant)
// Installed before: check it still runs, and trust nothing else about it.
if _, err := os.Stat(filepath.Join(dir, marker)); err == nil {
// The packages are checked too. Removing one is exactly the kind of
// change an unrelated upgrade makes, and ldd below would catch it only
// as a missing library with no hint of which package held it.
if err := in.packages(ctx, build.Packages, &st); err != nil {
st.State, st.Detail = Failed, err.Error()
return st
}
bin, version, err := in.verify(ctx, dir, build)
if err == nil {
return in.ready(st, build, bin, version)
}
in.log().Warn("an installed engine no longer runs; installing it again",
"engine", w.Engine, "version", w.Version, "err", err)
}
if err := in.install(ctx, recipe, w, build, dir, &st); err != nil {
st.State, st.Detail, st.Binary = Failed, err.Error(), ""
return st
}
bin, version, err := in.verify(ctx, dir, build)
if err != nil {
st.State, st.Detail = Failed, err.Error()
return st
}
return in.ready(st, build, bin, version)
}
func (in *Installer) ready(st Status, b Build, bin, version string) Status {
st.State, st.Binary = Ready, bin
st.Detail = strings.TrimSpace(filepath.Base(bin) + " " + version)
var missing []string
for _, f := range b.CPUFlags {
if !in.Host.HasCPUFlag(f) {
missing = append(missing, f)
}
}
if len(missing) > 0 {
// Said on every report, not once in a log: it is the difference between
// five tokens a second and forty, and a VM given QEMU's default CPU
// model looks entirely healthy otherwise.
st.Detail += "; this CPU lacks " + strings.Join(missing, ", ") +
", so it will run several times slower (a VM needs CPU type \"host\")"
}
return st
}
// install fetches, checks and unpacks a build into dir.
//
// Into a scratch directory renamed into place at the end, so a failure halfway
// leaves nothing that looks installed.
func (in *Installer) install(ctx context.Context, r Recipe, w Want, b Build, dir string, st *Status) error {
if err := in.packages(ctx, b.Packages, st); err != nil {
return err
}
in.progress(st, "asking GitHub for the checksums of %s %s", w.Engine, w.Version)
sums, err := in.checksums(ctx, r.Repo, w.Version)
if err != nil {
return err
}
tmp := dir + ".partial"
_ = os.RemoveAll(tmp)
if err := os.MkdirAll(tmp, 0o755); err != nil {
return err
}
defer os.RemoveAll(tmp) //nolint:errcheck // gone either way once renamed
got := map[string]string{}
for _, asset := range b.Assets {
want, ok := sums[asset]
if !ok {
// Refused rather than installed unchecked: this is a binary that
// will run as root on every node.
return fmt.Errorf("%s %s publishes no checksum for %s, so it cannot be verified",
w.Engine, w.Version, asset)
}
archive := filepath.Join(tmp, asset+".download")
if err := in.download(ctx, r.Repo, w.Version, asset, archive, want, st); err != nil {
return err
}
in.progress(st, "unpacking %s", asset)
if err := untarGz(archive, tmp); err != nil {
return fmt.Errorf("unpacking %s: %w", asset, err)
}
_ = os.Remove(archive)
got[asset] = want
}
in.progress(st, "checking that %s runs", b.Binary)
bin, _, err := in.verify(ctx, tmp, b)
if err != nil {
return err
}
rel, _ := filepath.Rel(tmp, bin)
m, _ := json.MarshalIndent(markerFile{
Engine: w.Engine, Version: w.Version, Variant: b.Variant,
Binary: rel, SHA256: got, Installed: time.Now().UTC(),
}, "", " ")
if err := os.WriteFile(filepath.Join(tmp, marker), m, 0o644); err != nil { //nolint:gosec // not secret
return err
}
_ = os.RemoveAll(dir)
if err := os.Rename(tmp, dir); err != nil {
return err
}
in.log().Info("installed an engine", "engine", w.Engine, "version", w.Version,
"variant", b.Variant, "dir", dir)
return nil
}
// verify finds the binary and runs it.
//
// Two checks, because they fail differently. ldd names a missing library
// outright, where running the binary would only say it could not start. And
// running it is the only proof that the rest — the right instruction set, the
// right driver — is in place.
func (in *Installer) verify(ctx context.Context, dir string, b Build) (bin, version string, err error) {
bin, err = findFile(dir, b.Binary)
if err != nil {
return "", "", err
}
if out, err := in.run(ctx, nil, "ldd", bin); err == nil {
if missing := lddMissing(string(out)); len(missing) > 0 {
return "", "", fmt.Errorf("%s cannot run on this system: %s", b.Binary, strings.Join(missing, "; "))
}
}
tctx, cancel := context.WithTimeout(ctx, 60*time.Second)
defer cancel()
out, err := in.run(tctx, nil, bin, "--version")
if err != nil {
return "", "", fmt.Errorf("%s --version failed: %v: %s", b.Binary, err, lastLine(out))
}
return bin, firstVersionLine(out), nil
}
// packages installs the operating system packages a build needs.
//
// Only those missing, and from the node's own package sources: a package the
// system already has is not touched, so an engine never upgrades something
// the rest of the machine depends on.
func (in *Installer) packages(ctx context.Context, byFamily map[string][]string, st *Status) error {
family := in.Host.Family()
pkgs := byFamily[family]
if len(pkgs) == 0 {
// A system of no family this knows is not refused: its packages may
// well be there, and if they are not, ldd names the missing library.
return nil
}
pm := packageManagers[family]
var missing []string
for _, p := range pkgs {
if _, err := in.run(ctx, nil, pm.query[0], append(pm.query[1:], p)...); err != nil {
missing = append(missing, p)
}
}
if len(missing) == 0 {
return nil
}
tool, cmds := pm.install(in, missing)
if tool == "" {
return fmt.Errorf("%s is missing and no package manager was found to install it",
strings.Join(missing, ", "))
}
in.progress(st, "installing %s with %s", strings.Join(missing, ", "), tool)
for _, c := range cmds {
if out, err := in.run(ctx, pm.env, c[0], c[1:]...); err != nil {
return fmt.Errorf("%s: %v: %s", strings.Join(c, " "), err, lastLine(out))
}
}
return nil
}
// packageManager is how one packaging family asks and installs.
type packageManager struct {
// query exits zero when the package named after it is installed.
query []string
env []string
// install picks the tool present on this machine and the commands to run.
install func(in *Installer, pkgs []string) (tool string, cmds [][]string)
}
var packageManagers = map[string]packageManager{
"deb": {
// dpkg -s exits zero for a package merely known about, so the status
// is asked for and checked by the command itself: ${db:Status-Abbrev}
// is "ii " only when installed.
query: []string{"sh", "-c", `test "$(dpkg-query -W -f='${db:Status-Abbrev}' "$0" 2>/dev/null)" = "ii "`},
env: []string{"DEBIAN_FRONTEND=noninteractive"},
install: func(_ *Installer, pkgs []string) (string, [][]string) {
// A lock timeout rather than failing on the lock: a machine a
// minute old is running cloud-init's and unattended-upgrades' own
// apt, and the first attempt on every new node would fail.
lock := "-o=DPkg::Lock::Timeout=600"
return "apt-get", [][]string{
{"apt-get", lock, "update"},
append([]string{"apt-get", lock, "install", "-y", "--no-install-recommends"}, pkgs...),
}
},
},
"rpm": {
query: []string{"rpm", "-q"},
install: func(in *Installer, pkgs []string) (string, [][]string) {
for _, tool := range []string{"dnf", "yum"} {
if in.has(tool) {
return tool, [][]string{append([]string{tool, "install", "-y"}, pkgs...)}
}
}
if in.has("zypper") {
return "zypper", [][]string{append([]string{"zypper", "--non-interactive", "install"}, pkgs...)}
}
return "", nil
},
},
}
// has reports whether a command is on PATH. Asked through the same seam as
// every other command, so a test can say which tools a machine has.
func (in *Installer) has(tool string) bool {
_, err := in.run(context.Background(), nil, "sh", "-c", `command -v "$0"`, tool)
return err == nil
}
// checksums reads the sha256 GitHub publishes for each asset of a release.
func (in *Installer) checksums(ctx context.Context, repo, tag string) (map[string]string, error) {
url := strings.TrimSuffix(in.apiBase(), "/") + "/repos/" + repo + "/releases/tags/" + tag
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept", "application/vnd.github+json")
resp, err := in.client().Do(req)
if err != nil {
return nil, fmt.Errorf("reaching GitHub for %s %s: %w", repo, tag, err)
}
defer resp.Body.Close()
switch {
case resp.StatusCode == http.StatusNotFound:
return nil, fmt.Errorf("%s has no release %s", repo, tag)
case resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusTooManyRequests:
// Unauthenticated callers get sixty requests an hour per address, and
// a whole fleet behind one NAT shares them.
return nil, fmt.Errorf("GitHub is rate-limiting this address (%s); it retries later", resp.Status)
case resp.StatusCode != http.StatusOK:
return nil, fmt.Errorf("GitHub answered %s for %s %s", resp.Status, repo, tag)
}
var rel struct {
Assets []struct {
Name string `json:"name"`
Digest string `json:"digest"`
} `json:"assets"`
}
if err := json.NewDecoder(resp.Body).Decode(&rel); err != nil {
return nil, fmt.Errorf("reading %s %s from GitHub: %w", repo, tag, err)
}
out := map[string]string{}
for _, a := range rel.Assets {
if sum, ok := strings.CutPrefix(a.Digest, "sha256:"); ok && len(sum) == 64 {
out[a.Name] = sum
}
}
return out, nil
}
func (in *Installer) download(ctx context.Context, repo, tag, asset, dst, want string, st *Status) error {
url := strings.TrimSuffix(in.downloadBase(), "/") + "/" + repo + "/releases/download/" + tag + "/" + asset
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return err
}
resp, err := in.client().Do(req)
if err != nil {
return fmt.Errorf("downloading %s: %w", asset, err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("downloading %s: %s", asset, resp.Status)
}
if resp.ContentLength > 0 {
in.progress(st, "downloading %s (%d MB)", asset, resp.ContentLength>>20)
} else {
in.progress(st, "downloading %s", asset)
}
f, err := os.Create(dst) //nolint:gosec // inside the install root
if err != nil {
return err
}
h := sha256.New()
_, err = io.Copy(io.MultiWriter(f, h), resp.Body)
if cerr := f.Close(); err == nil {
err = cerr
}
if err != nil {
return fmt.Errorf("downloading %s: %w", asset, err)
}
if got := hex.EncodeToString(h.Sum(nil)); got != want {
return fmt.Errorf("%s does not match the checksum its publisher gave (got %s, expected %s)",
asset, got[:12], want[:12])
}
return nil
}
// untarGz unpacks an archive under dir, refusing anything that would land
// outside it. A release archive is somebody else's file, and a path of
// "../../etc/cron.d/x" in it would otherwise be written as root.
func untarGz(archive, dir string) error {
f, err := os.Open(archive) //nolint:gosec // inside the install root
if err != nil {
return err
}
defer f.Close()
gz, err := gzip.NewReader(f)
if err != nil {
return err
}
tr := tar.NewReader(gz)
for {
h, err := tr.Next()
if errors.Is(err, io.EOF) {
return nil
}
if err != nil {
return err
}
target, err := within(dir, h.Name)
if err != nil {
return err
}
switch h.Typeflag {
case tar.TypeDir:
if err := os.MkdirAll(target, 0o755); err != nil {
return err
}
case tar.TypeReg:
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return err
}
out, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, fs.FileMode(h.Mode)&0o755) //nolint:gosec // checked above
if err != nil {
return err
}
if _, err := io.Copy(out, tr); err != nil { //nolint:gosec // a release archive, size bounded by its checksum
out.Close()
return err
}
if err := out.Close(); err != nil {
return err
}
case tar.TypeSymlink:
// Libraries are versioned through symlinks, and those are
// relative. One pointing anywhere else is refused.
if filepath.IsAbs(h.Linkname) {
return fmt.Errorf("%s links outside the archive", h.Name)
}
if _, err := within(dir, filepath.Join(filepath.Dir(h.Name), h.Linkname)); err != nil {
return fmt.Errorf("%s links outside the archive", h.Name)
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return err
}
_ = os.Remove(target)
if err := os.Symlink(h.Linkname, target); err != nil {
return err
}
default:
// Devices, FIFOs and hard links have no business in a release.
}
}
}
func within(dir, name string) (string, error) {
target := filepath.Join(dir, name) //nolint:gosec // checked on the next line
rel, err := filepath.Rel(dir, target)
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return "", fmt.Errorf("%s would unpack outside the install directory", name)
}
return target, nil
}
// lddMissing reads what ldd could not satisfy, in two shapes:
//
// libgomp.so.1 => not found
// /lib64/libstdc++.so.6: version `GLIBCXX_3.4.30' not found (required by ...)
//
// The second is a system too old for the build — RHEL 9 against a build that
// wants GCC 12's libstdc++ — and is said as such, rather than as a library
// that is plainly there being "not found".
func lddMissing(out string) []string {
seen := map[string]bool{}
var missing []string
for _, line := range strings.Split(out, "\n") {
line = strings.TrimSpace(line)
if !strings.Contains(line, "not found") {
continue
}
var what string
if lib, rest, ok := strings.Cut(line, ": version `"); ok {
sym, _, _ := strings.Cut(rest, "'")
what = filepath.Base(lib) + " is too old (needs " + sym + ")"
} else {
what = strings.TrimSpace(strings.SplitN(line, "=>", 2)[0]) + " is not installed"
}
if !seen[what] {
seen[what] = true
missing = append(missing, what)
}
}
return missing
}
func findFile(dir, name string) (string, error) {
var found string
_ = filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
if err != nil || found != "" {
return nil //nolint:nilerr // an unreadable subtree is not the binary
}
if !d.IsDir() && d.Name() == name {
found = p
}
return nil
})
if found == "" {
return "", fmt.Errorf("the build has no %s in it", name)
}
return found, nil
}
func (in *Installer) progress(st *Status, format string, args ...any) {
st.State, st.Detail = Installing, fmt.Sprintf(format, args...)
in.report(*st)
}
func (in *Installer) report(st Status) {
if in.OnStatus != nil {
in.OnStatus(st)
}
}
func (in *Installer) run(ctx context.Context, env []string, name string, args ...string) ([]byte, error) {
if in.Run != nil {
return in.Run(ctx, env, name, args...)
}
cmd := exec.CommandContext(ctx, name, args...) //nolint:gosec // fixed commands and paths we installed
cmd.Env = append(os.Environ(), env...)
return cmd.CombinedOutput()
}
func (in *Installer) root() string {
if in.Root != "" {
return in.Root
}
return DefaultRoot
}
func (in *Installer) log() *slog.Logger {
if in.Log != nil {
return in.Log
}
return slog.Default()
}
func (in *Installer) client() *http.Client {
if in.HTTP != nil {
return in.HTTP
}
return &http.Client{Timeout: 30 * time.Minute}
}
func (in *Installer) apiBase() string {
if in.APIBase != "" {
return in.APIBase
}
return "https://api.github.com"
}
func (in *Installer) downloadBase() string {
if in.DownloadBase != "" {
return in.DownloadBase
}
return "https://github.com"
}
func lastLine(b []byte) string {
lines := strings.Split(strings.TrimSpace(string(b)), "\n")
return strings.TrimSpace(lines[len(lines)-1])
}
func firstVersionLine(b []byte) string {
for _, l := range strings.Split(string(b), "\n") {
if strings.Contains(strings.ToLower(l), "version") {
return strings.TrimSpace(l)
}
}
return ""
}
+302
View File
@@ -0,0 +1,302 @@
package provision
import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
)
var ubuntu = discovery.Host{OSID: "ubuntu", OSVersion: "24.04", Arch: "amd64", CPUFlags: []string{"avx2"}}
// release is a fake GitHub: one release, its assets and their digests.
type release struct {
assets map[string][]byte
digests map[string]string
gets int
mu sync.Mutex
}
func tarGz(t *testing.T, files map[string]string, links map[string]string) []byte {
t.Helper()
var buf bytes.Buffer
gz := gzip.NewWriter(&buf)
tw := tar.NewWriter(gz)
for name, body := range files {
_ = tw.WriteHeader(&tar.Header{Name: name, Mode: 0o755, Size: int64(len(body)), Typeflag: tar.TypeReg})
_, _ = tw.Write([]byte(body))
}
for name, target := range links {
_ = tw.WriteHeader(&tar.Header{Name: name, Linkname: target, Typeflag: tar.TypeSymlink})
}
_ = tw.Close()
_ = gz.Close()
return buf.Bytes()
}
func (r *release) serve(t *testing.T) *httptest.Server {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
r.mu.Lock()
r.gets++
r.mu.Unlock()
if strings.HasPrefix(req.URL.Path, "/repos/") {
var out struct {
Assets []map[string]string `json:"assets"`
}
for name := range r.assets {
out.Assets = append(out.Assets, map[string]string{"name": name, "digest": r.digests[name]})
}
_ = json.NewEncoder(w).Encode(out)
return
}
body, ok := r.assets[filepath.Base(req.URL.Path)]
if !ok {
http.NotFound(w, req)
return
}
_, _ = w.Write(body)
}))
t.Cleanup(srv.Close)
return srv
}
func newRelease(assets map[string][]byte) *release {
r := &release{assets: assets, digests: map[string]string{}}
for name, b := range assets {
sum := sha256.Sum256(b)
r.digests[name] = "sha256:" + hex.EncodeToString(sum[:])
}
return r
}
// fakeSystem answers the package tools, ldd and the engine itself.
type fakeSystem struct {
mu sync.Mutex
installed map[string]bool
tools map[string]bool
lddOut string
calls []string
}
var errExit = errors.New("exit 1")
func (f *fakeSystem) run(_ context.Context, _ []string, name string, args ...string) ([]byte, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.calls = append(f.calls, name+" "+strings.Join(args, " "))
switch {
case name == "sh" && strings.Contains(args[1], "dpkg-query"),
name == "rpm" && args[0] == "-q":
if f.installed[args[len(args)-1]] {
return nil, nil
}
return nil, errExit
case name == "sh" && strings.Contains(args[1], "command -v"):
if f.tools[args[2]] {
return nil, nil
}
return nil, errExit
case name == "apt-get" || name == "dnf":
for i, a := range args {
if a == "install" {
for _, p := range args[i+1:] {
if !strings.HasPrefix(p, "-") {
f.installed[p] = true
}
}
}
}
return nil, nil
case name == "ldd":
return []byte(f.lddOut), nil
case strings.HasSuffix(name, "llama-server"):
return []byte("version: 10991 (abc)\n"), nil
}
return nil, errors.New("unexpected command " + name)
}
func installer(t *testing.T, srv *httptest.Server, sys *fakeSystem, host discovery.Host) (*Installer, *[]Status) {
var seen []Status
var mu sync.Mutex
return &Installer{
Root: t.TempDir(), Host: host, Run: sys.run,
APIBase: srv.URL, DownloadBase: srv.URL, HTTP: srv.Client(),
OnStatus: func(s Status) { mu.Lock(); seen = append(seen, s); mu.Unlock() },
}, &seen
}
func llamaRelease(t *testing.T) *release {
return newRelease(map[string][]byte{
"llama-b1-bin-ubuntu-x64.tar.gz": tarGz(t,
map[string]string{"llama-b1/llama-server": "#!/bin/sh\n", "llama-b1/libllama.so.0": "x"},
map[string]string{"llama-b1/libllama.so": "libllama.so.0"}),
})
}
func TestAnEngineIsInstalledFromNothing(t *testing.T) {
// A node rebuilt from a bare image: no engine, no OpenMP runtime. What it
// ends with has to have come from the publisher, checked, and been run.
rel := llamaRelease(t)
sys := &fakeSystem{installed: map[string]bool{}}
in, seen := installer(t, rel.serve(t), sys, ubuntu)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if len(got) != 1 || got[0].State != Ready {
t.Fatalf("not ready: %+v", got)
}
if !sys.installed["libgomp1"] {
t.Error("the OpenMP runtime the build links against was not installed")
}
if !strings.HasPrefix(got[0].Binary, in.Root) {
t.Errorf("binary %s is not the one installed under %s", got[0].Binary, in.Root)
}
if _, err := os.Lstat(filepath.Join(filepath.Dir(got[0].Binary), "libllama.so")); err != nil {
t.Errorf("the library symlink was not unpacked: %v", err)
}
var states []string
for _, s := range *seen {
states = append(states, s.State+": "+s.Detail)
}
if !strings.Contains(strings.Join(states, "\n"), "downloading") {
t.Errorf("a download was not reported while it ran:\n%s", strings.Join(states, "\n"))
}
}
func TestAnInstalledEngineIsCheckedNotFetchedAgain(t *testing.T) {
rel := llamaRelease(t)
sys := &fakeSystem{installed: map[string]bool{}}
in, _ := installer(t, rel.serve(t), sys, ubuntu)
w := []Want{{Engine: "llamacpp", Version: "b1"}}
in.Converge(t.Context(), w)
before := rel.gets
got := in.Converge(t.Context(), w)
if got[0].State != Ready {
t.Fatalf("second pass: %+v", got[0])
}
if rel.gets != before {
t.Errorf("a ready engine was downloaded again (%d requests)", rel.gets-before)
}
}
func TestAnEngineThatCannotLoadALibraryIsNotReady(t *testing.T) {
// Found on a node before: llama-server present, executable, and unable to
// start for want of a library — while the node reported ready.
rel := llamaRelease(t)
sys := &fakeSystem{installed: map[string]bool{"libgomp1": true},
lddOut: "\tlibcuda.so.1 => not found\n"}
in, _ := installer(t, rel.serve(t), sys, ubuntu)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if got[0].State != Failed || !strings.Contains(got[0].Detail, "libcuda.so.1") {
t.Fatalf("expected a failure naming the library: %+v", got[0])
}
if _, err := os.Stat(filepath.Join(in.Root, "llamacpp", "b1-cpu", marker)); err == nil {
t.Error("a build that does not run was marked installed")
}
}
func TestADownloadThatDoesNotMatchItsChecksumIsRefused(t *testing.T) {
rel := llamaRelease(t)
for name := range rel.digests {
rel.digests[name] = "sha256:" + strings.Repeat("0", 64)
}
sys := &fakeSystem{installed: map[string]bool{"libgomp1": true}}
in, _ := installer(t, rel.serve(t), sys, ubuntu)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if got[0].State != Failed || !strings.Contains(got[0].Detail, "checksum") {
t.Fatalf("a tampered download was not refused: %+v", got[0])
}
}
func TestAnArchiveCannotWriteOutsideItsDirectory(t *testing.T) {
rel := newRelease(map[string][]byte{
"llama-b1-bin-ubuntu-x64.tar.gz": tarGz(t,
map[string]string{"../../escaped": "x", "llama-server": ""}, nil),
})
sys := &fakeSystem{installed: map[string]bool{"libgomp1": true}}
in, _ := installer(t, rel.serve(t), sys, ubuntu)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if got[0].State != Failed {
t.Fatalf("an archive escaping its directory was installed: %+v", got[0])
}
}
func TestTheBuildFollowsTheHardware(t *testing.T) {
gpu := ubuntu
gpu.NVIDIADriver = "570.86.15"
b, err := chooseLlamaCpp("b1", gpu)
if err != nil || !strings.HasPrefix(b.Variant, "cuda") || len(b.Assets) != 2 {
t.Errorf("an NVIDIA node got %+v, %v; expected the CUDA build and its runtime", b, err)
}
if _, err := chooseLlamaCpp("b1", discovery.Host{OSID: "ubuntu", Arch: "riscv64"}); err == nil {
t.Error("a build was chosen for an architecture llama.cpp does not publish for")
}
slow := ubuntu
slow.CPUFlags = nil
st := (&Installer{Host: slow}).ready(Status{}, Build{CPUFlags: []string{"avx2"}}, "/x/llama-server", "")
if !strings.Contains(st.Detail, "avx2") {
t.Errorf("a CPU without AVX2 was not called out: %q", st.Detail)
}
}
func TestAnEngineTheAgentCannotInstallSaysSo(t *testing.T) {
in := &Installer{Root: t.TempDir(), Host: ubuntu}
got := in.Converge(t.Context(), []Want{{Engine: "vllm", Version: "0.29.0"}})
if got[0].State != Unsupported {
t.Errorf("got %+v", got[0])
}
}
func TestARedHatFamilyNodeInstallsWithItsOwnPackageManager(t *testing.T) {
// The build is not Ubuntu's in any way that matters; the package that
// carries OpenMP's runtime is named differently, and installed by dnf.
rocky := discovery.Host{OSID: "rocky", OSVersion: "10.0", OSLike: []string{"rhel", "centos", "fedora"},
Arch: "amd64", CPUFlags: []string{"avx2"}}
rel := llamaRelease(t)
sys := &fakeSystem{installed: map[string]bool{}, tools: map[string]bool{"dnf": true}}
in, _ := installer(t, rel.serve(t), sys, rocky)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if got[0].State != Ready {
t.Fatalf("not ready: %+v", got[0])
}
if !sys.installed["libgomp"] || sys.installed["libgomp1"] {
t.Errorf("installed %v; expected the rpm name, libgomp", sys.installed)
}
}
func TestASystemTooOldForTheBuildSaysWhatItLacks(t *testing.T) {
// RHEL 9: glibc is new enough, libstdc++ is one GCC release short. ldd
// finds the library and reports the symbol, and "libstdc++ not found"
// would send somebody looking for a package that is installed.
rel := llamaRelease(t)
sys := &fakeSystem{installed: map[string]bool{"libgomp1": true},
lddOut: "\t/lib64/libstdc++.so.6: version `GLIBCXX_3.4.30' not found (required by /x/libllama.so)\n" +
"\t/lib64/libstdc++.so.6: version `GLIBCXX_3.4.30' not found (required by /x/llama-server)\n"}
in, _ := installer(t, rel.serve(t), sys, ubuntu)
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: "b1"}})
if got[0].State != Failed || !strings.Contains(got[0].Detail, "libstdc++.so.6 is too old (needs GLIBCXX_3.4.30)") {
t.Fatalf("detail %q", got[0].Detail)
}
if strings.Count(got[0].Detail, "GLIBCXX") != 1 {
t.Errorf("the same shortfall was listed more than once: %q", got[0].Detail)
}
}
+109
View File
@@ -0,0 +1,109 @@
// Package provision installs the engines a node is asked to have.
//
// The control plane says which engines at which versions; everything else is
// worked out here, on the node, because only the node knows it: which build
// fits its processor and driver, what its operating system is missing, and
// whether the result actually runs. Nothing already on the machine is taken
// on trust. An engine counts as installed when this package installed it,
// checked every file against the checksum its publisher gave, and watched it
// start.
package provision
import (
"fmt"
"strings"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
)
// Build is one published build of an engine, chosen for one machine.
type Build struct {
// Variant names the build: "cpu", "cuda-12.8".
Variant string
// Assets are the release files to fetch, unpacked in order into one
// directory.
Assets []string
// Packages are operating system packages the build links against and a
// minimal install may not carry, by packaging family ("deb", "rpm"): the
// same library is libgomp1 on one and libgomp on the other.
Packages map[string][]string
// Binary is the executable's file name, found anywhere in the unpacked
// tree.
Binary string
// CPUFlags are extensions the build is slow without. Missing ones are
// reported, not refused: slow is still running.
CPUFlags []string
}
// Recipe knows how to install one engine.
type Recipe struct {
// Repo is the GitHub repository whose releases carry the builds.
Repo string
// Choose picks the build for a machine, or says why none fits.
Choose func(version string, h discovery.Host) (Build, error)
}
// Recipes are the engines this agent can install.
//
// vLLM and SGLang are not here yet: they are Python environments of several
// gigabytes rather than an archive, and are installed by hand and named in the
// agent's configuration until they are.
var Recipes = map[string]Recipe{
"llamacpp": {Repo: "ggml-org/llama.cpp", Choose: chooseLlamaCpp},
}
// cudaForLlamaCpp is the CUDA build taken on an NVIDIA node. The release also
// carries a matching cudart archive, so the node needs a driver and nothing
// else — no toolkit.
const cudaForLlamaCpp = "12.8"
func chooseLlamaCpp(version string, h discovery.Host) (Build, error) {
// The published Linux builds are built on Ubuntu, but nothing about them
// is Ubuntu's: they need glibc 2.34 and the libstdc++ of GCC 12, which
// Debian 12 and RHEL 10 have and RHEL 9 does not. So the distribution's
// name decides nothing here. The build is tried everywhere, and ldd says
// whether this system can load it — in the words of the missing symbol.
var arch string
switch h.Arch {
case "amd64":
arch = "x64"
case "arm64":
arch = "arm64"
default:
return Build{}, fmt.Errorf("llama.cpp publishes no Linux build for %s", orUnknown(h.Arch))
}
b := Build{
Variant: "cpu",
Assets: []string{fmt.Sprintf("llama-%s-bin-ubuntu-%s.tar.gz", version, arch)},
// Everything else the build links is in the archive or in any current
// glibc system; OpenMP's runtime is the one a cloud image lacks, and
// the server does not start without it.
Packages: map[string][]string{"deb": {"libgomp1"}, "rpm": {"libgomp"}},
Binary: "llama-server",
}
if arch == "x64" {
// The archive carries a backend per CPU generation and loads the best
// one, down to SSE4.2. Below AVX2 it runs, several times slower.
b.CPUFlags = []string{"avx2"}
}
// The CUDA build is published for x64 at this version; on arm64 it is a
// newer CUDA that the driver may not support, so an arm64 GPU node runs
// the CPU build until someone decides otherwise.
if h.NVIDIADriver != "" && arch == "x64" {
name := fmt.Sprintf("bin-ubuntu-cuda-%s-x64.tar.gz", cudaForLlamaCpp)
b.Variant = "cuda-" + cudaForLlamaCpp
b.Assets = []string{
fmt.Sprintf("llama-%s-%s", version, name),
fmt.Sprintf("cudart-llama-%s-%s", version, name),
}
}
return b, nil
}
func orUnknown(s string) string {
if strings.TrimSpace(s) == "" {
return "an unrecognised system"
}
return s
}
+42 -3
View File
@@ -1,9 +1,11 @@
package agent
import (
"context"
"os"
"slices"
"testing"
"time"
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
)
@@ -60,15 +62,15 @@ func TestAHeartbeatSaysWhenTheEnginesChanged(t *testing.T) {
engines := func(m *llmgwv1.AgentMessage) *llmgwv1.Heartbeat { return m.GetHeartbeat() }
a.mu.Lock()
a.engineVers = map[string]string{"llamacpp": "unknown"}
a.enginesSent = enginesFingerprint(a.engineVers)
a.configuredVers = map[string]string{"llamacpp": "unknown"}
a.enginesSent = enginesFingerprint(a.configuredVers)
a.mu.Unlock()
if hb := engines(a.heartbeat()); hb.GetEnginesChanged() {
t.Errorf("unchanged engines were resent")
}
a.mu.Lock()
a.engineVers = map[string]string{"llamacpp": "unknown", "vllm": "unknown"}
a.configuredVers = map[string]string{"llamacpp": "unknown", "vllm": "unknown"}
a.mu.Unlock()
hb := engines(a.heartbeat())
if !hb.GetEnginesChanged() || len(hb.GetEngineVersions()) != 2 {
@@ -78,3 +80,40 @@ func TestAHeartbeatSaysWhenTheEnginesChanged(t *testing.T) {
t.Errorf("reported again on the next heartbeat")
}
}
func TestWhatTheControlPlaneAsksToInstallIsReportedBack(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)
}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
go a.provisionLoop(ctx)
// An engine this agent cannot install: no network is needed to learn so,
// and the answer has to reach the control plane rather than a log.
a.setProvision(&llmgwv1.Provision{Engines: []*llmgwv1.EngineWant{{Engine: "vllm", Version: "0.29.0"}}})
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
hb := a.heartbeat().GetHeartbeat()
if hb.GetProvisionChanged() {
p := hb.GetProvision()
if len(p) != 1 || p[0].GetState() != "unsupported" {
t.Fatalf("reported %v", p)
}
a.mu.Lock()
_, reported := a.engineReportLocked()["vllm"]
a.mu.Unlock()
if reported {
t.Error("an engine that was not installed is reported as present")
}
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatal("the installation's outcome never reached a heartbeat")
}
+16 -11
View File
@@ -189,13 +189,27 @@ func newNodesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
_ = w.Flush()
// Unhealthy GPUs are usually the answer when a node looks idle but
// nothing schedules onto it.
// nothing schedules onto it. An engine still installing, or one
// that failed to, is the other usual answer.
said := false
for _, n := range nodes {
for _, p := range n.Unhealthy {
fmt.Printf("\n%s: %s (kept out of placement)", n.ID, p)
said = true
}
for _, p := range n.Provision {
if p.State == "ready" {
continue
}
fmt.Printf("\n%s: %s %s %s: %s", n.ID, p.Engine, p.Version, p.State, p.Detail)
said = true
}
for _, w := range n.Warnings {
fmt.Printf("\n%s: %s", n.ID, w)
said = true
}
}
if hasUnhealthy(nodes) {
if said {
fmt.Println()
}
return nil
@@ -478,12 +492,3 @@ func dash(s string) string {
}
return s
}
func hasUnhealthy(nodes []adminapi.NodeView) bool {
for _, n := range nodes {
if len(n.Unhealthy) > 0 {
return true
}
}
return false
}
+2
View File
@@ -401,6 +401,8 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live
}
}
svc := agentsvc.New(ctl.Inventory(), ctl.Reconciler(), requireIdentity, log)
svc.SetEngines(cfg.WantedEngines())
log.Info("nodes install these engines", "engines", cfg.WantedEngines())
// A standby must refuse agents as well as refusing to place. Commands
// travel down the stream the agent opened, so a standby holding one is a
// node the leader cannot command — and the leader's placements would fail
+31
View File
@@ -72,6 +72,15 @@ type Server struct {
// Empty means there is none, and /admin/mirror stays unmounted.
Mirror Mirror `koanf:"mirror"`
// Engines are the engines every node installs, and at which version. The
// agents install them themselves — the right build for their hardware,
// checked against its publisher's checksum and seen to run — so a node is
// never trusted to already have what somebody once put there.
//
// Unset keeps DefaultEngines. A version replaces the default; an empty
// one leaves that engine to be installed by hand.
Engines map[string]string `koanf:"engines"`
Router Router `koanf:"router"`
Control Control `koanf:"control"`
Models []catalog.Model `koanf:"models"`
@@ -542,3 +551,25 @@ func (s Server) ControlPlaneMode() bool {
}
return false
}
// DefaultEngines are the engine versions this release was checked against:
// `llmgw ctl engine-check` accepted every flag llmgw builds for them.
var DefaultEngines = map[string]string{
"llamacpp": "b10991",
}
// WantedEngines is DefaultEngines with the configuration laid over it.
func (s Server) WantedEngines() map[string]string {
out := map[string]string{}
for k, v := range DefaultEngines {
out[k] = v
}
for k, v := range s.Engines {
if v == "" {
delete(out, k)
continue
}
out[k] = v
}
return out
}
+17
View File
@@ -182,3 +182,20 @@ func TestAModelWithProfilesStillMakesAControlPlane(t *testing.T) {
t.Error("a declared model with profiles is still a control plane")
}
}
func TestEnginesDefaultToTheCheckedVersions(t *testing.T) {
var s Server
if got := s.WantedEngines()["llamacpp"]; got != DefaultEngines["llamacpp"] {
t.Errorf("unset config wants llamacpp %q", got)
}
s.Engines = map[string]string{"llamacpp": "b11000", "vllm": "0.29.0"}
got := s.WantedEngines()
if got["llamacpp"] != "b11000" || got["vllm"] != "0.29.0" {
t.Errorf("configured versions not taken: %v", got)
}
// Empty leaves an engine to be installed by hand.
s.Engines = map[string]string{"llamacpp": ""}
if _, ok := s.WantedEngines()["llamacpp"]; ok {
t.Error("an emptied engine is still wanted")
}
}
+18
View File
@@ -13,6 +13,7 @@ import (
"io"
"net/http"
"slices"
"strings"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/bench"
@@ -78,6 +79,15 @@ type NodeView struct {
// which is not the same as having none — placement treats it that way too,
// or a control plane upgrade would empty a fleet.
Engines []string `json:"engines,omitempty"`
// Provision is where installing each wanted engine stands. An engine is in
// Engines only once its installation here is ready.
Provision []inventory.EngineInstall `json:"provision,omitempty"`
// OS is the operating system and architecture, as the agent read them.
OS string `json:"os,omitempty"`
// Warnings are facts about the machine that make it slow or unusable
// without making it unhealthy: a VM with no vector instructions runs every
// model, at a fifth of the speed.
Warnings []string `json:"warnings,omitempty"`
// Models are the weights this node can see. Absent means it was not asked
// to look, which is not the same as holding none.
@@ -230,6 +240,14 @@ func (a *API) Nodes() []NodeView {
v.Engines = append(v.Engines, name)
}
slices.Sort(v.Engines)
v.Provision = n.Provision
if h := n.Host; h.OSID != "" {
v.OS = strings.TrimSpace(h.OSID + " " + h.OSVersion + " " + h.Arch)
if h.Arch == "amd64" && !h.HasCPUFlag("avx2") {
v.Warnings = append(v.Warnings, "no AVX2: inference runs several times slower "+
"(a VM needs CPU type \"host\")")
}
}
for _, w := range n.Weights {
v.Models = append(v.Models, w.Path)
}
+7
View File
@@ -23,8 +23,15 @@ func NodeFromRegister(r *llmgwv1.Register) inventory.Node {
State: inventory.Ready,
AgentVersion: r.GetAgentVersion(),
EngineVersions: r.GetEngineVersions(),
EnginesProbed: r.GetEnginesProbed(),
MemoryTotalMB: int(r.GetMemoryTotalMb()),
}
if h := r.GetHost(); h != nil {
n.Host = discovery.Host{
OSID: h.GetOsId(), OSVersion: h.GetOsVersion(), OSCodename: h.GetOsCodename(),
Arch: h.GetArch(), CPUFlags: h.GetCpuFlags(), NVIDIADriver: h.GetNvidiaDriver(),
}
}
for _, w := range r.GetWeights() {
n.Weights = append(n.Weights, discovery.LocalWeights{
Path: w.GetPath(), SizeBytes: w.GetSizeBytes(), Shared: w.GetShared(),
+47 -1
View File
@@ -49,8 +49,14 @@ type Service struct {
mu sync.RWMutex
conns map[string]*conn
// engines is what every node is told to install, set once at start.
engines map[string]string
}
// SetEngines sets the engines every node installs, by name and version.
func (s *Service) SetEngines(e map[string]string) { s.engines = maps.Clone(e) }
// conn is one agent's open stream.
type conn struct {
nodeID string
@@ -157,6 +163,21 @@ func (s *Service) Connect(
return err
}
// What this node should have installed. Sent on every registration, so a
// node rebuilt from nothing is told the same as one that has been here a
// month, and a changed version reaches every node on its next connect.
if s.engines != nil {
p := &llmgwv1.Provision{}
for _, e := range slices.Sorted(maps.Keys(s.engines)) {
p.Engines = append(p.Engines, &llmgwv1.EngineWant{Engine: e, Version: s.engines[e]})
}
if err := stream.Send(&llmgwv1.ControlMessage{
Msg: &llmgwv1.ControlMessage_Provision{Provision: p},
}); err != nil {
return err
}
}
// Recover replica state from what this node is already running, before any
// reconciliation pass can conclude the cluster is empty and place
// duplicates (plan §6.5).
@@ -272,7 +293,8 @@ func (s *Service) applyHeartbeat(nodeID string, hb *llmgwv1.Heartbeat) {
storage := hb.GetStorage()
memMB := int(hb.GetMemoryTotalMb())
if len(used) == 0 && len(problems) == 0 && len(storage) == 0 &&
!hb.GetWeightsChanged() && memMB == 0 && !hb.GetEnginesChanged() {
!hb.GetWeightsChanged() && memMB == 0 && !hb.GetEnginesChanged() &&
!hb.GetProvisionChanged() {
return
}
@@ -339,6 +361,30 @@ func (s *Service) applyHeartbeat(nodeID string, hb *llmgwv1.Heartbeat) {
}
n.EngineVersions = vers
}
if hb.GetProvisionChanged() {
prev := map[string]inventory.EngineInstall{}
for _, p := range n.Provision {
prev[p.Engine] = p
}
n.Provision = n.Provision[:0]
for _, p := range hb.GetProvision() {
e := inventory.EngineInstall{
Engine: p.GetEngine(), Version: p.GetVersion(), Variant: p.GetVariant(),
State: p.GetState(), Detail: p.GetDetail(),
}
// A state change is worth a line; a download's progress is not.
if old, ok := prev[e.Engine]; !ok || old.State != e.State {
lvl := slog.LevelInfo
if e.State == "failed" || e.State == "unsupported" {
lvl = slog.LevelWarn
}
s.log.Log(context.Background(), lvl, "engine installation on a node",
"node", nodeID, "engine", e.Engine, "version", e.Version,
"variant", e.Variant, "state", e.State, "detail", e.Detail)
}
n.Provision = append(n.Provision, e)
}
}
s.inv.Upsert(n)
}
+18
View File
@@ -102,6 +102,24 @@ type Node struct {
// version decides which metric names and CLI flags apply (plan §14).
AgentVersion string `json:"agent_version,omitempty"`
EngineVersions map[string]string `json:"engine_versions,omitempty"`
// EnginesProbed says EngineVersions is an answer, so an empty map means
// "has none" rather than "an agent too old to say".
EnginesProbed bool `json:"engines_probed,omitempty"`
// Host is the operating system and processor, as the agent reads them.
Host discovery.Host `json:"host"`
// Provision is where installing each wanted engine stands on this node.
Provision []EngineInstall `json:"provision,omitempty"`
}
// EngineInstall is where one engine's installation stands on a node.
type EngineInstall struct {
Engine string `json:"engine"`
Version string `json:"version"`
Variant string `json:"variant,omitempty"`
// State is "installing", "ready", "failed" or "unsupported".
State string `json:"state"`
Detail string `json:"detail,omitempty"`
}
// HasFabric reports whether the node is attached to a fabric.
+17 -2
View File
@@ -403,9 +403,14 @@ func (s *Scheduler) eligibleNodes(p Profile, fabric string, avoid []string, reje
// A node that reports no engines at all is not excluded. That is what
// every agent older than this reports, and emptying a fleet on a
// control plane upgrade is a worse failure than the one being fixed.
if p.Engine != "" && len(n.EngineVersions) > 0 {
//
// An agent that says it probed is believed when it reports nothing:
// that is a node still installing, or one where installing failed,
// and either way not a place to start an engine. sim is built into
// the agent and never installed.
if p.Engine != "" && p.Engine != "sim" && (n.EnginesProbed || len(n.EngineVersions) > 0) {
if _, ok := n.EngineVersions[p.Engine]; !ok {
reject("does not have " + p.Engine + " installed")
reject("does not have " + p.Engine + " installed" + installing(n, p.Engine))
continue
}
}
@@ -526,3 +531,13 @@ func (s *Scheduler) claimedMemory(nodeID string) int {
}
return total
}
// installing says why an engine is missing, when the node has said.
func installing(n inventory.Node, engine string) string {
for _, p := range n.Provision {
if p.Engine == engine && p.Detail != "" {
return " (" + p.State + ": " + p.Detail + ")"
}
}
return ""
}
@@ -797,3 +797,33 @@ func TestAProfileThatDeclaresNothingClaimsNothing(t *testing.T) {
t.Errorf("an undeclared neighbour should not fill the node:\n%s", res.Explain())
}
}
func TestANodeStillInstallingItsEngineIsNotACandidate(t *testing.T) {
// A node rebuilt from nothing registers before its agent has installed
// anything, and reports no engines. Treated as "too old to say", it was a
// candidate for every model at the moment it could run none of them.
fresh := node("fresh", "h100-80g", 8)
fresh.EnginesProbed = true
fresh.Provision = []inventory.EngineInstall{{
Engine: "vllm", Version: "0.29.0", State: "installing", Detail: "downloading",
}}
res := New(store(fresh)).Place([]Profile{{
Name: "tp1", Engine: "vllm", GPUClass: "h100-80g", GPUsPerNode: 1, Nodes: 1,
}}, nil)
if res.Placement != nil {
t.Fatal("placed on a node whose engine is not installed yet")
}
if !strings.Contains(res.Explain(), "installing: downloading") {
t.Errorf("the rejection should say the engine is on its way:\n%s", res.Explain())
}
// The simulated engine is part of the agent, so it is never installed and
// never missing.
res = New(store(fresh)).Place([]Profile{{
Name: "sim", Engine: "sim", GPUClass: "h100-80g", GPUsPerNode: 1, Nodes: 1,
}}, nil)
if res.Placement == nil {
t.Errorf("sim was refused:\n%s", res.Explain())
}
}
+5 -2
View File
@@ -101,7 +101,7 @@ requested whenever a fallback served it.</p>
<h2>Nodes</h2>
{{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>Classes</th><th>Fabrics</th><th class="num">Models</th><th class="num">Replicas</th><th>Last seen</th></tr>
<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 class="state state-{{.State}}">{{.State}}</td>
@@ -111,12 +111,15 @@ requested whenever a fallback served it.</p>
second, and an em dash where nothing was reported: an agent that
was never asked about its disk must not read as a full one. */}}
<td class="num">{{if .DiskTotalGB}}{{.DiskFreeGB}}G/{{.DiskTotalGB}}G{{else}}—{{end}}</td>
<td>{{if .Engines}}{{join .Engines}}{{else}}<span class="note">none yet</span>{{end}}</td>
<td>{{join .Classes}}</td>
<td>{{join .Fabrics}}</td>
<td class="num">{{if .Models}}{{len .Models}}{{else}}—{{end}}</td>
<td class="num">{{.Replicas}}</td>
<td>{{or .LastSeenAgo "—"}}</td>
</tr>{{if .Unhealthy}}<tr><td colspan="9" class="note bad">{{join .Unhealthy}}</td></tr>{{end}}
</tr>{{if .Unhealthy}}<tr><td colspan="11" class="note bad">{{join .Unhealthy}}</td></tr>{{end}}
{{range .Provision}}{{if ne .State "ready"}}<tr><td colspan="11" class="note{{if or (eq .State "failed") (eq .State "unsupported")}} bad{{end}}">{{.Engine}} {{.Version}}{{if .Variant}} ({{.Variant}}){{end}} {{.State}}: {{.Detail}}</td></tr>{{end}}{{end}}
{{range .Warnings}}<tr><td colspan="11" class="note bad">{{.}}</td></tr>{{end}}
{{end}}</table>
{{else}}<p class="empty">No agents have registered.</p>{{end}}
+103
View File
@@ -0,0 +1,103 @@
#!/usr/bin/env python3
"""Put each version's CHANGELOG.md section on its Gitea release.
Gitea materialises a release for every pushed tag and leaves its body empty,
and the release workflow only ever attached binaries — so every release page
was a list of files with nothing saying what had changed. The changelog is
where that is written, and it stays the one place it is written: this copies
it, it does not replace it.
Idempotent: a release whose body already matches is left alone, a version with
no release is skipped, and nothing is created or deleted.
Usage:
GITEA_TOKEN=<token> scripts/gitea/release_notes.py [--tag v0.8.1] [--dry-run]
Without --tag, every version in the changelog is synced. The release workflow
runs it with the tag it just built; running the workflow by hand runs it for
all of them, which is how releases made before this existed get their notes.
"""
import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
DEFAULT_INSTANCE = "https://gitea.mynodes.xyz"
DEFAULT_REPO = "ilker/llmgw"
# "## v0.8.1 — 2026-09-28". The dash is whatever was typed; the version is what
# matters.
HEADING = re.compile(r"^## (v\d+\.\d+\.\d+)\b.*$", re.M)
def sections(changelog):
"""Map each version to the text under its heading, up to the next one."""
out = {}
marks = list(HEADING.finditer(changelog))
for i, m in enumerate(marks):
end = marks[i + 1].start() if i + 1 < len(marks) else len(changelog)
out[m.group(1)] = changelog[m.end():end].strip() + "\n"
return out
def request(method, url, token, payload=None):
data = json.dumps(payload).encode() if payload is not None else None
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Authorization", f"token {token}")
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
body = resp.read()
return json.loads(body) if body else None
except urllib.error.HTTPError as e:
if e.code == 404:
return None
detail = e.read().decode(errors="replace")[:300]
raise SystemExit(f"{method} {url} failed: {e.code} {detail}") from e
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--instance", default=os.environ.get("GITEA_INSTANCE", DEFAULT_INSTANCE))
ap.add_argument("--repo", default=os.environ.get("GITEA_REPO", DEFAULT_REPO))
ap.add_argument("--changelog", default="CHANGELOG.md")
ap.add_argument("--tag", help="sync only this version")
ap.add_argument("--dry-run", action="store_true", help="show what would change")
args = ap.parse_args()
token = os.environ.get("GITEA_TOKEN", "")
if not token and not args.dry_run:
sys.exit("GITEA_TOKEN is not set (use --dry-run to preview)")
with open(args.changelog, encoding="utf-8") as f:
notes = sections(f.read())
if args.tag:
if args.tag not in notes:
# A tag with no section is a release somebody forgot to describe.
# Failing says so where it is seen; an empty page says nothing.
sys.exit(f"{args.changelog} has no section for {args.tag}")
notes = {args.tag: notes[args.tag]}
base = f"{args.instance.rstrip('/')}/api/v1/repos/{args.repo}/releases"
for tag, body in notes.items():
if args.dry_run:
print(f"would set {tag} ({len(body.splitlines())} lines)")
continue
rel = request("GET", f"{base}/tags/{urllib.parse.quote(tag)}", token)
if rel is None:
print(f"skip {tag} (no release)")
continue
if (rel.get("body") or "").strip() == body.strip():
print(f"same {tag}")
continue
request("PATCH", f"{base}/{rel['id']}", token, {"body": body})
print(f"set {tag}")
if __name__ == "__main__":
main()