v0.83.0: observability pass — always-DEBUG capture ring + GET /debug/logs + heartbeat log-pull + gap-fill sweep

Capture layer: applog.New returns (logger, Ring) — slog fan-out, stderr at the
configured level, ~1000-entry ring fixed at LevelDebug (remote diagnostics
without a config flip). GET /debug/logs (token-authed, ?raw=1) + request-level
DEBUG middleware. Heartbeat log-pull mirrors the report logtail pattern:
envelope log_tail_requested -> next heartbeat carries log_tail (128KB cap,
consume-once, failed-push retry proven). Gap-fill sweep over netverify/
netstorage/netmount/signedjobs/selfupdate/disks/controller-swap/desired/loop.
Red-proofs: ring-at-emit-level FAILs capture test; drain removed FAILs
consume-once; dropped phase line FAILs the S7 log-sequence smoke.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
This commit is contained in:
2026-07-11 16:24:07 +02:00
parent 461eaf42c1
commit cb692f8788
20 changed files with 902 additions and 29 deletions
+204 -9
View File
@@ -1,28 +1,223 @@
// Package log builds the agent's slog logger. Kept tiny on purpose; the agent is
// a host service, so logs go to stderr (journald-friendly). Secrets must never be
// passed to the logger — config is logged only via Config.Redacted (see config).
//
// Observability pass (v0.83.0): New now fans out to TWO handlers — stderr at the
// configured level (journald behavior unchanged) and an in-memory debug Ring fixed
// at LevelDebug. The ring is the remote-diagnostics capture layer: DEBUG detail
// exists for GET /debug/logs and the heartbeat log-pull WITHOUT a config flip.
package log
import (
"context"
"io"
"log/slog"
"os"
"strings"
"sync"
"time"
)
// DefaultRingSize is the debug ring's entry capacity (~1000 lines ≈ a few hours of
// normal operation; the heartbeat tail is additionally byte-capped by its caller).
const DefaultRingSize = 1000
// New returns a text slog.Logger at the given level ("debug"|"info"|"warn"|
// "error"; unknown falls back to info), writing to stderr.
func New(level string) *slog.Logger {
var lvl slog.Level
// "error"; unknown falls back to info) writing to stderr, plus the debug Ring
// that captures EVERY record at LevelDebug regardless of the stderr level.
func New(level string) (*slog.Logger, *Ring) {
return NewWithWriter(os.Stderr, level, DefaultRingSize)
}
// NewWithWriter is the injectable constructor (tests capture the stderr stream).
func NewWithWriter(w io.Writer, level string, ringSize int) (*slog.Logger, *Ring) {
lvl := ParseLevel(level)
ring := NewRing(ringSize)
stderrH := slog.NewTextHandler(w, &slog.HandlerOptions{Level: lvl})
// The ring handler is FIXED at LevelDebug — the capture layer must hold flow
// detail even when the emit level is info (the motivating incident: an info box
// showed nothing about a refused NAS verify because the detail never existed).
ringH := slog.NewTextHandler(ring, &slog.HandlerOptions{Level: slog.LevelDebug})
return slog.New(fanout{stderrH, ringH}), ring
}
// ParseLevel maps a config level string to a slog.Level (unknown → info).
func ParseLevel(level string) slog.Level {
switch strings.ToLower(level) {
case "debug":
lvl = slog.LevelDebug
return slog.LevelDebug
case "warn", "warning":
lvl = slog.LevelWarn
return slog.LevelWarn
case "error":
lvl = slog.LevelError
return slog.LevelError
default:
lvl = slog.LevelInfo
return slog.LevelInfo
}
h := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: lvl})
return slog.New(h)
}
// fanout dispatches one record to every handler whose level admits it. Each
// handler keeps its own level, so the ring can capture DEBUG while stderr stays
// at the configured emit level.
type fanout []slog.Handler
func (f fanout) Enabled(ctx context.Context, lvl slog.Level) bool {
for _, h := range f {
if h.Enabled(ctx, lvl) {
return true
}
}
return false
}
func (f fanout) Handle(ctx context.Context, r slog.Record) error {
var firstErr error
for _, h := range f {
if !h.Enabled(ctx, r.Level) {
continue
}
if err := h.Handle(ctx, r.Clone()); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func (f fanout) WithAttrs(attrs []slog.Attr) slog.Handler {
out := make(fanout, len(f))
for i, h := range f {
out[i] = h.WithAttrs(attrs)
}
return out
}
func (f fanout) WithGroup(name string) slog.Handler {
out := make(fanout, len(f))
for i, h := range f {
out[i] = h.WithGroup(name)
}
return out
}
// Entry is one captured log record, parsed from the slog text line (the port of
// the controller's web.LogBuffer entry, adapted to slog's text format).
type Entry struct {
Time time.Time `json:"timestamp"`
Level string `json:"level"` // DEBUG | INFO | WARN | ERROR
Message string `json:"message"` // msg=… plus attrs, verbatim from the text handler
}
// Ring is a thread-safe fixed-size ring of Entries. It implements io.Writer so a
// slog.TextHandler can feed it (one Write per record — TextHandler writes each
// record as a single line).
type Ring struct {
mu sync.RWMutex
entries []Entry
pos int
full bool
}
// NewRing creates a ring keeping the last size entries (size ≤ 0 → DefaultRingSize).
func NewRing(size int) *Ring {
if size <= 0 {
size = DefaultRingSize
}
return &Ring{entries: make([]Entry, size)}
}
// Write parses one slog text line ("time=… level=… msg=… k=v …") into an Entry.
func (r *Ring) Write(p []byte) (int, error) {
line := strings.TrimRight(string(p), "\r\n")
if line == "" {
return len(p), nil
}
e := parseSlogLine(line)
r.mu.Lock()
r.entries[r.pos] = e
r.pos = (r.pos + 1) % len(r.entries)
if r.pos == 0 && !r.full {
r.full = true
}
r.mu.Unlock()
return len(p), nil
}
// Entries returns up to limit entries in chronological order (newest kept when
// truncating; limit ≤ 0 or > cap → everything held) plus the total held count.
func (r *Ring) Entries(limit int) ([]Entry, int) {
r.mu.RLock()
defer r.mu.RUnlock()
total := r.pos
start := 0
if r.full {
total = len(r.entries)
start = r.pos
}
out := make([]Entry, 0, total)
for i := 0; i < total; i++ {
out = append(out, r.entries[(start+i)%len(r.entries)])
}
if limit > 0 && len(out) > limit {
out = out[len(out)-limit:]
}
return out, total
}
// Lines renders the held entries as plain text lines (chronological), dropping
// from the HEAD (oldest) to honor maxBytes so the newest lines survive — the
// heartbeat log-tail budget (maxBytes ≤ 0 → no byte cap).
func (r *Ring) Lines(maxBytes int) []string {
entries, _ := r.Entries(0)
lines := make([]string, len(entries))
for i, e := range entries {
lines[i] = e.Time.Format(time.RFC3339) + " [" + e.Level + "] " + e.Message
}
if maxBytes <= 0 {
return lines
}
total := 0
start := len(lines)
for i := len(lines) - 1; i >= 0; i-- {
total += len(lines[i]) + 1 // +1 for the newline it represents
if total > maxBytes {
break
}
start = i
}
return lines[start:]
}
// parseSlogLine splits a TextHandler line into time / level / the rest. Attrs and
// the quoted msg stay verbatim in Message — honest, and immune to quoting edge
// cases the viewer doesn't need parsed.
func parseSlogLine(line string) Entry {
e := Entry{Level: "INFO", Message: line, Time: time.Now()}
rest := line
if v, r2, ok := cutField(rest, "time="); ok {
if t, err := time.Parse(time.RFC3339Nano, v); err == nil {
e.Time = t
}
rest = r2
}
if v, r2, ok := cutField(rest, "level="); ok {
switch v {
case "DEBUG", "INFO", "WARN", "ERROR":
e.Level = v
}
rest = r2
}
e.Message = rest
return e
}
// cutField extracts a leading `key=value ` field (unquoted — slog never quotes
// its own time/level values) and returns the value + the remainder.
func cutField(s, key string) (val, rest string, ok bool) {
if !strings.HasPrefix(s, key) {
return "", s, false
}
s = s[len(key):]
if i := strings.IndexByte(s, ' '); i >= 0 {
return s[:i], s[i+1:], true
}
return s, "", true
}
+121
View File
@@ -0,0 +1,121 @@
package log
import (
"bytes"
"fmt"
"strings"
"testing"
)
// S1 capture-at-info (the observability pass's load-bearing property): at emit level
// "info" a DEBUG line reaches the ring AND is absent from the stderr stream. Companion
// red-proof: revert the fan-out (New returning a single stderr handler) → the ring
// misses the DEBUG entry → the first assertion fails.
func TestCaptureAtInfo_RingHoldsDebugStderrDoesNot(t *testing.T) {
var stderr bytes.Buffer
logger, ring := NewWithWriter(&stderr, "info", 50)
logger.Debug("netverify: /proc/mounts verdict", "mounted", false)
logger.Info("netmount: ensured network mount", "name", "nas-media")
entries, total := ring.Entries(0)
if total != 2 {
t.Fatalf("ring holds %d entries, want 2 (DEBUG must be captured at emit level info)", total)
}
if entries[0].Level != "DEBUG" || !strings.Contains(entries[0].Message, "proc/mounts verdict") {
t.Errorf("ring entry 0 = %+v, want the DEBUG verdict line", entries[0])
}
if entries[1].Level != "INFO" {
t.Errorf("ring entry 1 level = %q, want INFO", entries[1].Level)
}
if strings.Contains(stderr.String(), "proc/mounts verdict") {
t.Errorf("stderr contains the DEBUG line at emit level info:\n%s", stderr.String())
}
if !strings.Contains(stderr.String(), "ensured network mount") {
t.Errorf("stderr missing the INFO line:\n%s", stderr.String())
}
}
// At emit level "debug" both sinks carry the line (journald behavior unchanged).
func TestCaptureAtDebug_BothSinks(t *testing.T) {
var stderr bytes.Buffer
logger, ring := NewWithWriter(&stderr, "debug", 50)
logger.Debug("flow detail", "k", "v")
if _, total := ring.Entries(0); total != 1 {
t.Fatalf("ring total = %d, want 1", total)
}
if !strings.Contains(stderr.String(), "flow detail") {
t.Errorf("stderr missing the DEBUG line at emit level debug:\n%s", stderr.String())
}
}
// The ring wraps: with capacity 5 and 8 writes, the NEWEST 5 survive in order.
func TestRing_WrapKeepsNewest(t *testing.T) {
var stderr bytes.Buffer
logger, ring := NewWithWriter(&stderr, "error", 5)
for i := 0; i < 8; i++ {
logger.Info(fmt.Sprintf("line-%d", i))
}
entries, total := ring.Entries(0)
if total != 5 || len(entries) != 5 {
t.Fatalf("total=%d len=%d, want 5", total, len(entries))
}
for i, e := range entries {
want := fmt.Sprintf("line-%d", i+3)
if !strings.Contains(e.Message, want) {
t.Errorf("entry %d = %q, want it to contain %q (chronological, newest kept)", i, e.Message, want)
}
}
}
// Entries(limit) keeps the most recent `limit` entries.
func TestRing_LimitKeepsNewest(t *testing.T) {
logger, ring := NewWithWriter(&bytes.Buffer{}, "error", 10)
for i := 0; i < 6; i++ {
logger.Info(fmt.Sprintf("line-%d", i))
}
entries, _ := ring.Entries(2)
if len(entries) != 2 || !strings.Contains(entries[1].Message, "line-5") || !strings.Contains(entries[0].Message, "line-4") {
t.Errorf("Entries(2) = %+v, want the two newest lines", entries)
}
}
// Lines honors the byte budget by dropping the OLDEST lines (the heartbeat 128 KB cap).
func TestRing_LinesByteBudgetKeepsNewest(t *testing.T) {
logger, ring := NewWithWriter(&bytes.Buffer{}, "error", 10)
for i := 0; i < 5; i++ {
logger.Info(fmt.Sprintf("line-%d %s", i, strings.Repeat("x", 100)))
}
all := ring.Lines(0)
if len(all) != 5 {
t.Fatalf("uncapped lines = %d, want 5", len(all))
}
// Budget for roughly two lines.
budget := len(all[3]) + len(all[4]) + 2
capped := ring.Lines(budget)
if len(capped) >= 5 {
t.Fatalf("byte budget did not truncate: %d lines", len(capped))
}
if !strings.Contains(capped[len(capped)-1], "line-4") {
t.Errorf("newest line missing after byte cap: %q", capped[len(capped)-1])
}
}
// The slog text line parser extracts time + level and preserves the remainder verbatim.
func TestParseSlogLine(t *testing.T) {
e := parseSlogLine(`time=2026-07-11T10:30:00.123+02:00 level=WARN msg="verify failed" code=auth_failed`)
if e.Level != "WARN" {
t.Errorf("level = %q, want WARN", e.Level)
}
if e.Time.Year() != 2026 || e.Time.Minute() != 30 {
t.Errorf("time not parsed: %v", e.Time)
}
if !strings.Contains(e.Message, `msg="verify failed"`) || !strings.Contains(e.Message, "code=auth_failed") {
t.Errorf("message lost content: %q", e.Message)
}
// A non-slog line degrades to INFO with the line verbatim (never dropped).
raw := parseSlogLine("plain text line")
if raw.Level != "INFO" || raw.Message != "plain text line" {
t.Errorf("raw line entry = %+v", raw)
}
}