Ilker ManapandClaude Opus 5.5
33e678ab84
Seeing what a node did meant ssh through a jump host, to a machine whose address nobody remembers, to read a journal only root may open. When an engine install failed on a new node today, the reason was one line — and it was four steps away from the page that said the install had failed. Each node's row on the cluster page now links to its logs: the agent's own, and every engine running there, chosen from the list the node itself sends. The page can follow, refreshing every five seconds, and is on both the admin listener and /admin. `llmgw ctl logs NODE [--source SHARD] [-n N]` reads the same. They travel over the stream the agent already holds, so they need nothing the control plane does not already have: a request goes down, the lines come back, and a reply that does not come in ten seconds is an agent too old to answer. The agent keeps its last 2000 lines in memory as well as writing them wherever it did before; the engines' last few hundred lines were already kept and sent only when one crashed. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
85 lines
2.0 KiB
Go
85 lines
2.0 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
// agentLogLines is how much of its own log an agent keeps for the admin page.
|
|
// Enough to cover an engine install or a failed start; the journal keeps the
|
|
// rest for anyone who can reach it.
|
|
const agentLogLines = 2000
|
|
|
|
// logRing keeps the most recent lines written to it.
|
|
type logRing struct {
|
|
mu sync.Mutex
|
|
lines []string
|
|
next int
|
|
full bool
|
|
}
|
|
|
|
func newLogRing(n int) *logRing { return &logRing{lines: make([]string, n)} }
|
|
|
|
// Write takes whole lines, as a text handler writes them.
|
|
func (r *logRing) Write(p []byte) (int, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
|
|
r.lines[r.next] = line
|
|
r.next = (r.next + 1) % len(r.lines)
|
|
if r.next == 0 {
|
|
r.full = true
|
|
}
|
|
}
|
|
return len(p), nil
|
|
}
|
|
|
|
// tail returns up to n of the most recent lines, oldest first.
|
|
func (r *logRing) tail(n int) []string {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
size := r.next
|
|
if r.full {
|
|
size = len(r.lines)
|
|
}
|
|
n = min(n, size)
|
|
out := make([]string, 0, n)
|
|
for i := size - n; i < size; i++ {
|
|
idx := i
|
|
if r.full {
|
|
idx = (r.next + i) % len(r.lines)
|
|
}
|
|
out = append(out, r.lines[idx])
|
|
}
|
|
return out
|
|
}
|
|
|
|
// teeHandler sends every record to two handlers: wherever the agent already
|
|
// logs, and the ring the admin page reads.
|
|
type teeHandler struct{ a, b slog.Handler }
|
|
|
|
func (t teeHandler) Enabled(ctx context.Context, l slog.Level) bool {
|
|
return t.a.Enabled(ctx, l) || t.b.Enabled(ctx, l)
|
|
}
|
|
|
|
func (t teeHandler) Handle(ctx context.Context, r slog.Record) error {
|
|
var err error
|
|
if t.a.Enabled(ctx, r.Level) {
|
|
err = t.a.Handle(ctx, r.Clone())
|
|
}
|
|
if t.b.Enabled(ctx, r.Level) {
|
|
_ = t.b.Handle(ctx, r.Clone())
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (t teeHandler) WithAttrs(as []slog.Attr) slog.Handler {
|
|
return teeHandler{t.a.WithAttrs(as), t.b.WithAttrs(as)}
|
|
}
|
|
|
|
func (t teeHandler) WithGroup(name string) slog.Handler {
|
|
return teeHandler{t.a.WithGroup(name), t.b.WithGroup(name)}
|
|
}
|