diff --git a/internal/app/app.go b/internal/app/app.go index 0deb387..bb3090b 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -13,6 +13,7 @@ import ( "github.com/prop4n/proxmops/internal/auth" "github.com/prop4n/proxmops/internal/config" "github.com/prop4n/proxmops/internal/crypt" + "github.com/prop4n/proxmops/internal/logbuf" "github.com/prop4n/proxmops/internal/manifest" "github.com/prop4n/proxmops/internal/proxmox" "github.com/prop4n/proxmops/internal/reconcile" @@ -190,6 +191,11 @@ func deleteManaged(ctx context.Context, desired []manifest.Resource, del cluster } func (a *App) Run(ctx context.Context, addr string) error { + // Tee the daemon's logs into an in-memory ring buffer so the web UI can show + // a live tail; lines still go to the original handler (stdout). + logs := logbuf.New(1000) + a.log = slog.New(logs.Handler(a.log.Handler())) + st, err := store.Open(a.cfg.Server.DatabasePath) if err != nil { return fmt.Errorf("open store: %w", err) @@ -217,6 +223,7 @@ func (a *App) Run(ctx context.Context, addr string) error { Deleter: a, Detailer: a, Events: st, + Logs: logs, CookieSecure: a.cfg.Server.CookieSecure, }, a.log) diff --git a/internal/logbuf/logbuf.go b/internal/logbuf/logbuf.go new file mode 100644 index 0000000..b52610e --- /dev/null +++ b/internal/logbuf/logbuf.go @@ -0,0 +1,120 @@ +// Package logbuf keeps the daemon's most recent log lines in memory and fans +// new ones out to subscribers, so the web UI can show a live tail. It wraps an +// existing slog.Handler rather than replacing it: lines still go to stdout. +package logbuf + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + "time" +) + +// Entry is one captured log line. +type Entry struct { + Time time.Time `json:"time"` + Level string `json:"level"` + Message string `json:"message"` +} + +// Buffer is a bounded ring of recent log entries with live subscribers. It is +// safe for concurrent use. +type Buffer struct { + mu sync.Mutex + entries []Entry + cap int + + subsMu sync.Mutex + subs map[chan Entry]struct{} +} + +// New returns a Buffer retaining up to capacity entries. +func New(capacity int) *Buffer { + if capacity <= 0 { + capacity = 1000 + } + return &Buffer{ + entries: make([]Entry, 0, capacity), + cap: capacity, + subs: make(map[chan Entry]struct{}), + } +} + +// Snapshot returns a copy of the retained entries, oldest first. +func (b *Buffer) Snapshot() []Entry { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]Entry, len(b.entries)) + copy(out, b.entries) + return out +} + +// Subscribe registers a listener for new entries. The channel is buffered; a +// slow consumer drops entries rather than blocking the daemon. The returned +// func unregisters it. +func (b *Buffer) Subscribe() (<-chan Entry, func()) { + ch := make(chan Entry, 64) + b.subsMu.Lock() + b.subs[ch] = struct{}{} + b.subsMu.Unlock() + return ch, func() { + b.subsMu.Lock() + delete(b.subs, ch) + b.subsMu.Unlock() + } +} + +// add appends an entry, dropping the oldest past capacity, and broadcasts it. +func (b *Buffer) add(e Entry) { + b.mu.Lock() + if len(b.entries) == b.cap { + b.entries = append(b.entries[:0], b.entries[1:]...) + } + b.entries = append(b.entries, e) + b.mu.Unlock() + + b.subsMu.Lock() + for ch := range b.subs { + select { + case ch <- e: + default: // drop for a slow consumer + } + } + b.subsMu.Unlock() +} + +// Handler wraps inner so every record is also captured in the buffer. +func (b *Buffer) Handler(inner slog.Handler) slog.Handler { + return &handler{inner: inner, buf: b} +} + +// handler tees slog records into the buffer, then delegates to inner. +type handler struct { + inner slog.Handler + buf *Buffer +} + +func (h *handler) Enabled(ctx context.Context, l slog.Level) bool { + return h.inner.Enabled(ctx, l) +} + +func (h *handler) Handle(ctx context.Context, r slog.Record) error { + var sb strings.Builder + sb.WriteString(r.Message) + r.Attrs(func(a slog.Attr) bool { + fmt.Fprintf(&sb, " %s=%v", a.Key, a.Value) + return true + }) + h.buf.add(Entry{Time: r.Time, Level: r.Level.String(), Message: sb.String()}) + return h.inner.Handle(ctx, r) +} + +func (h *handler) WithAttrs(as []slog.Attr) slog.Handler { + return &handler{inner: h.inner.WithAttrs(as), buf: h.buf} +} + +func (h *handler) WithGroup(name string) slog.Handler { + return &handler{inner: h.inner.WithGroup(name), buf: h.buf} +} diff --git a/internal/logbuf/logbuf_test.go b/internal/logbuf/logbuf_test.go new file mode 100644 index 0000000..22f148d --- /dev/null +++ b/internal/logbuf/logbuf_test.go @@ -0,0 +1,56 @@ +package logbuf + +import ( + "context" + "io" + "log/slog" + "testing" + "time" +) + +func TestBufferCapAndSnapshot(t *testing.T) { + b := New(3) + log := slog.New(b.Handler(slog.NewTextHandler(io.Discard, nil))) + for _, m := range []string{"a", "b", "c", "d"} { + log.Info(m) + } + snap := b.Snapshot() + if len(snap) != 3 { + t.Fatalf("len = %d, want 3 (capped)", len(snap)) + } + // Oldest dropped; order preserved. + if snap[0].Message != "b" || snap[2].Message != "d" { + t.Fatalf("snapshot = %v, want [b c d]", []string{snap[0].Message, snap[1].Message, snap[2].Message}) + } +} + +func TestHandlerCapturesAttrs(t *testing.T) { + b := New(10) + log := slog.New(b.Handler(slog.NewTextHandler(io.Discard, nil))) + log.Info("applied", "action", "create", "vmid", 101) + snap := b.Snapshot() + if len(snap) != 1 { + t.Fatalf("len = %d, want 1", len(snap)) + } + if snap[0].Level != "INFO" || snap[0].Message != "applied action=create vmid=101" { + t.Fatalf("entry = %+v", snap[0]) + } +} + +func TestSubscribeReceivesNewEntries(t *testing.T) { + b := New(10) + log := slog.New(b.Handler(slog.NewTextHandler(io.Discard, nil))) + ch, cancel := b.Subscribe() + defer cancel() + + log.Info("live") + select { + case e := <-ch: + if e.Message != "live" { + t.Fatalf("got %q, want live", e.Message) + } + case <-time.After(time.Second): + t.Fatal("subscriber did not receive the entry") + } + _ = context.Background() +} diff --git a/internal/server/logs.go b/internal/server/logs.go new file mode 100644 index 0000000..2e4a5c3 --- /dev/null +++ b/internal/server/logs.go @@ -0,0 +1,79 @@ +package server + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/prop4n/proxmops/internal/logbuf" +) + +// LogSource provides the daemon's recent log lines and a live feed. +type LogSource interface { + Snapshot() []logbuf.Entry + Subscribe() (<-chan logbuf.Entry, func()) +} + +// handleLogs returns the retained log lines, oldest first. +func (s *Server) handleLogs(w http.ResponseWriter, _ *http.Request) { + if s.logs == nil { + writeJSON(w, http.StatusOK, []logbuf.Entry{}) + return + } + writeJSON(w, http.StatusOK, s.logs.Snapshot()) +} + +// handleLogStream streams new log lines over Server-Sent Events. +func (s *Server) handleLogStream(w http.ResponseWriter, r *http.Request) { + if s.logs == nil { + writeError(w, http.StatusServiceUnavailable, "logs unavailable") + return + } + flusher, ok := w.(http.Flusher) + if !ok { + writeError(w, http.StatusInternalServerError, "streaming unsupported") + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + entries, cancel := s.logs.Subscribe() + defer cancel() + + emit := func(e logbuf.Entry) bool { + data, err := json.Marshal(e) + if err != nil { + return true + } + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(sseWriteTimeout)) + if _, err := fmt.Fprintf(w, "event: log\ndata: %s\n\n", data); err != nil { + return false + } + flusher.Flush() + return true + } + + keepAlive := time.NewTicker(sseKeepAlive) + defer keepAlive.Stop() + + for { + select { + case <-r.Context().Done(): + return + case e := <-entries: + if !emit(e) { + return + } + case <-keepAlive.C: + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(sseWriteTimeout)) + if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil { + return + } + flusher.Flush() + } + } +} diff --git a/internal/server/routes.go b/internal/server/routes.go index 7a7839c..87b7577 100644 --- a/internal/server/routes.go +++ b/internal/server/routes.go @@ -34,6 +34,8 @@ func (s *Server) routes() http.Handler { r.Delete("/resources/{kind}/{name}", s.handleDeleteResource) r.Get("/resources/{kind}/{name}/events", s.handleResourceEvents) r.Get("/events", s.handleEvents) + r.Get("/logs", s.handleLogs) + r.Get("/logs/stream", s.handleLogStream) r.Get("/settings", s.handleGetSettings) r.Put("/settings", s.handlePutSettings) r.Post("/settings/test", s.handleTestSettings) diff --git a/internal/server/server.go b/internal/server/server.go index adaacad..49289c6 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -24,6 +24,7 @@ type Server struct { deleter ResourceDeleter detailer ResourceDetailer events EventsReader + logs LogSource cookieSecure bool } @@ -36,6 +37,7 @@ type Options struct { Deleter ResourceDeleter Detailer ResourceDetailer Events EventsReader + Logs LogSource CookieSecure bool } @@ -49,6 +51,7 @@ func New(opts Options, log *slog.Logger) *Server { deleter: opts.Deleter, detailer: opts.Detailer, events: opts.Events, + logs: opts.Logs, cookieSecure: opts.CookieSecure, } s.http = &http.Server{ diff --git a/web/ui/src/components/layout/AppShell.vue b/web/ui/src/components/layout/AppShell.vue index 93a4156..18555df 100644 --- a/web/ui/src/components/layout/AppShell.vue +++ b/web/ui/src/components/layout/AppShell.vue @@ -1,7 +1,7 @@ + + diff --git a/web/ui/src/router/index.ts b/web/ui/src/router/index.ts index 73749d9..d0344e1 100644 --- a/web/ui/src/router/index.ts +++ b/web/ui/src/router/index.ts @@ -10,6 +10,7 @@ const router = createRouter({ component: () => import('@/components/layout/AppShell.vue'), children: [ { path: '', name: 'dashboard', component: () => import('@/pages/DashboardView.vue'), meta: { requiresAuth: true } }, + { path: 'logs', name: 'logs', component: () => import('@/pages/LogsView.vue'), meta: { requiresAuth: true } }, { path: 'settings', name: 'settings', component: () => import('@/pages/SettingsView.vue'), meta: { requiresAuth: true } }, ], },