Fixes from review: setup's memory check, MCP secrets, VS Code, bandwidth
CI / test (push) Successful in 9m47s
CI / test (push) Successful in 9m47s
- server init run again on a machine already set up is not refused its memory (the docs fix a certificate with --host so), and judges count only when one will be installed (judgeComes: not without a mirror, nor from a bundle without it). - A server's sealed secrets that it no longer reads are removed when it is put or deleted; replacing a credential replaces every header it was sent, rather than adding one beside the old. - llmgw.gatewayUrl is a machine setting and restricted in untrusted workspaces: a cloned repository's settings cannot send a sign-in to another gateway. - The bandwidth buffer is at least eight times the CPU's last cache and at most a quarter of free memory, else nothing is measured until there is room; a registration carries the figure measured so far. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
54c6b66cb9
commit
b986ea180d
12 files changed
+197
-20
No files matched your search
@@ -39,6 +39,18 @@ APIs change between minors without a deprecation period.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **`server init` run again on a machine already set up was refused its
|
||||
memory**, as on an 8 GB install fixing its certificate with `--host`, and
|
||||
judges that would not be installed (no mirror) were counted.
|
||||
- **Removing an MCP server left its sealed credential in the store**, and
|
||||
replacing a credential under another header kept the old one sent. A
|
||||
server's secrets now go when it no longer reads them.
|
||||
- **A cloned repository could point the VS Code extension at another
|
||||
gateway.** `llmgw.gatewayUrl` is a machine setting now, never taken from a
|
||||
workspace.
|
||||
- **A node's memory bandwidth could be measured in its cache** on a busy
|
||||
node, and was 0 after a reconnect until the next heartbeat. The buffer is
|
||||
at least eight times the processor's last cache.
|
||||
- **A project kept in the cluster could have its questions embedded outside
|
||||
it.** The similar-question cache now asks the embedding model as the
|
||||
question's project would, external fallbacks refused, and skips a model
|
||||
|
||||
@@ -571,7 +571,9 @@ func (a *Agent) session(ctx context.Context, server string) error {
|
||||
reg.Host = &llmgwv1.Host{
|
||||
OsId: h.OSID, OsVersion: h.OSVersion, OsCodename: h.OSCodename,
|
||||
Arch: h.Arch, CpuFlags: h.CPUFlags, NvidiaDriver: h.NVIDIADriver,
|
||||
MemBandwidthGbs: h.MemBandwidthGBs,
|
||||
// What is measured by now, not what the startup scan found before
|
||||
// the measurement ended: a reconnect would otherwise say 0.
|
||||
MemBandwidthGbs: a.memBandwidth(),
|
||||
}
|
||||
|
||||
if err := stream.Send(&llmgwv1.AgentMessage{
|
||||
|
||||
@@ -3,6 +3,7 @@ package discovery
|
||||
import (
|
||||
"bufio"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -18,19 +19,27 @@ import (
|
||||
// too large for any cache — a VM sharing its host's memory with others gets
|
||||
// its share, which is the number that matters for it.
|
||||
|
||||
// bandwidthBuffer is what is read: well past any processor's cache.
|
||||
const bandwidthBuffer = 512 << 20
|
||||
// bandwidthBuffer is what is read at most; the least is eight times the
|
||||
// processor's last cache, and never under minBuffer. Smaller, it would be
|
||||
// the cache that was measured — some hold a gigabyte — and the figure kept
|
||||
// would be too high.
|
||||
const (
|
||||
bandwidthBuffer = 2 << 30
|
||||
minBuffer = 256 << 20
|
||||
)
|
||||
|
||||
// MeasureMemoryBandwidth is the memory's read rate in GB/s, the best of a
|
||||
// few passes; 0 when the machine has too little free memory to measure.
|
||||
func MeasureMemoryBandwidth() float64 {
|
||||
size := int64(bandwidthBuffer)
|
||||
// Never more than a tenth of what is free: a node already holding models
|
||||
// must not be pushed into swap by a measurement.
|
||||
// Never more than a quarter of what is free: a node already holding
|
||||
// models must not be pushed into swap by a measurement. Without room
|
||||
// for a buffer past the cache, none is measured: the hourly retry
|
||||
// measures it when there is.
|
||||
if avail := memAvailable("/proc/meminfo"); avail > 0 {
|
||||
size = min(size, avail/10)
|
||||
size = min(size, avail/4)
|
||||
}
|
||||
if size < 64<<20 {
|
||||
if size < max(minBuffer, 8*lastCacheBytes("/sys/devices/system/cpu")) {
|
||||
return 0
|
||||
}
|
||||
buf := make([]uint64, size/8)
|
||||
@@ -111,3 +120,27 @@ func memAvailable(path string) int64 {
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// lastCacheBytes is the largest cache any CPU reports, 0 when unknown.
|
||||
func lastCacheBytes(root string) int64 {
|
||||
paths, _ := filepath.Glob(filepath.Join(root, "cpu*", "cache", "index*", "size"))
|
||||
var most int64
|
||||
for _, p := range paths {
|
||||
b, err := os.ReadFile(p) //nolint:gosec // under /sys
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
v := strings.TrimSpace(string(b))
|
||||
mult := int64(1)
|
||||
switch {
|
||||
case strings.HasSuffix(v, "K"):
|
||||
mult, v = 1<<10, strings.TrimSuffix(v, "K")
|
||||
case strings.HasSuffix(v, "M"):
|
||||
mult, v = 1<<20, strings.TrimSuffix(v, "M")
|
||||
}
|
||||
if n, err := strconv.ParseInt(v, 10, 64); err == nil {
|
||||
most = max(most, n*mult)
|
||||
}
|
||||
}
|
||||
return most
|
||||
}
|
||||
@@ -1,6 +1,10 @@
|
||||
package discovery
|
||||
|
||||
import "testing"
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMemoryBandwidthIsMeasured(t *testing.T) {
|
||||
if testing.Short() {
|
||||
@@ -14,3 +18,15 @@ func TestMemoryBandwidthIsMeasured(t *testing.T) {
|
||||
}
|
||||
t.Logf("%.1f GB/s", bw)
|
||||
}
|
||||
|
||||
func TestTheLastCacheIsReadFromSys(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
for path, size := range map[string]string{"cpu0/cache/index0/size": "48K", "cpu0/cache/index3/size": "96M"} {
|
||||
p := filepath.Join(root, path)
|
||||
_ = os.MkdirAll(filepath.Dir(p), 0o755)
|
||||
_ = os.WriteFile(p, []byte(size+"\n"), 0o644)
|
||||
}
|
||||
if got := lastCacheBytes(root); got != 96<<20 {
|
||||
t.Errorf("%d", got)
|
||||
}
|
||||
}
|
||||
@@ -365,12 +365,7 @@ func (in *serverInit) run(ctx context.Context) error {
|
||||
// judgeWanted is whether this install fetches the judge model, and so
|
||||
// whether the baseline policy asks it.
|
||||
func (in *serverInit) judgeWanted() bool {
|
||||
if in.noGuardModel || (in.noMirror && in.mirrorURL == "") || in.unitPath == "" {
|
||||
return false
|
||||
}
|
||||
// Offline, only a judge the bundle carries: a policy asking a judge
|
||||
// that never comes refuses every request.
|
||||
if in.bundle != "" && !in.bundleCarries(guardModelName) {
|
||||
if !in.judgeComes() || in.unitPath == "" {
|
||||
return false
|
||||
}
|
||||
_, err := exec.LookPath("systemctl")
|
||||
@@ -1000,3 +995,13 @@ func virtualInterface(name string) bool {
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// judgeComes is whether this setup has a judge to install: one not left
|
||||
// out, a mirror to fetch it from, and, offline, a bundle that carries it —
|
||||
// a policy asking a judge that never comes refuses every request.
|
||||
func (in *serverInit) judgeComes() bool {
|
||||
if in.noGuardModel || (in.noMirror && in.mirrorURL == "") {
|
||||
return false
|
||||
}
|
||||
return in.bundle == "" || in.bundleCarries(guardModelName)
|
||||
}
|
||||
@@ -176,7 +176,9 @@ func (in *serverInit) servingMemoryMB(judgesElsewhere bool) (mb int, what string
|
||||
gb += in.starterModelSpec().minMemoryGB()
|
||||
parts = append(parts, "the starter model")
|
||||
}
|
||||
if !in.noGuardModel && !judgesElsewhere {
|
||||
// Only judges that will be installed: none without a mirror to fetch
|
||||
// them from, or from a bundle that does not carry one.
|
||||
if !judgesElsewhere && in.judgeComes() {
|
||||
n := in.judgeFor().Replicas
|
||||
gb += n * judgeMemoryGB
|
||||
parts = append(parts, fmt.Sprintf("%d judges", n))
|
||||
@@ -193,6 +195,11 @@ func (in *serverInit) checkMemory() error {
|
||||
if in.unitPath == "" {
|
||||
return nil // a development run, served by nothing
|
||||
}
|
||||
// Run again on a machine already set up — to change --host, say, as
|
||||
// the docs advise — it installs no models it does not already serve.
|
||||
if _, err := os.Stat(in.configPath()); err == nil {
|
||||
return nil
|
||||
}
|
||||
meminfo, err := os.ReadFile("/proc/meminfo")
|
||||
if err != nil {
|
||||
return nil
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/preflight"
|
||||
@@ -24,4 +25,22 @@ func TestTheMemoryAMachineNeedsCountsWhatItServes(t *testing.T) {
|
||||
if mb, _ := (&serverInit{noModel: true, noGuardModel: true}).servingMemoryMB(false); mb != 0 {
|
||||
t.Errorf("nothing served: %d", mb)
|
||||
}
|
||||
// Without a mirror no judge is fetched, so none is counted.
|
||||
if mb, what := (&serverInit{noMirror: true}).servingMemoryMB(false); mb != 0 {
|
||||
t.Errorf("no mirror, nothing to fetch: %d %q", mb, what)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetupRunAgainOnAMachineAlreadySetUpIsNotRefusedItsMemory(t *testing.T) {
|
||||
// The docs fix a certificate with server init --host on a machine set up
|
||||
// before the check: it installs nothing it does not already serve.
|
||||
dir := t.TempDir()
|
||||
cfg := dir + "/server.yaml"
|
||||
if err := os.WriteFile(cfg, []byte("listen: :8080\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
in := &serverInit{unitPath: dir + "/llmgw-server.service", etcDir: dir}
|
||||
if err := in.checkMemory(); err != nil {
|
||||
t.Errorf("refused: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -386,7 +386,7 @@ func (m *consoleModel) mcpCredentialForm(s adminapi.MCPServerView) {
|
||||
return nil, err
|
||||
}
|
||||
return m.call("Credential of "+s.Name, http.MethodPut, mcpPath(s.Name),
|
||||
adminapi.MCPServerRequest{URL: s.URL, Secrets: secrets}, nil), nil
|
||||
adminapi.MCPServerRequest{URL: s.URL, Secrets: secrets, HeaderRefs: map[string]string{}}, nil), nil
|
||||
}})
|
||||
}
|
||||
|
||||
|
||||
+42
-2
@@ -59,7 +59,47 @@ func (s *DB) PutMCPServer(ctx context.Context, m MCPServer) error {
|
||||
rate_per_minute = excluded.rate_per_minute`,
|
||||
m.Name, m.URL, m.Description, string(h), string(r), boolInt(m.Enabled), m.Timeout.Milliseconds(), m.RatePerMinute,
|
||||
m.CreatedAt.UnixMilli(), m.CreatedBy)
|
||||
return err
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.pruneMCPSecrets(ctx, m.Name, m.HeaderRefs)
|
||||
}
|
||||
|
||||
// pruneMCPSecrets removes the server's sealed secrets its headers no longer
|
||||
// read: a credential replaced under another header, or every one of a
|
||||
// server removed, is not left at rest.
|
||||
func (s *DB) pruneMCPSecrets(ctx context.Context, name string, refs map[string]string) error {
|
||||
prefix := "mcp/" + name + "/"
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
`SELECT name FROM cluster_secrets WHERE substr(name, 1, ?) = ?`, len(prefix), prefix)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var stale []string
|
||||
for rows.Next() {
|
||||
var n string
|
||||
if err := rows.Scan(&n); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
used := false
|
||||
for _, ref := range refs {
|
||||
used = used || ref == "store:"+n
|
||||
}
|
||||
if !used {
|
||||
stale = append(stale, n)
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, n := range stale {
|
||||
if _, err := s.db.ExecContext(ctx, `DELETE FROM cluster_secrets WHERE name = ?`, n); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MCPServers are every server registered, by name.
|
||||
@@ -90,7 +130,7 @@ func (s *DB) DeleteMCPServer(ctx context.Context, name string) error {
|
||||
if n, _ := res.RowsAffected(); n == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
return s.pruneMCPSecrets(ctx, name, nil)
|
||||
}
|
||||
|
||||
func scanMCPServer(rows *sql.Rows) (MCPServer, error) {
|
||||
|
||||
@@ -3,6 +3,7 @@ package store
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
@@ -71,3 +72,37 @@ func TestACompanyMCPServerIsKeptReplacedAndRemoved(t *testing.T) {
|
||||
t.Errorf("removing twice: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAServersSecretsGoWhenItNoLongerReadsThem(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s, err := OpenSQLite(ctx, filepath.Join(t.TempDir(), "s.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer s.Close()
|
||||
kept := func(name string) bool {
|
||||
_, err := s.StoredSecret(ctx, name)
|
||||
return err == nil
|
||||
}
|
||||
_ = s.SetClusterSecret(ctx, "mcp/jira/X-Api-Key", "k1")
|
||||
_ = s.SetClusterSecret(ctx, "mcp/jira-old/Authorization", "other server's")
|
||||
if err := s.PutMCPServer(ctx, MCPServer{Name: "jira", URL: "https://j/mcp",
|
||||
HeaderRefs: map[string]string{"X-Api-Key": "store:mcp/jira/X-Api-Key"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !kept("mcp/jira/X-Api-Key") {
|
||||
t.Fatal("a secret the server reads was removed")
|
||||
}
|
||||
// Replaced by a bearer token: the key it was sent before goes.
|
||||
_ = s.SetClusterSecret(ctx, "mcp/jira/Authorization", "Bearer t")
|
||||
_ = s.PutMCPServer(ctx, MCPServer{Name: "jira", URL: "https://j/mcp",
|
||||
HeaderRefs: map[string]string{"Authorization": "store:mcp/jira/Authorization"}})
|
||||
if kept("mcp/jira/X-Api-Key") || !kept("mcp/jira/Authorization") {
|
||||
t.Error("the replaced key is still kept, or the new one is not")
|
||||
}
|
||||
// Removed: its secrets go with it, another server's of a like name stay.
|
||||
_ = s.DeleteMCPServer(ctx, "jira")
|
||||
if kept("mcp/jira/Authorization") || !kept("mcp/jira-old/Authorization") {
|
||||
t.Error("removing jira left its secret, or took jira-old's")
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,11 @@
|
||||
"extensionKind": ["ui", "workspace"],
|
||||
"activationEvents": ["onStartupFinished", "onUri"],
|
||||
"capabilities": {
|
||||
"untrustedWorkspaces": { "supported": true },
|
||||
"untrustedWorkspaces": {
|
||||
"supported": "limited",
|
||||
"description": "Where the gateway is, and so where your key is sent, is never taken from a workspace.",
|
||||
"restrictedConfigurations": ["llmgw.gatewayUrl"]
|
||||
},
|
||||
"virtualWorkspaces": true
|
||||
},
|
||||
"contributes": {
|
||||
@@ -33,6 +37,7 @@
|
||||
"llmgw.gatewayUrl": {
|
||||
"type": "string",
|
||||
"default": "",
|
||||
"scope": "machine",
|
||||
"markdownDescription": "The gateway's address, such as `https://llm.example.com`: the one the people running it gave you, without `/v1`."
|
||||
},
|
||||
"llmgw.showModelsWithoutTools": {
|
||||
|
||||
@@ -210,7 +210,10 @@ func (p *AccountPages) RegisterAdminMCP(mux *http.ServeMux, s MCPAdmin, local fu
|
||||
err = errors.New(l.T("mcp.err.no_secret"))
|
||||
}
|
||||
if err == nil {
|
||||
err = put(r, admin, name, adminapi.MCPServerRequest{Secrets: secrets})
|
||||
// In place of every credential it was sent, not beside them: a
|
||||
// bearer token given where an X-Api-Key was must not leave the
|
||||
// key still sent.
|
||||
err = put(r, admin, name, adminapi.MCPServerRequest{Secrets: secrets, HeaderRefs: map[string]string{}})
|
||||
}
|
||||
done(w, r, admin, "mcp server "+name, l.T("mcp.credential_replaced", name), err)
|
||||
}))
|
||||
|
||||
Reference in new issue
Block a user